replace slt with deterministic integration test

Author: uddhavdave

datafusion/core/tests/parquet/filter_pushdown.rs

@@ -26,10 +26,11 @@
 //! select * from data limit 10;
 //! ```
 
-use arrow::array::{ArrayRef, Int32Array, StringArray};
+use arrow::array::{ArrayRef, Int32Array, Int64Array, StringArray};
 use arrow::compute::concat_batches;
 use arrow::error::ArrowError;
 use arrow::record_batch::RecordBatch;
+use datafusion::execution::SessionStateBuilder;
 use datafusion::physical_plan::metrics::{MetricValue, MetricsSet};
 use datafusion::physical_plan::{collect, displayable};
 use datafusion::prelude::{
@@ -809,3 +810,91 @@ async fn pushed_down_predicate_reports_the_original_error() {
         "expected the original cast error, got {root:?}"
     );
 }
+
+/// A MIN/MAX dynamic filter must survive a file that lacks the aggregated
+/// column. That file's Partial aggregate evaluates to a typed null
+/// (`Int64(NULL)`), which must not replace the real MIN published later.
+///
+/// With one partition the files are read in order `01_missing`, `02_high`,
+/// `03_low`, so the bad interleaving happens on every run. Without the fix,
+/// the MIN bound stays null, the filter becomes `latency_ms > 204`, and
+/// `03_low` is pruned, returning MIN = 200.
+#[tokio::test]
+async fn aggregate_dynamic_filter_ignores_typed_null_bound() {
+    let tempdir = TempDir::new_in(Path::new(".")).unwrap();
+    let dir = tempdir.path();
+
+    let write = |name: &str, col_name: &str, array: ArrayRef| {
+        let batch = RecordBatch::try_from_iter([(col_name, array)]).unwrap();
+        let file = File::create(dir.join(name)).unwrap();
+        let mut writer = ArrowWriter::try_new(file, batch.schema(), None).unwrap();
+        writer.write(&batch).unwrap();
+        writer.close().unwrap();
+    };
+    write(
+        "01_missing.parquet",
+        "host",
+        Arc::new(StringArray::from(vec!["h1"; 5])),
+    );
+    write(
+        "02_high.parquet",
+        "latency_ms",
+        Arc::new(Int64Array::from(vec![200, 201, 202, 203, 204])),
+    );
+    write(
+        "03_low.parquet",
+        "latency_ms",
+        Arc::new(Int64Array::from(vec![100, 101, 102, 103, 104])),
+    );
+
+    // One partition keeps the file order fixed. Drop the rule that would
+    // otherwise fold Partial/Final into a Single aggregate, which does not
+    // create a dynamic filter.
+    let default_state = SessionStateBuilder::new().with_default_features().build();
+    let rules = default_state
+        .physical_optimizers()
+        .iter()
+        .filter(|rule| rule.name() != "CombinePartialFinalAggregate")
+        .cloned()
+        .collect();
+    let config = SessionConfig::new().with_target_partitions(1);
+    let state = SessionStateBuilder::new()
+        .with_config(config)
+        .with_default_features()
+        .with_physical_optimizer_rules(rules)
+        .build();
+    let ctx = SessionContext::new_with_state(state);
+
+    ctx.sql(&format!(
+        "CREATE EXTERNAL TABLE t (latency_ms BIGINT, host VARCHAR) \
+         STORED AS PARQUET LOCATION '{}/'",
+        dir.display()
+    ))
+    .await
+    .unwrap();
+
+    let df = ctx
+        .sql("SELECT min(latency_ms), max(latency_ms) FROM t")
+        .await
+        .unwrap();
+    let plan = df.create_physical_plan().await.unwrap();
+    let plan_str = displayable(plan.as_ref()).indent(false).to_string();
+    assert!(
+        plan_str.contains("mode=Partial") && plan_str.contains("DynamicFilter"),
+        "expected a Partial aggregate with a dynamic filter:\n{plan_str}"
+    );
+
+    let batches = collect(plan, ctx.task_ctx()).await.unwrap();
+    let batch = concat_batches(&batches[0].schema(), &batches).unwrap();
+    let min = batch
+        .column(0)
+        .as_any()
+        .downcast_ref::<Int64Array>()
+        .unwrap();
+    let max = batch
+        .column(1)
+        .as_any()
+        .downcast_ref::<Int64Array>()
+        .unwrap();
+    assert_eq!((min.value(0), max.value(0)), (100, 204), "\n{plan_str}");
+}

datafusion/sqllogictest/test_files/push_down_filter_regression.slt

@@ -571,100 +571,6 @@ reset datafusion.optimizer.max_passes;
 statement ok
 drop table agg_filter_pushdown;
 
-########
-# MIN/MAX dynamic filter over a schema-evolved dataset.
-#
-# One of the files does not contain the aggregated column at all, so that
-# partition's Partial aggregate evaluates to a typed null (Int64(NULL)) rather
-# than ScalarValue::Null. Merging that bound into the shared dynamic filter
-# bound must leave any real MIN/MAX from other partitions untouched. If the
-# typed null were compared as a value, it would win the MIN comparison, the
-# filter would collapse to `latency_ms > 204`, and the file holding the true
-# minimum would be pruned, returning 200 instead of 100.
-#
-# The wrong answer needs the file holding the minimum to be opened after the
-# other two partitions have published their bounds, so that file is named to
-# sort last. Use as many partitions as files so every file is read by its own
-# partition. The outcome still depends on scheduling, so the query is repeated
-# a few times; `scalar_min_max_ignore_typed_nulls` in
-# datafusion/physical-plan/src/aggregates/aggregate_stream.rs covers the
-# merge deterministically.
-
-statement ok
-set datafusion.execution.target_partitions = 8;
-
-statement ok
-COPY (
-  SELECT * FROM (VALUES ('h1'), ('h1'), ('h1'), ('h1'), ('h1')) AS t(host)
-) TO 'test_files/scratch/push_down_filter_regression/agg_dyn_schema_evolution/01_missing.parquet'
-STORED AS PARQUET;
-
-statement ok
-COPY (
-  SELECT * FROM (VALUES (200), (201), (202), (203), (204)) AS t(latency_ms)
-) TO 'test_files/scratch/push_down_filter_regression/agg_dyn_schema_evolution/02_high.parquet'
-STORED AS PARQUET;
-
-statement ok
-COPY (
-  SELECT * FROM (VALUES (100), (101), (102), (103), (104)) AS t(latency_ms)
-) TO 'test_files/scratch/push_down_filter_regression/agg_dyn_schema_evolution/03_low.parquet'
-STORED AS PARQUET;
-
-statement ok
-CREATE EXTERNAL TABLE agg_dyn_schema_evolution (latency_ms BIGINT, host VARCHAR)
-STORED AS PARQUET
-LOCATION 'test_files/scratch/push_down_filter_regression/agg_dyn_schema_evolution/';
-
-# Sanity check that the plan uses a Partial/Final aggregate with a dynamic
-# filter pushed into the scan, and one partition per file.
-query TT
-explain select min(latency_ms), max(latency_ms) from agg_dyn_schema_evolution;
-----
-physical_plan
-01)AggregateExec: mode=Final, gby=[], aggr=[min(agg_dyn_schema_evolution.latency_ms), max(agg_dyn_schema_evolution.latency_ms)]
-02)--CoalescePartitionsExec
-03)----AggregateExec: mode=Partial, gby=[], aggr=[min(agg_dyn_schema_evolution.latency_ms), max(agg_dyn_schema_evolution.latency_ms)]
-04)------DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_regression/agg_dyn_schema_evolution/01_missing.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_regression/agg_dyn_schema_evolution/02_high.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_regression/agg_dyn_schema_evolution/03_low.parquet]]}, projection=[latency_ms], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible
-
-query II
-select min(latency_ms), max(latency_ms) from agg_dyn_schema_evolution;
-----
-100 204
-
-query II
-select min(latency_ms), max(latency_ms) from agg_dyn_schema_evolution;
-----
-100 204
-
-query II
-select min(latency_ms), max(latency_ms) from agg_dyn_schema_evolution;
-----
-100 204
-
-query II
-select min(latency_ms), max(latency_ms) from agg_dyn_schema_evolution;
-----
-100 204
-
-query II
-select min(latency_ms), max(latency_ms) from agg_dyn_schema_evolution;
-----
-100 204
-
-query I
-select min(latency_ms) from agg_dyn_schema_evolution;
-----
-100
-
-query I
-select max(latency_ms) from agg_dyn_schema_evolution;
-----
-204
-
-statement ok
-drop table agg_dyn_schema_evolution;
-
 # Config reset
 
 # The SLT runner sets `target_partitions` to 4 instead of using the default, so