test: run microbenchmarks in extended CI

.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)"

.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)

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

datafusion/core/benches/sql_planner.rs

@@ -203,17 +203,10 @@ fn register_defs(ctx: SessionContext, defs: Vec<TableDef>) -> 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::<Vec<_>>();
 
-    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| {

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,15 +108,19 @@ 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();
     options.execution.time_zone = Some("UTC".parse().unwrap());
     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);

datafusion/functions/benches/string_expressions/repeat.rs

@@ -60,7 +60,6 @@ fn create_args<O: OffsetSizeTrait>(
 
 fn invoke_repeat_with_args(
     args: Vec<ColumnarValue>,
-    repeat_times: i64,
 ) -> Result<ColumnarValue, DataFusionError> {
     let arg_fields = args
         .iter()
@@ -69,11 +68,22 @@ fn invoke_repeat_with_args(
         .collect::<Vec<_>>();
     let config_options = Arc::new(ConfigOptions::default());
 
+    let return_type = string::repeat().return_type(
+        &args
+            .iter()
+            .map(ColumnarValue::data_type)
+            .collect::<Vec<_>>(),
+    )?;
+    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())
                 })
             },
         );
@@ -210,12 +220,25 @@ fn criterion_benchmark(c: &mut Criterion) {
         group.measurement_time(Duration::from_secs(10));
 
         let args = create_args::<i32>(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| {
                 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",
+                        ),
+                    )
                 })
             },
         );

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(
             <Arc<Field>>::from(Field::new(format!("arg_{idx}"), arg.data_type(), true))
         })
         .collect::<Vec<_>>();
+    // 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::<Vec<_>>();
+    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(),
@@ -161,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"),