adds sandbox query/cte support for clickhouse (#2834)

Co-authored-by: Ziinc <Ziinc@users.noreply.github.com>

Author: amokan

docs/docs.logflare.com/docs/backends/clickhouse.mdx

@@ -3,7 +3,7 @@ sidebar_position: 7
 ---
 # ClickHouse
 
-The ClickHouse backend is **ingest-only** to a ClickHouse [HTTP](https://clickhouse.com/docs/interfaces/http) endpoint.
+The ClickHouse backend supports both **ingestion** and **querying** via a ClickHouse [HTTP](https://clickhouse.com/docs/interfaces/http) endpoint.
 
 ## Behavior and Configuration
 
@@ -50,3 +50,38 @@ The ingest table schema is as follows:
 By default, the ClickHouse backends will utilize the [`MergeTree` engine](https://clickhouse.com/docs/engines/table-engines/mergetree-family/mergetree).
 
 Note that when using ClickHouse Cloud, replication is handled automatically as mentioned in the [data replication documentaion](https://clickhouse.com/docs/engines/table-engines/mergetree-family/replication#creating-replicated-tables).
+
+## Querying
+
+ClickHouse backends support SQL querying through Logflare Endpoints and Alerts. The backend uses ClickHouse SQL dialect, which supports standard SQL features including:
+
+- Common Table Expressions (CTEs) with `WITH` clauses
+- Complex aggregations and window functions
+- Array and nested data type operations
+- ClickHouse-specific functions (e.g., `tuple()`, `arraySlice()`, `JSONExtractString()`)
+
+### Sandboxed Queries
+
+ClickHouse backends fully support [sandboxed queries](/concepts/endpoints#query-sandboxing) within Endpoints, allowing you to create secure, parameterized API endpoints where consumers can provide custom SQL while being restricted to pre-defined data subsets.
+
+Example sandboxed ClickHouse endpoint:
+
+```sql
+WITH filtered_logs AS (
+    SELECT id, event_message, timestamp
+    FROM my_clickhouse_source
+    WHERE timestamp > now() - interval 1 day
+)
+SELECT * FROM filtered_logs
+```
+
+Consumers can then query within the sandbox via the `sql=` parameter:
+
+```sql
+SELECT event_message, count(*) as count
+FROM filtered_logs
+GROUP BY event_message
+ORDER BY count DESC
+```
+
+See the [Endpoints documentation](/concepts/endpoints) for more details on sandboxed queries and security features.

docs/docs.logflare.com/docs/concepts/endpoints.md

@@ -59,6 +59,10 @@ Parameters that do not match any that are declared in the SQL template will be i
 
 You can create sandboxed queries by using a CTE within the query. It allows the Endpoint consumer to provide a custom SQL query through the `sql=` query parameter.
 
+:::note
+Sandboxed queries are supported for BigQuery and ClickHouse backends. PostgreSQL backends do not currently support this feature.
+:::
+
 For example, this sandboxed query creates a temporary result called `errors`, which limits the results to containing the `"ERROR"` string as well as being before the year `2020` .
 
 ```sql

lib/logflare/sql.ex

@@ -23,6 +23,9 @@ defmodule Logflare.Sql do
 
   @typep query_language :: :bq_sql | :ch_sql | :pg_sql
 
+  @bq_restricted_functions ~w(external_query session_user)
+  @ch_restricted_functions ~w(azureblobstorage cluster currentuser deltalake file gcs hdfs hudi iceberg jdbc mongodb mysql odbc postgresql redis remote remotesecure s3 sqlite url)
+
   @doc """
   Converts a language atom to its corresponding dialect.
 
@@ -66,22 +69,49 @@ defmodule Logflare.Sql do
   end
 
   @doc """
-  Transforms and validates an SQL query for querying with bigquery.any()
-  The resultant SQL is BigQuery compatible.
+  Transforms and validates a SQL query for the specified dialect,
+  which can be BigQuery (`:bq_sql`), ClickHouse (`:ch_sql`), or PostgreSQL (`:pg_sql`).
 
+  The query is parsed, validated, and transformed to include fully-qualified table names
+  appropriate for the target backend.
 
-  DML is blocked
-  - https://cloud.google.com/bigquery/docs/reference/standard-sql/dml-syntax
+  ## Validation Rules
 
-  ### Example
+  All queries are validated to ensure:
+  - Only SELECT statements are allowed (DML is blocked)
+  - Single query only (no multiple statements)
+  - No wildcard selects (`SELECT *`)
+  - No restricted functions (`SESSION_USER`, `EXTERNAL_QUERY`)
+  - All referenced tables/sources exist
+
+  ## Sandboxed Queries
+
+  BigQuery and ClickHouse support sandboxed queries via tuple input `{cte_query, consumer_query}`.
+  This allows secure, parameterized endpoints where consumers can provide custom SQL while
+  being restricted to pre-defined data subsets via CTEs.
+
+  PostgreSQL does not currently support sandboxed queries.
+
+  ## Examples
 
-    iex> transform("select a from my_table", %User{...})
-    {:ok, "select a from `my_project.my_dataset.source_token`"}
+  Basic query transformation:
 
-  With a sandboxed query
-    iex> cte = "..."
-    iex> transform({cte, "select a from my_alias"}, %User{...})
-    {:ok, "..."}
+      transform(:bq_sql, "select a from my_table", user)
+      # => {:ok, "select a from `my_project.my_dataset.source_token`"}
+
+      transform(:ch_sql, "select a from my_table", user)
+      # => {:ok, "select a from my_clickhouse_table"}
+
+  Sandboxed query (BigQuery and ClickHouse only):
+
+      cte = "with filtered as (select a from my_table where a > 0) select a from filtered"
+      consumer_query = "select a from filtered where a < 100"
+
+      transform(:bq_sql, {cte, consumer_query}, user)
+      # => {:ok, "with filtered as (select a from `project.dataset.token` where a > 0) select a from filtered where a < 100"}
+
+      transform(:ch_sql, {cte, consumer_query}, user)
+      # => {:ok, "with filtered as (select a from my_clickhouse_table where a > 0) select a from filtered where a < 100"}
   """
   @typep input :: String.t() | {String.t(), String.t()}
   @spec transform(
@@ -95,8 +125,44 @@ defmodule Logflare.Sql do
     transform(lang, input, user)
   end
 
-  # clickhouse and postgres
-  def transform(language, query, %User{} = user) when language in ~w(ch_sql pg_sql)a do
+  # clickhouse with sandboxed query support
+  def transform(:ch_sql = language, input, %User{} = user) do
+    {query, sandboxed_query} =
+      case input do
+        q when is_non_empty_binary(q) -> {q, nil}
+        other when is_tuple(other) -> other
+      end
+
+    sql_dialect = to_dialect(language)
+    sources = Sources.list_sources_by_user(user)
+    source_mapping = source_mapping(sources)
+
+    Logger.metadata(query_string: query)
+
+    with {:ok, statements} <- Parser.parse(sql_dialect, query),
+         {:ok, sandboxed_query_ast} <- sandboxed_ast(sandboxed_query, sql_dialect),
+         base_data = %{
+           sources: sources,
+           source_mapping: source_mapping,
+           source_names: Map.keys(source_mapping),
+           sandboxed_query: sandboxed_query,
+           sandboxed_query_ast: sandboxed_query_ast,
+           ast: statements,
+           dialect: sql_dialect
+         },
+         data = DialectTransformer.Clickhouse.build_transformation_data(user, base_data),
+         :ok <- validate_query(statements, data),
+         :ok <- maybe_validate_sandboxed_query_ast({statements, sandboxed_query_ast}, data) do
+      data = %{data | sandboxed_query_ast: sandboxed_query_ast}
+
+      statements
+      |> do_transform(data)
+      |> Parser.to_string()
+    end
+  end
+
+  # postgres (no sandboxed query support)
+  def transform(:pg_sql = language, query, %User{} = user) do
     sql_dialect = to_dialect(language)
     sources = Sources.list_sources_by_user(user)
     source_mapping = source_mapping(sources)
@@ -347,16 +413,22 @@ defmodule Logflare.Sql do
   defp sandboxed_ast(_, _), do: {:ok, nil}
 
   # applies to both ctes, sandboxed queries, and non-ctes
-  defp validate_query(ast, data) when is_list(ast) do
+  defp validate_query(ast, %{dialect: dialect} = data) when is_list(ast) do
     with :ok <- check_select_statement_only(ast),
          :ok <- check_single_query_only(ast),
-         :ok <- has_restricted_functions(ast),
+         :ok <- maybe_check_restricted_functions(ast, dialect, data),
          :ok <- has_wildcard_in_select(ast),
          :ok <- check_all_sources_allowed(ast, data) do
       :ok
     end
   end
 
+  defp maybe_check_restricted_functions(ast, dialect, data)
+       when dialect in ~w(bigquery clickhouse),
+       do: has_restricted_functions(ast, data)
+
+  defp maybe_check_restricted_functions(_ast, _dialect, _data), do: :ok
+
   # applies only to the sandboed query
   defp maybe_validate_sandboxed_query_ast({cte_ast, ast}, data) when is_list(ast) do
     with :ok <- validate_query(ast, data),
@@ -481,33 +553,58 @@ defmodule Logflare.Sql do
     end
   end
 
-  defp has_restricted_functions(ast) when is_list(ast), do: has_restricted_functions(ast, :ok)
+  defp has_restricted_functions(ast, data) when is_list(ast),
+    do: has_restricted_functions(ast, :ok, data)
 
-  defp has_restricted_functions({"Function", %{"name" => [%{"value" => _} | _] = names}}, :ok) do
-    restricted =
+  defp has_restricted_functions({"Function", %{"name" => [%{"value" => _} | _] = names}}, :ok, %{
+         dialect: dialect
+       }) do
+    restricted_list = get_restricted_functions_for_dialect(dialect)
+
+    found_restricted =
       for name <- names,
           normalized = String.downcase(name["value"]),
-          normalized in ["session_user", "external_query"] do
+          normalized in restricted_list do
         normalized
       end
 
-    if Enum.empty?(restricted) do
+    if Enum.empty?(found_restricted) do
       :ok
     else
-      {:error, "Restricted function #{Enum.join(restricted, ", ")}"}
+      {:error, "Restricted function #{Enum.join(found_restricted, ", ")}"}
     end
   end
 
-  defp has_restricted_functions(kv, :ok = acc) when is_list(kv) or is_map(kv) do
+  defp has_restricted_functions(
+         {"Table", %{"args" => [_ | _], "name" => [%{"value" => name} | _]}},
+         :ok,
+         %{dialect: dialect}
+       ) do
+    restricted_list = get_restricted_functions_for_dialect(dialect)
+    normalized = String.downcase(name)
+
+    if normalized in restricted_list do
+      {:error, "Restricted function #{normalized}"}
+    else
+      :ok
+    end
+  end
+
+  defp has_restricted_functions(kv, :ok = acc, data) when is_list(kv) or is_map(kv) do
     kv
-    |> Enum.reduce(acc, fn kv, nested_acc -> has_restricted_functions(kv, nested_acc) end)
+    |> Enum.reduce(acc, fn kv, nested_acc -> has_restricted_functions(kv, nested_acc, data) end)
   end
 
-  defp has_restricted_functions({_k, v}, :ok = acc) when is_list(v) or is_map(v) do
-    has_restricted_functions(v, acc)
+  defp has_restricted_functions({_k, v}, :ok = acc, data) when is_list(v) or is_map(v) do
+    has_restricted_functions(v, acc, data)
   end
 
-  defp has_restricted_functions(_kv, acc), do: acc
+  defp has_restricted_functions(_kv, acc, _data), do: acc
+
+  @spec get_restricted_functions_for_dialect(String.t() | nil) :: [String.t()]
+  defp get_restricted_functions_for_dialect("bigquery"), do: @bq_restricted_functions
+  defp get_restricted_functions_for_dialect("clickhouse"), do: @ch_restricted_functions
+  defp get_restricted_functions_for_dialect(_), do: []
 
   defp has_restricted_sources(cte_ast, ast) when is_list(ast) do
     aliases =

lib/logflare/sql/dialect_transformer/clickhouse.ex

@@ -6,6 +6,7 @@ defmodule Logflare.Sql.DialectTransformer.Clickhouse do
   @behaviour Logflare.Sql.DialectTransformer
 
   alias Logflare.Backends.Adaptor.ClickhouseAdaptor
+  alias Logflare.User
 
   @impl true
   def quote_style, do: nil
@@ -18,4 +19,12 @@ defmodule Logflare.Sql.DialectTransformer.Clickhouse do
     source = Enum.find(sources, fn s -> s.name == source_name end)
     ClickhouseAdaptor.clickhouse_ingest_table_name(source)
   end
+
+  @doc """
+  Builds transformation data for ClickHouse from a user and base data.
+
+  Since ClickHouse does not require project/dataset metadata, we can just pass through the base data.
+  """
+  @spec build_transformation_data(User.t(), map()) :: map()
+  def build_transformation_data(%User{}, base_data), do: base_data
 end

test/logflare/sql/dialect_transformer/clickhouse_test.exs

@@ -63,4 +63,21 @@ defmodule Logflare.Sql.DialectTransformer.ClickhouseTest do
       assert result == expected
     end
   end
+
+  describe "build_transformation_data/2" do
+    test "passes through base data unchanged" do
+      user = build(:user)
+
+      base_data = %{
+        sources: [],
+        dialect: "clickhouse",
+        ast: [],
+        sandboxed_query: nil
+      }
+
+      result = Clickhouse.build_transformation_data(user, base_data)
+
+      assert result == base_data
+    end
+  end
 end