From 20287fc56a28b59e9e5863b959913e8cda7ce014 Mon Sep 17 00:00:00 2001 From: aetos Date: Mon, 28 Sep 2026 20:58:48 +0530 Subject: [PATCH 1/3] test: run microbenchmarks in extended CI Execute Criterion benchmarks in test mode with debug checks and a small planner fixture. Repair stale benchmark argument and return types exposed by the new check. Closes apache/datafusion#24868 --- .asf.yaml | 1 + .github/workflows/extended.yml | 22 ++++++++ Cargo.toml | 5 ++ datafusion/core/benches/sql_planner.rs | 35 ++++++------ .../datetime_expressions/to_timestamp.rs | 10 +++- .../benches/string_expressions/repeat.rs | 36 ++++++++---- datafusion/spark/benches/slice.rs | 19 ++++++- xtask/src/ci_steps.rs | 55 ++++++++++++++++++- 8 files changed, 147 insertions(+), 36 deletions(-) diff --git a/.asf.yaml b/.asf.yaml index 4515eb3feb6f2..435e65ccc64dd 100644 --- a/.asf.yaml +++ b/.asf.yaml @@ -69,6 +69,7 @@ github: - "cargo check datafusion-spark features" - "cargo test (amd64)" - "cargo test 'extended_tests' (amd64)" + - "microbenchmark tests (amd64)" - "cargo test hash collisions (amd64)" - "cargo test datafusion-cli (amd64)" - "cargo examples (amd64)" diff --git a/.github/workflows/extended.yml b/.github/workflows/extended.yml index 119a6da234e71..46195097ac770 100644 --- a/.github/workflows/extended.yml +++ b/.github/workflows/extended.yml @@ -90,6 +90,28 @@ jobs: - name: Cleanup run: cargo clean + # Execute Criterion benchmarks once to catch panics without measuring performance. + linux-test-microbenchmarks: + name: microbenchmark tests (amd64) + # Long-running job: run in the merge queue, on release pushes, or manually. + if: github.event_name != 'pull_request' + runs-on: ${{ vars.USE_RUNS_ON == 'true' && format('runs-on={0},family=m8a+m7a+c8a,cpu=16,image=ubuntu24-full-x64,extras=s3-cache,disk=large,tag=datafusion', github.run_id) || 'ubuntu-latest' }} + container: + image: amd64/rust + steps: + - uses: runs-on/action@efac073ea2507ec18797de3a81704201ade11d9d # v2.3.1 + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + ref: ${{ github.event.inputs.pr_head_sha }} # otherwise use the event's commit, including the merge group + submodules: true + fetch-depth: 1 + - name: Setup Rust toolchain + uses: ./.github/actions/setup-builder + with: + rust-version: stable + - name: Test microbenchmarks + run: cargo xtask ci step test microbenchmarks + # Check answers are correct when hash values collide hash-collisions: name: cargo test hash collisions (amd64) diff --git a/Cargo.toml b/Cargo.toml index f3cc9aed9734f..c98574c02f12c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -408,3 +408,8 @@ debug = false debug-assertions = false strip = "debuginfo" incremental = false + +# Execute benchmark tests with debug checks, without release-mode build costs. +[profile.ci-bench] +inherits = "ci" +opt-level = 1 diff --git a/datafusion/core/benches/sql_planner.rs b/datafusion/core/benches/sql_planner.rs index 5e5a63c0e9166..12d50840ac607 100644 --- a/datafusion/core/benches/sql_planner.rs +++ b/datafusion/core/benches/sql_planner.rs @@ -203,17 +203,10 @@ fn register_defs(ctx: SessionContext, defs: Vec) -> SessionContext { ctx } -fn register_clickbench_hits_table(rt: &Runtime) -> SessionContext { +fn register_clickbench_hits_table(rt: &Runtime, path: &str) -> SessionContext { let ctx = SessionContext::new(); // use an external table for clickbench benchmarks - let path = - if PathBuf::from(format!("{BENCHMARKS_PATH_1}{CLICKBENCH_DATA_PATH}")).exists() { - format!("{BENCHMARKS_PATH_1}{CLICKBENCH_DATA_PATH}") - } else { - format!("{BENCHMARKS_PATH_2}{CLICKBENCH_DATA_PATH}") - }; - let sql = format!("CREATE EXTERNAL TABLE hits_raw STORED AS PARQUET LOCATION '{path}'"); @@ -355,15 +348,23 @@ fn union_orderby_query(n: usize) -> String { } fn criterion_benchmark(c: &mut Criterion) { + // CI uses the small checked-in fixture; performance runs use the full dataset. + let clickbench_data_path = + std::env::var("CLICKBENCH_DATA_PATH").unwrap_or_else(|_| { + let path = format!("{BENCHMARKS_PATH_1}{CLICKBENCH_DATA_PATH}"); + if PathBuf::from(&path).exists() { + path + } else { + format!("{BENCHMARKS_PATH_2}{CLICKBENCH_DATA_PATH}") + } + }); // verify that we can load the clickbench data prior to running the benchmark - if !PathBuf::from(format!("{BENCHMARKS_PATH_1}{CLICKBENCH_DATA_PATH}")).exists() - && !PathBuf::from(format!("{BENCHMARKS_PATH_2}{CLICKBENCH_DATA_PATH}")).exists() - { - panic!( - "benchmarks/data/hits_partitioned/ could not be loaded. Please run \ - 'benchmarks/bench.sh data clickbench_partitioned' prior to running this benchmark" - ) - } + assert!( + PathBuf::from(&clickbench_data_path).exists(), + "{clickbench_data_path} could not be loaded. Please run \ + 'benchmarks/bench.sh data clickbench_partitioned' prior to running this benchmark \ + or set CLICKBENCH_DATA_PATH to an existing dataset" + ); let ctx = create_context(); let rt = Runtime::new().unwrap(); @@ -693,7 +694,7 @@ fn criterion_benchmark(c: &mut Criterion) { ))) .collect::>(); - let clickbench_ctx = register_clickbench_hits_table(&rt); + let clickbench_ctx = register_clickbench_hits_table(&rt, &clickbench_data_path); // for (i, sql) in clickbench_queries.iter().enumerate() { // c.bench_function(&format!("logical_plan_clickbench_q{}", i + 1), |b| { diff --git a/datafusion/functions/benches/datetime_expressions/to_timestamp.rs b/datafusion/functions/benches/datetime_expressions/to_timestamp.rs index aaf6610e1cbfe..bcfdb268432a0 100644 --- a/datafusion/functions/benches/datetime_expressions/to_timestamp.rs +++ b/datafusion/functions/benches/datetime_expressions/to_timestamp.rs @@ -21,7 +21,7 @@ use std::sync::Arc; use arrow::array::builder::StringBuilder; use arrow::array::{Array, ArrayRef, StringArray}; use arrow::compute::cast; -use arrow::datatypes::{DataType, Field, TimeUnit}; +use arrow::datatypes::{DataType, Field}; use criterion::{Criterion, criterion_group}; use datafusion_common::config::ConfigOptions; use datafusion_expr::{ColumnarValue, ScalarFunctionArgs}; @@ -108,8 +108,6 @@ fn data_with_formats() -> (StringArray, StringArray, StringArray, StringArray) { ) } fn criterion_benchmark(c: &mut Criterion) { - let return_field = - Field::new("f", DataType::Timestamp(TimeUnit::Nanosecond, None), true).into(); let arg_field = Field::new("a", DataType::Utf8, false).into(); let arg_fields = vec![arg_field]; let mut options = ConfigOptions::default(); @@ -117,6 +115,12 @@ fn criterion_benchmark(c: &mut Criterion) { let config_options = Arc::new(options); let to_timestamp_udf = to_timestamp(config_options.as_ref()); + let return_field = Field::new( + "f", + to_timestamp_udf.return_type(&[DataType::Utf8]).unwrap(), + true, + ) + .into(); c.bench_function("to_timestamp_no_formats_utf8", |b| { let to_timestamp_udf = Arc::clone(&to_timestamp_udf); diff --git a/datafusion/functions/benches/string_expressions/repeat.rs b/datafusion/functions/benches/string_expressions/repeat.rs index faa4953838da9..244e52d2db65d 100644 --- a/datafusion/functions/benches/string_expressions/repeat.rs +++ b/datafusion/functions/benches/string_expressions/repeat.rs @@ -60,7 +60,6 @@ fn create_args( fn invoke_repeat_with_args( args: Vec, - repeat_times: i64, ) -> Result { let arg_fields = args .iter() @@ -69,11 +68,22 @@ fn invoke_repeat_with_args( .collect::>(); let config_options = Arc::new(ConfigOptions::default()); + let return_type = string::repeat().return_type( + &args + .iter() + .map(ColumnarValue::data_type) + .collect::>(), + )?; + let number_rows = match &args[0] { + ColumnarValue::Array(array) => array.len(), + ColumnarValue::Scalar(_) => 1, + }; + string::repeat().invoke_with_args(ScalarFunctionArgs { args, arg_fields, - number_rows: repeat_times as usize, - return_field: Field::new("f", DataType::Utf8, true).into(), + number_rows, + return_field: Field::new("f", return_type, true).into(), config_options: Arc::clone(&config_options), }) } @@ -111,7 +121,7 @@ fn criterion_benchmark(c: &mut Criterion) { Field::new("b", DataType::Int64, false).into(), ], number_rows: 1, - return_field: Field::new("f", DataType::Utf8, true).into(), + return_field: Field::new("f", DataType::Utf8View, true).into(), config_options: Arc::clone(&config_options), }; b.iter(|| black_box(repeat_fn.invoke_with_args(args.clone()).unwrap())) @@ -131,7 +141,7 @@ fn criterion_benchmark(c: &mut Criterion) { |b| { b.iter(|| { let args_cloned = args.clone(); - black_box(invoke_repeat_with_args(args_cloned, repeat_times)) + black_box(invoke_repeat_with_args(args_cloned).unwrap()) }) }, ); @@ -142,7 +152,7 @@ fn criterion_benchmark(c: &mut Criterion) { |b| { b.iter(|| { let args_cloned = args.clone(); - black_box(invoke_repeat_with_args(args_cloned, repeat_times)) + black_box(invoke_repeat_with_args(args_cloned).unwrap()) }) }, ); @@ -153,7 +163,7 @@ fn criterion_benchmark(c: &mut Criterion) { |b| { b.iter(|| { let args_cloned = args.clone(); - black_box(invoke_repeat_with_args(args_cloned, repeat_times)) + black_box(invoke_repeat_with_args(args_cloned).unwrap()) }) }, ); @@ -173,7 +183,7 @@ fn criterion_benchmark(c: &mut Criterion) { |b| { b.iter(|| { let args_cloned = args.clone(); - black_box(invoke_repeat_with_args(args_cloned, repeat_times)) + black_box(invoke_repeat_with_args(args_cloned).unwrap()) }) }, ); @@ -184,7 +194,7 @@ fn criterion_benchmark(c: &mut Criterion) { |b| { b.iter(|| { let args_cloned = args.clone(); - black_box(invoke_repeat_with_args(args_cloned, repeat_times)) + black_box(invoke_repeat_with_args(args_cloned).unwrap()) }) }, ); @@ -195,7 +205,7 @@ fn criterion_benchmark(c: &mut Criterion) { |b| { b.iter(|| { let args_cloned = args.clone(); - black_box(invoke_repeat_with_args(args_cloned, repeat_times)) + black_box(invoke_repeat_with_args(args_cloned).unwrap()) }) }, ); @@ -215,7 +225,11 @@ fn criterion_benchmark(c: &mut Criterion) { |b| { b.iter(|| { let args_cloned = args.clone(); - black_box(invoke_repeat_with_args(args_cloned, repeat_times)) + black_box( + invoke_repeat_with_args(args_cloned).expect_err( + "repeat should reject an overflowing output size", + ), + ) }) }, ); diff --git a/datafusion/spark/benches/slice.rs b/datafusion/spark/benches/slice.rs index da392dc042f92..2686c968ab18c 100644 --- a/datafusion/spark/benches/slice.rs +++ b/datafusion/spark/benches/slice.rs @@ -23,7 +23,8 @@ use arrow::datatypes::{DataType, Field, Int64Type}; use criterion::{Criterion, criterion_group, criterion_main}; use datafusion_common::ScalarValue; use datafusion_common::config::ConfigOptions; -use datafusion_expr::{ColumnarValue, ScalarFunctionArgs}; +use datafusion_expr::type_coercion::functions::fields_with_udf; +use datafusion_expr::{ColumnarValue, ReturnFieldArgs, ScalarFunctionArgs}; use datafusion_spark::function::array::slice; use rand::rngs::StdRng; use rand::{Rng, SeedableRng}; @@ -115,6 +116,19 @@ fn array_slice_benchmark( >::from(Field::new(format!("arg_{idx}"), arg.data_type(), true)) }) .collect::>(); + // Apply planner coercion (including ListView to List) before timing the UDF. + let arg_fields = fields_with_udf(&arg_fields, array_slice.as_ref()).unwrap(); + let args = args + .iter() + .zip(&arg_fields) + .map(|(arg, field)| arg.cast_to(field.data_type(), None).unwrap()) + .collect::>(); + let return_field = array_slice + .return_field_from_args(ReturnFieldArgs { + arg_fields: &arg_fields, + scalar_arguments: &vec![None; arg_fields.len()], + }) + .unwrap(); c.bench_function(name, |b| { b.iter(|| { black_box( @@ -123,8 +137,7 @@ fn array_slice_benchmark( args: args.clone(), arg_fields: arg_fields.clone(), number_rows: size, - return_field: Field::new_list_field(args[0].data_type(), true) - .into(), + return_field: Arc::clone(&return_field), config_options: Arc::new(ConfigOptions::default()), }) .unwrap(), diff --git a/xtask/src/ci_steps.rs b/xtask/src/ci_steps.rs index 1f0a2406d6d69..9340f9a5501c5 100644 --- a/xtask/src/ci_steps.rs +++ b/xtask/src/ci_steps.rs @@ -55,7 +55,7 @@ static CI_STEPS: &[StepInfo] = &[ }, StepInfo { command: "test", - help_usage: "test [--explain]", + help_usage: "test [--explain]", help_examples: &["test cli", "test workspace"], help_description: "Run a CI test suite", error_message: "Cargo test step failed", @@ -288,6 +288,38 @@ impl StepContext { "integration-tests", ]); } + "microbenchmarks" => { + // The SQL benchmark harness needs separately generated datasets + // and is covered by the benchmark result verification job. + // Select only bench targets (not lib/bin unit tests), and tell + // Criterion to execute each benchmark once instead of timing it. + // Keep debug assertions and overflow checks enabled without the + // cost of unoptimized execution on the larger benchmark datasets. + command.env( + "CLICKBENCH_DATA_PATH", + self.root + .join("datafusion/core/tests/data/clickbench_hits_10.parquet"), + ); + command.args([ + "test", + "--profile", + "ci-bench", + "--no-fail-fast", + "--workspace", + "--exclude", + "datafusion-examples", + "--exclude", + "datafusion-benchmarks", + "--exclude", + "datafusion-cli", + "--bench", + "*", + "--features", + "datafusion-physical-plan/test_utils", + "--", + "--test", + ]); + } "benchmark-plan" => { command .args([ @@ -614,7 +646,7 @@ fn required_env(name: &str) -> Result { mod tests { use super::*; - const TEST_USAGE: &str = "usage: cargo xtask ci step test [--explain]"; + const TEST_USAGE: &str = "usage: cargo xtask ci step test [--explain]"; fn args(args: &[&str]) -> Vec { args.iter().map(|arg| (*arg).to_string()).collect() @@ -793,6 +825,25 @@ cargo test \ --lib \ --tests \ --features integration-tests +") + }, + }, + Cmd { + command: &["test", "microbenchmarks"], + expected: |actual| { + insta::assert_snapshot!(actual, @r" +cd /workspace && \ +CLICKBENCH_DATA_PATH=/workspace/datafusion/core/tests/data/clickbench_hits_10.parquet \ +cargo test \ +--profile ci-bench \ +--no-fail-fast \ +--workspace \ +--exclude datafusion-examples \ +--exclude datafusion-benchmarks \ +--exclude datafusion-cli \ +--bench '*' \ +--features datafusion-physical-plan/test_utils \ +-- --test ") }, }, From 00c4b0209f3cdb1db09c484e772946544d44aaaf Mon Sep 17 00:00:00 2001 From: aetos Date: Wed, 21 Oct 2026 15:01:55 +0530 Subject: [PATCH 2/3] test: clarify microbenchmark labels and overflow checks --- .../functions/benches/string_expressions/repeat.rs | 9 +++++++++ datafusion/spark/benches/slice.rs | 5 ++++- 2 files changed, 13 insertions(+), 1 deletion(-) diff --git a/datafusion/functions/benches/string_expressions/repeat.rs b/datafusion/functions/benches/string_expressions/repeat.rs index 244e52d2db65d..a374be1f8952d 100644 --- a/datafusion/functions/benches/string_expressions/repeat.rs +++ b/datafusion/functions/benches/string_expressions/repeat.rs @@ -220,6 +220,15 @@ fn criterion_benchmark(c: &mut Criterion) { group.measurement_time(Duration::from_secs(10)); let args = create_args::(size, 2, repeat_times, false); + // Verify once outside the timed loop that the failure is the expected + // overflow error, and not an unrelated error that also happens to fail. + let err = invoke_repeat_with_args(args.clone()) + .expect_err("repeat should reject an overflowing output size") + .to_string(); + assert!( + err.contains("string size overflow on repeat"), + "unexpected error: {err}" + ); group.bench_function( format!("repeat_string overflow [size={size}, repeat_times={repeat_times}]"), |b| { diff --git a/datafusion/spark/benches/slice.rs b/datafusion/spark/benches/slice.rs index 2686c968ab18c..dd2ab787e8458 100644 --- a/datafusion/spark/benches/slice.rs +++ b/datafusion/spark/benches/slice.rs @@ -174,7 +174,10 @@ fn criterion_benchmark(c: &mut Criterion) { let scalar_to = ColumnarValue::Scalar(ScalarValue::from(child_array_size as i64 / 2)); for input in [list_array, list_view_array] { - let input_type = input.data_type().to_string(); + let input_type = match input.data_type() { + DataType::ListView(_) => "ListView -> List (coercion excluded)".to_string(), + _ => input.data_type().to_string(), + }; array_slice_benchmark( &format!("slice: input {input_type}, array args, no stride"), From 103a62f75c81dcdbd9d1d9b601f5c1b2818902ec Mon Sep 17 00:00:00 2001 From: aetos Date: Wed, 21 Oct 2026 15:01:55 +0530 Subject: [PATCH 3/3] chore(deps): patch certificate verification dependencies --- Cargo.lock | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 714fc492fbe74..c831dee883a55 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5601,9 +5601,9 @@ checksum = "f87165f0995f63a9fbeea62b64d10b4d9d8e78ec6d7d51fb2125fda7bb36788f" [[package]] name = "rustls-webpki" -version = "0.103.15" +version = "0.103.16" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f3c3cf1d8b1e7d4927e2d154c3fcb02979afb9939629c62cd9048d4f07b60ac2" +checksum = "4443b1b6d39d7c663139942cc4558262562d2c84123bb6282ef177009ca8b281" dependencies = [ "aws-lc-rs", "ring", @@ -7244,9 +7244,9 @@ dependencies = [ [[package]] name = "webpki-root-certs" -version = "1.0.9" +version = "1.0.10" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b96554aa2acc8ccdb7e1c9a58a7a68dd5d13bccc69cd124cb09406db612a1c9b" +checksum = "58ec80e73fd8baf57bf0d9615554c75348bda990d6d5bbeab964dd068319fecb" dependencies = [ "rustls-pki-types", ]