From 182a9bf45186669654ff2eec0dd520beefbd1e75 Mon Sep 17 00:00:00 2001 From: Trung Nguyen Date: Mon, 27 Jul 2026 23:32:00 +0200 Subject: [PATCH 1/4] feat(drivers): add ClickHouse driver via official clickhouse crate Co-authored-by: Cursor --- linux/Cargo.lock | 224 +++++++ linux/Cargo.toml | 1 + linux/crates/app/Cargo.toml | 1 + linux/crates/app/src/main.rs | 1 + linux/crates/core/src/sql_dialect.rs | 2 +- linux/crates/drivers/clickhouse/Cargo.toml | 28 + linux/crates/drivers/clickhouse/src/lib.rs | 624 ++++++++++++++++++ .../drivers/clickhouse/tests/integration.rs | 155 +++++ linux/docs/adding-drivers.md | 2 +- 9 files changed, 1036 insertions(+), 2 deletions(-) create mode 100644 linux/crates/drivers/clickhouse/Cargo.toml create mode 100644 linux/crates/drivers/clickhouse/src/lib.rs create mode 100644 linux/crates/drivers/clickhouse/tests/integration.rs diff --git a/linux/Cargo.lock b/linux/Cargo.lock index a1270c316..c79466419 100644 --- a/linux/Cargo.lock +++ b/linux/Cargo.lock @@ -400,6 +400,12 @@ dependencies = [ "cipher", ] +[[package]] +name = "bnum" +version = "0.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "119771309b95163ec7aaf79810da82f7cd0599c19722d48b9c03894dca833966" + [[package]] name = "bollard" version = "0.20.2" @@ -498,6 +504,15 @@ dependencies = [ "syn 2.0.117", ] +[[package]] +name = "bstr" +version = "1.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1f7dc094d718f2e1c1559ad110e27eeaae14a5465d3d56dd6dbd793079fbd530" +dependencies = [ + "memchr", +] + [[package]] name = "bumpalo" version = "3.20.2" @@ -537,6 +552,9 @@ name = "bytes" version = "1.11.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1e748733b7cbc798e1434b6ac524f0c1ff2ab456fe201501e6497c8417a4fc33" +dependencies = [ + "serde", +] [[package]] name = "cairo-rs" @@ -651,6 +669,55 @@ dependencies = [ "zeroize", ] +[[package]] +name = "clickhouse" +version = "0.15.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d8063696febb0a10a6fb9df1460c52c509188f1d19da2157694ab3ca0feffc74" +dependencies = [ + "bnum", + "bstr", + "bytes", + "clickhouse-macros", + "clickhouse-types", + "futures-channel", + "futures-util", + "http-body-util", + "hyper", + "hyper-rustls", + "hyper-util", + "polonius-the-crab", + "rustls 0.23.39", + "serde", + "serde_json", + "thiserror 2.0.18", + "tokio", + "tracing", + "url", +] + +[[package]] +name = "clickhouse-macros" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ff6669899e23cb87b43daf7996f0ea3b9c07d0fb933d745bb7b815b052515ae3" +dependencies = [ + "proc-macro2", + "quote", + "serde_derive_internals", + "syn 2.0.117", +] + +[[package]] +name = "clickhouse-types" +version = "0.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "30a5efddc880ce9e2573bd867413d9056fa2bea0206af88dec21e72178b9dc74" +dependencies = [ + "bytes", + "thiserror 2.0.18", +] + [[package]] name = "cmake" version = "0.1.58" @@ -1842,6 +1909,17 @@ version = "0.4.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6fe2267d4ed49bc07b63801559be28c718ea06c4738b7a03c94df7386d2cde46" +[[package]] +name = "higher-kinded-types" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e690f8474c6c5d8ff99656fcbc195a215acc3949481a8b0b3351c838972dc776" +dependencies = [ + "macro_rules_attribute", + "never-say-never", + "paste", +] + [[package]] name = "hkdf" version = "0.12.4" @@ -1964,6 +2042,7 @@ dependencies = [ "tokio", "tokio-rustls 0.26.4", "tower-service", + "webpki-roots 1.0.7", ] [[package]] @@ -1985,13 +2064,16 @@ version = "0.1.20" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "96547c2556ec9d12fb1578c4eaf448b04993e7fb79cbaad930a656880a6bdfa0" dependencies = [ + "base64 0.22.1", "bytes", "futures-channel", "futures-util", "http", "http-body", "hyper", + "ipnet", "libc", + "percent-encoding", "pin-project-lite", "socket2", "tokio", @@ -2214,6 +2296,12 @@ dependencies = [ "zeroize", ] +[[package]] +name = "ipnet" +version = "2.12.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d98f6fed1fde3f8c21bc40a1abb88dd75e67924f9cffc3ef95607bad8017f8e2" + [[package]] name = "itertools" version = "0.14.0" @@ -2438,6 +2526,22 @@ version = "0.4.29" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5e5032e24019045c762d3c0f28f5b6b8bbf38563a65908389bf7978758920897" +[[package]] +name = "macro_rules_attribute" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "65049d7923698040cd0b1ddcced9b0eb14dd22c5f86ae59c3740eab64a676520" +dependencies = [ + "macro_rules_attribute-proc_macro", + "paste", +] + +[[package]] +name = "macro_rules_attribute-proc_macro" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "670fdfda89751bc4a84ac13eaa63e205cf0fd22b4c9a5fbfa085b63c1f1d3a30" + [[package]] name = "malloc_buf" version = "0.0.6" @@ -2520,6 +2624,12 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "never-say-never" +version = "6.6.666" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cf5a574dadd7941adeaa71823ecba5e28331b8313fb2e1c6a5c7e5981ea53ad6" + [[package]] name = "nix" version = "0.29.0" @@ -2895,6 +3005,12 @@ dependencies = [ "subtle", ] +[[package]] +name = "paste" +version = "1.0.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "57c0d7b74b563b49d38dae00a0c37d4d6de9b432382b2892f0574ddcae73fd0a" + [[package]] name = "pastey" version = "0.1.1" @@ -3002,6 +3118,16 @@ version = "0.2.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b4596b6d070b27117e987119b4dac604f3c58cfb0b191112e24771b2faeac1a6" +[[package]] +name = "polonius-the-crab" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ec242d7eccbb2fd8b3b5b6e3cf89f94a91a800f469005b44d154359609f8af72" +dependencies = [ + "higher-kinded-types", + "never-say-never", +] + [[package]] name = "poly1305" version = "0.8.0" @@ -3379,6 +3505,41 @@ dependencies = [ "bytecheck", ] +[[package]] +name = "reqwest" +version = "0.13.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "219c5811de6525e5416c7d5d53bb656d3afdbc6c5af816e0802bcfa42dbdc1c3" +dependencies = [ + "base64 0.22.1", + "bytes", + "encoding_rs", + "futures-core", + "h2", + "http", + "http-body", + "http-body-util", + "hyper", + "hyper-rustls", + "hyper-util", + "js-sys", + "log", + "mime", + "percent-encoding", + "pin-project-lite", + "serde", + "serde_json", + "sync_wrapper", + "tokio", + "tower", + "tower-http", + "tower-service", + "url", + "wasm-bindgen", + "wasm-bindgen-futures", + "web-sys", +] + [[package]] name = "rfc6979" version = "0.4.0" @@ -3603,6 +3764,7 @@ version = "0.23.39" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7c2c118cb077cca2822033836dfb1b975355dfb784b5e8da48f7b6c5db74e60e" dependencies = [ + "aws-lc-rs", "log", "once_cell", "ring", @@ -3670,6 +3832,7 @@ version = "0.103.13" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "61c429a8649f110dddef65e2a5ad240f747e85f7758a6bccc7e5777bd33f756e" dependencies = [ + "aws-lc-rs", "ring", "rustls-pki-types", "untrusted 0.9.0", @@ -3868,6 +4031,17 @@ dependencies = [ "syn 2.0.117", ] +[[package]] +name = "serde_derive_internals" +version = "0.29.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "18d26a20a969b9e3fdf2fc2d9f21eda6c40e2de84c9408bb5d3b05d499aae711" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.117", +] + [[package]] name = "serde_json" version = "1.0.149" @@ -4411,6 +4585,9 @@ name = "sync_wrapper" version = "1.0.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0bf256ce5efdfa370213c1dabab5935a12e49f2c58d15e9eac2870d3b4f27263" +dependencies = [ + "futures-core", +] [[package]] name = "synstructure" @@ -4456,6 +4633,7 @@ dependencies = [ "sourceview5", "sqlformat", "tablepro-core", + "tablepro-driver-clickhouse", "tablepro-driver-mssql", "tablepro-driver-mysql", "tablepro-driver-postgres", @@ -4485,6 +4663,23 @@ dependencies = [ "uuid", ] +[[package]] +name = "tablepro-driver-clickhouse" +version = "0.1.0" +dependencies = [ + "async-trait", + "chrono", + "clickhouse", + "rust_decimal", + "secrecy", + "serde", + "serde_json", + "tablepro-core", + "testcontainers", + "tokio", + "uuid", +] + [[package]] name = "tablepro-driver-mssql" version = "0.1.0" @@ -4644,6 +4839,7 @@ dependencies = [ "memchr", "parse-display", "pin-project-lite", + "reqwest", "serde", "serde_json", "serde_with", @@ -5001,6 +5197,24 @@ dependencies = [ "tracing", ] +[[package]] +name = "tower-http" +version = "0.6.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4cfcf7e2740e6fc6d4d688b4ef00650406bb94adf4731e43c096c3a19fe40840" +dependencies = [ + "bitflags", + "bytes", + "futures-util", + "http", + "http-body", + "pin-project-lite", + "tower", + "tower-layer", + "tower-service", + "url", +] + [[package]] name = "tower-layer" version = "0.3.3" @@ -5376,6 +5590,16 @@ dependencies = [ "semver", ] +[[package]] +name = "web-sys" +version = "0.3.95" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4f2dfbb17949fa2088e5d39408c48368947b86f7834484e87b73de55bc14d97d" +dependencies = [ + "js-sys", + "wasm-bindgen", +] + [[package]] name = "web-time" version = "1.1.0" diff --git a/linux/Cargo.toml b/linux/Cargo.toml index 27aa71c5e..ec5b95d19 100644 --- a/linux/Cargo.toml +++ b/linux/Cargo.toml @@ -9,6 +9,7 @@ members = [ "crates/drivers/mysql", "crates/drivers/postgres", "crates/drivers/sqlite", + "crates/drivers/clickhouse", ] [workspace.package] diff --git a/linux/crates/app/Cargo.toml b/linux/crates/app/Cargo.toml index 39db1d6a4..3e11a235b 100644 --- a/linux/crates/app/Cargo.toml +++ b/linux/crates/app/Cargo.toml @@ -11,6 +11,7 @@ path = "src/main.rs" [dependencies] tablepro-core = { path = "../core" } +tablepro-driver-clickhouse = { path = "../drivers/clickhouse" } tablepro-driver-mssql = { path = "../drivers/mssql" } tablepro-driver-mysql = { path = "../drivers/mysql" } tablepro-driver-postgres = { path = "../drivers/postgres" } diff --git a/linux/crates/app/src/main.rs b/linux/crates/app/src/main.rs index 4d6d59af0..62c1675c9 100644 --- a/linux/crates/app/src/main.rs +++ b/linux/crates/app/src/main.rs @@ -70,6 +70,7 @@ fn main() { fn build_registry() -> DriverRegistry { let mut r = DriverRegistry::new(); + r.register(Arc::new(drivers_clickhouse::ClickhouseDriver)); r.register(Arc::new(drivers_mssql::MssqlDriver)); r.register(Arc::new(drivers_mysql::MysqlDriver)); r.register(Arc::new(drivers_postgres::PgDriver)); diff --git a/linux/crates/core/src/sql_dialect.rs b/linux/crates/core/src/sql_dialect.rs index 4591e4d60..3149588da 100644 --- a/linux/crates/core/src/sql_dialect.rs +++ b/linux/crates/core/src/sql_dialect.rs @@ -16,7 +16,7 @@ pub enum BuildSqlError { pub fn quote_ident(driver_id: &str, name: &str) -> String { match driver_id { - "mysql" => format!("`{}`", name.replace('`', "``")), + "mysql" | "clickhouse" => format!("`{}`", name.replace('`', "``")), "mssql" => format!("[{}]", name.replace(']', "]]")), _ => format!("\"{}\"", name.replace('"', "\"\"")), } diff --git a/linux/crates/drivers/clickhouse/Cargo.toml b/linux/crates/drivers/clickhouse/Cargo.toml new file mode 100644 index 000000000..4997024ba --- /dev/null +++ b/linux/crates/drivers/clickhouse/Cargo.toml @@ -0,0 +1,28 @@ +[package] +name = "tablepro-driver-clickhouse" +version = "0.1.0" +edition.workspace = true +rust-version.workspace = true +publish = false + +[lib] +name = "drivers_clickhouse" +path = "src/lib.rs" + +[dependencies] +tablepro-core = { path = "../../core" } +async-trait.workspace = true +chrono.workspace = true +clickhouse = { version = "0.15", default-features = false, features = ["rustls-tls"] } +rust_decimal.workspace = true +secrecy.workspace = true +serde.workspace = true +serde_json.workspace = true +tokio.workspace = true +uuid.workspace = true + +[dev-dependencies] +chrono.workspace = true +secrecy.workspace = true +testcontainers = { version = "0.27", features = ["http_wait_plain"] } +tokio = { workspace = true, features = ["macros", "rt-multi-thread"] } diff --git a/linux/crates/drivers/clickhouse/src/lib.rs b/linux/crates/drivers/clickhouse/src/lib.rs new file mode 100644 index 000000000..ee02be9bc --- /dev/null +++ b/linux/crates/drivers/clickhouse/src/lib.rs @@ -0,0 +1,624 @@ +use async_trait::async_trait; +use secrecy::ExposeSecret; +use serde::Deserialize; + +use tablepro_core::{ + ColumnInfo, ConnectOptions, Connection, DatabaseDriver, DriverError, ExecResult, ForeignKeyInfo, IndexInfo, + MAX_QUERY_ROWS, QueryResult, TableInfo, Value, +}; + +pub struct ClickhouseDriver; + +#[async_trait] +impl DatabaseDriver for ClickhouseDriver { + fn id(&self) -> &'static str { + "clickhouse" + } + + fn display_name(&self) -> &'static str { + "ClickHouse" + } + + fn default_port(&self) -> u16 { + 8123 + } + + async fn connect(&self, opts: ConnectOptions) -> Result, DriverError> { + let scheme = if opts.use_tls { "https" } else { "http" }; + let url = format!("{scheme}://{}:{}", opts.host, opts.port); + let mut client = clickhouse::Client::default() + .with_url(url) + .with_product_info("tablepro-linux", env!("CARGO_PKG_VERSION")); + if !opts.username.is_empty() { + client = client.with_user(opts.username); + } + if !opts.password.expose_secret().is_empty() { + client = client.with_password(opts.password.expose_secret()); + } + if !opts.database.is_empty() { + client = client.with_database(opts.database.clone()); + } + + // Prove the HTTP endpoint is reachable and credentials work. + client + .query("SELECT 1") + .execute() + .await + .map_err(map_clickhouse_error)?; + + Ok(Box::new(ClickhouseConnection { + client, + database: if opts.database.is_empty() { + "default".into() + } else { + opts.database + }, + })) + } +} + +struct ClickhouseConnection { + client: clickhouse::Client, + database: String, +} + +#[async_trait] +impl Connection for ClickhouseConnection { + async fn list_tables(&self) -> Result, DriverError> { + #[derive(Debug, Deserialize, clickhouse::Row)] + struct Row { + database: String, + name: String, + } + + let rows = self + .client + .query( + "SELECT database, name + FROM system.tables + WHERE database = currentDatabase() + AND is_temporary = 0 + AND engine NOT LIKE '%View' + ORDER BY name", + ) + .fetch_all::() + .await + .map_err(map_clickhouse_error)?; + + Ok(rows + .into_iter() + .map(|r| TableInfo { + schema: Some(r.database), + name: r.name, + }) + .collect()) + } + + async fn fetch_columns(&self, schema: Option<&str>, table: &str) -> Result, DriverError> { + #[derive(Debug, Deserialize, clickhouse::Row)] + struct Row { + name: String, + #[serde(rename = "type")] + data_type: String, + is_in_primary_key: u8, + default_kind: String, + default_expression: String, + } + + let db = schema.unwrap_or(self.database.as_str()); + let rows = self + .client + .query( + "SELECT + name, + type, + is_in_primary_key, + default_kind, + default_expression + FROM system.columns + WHERE database = ? + AND table = ? + ORDER BY position", + ) + .bind(db) + .bind(table) + .fetch_all::() + .await + .map_err(map_clickhouse_error)?; + + Ok(rows + .into_iter() + .map(|r| { + let is_generated = matches!( + r.default_kind.as_str(), + "MATERIALIZED" | "ALIAS" | "EPHEMERAL" + ); + let default_value = if r.default_expression.is_empty() { + None + } else { + Some(r.default_expression) + }; + let nullable = r.data_type.starts_with("Nullable("); + ColumnInfo { + name: r.name, + data_type: r.data_type, + nullable, + primary_key: r.is_in_primary_key != 0, + is_auto_increment: false, + default_value, + is_generated, + } + }) + .collect()) + } + + async fn fetch_rows( + &self, + schema: Option<&str>, + table: &str, + offset: u64, + limit: u64, + ) -> Result { + let qualified = qualify(schema.unwrap_or(self.database.as_str()), table); + let sql = format!("SELECT * FROM {qualified} LIMIT {limit} OFFSET {offset}"); + self.query(&sql).await + } + + async fn query(&self, sql: &str) -> Result { + query_json(&self.client, sql).await + } + + async fn query_params(&self, sql: &str, params: &[Value]) -> Result { + if params.is_empty() { + return self.query(sql).await; + } + let bound = bind_placeholders(sql, params)?; + self.query(&bound).await + } + + async fn execute(&self, sql: &str) -> Result { + self.client + .query(sql) + .execute() + .await + .map_err(map_clickhouse_error)?; + // ClickHouse HTTP does not return rows_affected for arbitrary + // mutations reliably; report 0 rather than invent a count. + Ok(ExecResult { rows_affected: 0 }) + } + + async fn execute_params(&self, sql: &str, params: &[Value]) -> Result { + let bound = bind_placeholders(sql, params)?; + self.execute(&bound).await + } + + async fn execute_in_transaction(&self, statements: &[(String, Vec)]) -> Result, DriverError> { + // ClickHouse has no multi-statement ACID transaction for DML. + // Run statements sequentially; on failure surface which index + // failed. Callers already treat this as best-effort for CH. + let mut affected = Vec::with_capacity(statements.len()); + for (i, (sql, params)) in statements.iter().enumerate() { + match self.execute_params(sql, params).await { + Ok(r) => affected.push(r.rows_affected), + Err(e) => { + return Err(DriverError::Transaction { + statement_index: i, + source: Box::new(e), + }); + } + } + } + Ok(affected) + } + + async fn fetch_indexes(&self, schema: Option<&str>, table: &str) -> Result, DriverError> { + #[derive(Debug, Deserialize, clickhouse::Row)] + struct Row { + name: String, + primary_key: String, + } + + let db = schema.unwrap_or(self.database.as_str()); + let rows = self + .client + .query( + "SELECT name, primary_key + FROM system.tables + WHERE database = ? + AND name = ? + LIMIT 1", + ) + .bind(db) + .bind(table) + .fetch_all::() + .await + .map_err(map_clickhouse_error)?; + + let Some(row) = rows.into_iter().next() else { + return Ok(Vec::new()); + }; + if row.primary_key.is_empty() { + return Ok(Vec::new()); + } + let columns: Vec = row + .primary_key + .split(',') + .map(|s| s.trim().trim_matches('`').trim_matches('"').to_string()) + .filter(|s| !s.is_empty()) + .collect(); + if columns.is_empty() { + return Ok(Vec::new()); + } + Ok(vec![IndexInfo { + name: format!("{}_pk", row.name), + columns, + unique: true, + primary: true, + }]) + } + + async fn fetch_foreign_keys( + &self, + _schema: Option<&str>, + _table: &str, + ) -> Result, DriverError> { + // ClickHouse has no classical FK constraints. + Ok(Vec::new()) + } + + async fn ping(&self) -> Result<(), DriverError> { + self.client + .query("SELECT 1") + .execute() + .await + .map_err(map_clickhouse_error) + } + + async fn close(self: Box) -> Result<(), DriverError> { + Ok(()) + } +} + +#[derive(Debug, Deserialize)] +struct JsonMeta { + name: String, + #[serde(rename = "type")] + data_type: String, +} + +#[derive(Debug, Deserialize)] +struct JsonResponse { + #[serde(default)] + meta: Vec, + #[serde(default)] + data: Vec>, + #[serde(default)] + rows: u64, +} + +async fn query_json(client: &clickhouse::Client, sql: &str) -> Result { + let mut cursor = client + .query(sql) + .fetch_bytes("JSON") + .map_err(map_clickhouse_error)?; + let bytes = cursor.collect().await.map_err(map_clickhouse_error)?; + if bytes.is_empty() { + return Ok(QueryResult { + columns: Vec::new(), + rows: Vec::new(), + truncated: false, + }); + } + + let parsed: JsonResponse = serde_json::from_slice(&bytes) + .map_err(|e| DriverError::Internal(format!("clickhouse JSON parse: {e}")))?; + + let columns: Vec = parsed + .meta + .iter() + .map(|m| ColumnInfo { + name: m.name.clone(), + data_type: m.data_type.clone(), + nullable: m.data_type.starts_with("Nullable(") || m.data_type == "Nullable", + primary_key: false, + is_auto_increment: false, + default_value: None, + is_generated: false, + }) + .collect(); + + let mut rows = Vec::with_capacity(parsed.data.len().min(MAX_QUERY_ROWS)); + for obj in parsed.data.into_iter().take(MAX_QUERY_ROWS) { + let mut row = Vec::with_capacity(columns.len()); + for col in &columns { + let raw = obj.get(&col.name).cloned().unwrap_or(serde_json::Value::Null); + row.push(json_to_value(raw, &col.data_type)); + } + rows.push(row); + } + + let truncated = parsed.rows as usize > rows.len() || rows.len() >= MAX_QUERY_ROWS; + Ok(QueryResult { + columns, + rows, + truncated, + }) +} + +fn json_to_value(raw: serde_json::Value, type_name: &str) -> Value { + if raw.is_null() { + return Value::Null; + } + let base = strip_nullable(type_name); + match base { + "Bool" => raw + .as_bool() + .map(Value::Bool) + .or_else(|| raw.as_u64().map(|n| Value::Bool(n != 0))) + .unwrap_or_else(|| Value::Text(raw.to_string())), + "Int8" | "Int16" | "Int32" | "Int64" | "UInt8" | "UInt16" | "UInt32" | "UInt64" => raw + .as_i64() + .or_else(|| raw.as_u64().and_then(|n| i64::try_from(n).ok())) + .map(Value::Int) + .or_else(|| { + raw.as_str() + .and_then(|s| s.parse::().ok()) + .map(Value::Int) + }) + .unwrap_or_else(|| Value::Text(raw.to_string())), + "Float32" | "Float64" => raw + .as_f64() + .map(Value::Float) + .or_else(|| { + raw.as_str() + .and_then(|s| s.parse::().ok()) + .map(Value::Float) + }) + .unwrap_or_else(|| Value::Text(raw.to_string())), + "Decimal" | "Decimal32" | "Decimal64" | "Decimal128" | "Decimal256" => raw + .as_str() + .and_then(|s| s.parse::().ok()) + .or_else(|| { + raw.as_f64() + .and_then(|f| rust_decimal::Decimal::try_from(f).ok()) + }) + .map(Value::Decimal) + .unwrap_or_else(|| Value::Text(raw.to_string())), + "Date" => raw + .as_str() + .and_then(|s| chrono::NaiveDate::parse_from_str(s, "%Y-%m-%d").ok()) + .map(Value::Date) + .unwrap_or_else(|| Value::Text(raw.as_str().unwrap_or(&raw.to_string()).to_string())), + "DateTime" | "DateTime64" => { + if let Some(s) = raw.as_str() { + if let Ok(dt) = chrono::DateTime::parse_from_rfc3339(s) { + return Value::TimestampTz(dt.with_timezone(&chrono::Utc)); + } + if let Ok(dt) = chrono::NaiveDateTime::parse_from_str(s, "%Y-%m-%d %H:%M:%S") { + return Value::DateTime(dt); + } + if let Ok(dt) = chrono::NaiveDateTime::parse_from_str(s, "%Y-%m-%d %H:%M:%S%.f") { + return Value::DateTime(dt); + } + return Value::Text(s.to_string()); + } + Value::Text(raw.to_string()) + } + "UUID" => raw + .as_str() + .and_then(|s| s.parse::().ok()) + .map(Value::Uuid) + .unwrap_or_else(|| Value::Text(raw.as_str().unwrap_or(&raw.to_string()).to_string())), + "String" | "FixedString" | "LowCardinality(String)" => { + Value::Text(raw.as_str().unwrap_or(&raw.to_string()).to_string()) + } + t if t.starts_with("Array") || t.starts_with("Map") || t.starts_with("Tuple") || t == "JSON" + || t.starts_with("Object") => + { + Value::Json(raw) + } + _ => { + if let Some(s) = raw.as_str() { + Value::Text(s.to_string()) + } else if let Some(n) = raw.as_i64() { + Value::Int(n) + } else if let Some(f) = raw.as_f64() { + Value::Float(f) + } else if let Some(b) = raw.as_bool() { + Value::Bool(b) + } else { + Value::Json(raw) + } + } + } +} + +fn strip_nullable(type_name: &str) -> &str { + type_name + .strip_prefix("Nullable(") + .and_then(|s| s.strip_suffix(')')) + .unwrap_or(type_name) +} + +fn qualify(schema: &str, table: &str) -> String { + format!("{}.{}", quote_ident(schema), quote_ident(table)) +} + +fn quote_ident(name: &str) -> String { + format!("`{}`", name.replace('`', "``")) +} + +fn bind_placeholders(sql: &str, params: &[Value]) -> Result { + let mut out = String::with_capacity(sql.len() + params.len() * 8); + let mut param_idx = 0; + let mut chars = sql.chars().peekable(); + while let Some(ch) = chars.next() { + if ch == '?' { + let Some(value) = params.get(param_idx) else { + return Err(DriverError::Internal(format!( + "not enough bind parameters: need at least {}", + param_idx + 1 + ))); + }; + out.push_str(&literal(value)); + param_idx += 1; + continue; + } + if ch == '$' { + let mut digits = String::new(); + while let Some(d) = chars.peek().copied().filter(|c| c.is_ascii_digit()) { + digits.push(d); + chars.next(); + } + if !digits.is_empty() { + let n: usize = digits + .parse() + .map_err(|_| DriverError::Internal(format!("bad placeholder ${digits}")))?; + let Some(value) = params.get(n.saturating_sub(1)) else { + return Err(DriverError::Internal(format!( + "bind parameter ${n} out of range (have {})", + params.len() + ))); + }; + out.push_str(&literal(value)); + continue; + } + out.push('$'); + continue; + } + out.push(ch); + } + if param_idx > 0 && param_idx != params.len() { + return Err(DriverError::Internal(format!( + "bind parameter count mismatch: used {param_idx}, have {}", + params.len() + ))); + } + Ok(out) +} + +fn literal(value: &Value) -> String { + match value { + Value::Null => "NULL".into(), + Value::Bool(b) => if *b { "true" } else { "false" }.into(), + Value::Int(i) => i.to_string(), + Value::Float(f) => { + if f.is_finite() { + f.to_string() + } else { + "NULL".into() + } + } + Value::Text(s) => format!("'{}'", escape_str(s)), + Value::Bytes(b) => format!("unhex('{}')", hex::encode(b)), + Value::Date(d) => format!("toDate('{}')", d), + Value::Time(t) => format!("'{}'", t), + Value::DateTime(dt) => format!("toDateTime('{}')", dt.format("%Y-%m-%d %H:%M:%S")), + Value::TimestampTz(ts) => format!( + "toDateTime64('{}', 3)", + ts.format("%Y-%m-%d %H:%M:%S%.3f") + ), + Value::Decimal(d) => d.to_string(), + Value::Uuid(u) => format!("toUUID('{}')", u), + Value::Json(j) => format!("'{}'", escape_str(&j.to_string())), + } +} + +fn escape_str(s: &str) -> String { + s.replace('\\', "\\\\").replace('\'', "\\'") +} + +fn map_clickhouse_error(err: clickhouse::error::Error) -> DriverError { + let msg = err.to_string(); + let lower = msg.to_lowercase(); + if lower.contains("connection refused") || lower.contains("connect error") { + return DriverError::ConnectionRefused; + } + if lower.contains("authentication") || lower.contains("password is incorrect") || lower.contains("code: 516") + { + return DriverError::AuthFailed; + } + if lower.contains("tls") || lower.contains("certificate") { + return DriverError::Tls(msg); + } + DriverError::Query { + message: msg, + sqlstate: None, + } +} + +// Minimal hex encoder so we do not pull the `hex` crate for one call site. +mod hex { + pub fn encode(bytes: &[u8]) -> String { + const LUT: &[u8; 16] = b"0123456789abcdef"; + let mut out = String::with_capacity(bytes.len() * 2); + for b in bytes { + out.push(LUT[(b >> 4) as usize] as char); + out.push(LUT[(b & 0xf) as usize] as char); + } + out + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn driver_metadata() { + let d = ClickhouseDriver; + assert_eq!(d.id(), "clickhouse"); + assert_eq!(d.display_name(), "ClickHouse"); + assert_eq!(d.default_port(), 8123); + } + + #[test] + fn quote_ident_escapes_backticks() { + assert_eq!(quote_ident("users"), "`users`"); + assert_eq!(quote_ident("a`b"), "`a``b`"); + } + + #[test] + fn bind_question_marks() { + let sql = bind_placeholders( + "UPDATE t SET a = ? WHERE id = ?", + &[Value::Text("x".into()), Value::Int(1)], + ) + .unwrap(); + assert_eq!(sql, "UPDATE t SET a = 'x' WHERE id = 1"); + } + + #[test] + fn bind_dollar_placeholders() { + let sql = bind_placeholders( + "UPDATE t SET a = $1 WHERE id = $2", + &[Value::Text("x".into()), Value::Int(1)], + ) + .unwrap(); + assert_eq!(sql, "UPDATE t SET a = 'x' WHERE id = 1"); + } + + #[test] + fn json_to_value_maps_common_types() { + assert_eq!(json_to_value(serde_json::json!(true), "Bool"), Value::Bool(true)); + assert_eq!(json_to_value(serde_json::json!(42), "Int64"), Value::Int(42)); + assert_eq!( + json_to_value(serde_json::json!("hello"), "String"), + Value::Text("hello".into()) + ); + assert_eq!(json_to_value(serde_json::Value::Null, "Nullable(String)"), Value::Null); + assert_eq!( + json_to_value(serde_json::json!("2024-06-15"), "Date"), + Value::Date(chrono::NaiveDate::from_ymd_opt(2024, 6, 15).unwrap()) + ); + } + + #[test] + fn map_error_classifies_auth() { + // Construct via Display path: we only have Query/Auth heuristics on strings. + let err = map_clickhouse_error(clickhouse::error::Error::BadResponse( + "Code: 516. Authentication failed: password is incorrect".into(), + )); + assert!(matches!(err, DriverError::AuthFailed)); + } +} diff --git a/linux/crates/drivers/clickhouse/tests/integration.rs b/linux/crates/drivers/clickhouse/tests/integration.rs new file mode 100644 index 000000000..f719a7227 --- /dev/null +++ b/linux/crates/drivers/clickhouse/tests/integration.rs @@ -0,0 +1,155 @@ +use drivers_clickhouse::ClickhouseDriver; +use tablepro_core::{ConnectOptions, DatabaseDriver, Value}; +use testcontainers::core::{IntoContainerPort, WaitFor}; +use testcontainers::core::wait::HttpWaitStrategy; +use testcontainers::runners::AsyncRunner; +use testcontainers::{ContainerAsync, GenericImage, ImageExt}; + +async fn start_clickhouse() -> (ContainerAsync, ConnectOptions) { + let container = GenericImage::new("clickhouse/clickhouse-server", "24.8") + .with_exposed_port(8123.tcp()) + .with_wait_for(WaitFor::http( + HttpWaitStrategy::new("/ping") + .with_port(8123.tcp()) + .with_expected_status_code(200u16), + )) + .with_env_var("CLICKHOUSE_USER", "default") + .with_env_var("CLICKHOUSE_PASSWORD", "tablepro") + .with_env_var("CLICKHOUSE_DB", "default") + .with_env_var("CLICKHOUSE_DEFAULT_ACCESS_MANAGEMENT", "1") + .start() + .await + .expect("start clickhouse container"); + let host = container.get_host().await.expect("host").to_string(); + let port = container.get_host_port_ipv4(8123).await.expect("port"); + let opts = ConnectOptions { + host, + port, + database: "default".into(), + username: "default".into(), + password: secrecy::SecretString::new("tablepro".to_string().into()), + use_tls: false, + }; + (container, opts) +} + +async fn connect(opts: ConnectOptions) -> Box { + ClickhouseDriver.connect(opts).await.expect("connect") +} + +#[tokio::test] +#[ignore = "requires docker"] +async fn connect_list_tables_and_pk_detection() { + let (_c, opts) = start_clickhouse().await; + let conn = connect(opts).await; + + conn.execute( + "CREATE TABLE pk_demo ( + id UInt64, + name String, + note Nullable(String) + ) ENGINE = MergeTree + ORDER BY id", + ) + .await + .unwrap(); + conn.execute("INSERT INTO pk_demo (id, name, note) VALUES (1, 'a', NULL), (2, 'b', 'second')") + .await + .unwrap(); + + let tables = conn.list_tables().await.unwrap(); + assert!(tables.iter().any(|t| t.name == "pk_demo")); + + let cols = conn.fetch_columns(None, "pk_demo").await.unwrap(); + assert_eq!(cols.len(), 3); + let id_col = cols.iter().find(|c| c.name == "id").unwrap(); + assert!(id_col.primary_key, "ORDER BY key must be detected as primary_key"); + assert!(!id_col.nullable); + let note_col = cols.iter().find(|c| c.name == "note").unwrap(); + assert!(!note_col.primary_key); + assert!(note_col.nullable); + + let result = conn.fetch_rows(None, "pk_demo", 0, 100).await.unwrap(); + assert_eq!(result.rows.len(), 2); + assert!(!result.truncated); +} + +#[tokio::test] +#[ignore = "requires docker"] +async fn value_roundtrip_common_types() { + let (_c, opts) = start_clickhouse().await; + let conn = connect(opts).await; + + conn.execute( + "CREATE TABLE roundtrip ( + id UInt64, + b Bool, + i64 Int64, + f64 Float64, + t String, + d Date, + nullable_text Nullable(String) + ) ENGINE = MergeTree + ORDER BY id", + ) + .await + .unwrap(); + + conn.execute_params( + "INSERT INTO roundtrip (id, b, i64, f64, t, d, nullable_text) VALUES (?, ?, ?, ?, ?, ?, ?)", + &[ + Value::Int(1), + Value::Bool(true), + Value::Int(42), + Value::Float(1.5), + Value::Text("hello".into()), + Value::Date(chrono::NaiveDate::from_ymd_opt(2024, 6, 15).unwrap()), + Value::Null, + ], + ) + .await + .unwrap(); + + let result = conn + .query("SELECT id, b, i64, f64, t, d, nullable_text FROM roundtrip WHERE id = 1") + .await + .unwrap(); + assert_eq!(result.rows.len(), 1); + let row = &result.rows[0]; + assert_eq!(row[0], Value::Int(1)); + assert_eq!(row[1], Value::Bool(true)); + assert_eq!(row[2], Value::Int(42)); + assert!(matches!(row[3], Value::Float(f) if (f - 1.5).abs() < 1e-9)); + assert_eq!(row[4], Value::Text("hello".into())); + assert_eq!( + row[5], + Value::Date(chrono::NaiveDate::from_ymd_opt(2024, 6, 15).unwrap()) + ); + assert_eq!(row[6], Value::Null); +} + +#[tokio::test] +#[ignore = "requires docker"] +async fn pagination_and_truncated_flag() { + let (_c, opts) = start_clickhouse().await; + let conn = connect(opts).await; + + conn.execute("CREATE TABLE n (i UInt64) ENGINE = MergeTree ORDER BY i") + .await + .unwrap(); + for i in 1..=10 { + conn.execute(&format!("INSERT INTO n VALUES ({i})")).await.unwrap(); + } + + let page = conn.fetch_rows(None, "n", 5, 3).await.unwrap(); + assert_eq!(page.rows.len(), 3); +} + +#[tokio::test] +#[ignore = "requires docker"] +async fn bad_sql_returns_query_error() { + let (_c, opts) = start_clickhouse().await; + let conn = connect(opts).await; + let err = conn.query("SELECT * FROM definitely_missing_table_xyz").await.unwrap_err(); + assert!(matches!(err, tablepro_core::DriverError::Query { .. })); +} diff --git a/linux/docs/adding-drivers.md b/linux/docs/adding-drivers.md index 91bb4d884..014ad8887 100644 --- a/linux/docs/adding-drivers.md +++ b/linux/docs/adding-drivers.md @@ -22,7 +22,7 @@ Each step is small. The whole task takes between half a day (PG-shaped engines) | SQLite | `sqlx` with `sqlite` feature | File-based, no network. | | MSSQL | `tiberius` | Pure Rust TDS. Watch governance — `praxiomlabs/rust-mssql-driver` is a credible alternative. | | Oracle | `oracle` (rust-oracle, kubo) | Wraps ODPI-C. Requires Oracle Instant Client on the build host. | -| ClickHouse | `clickhouse-arrow` | Faster than `clickhouse-rs`. | +| ClickHouse | official `clickhouse` crate | HTTP interface (8123). Dynamic results via `FORMAT JSON`. | | Redis | `fred` | Modern tokio rewrite of redis-rs. | | MongoDB | official `mongodb` | Mature, OpenTelemetry support. | | DuckDB | `duckdb` (official) | Bundled native lib, edition 2024. | From cf5f4c0d957696f4ab288ac2449b56ad15d3d5f5 Mon Sep 17 00:00:00 2001 From: Trung Nguyen Date: Mon, 27 Jul 2026 23:41:46 +0200 Subject: [PATCH 2/4] test(core): cover clickhouse quote_ident dialect Co-authored-by: Cursor --- linux/crates/core/src/sql_dialect.rs | 2 ++ 1 file changed, 2 insertions(+) diff --git a/linux/crates/core/src/sql_dialect.rs b/linux/crates/core/src/sql_dialect.rs index 3149588da..eee96185b 100644 --- a/linux/crates/core/src/sql_dialect.rs +++ b/linux/crates/core/src/sql_dialect.rs @@ -269,6 +269,8 @@ mod tests { assert_eq!(quote_ident("postgres", "users"), "\"users\""); assert_eq!(quote_ident("sqlite", "users"), "\"users\""); assert_eq!(quote_ident("mysql", "users"), "`users`"); + assert_eq!(quote_ident("clickhouse", "users"), "`users`"); + assert_eq!(quote_ident("clickhouse", "a`b"), "`a``b`"); } #[test] From 9be9ce32fb2b3980da89eff711dfdd0860ab4638 Mon Sep 17 00:00:00 2001 From: Ngo Quoc Dat Date: Wed, 29 Jul 2026 12:30:31 +0700 Subject: [PATCH 3/4] feat(core): add an UPDATE dialect hook and a rows-affected capability Claude-Session: https://claude.ai/code/session_01Jp9Vhikm6vJFoiScLVAM4L --- .../crates/app/src/services/change_tracker.rs | 25 ++++---- linux/crates/app/src/ui/app/row_ops.rs | 13 +++- linux/crates/core/src/driver.rs | 10 ++++ linux/crates/core/src/sql_dialect.rs | 60 +++++++++++++++---- 4 files changed, 81 insertions(+), 27 deletions(-) diff --git a/linux/crates/app/src/services/change_tracker.rs b/linux/crates/app/src/services/change_tracker.rs index 9c4e252a6..114498e61 100644 --- a/linux/crates/app/src/services/change_tracker.rs +++ b/linux/crates/app/src/services/change_tracker.rs @@ -32,7 +32,7 @@ use uuid::Uuid; use tablepro_core::{ ColumnInfo, Value, - sql_dialect::{BuildSqlError, build_full_row_update, build_insert_from_draft, placeholder_for, quote_ident}, + sql_dialect::{BuildSqlError, build_insert_from_draft, build_update, placeholder_for, quote_ident}, }; const UNDO_LIMIT: usize = 50; @@ -582,12 +582,10 @@ impl TabChangeTracker { } for (row_key, mut edits) in per_row { edits.sort_by_key(|e| e.0); - // Build a synthetic original_row + new_row pair so we can - // reuse `build_full_row_update`. For non-edited columns we - // pass `Value::Null` for the new value but the UPDATE only - // touches non-PK columns and we want it to leave un-edited - // columns alone — so pass `original_row` for them too. - // Easier: build the SQL directly here. + // `build_full_row_update` writes every non-PK column, which + // would clobber concurrent edits to columns this user never + // touched. Build the SET list from the tracked edits instead + // and render it through the shared dialect helper. let RowKey::Persisted(pk_keyvalues) = &row_key else { continue; // Drafts don't go through UPDATE }; @@ -642,15 +640,12 @@ impl TabChangeTracker { Some(s) => format!("{}.{}", quote_ident(driver_id, s), quote_ident(driver_id, table)), None => quote_ident(driver_id, table), }; - let sql = format!( - "UPDATE {} SET {} WHERE {}", - qualified, - set_clauses.join(", "), - where_clauses.join(" AND ") + let sql = build_update( + driver_id, + &qualified, + &set_clauses.join(", "), + &where_clauses.join(" AND "), ); - // Suppress unused-variable warning while keeping the same - // build pattern as build_full_row_update. - let _ = build_full_row_update; out.push((sql, params)); sources.push(StatementSource::Update { row_key: row_key.clone(), diff --git a/linux/crates/app/src/ui/app/row_ops.rs b/linux/crates/app/src/ui/app/row_ops.rs index 5c53ea0b9..2e1d1e2ee 100644 --- a/linux/crates/app/src/ui/app/row_ops.rs +++ b/linux/crates/app/src/ui/app/row_ops.rs @@ -29,6 +29,15 @@ impl App { self.dispatch_to_tab(tab_id, BrowseTabInput::SaveFailed(crate::tr!("No active connection"))); return; }; + // Drivers that cannot report a row count for UPDATE / DELETE + // return 0 for every statement, which the concurrency guard + // below would read as "every row vanished". Skip the guard for + // them rather than warn on every successful save. + let reports_rows_affected = self + .current_driver_id + .as_ref() + .and_then(|id| self.registry.get(id)) + .is_none_or(|driver| driver.reports_rows_affected()); self.set_row_op_in_flight(true); // Increment the in-flight counter so window-close blocks until // the transaction resolves. Decrement happens in the @@ -51,7 +60,9 @@ impl App { // — there's nothing to roll back — but the // user must hear about it so a phantom save // doesn't pass silently. - let warning = compute_concurrency_warning(&statements, &affected); + let warning = reports_rows_affected + .then(|| compute_concurrency_warning(&statements, &affected)) + .flatten(); sender_for_cmd.input(AppMsg::RowOpStarted); sender_for_cmd.input(AppMsg::WorkspaceSchemaWordsChanged); sender_for_cmd.input(AppMsg::SaveCompletedForTab(tab_id, warning)); diff --git a/linux/crates/core/src/driver.rs b/linux/crates/core/src/driver.rs index c98e26383..c6921857e 100644 --- a/linux/crates/core/src/driver.rs +++ b/linux/crates/core/src/driver.rs @@ -24,5 +24,15 @@ pub trait DatabaseDriver: Send + Sync { false } + /// Whether `ExecResult::rows_affected` carries a real count for + /// UPDATE and DELETE. The inline-edit Save path reads a zero count + /// as an optimistic-concurrency conflict, so a driver that cannot + /// produce one must say so or every successful save reports a lost + /// update. ClickHouse applies both as asynchronous mutations and + /// returns no row count for either. + fn reports_rows_affected(&self) -> bool { + true + } + async fn connect(&self, opts: ConnectOptions) -> Result, DriverError>; } diff --git a/linux/crates/core/src/sql_dialect.rs b/linux/crates/core/src/sql_dialect.rs index eee96185b..c54726b8c 100644 --- a/linux/crates/core/src/sql_dialect.rs +++ b/linux/crates/core/src/sql_dialect.rs @@ -30,6 +30,18 @@ pub fn placeholder_for(driver_id: &str, index: usize) -> String { } } +/// Render an `UPDATE`. ClickHouse only accepted standard `UPDATE` +/// syntax from 25.7; the spelling that works across every supported +/// release is `ALTER TABLE … UPDATE`, which the server applies as a +/// mutation. `qualified_table`, `set_clause` and `where_clause` are +/// pre-built SQL, not identifiers. +pub fn build_update(driver_id: &str, qualified_table: &str, set_clause: &str, where_clause: &str) -> String { + match driver_id { + "clickhouse" => format!("ALTER TABLE {qualified_table} UPDATE {set_clause} WHERE {where_clause}"), + _ => format!("UPDATE {qualified_table} SET {set_clause} WHERE {where_clause}"), + } +} + /// Render the `ORDER BY` and row-window tail of a paged `SELECT`, /// including the leading space. The two clauses are built together /// because SQL Server couples them: `OFFSET … FETCH` is defined as a @@ -90,12 +102,7 @@ pub fn build_single_cell_update( &mut params, ); - let sql = format!( - "UPDATE {} SET {} WHERE {}", - quote_ident(driver_id, table), - set_clause, - where_clause - ); + let sql = build_update(driver_id, "e_ident(driver_id, table), &set_clause, &where_clause); Ok((sql, params)) } @@ -152,11 +159,11 @@ pub fn build_full_row_update( &mut params, ); - let sql = format!( - "UPDATE {} SET {} WHERE {}", - quote_ident(driver_id, table), - set_clauses.join(", "), - where_clause + let sql = build_update( + driver_id, + "e_ident(driver_id, table), + &set_clauses.join(", "), + &where_clause, ); Ok((sql, params)) } @@ -363,6 +370,37 @@ mod tests { assert_eq!(params, vec![Value::Text("bob".into()), Value::Int(7)]); } + #[test] + fn single_cell_update_clickhouse() { + let columns = vec![col("id", true), col("name", false)]; + let original = vec![Value::Int(7), Value::Text("alice".into())]; + let (sql, params) = + build_single_cell_update("clickhouse", "u", &columns, &original, 1, Value::Text("bob".into())).unwrap(); + assert_eq!(sql, "ALTER TABLE `u` UPDATE `name` = ? WHERE `id` = ?"); + assert_eq!(params, vec![Value::Text("bob".into()), Value::Int(7)]); + } + + #[test] + fn full_row_update_clickhouse() { + let columns = vec![col("id", true), col("a", false), col("b", false)]; + let original = vec![Value::Int(1), Value::Text("x".into()), Value::Text("y".into())]; + let new_values = vec![Value::Int(1), Value::Text("x2".into()), Value::Text("y2".into())]; + let (sql, _) = build_full_row_update("clickhouse", "t", &columns, &original, &new_values).unwrap(); + assert_eq!(sql, "ALTER TABLE `t` UPDATE `a` = ?, `b` = ? WHERE `id` = ?"); + } + + #[test] + fn build_update_keeps_standard_syntax_for_other_dialects() { + assert_eq!( + build_update("postgres", "\"t\"", "\"a\" = $1", "\"id\" = $2"), + "UPDATE \"t\" SET \"a\" = $1 WHERE \"id\" = $2" + ); + assert_eq!( + build_update("clickhouse", "`t`", "`a` = ?", "`id` = ?"), + "ALTER TABLE `t` UPDATE `a` = ? WHERE `id` = ?" + ); + } + #[test] fn single_cell_update_sqlite() { let columns = vec![col("id", true), col("v", false)]; From 253cf2d096aab8d5f3fe01ef848cc9a48b2fa713 Mon Sep 17 00:00:00 2001 From: Ngo Quoc Dat Date: Wed, 29 Jul 2026 12:30:51 +0700 Subject: [PATCH 4/4] fix(drivers): correct ClickHouse type mapping, parameter binding, and result streaming Claude-Session: https://claude.ai/code/session_01Jp9Vhikm6vJFoiScLVAM4L --- .github/workflows/build-linux.yml | 2 + linux/Cargo.toml | 3 +- linux/README.md | 2 +- linux/ROADMAP.md | 2 +- linux/crates/drivers/clickhouse/Cargo.toml | 4 +- linux/crates/drivers/clickhouse/src/lib.rs | 902 +++++++++++++----- .../drivers/clickhouse/tests/integration.rs | 197 +++- linux/docs/adding-drivers.md | 19 +- 8 files changed, 856 insertions(+), 275 deletions(-) diff --git a/.github/workflows/build-linux.yml b/.github/workflows/build-linux.yml index 7443ffd19..4a937184b 100644 --- a/.github/workflows/build-linux.yml +++ b/.github/workflows/build-linux.yml @@ -96,3 +96,5 @@ jobs: run: cargo test --test integration -p tablepro-driver-postgres -- --include-ignored --test-threads=1 - name: MySQL integration tests run: cargo test --test integration -p tablepro-driver-mysql -- --include-ignored --test-threads=1 + - name: ClickHouse integration tests + run: cargo test --test integration -p tablepro-driver-clickhouse -- --include-ignored --test-threads=1 diff --git a/linux/Cargo.toml b/linux/Cargo.toml index ec5b95d19..9235fd13c 100644 --- a/linux/Cargo.toml +++ b/linux/Cargo.toml @@ -5,11 +5,11 @@ members = [ "crates/core", "crates/ssh", "crates/storage", + "crates/drivers/clickhouse", "crates/drivers/mssql", "crates/drivers/mysql", "crates/drivers/postgres", "crates/drivers/sqlite", - "crates/drivers/clickhouse", ] [workspace.package] @@ -30,6 +30,7 @@ russh = "0.55" secrecy = { version = "0.10", features = ["serde"] } tracing = "0.1" tracing-subscriber = { version = "0.3", features = ["env-filter"] } +clickhouse = { version = "0.15", default-features = false, features = ["rustls-tls"] } sqlx = { version = "0.8", default-features = false, features = ["runtime-tokio", "tls-rustls", "chrono", "rust_decimal", "uuid", "json"] } tiberius = { version = "0.12", default-features = false, features = ["tds73", "rustls", "chrono", "rust_decimal"] } chrono = { version = "0.4", default-features = false, features = ["clock", "serde"] } diff --git a/linux/README.md b/linux/README.md index bd8467afd..8e972a696 100644 --- a/linux/README.md +++ b/linux/README.md @@ -14,7 +14,7 @@ Phase 0 — foundation. The technology stack was validated by a 2-day spike in A | GUI toolkit | GTK4 4.14+ + libadwaita 1.5+ | | App architecture | [Relm4](https://relm4.org) — Elm-style components on gtk4-rs | | Async | tokio (DB drivers) bridged to glib main loop (UI) | -| DB drivers | sqlx (PG / MySQL / SQLite), fred (Redis), official mongodb crate, clickhouse-arrow, etc. | +| DB drivers | sqlx (PG / MySQL / SQLite), tiberius (MSSQL), official clickhouse and mongodb crates, fred (Redis), etc. | | Persistence | libsecret (passwords), gio::Settings (prefs), JSON files (connection metadata) | | Distribution | Flathub primary, .deb / .rpm / AppImage secondary | diff --git a/linux/ROADMAP.md b/linux/ROADMAP.md index 19f972c45..80c756676 100644 --- a/linux/ROADMAP.md +++ b/linux/ROADMAP.md @@ -232,7 +232,7 @@ This is where the project sits today. The app is interesting but **not productio ### Additional drivers (parallelizable, ~1 week each) -- [ ] ClickHouse via `clickhouse-arrow` +- [x] ClickHouse via the official `clickhouse` crate - [x] MSSQL via `tiberius` - [ ] Oracle via `oracle` crate (ODPI-C) - [ ] Redis via `fred` diff --git a/linux/crates/drivers/clickhouse/Cargo.toml b/linux/crates/drivers/clickhouse/Cargo.toml index 4997024ba..19f86307f 100644 --- a/linux/crates/drivers/clickhouse/Cargo.toml +++ b/linux/crates/drivers/clickhouse/Cargo.toml @@ -13,7 +13,7 @@ path = "src/lib.rs" tablepro-core = { path = "../../core" } async-trait.workspace = true chrono.workspace = true -clickhouse = { version = "0.15", default-features = false, features = ["rustls-tls"] } +clickhouse.workspace = true rust_decimal.workspace = true secrecy.workspace = true serde.workspace = true @@ -24,5 +24,5 @@ uuid.workspace = true [dev-dependencies] chrono.workspace = true secrecy.workspace = true -testcontainers = { version = "0.27", features = ["http_wait_plain"] } +testcontainers = { workspace = true, features = ["http_wait_plain"] } tokio = { workspace = true, features = ["macros", "rt-multi-thread"] } diff --git a/linux/crates/drivers/clickhouse/src/lib.rs b/linux/crates/drivers/clickhouse/src/lib.rs index ee02be9bc..54dc1f03c 100644 --- a/linux/crates/drivers/clickhouse/src/lib.rs +++ b/linux/crates/drivers/clickhouse/src/lib.rs @@ -1,18 +1,35 @@ +use std::time::Duration; + use async_trait::async_trait; use secrecy::ExposeSecret; use serde::Deserialize; use tablepro_core::{ ColumnInfo, ConnectOptions, Connection, DatabaseDriver, DriverError, ExecResult, ForeignKeyInfo, IndexInfo, - MAX_QUERY_ROWS, QueryResult, TableInfo, Value, + MAX_QUERY_ROWS, QueryResult, TableInfo, Value, sql_dialect::quote_ident, }; +const DRIVER_ID: &str = "clickhouse"; + +/// Applies to the reachability probe in `connect` and to `ping`. The +/// `clickhouse` crate drives hyper, which has no default timeout, so a +/// black-holed host would otherwise hang the connect dialog forever. +/// Queries stay unbounded: an analytical query that runs for minutes is +/// legitimate and the user can close the tab to drop it. +const PROBE_TIMEOUT: Duration = Duration::from_secs(5); + +/// Newline-delimited format carrying column names on line 1 and column +/// types on line 2. Preferred over `JSON` because it streams: rows +/// arrive one line at a time, so a `SELECT *` over a billion-row table +/// stops at `MAX_QUERY_ROWS` instead of buffering the whole result. +const ROW_FORMAT: &str = "JSONCompactEachRowWithNamesAndTypes"; + pub struct ClickhouseDriver; #[async_trait] impl DatabaseDriver for ClickhouseDriver { fn id(&self) -> &'static str { - "clickhouse" + DRIVER_ID } fn display_name(&self) -> &'static str { @@ -23,12 +40,22 @@ impl DatabaseDriver for ClickhouseDriver { 8123 } + fn reports_rows_affected(&self) -> bool { + false + } + async fn connect(&self, opts: ConnectOptions) -> Result, DriverError> { let scheme = if opts.use_tls { "https" } else { "http" }; let url = format!("{scheme}://{}:{}", opts.host, opts.port); let mut client = clickhouse::Client::default() .with_url(url) - .with_product_info("tablepro-linux", env!("CARGO_PKG_VERSION")); + .with_product_info("tablepro-linux", env!("CARGO_PKG_VERSION")) + // `ALTER TABLE … UPDATE` and `DELETE FROM` are queued as + // asynchronous mutations by default, so a save would return + // before the rows changed and the grid would refetch stale + // values. Wait for the mutation to finish on the server we + // are talking to. + .with_setting("mutations_sync", "1"); if !opts.username.is_empty() { client = client.with_user(opts.username); } @@ -39,29 +66,61 @@ impl DatabaseDriver for ClickhouseDriver { client = client.with_database(opts.database.clone()); } - // Prove the HTTP endpoint is reachable and credentials work. - client - .query("SELECT 1") - .execute() - .await - .map_err(map_clickhouse_error)?; + let probe = client.query("SELECT 1").execute(); + match tokio::time::timeout(PROBE_TIMEOUT, probe).await { + Ok(result) => result.map_err(map_clickhouse_error)?, + Err(_) => return Err(DriverError::ConnectionRefused), + } - Ok(Box::new(ClickhouseConnection { - client, - database: if opts.database.is_empty() { - "default".into() - } else { - opts.database - }, - })) + let database = if opts.database.is_empty() { + resolve_current_database(&client).await + } else { + opts.database + }; + Ok(Box::new(ClickhouseConnection { client, database })) } } +/// The catalog queries filter `system.tables` / `system.columns` by an +/// explicit database name, so an empty `ConnectOptions::database` has to +/// resolve to whatever the server picked for this user rather than being +/// assumed to be `default`. +async fn resolve_current_database(client: &clickhouse::Client) -> String { + client + .query("SELECT currentDatabase()") + .fetch_one::() + .await + .unwrap_or_else(|_| "default".into()) +} + struct ClickhouseConnection { client: clickhouse::Client, database: String, } +impl ClickhouseConnection { + fn database_of<'a>(&'a self, schema: Option<&'a str>) -> &'a str { + schema.unwrap_or(self.database.as_str()) + } + + /// Runs a statement and reports the row count the server put in + /// `X-ClickHouse-Summary`. Meaningful for INSERT; mutations report + /// nothing, which is why the driver declares + /// `reports_rows_affected() == false`. + async fn execute_reporting(&self, sql: &str) -> Result { + let mut cursor = self + .client + .query(&escape_bind_markers(sql)) + // The summary header is sent before the body, so its counts + // are only complete once the server has finished the query. + .with_setting("wait_end_of_query", "1") + .fetch_bytes(ROW_FORMAT) + .map_err(map_clickhouse_error)?; + while cursor.next().await.map_err(map_clickhouse_error)?.is_some() {} + Ok(cursor.summary().and_then(|s| s.written_rows()).unwrap_or(0)) + } +} + #[async_trait] impl Connection for ClickhouseConnection { async fn list_tables(&self) -> Result, DriverError> { @@ -76,11 +135,11 @@ impl Connection for ClickhouseConnection { .query( "SELECT database, name FROM system.tables - WHERE database = currentDatabase() + WHERE database = ? AND is_temporary = 0 - AND engine NOT LIKE '%View' ORDER BY name", ) + .bind(self.database.as_str()) .fetch_all::() .await .map_err(map_clickhouse_error)?; @@ -105,7 +164,6 @@ impl Connection for ClickhouseConnection { default_expression: String, } - let db = schema.unwrap_or(self.database.as_str()); let rows = self .client .query( @@ -120,7 +178,7 @@ impl Connection for ClickhouseConnection { AND table = ? ORDER BY position", ) - .bind(db) + .bind(self.database_of(schema)) .bind(table) .fetch_all::() .await @@ -129,20 +187,21 @@ impl Connection for ClickhouseConnection { Ok(rows .into_iter() .map(|r| { - let is_generated = matches!( - r.default_kind.as_str(), - "MATERIALIZED" | "ALIAS" | "EPHEMERAL" - ); + let is_generated = matches!(r.default_kind.as_str(), "MATERIALIZED" | "ALIAS" | "EPHEMERAL"); let default_value = if r.default_expression.is_empty() { None } else { Some(r.default_expression) }; - let nullable = r.data_type.starts_with("Nullable("); ColumnInfo { + nullable: type_is_nullable(&r.data_type), name: r.name, data_type: r.data_type, - nullable, + // A MergeTree sorting key is the closest thing to a + // row identifier ClickHouse has, but it is not + // unique. The edit path needs *some* key to build a + // WHERE from; see `fetch_indexes` for why it is not + // advertised as unique. primary_key: r.is_in_primary_key != 0, is_auto_increment: false, default_value, @@ -159,13 +218,13 @@ impl Connection for ClickhouseConnection { offset: u64, limit: u64, ) -> Result { - let qualified = qualify(schema.unwrap_or(self.database.as_str()), table); + let qualified = qualify(self.database_of(schema), table); let sql = format!("SELECT * FROM {qualified} LIMIT {limit} OFFSET {offset}"); - self.query(&sql).await + fetch_result(&self.client, &sql, limit as usize).await } async fn query(&self, sql: &str) -> Result { - query_json(&self.client, sql).await + fetch_result(&self.client, sql, MAX_QUERY_ROWS).await } async fn query_params(&self, sql: &str, params: &[Value]) -> Result { @@ -177,14 +236,8 @@ impl Connection for ClickhouseConnection { } async fn execute(&self, sql: &str) -> Result { - self.client - .query(sql) - .execute() - .await - .map_err(map_clickhouse_error)?; - // ClickHouse HTTP does not return rows_affected for arbitrary - // mutations reliably; report 0 rather than invent a count. - Ok(ExecResult { rows_affected: 0 }) + let rows_affected = self.execute_reporting(sql).await?; + Ok(ExecResult { rows_affected }) } async fn execute_params(&self, sql: &str, params: &[Value]) -> Result { @@ -193,9 +246,11 @@ impl Connection for ClickhouseConnection { } async fn execute_in_transaction(&self, statements: &[(String, Vec)]) -> Result, DriverError> { - // ClickHouse has no multi-statement ACID transaction for DML. - // Run statements sequentially; on failure surface which index - // failed. Callers already treat this as best-effort for CH. + // ClickHouse has no multi-statement ACID transaction for DML, so + // this cannot honour the trait's rollback contract: statements + // before the failing one stay applied. The index in the returned + // error is still the one that failed, which is what the caller + // uses to flag the offending row. let mut affected = Vec::with_capacity(statements.len()); for (i, (sql, params)) in statements.iter().enumerate() { match self.execute_params(sql, params).await { @@ -218,7 +273,6 @@ impl Connection for ClickhouseConnection { primary_key: String, } - let db = schema.unwrap_or(self.database.as_str()); let rows = self .client .query( @@ -228,7 +282,7 @@ impl Connection for ClickhouseConnection { AND name = ? LIMIT 1", ) - .bind(db) + .bind(self.database_of(schema)) .bind(table) .fetch_all::() .await @@ -237,22 +291,18 @@ impl Connection for ClickhouseConnection { let Some(row) = rows.into_iter().next() else { return Ok(Vec::new()); }; - if row.primary_key.is_empty() { - return Ok(Vec::new()); - } - let columns: Vec = row - .primary_key - .split(',') - .map(|s| s.trim().trim_matches('`').trim_matches('"').to_string()) - .filter(|s| !s.is_empty()) - .collect(); + let columns = split_key_expression(&row.primary_key); if columns.is_empty() { return Ok(Vec::new()); } Ok(vec![IndexInfo { - name: format!("{}_pk", row.name), + name: format!("{}_sorting_key", row.name), columns, - unique: true, + // A MergeTree primary key is a sparse sorting key, not a + // uniqueness constraint: duplicate keys are legal and common. + // Advertising it as unique would tell the UI that an UPDATE + // built from these columns touches exactly one row. + unique: false, primary: true, }]) } @@ -267,11 +317,11 @@ impl Connection for ClickhouseConnection { } async fn ping(&self) -> Result<(), DriverError> { - self.client - .query("SELECT 1") - .execute() - .await - .map_err(map_clickhouse_error) + let probe = self.client.query("SELECT 1").execute(); + match tokio::time::timeout(PROBE_TIMEOUT, probe).await { + Ok(result) => result.map_err(map_clickhouse_error), + Err(_) => Err(DriverError::Disconnected), + } } async fn close(self: Box) -> Result<(), DriverError> { @@ -279,47 +329,78 @@ impl Connection for ClickhouseConnection { } } -#[derive(Debug, Deserialize)] -struct JsonMeta { - name: String, - #[serde(rename = "type")] - data_type: String, +/// Reads a `ROW_FORMAT` response one line at a time so the caller can +/// stop at `max_rows` without materialising the rest of the result. +struct LineReader { + cursor: clickhouse::query::BytesCursor, + buf: Vec, + consumed: usize, + eof: bool, +} + +impl LineReader { + fn new(cursor: clickhouse::query::BytesCursor) -> Self { + Self { + cursor, + buf: Vec::new(), + consumed: 0, + eof: false, + } + } + + async fn next_line(&mut self) -> Result>, DriverError> { + loop { + if let Some(idx) = self.buf[self.consumed..].iter().position(|b| *b == b'\n') { + let end = self.consumed + idx; + let line = self.buf[self.consumed..end].to_vec(); + self.consumed = end + 1; + return Ok(Some(line)); + } + if self.eof { + let rest = self.buf[self.consumed..].to_vec(); + self.consumed = self.buf.len(); + return Ok((!rest.is_empty()).then_some(rest)); + } + match self.cursor.next().await.map_err(map_clickhouse_error)? { + Some(chunk) => { + self.buf.drain(..self.consumed); + self.consumed = 0; + self.buf.extend_from_slice(&chunk); + } + None => self.eof = true, + } + } + } } -#[derive(Debug, Deserialize)] -struct JsonResponse { - #[serde(default)] - meta: Vec, - #[serde(default)] - data: Vec>, - #[serde(default)] - rows: u64, +fn parse_line(line: &[u8]) -> Result { + serde_json::from_slice(line).map_err(|e| DriverError::Internal(format!("clickhouse response parse: {e}"))) } -async fn query_json(client: &clickhouse::Client, sql: &str) -> Result { - let mut cursor = client - .query(sql) - .fetch_bytes("JSON") +async fn fetch_result(client: &clickhouse::Client, sql: &str, max_rows: usize) -> Result { + let cursor = client + .query(&escape_bind_markers(sql)) + .fetch_bytes(ROW_FORMAT) .map_err(map_clickhouse_error)?; - let bytes = cursor.collect().await.map_err(map_clickhouse_error)?; - if bytes.is_empty() { - return Ok(QueryResult { - columns: Vec::new(), - rows: Vec::new(), - truncated: false, - }); - } - - let parsed: JsonResponse = serde_json::from_slice(&bytes) - .map_err(|e| DriverError::Internal(format!("clickhouse JSON parse: {e}")))?; - - let columns: Vec = parsed - .meta - .iter() - .map(|m| ColumnInfo { - name: m.name.clone(), - data_type: m.data_type.clone(), - nullable: m.data_type.starts_with("Nullable(") || m.data_type == "Nullable", + let mut reader = LineReader::new(cursor); + + // A statement with no result set (DDL, INSERT) sends an empty body. + let Some(names_line) = reader.next_line().await? else { + return Ok(empty_result()); + }; + let names: Vec = parse_line(&names_line)?; + let Some(types_line) = reader.next_line().await? else { + return Ok(empty_result()); + }; + let types: Vec = parse_line(&types_line)?; + + let columns: Vec = names + .into_iter() + .zip(types) + .map(|(name, data_type)| ColumnInfo { + nullable: type_is_nullable(&data_type), + name, + data_type, primary_key: false, is_auto_increment: false, default_value: None, @@ -327,17 +408,27 @@ async fn query_json(client: &clickhouse::Client, sql: &str) -> Result> = Vec::new(); + let mut truncated = false; + while let Some(line) = reader.next_line().await? { + if line.is_empty() { + continue; + } + // Read one line past the cap so the flag reflects rows the + // server actually had, not a result that happens to land on it. + if rows.len() == max_rows { + truncated = true; + break; + } + let raw: Vec = parse_line(&line)?; let mut row = Vec::with_capacity(columns.len()); - for col in &columns { - let raw = obj.get(&col.name).cloned().unwrap_or(serde_json::Value::Null); - row.push(json_to_value(raw, &col.data_type)); + for (i, col) in columns.iter().enumerate() { + let cell = raw.get(i).cloned().unwrap_or(serde_json::Value::Null); + row.push(json_to_value(cell, &col.data_type)); } rows.push(row); } - let truncated = parsed.rows as usize > rows.len() || rows.len() >= MAX_QUERY_ROWS; Ok(QueryResult { columns, rows, @@ -345,218 +436,366 @@ async fn query_json(client: &clickhouse::Client, sql: &str) -> Result QueryResult { + QueryResult { + columns: Vec::new(), + rows: Vec::new(), + truncated: false, + } +} + +/// Peels the wrappers that do not change how a value is encoded, then +/// drops any type arguments. `LowCardinality(Nullable(String))` becomes +/// `String`, `Decimal(9, 2)` becomes `Decimal`, `DateTime64(3, 'UTC')` +/// becomes `DateTime64`. Matching the raw name instead would miss every +/// parameterised type, since the server always reports its arguments. +fn base_type(type_name: &str) -> &str { + let mut t = type_name.trim(); + while let Some(inner) = unwrap_type(t, "LowCardinality").or_else(|| unwrap_type(t, "Nullable")) { + t = inner; + } + t.split('(').next().unwrap_or(t).trim() +} + +/// `Nullable(T)` survives inside `LowCardinality`, so the wrapper has to +/// come off before the nullability test. +fn type_is_nullable(type_name: &str) -> bool { + let t = type_name.trim(); + let inner = unwrap_type(t, "LowCardinality").unwrap_or(t); + unwrap_type(inner, "Nullable").is_some() +} + +fn unwrap_type<'a>(type_name: &'a str, wrapper: &str) -> Option<&'a str> { + type_name + .strip_prefix(wrapper)? + .strip_prefix('(')? + .strip_suffix(')') + .map(str::trim) +} + fn json_to_value(raw: serde_json::Value, type_name: &str) -> Value { if raw.is_null() { return Value::Null; } - let base = strip_nullable(type_name); - match base { + match base_type(type_name) { "Bool" => raw .as_bool() .map(Value::Bool) .or_else(|| raw.as_u64().map(|n| Value::Bool(n != 0))) - .unwrap_or_else(|| Value::Text(raw.to_string())), + .unwrap_or_else(|| fallback_text(&raw)), "Int8" | "Int16" | "Int32" | "Int64" | "UInt8" | "UInt16" | "UInt32" | "UInt64" => raw .as_i64() .or_else(|| raw.as_u64().and_then(|n| i64::try_from(n).ok())) + .or_else(|| raw.as_str().and_then(|s| s.parse::().ok())) .map(Value::Int) - .or_else(|| { - raw.as_str() - .and_then(|s| s.parse::().ok()) - .map(Value::Int) - }) - .unwrap_or_else(|| Value::Text(raw.to_string())), + // Int128 / UInt64 past i64::MAX have no lossless `Value`. + // Text keeps every digit; Float would round. + .unwrap_or_else(|| fallback_text(&raw)), "Float32" | "Float64" => raw .as_f64() + .or_else(|| raw.as_str().and_then(|s| s.parse::().ok())) .map(Value::Float) - .or_else(|| { - raw.as_str() - .and_then(|s| s.parse::().ok()) - .map(Value::Float) - }) - .unwrap_or_else(|| Value::Text(raw.to_string())), + .unwrap_or_else(|| fallback_text(&raw)), "Decimal" | "Decimal32" | "Decimal64" | "Decimal128" | "Decimal256" => raw .as_str() .and_then(|s| s.parse::().ok()) - .or_else(|| { - raw.as_f64() - .and_then(|f| rust_decimal::Decimal::try_from(f).ok()) - }) + .or_else(|| raw.as_f64().and_then(|f| rust_decimal::Decimal::try_from(f).ok())) .map(Value::Decimal) - .unwrap_or_else(|| Value::Text(raw.to_string())), - "Date" => raw + .unwrap_or_else(|| fallback_text(&raw)), + "Date" | "Date32" => raw .as_str() .and_then(|s| chrono::NaiveDate::parse_from_str(s, "%Y-%m-%d").ok()) .map(Value::Date) - .unwrap_or_else(|| Value::Text(raw.as_str().unwrap_or(&raw.to_string()).to_string())), - "DateTime" | "DateTime64" => { - if let Some(s) = raw.as_str() { - if let Ok(dt) = chrono::DateTime::parse_from_rfc3339(s) { - return Value::TimestampTz(dt.with_timezone(&chrono::Utc)); - } - if let Ok(dt) = chrono::NaiveDateTime::parse_from_str(s, "%Y-%m-%d %H:%M:%S") { - return Value::DateTime(dt); - } - if let Ok(dt) = chrono::NaiveDateTime::parse_from_str(s, "%Y-%m-%d %H:%M:%S%.f") { - return Value::DateTime(dt); - } - return Value::Text(s.to_string()); - } - Value::Text(raw.to_string()) - } + .unwrap_or_else(|| fallback_text(&raw)), + "DateTime" | "DateTime64" => parse_datetime(&raw), "UUID" => raw .as_str() .and_then(|s| s.parse::().ok()) .map(Value::Uuid) - .unwrap_or_else(|| Value::Text(raw.as_str().unwrap_or(&raw.to_string()).to_string())), - "String" | "FixedString" | "LowCardinality(String)" => { - Value::Text(raw.as_str().unwrap_or(&raw.to_string()).to_string()) - } - t if t.starts_with("Array") || t.starts_with("Map") || t.starts_with("Tuple") || t == "JSON" - || t.starts_with("Object") => - { - Value::Json(raw) - } - _ => { - if let Some(s) = raw.as_str() { - Value::Text(s.to_string()) - } else if let Some(n) = raw.as_i64() { - Value::Int(n) - } else if let Some(f) = raw.as_f64() { - Value::Float(f) - } else if let Some(b) = raw.as_bool() { - Value::Bool(b) - } else { - Value::Json(raw) - } - } + .unwrap_or_else(|| fallback_text(&raw)), + "String" | "FixedString" | "Enum8" | "Enum16" | "IPv4" | "IPv6" => fallback_text(&raw), + "Array" | "Map" | "Tuple" | "Nested" | "JSON" | "Object" | "Variant" | "Dynamic" => Value::Json(raw), + _ => fallback_text(&raw), } } -fn strip_nullable(type_name: &str) -> &str { - type_name - .strip_prefix("Nullable(") - .and_then(|s| s.strip_suffix(')')) - .unwrap_or(type_name) +fn parse_datetime(raw: &serde_json::Value) -> Value { + let Some(s) = raw.as_str() else { + return fallback_text(raw); + }; + if let Ok(dt) = chrono::DateTime::parse_from_rfc3339(s) { + return Value::TimestampTz(dt.with_timezone(&chrono::Utc)); + } + // `%.f` also matches a whole-second timestamp, so one pattern covers + // both `DateTime` and every `DateTime64` precision. + if let Ok(dt) = chrono::NaiveDateTime::parse_from_str(s, "%Y-%m-%d %H:%M:%S%.f") { + return Value::DateTime(dt); + } + Value::Text(s.to_string()) +} + +/// A JSON string keeps its own text; anything else keeps its JSON +/// spelling so no digits are lost on the way to the grid. +fn fallback_text(raw: &serde_json::Value) -> Value { + match raw.as_str() { + Some(s) => Value::Text(s.to_string()), + None => Value::Text(raw.to_string()), + } +} + +/// Splits a `primary_key` expression from `system.tables` on top-level +/// commas only. A naive split breaks `toYYYYMM(d), id` into +/// `toYYYYMM(d` and `d), id`. +fn split_key_expression(expression: &str) -> Vec { + let mut parts = Vec::new(); + let mut depth = 0usize; + let mut current = String::new(); + for ch in expression.chars() { + match ch { + '(' => { + depth += 1; + current.push(ch); + } + ')' => { + depth = depth.saturating_sub(1); + current.push(ch); + } + ',' if depth == 0 => { + parts.push(std::mem::take(&mut current)); + } + _ => current.push(ch), + } + } + parts.push(current); + parts + .into_iter() + .map(|s| s.trim().trim_matches('`').trim_matches('"').to_string()) + .filter(|s| !s.is_empty()) + .collect() } fn qualify(schema: &str, table: &str) -> String { - format!("{}.{}", quote_ident(schema), quote_ident(table)) + format!("{}.{}", quote_ident(DRIVER_ID, schema), quote_ident(DRIVER_ID, table)) } -fn quote_ident(name: &str) -> String { - format!("`{}`", name.replace('`', "``")) +#[derive(Clone, Copy, PartialEq, Eq)] +enum ScanState { + Sql, + SingleQuote, + DoubleQuote, + Backtick, + LineComment, + BlockComment, } +/// Inlines `Value`s as escaped SQL literals. ClickHouse's HTTP interface +/// has no positional binding, so the driver has to do the substitution +/// itself, which means it also has to know where SQL ends and a string +/// literal begins: a `?` inside `'what?'` is data, not a placeholder. +/// Filters carry user-typed text (`FilterOp::Raw` concatenates a whole +/// clause), so scanning blind would let one apostrophe shift every +/// binding after it. fn bind_placeholders(sql: &str, params: &[Value]) -> Result { let mut out = String::with_capacity(sql.len() + params.len() * 8); - let mut param_idx = 0; + let mut used = vec![false; params.len()]; + let mut next_positional = 0usize; + let mut state = ScanState::Sql; let mut chars = sql.chars().peekable(); + while let Some(ch) = chars.next() { - if ch == '?' { - let Some(value) = params.get(param_idx) else { - return Err(DriverError::Internal(format!( - "not enough bind parameters: need at least {}", - param_idx + 1 - ))); - }; - out.push_str(&literal(value)); - param_idx += 1; - continue; - } - if ch == '$' { - let mut digits = String::new(); - while let Some(d) = chars.peek().copied().filter(|c| c.is_ascii_digit()) { - digits.push(d); - chars.next(); - } - if !digits.is_empty() { - let n: usize = digits - .parse() - .map_err(|_| DriverError::Internal(format!("bad placeholder ${digits}")))?; - let Some(value) = params.get(n.saturating_sub(1)) else { - return Err(DriverError::Internal(format!( - "bind parameter ${n} out of range (have {})", - params.len() - ))); + match state { + ScanState::SingleQuote | ScanState::DoubleQuote | ScanState::Backtick => { + out.push(ch); + let closer = match state { + ScanState::SingleQuote => '\'', + ScanState::DoubleQuote => '"', + _ => '`', }; - out.push_str(&literal(value)); - continue; + if ch == '\\' { + // ClickHouse honours backslash escapes inside every + // quoted form, so the next character is literal. + if let Some(escaped) = chars.next() { + out.push(escaped); + } + } else if ch == closer { + // A doubled quote is an escaped quote, not a close. + if chars.peek() == Some(&closer) { + out.push(closer); + chars.next(); + } else { + state = ScanState::Sql; + } + } } - out.push('$'); - continue; + ScanState::LineComment => { + out.push(ch); + if ch == '\n' { + state = ScanState::Sql; + } + } + ScanState::BlockComment => { + out.push(ch); + if ch == '*' && chars.peek() == Some(&'/') { + out.push('/'); + chars.next(); + state = ScanState::Sql; + } + } + ScanState::Sql => match ch { + '\'' | '"' | '`' => { + out.push(ch); + state = match ch { + '\'' => ScanState::SingleQuote, + '"' => ScanState::DoubleQuote, + _ => ScanState::Backtick, + }; + } + '-' if chars.peek() == Some(&'-') => { + out.push_str("--"); + chars.next(); + state = ScanState::LineComment; + } + '/' if chars.peek() == Some(&'*') => { + out.push_str("/*"); + chars.next(); + state = ScanState::BlockComment; + } + '?' => { + let Some(value) = params.get(next_positional) else { + return Err(DriverError::Internal(format!( + "not enough bind parameters: need at least {}", + next_positional + 1 + ))); + }; + out.push_str(&literal(value)?); + used[next_positional] = true; + next_positional += 1; + } + '$' => { + let mut digits = String::new(); + while let Some(d) = chars.peek().copied().filter(char::is_ascii_digit) { + digits.push(d); + chars.next(); + } + if digits.is_empty() { + out.push('$'); + continue; + } + let n: usize = digits + .parse() + .map_err(|_| DriverError::Internal(format!("bad placeholder ${digits}")))?; + let Some(index) = n.checked_sub(1) else { + return Err(DriverError::Internal("bind placeholders start at $1".into())); + }; + let Some(value) = params.get(index) else { + return Err(DriverError::Internal(format!( + "bind parameter ${n} out of range (have {})", + params.len() + ))); + }; + out.push_str(&literal(value)?); + used[index] = true; + } + _ => out.push(ch), + }, } - out.push(ch); } - if param_idx > 0 && param_idx != params.len() { + + if let Some(unused) = used.iter().position(|u| !u) { return Err(DriverError::Internal(format!( - "bind parameter count mismatch: used {param_idx}, have {}", + "bind parameter {} of {} was never referenced", + unused + 1, params.len() ))); } Ok(out) } -fn literal(value: &Value) -> String { - match value { +fn literal(value: &Value) -> Result { + let rendered = match value { Value::Null => "NULL".into(), Value::Bool(b) => if *b { "true" } else { "false" }.into(), Value::Int(i) => i.to_string(), Value::Float(f) => { - if f.is_finite() { - f.to_string() + // ClickHouse spells these out; silently substituting NULL + // would write a different value than the user typed. + if f.is_nan() { + "nan".into() + } else if f.is_infinite() { + if f.is_sign_negative() { "-inf" } else { "inf" }.into() } else { - "NULL".into() + f.to_string() } } Value::Text(s) => format!("'{}'", escape_str(s)), - Value::Bytes(b) => format!("unhex('{}')", hex::encode(b)), - Value::Date(d) => format!("toDate('{}')", d), - Value::Time(t) => format!("'{}'", t), + Value::Bytes(b) => format!("unhex('{}')", hex_encode(b)), + Value::Date(d) => format!("toDate('{}')", d.format("%Y-%m-%d")), + Value::Time(t) => format!("'{}'", t.format("%H:%M:%S%.f")), Value::DateTime(dt) => format!("toDateTime('{}')", dt.format("%Y-%m-%d %H:%M:%S")), - Value::TimestampTz(ts) => format!( - "toDateTime64('{}', 3)", - ts.format("%Y-%m-%d %H:%M:%S%.3f") - ), - Value::Decimal(d) => d.to_string(), - Value::Uuid(u) => format!("toUUID('{}')", u), + Value::TimestampTz(ts) => format!("toDateTime64('{}', 3)", ts.format("%Y-%m-%d %H:%M:%S%.3f")), + Value::Decimal(d) => format!("toDecimal128('{d}', {})", d.scale()), + Value::Uuid(u) => format!("toUUID('{u}')"), Value::Json(j) => format!("'{}'", escape_str(&j.to_string())), - } + }; + Ok(rendered) } fn escape_str(s: &str) -> String { s.replace('\\', "\\\\").replace('\'', "\\'") } -fn map_clickhouse_error(err: clickhouse::error::Error) -> DriverError { - let msg = err.to_string(); - let lower = msg.to_lowercase(); - if lower.contains("connection refused") || lower.contains("connect error") { - return DriverError::ConnectionRefused; - } - if lower.contains("authentication") || lower.contains("password is incorrect") || lower.contains("code: 516") - { - return DriverError::AuthFailed; - } - if lower.contains("tls") || lower.contains("certificate") { - return DriverError::Tls(msg); - } - DriverError::Query { - message: msg, - sqlstate: None, +/// The `clickhouse` crate treats every `?` in a query template as one of +/// its own bind markers and `??` as an escaped literal. Statements this +/// driver sends are already fully rendered, so any `?` left in them is +/// data: a value inlined by `bind_placeholders`, or a question mark the +/// user typed in the SQL editor. Without escaping, the crate rejects the +/// query as having unbound arguments before it ever reaches the server. +fn escape_bind_markers(sql: &str) -> String { + sql.replace('?', "??") +} + +fn hex_encode(bytes: &[u8]) -> String { + const LUT: &[u8; 16] = b"0123456789abcdef"; + let mut out = String::with_capacity(bytes.len() * 2); + for b in bytes { + out.push(LUT[(b >> 4) as usize] as char); + out.push(LUT[(b & 0xf) as usize] as char); } + out } -// Minimal hex encoder so we do not pull the `hex` crate for one call site. -mod hex { - pub fn encode(bytes: &[u8]) -> String { - const LUT: &[u8; 16] = b"0123456789abcdef"; - let mut out = String::with_capacity(bytes.len() * 2); - for b in bytes { - out.push(LUT[(b >> 4) as usize] as char); - out.push(LUT[(b & 0xf) as usize] as char); +/// ClickHouse error codes that mean the credentials were rejected. +/// 192 UNKNOWN_USER, 193 WRONG_PASSWORD, 194 REQUIRED_PASSWORD, +/// 497 ACCESS_DENIED, 516 AUTHENTICATION_FAILED. +const AUTH_CODES: [&str; 5] = ["code: 192", "code: 193", "code: 194", "code: 497", "code: 516"]; + +fn map_clickhouse_error(err: clickhouse::error::Error) -> DriverError { + let msg = err.to_string(); + let lower = msg.to_lowercase(); + match &err { + // Transport failures are the only place a TLS or refused-connect + // diagnosis can come from. Matching those words against a server + // response would misclassify a query that merely mentions them. + clickhouse::error::Error::Network(_) => { + if lower.contains("certificate") || lower.contains("tls") || lower.contains("handshake") { + DriverError::Tls(msg) + } else if lower.contains("connection refused") || lower.contains("connect error") { + DriverError::ConnectionRefused + } else { + DriverError::Disconnected + } + } + clickhouse::error::Error::TimedOut => DriverError::Disconnected, + _ => { + if AUTH_CODES.iter().any(|code| lower.contains(code)) { + DriverError::AuthFailed + } else { + DriverError::Query { + message: msg, + sqlstate: None, + } + } } - out } } @@ -570,55 +809,202 @@ mod tests { assert_eq!(d.id(), "clickhouse"); assert_eq!(d.display_name(), "ClickHouse"); assert_eq!(d.default_port(), 8123); + assert!(!d.reports_rows_affected()); + } + + #[test] + fn qualify_escapes_backticks() { + assert_eq!(qualify("db", "users"), "`db`.`users`"); + assert_eq!(qualify("db", "a`b"), "`db`.`a``b`"); + } + + #[test] + fn base_type_strips_arguments_and_wrappers() { + assert_eq!(base_type("String"), "String"); + assert_eq!(base_type("Decimal(9, 2)"), "Decimal"); + assert_eq!(base_type("DateTime64(3, 'UTC')"), "DateTime64"); + assert_eq!(base_type("FixedString(16)"), "FixedString"); + assert_eq!(base_type("Nullable(Decimal(18, 4))"), "Decimal"); + assert_eq!(base_type("LowCardinality(Nullable(String))"), "String"); + assert_eq!(base_type("Array(Nullable(String))"), "Array"); + assert_eq!(base_type("Map(String, UInt64)"), "Map"); + } + + #[test] + fn nullability_survives_low_cardinality() { + assert!(!type_is_nullable("String")); + assert!(type_is_nullable("Nullable(String)")); + assert!(type_is_nullable("LowCardinality(Nullable(String))")); + assert!(!type_is_nullable("LowCardinality(String)")); + // The inner Nullable belongs to the element, not the column. + assert!(!type_is_nullable("Array(Nullable(String))")); } #[test] - fn quote_ident_escapes_backticks() { - assert_eq!(quote_ident("users"), "`users`"); - assert_eq!(quote_ident("a`b"), "`a``b`"); + fn parameterised_types_decode_to_typed_values() { + assert_eq!( + json_to_value(serde_json::json!("12.34"), "Decimal(9, 2)"), + Value::Decimal("12.34".parse().unwrap()) + ); + assert_eq!( + json_to_value(serde_json::json!("2024-06-15 08:30:00.123"), "DateTime64(3)"), + Value::DateTime( + chrono::NaiveDate::from_ymd_opt(2024, 6, 15) + .unwrap() + .and_hms_milli_opt(8, 30, 0, 123) + .unwrap() + ) + ); + assert_eq!( + json_to_value(serde_json::json!("abc"), "LowCardinality(Nullable(String))"), + Value::Text("abc".into()) + ); + assert_eq!( + json_to_value(serde_json::json!("2024-06-15 08:30:00"), "DateTime"), + Value::DateTime( + chrono::NaiveDate::from_ymd_opt(2024, 6, 15) + .unwrap() + .and_hms_opt(8, 30, 0) + .unwrap() + ) + ); + } + + #[test] + fn json_to_value_maps_common_types() { + assert_eq!(json_to_value(serde_json::json!(true), "Bool"), Value::Bool(true)); + assert_eq!(json_to_value(serde_json::json!(42), "Int64"), Value::Int(42)); + assert_eq!( + json_to_value(serde_json::json!("hello"), "String"), + Value::Text("hello".into()) + ); + assert_eq!(json_to_value(serde_json::Value::Null, "Nullable(String)"), Value::Null); + assert_eq!( + json_to_value(serde_json::json!("2024-06-15"), "Date"), + Value::Date(chrono::NaiveDate::from_ymd_opt(2024, 6, 15).unwrap()) + ); + assert_eq!( + json_to_value(serde_json::json!([1, 2]), "Array(UInt8)"), + Value::Json(serde_json::json!([1, 2])) + ); } #[test] fn bind_question_marks() { let sql = bind_placeholders( - "UPDATE t SET a = ? WHERE id = ?", + "ALTER TABLE t UPDATE a = ? WHERE id = ?", &[Value::Text("x".into()), Value::Int(1)], ) .unwrap(); - assert_eq!(sql, "UPDATE t SET a = 'x' WHERE id = 1"); + assert_eq!(sql, "ALTER TABLE t UPDATE a = 'x' WHERE id = 1"); } #[test] fn bind_dollar_placeholders() { let sql = bind_placeholders( - "UPDATE t SET a = $1 WHERE id = $2", + "ALTER TABLE t UPDATE a = $1 WHERE id = $2", &[Value::Text("x".into()), Value::Int(1)], ) .unwrap(); - assert_eq!(sql, "UPDATE t SET a = 'x' WHERE id = 1"); + assert_eq!(sql, "ALTER TABLE t UPDATE a = 'x' WHERE id = 1"); } #[test] - fn json_to_value_maps_common_types() { - assert_eq!(json_to_value(serde_json::json!(true), "Bool"), Value::Bool(true)); - assert_eq!(json_to_value(serde_json::json!(42), "Int64"), Value::Int(42)); + fn placeholders_inside_literals_are_left_alone() { + let sql = bind_placeholders("SELECT * FROM t WHERE note = 'what? $1' AND id = ?", &[Value::Int(7)]).unwrap(); + assert_eq!(sql, "SELECT * FROM t WHERE note = 'what? $1' AND id = 7"); + } + + #[test] + fn placeholders_inside_comments_and_identifiers_are_left_alone() { + let sql = bind_placeholders( + "SELECT `we?rd`, /* $1 ? */ x -- ?\n FROM t WHERE id = ?", + &[Value::Int(3)], + ) + .unwrap(); + assert_eq!(sql, "SELECT `we?rd`, /* $1 ? */ x -- ?\n FROM t WHERE id = 3"); + } + + #[test] + fn escaped_quote_does_not_end_a_literal() { + let sql = bind_placeholders("SELECT * FROM t WHERE a = 'it''s ?' AND b = ?", &[Value::Int(1)]).unwrap(); + assert_eq!(sql, "SELECT * FROM t WHERE a = 'it''s ?' AND b = 1"); + + let sql = bind_placeholders("SELECT * FROM t WHERE a = 'it\\'s ?' AND b = ?", &[Value::Int(1)]).unwrap(); + assert_eq!(sql, "SELECT * FROM t WHERE a = 'it\\'s ?' AND b = 1"); + } + + #[test] + fn unreferenced_parameter_is_an_error() { + let err = bind_placeholders("SELECT * FROM t WHERE id = ?", &[Value::Int(1), Value::Int(2)]).unwrap_err(); + assert!(matches!(err, DriverError::Internal(_))); + } + + #[test] + fn missing_parameter_is_an_error() { + let err = bind_placeholders("SELECT * FROM t WHERE a = ? AND b = ?", &[Value::Int(1)]).unwrap_err(); + assert!(matches!(err, DriverError::Internal(_))); + + let err = bind_placeholders("SELECT * FROM t WHERE a = $3", &[Value::Int(1)]).unwrap_err(); + assert!(matches!(err, DriverError::Internal(_))); + } + + #[test] + fn text_literals_escape_quotes_and_backslashes() { + assert_eq!(literal(&Value::Text("it's \\ ok".into())).unwrap(), "'it\\'s \\\\ ok'"); assert_eq!( - json_to_value(serde_json::json!("hello"), "String"), - Value::Text("hello".into()) + literal(&Value::Bytes(vec![0x00, 0xff, 0x0a])).unwrap(), + "unhex('00ff0a')" ); - assert_eq!(json_to_value(serde_json::Value::Null, "Nullable(String)"), Value::Null); + } + + #[test] + fn non_finite_floats_use_clickhouse_spellings() { + assert_eq!(literal(&Value::Float(f64::NAN)).unwrap(), "nan"); + assert_eq!(literal(&Value::Float(f64::INFINITY)).unwrap(), "inf"); + assert_eq!(literal(&Value::Float(f64::NEG_INFINITY)).unwrap(), "-inf"); + } + + #[test] + fn decimal_literals_keep_their_scale() { assert_eq!( - json_to_value(serde_json::json!("2024-06-15"), "Date"), - Value::Date(chrono::NaiveDate::from_ymd_opt(2024, 6, 15).unwrap()) + literal(&Value::Decimal("12.3400".parse().unwrap())).unwrap(), + "toDecimal128('12.3400', 4)" ); } + #[test] + fn question_marks_are_escaped_for_the_client_template() { + assert_eq!( + escape_bind_markers("SELECT * FROM t WHERE note = 'what?'"), + "SELECT * FROM t WHERE note = 'what??'" + ); + // `?fields` is the crate's other marker; escaping covers it too. + assert_eq!(escape_bind_markers("SELECT ?fields FROM t"), "SELECT ??fields FROM t"); + assert_eq!(escape_bind_markers("SELECT 1"), "SELECT 1"); + } + + #[test] + fn key_expression_splits_at_top_level_only() { + assert_eq!(split_key_expression("id"), vec!["id"]); + assert_eq!(split_key_expression("`a`, `b`"), vec!["a", "b"]); + assert_eq!(split_key_expression("toYYYYMM(d), id"), vec!["toYYYYMM(d)", "id"]); + assert!(split_key_expression("").is_empty()); + } + #[test] fn map_error_classifies_auth() { - // Construct via Display path: we only have Query/Auth heuristics on strings. let err = map_clickhouse_error(clickhouse::error::Error::BadResponse( "Code: 516. Authentication failed: password is incorrect".into(), )); assert!(matches!(err, DriverError::AuthFailed)); } + + #[test] + fn server_error_mentioning_certificate_stays_a_query_error() { + let err = map_clickhouse_error(clickhouse::error::Error::BadResponse( + "Code: 47. Unknown identifier: certificate".into(), + )); + assert!(matches!(err, DriverError::Query { .. })); + } } diff --git a/linux/crates/drivers/clickhouse/tests/integration.rs b/linux/crates/drivers/clickhouse/tests/integration.rs index f719a7227..fb3870fba 100644 --- a/linux/crates/drivers/clickhouse/tests/integration.rs +++ b/linux/crates/drivers/clickhouse/tests/integration.rs @@ -1,7 +1,8 @@ use drivers_clickhouse::ClickhouseDriver; -use tablepro_core::{ConnectOptions, DatabaseDriver, Value}; -use testcontainers::core::{IntoContainerPort, WaitFor}; +use tablepro_core::sql_dialect::{build_full_row_update, build_single_cell_update}; +use tablepro_core::{ColumnInfo, ConnectOptions, DatabaseDriver, Value}; use testcontainers::core::wait::HttpWaitStrategy; +use testcontainers::core::{IntoContainerPort, WaitFor}; use testcontainers::runners::AsyncRunner; use testcontainers::{ContainerAsync, GenericImage, ImageExt}; @@ -69,11 +70,159 @@ async fn connect_list_tables_and_pk_detection() { assert!(!note_col.primary_key); assert!(note_col.nullable); + // A MergeTree sorting key allows duplicates, so the index must not + // claim uniqueness the engine does not enforce. + let indexes = conn.fetch_indexes(None, "pk_demo").await.unwrap(); + assert_eq!(indexes.len(), 1); + assert_eq!(indexes[0].columns, vec!["id".to_string()]); + assert!(indexes[0].primary); + assert!(!indexes[0].unique); + let result = conn.fetch_rows(None, "pk_demo", 0, 100).await.unwrap(); assert_eq!(result.rows.len(), 2); assert!(!result.truncated); } +#[tokio::test] +#[ignore = "requires docker"] +async fn views_appear_in_the_table_list() { + let (_c, opts) = start_clickhouse().await; + let conn = connect(opts).await; + + conn.execute("CREATE TABLE base (id UInt64) ENGINE = MergeTree ORDER BY id") + .await + .unwrap(); + conn.execute("CREATE VIEW base_view AS SELECT id FROM base") + .await + .unwrap(); + + let tables = conn.list_tables().await.unwrap(); + assert!(tables.iter().any(|t| t.name == "base")); + assert!( + tables.iter().any(|t| t.name == "base_view"), + "views must be listed alongside tables" + ); +} + +/// The inline-edit Save path renders its UPDATE through +/// `sql_dialect`, which has to emit `ALTER TABLE … UPDATE` for +/// ClickHouse. A plain `UPDATE` is a syntax error before 25.7, so this +/// covers the dialect and the driver's bind path together. +#[tokio::test] +#[ignore = "requires docker"] +async fn inline_edit_update_applies() { + let (_c, opts) = start_clickhouse().await; + let conn = connect(opts).await; + + conn.execute("CREATE TABLE edits (id UInt64, name String) ENGINE = MergeTree ORDER BY id") + .await + .unwrap(); + conn.execute("INSERT INTO edits VALUES (1, 'before'), (2, 'other')") + .await + .unwrap(); + + let columns = conn.fetch_columns(None, "edits").await.unwrap(); + let original = vec![Value::Int(1), Value::Text("before".into())]; + let (sql, params) = build_single_cell_update( + "clickhouse", + "edits", + &columns, + &original, + 1, + Value::Text("after".into()), + ) + .unwrap(); + assert!(sql.starts_with("ALTER TABLE"), "unexpected dialect: {sql}"); + conn.execute_in_transaction(&[(sql, params)]).await.unwrap(); + + let result = conn.query("SELECT name FROM edits ORDER BY id").await.unwrap(); + assert_eq!(result.rows[0][0], Value::Text("after".into())); + assert_eq!(result.rows[1][0], Value::Text("other".into())); +} + +#[tokio::test] +#[ignore = "requires docker"] +async fn full_row_update_applies() { + let (_c, opts) = start_clickhouse().await; + let conn = connect(opts).await; + + conn.execute("CREATE TABLE rows_edit (id UInt64, a String, b Int64) ENGINE = MergeTree ORDER BY id") + .await + .unwrap(); + conn.execute("INSERT INTO rows_edit VALUES (1, 'x', 10)").await.unwrap(); + + let columns: Vec = conn.fetch_columns(None, "rows_edit").await.unwrap(); + let original = vec![Value::Int(1), Value::Text("x".into()), Value::Int(10)]; + let new_values = vec![Value::Int(1), Value::Text("y".into()), Value::Int(20)]; + let (sql, params) = build_full_row_update("clickhouse", "rows_edit", &columns, &original, &new_values).unwrap(); + conn.execute_in_transaction(&[(sql, params)]).await.unwrap(); + + let result = conn.query("SELECT a, b FROM rows_edit WHERE id = 1").await.unwrap(); + assert_eq!(result.rows[0][0], Value::Text("y".into())); + assert_eq!(result.rows[0][1], Value::Int(20)); +} + +/// A row whose text contains an apostrophe and a `?` would corrupt the +/// bind pass if the scanner walked the SQL blind. +#[tokio::test] +#[ignore = "requires docker"] +async fn literals_with_quotes_and_placeholders_round_trip() { + let (_c, opts) = start_clickhouse().await; + let conn = connect(opts).await; + + conn.execute("CREATE TABLE quoting (id UInt64, note String) ENGINE = MergeTree ORDER BY id") + .await + .unwrap(); + let tricky = "it's a ? and a $1 \\ backslash"; + conn.execute_params( + "INSERT INTO quoting (id, note) VALUES (?, ?)", + &[Value::Int(1), Value::Text(tricky.into())], + ) + .await + .unwrap(); + + let result = conn + .query_params("SELECT note FROM quoting WHERE id = ?", &[Value::Int(1)]) + .await + .unwrap(); + assert_eq!(result.rows[0][0], Value::Text(tricky.into())); +} + +#[tokio::test] +#[ignore = "requires docker"] +async fn parameterised_types_decode_to_typed_values() { + let (_c, opts) = start_clickhouse().await; + let conn = connect(opts).await; + + conn.execute( + "CREATE TABLE typed ( + id UInt64, + price Decimal(9, 2), + stamp DateTime64(3), + label LowCardinality(Nullable(String)) + ) ENGINE = MergeTree + ORDER BY id", + ) + .await + .unwrap(); + conn.execute("INSERT INTO typed VALUES (1, 12.34, '2024-06-15 08:30:00.123', 'tag')") + .await + .unwrap(); + + let cols = conn.fetch_columns(None, "typed").await.unwrap(); + let label = cols.iter().find(|c| c.name == "label").unwrap(); + assert!(label.nullable, "LowCardinality(Nullable(T)) must read as nullable"); + + let result = conn.query("SELECT price, stamp, label FROM typed").await.unwrap(); + assert_eq!(result.rows[0][0], Value::Decimal("12.34".parse().unwrap())); + assert!( + matches!(result.rows[0][1], Value::DateTime(_)), + "DateTime64(3) decoded as {:?}", + result.rows[0][1] + ); + assert_eq!(result.rows[0][2], Value::Text("tag".into())); +} + #[tokio::test] #[ignore = "requires docker"] async fn value_roundtrip_common_types() { @@ -137,12 +286,45 @@ async fn pagination_and_truncated_flag() { conn.execute("CREATE TABLE n (i UInt64) ENGINE = MergeTree ORDER BY i") .await .unwrap(); - for i in 1..=10 { - conn.execute(&format!("INSERT INTO n VALUES ({i})")).await.unwrap(); - } + conn.execute("INSERT INTO n SELECT number + 1 FROM numbers(10)") + .await + .unwrap(); + // ClickHouse applies OFFSET after the sort key, so rows 6..8 are + // the deterministic third page of three. let page = conn.fetch_rows(None, "n", 5, 3).await.unwrap(); assert_eq!(page.rows.len(), 3); + assert_eq!(page.rows[0][0], Value::Int(6)); + assert_eq!(page.rows[2][0], Value::Int(8)); + // A page carries its own LIMIT, so the server never sends a row past + // it and the cap has nothing to cut. Same as the sqlx drivers: + // `truncated` describes the row cap, not the page window. + assert!(!page.truncated); + + let last = conn.fetch_rows(None, "n", 8, 3).await.unwrap(); + assert_eq!(last.rows.len(), 2); + assert!(!last.truncated); +} + +/// `MAX_QUERY_ROWS` bounds an arbitrary `query`; the flag has to fire +/// on the row past the cap, not on a result that merely fills it. +#[tokio::test] +#[ignore = "requires docker"] +async fn query_truncates_at_the_row_cap() { + let (_c, opts) = start_clickhouse().await; + let conn = connect(opts).await; + + let cap = tablepro_core::MAX_QUERY_ROWS; + let exact = conn.query(&format!("SELECT number FROM numbers({cap})")).await.unwrap(); + assert_eq!(exact.rows.len(), cap); + assert!(!exact.truncated, "a result of exactly the cap is complete"); + + let over = conn + .query(&format!("SELECT number FROM numbers({})", cap + 1)) + .await + .unwrap(); + assert_eq!(over.rows.len(), cap); + assert!(over.truncated); } #[tokio::test] @@ -150,6 +332,9 @@ async fn pagination_and_truncated_flag() { async fn bad_sql_returns_query_error() { let (_c, opts) = start_clickhouse().await; let conn = connect(opts).await; - let err = conn.query("SELECT * FROM definitely_missing_table_xyz").await.unwrap_err(); + let err = conn + .query("SELECT * FROM definitely_missing_table_xyz") + .await + .unwrap_err(); assert!(matches!(err, tablepro_core::DriverError::Query { .. })); } diff --git a/linux/docs/adding-drivers.md b/linux/docs/adding-drivers.md index 014ad8887..088de7bb0 100644 --- a/linux/docs/adding-drivers.md +++ b/linux/docs/adding-drivers.md @@ -22,7 +22,7 @@ Each step is small. The whole task takes between half a day (PG-shaped engines) | SQLite | `sqlx` with `sqlite` feature | File-based, no network. | | MSSQL | `tiberius` | Pure Rust TDS. Watch governance — `praxiomlabs/rust-mssql-driver` is a credible alternative. | | Oracle | `oracle` (rust-oracle, kubo) | Wraps ODPI-C. Requires Oracle Instant Client on the build host. | -| ClickHouse | official `clickhouse` crate | HTTP interface (8123). Dynamic results via `FORMAT JSON`. | +| ClickHouse | official `clickhouse` crate | HTTP interface (8123). Dynamic results streamed via `FORMAT JSONCompactEachRowWithNamesAndTypes`. | | Redis | `fred` | Modern tokio rewrite of redis-rs. | | MongoDB | official `mongodb` | Mature, OpenTelemetry support. | | DuckDB | `duckdb` (official) | Bundled native lib, edition 2024. | @@ -60,7 +60,7 @@ path = "src/lib.rs" [dependencies] tablepro-core = { path = "../../core" } async-trait = "0.1" -clickhouse-arrow = "0.x" +clickhouse = { version = "0.15", default-features = false, features = ["rustls-tls"] } tokio = { version = "1", features = ["rt", "macros", "net", "time"] } thiserror = "2" ``` @@ -111,7 +111,7 @@ pub struct ClickhouseDriver; impl DatabaseDriver for ClickhouseDriver { fn id(&self) -> &'static str { "clickhouse" } fn display_name(&self) -> &'static str { "ClickHouse" } - fn default_port(&self) -> u16 { 9000 } + fn default_port(&self) -> u16 { 8123 } async fn connect(&self, opts: ConnectOptions) -> Result, DriverError> { let client = build_client(opts).await?; @@ -120,7 +120,7 @@ impl DatabaseDriver for ClickhouseDriver { } struct ClickhouseConnection { - client: clickhouse_arrow::Client, + client: clickhouse::Client, } #[async_trait] @@ -140,6 +140,13 @@ Notes: - `default_port()` is what the connection dialog pre-fills. - `DriverError` is a `thiserror` enum in `tablepro-core`. Map underlying crate errors into the variants. Add a new variant only after PR discussion. +`DatabaseDriver` also has defaulted hooks for engines that break an assumption the app otherwise makes. Override one only when the default is wrong for your engine: + +- `ddl_is_transactional()`: the structure editor batches DDL into one transaction when true. False for engines that commit implicitly on every DDL statement. +- `reports_rows_affected()`: the inline-edit Save path reads a zero `rows_affected` on an UPDATE or DELETE as another session having changed the row. Return false if the engine cannot produce a count, or every successful save warns about a lost update. + +If your engine needs a different SQL spelling for a statement the app builds centrally, add the dialect branch in `core::sql_dialect` (`quote_ident`, `placeholder_for`, `build_update`, `build_order_and_pagination`) rather than rewriting the SQL inside the driver. ClickHouse takes `build_update`'s `ALTER TABLE … UPDATE` branch for this reason. + ## 4. Add the crate to the workspace Edit `linux/Cargo.toml`: @@ -194,9 +201,9 @@ use testcontainers::images::generic::GenericImage; async fn list_tables_returns_seeded_tables() { let docker = Cli::default(); let image = GenericImage::new("clickhouse/clickhouse-server", "latest") - .with_exposed_port(9000); + .with_exposed_port(8123); let node = docker.run(image); - let port = node.get_host_port_ipv4(9000); + let port = node.get_host_port_ipv4(8123); let driver = ClickhouseDriver; let conn = driver.connect(ConnectOptions {