@spec cancel_stream(:gen_tcp.socket(), map()) :: :ok | {:error, term()}

Cleans up a stream previously started by start_stream/3, for ChDriver.DBConnection's handle_deallocate/4.

If the stream already reached :end_of_stream (stream.done is true -- the caller consumed every block via stream_fetch/2), there is nothing left in flight and this is a no-op returning :ok immediately.

Otherwise (the caller stopped early, e.g. Enum.take/2 on a partially consumed stream) there is no server-side portal/cursor to close the way Postgres has -- but ClickHouse's native protocol does have a Cancel packet (Client packet type 3): this sends it, then keeps draining and discarding blocks off the same socket/buffer until :end_of_stream (blocks already in flight before the server notices the cancellation can still arrive), so the connection is left in a clean, byte-position-correct state for the next query instead of leaking unread bytes that would desync the very next request sent on this socket.

Returns :ok or {:error, reason}.

@spec close(:gen_tcp.socket()) :: :ok

Closes a socket previously returned by connect/1.

@spec connect(keyword()) ::
  {:ok,
   %{
     socket: :gen_tcp.socket(),
     server_info: ChDriver.Protocol.ServerHello.t(),
     compression: ChDriver.Protocol.Block.Compressed.method()
   }}
  | {:error, term()}

Opens a TCP connection to a ClickHouse server and performs the Hello handshake.

Options

• :hostname - defaults to ~c"localhost"
• :port - defaults to 9000
• :database - default_database to advertise, defaults to "default"
• :username - defaults to "default"
• :password - defaults to ""
• :connect_timeout - TCP connect timeout in ms, defaults to 5_000
• :recv_timeout - :gen_tcp.recv/3 timeout in ms, defaults to 5_000
• :max_buffer_size - maximum accumulated receive-buffer size in bytes before the handshake fails fast with {:error, %ChDriver.Error{}} rather than growing the buffer unbounded, defaults to 67108864 (64MB, matching Postgrex's @max_packet)
• :compression - :none (default) or :lz4. Opt-in wire compression for query/3's outbound/inbound Data blocks -- off by default so existing callers see byte-for-byte unchanged behavior. This is a per-connection default; query/3's own opts can override it per call. See ChDriver.Protocol.Messages.encode_query/2's moduledoc for exactly how this is negotiated with the server and why both directions of block traffic are affected. The only unsupported value besides these two raises ArgumentError rather than silently falling back to uncompressed.

Returns {:ok, %{socket: socket, server_info: %ChDriver.Protocol.ServerHello{}, compression: :none | :lz4}} on success, or {:error, reason} on failure. The caller owns the returned socket and is responsible for closing it (see ChDriver.Connection.close/1).

@spec ping(
  map(),
  keyword()
) :: :ok | {:error, term()}

Sends a Ping packet and waits for the server's Pong. Returns :ok on a successful round-trip or {:error, term} on a socket error or an unexpected response.

@spec query(map(), binary(), keyword()) :: {:ok, map()} | {:error, term()}

Runs query_string against the connection returned by connect/1 and collects the result.

Sends a Query packet followed by an empty Data packet (required even for plain SELECTs -- see ChDriver.Protocol.encode_empty_data_packet/0), then loops reading and dispatching response packets. opts is forwarded to ChDriver.Protocol.encode_query/2 unchanged, so a :params list of {name, raw_text} or {name, raw_text, escape_rounds} tuples binds that query's {name:Type} placeholders (see ChDriver.Params.text/1 and ChDriver.Params.escape_rounds/1).

• Data/ProfileEvents packets are accumulated (columns come from the first Data packet seen with a non-empty column list -- ClickHouse sends an empty "header" block first, then one or more Data blocks with actual rows; ProfileEvents blocks are ignored for the returned result but decoded/consumed so the wire stays in sync).
• Progress packets are ignored (no streaming/partial-result API yet).
• Pong is ignored (only relevant to ping/1, not implemented yet).
• EndOfStream ends the loop successfully.
• Exception ends the loop with {:error, %ChDriver.Error{}}.

opts[:compression] (:none or :lz4) overrides the connection's default (the :compression passed to connect/1, :none if that wasn't given either) for this one query -- see connect/1's docs and ChDriver.Protocol.Messages.encode_query/2's moduledoc for what this negotiates with the server.

Returns {:ok, %{columns: [{name, type}], rows: [[term]]}} or {:error, term} (either a socket error or a %ChDriver.Error{}).

@spec start_stream(map(), binary(), keyword()) :: {:ok, map()} | {:error, term()}

Starts query_string the same way query/3 does (Query packet + empty Data packet), but instead of looping internally to :end_of_stream and accumulating every row into memory, stops after receiving just one block's worth of response -- ClickHouse's own "header" block (0 rows, but with the result's column names/types) -- and returns a small, stateful "stream" map that stream_fetch/2 resumes from on demand.

This is what ChDriver.DBConnection's handle_declare/4 calls: the returned map (%{columns:, buffer:, done:, recv_timeout:, max_buffer_size:, compression:, pending:}) is exactly the "cursor" state that needs to persist across handle_fetch/4 calls -- the socket itself is conn.socket (unchanged, owned by the connection the whole time), buffer is the unconsumed tail of whatever was already read off it, and pending holds the header block's own rows (almost always [], but a tiny result can in principle arrive combined with the header in a single block -- see the moduledoc note on receive_stream_block/6 -- so this is never silently dropped) for the first stream_fetch/2 call to hand back without doing any socket I/O.

Returns {:ok, stream} or {:error, reason} (a socket error or a %ChDriver.Error{}, exactly like query/3).

@spec stream_fetch(:gen_tcp.socket(), map()) ::
  {:cont | :halt, map(), map()} | {:error, term()}

Resumes a stream previously started by start_stream/3 (or a prior call to this function), returning exactly one Data block's worth of rows -- the natural unit the wire protocol already delivers, so there's no artificial row-count-based rebatching here.

If stream.pending is set (the header block's own rows, stashed by start_stream/3 and not yet handed back), returns that directly without touching the socket at all. Otherwise resumes the receive loop from stream.buffer on socket.

Returns {:cont, %{columns:, rows:}, stream} while more blocks remain, {:halt, %{columns:, rows:}, stream} once :end_of_stream is reached (stream.done is true from then on, and every further call returns the same empty :halt immediately), or {:error, reason}.