Replace Ch with ReqCH for ClickHouse integration (#83)

Co-authored-by: Jonatan Kłosko <[email protected]>

Author: Philip Sampaio

.github/workflows/test.yml

@@ -12,18 +12,23 @@ jobs:
       matrix:
         include:
           - pair:
-              elixir: 1.14.2
-              otp: 25.0
+              elixir: "1.14.2"
+              otp: "25.0"
+
+          - pair:
+              elixir: "1.17.3"
+              otp: "27.1.2"
             lint: true
+
     steps:
-      - uses: actions/checkout@v3
+      - uses: actions/checkout@v4
 
       - uses: erlef/setup-beam@v1
         with:
           otp-version: ${{matrix.pair.otp}}
           elixir-version: ${{matrix.pair.elixir}}
 
-      - uses: actions/cache@v3
+      - uses: actions/cache@v4
         with:
           path: |
             deps

lib/assets/connection_cell/main.js

@@ -821,13 +821,10 @@ export function init(ctx, info) {
 
     template: `
     <div class="row">
-      <BaseInput
-        name="schema"
-        label="Schema"
-        type="text"
-        v-model="fields.schema"
-        inputClass="input"
-        :grow
+      <BaseSwitch
+        name="use_ssl"
+        label="HTTPS"
+        v-model="fields.use_ssl"
       />
       <BaseInput
         name="hostname"
@@ -847,7 +844,7 @@ export function init(ctx, info) {
         :grow
         :required
       />
-     </div>
+    </div>
     <div class="row">
       <BaseInput
         name="database"

lib/kino_db/connection_cell.ex

@@ -150,7 +150,9 @@ defmodule KinoDB.ConnectionCell do
               ~w|database hostname port use_ipv6 username password use_ssl cacertfile instance|
 
         "clickhouse" ->
-          ~w|scheme username password_secret hostname port database|
+          if fields["use_password_secret"],
+            do: ~w|hostname port use_ssl username password_secret database|,
+            else: ~w|hostname port use_ssl username password database|
 
         type when type in ["postgres", "mysql"] ->
           if fields["use_password_secret"],
@@ -330,10 +332,15 @@ defmodule KinoDB.ConnectionCell do
   end
 
   defp to_quoted(%{"type" => "clickhouse"} = attrs) do
+    trimmed = attrs |> trim_opts() |> Map.new()
+    shared_opts = shared_options(trimmed)
+
+    clickhouse_opts = trimmed |> clickhouse_options(shared_opts)
+
     quote do
-      opts = unquote(trim_opts(shared_options(attrs) ++ clickhouse_options(attrs)))
+      unquote(quoted_var(attrs["variable"])) = ReqCH.new(unquote(clickhouse_opts))
 
-      {:ok, unquote(quoted_var(attrs["variable"]))} = Kino.start_child({Ch, opts})
+      :ok
     end
   end
 
@@ -438,9 +445,58 @@ defmodule KinoDB.ConnectionCell do
   end
 
   defp clickhouse_options(attrs) do
-    [
-      scheme: attrs["scheme"] || "http"
-    ]
+    scheme = if attrs["use_ssl"], do: "https", else: "http"
+
+    [scheme: scheme]
+  end
+
+  defp clickhouse_options(attrs, shared_options) do
+    attrs
+    |> clickhouse_options()
+    |> build_clickhouse_base_url(shared_options)
+    |> maybe_add_req_basic_auth(shared_options)
+    |> maybe_add_clickhouse_database(shared_options)
+  end
+
+  defp build_clickhouse_base_url(opts, shared_opts) do
+    host = Keyword.fetch!(shared_opts, :hostname)
+    port = Keyword.fetch!(shared_opts, :port)
+    scheme = Keyword.fetch!(opts, :scheme)
+
+    uri = %URI{scheme: scheme, port: port, host: host}
+
+    opts
+    |> Keyword.put_new(:base_url, URI.to_string(uri))
+    |> Keyword.delete(:scheme)
+  end
+
+  defp maybe_add_req_basic_auth(opts, shared_opts) do
+    username = shared_opts[:username]
+
+    if username != "" do
+      password = shared_opts[:password]
+
+      auth =
+        if is_binary(password) do
+          "#{username}:#{password}"
+        else
+          quote do
+            unquote(username) <> ":" <> unquote(password)
+          end
+        end
+
+      Keyword.put_new(opts, :auth, {:basic, auth})
+    else
+      opts
+    end
+  end
+
+  defp maybe_add_clickhouse_database(opts, shared_opts) do
+    if shared_opts[:database] != "" do
+      Keyword.put_new(opts, :database, shared_opts[:database])
+    else
+      opts
+    end
   end
 
   defp quoted_var(string), do: {String.to_atom(string), [], nil}
@@ -462,6 +518,7 @@ defmodule KinoDB.ConnectionCell do
       Code.ensure_loaded?(Exqlite) -> "sqlite"
       Code.ensure_loaded?(ReqBigQuery) -> "bigquery"
       Code.ensure_loaded?(ReqAthena) -> "athena"
+      Code.ensure_loaded?(ReqCH) -> "clickhouse"
       Code.ensure_loaded?(Adbc) -> "duckdb"
       Code.ensure_loaded?(Tds) -> "sqlserver"
       true -> "postgres"
@@ -493,16 +550,10 @@ defmodule KinoDB.ConnectionCell do
   end
 
   defp missing_dep(%{"type" => "athena"}) do
-    deps = [
+    missing_many_deps([
       {ReqAthena, ~s|{:req_athena, "~> 0.1"}|},
       {Explorer, ~s|{:explorer, "~> 0.9"}|}
-    ]
-
-    deps = for {module, dep} <- deps, not Code.ensure_loaded?(module), do: dep
-
-    if deps != [] do
-      Enum.join(deps, ", ")
-    end
+    ])
   end
 
   defp missing_dep(%{"type" => adbc}) when adbc in ~w[snowflake duckdb] do
@@ -518,13 +569,22 @@ defmodule KinoDB.ConnectionCell do
   end
 
   defp missing_dep(%{"type" => "clickhouse"}) do
-    unless Code.ensure_loaded?(Ch) do
-      ~s|{:ch, "~> 0.2"}|
-    end
+    missing_many_deps([
+      {ReqCH, ~s|{:req_ch, "~> 0.1"}|},
+      {Explorer, ~s|{:explorer, "~> 0.10"}|}
+    ])
   end
 
   defp missing_dep(_ctx), do: nil
 
+  defp missing_many_deps(deps) do
+    deps = for {module, dep} <- deps, not Code.ensure_loaded?(module), do: dep
+
+    if deps != [] do
+      Enum.join(deps, ", ")
+    end
+  end
+
   defp join_quoted(quoted_blocks) do
     asts =
       Enum.flat_map(quoted_blocks, fn

lib/kino_db/sql_cell.ex

@@ -167,6 +167,7 @@ defmodule KinoDB.SQLCell do
     cond do
       Keyword.has_key?(connection.request_steps, :bigquery_run) -> "bigquery"
       Keyword.has_key?(connection.request_steps, :athena_run) -> "athena"
+      Keyword.has_key?(connection.request_steps, :clickhouse_run) -> "clickhouse"
       true -> nil
     end
   end
@@ -219,8 +220,18 @@ defmodule KinoDB.SQLCell do
     to_quoted(attrs, quote(do: Tds), fn n -> "@#{n}" end)
   end
 
+  # query!/4 based that returns a Req response.
   defp to_quoted(%{"connection" => %{"type" => "clickhouse"}} = attrs) do
-    to_quoted(attrs, quote(do: Ch), fn n -> "{$#{n}:String}" end)
+    to_quoted_query_req(attrs, quote(do: ReqCH), fn n, inner ->
+      name =
+        if String.match?(inner, ~r/[^a-z0-9_]/) do
+          "param_#{n}"
+        else
+          inner
+        end
+
+      "{#{name}:String}"
+    end)
   end
 
   # Explorer-based
@@ -246,6 +257,21 @@ defmodule KinoDB.SQLCell do
     end
   end
 
+  defp to_quoted_query_req(attrs, quoted_module, next) do
+    {query, params} = parameterize(attrs["query"], attrs["connection"]["type"], next)
+    opts_args = query_opts_args(attrs)
+
+    quote do
+      unquote(quoted_var(attrs["result_variable"])) =
+        unquote(quoted_module).query!(
+          unquote(quoted_var(attrs["connection"]["variable"])),
+          unquote(quoted_query(query)),
+          unquote(params),
+          unquote_splicing(opts_args)
+        ).body
+    end
+  end
+
   defp to_quoted(attrs, quoted_module, next) do
     {query, params} = parameterize(attrs["query"], attrs["connection"]["type"], next)
     opts_args = query_opts_args(attrs)
@@ -310,6 +336,9 @@ defmodule KinoDB.SQLCell do
   defp query_opts_args(%{"connection" => %{"type" => "athena"}, "cache_query" => cache_query}),
     do: [[cache_query: cache_query]]
 
+  defp query_opts_args(%{"connection" => %{"type" => "clickhouse"}}),
+    do: [[format: :explorer]]
+
   defp query_opts_args(_attrs), do: []
 
   defp parameterize(query, type, next) do
@@ -342,7 +371,7 @@ defmodule KinoDB.SQLCell do
 
   defp parameterize("{{" <> rest = query, raw, params, n, type, next) do
     with [inner, rest] <- String.split(rest, "}}", parts: 2),
-         sql_param <- next.(n),
+         sql_param <- apply_next(next, n, inner),
          {:ok, param} <- quote_param(type, inner, sql_param) do
       parameterize(rest, raw <> sql_param, [param | params], n + 1, type, next)
     else
@@ -354,6 +383,21 @@ defmodule KinoDB.SQLCell do
     parameterize(rest, <<raw::binary, char::utf8>>, params, n, type, next)
   end
 
+  defp apply_next(next, n, _inner) when is_function(next, 1), do: next.(n)
+  defp apply_next(next, n, inner) when is_function(next, 2), do: next.(n, inner)
+
+  defp quote_param("clickhouse", inner, sql_param) do
+    with {:ok, inner_ast} <- Code.string_to_quoted(inner) do
+      name =
+        sql_param |> String.trim_leading("{") |> String.split(":", parts: 2) |> List.first()
+
+      {:ok,
+       quote do
+         {unquote(name), unquote(inner_ast)}
+       end}
+    end
+  end
+
   defp quote_param("sqlserver", inner, sql_param) do
     with {:ok, inner_ast} <- Code.string_to_quoted(inner) do
       {:ok,

mix.exs

@@ -34,11 +34,12 @@ defmodule KinoDB.MixProject do
       {:myxql, "~> 0.7", optional: true},
       {:postgrex, "~> 0.18 or ~> 1.0", optional: true},
       {:tds, "~> 2.3.4 or ~> 2.4", optional: true},
-      {:explorer, "~> 0.8", optional: true},
+      {:explorer, "~> 0.10", optional: true},
 
       # Those dependecies are new, so we use stricter versions
       {:req_bigquery, "~> 0.1.0", optional: true},
       {:req_athena, "~> 0.2.0", optional: true},
+      {:req_ch, "~> 0.1.0", optional: true},
 
       # Dev only
       {:ex_doc, "~> 0.28", only: :dev, runtime: false}