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 && ( + <> + +