Support config matrix for SLT tests

datafusion/sqllogictest/README.md

@@ -184,6 +184,47 @@ EXPLAIN ANALYZE SELECT * FROM generate_series(100);
 Plan with Metrics LazyMemoryExec: partitions=1, batch_generators=[generate_series: start=0, end=100, batch_size=8192], metrics=[output_rows=101, elapsed_compute=<slt:ignore>, output_bytes=<slt:ignore>]
 ```
 
+## Cookbook: Sweeping config with `configMatrix`
+
+Runs the same `.slt` once per combination of config values. Each directive is a comment:
+
+```text
+# configMatrix: <key>=<v1>,<v2>[,...]
+```
+
+- Repeat the directive to nest keys. Values are the cartesian product.
+- Whitespace-trimmed and deduped; repeated keys merge value lists.
+- Unknown key or invalid value fails fast, naming the file, key, and value.
+- Supported by the default runner and by `--substrait-round-trip`. `--complete`
+  rejects a file that declares directives, since it would overwrite the file
+  with the output of a single combination. The Postgres runner ignores them.
+- Test failures are prefixed with the combination that produced them.
+
+Nested example (2 × 2 = 4 runs):
+
+```text
+# configMatrix: datafusion.execution.parquet.coerce_int96=ms,us
+# configMatrix: datafusion.execution.parquet.coerce_int96_tz=UTC,America/New_York
+
+statement ok
+CREATE EXTERNAL TABLE int96_from_spark
+STORED AS PARQUET
+LOCATION '../../parquet-testing/data/int96_from_spark.parquet';
+
+query I
+select count(*) from int96_from_spark
+----
+6
+```
+
+Failure output:
+
+```text
+[configMatrix: datafusion.execution.parquet.coerce_int96=ms, datafusion.execution.parquet.coerce_int96_tz=UTC]
+caused by
+External error: 1 errors in file .../parquet_int96_matrix.slt
+```
+
 # Reference
 
 ## Running tests: Validation Mode

datafusion/sqllogictest/bin/sqllogictests.rs

@@ -23,9 +23,9 @@ use datafusion::common::{DataFusionError, Result, exec_datafusion_err, exec_err}
 use datafusion_sqllogictest::DataFusionSubstraitRoundTrip;
 use datafusion_sqllogictest::TestFile;
 use datafusion_sqllogictest::{
-    CurrentlyExecutingSqlTracker, DataFusion, Filter, TestContext, df_value_validator,
-    read_dir_recursive, setup_scratch_dir, should_skip_file, should_skip_record,
-    value_normalizer,
+    CurrentlyExecutingSqlTracker, DFColumnType, DataFusion, Filter, TestContext,
+    df_value_validator, read_dir_recursive, setup_scratch_dir, should_skip_file,
+    should_skip_record, test_configurations, value_normalizer,
 };
 use futures::stream::StreamExt;
 use indicatif::{
@@ -443,33 +443,48 @@ async fn run_test_file_substrait_round_trip(
         path,
         relative_path,
     } = test_file;
-    let Some(test_ctx) = TestContext::try_new_for_test_file(&relative_path).await else {
-        info!("Skipping: {}", path.display());
-        return Ok(());
-    };
-    setup_scratch_dir(&relative_path)?;
 
-    let count: u64 = get_record_count(&path, "DatafusionSubstraitRoundTrip".to_string());
-    let pb = mp.add(ProgressBar::new(count));
+    // Parsed once and replayed for every configuration.
+    let records = parse_records(&path)?;
+    let count = count_records(&records, "DatafusionSubstraitRoundTrip");
 
-    pb.set_style(mp_style);
-    pb.set_message(relative_path.display().to_string());
+    for test_configuration in test_configurations(&path)? {
+        let Some(test_ctx) = TestContext::try_new_for_test_file(&relative_path).await
+        else {
+            info!("Skipping: {}", path.display());
+            return Ok(());
+        };
+        setup_scratch_dir(&relative_path)?;
+        // Before the engine is built: it snapshots config to detect drift.
+        test_ctx.apply_config_overrides(test_configuration.settings(), &relative_path)?;
+
+        let pb = mp.add(ProgressBar::new(count));
+        pb.set_style(mp_style.clone());
+        pb.set_message(relative_path.display().to_string());
+
+        let mut runner = sqllogictest::Runner::new(|| async {
+            Ok(DataFusionSubstraitRoundTrip::new(
+                test_ctx.session_ctx().clone(),
+                relative_path.clone(),
+                pb.clone(),
+            )
+            .with_currently_executing_sql_tracker(
+                currently_executing_sql_tracker.clone(),
+            ))
+        });
+        runner.add_label("DatafusionSubstraitRoundTrip");
+        runner.with_column_validator(strict_column_validator);
+        runner.with_normalizer(value_normalizer);
+        runner.with_validator(validator);
+        let result =
+            run_file_in_runner(&path, &records, &mut runner, filters, colored_output)
+                .await;
+        pb.finish_and_clear();
+
+        test_configuration.attribute_failure(result)?;
+    }
 
-    let mut runner = sqllogictest::Runner::new(|| async {
-        Ok(DataFusionSubstraitRoundTrip::new(
-            test_ctx.session_ctx().clone(),
-            relative_path.clone(),
-            pb.clone(),
-        )
-        .with_currently_executing_sql_tracker(currently_executing_sql_tracker.clone()))
-    });
-    runner.add_label("DatafusionSubstraitRoundTrip");
-    runner.with_column_validator(strict_column_validator);
-    runner.with_normalizer(value_normalizer);
-    runner.with_validator(validator);
-    let res = run_file_in_runner(path, &mut runner, filters, colored_output).await;
-    pb.finish_and_clear();
-    res
+    Ok(())
 }
 
 #[cfg(not(feature = "substrait"))]
@@ -502,64 +517,79 @@ async fn run_test_file(
         path,
         relative_path,
     } = test_file;
-    let Some(test_ctx) = TestContext::try_new_for_test_file(&relative_path).await else {
-        info!("Skipping: {}", path.display());
-        return Ok(());
-    };
-    setup_scratch_dir(&relative_path)?;
-
-    let count: u64 = get_record_count(&path, "Datafusion".to_string());
-    let pb = mp.add(ProgressBar::new(count));
 
-    pb.set_style(mp_style);
-    pb.set_message(relative_path.display().to_string());
+    // Parsed once and replayed for every configuration.
+    let records = parse_records(&path)?;
+    let count = count_records(&records, "Datafusion");
 
-    // If DataFusion configuration has changed during test file runs, errors will be
-    // pushed to this vec.
-    // HACK: managed externally because `sqllogictest` is an external dependency, and
-    // it doesn't have an API to directly access the inner runner.
-    let config_change_errors = Arc::new(Mutex::new(Vec::new()));
-    let mut runner = sqllogictest::Runner::new(|| async {
-        Ok(DataFusion::new(
-            test_ctx.session_ctx().clone(),
-            relative_path.clone(),
-            pb.clone(),
-        )
-        .with_currently_executing_sql_tracker(currently_executing_sql_tracker.clone())
-        .with_config_change_errors(Arc::clone(&config_change_errors)))
-    });
-    runner.add_label("Datafusion");
-    runner.with_column_validator(strict_column_validator);
-    runner.with_normalizer(value_normalizer);
-    runner.with_validator(validator);
-    let result = run_file_in_runner(path, &mut runner, filters, colored_output).await;
-    pb.finish_and_clear();
-
-    result?;
+    for test_configuration in test_configurations(&path)? {
+        let Some(test_ctx) = TestContext::try_new_for_test_file(&relative_path).await
+        else {
+            info!("Skipping: {}", path.display());
+            return Ok(());
+        };
+        setup_scratch_dir(&relative_path)?;
+        // Before the engine is built: it snapshots config to detect drift.
+        test_ctx.apply_config_overrides(test_configuration.settings(), &relative_path)?;
+
+        let pb = mp.add(ProgressBar::new(count));
+        pb.set_style(mp_style.clone());
+        pb.set_message(relative_path.display().to_string());
+
+        // If DataFusion configuration has changed during test file runs, errors will be
+        // pushed to this vec.
+        // HACK: managed externally because `sqllogictest` is an external dependency, and
+        // it doesn't have an API to directly access the inner runner.
+        let config_change_errors = Arc::new(Mutex::new(Vec::new()));
+        let mut runner = sqllogictest::Runner::new(|| async {
+            Ok(DataFusion::new(
+                test_ctx.session_ctx().clone(),
+                relative_path.clone(),
+                pb.clone(),
+            )
+            .with_currently_executing_sql_tracker(currently_executing_sql_tracker.clone())
+            .with_config_change_errors(Arc::clone(&config_change_errors)))
+        });
+        runner.add_label("Datafusion");
+        runner.with_column_validator(strict_column_validator);
+        runner.with_normalizer(value_normalizer);
+        runner.with_validator(validator);
+        let result =
+            run_file_in_runner(&path, &records, &mut runner, filters, colored_output)
+                .await;
+        pb.finish_and_clear();
+
+        test_configuration.attribute_failure(result)?;
+
+        // If there was no correctness error, check that the config is unchanged.
+        runner.shutdown_async().await;
+        test_configuration
+            .attribute_failure(config_change_result(&config_change_errors))?;
+    }
 
-    // If there was no correctness error, check that the config is unchanged.
-    runner.shutdown_async().await;
-    config_change_result(&config_change_errors)
+    Ok(())
 }
 
-async fn run_file_in_runner<D: AsyncDB, M: MakeConnection<Conn = D>>(
-    path: PathBuf,
+async fn run_file_in_runner<D, M>(
+    path: &Path,
+    records: &[Record<DFColumnType>],
     runner: &mut sqllogictest::Runner<D, M>,
     filters: &[Filter],
     colored_output: bool,
-) -> Result<()> {
-    let path = path.canonicalize()?;
-    let records =
-        parse_file(&path).map_err(|e| DataFusionError::External(Box::new(e)))?;
+) -> Result<()>
+where
+    D: AsyncDB<ColumnType = DFColumnType>,
+    M: MakeConnection<Conn = D>,
+{
     let mut errs = vec![];
-    for record in records.into_iter() {
+    for record in records {
         if let Record::Halt { .. } = record {
             break;
         }
-        if should_skip_record::<D>(&record, filters) {
+        if should_skip_record::<D>(record, filters) {
             continue;
         }
-        if let Err(err) = runner.run_async(record).await {
+        if let Err(err) = runner.run_async(record.clone()).await {
             if colored_output {
                 errs.push(format!("{}", err.display(true)));
             } else {
@@ -569,6 +599,7 @@ async fn run_file_in_runner<D: AsyncDB, M: MakeConnection<Conn = D>>(
     }
 
     if !errs.is_empty() {
+        let path = path.canonicalize()?;
         let mut msg = format!("{} errors in file {}\n\n", errs.len(), path.display());
         for (i, err) in errs.iter().enumerate() {
             if i >= ERRS_PER_FILE_LIMIT {
@@ -587,30 +618,28 @@ async fn run_file_in_runner<D: AsyncDB, M: MakeConnection<Conn = D>>(
     Ok(())
 }
 
-#[expect(clippy::needless_pass_by_value)]
-fn get_record_count(path: &PathBuf, label: String) -> u64 {
-    let records: Vec<Record<<DataFusion as AsyncDB>::ColumnType>> =
-        parse_file(path).unwrap();
-    let mut count: u64 = 0;
-
-    for rec in &records {
-        match rec {
-            Record::Query { conditions, .. } | Record::Statement { conditions, .. }
-                if conditions.is_empty()
-                    || !conditions.contains(&Condition::SkipIf {
-                        label: label.clone(),
-                    })
-                    || conditions.contains(&Condition::OnlyIf {
-                        label: label.clone(),
-                    }) =>
-            {
-                count += 1;
-            }
-            _ => {}
-        }
-    }
+fn parse_records(path: &Path) -> Result<Vec<Record<DFColumnType>>> {
+    parse_file(path).map_err(|e| DataFusionError::External(Box::new(e)))
+}
 
-    count
+fn count_records(records: &[Record<DFColumnType>], label: &str) -> u64 {
+    let skip_if = Condition::SkipIf {
+        label: label.to_string(),
+    };
+    let only_if = Condition::OnlyIf {
+        label: label.to_string(),
+    };
+    records
+        .iter()
+        .filter(|rec| match rec {
+            Record::Query { conditions, .. } | Record::Statement { conditions, .. } => {
+                conditions.is_empty()
+                    || !conditions.contains(&skip_if)
+                    || conditions.contains(&only_if)
+            }
+            _ => false,
+        })
+        .count() as u64
 }
 
 #[cfg(feature = "postgres")]
@@ -629,7 +658,8 @@ async fn run_test_file_with_postgres(
     } = test_file;
     setup_scratch_dir(&relative_path)?;
 
-    let count: u64 = get_record_count(&path, "postgresql".to_string());
+    let records = parse_records(&path)?;
+    let count = count_records(&records, "postgresql");
     let pb = mp.add(ProgressBar::new(count));
 
     pb.set_style(mp_style);
@@ -646,7 +676,7 @@ async fn run_test_file_with_postgres(
     runner.with_column_validator(strict_column_validator);
     runner.with_normalizer(value_normalizer);
     runner.with_validator(validator);
-    let result = run_file_in_runner(path, &mut runner, filters, false).await;
+    let result = run_file_in_runner(&path, &records, &mut runner, filters, false).await;
     pb.finish_and_clear();
     result
 }
@@ -682,13 +712,25 @@ async fn run_complete_file(
 
     info!("Using complete mode to complete: {}", path.display());
 
+    // `update_test_file` rewrites the file from a single run, so it cannot hold
+    // one expected-output set per configuration.
+    if test_configurations(&path)?.iter().any(|c| !c.is_empty()) {
+        return exec_err!(
+            "Cannot use --complete on {}: it declares `# configMatrix:` \
+             directives, and completion would overwrite the file with the \
+             output of a single configuration",
+            relative_path.display()
+        );
+    }
+
     let Some(test_ctx) = TestContext::try_new_for_test_file(&relative_path).await else {
         info!("Skipping: {}", path.display());
         return Ok(());
     };
     setup_scratch_dir(&relative_path)?;
 
-    let count: u64 = get_record_count(&path, "Datafusion".to_string());
+    // `update_test_file` re-parses the file itself, so only the count is needed.
+    let count = count_records(&parse_records(&path)?, "Datafusion");
     let pb = mp.add(ProgressBar::new(count));
 
     pb.set_style(mp_style);
@@ -744,7 +786,7 @@ async fn run_complete_file_with_postgres(
     );
     setup_scratch_dir(&relative_path)?;
 
-    let count: u64 = get_record_count(&path, "postgresql".to_string());
+    let count = count_records(&parse_records(&path)?, "postgresql");
     let pb = mp.add(ProgressBar::new(count));
 
     pb.set_style(mp_style);