This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-rust.git
The following commit(s) were added to refs/heads/main by this push:
new 054ce25d [benchmark] Add DataFusion TPC-DS benchmark harness (#542)
054ce25d is described below
commit 054ce25d984f9f937365192169038ef45cec50d1
Author: Jingsong Lee <[email protected]>
AuthorDate: Sun Jul 19 09:47:53 2026 +0800
[benchmark] Add DataFusion TPC-DS benchmark harness (#542)
---
Cargo.toml | 2 +-
benchmarks/tpcds/Cargo.toml | 38 ++++++
benchmarks/tpcds/README.md | 212 +++++++++++++++++++++++++++++
benchmarks/tpcds/src/cli.rs | 166 +++++++++++++++++++++++
benchmarks/tpcds/src/command.rs | 141 +++++++++++++++++++
benchmarks/tpcds/src/context.rs | 101 ++++++++++++++
benchmarks/tpcds/src/lib.rs | 292 ++++++++++++++++++++++++++++++++++++++++
benchmarks/tpcds/src/load.rs | 169 +++++++++++++++++++++++
benchmarks/tpcds/src/main.rs | 24 ++++
benchmarks/tpcds/src/report.rs | 109 +++++++++++++++
benchmarks/tpcds/src/run.rs | 273 +++++++++++++++++++++++++++++++++++++
benchmarks/tpcds/tests/smoke.rs | 225 +++++++++++++++++++++++++++++++
12 files changed, 1751 insertions(+), 1 deletion(-)
diff --git a/Cargo.toml b/Cargo.toml
index b8d4edc3..168a5890 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -17,7 +17,7 @@
[workspace]
resolver = "2"
-members = ["crates/paimon", "crates/paimon-rest-server",
"crates/integration_tests", "bindings/c", "bindings/python",
"crates/integrations/datafusion"]
+members = ["crates/paimon", "crates/paimon-rest-server",
"crates/integration_tests", "bindings/c", "bindings/python",
"crates/integrations/datafusion", "benchmarks/tpcds"]
[workspace.package]
version = "0.3.0"
diff --git a/benchmarks/tpcds/Cargo.toml b/benchmarks/tpcds/Cargo.toml
new file mode 100644
index 00000000..9765b19a
--- /dev/null
+++ b/benchmarks/tpcds/Cargo.toml
@@ -0,0 +1,38 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+[package]
+name = "paimon-tpcds-bench"
+edition.workspace = true
+version.workspace = true
+publish = false
+license.workspace = true
+
+[dependencies]
+clap = { version = "4", features = ["derive"] }
+datafusion = { workspace = true }
+paimon = { workspace = true }
+paimon-datafusion = { path = "../../crates/integrations/datafusion" }
+serde = { version = "1", features = ["derive"] }
+serde_json = "1"
+tokio = { workspace = true, features = ["macros", "rt-multi-thread"] }
+
+[dev-dependencies]
+arrow-array = { workspace = true }
+arrow-schema = { workspace = true }
+parquet = { workspace = true }
+tempfile = "3"
diff --git a/benchmarks/tpcds/README.md b/benchmarks/tpcds/README.md
new file mode 100644
index 00000000..f5872214
--- /dev/null
+++ b/benchmarks/tpcds/README.md
@@ -0,0 +1,212 @@
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+# DataFusion + Paimon TPC-DS-Derived Benchmark
+
+This crate loads generated TPC-DS Parquet data into Apache Paimon and runs the
+same external query files against Paimon or Parquet through Apache DataFusion.
+
+> **Disclosure:** This is a TPC-DS-derived non-TPC benchmark. Its results are
+> not official TPC results and must not be compared with official TPC results.
+
+The crate does not include or download TPC tools, generated data, or query
+text. Obtain those materials separately and follow their licenses.
+
+## Prerequisites
+
+- Build and run on the machine being measured; do not benchmark a debug build.
+- Use the TPC-DS data generator and conversion instructions from
+
[`apache/datafusion-benchmarks`](https://github.com/apache/datafusion-benchmarks/tree/main/tpcds).
+- Keep the upstream query directory available, normally
+ `datafusion-benchmarks/tpcds/queries`.
+- Provide enough storage for the source data, Paimon copy, and DataFusion
+ spill files. For SF1000, fast local NVMe is strongly recommended.
+
+The expected generated-data layout is:
+
+```text
+/data/tpcds-sf1000/
+ call_center.parquet/
+ catalog_page.parquet/
+ ...
+ web_site.parquet/
+```
+
+The upstream generation flow uses scale factor 1000 for approximately 1 TB of
+uncompressed generated data. Choose the generator partition count for the
+target machine and retain it in the test notes. For example:
+
+```bash
+tpctools generate --benchmark tpcds \
+ --scale 1000 \
+ --partitions 64 \
+ --generator-path /path/to/DSGen-software-code/tools \
+ --output /data/tpcds
+
+python3 tpcdsgen.py convert --scale-factor 1000 --partitions 64
+```
+
+The upstream conversion script currently contains environment-specific paths;
+inspect and update them before conversion.
+
+## Build
+
+```bash
+cargo build --release -p paimon-tpcds-bench
+target/release/paimon-tpcds-bench --help
+```
+
+Keep the exact `paimon-rust` commit, generated-data scale, generator version,
+file counts, physical bytes, operating system, CPU, memory, and storage model
+with every published report.
+
+## Compatibility Pass
+
+Do not start with SF1000. Use progressively larger datasets:
+
+1. SF10: validate table schemas and all query files.
+2. SF100: validate correctness, spill configuration, and stable timings.
+3. SF1000: run the final measurement without changing the validated SQL.
+
+Use `--tables` and `--query` to isolate failures:
+
+```bash
+target/release/paimon-tpcds-bench load \
+ --data /data/tpcds-sf10 \
+ --warehouse /data/paimon-sf10 \
+ --tables store_sales,date_dim,item
+
+target/release/paimon-tpcds-bench run \
+ --source paimon \
+ --warehouse /data/paimon-sf10 \
+ --queries /src/datafusion-benchmarks/tpcds/queries \
+ --query 1,3-5 \
+ --output results/paimon-sf10.json
+```
+
+Query selection accepts comma-separated numbers and inclusive ranges. Query
+files may contain multiple SQL statements; their statements execute
+sequentially and the file remains one timed benchmark unit.
+
+## Load Paimon Tables
+
+The loader infers each Parquet schema and creates an unpartitioned append-only
+Paimon table. It processes and commits one table at a time.
+
+```bash
+target/release/paimon-tpcds-bench load \
+ --data /data/tpcds-sf1000 \
+ --warehouse /data/paimon-sf1000 \
+ --database tpcds \
+ --if-exists error \
+ --target-partitions 64 \
+ --memory-limit-gib 192 \
+ --spill-dir /nvme/datafusion-spill \
+ --max-spill-gib 1024
+```
+
+`--if-exists` is deliberately explicit:
+
+- `error` stops before changing an existing table;
+- `skip` leaves the existing table untouched;
+- `overwrite` runs `INSERT OVERWRITE` from the Parquet source.
+
+The loader's elapsed time is operational information, not part of query
+performance results.
+
+## Run the Paimon Benchmark
+
+```bash
+target/release/paimon-tpcds-bench run \
+ --source paimon \
+ --warehouse /data/paimon-sf1000 \
+ --database tpcds \
+ --queries /src/datafusion-benchmarks/tpcds/queries \
+ --output results/datafusion-paimon-sf1000.json \
+ --warmup 1 \
+ --iterations 3 \
+ --target-partitions 64 \
+ --memory-limit-gib 192 \
+ --spill-dir /nvme/datafusion-spill \
+ --max-spill-gib 1024
+```
+
+## Run the Parquet Baseline
+
+Use the same binary, runtime settings, query files, and generated data. The
+`--warehouse` path is only a lightweight catalog location for session-scoped
+Parquet tables; use an empty location separate from the measured data.
+
+```bash
+target/release/paimon-tpcds-bench run \
+ --source parquet \
+ --data /data/tpcds-sf1000 \
+ --warehouse /data/parquet-benchmark-catalog \
+ --database tpcds \
+ --queries /src/datafusion-benchmarks/tpcds/queries \
+ --output results/datafusion-parquet-sf1000.json \
+ --warmup 1 \
+ --iterations 3 \
+ --target-partitions 64 \
+ --memory-limit-gib 192 \
+ --spill-dir /nvme/datafusion-spill \
+ --max-spill-gib 1024
+```
+
+This is an end-to-end source comparison. Loading the data into Paimon rewrites
+the physical files, so it is not a pure measurement of catalog or manifest
+overhead.
+
+## Cache Protocol
+
+Run and label cold and warm experiments separately:
+
+- Warm: use `--warmup 1` or more and report only measured iterations.
+- Cold: use `--warmup 0`, start from a documented cache state, and perform OS
+ page-cache eviction outside this tool only when the test operator can do so
+ safely.
+
+The runner does not drop the OS page cache. Do not mix cold and warm timings in
+one aggregate.
+
+## JSON Report
+
+The versioned report records:
+
+- source, paths, versions, runtime limits, and query iteration counts;
+- logical planning, physical planning, execution, and total wall-clock time;
+- output rows and errors;
+- spill count, spilled rows, spilled bytes, bytes scanned, and summed operator
+ peak-memory metrics when DataFusion exposes them.
+
+`operator_peak_memory_bytes` is the sum of available operator metrics, not a
+process-wide peak RSS. A zero value can mean that the physical operators did
+not publish that metric. Paimon scans currently do not publish a precise
+`bytes_scanned` execution metric, so that field may also be zero.
+
+The report is written even when a measured query fails; the command then exits
+non-zero. Preserve failed queries in comparisons instead of silently excluding
+them.
+
+## Interpreting Results
+
+Report per-query distributions and failure counts. Useful summaries include
+median, p95, total elapsed time, and geometric mean over the same successful
+query set. Never hide OOM, timeout, unsupported SQL, or correctness failures by
+computing an aggregate only from the remaining queries.
diff --git a/benchmarks/tpcds/src/cli.rs b/benchmarks/tpcds/src/cli.rs
new file mode 100644
index 00000000..878819f2
--- /dev/null
+++ b/benchmarks/tpcds/src/cli.rs
@@ -0,0 +1,166 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::path::PathBuf;
+
+use clap::{Args, Parser, Subcommand, ValueEnum};
+
+use crate::{BenchmarkRuntimeConfig, ExistingTablePolicy, SourceKind};
+
+#[derive(Debug, Parser)]
+#[command(
+ name = "paimon-tpcds-bench",
+ about = "TPC-DS-derived non-TPC benchmark for DataFusion and Paimon"
+)]
+pub struct Cli {
+ #[command(subcommand)]
+ pub command: Command,
+}
+
+#[derive(Debug, Subcommand)]
+pub enum Command {
+ /// Load generated TPC-DS Parquet data into append-only Paimon tables.
+ Load(LoadArgs),
+ /// Run external TPC-DS-derived query files against Paimon or Parquet.
+ Run(RunArgs),
+}
+
+#[derive(Debug, Args)]
+pub struct LoadArgs {
+ /// Directory containing <table>.parquet paths.
+ #[arg(long)]
+ pub data: PathBuf,
+ /// Local Paimon filesystem warehouse directory.
+ #[arg(long)]
+ pub warehouse: PathBuf,
+ /// Paimon database name.
+ #[arg(long, default_value = "tpcds")]
+ pub database: String,
+ /// Comma-separated table names; defaults to all 24 tables.
+ #[arg(long)]
+ pub tables: Option<String>,
+ /// Behavior when a target table already exists.
+ #[arg(long, value_enum, default_value_t = ExistingPolicyArg::Error)]
+ pub if_exists: ExistingPolicyArg,
+ #[command(flatten)]
+ pub runtime: RuntimeArgs,
+}
+
+#[derive(Debug, Args)]
+pub struct RunArgs {
+ /// Source being measured.
+ #[arg(long, value_enum)]
+ pub source: SourceKind,
+ /// Parquet data directory. Required when --source=parquet.
+ #[arg(long)]
+ pub data: Option<PathBuf>,
+ /// Local Paimon warehouse, or a temporary catalog directory for Parquet.
+ #[arg(long)]
+ pub warehouse: PathBuf,
+ /// Directory containing q1.sql through q99.sql.
+ #[arg(long)]
+ pub queries: PathBuf,
+ /// JSON output path.
+ #[arg(long)]
+ pub output: PathBuf,
+ /// Catalog database containing the Paimon tables.
+ #[arg(long, default_value = "tpcds")]
+ pub database: String,
+ /// Query numbers, comma lists, or inclusive ranges (for example 1,3-5).
+ #[arg(long)]
+ pub query: Option<String>,
+ /// Parquet tables to register; defaults to all 24 tables.
+ #[arg(long)]
+ pub tables: Option<String>,
+ /// Warmup executions per query file.
+ #[arg(long, default_value_t = 1)]
+ pub warmup: usize,
+ /// Measured executions per query file.
+ #[arg(long, default_value_t = 3)]
+ pub iterations: usize,
+ #[command(flatten)]
+ pub runtime: RuntimeArgs,
+}
+
+impl RunArgs {
+ pub fn validate(&self) -> Result<(), String> {
+ if self.source == SourceKind::Parquet && self.data.is_none() {
+ return Err("--data is required when --source=parquet".to_string());
+ }
+ if self.iterations == 0 {
+ return Err("--iterations must be greater than zero".to_string());
+ }
+ Ok(())
+ }
+}
+
+#[derive(Debug, Clone, Args)]
+pub struct RuntimeArgs {
+ /// DataFusion execution partitions. Defaults to available CPUs.
+ #[arg(long)]
+ pub target_partitions: Option<usize>,
+ /// DataFusion memory limit in GiB. Omit for an unbounded pool.
+ #[arg(long)]
+ pub memory_limit_gib: Option<u64>,
+ /// Directory used for DataFusion spill files.
+ #[arg(long)]
+ pub spill_dir: Option<PathBuf>,
+ /// Maximum spill-directory usage in GiB.
+ #[arg(long)]
+ pub max_spill_gib: Option<u64>,
+}
+
+impl RuntimeArgs {
+ pub fn to_config(&self) -> Result<BenchmarkRuntimeConfig, String> {
+ let defaults = BenchmarkRuntimeConfig::default();
+ Ok(BenchmarkRuntimeConfig {
+ target_partitions: self
+ .target_partitions
+ .unwrap_or(defaults.target_partitions)
+ .max(1),
+ memory_limit_bytes:
self.memory_limit_gib.map(gib_to_usize).transpose()?,
+ spill_dir: self.spill_dir.clone(),
+ max_spill_bytes: self.max_spill_gib.map(gib_to_u64).transpose()?,
+ })
+ }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, ValueEnum)]
+pub enum ExistingPolicyArg {
+ Error,
+ Skip,
+ Overwrite,
+}
+
+impl From<ExistingPolicyArg> for ExistingTablePolicy {
+ fn from(value: ExistingPolicyArg) -> Self {
+ match value {
+ ExistingPolicyArg::Error => ExistingTablePolicy::Error,
+ ExistingPolicyArg::Skip => ExistingTablePolicy::Skip,
+ ExistingPolicyArg::Overwrite => ExistingTablePolicy::Overwrite,
+ }
+ }
+}
+
+fn gib_to_usize(gib: u64) -> Result<usize, String> {
+ usize::try_from(gib_to_u64(gib)?).map_err(|_| format!("{gib} GiB exceeds
usize"))
+}
+
+fn gib_to_u64(gib: u64) -> Result<u64, String> {
+ gib.checked_mul(1024 * 1024 * 1024)
+ .ok_or_else(|| format!("{gib} GiB exceeds u64"))
+}
diff --git a/benchmarks/tpcds/src/command.rs b/benchmarks/tpcds/src/command.rs
new file mode 100644
index 00000000..82febf8b
--- /dev/null
+++ b/benchmarks/tpcds/src/command.rs
@@ -0,0 +1,141 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::collections::HashSet;
+use std::io::{Error as IoError, ErrorKind};
+
+use crate::{
+ load_parquet_table, load_query_files, open_catalog_session,
parse_number_selection,
+ register_parquet_tables, run_query_file, write_report, BenchmarkReport,
Command,
+ ExistingTablePolicy, QueryRunConfig, SourceKind, TPCDS_TABLES,
+};
+
+type BoxError = Box<dyn std::error::Error + Send + Sync>;
+
+pub async fn execute_command(command: Command) -> Result<(), BoxError> {
+ match command {
+ Command::Load(args) => {
+ let runtime = args.runtime.to_config().map_err(invalid_input)?;
+ let tables = select_tables(args.tables.as_deref())?;
+ let session = open_catalog_session(&runtime, &args.warehouse,
&args.database).await?;
+ let policy: ExistingTablePolicy = args.if_exists.into();
+ for table in tables {
+ println!("Loading {table} ...");
+ let result = load_parquet_table(&session, &args.data, table,
policy).await?;
+ println!(
+ " status={:?} rows={} elapsed_ms={}",
+ result.status, result.rows, result.elapsed_ms
+ );
+ }
+ Ok(())
+ }
+ Command::Run(args) => {
+ args.validate().map_err(invalid_input)?;
+ let runtime = args.runtime.to_config().map_err(invalid_input)?;
+ let session = open_catalog_session(&runtime, &args.warehouse,
&args.database).await?;
+ if args.source == SourceKind::Parquet {
+ let tables = select_tables(args.tables.as_deref())?;
+ let data = args
+ .data
+ .as_deref()
+ .ok_or_else(|| invalid_input("--data is required when
--source=parquet"))?;
+ register_parquet_tables(&session, data, &tables).await?;
+ }
+
+ let query_numbers =
+ parse_number_selection(args.query.as_deref(), 1,
99).map_err(invalid_input)?;
+ let query_files =
+ load_query_files(&args.queries,
&query_numbers).map_err(invalid_input)?;
+ let query_run = QueryRunConfig {
+ warmup_iterations: args.warmup,
+ measured_iterations: args.iterations,
+ };
+ let mut query_results = Vec::with_capacity(query_files.len());
+ for query in &query_files {
+ println!("Running q{} ...", query.number);
+ let result = run_query_file(&session, query, &query_run).await;
+ for iteration in &result.iterations {
+ if let Some(error) = &iteration.error {
+ println!(
+ " iteration={} failed after {:.3} ms: {error}",
+ iteration.iteration, iteration.total_ms
+ );
+ } else {
+ println!(
+ " iteration={} total_ms={:.3} rows={}
spilled_bytes={}",
+ iteration.iteration,
+ iteration.total_ms,
+ iteration.output_rows,
+ iteration.metrics.spilled_bytes
+ );
+ }
+ }
+ query_results.push(result);
+ }
+
+ let report = BenchmarkReport::new(
+ args.source,
+ runtime,
+ query_run,
+ args.warehouse.display().to_string(),
+ args.data
+ .as_ref()
+ .map(|path| path.display().to_string())
+ .unwrap_or_default(),
+ args.queries.display().to_string(),
+ args.database,
+ query_results,
+ );
+ write_report(&report, &args.output)?;
+ println!("Report written to {}", args.output.display());
+ if report.has_failures() {
+ return Err(IoError::other(format!(
+ "one or more queries failed; see {}",
+ args.output.display()
+ ))
+ .into());
+ }
+ Ok(())
+ }
+ }
+}
+
+fn select_tables(selection: Option<&str>) -> Result<Vec<&'static str>,
BoxError> {
+ let Some(selection) = selection else {
+ return Ok(TPCDS_TABLES.to_vec());
+ };
+ let selected = selection
+ .split(',')
+ .map(str::trim)
+ .filter(|name| !name.is_empty())
+ .collect::<HashSet<_>>();
+ if selected.is_empty() {
+ return Ok(TPCDS_TABLES.to_vec());
+ }
+ if let Some(unknown) = selected.iter().find(|name|
!TPCDS_TABLES.contains(name)) {
+ return Err(invalid_input(format!("unknown TPC-DS table '{unknown}'")));
+ }
+ Ok(TPCDS_TABLES
+ .iter()
+ .copied()
+ .filter(|name| selected.contains(name))
+ .collect())
+}
+
+fn invalid_input(message: impl Into<String>) -> BoxError {
+ IoError::new(ErrorKind::InvalidInput, message.into()).into()
+}
diff --git a/benchmarks/tpcds/src/context.rs b/benchmarks/tpcds/src/context.rs
new file mode 100644
index 00000000..a4a4cd4e
--- /dev/null
+++ b/benchmarks/tpcds/src/context.rs
@@ -0,0 +1,101 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::path::{Path, PathBuf};
+use std::sync::Arc;
+
+use datafusion::error::Result as DataFusionResult;
+use datafusion::execution::runtime_env::RuntimeEnvBuilder;
+use datafusion::execution::SessionStateBuilder;
+use paimon::{CatalogOptions, FileSystemCatalog, Options};
+use paimon_datafusion::SQLContext;
+use serde::{Deserialize, Serialize};
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct BenchmarkRuntimeConfig {
+ pub target_partitions: usize,
+ pub memory_limit_bytes: Option<usize>,
+ pub spill_dir: Option<PathBuf>,
+ pub max_spill_bytes: Option<u64>,
+}
+
+impl Default for BenchmarkRuntimeConfig {
+ fn default() -> Self {
+ Self {
+ target_partitions: std::thread::available_parallelism()
+ .map(usize::from)
+ .unwrap_or(1),
+ memory_limit_bytes: None,
+ spill_dir: None,
+ max_spill_bytes: None,
+ }
+ }
+}
+
+pub fn build_sql_context(config: &BenchmarkRuntimeConfig) ->
DataFusionResult<SQLContext> {
+ let mut runtime = RuntimeEnvBuilder::new();
+ if let Some(memory_limit) = config.memory_limit_bytes {
+ runtime = runtime.with_memory_limit(memory_limit, 1.0);
+ }
+ if let Some(spill_dir) = &config.spill_dir {
+ runtime = runtime.with_temp_file_path(spill_dir);
+ }
+ if let Some(max_spill_bytes) = config.max_spill_bytes {
+ runtime = runtime.with_max_temp_directory_size(max_spill_bytes);
+ }
+ let sql = SQLContext::new();
+ let state_ref = sql.ctx().state_ref();
+ let current_state = state_ref.read().clone();
+ let session_config = current_state
+ .config()
+ .clone()
+ .with_target_partitions(config.target_partitions.max(1));
+ let state = SessionStateBuilder::from(current_state)
+ .with_config(session_config)
+ .with_runtime_env(Arc::new(runtime.build()?))
+ .build();
+ *state_ref.write() = state;
+ drop(state_ref);
+ Ok(sql)
+}
+
+pub struct CatalogSession {
+ pub sql: SQLContext,
+ pub catalog: Arc<FileSystemCatalog>,
+ pub catalog_name: String,
+ pub database: String,
+}
+
+pub async fn open_catalog_session(
+ runtime_config: &BenchmarkRuntimeConfig,
+ warehouse: &Path,
+ database: &str,
+) -> Result<CatalogSession, Box<dyn std::error::Error + Send + Sync>> {
+ std::fs::create_dir_all(warehouse)?;
+ let mut options = Options::new();
+ options.set(CatalogOptions::WAREHOUSE, warehouse.display().to_string());
+ let catalog = Arc::new(FileSystemCatalog::new(options)?);
+ let mut sql = build_sql_context(runtime_config)?;
+ sql.register_catalog_with_default_db("paimon", catalog.clone(),
Some(database))
+ .await?;
+ Ok(CatalogSession {
+ sql,
+ catalog,
+ catalog_name: "paimon".to_string(),
+ database: database.to_string(),
+ })
+}
diff --git a/benchmarks/tpcds/src/lib.rs b/benchmarks/tpcds/src/lib.rs
new file mode 100644
index 00000000..6229b550
--- /dev/null
+++ b/benchmarks/tpcds/src/lib.rs
@@ -0,0 +1,292 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::collections::BTreeSet;
+use std::fs;
+use std::path::{Path, PathBuf};
+
+use datafusion::sql::sqlparser::dialect::GenericDialect;
+use datafusion::sql::sqlparser::parser::Parser;
+
+mod cli;
+mod command;
+mod context;
+mod load;
+mod report;
+mod run;
+
+pub use cli::{Cli, Command, ExistingPolicyArg, LoadArgs, RunArgs, RuntimeArgs};
+pub use command::execute_command;
+pub use context::{
+ build_sql_context, open_catalog_session, BenchmarkRuntimeConfig,
CatalogSession,
+};
+pub use load::{load_parquet_table, ExistingTablePolicy, LoadStatus,
TableLoadResult};
+pub use report::{write_report, BenchmarkReport, SourceKind,
NON_TPC_DISCLOSURE};
+pub use run::{
+ register_parquet_tables, run_query_file, IterationResult, PhysicalMetrics,
QueryRunConfig,
+ QueryRunResult,
+};
+
+pub const TPCDS_TABLES: [&str; 24] = [
+ "call_center",
+ "catalog_page",
+ "catalog_returns",
+ "catalog_sales",
+ "customer",
+ "customer_address",
+ "customer_demographics",
+ "date_dim",
+ "time_dim",
+ "household_demographics",
+ "income_band",
+ "inventory",
+ "item",
+ "promotion",
+ "reason",
+ "ship_mode",
+ "store",
+ "store_returns",
+ "store_sales",
+ "warehouse",
+ "web_page",
+ "web_returns",
+ "web_sales",
+ "web_site",
+];
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct QueryFile {
+ pub number: u32,
+ pub path: PathBuf,
+ pub statements: Vec<String>,
+}
+
+pub fn load_query_files(query_dir: &Path, queries: &[u32]) ->
Result<Vec<QueryFile>, String> {
+ queries
+ .iter()
+ .map(|number| {
+ let path = query_dir.join(format!("q{number}.sql"));
+ let sql = fs::read_to_string(&path)
+ .map_err(|error| format!("failed to read {}: {error}",
path.display()))?;
+ let statements = Parser::parse_sql(&GenericDialect, &sql)
+ .map_err(|error| format!("failed to parse {}: {error}",
path.display()))?
+ .into_iter()
+ .map(|statement| statement.to_string())
+ .collect::<Vec<_>>();
+ if statements.is_empty() {
+ return Err(format!("query file {} is empty", path.display()));
+ }
+ Ok(QueryFile {
+ number: *number,
+ path,
+ statements,
+ })
+ })
+ .collect()
+}
+
+pub fn parse_number_selection(
+ selection: Option<&str>,
+ min: u32,
+ max: u32,
+) -> Result<Vec<u32>, String> {
+ let mut values = BTreeSet::new();
+ let selection = selection.unwrap_or("");
+ for part in selection
+ .split(',')
+ .map(str::trim)
+ .filter(|s| !s.is_empty())
+ {
+ if let Some((start, end)) = part.split_once('-') {
+ let start = start
+ .trim()
+ .parse::<u32>()
+ .map_err(|_| format!("invalid selection value '{start}'"))?;
+ let end = end
+ .trim()
+ .parse::<u32>()
+ .map_err(|_| format!("invalid selection value '{end}'"))?;
+ if start > end {
+ return Err(format!("invalid descending range '{part}'"));
+ }
+ values.extend(start..=end);
+ } else {
+ values.insert(
+ part.parse::<u32>()
+ .map_err(|_| format!("invalid selection value '{part}'"))?,
+ );
+ }
+ }
+
+ if values.is_empty() {
+ values.extend(min..=max);
+ }
+ if let Some(value) = values.iter().find(|value| **value < min || **value >
max) {
+ return Err(format!("selection value {value} is outside
{min}..={max}"));
+ }
+ Ok(values.into_iter().collect())
+}
+
+#[cfg(test)]
+mod tests {
+ use std::fs;
+
+ use clap::Parser;
+ use datafusion::execution::memory_pool::MemoryLimit;
+
+ use super::{
+ build_sql_context, load_query_files, parse_number_selection,
BenchmarkReport,
+ BenchmarkRuntimeConfig, Cli, Command, QueryRunConfig, SourceKind,
TPCDS_TABLES,
+ };
+ use tempfile::TempDir;
+
+ #[test]
+ fn query_selection_supports_lists_and_ranges() {
+ assert_eq!(
+ parse_number_selection(Some("1,3-5"), 1, 99).unwrap(),
+ vec![1, 3, 4, 5]
+ );
+ }
+
+ #[test]
+ fn query_files_are_external_and_can_contain_multiple_statements() {
+ let dir = TempDir::new().unwrap();
+ fs::write(dir.path().join("q1.sql"), "SELECT 1; SELECT 2;").unwrap();
+ fs::write(dir.path().join("q3.sql"), "SELECT 3;").unwrap();
+
+ let files = load_query_files(dir.path(), &[1, 3]).unwrap();
+
+ assert_eq!(
+ files.iter().map(|file| file.number).collect::<Vec<_>>(),
+ vec![1, 3]
+ );
+ assert_eq!(files[0].statements, vec!["SELECT 1", "SELECT 2"]);
+ assert_eq!(files[1].statements, vec!["SELECT 3"]);
+ }
+
+ #[test]
+ fn runtime_config_controls_parallelism_memory_and_spill_path() {
+ let spill_dir = TempDir::new().unwrap();
+ let ctx = build_sql_context(&BenchmarkRuntimeConfig {
+ target_partitions: 3,
+ memory_limit_bytes: Some(32 * 1024 * 1024),
+ spill_dir: Some(spill_dir.path().to_path_buf()),
+ max_spill_bytes: Some(64 * 1024 * 1024),
+ })
+ .unwrap();
+
+ assert_eq!(
+ ctx.ctx()
+ .state()
+ .config_options()
+ .execution
+ .target_partitions,
+ 3
+ );
+ assert!(matches!(
+ ctx.ctx().runtime_env().memory_pool.memory_limit(),
+ MemoryLimit::Finite(size) if size == 32 * 1024 * 1024
+ ));
+ assert!(
+
ctx.ctx().runtime_env().disk_manager.temp_dir_paths()[0].starts_with(spill_dir.path())
+ );
+ assert_eq!(
+ ctx.ctx()
+ .runtime_env()
+ .disk_manager
+ .max_temp_directory_size(),
+ 64 * 1024 * 1024
+ );
+ }
+
+ #[test]
+ fn canonical_table_list_contains_all_24_tpcds_tables() {
+ assert_eq!(TPCDS_TABLES.len(), 24);
+ assert_eq!(TPCDS_TABLES.first(), Some(&"call_center"));
+ assert_eq!(TPCDS_TABLES.last(), Some(&"web_site"));
+ assert!(TPCDS_TABLES.contains(&"store_sales"));
+ }
+
+ #[test]
+ fn report_json_contains_non_tpc_disclosure_and_round_trips() {
+ let report = BenchmarkReport::new(
+ SourceKind::Paimon,
+ BenchmarkRuntimeConfig::default(),
+ QueryRunConfig::default(),
+ "/warehouse".to_string(),
+ "/data".to_string(),
+ "/queries".to_string(),
+ "tpcds".to_string(),
+ vec![],
+ );
+
+ let json = serde_json::to_string(&report).unwrap();
+ let decoded: BenchmarkReport = serde_json::from_str(&json).unwrap();
+
+ assert_eq!(decoded.source, SourceKind::Paimon);
+ assert!(decoded.disclosure.contains("non-TPC"));
+ assert_eq!(decoded.datafusion_version, "54.0.0");
+ }
+
+ #[test]
+ fn cli_parses_a_paimon_run_without_a_parquet_data_path() {
+ let cli = Cli::try_parse_from([
+ "paimon-tpcds-bench",
+ "run",
+ "--source",
+ "paimon",
+ "--warehouse",
+ "/warehouse",
+ "--queries",
+ "/queries",
+ "--output",
+ "/report.json",
+ "--query",
+ "1,3-5",
+ ])
+ .unwrap();
+
+ let Command::Run(args) = cli.command else {
+ panic!("expected run command");
+ };
+ assert_eq!(args.source, SourceKind::Paimon);
+ assert_eq!(args.query.as_deref(), Some("1,3-5"));
+ assert!(args.data.is_none());
+ }
+
+ #[test]
+ fn parquet_run_requires_a_data_path() {
+ let cli = Cli::try_parse_from([
+ "paimon-tpcds-bench",
+ "run",
+ "--source",
+ "parquet",
+ "--warehouse",
+ "/warehouse",
+ "--queries",
+ "/queries",
+ "--output",
+ "/report.json",
+ ])
+ .unwrap();
+ let Command::Run(args) = cli.command else {
+ panic!("expected run command");
+ };
+
+ assert!(args.validate().unwrap_err().contains("--data"));
+ }
+}
diff --git a/benchmarks/tpcds/src/load.rs b/benchmarks/tpcds/src/load.rs
new file mode 100644
index 00000000..4a52105c
--- /dev/null
+++ b/benchmarks/tpcds/src/load.rs
@@ -0,0 +1,169 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::path::Path;
+use std::time::Instant;
+
+use datafusion::arrow::array::UInt64Array;
+use datafusion::prelude::ParquetReadOptions;
+use paimon::arrow::arrow_to_paimon_type;
+use paimon::catalog::Identifier;
+use paimon::spec::Schema;
+use paimon::{Catalog, Error as PaimonError};
+use serde::{Deserialize, Serialize};
+
+use crate::context::CatalogSession;
+
+type BoxError = Box<dyn std::error::Error + Send + Sync>;
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum ExistingTablePolicy {
+ Error,
+ Skip,
+ Overwrite,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum LoadStatus {
+ Loaded,
+ Skipped,
+ Overwritten,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct TableLoadResult {
+ pub table: String,
+ pub status: LoadStatus,
+ pub rows: u64,
+ pub elapsed_ms: u128,
+}
+
+pub async fn load_parquet_table(
+ session: &CatalogSession,
+ data_root: &Path,
+ table_name: &str,
+ existing_policy: ExistingTablePolicy,
+) -> Result<TableLoadResult, BoxError> {
+ let started = Instant::now();
+ let identifier = Identifier::new(&session.database, table_name);
+ let exists = match session.catalog.get_table(&identifier).await {
+ Ok(_) => true,
+ Err(PaimonError::TableNotExist { .. }) => false,
+ Err(error) => return Err(error.into()),
+ };
+
+ if exists && existing_policy == ExistingTablePolicy::Skip {
+ return Ok(TableLoadResult {
+ table: table_name.to_string(),
+ status: LoadStatus::Skipped,
+ rows: 0,
+ elapsed_ms: started.elapsed().as_millis(),
+ });
+ }
+ if exists && existing_policy == ExistingTablePolicy::Error {
+ return Err(PaimonError::TableAlreadyExist {
+ full_name: identifier.full_name(),
+ }
+ .into());
+ }
+
+ let source_path = data_root.join(format!("{table_name}.parquet"));
+ let source_path = source_path
+ .to_str()
+ .ok_or_else(|| format!("source path is not valid UTF-8: {}",
source_path.display()))?;
+ let source = session
+ .sql
+ .ctx()
+ .read_parquet(source_path, ParquetReadOptions::default())
+ .await?;
+ let arrow_schema = source.schema().inner().clone();
+
+ if !exists {
+ let mut schema = Schema::builder();
+ for field in arrow_schema.fields() {
+ schema = schema.column(
+ field.name(),
+ arrow_to_paimon_type(field.data_type(), field.is_nullable())?,
+ );
+ }
+ session
+ .catalog
+ .create_table(&identifier, schema.build()?, false)
+ .await?;
+ }
+
+ let source_name = format!("__tpcds_source_{table_name}");
+ let source_reference = format!(
+ "{}.{}.{}",
+ quote_identifier(&session.catalog_name),
+ quote_identifier(&session.database),
+ quote_identifier(&source_name)
+ );
+ if session.sql.temp_table_exist(source_reference.as_str())? {
+ session
+ .sql
+ .deregister_temp_table(source_reference.as_str())?;
+ }
+ session
+ .sql
+ .register_temp_table(source_reference.as_str(), source.into_view())?;
+
+ let target_reference = format!(
+ "{}.{}.{}",
+ quote_identifier(&session.catalog_name),
+ quote_identifier(&session.database),
+ quote_identifier(table_name)
+ );
+ let operation = if exists {
+ "INSERT OVERWRITE"
+ } else {
+ "INSERT INTO"
+ };
+ let load_result = session
+ .sql
+ .sql(&format!(
+ "{operation} {target_reference} SELECT * FROM {source_reference}"
+ ))
+ .await;
+ let batches = match load_result {
+ Ok(frame) => frame.collect().await,
+ Err(error) => Err(error),
+ };
+ let _ = session.sql.deregister_temp_table(source_reference.as_str());
+ let batches = batches?;
+ let rows = batches
+ .first()
+ .and_then(|batch|
batch.column(0).as_any().downcast_ref::<UInt64Array>())
+ .map(|counts| counts.value(0))
+ .ok_or("DataFusion INSERT did not return a UInt64 row count")?;
+
+ Ok(TableLoadResult {
+ table: table_name.to_string(),
+ status: if exists {
+ LoadStatus::Overwritten
+ } else {
+ LoadStatus::Loaded
+ },
+ rows,
+ elapsed_ms: started.elapsed().as_millis(),
+ })
+}
+
+fn quote_identifier(identifier: &str) -> String {
+ format!("\"{}\"", identifier.replace('"', "\"\""))
+}
diff --git a/benchmarks/tpcds/src/main.rs b/benchmarks/tpcds/src/main.rs
new file mode 100644
index 00000000..27f3267c
--- /dev/null
+++ b/benchmarks/tpcds/src/main.rs
@@ -0,0 +1,24 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use clap::Parser;
+use paimon_tpcds_bench::{execute_command, Cli};
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+ execute_command(Cli::parse().command).await
+}
diff --git a/benchmarks/tpcds/src/report.rs b/benchmarks/tpcds/src/report.rs
new file mode 100644
index 00000000..b026f19a
--- /dev/null
+++ b/benchmarks/tpcds/src/report.rs
@@ -0,0 +1,109 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::path::Path;
+use std::time::{SystemTime, UNIX_EPOCH};
+
+use clap::ValueEnum;
+use serde::{Deserialize, Serialize};
+
+use crate::{BenchmarkRuntimeConfig, QueryRunConfig, QueryRunResult};
+
+pub const NON_TPC_DISCLOSURE: &str =
+ "TPC-DS-derived non-TPC benchmark; these results are not official TPC
results.";
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, ValueEnum)]
+#[serde(rename_all = "snake_case")]
+pub enum SourceKind {
+ Paimon,
+ Parquet,
+}
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct BenchmarkReport {
+ pub schema_version: u32,
+ pub disclosure: String,
+ pub engine: String,
+ pub datafusion_version: String,
+ pub paimon_version: String,
+ pub created_unix_ms: u128,
+ pub source: SourceKind,
+ pub runtime: BenchmarkRuntimeConfig,
+ pub query_run: QueryRunConfig,
+ pub warehouse: String,
+ pub data_root: String,
+ pub query_dir: String,
+ pub database: String,
+ pub queries: Vec<QueryRunResult>,
+}
+
+impl BenchmarkReport {
+ #[allow(clippy::too_many_arguments)]
+ pub fn new(
+ source: SourceKind,
+ runtime: BenchmarkRuntimeConfig,
+ query_run: QueryRunConfig,
+ warehouse: String,
+ data_root: String,
+ query_dir: String,
+ database: String,
+ queries: Vec<QueryRunResult>,
+ ) -> Self {
+ Self {
+ schema_version: 1,
+ disclosure: NON_TPC_DISCLOSURE.to_string(),
+ engine: "datafusion+paimon-rust".to_string(),
+ datafusion_version: datafusion::DATAFUSION_VERSION.to_string(),
+ paimon_version: env!("CARGO_PKG_VERSION").to_string(),
+ created_unix_ms: SystemTime::now()
+ .duration_since(UNIX_EPOCH)
+ .unwrap_or_default()
+ .as_millis(),
+ source,
+ runtime,
+ query_run,
+ warehouse,
+ data_root,
+ query_dir,
+ database,
+ queries,
+ }
+ }
+
+ pub fn has_failures(&self) -> bool {
+ self.queries.iter().any(|query| {
+ !query.warmup_failures.is_empty()
+ || query
+ .iterations
+ .iter()
+ .any(|iteration| iteration.error.is_some())
+ })
+ }
+}
+
+pub fn write_report(
+ report: &BenchmarkReport,
+ path: &Path,
+) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+ if let Some(parent) = path.parent() {
+ if !parent.as_os_str().is_empty() {
+ std::fs::create_dir_all(parent)?;
+ }
+ }
+ std::fs::write(path, serde_json::to_vec_pretty(report)?)?;
+ Ok(())
+}
diff --git a/benchmarks/tpcds/src/run.rs b/benchmarks/tpcds/src/run.rs
new file mode 100644
index 00000000..af5b063a
--- /dev/null
+++ b/benchmarks/tpcds/src/run.rs
@@ -0,0 +1,273 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::path::Path;
+use std::time::Instant;
+
+use datafusion::physical_plan::{collect, ExecutionPlan};
+use datafusion::prelude::ParquetReadOptions;
+use serde::{Deserialize, Serialize};
+
+use crate::context::CatalogSession;
+use crate::QueryFile;
+
+type BoxError = Box<dyn std::error::Error + Send + Sync>;
+
+pub async fn register_parquet_tables<S: AsRef<str>>(
+ session: &CatalogSession,
+ data_root: &Path,
+ tables: &[S],
+) -> Result<(), BoxError> {
+ for table in tables {
+ let table = table.as_ref();
+ let path = data_root.join(format!("{table}.parquet"));
+ let path = path
+ .to_str()
+ .ok_or_else(|| format!("source path is not valid UTF-8: {}",
path.display()))?;
+ let frame = session
+ .sql
+ .ctx()
+ .read_parquet(path, ParquetReadOptions::default())
+ .await?;
+ let table_reference = format!(
+ "{}.{}.{}",
+ quote_identifier(&session.catalog_name),
+ quote_identifier(&session.database),
+ quote_identifier(table)
+ );
+ if session.sql.temp_table_exist(table_reference.as_str())? {
+ session
+ .sql
+ .deregister_temp_table(table_reference.as_str())?;
+ }
+ session
+ .sql
+ .register_temp_table(table_reference.as_str(), frame.into_view())?;
+ }
+ Ok(())
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+pub struct QueryRunConfig {
+ pub warmup_iterations: usize,
+ pub measured_iterations: usize,
+}
+
+impl Default for QueryRunConfig {
+ fn default() -> Self {
+ Self {
+ warmup_iterations: 1,
+ measured_iterations: 3,
+ }
+ }
+}
+
+#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
+pub struct PhysicalMetrics {
+ pub spill_count: u64,
+ pub spilled_rows: u64,
+ pub spilled_bytes: u64,
+ pub bytes_scanned: u64,
+ pub operator_peak_memory_bytes: u64,
+}
+
+impl PhysicalMetrics {
+ fn add_plan(&mut self, plan: &dyn ExecutionPlan) {
+ if let Some(metrics) = plan.metrics() {
+ self.spill_count += metrics.spill_count().unwrap_or(0) as u64;
+ self.spilled_rows += metrics.spilled_rows().unwrap_or(0) as u64;
+ self.spilled_bytes += metrics.spilled_bytes().unwrap_or(0) as u64;
+ self.bytes_scanned += metrics
+ .sum_by_name("bytes_scanned")
+ .map(|value| value.as_usize() as u64)
+ .unwrap_or(0);
+ self.operator_peak_memory_bytes += metrics
+ .sum_by_name("peak_mem_used")
+ .map(|value| value.as_usize() as u64)
+ .unwrap_or(0);
+ }
+ for child in plan.children() {
+ self.add_plan(child.as_ref());
+ }
+ }
+}
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct IterationResult {
+ pub iteration: usize,
+ pub logical_planning_ms: f64,
+ pub physical_planning_ms: f64,
+ pub execution_ms: f64,
+ pub total_ms: f64,
+ pub output_rows: u64,
+ pub metrics: PhysicalMetrics,
+ pub error: Option<String>,
+}
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct QueryRunResult {
+ pub query: u32,
+ pub path: String,
+ pub warmup_failures: Vec<String>,
+ pub iterations: Vec<IterationResult>,
+}
+
+pub async fn run_query_file(
+ session: &CatalogSession,
+ query: &QueryFile,
+ config: &QueryRunConfig,
+) -> QueryRunResult {
+ let mut warmup_failures = Vec::new();
+ for iteration in 0..config.warmup_iterations {
+ let result = execute_iteration(session, query, iteration).await;
+ if let Some(error) = result.error {
+ warmup_failures.push(error);
+ }
+ }
+
+ let mut iterations = Vec::with_capacity(config.measured_iterations);
+ for iteration in 0..config.measured_iterations {
+ iterations.push(execute_iteration(session, query, iteration +
1).await);
+ }
+
+ QueryRunResult {
+ query: query.number,
+ path: query.path.display().to_string(),
+ warmup_failures,
+ iterations,
+ }
+}
+
+async fn execute_iteration(
+ session: &CatalogSession,
+ query: &QueryFile,
+ iteration: usize,
+) -> IterationResult {
+ let total_started = Instant::now();
+ let mut logical_planning_ms = 0.0;
+ let mut physical_planning_ms = 0.0;
+ let mut execution_ms = 0.0;
+ let mut output_rows = 0u64;
+ let mut metrics = PhysicalMetrics::default();
+
+ for statement in &query.statements {
+ let logical_started = Instant::now();
+ let frame = match session.sql.sql(statement).await {
+ Ok(frame) => frame,
+ Err(error) => {
+ logical_planning_ms += elapsed_ms(logical_started);
+ return failed_iteration(
+ iteration,
+ logical_planning_ms,
+ physical_planning_ms,
+ execution_ms,
+ total_started,
+ output_rows,
+ metrics,
+ error.to_string(),
+ );
+ }
+ };
+ logical_planning_ms += elapsed_ms(logical_started);
+
+ let physical_started = Instant::now();
+ let plan = match frame.create_physical_plan().await {
+ Ok(plan) => plan,
+ Err(error) => {
+ physical_planning_ms += elapsed_ms(physical_started);
+ return failed_iteration(
+ iteration,
+ logical_planning_ms,
+ physical_planning_ms,
+ execution_ms,
+ total_started,
+ output_rows,
+ metrics,
+ error.to_string(),
+ );
+ }
+ };
+ physical_planning_ms += elapsed_ms(physical_started);
+
+ let execution_started = Instant::now();
+ let batches = match collect(plan.clone(),
session.sql.ctx().task_ctx()).await {
+ Ok(batches) => batches,
+ Err(error) => {
+ execution_ms += elapsed_ms(execution_started);
+ metrics.add_plan(plan.as_ref());
+ return failed_iteration(
+ iteration,
+ logical_planning_ms,
+ physical_planning_ms,
+ execution_ms,
+ total_started,
+ output_rows,
+ metrics,
+ error.to_string(),
+ );
+ }
+ };
+ execution_ms += elapsed_ms(execution_started);
+ output_rows += batches
+ .iter()
+ .map(|batch| batch.num_rows() as u64)
+ .sum::<u64>();
+ metrics.add_plan(plan.as_ref());
+ }
+
+ IterationResult {
+ iteration,
+ logical_planning_ms,
+ physical_planning_ms,
+ execution_ms,
+ total_ms: elapsed_ms(total_started),
+ output_rows,
+ metrics,
+ error: None,
+ }
+}
+
+#[allow(clippy::too_many_arguments)]
+fn failed_iteration(
+ iteration: usize,
+ logical_planning_ms: f64,
+ physical_planning_ms: f64,
+ execution_ms: f64,
+ total_started: Instant,
+ output_rows: u64,
+ metrics: PhysicalMetrics,
+ error: String,
+) -> IterationResult {
+ IterationResult {
+ iteration,
+ logical_planning_ms,
+ physical_planning_ms,
+ execution_ms,
+ total_ms: elapsed_ms(total_started),
+ output_rows,
+ metrics,
+ error: Some(error),
+ }
+}
+
+fn elapsed_ms(started: Instant) -> f64 {
+ started.elapsed().as_secs_f64() * 1_000.0
+}
+
+fn quote_identifier(identifier: &str) -> String {
+ format!("\"{}\"", identifier.replace('"', "\"\""))
+}
diff --git a/benchmarks/tpcds/tests/smoke.rs b/benchmarks/tpcds/tests/smoke.rs
new file mode 100644
index 00000000..ff175b0f
--- /dev/null
+++ b/benchmarks/tpcds/tests/smoke.rs
@@ -0,0 +1,225 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::fs::{self, File};
+use std::sync::Arc;
+
+use arrow_array::{Int32Array, Int64Array, RecordBatch, StringArray};
+use arrow_schema::{DataType, Field, Schema};
+use paimon_tpcds_bench::{
+ execute_command, load_parquet_table, load_query_files,
open_catalog_session,
+ register_parquet_tables, run_query_file, BenchmarkReport,
BenchmarkRuntimeConfig, Command,
+ ExistingPolicyArg, ExistingTablePolicy, LoadArgs, LoadStatus,
QueryRunConfig, RunArgs,
+ RuntimeArgs, SourceKind,
+};
+use parquet::arrow::ArrowWriter;
+use tempfile::TempDir;
+
+fn write_fixture(root: &TempDir, table: &str) {
+ let table_dir = root.path().join(format!("{table}.parquet"));
+ fs::create_dir_all(&table_dir).unwrap();
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("id", DataType::Int32, false),
+ Field::new("name", DataType::Utf8, true),
+ ]));
+ let batch = RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![
+ Arc::new(Int32Array::from(vec![1, 2, 3])),
+ Arc::new(StringArray::from(vec!["a", "b", "c"])),
+ ],
+ )
+ .unwrap();
+ let mut writer = ArrowWriter::try_new(
+ File::create(table_dir.join("part-0.parquet")).unwrap(),
+ schema,
+ None,
+ )
+ .unwrap();
+ writer.write(&batch).unwrap();
+ writer.close().unwrap();
+}
+
+#[tokio::test]
+async fn parquet_fixture_loads_into_paimon() {
+ let data = TempDir::new().unwrap();
+ let warehouse = TempDir::new().unwrap();
+ write_fixture(&data, "store_sales");
+ let session = open_catalog_session(
+ &BenchmarkRuntimeConfig::default(),
+ warehouse.path(),
+ "tpcds",
+ )
+ .await
+ .unwrap();
+
+ let loaded = load_parquet_table(
+ &session,
+ data.path(),
+ "store_sales",
+ ExistingTablePolicy::Error,
+ )
+ .await
+ .unwrap();
+
+ assert_eq!(loaded.status, LoadStatus::Loaded);
+ assert_eq!(loaded.rows, 3);
+ let batches = session
+ .sql
+ .sql("SELECT COUNT(*) FROM paimon.tpcds.store_sales")
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ let counts = batches[0]
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int64Array>()
+ .unwrap();
+ assert_eq!(counts.value(0), 3);
+}
+
+#[tokio::test]
+async fn loaded_paimon_table_runs_warmups_and_measured_iterations() {
+ let data = TempDir::new().unwrap();
+ let warehouse = TempDir::new().unwrap();
+ let queries = TempDir::new().unwrap();
+ write_fixture(&data, "store_sales");
+ fs::write(
+ queries.path().join("q1.sql"),
+ "SELECT COUNT(*), SUM(id) FROM store_sales;",
+ )
+ .unwrap();
+ let session = open_catalog_session(
+ &BenchmarkRuntimeConfig::default(),
+ warehouse.path(),
+ "tpcds",
+ )
+ .await
+ .unwrap();
+ load_parquet_table(
+ &session,
+ data.path(),
+ "store_sales",
+ ExistingTablePolicy::Error,
+ )
+ .await
+ .unwrap();
+ let query = load_query_files(queries.path(), &[1]).unwrap().remove(0);
+
+ let result = run_query_file(
+ &session,
+ &query,
+ &QueryRunConfig {
+ warmup_iterations: 1,
+ measured_iterations: 2,
+ },
+ )
+ .await;
+
+ assert_eq!(result.query, 1);
+ assert_eq!(result.warmup_failures.len(), 0);
+ assert_eq!(result.iterations.len(), 2);
+ assert!(result.iterations.iter().all(|iteration| {
+ iteration.error.is_none() && iteration.output_rows == 1 &&
iteration.total_ms > 0.0
+ }));
+}
+
+#[tokio::test]
+async fn parquet_baseline_uses_the_same_query_runner() {
+ let data = TempDir::new().unwrap();
+ let catalog = TempDir::new().unwrap();
+ let queries = TempDir::new().unwrap();
+ write_fixture(&data, "store_sales");
+ fs::write(
+ queries.path().join("q1.sql"),
+ "SELECT COUNT(*), SUM(id) FROM store_sales;",
+ )
+ .unwrap();
+ let session = open_catalog_session(&BenchmarkRuntimeConfig::default(),
catalog.path(), "tpcds")
+ .await
+ .unwrap();
+ register_parquet_tables(&session, data.path(), &["store_sales"])
+ .await
+ .unwrap();
+ let query = load_query_files(queries.path(), &[1]).unwrap().remove(0);
+
+ let result = run_query_file(
+ &session,
+ &query,
+ &QueryRunConfig {
+ warmup_iterations: 0,
+ measured_iterations: 1,
+ },
+ )
+ .await;
+
+ assert_eq!(result.iterations[0].output_rows, 1);
+ assert!(result.iterations[0].error.is_none());
+}
+
+#[tokio::test]
+async fn command_orchestration_loads_and_writes_a_run_report() {
+ let data = TempDir::new().unwrap();
+ let warehouse = TempDir::new().unwrap();
+ let queries = TempDir::new().unwrap();
+ let output = warehouse.path().join("report.json");
+ write_fixture(&data, "store_sales");
+ fs::write(
+ queries.path().join("q1.sql"),
+ "SELECT COUNT(*) FROM store_sales;",
+ )
+ .unwrap();
+ let runtime = RuntimeArgs {
+ target_partitions: Some(2),
+ memory_limit_gib: None,
+ spill_dir: None,
+ max_spill_gib: None,
+ };
+
+ execute_command(Command::Load(LoadArgs {
+ data: data.path().to_path_buf(),
+ warehouse: warehouse.path().to_path_buf(),
+ database: "tpcds".to_string(),
+ tables: Some("store_sales".to_string()),
+ if_exists: ExistingPolicyArg::Error,
+ runtime: runtime.clone(),
+ }))
+ .await
+ .unwrap();
+ execute_command(Command::Run(RunArgs {
+ source: SourceKind::Paimon,
+ data: None,
+ warehouse: warehouse.path().to_path_buf(),
+ queries: queries.path().to_path_buf(),
+ output: output.clone(),
+ database: "tpcds".to_string(),
+ query: Some("1".to_string()),
+ tables: None,
+ warmup: 0,
+ iterations: 1,
+ runtime,
+ }))
+ .await
+ .unwrap();
+
+ let report: BenchmarkReport =
serde_json::from_slice(&fs::read(output).unwrap()).unwrap();
+ assert_eq!(report.source, SourceKind::Paimon);
+ assert_eq!(report.queries.len(), 1);
+ assert!(!report.has_failures());
+}
