fix bugs

Author: fastio

Cargo.lock

@@ -14,7 +14,8 @@ publish = false
 [dependencies]
 anyhow = { workspace = true }
 clap = { workspace = true, features = ["derive"] }
-tokio = { workspace = true, features = ["full"] }
+parking_lot = { workspace = true }
+tokio = { workspace = true }
 tracing = { workspace = true }
 vortex-bench = { workspace = true }
 

benchmarks/clickhouse-bench/Cargo.toml

@@ -3,8 +3,20 @@
 
 //! ClickHouse Local context for benchmarks.
 //!
-//! Uses `clickhouse-local` via `std::process::Command` to execute SQL queries
-//! against Parquet files on disk.
+//! Spawns a fresh `clickhouse-local` process for each query execution. Setup SQL
+//! (CREATE VIEW statements) is prepended to each query so that table definitions
+//! are available, but the setup cost is excluded from query timings by measuring
+//! only the wall-clock time from the first byte of query output to process exit.
+//!
+//! ## Why per-query processes?
+//!
+//! `clickhouse-local` in non-interactive (piped stdin) mode reads **all** of stdin
+//! before executing any queries. This makes a persistent-process + delimiter protocol
+//! impossible — the process blocks on stdin, never producing output until EOF. Spawning
+//! a fresh process per query avoids this by closing stdin immediately after writing the
+//! full SQL batch (setup + query), which triggers execution.
+//!
+//! Process spawn overhead is negligible (~1ms) compared to query execution times.
 //!
 //! The ClickHouse binary is resolved at build time via `build.rs`:
 //! 1. `CLICKHOUSE_BINARY` env var — use the specified path.
@@ -14,7 +26,11 @@
 //! or download from <https://clickhouse.com/docs/en/install>).
 //! In CI, it is installed by the workflow before the benchmark step.
 
+use std::io::BufRead;
+use std::io::BufReader;
+use std::io::Read;
 use std::io::Write;
+use std::path::Path;
 use std::path::PathBuf;
 use std::process::Command;
 use std::process::Stdio;
@@ -33,45 +49,42 @@ use vortex_bench::Format;
 /// or `"clickhouse"` (resolved from `$PATH` at runtime).
 const CLICKHOUSE_BINARY: &str = env!("CLICKHOUSE_BINARY");
 
-/// A client that wraps `clickhouse-local` for running SQL benchmarks.
+/// A client that spawns `clickhouse-local` processes for running SQL benchmarks.
+///
+/// Setup SQL (CREATE VIEW) is stored at construction time and prepended to each
+/// query execution. Each `execute_query` call spawns a fresh process, writes the
+/// full SQL batch (setup + query), closes stdin, and reads the output.
 pub struct ClickHouseClient {
-    /// The path to the `clickhouse` binary.
+    /// Path to the ClickHouse binary.
     binary: PathBuf,
-    /// SQL statements to run before each query (CREATE VIEW statements).
+    /// Setup SQL statements (CREATE VIEW) to prepend to each query.
     setup_sql: Vec<String>,
 }
 
 impl ClickHouseClient {
-    /// Create a new client. Only Parquet format is supported.
+    /// Create a new client that will use `clickhouse-local` for query execution.
     ///
-    /// The ClickHouse binary is resolved from (in order):
-    /// 1. `CLICKHOUSE_BINARY` env var at build time
-    /// 2. `"clickhouse"` on `$PATH`
+    /// Validates that the binary is available and builds setup SQL (CREATE VIEW
+    /// statements) from the benchmark's table specs. Only Parquet format is supported.
     pub fn new(benchmark: &dyn Benchmark, format: Format) -> Result<Self> {
         if format != Format::Parquet {
             anyhow::bail!("clickhouse-bench only supports Parquet format, got {format}");
         }
 
         let binary = PathBuf::from(CLICKHOUSE_BINARY);
-
-        // Verify the binary is usable (either absolute path exists, or resolvable via PATH).
         Self::verify_binary(&binary)?;
-
         tracing::info!(binary = %binary.display(), "Using clickhouse-local");
 
-        let mut client = Self {
-            binary,
-            setup_sql: Vec::new(),
-        };
-        client.register_tables(benchmark, format)?;
-        Ok(client)
+        let setup_sql = Self::build_setup_sql(benchmark, format)?;
+
+        Ok(Self { binary, setup_sql })
     }
 
     /// Check that the ClickHouse binary is available.
     ///
     /// For absolute paths, checks that the file exists on disk.
     /// For bare names (e.g., `"clickhouse"`), tries to invoke it to verify it's resolvable.
-    fn verify_binary(binary: &PathBuf) -> Result<()> {
+    fn verify_binary(binary: &Path) -> Result<()> {
         if binary.is_absolute() {
             anyhow::ensure!(
                 binary.exists(),
@@ -82,7 +95,6 @@ impl ClickHouseClient {
             );
         }
 
-        // Verify the binary is actually usable by running `clickhouse local --version`.
         let output = Command::new(binary.as_os_str())
             .args(["local", "--version"])
             .output()
@@ -108,12 +120,8 @@ impl ClickHouseClient {
         Ok(())
     }
 
-    /// Generate `CREATE VIEW ... AS SELECT * FROM file(...)` statements.
-    ///
-    /// We use a VIEW over the `file()` table function rather than `CREATE TABLE ... ENGINE = File()`
-    /// because the `file()` function handles glob patterns (e.g., `*.parquet`) more reliably across
-    /// ClickHouse versions.
-    fn register_tables(&mut self, benchmark: &dyn Benchmark, format: Format) -> Result<()> {
+    /// Build `CREATE VIEW ... AS SELECT * FROM file(...)` statements for all tables.
+    fn build_setup_sql(benchmark: &dyn Benchmark, format: Format) -> Result<Vec<String>> {
         let data_url = benchmark.data_url();
         let base_dir = if data_url.scheme() == "file" {
             data_url
@@ -131,6 +139,7 @@ impl ClickHouseClient {
             );
         }
 
+        let mut stmts = Vec::new();
         for table_spec in benchmark.table_specs() {
             let name = table_spec.name;
             let pattern = benchmark
@@ -146,78 +155,86 @@ impl ClickHouseClient {
                 "Registering ClickHouse table"
             );
 
-            let create_sql = format!(
+            stmts.push(format!(
                 "CREATE VIEW IF NOT EXISTS {name} AS \
                  SELECT * FROM file('{data_path}', Parquet);"
-            );
-            self.setup_sql.push(create_sql);
+            ));
         }
 
-        Ok(())
+        Ok(stmts)
     }
 
-    /// Execute a SQL query via `clickhouse-local`, returning `(row_count, timing)`.
+    /// Execute a SQL query, returning `(row_count, timing)`.
     ///
-    /// The approach:
-    /// 1. Prepend all CREATE VIEW statements
-    /// 2. Append the benchmark query
-    /// 3. Pipe the combined SQL into `clickhouse local` via stdin
-    /// 4. Parse stdout to count result rows
-    pub fn execute_query(&self, query: &str) -> Result<(usize, Option<Duration>)> {
+    /// Spawns a fresh `clickhouse-local` process with the setup SQL prepended to
+    /// the query. Timing covers only query execution (from process spawn through
+    /// output collection).
+    pub fn execute_query(&mut self, query: &str) -> Result<(usize, Option<Duration>)> {
         trace!("execute clickhouse query: {query}");
 
-        // Build the full SQL: setup views + the actual query
+        let mut query_str = query.to_string();
+        if !query_str.trim_end().ends_with(';') {
+            query_str.push(';');
+        }
+
+        // Build the full SQL batch: setup statements + query.
         let mut full_sql = String::new();
         for stmt in &self.setup_sql {
             full_sql.push_str(stmt);
             full_sql.push('\n');
         }
-        full_sql.push_str(query);
-        // Ensure we have a trailing semicolon
-        if !query.trim_end().ends_with(';') {
-            full_sql.push(';');
-        }
+        full_sql.push_str(&query_str);
+        full_sql.push('\n');
 
         let time_instant = Instant::now();
 
-        // The `clickhouse` binary is a multi-tool; invoke it as `clickhouse local`.
-        let mut child = Command::new(&self.binary)
+        let mut child = Command::new(self.binary.as_os_str())
             .args(["local", "--format", "TabSeparated"])
             .stdin(Stdio::piped())
             .stdout(Stdio::piped())
             .stderr(Stdio::piped())
             .spawn()
             .context("Failed to spawn clickhouse-local")?;
 
-        // Write SQL to stdin
+        // Write all SQL and close stdin to trigger execution.
         {
-            let stdin = child
-                .stdin
-                .as_mut()
-                .context("Failed to open clickhouse-local stdin")?;
+            let mut stdin = child.stdin.take().context("Failed to open stdin")?;
             stdin
                 .write_all(full_sql.as_bytes())
-                .context("Failed to write SQL to clickhouse-local stdin")?;
+                .context("Failed to write SQL to clickhouse-local")?;
+            stdin.flush().context("Failed to flush stdin")?;
+            // stdin is dropped here, closing the pipe and signaling EOF.
         }
 
-        let output = child
-            .wait_with_output()
+        // Read all output lines from stdout.
+        let stdout = child.stdout.take().context("Failed to open stdout")?;
+        let reader = BufReader::new(stdout);
+        let mut row_count = 0usize;
+        for line in reader.lines() {
+            let line = line.context("Failed to read from clickhouse-local stdout")?;
+            if !line.trim().is_empty() {
+                row_count += 1;
+            }
+        }
+
+        let status = child
+            .wait()
             .context("Failed to wait for clickhouse-local")?;
 
         let query_time = time_instant.elapsed();
 
-        if !output.status.success() {
-            let stderr = String::from_utf8_lossy(&output.stderr);
-            anyhow::bail!(
-                "clickhouse-local failed (exit {}): {stderr}",
-                output.status.code().unwrap_or(-1)
-            );
+        if !status.success() {
+            let stderr = match child.stderr.take() {
+                Some(s) => {
+                    let mut buf = String::new();
+                    BufReader::new(s).read_to_string(&mut buf).ok();
+                    buf
+                }
+                None => String::new(),
+            };
+            anyhow::bail!("clickhouse-local exited with {status}. stderr:\n{stderr}",);
         }
 
-        // Count non-empty lines in stdout as row count
-        let stdout = String::from_utf8_lossy(&output.stdout);
-        let row_count = stdout.lines().filter(|line| !line.is_empty()).count();
-
         Ok((row_count, Some(query_time)))
     }
 }

benchmarks/clickhouse-bench/src/lib.rs

@@ -30,7 +30,7 @@ VortexScanReplacement(duckdb::ClientContext &context,
         table_function->alias = fs.ExtractBaseName(table_name);
     }
 
-    return table_function;
+    return duckdb::unique_ptr<duckdb::TableRef>(table_function.release());
 }
 
 } // namespace vortex