From e04e5dd57d8ee5a42270d29e5597f431679ad3e5 Mon Sep 17 00:00:00 2001 From: kkmihai Date: Wed, 4 Nov 2026 06:21:03 +0200 Subject: [PATCH] feat(database): add ClickHouse data sources (#865) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds ClickHouse to Data Sources over its HTTP interface (HTTP/HTTPS, SSH tunnels, session settings, server-side cancellation): browse databases, tables and columns, inspect DDL, run queries and EXPLAIN plans, with ClickHouse dialect completion. Table data stays read-only and manual transactions are rejected, since ClickHouse keys are not unique and HTTP has no transactions. Results are streamed as JSONCompactEachRowWithNamesAndTypes and parsing stops at the row limit, because result_overflow_mode=break still returns a whole block; errors show only the server's "Code: …" message, including failures raised mid-stream. Fixes #862 Co-authored-by: kkMihai <59262001+kkMihai@users.noreply.github.com> --- Cargo.lock | 23 + crates/ide-db/Cargo.toml | 6 +- crates/ide-db/src/clickhouse.rs | 844 ++++++++++++++++++++ crates/ide-db/src/conn.rs | 40 +- crates/ide-db/src/lib.rs | 88 +- crates/ide-db/src/plan.rs | 19 +- crates/ide-db/src/source.rs | 41 +- crates/ide-db/src/split.rs | 19 +- crates/ide-db/src/sql.rs | 108 ++- crates/ide-db/tests/clickhouse.rs | 136 ++++ crates/ide-db/tests/sqlite.rs | 6 +- crates/ide-host/src/ops/db.rs | 3 +- e2e/clickhouse-source.spec.ts | 42 + src/bindings.ts | 2 +- src/features/database/ObjectDialogs.tsx | 24 +- src/features/database/SourceDialog.tsx | 50 +- src/features/database/TableEditor.tsx | 11 +- src/features/database/console.tsx | 29 +- src/features/database/database.test.ts | 40 +- src/features/database/export.ts | 6 +- src/features/database/model.ts | 4 + src/features/database/objects.test.ts | 48 ++ src/features/database/objects.ts | 37 +- src/features/database/sql/analyze.ts | 8 +- src/features/database/sql/complete.ts | 40 +- src/features/database/sql/dialect.ts | 86 ++ src/features/database/sql/lexer.ts | 22 +- src/features/database/sql/params.ts | 11 + src/features/database/sql/resolve.ts | 22 +- src/features/database/sql/sql.test.ts | 66 +- src/features/database/store.ts | 8 +- src/features/database/table-editor.test.tsx | 90 +++ src/features/database/treeActions.ts | 18 +- src/features/database/treeMenu.ts | 18 +- src/features/database/vendors.ts | 20 + src/platform/locales/text/zh.json | 3 + src/platform/locales/textCatalog.test.ts | 1 + src/ui/icons.tsx | 1 + src/ui/openidea/database/clickhouse.svg | 1 + 39 files changed, 1921 insertions(+), 120 deletions(-) create mode 100644 crates/ide-db/src/clickhouse.rs create mode 100644 crates/ide-db/tests/clickhouse.rs create mode 100644 e2e/clickhouse-source.spec.ts create mode 100644 src/features/database/table-editor.test.tsx create mode 100644 src/ui/openidea/database/clickhouse.svg diff --git a/Cargo.lock b/Cargo.lock index 428910432..ec56cb175 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4154,8 +4154,11 @@ dependencies = [ "dirs 6.0.0", "futures", "ide-core", + "ide-http", "ide-ssh", "parking_lot", + "reqwest", + "rustls", "serde", "serde_json", "specta", @@ -4166,6 +4169,7 @@ dependencies = [ "tokio-util", "toml 0.9.12+spec-1.1.0", "tracing", + "uuid", ] [[package]] @@ -7133,6 +7137,7 @@ dependencies = [ "rustls-platform-verifier", "serde", "serde_json", + "serde_urlencoded", "sync_wrapper", "tokio", "tokio-rustls", @@ -7491,6 +7496,12 @@ dependencies = [ "wait-timeout", ] +[[package]] +name = "ryu" +version = "1.0.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9774ba4a74de5f7b1c1451ed6cd5285a32eddb5cccb8cc655a4e50009e06477f" + [[package]] name = "saa" version = "5.6.2" @@ -7795,6 +7806,18 @@ dependencies = [ "serde_core", ] +[[package]] +name = "serde_urlencoded" +version = "0.7.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d3491c14715ca2294c4d6a88f15e84739788c1d030eed8c110436aafdaa2f3fd" +dependencies = [ + "form_urlencoded", + "itoa", + "ryu", + "serde", +] + [[package]] name = "serde_with" version = "3.24.0" diff --git a/crates/ide-db/Cargo.toml b/crates/ide-db/Cargo.toml index 6a5cb4407..8f9853930 100644 --- a/crates/ide-db/Cargo.toml +++ b/crates/ide-db/Cargo.toml @@ -8,7 +8,7 @@ license.workspace = true [dependencies] ide-core.workspace = true serde.workspace = true -serde_json.workspace = true +serde_json = { workspace = true, features = ["raw_value"] } specta.workspace = true toml.workspace = true tokio.workspace = true @@ -21,6 +21,10 @@ bb8.workspace = true tokio-util = { workspace = true, features = ["compat"] } dirs.workspace = true ide-ssh.workspace = true +ide-http.workspace = true +reqwest = { workspace = true, features = ["query"] } +rustls.workspace = true +uuid.workspace = true [dev-dependencies] tokio = { workspace = true, features = ["rt-multi-thread", "macros"] } diff --git a/crates/ide-db/src/clickhouse.rs b/crates/ide-db/src/clickhouse.rs new file mode 100644 index 000000000..34e869dce --- /dev/null +++ b/crates/ide-db/src/clickhouse.rs @@ -0,0 +1,844 @@ +use crate::conn::{QueryResult, ResultColumn, Setup}; +use crate::source::{DataSource, DbSslMode, DbTxMode}; +use ide_core::{CancellationToken, IdeError, IdeResult, error::ErrorKind}; +use parking_lot::RwLock; +use reqwest::{Client, RequestBuilder, StatusCode, Url}; +use std::sync::Arc; +use std::time::{Duration, Instant}; +use tokio::sync::Mutex; + +#[derive(Clone)] +pub struct ClickhousePool(Arc); + +struct State { + client: RwLock>, + endpoint: Url, + user: String, + password: String, + database: String, + settings: Vec<(String, String)>, + read_only: bool, + session_id: String, + session: Mutex, + closed: CancellationToken, +} + +/// The response format: a names line, a types line, then one JSON array per row, so rows past +/// the grid's limit are skipped without being parsed. +const FORMAT: &str = "JSONCompactEachRowWithNamesAndTypes"; + +pub(crate) fn transaction_error() -> IdeError { + IdeError::invalid("ClickHouse HTTP does not support SQL transactions; use auto-commit and run writes directly") +} + +impl ClickhousePool { + pub(crate) async fn connect(source: &DataSource, password: Option<&str>, setup: Setup) -> IdeResult { + if source.options.transaction == DbTxMode::Manual { + return Err(transaction_error()); + } + for (name, _) in &setup.params { + if reserved_setting(name) { + return Err(IdeError::invalid(format!("ClickHouse HTTP manages the {name} setting"))); + } + } + let secure = matches!(source.ssl, DbSslMode::Require | DbSslMode::VerifyCa | DbSslMode::VerifyFull); + let tls_files = [&source.ssl_ca_file, &source.ssl_cert_file, &source.ssl_key_file]; + if !secure && tls_files.iter().any(|p| !p.trim().is_empty()) { + return Err(IdeError::invalid( + "ClickHouse TLS certificate files require an SSL mode of Require, Verify CA, or Verify Full", + )); + } + let host = source.host.trim(); + if host.is_empty() || host.contains(['/', '@', '?', '#']) { + return Err(IdeError::invalid("ClickHouse host must contain only a hostname or IP address")); + } + let host = if host.contains(':') && !host.starts_with('[') { format!("[{host}]") } else { host.to_owned() }; + let scheme = if secure { "https" } else { "http" }; + let port = source.port.unwrap_or(if secure { 8443 } else { 8123 }); + let endpoint = Url::parse(&format!("{scheme}://{host}:{port}/")) + .map_err(|_| IdeError::invalid("invalid ClickHouse HTTP host or port"))?; + let timeout = setup.connect_timeout.unwrap_or(Duration::from_secs(30)); + let _ = rustls::crypto::ring::default_provider().install_default(); + let mut builder = Client::builder() + .connect_timeout(timeout) + .pool_max_idle_per_host(4) + .redirect(reqwest::redirect::Policy::none()) + .user_agent(concat!("OpenIDEA/", env!("CARGO_PKG_VERSION"))); + let ca = source.ssl_ca_file.trim(); + if !ca.is_empty() { + let pem = read_pem(ca, "CA certificate")?; + let certificates = reqwest::Certificate::from_pem_bundle(&pem) + .map_err(|_| IdeError::invalid("invalid ClickHouse CA certificate PEM"))?; + if certificates.is_empty() { + return Err(IdeError::invalid("ClickHouse CA certificate PEM contains no certificates")); + } + builder = builder.tls_certs_only(certificates); + } + if source.ssl == DbSslMode::VerifyCa { + if ca.is_empty() { + return Err(IdeError::invalid("ClickHouse Verify CA requires a CA certificate file")); + } + builder = builder.tls_danger_accept_invalid_hostnames(true); + } + let cert = source.ssl_cert_file.trim(); + let key = source.ssl_key_file.trim(); + if cert.is_empty() != key.is_empty() { + return Err(IdeError::invalid("ClickHouse client TLS requires both certificate and key files")); + } + if !cert.is_empty() { + let mut pem = read_pem(cert, "client certificate")?; + pem.push(b'\n'); + pem.extend(read_pem(key, "client key")?); + let identity = reqwest::Identity::from_pem(&pem) + .map_err(|_| IdeError::invalid("invalid ClickHouse client certificate or key PEM"))?; + builder = builder.identity(identity); + } + let proxy = ide_core::proxy_settings::proxy(); + let client = ide_http::client::with_proxy(builder, &proxy).build().map_err(http_error)?; + let database = if source.options.default_schema.trim().is_empty() { + source.database.trim() + } else { + source.options.default_schema.trim() + }; + let pool = Self(Arc::new(State { + client: RwLock::new(Some(client)), + endpoint, + user: if source.user.is_empty() { "default".into() } else { source.user.clone() }, + password: password.unwrap_or_default().to_owned(), + database: if database.is_empty() { "default".into() } else { database.into() }, + settings: setup.params, + read_only: source.read_only, + session_id: uuid::Uuid::new_v4().to_string(), + session: Mutex::new(false), + closed: CancellationToken::new(), + })); + let cancel = CancellationToken::new(); + let initialize = async { + pool.execute("SELECT 1", 1, &cancel).await?; + for sql in setup.statements { + pool.execute(&sql, 0, &cancel).await?; + } + Ok(pool.clone()) + }; + tokio::pin!(initialize); + tokio::select! { + result = &mut initialize => { + if result.is_err() { + pool.close(); + } + result + }, + () = tokio::time::sleep(timeout) => { + cancel.cancel(); + let _ = initialize.await; + pool.close(); + Err(IdeError::new(ErrorKind::TimedOut, "timed out connecting to ClickHouse HTTP")) + } + } + } + + pub fn close(&self) { + self.0.closed.cancel(); + self.0.client.write().take(); + } + + pub fn is_closed(&self) -> bool { + self.0.closed.is_cancelled() + } + + fn client(&self) -> IdeResult { + self.0.client.read().clone().ok_or_else(|| IdeError::new(ErrorKind::Io, "ClickHouse connection is closed")) + } + + fn request(&self, client: &Client, query_id: &str, max_rows: usize, initialized: bool) -> RequestBuilder { + let mut request = client + .post(self.0.endpoint.clone()) + .basic_auth(&self.0.user, Some(&self.0.password)) + .header(reqwest::header::CONTENT_TYPE, "text/plain; charset=utf-8") + .query(&self.0.settings) + .query(&[ + ("session_id", self.0.session_id.clone()), + ("session_timeout", "3600".into()), + ("query_id", query_id.to_owned()), + ("default_format", FORMAT.into()), + ("output_format_json_quote_denormals", "1".into()), + ("max_result_rows", max_rows.saturating_add(1).to_string()), + ("result_overflow_mode", "break".into()), + ]); + if initialized { + request = request.query(&[("session_check", "1")]); + } else { + request = request.query(&[("database", &self.0.database)]); + } + if self.0.read_only { + request = request.query(&[("readonly", "1")]); + } + request + } + + pub async fn execute(&self, sql: &str, max_rows: usize, cancel: &CancellationToken) -> IdeResult { + if transaction_statement(sql) { + return Err(transaction_error()); + } + let mut session = tokio::select! { + biased; + () = cancel.cancelled() => return Err(IdeError::cancelled()), + () = self.0.closed.cancelled() => return Err(IdeError::new(ErrorKind::Io, "ClickHouse connection is closed")), + session = self.0.session.lock() => session, + }; + let client = self.client()?; + let query_id = uuid::Uuid::new_v4().to_string(); + let request = self.request(&client, &query_id, max_rows, *session).body(sql.to_owned()); + let started = Instant::now(); + let fetch = async { + let mut response = request.send().await.map_err(http_error)?; + let execution_ms = started.elapsed().as_secs_f64() * 1000.0; + let status = response.status(); + let header = |name: &str| response.headers().get(name).and_then(|v| v.to_str().ok()).map(str::to_owned); + let exception_code = header("X-ClickHouse-Exception-Code"); + let format = header("X-ClickHouse-Format"); + let affected = header("X-ClickHouse-Summary") + .and_then(|v| serde_json::from_str::(&v).ok()) + .and_then(|v| { + v.get("written_rows") + .and_then(|n| n.as_str().and_then(|n| n.parse::().ok()).or_else(|| n.as_u64())) + }); + if !status.is_success() || exception_code.as_deref().is_some_and(|code| code != "0") { + let body = response.bytes().await.map_err(http_error)?; + return Err(self.server_error(status, exception_code.as_deref(), &body)); + } + let (columns, rows, has_more) = if format.as_deref().is_some_and(|f| f != FORMAT) { + let body = response.bytes().await.map_err(http_error)?; + if !body.iter().all(u8::is_ascii_whitespace) { + return Err(IdeError::invalid( + "ClickHouse result grids need the default output format; remove the query's FORMAT clause", + )); + } + (Vec::new(), Vec::new(), false) + } else { + let mut reader = RowReader::new(max_rows); + while let Some(chunk) = response.chunk().await.map_err(http_error)? { + reader.push(&chunk)?; + } + reader.finish().map_err(|error| self.redact_error(error))? + }; + let affected = if columns.is_empty() { affected.map(|n| n as f64) } else { None }; + Ok(QueryResult { + statement: sql.to_owned(), + columns, + rows, + affected, + has_more, + elapsed_ms: started.elapsed().as_secs_f64() * 1000.0, + execution_ms, + }) + }; + let result = tokio::select! { + biased; + () = cancel.cancelled() => None, + () = self.0.closed.cancelled() => None, + result = fetch => Some(result), + }; + match result { + Some(result) => { + if result.is_ok() { + *session = true; + } + result + } + None => { + let killed = self.kill(&client, &query_id).await; + match killed { + Ok(()) => Err(IdeError::cancelled()), + Err(error) => Err(IdeError::new( + ErrorKind::Cancelled, + format!("cancelled locally; ClickHouse server cancellation failed: {}", error.message), + )), + } + } + } + } + + async fn kill(&self, client: &Client, query_id: &str) -> IdeResult<()> { + let response = client + .post(self.0.endpoint.clone()) + .basic_auth(&self.0.user, Some(&self.0.password)) + .timeout(Duration::from_secs(3)) + .body(format!("KILL QUERY WHERE query_id = '{query_id}' SYNC")) + .send() + .await + .map_err(http_error)?; + let status = response.status(); + let code = response + .headers() + .get("X-ClickHouse-Exception-Code") + .and_then(|value| value.to_str().ok()) + .map(str::to_owned); + let body = response.bytes().await.map_err(http_error)?; + if status.is_success() && !code.as_deref().is_some_and(|code| code != "0") && !body.starts_with(b"Code:") { + Ok(()) + } else { + Err(self.server_error(status, code.as_deref(), &body)) + } + } + + fn redact_error(&self, mut error: IdeError) -> IdeError { + if !self.0.password.is_empty() { + error.message = error.message.replace(&self.0.password, "[redacted]"); + } + error + } + + fn server_error(&self, status: StatusCode, code: Option<&str>, body: &[u8]) -> IdeError { + let message = String::from_utf8_lossy(body); + let message: String = exception_text(&message).chars().take(16_384).collect(); + let auth = matches!(status, StatusCode::UNAUTHORIZED | StatusCode::FORBIDDEN) + || matches!(code, Some("193" | "194" | "195" | "516" | "497")); + let detail = if message.is_empty() { status.canonical_reason().unwrap_or("request failed") } else { &message }; + let detail = if detail.contains("SESSION_NOT_FOUND") || detail.contains("Session not found") { + "ClickHouse HTTP session expired; reconnect to restore session settings" + } else { + detail + }; + self.redact_error(IdeError::new( + if auth { ErrorKind::PermissionDenied } else { ErrorKind::InvalidInput }, + format!("ClickHouse HTTP {}: {detail}", status.as_u16()), + )) + } +} + +type DecodedResult = (Vec, Vec>>, bool); + +/// Decodes a streamed [`FORMAT`] body. Keeps the first `max_rows` rows; later rows only set +/// `has_more` and are read past unparsed, so a grid limit bounds memory however large the +/// block `result_overflow_mode=break` lets the server send. A line that is not a JSON array +/// starts the exception ClickHouse writes when a query fails mid-stream. +struct RowReader { + max_rows: usize, + pending: Vec, + names: Option>, + types: Option>, + rows: Vec>>, + more: bool, + exception: Option, +} + +impl RowReader { + fn new(max_rows: usize) -> Self { + Self { max_rows, pending: Vec::new(), names: None, types: None, rows: Vec::new(), more: false, exception: None } + } + + fn push(&mut self, chunk: &[u8]) -> IdeResult<()> { + self.pending.extend_from_slice(chunk); + let buffer = std::mem::take(&mut self.pending); + let mut start = 0; + while let Some(end) = buffer[start..].iter().position(|&b| b == b'\n') { + self.line(&buffer[start..start + end])?; + start += end + 1; + } + self.pending = buffer[start..].to_vec(); + Ok(()) + } + + fn line(&mut self, line: &[u8]) -> IdeResult<()> { + if let Some(exception) = &mut self.exception { + if exception.len() < 16_384 { + exception.push_str(&String::from_utf8_lossy(line)); + exception.push('\n'); + } + return Ok(()); + } + let text = line.trim_ascii(); + if text.is_empty() { + return Ok(()); + } + if text[0] != b'[' { + self.exception = Some(format!("{}\n", String::from_utf8_lossy(line))); + return Ok(()); + } + if self.names.is_some() && self.types.is_some() && self.rows.len() >= self.max_rows { + self.more = true; + return Ok(()); + } + let Ok(values) = serde_json::from_slice::>>(text) else { + self.exception = Some(format!("{}\n", String::from_utf8_lossy(line))); + return Ok(()); + }; + let protocol = |message: String| IdeError::new(ErrorKind::Protocol, message); + let header = |values: Vec>| { + values + .iter() + .map(|v| serde_json::from_str::(v.get())) + .collect::, _>>() + .map_err(|e| protocol(format!("invalid ClickHouse result header: {e}"))) + }; + let Some(names) = &self.names else { + self.names = Some(header(values)?); + return Ok(()); + }; + if self.types.is_none() { + self.types = Some(header(values)?); + return Ok(()); + } + if values.len() != names.len() { + return Err(protocol("ClickHouse result row width does not match its metadata".into())); + } + let row = values + .into_iter() + .map(|value| { + let text = value.get(); + if text == "null" { + Ok(None) + } else if text.starts_with('"') { + serde_json::from_str::(text).map(Some).map_err(|e| protocol(e.to_string())) + } else { + Ok(Some(text.to_owned())) + } + }) + .collect::>()?; + self.rows.push(row); + Ok(()) + } + + fn finish(mut self) -> IdeResult { + let rest = std::mem::take(&mut self.pending); + self.line(&rest)?; + if let Some(exception) = self.exception { + return Err(IdeError::invalid(format!("ClickHouse: {}", exception_text(&exception)))); + } + let Some(names) = self.names else { return Ok((Vec::new(), Vec::new(), false)) }; + let types = self.types.unwrap_or_default(); + if types.len() != names.len() { + return Err(IdeError::new(ErrorKind::Protocol, "ClickHouse result types do not match its columns")); + } + let columns = names.into_iter().zip(types).map(|(name, type_name)| ResultColumn { name, type_name }).collect(); + Ok((columns, self.rows, self.more)) + } +} + +/// The server's message in a failed response: rows already streamed and the +/// `__exception__` block markers ClickHouse wraps a mid-stream error in are dropped. +fn exception_text(body: &str) -> &str { + let Some(at) = body.find("Code:") else { return body.trim() }; + let text = &body[at..]; + match text.find("\n__exception__") { + // The line before the closing marker is ` `. + Some(end) => text[..end].trim_end().rsplit_once('\n').map_or(text[..end].trim(), |(message, _)| message.trim()), + None => text.trim(), + } +} + +fn http_error(error: reqwest::Error) -> IdeError { + use std::error::Error; + let kind = if error.is_timeout() { ErrorKind::TimedOut } else { ErrorKind::Io }; + let error = error.without_url(); + let mut message = error.to_string(); + let mut source = error.source(); + while let Some(error) = source { + let detail = error.to_string(); + if !message.contains(&detail) { + message.push_str(": "); + message.push_str(&detail); + } + source = error.source(); + } + IdeError::new(kind, format!("ClickHouse HTTP: {message}")) +} + +fn read_pem(path: &str, label: &str) -> IdeResult> { + std::fs::read(ide_ssh::expand(path)) + .map_err(|error| IdeError::new(ErrorKind::Io, format!("cannot read ClickHouse {label}: {error}"))) +} + +fn reserved_setting(name: &str) -> bool { + let name = name.to_ascii_lowercase(); + name.starts_with("output_format_json_") + || matches!( + name.as_str(), + "user" + | "password" + | "query" + | "database" + | "session_id" + | "session_check" + | "session_timeout" + | "query_id" + | "default_format" + | "max_result_rows" + | "result_overflow_mode" + | "readonly" + ) +} + +fn transaction_statement(sql: &str) -> bool { + let mut sql = sql.trim_start(); + loop { + if let Some(rest) = sql.strip_prefix("--").or_else(|| sql.strip_prefix('#')) { + sql = rest.split_once('\n').map_or("", |(_, rest)| rest).trim_start(); + } else if let Some(rest) = sql.strip_prefix("/*") { + sql = rest.split_once("*/").map_or("", |(_, rest)| rest).trim_start(); + } else { + break; + } + } + let mut words = sql.split(|c: char| !c.is_ascii_alphanumeric() && c != '_').filter(|s| !s.is_empty()); + let first = words.next().unwrap_or_default(); + ["BEGIN", "COMMIT", "ROLLBACK", "SAVEPOINT", "RELEASE"].iter().any(|s| first.eq_ignore_ascii_case(s)) + || (first.eq_ignore_ascii_case("START") && words.next().is_some_and(|s| s.eq_ignore_ascii_case("TRANSACTION"))) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::conn::Pool; + use crate::source::{DbDriver, DbOptions, DbProperty}; + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + const ONE: &str = "[\"1\"]\n[\"UInt8\"]\n[1]\n"; + + async fn read_request(stream: &mut tokio::net::TcpStream) -> String { + let mut bytes = Vec::new(); + let mut buffer = [0; 4096]; + loop { + let count = tokio::time::timeout(Duration::from_secs(5), stream.read(&mut buffer)).await.unwrap().unwrap(); + assert!(count > 0, "request ended early"); + bytes.extend_from_slice(&buffer[..count]); + if let Some(end) = bytes.windows(4).position(|s| s == b"\r\n\r\n") { + let header = String::from_utf8_lossy(&bytes[..end]); + let length: usize = header + .lines() + .find_map(|line| { + line.to_ascii_lowercase().strip_prefix("content-length:").map(|v| v.trim().parse().unwrap()) + }) + .unwrap_or(0); + if bytes.len() >= end + 4 + length { + return String::from_utf8(bytes).unwrap(); + } + } + } + } + + async fn reply(stream: &mut tokio::net::TcpStream, status: &str, headers: &str, body: &str) { + let response = + format!("HTTP/1.1 {status}\r\nContent-Length: {}\r\nConnection: close\r\n{headers}\r\n{body}", body.len()); + stream.write_all(response.as_bytes()).await.unwrap(); + } + + async fn server( + responses: Vec<(&'static str, &'static str, &'static str)>, + ) -> (u16, tokio::task::JoinHandle>) { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + let task = tokio::spawn(async move { + let mut requests = Vec::new(); + for (status, headers, body) in responses { + let (mut stream, _) = + tokio::time::timeout(Duration::from_secs(5), listener.accept()).await.unwrap().unwrap(); + requests.push(read_request(&mut stream).await); + reply(&mut stream, status, headers, body).await; + } + requests + }); + (port, task) + } + + fn source(port: u16) -> DataSource { + DataSource { driver: DbDriver::Clickhouse, host: "127.0.0.1".into(), port: Some(port), ..Default::default() } + } + + fn decode(body: &[u8], max_rows: usize, chunk: usize) -> IdeResult { + let mut reader = RowReader::new(max_rows); + for part in body.chunks(chunk) { + reader.push(part)?; + } + reader.finish() + } + + #[test] + fn rows_preserve_precision_nulls_and_strings_across_chunks() { + let body = concat!( + r#"["n", "d", "v", "v", "v"]"#, + "\n", + r#"["UInt64", "Decimal(38, 9)", "Nullable(String)", "Nullable(String)", "Nullable(String)"]"#, + "\n", + r#"[18446744073709551615, 12345678901234567890.123456789, null, "null", "ᴺᵁᴸᴸ"]"#, + ); + for chunk in [1, 7, body.len()] { + let (columns, rows, more) = decode(body.as_bytes(), 1, chunk).unwrap(); + assert_eq!(columns.len(), 5); + assert_eq!(columns[1].type_name, "Decimal(38, 9)"); + assert_eq!( + rows[0], + vec![ + Some("18446744073709551615".into()), + Some("12345678901234567890.123456789".into()), + None, + Some("null".into()), + Some("ᴺᵁᴸᴸ".into()), + ] + ); + assert!(!more); + } + let (columns, rows, more) = decode(b"[\"x\"]\n[\"String\"]\n", 0, 64).unwrap(); + assert_eq!(columns[0].name, "x"); + assert!(rows.is_empty() && !more); + assert_eq!(decode(b"", 5, 64).unwrap().0.len(), 0); + assert!(decode(b"[\"x\"]\n[\"UInt8\"]\n[1, 2]\n", 1, 64).is_err()); + } + + #[test] + fn rows_past_the_limit_are_skipped_but_a_later_exception_still_fails() { + let mut body = String::from("[\"n\"]\n[\"UInt64\"]\n"); + for n in 0..1000 { + body.push_str(&format!("[{n}]\n")); + } + let (_, rows, more) = decode(body.as_bytes(), 2, 4096).unwrap(); + assert_eq!(rows, vec![vec![Some("0".into())], vec![Some("1".into())]]); + assert!(more); + body.push_str("\n__exception__\nabcdefgh\nCode: 395. DB::Exception: Value passed to 'throwIf' is non-zero\n"); + body.push_str("62 abcdefgh\n__exception__\n"); + let error = decode(body.as_bytes(), 2, 4096).unwrap_err(); + assert_eq!(error.message, "ClickHouse: Code: 395. DB::Exception: Value passed to 'throwIf' is non-zero"); + let multiline = "Code: 62. DB::Exception: Syntax error\nat line 2"; + assert_eq!(exception_text(multiline), multiline); + } + + #[tokio::test] + async fn http_requests_use_auth_settings_limits_and_a_shared_session() { + let (port, requests) = server(vec![ + ("200 OK", "X-ClickHouse-Format: JSONCompactEachRowWithNamesAndTypes\r\n", ONE), + ("200 OK", "", ""), + ( + "200 OK", + "X-ClickHouse-Format: JSONCompactEachRowWithNamesAndTypes\r\n", + "[\"n\"]\n[\"UInt64\"]\n[\"1\"]\n[\"2\"]\n[\"3\"]\n", + ), + ]) + .await; + let source = DataSource { + user: "reader".into(), + read_only: true, + options: DbOptions { default_schema: "analytics".into(), time_zone: "UTC".into(), ..Default::default() }, + properties: vec![DbProperty { name: "max_threads".into(), value: "2".into() }], + ..source(port) + }; + let pool = Pool::connect(&source, Some("mock-password"), None).await.unwrap(); + pool.execute("USE other", 0, &CancellationToken::new()).await.unwrap(); + let result = pool.execute("SELECT number FROM numbers(3)", 1, &CancellationToken::new()).await.unwrap(); + assert_eq!(result.rows, vec![vec![Some("1".into())]]); + assert!(result.has_more && result.affected.is_none()); + let requests = requests.await.unwrap(); + let target = |request: &String| { + Url::parse(&format!( + "http://localhost{}", + request.lines().next().unwrap().split_whitespace().nth(1).unwrap() + )) + .unwrap() + }; + let urls: Vec<_> = requests.iter().map(target).collect(); + let value = |url: &Url, key: &str| url.query_pairs().find(|(k, _)| k == key).map(|(_, v)| v.into_owned()); + assert_eq!(value(&urls[0], "database").as_deref(), Some("analytics")); + assert_eq!(value(&urls[1], "database"), None); + assert_eq!(value(&urls[1], "session_check").as_deref(), Some("1")); + assert_eq!(value(&urls[0], "session_id"), value(&urls[2], "session_id")); + assert_ne!(value(&urls[0], "query_id"), value(&urls[2], "query_id")); + assert_eq!(value(&urls[2], "max_result_rows").as_deref(), Some("2")); + for (request, url) in requests.iter().zip(&urls) { + assert!(request.starts_with("POST /?")); + assert!(request.contains("Basic cmVhZGVyOm1vY2stcGFzc3dvcmQ=")); + assert!(!url.as_str().contains("reader") && !url.as_str().contains("mock-password")); + assert_eq!(value(url, "readonly").as_deref(), Some("1")); + assert_eq!(value(url, "session_timezone").as_deref(), Some("UTC")); + assert_eq!(value(url, "max_threads").as_deref(), Some("2")); + } + assert!(requests[0].ends_with("SELECT 1")); + assert!(pool.transaction(&[]).await.is_err()); + let cancel = CancellationToken::new(); + cancel.cancel(); + assert_eq!(pool.execute("SELECT 1", 1, &cancel).await.unwrap_err().kind, ErrorKind::Cancelled); + pool.close().await; + assert!(pool.is_closed()); + assert!(pool.execute("SELECT 1", 1, &CancellationToken::new()).await.is_err()); + } + + #[tokio::test] + async fn sequential_queries_reuse_the_http_connection() { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + let task = tokio::spawn(async move { + let (mut stream, _) = listener.accept().await.unwrap(); + for _ in 0..2 { + read_request(&mut stream).await; + let response = format!("HTTP/1.1 200 OK\r\nContent-Length: {}\r\n\r\n{ONE}", ONE.len()); + stream.write_all(response.as_bytes()).await.unwrap(); + } + }); + let pool = Pool::connect(&source(port), None, None).await.unwrap(); + assert_eq!( + pool.execute("SELECT 1", 1, &CancellationToken::new()).await.unwrap().rows, + vec![vec![Some("1".into())]] + ); + task.await.unwrap(); + } + + #[tokio::test] + async fn server_errors_keep_details_and_hide_passwords() { + let (port, requests) = server(vec![( + "403 Forbidden", + "X-ClickHouse-Exception-Code: 516\r\n", + "Code: 516. DB::Exception: bad mock-password", + )]) + .await; + let error = match Pool::connect(&source(port), Some("mock-password"), None).await { + Ok(_) => panic!("expected login failure"), + Err(error) => error, + }; + assert_eq!(error.kind, ErrorKind::PermissionDenied); + assert!(error.message.contains("516") && error.message.contains("[redacted]")); + assert!(!error.message.contains("mock-password")); + requests.await.unwrap(); + let (port, requests) = server(vec![("200 OK", "", "Code: 241. DB::Exception: memory limit")]).await; + let error = match Pool::connect(&source(port), None, None).await { + Ok(_) => panic!("expected late query failure"), + Err(error) => error, + }; + assert!(error.message.contains("memory limit")); + requests.await.unwrap(); + let (port, requests) = server(vec![( + "500 Internal Server Error", + "X-ClickHouse-Exception-Code: 395\r\n", + "[\"1\"]\n[\"UInt8\"]\n[1]\n\n__exception__\ntag\nCode: 395. DB::Exception: thrown\n32 tag\n__exception__\n", + )]) + .await; + let error = match Pool::connect(&source(port), None, None).await { + Ok(_) => panic!("expected mid-stream failure"), + Err(error) => error, + }; + assert_eq!(error.message, "ClickHouse HTTP 500: Code: 395. DB::Exception: thrown"); + requests.await.unwrap(); + } + + #[tokio::test] + async fn cancellation_sends_kill_outside_the_busy_session() { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + let (started, received) = tokio::sync::oneshot::channel(); + let server = tokio::spawn(async move { + let (mut health, _) = listener.accept().await.unwrap(); + read_request(&mut health).await; + reply(&mut health, "200 OK", "", ONE).await; + drop(health); + let (mut query, _) = listener.accept().await.unwrap(); + let request = read_request(&mut query).await; + started.send(()).unwrap(); + let (mut kill, _) = tokio::time::timeout(Duration::from_secs(5), listener.accept()).await.unwrap().unwrap(); + let kill_request = read_request(&mut kill).await; + reply(&mut kill, "200 OK", "", "").await; + (request, kill_request) + }); + let pool = Pool::connect(&source(port), None, None).await.unwrap(); + let cancel = CancellationToken::new(); + let query = pool.execute("SELECT sleep(30)", 1, &cancel); + let cancelled = async { + received.await.unwrap(); + cancel.cancel(); + }; + let (result, ()) = tokio::join!(query, cancelled); + assert_eq!(result.unwrap_err().kind, ErrorKind::Cancelled); + let (query, kill) = server.await.unwrap(); + let target = query.lines().next().unwrap().split_whitespace().nth(1).unwrap(); + let url = Url::parse(&format!("http://localhost{target}")).unwrap(); + let query_id = url.query_pairs().find(|(k, _)| k == "query_id").unwrap().1.into_owned(); + assert!(kill.ends_with(&format!("KILL QUERY WHERE query_id = '{query_id}' SYNC"))); + assert!(!kill.contains("session_id=")); + } + + #[tokio::test] + async fn unsupported_settings_and_transactions_fail_before_connecting() { + for statement in ["BEGIN", "-- comment\n COMMIT", "/* comment */ ROLLBACK", "START TRANSACTION", "SAVEPOINT s"] + { + assert!(transaction_statement(statement)); + } + assert!(!transaction_statement("SELECT 'BEGIN'")); + let sources = [ + DataSource { options: DbOptions { transaction: DbTxMode::Manual, ..Default::default() }, ..source(1) }, + DataSource { + properties: vec![DbProperty { name: "statement_cache_capacity".into(), value: "4".into() }], + ..source(1) + }, + DataSource { properties: vec![DbProperty { name: "password".into(), value: "x".into() }], ..source(1) }, + DataSource { ssl_ca_file: "unused.pem".into(), ..source(1) }, + DataSource { ssl: DbSslMode::VerifyCa, ..source(1) }, + DataSource { ssl: DbSslMode::Require, ssl_cert_file: "unused.pem".into(), ..source(1) }, + ]; + for source in sources { + let error = match Pool::connect(&source, None, None).await { + Ok(_) => panic!("unsupported settings were accepted"), + Err(error) => error, + }; + assert_eq!(error.kind, ErrorKind::InvalidInput); + } + } + + #[tokio::test] + #[ignore = "requires CLICKHOUSE_TEST_HOST, CLICKHOUSE_TEST_USER, and CLICKHOUSE_TEST_PASSWORD"] + async fn clickhouse_cloud_transport() { + let source = DataSource { + driver: DbDriver::Clickhouse, + host: std::env::var("CLICKHOUSE_TEST_HOST").expect("CLICKHOUSE_TEST_HOST"), + port: Some(std::env::var("CLICKHOUSE_TEST_PORT").ok().and_then(|v| v.parse().ok()).unwrap_or(8443)), + user: std::env::var("CLICKHOUSE_TEST_USER").expect("CLICKHOUSE_TEST_USER"), + ssl: DbSslMode::Require, + options: DbOptions { + time_zone: "UTC".into(), + startup_script: "SET max_threads = 2".into(), + ..Default::default() + }, + properties: vec![DbProperty { name: "max_block_size".into(), value: "50".into() }], + ..Default::default() + }; + let password = std::env::var("CLICKHOUSE_TEST_PASSWORD").expect("CLICKHOUSE_TEST_PASSWORD"); + let pool = Pool::connect(&source, Some(&password), None).await.unwrap(); + let cancel = CancellationToken::new(); + let settings = pool + .execute("SELECT getSetting('max_threads'), getSetting('max_block_size'), timezone()", 1, &cancel) + .await + .unwrap(); + assert_eq!(settings.rows, vec![vec![Some("2".into()), Some("50".into()), Some("UTC".into())]]); + let result = pool.execute("SELECT toUInt64('18446744073709551615') AS n, CAST(NULL AS Nullable(String)) AS missing, CAST('ᴺᵁᴸᴸ' AS Nullable(String)) AS text, 'a\\nb' AS lines", 1, &cancel).await.unwrap(); + assert_eq!( + result.rows, + vec![vec![Some("18446744073709551615".into()), None, Some("ᴺᵁᴸᴸ".into()), Some("a\nb".into())]] + ); + assert_eq!(result.columns[1].type_name, "Nullable(String)"); + let limited = pool.execute("SELECT number FROM numbers(100)", 2, &cancel).await.unwrap(); + assert_eq!(limited.rows.len(), 2); + assert!(limited.has_more); + let marker = format!("transport_cancel_{}", uuid::Uuid::new_v4().simple()); + let sql = format!("SELECT sleepEachRow(0.05) FROM numbers(50) /* {marker} */"); + let stop = CancellationToken::new(); + let query = pool.execute(&sql, 1, &stop); + let cancel_query = async { + tokio::time::sleep(Duration::from_millis(500)).await; + stop.cancel(); + }; + let (cancelled, ()) = tokio::join!(query, cancel_query); + let cancelled = cancelled.unwrap_err(); + assert_eq!(cancelled.kind, ErrorKind::Cancelled); + assert_eq!(cancelled.message, "cancelled"); + let active = pool.execute(&format!("SELECT count() FROM system.processes WHERE position(query, '{marker}') > 0 AND query_id != currentQueryID()"), 1, &cancel).await.unwrap(); + assert_eq!(active.rows, vec![vec![Some("0".into())]]); + let table = format!("transport_test_{}", uuid::Uuid::new_v4().simple()); + pool.execute(&format!("CREATE TEMPORARY TABLE {table} (n UInt8)"), 0, &cancel).await.unwrap(); + let continuity = async { + pool.execute(&format!("INSERT INTO {table} VALUES (7)"), 0, &cancel).await?; + pool.execute(&format!("SELECT n FROM {table}"), 1, &cancel).await + } + .await; + let cleanup = pool.execute(&format!("DROP TEMPORARY TABLE {table}"), 0, &cancel).await; + assert_eq!(continuity.unwrap().rows, vec![vec![Some("7".into())]]); + cleanup.unwrap(); + pool.close().await; + assert!(pool.is_closed()); + } +} diff --git a/crates/ide-db/src/conn.rs b/crates/ide-db/src/conn.rs index aa9fd7f43..94bd01348 100644 --- a/crates/ide-db/src/conn.rs +++ b/crates/ide-db/src/conn.rs @@ -106,6 +106,7 @@ pub enum Pool { Postgres(PgPool), Mysql(MySqlPool), SqlServer(Bb8Pool), + Clickhouse(crate::clickhouse::ClickhousePool), } /// Runs `$body` with `$p` bound to the concrete pool of each driver. @@ -115,7 +116,7 @@ macro_rules! with_pool { Pool::Sqlite($p) => $body, Pool::Postgres($p) => $body, Pool::Mysql($p) => $body, - Pool::SqlServer(_) => unreachable!(), + Pool::SqlServer(_) | Pool::Clickhouse(_) => unreachable!(), } }; } @@ -503,6 +504,9 @@ impl Setup { value.parse::().map_err(|_| IdeError::invalid(format!("{what} must be a whole number: {value}"))) }; match (source.driver, name.to_ascii_lowercase().as_str()) { + (DbDriver::Clickhouse, "statement_cache_capacity") => { + return Err(IdeError::invalid("ClickHouse HTTP does not support statement caching")); + } (_, "statement_cache_capacity") => s.statement_cache = Some(number(&name)? as usize), (_, "connect_timeout") => s.connect_timeout = Some(Duration::from_secs(number(&name)?)), (DbDriver::Postgres, "application_name") => s.application_name = Some(value.to_owned()), @@ -512,6 +516,9 @@ impl Setup { _ => s.params.push((name, value.to_owned())), } } + if source.driver == DbDriver::Clickhouse && !o.time_zone.trim().is_empty() { + s.params.push(("session_timezone".into(), o.time_zone.trim().into())); + } let script = o.startup_script.trim(); if !script.is_empty() { s.statements.extend(crate::split::split(script, source.driver).into_iter().map(|st| st.text)); @@ -527,6 +534,7 @@ impl Pool { Pool::Postgres(_) => DbDriver::Postgres, Pool::Mysql(_) => DbDriver::Mysql, Pool::SqlServer(_) => DbDriver::SqlServer, + Pool::Clickhouse(_) => DbDriver::Clickhouse, } } @@ -534,6 +542,9 @@ impl Pool { pub async fn connect(source: &DataSource, password: Option<&str>, root: Option<&Path>) -> IdeResult { const MAX: u32 = 4; let setup = Setup::of(source)?; + if source.driver == DbDriver::Clickhouse { + return crate::clickhouse::ClickhousePool::connect(source, password, setup).await.map(Pool::Clickhouse); + } let timeout = setup.connect_timeout.unwrap_or(Duration::from_secs(30)); let init = Arc::new(setup.statements); // Runs the session statements (startup script…) on every new connection. @@ -695,12 +706,14 @@ impl Pool { conn.query("SELECT 1", &[]).await.map_err(mssql_err)?.into_results().await.map_err(mssql_err)?; Ok(Pool::SqlServer(pool)) } + DbDriver::Clickhouse => unreachable!(), } } pub async fn close(&self) { match self { Pool::SqlServer(_) => {} + Pool::Clickhouse(p) => p.close(), _ => with_pool!(self, p => p.close().await), } } @@ -708,12 +721,16 @@ impl Pool { pub fn is_closed(&self) -> bool { match self { Pool::SqlServer(_) => false, + Pool::Clickhouse(p) => p.is_closed(), _ => with_pool!(self, p => p.is_closed()), } } /// Runs `sql` (no row limit beyond `max`), for introspection: cells are text (see [`Cells::Text`]). pub async fn rows(&self, sql: &str, max: usize) -> IdeResult>>> { + if let Pool::Clickhouse(p) = self { + return Ok(p.execute(sql, max, &CancellationToken::new()).await?.rows); + } let fetched: IdeResult = match self { Pool::SqlServer(p) => { let mut conn = p.get_owned().await.map_err(mssql_pool_err)?; @@ -727,6 +744,9 @@ impl Pool { /// Runs one statement and keeps at most `max_rows` rows. Cancelling stops /// the statement on the server (PostgreSQL, MySQL) and drops the connection. pub async fn execute(&self, sql: &str, max_rows: usize, cancel: &CancellationToken) -> IdeResult { + if let Pool::Clickhouse(p) = self { + return p.execute(sql, max_rows, cancel).await; + } let mut conn = Held::acquire(self).await?; let r = conn.execute(self, sql, max_rows, cancel).await; if cancel.is_cancelled() { @@ -740,6 +760,7 @@ impl Pool { /// row when flagged (table editor submits). pub async fn transaction(&self, statements: &[(String, bool)]) -> IdeResult { match self { + Pool::Clickhouse(_) => Err(crate::clickhouse::transaction_error()), Pool::SqlServer(p) => { let mut conn = p.get_owned().await.map_err(mssql_pool_err)?; let result = async { @@ -790,6 +811,7 @@ pub enum Held { Postgres(PoolConnection), Mysql(PoolConnection), SqlServer(Box>), + Clickhouse(crate::clickhouse::ClickhousePool), } impl Held { @@ -799,6 +821,12 @@ impl Held { Pool::Postgres(p) => Held::Postgres(p.acquire().await.map_err(db_err)?), Pool::Mysql(p) => Held::Mysql(p.acquire().await.map_err(db_err)?), Pool::SqlServer(p) => Held::SqlServer(Box::new(p.get_owned().await.map_err(mssql_pool_err)?)), + Pool::Clickhouse(p) => { + if p.is_closed() { + return Err(IdeError::new(ErrorKind::Io, "ClickHouse connection is closed")); + } + Held::Clickhouse(p.clone()) + } }) } @@ -806,13 +834,14 @@ impl Held { pub async fn command(&mut self, sql: &str) -> IdeResult<()> { match self { Held::SqlServer(c) => mssql_command(c, sql).await, + Held::Clickhouse(_) => Err(crate::clickhouse::transaction_error()), other => { let sql = sqlx::AssertSqlSafe(sql.to_owned()); match other { Held::Sqlite(c) => sqlx::raw_sql(sql).execute(&mut **c).await.map(drop), Held::Postgres(c) => sqlx::raw_sql(sql).execute(&mut **c).await.map(drop), Held::Mysql(c) => sqlx::raw_sql(sql).execute(&mut **c).await.map(drop), - Held::SqlServer(_) => unreachable!(), + Held::SqlServer(_) | Held::Clickhouse(_) => unreachable!(), } .map_err(db_err) } @@ -833,6 +862,7 @@ impl Held { Held::SqlServer(mut c) => { c.0.take(); } + Held::Clickhouse(_) => {} } } @@ -845,6 +875,9 @@ impl Held { max_rows: usize, cancel: &CancellationToken, ) -> IdeResult { + if let Held::Clickhouse(c) = self { + return c.execute(sql, max_rows, cancel).await; + } let started = Instant::now(); // `$id`: the query naming the server process, `$kill`: how another connection stops it. macro_rules! run { @@ -887,6 +920,7 @@ impl Held { () = cancel.cancelled() => return Err(IdeError::cancelled()), } } + Held::Clickhouse(_) => unreachable!(), }; let (columns, rows, affected, has_more, first) = fetched; let elapsed_ms = started.elapsed().as_secs_f64() * 1000.0; @@ -907,7 +941,7 @@ pub fn begin_statement(driver: DbDriver) -> &'static str { match driver { DbDriver::Mysql => "START TRANSACTION", DbDriver::SqlServer => "BEGIN TRAN", - DbDriver::Sqlite | DbDriver::Postgres => "BEGIN", + DbDriver::Sqlite | DbDriver::Postgres | DbDriver::Clickhouse => "BEGIN", } } diff --git a/crates/ide-db/src/lib.rs b/crates/ide-db/src/lib.rs index 5f7034c5e..f12f6a63f 100644 --- a/crates/ide-db/src/lib.rs +++ b/crates/ide-db/src/lib.rs @@ -2,6 +2,7 @@ //! execution ([`conn`]), introspection, table data paging / editing ([`sql`]) //! and statement splitting ([`split`]), for SQLite, PostgreSQL and MySQL/MariaDB. +mod clickhouse; pub mod conn; pub mod history; pub mod plan; @@ -547,12 +548,14 @@ impl DbManager { })?; let version = match source.driver { DbDriver::Sqlite => "SELECT 'SQLite ' || sqlite_version()", - DbDriver::Postgres => "SELECT version()", + DbDriver::Postgres | DbDriver::Clickhouse => "SELECT version()", DbDriver::Mysql => "SELECT CONCAT(@@version_comment, ' ', VERSION())", DbDriver::SqlServer => "SELECT @@VERSION", }; - let v = pool.rows(version, 1).await.ok().and_then(|r| r.into_iter().next()?.into_iter().next()?); + let result = pool.rows(version, 1).await; pool.close().await; + let rows = if source.driver == DbDriver::Clickhouse { result? } else { result.unwrap_or_default() }; + let v = rows.into_iter().next().and_then(|r| r.into_iter().next().flatten()); Ok(v.unwrap_or_else(|| "Connected".into())) } @@ -701,8 +704,11 @@ impl DbManager { _ => indexes.push(IndexInfo { name, unique: flag(&unique), columns: vec![col] }), } } - let fk_rows = - pool.rows(&sql::table_foreign_keys(driver, schema, table), MAX_META_ROWS).await.unwrap_or_default(); + let fk_rows = if driver == DbDriver::Clickhouse { + Vec::new() + } else { + pool.rows(&sql::table_foreign_keys(driver, schema, table), MAX_META_ROWS).await.unwrap_or_default() + }; let mut foreign_keys = group_foreign_keys(fk_rows, &[]); // SQLite keys that reference the primary key implicitly: look it up. for k in foreign_keys.iter_mut().filter(|k| k.ref_columns.is_empty()) { @@ -715,13 +721,15 @@ impl DbManager { .collect(); } foreign_keys.retain(|k| k.columns.len() == k.ref_columns.len()); - let triggers = pool - .rows(&sql::triggers(driver, schema, table), MAX_META_ROWS) - .await - .unwrap_or_else(|e| { + let trigger_rows = if driver == DbDriver::Clickhouse { + Vec::new() + } else { + pool.rows(&sql::triggers(driver, schema, table), MAX_META_ROWS).await.unwrap_or_else(|e| { tracing::warn!("triggers: {}", e.message); Vec::new() }) + }; + let triggers = trigger_rows .into_iter() .filter_map(|r| { let mut it = r.into_iter(); @@ -822,6 +830,11 @@ impl DbManager { let ddl = cells(pool.rows(&q, 1).await?, 1).into_iter().next(); ddl.map(|d| format!("{d};\n")).ok_or_else(missing) } + DbDriver::Clickhouse => { + let q = format!("SHOW CREATE TABLE {}", sql::qualified(driver, schema, name)); + let ddl = cells(pool.rows(&q, 1).await?, 0).into_iter().next(); + ddl.map(|d| format!("{};\n", d.trim_end().trim_end_matches(';'))).ok_or_else(missing) + } DbDriver::SqlServer => { let obj = format!("{}.{}", schema, name).replace('\'', "''"); let ddl = cells(pool.rows(&format!("SELECT OBJECT_DEFINITION(OBJECT_ID('{}'))", obj), 1).await?, 0) @@ -902,10 +915,14 @@ impl DbManager { } } // Foreign keys are a nicety (JOIN completion): a server that refuses the catalog query still completes. - let fk_rows = pool.rows(sql::foreign_keys(driver), MAX_META_ROWS).await.unwrap_or_else(|e| { - tracing::warn!("foreign keys: {}", e.message); + let fk_rows = if driver == DbDriver::Clickhouse { Vec::new() - }); + } else { + pool.rows(sql::foreign_keys(driver), MAX_META_ROWS).await.unwrap_or_else(|e| { + tracing::warn!("foreign keys: {}", e.message); + Vec::new() + }) + }; let foreign_keys = group_foreign_keys(fk_rows, &tables); let default_schema = match sql::current_schema(driver) { None => "main".to_owned(), @@ -980,6 +997,9 @@ impl DbManager { let Some(session) = session else { return pool.execute(statement, max_rows, cancel).await; }; + if pool.driver() == DbDriver::Clickhouse { + return Err(IdeError::invalid("ClickHouse manual transactions are unsupported")); + } let mut held = session.lock().await; if held.is_none() { let mut conn = Held::acquire(&pool).await?; @@ -998,8 +1018,11 @@ impl DbManager { /// Switches `id` between auto-commit and manual commit. Leaving manual /// mode rolls back what was not committed. - pub async fn set_manual_commit(&self, id: &str, manual: bool) { + pub async fn set_manual_commit(&self, id: &str, manual: bool) -> IdeResult<()> { if manual { + if self.source(id)?.driver == DbDriver::Clickhouse { + return Err(IdeError::invalid("ClickHouse manual transactions are unsupported")); + } self.sessions.lock().entry(id.to_owned()).or_default(); } else { let session = self.sessions.lock().remove(id); @@ -1007,10 +1030,14 @@ impl DbManager { end_session(&s).await; } } + Ok(()) } /// Commits (or rolls back) the open transaction of a manual-commit data source. pub async fn finish_transaction(&self, id: &str, commit: bool) -> IdeResult<()> { + if self.source(id).is_ok_and(|s| s.driver == DbDriver::Clickhouse) { + return Err(IdeError::invalid("ClickHouse manual transactions are unsupported")); + } let session = self.sessions.lock().get(id).cloned(); let Some(session) = session else { return Ok(()) }; let mut held = session.lock().await; @@ -1052,6 +1079,9 @@ impl DbManager { /// Applies the table editor's pending changes in one transaction. pub async fn apply_edits(&self, id: &str, schema: &str, table: &str, edits: &[DbRowEdit]) -> IdeResult { + if self.source(id)?.driver == DbDriver::Clickhouse { + return Err(IdeError::invalid("ClickHouse table editing is unsupported")); + } let pool = self.pool(id).await?; let driver = pool.driver(); let statements = edits @@ -1064,3 +1094,37 @@ impl DbManager { pool.transaction(&statements).await } } + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn clickhouse_rejects_manual_transactions_and_table_edits_without_connecting() { + let tmp = tempfile::tempdir().unwrap(); + let root = tmp.path().to_path_buf(); + let ssh = Arc::new(SshManager::new(root.join("ssh-passwords.json"))); + let db = DbManager::new(root.clone(), root.join("passwords.json"), root.join("history.json"), ssh); + db.save_sources(vec![DataSource { + id: "clickhouse".into(), + name: "clickhouse".into(), + driver: DbDriver::Clickhouse, + host: "127.0.0.1".into(), + port: Some(1), + ..Default::default() + }]) + .await + .unwrap(); + let error = db.set_manual_commit("clickhouse", true).await.unwrap_err(); + assert!(error.message.contains("manual transactions are unsupported")); + assert!(db.sessions.lock().is_empty()); + for commit in [false, true] { + let error = db.finish_transaction("clickhouse", commit).await.unwrap_err(); + assert!(error.message.contains("manual transactions are unsupported")); + } + let error = db.apply_edits("clickhouse", "default", "events", &[]).await.unwrap_err(); + assert!(error.message.contains("table editing is unsupported")); + assert!(!db.is_connected("clickhouse")); + db.set_manual_commit("clickhouse", false).await.unwrap(); + } +} diff --git a/crates/ide-db/src/plan.rs b/crates/ide-db/src/plan.rs index af3789137..32b217b6e 100644 --- a/crates/ide-db/src/plan.rs +++ b/crates/ide-db/src/plan.rs @@ -35,6 +35,7 @@ pub fn explain_statement(driver: DbDriver, statement: &str) -> String { match driver { DbDriver::Sqlite => format!("EXPLAIN QUERY PLAN {s}"), DbDriver::Postgres => format!("EXPLAIN (FORMAT JSON) {s}"), + DbDriver::Clickhouse => format!("EXPLAIN json = 1 {s}"), DbDriver::Mysql => format!("EXPLAIN {s}"), DbDriver::SqlServer => format!("SET SHOWPLAN_TEXT ON; {s}"), } @@ -44,7 +45,7 @@ pub fn explain_statement(driver: DbDriver, statement: &str) -> String { pub fn parse(driver: DbDriver, result: &QueryResult) -> IdeResult> { match driver { DbDriver::Sqlite => Ok(sqlite(&result.rows)), - DbDriver::Postgres => { + DbDriver::Postgres | DbDriver::Clickhouse => { let text: String = result.rows.iter().filter_map(|r| r.first().cloned().flatten()).collect::>().join("\n"); let json: Value = serde_json::from_str(&text).map_err(|e| IdeError::invalid(format!("plan: {e}")))?; @@ -199,6 +200,7 @@ mod tests { assert_eq!(explain_statement(DbDriver::Sqlite, "select 1;"), "EXPLAIN QUERY PLAN select 1"); assert_eq!(explain_statement(DbDriver::Postgres, " select 1 "), "EXPLAIN (FORMAT JSON) select 1"); assert_eq!(explain_statement(DbDriver::Mysql, "select 1"), "EXPLAIN select 1"); + assert_eq!(explain_statement(DbDriver::Clickhouse, " select 1;"), "EXPLAIN json = 1 select 1"); assert_eq!(explain_statement(DbDriver::SqlServer, "select 1"), "SET SHOWPLAN_TEXT ON; select 1"); } @@ -254,6 +256,21 @@ mod tests { assert_eq!((plan[3].parent, plan[3].operation.as_str()), (Some(2), "Seq Scan")); } + #[test] + fn reads_clickhouse_json_plans_and_rejects_invalid_json() { + let json = r#"[{"Plan": {"Node Type": "Expression", "Node Id": "Expression_5", + "Description": "Project names", "Plans": [ + {"Node Type": "ReadFromMergeTree", "Node Id": "ReadFromMergeTree_0", + "Description": "default.events"} + ]}}]"#; + let plan = parse(DbDriver::Clickhouse, &result(&["explain"], vec![vec![Some(json)]])).unwrap(); + assert_eq!(plan.len(), 2); + assert_eq!((plan[0].parent, plan[0].operation.as_str()), (None, "Expression")); + assert_eq!((plan[1].parent, plan[1].operation.as_str()), (Some(0), "ReadFromMergeTree")); + assert!(plan[1].properties.iter().any(|p| p.name == "Description" && p.value == "default.events")); + assert!(parse(DbDriver::Clickhouse, &result(&["explain"], vec![vec![Some("invalid")]])).is_err()); + } + #[test] fn reads_mysql_rows() { let r = result( diff --git a/crates/ide-db/src/source.rs b/crates/ide-db/src/source.rs index a3590ae6e..c048430d8 100644 --- a/crates/ide-db/src/source.rs +++ b/crates/ide-db/src/source.rs @@ -18,6 +18,7 @@ pub enum DbDriver { Mysql, /// Microsoft SQL Server. SqlServer, + Clickhouse, } impl DbDriver { @@ -27,6 +28,7 @@ impl DbDriver { DbDriver::Postgres => Some(5432), DbDriver::Mysql => Some(3306), DbDriver::SqlServer => Some(1433), + DbDriver::Clickhouse => Some(8123), } } } @@ -253,6 +255,7 @@ pub fn scheme_driver(scheme: &str) -> Option { "mysql" | "mariadb" | "tidb" | "singlestore" | "memsql" | "oceanbase" => Some(DbDriver::Mysql), "sqlserver" | "mssql" | "mssqlserver" => Some(DbDriver::SqlServer), "sqlite" | "sqlite3" => Some(DbDriver::Sqlite), + "clickhouse" | "http" | "https" => Some(DbDriver::Clickhouse), _ => None, } } @@ -260,7 +263,7 @@ pub fn scheme_driver(scheme: &str) -> Option { /// Parses `scheme://[user[:password]@]host[:port][/database][?k=v&…]`, `jdbc:` forms /// included, and `sqlite:path`. pub fn parse_url(raw: &str) -> IdeResult { - let bad = || IdeError::invalid(format!("not a database URL: {raw}")); + let bad = || IdeError::invalid("invalid database URL"); let url = raw.trim(); let url = url.strip_prefix("jdbc:").or_else(|| url.strip_prefix("JDBC:")).unwrap_or(url); let (scheme, rest) = url.split_once(':').ok_or_else(bad)?; @@ -367,6 +370,15 @@ impl DataSource { } properties.append(&mut s.properties); s.properties = properties; + if s.driver == DbDriver::Clickhouse { + let raw_url = self.url.trim().to_ascii_lowercase(); + let url = raw_url.strip_prefix("jdbc:").unwrap_or(&raw_url); + if url.starts_with("https://") { + s.ssl = DbSslMode::VerifyFull; + } else if url.starts_with("http://") { + s.ssl = DbSslMode::Disable; + } + } Ok(s) } @@ -519,4 +531,31 @@ mod tests { assert!(DataSource { url: "mysql://h/db".into(), ..s.clone() }.resolved().is_err()); assert_eq!(DataSource { url_only: false, ..s }.resolved().unwrap().host, "localhost"); } + + #[test] + fn clickhouse_urls_resolve_http_https_and_credentials() { + assert_eq!(DbDriver::Clickhouse.default_port(), Some(8123)); + let source = DataSource { + driver: DbDriver::Clickhouse, + url_only: true, + url: "https://reader:s%40cret@db.example:8443/analytics?max_execution_time=30".into(), + ..Default::default() + }; + let resolved = source.resolved().unwrap(); + assert_eq!(resolved.host, "db.example"); + assert_eq!(resolved.port, Some(8443)); + assert_eq!(resolved.database, "analytics"); + assert_eq!(resolved.user, "reader"); + assert_eq!(resolved.ssl, DbSslMode::VerifyFull); + assert_eq!(source.url_password().as_deref(), Some("s@cret")); + assert_eq!(resolved.properties[0].name, "max_execution_time"); + let http = DataSource { url: "http://localhost/default".into(), ..source.clone() }.resolved().unwrap(); + assert_eq!(http.ssl, DbSslMode::Disable); + let jdbc = DataSource { url: "jdbc:clickhouse://localhost/analytics".into(), ..source.clone() }; + assert_eq!(jdbc.resolved().unwrap().database, "analytics"); + let secure = DataSource { url: "https://db.example/default?sslmode=disable".into(), ..source }; + assert_eq!(secure.resolved().unwrap().ssl, DbSslMode::VerifyFull); + let error = parse_url("clickhouse://reader:secret@db.example:invalid/default").unwrap_err(); + assert!(!error.message.contains("secret")); + } } diff --git a/crates/ide-db/src/split.rs b/crates/ide-db/src/split.rs index e9cdbc84f..b2ee3a108 100644 --- a/crates/ide-db/src/split.rs +++ b/crates/ide-db/src/split.rs @@ -28,6 +28,7 @@ enum State { /// Every statement of `text` that contains more than whitespace and comments. pub fn split(text: &str, driver: DbDriver) -> Vec { let mysql = driver == DbDriver::Mysql; + let clickhouse = driver == DbDriver::Clickhouse; let chars: Vec<(usize, char)> = text.char_indices().collect(); // UTF-16 offset of each char index (plus the end). let mut u16_at = Vec::with_capacity(chars.len() + 1); @@ -81,7 +82,7 @@ pub fn split(text: &str, driver: DbDriver) -> Vec { state = State::Double; has_code = true; } - '`' if mysql => { + '`' if mysql || clickhouse => { state = State::Backtick; has_code = true; } @@ -117,7 +118,7 @@ pub fn split(text: &str, driver: DbDriver) -> Vec { _ => {} }, State::Single => { - if mysql && c == '\\' { + if (mysql || clickhouse) && c == '\\' { i += 1; } else if c == '\'' { if next == Some('\'') { @@ -128,7 +129,9 @@ pub fn split(text: &str, driver: DbDriver) -> Vec { } } State::Double => { - if c == '"' { + if clickhouse && c == '\\' { + i += 1; + } else if c == '"' { if next == Some('"') { i += 1; } else { @@ -137,7 +140,9 @@ pub fn split(text: &str, driver: DbDriver) -> Vec { } } State::Backtick => { - if c == '`' { + if clickhouse && (c == '\\' || (c == '`' && next == Some('`'))) { + i += 1; + } else if c == '`' { state = State::Code; } } @@ -224,6 +229,12 @@ mod tests { assert_eq!(texts(sql, DbDriver::Mysql), vec!["select `a;b`, 'it\\'s;' from t # x;y", "select 2"]); } + #[test] + fn clickhouse_backticks_and_escaped_quotes_keep_statements_intact() { + let query = r#"SELECT `a\`;b`, "c\";d", 'it\'s;safe'; SELECT 2"#; + assert_eq!(texts(query, DbDriver::Clickhouse), vec![r#"SELECT `a\`;b`, "c\";d", 'it\'s;safe'"#, "SELECT 2"]); + } + #[test] fn postgres_dollar_quotes() { let sql = diff --git a/crates/ide-db/src/sql.rs b/crates/ide-db/src/sql.rs index 79d6c1ec5..bd281a859 100644 --- a/crates/ide-db/src/sql.rs +++ b/crates/ide-db/src/sql.rs @@ -7,6 +7,7 @@ use specta::Type; pub fn quote_ident(driver: DbDriver, name: &str) -> String { match driver { + DbDriver::Clickhouse => format!("`{}`", name.replace('\\', "\\\\").replace('`', "\\`")), DbDriver::Mysql => format!("`{}`", name.replace('`', "``")), DbDriver::SqlServer => format!("[{}]", name.replace(']', "]]")), _ => format!("\"{}\"", name.replace('"', "\"\"")), @@ -15,7 +16,7 @@ pub fn quote_ident(driver: DbDriver, name: &str) -> String { pub fn quote_literal(driver: DbDriver, value: &str) -> String { let mut s = value.replace('\'', "''"); - if driver == DbDriver::Mysql { + if matches!(driver, DbDriver::Mysql | DbDriver::Clickhouse) { s = s.replace('\\', "\\\\"); } format!("'{s}'") @@ -48,6 +49,10 @@ pub const INTERNAL_DATABASES: &[&str] = &["rdsadmin", "azure_maintenance", "azur pub fn schemas(driver: DbDriver, system: bool) -> &'static str { match driver { DbDriver::Sqlite => "SELECT name FROM pragma_database_list WHERE name <> 'temp' ORDER BY seq", + DbDriver::Clickhouse if system => "SELECT name FROM system.databases ORDER BY name", + DbDriver::Clickhouse => { + "SELECT name FROM system.databases WHERE lower(name) NOT IN ('system', 'information_schema') ORDER BY name" + } DbDriver::Postgres if system => { "SELECT schema_name FROM information_schema.schemata \ WHERE schema_name NOT LIKE 'pg\\_toast%' AND schema_name NOT LIKE 'pg\\_temp%' ORDER BY 1" @@ -73,6 +78,12 @@ pub fn schemas(driver: DbDriver, system: bool) -> &'static str { /// Rows: `name, kind` where kind is `table`, `view` or `materialized view`. pub fn tables(driver: DbDriver, schema: &str) -> String { match driver { + DbDriver::Clickhouse => format!( + "SELECT name, CASE engine WHEN 'View' THEN 'view' WHEN 'MaterializedView' THEN 'materialized view' \ + WHEN 'WindowView' THEN 'view' ELSE 'table' END FROM system.tables \ + WHERE database = {} AND is_temporary = 0 ORDER BY name", + quote_literal(driver, schema) + ), DbDriver::Sqlite => format!( "SELECT name, type FROM {}.sqlite_master WHERE type IN ('table', 'view') \ AND name NOT LIKE 'sqlite\\_%' ESCAPE '\\' ORDER BY name", @@ -101,6 +112,12 @@ pub fn tables(driver: DbDriver, schema: &str) -> String { pub fn columns(driver: DbDriver, schema: &str, table: &str) -> String { let (s, t) = (quote_literal(driver, schema), quote_literal(driver, table)); match driver { + DbDriver::Clickhouse => format!( + "SELECT name, type, CASE WHEN startsWith(type, 'Nullable(') \ + OR startsWith(type, 'LowCardinality(Nullable(') THEN 'YES' ELSE 'NO' END, \ + if(default_kind = '', NULL, default_expression), is_in_primary_key \ + FROM system.columns WHERE database = {s} AND table = {t} ORDER BY position" + ), DbDriver::Sqlite => format!( "SELECT name, type, CASE WHEN \"notnull\" = 0 THEN 'YES' ELSE 'NO' END, dflt_value, \ CASE WHEN pk > 0 THEN 1 ELSE 0 END FROM pragma_table_info({t}, {s}) ORDER BY cid" @@ -138,6 +155,12 @@ pub fn columns(driver: DbDriver, schema: &str, table: &str) -> String { pub fn indexes(driver: DbDriver, schema: &str, table: &str) -> String { let (s, t) = (quote_literal(driver, schema), quote_literal(driver, table)); match driver { + DbDriver::Clickhouse => format!( + "SELECT * FROM (SELECT 'PRIMARY KEY' AS index_name, 0 AS is_unique, primary_key AS expression \ + FROM system.tables WHERE database = {s} AND name = {t} AND primary_key != '' \ + UNION ALL SELECT name, 0, expr FROM system.data_skipping_indices \ + WHERE database = {s} AND table = {t}) ORDER BY index_name" + ), DbDriver::Sqlite => format!( "SELECT il.name, il.\"unique\", ii.name FROM pragma_index_list({t}, {s}) il \ JOIN pragma_index_info(il.name, {s}) ii ORDER BY il.name, ii.seqno" @@ -169,6 +192,15 @@ pub fn indexes(driver: DbDriver, schema: &str, table: &str) -> String { /// for SQL completion, grouped by table in column order. pub fn completion_columns(driver: DbDriver) -> &'static str { match driver { + DbDriver::Clickhouse => { + "SELECT c.database, c.table, CASE t.engine WHEN 'View' THEN 'view' \ + WHEN 'MaterializedView' THEN 'materialized view' WHEN 'WindowView' THEN 'view' ELSE 'table' END, \ + c.name, c.type, CASE WHEN startsWith(c.type, 'Nullable(') \ + OR startsWith(c.type, 'LowCardinality(Nullable(') THEN 'YES' ELSE 'NO' END, c.is_in_primary_key, 1 \ + FROM system.columns c JOIN system.tables t ON t.database = c.database AND t.name = c.table \ + WHERE lower(c.database) NOT IN ('system', 'information_schema') AND t.is_temporary = 0 \ + ORDER BY c.database, c.table, c.position LIMIT 50000" + } DbDriver::Sqlite => { "SELECT 'main', m.name, m.type, p.name, p.type, CASE WHEN p.\"notnull\" = 0 THEN 'YES' ELSE 'NO' END, \ CASE WHEN p.pk > 0 THEN 1 ELSE 0 END, \ @@ -221,6 +253,7 @@ pub fn completion_columns(driver: DbDriver) -> &'static str { /// when the key references the primary key implicitly. pub fn foreign_keys(driver: DbDriver) -> &'static str { match driver { + DbDriver::Clickhouse => "SELECT '', '', '', '', '', '', '' WHERE 0", DbDriver::Sqlite => { "SELECT 'main', m.name, f.id, f.\"from\", 'main', f.\"table\", f.\"to\" \ FROM sqlite_master m JOIN pragma_foreign_key_list(m.name) f \ @@ -258,6 +291,7 @@ pub fn foreign_keys(driver: DbDriver) -> &'static str { pub fn table_foreign_keys(driver: DbDriver, schema: &str, table: &str) -> String { let (s, t) = (quote_literal(driver, schema), quote_literal(driver, table)); match driver { + DbDriver::Clickhouse => foreign_keys(driver).to_owned(), DbDriver::Sqlite => format!( "SELECT {s}, {t}, f.id, f.\"from\", {s}, f.\"table\", f.\"to\" FROM pragma_foreign_key_list({t}, {s}) f ORDER BY f.id, f.seq" ), @@ -276,6 +310,7 @@ pub fn table_foreign_keys(driver: DbDriver, schema: &str, table: &str) -> String /// The schema unqualified names resolve to (`None`: always `main`). pub fn current_schema(driver: DbDriver) -> Option<&'static str> { match driver { + DbDriver::Clickhouse => Some("SELECT currentDatabase()"), DbDriver::Sqlite => None, DbDriver::Postgres => Some("SELECT current_schema()"), DbDriver::Mysql => Some("SELECT DATABASE()"), @@ -362,6 +397,9 @@ fn where_key(driver: DbDriver, key: &[CellValue]) -> String { /// The DML statement for `edit`. Values are written as quoted literals, which /// every supported database coerces to the column type. pub fn edit_statement(driver: DbDriver, schema: &str, table: &str, edit: &DbRowEdit) -> Result { + if driver == DbDriver::Clickhouse { + return Err("ClickHouse table editing is unsupported: primary keys do not uniquely identify rows and edits require transactions".into()); + } let target = qualified(driver, schema, table); match edit { DbRowEdit::Update { key, values } => { @@ -450,6 +488,7 @@ pub fn postgres_ddl_indexes(oid: u32) -> String { pub fn triggers(driver: DbDriver, schema: &str, table: &str) -> String { let (s, t) = (quote_literal(driver, schema), quote_literal(driver, table)); match driver { + DbDriver::Clickhouse => "SELECT '', '', '' WHERE 0".to_owned(), DbDriver::Sqlite => format!( "SELECT name, sql FROM {}.sqlite_master WHERE type = 'trigger' AND tbl_name = {t} ORDER BY name", quote_ident(driver, schema) @@ -492,7 +531,7 @@ pub fn sqlite_trigger_event(sql: &str) -> (String, String) { pub fn routines(driver: DbDriver, schema: &str) -> Option { let s = quote_literal(driver, schema); match driver { - DbDriver::Sqlite => None, + DbDriver::Sqlite | DbDriver::Clickhouse => None, // Functions of extensions are left out, as in IDEA. DbDriver::Postgres => Some(format!( "SELECT p.proname, CASE WHEN p.prokind = 'p' THEN 'procedure' ELSE 'function' END, \ @@ -625,6 +664,71 @@ mod tests { assert_eq!(quote_literal(DbDriver::Mysql, "it's \\"), "'it''s \\\\'"); } + #[test] + fn clickhouse_escapes_backslashes_and_delimiters() { + let driver = DbDriver::Clickhouse; + assert_eq!(quote_literal(driver, r"path\O'Brien"), r"'path\\O''Brien'"); + assert_eq!(quote_literal(driver, r"\' OR 1 = 1 --"), r"'\\'' OR 1 = 1 --'"); + assert_eq!(quote_ident(driver, r"a\`b"), r"`a\\\`b`"); + assert_eq!(qualified(driver, "data base", "event`log"), r"`data base`.`event\`log`"); + } + + #[test] + fn clickhouse_metadata_preserves_types_defaults_and_nonunique_keys() { + let driver = DbDriver::Clickhouse; + assert_eq!(schemas(driver, true), "SELECT name FROM system.databases ORDER BY name"); + assert!(schemas(driver, false).contains("lower(name) NOT IN ('system', 'information_schema')")); + assert_eq!(current_schema(driver), Some("SELECT currentDatabase()")); + let relations = tables(driver, "event's\\db"); + assert!(relations.contains("WHEN 'MaterializedView' THEN 'materialized view'")); + assert!(relations.contains(r"WHERE database = 'event''s\\db' AND is_temporary = 0")); + let fields = columns(driver, "default", "events"); + assert!(fields.starts_with("SELECT name, type, CASE WHEN startsWith(type, 'Nullable(')")); + assert!(fields.contains("startsWith(type, 'LowCardinality(Nullable(')")); + assert!(fields.contains("if(default_kind = '', NULL, default_expression), is_in_primary_key")); + assert!(fields.ends_with("WHERE database = 'default' AND table = 'events' ORDER BY position")); + let keys = indexes(driver, "default", "events"); + assert!(keys.contains("'PRIMARY KEY' AS index_name, 0 AS is_unique, primary_key AS expression")); + assert!(keys.contains("SELECT name, 0, expr FROM system.data_skipping_indices")); + let completion = completion_columns(driver); + assert!(completion.contains("c.is_in_primary_key, 1")); + assert!(completion.contains("ORDER BY c.database, c.table, c.position LIMIT 50000")); + } + + #[test] + fn clickhouse_has_no_foreign_key_trigger_or_schema_routine_catalogs() { + let driver = DbDriver::Clickhouse; + assert_eq!(foreign_keys(driver), "SELECT '', '', '', '', '', '', '' WHERE 0"); + assert_eq!(table_foreign_keys(driver, "default", "events"), foreign_keys(driver)); + assert_eq!(triggers(driver, "default", "events"), "SELECT '', '', '' WHERE 0"); + assert!(routines(driver, "default").is_none()); + assert!(sequences(driver, "default").is_none()); + assert!(object_ddl(driver, "default", crate::DbObjectKind::Trigger, "t").is_none()); + } + + #[test] + fn clickhouse_pages_quote_keys_and_fetch_one_extra_row() { + let sort = [SortKey { column: "event`id".into(), descending: true }]; + assert_eq!( + table_page(DbDriver::Clickhouse, "default", "events", Some("WHERE id > 3"), &sort, 50, 25), + r"SELECT * FROM `default`.`events` WHERE id > 3 ORDER BY `event\`id` DESC LIMIT 26 OFFSET 50" + ); + } + + #[test] + fn clickhouse_keys_never_enable_transactional_table_edits() { + let key = vec![cv("id", Some("1"))]; + for edit in [ + DbRowEdit::Update { key: key.clone(), values: vec![cv("name", Some("new"))] }, + DbRowEdit::Delete { key }, + DbRowEdit::Insert { values: vec![cv("name", Some("new"))] }, + DbRowEdit::Insert { values: vec![] }, + ] { + let error = edit_statement(DbDriver::Clickhouse, "default", "events", &edit).unwrap_err(); + assert!(error.contains("primary keys do not uniquely identify rows")); + } + } + #[test] fn pages() { let sort = [SortKey { column: "id".into(), descending: true }]; diff --git a/crates/ide-db/tests/clickhouse.rs b/crates/ide-db/tests/clickhouse.rs new file mode 100644 index 000000000..d19a9ff16 --- /dev/null +++ b/crates/ide-db/tests/clickhouse.rs @@ -0,0 +1,136 @@ +use ide_core::{CancellationToken, IdeError, IdeResult}; +use ide_db::{CellValue, DataSource, DbDriver, DbManager, DbRowEdit, DbSslMode, SortKey, sql}; +use std::sync::Arc; +use std::time::{SystemTime, UNIX_EPOCH}; + +fn check(condition: bool, message: &str) -> IdeResult<()> { + if condition { Ok(()) } else { Err(IdeError::invalid(message)) } +} + +#[tokio::test] +#[ignore = "requires CLICKHOUSE_TEST_HOST, CLICKHOUSE_TEST_USER, and CLICKHOUSE_TEST_PASSWORD"] +async fn clickhouse_cloud_manager_catalogs_and_session_table() -> IdeResult<()> { + let tmp = tempfile::tempdir()?; + let root = tmp.path().to_path_buf(); + let ssh = Arc::new(ide_ssh::SshManager::new(root.join("ssh-passwords.json"))); + let db = DbManager::new(root.clone(), root.join("passwords.json"), root.join("history.json"), ssh); + let mut source = DataSource { + id: "clickhouse".into(), + name: "clickhouse".into(), + driver: DbDriver::Clickhouse, + host: std::env::var("CLICKHOUSE_TEST_HOST").expect("set CLICKHOUSE_TEST_HOST"), + port: Some( + std::env::var("CLICKHOUSE_TEST_PORT") + .map(|value| value.parse().expect("numeric CLICKHOUSE_TEST_PORT")) + .unwrap_or(8443), + ), + user: std::env::var("CLICKHOUSE_TEST_USER").expect("set CLICKHOUSE_TEST_USER"), + database: "default".into(), + ssl: DbSslMode::VerifyFull, + ..Default::default() + }; + source.introspection.show_system = true; + let password = std::env::var("CLICKHOUSE_TEST_PASSWORD").expect("set CLICKHOUSE_TEST_PASSWORD"); + check(!db.test(&source, Some(password.clone()), None).await?.is_empty(), "missing server version")?; + db.save_sources(vec![source]).await?; + db.set_password("clickhouse", Some(password), false).await?; + let cancel = CancellationToken::new(); + let suffix = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_nanos(); + let name = format!("manager_test_{}_{suffix}", std::process::id()); + let table = sql::quote_ident(DbDriver::Clickhouse, &name); + db.execute( + "clickhouse", + &format!( + "CREATE TEMPORARY TABLE {table} (id UInt64, label Nullable(String) DEFAULT NULL, \ + category LowCardinality(Nullable(String)) DEFAULT NULL, items Array(Nullable(UInt8)), \ + label_size UInt64 ALIAS length(ifNull(label, ''))) ENGINE = Memory" + ), + 0, + &cancel, + ) + .await?; + let checked: IdeResult<()> = async { + db.execute( + "clickhouse", + &format!("INSERT INTO {table} (id, label, items) VALUES (1, 'first', [1, NULL]), (1, NULL, [])"), + 0, + &cancel, + ) + .await?; + let schemas = db.schemas("clickhouse").await?; + check(schemas.iter().any(|schema| schema == "default"), "default database missing")?; + check(schemas.iter().any(|schema| schema == "system"), "system database missing")?; + let tables = db.tables("clickhouse", "system").await?; + check(tables.iter().any(|table| table.name == "one"), "system table listing missing one")?; + let details = db.table_details("clickhouse", "", &name).await?; + check(details.columns.len() == 5, "temporary table columns missing")?; + for (name, data_type, nullable) in [ + ("id", "UInt64", false), + ("label", "Nullable(String)", true), + ("category", "LowCardinality(Nullable(String))", true), + ("items", "Array(Nullable(UInt8))", false), + ] { + check( + details + .columns + .iter() + .any(|column| column.name == name && column.data_type == data_type && column.nullable == nullable), + "column type or nullability mismatch", + )?; + } + check( + details.columns.iter().any(|column| column.name == "label" && column.default.as_deref() == Some("NULL")), + "explicit default expression missing", + )?; + check(details.indexes.iter().all(|index| !index.unique), "ClickHouse index reported unique")?; + check(details.foreign_keys.is_empty() && details.triggers.is_empty(), "unsupported table objects returned")?; + let sort = [SortKey { column: "id".into(), descending: false }]; + let first = db.table_page("clickhouse", "", &name, None, &sort, 0, 1, &cancel).await?; + check(first.result.rows.len() == 1 && first.result.has_more, "first page limit failed")?; + let second = db.table_page("clickhouse", "", &name, None, &sort, 1, 1, &cancel).await?; + check(second.result.rows.len() == 1 && !second.result.has_more, "page offset failed")?; + let ddl = db.ddl("clickhouse", "", &name).await?; + check(ddl.contains(&name) && ddl.contains("Memory"), "temporary table DDL missing")?; + let completion = db.completion("clickhouse").await?; + check(completion.default_schema == "default", "completion default database mismatch")?; + check(completion.foreign_keys.is_empty(), "completion returned unsupported foreign keys")?; + if let Some(table) = completion.tables.first() { + let details = db.table_details("clickhouse", &table.schema, &table.name).await?; + check( + table.columns.iter().all(|column| { + details.columns.iter().any(|field| field.name == column.name && field.data_type == column.data_type) + }), + "completion columns differ from catalog", + )?; + } + let plan = db.explain("clickhouse", &format!("SELECT id FROM {table}"), &cancel).await?; + check(!plan.is_empty(), "query plan missing")?; + let value = r"path\O'Brien"; + let result = db + .execute("clickhouse", &format!("SELECT {}", sql::quote_literal(DbDriver::Clickhouse, value)), 1, &cancel) + .await?; + check(result.rows == vec![vec![Some(value.to_owned())]], "quoted literal changed value")?; + let edits = [DbRowEdit::Update { + key: vec![CellValue { column: "id".into(), value: Some("1".into()) }], + values: vec![CellValue { column: "label".into(), value: Some("changed".into()) }], + }]; + check( + db.apply_edits("clickhouse", "", &name, &edits) + .await + .err() + .is_some_and(|error| error.message.contains("table editing is unsupported")), + "supplied row key authorized table editing", + )?; + check(db.set_manual_commit("clickhouse", true).await.is_err(), "manual transactions were accepted")?; + for commit in [false, true] { + check(db.finish_transaction("clickhouse", commit).await.is_err(), "transaction control was accepted")?; + } + Ok(()) + } + .await; + let cleanup = db.execute("clickhouse", &format!("DROP TEMPORARY TABLE {table}"), 0, &cancel).await; + db.disconnect("clickhouse").await; + checked?; + cleanup?; + Ok(()) +} diff --git a/crates/ide-db/tests/sqlite.rs b/crates/ide-db/tests/sqlite.rs index f340ee536..b800cd154 100644 --- a/crates/ide-db/tests/sqlite.rs +++ b/crates/ide-db/tests/sqlite.rs @@ -197,7 +197,7 @@ async fn count(db: &DbManager) -> String { async fn manual_commit_keeps_a_transaction_until_commit_or_rollback() -> IdeResult<()> { let (_tmp, db) = setup().await; let c = CancellationToken::new(); - db.set_manual_commit("s1", true).await; + db.set_manual_commit("s1", true).await.unwrap(); db.execute("s1", "INSERT INTO users (name) VALUES ('dee')", 1, &c).await?; // The session sees its own change. assert_eq!(count(&db).await, "4"); @@ -208,10 +208,10 @@ async fn manual_commit_keeps_a_transaction_until_commit_or_rollback() -> IdeResu db.finish_transaction("s1", true).await?; // Leaving manual mode rolls back what is still open. db.execute("s1", "DELETE FROM users", 1, &c).await?; - db.set_manual_commit("s1", false).await; + db.set_manual_commit("s1", false).await.unwrap(); assert_eq!(count(&db).await, "4"); // Disconnecting with an open transaction does not hang. - db.set_manual_commit("s1", true).await; + db.set_manual_commit("s1", true).await.unwrap(); db.execute("s1", "DELETE FROM users", 1, &c).await?; db.disconnect("s1").await; assert_eq!(count(&db).await, "4"); diff --git a/crates/ide-host/src/ops/db.rs b/crates/ide-host/src/ops/db.rs index 63c8419a0..102bf2742 100644 --- a/crates/ide-host/src/ops/db.rs +++ b/crates/ide-host/src/ops/db.rs @@ -216,8 +216,7 @@ pub async fn connect(host: &Host, window: &str, id: &str) -> IdeResult<()> { /// manual mode rolls back what was not committed). pub async fn set_manual_commit(host: &Host, window: &str, id: &str, manual: bool) -> IdeResult<()> { let db = host.project(window)?.db.clone(); - db.set_manual_commit(id, manual).await; - Ok(()) + db.set_manual_commit(id, manual).await } /// Commits or rolls back the open transaction of a manual-commit data source. diff --git a/e2e/clickhouse-source.spec.ts b/e2e/clickhouse-source.spec.ts new file mode 100644 index 000000000..8a29e09ea --- /dev/null +++ b/e2e/clickhouse-source.spec.ts @@ -0,0 +1,42 @@ +import { expect, test } from "./fixtures"; + +test("configures a ClickHouse HTTP source and keeps transaction controls unavailable", async ({ + workbench: page, +}, info) => { + await page.evaluate(async () => { + const storePath = "/src/features/database/store.ts"; + const bindingsPath = "/src/bindings.ts"; + const { useDatabase } = await import(storePath); + const { commands } = await import(bindingsPath); + commands.dbTest = async (source: unknown) => { + document.documentElement.dataset.testedSource = JSON.stringify(source); + return { status: "ok", data: "ClickHouse" }; + }; + useDatabase.setState({ dialog: { kind: "source", source: null } }); + }); + const dialog = page.getByRole("dialog", { name: "New Data Source", exact: true }); + await dialog.getByRole("button", { name: "Data source type", exact: true }).click(); + await page.getByRole("option", { name: "ClickHouse", exact: true }).click(); + await expect(dialog.locator("#db-port")).toHaveValue("8123"); + await expect(dialog.locator("#db-user")).toHaveValue("default"); + await expect(dialog.locator("#db-database")).toHaveValue("default"); + await expect(dialog.locator("#db-url")).toHaveValue("http://localhost:8123/default"); + await dialog.locator("#db-url").fill("https://reader@db.example:8443/analytics"); + await expect(dialog.locator("#db-host")).toHaveValue("db.example"); + await expect(dialog.locator("#db-port")).toHaveValue("8443"); + await expect(dialog.locator("#db-user")).toHaveValue("reader"); + await expect(dialog.locator("#db-database")).toHaveValue("analytics"); + await dialog.getByRole("button", { name: "Test Connection", exact: true }).click(); + await expect(dialog.getByRole("status")).toHaveText("Succeeded: ClickHouse"); + expect(await page.evaluate(() => JSON.parse(document.documentElement.dataset.testedSource!))).toMatchObject({ + driver: "clickhouse", + host: "db.example", + port: 8443, + user: "reader", + database: "analytics", + ssl: "verifyFull", + }); + await dialog.screenshot({ path: info.outputPath("clickhouse-source.png") }); + await dialog.getByRole("tab", { name: "Options", exact: true }).click(); + await expect(dialog.getByRole("button", { name: "Transaction control", exact: true })).toBeDisabled(); +}); diff --git a/src/bindings.ts b/src/bindings.ts index ab0d32e51..f8cde1141 100644 --- a/src/bindings.ts +++ b/src/bindings.ts @@ -2405,7 +2405,7 @@ export type DbDriver = "sqlite" | "postgres" | /** MySQL and MariaDB. */ "mysql" | /** Microsoft SQL Server. */ -"sqlServer"; +"sqlServer" | "clickhouse"; /** `schema.table (columns) REFERENCES ref_schema.ref_table (ref_columns)`. */ export type DbForeignKey = { diff --git a/src/features/database/ObjectDialogs.tsx b/src/features/database/ObjectDialogs.tsx index 08fb56d4d..648e7ec13 100644 --- a/src/features/database/ObjectDialogs.tsx +++ b/src/features/database/ObjectDialogs.tsx @@ -269,6 +269,7 @@ const TYPES: Record = { "bit", "varbinary(max)", ], + clickhouse: ["UInt64", "Int64", "Float64", "String", "Bool", "Date", "DateTime", "UUID", "Nullable(String)"], }; interface ModifyProps { @@ -328,7 +329,8 @@ function ModifyTableForm({ const driver = driverOf(id); const [name, setName] = useState(table ?? "table_name"); const [columns, setColumns] = useState(() => { - if (!initial) return [{ ...blankColumn(driver, 1), name: "id", primaryKey: true, notNull: true }]; + if (!initial) + return [{ ...blankColumn(driver, 1), name: "id", primaryKey: driver !== "clickhouse", notNull: true }]; const drafts = initial.map(draftOf); if (addColumn) drafts.push(blankColumn(driver, drafts.length + 1)); return drafts; @@ -448,14 +450,18 @@ function ModifyTableForm({ disabled={current.dropped || locked} onChange={(e) => patch({ dataType: e.target.value })} /> - - patch({ notNull })} - label="Not null" - /> - {!table && ( + {driver !== "clickhouse" && ( + <> + + patch({ notNull })} + label="Not null" + /> + + )} + {!table && driver !== "clickhouse" && ( <> > = [ /** Driver property names the Advanced tab suggests, and what the backend does with them. */ const PROPERTIES: Record = { + clickhouse: { + names: ["max_execution_time", "max_threads", "connect_timeout"], + hint: "ClickHouse settings sent with each HTTP request. connect_timeout is in seconds.", + }, postgres: { names: [ "statement_timeout", @@ -376,6 +380,7 @@ export function SourceDialog({ const vendor = vendorOf(s); const sqlite = s.driver === "sqlite"; const postgres = s.driver === "postgres"; + const clickhouse = s.driver === "clickhouse"; const urlOnly = !sqlite && !!s.urlOnly; const parsedUrl = urlOnly ? parseSourceUrl(s.url ?? "") : null; const urlValid = !!parsedUrl && parsedUrl.source.driver === s.driver; @@ -398,7 +403,20 @@ export function SourceDialog({ setS((x) => { const old = vendorOf(x); const port = x.port == null || x.port === old.port ? v.port : x.port; - return { ...x, driver: v.driver, vendor: v.id === v.driver ? "" : v.id, port, ssl: v.ssl ?? x.ssl }; + return { + ...x, + driver: v.driver, + vendor: v.id === v.driver ? "" : v.id, + port, + ssl: v.ssl ?? (v.driver === "clickhouse" ? "disable" : x.ssl), + ...(v.driver === "clickhouse" + ? { + user: x.user || "default", + database: x.database || "default", + options: { ...blankOptions(), ...x.options, transaction: "auto" as const }, + } + : {}), + }; }); }; @@ -614,8 +632,9 @@ export function SourceDialog({ bound && void setManualCommit(bound, v === "manual")} - disabled={!bound} - /> - {manual && ( + {supportsTransactions && ( + <> + +