From a293eebca09ae508c0e5738c45671fb1c5a5aa21 Mon Sep 17 00:00:00 2001 From: Alexey Milovidov Date: Sat, 5 Sep 2026 21:58:08 +0200 Subject: [PATCH 1/2] Add ClickHouse and chDB readers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two readers behind the new default-on `clickhouse` and `chdb` features, sharing one implementation (`ClickHouseSqlReader`): - `clickhouse://[user[:password]@]host[:port][/database][?setting=…]` (`clickhouses://` for TLS) talks to a server's HTTP interface with plain requests and exchanges data as Arrow IPC streams, so no driver is needed. Unknown URI parameters are forwarded as ClickHouse settings; host, user and password fall back to CLICKHOUSE_HOST/USER/PASSWORD. - `chdb://[path]` runs the embedded chDB engine in-process. libchdb is loaded at runtime via libloading (same approach as the ODBC driver manager), so the build has no new native dependency. chDB is also a new `CacheBackend` (`chdb+://`, `--cache chdb`), so a ClickHouse setup needs no other engine. Type handling: every SELECT is DESCRIBEd first and, when ClickHouse's Arrow output would lose the type (DateTime → UInt32, Enum → codes, UUID/IP/wide integers → bytes, Decimal), wrapped in `SELECT * REPLACE (…)` converting those columns server-side; timestamps are normalized to naive microseconds. Read-only accounts (e.g. play.clickhouse.com) are detected on connect via a CREATE TEMPORARY TABLE probe; `reader_from_uri` then keeps the executor's intermediate tables in an embedded chDB cache automatically. Dialect: Nullable(…) cast targets, TEMPORARY temp tables, quantiles via quantileExactInclusive (inline, no correlated subquery), greatest/least cast to Float64 (no UInt64/Float64 supertype), Memory-engine memo table for the caching layer via new `cache_meta_*` SqlDialect hooks, and a new `sql_null_safe_equals` hook because older ClickHouse only accepts IS NOT DISTINCT FROM in JOIN ON. Portable fixes found along the way: the Vega-Lite writer rescaled timestamps that were already converted to microseconds (overflow for any non-microsecond source); density and boxplot stat SQL now alias qualified projections, which ClickHouse otherwise names `cte.col`. Also: CLI `--cache chdb`, Jupyter kernel connection names, Positron connection drivers for ClickHouse and chDB, docs and CHANGELOG. Co-Authored-By: Claude Fable 5.1 --- CHANGELOG.md | 17 + Cargo.lock | 29 + Cargo.toml | 2 + doc/get_started/tooling/cli.qmd | 54 ++ ggsql-cli/CLAUDE.md | 4 +- ggsql-cli/Cargo.toml | 8 +- ggsql-cli/src/main.rs | 10 +- ggsql-jupyter/Cargo.toml | 4 +- ggsql-jupyter/src/executor.rs | 30 + ggsql-vscode/src/connections.ts | 101 +++ src/CLAUDE.md | 9 +- src/Cargo.toml | 13 +- src/plot/layer/geom/boxplot.rs | 4 +- src/plot/layer/geom/density.rs | 79 +- src/reader/cache.rs | 52 +- src/reader/clickhouse/chdb.rs | 549 ++++++++++++++ src/reader/clickhouse/http.rs | 456 ++++++++++++ src/reader/clickhouse/mod.rs | 1193 +++++++++++++++++++++++++++++++ src/reader/connection.rs | 131 +++- src/reader/mod.rs | 85 ++- src/writer/vegalite/data.rs | 16 +- 21 files changed, 2750 insertions(+), 96 deletions(-) create mode 100644 src/reader/clickhouse/chdb.rs create mode 100644 src/reader/clickhouse/http.rs create mode 100644 src/reader/clickhouse/mod.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index fb8f1ea6a..e1bc567e0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,23 @@ ### Added +- ClickHouse support, as two readers behind the new default-on `clickhouse` and + `chdb` features. `clickhouse://[user[:password]@]host[:port][/database]` (or + `clickhouses://` for TLS) talks to a server's HTTP interface and exchanges data + as Arrow IPC streams, so no driver installation is needed; unrecognised URI + query parameters are forwarded as ClickHouse settings, and the host, user and + password fall back to the `CLICKHOUSE_HOST`/`CLICKHOUSE_USER`/`CLICKHOUSE_PASSWORD` + environment variables. `chdb://[path]` runs the embedded chDB engine + in-process (libchdb is loaded at runtime, so the build has no new native + dependency), and `chdb` is also a new in-memory backend for the caching layer + (`chdb+://…`, `--cache chdb`), so a ClickHouse setup needs no other + database engine. Stat transforms run on the server through session-scoped + temporary tables; a read-only account such as `play.clickhouse.com` is + detected on connect and its intermediate tables are kept in an embedded chDB + cache automatically. `DateTime`, `Enum`, `UUID`, IP address, `FixedString`, + `Decimal` and wide-integer columns are converted server-side to types the plot + pipeline understands. The Jupyter kernel and Positron connections pane + recognise both schemes. - New caching layer that wraps any `Reader` with an in-memory, writeable cache backend (currently duckdb or sqlite), making write-constrained databases usable and avoiding repeated remote reads during interactive iteration. diff --git a/Cargo.lock b/Cargo.lock index 788e814dd..92ab68b64 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -322,6 +322,7 @@ dependencies = [ "arrow-select", "flatbuffers", "lz4_flex", + "zstd", ] [[package]] @@ -6654,3 +6655,31 @@ dependencies = [ "log", "simd-adler32", ] + +[[package]] +name = "zstd" +version = "0.13.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e91ee311a569c327171651566e07972200e76fcfe2242a4fa446149a3881c08a" +dependencies = [ + "zstd-safe", +] + +[[package]] +name = "zstd-safe" +version = "7.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "64d80649ab6db9d9f6f9c80a40becd948eda4714a0a5ac8c4d157a32231c7882" +dependencies = [ + "zstd-sys", +] + +[[package]] +name = "zstd-sys" +version = "2.1.0+zstd.1.5.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0ef0a8027ec3ee71300ab3bcbcd0393f434aa72b91ca6d635a39941deae8eea0" +dependencies = [ + "cc", + "pkg-config", +] diff --git a/Cargo.toml b/Cargo.toml index beaa88a36..7f2ea18bb 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -44,6 +44,8 @@ duckdb = { version = "~1.10502", features = ["bundled", "vtab-arrow"] } parquet = { version = "58", default-features = false, features = ["arrow", "snap"] } bytes = "1" rusqlite = { version = "0.38", features = ["bundled", "chrono", "load_extension"] } +# ClickHouse HTTP interface (also fetches the AI skill at CLI build time) +ureq = "3" # ODBC toml_edit = "0.22" diff --git a/doc/get_started/tooling/cli.qmd b/doc/get_started/tooling/cli.qmd index b22931b14..49cd8b423 100644 --- a/doc/get_started/tooling/cli.qmd +++ b/doc/get_started/tooling/cli.qmd @@ -65,6 +65,60 @@ col_a, col_b, col_c 12.5, 29.48, gamma ``` +### ClickHouse + +ggsql connects to [ClickHouse](https://clickhouse.com) over its HTTP interface, so no driver needs to be installed. The connection string is + +``` +clickhouse://[user[:password]@]host[:port][/database][?param=value&...] +``` + +The port defaults to 8123. Use `clickhouses://` (or add `?secure=1`) for a TLS connection, where the port defaults to 8443. Any other query parameter is forwarded to the server with every request, so ClickHouse settings and HTTP parameters such as `session_timezone`, `max_threads` or `session_timeout` can be set per connection. If the connection string does not name them, the host, user and password are taken from the `CLICKHOUSE_HOST`, `CLICKHOUSE_USER` and `CLICKHOUSE_PASSWORD` environment variables, so a password need not appear on the command line. + +```bash +$ ggsql exec --reader clickhouse://localhost:8123/default \ + "SELECT number AS x, number * number AS y FROM numbers(10) VISUALISE x, y DRAW line" +``` + +Each reader holds one ClickHouse session, and ggsql stores intermediate results in temporary tables inside that session, which is what lets stat transforms such as binning and quantiles run on the server. A read-only account without the `CREATE TEMPORARY TABLE` privilege is detected when connecting; ggsql then keeps the intermediate tables in an embedded [chDB](#chdb) engine and sends only your own query to the server. The public playground works this way: + +```bash +$ ggsql exec --reader clickhouses://explorer@play.clickhouse.com:443 \ + "SELECT toStartOfMonth(created_at) AS month, count() AS events + FROM github_events WHERE created_at >= '2024-01-01' + GROUP BY month ORDER BY month + VISUALISE month AS x, events AS y DRAW line" +``` + +The same arrangement can be requested explicitly with `chdb+clickhouse://…` (or `--cache chdb`). It also memoizes the server's answers, so re-running a query while iterating on the plot does not hit the server again until the cache entry expires. + +ClickHouse `DateTime` and `DateTime64` values are shown in UTC. `Enum`, `UUID`, `IPv4`/`IPv6`, `FixedString` and 128/256-bit integer columns arrive as strings, `Decimal` columns as floating-point numbers, and arrays, tuples and maps as their text form. + +### chDB {#chdb} + +[chDB](https://clickhouse.com/chdb) is ClickHouse as an in-process library. ggsql uses it in two roles: as a database of its own, and as the in-memory cache in front of a ClickHouse server. Either way it needs `libchdb` on the machine; install it with + +```bash +curl -sL https://lib.chdb.io | bash +``` + +or point the `GGSQL_CHDB_LIBRARY` environment variable at a `libchdb.so` (or `.dylib`) you downloaded yourself. Nothing else changes: the same `ggsql` binary works with or without the library, and only `chdb://` connections need it. + +``` +chdb:// in-memory engine (also chdb://memory) +chdb:///path/to/directory state persists under the directory +chdb://…?key=value extra engine arguments, passed as --key=value +``` + +Everything ClickHouse can read, chDB can read too, so a chDB connection is a quick way to plot local files: + +```bash +$ ggsql exec --reader chdb://memory \ + "SELECT * FROM 'trips.parquet' VISUALISE fare AS x DRAW histogram" +``` + +chDB keeps a single connection per process, so all `chdb://` connections in one process share one engine and one path. + ## Output format `ggsql exec` and `ggsql run` render with the writer named by `--writer` (short `-w`), defaulting to `--writer vegalite` (the Vega-Lite JSON above). A build that includes the optional `png` writer can also render straight to a PNG image with `--writer png`, which needs a GPU adapter available where it runs. diff --git a/ggsql-cli/CLAUDE.md b/ggsql-cli/CLAUDE.md index eee23793d..09d42de8f 100644 --- a/ggsql-cli/CLAUDE.md +++ b/ggsql-cli/CLAUDE.md @@ -34,7 +34,7 @@ The binary name is `ggsql` (not `ggsql-cli`) — that's what release artifacts a Only public `ggsql::*` API is used (`reader`, `writer`, `validate`, `parser`, `VERSION`) — this crate has no awareness of internal modules. -`exec`/`run` build their reader via the library factory `ggsql::reader::connection::reader_from_uri`. They accept an in-memory caching layer (off by default) selected either by the composite connection scheme `+://…` (e.g. `duckdb+odbc://…`) or the `--cache ` flag; the two cannot be combined. +`exec`/`run` build their reader via the library factory `ggsql::reader::connection::reader_from_uri`. They accept an in-memory caching layer (off by default) selected either by the composite connection scheme `+://…` (e.g. `duckdb+odbc://…`) or the `--cache ` flag; the two cannot be combined. `exec` and `run` share a `WriterSpec { name, options }`: `--writer` names the writer and repeated `--writer-option key=value` flags (short `-D`, visible alias `--writer-options`, several settings per flag when separated by `;`) become a `ggsql::writer::WriterOptions`, parsed up front in `main` so a malformed pair fails before any SQL runs. The two travel together down `cmd_exec` → `exec_with_reader` → `render_spec`, which dispatches on the name and hands the options to `Writer::from_options`. Adding a setting to a writer therefore needs no CLI change; which keys exist is the writer's business, and an unknown one is its error to report. User-facing keys are documented in [`/doc/get_started/tooling/cli.qmd`](../doc/get_started/tooling/cli.qmd). @@ -59,7 +59,7 @@ The macOS codesign step uses [`/entitlements.plist`](../entitlements.plist) at t ## Features ```toml -default = ["duckdb", "sqlite", "vegalite", "ipc", "parquet", "builtin-data", "odbc"] +default = ["duckdb", "sqlite", "clickhouse", "chdb", "vegalite", "parquet", "builtin-data", "odbc"] ``` Each feature passes through to `ggsql/`. The `vegalite` flag also gates the writer-rendering path in `main.rs` via `#[cfg(feature = "vegalite")]`. diff --git a/ggsql-cli/Cargo.toml b/ggsql-cli/Cargo.toml index 20c96f7c5..1f961bd84 100644 --- a/ggsql-cli/Cargo.toml +++ b/ggsql-cli/Cargo.toml @@ -41,18 +41,20 @@ termimad = "0.31" [build-dependencies] regex.workspace = true -ureq = "3" +ureq.workspace = true [features] -default = ["duckdb", "sqlite", "vegalite", "parquet", "builtin-data", "odbc"] +default = ["duckdb", "sqlite", "clickhouse", "chdb", "vegalite", "parquet", "builtin-data", "odbc"] duckdb = ["ggsql/duckdb"] parquet = ["ggsql/parquet"] sqlite = ["ggsql/sqlite"] odbc = ["ggsql/odbc"] +clickhouse = ["ggsql/clickhouse"] +chdb = ["ggsql/chdb"] vegalite = ["ggsql/vegalite"] png = ["ggsql/png"] builtin-data = ["ggsql/builtin-data"] -all-readers = ["duckdb", "sqlite", "odbc"] +all-readers = ["duckdb", "sqlite", "odbc", "clickhouse", "chdb"] # cargo-packager configuration for cross-platform installers [package.metadata.packager] diff --git a/ggsql-cli/src/main.rs b/ggsql-cli/src/main.rs index f04592cd7..587e89aa1 100644 --- a/ggsql-cli/src/main.rs +++ b/ggsql-cli/src/main.rs @@ -64,11 +64,11 @@ pub enum Commands { /// The ggsql query to execute query: String, - /// Data source connection string (duckdb://, sqlite://, odbc://) + /// Data source connection string (duckdb://, sqlite://, odbc://, clickhouse://) #[arg(short, long, default_value = "duckdb://memory")] reader: String, - /// In-memory cache backend wrapping the reader (duckdb, sqlite). Off by default. + /// In-memory cache backend wrapping the reader (duckdb, sqlite, chdb). Off by default. #[arg(long)] cache: Option, @@ -104,11 +104,11 @@ pub enum Commands { /// Path to .sql file containing ggsql query file: PathBuf, - /// Data source connection string (duckdb://, sqlite://, odbc://) + /// Data source connection string (duckdb://, sqlite://, odbc://, clickhouse://) #[arg(short, long, default_value = "duckdb://memory")] reader: String, - /// In-memory cache backend wrapping the reader (duckdb, sqlite). Off by default. + /// In-memory cache backend wrapping the reader (duckdb, sqlite, chdb). Off by default. #[arg(long)] cache: Option, @@ -154,7 +154,7 @@ pub enum Commands { /// The ggsql query to validate query: String, - /// Data source connection string for column validation (duckdb://, sqlite://, polars://) + /// Data source connection string for column validation (duckdb://, sqlite://, odbc://, clickhouse://) #[arg(short, long)] reader: Option, }, diff --git a/ggsql-jupyter/Cargo.toml b/ggsql-jupyter/Cargo.toml index 6be601eae..30dece751 100644 --- a/ggsql-jupyter/Cargo.toml +++ b/ggsql-jupyter/Cargo.toml @@ -59,10 +59,12 @@ uuid = { version = "1.0", features = ["v4"] } [features] default = ["all-readers"] -all-readers = ["sqlite", "odbc", "duckdb"] +all-readers = ["sqlite", "odbc", "duckdb", "clickhouse", "chdb"] odbc = ["ggsql/odbc"] sqlite = ["ggsql/sqlite"] duckdb = ["ggsql/duckdb"] +clickhouse = ["ggsql/clickhouse"] +chdb = ["ggsql/chdb"] [dev-dependencies] tempfile = "3.8" diff --git a/ggsql-jupyter/src/executor.rs b/ggsql-jupyter/src/executor.rs index 18f6e0116..80f28086f 100644 --- a/ggsql-jupyter/src/executor.rs +++ b/ggsql-jupyter/src/executor.rs @@ -52,9 +52,33 @@ pub fn display_name_for_uri(uri: &str) -> String { } return "ODBC".to_string(); } + if uri.starts_with("clickhouse://") || uri.starts_with("clickhouses://") { + return format!("ClickHouse ({})", clickhouse_host(uri)); + } + if let Some(rest) = uri.strip_prefix("chdb://") { + let path = rest.split('?').next().unwrap_or(rest); + if path.is_empty() || path == "memory" || path == ":memory:" { + return "chDB (memory)".to_string(); + } + return format!("chDB ({path})"); + } uri.to_string() } +/// The `host[:port]` part of a `clickhouse://` / `clickhouses://` URI, without +/// credentials, database or parameters. +fn clickhouse_host(uri: &str) -> String { + let rest = uri.split_once("://").map(|(_, r)| r).unwrap_or(uri); + let rest = rest.split('?').next().unwrap_or(rest); + let rest = rest.rsplit_once('@').map(|(_, h)| h).unwrap_or(rest); + let host = rest.split('/').next().unwrap_or(rest); + if host.is_empty() { + "localhost".to_string() + } else { + host.to_string() + } +} + /// Detect the database type name from a connection URI (e.g. "DuckDB", "Snowflake"). pub fn type_name_for_uri(uri: &str) -> String { if uri.starts_with("duckdb://") { @@ -63,6 +87,12 @@ pub fn type_name_for_uri(uri: &str) -> String { if uri.starts_with("sqlite://") { return "SQLite".to_string(); } + if uri.starts_with("clickhouse://") || uri.starts_with("clickhouses://") { + return "ClickHouse".to_string(); + } + if uri.starts_with("chdb://") { + return "chDB".to_string(); + } if let Some(odbc) = uri.strip_prefix("odbc://") { if let Some(driver) = extract_odbc_value(odbc, "driver") { let lower = driver.to_lowercase(); diff --git a/ggsql-vscode/src/connections.ts b/ggsql-vscode/src/connections.ts index ee15ddfed..91371b446 100644 --- a/ggsql-vscode/src/connections.ts +++ b/ggsql-vscode/src/connections.ts @@ -24,6 +24,8 @@ export function createConnectionDrivers( return [ createDuckDBDriver(positronApi), createSQLiteDriver(positronApi), + createClickHouseDriver(positronApi), + createChdbDriver(positronApi), createSnowflakeDefaultDriver(positronApi), createSnowflakePasswordDriver(positronApi), createSnowflakeSSODriver(positronApi), @@ -411,6 +413,105 @@ function createSnowflakePATDriver( }; } +// ============================================================================ +// ClickHouse +// ============================================================================ + +/** + * ClickHouse connection driver (HTTP interface). + * + * Inputs: host (required), port, database, user, password, TLS. Produces + * `clickhouse://` or `clickhouses://`. The password is omitted from the + * generated code when left blank so it can come from `CLICKHOUSE_PASSWORD`. + */ +function createClickHouseDriver( + positronApi: PositronApi +): positron.ConnectionsDriver { + return { + driverId: 'ggsql-clickhouse', + metadata: { + languageId: 'ggsql', + name: 'ClickHouse', + inputs: [ + { id: 'host', label: 'Host', type: 'string', value: 'localhost' }, + { id: 'port', label: 'Port', type: 'number', value: '8123' }, + { id: 'database', label: 'Database', type: 'string', value: '' }, + { id: 'user', label: 'User', type: 'string', value: 'default' }, + { id: 'password', label: 'Password', type: 'string', value: '' }, + { + id: 'secure', + label: 'Use TLS (clickhouses://)', + type: 'option', + options: [ + { identifier: 'no', title: 'No' }, + { identifier: 'yes', title: 'Yes' }, + ], + value: 'no', + }, + ], + } as ConnectionsDriverMetadata, + generateCode: (inputs) => { + const get = (id: string) => + inputs.find((i) => i.id === id)?.value?.trim() || ''; + const scheme = get('secure') === 'yes' ? 'clickhouses' : 'clickhouse'; + const user = get('user'); + const password = get('password'); + const auth = user + ? encodeURIComponent(user) + + (password ? `:${encodeURIComponent(password)}` : '') + + '@' + : ''; + const host = get('host') || 'localhost'; + const port = get('port') ? `:${get('port')}` : ''; + const database = get('database') ? `/${encodeURIComponent(get('database'))}` : ''; + return `-- @connect: ${scheme}://${auth}${host}${port}${database}`; + }, + connect: async (code: string) => { + await positronApi.runtime.executeCode('ggsql', code, false); + }, + }; +} + +// ============================================================================ +// chDB (embedded ClickHouse) +// ============================================================================ + +/** + * chDB connection driver. + * + * Inputs: data directory (optional; blank = in-memory engine). + */ +function createChdbDriver( + positronApi: PositronApi +): positron.ConnectionsDriver { + return { + driverId: 'ggsql-chdb', + metadata: { + languageId: 'ggsql', + name: 'chDB', + description: 'Embedded ClickHouse (libchdb)', + inputs: [ + { + id: 'path', + label: 'Data directory (blank for in-memory)', + type: 'string', + value: '', + }, + ], + } as ConnectionsDriverMetadata, + generateCode: (inputs) => { + const path = inputs.find((i) => i.id === 'path')?.value?.trim(); + if (!path) { + return '-- @connect: chdb://memory'; + } + return `-- @connect: chdb://${path}`; + }, + connect: async (code: string) => { + await positronApi.runtime.executeCode('ggsql', code, false); + }, + }; +} + // ============================================================================ // Generic ODBC // ============================================================================ diff --git a/src/CLAUDE.md b/src/CLAUDE.md index 06d8f93e9..f4db00616 100644 --- a/src/CLAUDE.md +++ b/src/CLAUDE.md @@ -47,13 +47,16 @@ Grammar lives in [`/tree-sitter-ggsql/`](../tree-sitter-ggsql/) — when adding | `duckdb.rs` | DuckDB (in-memory or file) | `duckdb` (default) | | `sqlite.rs` | SQLite | `sqlite` (default) | | `odbc.rs` | ODBC | `odbc` (default) | -| `cache.rs` | `CachingReader` — wraps any primary `Reader` with an in-memory cache | `duckdb` or `sqlite` | +| `clickhouse/` | ClickHouse: `http.rs` reaches a server over HTTP (`ureq` + `FORMAT ArrowStream`); `chdb.rs` runs embedded chDB via runtime-loaded libchdb, also a `CacheBackend` | `clickhouse` / `chdb` (both default) | +| `cache.rs` | `CachingReader` — wraps any primary `Reader` with an in-memory cache | `duckdb`, `sqlite` or `chdb` | | `connection.rs` | Connection-string parsing for all of the above | — | | `spec.rs` | `Spec` type returned by `execute()`, plus DataFrame conversion | — | | `data.rs` | Bundled sample datasets — the `ggsql:` builtins | `builtin-data` | `SqlDialect` trait in `mod.rs` lets each driver supply its own type names, information-schema queries, and spatial helper methods (`sql_st_transform`, `sql_geometry_to_wkb`, `sql_geometry_bbox`, `sql_ensure_geometry`, `sql_select_replace`, `sql_spatial_setup`). +**ClickHouse readers.** `clickhouse/mod.rs` holds `ClickHouseSqlReader`, one `Reader` implementation shared by two transports: `http.rs` (`ClickHouseReader`, one `POST` per statement against a server, results as `FORMAT ArrowStream`, `register()` = `CREATE TEMPORARY TABLE` + `INSERT … FORMAT ArrowStream`, one `session_id` per reader so temp tables and `SET` persist) and `chdb.rs` (`ChdbReader`, libchdb loaded at runtime with `libloading` like the ODBC driver manager — search order `GGSQL_CHDB_LIBRARY`, system path, usual install dirs — so the `chdb` feature adds no build dependency; inserts go through a temp file and `FROM INFILE` because the C API has no input-data channel; libchdb allows **one connection per process**, shared by all `ChdbReader`s on the same path). Because ClickHouse's Arrow output loses some of its own types (`DateTime` → `UInt32`, `Enum` → codes, `UUID`/`IPv4`/`Decimal` → bytes/decimals), every `SELECT`/`WITH` is first `DESCRIBE`d and, when needed, wrapped in `SELECT * REPLACE (…)` converting those columns server-side; all timestamps are then normalized to naive microseconds. `ClickHouseDialect` casts to `Nullable(…)` types (ClickHouse cannot cast `NULL` to a non-nullable type), uses `TEMPORARY` in temp-table DDL, maps quantiles to `quantileExactInclusive` (inline, no correlated subquery), casts `greatest`/`least` arguments to Float64 (no UInt64/Float64 supertype), and spells the caching layer's memo-table bookkeeping (the `cache_meta_*` `SqlDialect` hooks) as a `Memory` table with `ALTER TABLE … UPDATE/DELETE`. Schema introspection reads `system.databases/tables/columns`; a database is both catalog and schema. `ClickHouseSqlReader::supports_temporary_tables()` probes once; `reader_from_uri` wraps a plain `clickhouse://` reader that fails the probe (read-only account, e.g. `play.clickhouse.com`) in a `CachingReader` on a chDB backend, so no DuckDB is involved on the ClickHouse path. Arrow IPC batches from ClickHouse are LZ4-compressed by default, hence `arrow/ipc_compression`. Live tests: HTTP ones are gated on `GGSQL_CLICKHOUSE_URI`; chDB ones skip when libchdb cannot be loaded and are serialized on a static mutex because the executor's temp-table names are per process. + **Caching layer.** `CachingReader` (`cache.rs`) wraps a primary reader plus an in-memory `CacheBackend`, splitting work across two `Reader` surfaces. **`execute_sql` = source**: base reads of the user's data plus user setup/DML run on the primary (with result memoization), except `ggsql:` builtins, the `__ggsql_cache_meta__` table, and reads that reference a cache-resident internal table, which go to the cache. **`execute_sql_cached` = compute**: all dialect-generated/derived SQL (schema probes, stats, projection/map transforms, spatial setup, final layer queries — everything operating on `__ggsql_*` tables) runs on the cache; it defaults to `execute_sql` so a plain reader runs everything on one connection. Cache routing is by **exact-identifier membership** in the set of tables registered into the cache. Memoization keys on `hash(primary_uri + sql)` and is tracked in the `__ggsql_cache_meta__` table inside the cache backend. Each memoized read is bounded by a **TTL** (default 300s) and the whole memo by an **LRU byte budget** (default 512 MB); both are configurable via `CacheConfig` (env `GGSQL_CACHE_DISABLED`/`GGSQL_CACHE_TTL`/`GGSQL_CACHE_MAX_BYTES`, or per-connection URI query parameters `?cache_ttl=…&cache_max_bytes=…&cache_disabled=…`). The `__ggsql_cache_meta__` table is queryable for introspection (`SELECT * FROM __ggsql_cache_meta__`). Pure/non-visual SQL (CLI table fallback, Jupyter) goes through `execute_sql` so it reads the primary rather than the empty cache. `Reader::materialize_table` (default = `CREATE TEMP TABLE` on the reader, no Rust roundtrip) is overridden to read the body via the source surface and `register()` the result into the cache, so the primary is never written to; `Reader::caches_sources()` (default `false`, `true` for `CachingReader`) gates the executor's per-layer source staging: file sources are staged on the cache surface, while identifiers go through `materialize_table`, which routes the read to the cache (CTEs, builtins, cache-resident tables) or the primary as needed. `dialect()` returns the **cache** dialect, and every compute-surface failure is prefixed with the cache backend's scheme (``on the `duckdb` cache backend: …``) so a cache-dialect driver error is not mistaken for one from the user's own connection. Selected via the composite `+://` scheme (`reader_from_uri` / `split_cache_uri`) or the CLI `--cache` flag; off by default. ### `execute/` @@ -105,12 +108,14 @@ Defined in `Cargo.toml`: | `duckdb` | ✓ | DuckDB reader | | `sqlite` | ✓ | SQLite reader | | `odbc` | ✓ | ODBC reader | +| `clickhouse` | ✓ | ClickHouse server reader (HTTP + Arrow IPC; enables `arrow/ipc`, `arrow/ipc_compression`) | +| `chdb` | ✓ | Embedded chDB reader + cache backend (runtime-loaded libchdb) | | `parquet` | ✓ | Parquet support in readers/data | | `spatial` | ✓ | Spatial/geometry support (geozero for WKT↔GeoJSON) | | `vegalite` | ✓ | Vega-Lite writer | | `png` | — | PNG raster writer (GPU; excluded from the MSRV build) | | `builtin-data` | ✓ | Bundled penguins/airquality datasets | -| `all-readers` | — | `duckdb` + `sqlite` + `odbc` | +| `all-readers` | — | `duckdb` + `sqlite` + `odbc` + `clickhouse` + `chdb` | `ggsql-wasm` builds with `default-features = false` plus `vegalite`, `sqlite`, `builtin-data`. `ggsql-jupyter` builds with `duckdb`, `vegalite`. diff --git a/src/Cargo.toml b/src/Cargo.toml index 3c8c8ec58..68f186a21 100644 --- a/src/Cargo.toml +++ b/src/Cargo.toml @@ -36,6 +36,9 @@ bytes = { workspace = true } # ADBC reader adbc_core = { version = "0.23", optional = true } +# ClickHouse reader (HTTP interface) +ureq = { workspace = true, optional = true } + # Spatial geozero = { workspace = true, optional = true, features = ["with-wkb", "with-wkt", "with-geojson"] } @@ -60,19 +63,23 @@ uuid.workspace = true [dev-dependencies] jsonschema = { version = "0.44", default-features = false, features = ["resolve-file"] } tempfile = "3.8" -ureq = "3" +ureq.workspace = true adbc_datafusion = "0.23" adbc_driver_manager = "0.23" [features] -default = ["adbc", "duckdb", "sqlite", "vegalite", "parquet", "builtin-data", "odbc", "spatial"] +default = ["adbc", "duckdb", "sqlite", "clickhouse", "chdb", "vegalite", "parquet", "builtin-data", "odbc", "spatial"] duckdb = ["dep:duckdb"] parquet = ["dep:parquet"] sqlite = ["dep:rusqlite"] adbc = ["dep:adbc_core"] +# ClickHouse writes Arrow IPC batches LZ4-compressed by default (and a +# read-only account may not change that), so the decoder needs `ipc_compression`. +clickhouse = ["dep:ureq", "arrow/ipc", "arrow/ipc_compression"] +chdb = ["dep:libloading", "arrow/ipc", "arrow/ipc_compression"] odbc = ["dep:toml_edit", "dep:libloading"] spatial = ["dep:geozero", "rusqlite?/load_extension"] vegalite = [] png = ["dep:hephaestus"] builtin-data = [] -all-readers = ["duckdb", "sqlite", "odbc"] +all-readers = ["duckdb", "sqlite", "odbc", "clickhouse", "chdb"] diff --git a/src/plot/layer/geom/boxplot.rs b/src/plot/layer/geom/boxplot.rs index 0c5bcd450..0932b55c5 100644 --- a/src/plot/layer/geom/boxplot.rs +++ b/src/plot/layer/geom/boxplot.rs @@ -246,7 +246,9 @@ fn boxplot_sql_filter_outliers(groups: &[String], value: &str, from: &str) -> St for column in groups { let quoted = naming::quote_ident(column); join_pairs.push(format!("raw.{} = summary.{}", quoted, quoted)); - keep_columns.push(format!("raw.{}", quoted)); + // Aliased explicitly: some engines (ClickHouse) otherwise name an + // unaliased `raw.col` projection `raw.col`. + keep_columns.push(format!("raw.{quoted} AS {quoted}")); } let quoted_value = naming::quote_ident(value); diff --git a/src/plot/layer/geom/density.rs b/src/plot/layer/geom/density.rs index c81139b82..18341dafd 100644 --- a/src/plot/layer/geom/density.rs +++ b/src/plot/layer/geom/density.rs @@ -187,6 +187,7 @@ pub(crate) fn stat_density( &bw_cte, &data_cte, &grid_cte, + dialect, ); let mut consumed = vec![value_aesthetic.to_string()]; @@ -455,12 +456,16 @@ fn build_grid_cte( .iter() .map(|g| { let q = naming::quote_ident(g); - format!("full_grid.{q} IS NOT DISTINCT FROM bandwidth.{q}") + dialect + .sql_null_safe_equals(&format!("full_grid.{q}"), &format!("bandwidth.{q}")) }) .collect(); let grid_groups_select: Vec = groups .iter() - .map(|g| format!("full_grid.{}", naming::quote_ident(g))) + .map(|g| { + let q = naming::quote_ident(g); + format!("full_grid.{q} AS {q}") + }) .collect(); format!( @@ -511,6 +516,7 @@ fn compute_density( bandwidth_cte: &str, data_cte: &str, grid_cte: &str, + dialect: &dyn SqlDialect, ) -> String { // Build bandwidth join condition (NULL-safe) let bandwidth_conditions = if group_by.is_empty() { @@ -520,7 +526,7 @@ fn compute_density( .iter() .map(|g| { let q = naming::quote_ident(g); - format!("data.{q} IS NOT DISTINCT FROM bandwidth.{q}") + dialect.sql_null_safe_equals(&format!("data.{q}"), &format!("bandwidth.{q}")) }) .collect::>() .join(" AND ") @@ -534,7 +540,7 @@ fn compute_density( .iter() .map(|g| { let q = naming::quote_ident(g); - format!("grid.{q} IS NOT DISTINCT FROM data.{q}") + dialect.sql_null_safe_equals(&format!("grid.{q}"), &format!("data.{q}")) }) .collect(); format!("WHERE {}", grid_data_conds.join(" AND ")) @@ -553,6 +559,15 @@ fn compute_density( .iter() .map(|g| format!("grid.{}", naming::quote_ident(g))) .collect(); + // Projected with an explicit alias: some engines (ClickHouse) otherwise + // name an unaliased `grid.col` projection `grid.col`. + let grid_groups_select: Vec = group_by + .iter() + .map(|g| { + let q = naming::quote_ident(g); + format!("grid.{q} AS {q}") + }) + .collect(); let aggregation = format!( "GROUP BY grid.x{grid_group_by} ORDER BY grid.x{grid_group_by}", @@ -597,7 +612,7 @@ fn compute_density( intensity_column = intensity_column, density_column = density_column, aggregation = aggregation, - grid_groups = with_trailing_comma(&grid_groups.join(", ")) + grid_groups = with_trailing_comma(&grid_groups_select.join(", ")) ) } @@ -625,7 +640,15 @@ mod tests { let data_cte = build_data_cte("x", None, None, query, &groups); let grid_cte = build_grid_cte(&groups, 512, None, &AnsiDialect); let kernel = choose_kde_kernel(¶meters, None).expect("kernel should be valid"); - let sql = compute_density("x", &groups, kernel, &bw_cte, &data_cte, &grid_cte); + let sql = compute_density( + "x", + &groups, + kernel, + &bw_cte, + &data_cte, + &grid_cte, + &AnsiDialect, + ); let expected = r#"WITH RECURSIVE bandwidth AS ( @@ -701,7 +724,15 @@ mod tests { let data_cte = build_data_cte("x", None, None, query, &groups); let grid_cte = build_grid_cte(&groups, 512, None, &AnsiDialect); let kernel = choose_kde_kernel(¶meters, None).expect("kernel should be valid"); - let sql = compute_density("x", &groups, kernel, &bw_cte, &data_cte, &grid_cte); + let sql = compute_density( + "x", + &groups, + kernel, + &bw_cte, + &data_cte, + &grid_cte, + &AnsiDialect, + ); let expected = r#"WITH RECURSIVE bandwidth AS ( @@ -740,7 +771,7 @@ mod tests { FROM ( SELECT grid.x AS "__ggsql_stat_x", - grid."region", grid."category", + grid."region" AS "region", grid."category" AS "category", SUM(data.weight * ((EXP(-0.5 * (grid.x - data.val) * (grid.x - data.val) / (bandwidth.bw * bandwidth.bw))) * 0.3989422804014327)) / MIN(bandwidth.bw) AS "__ggsql_stat_intensity", SUM(data.weight) AS "__norm" FROM data @@ -880,7 +911,15 @@ mod tests { // Use wide range to capture essentially all density mass let grid_cte = build_grid_cte(&groups, 512, None, &AnsiDialect); let kernel = choose_kde_kernel(¶meters, None).expect("kernel should be valid"); - let sql = compute_density("x", &groups, kernel, &bw_cte, &data_cte, &grid_cte); + let sql = compute_density( + "x", + &groups, + kernel, + &bw_cte, + &data_cte, + &grid_cte, + &AnsiDialect, + ); // Execute query let reader = DuckDBReader::from_connection_string("duckdb://memory").unwrap(); @@ -1007,6 +1046,7 @@ mod tests { &bw_cte, &data_cte_unweighted, &grid_cte, + &AnsiDialect, ); let reader = DuckDBReader::from_connection_string("duckdb://memory").unwrap(); @@ -1017,8 +1057,15 @@ mod tests { // With explicit uniform weights (should be equivalent) let query_weighted = "SELECT x, 1.0 AS weight FROM (VALUES (1.0), (2.0), (3.0)) AS t(x)"; let data_cte_weighted = build_data_cte("x", None, Some("weight"), query_weighted, &groups); - let sql_weighted = - compute_density("x", &groups, kernel, &bw_cte, &data_cte_weighted, &grid_cte); + let sql_weighted = compute_density( + "x", + &groups, + kernel, + &bw_cte, + &data_cte_weighted, + &grid_cte, + &AnsiDialect, + ); let df_weighted = reader .execute_sql(&sql_weighted) .expect("SQL should execute"); @@ -1155,7 +1202,15 @@ mod tests { let data_cte = build_data_cte("x", None, None, query, &groups); let grid_cte = build_grid_cte(&groups, 512, None, &AnsiDialect); let kernel = choose_kde_kernel(¶meters, None).expect("kernel should be valid"); - let sql = compute_density("x", &groups, kernel, &bw_cte, &data_cte, &grid_cte); + let sql = compute_density( + "x", + &groups, + kernel, + &bw_cte, + &data_cte, + &grid_cte, + &AnsiDialect, + ); // Warm-up run reader.execute_sql(&sql).expect("Warm-up failed"); diff --git a/src/reader/cache.rs b/src/reader/cache.rs index 411205cff..c0a12c261 100644 --- a/src/reader/cache.rs +++ b/src/reader/cache.rs @@ -205,13 +205,10 @@ impl CachingReader { if self.meta_ready.get() { return Ok(()); } - let sql = format!( - "CREATE TABLE IF NOT EXISTS {} (\ - cache_key VARCHAR PRIMARY KEY, sql VARCHAR NOT NULL, table_name VARCHAR NOT NULL, \ - fetched_at_epoch_ms BIGINT NOT NULL, last_accessed_epoch_ms BIGINT NOT NULL, \ - byte_estimate BIGINT NOT NULL, row_count BIGINT NOT NULL)", - naming::quote_ident(naming::CACHE_META_TABLE) - ); + let sql = self + .cache + .dialect() + .cache_meta_table_sql(naming::CACHE_META_TABLE); self.cache.execute_sql(&sql)?; self.meta_ready.set(true); Ok(()) @@ -245,33 +242,27 @@ impl CachingReader { byte_estimate: i64, row_count: i64, ) -> Result<()> { - let now = now_ms(); - let stmt = format!( - "INSERT OR REPLACE INTO {} \ - (cache_key, sql, table_name, fetched_at_epoch_ms, last_accessed_epoch_ms, \ - byte_estimate, row_count) \ - VALUES ({}, {}, {}, {}, {}, {}, {})", - naming::quote_ident(naming::CACHE_META_TABLE), - naming::quote_literal(key), - naming::quote_literal(sql), - naming::quote_literal(table), - now, - now, + let stmts = self.cache.dialect().cache_meta_upsert_sql( + naming::CACHE_META_TABLE, + key, + sql, + table, + now_ms(), byte_estimate, row_count, ); - self.cache.execute_sql(&stmt)?; + for stmt in stmts { + self.cache.execute_sql(&stmt)?; + } Ok(()) } /// Advance the last-accessed timestamp for `key` (LRU bookkeeping). fn touch(&self, key: &str) -> Result<()> { - let stmt = format!( - "UPDATE {} SET last_accessed_epoch_ms = {} WHERE cache_key = {}", - naming::quote_ident(naming::CACHE_META_TABLE), - now_ms(), - naming::quote_literal(key), - ); + let stmt = + self.cache + .dialect() + .cache_meta_touch_sql(naming::CACHE_META_TABLE, key, now_ms()); self.cache.execute_sql(&stmt)?; Ok(()) } @@ -279,11 +270,10 @@ impl CachingReader { /// Drop a single memo entry: unregister the table, then delete its meta row. fn drop_entry(&self, key: &str, table: &str) -> Result<()> { self.cache.unregister(table)?; - let del = format!( - "DELETE FROM {} WHERE cache_key = {}", - naming::quote_ident(naming::CACHE_META_TABLE), - naming::quote_literal(key), - ); + let del = self + .cache + .dialect() + .cache_meta_delete_sql(naming::CACHE_META_TABLE, key); self.cache.execute_sql(&del)?; Ok(()) } diff --git a/src/reader/clickhouse/chdb.rs b/src/reader/clickhouse/chdb.rs new file mode 100644 index 000000000..7b143791c --- /dev/null +++ b/src/reader/clickhouse/chdb.rs @@ -0,0 +1,549 @@ +//! Embedded ClickHouse via [chDB](https://clickhouse.com/chdb) (libchdb). +//! +//! libchdb is loaded at runtime with `libloading`, so the `chdb` feature adds +//! no build-time dependency: a build with the feature enabled works everywhere, +//! and only a `chdb://` connection (or a `chdb+…://` cache) needs the library +//! present. It is searched in `GGSQL_CHDB_LIBRARY`, then the system library +//! path and the usual install locations. Install it with +//! `curl -sL https://lib.chdb.io | bash`. +//! +//! # Connection strings +//! +//! ```text +//! chdb:// in-memory engine (also chdb://memory, chdb://:memory:) +//! chdb:///path/to/dir persistent state under a directory +//! chdb://…?key=value extra `--key=value` engine arguments +//! ``` +//! +//! libchdb keeps a single connection per process. Readers opened on the same +//! path share it (the connection closes when the last reader drops); opening a +//! second path while one is in use is an error. + +use std::ffi::{c_char, c_int, c_void, CStr, CString}; +use std::path::PathBuf; +use std::sync::{Arc, Mutex, OnceLock, Weak}; + +use super::{ch_literal, percent_decode, ClickHouseSqlReader, Transport}; +use crate::reader::CacheBackend; +use crate::{GgsqlError, Result}; + +/// Reader for the embedded chDB engine. +pub type ChdbReader = ClickHouseSqlReader; + +/// Path spelling that selects the in-memory engine. +const MEMORY: &str = ":memory:"; + +impl ChdbReader { + /// Create a reader from a `chdb://` connection string. + pub fn from_connection_string(uri: &str) -> Result { + let rest = uri.strip_prefix("chdb://").ok_or_else(|| { + GgsqlError::ReaderError(format!( + "Invalid chDB connection string '{uri}': expected chdb://" + )) + })?; + let (path, query) = match rest.split_once('?') { + Some((p, q)) => (p, Some(q)), + None => (rest, None), + }; + let path = match path { + "" | "memory" | MEMORY => MEMORY.to_string(), + p => percent_decode(p), + }; + let args: Vec = query + .map(|q| { + q.split('&') + .filter(|s| !s.is_empty()) + .map(|s| match s.split_once('=') { + Some((k, v)) => format!("--{k}={}", percent_decode(v)), + None => format!("--{s}"), + }) + .collect() + }) + .unwrap_or_default(); + Self::from_transport(ChdbTransport::open(&path, &args)?) + } + + /// An in-memory engine. + pub fn in_memory() -> Result { + Self::from_transport(ChdbTransport::open(MEMORY, &[])?) + } + + /// An engine whose state persists under `path`. + pub fn with_path(path: &str) -> Result { + Self::from_transport(ChdbTransport::open(path, &[])?) + } + + /// Version of the loaded libchdb, e.g. `26.7.0`. + pub fn library_version() -> Result { + let api = Api::get()?; + match api.version { + Some(f) => Ok(unsafe { CStr::from_ptr(f()) } + .to_string_lossy() + .into_owned()), + None => Ok("unknown".to_string()), + } + } +} + +impl CacheBackend for ChdbReader { + fn new_in_memory() -> Result { + Self::in_memory() + } +} + +// ============================================================================= +// FFI +// ============================================================================= + +// `chdb_connection` is `struct chdb_connection_ *`; `chdb_connect` returns a +// pointer to one (`chdb_connection *`), which is also what `chdb_close_conn` +// takes. Queries take the dereferenced `chdb_connection`. +type Conn = *mut c_void; +type ConnHandle = *mut Conn; +type QueryResult = *mut c_void; + +struct Api { + _lib: libloading::Library, + connect: unsafe extern "C" fn(c_int, *mut *mut c_char) -> ConnHandle, + close_conn: unsafe extern "C" fn(ConnHandle), + query: unsafe extern "C" fn(Conn, *const c_char, *const c_char) -> QueryResult, + destroy_result: unsafe extern "C" fn(QueryResult), + result_buffer: unsafe extern "C" fn(QueryResult) -> *mut c_char, + result_length: unsafe extern "C" fn(QueryResult) -> usize, + result_error: unsafe extern "C" fn(QueryResult) -> *const c_char, + version: Option *const c_char>, +} + +static API: OnceLock> = OnceLock::new(); + +impl Api { + fn get() -> Result<&'static Api> { + API.get_or_init(Self::load).as_ref().map_err(|e| { + GgsqlError::ReaderError(format!( + "chDB is not available: {e}. Install libchdb with `curl -sL https://lib.chdb.io | bash`, \ + or set GGSQL_CHDB_LIBRARY to the path of libchdb.so" + )) + }) + } + + fn candidates() -> Vec { + let mut paths: Vec = Vec::new(); + if let Ok(p) = std::env::var("GGSQL_CHDB_LIBRARY") { + paths.push(PathBuf::from(p)); + } + let home = std::env::var("HOME").ok().map(PathBuf::from); + #[cfg(target_os = "macos")] + { + paths.push("libchdb.dylib".into()); + paths.push("/usr/local/lib/libchdb.dylib".into()); + paths.push("/opt/homebrew/lib/libchdb.dylib".into()); + if let Some(h) = &home { + paths.push(h.join(".local/lib/libchdb.dylib")); + } + } + #[cfg(target_os = "windows")] + { + paths.push("chdb.dll".into()); + paths.push("libchdb.dll".into()); + let _ = &home; + } + #[cfg(not(any(target_os = "macos", target_os = "windows")))] + { + paths.push("libchdb.so".into()); + paths.push("/usr/local/lib/libchdb.so".into()); + paths.push("/usr/lib/libchdb.so".into()); + if let Some(h) = &home { + paths.push(h.join(".local/lib/libchdb.so")); + } + } + paths + } + + fn load() -> std::result::Result { + let mut errors = Vec::new(); + for path in Self::candidates() { + match unsafe { libloading::Library::new(&path) } { + Ok(lib) => return unsafe { Self::from_library(lib) }, + Err(e) => errors.push(format!("{}: {e}", path.display())), + } + } + Err(format!("libchdb not found ({})", errors.join("; "))) + } + + unsafe fn from_library(lib: libloading::Library) -> std::result::Result { + macro_rules! sym { + ($name:literal) => { + *lib.get(concat!($name, "\0").as_bytes()) + .map_err(|e| format!("libchdb lacks {}: {e}", $name))? + }; + } + let api = Api { + connect: sym!("chdb_connect"), + close_conn: sym!("chdb_close_conn"), + query: sym!("chdb_query"), + destroy_result: sym!("chdb_destroy_query_result"), + result_buffer: sym!("chdb_result_buffer"), + result_length: sym!("chdb_result_length"), + result_error: sym!("chdb_result_error"), + version: lib + .get:: *const c_char>(b"chdb_version\0") + .ok() + .map(|s| *s), + _lib: lib, + }; + // chDB installs process-wide signal handlers by default, which would + // hijack the CLI's and the Jupyter kernel's own handling. + if let Ok(f) = api + ._lib + .get::(b"chdb_set_signal_handlers_enabled\0") + { + f(0); + } + Ok(api) + } +} + +/// The process-wide libchdb connection. +struct Connection { + api: &'static Api, + handle: ConnHandle, + conn: Conn, + path: String, + /// Serializes statements: the engine is one session. + lock: Mutex<()>, +} + +// The raw pointers are only ever used through the API, which documents its +// functions as thread-safe, and statements are additionally serialized. +unsafe impl Send for Connection {} +unsafe impl Sync for Connection {} + +impl Drop for Connection { + fn drop(&mut self) { + unsafe { (self.api.close_conn)(self.handle) } + } +} + +static SHARED: Mutex> = Mutex::new(Weak::new()); + +impl Connection { + fn open(path: &str, args: &[String]) -> Result> { + let mut shared = SHARED.lock().unwrap_or_else(|p| p.into_inner()); + if let Some(existing) = shared.upgrade() { + if existing.path == path { + return Ok(existing); + } + return Err(GgsqlError::ReaderError(format!( + "chDB supports one connection per process, and one is already open at '{}'; \ + cannot also open '{path}'", + existing.path + ))); + } + + let api = Api::get()?; + let mut argv: Vec = vec![CString::new("clickhouse").unwrap()]; + if path != MEMORY { + argv.push(CString::new(format!("--path={path}")).map_err(bad_arg)?); + } + for a in args { + argv.push(CString::new(a.as_str()).map_err(bad_arg)?); + } + let mut argv_ptrs: Vec<*mut c_char> = + argv.iter().map(|a| a.as_ptr() as *mut c_char).collect(); + + let handle = unsafe { (api.connect)(argv_ptrs.len() as c_int, argv_ptrs.as_mut_ptr()) }; + if handle.is_null() { + return Err(GgsqlError::ReaderError(format!( + "chdb_connect failed for path '{path}'" + ))); + } + let conn = unsafe { *handle }; + if conn.is_null() { + return Err(GgsqlError::ReaderError(format!( + "chdb_connect returned no connection for path '{path}'" + ))); + } + + let connection = Arc::new(Connection { + api, + handle, + conn, + path: path.to_string(), + lock: Mutex::new(()), + }); + // Results never leave the process, so compressing the Arrow stream + // only costs CPU. + connection.query( + "SET output_format_arrow_compression_method = 'none'", + "TabSeparated", + )?; + *shared = Arc::downgrade(&connection); + Ok(connection) + } + + fn query(&self, sql: &str, format: &str) -> Result> { + let csql = CString::new(sql).map_err(bad_arg)?; + let cfmt = CString::new(format).map_err(bad_arg)?; + let _guard = self.lock.lock().unwrap_or_else(|p| p.into_inner()); + let result = unsafe { (self.api.query)(self.conn, csql.as_ptr(), cfmt.as_ptr()) }; + if result.is_null() { + return Err(GgsqlError::ReaderError( + "chDB returned no result (connection closed?)".into(), + )); + } + let outcome = unsafe { + let err = (self.api.result_error)(result); + if !err.is_null() { + let message = CStr::from_ptr(err).to_string_lossy().trim().to_string(); + Err(GgsqlError::ReaderError(format!( + "ClickHouse error: {message}" + ))) + } else { + let buf = (self.api.result_buffer)(result); + let len = (self.api.result_length)(result); + if buf.is_null() || len == 0 { + Ok(Vec::new()) + } else { + Ok(std::slice::from_raw_parts(buf as *const u8, len).to_vec()) + } + } + }; + unsafe { (self.api.destroy_result)(result) }; + outcome + } +} + +fn bad_arg(e: std::ffi::NulError) -> GgsqlError { + GgsqlError::ReaderError(format!("argument contains a NUL byte: {e}")) +} + +// ============================================================================= +// Transport +// ============================================================================= + +/// Statements run in-process on libchdb. +pub struct ChdbTransport { + connection: Arc, +} + +impl ChdbTransport { + /// Open (or share) the process-wide connection for `path` + /// (`":memory:"` for the in-memory engine), passing `args` as extra + /// `--key=value` engine arguments on first open. + pub fn open(path: &str, args: &[String]) -> Result { + Ok(Self { + connection: Connection::open(path, args)?, + }) + } + + /// The engine path this transport is attached to. + pub fn path(&self) -> &str { + &self.connection.path + } +} + +impl Transport for ChdbTransport { + fn run(&self, sql: &str, format: &str) -> Result> { + self.connection.query(sql, format) + } + + /// The C API has no input-data channel for `INSERT … FORMAT`, so the + /// stream goes through a temporary file read back with `FROM INFILE`. + fn insert_arrow(&self, table: &str, ipc: &[u8]) -> Result<()> { + let path = std::env::temp_dir().join(format!("ggsql-chdb-{}.arrows", uuid::Uuid::new_v4())); + std::fs::write(&path, ipc).map_err(|e| { + GgsqlError::ReaderError(format!( + "Failed to write staging file {}: {e}", + path.display() + )) + })?; + let sql = format!( + "INSERT INTO {table} FROM INFILE {} FORMAT ArrowStream", + ch_literal(&path.to_string_lossy()) + ); + let outcome = self.connection.query(&sql, "TabSeparated"); + let _ = std::fs::remove_file(&path); + outcome.map(|_| ()) + } + + fn assumes_temporary_tables(&self) -> bool { + true + } + + fn endpoint(&self) -> String { + format!("chdb ({})", self.connection.path) + } +} + +// ============================================================================= +// Tests +// ============================================================================= + +#[cfg(test)] +mod tests { + use super::super::live_tests as shared; + use super::*; + use crate::reader::Reader; + + /// libchdb is one connection per process, and the executor's temp-table + /// names are per process too, so tests that run ggsql queries on chDB + /// must not overlap. Each test holds this for its whole body. + static SERIAL: Mutex<()> = Mutex::new(()); + + /// The in-memory engine plus the serialization guard, or `None` (skipping + /// the test) when libchdb is not installed on this machine. + fn reader_or_skip() -> Option<(std::sync::MutexGuard<'static, ()>, ChdbReader)> { + if let Err(e) = Api::get() { + eprintln!("skipping chDB test: {e}"); + return None; + } + let guard = SERIAL.lock().unwrap_or_else(|p| p.into_inner()); + Some(( + guard, + ChdbReader::in_memory().expect("chDB in-memory engine"), + )) + } + + #[test] + fn test_uri_forms_share_one_connection() { + let Some((_guard, a)) = reader_or_skip() else { + return; + }; + // Every in-memory spelling maps to the same process-wide connection. + let b = ChdbReader::from_connection_string("chdb://").unwrap(); + let c = ChdbReader::from_connection_string("chdb://memory").unwrap(); + let d = ChdbReader::from_connection_string("chdb://:memory:").unwrap(); + for r in [&b, &c, &d] { + assert_eq!(r.transport().path(), MEMORY); + } + assert!(Arc::ptr_eq( + &a.transport().connection, + &d.transport().connection + )); + // A different path cannot be opened while the shared one is alive. + let err = ChdbReader::with_path("/tmp/ggsql-chdb-other") + .err() + .unwrap() + .to_string(); + assert!(err.contains("one connection per process"), "{err}"); + assert!(ChdbReader::from_connection_string("duckdb://memory").is_err()); + assert!(ChdbReader::library_version().unwrap().contains('.')); + } + + #[test] + fn live_basic_types() { + let Some((_guard, reader)) = reader_or_skip() else { + return; + }; + shared::basic_types(&reader); + } + + #[test] + fn live_empty_result_keeps_schema() { + let Some((_guard, reader)) = reader_or_skip() else { + return; + }; + shared::empty_result_keeps_schema(&reader); + } + + #[test] + fn live_ddl_and_errors() { + let Some((_guard, reader)) = reader_or_skip() else { + return; + }; + shared::ddl_and_errors(&reader); + } + + #[test] + fn live_register_roundtrip() { + let Some((_guard, reader)) = reader_or_skip() else { + return; + }; + shared::register_roundtrip(&reader, "__ggsql_chdb_reg__"); + } + + #[test] + fn live_temp_tables_persist() { + let Some((_guard, reader)) = reader_or_skip() else { + return; + }; + shared::temp_tables_persist(&reader, "__ggsql_chdb_mat__"); + } + + #[test] + fn live_schema_introspection() { + let Some((_guard, reader)) = reader_or_skip() else { + return; + }; + shared::schema_introspection(&reader); + } + + #[cfg(feature = "vegalite")] + #[test] + fn live_execute_pipeline() { + let Some((_guard, reader)) = reader_or_skip() else { + return; + }; + shared::execute_pipeline(&reader); + } + + #[cfg(all(feature = "builtin-data", feature = "parquet"))] + #[test] + fn live_builtin_dataset() { + let Some((_guard, reader)) = reader_or_skip() else { + return; + }; + shared::builtin_dataset(&reader); + } + + #[test] + fn live_file_source() { + let Some((_guard, reader)) = reader_or_skip() else { + return; + }; + let path = std::env::temp_dir().join(format!("ggsql-chdb-{}.csv", uuid::Uuid::new_v4())); + std::fs::write(&path, "a,b\n1,2\n3,4\n").unwrap(); + // The executor spells file sources as `FROM ''`. + let df = reader + .execute_sql(&format!("SELECT * FROM '{}' ORDER BY a", path.display())) + .unwrap(); + let _ = std::fs::remove_file(&path); + assert_eq!(df.height(), 2); + assert_eq!(df.get_column_names(), vec!["a", "b"]); + } + + /// chDB as the cache behind a read-only primary: the executor's temporary + /// tables, the memo table and its bookkeeping all live on the engine. + #[test] + fn live_as_cache_backend() { + use crate::reader::cache::CachingReader; + use crate::reader::test_support::ReadOnlyReader; + + let Some((_guard, cache)) = reader_or_skip() else { + return; + }; + let primary = ReadOnlyReader::new(Box::new(ChdbReader::in_memory().unwrap())); + let reader = CachingReader::new( + Box::new(primary), + Box::new(cache), + "test://readonly-primary", + "chdb", + ); + let query = "SELECT number AS x, number * 2 AS y FROM numbers(6) \ + VISUALISE x, y DRAW point"; + let spec = reader.execute(query).unwrap(); + assert_eq!(spec.metadata().rows, 6); + // Second run is served from the memo; the memo table is queryable. + let spec = reader.execute(query).unwrap(); + assert_eq!(spec.metadata().rows, 6); + let meta = reader + .execute_sql("SELECT cache_key, row_count FROM __ggsql_cache_meta__") + .unwrap(); + assert!(meta.height() >= 1, "memo rows expected"); + reader.clear_cache().unwrap(); + let meta = reader + .execute_sql("SELECT cache_key FROM __ggsql_cache_meta__") + .unwrap(); + assert_eq!(meta.height(), 0); + } +} diff --git a/src/reader/clickhouse/http.rs b/src/reader/clickhouse/http.rs new file mode 100644 index 000000000..5abf597b7 --- /dev/null +++ b/src/reader/clickhouse/http.rs @@ -0,0 +1,456 @@ +//! ClickHouse server access over the HTTP interface. +//! +//! Plain HTTP(S) requests: the SQL travels in the request body, results come +//! back as `FORMAT ArrowStream`, and registered DataFrames are bulk-loaded with +//! `INSERT … FORMAT ArrowStream`. No client library or native-protocol driver +//! is needed. +//! +//! # Connection strings +//! +//! ```text +//! clickhouse://[user[:password]@]host[:port][/database][?param=value&…] +//! clickhouses://… TLS; default port 8443 +//! ``` +//! +//! Every query parameter other than `secure`, `user`, `password` and +//! `database` is forwarded verbatim to the server on each request, so any +//! ClickHouse setting or HTTP-interface parameter can be set per connection +//! (`?session_timezone=UTC&max_threads=4&session_timeout=3600`). +//! +//! When the URI omits them, the host, user and password fall back to the +//! `CLICKHOUSE_HOST`, `CLICKHOUSE_USER` and `CLICKHOUSE_PASSWORD` environment +//! variables, so credentials need not appear in the connection string. +//! +//! Each transport owns one server session (a fresh `session_id`), so temporary +//! tables and `SET` statements persist across requests. + +use std::time::Duration; + +use super::{percent_decode, ClickHouseSqlReader, Transport}; +use crate::{GgsqlError, Result}; + +const DEFAULT_HTTP_PORT: u16 = 8123; +const DEFAULT_HTTPS_PORT: u16 = 8443; + +/// Reader for a ClickHouse server reached over HTTP. +pub type ClickHouseReader = ClickHouseSqlReader; + +impl ClickHouseReader { + /// Create a reader from a `clickhouse://` / `clickhouses://` connection + /// string. Connects eagerly. + pub fn from_connection_string(uri: &str) -> Result { + Self::new(ClickHouseConfig::from_uri(uri)?) + } + + /// Create a reader from an already-parsed configuration. Connects eagerly. + pub fn new(config: ClickHouseConfig) -> Result { + Self::from_transport(HttpTransport::new(config)) + } + + /// The parsed connection configuration. + pub fn config(&self) -> &ClickHouseConfig { + &self.transport().config + } +} + +// ============================================================================= +// Connection configuration +// ============================================================================= + +/// Parsed form of a `clickhouse://` / `clickhouses://` connection string. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ClickHouseConfig { + /// Server endpoint including scheme and port, e.g. `http://localhost:8123`. + pub base_url: String, + /// Default database for unqualified table names (`None` = server default). + pub database: Option, + pub user: String, + pub password: String, + /// Extra URL parameters forwarded to the server on every request. + pub params: Vec<(String, String)>, +} + +impl ClickHouseConfig { + /// Parse a connection string. See the [module docs](self) for the format. + pub fn from_uri(uri: &str) -> Result { + let (mut secure, rest) = if let Some(rest) = uri.strip_prefix("clickhouses://") { + (true, rest) + } else if let Some(rest) = uri.strip_prefix("clickhouse://") { + (false, rest) + } else { + return Err(GgsqlError::ReaderError(format!( + "Invalid ClickHouse connection string '{uri}': expected clickhouse:// or clickhouses://" + ))); + }; + + let (rest, query) = match rest.split_once('?') { + Some((r, q)) => (r, Some(q)), + None => (rest, None), + }; + + let (userinfo, hostpath) = match rest.rsplit_once('@') { + Some((u, h)) => (Some(u), h), + None => (None, rest), + }; + + let (hostport, database) = match hostpath.split_once('/') { + Some((h, d)) if !d.is_empty() => (h, Some(percent_decode(d))), + Some((h, _)) => (h, None), + None => (hostpath, None), + }; + + let (mut user, mut password) = match userinfo { + Some(info) => match info.split_once(':') { + Some((u, p)) => (Some(percent_decode(u)), Some(percent_decode(p))), + None => (Some(percent_decode(info)), None), + }, + None => (None, None), + }; + let mut database = database; + + let mut params = Vec::new(); + if let Some(query) = query { + for segment in query.split('&').filter(|s| !s.is_empty()) { + let (key, value) = segment.split_once('=').unwrap_or((segment, "")); + let value = percent_decode(value); + match key { + "secure" => { + secure = matches!( + value.to_ascii_lowercase().as_str(), + "" | "1" | "true" | "yes" + ) + } + "user" => user = Some(value), + "password" => password = Some(value), + "database" => database = Some(value), + _ => params.push((key.to_string(), value)), + } + } + } + + let (host, port) = split_host_port(hostport)?; + let host = if host.is_empty() { + std::env::var("CLICKHOUSE_HOST").unwrap_or_else(|_| "localhost".to_string()) + } else { + host.to_string() + }; + let port = port.unwrap_or(if secure { + DEFAULT_HTTPS_PORT + } else { + DEFAULT_HTTP_PORT + }); + let scheme = if secure { "https" } else { "http" }; + + Ok(Self { + base_url: format!("{scheme}://{host}:{port}"), + database, + user: user + .or_else(|| std::env::var("CLICKHOUSE_USER").ok()) + .unwrap_or_else(|| "default".to_string()), + password: password + .or_else(|| std::env::var("CLICKHOUSE_PASSWORD").ok()) + .unwrap_or_default(), + params, + }) + } +} + +/// Split `host[:port]`, accepting bracketed IPv6 literals (`[::1]:8123`). +fn split_host_port(hostport: &str) -> Result<(&str, Option)> { + let parse_port = |p: &str| -> Result { + p.parse::().map_err(|_| { + GgsqlError::ReaderError(format!( + "Invalid port '{p}' in ClickHouse connection string" + )) + }) + }; + if hostport.starts_with('[') { + let end = hostport.find(']').ok_or_else(|| { + GgsqlError::ReaderError(format!( + "Unterminated IPv6 literal in ClickHouse host '{hostport}'" + )) + })?; + let host = &hostport[..=end]; + return match hostport[end + 1..].strip_prefix(':') { + Some(p) => Ok((host, Some(parse_port(p)?))), + None => Ok((host, None)), + }; + } + match hostport.rsplit_once(':') { + Some((host, port)) => Ok((host, Some(parse_port(port)?))), + None => Ok((hostport, None)), + } +} + +// ============================================================================= +// Transport +// ============================================================================= + +/// One HTTP session against a ClickHouse server. +pub struct HttpTransport { + agent: ureq::Agent, + config: ClickHouseConfig, + session_id: String, +} + +/// Body limit for responses. ClickHouse results are read fully into memory. +const RESPONSE_LIMIT: u64 = u64::MAX; + +impl HttpTransport { + pub fn new(config: ClickHouseConfig) -> Self { + let agent_config = ureq::Agent::config_builder() + .http_status_as_error(false) + .timeout_connect(Some(Duration::from_secs(30))) + .user_agent(format!("ggsql/{}", crate::VERSION)) + .build(); + Self { + agent: ureq::Agent::new_with_config(agent_config), + config, + session_id: uuid::Uuid::new_v4().to_string(), + } + } + + /// Build a request carrying the session, credentials, output format and + /// per-connection parameters. `query` puts the SQL in the URL so the body + /// can carry data (used for `INSERT … FORMAT ArrowStream`). + fn request( + &self, + format: &str, + query: Option<&str>, + ) -> ureq::RequestBuilder { + let mut req = self + .agent + .post(format!("{}/", self.config.base_url)) + .header("X-ClickHouse-User", &self.config.user) + .header("X-ClickHouse-Key", &self.config.password) + .query("session_id", &self.session_id) + .query("default_format", format) + // Buffer the whole result server-side so an error raised while + // streaming still surfaces as a non-200 status with a message, + // instead of garbage appended to a partial Arrow stream. + .query("wait_end_of_query", "1") + // Compress responses; ureq decompresses transparently. + .query("enable_http_compression", "1"); + if let Some(db) = &self.config.database { + req = req.query("database", db); + } + if let Some(q) = query { + req = req.query("query", q); + } + for (k, v) in &self.config.params { + req = req.query(k, v); + } + req + } + + /// Send a request and return the response body, turning any non-200 + /// status into a `ReaderError` carrying the server's message. + fn send( + &self, + req: ureq::RequestBuilder, + body: impl ureq::AsSendBody, + ) -> Result> { + let mut resp = req.send(body).map_err(|e| { + GgsqlError::ReaderError(format!( + "ClickHouse request to {} failed: {e}", + self.config.base_url + )) + })?; + let status = resp.status().as_u16(); + let bytes = resp + .body_mut() + .with_config() + .limit(RESPONSE_LIMIT) + .read_to_vec() + .map_err(|e| { + GgsqlError::ReaderError(format!("Failed to read ClickHouse response: {e}")) + })?; + if status != 200 { + let message = String::from_utf8_lossy(&bytes).trim().to_string(); + let message = if message.is_empty() { + format!("HTTP status {status}") + } else { + message + }; + return Err(GgsqlError::ReaderError(format!( + "ClickHouse error: {message}" + ))); + } + Ok(bytes) + } +} + +impl Transport for HttpTransport { + fn run(&self, sql: &str, format: &str) -> Result> { + self.send(self.request(format, None), sql) + } + + fn insert_arrow(&self, table: &str, ipc: &[u8]) -> Result<()> { + let insert = format!("INSERT INTO {table} FORMAT ArrowStream"); + self.send(self.request("TabSeparated", Some(&insert)), ipc)?; + Ok(()) + } + + fn endpoint(&self) -> String { + self.config.base_url.clone() + } +} + +// ============================================================================= +// Tests +// ============================================================================= + +#[cfg(test)] +mod tests { + use super::*; + + // ---- Connection strings (no server needed) ----------------------------- + + #[test] + fn test_config_full_uri() { + let c = ClickHouseConfig::from_uri( + "clickhouse://alice:s%40cret@db.example.com:9999/analytics?max_threads=4&session_timezone=UTC", + ) + .unwrap(); + assert_eq!(c.base_url, "http://db.example.com:9999"); + assert_eq!(c.database.as_deref(), Some("analytics")); + assert_eq!(c.user, "alice"); + assert_eq!(c.password, "s@cret"); + assert_eq!( + c.params, + vec![ + ("max_threads".to_string(), "4".to_string()), + ("session_timezone".to_string(), "UTC".to_string()), + ] + ); + } + + #[test] + fn test_config_defaults() { + let c = ClickHouseConfig::from_uri("clickhouse://myhost").unwrap(); + assert_eq!(c.base_url, "http://myhost:8123"); + assert_eq!(c.database, None); + assert!(c.params.is_empty()); + + let c = ClickHouseConfig::from_uri("clickhouse://myhost/").unwrap(); + assert_eq!(c.database, None); + } + + #[test] + fn test_config_secure() { + let c = ClickHouseConfig::from_uri("clickhouses://play.clickhouse.com").unwrap(); + assert_eq!(c.base_url, "https://play.clickhouse.com:8443"); + + let c = + ClickHouseConfig::from_uri("clickhouses://explorer@play.clickhouse.com:443").unwrap(); + assert_eq!(c.base_url, "https://play.clickhouse.com:443"); + assert_eq!(c.user, "explorer"); + + let c = ClickHouseConfig::from_uri("clickhouse://h?secure=1").unwrap(); + assert_eq!(c.base_url, "https://h:8443"); + assert!(c.params.is_empty(), "secure is consumed, not forwarded"); + + let c = ClickHouseConfig::from_uri("clickhouses://h?secure=false").unwrap(); + assert_eq!(c.base_url, "http://h:8123"); + } + + #[test] + fn test_config_credentials_as_params() { + let c = + ClickHouseConfig::from_uri("clickhouse://h/?user=bob&password=pw&database=db").unwrap(); + assert_eq!(c.user, "bob"); + assert_eq!(c.password, "pw"); + assert_eq!(c.database.as_deref(), Some("db")); + assert!(c.params.is_empty()); + } + + #[test] + fn test_config_ipv6() { + let c = ClickHouseConfig::from_uri("clickhouse://[::1]:8124/db").unwrap(); + assert_eq!(c.base_url, "http://[::1]:8124"); + assert_eq!(c.database.as_deref(), Some("db")); + + let c = ClickHouseConfig::from_uri("clickhouse://[::1]").unwrap(); + assert_eq!(c.base_url, "http://[::1]:8123"); + } + + #[test] + fn test_config_password_with_at_sign() { + let c = ClickHouseConfig::from_uri("clickhouse://u:p@ss@h").unwrap(); + assert_eq!(c.user, "u"); + assert_eq!(c.password, "p@ss"); + assert_eq!(c.base_url, "http://h:8123"); + } + + #[test] + fn test_config_rejects_other_schemes_and_bad_ports() { + assert!(ClickHouseConfig::from_uri("duckdb://memory").is_err()); + assert!(ClickHouseConfig::from_uri("clickhouse://h:notaport").is_err()); + assert!(ClickHouseConfig::from_uri("clickhouse://[::1").is_err()); + } + + // ---- Against a live server --------------------------------------------- + // + // Set GGSQL_CLICKHOUSE_URI (e.g. `clickhouse://localhost:8123`) to run + // these; they are skipped otherwise. + + use super::super::live_tests as shared; + + fn live_reader() -> Option { + let uri = std::env::var("GGSQL_CLICKHOUSE_URI").ok()?; + Some( + ClickHouseReader::from_connection_string(&uri) + .unwrap_or_else(|e| panic!("cannot connect to {uri}: {e}")), + ) + } + + #[test] + fn live_basic_types() { + let Some(reader) = live_reader() else { return }; + shared::basic_types(&reader); + } + + #[test] + fn live_empty_result_keeps_schema() { + let Some(reader) = live_reader() else { return }; + shared::empty_result_keeps_schema(&reader); + } + + #[test] + fn live_ddl_and_errors() { + let Some(reader) = live_reader() else { return }; + shared::ddl_and_errors(&reader); + } + + #[test] + fn live_register_roundtrip() { + let Some(reader) = live_reader() else { return }; + shared::register_roundtrip(&reader, "__ggsql_http_reg__"); + } + + #[test] + fn live_temp_tables_persist_within_session() { + let Some(reader) = live_reader() else { return }; + shared::temp_tables_persist(&reader, "__ggsql_http_mat__"); + } + + #[test] + fn live_schema_introspection() { + let Some(reader) = live_reader() else { return }; + shared::schema_introspection(&reader); + } + + #[cfg(feature = "vegalite")] + #[test] + fn live_execute_pipeline() { + let Some(reader) = live_reader() else { return }; + shared::execute_pipeline(&reader); + } + + #[cfg(all(feature = "builtin-data", feature = "parquet"))] + #[test] + fn live_builtin_dataset() { + let Some(reader) = live_reader() else { return }; + shared::builtin_dataset(&reader); + } +} diff --git a/src/reader/clickhouse/mod.rs b/src/reader/clickhouse/mod.rs new file mode 100644 index 000000000..5413a8e8e --- /dev/null +++ b/src/reader/clickhouse/mod.rs @@ -0,0 +1,1193 @@ +//! ClickHouse readers: a remote server over HTTP, and the embedded chDB engine. +//! +//! Both speak ClickHouse SQL and both exchange data as Arrow IPC streams, so +//! they share one reader implementation, [`ClickHouseSqlReader`], that is +//! generic over a [`Transport`]: +//! +//! - [`ClickHouseReader`] (`clickhouse://`, feature `clickhouse`) sends each +//! statement to a server's HTTP interface ([`http::HttpTransport`]). +//! - [`ChdbReader`] (`chdb://`, feature `chdb`) runs statements in-process on +//! libchdb, loaded at runtime ([`chdb::ChdbTransport`]). It also serves as +//! the in-memory [`CacheBackend`] behind `chdb+://` connection +//! strings, so a ClickHouse setup never needs another database engine. +//! +//! # Types +//! +//! ClickHouse's Arrow output cannot express some of its own types: `DateTime` +//! arrives as a bare `UInt32`, `Enum` as its integer codes, `UUID`/`IPv4`/ +//! `IPv6`/128- and 256-bit integers as raw bytes, and `Decimal` as an Arrow +//! decimal the plot pipeline does not consume. Before running a `SELECT`, the +//! reader asks the engine for the result's column types (`DESCRIBE TABLE (…)`) +//! and, when any such column is present, wraps the query in +//! `SELECT * REPLACE (…)` that converts those columns server-side +//! (`DateTime` → `DateTime64`, everything else → `String` or `Float64`). +//! Timestamps keep their instant; ggsql renders them in UTC. +//! +//! # Temporary tables +//! +//! The executor materializes CTEs and the global query as temporary tables. +//! Each reader owns one session, so those tables (and `SET` statements) stay +//! visible across the statements of a ggsql query. A server account without +//! the `CREATE TEMPORARY TABLE` privilege is detected on connect +//! ([`ClickHouseSqlReader::supports_temporary_tables`]); the connection +//! factory then keeps intermediate tables in an embedded chDB cache instead. + +use std::cell::{OnceCell, RefCell}; +use std::collections::HashSet; +use std::io::Cursor; + +use arrow::array::RecordBatch; +use arrow::datatypes::{DataType, TimeUnit}; +use arrow::ipc::reader::StreamReader; +use arrow::ipc::writer::StreamWriter; + +use crate::array_util::value_to_string; +use crate::reader::{ColumnInfo, Reader, SqlDialect, TableInfo}; +use crate::{naming, DataFrame, GgsqlError, Result}; + +#[cfg(feature = "chdb")] +pub mod chdb; +#[cfg(feature = "clickhouse")] +pub mod http; + +#[cfg(feature = "chdb")] +pub use chdb::{ChdbReader, ChdbTransport}; +#[cfg(feature = "clickhouse")] +pub use http::{ClickHouseConfig, ClickHouseReader, HttpTransport}; + +// ============================================================================= +// Dialect +// ============================================================================= + +/// ClickHouse SQL dialect, shared by the HTTP and chDB readers. +/// +/// Cast targets are `Nullable(…)` because ClickHouse refuses to cast `NULL` to +/// a non-nullable type; temporary tables need the `TEMPORARY` keyword; and +/// quantiles use the native `quantileExactInclusive`, which matches the +/// linear-interpolation semantics of `QUANTILE_CONT`. The caching layer's memo +/// table is a `Memory` table maintained with synchronous `ALTER TABLE` +/// mutations, since ClickHouse has no `INSERT OR REPLACE`, `UPDATE` or +/// `DELETE FROM` for that engine. +pub struct ClickHouseDialect; + +/// ClickHouse string literal: single quotes doubled, backslashes escaped +/// (ClickHouse interprets backslash escapes inside string literals). +pub(crate) fn ch_literal(s: &str) -> String { + format!("'{}'", s.replace('\\', "\\\\").replace('\'', "''")) +} + +/// Print `sql` to stderr when `GGSQL_CLICKHOUSE_TRACE` is set, so the +/// statements actually sent (including the `DESCRIBE`/`REPLACE` rewrites) can +/// be inspected. +fn trace_sql(sql: &str) { + static ENABLED: std::sync::OnceLock = std::sync::OnceLock::new(); + if *ENABLED.get_or_init(|| std::env::var_os("GGSQL_CLICKHOUSE_TRACE").is_some()) { + eprintln!("[clickhouse] {sql}"); + } +} + +/// Comma-separated argument list with every expression cast to Float64. +fn float_args(exprs: &[&str]) -> String { + exprs + .iter() + .map(|e| format!("toFloat64({e})")) + .collect::>() + .join(", ") +} + +impl SqlDialect for ClickHouseDialect { + fn number_type_name(&self) -> Option<&str> { + Some("Nullable(Float64)") + } + + fn integer_type_name(&self) -> Option<&str> { + Some("Nullable(Int64)") + } + + fn date_type_name(&self) -> Option<&str> { + Some("Nullable(Date32)") + } + + fn datetime_type_name(&self) -> Option<&str> { + Some("Nullable(DateTime64(6))") + } + + /// ClickHouse has no portable time-of-day type; time columns are left as is. + fn time_type_name(&self) -> Option<&str> { + None + } + + fn string_type_name(&self) -> Option<&str> { + Some("Nullable(String)") + } + + fn boolean_type_name(&self) -> Option<&str> { + Some("Nullable(Bool)") + } + + // The executor only asks for greatest/least of numeric expressions. The + // arguments are cast to Float64 because ClickHouse has no common supertype + // for UInt64 (the type of `count()` and `numbers()`) and Float64. + + fn sql_greatest(&self, exprs: &[&str]) -> String { + if exprs.len() == 1 { + return exprs[0].to_string(); + } + format!("greatest({})", float_args(exprs)) + } + + fn sql_least(&self, exprs: &[&str]) -> String { + if exprs.len() == 1 { + return exprs[0].to_string(); + } + format!("least({})", float_args(exprs)) + } + + /// Older ClickHouse versions only accept `IS NOT DISTINCT FROM` inside + /// `JOIN ON`; this spelling works in any clause on every version. + fn sql_null_safe_equals(&self, left: &str, right: &str) -> String { + format!("(({left} = {right}) OR ({left} IS NULL AND {right} IS NULL))") + } + + fn sql_select_replace( + &self, + expr: &str, + col: &str, + from: &str, + _all_columns: &[String], + ) -> String { + format!("SELECT * REPLACE ({expr} AS {col}) FROM ({from})") + } + + fn sql_generate_series(&self, n: usize) -> String { + format!("\"__ggsql_seq__\"(n) AS (SELECT toFloat64(number) AS n FROM numbers({n}))") + } + + fn sql_quantile_inline(&self, column: &str, fraction: f64) -> Option { + Some(format!( + "quantileExactInclusive({})({})", + fraction, + naming::quote_ident(column) + )) + } + + /// Every caller embeds this in a `GROUP BY {groups}` query over `from`, so + /// the native aggregate is equivalent to the correlated scalar subquery + /// other dialects produce, and it also runs on ClickHouse versions without + /// correlated-subquery support. + fn sql_percentile( + &self, + column: &str, + fraction: f64, + _from: &str, + _groups: &[String], + ) -> String { + format!( + "quantileExactInclusive({fraction})({})", + naming::quote_ident(column) + ) + } + + fn sql_date_literal(&self, days_since_epoch: i32) -> String { + format!("toDate32({days_since_epoch})") + } + + fn sql_datetime_literal(&self, microseconds_since_epoch: i64) -> String { + format!("fromUnixTimestamp64Micro({microseconds_since_epoch})") + } + + fn create_or_replace_temp_table_sql( + &self, + name: &str, + column_aliases: &[String], + body_sql: &str, + ) -> Vec { + let qname = naming::quote_ident(name); + let body = super::wrap_with_column_aliases(body_sql, column_aliases); + vec![ + format!("DROP TEMPORARY TABLE IF EXISTS {qname}"), + format!("CREATE TEMPORARY TABLE {qname} AS {body}"), + ] + } + + fn cache_meta_table_sql(&self, table: &str) -> String { + format!( + "CREATE TABLE IF NOT EXISTS {} (\ + cache_key String, sql String, table_name String, \ + fetched_at_epoch_ms Int64, last_accessed_epoch_ms Int64, \ + byte_estimate Int64, row_count Int64) ENGINE = Memory", + naming::quote_ident(table) + ) + } + + fn cache_meta_upsert_sql( + &self, + table: &str, + key: &str, + sql: &str, + table_name: &str, + now_ms: i64, + byte_estimate: i64, + row_count: i64, + ) -> Vec { + let q = naming::quote_ident(table); + vec![ + format!( + "ALTER TABLE {q} DELETE WHERE cache_key = {}", + ch_literal(key) + ), + format!( + "INSERT INTO {q} \ + (cache_key, sql, table_name, fetched_at_epoch_ms, last_accessed_epoch_ms, \ + byte_estimate, row_count) \ + VALUES ({}, {}, {}, {now_ms}, {now_ms}, {byte_estimate}, {row_count})", + ch_literal(key), + ch_literal(sql), + ch_literal(table_name), + ), + ] + } + + fn cache_meta_touch_sql(&self, table: &str, key: &str, now_ms: i64) -> String { + format!( + "ALTER TABLE {} UPDATE last_accessed_epoch_ms = {now_ms} WHERE cache_key = {}", + naming::quote_ident(table), + ch_literal(key) + ) + } + + fn cache_meta_delete_sql(&self, table: &str, key: &str) -> String { + format!( + "ALTER TABLE {} DELETE WHERE cache_key = {}", + naming::quote_ident(table), + ch_literal(key) + ) + } +} + +// ============================================================================= +// Transport +// ============================================================================= + +/// How statements reach a ClickHouse engine. +/// +/// Implementations run one statement at a time within a single session, so +/// temporary tables and `SET` statements persist between calls. +pub trait Transport: Send { + /// Run one statement and return the raw bytes of its output in `format` + /// (a ClickHouse output format name such as `ArrowStream`). Statements + /// without a result set return an empty buffer. + fn run(&self, sql: &str, format: &str) -> Result>; + + /// Bulk-load an Arrow IPC stream into the existing table `table` + /// (already quoted for SQL). + fn insert_arrow(&self, table: &str, ipc: &[u8]) -> Result<()>; + + /// Whether `CREATE TEMPORARY TABLE` is known to work without probing the + /// engine. `false` makes the reader check once on first use. + fn assumes_temporary_tables(&self) -> bool { + false + } + + /// Short description of the endpoint for error messages. + fn endpoint(&self) -> String; +} + +// ============================================================================= +// Reader +// ============================================================================= + +/// Reader for any ClickHouse engine reachable through a [`Transport`]. +/// +/// Use the concrete aliases: [`ClickHouseReader`] for a server, [`ChdbReader`] +/// for the embedded engine. +pub struct ClickHouseSqlReader { + transport: T, + registered_tables: RefCell>, + temp_tables_supported: OnceCell, +} + +impl ClickHouseSqlReader { + /// Wrap a transport. Connects eagerly: a `SELECT 1` round trip verifies + /// the endpoint and credentials so a bad connection string fails here + /// rather than on the first query. + pub fn from_transport(transport: T) -> Result { + let reader = Self { + transport, + registered_tables: RefCell::new(HashSet::new()), + temp_tables_supported: OnceCell::new(), + }; + reader.query_arrow("SELECT 1")?; + Ok(reader) + } + + /// The underlying transport. + pub fn transport(&self) -> &T { + &self.transport + } + + /// Whether this session may create temporary tables, which the executor + /// needs for CTEs and the global query. Probed once (a `CREATE TEMPORARY + /// TABLE` that is dropped again); a read-only account answers `false`. + pub fn supports_temporary_tables(&self) -> bool { + *self.temp_tables_supported.get_or_init(|| { + if self.transport.assumes_temporary_tables() { + return true; + } + let probe = naming::quote_ident(&format!("__ggsql_probe_{}__", naming::session_id())); + let ok = self + .execute_raw(&format!("CREATE TEMPORARY TABLE {probe} (x UInt8)")) + .is_ok(); + if ok { + let _ = self.execute_raw(&format!("DROP TEMPORARY TABLE IF EXISTS {probe}")); + } + ok + }) + } + + /// Run a statement that returns no rows. + fn execute_raw(&self, sql: &str) -> Result<()> { + trace_sql(sql); + self.transport.run(sql, "TabSeparated")?; + Ok(()) + } + + /// Run a row-returning statement as is (no type rewriting) and decode the + /// Arrow stream. + fn query_arrow(&self, sql: &str) -> Result { + trace_sql(sql); + decode_arrow_stream(&self.transport.run(sql, "ArrowStream")?) + } + + /// Column names and ClickHouse type names of a query's result. + fn describe(&self, sql: &str) -> Result> { + let df = self.query_arrow(&format!("DESCRIBE TABLE ({sql})"))?; + let names = df.column("name")?; + let types = df.column("type")?; + Ok((0..df.height()) + .map(|i| (value_to_string(names, i), value_to_string(types, i))) + .collect()) + } + + /// Wrap a `SELECT`/`WITH` query so that columns whose ClickHouse type has + /// no faithful Arrow representation are converted server-side. Any other + /// statement, or a query the engine cannot describe, is returned unchanged + /// so the real execution reports the real error. + fn with_arrow_friendly_types(&self, sql: &str) -> String { + let first = sql + .split_whitespace() + .next() + .unwrap_or("") + .to_ascii_uppercase(); + if first != "SELECT" && first != "WITH" { + return sql.to_string(); + } + let Ok(columns) = self.describe(sql) else { + return sql.to_string(); + }; + let replacements: Vec = columns + .iter() + .filter_map(|(name, ty)| arrow_friendly_expr(name, ty)) + .collect(); + if replacements.is_empty() { + return sql.to_string(); + } + format!( + "SELECT * REPLACE ({}) FROM ({sql})", + replacements.join(", ") + ) + } + + fn temporary_table_exists(&self, name: &str) -> Result { + let df = self.query_arrow(&format!( + "EXISTS TEMPORARY TABLE {}", + naming::quote_ident(name) + ))?; + Ok(df.height() == 1 && value_to_string(df.column("result")?, 0) == "1") + } + + /// Load any `ggsql:` builtin datasets referenced by `sql` into the + /// session as temporary tables. + #[cfg(all(feature = "builtin-data", feature = "parquet"))] + fn ensure_builtin_datasets(&self, sql: &str) -> Result<()> { + for name in crate::parser::extract_builtin_dataset_names(sql)? { + let table = naming::builtin_data_table(&name); + if !self.temporary_table_exists(&table)? { + let df = super::data::load_builtin_dataframe(&name)?; + self.register(&table, df, true)?; + } + } + Ok(()) + } +} + +impl Reader for ClickHouseSqlReader { + fn execute_sql(&self, sql: &str) -> Result { + #[cfg(all(feature = "builtin-data", feature = "parquet"))] + self.ensure_builtin_datasets(sql)?; + + let sql = crate::parser::rewrite_namespaced_sql(sql)?; + // One statement per call; a trailing terminator is harmless to the + // engine but would break the `DESCRIBE TABLE (…)` / + // `SELECT * REPLACE … FROM (…)` wrapping. + let sql = sql.trim().trim_end_matches(';').trim(); + + if !super::returns_rows(sql) { + self.execute_raw(sql)?; + return Ok(DataFrame::empty()); + } + + self.query_arrow(&self.with_arrow_friendly_types(sql)) + } + + fn register(&self, name: &str, df: DataFrame, replace: bool) -> Result<()> { + super::validate_table_name(name)?; + let qname = naming::quote_ident(name); + + if replace { + self.execute_raw(&format!("DROP TEMPORARY TABLE IF EXISTS {qname}"))?; + } else if self.temporary_table_exists(name)? { + return Err(GgsqlError::ReaderError(format!( + "Table '{name}' already exists" + ))); + } + + let schema = df.schema(); + let col_defs: Vec = schema + .fields() + .iter() + .map(|f| { + format!( + "{} {}", + naming::quote_ident(f.name()), + arrow_type_to_clickhouse(f.data_type(), f.is_nullable()) + ) + }) + .collect(); + self.execute_raw(&format!( + "CREATE TEMPORARY TABLE {qname} ({})", + col_defs.join(", ") + )) + .map_err(|e| { + GgsqlError::ReaderError(format!("Failed to create temp table '{name}': {e}")) + })?; + + if df.height() > 0 { + let ipc = encode_arrow_stream(df)?; + self.transport.insert_arrow(&qname, &ipc).map_err(|e| { + GgsqlError::ReaderError(format!("Failed to insert into '{name}': {e}")) + })?; + } + + self.registered_tables.borrow_mut().insert(name.to_string()); + Ok(()) + } + + fn unregister(&self, name: &str) -> Result<()> { + if !self.registered_tables.borrow().contains(name) { + return Err(GgsqlError::ReaderError(format!( + "Table '{name}' was not registered via this reader" + ))); + } + self.execute_raw(&format!( + "DROP TEMPORARY TABLE IF EXISTS {}", + naming::quote_ident(name) + ))?; + self.registered_tables.borrow_mut().remove(name); + Ok(()) + } + + fn execute(&self, query: &str) -> Result { + super::execute_with_reader(self, query) + } + + fn dialect(&self) -> &dyn SqlDialect { + &ClickHouseDialect + } + + // ClickHouse has a single level of namespacing: a database is both the + // catalog and the schema. `system.*` is more reliable than + // `information_schema`, which lists every column twice (upper- and + // lower-case names). + + fn list_catalogs(&self) -> Result> { + let df = self.query_arrow( + "SELECT name FROM system.databases \ + WHERE name NOT IN ('system', 'INFORMATION_SCHEMA', 'information_schema') \ + ORDER BY name", + )?; + column_strings(&df, "name") + } + + fn list_schemas(&self, catalog: &str) -> Result> { + Ok(vec![catalog.to_string()]) + } + + fn list_tables(&self, _catalog: &str, schema: &str) -> Result> { + let df = self.query_arrow(&format!( + "SELECT name, \ + CASE WHEN engine IN ('View', 'MaterializedView', 'LiveView', 'WindowView') \ + THEN 'VIEW' ELSE 'BASE TABLE' END AS table_type \ + FROM system.tables WHERE database = {} ORDER BY name", + naming::quote_literal(schema) + ))?; + let names = column_strings(&df, "name")?; + let types = column_strings(&df, "table_type")?; + Ok(names + .into_iter() + .zip(types) + .map(|(name, table_type)| TableInfo { name, table_type }) + .collect()) + } + + fn list_columns(&self, _catalog: &str, schema: &str, table: &str) -> Result> { + let df = self.query_arrow(&format!( + "SELECT name, type FROM system.columns \ + WHERE database = {} AND table = {} ORDER BY position", + naming::quote_literal(schema), + naming::quote_literal(table) + ))?; + let names = column_strings(&df, "name")?; + let types = column_strings(&df, "type")?; + Ok(names + .into_iter() + .zip(types) + .map(|(name, data_type)| ColumnInfo { name, data_type }) + .collect()) + } +} + +// ============================================================================= +// Arrow helpers +// ============================================================================= + +/// Decode an `ArrowStream` payload into a DataFrame. An empty body (a +/// statement without a result set) yields an empty DataFrame; a stream with a +/// schema but no batches keeps its column names and types. +pub(crate) fn decode_arrow_stream(bytes: &[u8]) -> Result { + if bytes.is_empty() { + return Ok(DataFrame::empty()); + } + let reader = StreamReader::try_new(Cursor::new(bytes), None).map_err(|e| { + GgsqlError::ReaderError(format!("Failed to decode ClickHouse Arrow stream: {e}")) + })?; + let schema = reader.schema(); + let batches = reader + .collect::, _>>() + .map_err(|e| { + GgsqlError::ReaderError(format!("Failed to decode ClickHouse Arrow batch: {e}")) + })?; + let merged = match batches.len() { + 0 => RecordBatch::new_empty(schema), + 1 => batches.into_iter().next().unwrap(), + _ => arrow::compute::concat_batches(&schema, &batches) + .map_err(|e| GgsqlError::ReaderError(format!("concat_batches: {e}")))?, + }; + Ok(DataFrame::from_record_batch(normalize_timestamps(merged)?)) +} + +/// Cast every timestamp column to microseconds without a time zone, the one +/// timestamp representation the rest of the pipeline works with. ClickHouse +/// emits `DateTime64(p)` in the unit matching `p` and tags it with the server +/// time zone; the instant is preserved, and ggsql renders instants in UTC. +fn normalize_timestamps(batch: RecordBatch) -> Result { + use arrow::datatypes::{Field, Schema}; + use std::sync::Arc; + + let target = DataType::Timestamp(TimeUnit::Microsecond, None); + if !batch + .schema() + .fields() + .iter() + .any(|f| matches!(f.data_type(), DataType::Timestamp(_, _)) && f.data_type() != &target) + { + return Ok(batch); + } + let mut fields = Vec::with_capacity(batch.num_columns()); + let mut columns = Vec::with_capacity(batch.num_columns()); + for (field, column) in batch.schema().fields().iter().zip(batch.columns()) { + if matches!(field.data_type(), DataType::Timestamp(_, _)) && field.data_type() != &target { + let cast = arrow::compute::cast(column, &target).map_err(|e| { + GgsqlError::ReaderError(format!( + "Failed to normalize timestamp column '{}': {e}", + field.name() + )) + })?; + fields.push(Arc::new(Field::new( + field.name(), + target.clone(), + field.is_nullable(), + ))); + columns.push(cast); + } else { + fields.push(field.clone()); + columns.push(column.clone()); + } + } + RecordBatch::try_new(Arc::new(Schema::new(fields)), columns) + .map_err(|e| GgsqlError::ReaderError(format!("Failed to rebuild record batch: {e}"))) +} + +/// Encode a DataFrame as an Arrow IPC stream for `INSERT … FORMAT ArrowStream`. +fn encode_arrow_stream(df: DataFrame) -> Result> { + let batch = df.into_inner(); + let mut buf = Vec::new(); + let mut writer = StreamWriter::try_new(&mut buf, &batch.schema()) + .map_err(|e| GgsqlError::ReaderError(format!("Failed to encode Arrow stream: {e}")))?; + writer + .write(&batch) + .map_err(|e| GgsqlError::ReaderError(format!("Failed to encode Arrow batch: {e}")))?; + writer + .finish() + .map_err(|e| GgsqlError::ReaderError(format!("Failed to finish Arrow stream: {e}")))?; + drop(writer); + Ok(buf) +} + +/// The innermost type name of a ClickHouse type, with `Nullable(…)` and +/// `LowCardinality(…)` wrappers and any parameter list removed: +/// `LowCardinality(Nullable(FixedString(16)))` → `FixedString`. +fn base_type_name(ch_type: &str) -> &str { + let mut t = ch_type.trim(); + loop { + let inner = ["Nullable(", "LowCardinality("] + .iter() + .find_map(|w| t.strip_prefix(w)) + .map(|s| s.strip_suffix(')').unwrap_or(s).trim()); + match inner { + Some(s) => t = s, + None => break, + } + } + t.split('(').next().unwrap_or(t).trim() +} + +/// A `SELECT * REPLACE` item converting a column to a type ClickHouse can +/// export to Arrow faithfully, or `None` if the column needs no conversion. +fn arrow_friendly_expr(name: &str, ch_type: &str) -> Option { + let q = naming::quote_ident(name); + let expr = match base_type_name(ch_type) { + // Exported as a bare UInt32 otherwise. + "DateTime" => format!("toDateTime64({q}, 0)"), + // Exported as integer codes or raw bytes otherwise. + "Enum" | "Enum8" | "Enum16" | "UUID" | "IPv4" | "IPv6" | "Int128" | "UInt128" + | "Int256" | "UInt256" | "Time" | "Time64" => format!("toString({q})"), + // Nested containers have no visual encoding; show their text form. + "Array" | "Tuple" | "Map" | "Nested" | "Variant" | "Dynamic" | "JSON" | "Object" + | "Point" | "Ring" | "LineString" | "MultiLineString" | "Polygon" | "MultiPolygon" => { + format!("toString({q})") + } + "FixedString" => format!("toStringCutToZero({q})"), + "Decimal" | "Decimal32" | "Decimal64" | "Decimal128" | "Decimal256" => { + format!("toFloat64({q})") + } + // The type of a bare NULL literal. + "Nothing" => format!("CAST({q} AS Nullable(String))"), + _ => return None, + }; + Some(format!("{expr} AS {q}")) +} + +/// ClickHouse column type for an Arrow field, used when creating the +/// temporary table that receives a registered DataFrame. +fn arrow_type_to_clickhouse(dtype: &DataType, nullable: bool) -> String { + let base: String = match dtype { + DataType::Boolean => "Bool".into(), + DataType::Int8 => "Int8".into(), + DataType::Int16 => "Int16".into(), + DataType::Int32 => "Int32".into(), + DataType::Int64 => "Int64".into(), + DataType::UInt8 => "UInt8".into(), + DataType::UInt16 => "UInt16".into(), + DataType::UInt32 => "UInt32".into(), + DataType::UInt64 => "UInt64".into(), + DataType::Float16 | DataType::Float32 => "Float32".into(), + DataType::Float64 => "Float64".into(), + DataType::Utf8 + | DataType::LargeUtf8 + | DataType::Utf8View + | DataType::Binary + | DataType::LargeBinary + | DataType::BinaryView + | DataType::FixedSizeBinary(_) => "String".into(), + DataType::Date32 => "Date32".into(), + DataType::Date64 => "DateTime64(3)".into(), + DataType::Timestamp(unit, _) => { + let precision = match unit { + TimeUnit::Second => 0, + TimeUnit::Millisecond => 3, + TimeUnit::Microsecond => 6, + TimeUnit::Nanosecond => 9, + }; + format!("DateTime64({precision})") + } + DataType::Decimal128(p, s) | DataType::Decimal256(p, s) => format!("Decimal({p}, {s})"), + DataType::Dictionary(_, value) => return arrow_type_to_clickhouse(value, nullable), + // Containers cannot be Nullable in ClickHouse; nullability lives on + // the element type. + DataType::List(f) | DataType::LargeList(f) | DataType::FixedSizeList(f, _) => { + return format!( + "Array({})", + arrow_type_to_clickhouse(f.data_type(), f.is_nullable()) + ) + } + DataType::Struct(fields) => { + let items: Vec = fields + .iter() + .map(|f| { + format!( + "{} {}", + naming::quote_ident(f.name()), + arrow_type_to_clickhouse(f.data_type(), f.is_nullable()) + ) + }) + .collect(); + return format!("Tuple({})", items.join(", ")); + } + DataType::Null => return "Nullable(String)".into(), + // Durations and times of day have no stable Arrow mapping in + // ClickHouse; store the raw integer. + DataType::Time32(_) | DataType::Time64(_) | DataType::Duration(_) => "Int64".into(), + _ => "String".into(), + }; + if nullable { + format!("Nullable({base})") + } else { + base + } +} + +/// Decode `%XX` escapes in a connection-string component; anything else is +/// passed through unchanged. +pub(crate) fn percent_decode(s: &str) -> String { + let bytes = s.as_bytes(); + let mut out = Vec::with_capacity(bytes.len()); + let mut i = 0; + while i < bytes.len() { + if bytes[i] == b'%' && i + 2 < bytes.len() { + let hex = std::str::from_utf8(&bytes[i + 1..i + 3]).ok(); + if let Some(v) = hex.and_then(|h| u8::from_str_radix(h, 16).ok()) { + out.push(v); + i += 3; + continue; + } + } + out.push(bytes[i]); + i += 1; + } + String::from_utf8_lossy(&out).into_owned() +} + +/// Non-null values of a string column, in row order. +fn column_strings(df: &DataFrame, column: &str) -> Result> { + let col = df.column(column)?; + Ok((0..df.height()) + .filter(|&i| !col.is_null(i)) + .map(|i| value_to_string(col, i)) + .collect()) +} + +// ============================================================================= +// Tests +// ============================================================================= + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_base_type_name() { + assert_eq!(base_type_name("DateTime"), "DateTime"); + assert_eq!(base_type_name("DateTime('UTC')"), "DateTime"); + assert_eq!(base_type_name("DateTime64(3)"), "DateTime64"); + assert_eq!(base_type_name("Nullable(DateTime)"), "DateTime"); + assert_eq!( + base_type_name("LowCardinality(Nullable(FixedString(16)))"), + "FixedString" + ); + assert_eq!(base_type_name("Enum8('a' = 1, 'b' = 2)"), "Enum8"); + assert_eq!(base_type_name("Array(Nullable(String))"), "Array"); + } + + #[test] + fn test_arrow_friendly_expr() { + assert_eq!( + arrow_friendly_expr("ts", "Nullable(DateTime)").as_deref(), + Some("toDateTime64(\"ts\", 0) AS \"ts\"") + ); + assert_eq!( + arrow_friendly_expr("e", "Enum8('a' = 1)").as_deref(), + Some("toString(\"e\") AS \"e\"") + ); + assert_eq!( + arrow_friendly_expr("d", "Decimal(18, 2)").as_deref(), + Some("toFloat64(\"d\") AS \"d\"") + ); + assert_eq!( + arrow_friendly_expr("f", "FixedString(4)").as_deref(), + Some("toStringCutToZero(\"f\") AS \"f\"") + ); + assert_eq!(arrow_friendly_expr("x", "UInt64"), None); + assert_eq!(arrow_friendly_expr("x", "DateTime64(6, 'UTC')"), None); + assert_eq!(arrow_friendly_expr("x", "LowCardinality(String)"), None); + assert_eq!(arrow_friendly_expr("x", "Date"), None); + assert_eq!(arrow_friendly_expr("x", "Bool"), None); + } + + #[test] + fn test_arrow_type_to_clickhouse() { + assert_eq!(arrow_type_to_clickhouse(&DataType::Int64, false), "Int64"); + assert_eq!( + arrow_type_to_clickhouse(&DataType::Float64, true), + "Nullable(Float64)" + ); + assert_eq!( + arrow_type_to_clickhouse(&DataType::Utf8, true), + "Nullable(String)" + ); + assert_eq!(arrow_type_to_clickhouse(&DataType::Date32, false), "Date32"); + assert_eq!( + arrow_type_to_clickhouse(&DataType::Timestamp(TimeUnit::Microsecond, None), true), + "Nullable(DateTime64(6))" + ); + assert_eq!(arrow_type_to_clickhouse(&DataType::Boolean, false), "Bool"); + let list = DataType::List(std::sync::Arc::new(arrow::datatypes::Field::new( + "item", + DataType::Int32, + true, + ))); + assert_eq!( + arrow_type_to_clickhouse(&list, true), + "Array(Nullable(Int32))", + "arrays are never Nullable themselves" + ); + } + + #[test] + fn test_percent_decode() { + assert_eq!(percent_decode("a%20b%2Fc"), "a b/c"); + assert_eq!(percent_decode("plain+text"), "plain+text"); + assert_eq!(percent_decode("bad%zz%"), "bad%zz%"); + } + + #[test] + fn test_dialect_temp_table_sql() { + let stmts = ClickHouseDialect.create_or_replace_temp_table_sql( + "__ggsql_t__", + &["a".to_string(), "b".to_string()], + "SELECT 1, 2", + ); + assert_eq!(stmts.len(), 2); + assert_eq!(stmts[0], "DROP TEMPORARY TABLE IF EXISTS \"__ggsql_t__\""); + assert!(stmts[1].starts_with("CREATE TEMPORARY TABLE \"__ggsql_t__\" AS WITH")); + assert!(stmts[1].contains("\"a\", \"b\"")); + } + + #[test] + fn test_dialect_quantile_and_series() { + assert_eq!( + ClickHouseDialect.sql_quantile_inline("v", 0.25).unwrap(), + "quantileExactInclusive(0.25)(\"v\")" + ); + assert_eq!( + ClickHouseDialect.sql_generate_series(5), + "\"__ggsql_seq__\"(n) AS (SELECT toFloat64(number) AS n FROM numbers(5))" + ); + assert_eq!( + ClickHouseDialect.sql_greatest(&["a", "b"]), + "greatest(toFloat64(a), toFloat64(b))" + ); + assert_eq!(ClickHouseDialect.sql_least(&["a"]), "a"); + assert_eq!( + ClickHouseDialect.sql_percentile("v", 0.5, "ignored", &["g".to_string()]), + "quantileExactInclusive(0.5)(\"v\")" + ); + assert_eq!( + ClickHouseDialect.sql_null_safe_equals("a.k", "b.k"), + "((a.k = b.k) OR (a.k IS NULL AND b.k IS NULL))" + ); + assert_eq!(ClickHouseDialect.sql_date_literal(-1), "toDate32(-1)"); + assert_eq!( + ClickHouseDialect.sql_datetime_literal(1_000_000), + "fromUnixTimestamp64Micro(1000000)" + ); + assert_eq!(ClickHouseDialect.time_type_name(), None); + } + + #[test] + fn test_dialect_cache_meta_sql() { + let create = ClickHouseDialect.cache_meta_table_sql("__ggsql_cache_meta__"); + assert!(create.starts_with("CREATE TABLE IF NOT EXISTS \"__ggsql_cache_meta__\"")); + assert!(create.ends_with("ENGINE = Memory")); + + let upsert = ClickHouseDialect.cache_meta_upsert_sql( + "m", + "k1", + "SELECT 'it''s \\ done'", + "__ggsql_cache_k1__", + 10, + 20, + 30, + ); + assert_eq!(upsert.len(), 2); + assert_eq!(upsert[0], "ALTER TABLE \"m\" DELETE WHERE cache_key = 'k1'"); + assert!(upsert[1].starts_with("INSERT INTO \"m\" ")); + assert!( + upsert[1].contains("'SELECT ''it''''s \\\\ done'''"), + "quotes doubled and backslashes escaped: {}", + upsert[1] + ); + assert!(upsert[1].ends_with( + "VALUES ('k1', 'SELECT ''it''''s \\\\ done''', '__ggsql_cache_k1__', 10, 10, 20, 30)" + )); + + assert_eq!( + ClickHouseDialect.cache_meta_touch_sql("m", "k1", 99), + "ALTER TABLE \"m\" UPDATE last_accessed_epoch_ms = 99 WHERE cache_key = 'k1'" + ); + assert_eq!( + ClickHouseDialect.cache_meta_delete_sql("m", "k1"), + "ALTER TABLE \"m\" DELETE WHERE cache_key = 'k1'" + ); + } +} + +/// Behavioural tests shared by every transport. Each transport's test module +/// calls these with a connected reader; they use table names unique to the +/// caller so transports sharing one engine (chDB has a single in-process +/// connection) do not collide. +#[cfg(test)] +pub(crate) mod live_tests { + use super::*; + + pub(crate) fn basic_types(reader: &ClickHouseSqlReader) { + let df = reader + .execute_sql( + "SELECT toDateTime('2024-01-02 03:04:05', 'UTC') AS dt, \ + toDate('2024-01-02') AS d, \ + 'x' AS s, toLowCardinality('lc') AS lc, \ + 1 AS u8, toInt64(-5) AS i64, 1.5 AS f, true AS b, \ + CAST(NULL AS Nullable(Int32)) AS n, \ + CAST('a' AS Enum8('a' = 1, 'b' = 2)) AS e, \ + toUUID('61f0c404-5cb3-11e7-907b-a6006ad3dba0') AS uid, \ + toIPv4('1.2.3.4') AS ip, toDecimal64(2.5, 2) AS dec, \ + toFixedString('ab', 4) AS fs, toInt128(7) AS big, \ + [1, 2] AS arr, toDateTime64('2024-01-02 03:04:05.123', 3, 'UTC') AS dt64", + ) + .unwrap(); + assert_eq!(df.height(), 1); + // Every timestamp arrives normalized to naive microseconds. + let us = DataType::Timestamp(TimeUnit::Microsecond, None); + assert_eq!(df.column_dtype("dt").unwrap(), us); + assert_eq!(df.column_dtype("dt64").unwrap(), us); + let dt64 = crate::array_util::as_timestamp_us(df.column("dt64").unwrap()).unwrap(); + assert_eq!(dt64.value(0), 1_704_164_645_123_000); + assert_eq!(df.column_dtype("d").unwrap(), DataType::Date32); + assert_eq!(df.column_dtype("s").unwrap(), DataType::Utf8); + assert_eq!(df.column_dtype("lc").unwrap(), DataType::Utf8); + assert_eq!(df.column_dtype("u8").unwrap(), DataType::UInt8); + assert_eq!(df.column_dtype("i64").unwrap(), DataType::Int64); + assert_eq!(df.column_dtype("f").unwrap(), DataType::Float64); + assert_eq!(df.column_dtype("b").unwrap(), DataType::Boolean); + assert_eq!(df.column_dtype("n").unwrap(), DataType::Int32); + assert!(df.column("n").unwrap().is_null(0)); + assert_eq!(df.column_dtype("e").unwrap(), DataType::Utf8); + assert_eq!(value_to_string(df.column("e").unwrap(), 0), "a"); + assert_eq!( + value_to_string(df.column("uid").unwrap(), 0), + "61f0c404-5cb3-11e7-907b-a6006ad3dba0" + ); + assert_eq!(value_to_string(df.column("ip").unwrap(), 0), "1.2.3.4"); + assert_eq!(df.column_dtype("dec").unwrap(), DataType::Float64); + assert_eq!(value_to_string(df.column("fs").unwrap(), 0), "ab"); + assert_eq!(value_to_string(df.column("big").unwrap(), 0), "7"); + assert_eq!(value_to_string(df.column("arr").unwrap(), 0), "[1,2]"); + + // The DateTime survived as an instant: 2024-01-02T03:04:05Z. + let ts = arrow::compute::cast( + df.column("dt").unwrap(), + &DataType::Timestamp(TimeUnit::Microsecond, None), + ) + .unwrap(); + let ts = crate::array_util::as_timestamp_us(&ts).unwrap(); + assert_eq!(ts.value(0), 1_704_164_645_000_000); + } + + pub(crate) fn empty_result_keeps_schema(reader: &ClickHouseSqlReader) { + let df = reader + .execute_sql("SELECT number AS x, toString(number) AS s FROM numbers(3) WHERE 0") + .unwrap(); + assert_eq!(df.height(), 0); + assert_eq!(df.get_column_names(), vec!["x", "s"]); + } + + pub(crate) fn ddl_and_errors(reader: &ClickHouseSqlReader) { + assert_eq!( + reader.execute_sql("SET max_threads = 2").unwrap().height(), + 0 + ); + let df = reader + .execute_sql("SELECT getSetting('max_threads') AS v") + .unwrap(); + assert_eq!(value_to_string(df.column("v").unwrap(), 0), "2"); + + let err = reader + .execute_sql("SELECT * FROM __ggsql_no_such_table__") + .unwrap_err() + .to_string(); + assert!( + err.contains("UNKNOWN_TABLE") || err.contains("does not exist"), + "{err}" + ); + + let err = reader.execute_sql("SELEC 1").unwrap_err().to_string(); + assert!(err.contains("Syntax error"), "{err}"); + } + + pub(crate) fn register_roundtrip(reader: &ClickHouseSqlReader, table: &str) { + let q = naming::quote_ident(table); + let df = crate::df! { + "x" => vec![1_i64, 2, 3], + "label" => vec!["a", "b", "c"], + "v" => vec![1.5_f64, 2.5, 3.5], + } + .unwrap(); + reader.register(table, df, true).unwrap(); + let back = reader + .execute_sql(&format!("SELECT x, label, v FROM {q} ORDER BY x")) + .unwrap(); + assert_eq!(back.height(), 3); + assert_eq!(back.column_dtype("x").unwrap(), DataType::Int64); + assert_eq!(back.column_dtype("label").unwrap(), DataType::Utf8); + assert_eq!(back.column_dtype("v").unwrap(), DataType::Float64); + assert_eq!(value_to_string(back.column("label").unwrap(), 2), "c"); + + // Without replace the second registration is refused. + let df2 = crate::df! { "x" => vec![9_i64] }.unwrap(); + assert!(reader.register(table, df2, false).is_err()); + + reader.unregister(table).unwrap(); + assert!(reader.execute_sql(&format!("SELECT * FROM {q}")).is_err()); + assert!(reader.unregister(table).is_err()); + } + + pub(crate) fn temp_tables_persist(reader: &ClickHouseSqlReader, table: &str) { + let q = naming::quote_ident(table); + assert!(reader.supports_temporary_tables()); + reader + .materialize_table(table, &[], "SELECT number AS n FROM numbers(4)") + .unwrap(); + let df = reader + .execute_sql(&format!("SELECT count() AS c FROM {q}")) + .unwrap(); + assert_eq!(value_to_string(df.column("c").unwrap(), 0), "4"); + // Re-materializing replaces rather than failing on "already exists". + reader + .materialize_table(table, &["m".to_string()], "SELECT 1") + .unwrap(); + let df = reader.execute_sql(&format!("SELECT m FROM {q}")).unwrap(); + assert_eq!(df.height(), 1); + reader + .execute_sql(&format!("DROP TEMPORARY TABLE {q}")) + .unwrap(); + } + + pub(crate) fn schema_introspection(reader: &ClickHouseSqlReader) { + let catalogs = reader.list_catalogs().unwrap(); + assert!(catalogs.iter().any(|c| c == "default"), "{catalogs:?}"); + assert!(!catalogs.iter().any(|c| c == "system")); + assert_eq!(reader.list_schemas("default").unwrap(), vec!["default"]); + let tables = reader.list_tables("system", "system").unwrap(); + assert!( + tables.iter().any(|t| t.name == "numbers"), + "system.numbers missing" + ); + let cols = reader.list_columns("system", "system", "numbers").unwrap(); + assert_eq!(cols.len(), 1); + assert_eq!(cols[0].name, "number"); + assert_eq!(cols[0].data_type, "UInt64"); + } + + #[cfg(feature = "vegalite")] + pub(crate) fn execute_pipeline(reader: &ClickHouseSqlReader) { + use crate::writer::{VegaLiteWriter, Writer}; + + // Scatter with a DateTime axis and an Enum colour: both go through the + // type rewrite and a scale-driven cast. The Enum column lives in a + // temp table because the ggsql grammar does not parse `Enum8('a' = 1)` + // type arguments inside a query. + let table = naming::quote_ident(&format!("__ggsql_enum_{}__", naming::session_id())); + reader + .execute_sql(&format!( + "CREATE TEMPORARY TABLE {table} \ + (day DateTime('UTC'), y UInt64, parity Enum8('even' = 1, 'odd' = 2))" + )) + .unwrap(); + reader + .execute_sql(&format!( + "INSERT INTO {table} SELECT toDateTime('2024-01-01 00:00:00', 'UTC') + number * 86400, \ + number * number, if(number % 2 = 0, 'even', 'odd') FROM numbers(10)" + )) + .unwrap(); + let spec = reader + .execute(&format!( + "SELECT day, y, parity FROM {table} \ + VISUALISE day AS x, y AS y, parity AS color DRAW point" + )) + .unwrap(); + reader + .execute_sql(&format!("DROP TEMPORARY TABLE {table}")) + .unwrap(); + assert_eq!(spec.metadata().rows, 10); + let json = VegaLiteWriter::new().render(&spec).unwrap(); + assert!( + json.contains("\"temporal\""), + "date axis should be temporal" + ); + assert!(json.contains("even"), "enum labels should survive: {json}"); + + // Stat transforms run on the engine: histogram binning and a boxplot + // (quantiles) over a global temp table. + let spec = reader + .execute( + "SELECT number % 7 AS g, sin(number) * 10 AS v FROM numbers(200) \ + VISUALISE v AS x DRAW histogram SETTING bins => 10", + ) + .unwrap(); + assert!(spec.layer_data(0).unwrap().height() > 1); + + let spec = reader + .execute( + "SELECT toString(number % 3) AS g, number AS v FROM numbers(30) \ + VISUALISE g AS x, v AS y DRAW boxplot", + ) + .unwrap(); + // Three groups, each summarized as box, median and two whiskers. + assert_eq!(spec.layer_data(0).unwrap().height(), 12); + + // Bar with count stat plus a CTE that is materialized as a temp table. + let spec = reader + .execute( + "WITH t AS (SELECT number % 4 AS k FROM numbers(40)) \ + SELECT toString(k) AS k FROM t \ + VISUALISE k AS x DRAW bar", + ) + .unwrap(); + assert_eq!(spec.layer_data(0).unwrap().height(), 4); + } + + #[cfg(all(feature = "builtin-data", feature = "parquet"))] + pub(crate) fn builtin_dataset(reader: &ClickHouseSqlReader) { + let df = reader + .execute_sql("SELECT count() AS c FROM ggsql:penguins") + .unwrap(); + assert_eq!(value_to_string(df.column("c").unwrap(), 0), "344"); + // Second reference reuses the session's temp table. + let df = reader + .execute_sql( + "SELECT species, count() AS c FROM ggsql:penguins GROUP BY species ORDER BY species", + ) + .unwrap(); + assert_eq!(df.height(), 3); + } +} diff --git a/src/reader/connection.rs b/src/reader/connection.rs index 552257680..961d4fe8c 100644 --- a/src/reader/connection.rs +++ b/src/reader/connection.rs @@ -1,7 +1,8 @@ //! Connection string handling for data sources. //! -//! Maps URI-style connection strings (`duckdb://…`, `sqlite://…`, `odbc://…`) and -//! the composite caching form (`+://…`) to readers. +//! Maps URI-style connection strings (`duckdb://…`, `sqlite://…`, `odbc://…`, +//! `clickhouse://…`) and the composite caching form (`+://…`) +//! to readers. use crate::reader::Reader; use crate::{GgsqlError, Result}; @@ -34,12 +35,12 @@ pub fn split_cache_uri(uri: &str) -> Option<(String, String)> { } /// Cache-config keys recognised in a connection URI's trailing `?` query string. -#[cfg(any(feature = "duckdb", feature = "sqlite"))] +#[cfg(any(feature = "duckdb", feature = "sqlite", feature = "chdb"))] const KNOWN_CACHE_PARAMS: &[&str] = &["cache_ttl", "cache_max_bytes", "cache_disabled"]; /// Pull cache-config keys out of a connection URI's trailing `?key=value&…` /// query string, returning the URI with those keys removed plus the overrides. -#[cfg(any(feature = "duckdb", feature = "sqlite"))] +#[cfg(any(feature = "duckdb", feature = "sqlite", feature = "chdb"))] fn strip_cache_params(uri: &str) -> (String, crate::reader::cache::CacheConfigOverride) { use crate::reader::cache::{parse_human_bytes, CacheConfigOverride}; @@ -73,13 +74,14 @@ fn strip_cache_params(uri: &str) -> (String, crate::reader::cache::CacheConfigOv } /// Map a cache-backend scheme to its in-memory connection URI. -#[cfg(any(feature = "duckdb", feature = "sqlite"))] +#[cfg(any(feature = "duckdb", feature = "sqlite", feature = "chdb"))] fn cache_uri(scheme: &str) -> Result<&'static str> { match scheme { "duckdb" => Ok("duckdb://memory"), "sqlite" => Ok("sqlite://memory"), + "chdb" => Ok("chdb://memory"), _ => Err(GgsqlError::ReaderError(format!( - "Unsupported cache backend '{}'. Supported: duckdb, sqlite", + "Unsupported cache backend '{}'. Supported: duckdb, sqlite, chdb", scheme ))), } @@ -129,13 +131,41 @@ pub fn build_reader(uri: &str) -> Result> { )); } } + if uri.starts_with("clickhouse://") || uri.starts_with("clickhouses://") { + #[cfg(feature = "clickhouse")] + { + return Ok(Box::new( + crate::reader::ClickHouseReader::from_connection_string(uri)?, + )); + } + #[cfg(not(feature = "clickhouse"))] + { + return Err(GgsqlError::ReaderError( + "ClickHouse reader not compiled in. Rebuild with --features clickhouse".to_string(), + )); + } + } + if uri.starts_with("chdb://") { + #[cfg(feature = "chdb")] + { + return Ok(Box::new(crate::reader::ChdbReader::from_connection_string( + uri, + )?)); + } + #[cfg(not(feature = "chdb"))] + { + return Err(GgsqlError::ReaderError( + "chDB reader not compiled in. Rebuild with --features chdb".to_string(), + )); + } + } if uri.starts_with("postgres://") || uri.starts_with("postgresql://") { return Err(GgsqlError::ReaderError( "PostgreSQL reader is not yet implemented".to_string(), )); } Err(GgsqlError::ReaderError(format!( - "Unsupported connection string: {}. Supported: duckdb://, sqlite://, odbc://", + "Unsupported connection string: {}. Supported: duckdb://, sqlite://, odbc://, clickhouse://, chdb://", uri ))) } @@ -146,33 +176,68 @@ pub fn build_reader(uri: &str) -> Result> { /// [`CachingReader`]: crate::reader::CachingReader pub fn reader_from_uri(uri: &str) -> Result> { if let Some((primary_uri, cache_scheme)) = split_cache_uri(uri) { - #[cfg(any(feature = "duckdb", feature = "sqlite"))] + #[cfg(any(feature = "duckdb", feature = "sqlite", feature = "chdb"))] { - use crate::reader::cache::CacheConfig; - let (primary_uri, over) = strip_cache_params(&primary_uri); - let config = CacheConfig::from_env().merge(over); let primary = build_reader(&primary_uri)?; - let cache = build_reader(cache_uri(&cache_scheme)?)?; - return Ok(Box::new(crate::reader::CachingReader::with_config( - primary, - cache, - primary_uri, - cache_scheme, - config, - ))); + return wrap_in_cache(primary, primary_uri, &cache_scheme, over); } - #[cfg(not(any(feature = "duckdb", feature = "sqlite")))] + #[cfg(not(any(feature = "duckdb", feature = "sqlite", feature = "chdb")))] { let _ = (&primary_uri, &cache_scheme); return Err(GgsqlError::ReaderError( - "Caching layer requires the duckdb or sqlite feature".to_string(), + "Caching layer requires the duckdb, sqlite or chdb feature".to_string(), )); } } + + // A ClickHouse account that may not create temporary tables (read-only + // servers such as play.clickhouse.com) cannot hold the executor's + // intermediate tables. Keep them in an embedded chDB engine instead, so + // the plain `clickhouse://` URI works there too and only the user's own + // reads reach the server. + #[cfg(all(feature = "clickhouse", feature = "chdb"))] + if uri.starts_with("clickhouse://") || uri.starts_with("clickhouses://") { + let (primary_uri, over) = strip_cache_params(uri); + let reader = crate::reader::ClickHouseReader::from_connection_string(&primary_uri)?; + if reader.supports_temporary_tables() { + return Ok(Box::new(reader)); + } + return wrap_in_cache(Box::new(reader), primary_uri, "chdb", over).map_err(|e| { + GgsqlError::ReaderError(format!( + "This ClickHouse account cannot create temporary tables, so ggsql needs the \ + embedded chDB engine to hold intermediate results, but it could not be set up: {e}" + )) + }); + } + build_reader(uri) } +/// Wrap `primary` in a [`CachingReader`] on a fresh in-memory `cache_scheme` +/// backend, applying cache settings from the environment and `over`. +/// +/// [`CachingReader`]: crate::reader::CachingReader +#[cfg(any(feature = "duckdb", feature = "sqlite", feature = "chdb"))] +fn wrap_in_cache( + primary: Box, + primary_uri: String, + cache_scheme: &str, + over: crate::reader::cache::CacheConfigOverride, +) -> Result> { + use crate::reader::cache::CacheConfig; + + let config = CacheConfig::from_env().merge(over); + let cache = build_reader(cache_uri(cache_scheme)?)?; + Ok(Box::new(crate::reader::CachingReader::with_config( + primary, + cache, + primary_uri, + cache_scheme.to_string(), + config, + ))) +} + /// Extract a value from an ODBC connection string by key, stripping braces. pub fn extract_odbc_value(conn_str: &str, key: &str) -> Option { let lower = conn_str.to_lowercase(); @@ -210,6 +275,26 @@ mod tests { assert!(err.contains("not yet implemented"), "got: {err}"); } + #[cfg(feature = "clickhouse")] + #[test] + fn test_build_reader_clickhouse_dispatch() { + // A malformed URI is rejected by the ClickHouse parser, not the + // generic "unsupported scheme" path. + let err = build_reader("clickhouse://host:notaport") + .err() + .unwrap() + .to_string(); + assert!(err.contains("Invalid port"), "got: {err}"); + // A composite cache URI routes the primary to ClickHouse. + assert_eq!( + split_cache_uri("duckdb+clickhouses://explorer@play.clickhouse.com:443"), + Some(( + "clickhouses://explorer@play.clickhouse.com:443".to_string(), + "duckdb".to_string() + )) + ); + } + #[cfg(feature = "duckdb")] #[test] fn test_build_reader_duckdb_memory_and_empty() { @@ -268,7 +353,7 @@ mod tests { assert_eq!(split_cache_uri("odbc+://x"), None); } - #[cfg(any(feature = "duckdb", feature = "sqlite"))] + #[cfg(any(feature = "duckdb", feature = "sqlite", feature = "chdb"))] #[test] fn test_strip_cache_params_parses_known_keys() { let (uri, over) = strip_cache_params("duckdb://memory?cache_ttl=600"); @@ -284,7 +369,7 @@ mod tests { assert_eq!(over.enabled, Some(false)); } - #[cfg(any(feature = "duckdb", feature = "sqlite"))] + #[cfg(any(feature = "duckdb", feature = "sqlite", feature = "chdb"))] #[test] fn test_strip_cache_params_keeps_non_cache_segments() { // A non-cache `?key=` tail contributes no overrides and is left in place. diff --git a/src/reader/mod.rs b/src/reader/mod.rs index aaa961a83..cb23c8c35 100644 --- a/src/reader/mod.rs +++ b/src/reader/mod.rs @@ -94,6 +94,13 @@ pub trait SqlDialect { } } + /// Null-safe equality of two expressions: true when both are equal or + /// both NULL. Override for backends that restrict `IS NOT DISTINCT FROM` + /// (ClickHouse only accepts it in `JOIN ON`). + fn sql_null_safe_equals(&self, left: &str, right: &str) -> String { + format!("{left} IS NOT DISTINCT FROM {right}") + } + /// Scalar MAX across any number of SQL expressions. fn sql_greatest(&self, exprs: &[&str]) -> String { let mut result = exprs[0].to_string(); @@ -356,6 +363,71 @@ pub trait SqlDialect { format!("CREATE TEMP TABLE {} AS {}", qname, body), ] } + + // ------------------------------------------------------------------------- + // Caching-layer memo table + // + // `CachingReader` keeps one row per memoized read in a table on the cache + // backend. Backends without `INSERT OR REPLACE` / `UPDATE` / `DELETE FROM` + // (ClickHouse) override these to spell the same operations natively. + // ------------------------------------------------------------------------- + + /// DDL creating the memo table `table` if it does not exist. Columns: + /// `cache_key` (text, unique), `sql`, `table_name` (text), + /// `fetched_at_epoch_ms`, `last_accessed_epoch_ms`, `byte_estimate`, + /// `row_count` (64-bit integers). + fn cache_meta_table_sql(&self, table: &str) -> String { + format!( + "CREATE TABLE IF NOT EXISTS {} (\ + cache_key VARCHAR PRIMARY KEY, sql VARCHAR NOT NULL, table_name VARCHAR NOT NULL, \ + fetched_at_epoch_ms BIGINT NOT NULL, last_accessed_epoch_ms BIGINT NOT NULL, \ + byte_estimate BIGINT NOT NULL, row_count BIGINT NOT NULL)", + naming::quote_ident(table) + ) + } + + /// Statements that insert the memo row for `key`, replacing any existing + /// one. Both timestamps are set to `now_ms`. + #[allow(clippy::too_many_arguments)] + fn cache_meta_upsert_sql( + &self, + table: &str, + key: &str, + sql: &str, + table_name: &str, + now_ms: i64, + byte_estimate: i64, + row_count: i64, + ) -> Vec { + vec![format!( + "INSERT OR REPLACE INTO {} \ + (cache_key, sql, table_name, fetched_at_epoch_ms, last_accessed_epoch_ms, \ + byte_estimate, row_count) \ + VALUES ({}, {}, {}, {now_ms}, {now_ms}, {byte_estimate}, {row_count})", + naming::quote_ident(table), + naming::quote_literal(key), + naming::quote_literal(sql), + naming::quote_literal(table_name), + )] + } + + /// Statement advancing `last_accessed_epoch_ms` of the memo row for `key`. + fn cache_meta_touch_sql(&self, table: &str, key: &str, now_ms: i64) -> String { + format!( + "UPDATE {} SET last_accessed_epoch_ms = {now_ms} WHERE cache_key = {}", + naming::quote_ident(table), + naming::quote_literal(key), + ) + } + + /// Statement deleting the memo row for `key`. + fn cache_meta_delete_sql(&self, table: &str, key: &str) -> String { + format!( + "DELETE FROM {} WHERE cache_key = {}", + naming::quote_ident(table), + naming::quote_literal(key), + ) + } } /// Wrap a body SQL in a CTE with a column alias list when aliases are present. @@ -429,7 +501,10 @@ pub mod odbc; #[cfg(feature = "adbc")] pub mod adbc; -#[cfg(any(feature = "duckdb", feature = "sqlite"))] +#[cfg(any(feature = "clickhouse", feature = "chdb"))] +pub mod clickhouse; + +#[cfg(any(feature = "duckdb", feature = "sqlite", feature = "chdb"))] pub mod cache; #[cfg(all(test, feature = "duckdb", feature = "sqlite"))] @@ -451,7 +526,13 @@ pub use odbc::OdbcReader; #[cfg(feature = "adbc")] pub use adbc::AdbcReader; -#[cfg(any(feature = "duckdb", feature = "sqlite"))] +#[cfg(feature = "clickhouse")] +pub use clickhouse::ClickHouseReader; + +#[cfg(feature = "chdb")] +pub use clickhouse::ChdbReader; + +#[cfg(any(feature = "duckdb", feature = "sqlite", feature = "chdb"))] pub use cache::CachingReader; // ============================================================================ diff --git a/src/writer/vegalite/data.rs b/src/writer/vegalite/data.rs index 3953b68c8..be4bb2052 100644 --- a/src/writer/vegalite/data.rs +++ b/src/writer/vegalite/data.rs @@ -81,20 +81,14 @@ pub(super) fn series_value_at(array: &ArrayRef, idx: usize) -> Result { let date = unix_epoch + chrono::Duration::days(days as i64); Ok(json!(date.format("%Y-%m-%d").to_string())) } - DataType::Timestamp(time_unit, _) => { - // Convert timestamp to ISO datetime: "YYYY-MM-DDTHH:MM:SS.sssZ" - let timestamp = as_timestamp_us(array).map(|a| a.value(idx)).or_else(|_| { - // Try casting to microsecond timestamp first + DataType::Timestamp(_, _) => { + // Convert timestamp to ISO datetime: "YYYY-MM-DDTHH:MM:SS.sssZ". + // Arrays in another unit (or with a time zone) are cast to + // microseconds first, so `micros` is always in microseconds. + let micros = as_timestamp_us(array).map(|a| a.value(idx)).or_else(|_| { let cast = cast_array(array, &DataType::Timestamp(TimeUnit::Microsecond, None))?; Ok(as_timestamp_us(&cast)?.value(idx)) })?; - // timestamp is in microseconds for TimestampMicrosecondArray - let micros = match time_unit { - TimeUnit::Microsecond => timestamp, - TimeUnit::Millisecond => timestamp * 1_000, - TimeUnit::Nanosecond => timestamp / 1_000, - TimeUnit::Second => timestamp * 1_000_000, - }; let secs = micros / 1_000_000; let nsecs = ((micros % 1_000_000) * 1000) as u32; let dt = chrono::DateTime::::from_timestamp(secs, nsecs) From 536f3b5b81445efdd7a1ad485897f19501100433 Mon Sep 17 00:00:00 2001 From: Alexey Milovidov Date: Sat, 5 Sep 2026 22:06:48 +0200 Subject: [PATCH 2/2] Target ClickHouse 26.8+; drop old-version shims Remove the `sql_null_safe_equals` dialect hook and the dialect threading it required in the density stat SQL: ClickHouse 26.8+ accepts `IS NOT DISTINCT FROM` in any clause and supports correlated subqueries, so no compatibility spelling is needed. Document the version requirement. Co-Authored-By: Claude Fable 5.1 --- CHANGELOG.md | 2 +- doc/get_started/tooling/cli.qmd | 2 +- src/CLAUDE.md | 2 +- src/plot/layer/geom/density.rs | 61 +++++---------------------------- src/reader/clickhouse/mod.rs | 16 +++------ src/reader/mod.rs | 7 ---- 6 files changed, 16 insertions(+), 74 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e1bc567e0..5d5bd742f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -18,7 +18,7 @@ cache automatically. `DateTime`, `Enum`, `UUID`, IP address, `FixedString`, `Decimal` and wide-integer columns are converted server-side to types the plot pipeline understands. The Jupyter kernel and Positron connections pane - recognise both schemes. + recognise both schemes. ClickHouse 26.8 or newer is required. - New caching layer that wraps any `Reader` with an in-memory, writeable cache backend (currently duckdb or sqlite), making write-constrained databases usable and avoiding repeated remote reads during interactive iteration. diff --git a/doc/get_started/tooling/cli.qmd b/doc/get_started/tooling/cli.qmd index 49cd8b423..bab2a5cb3 100644 --- a/doc/get_started/tooling/cli.qmd +++ b/doc/get_started/tooling/cli.qmd @@ -67,7 +67,7 @@ col_a, col_b, col_c ### ClickHouse -ggsql connects to [ClickHouse](https://clickhouse.com) over its HTTP interface, so no driver needs to be installed. The connection string is +ggsql connects to [ClickHouse](https://clickhouse.com) 26.8 or newer over its HTTP interface, so no driver needs to be installed. The connection string is ``` clickhouse://[user[:password]@]host[:port][/database][?param=value&...] diff --git a/src/CLAUDE.md b/src/CLAUDE.md index f4db00616..725eace51 100644 --- a/src/CLAUDE.md +++ b/src/CLAUDE.md @@ -55,7 +55,7 @@ Grammar lives in [`/tree-sitter-ggsql/`](../tree-sitter-ggsql/) — when adding `SqlDialect` trait in `mod.rs` lets each driver supply its own type names, information-schema queries, and spatial helper methods (`sql_st_transform`, `sql_geometry_to_wkb`, `sql_geometry_bbox`, `sql_ensure_geometry`, `sql_select_replace`, `sql_spatial_setup`). -**ClickHouse readers.** `clickhouse/mod.rs` holds `ClickHouseSqlReader`, one `Reader` implementation shared by two transports: `http.rs` (`ClickHouseReader`, one `POST` per statement against a server, results as `FORMAT ArrowStream`, `register()` = `CREATE TEMPORARY TABLE` + `INSERT … FORMAT ArrowStream`, one `session_id` per reader so temp tables and `SET` persist) and `chdb.rs` (`ChdbReader`, libchdb loaded at runtime with `libloading` like the ODBC driver manager — search order `GGSQL_CHDB_LIBRARY`, system path, usual install dirs — so the `chdb` feature adds no build dependency; inserts go through a temp file and `FROM INFILE` because the C API has no input-data channel; libchdb allows **one connection per process**, shared by all `ChdbReader`s on the same path). Because ClickHouse's Arrow output loses some of its own types (`DateTime` → `UInt32`, `Enum` → codes, `UUID`/`IPv4`/`Decimal` → bytes/decimals), every `SELECT`/`WITH` is first `DESCRIBE`d and, when needed, wrapped in `SELECT * REPLACE (…)` converting those columns server-side; all timestamps are then normalized to naive microseconds. `ClickHouseDialect` casts to `Nullable(…)` types (ClickHouse cannot cast `NULL` to a non-nullable type), uses `TEMPORARY` in temp-table DDL, maps quantiles to `quantileExactInclusive` (inline, no correlated subquery), casts `greatest`/`least` arguments to Float64 (no UInt64/Float64 supertype), and spells the caching layer's memo-table bookkeeping (the `cache_meta_*` `SqlDialect` hooks) as a `Memory` table with `ALTER TABLE … UPDATE/DELETE`. Schema introspection reads `system.databases/tables/columns`; a database is both catalog and schema. `ClickHouseSqlReader::supports_temporary_tables()` probes once; `reader_from_uri` wraps a plain `clickhouse://` reader that fails the probe (read-only account, e.g. `play.clickhouse.com`) in a `CachingReader` on a chDB backend, so no DuckDB is involved on the ClickHouse path. Arrow IPC batches from ClickHouse are LZ4-compressed by default, hence `arrow/ipc_compression`. Live tests: HTTP ones are gated on `GGSQL_CLICKHOUSE_URI`; chDB ones skip when libchdb cannot be loaded and are serialized on a static mutex because the executor's temp-table names are per process. +**ClickHouse readers.** `clickhouse/mod.rs` holds `ClickHouseSqlReader`, one `Reader` implementation shared by two transports: `http.rs` (`ClickHouseReader`, one `POST` per statement against a server, results as `FORMAT ArrowStream`, `register()` = `CREATE TEMPORARY TABLE` + `INSERT … FORMAT ArrowStream`, one `session_id` per reader so temp tables and `SET` persist) and `chdb.rs` (`ChdbReader`, libchdb loaded at runtime with `libloading` like the ODBC driver manager — search order `GGSQL_CHDB_LIBRARY`, system path, usual install dirs — so the `chdb` feature adds no build dependency; inserts go through a temp file and `FROM INFILE` because the C API has no input-data channel; libchdb allows **one connection per process**, shared by all `ChdbReader`s on the same path). Because ClickHouse's Arrow output loses some of its own types (`DateTime` → `UInt32`, `Enum` → codes, `UUID`/`IPv4`/`Decimal` → bytes/decimals), every `SELECT`/`WITH` is first `DESCRIBE`d and, when needed, wrapped in `SELECT * REPLACE (…)` converting those columns server-side; all timestamps are then normalized to naive microseconds. `ClickHouseDialect` casts to `Nullable(…)` types (ClickHouse cannot cast `NULL` to a non-nullable type), uses `TEMPORARY` in temp-table DDL, maps quantiles to `quantileExactInclusive` (inline, no correlated subquery), casts `greatest`/`least` arguments to Float64 (no UInt64/Float64 supertype), and spells the caching layer's memo-table bookkeeping (the `cache_meta_*` `SqlDialect` hooks) as a `Memory` table with `ALTER TABLE … UPDATE/DELETE`. ClickHouse 26.8+ is assumed; there are no shims for older servers. Schema introspection reads `system.databases/tables/columns`; a database is both catalog and schema. `ClickHouseSqlReader::supports_temporary_tables()` probes once; `reader_from_uri` wraps a plain `clickhouse://` reader that fails the probe (read-only account, e.g. `play.clickhouse.com`) in a `CachingReader` on a chDB backend, so no DuckDB is involved on the ClickHouse path. Arrow IPC batches from ClickHouse are LZ4-compressed by default, hence `arrow/ipc_compression`. Live tests: HTTP ones are gated on `GGSQL_CLICKHOUSE_URI`; chDB ones skip when libchdb cannot be loaded and are serialized on a static mutex because the executor's temp-table names are per process. **Caching layer.** `CachingReader` (`cache.rs`) wraps a primary reader plus an in-memory `CacheBackend`, splitting work across two `Reader` surfaces. **`execute_sql` = source**: base reads of the user's data plus user setup/DML run on the primary (with result memoization), except `ggsql:` builtins, the `__ggsql_cache_meta__` table, and reads that reference a cache-resident internal table, which go to the cache. **`execute_sql_cached` = compute**: all dialect-generated/derived SQL (schema probes, stats, projection/map transforms, spatial setup, final layer queries — everything operating on `__ggsql_*` tables) runs on the cache; it defaults to `execute_sql` so a plain reader runs everything on one connection. Cache routing is by **exact-identifier membership** in the set of tables registered into the cache. Memoization keys on `hash(primary_uri + sql)` and is tracked in the `__ggsql_cache_meta__` table inside the cache backend. Each memoized read is bounded by a **TTL** (default 300s) and the whole memo by an **LRU byte budget** (default 512 MB); both are configurable via `CacheConfig` (env `GGSQL_CACHE_DISABLED`/`GGSQL_CACHE_TTL`/`GGSQL_CACHE_MAX_BYTES`, or per-connection URI query parameters `?cache_ttl=…&cache_max_bytes=…&cache_disabled=…`). The `__ggsql_cache_meta__` table is queryable for introspection (`SELECT * FROM __ggsql_cache_meta__`). Pure/non-visual SQL (CLI table fallback, Jupyter) goes through `execute_sql` so it reads the primary rather than the empty cache. `Reader::materialize_table` (default = `CREATE TEMP TABLE` on the reader, no Rust roundtrip) is overridden to read the body via the source surface and `register()` the result into the cache, so the primary is never written to; `Reader::caches_sources()` (default `false`, `true` for `CachingReader`) gates the executor's per-layer source staging: file sources are staged on the cache surface, while identifiers go through `materialize_table`, which routes the read to the cache (CTEs, builtins, cache-resident tables) or the primary as needed. `dialect()` returns the **cache** dialect, and every compute-surface failure is prefixed with the cache backend's scheme (``on the `duckdb` cache backend: …``) so a cache-dialect driver error is not mistaken for one from the user's own connection. Selected via the composite `+://` scheme (`reader_from_uri` / `split_cache_uri`) or the CLI `--cache` flag; off by default. diff --git a/src/plot/layer/geom/density.rs b/src/plot/layer/geom/density.rs index 18341dafd..3ac2a2ea9 100644 --- a/src/plot/layer/geom/density.rs +++ b/src/plot/layer/geom/density.rs @@ -187,7 +187,6 @@ pub(crate) fn stat_density( &bw_cte, &data_cte, &grid_cte, - dialect, ); let mut consumed = vec![value_aesthetic.to_string()]; @@ -456,8 +455,7 @@ fn build_grid_cte( .iter() .map(|g| { let q = naming::quote_ident(g); - dialect - .sql_null_safe_equals(&format!("full_grid.{q}"), &format!("bandwidth.{q}")) + format!("full_grid.{q} IS NOT DISTINCT FROM bandwidth.{q}") }) .collect(); let grid_groups_select: Vec = groups @@ -516,7 +514,6 @@ fn compute_density( bandwidth_cte: &str, data_cte: &str, grid_cte: &str, - dialect: &dyn SqlDialect, ) -> String { // Build bandwidth join condition (NULL-safe) let bandwidth_conditions = if group_by.is_empty() { @@ -526,7 +523,7 @@ fn compute_density( .iter() .map(|g| { let q = naming::quote_ident(g); - dialect.sql_null_safe_equals(&format!("data.{q}"), &format!("bandwidth.{q}")) + format!("data.{q} IS NOT DISTINCT FROM bandwidth.{q}") }) .collect::>() .join(" AND ") @@ -540,7 +537,7 @@ fn compute_density( .iter() .map(|g| { let q = naming::quote_ident(g); - dialect.sql_null_safe_equals(&format!("grid.{q}"), &format!("data.{q}")) + format!("grid.{q} IS NOT DISTINCT FROM data.{q}") }) .collect(); format!("WHERE {}", grid_data_conds.join(" AND ")) @@ -640,15 +637,7 @@ mod tests { let data_cte = build_data_cte("x", None, None, query, &groups); let grid_cte = build_grid_cte(&groups, 512, None, &AnsiDialect); let kernel = choose_kde_kernel(¶meters, None).expect("kernel should be valid"); - let sql = compute_density( - "x", - &groups, - kernel, - &bw_cte, - &data_cte, - &grid_cte, - &AnsiDialect, - ); + let sql = compute_density("x", &groups, kernel, &bw_cte, &data_cte, &grid_cte); let expected = r#"WITH RECURSIVE bandwidth AS ( @@ -724,15 +713,7 @@ mod tests { let data_cte = build_data_cte("x", None, None, query, &groups); let grid_cte = build_grid_cte(&groups, 512, None, &AnsiDialect); let kernel = choose_kde_kernel(¶meters, None).expect("kernel should be valid"); - let sql = compute_density( - "x", - &groups, - kernel, - &bw_cte, - &data_cte, - &grid_cte, - &AnsiDialect, - ); + let sql = compute_density("x", &groups, kernel, &bw_cte, &data_cte, &grid_cte); let expected = r#"WITH RECURSIVE bandwidth AS ( @@ -911,15 +892,7 @@ mod tests { // Use wide range to capture essentially all density mass let grid_cte = build_grid_cte(&groups, 512, None, &AnsiDialect); let kernel = choose_kde_kernel(¶meters, None).expect("kernel should be valid"); - let sql = compute_density( - "x", - &groups, - kernel, - &bw_cte, - &data_cte, - &grid_cte, - &AnsiDialect, - ); + let sql = compute_density("x", &groups, kernel, &bw_cte, &data_cte, &grid_cte); // Execute query let reader = DuckDBReader::from_connection_string("duckdb://memory").unwrap(); @@ -1046,7 +1019,6 @@ mod tests { &bw_cte, &data_cte_unweighted, &grid_cte, - &AnsiDialect, ); let reader = DuckDBReader::from_connection_string("duckdb://memory").unwrap(); @@ -1057,15 +1029,8 @@ mod tests { // With explicit uniform weights (should be equivalent) let query_weighted = "SELECT x, 1.0 AS weight FROM (VALUES (1.0), (2.0), (3.0)) AS t(x)"; let data_cte_weighted = build_data_cte("x", None, Some("weight"), query_weighted, &groups); - let sql_weighted = compute_density( - "x", - &groups, - kernel, - &bw_cte, - &data_cte_weighted, - &grid_cte, - &AnsiDialect, - ); + let sql_weighted = + compute_density("x", &groups, kernel, &bw_cte, &data_cte_weighted, &grid_cte); let df_weighted = reader .execute_sql(&sql_weighted) .expect("SQL should execute"); @@ -1202,15 +1167,7 @@ mod tests { let data_cte = build_data_cte("x", None, None, query, &groups); let grid_cte = build_grid_cte(&groups, 512, None, &AnsiDialect); let kernel = choose_kde_kernel(¶meters, None).expect("kernel should be valid"); - let sql = compute_density( - "x", - &groups, - kernel, - &bw_cte, - &data_cte, - &grid_cte, - &AnsiDialect, - ); + let sql = compute_density("x", &groups, kernel, &bw_cte, &data_cte, &grid_cte); // Warm-up run reader.execute_sql(&sql).expect("Warm-up failed"); diff --git a/src/reader/clickhouse/mod.rs b/src/reader/clickhouse/mod.rs index 5413a8e8e..ee23b8067 100644 --- a/src/reader/clickhouse/mod.rs +++ b/src/reader/clickhouse/mod.rs @@ -11,6 +11,9 @@ //! the in-memory [`CacheBackend`] behind `chdb+://` connection //! strings, so a ClickHouse setup never needs another database engine. //! +//! ClickHouse 26.8 or newer is assumed (correlated subqueries, `IS NOT +//! DISTINCT FROM` in any clause); older servers are not supported. +//! //! # Types //! //! ClickHouse's Arrow output cannot express some of its own types: `DateTime` @@ -143,12 +146,6 @@ impl SqlDialect for ClickHouseDialect { format!("least({})", float_args(exprs)) } - /// Older ClickHouse versions only accept `IS NOT DISTINCT FROM` inside - /// `JOIN ON`; this spelling works in any clause on every version. - fn sql_null_safe_equals(&self, left: &str, right: &str) -> String { - format!("(({left} = {right}) OR ({left} IS NULL AND {right} IS NULL))") - } - fn sql_select_replace( &self, expr: &str, @@ -173,8 +170,7 @@ impl SqlDialect for ClickHouseDialect { /// Every caller embeds this in a `GROUP BY {groups}` query over `from`, so /// the native aggregate is equivalent to the correlated scalar subquery - /// other dialects produce, and it also runs on ClickHouse versions without - /// correlated-subquery support. + /// other dialects produce, and far cheaper. fn sql_percentile( &self, column: &str, @@ -902,10 +898,6 @@ mod tests { ClickHouseDialect.sql_percentile("v", 0.5, "ignored", &["g".to_string()]), "quantileExactInclusive(0.5)(\"v\")" ); - assert_eq!( - ClickHouseDialect.sql_null_safe_equals("a.k", "b.k"), - "((a.k = b.k) OR (a.k IS NULL AND b.k IS NULL))" - ); assert_eq!(ClickHouseDialect.sql_date_literal(-1), "toDate32(-1)"); assert_eq!( ClickHouseDialect.sql_datetime_literal(1_000_000), diff --git a/src/reader/mod.rs b/src/reader/mod.rs index cb23c8c35..969816d23 100644 --- a/src/reader/mod.rs +++ b/src/reader/mod.rs @@ -94,13 +94,6 @@ pub trait SqlDialect { } } - /// Null-safe equality of two expressions: true when both are equal or - /// both NULL. Override for backends that restrict `IS NOT DISTINCT FROM` - /// (ClickHouse only accepts it in `JOIN ON`). - fn sql_null_safe_equals(&self, left: &str, right: &str) -> String { - format!("{left} IS NOT DISTINCT FROM {right}") - } - /// Scalar MAX across any number of SQL expressions. fn sql_greatest(&self, exprs: &[&str]) -> String { let mut result = exprs[0].to_string();