Packages

caterpillar

0.40.1
0.40.1 0.40.0

Elixir client for ClickHouse, a fast open-source Online Analytical Processing (OLAP) database management system. Forked from https://github.com/balance-platform/pillar to support CH25+

HexDocs

Current section

2 Versions

Jump to
Readme
2 Versions
6 Dependencies
Dependants
Files
Activity
Hide whitespace

Compare versions

16 files changed
+2396 additions
-79 deletions
  @@ -10,8 +10,12 @@
10 10 Elixir client for [ClickHouse](https://clickhouse.tech/), a fast open-source
11 11 Online Analytical Processing (OLAP) database management system.
12 12
13 - # Features
13 + ## Table of Contents
14 14
15 + - [Getting Started](#getting-started)
16 + - [Installation](#installation)
17 + - [Basic Usage](#basic-usage)
18 + - [Features](#features)
15 19 - [Direct Usage with connection structure](#direct-usage-with-connection-structure)
16 20 - [Pool of workers](#pool-of-workers)
17 21 - [Async insert](#async-insert)
  @@ -19,16 +23,70 @@ Online Analytical Processing (OLAP) database management system.
19 23 - [Migrations](#migrations)
20 24 - [DateTime Timezones](#timezones)
21 25 - [Switching between HTTP adapters](#http-adapters)
26 + - [Configuration](#configuration)
27 + - [Connection Options](#connection-options)
28 + - [Pool Configuration](#pool-configuration)
29 + - [HTTP Adapters](#http-adapters)
30 + - [Advanced Usage](#advanced-usage)
31 + - [Bulk Insert Strategies](#bulk-insert-strategies)
32 + - [Custom Type Conversions](#custom-type-conversions)
33 + - [Troubleshooting](#troubleshooting)
34 + - [Common Issues](#common-issues)
35 + - [Performance Optimization](#performance-optimization)
36 + - [Contribution](#contribution)
22 37
23 - ## Usage
38 + ## Getting Started
39 +
40 + ### Installation
41 +
42 + Add `pillar` to your list of dependencies in `mix.exs`:
43 +
44 + ```elixir
45 + def deps do
46 + [
47 + {:caterpillar, "~> 0.40"}
48 + ]
49 + end
50 + ```
51 +
52 + ### Basic Usage
53 +
54 + Here's a simple example to get started with Pillar:
55 +
56 + ```elixir
57 + # Create a direct connection
58 + conn = Pillar.Connection.new("http://user:password@localhost:8123/database")
59 +
60 + # Execute a SELECT query
61 + {:ok, result} = Pillar.select(conn, "SELECT * FROM users LIMIT 10")
62 +
63 + # Execute a parameterized query
64 + {:ok, result} = Pillar.query(conn,
65 + "SELECT * FROM users WHERE created_at > {min_date} LIMIT {limit}",
66 + %{min_date: "2023-01-01", limit: 100}
67 + )
68 +
69 + # Insert data
70 + {:ok, _} = Pillar.insert(conn,
71 + "INSERT INTO events (user_id, event_type, created_at) VALUES ({user_id}, {event_type}, {created_at})",
72 + %{user_id: 123, event_type: "login", created_at: DateTime.utc_now()}
73 + )
74 + ```
75 +
76 + ## Features
77 +
78 + Pillar offers a comprehensive set of features for working with ClickHouse:
24 79
25 80 ### Direct Usage with connection structure
26 81
82 + The most straightforward way to use Pillar is by creating a direct connection to your ClickHouse server:
83 +
27 84 ```elixir
85 + # Create a connection to your ClickHouse server
28 86 conn = Pillar.Connection.new("http://user:password@localhost:8123/database")
29 87
30 - # Params are passed in brackets {} in SQL query, and map strtucture does fill
31 - # query by values.
88 + # Parameters are passed in curly braces {} in the SQL query
89 + # The map structure provides values for these parameters
32 90 sql = "SELECT count(*) FROM users WHERE lastname = {lastname}"
33 91
34 92 params = %{lastname: "Smith"}
  @@ -37,12 +95,28 @@ params = %{lastname: "Smith"}
37 95
38 96 result
39 97 #=> [%{"count(*)" => 347}]
98 + ```
40 99
100 + You can also specify additional connection options:
101 +
102 + ```elixir
103 + conn = Pillar.Connection.new(
104 + "http://user:password@localhost:8123/database",
105 + %{
106 + timeout: 30_000, # Connection timeout in milliseconds
107 + pool_timeout: 5_000, # Pool timeout in milliseconds
108 + request_timeout: 60_000 # Request timeout in milliseconds
109 + }
110 + )
41 111 ```
42 112
43 113 ### Pool of workers
44 114
45 - Recommended usage, because of limited connections and supervised workers.
115 + For production environments, using a connection pool is highly recommended. This approach provides:
116 + - Efficient connection management
117 + - Supervised workers
118 + - Load balancing across multiple ClickHouse servers
119 + - Better performance under high load
46 120
47 121 ```elixir
48 122 defmodule ClickhouseMaster do
  @@ -52,130 +126,516 @@ defmodule ClickhouseMaster do
52 126 "http://user:password@host-master-2:8123/database"
53 127 ],
54 128 name: __MODULE__,
55 - pool_size: 15
129 + pool_size: 15,
130 + pool_timeout: 10_000, # Time to wait for a connection from the pool
131 + timeout: 30_000 # Default query timeout
56 132 end
57 133
134 + # Start the connection pool as part of your supervision tree
58 135 ClickhouseMaster.start_link()
59 136
60 - {:ok, result} = ClickhouseMaster.select(sql, %{param: value})
137 + # Execute queries using the pool
138 + {:ok, result} = ClickhouseMaster.select("SELECT * FROM users WHERE age > {min_age}", %{min_age: 21})
139 + {:ok, _} = ClickhouseMaster.insert("INSERT INTO logs (message) VALUES ({message})", %{message: "User logged in"})
61 140 ```
62 141
142 + The pool automatically manages connection acquisition and release, and can handle multiple ClickHouse servers for load balancing and high availability.
143 +
63 144 ### Async insert
64 145
65 - ```elixir
66 - connection = Pillar.Connection.new("http://user:password@host-master-1:8123/database")
146 + Asynchronous inserts are useful for non-blocking operations when you don't need to wait for a response. This is particularly valuable for:
147 + - Logging events
148 + - Metrics collection
149 + - Any high-volume insert operations where immediate confirmation isn't required
67 150
68 - Pillar.async_insert(connection, "INSERT INTO events (user_id, event) SELECT {user_id}, {event}", %{
69 - user_id: user.id,
70 - event: "password_changed"
71 - }) # => :ok
151 + ```elixir
152 + # Using a connection pool (recommended approach)
153 + ClickhouseMaster.async_insert(
154 + "INSERT INTO events (user_id, event, timestamp) VALUES ({user_id}, {event}, {timestamp})",
155 + %{
156 + user_id: user.id,
157 + event: "password_changed",
158 + timestamp: DateTime.utc_now()
159 + }
160 + ) # => :ok
161 +
162 + # The request is sent and the function returns immediately without waiting for a response
72 163 ```
73 164
165 + Note: Async inserts are only available when using a connection pool created with `use Pillar`. If you attempt to use `Pillar.async_insert/4` directly with a connection structure, it will raise an error.
166 +
74 167 ### Buffer for periodical bulk inserts
75 168
76 - For this feature required [Pool of workers](#pool-of-workers).
169 + The bulk insert buffer feature allows you to collect records in memory and insert them in batches at specified intervals. This is highly efficient for:
170 + - High-frequency event logging
171 + - Metrics collection
172 + - Any scenario where you need to insert many small records
173 +
174 + This feature requires a [Pool of workers](#pool-of-workers) to be set up first.
77 175
78 176 ```elixir
79 177 defmodule BulkToLogs do
80 178 use Pillar.BulkInsertBuffer,
179 + # Reference to your Pillar connection pool
81 180 pool: ClickhouseMaster,
181 +
182 + # Target table for inserts
82 183 table_name: "logs",
83 - # interval_between_inserts_in_seconds, by default -> 5
184 +
185 + # How often to flush buffered records (seconds)
186 + # Default is 5 seconds if not specified
84 187 interval_between_inserts_in_seconds: 5,
85 - # on_errors is optional
86 - on_errors: &__MODULE__.dump_to_file/2
188 +
189 + # Optional error handler function
190 + on_errors: &__MODULE__.dump_to_file/2,
191 +
192 + # Maximum records to buffer before forcing a flush
193 + # Optional, defaults to 10000
194 + max_buffer_size: 5000
87 195
88 196 @doc """
89 - dump to file function store failed inserts into file
197 + Error handler that stores failed inserts into a file
198 +
199 + Parameters:
200 + - result: The error result from ClickHouse
201 + - records: The batch of records that failed to insert
90 202 """
91 203 def dump_to_file(_result, records) do
92 - File.write("bad_inserts/#{DateTime.utc_now()}", inspect(records))
204 + timestamp = DateTime.utc_now() |> DateTime.to_string() |> String.replace(":", "-")
205 + directory = "bad_inserts"
206 +
207 + # Ensure the directory exists
208 + File.mkdir_p!(directory)
209 +
210 + # Write failed records to a file
211 + File.write("#{directory}/#{timestamp}.log", inspect(records, pretty: true))
93 212 end
94 213
95 214 @doc """
96 - retry insert is dangerous (but it is possible and listed as proof of concept)
97 -
98 - this function may be used in `on_errors` option
215 + Alternative error handler that attempts to retry failed inserts
216 +
217 + Note: Retrying can be risky in case of persistent errors
99 218 """
100 219 def retry_insert(_result, records) do
220 + # Add a short delay before retrying
221 + Process.sleep(1000)
101 222 __MODULE__.insert(records)
102 223 end
103 224 end
104 225 ```
105 226
227 + Usage example:
228 +
106 229 ```elixir
230 + # Records are buffered in memory until the flush interval
107 231 :ok = BulkToLogs.insert(%{value: "online", count: 133, datetime: DateTime.utc_now()})
108 232 :ok = BulkToLogs.insert(%{value: "online", count: 134, datetime: DateTime.utc_now()})
109 - :ok = BulkToLogs.insert(%{value: "online", count: 132, datetime: DateTime.utc_now()})
110 - ....
233 + :ok = BulkToLogs.insert(%{value: "offline", count: 42, datetime: DateTime.utc_now()})
111 234
112 - # All this records will be inserted with 5 second interval.
235 + # All these records will be inserted in a single batch after the configured interval (5 seconds by default)
113 236 ```
114 237
115 - *on_errors* parameter allows you to catch any error of bulk insert (for example: one of batch is bad or clickhouse was not available )
238 + The `on_errors` parameter is a callback function that will be invoked if an error occurs during bulk insert. This is useful for:
239 + - Logging failed inserts
240 + - Writing failed records to a backup location
241 + - Implementing custom retry logic
116 242
243 + The callback receives two parameters:
244 + 1. The error result from ClickHouse
245 + 2. The batch of records that failed to insert
117 246
118 247 ### Migrations
119 248
120 - Migrations can be generated with mix task `mix pillar.gen.migration migration_name`.
249 + Pillar provides a migrations system to help you manage your ClickHouse database schema changes in a version-controlled manner. This feature is particularly useful for:
250 + - Creating tables
251 + - Modifying schema
252 + - Ensuring consistent database setup across environments
253 + - Tracking schema changes over time
121 254
122 - Multi-statement migration Example for this [UseCase](https://github.com/balance-platform/pillar/issues/61)
255 + #### Generating Migrations
256 +
257 + Migrations can be generated with the mix task:
258 +
259 + ```bash
260 + mix pillar.gen.migration create_events_table
261 + ```
262 +
263 + This creates a new migration file in `priv/pillar_migrations` with a timestamp prefix, for example:
264 + `priv/pillar_migrations/20250528120000_create_events_table.exs`
265 +
266 + #### Basic Migration Structure
267 +
268 + ```elixir
269 + defmodule Pillar.Migrations.CreateEventsTable do
270 + def up do
271 + """
272 + CREATE TABLE IF NOT EXISTS events (
273 + id UUID,
274 + user_id UInt64,
275 + event_type String,
276 + payload String,
277 + created_at DateTime
278 + ) ENGINE = MergeTree()
279 + ORDER BY (created_at, id)
280 + """
281 + end
282 +
283 + # Optional: Implement a down function for rollbacks
284 + def down do
285 + "DROP TABLE IF EXISTS events"
286 + end
287 + end
288 + ```
289 +
290 + #### Multi-Statement Migrations
291 +
292 + For complex scenarios where you need to execute multiple statements in a single migration, return a list of strings:
123 293
124 294 ```elixir
125 295 defmodule Pillar.Migrations.CreateMultipleTables do
126 296 def up do
127 - # for MultiStatement migration result of this function should be List of Strings
297 + # For multi-statement migrations, return a list of strings
298 + [
299 + """
300 + CREATE TABLE IF NOT EXISTS events (
301 + id UUID,
302 + user_id UInt64,
303 + event_type String,
304 + created_at DateTime
305 + ) ENGINE = MergeTree()
306 + ORDER BY (created_at, id)
307 + """,
308 +
309 + """
310 + CREATE TABLE IF NOT EXISTS event_metrics (
311 + date Date,
312 + event_type String,
313 + count UInt64
314 + ) ENGINE = SummingMergeTree(count)
315 + ORDER BY (date, event_type)
316 + """
317 + ]
318 + end
319 + end
320 + ```
321 +
322 + You can also dynamically generate migrations:
323 +
324 + ```elixir
325 + defmodule Pillar.Migrations.CreateShardedTables do
326 + def up do
327 + # Generate 5 sharded tables
128 328 (0..4) |> Enum.map(fn i ->
129 - "CREATE TABLE IF NOT EXISTS shard_#{i} (field FixedString(10)) ENGINE = Memory"
329 + """
330 + CREATE TABLE IF NOT EXISTS events_shard_#{i} (
331 + id UUID,
332 + user_id UInt64,
333 + event_type String,
334 + created_at DateTime
335 + ) ENGINE = MergeTree()
336 + ORDER BY (created_at, id)
337 + """
130 338 end)
131 339 end
132 340 end
133 341 ```
134 342
135 - ```bash
136 - mix pillar.gen.migration events_table
137 - ```
343 + #### Running Migrations
138 344
139 - But for launching them we have to write own task, like this:
345 + To run migrations, create a mix task:
140 346
141 347 ```elixir
142 348 defmodule Mix.Tasks.MigrateClickhouse do
143 349 use Mix.Task
144 - def run(_args) do
145 - connection_string = Application.get_env(:my_project, :clickhouse_url)
350 +
351 + @shortdoc "Runs ClickHouse migrations"
352 +
353 + def run(args) do
354 + # Start any necessary applications
355 + Application.ensure_all_started(:caterpillar)
356 +
357 + # Parse command line arguments if needed
358 + {opts, _, _} = OptionParser.parse(args, strict: [env: :string])
359 + env = Keyword.get(opts, :env, "dev")
360 +
361 + # Get connection details from your application config
362 + connection_string = Application.get_env(:my_project, String.to_atom("clickhouse_#{env}_url"))
146 363 conn = Pillar.Connection.new(connection_string)
147 - Pillar.Migrations.migrate(conn)
364 +
365 + # Run the migrations
366 + case Pillar.Migrations.migrate(conn) do
367 + :ok ->
368 + Mix.shell().info("Migrations completed successfully")
369 + {:error, reason} ->
370 + Mix.shell().error("Migration failed: #{inspect(reason)}")
371 + exit({:shutdown, 1})
372 + end
148 373 end
149 374 end
150 375 ```
151 376
152 - And launch this via command.
377 + Then run the migrations with:
153 378
154 379 ```bash
155 380 mix migrate_clickhouse
381 + # Or with environment specification:
382 + mix migrate_clickhouse --env=prod
156 383 ```
157 384
385 + #### Migration Tracking
386 +
387 + Pillar automatically tracks applied migrations in a special table named `pillar_migrations` in your ClickHouse database. This table is created automatically and contains:
388 + - The migration version (derived from the timestamp)
389 + - The migration name
390 + - When the migration was applied
391 +
158 392 ### Timezones
159 393
160 - In order to be able to use Timezones add timezones database to your project and configure your app:
394 + Pillar supports timezone-aware DateTime operations when working with ClickHouse. This is particularly important when:
395 + - Storing and retrieving DateTime values
396 + - Performing date/time calculations across different time zones
397 + - Ensuring consistent timestamp handling
398 +
399 + To enable timezone support:
400 +
401 + 1. Add the `tzdata` dependency to your project:
161 402
162 403 ```elixir
404 + defp deps do
405 + [
406 + {:caterpillar, "~> 0.40"},
407 + {:tzdata, "~> 1.1"}
408 + ]
409 + end
410 + ```
411 +
412 + 2. Configure Elixir to use the Tzdata timezone database:
413 +
414 + ```elixir
415 + # In your config/config.exs
163 416 config :elixir, :time_zone_database, Tzdata.TimeZoneDatabase
164 417 ```
165 418
166 - Details here https://hexdocs.pm/elixir/1.12/DateTime.html#module-time-zone-database
419 + 3. Use timezone-aware DateTime functions in your code:
420 +
421 + ```elixir
422 + # Create a timezone-aware DateTime
423 + datetime = DateTime.from_naive!(~N[2023-01-15 10:30:00], "Europe/Berlin")
424 +
425 + # Insert it into ClickHouse
426 + Pillar.insert(conn,
427 + "INSERT INTO events (id, timestamp) VALUES ({id}, {timestamp})",
428 + %{id: 123, timestamp: datetime}
429 + )
430 + ```
431 +
432 + For more details on DateTime and timezone handling in Elixir, see the [official documentation](https://hexdocs.pm/elixir/1.12/DateTime.html#module-time-zone-database).
433 +
434 + ## Configuration
435 +
436 + ### Connection Options
437 +
438 + When creating a new connection with `Pillar.Connection.new/2`, you can specify various options (parameters):
439 +
440 + ```elixir
441 + Pillar.Connection.new(
442 + "http://user:password@localhost:8123/database",
443 + # Params take precedence over URI string
444 +
445 + # Authentication options
446 + user: "default", # Override username in URL
447 + password: "secret", # Override password in URL
448 +
449 + # Timeout options
450 + timeout: 30_000, # Connection timeout in milliseconds
451 +
452 + # ClickHouse specific options
453 + database: "my_database", # Override database in URL
454 + default_format: "JSON", # Default response format
455 +
456 + # Query execution options
457 + max_execution_time: 60, # Maximum query execution time in seconds
458 + max_memory_usage: 10000000, # Maximum memory usage for query in bytes
459 + )
460 + ```
461 +
462 + ### Pool Configuration
463 +
464 + When using a connection pool with `use Pillar`, you can configure the following options:
465 +
466 + ```elixir
467 + defmodule MyApp.ClickHouse do
468 + use Pillar,
469 + # List of ClickHouse servers for load balancing and high availability
470 + connection_strings: [
471 + "http://user:password@clickhouse-1:8123/database",
472 + "http://user:password@clickhouse-2:8123/database"
473 + ],
474 +
475 + # Pool name (defaults to module name if not specified)
476 + name: __MODULE__,
477 +
478 + # Number of connections to maintain in the pool (default: 10)
479 + pool_size: 15,
480 +
481 + # Maximum overflow connections (calculated as pool_size * 0.3 by default)
482 + max_overflow: 5,
483 +
484 + # Time to wait when acquiring a connection from the pool in ms (default: 5000)
485 + pool_timeout: 10_000,
486 +
487 + # Default query timeout in ms (default: 5000)
488 + timeout: 30_000
489 + end
490 + ```
167 491
168 492 ### HTTP Adapters
169 493
170 - If you have problems with default Pillar HTTP Adapter (Mint over Tesla), you can use alternative one, based on :httpc or define your own and pass it
171 - through config.
494 + Pillar provides multiple HTTP adapters for communicating with ClickHouse:
172 495
173 - ```
174 - config :pillar, Pillar.HttpClient, http_adapter: Pillar.HttpClient.TeslaMintAdapter
496 + 1. **TeslaMintAdapter** (default): Uses Tesla with Mint HTTP client
497 + 2. **HttpcAdapter**: Uses Erlang's built-in `:httpc` module
498 +
499 + If you encounter issues with the default adapter, you can switch to an alternative:
500 +
501 + ```elixir
502 + # In your config/config.exs
503 + config :caterpillar, Pillar.HttpClient, http_adapter: Pillar.HttpClient.HttpcAdapter
175 504 ```
176 505
177 - Adapter should define one function `post/3` and return 2 possible results (`%Pillar.HttpClient.Response{}`, `%Pillar.HttpClient.TransportError{}`)
506 + You can also implement your own HTTP adapter by creating a module that implements the `post/3` function and returns either a `%Pillar.HttpClient.Response{}` or a `%Pillar.HttpClient.TransportError{}` struct:
178 507
179 - # Contribution
508 + ```elixir
509 + defmodule MyApp.CustomHttpAdapter do
510 + @behaviour Pillar.HttpClient.Adapter
511 +
512 + @impl true
513 + def post(url, body, options) do
514 + # Implement your custom HTTP client logic
515 + # ...
516 +
517 + # Return a response struct
518 + %Pillar.HttpClient.Response{
519 + body: response_body,
520 + status: 200,
521 + headers: [{"content-type", "application/json"}]
522 + }
523 + end
524 + end
525 +
526 + # Configure Pillar to use your custom adapter
527 + config :caterpillar, Pillar.HttpClient, http_adapter: MyApp.CustomHttpAdapter
528 + ```
529 +
530 + ## Advanced Usage
531 +
532 + ### Bulk Insert Strategies
533 +
534 + Pillar provides several strategies for handling bulk inserts:
535 +
536 + 1. **Direct batch insert**: Insert multiple records in a single query
537 + ```elixir
538 + records = [
539 + %{id: 1, name: "Alice", score: 85},
540 + %{id: 2, name: "Bob", score: 92},
541 + %{id: 3, name: "Charlie", score: 78}
542 + ]
543 +
544 + Pillar.insert_to_table(conn, "students", records)
545 + ```
546 +
547 + 2. **Buffered inserts**: Use `Pillar.BulkInsertBuffer` for timed batch processing
548 + ```elixir
549 + # Define a buffer module as shown in the Buffer section
550 +
551 + # Then insert records that will be buffered and flushed periodically
552 + StudentMetricsBuffer.insert(%{student_id: 123, metric: "login", count: 1})
553 + ```
554 +
555 + 3. **Async inserts**: Use `async_insert` for fire-and-forget operations
556 + ```elixir
557 + ClickHouseMaster.async_insert_to_table("event_logs", %{
558 + event: "page_view",
559 + user_id: 42,
560 + timestamp: DateTime.utc_now()
561 + })
562 + ```
563 +
564 + ### Custom Type Conversions
565 +
566 + Pillar handles type conversions between Elixir and ClickHouse automatically, but you can extend or customize this behavior:
567 +
568 + ```elixir
569 + # Convert a custom Elixir struct to a ClickHouse-compatible format
570 + defimpl Pillar.TypeConvert.ToClickHouse, for: MyApp.User do
571 + def convert(user) do
572 + %{
573 + id: user.id,
574 + name: user.full_name,
575 + email: user.email,
576 + created_at: DateTime.to_iso8601(user.inserted_at)
577 + }
578 + end
579 + end
580 + ```
581 +
582 + ## Troubleshooting
583 +
584 + ### Common Issues
585 +
586 + #### Connection Timeouts
587 +
588 + If you experience connection timeouts, consider:
589 +
590 + 1. Increasing the timeout values:
591 + ```elixir
592 + conn = Pillar.Connection.new(url, %{timeout: 30_000})
593 + ```
594 +
595 + 2. Checking network connectivity between your application and ClickHouse
596 +
597 + 3. Verifying ClickHouse server is running and accepting connections:
598 + ```bash
599 + curl http://clickhouse-server:8123/ping
600 + ```
601 +
602 + #### Memory Limitations
603 +
604 + For large queries that consume significant memory:
605 +
606 + 1. Add query settings to limit memory usage:
607 + ```elixir
608 + Pillar.query(conn, "SELECT * FROM huge_table", %{}, %{
609 + max_memory_usage: 10000000000, # 10GB
610 + max_execution_time: 300 # 5 minutes
611 + })
612 + ```
613 +
614 + 2. Consider using streaming queries (with FORMAT CSV or TabSeparated) for very large result sets
615 +
616 + #### Bulk Insert Failures
617 +
618 + If bulk inserts fail:
619 +
620 + 1. Check your error handler in the `BulkInsertBuffer` configuration
621 + 2. Verify data types match the table schema
622 + 3. Consider reducing batch sizes or increasing the interval between inserts
623 +
624 + ### Performance Optimization
625 +
626 + 1. **Use connection pooling**: Always use a connection pool in production environments
627 +
628 + 2. **Batch inserts**: Group multiple inserts into a single operation when possible
629 +
630 + 3. **Use async operations**: For high-volume inserts where immediate confirmation isn't necessary
631 +
632 + 4. **Query optimization**: Leverage ClickHouse's strengths:
633 + - Use proper table engines based on your query patterns
634 + - Ensure you have appropriate indices
635 + - Filter on columns used in the ORDER BY clause
636 +
637 + 5. **Connection reuse**: Avoid creating new connections for each query
638 +
639 + ## Contribution
180 640
181 641 Feel free to make a pull request. All contributions are appreciated!
  @@ -1,6 +1,6 @@
1 1 {<<"links">>,[{<<"GitHub">>,<<"https://github.com/am-kantox/pillar">>}]}.
2 2 {<<"name">>,<<"caterpillar">>}.
3 - {<<"version">>,<<"0.40.0">>}.
3 + {<<"version">>,<<"0.40.1">>}.
4 4 {<<"description">>,
5 5 <<"Elixir client for ClickHouse, a fast open-source Online Analytical\nProcessing (OLAP) database management system.\n\nForked from https://github.com/balance-platform/pillar to support CH25+">>}.
6 6 {<<"elixir">>,<<"~> 1.9">>}.
  @@ -23,7 +23,11 @@
23 23 <<"lib/pillar/migrations/base.ex">>,<<"lib/pillar/migrations/migrate.ex">>,
24 24 <<"lib/pillar/migrations/generator.ex">>,<<"lib/tasks">>,
25 25 <<"lib/tasks/gen_migration.ex">>,<<"lib/pillar.ex">>,<<".formatter.exs">>,
26 - <<"mix.exs">>,<<"README.md">>]}.
26 + <<"mix.exs">>,<<"README.md">>,<<"stuff/advanced">>,
27 + <<"stuff/advanced/http_adapters.md">>,<<"stuff/advanced/custom_types.md">>,
28 + <<"stuff/guides">>,<<"stuff/guides/migrations.md">>,
29 + <<"stuff/guides/connection_pool.md">>,<<"stuff/guides/bulk_inserts.md">>,
30 + <<"stuff/guides/getting_started.md">>,<<"stuff/troubleshooting.md">>]}.
27 31 {<<"licenses">>,[<<"MIT">>]}.
28 32 {<<"requirements">>,
29 33 [[{<<"name">>,<<"jason">>},
  @@ -1,5 +1,79 @@
1 1 defmodule Pillar do
2 - @moduledoc false
2 + @moduledoc """
3 + Elixir client for [ClickHouse](https://clickhouse.tech/), a fast open-source
4 + Online Analytical Processing (OLAP) database management system.
5 +
6 + ## Overview
7 +
8 + Pillar provides a straightforward way to interact with ClickHouse databases from Elixir,
9 + supporting both direct connections and connection pools. It handles query building,
10 + parameter substitution, and response parsing.
11 +
12 + The library offers the following core features:
13 + - Direct connection to ClickHouse servers
14 + - Connection pooling for improved performance
15 + - Synchronous and asynchronous query execution
16 + - Structured data insertion
17 + - Parameter substitution
18 + - Query result parsing
19 + - Migration support
20 +
21 + ## Direct Usage
22 +
23 + For simple usage with a direct connection:
24 +
25 + ```elixir
26 + # Create a connection
27 + conn = Pillar.Connection.new("http://user:password@localhost:8123/database")
28 +
29 + # Execute a query with parameter substitution
30 + sql = "SELECT count(*) FROM users WHERE lastname = {lastname}"
31 + params = %{lastname: "Smith"}
32 +
33 + {:ok, result} = Pillar.query(conn, sql, params)
34 + # => [%{"count(*)" => 347}]
35 + ```
36 +
37 + ## Connection Pool Usage
38 +
39 + For production applications, using a connection pool is recommended:
40 +
41 + ```elixir
42 + defmodule MyApp.ClickHouse do
43 + use Pillar,
44 + connection_strings: [
45 + "http://user:password@host-1:8123/database",
46 + "http://user:password@host-2:8123/database"
47 + ],
48 + name: __MODULE__,
49 + pool_size: 15
50 + end
51 +
52 + # Start the connection pool
53 + MyApp.ClickHouse.start_link()
54 +
55 + # Execute queries using the pool
56 + {:ok, result} = MyApp.ClickHouse.select("SELECT * FROM users WHERE id = {id}", %{id: 123})
57 + ```
58 +
59 + ## Asynchronous Operations
60 +
61 + For non-blocking inserts:
62 +
63 + ```elixir
64 + # Using direct connection
65 + Pillar.async_insert(conn, "INSERT INTO events (user_id, event) VALUES ({user_id}, {event})", %{
66 + user_id: 42,
67 + event: "login"
68 + })
69 +
70 + # Using connection pool
71 + MyApp.ClickHouse.async_insert("INSERT INTO events (user_id, event) VALUES ({user_id}, {event})", %{
72 + user_id: 42,
73 + event: "login"
74 + })
75 + ```
76 + """
3 77
4 78 alias Pillar.Connection
5 79 alias Pillar.HttpClient
  @@ -9,11 +83,82 @@ defmodule Pillar do
9 83
10 84 @default_timeout_ms 5_000
11 85
86 + @doc """
87 + Executes an INSERT statement against a ClickHouse database.
88 +
89 + ## Parameters
90 +
91 + - `connection` - A `Pillar.Connection` struct representing the database connection
92 + - `query` - The SQL query string with optional parameter placeholders in curly braces `{param}`
93 + - `params` - A map of parameters to be substituted in the query (default: `%{}`)
94 + - `options` - A map of options to customize the request (default: `%{}`)
95 +
96 + ## Options
97 +
98 + - `:timeout` - Request timeout in milliseconds (default: `5000`)
99 + - Other options are passed to the connection's URL builder
100 +
101 + ## Returns
102 +
103 + - `{:ok, result}` on success
104 + - `{:error, reason}` on failure
105 +
106 + ## Examples
107 +
108 + ```elixir
109 + Pillar.insert(conn, "INSERT INTO users (name, email) VALUES ({name}, {email})", %{
110 + name: "John Doe",
111 + email: "john@example.com"
112 + })
113 + ```
114 + """
12 115 def insert(%Connection{} = connection, query, params \\ %{}, options \\ %{}) do
13 116 final_sql = QueryBuilder.query(query, params)
14 117 execute_sql(connection, final_sql, options)
15 118 end
16 119
120 + @doc """
121 + Inserts data into a specified table, automatically generating the INSERT statement.
122 +
123 + This function allows inserting one record (as a map) or multiple records (as a list of maps)
124 + into a ClickHouse table without having to manually construct the SQL INSERT statement.
125 +
126 + ## Parameters
127 +
128 + - `connection` - A `Pillar.Connection` struct representing the database connection
129 + - `table_name` - The name of the table to insert data into
130 + - `record_or_records` - A map or list of maps representing the data to insert
131 + - `options` - A map of options to customize the request (default: `%{}`)
132 +
133 + ## Options
134 +
135 + - `:timeout` - Request timeout in milliseconds (default: `5000`)
136 + - `:query_options` - Options for the query builder (e.g., `%{format: :json}`)
137 +
138 + ## Returns
139 +
140 + - `{:ok, result}` on success
141 + - `{:error, reason}` on failure
142 +
143 + ## Examples
144 +
145 + Single record:
146 + ```elixir
147 + Pillar.insert_to_table(conn, "users", %{
148 + name: "John Doe",
149 + email: "john@example.com",
150 + created_at: DateTime.utc_now()
151 + })
152 + ```
153 +
154 + Multiple records:
155 + ```elixir
156 + Pillar.insert_to_table(conn, "users", [
157 + %{name: "John Doe", email: "john@example.com"},
158 + %{name: "Jane Smith", email: "jane@example.com"}
159 + ])
160 + ```
161 + """
17 162 def insert_to_table(
18 163 %Connection{version: version} = connection,
19 164 table_name,
  @@ -36,11 +181,81 @@ defmodule Pillar do
36 181 execute_sql(connection, final_sql, options)
37 182 end
38 183
184 + @doc """
185 + Executes an arbitrary SQL query against a ClickHouse database.
186 +
187 + This function can be used for any type of query (SELECT, CREATE, ALTER, etc.)
188 + but doesn't format the response as JSON. For SELECT queries with formatted
189 + results, consider using `select/4` instead.
190 +
191 + ## Parameters
192 +
193 + - `connection` - A `Pillar.Connection` struct representing the database connection
194 + - `query` - The SQL query string with optional parameter placeholders in curly braces `{param}`
195 + - `params` - A map of parameters to be substituted in the query (default: `%{}`)
196 + - `options` - A map of options to customize the request (default: `%{}`)
197 +
198 + ## Options
199 +
200 + - `:timeout` - Request timeout in milliseconds (default: `5000`)
201 + - Other options are passed to the connection's URL builder
202 +
203 + ## Returns
204 +
205 + - `{:ok, result}` on success
206 + - `{:error, reason}` on failure
207 +
208 + ## Examples
209 +
210 + ```elixir
211 + Pillar.query(conn, "CREATE TABLE users (id UInt64, name String, email String) ENGINE = MergeTree() ORDER BY id")
212 +
213 + Pillar.query(conn, "ALTER TABLE users ADD COLUMN created_at DateTime")
214 + ```
215 + """
39 216 def query(%Connection{} = connection, query, params \\ %{}, options \\ %{}) do
40 217 final_sql = QueryBuilder.query(query, params)
41 218 execute_sql(connection, final_sql, options)
42 219 end
43 220
221 + @doc """
222 + Executes a SELECT query and returns the results in a structured format.
223 +
224 + This function appends 'FORMAT JSON' to the query, which makes ClickHouse return
225 + the results in JSON format, which are then parsed into Elixir data structures.
226 +
227 + ## Parameters
228 +
229 + - `connection` - A `Pillar.Connection` struct representing the database connection
230 + - `query` - The SQL query string with optional parameter placeholders in curly braces `{param}`
231 + - `params` - A map of parameters to be substituted in the query (default: `%{}`)
232 + - `options` - A map of options to customize the request (default: `%{}`)
233 +
234 + ## Options
235 +
236 + - `:timeout` - Request timeout in milliseconds (default: `5000`)
237 + - Other options are passed to the connection's URL builder
238 +
239 + ## Returns
240 +
241 + - `{:ok, result}` on success, where result is a list of maps representing rows
242 + - `{:error, reason}` on failure
243 +
244 + ## Examples
245 +
246 + ```elixir
247 + {:ok, users} = Pillar.select(conn,
248 + "SELECT id, name, email FROM users WHERE signup_date > {date} LIMIT {limit}",
249 + %{date: "2023-01-01", limit: 100}
250 + )
251 +
252 + # users is a list of maps like:
253 + # [
254 + # %{"id" => 1, "name" => "John Doe", "email" => "john@example.com"},
255 + # %{"id" => 2, "name" => "Jane Smith", "email" => "jane@example.com"}
256 + # ]
257 + ```
258 + """
44 259 def select(%Connection{} = connection, query, params \\ %{}, options \\ %{}) do
45 260 final_sql = QueryBuilder.query(query, params) <> "\n FORMAT JSON"
46 261 execute_sql(connection, final_sql, options)
  @@ -55,6 +270,88 @@ defmodule Pillar do
55 270 |> ResponseParser.parse()
56 271 end
57 272
273 + @doc """
274 + Async version of `insert/4` that doesn't wait for a response.
275 +
276 + This function sends an INSERT query to ClickHouse but doesn't wait for a response,
277 + making it suitable for fire-and-forget operations where you don't need to confirm
278 + the result. Available only when using connection pools through the `use Pillar` macro.
279 + """
280 + def async_insert(%Connection{} = _connection, _query, _params \\ %{}, _options \\ %{}) do
281 + raise "async_insert/4 is only available through a connection pool created with `use Pillar`"
282 + end
283 +
284 + @doc """
285 + Async version of `insert_to_table/4` that doesn't wait for a response.
286 +
287 + This function inserts data into a specified table but doesn't wait for a response,
288 + making it suitable for fire-and-forget operations where you don't need to confirm
289 + the result. Available only when using connection pools through the `use Pillar` macro.
290 + """
291 + def async_insert_to_table(
292 + %Connection{} = _connection,
293 + _table_name,
294 + _record_or_records,
295 + _options \\ %{}
296 + ) do
297 + raise "async_insert_to_table/4 is only available through a connection pool created with `use Pillar`"
298 + end
299 +
300 + @doc """
301 + Async version of `query/4` that doesn't wait for a response.
302 +
303 + This function sends a query to ClickHouse but doesn't wait for a response,
304 + making it suitable for fire-and-forget operations where you don't need to confirm
305 + the result. Available only when using connection pools through the `use Pillar` macro.
306 + """
307 + def async_query(%Connection{} = _connection, _query, _params \\ %{}, _options \\ %{}) do
308 + raise "async_query/4 is only available through a connection pool created with `use Pillar`"
309 + end
310 +
311 + @doc """
312 + Defines a ClickHouse connection pool module.
313 +
314 + This macro sets up a module that manages a pool of ClickHouse connections.
315 + The generated module provides functions to execute queries against the connection pool,
316 + handling connection acquisition and release automatically.
317 +
318 + ## Options
319 +
320 + - `:connection_strings` - Required. A list of ClickHouse connection URLs
321 + - `:name` - Optional. The name of the pool (defaults to "PillarPool")
322 + - `:pool_size` - Optional. The number of connections to maintain (defaults to 10)
323 + - `:pool_timeout` - Optional. Timeout for acquiring a connection from the pool (defaults to 5000ms)
324 + - `:timeout` - Optional. Default query timeout (defaults to 5000ms)
325 +
326 + ## Generated Functions
327 +
328 + The macro generates the following functions in the module:
329 +
330 + - `start_link/1` - Starts the connection pool
331 + - `select/3` - Executes a SELECT query with JSON formatting
332 + - `query/3` - Executes an arbitrary SQL query
333 + - `insert/3` - Executes an INSERT statement
334 + - `insert_to_table/3` - Inserts data into a specified table
335 + - `async_query/3` - Asynchronously executes a query
336 + - `async_insert/3` - Asynchronously executes an INSERT statement
337 + - `async_insert_to_table/3` - Asynchronously inserts data into a specified table
338 +
339 + ## Example
340 +
341 + ```elixir
342 + defmodule MyApp.ClickHouse do
343 + use Pillar,
344 + connection_strings: [
345 + "http://user:password@host-1:8123/database",
346 + "http://user:password@host-2:8123/database"
347 + ],
348 + name: __MODULE__,
349 + pool_size: 15,
350 + pool_timeout: 10_000,
351 + timeout: 30_000
352 + end
353 + ```
354 + """
58 355 defmacro __using__(options) do
59 356 quote do
60 357 use GenServer
  @@ -11,6 +11,7 @@ defmodule Pillar.Connection do
11 11 }
12 12
13 13 @type t :: %{
14 + __struct__: __MODULE__,
14 15 host: String.t(),
15 16 port: integer,
16 17 scheme: String.t(),
  @@ -41,8 +42,8 @@ defmodule Pillar.Connection do
41 42 %Pillar.Connection{} = Pillar.Connection.new("https://localhost:8123")
42 43 ```
43 44 """
44 - @spec new(String.t()) :: t()
45 - def new(str) do
45 + @spec new(url :: String.t(), params :: keyword() | map()) :: t()
46 + def new(str, params \\ %{}) do
46 47 uri = URI.parse(str)
47 48
48 49 info = uri.userinfo
  @@ -54,17 +55,23 @@ defmodule Pillar.Connection do
54 55 :else -> String.split(info, ":")
55 56 end
56 57
57 - params = URI.decode_query(uri.query || "")
58 + query_params = URI.decode_query(uri.query || "")
59 + max_query_size = nil_or_string_to_int(query_params["max_query_size"])
58 60
59 - %__MODULE__{
60 - host: uri.host,
61 - port: uri.port,
62 - scheme: uri.scheme,
63 - database: Path.basename(uri.path || "default"),
64 - user: user,
65 - password: password,
66 - max_query_size: nil_or_string_to_int(params["max_query_size"])
67 - }
61 + data =
62 + %{
63 + host: uri.host,
64 + port: uri.port,
65 + scheme: uri.scheme,
66 + database: Path.basename(uri.path || "default"),
67 + user: user,
68 + password: password,
69 + max_query_size: max_query_size
70 + }
71 + |> Map.merge(Map.new(params))
72 +
73 + __MODULE__
74 + |> struct(data)
68 75 |> add_version()
69 76 end
  @@ -1,11 +1,25 @@
1 1 defmodule Pillar.HttpClient do
2 2 @moduledoc false
3 - @default_http_adapter Pillar.HttpClient.TeslaMintAdapter
4 3
5 - @spec post(String.t(), String.t(), Keyword.t()) ::
6 - Pillar.HttpClient.Response.t()
7 - | Pillar.HttpClient.TransportError.t()
8 - | %{__struct__: RuntimeError, message: String.t()}
4 + defmodule Adapter do
5 + @moduledoc "A behaviour to be implemented by adapters"
6 +
7 + @doc "A callback to be implemented by adapters"
8 + @callback post(url :: String.t(), post_body :: String.t(), options :: keyword()) ::
9 + Pillar.HttpClient.Response.t()
10 + | Pillar.HttpClient.TransportError.t()
11 + | %{__struct__: RuntimeError, message: String.t()}
12 + end
13 +
14 + @default_http_adapter Application.compile_env(
15 + :caterpillar,
16 + :http_adapter,
17 + Pillar.HttpClient.TeslaMintAdapter
18 + )
19 +
20 + @behaviour Adapter
21 +
22 + @impl Adapter
9 23 def post(url, post_body \\ "", options \\ [timeout: 10_000]) do
10 24 http_adapter = adapter()
11 25
  @@ -17,11 +31,9 @@ defmodule Pillar.HttpClient do
17 31 end
18 32
19 33 def adapter do
20 - module =
21 - Application.get_all_env(:pillar)
22 - |> Access.get(__MODULE__, [])
23 - |> Access.get(:http_adapter)
24 -
25 - module || @default_http_adapter
34 + :caterpillar
35 + |> Application.get_env(__MODULE__, [])
36 + |> Keyword.get(:http_adapter)
37 + |> Kernel.||(@default_http_adapter)
26 38 end
27 39 end
Loading more files…