From 677cb85643b023ea5ea647bdd100e7050ebe5530 Mon Sep 17 00:00:00 2001
From: ruslandoga <67764432+ruslandoga@users.noreply.github.com>
Date: Fri, 29 Dec 2023 19:25:45 +0800
Subject: [PATCH 1/9] move data to sql

---
 CHANGELOG.md                 |   5 +
 README.md                    |  92 +++++----
 lib/ch.ex                    |  59 ++++--
 lib/ch/connection.ex         |  39 ++--
 lib/ch/query.ex              | 159 ++++----------
 test/ch/aggregation_test.exs |  73 ++++---
 test/ch/connection_test.exs  | 388 +++++++++++++++++++++--------------
 test/ch/faults_test.exs      |  28 +--
 test/ch/query_test.exs       |  83 +++++---
 9 files changed, 510 insertions(+), 416 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 200604e..f7169b9 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,10 @@
 # Changelog
 
+## Unreleased
+
+- move rows payload (RowBinary, CSV, etc.) to SQL statement
+- remove pseudo-positional binds, make param names explicit
+
 ## 0.2.2 (2023-12-23)
 
 - fix query encoding for datetimes with zeroed microseconds `~U[****-**-** **:**:**.000000]` https://github.com/plausible/ch/pull/138
diff --git a/README.md b/README.md
index 4f55203..e6121cf 100644
--- a/README.md
+++ b/README.md
@@ -85,7 +85,7 @@ Ch.query!(pid, "CREATE TABLE IF NOT EXISTS ch_demo(id UInt64) ENGINE Null")
   Ch.query!(pid, "INSERT INTO ch_demo(id) SELECT number FROM system.numbers LIMIT {limit:UInt8}", %{"limit" => 2})
 ```
 
-#### Insert rows as [RowBinary](https://clickhouse.com/docs/en/interfaces/formats#rowbinary) (efficient)
+#### Insert [RowBinary](https://clickhouse.com/docs/en/interfaces/formats#rowbinary)
 
 ```elixir
 {:ok, pid} = Ch.start_link()
@@ -98,8 +98,11 @@ types = [Ch.Types.u64()]
 # or
 types = [:u64]
 
+rows = [[0], [1]]
+row_binary = Ch.RowBinary.encode_rows(rows, types)
+
 %Ch.Result{num_rows: 2} =
-  Ch.query!(pid, "INSERT INTO ch_demo(id) FORMAT RowBinary", [[0], [1]], types: types)
+  Ch.query!(pid, ["INSERT INTO ch_demo(id) FORMAT RowBinary\n" | row_binary])
 ```
 
 Note that RowBinary format encoding requires `:types` option to be provided.
@@ -107,11 +110,14 @@ Note that RowBinary format encoding requires `:types` option to be provided.
 Similarly, you can use [`RowBinaryWithNamesAndTypes`](https://clickhouse.com/docs/en/interfaces/formats#rowbinarywithnamesandtypes) which would additionally do something like a type check.
 
 ```elixir
-sql = "INSERT INTO ch_demo FORMAT RowBinaryWithNamesAndTypes"
-opts = [names: ["id"], types: ["UInt64"]]
-rows = [[0], [1]]
+names = ["id"]
+types = ["UInt64"]
+
+header = Ch.RowBinary.encode_names_and_types(names, types)
+row_binary = Ch.RowBinary.encode_rows(rows, types)
 
-%Ch.Result{num_rows: 2} = Ch.query!(pid, sql, rows, opts)
+%Ch.Result{num_rows: 3} =
+  Ch.query!(pid, ["INSERT INTO ch_demo FORMAT RowBinaryWithNamesAndTypes\n", header | row_binary])
 ```
 
 #### Insert rows in custom [format](https://clickhouse.com/docs/en/interfaces/formats)
@@ -121,29 +127,42 @@ rows = [[0], [1]]
 
 Ch.query!(pid, "CREATE TABLE IF NOT EXISTS ch_demo(id UInt64) ENGINE Null")
 
-csv = [0, 1] |> Enum.map(&to_string/1) |> Enum.intersperse(?\n)
+csv = "0\n1"
 
 %Ch.Result{num_rows: 2} =
-  Ch.query!(pid, "INSERT INTO ch_demo(id) FORMAT CSV", csv, encode: false)
+  Ch.query!(pid, ["INSERT INTO ch_demo(id) FORMAT CSV\n" | csv])
 ```
 
-#### Insert rows as chunked RowBinary stream
+#### Insert rows as [chunked](https://en.wikipedia.org/wiki/Chunked_transfer_encoding) RowBinary stream
 
 ```elixir
 {:ok, pid} = Ch.start_link()
 
 Ch.query!(pid, "CREATE TABLE IF NOT EXISTS ch_demo(id UInt64) ENGINE Null")
 
-stream = Stream.repeatedly(fn -> [:rand.uniform(100)] end)
-chunked = Stream.chunk_every(stream, 100)
-encoded = Stream.map(chunked, fn chunk -> Ch.RowBinary.encode_rows(chunk, _types = ["UInt64"]) end)
-ten_encoded_chunks = Stream.take(encoded, 10)
+row_binary =
+  Stream.repeatedly(fn -> [:rand.uniform(100)] end)
+  |> Stream.chunk_every(100_000)
+  |> Stream.map(fn chunk -> Ch.RowBinary.encode_rows(chunk, _types = ["UInt64"]) end)
+  |> Stream.take(10)
 
-%Ch.Result{num_rows: 1000} =
-  Ch.query(pid, "INSERT INTO ch_demo(id) FORMAT RowBinary", ten_encoded_chunks, encode: false)
+%Ch.Result{num_rows: 1_000_000} =
+  Ch.query(pid, Stream.concat(["INSERT INTO ch_demo(id) FORMAT RowBinary\n"], row_binary))
 ```
 
-This query makes a [`transfer-encoding: chunked`](https://en.wikipedia.org/wiki/Chunked_transfer_encoding) HTTP request while unfolding the stream resulting in lower memory usage.
+#### Insert rows via [input](https://clickhouse.com/docs/en/sql-reference/table-functions/input) function
+
+```elixir
+{:ok, pid} = Ch.start_link()
+
+Ch.query!(pid, "CREATE TABLE IF NOT EXISTS ch_demo(id UInt64) ENGINE Null")
+
+sql = "INSERT INTO ch_demo SELECT id + {ego:Int64} FROM input('id UInt64') FORMAT RowBinary\n"
+row_binary = Ch.RowBinary.encode_rows([[1], [2], [3]], ["UInt64"])
+
+%Ch.Result{num_rows: 3} =
+  Ch.query!(pid, [sql | row_binary], %{"ego" => -1})
+```
 
 #### Query with custom [settings](https://clickhouse.com/docs/en/operations/settings/settings)
 
@@ -156,7 +175,7 @@ settings = [async_insert: 1]
   Ch.query!(pid, "SHOW SETTINGS LIKE 'async_insert'")
 
 %Ch.Result{rows: [["async_insert", "Bool", "1"]]} =
-  Ch.query!(pid, "SHOW SETTINGS LIKE 'async_insert'", [], settings: settings)
+  Ch.query!(pid, "SHOW SETTINGS LIKE 'async_insert'", _params = [], settings: settings)
 ```
 
 ## Caveats
@@ -179,13 +198,13 @@ CREATE TABLE ch_nulls (
 """)
 
 types = ["Nullable(UInt8)", "UInt8", "UInt8"]
-inserted_rows = [[nil, nil, nil]]
-selected_rows = [[nil, 0, 0]]
+rows = [[nil, nil, nil]]
+row_binary = Ch.RowBinary.encode_rows(rows, types)
 
 %Ch.Result{num_rows: 1} =
-  Ch.query!(pid, "INSERT INTO ch_nulls(a, b, c) FORMAT RowBinary", inserted_rows, types: types)
+  Ch.query!(pid, ["INSERT INTO ch_nulls(a, b, c) FORMAT RowBinary\n" | row_binary])
 
-%Ch.Result{rows: ^selected_rows} =
+%Ch.Result{rows: [[nil, 0, 0]]} =
   Ch.query!(pid, "SELECT * FROM ch_nulls")
 ```
 
@@ -197,13 +216,17 @@ However, [`input()`](https://clickhouse.com/docs/en/sql-reference/table-function
 sql = """
 INSERT INTO ch_nulls
   SELECT * FROM input('a Nullable(UInt8), b Nullable(UInt8), c UInt8')
-  FORMAT RowBinary\
+  FORMAT RowBinary
 """
 
-Ch.query!(pid, sql, inserted_rows, types: ["Nullable(UInt8)", "Nullable(UInt8)", "UInt8"])
+types = ["Nullable(UInt8)", "Nullable(UInt8)", "UInt8"]
+row_binary = Ch.RowBinary.encode_rows(rows, types)
+
+%Ch.Result{num_rows: 1} =
+  Ch.query!(pid, [sql | row_binary])
 
-%Ch.Result{rows: [[0], [10]]} =
-  Ch.query!(pid, "SELECT b FROM ch_nulls ORDER BY b")
+%Ch.Result{rows: [_before = [nil, 0, 0], _after = [nil, 10, 0]]} =
+  Ch.query!(pid, "SELECT * FROM ch_nulls ORDER BY b")
 ```
 
 #### UTF-8 in RowBinary
@@ -215,26 +238,19 @@ When decoding [`String`](https://clickhouse.com/docs/en/sql-reference/data-types
 
 Ch.query!(pid, "CREATE TABLE ch_utf8(str String) ENGINE Memory")
 
-bin = "\x61\xF0\x80\x80\x80b"
-utf8 = "a�b"
+# "\x61\xF0\x80\x80\x80b" will become "a�b" on SELECT
+row_binary = Ch.RowBinary.encode(:string, "\x61\xF0\x80\x80\x80b")
 
 %Ch.Result{num_rows: 1} =
-  Ch.query!(pid, "INSERT INTO ch_utf8(str) FORMAT RowBinary", [[bin]], types: ["String"])
+  Ch.query!(pid, ["INSERT INTO ch_utf8(str) FORMAT RowBinary\n" | row_binary])
 
-%Ch.Result{rows: [[^utf8]]} =
+%Ch.Result{rows: [["a�b"]]} =
   Ch.query!(pid, "SELECT * FROM ch_utf8")
 
-%Ch.Result{rows: %{"data" => [[^utf8]]}} =
+%Ch.Result{rows: %{"data" => [["a�b"]]}} =
   pid |> Ch.query!("SELECT * FROM ch_utf8 FORMAT JSONCompact") |> Map.update!(:rows, &Jason.decode!/1)
 ```
 
-To get raw binary from `String` columns use `:binary` type that skips UTF-8 checks.
-
-```elixir
-%Ch.Result{rows: [[^bin]]} =
-  Ch.query!(pid, "SELECT * FROM ch_utf8", [], types: [:binary])
-```
-
 #### Timezones in RowBinary
 
 Decoding non-UTC datetimes like `DateTime('Asia/Taipei')` requires a [timezone database.](https://hexdocs.pm/elixir/DateTime.html#module-time-zone-database)
@@ -268,7 +284,7 @@ utc = DateTime.utc_now()
 taipei = DateTime.shift_zone!(utc, "Asia/Taipei")
 
 # ** (ArgumentError) non-UTC timezones are not supported for encoding: 2023-04-26 01:49:43.044569+08:00 CST Asia/Taipei
-Ch.query!(pid, "INSERT INTO ch_datetimes(datetime) FORMAT RowBinary", [[naive], [utc], [taipei]], types: ["DateTime"])
+Ch.RowBinary.encode_rows([[naive], [utc], [taipei]], ["DateTime"])
 ```
 
 ## Benchmarks
diff --git a/lib/ch.ex b/lib/ch.ex
index a027102..0a8ac52 100644
--- a/lib/ch.ex
+++ b/lib/ch.ex
@@ -2,49 +2,85 @@ defmodule Ch do
   @moduledoc "Minimal HTTP ClickHouse client."
   alias Ch.{Connection, Query, Result}
 
+  @type common_option ::
+          {:database, String.t()}
+          | {:username, String.t()}
+          | {:password, String.t()}
+          | {:settings, Keyword.t()}
+          | {:timeout, timeout}
+
+  @type start_option ::
+          common_option
+          | {:scheme, String.t()}
+          | {:hostname, String.t()}
+          | {:port, :inet.port_number()}
+          | {:transport_opts, :gen_tcp.connect_option()}
+          | DBConnection.start_option()
+
   @doc """
   Start the connection process and connect to ClickHouse.
 
   ## Options
 
+    * `:scheme` - HTTP scheme, defaults to `"http"`
     * `:hostname` - server hostname, defaults to `"localhost"`
     * `:port` - HTTP port, defualts to `8123`
-    * `:scheme` - HTTP scheme, defaults to `"http"`
+    * `:transport_opts` - options to be given to the transport being used. See `Mint.HTTP1.connect/4` for more info
     * `:database` - Database, defaults to `"default"`
     * `:username` - Username
     * `:password` - User password
     * `:settings` - Keyword list of ClickHouse settings
     * `:timeout` - HTTP receive timeout in milliseconds
-    * `:transport_opts` - options to be given to the transport being used. See `Mint.HTTP1.connect/4` for more info
+    * [`DBConnection.start_option()`](https://hexdocs.pm/db_connection/DBConnection.html#t:start_option/0)
 
   """
+  @spec start_link([start_option]) :: GenServer.on_start()
   def start_link(opts \\ []) do
     DBConnection.start_link(Connection, opts)
   end
 
   @doc """
   Returns a supervisor child specification for a DBConnection pool.
+
+  See `start_link/1` for supported options.
   """
+  @spec child_spec([start_option]) :: :supervisor.child_spec()
   def child_spec(opts) do
     DBConnection.child_spec(Connection, opts)
   end
 
+  # TODO move streaming to Ch.stream/4
+  @type statement :: iodata | Enumerable.t()
+  @type params :: %{String.t() => term} | [{String.t(), term}]
+
+  @type query_option ::
+          common_option
+          | {:command, Ch.Query.command()}
+          | {:headers, [{String.t(), String.t()}]}
+          | {:format, String.t()}
+          | {:decode, boolean}
+          | DBConnection.connection_option()
+
   @doc """
   Runs a query and returns the result as `{:ok, %Ch.Result{}}` or
   `{:error, Exception.t()}` if there was a database error.
 
   ## Options
 
-    * `:timeout` - Query request timeout
-    * `:settings` - Keyword list of settings
     * `:database` - Database
     * `:username` - Username
     * `:password` - User password
+    * `:settings` - Keyword list of settings
+    * `:timeout` - Query request timeout
+    * `:command` - Command tag for the query
+    * `:headers` - Custom HTTP headers for the request
+    * `:format` - Custom response format for the request
+    * `:decode` - Whether to automatically decode the response
+    * [`DBConnection.connection_option()`](https://hexdocs.pm/db_connection/DBConnection.html#t:connection_option/0)
 
   """
-  @spec query(DBConnection.conn(), iodata, params, Keyword.t()) ::
+  @spec query(DBConnection.conn(), statement, params, [query_option]) ::
           {:ok, Result.t()} | {:error, Exception.t()}
-        when params: map | [term] | [row :: [term]] | iodata | Enumerable.t()
   def query(conn, statement, params \\ [], opts \\ []) do
     query = Query.build(statement, opts)
 
@@ -57,26 +93,19 @@ defmodule Ch do
   Runs a query and returns the result or raises `Ch.Error` if
   there was an error. See `query/4`.
   """
-  @spec query!(DBConnection.conn(), iodata, params, Keyword.t()) :: Result.t()
-        when params: map | [term] | [row :: [term]] | iodata | Enumerable.t()
+  @spec query!(DBConnection.conn(), statement, params, [query_option]) :: Result.t()
   def query!(conn, statement, params \\ [], opts \\ []) do
     query = Query.build(statement, opts)
     DBConnection.execute!(conn, query, params, opts)
   end
 
   @doc false
-  @spec stream(DBConnection.t(), iodata, map | [term], Keyword.t()) :: DBConnection.Stream.t()
+  @spec stream(DBConnection.t(), statement, params, [query_option]) :: DBConnection.Stream.t()
   def stream(conn, statement, params \\ [], opts \\ []) do
     query = Query.build(statement, opts)
     DBConnection.stream(conn, query, params, opts)
   end
 
-  @doc false
-  @spec run(DBConnection.conn(), (DBConnection.t() -> any), Keyword.t()) :: any
-  def run(conn, f, opts \\ []) when is_function(f, 1) do
-    DBConnection.run(conn, f, opts)
-  end
-
   if Code.ensure_loaded?(Ecto.ParameterizedType) do
     @behaviour Ecto.ParameterizedType
 
diff --git a/lib/ch/connection.ex b/lib/ch/connection.ex
index 6c1b352..ad8ea4b 100644
--- a/lib/ch/connection.ex
+++ b/lib/ch/connection.ex
@@ -127,17 +127,17 @@ defmodule Ch.Connection do
   end
 
   @impl true
-  def handle_execute(%Query{command: :insert} = query, params, opts, conn) do
-    {query_params, extra_headers, body} = params
+  def handle_execute(%Query{statement: statement} = query, params, opts, conn) do
+    {query_params, extra_headers} = params
 
     path = path(conn, query_params, opts)
     headers = headers(conn, extra_headers, opts)
 
     result =
-      if is_function(body, 2) do
-        request_chunked(conn, "POST", path, headers, body, opts)
+      if is_list(statement) or is_binary(statement) do
+        request(conn, "POST", path, headers, statement, opts)
       else
-        request(conn, "POST", path, headers, body, opts)
+        request_chunked(conn, "POST", path, headers, statement, opts)
       end
 
     with {:ok, conn, responses} <- result do
@@ -145,17 +145,6 @@ defmodule Ch.Connection do
     end
   end
 
-  def handle_execute(query, params, opts, conn) do
-    {query_params, extra_headers, body} = params
-
-    path = path(conn, query_params, opts)
-    headers = headers(conn, extra_headers, opts)
-
-    with {:ok, conn, responses} <- request(conn, "POST", path, headers, body, opts) do
-      {:ok, query, responses, conn}
-    end
-  end
-
   @impl true
   def disconnect(_error, conn) do
     {:ok = ok, _conn} = HTTP.close(conn)
@@ -164,7 +153,14 @@ defmodule Ch.Connection do
 
   @typep response :: Mint.Types.status() | Mint.Types.headers() | binary
 
-  @spec request(conn, binary, binary, Mint.Types.headers(), iodata, Keyword.t()) ::
+  @spec request(
+          conn,
+          method :: String.t(),
+          path :: String.t(),
+          Mint.Types.headers(),
+          body :: iodata,
+          [Ch.query_option()]
+        ) ::
           {:ok, conn, [response]}
           | {:error, Error.t(), conn}
           | {:disconnect, Mint.Types.error(), conn}
@@ -174,7 +170,14 @@ defmodule Ch.Connection do
     end
   end
 
-  @spec request_chunked(conn, binary, binary, Mint.Types.headers(), Enumerable.t(), Keyword.t()) ::
+  @spec request_chunked(
+          conn,
+          method :: String.t(),
+          path :: String.t(),
+          Mint.Types.headers(),
+          body :: Enumerable.t(),
+          [Ch.query_option()]
+        ) ::
           {:ok, conn, [response]}
           | {:error, Error.t(), conn}
           | {:disconnect, Mint.Types.error(), conn}
diff --git a/lib/ch/query.ex b/lib/ch/query.ex
index bac1004..65b3ad8 100644
--- a/lib/ch/query.ex
+++ b/lib/ch/query.ex
@@ -1,16 +1,15 @@
 defmodule Ch.Query do
   @moduledoc "Query struct wrapping the SQL statement."
-  defstruct [:statement, :command, :encode, :decode]
+  defstruct [:statement, :command]
 
-  @type t :: %__MODULE__{statement: iodata, command: atom, encode: boolean, decode: boolean}
+  @type t :: %__MODULE__{statement: Ch.statement(), command: command}
+  @type params :: [{String.t(), String.t()}]
 
   @doc false
-  @spec build(iodata, Keyword.t()) :: t
+  @spec build(Ch.statement(), [Ch.query_option()]) :: t
   def build(statement, opts \\ []) do
     command = Keyword.get(opts, :command) || extract_command(statement)
-    encode = Keyword.get(opts, :encode, true)
-    decode = Keyword.get(opts, :decode, true)
-    %__MODULE__{statement: statement, command: command, encode: encode, decode: decode}
+    %__MODULE__{statement: statement, command: command}
   end
 
   statements = [
@@ -43,6 +42,13 @@ defmodule Ch.Query do
     {"WATCH", :watch}
   ]
 
+  command_union =
+    statements
+    |> Enum.map(fn {_, command} -> command end)
+    |> Enum.reduce(&{:|, [], [&1, &2]})
+
+  @type command :: unquote(command_union)
+
   defp extract_command(statement)
 
   for {statement, command} <- statements do
@@ -54,8 +60,8 @@ defmodule Ch.Query do
     extract_command(rest)
   end
 
-  defp extract_command([first_segment | _] = statement) do
-    extract_command(first_segment) || extract_command(IO.iodata_to_binary(statement))
+  defp extract_command([first_segment | _]) do
+    extract_command(first_segment)
   end
 
   defp extract_command(_other), do: nil
@@ -64,118 +70,46 @@ end
 defimpl DBConnection.Query, for: Ch.Query do
   alias Ch.{Query, Result, RowBinary}
 
-  @spec parse(Query.t(), Keyword.t()) :: Query.t()
+  @spec parse(Query.t(), [Ch.query_option()]) :: Query.t()
   def parse(query, _opts), do: query
 
-  @spec describe(Query.t(), Keyword.t()) :: Query.t()
+  @spec describe(Query.t(), [Ch.query_option()]) :: Query.t()
   def describe(query, _opts), do: query
 
-  @spec encode(Query.t(), params, Keyword.t()) :: {query_params, Mint.Types.headers(), body}
-        when params: map | [term] | [row :: [term]] | iodata | Enumerable.t(),
-             query_params: [{String.t(), String.t()}],
-             body: iodata | Enumerable.t()
-
-  def encode(%Query{command: :insert, encode: false, statement: statement}, data, opts) do
-    body =
-      case data do
-        _ when is_list(data) or is_binary(data) -> [statement, ?\n | data]
-        _ -> Stream.concat([[statement, ?\n]], data)
-      end
-
-    {_query_params = [], headers(opts), body}
-  end
-
-  def encode(%Query{command: :insert, statement: statement}, params, opts) do
-    cond do
-      names = Keyword.get(opts, :names) ->
-        types = Keyword.fetch!(opts, :types)
-        header = RowBinary.encode_names_and_types(names, types)
-        data = RowBinary.encode_rows(params, types)
-        {_query_params = [], headers(opts), [statement, ?\n, header | data]}
-
-      format_row_binary?(statement) ->
-        types = Keyword.fetch!(opts, :types)
-        data = RowBinary.encode_rows(params, types)
-        {_query_params = [], headers(opts), [statement, ?\n | data]}
-
-      true ->
-        {query_params(params), headers(opts), statement}
-    end
-  end
-
-  def encode(%Query{statement: statement}, params, opts) do
-    types = Keyword.get(opts, :types)
-    default_format = if types, do: "RowBinary", else: "RowBinaryWithNamesAndTypes"
-    format = Keyword.get(opts, :format) || default_format
-    {query_params(params), [{"x-clickhouse-format", format} | headers(opts)], statement}
-  end
-
-  defp format_row_binary?(statement) when is_binary(statement) do
-    statement |> String.trim_trailing() |> String.ends_with?("RowBinary")
-  end
-
-  defp format_row_binary?(statement) when is_list(statement) do
-    statement
-    |> IO.iodata_to_binary()
-    |> format_row_binary?()
+  @spec encode(Query.t(), Ch.params(), [Ch.query_option()]) ::
+          {Ch.Query.params(), Mint.Types.headers()}
+  def encode(%Query{}, params, opts) do
+    format = Keyword.get(opts, :format, "RowBinaryWithNamesAndTypes")
+    headers = Keyword.get(opts, :headers, [])
+    {query_params(params), [{"x-clickhouse-format", format} | headers]}
   end
 
-  @spec decode(Query.t(), [response], Keyword.t()) :: Result.t()
+  @spec decode(Query.t(), [response], [Ch.query_option()]) :: Result.t()
         when response: Mint.Types.status() | Mint.Types.headers() | binary
-  def decode(%Query{command: :insert}, responses, _opts) do
-    [_status, headers | _data] = responses
-
-    num_rows =
-      if summary = get_header(headers, "x-clickhouse-summary") do
-        %{"written_rows" => written_rows} = Jason.decode!(summary)
-        String.to_integer(written_rows)
-      end
-
-    %Result{num_rows: num_rows, rows: nil, command: :insert, headers: headers}
-  end
-
-  def decode(%Query{decode: false, command: command}, responses, _opts) when is_list(responses) do
-    # TODO potentially fails on x-progress-headers
-    [_status, headers | data] = responses
-    %Result{rows: data, command: command, headers: headers}
-  end
-
-  def decode(%Query{command: command}, responses, opts) when is_list(responses) do
-    # TODO potentially fails on x-progress-headers
+  def decode(%Query{command: command}, responses, opts) do
     [_status, headers | data] = responses
+    format = get_header(headers, "x-clickhouse-format")
+    decode = Keyword.get(opts, :decode, true)
 
-    case get_header(headers, "x-clickhouse-format") do
-      "RowBinary" ->
-        types = Keyword.fetch!(opts, :types)
-        rows = data |> IO.iodata_to_binary() |> RowBinary.decode_rows(types)
-        %Result{num_rows: length(rows), rows: rows, command: command, headers: headers}
-
-      "RowBinaryWithNamesAndTypes" ->
+    cond do
+      decode and format == "RowBinaryWithNamesAndTypes" ->
         rows = data |> IO.iodata_to_binary() |> RowBinary.decode_rows()
-        %Result{num_rows: length(rows), rows: rows, command: command, headers: headers}
+        %Result{num_rows: length(rows), rows: rows, headers: headers, command: command}
 
-      _other ->
-        %Result{rows: data, command: command, headers: headers}
-    end
-  end
+      format == nil ->
+        num_rows =
+          if summary = get_header(headers, "x-clickhouse-summary") do
+            %{"written_rows" => written_rows} = Jason.decode!(summary)
+            String.to_integer(written_rows)
+          end
 
-  # TODO merge :stream `decode/3` with "normal" `decode/3` clause above
-  @spec decode(Query.t(), {:stream, nil, responses}, Keyword.t()) :: responses
-        when responses: [Mint.Types.response()]
-  def decode(_query, {:stream, nil, responses}, _opts), do: responses
+        %Result{num_rows: num_rows, headers: headers, command: command}
 
-  @spec decode(Query.t(), {:stream, [atom], [Mint.Types.response()]}, Keyword.t()) :: [[term]]
-  def decode(_query, {:stream, types, responses}, _opts) do
-    decode_stream_data(responses, types)
-  end
-
-  defp decode_stream_data([{:data, _ref, data} | rest], types) do
-    [RowBinary.decode_rows(data, types) | decode_stream_data(rest, types)]
+      true ->
+        %Result{rows: data, headers: headers, command: command}
+    end
   end
 
-  defp decode_stream_data([_ | rest], types), do: decode_stream_data(rest, types)
-  defp decode_stream_data([] = done, _types), do: done
-
   defp get_header(headers, key) do
     case List.keyfind(headers, key, 0) do
       {_, value} -> value
@@ -183,15 +117,9 @@ defimpl DBConnection.Query, for: Ch.Query do
     end
   end
 
-  defp query_params(params) when is_map(params) do
-    Enum.map(params, fn {k, v} -> {"param_#{k}", encode_param(v)} end)
-  end
-
-  defp query_params(params) when is_list(params) do
-    params
-    |> Enum.with_index()
-    |> Enum.map(fn {v, idx} -> {"param_$#{idx}", encode_param(v)} end)
-  end
+  @compile inline: [query_params: 1]
+  defp query_params(params), do: Enum.map(params, &query_param/1)
+  defp query_param({k, v}), do: {"param_#{k}", encode_param(v)}
 
   defp encode_param(n) when is_integer(n), do: Integer.to_string(n)
   defp encode_param(f) when is_float(f), do: Float.to_string(f)
@@ -272,9 +200,6 @@ defimpl DBConnection.Query, for: Ch.Query do
   end
 
   defp escape_param([], param), do: param
-
-  @spec headers(Keyword.t()) :: Mint.Types.headers()
-  defp headers(opts), do: Keyword.get(opts, :headers, [])
 end
 
 defimpl String.Chars, for: Ch.Query do
diff --git a/test/ch/aggregation_test.exs b/test/ch/aggregation_test.exs
index 37bd8a3..996d03e 100644
--- a/test/ch/aggregation_test.exs
+++ b/test/ch/aggregation_test.exs
@@ -95,14 +95,15 @@ defmodule Ch.AggregationTest do
     assert %{num_rows: 2} =
              Ch.query!(
                conn,
-               """
-               INSERT INTO test_insert_aggregate_function
-                 SELECT uid, updated, arrayReduce('argMaxState', [name], [updated])
-                 FROM input('uid Int16, updated DateTime, name String')
-                 FORMAT RowBinary\
-               """,
-               rows,
-               types: ["Int16", "DateTime", "String"]
+               [
+                 """
+                 INSERT INTO test_insert_aggregate_function
+                   SELECT uid, updated, arrayReduce('argMaxState', [name], [updated])
+                   FROM input('uid Int16, updated DateTime, name String')
+                   FORMAT RowBinary
+                 """
+                 | Ch.RowBinary.encode_rows(rows, ["Int16", "DateTime", "String"])
+               ]
              )
 
     assert Ch.query!(conn, """
@@ -126,12 +127,16 @@ defmodule Ch.AggregationTest do
 
       Ch.query!(
         conn,
-        "INSERT INTO test_users_ephemeral_column(uid, updated, name_stub) FORMAT RowBinary",
-        _rows = [
-          [1231, ~N[2020-01-02 00:00:00], "Jane"],
-          [1231, ~N[2020-01-01 00:00:00], "John"]
-        ],
-        types: ["Int16", "DateTime", "String"]
+        [
+          "INSERT INTO test_users_ephemeral_column(uid, updated, name_stub) FORMAT RowBinary\n"
+          | Ch.RowBinary.encode_rows(
+              [
+                [1231, ~N[2020-01-02 00:00:00], "Jane"],
+                [1231, ~N[2020-01-01 00:00:00], "John"]
+              ],
+              ["Int16", "DateTime", "String"]
+            )
+        ]
       )
 
       assert Ch.query!(conn, """
@@ -152,16 +157,20 @@ defmodule Ch.AggregationTest do
 
       Ch.query!(
         conn,
-        """
-        INSERT INTO test_users_input_function
-          SELECT uid, updated, arrayReduce('argMaxState', [name], [updated])
-          FROM input('uid Int16, updated DateTime, name String') FORMAT RowBinary\
-        """,
-        _rows = [
-          [1231, ~N[2020-01-02 00:00:00], "Jane"],
-          [1231, ~N[2020-01-01 00:00:00], "John"]
-        ],
-        types: ["Int16", "DateTime", "String"]
+        [
+          """
+          INSERT INTO test_users_input_function
+            SELECT uid, updated, arrayReduce('argMaxState', [name], [updated])
+            FROM input('uid Int16, updated DateTime, name String') FORMAT RowBinary
+          """
+          | Ch.RowBinary.encode_rows(
+              [
+                [1231, ~N[2020-01-02 00:00:00], "Jane"],
+                [1231, ~N[2020-01-01 00:00:00], "John"]
+              ],
+              ["Int16", "DateTime", "String"]
+            )
+        ]
       )
 
       assert Ch.query!(conn, """
@@ -196,12 +205,16 @@ defmodule Ch.AggregationTest do
 
       Ch.query!(
         conn,
-        "INSERT INTO test_users_ne FORMAT RowBinary",
-        _rows = [
-          [1231, ~N[2020-01-02 00:00:00], "Jane"],
-          [1231, ~N[2020-01-01 00:00:00], "John"]
-        ],
-        types: ["Int16", "DateTime", "String"]
+        [
+          "INSERT INTO test_users_ne FORMAT RowBinary\n"
+          | Ch.RowBinary.encode_rows(
+              [
+                [1231, ~N[2020-01-02 00:00:00], "Jane"],
+                [1231, ~N[2020-01-01 00:00:00], "John"]
+              ],
+              ["Int16", "DateTime", "String"]
+            )
+        ]
       )
 
       assert Ch.query!(conn, """
diff --git a/test/ch/connection_test.exs b/test/ch/connection_test.exs
index 360bed2..89f1140 100644
--- a/test/ch/connection_test.exs
+++ b/test/ch/connection_test.exs
@@ -10,10 +10,6 @@ defmodule Ch.ConnectionTest do
     assert {:ok, %{num_rows: 1, rows: [[1]]}} = Ch.query(conn, "select 1")
   end
 
-  test "select with types", %{conn: conn} do
-    assert {:ok, %{num_rows: 1, rows: [[1]]}} = Ch.query(conn, "select 1", [], types: ["UInt8"])
-  end
-
   test "select with params", %{conn: conn} do
     assert {:ok, %{num_rows: 1, rows: [[1]]}} = Ch.query(conn, "select {a:UInt8}", %{"a" => 1})
 
@@ -66,17 +62,17 @@ defmodule Ch.ConnectionTest do
 
     # when the timezone information is provided in the type, we don't need to rely on server timezone
     assert {:ok, %{num_rows: 1, rows: [[bkk_datetime]]}} =
-             Ch.query(conn, "select {$0:DateTime('Asia/Bangkok')}", [naive_noon])
+             Ch.query(conn, "select {naive:DateTime('Asia/Bangkok')}", [{"naive", naive_noon}])
 
     assert bkk_datetime == DateTime.from_naive!(naive_noon, "Asia/Bangkok")
 
     assert {:ok, %{num_rows: 1, rows: [[~U[2022-01-01 12:00:00Z]]]}} =
-             Ch.query(conn, "select {$0:DateTime('UTC')}", [naive_noon])
+             Ch.query(conn, "select {naive:DateTime('UTC')}", [{"naive", naive_noon}])
 
     naive_noon_ms = ~N[2022-01-01 12:00:00.123]
 
     assert {:ok, %{num_rows: 1, rows: [[naive_datetime]]}} =
-             Ch.query(conn, "select {$0:DateTime64(3)}", [naive_noon_ms])
+             Ch.query(conn, "select {naive:DateTime64(3)}", [{"naive", naive_noon_ms}])
 
     assert NaiveDateTime.compare(
              naive_datetime,
@@ -109,7 +105,7 @@ defmodule Ch.ConnectionTest do
     #          Ch.query(conn, "select {a:UUID}", %{"a" => uuid_bin})
 
     # pseudo-positional bind
-    assert {:ok, %{num_rows: 1, rows: [[1]]}} = Ch.query(conn, "select {$0:UInt8}", [1])
+    assert {:ok, %{num_rows: 1, rows: [[1]]}} = Ch.query(conn, "select {$0:UInt8}", [{"$0", 1}])
   end
 
   test "utc datetime query param encoding", %{conn: conn} do
@@ -117,13 +113,17 @@ defmodule Ch.ConnectionTest do
     msk = DateTime.new!(~D[2021-01-01], ~T[15:00:00], "Europe/Moscow")
     naive = utc |> DateTime.shift_zone!(Ch.Test.clickhouse_tz(conn)) |> DateTime.to_naive()
 
-    assert Ch.query!(conn, "select {$0:DateTime} as d, toString(d)", [utc]).rows ==
+    assert Ch.query!(conn, "select {utc:DateTime} as d, toString(d)", %{"utc" => utc}).rows ==
              [[~N[2021-01-01 12:00:00], to_string(naive)]]
 
-    assert Ch.query!(conn, "select {$0:DateTime('UTC')} as d, toString(d)", [utc]).rows ==
+    assert Ch.query!(conn, "select {utc:DateTime('UTC')} as d, toString(d)", %{"utc" => utc}).rows ==
              [[utc, "2021-01-01 12:00:00"]]
 
-    assert Ch.query!(conn, "select {$0:DateTime('Europe/Moscow')} as d, toString(d)", [utc]).rows ==
+    assert Ch.query!(
+             conn,
+             "select {utc:DateTime('Europe/Moscow')} as d, toString(d)",
+             %{"utc" => utc}
+           ).rows ==
              [[msk, "2021-01-01 15:00:00"]]
   end
 
@@ -132,13 +132,17 @@ defmodule Ch.ConnectionTest do
     msk = DateTime.new!(~D[2021-01-01], ~T[15:00:00.123456], "Europe/Moscow")
     naive = utc |> DateTime.shift_zone!(Ch.Test.clickhouse_tz(conn)) |> DateTime.to_naive()
 
-    assert Ch.query!(conn, "select {$0:DateTime64(6)} as d, toString(d)", [utc]).rows ==
+    assert Ch.query!(conn, "select {utc:DateTime64(6)} as d, toString(d)", %{"utc" => utc}).rows ==
              [[~N[2021-01-01 12:00:00.123456], to_string(naive)]]
 
-    assert Ch.query!(conn, "select {$0:DateTime64(6, 'UTC')} as d, toString(d)", [utc]).rows ==
+    assert Ch.query!(conn, "select {utc:DateTime64(6, 'UTC')} as d, toString(d)", %{"utc" => utc}).rows ==
              [[utc, "2021-01-01 12:00:00.123456"]]
 
-    assert Ch.query!(conn, "select {$0:DateTime64(6,'Europe/Moscow')} as d, toString(d)", [utc]).rows ==
+    assert Ch.query!(
+             conn,
+             "select {utc:DateTime64(6,'Europe/Moscow')} as d, toString(d)",
+             %{"utc" => utc}
+           ).rows ==
              [[msk, "2021-01-01 15:00:00.123456"]]
   end
 
@@ -148,7 +152,7 @@ defmodule Ch.ConnectionTest do
     utc = ~U[2021-01-01 12:00:00.000000Z]
     naive = utc |> DateTime.shift_zone!(Ch.Test.clickhouse_tz(conn)) |> DateTime.to_naive()
 
-    assert Ch.query!(conn, "select {$0:DateTime64(6)} as d, toString(d)", [utc]).rows ==
+    assert Ch.query!(conn, "select {utc:DateTime64(6)} as d, toString(d)", %{"utc" => utc}).rows ==
              [[~N[2021-01-01 12:00:00.000000], to_string(naive)]]
   end
 
@@ -161,13 +165,16 @@ defmodule Ch.ConnectionTest do
   end
 
   test "create", %{conn: conn} do
-    assert {:ok, %{num_rows: nil, rows: []}} =
+    assert {:ok, %{command: :create}} =
              Ch.query(conn, "create table create_example(a UInt8) engine = Memory")
   end
 
   test "create with options", %{conn: conn} do
     assert {:error, %Ch.Error{code: 164, message: message}} =
-             Ch.query(conn, "create table create_example(a UInt8) engine = Memory", [],
+             Ch.query(
+               conn,
+               "create table create_example(a UInt8) engine = Memory",
+               _param = [],
                settings: [readonly: 1]
              )
 
@@ -202,8 +209,8 @@ defmodule Ch.ConnectionTest do
       assert {:ok, %{num_rows: 2}} =
                Ch.query(
                  conn,
-                 "insert into {$0:Identifier}(a, b) values ({$1:UInt8},{$2:String}),({$3:UInt8},{$4:String})",
-                 [table, 4, "d", 5, "e"]
+                 "insert into {table:Identifier}(a, b) values ({a:UInt8},{b:String}),({c:UInt8},{d:String})",
+                 %{"table" => table, "a" => 4, "b" => "d", "c" => 5, "d" => "e"}
                )
 
       assert {:ok, %{rows: rows}} =
@@ -226,26 +233,15 @@ defmodule Ch.ConnectionTest do
       assert message =~ "Cannot execute query in readonly mode."
     end
 
-    test "automatic RowBinary", %{conn: conn, table: table} do
-      stmt = "insert into #{table}(a, b) format RowBinary"
+    test "RowBinary", %{conn: conn, table: table} do
       types = ["UInt8", "String"]
       rows = [[1, "a"], [2, "b"]]
-      assert %{num_rows: 2} = Ch.query!(conn, stmt, rows, types: types)
-
-      assert %{rows: rows} =
-               Ch.query!(conn, "select * from {table:Identifier}", %{"table" => table})
-
-      assert rows == [[1, "a"], [2, "b"]]
-    end
 
-    test "manual RowBinary", %{conn: conn, table: table} do
-      stmt = "insert into #{table}(a, b) format RowBinary"
-
-      types = ["UInt8", "String"]
-      rows = [[1, "a"], [2, "b"]]
-      data = RowBinary.encode_rows(rows, types)
-
-      assert %{num_rows: 2} = Ch.query!(conn, stmt, data, encode: false)
+      assert %{num_rows: 2} =
+               Ch.query!(conn, [
+                 "insert into #{table}(a, b) format RowBinary\n"
+                 | RowBinary.encode_rows(rows, types)
+               ])
 
       assert %{rows: rows} =
                Ch.query!(conn, "select * from {table:Identifier}", %{"table" => table})
@@ -257,7 +253,7 @@ defmodule Ch.ConnectionTest do
       types = ["UInt8", "String"]
       rows = [[1, "a"], [2, "b"], [3, "c"]]
 
-      stream =
+      row_binary =
         rows
         |> Stream.chunk_every(2)
         |> Stream.map(fn chunk -> RowBinary.encode_rows(chunk, types) end)
@@ -265,9 +261,7 @@ defmodule Ch.ConnectionTest do
       assert {:ok, %{num_rows: 3}} =
                Ch.query(
                  conn,
-                 "insert into #{table}(a, b) format RowBinary",
-                 stream,
-                 encode: false
+                 Stream.concat(["insert into #{table}(a, b) format RowBinary\n"], row_binary)
                )
 
       assert {:ok, %{rows: rows}} =
@@ -299,8 +293,8 @@ defmodule Ch.ConnectionTest do
       assert {:ok, %{num_rows: 2}} =
                Ch.query(
                  conn,
-                 "insert into {$0:Identifier}(a, b) select a, b from {$0:Identifier} where a > {$1:UInt8}",
-                 [table, 1]
+                 "insert into {table:Identifier}(a, b) select a, b from {table:Identifier} where a > {a:UInt8}",
+                 %{"table" => table, "a" => 1}
                )
 
       assert {:ok, %{rows: new_rows}} =
@@ -320,7 +314,7 @@ defmodule Ch.ConnectionTest do
 
     settings = [allow_experimental_lightweight_delete: 1]
 
-    assert {:ok, %{rows: [], command: :delete}} =
+    assert {:ok, %{command: :delete}} =
              Ch.query(conn, "delete from delete_t where 1", [], settings: settings)
   end
 
@@ -379,14 +373,18 @@ defmodule Ch.ConnectionTest do
       assert {:ok, %{num_rows: 4}} =
                Ch.query(
                  conn,
-                 "insert into fixed_string_t(a) format RowBinary",
                  [
-                   [""],
-                   ["a"],
-                   ["aa"],
-                   ["aaa"]
-                 ],
-                 types: ["FixedString(3)"]
+                   "insert into fixed_string_t(a) format RowBinary\n"
+                   | RowBinary.encode_rows(
+                       [
+                         [""],
+                         ["a"],
+                         ["aa"],
+                         ["aaa"]
+                       ],
+                       ["FixedString(3)"]
+                     )
+                 ]
                )
 
       assert Ch.query!(conn, "select * from fixed_string_t").rows == [
@@ -423,13 +421,13 @@ defmodule Ch.ConnectionTest do
       assert %{num_rows: 3} =
                Ch.query!(
                  conn,
-                 "insert into decimal_t(d) format RowBinary",
-                 _rows = [
-                   [Decimal.new("2.66")],
-                   [Decimal.new("2.6666")],
-                   [Decimal.new("2.66666")]
-                 ],
-                 types: ["Decimal32(4)"]
+                 [
+                   "insert into decimal_t(d) format RowBinary\n"
+                   | RowBinary.encode_rows(
+                       [[Decimal.new("2.66")], [Decimal.new("2.6666")], [Decimal.new("2.66666")]],
+                       ["Decimal32(4)"]
+                     )
+                 ]
                )
 
       assert Ch.query!(conn, "select * from decimal_t").rows == [
@@ -454,9 +452,13 @@ defmodule Ch.ConnectionTest do
 
       Ch.query!(
         conn,
-        "insert into test_bool(A, B) format RowBinary",
-        _rows = [[3, true], [4, false]],
-        types: ["Int64", "Bool"]
+        [
+          "insert into test_bool(A, B) format RowBinary\n"
+          | RowBinary.encode_rows(
+              [[3, true], [4, false]],
+              ["Int64", "Bool"]
+            )
+        ]
       )
 
       # anything > 0 is `true`, here `2` is `true`
@@ -497,9 +499,13 @@ defmodule Ch.ConnectionTest do
 
       Ch.query!(
         conn,
-        "insert into t_uuid(x,y) format RowBinary",
-        _rows = [[uuid, "Example 3"]],
-        types: ["UUID", "String"]
+        [
+          "insert into t_uuid(x,y) format RowBinary\n"
+          | RowBinary.encode_rows(
+              [[uuid, "Example 3"]],
+              ["UUID", "String"]
+            )
+        ]
       )
 
       assert {:ok,
@@ -559,9 +565,13 @@ defmodule Ch.ConnectionTest do
 
       Ch.query!(
         conn,
-        "INSERT INTO t_enum(i, x) FORMAT RowBinary",
-        _rows = [[3, "hello"], [4, "world"], [5, 1], [6, 2]],
-        types: ["UInt8", "Enum8('hello' = 1, 'world' = 2)"]
+        [
+          "INSERT INTO t_enum(i, x) FORMAT RowBinary\n"
+          | RowBinary.encode_rows(
+              _rows = [[3, "hello"], [4, "world"], [5, 1], [6, 2]],
+              ["UInt8", "Enum8('hello' = 1, 'world' = 2)"]
+            )
+        ]
       )
 
       assert Ch.query!(conn, "SELECT *, CAST(x, 'Int8') FROM t_enum ORDER BY i").rows == [
@@ -608,18 +618,22 @@ defmodule Ch.ConnectionTest do
 
       assert Ch.query!(
                conn,
-               "INSERT INTO table_map FORMAT RowBinary",
-               _rows = [
-                 [%{"key10" => 20, "key20" => 40}],
-                 # empty map
-                 [%{}],
-                 # null map
-                 [nil],
-                 # empty proplist map
-                 [[]],
-                 [[{"key50", 100}]]
-               ],
-               types: ["Map(String, UInt64)"]
+               [
+                 "INSERT INTO table_map FORMAT RowBinary\n"
+                 | RowBinary.encode_rows(
+                     [
+                       [%{"key10" => 20, "key20" => 40}],
+                       # empty map
+                       [%{}],
+                       # null map
+                       [nil],
+                       # empty proplist map
+                       [[]],
+                       [[{"key50", 100}]]
+                     ],
+                     ["Map(String, UInt64)"]
+                   )
+               ]
              )
 
       assert Ch.query!(conn, "SELECT * FROM table_map ORDER BY a ASC").rows == [
@@ -641,7 +655,7 @@ defmodule Ch.ConnectionTest do
                [{1, "a"}, "Tuple(UInt8, String)"]
              ]
 
-      assert Ch.query!(conn, "SELECT {$0:Tuple(Int8, String)}", [{-1, "abs"}]).rows == [
+      assert Ch.query!(conn, "SELECT {t:Tuple(Int8, String)}", %{"t" => {-1, "abs"}}).rows == [
                [{-1, "abs"}]
              ]
 
@@ -660,9 +674,13 @@ defmodule Ch.ConnectionTest do
       assert %{num_rows: 2} =
                Ch.query!(
                  conn,
-                 "INSERT INTO tuples_t FORMAT RowBinary",
-                 _rows = [[{"a", 20}], [{"b", 30}]],
-                 types: ["Tuple(String, Int64)"]
+                 [
+                   "INSERT INTO tuples_t FORMAT RowBinary\n"
+                   | RowBinary.encode_rows(
+                       [[{"a", 20}], [{"b", 30}]],
+                       ["Tuple(String, Int64)"]
+                     )
+                 ]
                )
 
       assert Ch.query!(conn, "SELECT a FROM tuples_t ORDER BY a.1 ASC").rows == [
@@ -704,7 +722,11 @@ defmodule Ch.ConnectionTest do
       #     https://kb.altinity.com/altinity-kb-queries-and-syntax/time-zones/
       assert {:ok,
               %{num_rows: 1, rows: [[naive_datetime, "2022-12-12 12:00:00"]], headers: headers}} =
-               Ch.query(conn, "select {$0:DateTime} as d, toString(d)", [naive_noon])
+               Ch.query(
+                 conn,
+                 "select {naive:DateTime} as d, toString(d)",
+                 %{"naive" => naive_noon}
+               )
 
       # to make this test pass for contributors with non UTC timezone we perform the same steps as ClickHouse
       # i.e. we give server timezone to the naive datetime and shift it to UTC before comparing with the result
@@ -717,12 +739,18 @@ defmodule Ch.ConnectionTest do
                |> DateTime.to_naive()
 
       assert {:ok, %{num_rows: 1, rows: [[~U[2022-12-12 12:00:00Z], "2022-12-12 12:00:00"]]}} =
-               Ch.query(conn, "select {$0:DateTime('UTC')} as d, toString(d)", [naive_noon])
+               Ch.query(
+                 conn,
+                 "select {naive:DateTime('UTC')} as d, toString(d)",
+                 %{"naive" => naive_noon}
+               )
 
       assert {:ok, %{num_rows: 1, rows: rows}} =
-               Ch.query(conn, "select {$0:DateTime('Asia/Bangkok')} as d, toString(d)", [
-                 naive_noon
-               ])
+               Ch.query(
+                 conn,
+                 "select {naive:DateTime('Asia/Bangkok')} as d, toString(d)",
+                 %{"naive" => naive_noon}
+               )
 
       assert rows == [
                [
@@ -737,7 +765,7 @@ defmodule Ch.ConnectionTest do
       on_exit(fn -> Calendar.put_time_zone_database(prev_tz_db) end)
 
       assert_raise ArgumentError, ~r/:utc_only_time_zone_database/, fn ->
-        Ch.query(conn, "select {$0:DateTime('Asia/Tokyo')}", [naive_noon])
+        Ch.query(conn, "select {naive:DateTime('Asia/Tokyo')}", %{"naive" => naive_noon})
       end
     end
 
@@ -759,21 +787,37 @@ defmodule Ch.ConnectionTest do
              ]
 
       assert {:ok, %{num_rows: 1, rows: [[~D[1900-01-01], "1900-01-01"]]}} =
-               Ch.query(conn, "select {$0:Date32} as d, toString(d)", [~D[1900-01-01]])
+               Ch.query(
+                 conn,
+                 "select {date:Date32} as d, toString(d)",
+                 %{"date" => ~D[1900-01-01]}
+               )
 
       # max
       assert {:ok, %{num_rows: 1, rows: [[~D[2299-12-31], "2299-12-31"]]}} =
-               Ch.query(conn, "select {$0:Date32} as d, toString(d)", [~D[2299-12-31]])
+               Ch.query(
+                 conn,
+                 "select {date:Date32} as d, toString(d)",
+                 %{"date" => ~D[2299-12-31]}
+               )
 
       # min
       assert {:ok, %{num_rows: 1, rows: [[~D[1900-01-01], "1900-01-01"]]}} =
-               Ch.query(conn, "select {$0:Date32} as d, toString(d)", [~D[1900-01-01]])
+               Ch.query(
+                 conn,
+                 "select {date:Date32} as d, toString(d)",
+                 %{"date" => ~D[1900-01-01]}
+               )
 
       Ch.query!(
         conn,
-        "insert into new(timestamp, event_id) format RowBinary",
-        _rows = [[~D[1960-01-01], 3]],
-        types: ["Date32", "UInt8"]
+        [
+          "insert into new(timestamp, event_id) format RowBinary\n"
+          | RowBinary.encode_rows(
+              [[~D[1960-01-01], 3]],
+              ["Date32", "UInt8"]
+            )
+        ]
       )
 
       assert %{
@@ -829,12 +873,16 @@ defmodule Ch.ConnectionTest do
 
       Ch.query!(
         conn,
-        "insert into datetime64_t(event_id, timestamp) format RowBinary",
-        _rows = [
-          [4, ~N[2021-01-01 12:00:00.123456]],
-          [5, ~N[2021-01-01 12:00:00]]
-        ],
-        types: ["UInt8", "DateTime64(3)"]
+        [
+          "insert into datetime64_t(event_id, timestamp) format RowBinary\n"
+          | RowBinary.encode_rows(
+              [
+                [4, ~N[2021-01-01 12:00:00.123456]],
+                [5, ~N[2021-01-01 12:00:00]]
+              ],
+              ["UInt8", "DateTime64(3)"]
+            )
+        ]
       )
 
       assert {:ok, %{num_rows: 2, rows: rows}} =
@@ -863,7 +911,11 @@ defmodule Ch.ConnectionTest do
         # see https://clickhouse.com/docs/en/sql-reference/data-types/datetime
         #     https://kb.altinity.com/altinity-kb-queries-and-syntax/time-zones/
         assert {:ok, %{num_rows: 1, rows: [[naive_datetime]], headers: headers}} =
-                 Ch.query(conn, "select {$0:DateTime64(#{precision})}", [naive_noon])
+                 Ch.query(
+                   conn,
+                   "select {naive:DateTime64(#{precision})}",
+                   %{"naive" => naive_noon}
+                 )
 
         # to make this test pass for contributors with non UTC timezone we perform the same steps as ClickHouse
         # i.e. we give server timezone to the naive datetime and shift it to UTC before comparing with the result
@@ -919,9 +971,10 @@ defmodule Ch.ConnectionTest do
       assert {:ok, %{num_rows: 5}} =
                Ch.query(
                  conn,
-                 "insert into nullable format RowBinary",
-                 <<1, 2, 3, 4, 5>>,
-                 encode: false
+                 [
+                   "insert into nullable format RowBinary\n"
+                   | <<1, 2, 3, 4, 5>>
+                 ]
                )
 
       assert %{num_rows: 1, rows: [[count]]} =
@@ -942,9 +995,13 @@ defmodule Ch.ConnectionTest do
 
       Ch.query!(
         conn,
-        "INSERT INTO ch_nulls(a, b, c, d) FORMAT RowBinary",
-        [[nil, nil, nil, nil]],
-        types: ["UInt8", "Nullable(UInt8)", "UInt8", "Nullable(UInt8)"]
+        [
+          "INSERT INTO ch_nulls(a, b, c, d) FORMAT RowBinary\n"
+          | RowBinary.encode_rows(
+              [[nil, nil, nil, nil]],
+              ["UInt8", "Nullable(UInt8)", "UInt8", "Nullable(UInt8)"]
+            )
+        ]
       )
 
       # default is ignored...
@@ -962,14 +1019,18 @@ defmodule Ch.ConnectionTest do
 
       Ch.query!(
         conn,
-        """
-        INSERT INTO test_insert_default_value
-          SELECT id, name
-          FROM input('id UInt32, name Nullable(String)')
-          FORMAT RowBinary\
-        """,
-        [[1, nil], [-1, nil]],
-        types: ["UInt32", "Nullable(String)"]
+        [
+          """
+          INSERT INTO test_insert_default_value
+            SELECT id, name
+            FROM input('id UInt32, name Nullable(String)')
+            FORMAT RowBinary
+          """
+          | RowBinary.encode_rows(
+              [[1, nil], [-1, nil]],
+              ["UInt32", "Nullable(String)"]
+            )
+        ]
       )
 
       assert Ch.query!(conn, "SELECT * FROM test_insert_default_value ORDER BY n").rows ==
@@ -986,7 +1047,7 @@ defmodule Ch.ConnectionTest do
     end
 
     test "can encode and then decode Point in query params", %{conn: conn} do
-      assert Ch.query!(conn, "select {$0:Point}", [{10, 10}]).rows == [
+      assert Ch.query!(conn, "select {p:Point}", %{"p" => {10, 10}}).rows == [
                _row = [_point = {10.0, 10.0}]
              ]
     end
@@ -994,7 +1055,11 @@ defmodule Ch.ConnectionTest do
     test "can insert and select Point", %{conn: conn} do
       Ch.query!(conn, "CREATE TABLE geo_point (p Point) ENGINE = Memory()")
       Ch.query!(conn, "INSERT INTO geo_point VALUES((10, 10))")
-      Ch.query!(conn, "INSERT INTO geo_point FORMAT RowBinary", [[{20, 20}]], types: ["Point"])
+
+      Ch.query!(conn, [
+        "INSERT INTO geo_point FORMAT RowBinary\n"
+        | RowBinary.encode_rows([[{20, 20}]], ["Point"])
+      ])
 
       assert Ch.query!(conn, "SELECT p, toTypeName(p) FROM geo_point ORDER BY p ASC").rows == [
                [{10.0, 10.0}, "Point"],
@@ -1021,7 +1086,7 @@ defmodule Ch.ConnectionTest do
 
     test "can encode and then decode Ring in query params", %{conn: conn} do
       ring = [{0.0, 1.0}, {10.0, 3.0}]
-      assert Ch.query!(conn, "select {$0:Ring}", [ring]).rows == [_row = [ring]]
+      assert Ch.query!(conn, "select {r:Ring}", %{"r" => ring}).rows == [_row = [ring]]
     end
 
     test "can insert and select Ring", %{conn: conn} do
@@ -1029,7 +1094,10 @@ defmodule Ch.ConnectionTest do
       Ch.query!(conn, "INSERT INTO geo_ring VALUES([(0, 0), (10, 0), (10, 10), (0, 10)])")
 
       ring = [{20, 20}, {0, 0}, {0, 20}]
-      Ch.query!(conn, "INSERT INTO geo_ring FORMAT RowBinary", [[ring]], types: ["Ring"])
+
+      Ch.query!(conn, [
+        "INSERT INTO geo_ring FORMAT RowBinary\n" | RowBinary.encode_rows([[ring]], ["Ring"])
+      ])
 
       assert Ch.query!(conn, "SELECT r, toTypeName(r) FROM geo_ring ORDER BY r ASC").rows == [
                [[{0.0, 0.0}, {10.0, 0.0}, {10.0, 10.0}, {0.0, 10.0}], "Ring"],
@@ -1058,7 +1126,7 @@ defmodule Ch.ConnectionTest do
 
     test "can encode and then decode Polygon in query params", %{conn: conn} do
       polygon = [[{0.0, 1.0}, {10.0, 3.0}], [], [{2, 2}]]
-      assert Ch.query!(conn, "select {$0:Polygon}", [polygon]).rows == [_row = [polygon]]
+      assert Ch.query!(conn, "select {p:Polygon}", %{"p" => polygon}).rows == [_row = [polygon]]
     end
 
     test "can insert and select Polygon", %{conn: conn} do
@@ -1070,7 +1138,11 @@ defmodule Ch.ConnectionTest do
       )
 
       polygon = [[{0, 1.0}, {10, 3.2}], [], [{2, 2}]]
-      Ch.query!(conn, "INSERT INTO geo_polygon FORMAT RowBinary", [[polygon]], types: ["Polygon"])
+
+      Ch.query!(conn, [
+        "INSERT INTO geo_polygon FORMAT RowBinary\n"
+        | RowBinary.encode_rows([[polygon]], ["Polygon"])
+      ])
 
       assert Ch.query!(conn, "SELECT pg, toTypeName(pg) FROM geo_polygon ORDER BY pg ASC").rows ==
                [
@@ -1113,7 +1185,7 @@ defmodule Ch.ConnectionTest do
     test "can encode and then decode MultiPolygon in query params", %{conn: conn} do
       multipolygon = [[[{0.0, 1.0}, {10.0, 3.0}], [], [{2, 2}]], [], [[{3, 3}]]]
 
-      assert Ch.query!(conn, "select {$0:MultiPolygon}", [multipolygon]).rows == [
+      assert Ch.query!(conn, "select {mp:MultiPolygon}", %{"mp" => multipolygon}).rows == [
                _row = [multipolygon]
              ]
     end
@@ -1128,9 +1200,13 @@ defmodule Ch.ConnectionTest do
 
       multipolygon = [[[{0.0, 1.0}, {10.0, 3.0}], [], [{2, 2}]], [], [[{3, 3}]]]
 
-      Ch.query!(conn, "INSERT INTO geo_multipolygon FORMAT RowBinary", [[multipolygon]],
-        types: ["MultiPolygon"]
-      )
+      Ch.query!(conn, [
+        "INSERT INTO geo_multipolygon FORMAT RowBinary\n"
+        | RowBinary.encode_rows(
+            [[multipolygon]],
+            ["MultiPolygon"]
+          )
+      ])
 
       assert Ch.query!(conn, "SELECT mpg, toTypeName(mpg) FROM geo_multipolygon ORDER BY mpg ASC").rows ==
                [
@@ -1235,6 +1311,7 @@ defmodule Ch.ConnectionTest do
   end
 
   describe "stream" do
+    @tag :skip
     test "sends mint http packets", %{conn: conn} do
       stmt = "select number from system.numbers limit 1000"
 
@@ -1246,7 +1323,7 @@ defmodule Ch.ConnectionTest do
       end
 
       packets =
-        Ch.run(conn, fn conn ->
+        DBConnection.run(conn, fn conn ->
           conn
           |> Ch.stream(stmt)
           |> Enum.flat_map(drop_ref)
@@ -1267,11 +1344,12 @@ defmodule Ch.ConnectionTest do
       assert List.last(packets) == :done
     end
 
+    @tag :skip
     test "decodes RowBinary", %{conn: conn} do
       stmt = "select number from system.numbers limit 1000"
 
       rows =
-        Ch.run(conn, fn conn ->
+        DBConnection.run(conn, fn conn ->
           conn
           |> Ch.stream(stmt, _params = [], types: [:u64])
           |> Enum.into([])
@@ -1280,10 +1358,11 @@ defmodule Ch.ConnectionTest do
       assert List.flatten(rows) == Enum.into(0..999, [])
     end
 
+    @tag :skip
     test "disconnects on early halt", %{conn: conn} do
       logs =
         ExUnit.CaptureLog.capture_log(fn ->
-          Ch.run(conn, fn conn ->
+          DBConnection.run(conn, fn conn ->
             conn |> Ch.stream("select number from system.numbers") |> Enum.take(1)
           end)
 
@@ -1338,46 +1417,49 @@ defmodule Ch.ConnectionTest do
     end
 
     test "error on type mismatch", %{conn: conn} do
-      stmt = "insert into row_binary_names_and_types_t format RowBinaryWithNamesAndTypes"
-      rows = [["AB", "rare", -42]]
       names = ["country_code", "rare_string", "maybe_int32"]
+      types = [Ch.Types.fixed_string(2), Ch.Types.string(), Ch.Types.nullable(Ch.Types.u32())]
 
-      opts = [
-        names: names,
-        types: [Ch.Types.fixed_string(2), Ch.Types.string(), Ch.Types.nullable(Ch.Types.u32())]
-      ]
+      assert {:error, %Ch.Error{code: 117, message: message}} =
+               Ch.query(conn, [
+                 "insert into row_binary_names_and_types_t format RowBinaryWithNamesAndTypes\n",
+                 _header = RowBinary.encode_names_and_types(names, types)
+                 | RowBinary.encode_rows([["AB", "rare", -42]], types)
+               ])
 
-      assert {:error, %Ch.Error{code: 117, message: message}} = Ch.query(conn, stmt, rows, opts)
       assert message =~ "Type of 'rare_string' must be LowCardinality(String), not String"
 
-      opts = [
-        names: names,
-        types: [
-          Ch.Types.fixed_string(2),
-          Ch.Types.low_cardinality(Ch.Types.string()),
-          Ch.Types.nullable(Ch.Types.u32())
-        ]
+      types = [
+        Ch.Types.fixed_string(2),
+        Ch.Types.low_cardinality(Ch.Types.string()),
+        Ch.Types.nullable(Ch.Types.u32())
       ]
 
-      assert {:error, %Ch.Error{code: 117, message: message}} = Ch.query(conn, stmt, rows, opts)
+      assert {:error, %Ch.Error{code: 117, message: message}} =
+               Ch.query(conn, [
+                 "insert into row_binary_names_and_types_t format RowBinaryWithNamesAndTypes\n",
+                 _header = RowBinary.encode_names_and_types(names, types)
+                 | RowBinary.encode_rows([["AB", "rare", -42]], types)
+               ])
+
       assert message =~ "Type of 'maybe_int32' must be Nullable(Int32), not Nullable(UInt32)"
     end
 
     test "ok on valid types", %{conn: conn} do
-      stmt = "insert into row_binary_names_and_types_t format RowBinaryWithNamesAndTypes"
-      rows = [["AB", "rare", -42]]
       names = ["country_code", "rare_string", "maybe_int32"]
 
-      opts = [
-        names: names,
-        types: [
-          Ch.Types.fixed_string(2),
-          Ch.Types.low_cardinality(Ch.Types.string()),
-          Ch.Types.nullable(Ch.Types.i32())
-        ]
+      types = [
+        Ch.Types.fixed_string(2),
+        Ch.Types.low_cardinality(Ch.Types.string()),
+        Ch.Types.nullable(Ch.Types.i32())
       ]
 
-      assert {:ok, %{num_rows: 1}} = Ch.query(conn, stmt, rows, opts)
+      assert {:ok, %{num_rows: 1}} =
+               Ch.query(conn, [
+                 "insert into row_binary_names_and_types_t format RowBinaryWithNamesAndTypes\n",
+                 _header = RowBinary.encode_names_and_types(names, types)
+                 | RowBinary.encode_rows([["AB", "rare", -42]], types)
+               ])
     end
   end
 end
diff --git a/test/ch/faults_test.exs b/test/ch/faults_test.exs
index 69caf11..264612a 100644
--- a/test/ch/faults_test.exs
+++ b/test/ch/faults_test.exs
@@ -359,9 +359,10 @@ defmodule Ch.FaultsTest do
             assert {:error, %Mint.TransportError{reason: :closed}} =
                      Ch.query(
                        conn,
-                       "insert into unknown_table(a,b) format RowBinary",
-                       stream,
-                       encode: false
+                       Stream.concat(
+                         ["insert into unknown_table(a,b) format RowBinary\n"],
+                         stream
+                       )
                      )
           end)
 
@@ -376,9 +377,10 @@ defmodule Ch.FaultsTest do
             assert {:error, %Ch.Error{code: 60, message: message}} =
                      Ch.query(
                        conn,
-                       "insert into unknown_table(a,b) format RowBinary",
-                       stream,
-                       encode: false
+                       Stream.concat(
+                         ["insert into unknown_table(a,b) format RowBinary\n"],
+                         stream
+                       )
                      )
 
             assert message =~ ~r/UNKNOWN_TABLE/
@@ -418,9 +420,10 @@ defmodule Ch.FaultsTest do
             assert {:error, %Mint.TransportError{reason: :closed}} =
                      Ch.query(
                        conn,
-                       "insert into unknown_table(a,b) format RowBinary",
-                       stream,
-                       encode: false
+                       Stream.concat(
+                         ["insert into unknown_table(a,b) format RowBinary\n"],
+                         stream
+                       )
                      )
           end)
 
@@ -439,9 +442,10 @@ defmodule Ch.FaultsTest do
             assert {:error, %Ch.Error{code: 60, message: message}} =
                      Ch.query(
                        conn,
-                       "insert into unknown_table(a,b) format RowBinary",
-                       stream,
-                       encode: false
+                       Stream.concat(
+                         ["insert into unknown_table(a,b) format RowBinary\n"],
+                         stream
+                       )
                      )
 
             assert message =~ ~r/UNKNOWN_TABLE/
diff --git a/test/ch/query_test.exs b/test/ch/query_test.exs
index c8eec26..7ff7b9c 100644
--- a/test/ch/query_test.exs
+++ b/test/ch/query_test.exs
@@ -20,7 +20,9 @@ defmodule Ch.QueryTest do
              """).command == :select
 
       assert Query.build(["select 1+2"]).command == :select
-      assert Query.build([?S, ?E, ?L | "ECT 1"]).command == :select
+
+      # TODO?
+      refute Query.build([?S, ?E, ?L | "ECT 1"]).command == :select
 
       assert Query.build("with insert as (select 1) select * from insert").command == :select
       assert Query.build("insert into table(a, b) values(1, 2)").command == :insert
@@ -112,27 +114,34 @@ defmodule Ch.QueryTest do
 
     test "decoded binaries copy behaviour", %{conn: conn} do
       text = "hello world"
-      assert [[bin]] = Ch.query!(conn, "SELECT {$0:String}", [text]).rows
+      assert [[bin]] = Ch.query!(conn, "SELECT {t:String}", %{"t" => text}).rows
       assert :binary.referenced_byte_size(bin) == byte_size(text)
 
       # For OTP 20+ refc binaries up to 64 bytes might be copied during a GC
       text = String.duplicate("hello world", 6)
-      assert [[bin]] = Ch.query!(conn, "SELECT {$0:String}", [text]).rows
+      assert [[bin]] = Ch.query!(conn, "SELECT {t:String}", %{"t" => text}).rows
       assert :binary.referenced_byte_size(bin) == byte_size(text)
     end
 
     test "encode basic types", %{conn: conn} do
       # TODO
       # assert [[nil, nil]] = query("SELECT $1::text, $2::int", [nil, nil])
-      assert [[true, false]] = Ch.query!(conn, "SELECT {$0:bool}, {$1:Bool}", [true, false]).rows
-      assert [["ẽ"]] = Ch.query!(conn, "SELECT {$0:char}", ["ẽ"]).rows
-      assert [[42]] = Ch.query!(conn, "SELECT {$0:int}", [42]).rows
-      assert [[42.0, 43.0]] = Ch.query!(conn, "SELECT {$0:float}, {$1:float}", [42, 43.0]).rows
-      assert [[nil, nil]] = Ch.query!(conn, "SELECT {$0:float}, {$1:float}", ["NaN", "nan"]).rows
-      assert [[nil]] = Ch.query!(conn, "SELECT {$0:float}", ["inf"]).rows
-      assert [[nil]] = Ch.query!(conn, "SELECT {$0:float}", ["-inf"]).rows
-      assert [["ẽric"]] = Ch.query!(conn, "SELECT {$0:varchar}", ["ẽric"]).rows
-      assert [[<<1, 2, 3>>]] = Ch.query!(conn, "SELECT {$0:bytea}", [<<1, 2, 3>>]).rows
+      assert [[true, false]] =
+               Ch.query!(conn, "SELECT {a:bool}, {b:Bool}", [{"a", true}, {"b", false}]).rows
+
+      assert [["ẽ"]] = Ch.query!(conn, "SELECT {e:char}", [{"e", "ẽ"}]).rows
+      assert [[42]] = Ch.query!(conn, "SELECT {num:int}", [{"num", 42}]).rows
+
+      assert [[42.0, 43.0]] =
+               Ch.query!(conn, "SELECT {n:float}, {f:float}", [{"n", 42}, {"f", 43.0}]).rows
+
+      assert [[nil, nil]] =
+               Ch.query!(conn, "SELECT {a:float}, {b:float}", [{"a", "NaN"}, {"b", "nan"}]).rows
+
+      assert [[nil]] = Ch.query!(conn, "SELECT {i:float}", %{"i" => "inf"}).rows
+      assert [[nil]] = Ch.query!(conn, "SELECT {ni:float}", %{"ni" => "-inf"}).rows
+      assert [["ẽric"]] = Ch.query!(conn, "SELECT {name:varchar}", %{"name" => "ẽric"}).rows
+      assert [[<<1, 2, 3>>]] = Ch.query!(conn, "SELECT {b:bytea}", %{"b" => <<1, 2, 3>>}).rows
     end
 
     test "encode numeric", %{conn: conn} do
@@ -159,70 +168,78 @@ defmodule Ch.QueryTest do
 
       Enum.each(nums, fn {num, type} ->
         dec = Decimal.new(num)
-        assert [[dec]] == Ch.query!(conn, "SELECT {$0:#{type}}", [dec]).rows
+        assert [[dec]] == Ch.query!(conn, "SELECT {dec:#{type}}", %{"dec" => dec}).rows
       end)
     end
 
     test "encode integers and floats as numeric", %{conn: conn} do
       dec = Decimal.new(1)
-      assert [[dec]] == Ch.query!(conn, "SELECT {$0:numeric(1,0)}", [1]).rows
+      assert [[dec]] == Ch.query!(conn, "SELECT {dec:numeric(1,0)}", %{"dec" => 1}).rows
 
       dec = Decimal.from_float(1.0)
-      assert [[dec]] == Ch.query!(conn, "SELECT {$0:numeric(2,1)}", [1.0]).rows
+      assert [[dec]] == Ch.query!(conn, "SELECT {dec:numeric(2,1)}", %{"dec" => 1.0}).rows
     end
 
     @tag skip: true
     test "encode json/jsonb", %{conn: conn} do
       json = %{"foo" => 42}
-      assert [[json]] == Ch.query!(conn, "SELECT {$0::json}", [json]).rows
+      assert [[json]] == Ch.query!(conn, "SELECT {json::json}", %{"json" => json}).rows
     end
 
     test "encode uuid", %{conn: conn} do
       # TODO
       uuid = <<0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15>>
       uuid_hex = "00010203-0405-0607-0809-0a0b0c0d0e0f"
-      assert [[^uuid]] = Ch.query!(conn, "SELECT {$0:UUID}", [uuid_hex]).rows
+      assert [[^uuid]] = Ch.query!(conn, "SELECT {uuid:UUID}", %{"uuid" => uuid_hex}).rows
     end
 
     test "encode arrays", %{conn: conn} do
-      assert [[[]]] = Ch.query!(conn, "SELECT {$0:Array(integer)}", [[]]).rows
-      assert [[[1]]] = Ch.query!(conn, "SELECT {$0:Array(integer)}", [[1]]).rows
-      assert [[[1, 2]]] = Ch.query!(conn, "SELECT {$0:Array(integer)}", [[1, 2]]).rows
+      assert [[[]]] = Ch.query!(conn, "SELECT {a:Array(integer)}", %{"a" => []}).rows
+      assert [[[1]]] = Ch.query!(conn, "SELECT {a:Array(integer)}", %{"a" => [1]}).rows
+      assert [[[1, 2]]] = Ch.query!(conn, "SELECT {a:Array(integer)}", %{"a" => [1, 2]}).rows
 
-      assert [[["1"]]] = Ch.query!(conn, "SELECT {$0:Array(String)}", [["1"]]).rows
-      assert [[[true]]] = Ch.query!(conn, "SELECT {$0:Array(Bool)}", [[true]]).rows
+      assert [[["1"]]] = Ch.query!(conn, "SELECT {a:Array(String)}", %{"a" => ["1"]}).rows
+      assert [[[true]]] = Ch.query!(conn, "SELECT {a:Array(Bool)}", %{"a" => [true]}).rows
 
       assert [[[~D[2023-01-01]]]] =
-               Ch.query!(conn, "SELECT {$0:Array(Date)}", [[~D[2023-01-01]]]).rows
+               Ch.query!(conn, "SELECT {a:Array(Date)}", %{"a" => [~D[2023-01-01]]}).rows
 
       assert [[[Ch.Test.to_clickhouse_naive(conn, ~N[2023-01-01 12:00:00])]]] ==
-               Ch.query!(conn, "SELECT {$0:Array(DateTime)}", [[~N[2023-01-01 12:00:00]]]).rows
+               Ch.query!(conn, "SELECT {a:Array(DateTime)}", %{"a" => [~N[2023-01-01 12:00:00]]}).rows
 
       assert [[[~U[2023-01-01 12:00:00Z]]]] ==
-               Ch.query!(conn, "SELECT {$0:Array(DateTime('UTC'))}", [[~N[2023-01-01 12:00:00]]]).rows
+               Ch.query!(
+                 conn,
+                 "SELECT {a:Array(DateTime('UTC'))}",
+                 %{"a" => [~N[2023-01-01 12:00:00]]}
+               ).rows
 
       assert [[[~N[2023-01-01 12:00:00]]]] ==
-               Ch.query!(conn, "SELECT {$0:Array(DateTime)}", [[~U[2023-01-01 12:00:00Z]]]).rows
+               Ch.query!(conn, "SELECT {a:Array(DateTime)}", %{"a" => [~U[2023-01-01 12:00:00Z]]}).rows
 
       assert [[[~U[2023-01-01 12:00:00Z]]]] ==
-               Ch.query!(conn, "SELECT {$0:Array(DateTime('UTC'))}", [[~U[2023-01-01 12:00:00Z]]]).rows
+               Ch.query!(conn, "SELECT {a:Array(DateTime('UTC'))}", %{
+                 "a" => [~U[2023-01-01 12:00:00Z]]
+               }).rows
 
       assert [[[[0], [1]]]] =
-               Ch.query!(conn, "SELECT {$0:Array(Array(integer))}", [[[0], [1]]]).rows
+               Ch.query!(conn, "SELECT {a:Array(Array(integer))}", %{"a" => [[0], [1]]}).rows
+
+      assert [[[[0]]]] = Ch.query!(conn, "SELECT {a:Array(Array(integer))}", %{"a" => [[0]]}).rows
 
-      assert [[[[0]]]] = Ch.query!(conn, "SELECT {$0:Array(Array(integer))}", [[[0]]]).rows
-      # assert [[[1, nil, 3]]] = Ch.query!(conn, "SELECT {$0:Array(integer)}", [[1, nil, 3]]).rows
+      # assert [[[1, nil, 3]]] = Ch.query!(conn, "SELECT {a:Array(integer)}", %{"a" => [1, nil, 3]}).rows
     end
 
     test "encode network types", %{conn: conn} do
       # TODO, or wrap in custom struct like in postgrex
       # assert [["127.0.0.1/32"]] =
-      #          Ch.query!(conn, "SELECT {$0:inet4}::text", [{127, 0, 0, 1}]).rows
+      #          Ch.query!(conn, "SELECT {ip:inet4}::text", %{"ip" => {127, 0, 0, 1}}).rows
 
-      assert [[{127, 0, 0, 1}]] = Ch.query!(conn, "SELECT {$0:text}::inet4", ["127.0.0.1"]).rows
+      assert [[{127, 0, 0, 1}]] =
+               Ch.query!(conn, "SELECT {ip:text}::inet4", %{"ip" => "127.0.0.1"}).rows
 
       assert [[{0, 0, 0, 0, 0, 0, 0, 1}]] =
-               Ch.query!(conn, "SELECT {$0:text}::inet6", ["::1"]).rows
+               Ch.query!(conn, "SELECT {ip:text}::inet6", %{"ip" => "::1"}).rows
     end
 
     test "result struct", %{conn: conn} do

From 6f49d03ac8c91913a6fcd2e1a1161f19aab477cf Mon Sep 17 00:00:00 2001
From: ruslandoga <67764432+ruslandoga@users.noreply.github.com>
Date: Fri, 29 Dec 2023 19:35:33 +0800
Subject: [PATCH 2/9] update bench/insert.exs

---
 bench/insert.exs | 10 ++++++----
 1 file changed, 6 insertions(+), 4 deletions(-)

diff --git a/bench/insert.exs b/bench/insert.exs
index f6f01c4..b852196 100644
--- a/bench/insert.exs
+++ b/bench/insert.exs
@@ -6,7 +6,7 @@ scheme = System.get_env("CH_SCHEME") || "http"
 database = System.get_env("CH_DATABASE") || "ch_bench"
 
 {:ok, conn} = Ch.start_link(scheme: scheme, hostname: hostname, port: port)
-Ch.query!(conn, "CREATE DATABASE IF NOT EXISTS {$0:Identifier}", [database])
+Ch.query!(conn, "CREATE DATABASE IF NOT EXISTS {db:Identifier}", %{"db" => database})
 
 Ch.query!(conn, """
 CREATE TABLE IF NOT EXISTS #{database}.benchmark (
@@ -18,7 +18,7 @@ CREATE TABLE IF NOT EXISTS #{database}.benchmark (
 """)
 
 types = [Ch.Types.u64(), Ch.Types.string(), Ch.Types.array(Ch.Types.u8()), Ch.Types.datetime()]
-statement = "INSERT INTO #{database}.benchmark FORMAT RowBinary"
+statement = "INSERT INTO #{database}.benchmark FORMAT RowBinary\n"
 
 rows = fn count ->
   Enum.map(1..count, fn i ->
@@ -32,7 +32,9 @@ Benchee.run(
   %{
     # "control" => fn rows -> Enum.each(rows, fn _row -> :ok end) end,
     "encode" => fn rows -> RowBinary.encode_rows(rows, types) end,
-    "insert" => fn rows -> Ch.query!(conn, statement, rows, types: types) end,
+    "encode+insert" => fn rows ->
+      Ch.query!(conn, [statement | RowBinary.encode_rows(rows, types)])
+    end,
     # "control stream" => fn rows -> rows |> Stream.chunk_every(60_000) |> Stream.run() end,
     "encode stream" => fn rows ->
       rows
@@ -46,7 +48,7 @@ Benchee.run(
         |> Stream.chunk_every(60_000)
         |> Stream.map(fn chunk -> RowBinary.encode_rows(chunk, types) end)
 
-      Ch.query!(conn, statement, stream, encode: false)
+      Ch.query!(conn, Stream.concat([statement], stream))
     end
   },
   inputs: %{

From bfefa7ecc13c7a17077eb40666c397a7b9d3e039 Mon Sep 17 00:00:00 2001
From: ruslandoga <67764432+ruslandoga@users.noreply.github.com>
Date: Fri, 29 Dec 2023 19:37:47 +0800
Subject: [PATCH 3/9] update changelog

---
 CHANGELOG.md | 3 +--
 1 file changed, 1 insertion(+), 2 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index f7169b9..48cfdf8 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -2,8 +2,7 @@
 
 ## Unreleased
 
-- move rows payload (RowBinary, CSV, etc.) to SQL statement
-- remove pseudo-positional binds, make param names explicit
+- move rows payload (RowBinary, CSV, etc.) to SQL statement and remove pseudo-positional binds, making param names explicit https://github.com/plausible/ch/pull/143
 
 ## 0.2.2 (2023-12-23)
 

From b3cbb7664bcb7defa250cd9095449eb60f7ce3f3 Mon Sep 17 00:00:00 2001
From: ruslandoga <67764432+ruslandoga@users.noreply.github.com>
Date: Fri, 29 Dec 2023 19:46:45 +0800
Subject: [PATCH 4/9] Ch.Result: drop :headers, add :data

---
 CHANGELOG.md                |  1 +
 lib/ch/query.ex             |  6 +++---
 lib/ch/result.ex            | 20 +++++++++-----------
 test/ch/connection_test.exs | 28 +++++++++++++---------------
 test/ch/faults_test.exs     |  3 +--
 test/ch/headers_test.exs    | 24 ++++++------------------
 6 files changed, 33 insertions(+), 49 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 48cfdf8..161ae4f 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -3,6 +3,7 @@
 ## Unreleased
 
 - move rows payload (RowBinary, CSV, etc.) to SQL statement and remove pseudo-positional binds, making param names explicit https://github.com/plausible/ch/pull/143
+- drop `:headers` from `%Ch.Result{}` but add `:data`
 
 ## 0.2.2 (2023-12-23)
 
diff --git a/lib/ch/query.ex b/lib/ch/query.ex
index 65b3ad8..2685e70 100644
--- a/lib/ch/query.ex
+++ b/lib/ch/query.ex
@@ -94,7 +94,7 @@ defimpl DBConnection.Query, for: Ch.Query do
     cond do
       decode and format == "RowBinaryWithNamesAndTypes" ->
         rows = data |> IO.iodata_to_binary() |> RowBinary.decode_rows()
-        %Result{num_rows: length(rows), rows: rows, headers: headers, command: command}
+        %Result{num_rows: length(rows), rows: rows, data: data, command: command}
 
       format == nil ->
         num_rows =
@@ -103,10 +103,10 @@ defimpl DBConnection.Query, for: Ch.Query do
             String.to_integer(written_rows)
           end
 
-        %Result{num_rows: num_rows, headers: headers, command: command}
+        %Result{num_rows: num_rows, data: data, command: command}
 
       true ->
-        %Result{rows: data, headers: headers, command: command}
+        %Result{data: data, command: command}
     end
   end
 
diff --git a/lib/ch/result.ex b/lib/ch/result.ex
index ddca4df..2977aa9 100644
--- a/lib/ch/result.ex
+++ b/lib/ch/result.ex
@@ -2,21 +2,19 @@ defmodule Ch.Result do
   @moduledoc """
   Result struct returned from any successful query. Its fields are:
 
-    * `command` - An atom of the query command, for example: `:select`, `:insert`;
-    * `rows` - The result set. One of:
-      - a list of lists, each inner list corresponding to a
-        row, each element in the inner list corresponds to a column;
-      - raw iodata when the response is not automatically decoded, e.g. `x-clickhouse-format: CSV`
-    * `num_rows` - The number of fetched or affected rows;
-    * `headers` - The HTTP response headers
+    * `command` - An atom of the query command, for example: `:select`, `:insert`
+    * `num_rows` - The number of fetched or affected rows
+    * `rows` - A list of lists, each inner list corresponding to a row, each element in the inner list corresponds to a column
+    * `data` - The raw iodata from the response
+
   """
 
-  defstruct [:command, :num_rows, :rows, :headers]
+  defstruct [:command, :num_rows, :rows, :data]
 
   @type t :: %__MODULE__{
-          command: atom,
+          command: Ch.Query.command() | nil,
           num_rows: non_neg_integer | nil,
-          rows: [[term]] | iodata | nil,
-          headers: Mint.Types.headers()
+          rows: [[term]] | nil,
+          data: iodata
         }
 end
diff --git a/test/ch/connection_test.exs b/test/ch/connection_test.exs
index 89f1140..aa99a4a 100644
--- a/test/ch/connection_test.exs
+++ b/test/ch/connection_test.exs
@@ -47,12 +47,12 @@ defmodule Ch.ConnectionTest do
     # datetimes in params are sent in text and ClickHouse translates them to UTC from server timezone by default
     # see https://clickhouse.com/docs/en/sql-reference/data-types/datetime
     #     https://kb.altinity.com/altinity-kb-queries-and-syntax/time-zones/
-    assert {:ok, %{num_rows: 1, rows: [[naive_datetime]], headers: headers}} =
+    assert {:ok, %{num_rows: 1, rows: [[naive_datetime]]}} =
              Ch.query(conn, "select {naive:DateTime}", %{"naive" => naive_noon})
 
     # to make this test pass for contributors with non UTC timezone we perform the same steps as ClickHouse
     # i.e. we give server timezone to the naive datetime and shift it to UTC before comparing with the result
-    {_, timezone} = List.keyfind!(headers, "x-clickhouse-timezone", 0)
+    %Ch.Result{rows: [[timezone]]} = Ch.query!(conn, "select timezone()")
 
     assert naive_datetime ==
              naive_noon
@@ -720,8 +720,7 @@ defmodule Ch.ConnectionTest do
       # datetimes in params are sent in text and ClickHouse translates them to UTC from server timezone by default
       # see https://clickhouse.com/docs/en/sql-reference/data-types/datetime
       #     https://kb.altinity.com/altinity-kb-queries-and-syntax/time-zones/
-      assert {:ok,
-              %{num_rows: 1, rows: [[naive_datetime, "2022-12-12 12:00:00"]], headers: headers}} =
+      assert {:ok, %{num_rows: 1, rows: [[naive_datetime, "2022-12-12 12:00:00"]]}} =
                Ch.query(
                  conn,
                  "select {naive:DateTime} as d, toString(d)",
@@ -730,7 +729,7 @@ defmodule Ch.ConnectionTest do
 
       # to make this test pass for contributors with non UTC timezone we perform the same steps as ClickHouse
       # i.e. we give server timezone to the naive datetime and shift it to UTC before comparing with the result
-      {_, timezone} = List.keyfind!(headers, "x-clickhouse-timezone", 0)
+      %Ch.Result{rows: [[timezone]]} = Ch.query!(conn, "select timezone()")
 
       assert naive_datetime ==
                naive_noon
@@ -910,7 +909,7 @@ defmodule Ch.ConnectionTest do
         # datetimes in params are sent in text and ClickHouse translates them to UTC from server timezone by default
         # see https://clickhouse.com/docs/en/sql-reference/data-types/datetime
         #     https://kb.altinity.com/altinity-kb-queries-and-syntax/time-zones/
-        assert {:ok, %{num_rows: 1, rows: [[naive_datetime]], headers: headers}} =
+        assert {:ok, %{num_rows: 1, rows: [[naive_datetime]]}} =
                  Ch.query(
                    conn,
                    "select {naive:DateTime64(#{precision})}",
@@ -919,7 +918,7 @@ defmodule Ch.ConnectionTest do
 
         # to make this test pass for contributors with non UTC timezone we perform the same steps as ClickHouse
         # i.e. we give server timezone to the naive datetime and shift it to UTC before comparing with the result
-        {_, timezone} = List.keyfind!(headers, "x-clickhouse-timezone", 0)
+        %Ch.Result{rows: [[timezone]]} = Ch.query!(conn, "select timezone()")
 
         expected =
           naive_noon
@@ -1066,12 +1065,11 @@ defmodule Ch.ConnectionTest do
                [{20.0, 20.0}, "Point"]
              ]
 
-      # to make our RowBinary is not garbage in garbage out we also test a text format response
-      assert conn
-             |> Ch.query!(
+      # to make sure our RowBinary is not "garbage in, garbage out" we also test a "text format" response
+      assert Ch.query!(
+               conn,
                "SELECT p, toTypeName(p) FROM geo_point ORDER BY p ASC FORMAT JSONCompact"
-             )
-             |> Map.fetch!(:rows)
+             ).data
              |> Jason.decode!()
              |> Map.fetch!("data") == [
                [[10, 10], "Point"],
@@ -1108,7 +1106,7 @@ defmodule Ch.ConnectionTest do
       assert Ch.query!(
                conn,
                "SELECT r, toTypeName(r) FROM geo_ring ORDER BY r ASC FORMAT JSONCompact"
-             ).rows
+             ).data
              |> Jason.decode!()
              |> Map.fetch!("data") == [
                [[[0, 0], [10, 0], [10, 10], [0, 10]], "Ring"],
@@ -1160,7 +1158,7 @@ defmodule Ch.ConnectionTest do
       assert Ch.query!(
                conn,
                "SELECT pg, toTypeName(pg) FROM geo_polygon ORDER BY pg ASC FORMAT JSONCompact"
-             ).rows
+             ).data
              |> Jason.decode!()
              |> Map.fetch!("data") == [
                [[[[0, 1], [10, 3.2]], [], [[2, 2]]], "Polygon"],
@@ -1242,7 +1240,7 @@ defmodule Ch.ConnectionTest do
       assert Ch.query!(
                conn,
                "SELECT mpg, toTypeName(mpg) FROM geo_multipolygon ORDER BY mpg ASC FORMAT JSONCompact"
-             ).rows
+             ).data
              |> Jason.decode!()
              |> Map.fetch!("data") == [
                [
diff --git a/test/ch/faults_test.exs b/test/ch/faults_test.exs
index 264612a..bdf8d34 100644
--- a/test/ch/faults_test.exs
+++ b/test/ch/faults_test.exs
@@ -468,8 +468,7 @@ defmodule Ch.FaultsTest do
       test = self()
 
       header = "X-ClickHouse-Server-Display-Name"
-      {:ok, %Result{headers: headers}} = Ch.Test.sql_exec("select 1")
-      {_, expected_name} = List.keyfind!(headers, String.downcase(header), 0)
+      {:ok, %Result{rows: [[expected_name]]}} = Ch.Test.sql_exec("select hostName()")
 
       log =
         capture_async_log(fn ->
diff --git a/test/ch/headers_test.exs b/test/ch/headers_test.exs
index 52aa7c3..18c8b98 100644
--- a/test/ch/headers_test.exs
+++ b/test/ch/headers_test.exs
@@ -7,50 +7,38 @@ defmodule Ch.HeadersTest do
   end
 
   test "can request gzipped response through headers", %{conn: conn} do
-    assert {:ok, %{rows: rows, headers: headers}} =
+    assert {:ok, %{data: data}} =
              Ch.query(conn, "select number from system.numbers limit 100", [],
                decode: false,
                settings: [enable_http_compression: 1],
                headers: [{"accept-encoding", "gzip"}]
              )
 
-    assert :proplists.get_value("content-type", headers) == "application/octet-stream"
-    assert :proplists.get_value("content-encoding", headers) == "gzip"
-    assert :proplists.get_value("x-clickhouse-format", headers) == "RowBinaryWithNamesAndTypes"
-
     # https://en.wikipedia.org/wiki/Gzip
-    assert <<0x1F, 0x8B, _rest::bytes>> = IO.iodata_to_binary(rows)
+    assert <<0x1F, 0x8B, _rest::bytes>> = IO.iodata_to_binary(data)
   end
 
   test "can request lz4 response through headers", %{conn: conn} do
-    assert {:ok, %{rows: rows, headers: headers}} =
+    assert {:ok, %{data: data}} =
              Ch.query(conn, "select number from system.numbers limit 100", [],
                decode: false,
                settings: [enable_http_compression: 1],
                headers: [{"accept-encoding", "lz4"}]
              )
 
-    assert :proplists.get_value("content-type", headers) == "application/octet-stream"
-    assert :proplists.get_value("content-encoding", headers) == "lz4"
-    assert :proplists.get_value("x-clickhouse-format", headers) == "RowBinaryWithNamesAndTypes"
-
     # https://en.wikipedia.org/wiki/LZ4_(compression_algorithm)
-    assert <<0x04, 0x22, 0x4D, 0x18, _rest::bytes>> = IO.iodata_to_binary(rows)
+    assert <<0x04, 0x22, 0x4D, 0x18, _rest::bytes>> = IO.iodata_to_binary(data)
   end
 
   test "can request zstd response through headers", %{conn: conn} do
-    assert {:ok, %{rows: rows, headers: headers}} =
+    assert {:ok, %{data: data}} =
              Ch.query(conn, "select number from system.numbers limit 100", [],
                decode: false,
                settings: [enable_http_compression: 1],
                headers: [{"accept-encoding", "zstd"}]
              )
 
-    assert :proplists.get_value("content-type", headers) == "application/octet-stream"
-    assert :proplists.get_value("content-encoding", headers) == "zstd"
-    assert :proplists.get_value("x-clickhouse-format", headers) == "RowBinaryWithNamesAndTypes"
-
     # https://en.wikipedia.org/wiki/LZ4_(compression_algorithm)
-    assert <<0x28, 0xB5, 0x2F, 0xFD, _rest::bytes>> = IO.iodata_to_binary(rows)
+    assert <<0x28, 0xB5, 0x2F, 0xFD, _rest::bytes>> = IO.iodata_to_binary(data)
   end
 end

From df867e2fbaf6f8b3c75579da28c92b8caeacfa5f Mon Sep 17 00:00:00 2001
From: ruslandoga <67764432+ruslandoga@users.noreply.github.com>
Date: Fri, 29 Dec 2023 19:49:41 +0800
Subject: [PATCH 5/9] add pr to changelog

---
 CHANGELOG.md | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 161ae4f..bf1f46c 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -3,7 +3,7 @@
 ## Unreleased
 
 - move rows payload (RowBinary, CSV, etc.) to SQL statement and remove pseudo-positional binds, making param names explicit https://github.com/plausible/ch/pull/143
-- drop `:headers` from `%Ch.Result{}` but add `:data`
+- drop `:headers` from `%Ch.Result{}` but add `:data` https://github.com/plausible/ch/pull/144
 
 ## 0.2.2 (2023-12-23)
 

From 06665e6a3dfceade55083dfae264d55b46bb78e2 Mon Sep 17 00:00:00 2001
From: "dependabot[bot]" <49699333+dependabot[bot]@users.noreply.github.com>
Date: Fri, 29 Dec 2023 11:51:31 +0000
Subject: [PATCH 6/9] Bump dialyxir from 1.4.2 to 1.4.3

Bumps [dialyxir](https://github.com/jeremyjh/dialyxir) from 1.4.2 to 1.4.3.
- [Release notes](https://github.com/jeremyjh/dialyxir/releases)
- [Changelog](https://github.com/jeremyjh/dialyxir/blob/master/CHANGELOG.md)
- [Commits](https://github.com/jeremyjh/dialyxir/compare/1.4.2...1.4.3)

---
updated-dependencies:
- dependency-name: dialyxir
  dependency-type: direct:development
  update-type: version-update:semver-patch
...

Signed-off-by: dependabot[bot] <support@github.com>
---
 mix.lock | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/mix.lock b/mix.lock
index 0e8af8d..8c3b343 100644
--- a/mix.lock
+++ b/mix.lock
@@ -3,7 +3,7 @@
   "db_connection": {:hex, :db_connection, "2.6.0", "77d835c472b5b67fc4f29556dee74bf511bbafecdcaf98c27d27fa5918152086", [:mix], [{:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "c2f992d15725e721ec7fbc1189d4ecdb8afef76648c746a8e1cad35e3b8a35f3"},
   "decimal": {:hex, :decimal, "2.1.1", "5611dca5d4b2c3dd497dec8f68751f1f1a54755e8ed2a966c2633cf885973ad6", [:mix], [], "hexpm", "53cfe5f497ed0e7771ae1a475575603d77425099ba5faef9394932b35020ffcc"},
   "deep_merge": {:hex, :deep_merge, "1.0.0", "b4aa1a0d1acac393bdf38b2291af38cb1d4a52806cf7a4906f718e1feb5ee961", [:mix], [], "hexpm", "ce708e5f094b9cd4e8f2be4f00d2f4250c4095be93f8cd6d018c753894885430"},
-  "dialyxir": {:hex, :dialyxir, "1.4.2", "764a6e8e7a354f0ba95d58418178d486065ead1f69ad89782817c296d0d746a5", [:mix], [{:erlex, ">= 0.2.6", [hex: :erlex, repo: "hexpm", optional: false]}], "hexpm", "516603d8067b2fd585319e4b13d3674ad4f314a5902ba8130cd97dc902ce6bbd"},
+  "dialyxir": {:hex, :dialyxir, "1.4.3", "edd0124f358f0b9e95bfe53a9fcf806d615d8f838e2202a9f430d59566b6b53b", [:mix], [{:erlex, ">= 0.2.6", [hex: :erlex, repo: "hexpm", optional: false]}], "hexpm", "bf2cfb75cd5c5006bec30141b131663299c661a864ec7fbbc72dfa557487a986"},
   "earmark_parser": {:hex, :earmark_parser, "1.4.39", "424642f8335b05bb9eb611aa1564c148a8ee35c9c8a8bba6e129d51a3e3c6769", [:mix], [], "hexpm", "06553a88d1f1846da9ef066b87b57c6f605552cfbe40d20bd8d59cc6bde41944"},
   "ecto": {:hex, :ecto, "3.11.1", "4b4972b717e7ca83d30121b12998f5fcdc62ba0ed4f20fd390f16f3270d85c3e", [:mix], [{:decimal, "~> 2.0", [hex: :decimal, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "ebd3d3772cd0dfcd8d772659e41ed527c28b2a8bde4b00fe03e0463da0f1983b"},
   "erlex": {:hex, :erlex, "0.2.6", "c7987d15e899c7a2f34f5420d2a2ea0d659682c06ac607572df55a43753aa12e", [:mix], [], "hexpm", "2ed2e25711feb44d52b17d2780eabf998452f6efda104877a3881c2f8c0c0c75"},

From 6965b036e866a57ea021e56bd1efa6b824f2deae Mon Sep 17 00:00:00 2001
From: ruslandoga <67764432+ruslandoga@users.noreply.github.com>
Date: Fri, 29 Dec 2023 19:56:24 +0800
Subject: [PATCH 7/9] cleanup readme

---
 README.md | 19 ++++++++++++-------
 1 file changed, 12 insertions(+), 7 deletions(-)

diff --git a/README.md b/README.md
index e6121cf..7d84aeb 100644
--- a/README.md
+++ b/README.md
@@ -75,9 +75,6 @@ Ch.query!(pid, "CREATE TABLE IF NOT EXISTS ch_demo(id UInt64) ENGINE Null")
 %Ch.Result{num_rows: 2} =
   Ch.query!(pid, "INSERT INTO ch_demo(id) VALUES (0), (1)")
 
-%Ch.Result{num_rows: 2} =
-  Ch.query!(pid, "INSERT INTO ch_demo(id) VALUES ({$0:UInt8}), ({$1:UInt32})", [0, 1])
-
 %Ch.Result{num_rows: 2} =
   Ch.query!(pid, "INSERT INTO ch_demo(id) VALUES ({a:UInt16}), ({b:UInt64})", %{"a" => 0, "b" => 1})
 
@@ -117,7 +114,10 @@ header = Ch.RowBinary.encode_names_and_types(names, types)
 row_binary = Ch.RowBinary.encode_rows(rows, types)
 
 %Ch.Result{num_rows: 3} =
-  Ch.query!(pid, ["INSERT INTO ch_demo FORMAT RowBinaryWithNamesAndTypes\n", header | row_binary])
+  Ch.query!(pid, [
+    "INSERT INTO ch_demo FORMAT RowBinaryWithNamesAndTypes\n",
+    header | row_binary
+  ])
 ```
 
 #### Insert rows in custom [format](https://clickhouse.com/docs/en/interfaces/formats)
@@ -147,7 +147,12 @@ row_binary =
   |> Stream.take(10)
 
 %Ch.Result{num_rows: 1_000_000} =
-  Ch.query(pid, Stream.concat(["INSERT INTO ch_demo(id) FORMAT RowBinary\n"], row_binary))
+  Ch.query(pid,
+    Stream.concat(
+      ["INSERT INTO ch_demo(id) FORMAT RowBinary\n"],
+      row_binary
+    )
+  )
 ```
 
 #### Insert rows via [input](https://clickhouse.com/docs/en/sql-reference/table-functions/input) function
@@ -247,8 +252,8 @@ row_binary = Ch.RowBinary.encode(:string, "\x61\xF0\x80\x80\x80b")
 %Ch.Result{rows: [["a�b"]]} =
   Ch.query!(pid, "SELECT * FROM ch_utf8")
 
-%Ch.Result{rows: %{"data" => [["a�b"]]}} =
-  pid |> Ch.query!("SELECT * FROM ch_utf8 FORMAT JSONCompact") |> Map.update!(:rows, &Jason.decode!/1)
+%{"data" => [["a�b"]]} =
+  pid |> Ch.query!("SELECT * FROM ch_utf8 FORMAT JSONCompact").data |> Jason.decode!()
 ```
 
 #### Timezones in RowBinary

From 37317dc699d3a7f2fda883fdf78270e3cb87f483 Mon Sep 17 00:00:00 2001
From: ruslandoga <67764432+ruslandoga@users.noreply.github.com>
Date: Fri, 29 Dec 2023 21:09:23 +0800
Subject: [PATCH 8/9] stream

---
 README.md | 12 +++---------
 1 file changed, 3 insertions(+), 9 deletions(-)

diff --git a/README.md b/README.md
index 7d84aeb..e8cbf8c 100644
--- a/README.md
+++ b/README.md
@@ -140,19 +140,13 @@ csv = "0\n1"
 
 Ch.query!(pid, "CREATE TABLE IF NOT EXISTS ch_demo(id UInt64) ENGINE Null")
 
-row_binary =
+DBConnection.run(pid, fn conn ->
   Stream.repeatedly(fn -> [:rand.uniform(100)] end)
   |> Stream.chunk_every(100_000)
   |> Stream.map(fn chunk -> Ch.RowBinary.encode_rows(chunk, _types = ["UInt64"]) end)
   |> Stream.take(10)
-
-%Ch.Result{num_rows: 1_000_000} =
-  Ch.query(pid,
-    Stream.concat(
-      ["INSERT INTO ch_demo(id) FORMAT RowBinary\n"],
-      row_binary
-    )
-  )
+  |> Enum.into(Ch.stream(conn, "INSERT INTO ch_demo(id) FORMAT RowBinary"))
+end)
 ```
 
 #### Insert rows via [input](https://clickhouse.com/docs/en/sql-reference/table-functions/input) function

From e5c5b34e55b8b3e6c2a1e9693566a94e41193c2e Mon Sep 17 00:00:00 2001
From: ruslandoga <67764432+ruslandoga@users.noreply.github.com>
Date: Thu, 4 Jan 2024 11:29:28 +0900
Subject: [PATCH 9/9] wip

---
 README.md               | 20 ++++++++++++++--
 lib/ch.ex               |  8 ++++---
 lib/ch/connection.ex    |  3 ++-
 lib/ch/query.ex         | 12 +++++++++-
 lib/ch/stream.ex        | 44 ++++++++++++++++++++++++++++++++++
 test/ch/stream_test.exs | 52 +++++++++++++++++++++++++++++++++++++++++
 6 files changed, 132 insertions(+), 7 deletions(-)
 create mode 100644 lib/ch/stream.ex
 create mode 100644 test/ch/stream_test.exs

diff --git a/README.md b/README.md
index e8cbf8c..9998bbc 100644
--- a/README.md
+++ b/README.md
@@ -133,7 +133,7 @@ csv = "0\n1"
   Ch.query!(pid, ["INSERT INTO ch_demo(id) FORMAT CSV\n" | csv])
 ```
 
-#### Insert rows as [chunked](https://en.wikipedia.org/wiki/Chunked_transfer_encoding) RowBinary stream
+#### Insert [chunked](https://en.wikipedia.org/wiki/Chunked_transfer_encoding) RowBinary stream
 
 ```elixir
 {:ok, pid} = Ch.start_link()
@@ -145,10 +145,26 @@ DBConnection.run(pid, fn conn ->
   |> Stream.chunk_every(100_000)
   |> Stream.map(fn chunk -> Ch.RowBinary.encode_rows(chunk, _types = ["UInt64"]) end)
   |> Stream.take(10)
-  |> Enum.into(Ch.stream(conn, "INSERT INTO ch_demo(id) FORMAT RowBinary"))
+  |> Enum.into(Ch.stream(conn, "INSERT INTO ch_demo(id) FORMAT RowBinary\n"))
 end)
 ```
 
+#### Select rows as stream
+
+```elixir
+{:ok, pid} = Ch.start_link()
+
+DBConnection.run(pid, fn conn ->
+  Ch.stream(conn, "SELECT * FROM system.numbers LIMIT {limit:UInt64}", %{"limit" => 1_000_000})
+  |> Stream.each(&IO.inspect/1)
+  |> Stream.run()
+end)
+
+# %Ch.Result{rows: [[0], [1], [2], ...]}
+# %Ch.Result{rows: [[112123], [112124], [113125], ...]}
+# etc.
+```
+
 #### Insert rows via [input](https://clickhouse.com/docs/en/sql-reference/table-functions/input) function
 
 ```elixir
diff --git a/lib/ch.ex b/lib/ch.ex
index 0a8ac52..b5c60ca 100644
--- a/lib/ch.ex
+++ b/lib/ch.ex
@@ -99,11 +99,13 @@ defmodule Ch do
     DBConnection.execute!(conn, query, params, opts)
   end
 
-  @doc false
-  @spec stream(DBConnection.t(), statement, params, [query_option]) :: DBConnection.Stream.t()
+  @doc """
+  Returns a stream for a query on a connection.
+  """
+  @spec stream(DBConnection.t(), statement, params, [query_option]) :: Ch.Stream.t()
   def stream(conn, statement, params \\ [], opts \\ []) do
     query = Query.build(statement, opts)
-    DBConnection.stream(conn, query, params, opts)
+    %Ch.Stream{conn: conn, query: query, params: params, opts: opts}
   end
 
   if Code.ensure_loaded?(Ecto.ParameterizedType) do
diff --git a/lib/ch/connection.ex b/lib/ch/connection.ex
index ad8ea4b..dd9e168 100644
--- a/lib/ch/connection.ex
+++ b/lib/ch/connection.ex
@@ -85,7 +85,8 @@ defmodule Ch.Connection do
 
   @impl true
   def handle_declare(query, params, opts, conn) do
-    {query_params, extra_headers, body} = params
+    {query_params, extra_headers} = params
+    body = query.statement
 
     path = path(conn, query_params, opts)
     headers = headers(conn, extra_headers, opts)
diff --git a/lib/ch/query.ex b/lib/ch/query.ex
index 2685e70..3d82371 100644
--- a/lib/ch/query.ex
+++ b/lib/ch/query.ex
@@ -86,7 +86,7 @@ defimpl DBConnection.Query, for: Ch.Query do
 
   @spec decode(Query.t(), [response], [Ch.query_option()]) :: Result.t()
         when response: Mint.Types.status() | Mint.Types.headers() | binary
-  def decode(%Query{command: command}, responses, opts) do
+  def decode(%Query{command: command}, responses, opts) when is_list(responses) do
     [_status, headers | data] = responses
     format = get_header(headers, "x-clickhouse-format")
     decode = Keyword.get(opts, :decode, true)
@@ -110,6 +110,16 @@ defimpl DBConnection.Query, for: Ch.Query do
     end
   end
 
+  def decode(%Query{command: command}, {:stream, _types, responses}, _opts) do
+    %Result{command: command, data: stream_responses(responses)}
+  end
+
+  defp stream_responses([{:status, _, _} | rest]), do: stream_responses(rest)
+  defp stream_responses([{:headers, _, _} | rest]), do: stream_responses(rest)
+  defp stream_responses([{:data, _, data} | rest]), do: [data | stream_responses(rest)]
+  defp stream_responses([{:done, _}]), do: []
+  defp stream_responses([]), do: []
+
   defp get_header(headers, key) do
     case List.keyfind(headers, key, 0) do
       {_, value} -> value
diff --git a/lib/ch/stream.ex b/lib/ch/stream.ex
new file mode 100644
index 0000000..78a0c55
--- /dev/null
+++ b/lib/ch/stream.ex
@@ -0,0 +1,44 @@
+defmodule Ch.Stream do
+  @moduledoc """
+  Stream struct returned from stream commands.
+
+  All of its fields are private.
+  """
+
+  @derive {Inspect, only: []}
+  defstruct [:conn, :query, :params, :opts]
+
+  @type t :: %__MODULE__{
+          conn: DBConnection.conn(),
+          query: Ch.Query.t(),
+          params: Ch.params(),
+          opts: [Ch.query_option()]
+        }
+
+  defimpl Enumerable do
+    def reduce(stream, acc, fun) do
+      %Ch.Stream{conn: conn, query: query, params: params, opts: opts} = stream
+      stream = %DBConnection.Stream{conn: conn, query: query, params: params, opts: opts}
+      DBConnection.reduce(stream, acc, fun)
+    end
+
+    def member?(_, _), do: {:error, __MODULE__}
+    def count(_), do: {:error, __MODULE__}
+    def slice(_), do: {:error, __MODULE__}
+  end
+
+  defimpl Collectable do
+    def into(stream) do
+      %Ch.Stream{conn: conn, query: query, params: params, opts: opts} = stream
+      DBConnection.execute!(conn, query, params, opts)
+      {stream, &collect/2}
+    end
+
+    defp collect(%{conn: conn, query: query} = stream, {:cont, data}) do
+      DBConnection.execute!(conn, %{query | statement: data}, [])
+      stream
+    end
+
+    defp collect(conn, :done), do: HTTP.stream_request_body(conn, ref(conn), :eof)
+  end
+end
diff --git a/test/ch/stream_test.exs b/test/ch/stream_test.exs
new file mode 100644
index 0000000..b49f863
--- /dev/null
+++ b/test/ch/stream_test.exs
@@ -0,0 +1,52 @@
+defmodule Ch.StreamTest do
+  use ExUnit.Case
+
+  setup do
+    pool = start_supervised!({Ch, database: Ch.Test.database()})
+    {:ok, pool: pool}
+  end
+
+  describe "enumerable" do
+    test "works", %{pool: pool} do
+      result =
+        DBConnection.run(pool, fn conn ->
+          conn
+          |> Ch.stream("select * from system.numbers limit {limit:UInt32}", %{"limit" => 100})
+          |> Enum.into([])
+        end)
+
+      assert [
+               %Ch.Result{
+                 command: :select,
+                 data: [
+                   <<1, 6, 110, 117, 109, 98, 101, 114, 6, 85, 73, 110, 116, 54, 52>>,
+                   <<_::6400>>
+                 ]
+               },
+               %Ch.Result{command: :select, data: []}
+             ] = result
+    end
+  end
+
+  describe "collectable" do
+    test "works", %{pool: pool} do
+      Ch.query!(pool, "create table collectable_test(a UInt64, b String) engine Null")
+
+      rows =
+        Stream.repeatedly(fn -> [0, "0"] end)
+        |> Stream.chunk_every(10000)
+        |> Stream.map(fn chunk -> Ch.RowBinary.encode_rows(chunk, ["UInt64", "String"]) end)
+        |> Stream.take(3)
+
+      result =
+        DBConnection.run(pool, fn conn ->
+          Enum.into(
+            rows,
+            Ch.stream(conn, "insert into collectable_test(a, b) format RowBinary\n")
+          )
+        end)
+
+      assert result == :asdlkfhajsdf
+    end
+  end
+end