Another attempt at Ch

[ruslandoga]

TODOs:

• multihost #20 -- sessions might require more work, "With clustered deployments, due to lack of "sticky sessions", you need to be connected to a particular cluster node in order to properly utilize this feature, cause, for example, a round-robin load-balancer won't guarantee that the consequent requests will be processed by the same ClickHouse node." -- from https://clickhouse.com/docs/integrations/rust#session-id
• Experiment to simplify the API #140
• New api #156
• move rowbinary to sql #167
• next encoder #187
• fix: Reconnect on a closed connection when checked out #304
• nimble_options
• telemetry
• something similar to Stream.resource/3
• Collectable?

Three layers:

• Ch.Buffer -- data structure for accumulating rows for INSERT as RowBinary
• Ch.HTTP -- stateless helpers for Mint.HTTP1 with some ClickHouse specifics
• Ch.Pool -- NimblePool (at first) of Mint.HTTP1s

create_deadline = Ch.HTTP.to_deadline(to_timeout(second: 15))

with {:ok, conn} <- Mint.HTTP1.connect(:http, "localhost", 8123, mode: :passive, timeout: Ch.HTTP.to_timeout(create_deadline)) do
  try do
    {path, headers, body} = Ch.HTTP.encode("create table demo(a Int64, b String) engine Null")

    with {:ok, _ref, conn} <- Mint.HTTP1.request(conn, "POST", path, headers, body),
         # a helper to receive full response from a passive socket
         {:ok, response, conn} <- Ch.HTTP.recv_all(conn, create_deadline),
         # a helper to decode response from RowBinary or other known/supported format, convert error to Ch.Error, etc.
         {:ok, %Ch.Result{}} <- Ch.HTTP.decode(response), do: :ok
  after
    Mint.HTTP1.close(conn)
  end
end

rowbinary =
  Ch.Buffer.new(format: "RowBinaryWithNamesAndTypes", columns: [{"a", "Int64"}, {"b", "String"}])
  |> Ch.Buffer.add_row([?a, "a"])
  |> Ch.Buffer.add_row([?b, "b"])
  |> Ch.Buffer.add_rows([[?c, "c"], [?d, "d"]])
  |> Ch.Buffer.to_iodata()

{:ok, pool} = Ch.Pool.start_link(url: "http://localhost:8123", pool_size: 5)
insert = ["insert into demo format RowBinaryWithNamesAndTypes\n" | rowbinary]
Ch.Pool.query!(pool, :zstd.compress(insert), headers: [{"content-encoding", "zstd"}])

Ch.Pool.checkout(pool, fn conn ->
  Ch.HTTP.query!(conn, "select 1", reconnect: true) # idk
end)

[ruslandoga]

At first I thought about streaming compression each time we Ch.Buffer.add_row but it seems to be slower than doing it once in the end. And it complicates the API.

Results

Operating System: macOS
CPU Information: Apple M2
Number of Available Cores: 8
Available memory: 8 GB
Elixir 1.19.5
Erlang 28.3
JIT enabled: true

Benchmark suite executing with the following configuration:
warmup: 2 s
time: 5 s
memory time: 0 ns
reduction time: 0 ns
parallel: 1
inputs: 1 rows, 100,000 rows, 1000 rows
Estimated total run time: 1 min 3 s
Excluding outliers: false

Benchmarking nimble_lz4 once with input 1 rows ...
Benchmarking nimble_lz4 once with input 100,000 rows ...
Benchmarking nimble_lz4 once with input 1000 rows ...
Benchmarking zstd once with input 1 rows ...
Benchmarking zstd once with input 100,000 rows ...
Benchmarking zstd once with input 1000 rows ...
Benchmarking zstd stream with input 1 rows ...
Benchmarking zstd stream with input 100,000 rows ...
Benchmarking zstd stream with input 1000 rows ...
Calculating statistics...
Formatting results...

##### With input 1 rows #####
Name                      ips        average  deviation         median         99th %
nimble_lz4 once      409.21 K        2.44 μs   ±323.54%        2.33 μs        3.04 μs
zstd once            378.85 K        2.64 μs   ±625.79%        2.21 μs        7.08 μs
zstd stream           10.25 K       97.54 μs   ±357.19%       83.13 μs      203.34 μs

Comparison:
nimble_lz4 once      409.21 K
zstd once            378.85 K - 1.08x slower +0.196 μs
zstd stream           10.25 K - 39.91x slower +95.09 μs

##### With input 100,000 rows #####
Name                      ips        average  deviation         median         99th %
nimble_lz4 once         76.57       13.06 ms     ±3.87%       13.02 ms       13.38 ms
zstd once               72.66       13.76 ms     ±3.34%       13.72 ms       14.45 ms
zstd stream             14.45       69.20 ms     ±8.87%       65.38 ms       81.65 ms

Comparison:
nimble_lz4 once         76.57
zstd once               72.66 - 1.05x slower +0.70 ms
zstd stream             14.45 - 5.30x slower +56.14 ms

##### With input 1000 rows #####
Name                      ips        average  deviation         median         99th %
nimble_lz4 once        7.84 K      127.53 μs     ±4.95%      126.25 μs      147.81 μs
zstd once              7.46 K      134.09 μs     ±2.88%      133.58 μs      148.21 μs
zstd stream            1.91 K      524.74 μs    ±16.48%      534.88 μs      809.10 μs

Comparison:
nimble_lz4 once        7.84 K
zstd once              7.46 K - 1.05x slower +6.55 μs
zstd stream            1.91 K - 4.11x slower +397.20 μs

[ruslandoga]

Or rather it can be %Ch.Request{}.

req = Ch.Request.new(statement: ..., params: ..., headers: headers, etc.)
[{"x-clickhouse-format", "RowBinaryWithNamesAndTypes"} | _] = Ch.Request.headers(req)
"/?" <> _ = Ch.Request.path(req)
^statement = Ch.Request.body(req) # idk

[hkrutzer]

Just in case you haven't seen, Finch also has an HTTP1 connection pool that you could potentially copy things from 🙂

[ruslandoga]

Suggested change

	{:ok, query_result} | {:error, query_error}
	:ok | {:ok, query_result} | {:error, query_error}

[ruslandoga]

todo: :ok