Add support for :adbc format

lib/req_ch.ex

@@ -26,13 +26,13 @@ defmodule ReqCH do
   It is by default "http://localhost:8123".
 
     * `:format` - Optional. The format of the response. Default is `:tsv`.
-      This option accepts `:tsv`, `:csv`, `:json` or `:explorer` as atoms.
+      This option accepts `:tsv`, `:csv`, `:json`, `:explorer` and `:adbc` as atoms.
 
       It also accepts all formats described in the #{@formats_page} page.
       Use plain strings for these formats.
 
-      The `:explorer` format is special, and will build an Explorer dataframe
-      in case the `:explorer` dependency is installed.
+      The `:adbc` and `:explorer` formats are special, and will build an Result struct
+      in case the required dependency is installed.
 
     * `:database` - Optional. The database to use in the queries.
       Default is `nil`.
@@ -88,7 +88,29 @@ defmodule ReqCH do
       iex> response.body
       "0\\n1\\n2\\n"
 
-  With a specific format:
+  With the `:adbc` format:
+
+      iex> req = ReqCH.new(database: "system")
+      iex> {:ok, response} = ReqCH.query(req, "SELECT number FROM numbers LIMIT 3", [], [format: :adbc])
+      iex> response.body
+      %Adbc.Result{
+         data: [
+          [
+            %Adbc.Column{
+              field: %Adbc.Field{name: "number", type: :u64, metadata: nil},
+              data: %Adbc.BufferData{
+                data: <<0, 0, 0, 0, 0, 0, 0, 0, 1, 0, 0, 0, 0, 0, 0, 0, 2, 0, 0, 0, 0, 0, 0, 0>>,
+                validity: nil,
+                bit_offset: 0
+              },
+              size: nil
+            }
+          ]
+         ],
+         num_rows: nil
+       }
+
+  With the `:explorer` format:
 
       iex> req = ReqCH.new(database: "system")
       iex> {:ok, response} = ReqCH.query(req, "SELECT number FROM numbers LIMIT 3", [], [format: :explorer])
@@ -152,9 +174,18 @@ defmodule ReqCH do
   defp run(%Req.Request{} = request) do
     request = update_in(request.options, &Map.put_new(&1, :base_url, "http://localhost:8123"))
 
-    with %Req.Request{} = req1 <- add_format(request),
-         %Req.Request{} = req2 <- maybe_add_database(req1) do
-      Req.Request.append_response_steps(req2, clickhouse_result: &handle_clickhouse_result/1)
+    request
+    |> add_format()
+    |> maybe_remove_compression()
+    |> maybe_add_database()
+    |> Req.Request.append_response_steps(clickhouse_result: &handle_clickhouse_result/1)
+  end
+
+  defp maybe_remove_compression(request) do
+    if Req.Request.get_private(request, :clickhouse_format) == :adbc do
+      put_params(request, output_format_arrow_compression_method: "none")
+    else
+      request
     end
   end
 
@@ -231,41 +262,32 @@ defmodule ReqCH do
     [key, ?:, value]
   end
 
-  @valid_formats [:tsv, :csv, :json, :explorer]
+  @valid_formats [:tsv, :csv, :json, :explorer, :adbc]
 
   defp add_format(%Req.Request{} = request) do
-    format_option = Req.Request.get_option(request, :format, :tsv)
-    format = normalise_format(format_option)
+    format = Req.Request.get_option(request, :format, :tsv)
+    :ok = ensure_format!(format)
 
-    if format do
-      format_header = with :explorer <- format, do: "Parquet"
-
-      request
-      |> Req.Request.put_private(:clickhouse_format, format)
-      |> Req.Request.put_header("x-clickhouse-format", format_header)
-    else
-      raise ArgumentError,
-            "the given format #{inspect(format_option)} is invalid. Expecting one of #{inspect(@valid_formats)} " <>
-              "or one of the valid options described in #{@formats_page}"
-    end
+    request
+    |> Req.Request.put_private(:clickhouse_format, format)
+    |> Req.Request.put_header("x-clickhouse-format", format_to_string(format))
   end
 
-  defp normalise_format(:tsv), do: "TabSeparated"
-  defp normalise_format(:csv), do: "CSV"
-  defp normalise_format(:json), do: "JSON"
+  defp ensure_format!(format) when format in @supported_formats, do: :ok
+  defp ensure_format!(format) when format in @valid_formats, do: :ok
 
-  if Code.ensure_loaded?(Explorer) do
-    defp normalise_format(:explorer), do: :explorer
-  else
-    defp normalise_format(:explorer) do
-      raise ArgumentError,
-            "format: :explorer - you need to install Explorer as a dependency in order to use this format"
-    end
+  defp ensure_format!(format) do
+    raise ArgumentError,
+          "the given format #{inspect(format)} is invalid. Expecting one of #{inspect(@valid_formats)} " <>
+            "or one of the valid options described in #{@formats_page}"
   end
 
-  defp normalise_format(format) when format in @supported_formats, do: format
-
-  defp normalise_format(_), do: nil
+  defp format_to_string(:tsv), do: "TabSeparated"
+  defp format_to_string(:csv), do: "CSV"
+  defp format_to_string(:json), do: "JSON"
+  defp format_to_string(:adbc), do: "ArrowStream"
+  defp format_to_string(:explorer), do: "Parquet"
+  defp format_to_string(format) when format in @supported_formats, do: format
 
   defp maybe_add_database(%Req.Request{} = request) do
     if database = Req.Request.get_option(request, :database) do
@@ -276,18 +298,34 @@ defmodule ReqCH do
   end
 
   defp handle_clickhouse_result({request, %{status: 200} = response} = pair) do
-    want_explorer_df = Req.Request.get_private(request, :clickhouse_format) == :explorer
-    is_parquet_response = response.headers["x-clickhouse-format"] == ["Parquet"]
+    format = Req.Request.get_private(request, :clickhouse_format)
+    format_header = Req.Response.get_header(response, "x-clickhouse-format")
 
-    if want_explorer_df and is_parquet_response do
-      Req.Request.halt(request, update_in(response.body, &load_parquet/1))
-    else
-      pair
+    case {format, format_header} do
+      {:adbc, ["ArrowStream"]} ->
+        Req.Request.halt(request, update_in(response.body, &load_arrow/1))
+
+      {:explorer, ["Parquet"]} ->
+        Req.Request.halt(request, update_in(response.body, &load_parquet/1))
+
+      _otherwise ->
+        pair
     end
   end
 
   defp handle_clickhouse_result(request_response), do: request_response
 
+  if Code.ensure_loaded?(Adbc) do
+    defp load_arrow(body) do
+      Adbc.Result.from_ipc_stream!(body)
+    end
+  else
+    defp load_arrow(_body) do
+      raise ArgumentError,
+            "format: :adbc - you need to install Adbc as a dependency in order to use this format"
+    end
+  end
+
   if Code.ensure_loaded?(Explorer) do
     defp load_parquet(body) do
       Explorer.DataFrame.load_parquet!(body)

mix.exs

@@ -26,7 +26,8 @@ defmodule ReqCH.MixProject do
   defp deps do
     [
       {:req, "~> 0.5"},
-      {:explorer, "~> 0.10", optional: true},
+      {:explorer, "~> 0.11", optional: true},
+      {:adbc, "~> 0.12", optional: true},
       {:ex_doc, ">= 0.0.0", only: :docs, runtime: false}
     ]
   end

mix.lock

@@ -1,11 +1,15 @@
 %{
-  "aws_signature": {:hex, :aws_signature, "0.3.2", "adf33bc4af00b2089b7708bf20e3246f09c639a905a619b3689f0a0a22c3ef8f", [:rebar3], [], "hexpm", "b0daf61feb4250a8ab0adea60db3e336af732ff71dd3fb22e45ae3dcbd071e44"},
+  "adbc": {:hex, :adbc, "0.12.1", "64f04b2d4787f30b2ce1b9c5ff2f7ee393e611fbb8a8829fab402b767c133fa5", [:make, :mix], [{:cc_precompiler, "~> 0.1.8 or ~> 0.2", [hex: :cc_precompiler, repo: "hexpm", optional: false]}, {:decimal, "~> 2.1 or ~> 3.0", [hex: :decimal, repo: "hexpm", optional: false]}, {:elixir_make, "~> 0.9", [hex: :elixir_make, repo: "hexpm", optional: false]}, {:fine, "~> 0.1.0", [hex: :fine, repo: "hexpm", optional: false]}, {:pythonx, "~> 0.4.0", [hex: :pythonx, repo: "hexpm", optional: true]}, {:table, "~> 0.1.2", [hex: :table, repo: "hexpm", optional: false]}], "hexpm", "99a4f46e8af1c9dfc94c804c9f1d9e6c1923845a90af089765ea278d876da904"},
+  "aws_signature": {:hex, :aws_signature, "0.4.2", "1b35482c89ff5b91f5ead647a2bbc0d9620877479b44800915de92bacf9f1476", [:rebar3], [], "hexpm", "1df4a2d1dff200c7bdfa8f9f935efc71a51273adfc6dd39a9f2cc937e01baa01"},
   "castore": {:hex, :castore, "1.0.10", "43bbeeac820f16c89f79721af1b3e092399b3a1ecc8df1a472738fd853574911", [:mix], [], "hexpm", "1b0b7ea14d889d9ea21202c43a4fa015eb913021cb535e8ed91946f4b77a8848"},
-  "decimal": {:hex, :decimal, "2.2.0", "df3d06bb9517e302b1bd265c1e7f16cda51547ad9d99892049340841f3e15836", [:mix], [], "hexpm", "af8daf87384b51b7e611fb1a1f2c4d4876b65ef968fa8bd3adf44cff401c7f21"},
+  "cc_precompiler": {:hex, :cc_precompiler, "0.1.11", "8c844d0b9fb98a3edea067f94f616b3f6b29b959b6b3bf25fee94ffe34364768", [:mix], [{:elixir_make, "~> 0.7", [hex: :elixir_make, repo: "hexpm", optional: false]}], "hexpm", "3427232caf0835f94680e5bcf082408a70b48ad68a5f5c0b02a3bea9f3a075b9"},
+  "decimal": {:hex, :decimal, "2.4.1", "6c0fbede12fb122ba685e9ab41c6a40c129e322b3aa192f9e072e61f3a6ffaf2", [:mix], [], "hexpm", "7e618897933a8455f19a727d7c5e50a2c071a544b700e5e724298ecb4340187f"},
   "earmark_parser": {:hex, :earmark_parser, "1.4.41", "ab34711c9dc6212dda44fcd20ecb87ac3f3fce6f0ca2f28d4a00e4154f8cd599", [:mix], [], "hexpm", "a81a04c7e34b6617c2792e291b5a2e57ab316365c2644ddc553bb9ed863ebefa"},
+  "elixir_make": {:hex, :elixir_make, "0.10.0", "16577e2583a79bb79237bbff349619ef5d80afffc07eac6e4faf0d00e2ddaf7d", [:mix], [], "hexpm", "dc1f09fb7fa68866b886abd5f0f3c83553b1a19a52359a899e92af1bb3b31982"},
   "ex_doc": {:hex, :ex_doc, "0.35.1", "de804c590d3df2d9d5b8aec77d758b00c814b356119b3d4455e4b8a8687aecaf", [:mix], [{:earmark_parser, "~> 1.4.39", [hex: :earmark_parser, repo: "hexpm", optional: false]}, {:makeup_c, ">= 0.1.0", [hex: :makeup_c, repo: "hexpm", optional: true]}, {:makeup_elixir, "~> 0.14 or ~> 1.0", [hex: :makeup_elixir, repo: "hexpm", optional: false]}, {:makeup_erlang, "~> 0.1 or ~> 1.0", [hex: :makeup_erlang, repo: "hexpm", optional: false]}, {:makeup_html, ">= 0.1.0", [hex: :makeup_html, repo: "hexpm", optional: true]}], "hexpm", "2121c6402c8d44b05622677b761371a759143b958c6c19f6558ff64d0aed40df"},
-  "explorer": {:hex, :explorer, "0.10.0", "ba690afb59fce81746a1b6c9d25294aabcb9bae783ceeb43f5fd4834e1e16d78", [:mix], [{:adbc, "~> 0.1", [hex: :adbc, repo: "hexpm", optional: true]}, {:aws_signature, "~> 0.3", [hex: :aws_signature, repo: "hexpm", optional: false]}, {:castore, "~> 1.0", [hex: :castore, repo: "hexpm", optional: true]}, {:decimal, "~> 2.1", [hex: :decimal, repo: "hexpm", optional: false]}, {:flame, "~> 0.3", [hex: :flame, repo: "hexpm", optional: true]}, {:fss, "~> 0.1", [hex: :fss, repo: "hexpm", optional: false]}, {:nx, "~> 0.4", [hex: :nx, repo: "hexpm", optional: true]}, {:rustler, "~> 0.34.0", [hex: :rustler, repo: "hexpm", optional: true]}, {:rustler_precompiled, "~> 0.7", [hex: :rustler_precompiled, repo: "hexpm", optional: false]}, {:table, "~> 0.1.2", [hex: :table, repo: "hexpm", optional: false]}, {:table_rex, "~> 3.1.1 or ~> 4.0.0", [hex: :table_rex, repo: "hexpm", optional: false]}], "hexpm", "874b6af1f711186f391b507af293995f89311821486e01cbca4c99777ff20864"},
+  "explorer": {:hex, :explorer, "0.11.1", "f8fe87cdf37c4cca8fe7120fb5806f8327497f61290c18f9d4f33cef6b2fc5a0", [:mix], [{:adbc, "~> 0.1", [hex: :adbc, repo: "hexpm", optional: true]}, {:aws_signature, "~> 0.3", [hex: :aws_signature, repo: "hexpm", optional: false]}, {:decimal, "~> 2.1", [hex: :decimal, repo: "hexpm", optional: false]}, {:flame, "~> 0.3", [hex: :flame, repo: "hexpm", optional: true]}, {:fss, "~> 0.1", [hex: :fss, repo: "hexpm", optional: false]}, {:nx, "~> 0.4", [hex: :nx, repo: "hexpm", optional: true]}, {:rustler, "~> 0.36.0", [hex: :rustler, repo: "hexpm", optional: true]}, {:rustler_precompiled, "~> 0.7", [hex: :rustler_precompiled, repo: "hexpm", optional: false]}, {:table, "~> 0.1.2", [hex: :table, repo: "hexpm", optional: false]}, {:table_rex, "~> 4.1", [hex: :table_rex, repo: "hexpm", optional: false]}], "hexpm", "acc679ea15790d03d9a406bb45284bd4e30531d01a650d9194393cbadcdefccd"},
   "finch": {:hex, :finch, "0.19.0", "c644641491ea854fc5c1bbaef36bfc764e3f08e7185e1f084e35e0672241b76d", [:mix], [{:mime, "~> 1.0 or ~> 2.0", [hex: :mime, repo: "hexpm", optional: false]}, {:mint, "~> 1.6.2 or ~> 1.7", [hex: :mint, repo: "hexpm", optional: false]}, {:nimble_options, "~> 0.4 or ~> 1.0", [hex: :nimble_options, repo: "hexpm", optional: false]}, {:nimble_pool, "~> 1.1", [hex: :nimble_pool, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "fc5324ce209125d1e2fa0fcd2634601c52a787aff1cd33ee833664a5af4ea2b6"},
+  "fine": {:hex, :fine, "0.1.6", "4bf7151493443c454aac9f2fa2f34f5fefd0346a83fb5586a016c4a135c63247", [:mix], [], "hexpm", "5638eb4495488e885ebec167fa57973e5c35e1a50c344eb7666c90ec1c4e3b12"},
   "fss": {:hex, :fss, "0.1.1", "9db2344dbbb5d555ce442ac7c2f82dd975b605b50d169314a20f08ed21e08642", [:mix], [], "hexpm", "78ad5955c7919c3764065b21144913df7515d52e228c09427a004afe9c1a16b0"},
   "hpax": {:hex, :hpax, "1.0.0", "28dcf54509fe2152a3d040e4e3df5b265dcb6cb532029ecbacf4ce52caea3fd2", [:mix], [], "hexpm", "7f1314731d711e2ca5fdc7fd361296593fc2542570b3105595bb0bc6d0fad601"},
   "jason": {:hex, :jason, "1.4.4", "b9226785a9aa77b6857ca22832cffa5d5011a667207eb2a0ad56adb5db443b8a", [:mix], [{:decimal, "~> 1.0 or ~> 2.0", [hex: :decimal, repo: "hexpm", optional: true]}], "hexpm", "c5eb0cab91f094599f94d55bc63409236a8ec69a21a67814529e8d5f6cc90b3b"},
@@ -18,8 +22,8 @@
   "nimble_parsec": {:hex, :nimble_parsec, "1.4.0", "51f9b613ea62cfa97b25ccc2c1b4216e81df970acd8e16e8d1bdc58fef21370d", [:mix], [], "hexpm", "9c565862810fb383e9838c1dd2d7d2c437b3d13b267414ba6af33e50d2d1cf28"},
   "nimble_pool": {:hex, :nimble_pool, "1.1.0", "bf9c29fbdcba3564a8b800d1eeb5a3c58f36e1e11d7b7fb2e084a643f645f06b", [:mix], [], "hexpm", "af2e4e6b34197db81f7aad230c1118eac993acc0dae6bc83bac0126d4ae0813a"},
   "req": {:hex, :req, "0.5.7", "b722680e03d531a2947282adff474362a48a02aa54b131196fbf7acaff5e4cee", [:mix], [{:brotli, "~> 0.3.1", [hex: :brotli, repo: "hexpm", optional: true]}, {:ezstd, "~> 1.0", [hex: :ezstd, repo: "hexpm", optional: true]}, {:finch, "~> 0.17", [hex: :finch, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: false]}, {:mime, "~> 2.0.6 or ~> 2.1", [hex: :mime, repo: "hexpm", optional: false]}, {:nimble_csv, "~> 1.0", [hex: :nimble_csv, repo: "hexpm", optional: true]}, {:plug, "~> 1.0", [hex: :plug, repo: "hexpm", optional: true]}], "hexpm", "c6035374615120a8923e8089d0c21a3496cf9eda2d287b806081b8f323ceee29"},
-  "rustler_precompiled": {:hex, :rustler_precompiled, "0.8.2", "5f25cbe220a8fac3e7ad62e6f950fcdca5a5a5f8501835d2823e8c74bf4268d5", [:mix], [{:castore, "~> 0.1 or ~> 1.0", [hex: :castore, repo: "hexpm", optional: false]}, {:rustler, "~> 0.23", [hex: :rustler, repo: "hexpm", optional: true]}], "hexpm", "63d1bd5f8e23096d1ff851839923162096364bac8656a4a3c00d1fff8e83ee0a"},
+  "rustler_precompiled": {:hex, :rustler_precompiled, "0.9.0", "3a052eda09f3d2436364645cc1f13279cf95db310eb0c17b0d8f25484b233aa0", [:mix], [{:rustler, "~> 0.23", [hex: :rustler, repo: "hexpm", optional: true]}], "hexpm", "471d97315bd3bf7b64623418b3693eedd8e47de3d1cb79a0ac8f9da7d770d94c"},
   "table": {:hex, :table, "0.1.2", "87ad1125f5b70c5dea0307aa633194083eb5182ec537efc94e96af08937e14a8", [:mix], [], "hexpm", "7e99bc7efef806315c7e65640724bf165c3061cdc5d854060f74468367065029"},
-  "table_rex": {:hex, :table_rex, "4.0.0", "3c613a68ebdc6d4d1e731bc973c233500974ec3993c99fcdabb210407b90959b", [:mix], [], "hexpm", "c35c4d5612ca49ebb0344ea10387da4d2afe278387d4019e4d8111e815df8f55"},
+  "table_rex": {:hex, :table_rex, "4.1.0", "fbaa8b1ce154c9772012bf445bfb86b587430fb96f3b12022d3f35ee4a68c918", [:mix], [], "hexpm", "95932701df195d43bc2d1c6531178fc8338aa8f38c80f098504d529c43bc2601"},
   "telemetry": {:hex, :telemetry, "1.3.0", "fedebbae410d715cf8e7062c96a1ef32ec22e764197f70cda73d82778d61e7a2", [:rebar3], [], "hexpm", "7015fc8919dbe63764f4b4b87a95b7c0996bd539e0d499be6ec9d7f3875b79e6"},
 }

test/req_ch_test.exs

@@ -114,12 +114,32 @@ defmodule ReqCHTest do
                }
     end
 
-    test "with format option as :explorer but different format in the query" do
+    test "with format option as :adbc" do
+      req = ReqCH.new()
+
+      assert {:ok, %Req.Response{} = response} =
+               ReqCH.query(
+                 req,
+                 "SELECT number, number - 2 as less_two from system.numbers LIMIT 10",
+                 [],
+                 format: :adbc
+               )
+
+      assert %Adbc.Result{} = result = response.body
+
+      assert Adbc.Result.to_map(result) ==
+               %{
+                 "number" => [0, 1, 2, 3, 4, 5, 6, 7, 8, 9],
+                 "less_two" => [-2, -1, 0, 1, 2, 3, 4, 5, 6, 7]
+               }
+    end
+
+    test "with format option as :adbc but different format in the query" do
       assert {:ok, %Req.Response{} = response} =
                ReqCH.query(
                  ReqCH.new(),
                  "SELECT number, number - 2 as less_two from system.numbers LIMIT 10 FORMAT JSON",
-                 format: :explorer
+                 format: :adbc
                )
 
       assert response.status == 200
@@ -179,7 +199,7 @@ defmodule ReqCHTest do
 
     test "with invalid format" do
       error_message =
-        "the given format :invalid_format is invalid. Expecting one of [:tsv, :csv, :json, :explorer] " <>
+        "the given format :invalid_format is invalid. Expecting one of [:tsv, :csv, :json, :explorer, :adbc] " <>
           "or one of the valid options described in https://clickhouse.com/docs/en/interfaces/formats"
 
       assert_raise ArgumentError, error_message, fn ->