From e17689924bb2b246805dff0fc8fc63c7964f032a Mon Sep 17 00:00:00 2001 From: udev Date: Thu, 10 Sep 2026 15:59:43 +0530 Subject: [PATCH 1/3] fix: treat typed nulls as missing bounds in aggregate dynamic filter merge When aggregate dynamic filter pushdown is enabled with multiple partitions, a partition whose input lacks the aggregated column (for example a schema-evolved Parquet file) evaluates MIN/MAX to a typed null such as `Int64(NULL)`. The shared-bound merge only short-circuited on `ScalarValue::Null`, so the typed null fell through to `partial_cmp`, where `None` orders before `Some(_)`, and replaced a valid shared minimum. The dynamic filter then lost its lower bound and could prune the file holding the true MIN, returning a wrong result depending on partition scheduling. Use `ScalarValue::is_null()` so both untyped and typed nulls are ignored when merging bounds. Closes #25147 --- .../src/aggregates/aggregate_stream.rs | 63 +++++++++++- .../push_down_filter_regression.slt | 96 +++++++++++++++++++ 2 files changed, 155 insertions(+), 4 deletions(-) diff --git a/datafusion/physical-plan/src/aggregates/aggregate_stream.rs b/datafusion/physical-plan/src/aggregates/aggregate_stream.rs index 862a44cb20abe..542fe45019ccf 100644 --- a/datafusion/physical-plan/src/aggregates/aggregate_stream.rs +++ b/datafusion/physical-plan/src/aggregates/aggregate_stream.rs @@ -258,14 +258,24 @@ fn scalar_max(v1: &ScalarValue, v2: &ScalarValue) -> Result { } } +/// Short-circuits `scalar_min` / `scalar_max` when either side is null. +/// +/// Both the untyped [`ScalarValue::Null`] and typed nulls such as +/// `ScalarValue::Int64(None)` are treated as "no bound yet". Typed nulls show +/// up when a partition has no value for the aggregated column, for example a +/// schema-evolved Parquet file that lacks the column entirely. They must never +/// reach `partial_cmp`, where `None` orders before `Some(_)` and would replace a +/// valid shared minimum. fn scalar_cmp_null_short_circuit( v1: &ScalarValue, v2: &ScalarValue, ) -> Option { - match (v1, v2) { - (ScalarValue::Null, ScalarValue::Null) => Some(ScalarValue::Null), - (ScalarValue::Null, other) | (other, ScalarValue::Null) => Some(other.clone()), - _ => None, + if v1.is_null() { + Some(v2.clone()) + } else if v2.is_null() { + Some(v1.clone()) + } else { + None } } @@ -706,6 +716,51 @@ mod tests { Ok(()) } + /// Regression test for . + /// + /// A partition whose input lacks the aggregated column (e.g. a + /// schema-evolved Parquet file) yields a *typed* null bound such as + /// `Int64(None)`. Merging it into the shared bound must not replace a + /// valid value, otherwise the dynamic filter loses its lower/upper bound + /// and can prune files containing the true MIN/MAX. + #[test] + fn scalar_min_max_ignore_typed_nulls() -> Result<()> { + let value = ScalarValue::Int64(Some(100)); + let typed_null = ScalarValue::Int64(None); + let untyped_null = ScalarValue::Null; + + // typed null on either side is ignored + assert_eq!(scalar_min(&value, &typed_null)?, value); + assert_eq!(scalar_min(&typed_null, &value)?, value); + assert_eq!(scalar_max(&value, &typed_null)?, value); + assert_eq!(scalar_max(&typed_null, &value)?, value); + + // untyped null on either side is ignored + assert_eq!(scalar_min(&value, &untyped_null)?, value); + assert_eq!(scalar_min(&untyped_null, &value)?, value); + assert_eq!(scalar_max(&value, &untyped_null)?, value); + assert_eq!(scalar_max(&untyped_null, &value)?, value); + + // null vs null stays null + assert!(scalar_min(&untyped_null, &typed_null)?.is_null()); + assert!(scalar_max(&untyped_null, &typed_null)?.is_null()); + assert!(scalar_min(&typed_null, &typed_null)?.is_null()); + assert!(scalar_max(&typed_null, &typed_null)?.is_null()); + assert!(scalar_min(&typed_null, &untyped_null)?.is_null()); + assert!(scalar_max(&typed_null, &untyped_null)?.is_null()); + assert!(scalar_min(&untyped_null, &untyped_null)?.is_null()); + assert!(scalar_max(&untyped_null, &untyped_null)?.is_null()); + + // non-null values still compare normally + let smaller = ScalarValue::Int64(Some(1)); + assert_eq!(scalar_min(&value, &smaller)?, smaller); + assert_eq!(scalar_min(&smaller, &value)?, smaller); + assert_eq!(scalar_max(&value, &smaller)?, value); + assert_eq!(scalar_max(&smaller, &value)?, value); + + Ok(()) + } + #[tokio::test] async fn aggregate_stream_reports_partial_and_final_phases() -> Result<()> { let schema = Arc::new(Schema::new(vec![ diff --git a/datafusion/sqllogictest/test_files/push_down_filter_regression.slt b/datafusion/sqllogictest/test_files/push_down_filter_regression.slt index 5a038427d0c65..16ba9ea43d33c 100644 --- a/datafusion/sqllogictest/test_files/push_down_filter_regression.slt +++ b/datafusion/sqllogictest/test_files/push_down_filter_regression.slt @@ -571,6 +571,102 @@ reset datafusion.optimizer.max_passes; statement ok drop table agg_filter_pushdown; +######## +# Regression test for https://github.com/apache/datafusion/issues/25147 +# +# MIN/MAX dynamic filter with a schema-evolved dataset: one of the files does +# not contain the aggregated column at all. That partition's Partial +# aggregate evaluates to a *typed* null (Int64(NULL)) rather than +# ScalarValue::Null. The shared-bound merge used to only short-circuit on +# the untyped Null, so the typed null fell through to `partial_cmp`, where +# NULL orders before any value and replaced (or blocked) a valid shared MIN. +# The dynamic filter then lost its lower bound and became `latency_ms > 204` +# alone, which pruned the file holding the true minimum and returned 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 From 5a244aad9e0c77cf0198d1564c5fb15e5b727582 Mon Sep 17 00:00:00 2001 From: udev Date: Thu, 10 Sep 2026 16:13:57 +0530 Subject: [PATCH 2/3] Simplify null short-circuit and clarify comments --- .../src/aggregates/aggregate_stream.rs | 39 ++++++++++--------- .../push_down_filter_regression.slt | 18 ++++----- 2 files changed, 28 insertions(+), 29 deletions(-) diff --git a/datafusion/physical-plan/src/aggregates/aggregate_stream.rs b/datafusion/physical-plan/src/aggregates/aggregate_stream.rs index 542fe45019ccf..b903a6ed3ebcb 100644 --- a/datafusion/physical-plan/src/aggregates/aggregate_stream.rs +++ b/datafusion/physical-plan/src/aggregates/aggregate_stream.rs @@ -260,22 +260,25 @@ fn scalar_max(v1: &ScalarValue, v2: &ScalarValue) -> Result { /// Short-circuits `scalar_min` / `scalar_max` when either side is null. /// -/// Both the untyped [`ScalarValue::Null`] and typed nulls such as -/// `ScalarValue::Int64(None)` are treated as "no bound yet". Typed nulls show -/// up when a partition has no value for the aggregated column, for example a -/// schema-evolved Parquet file that lacks the column entirely. They must never -/// reach `partial_cmp`, where `None` orders before `Some(_)` and would replace a -/// valid shared minimum. +/// Returns the non-null side, or a null when both sides are null. Returns +/// `None` when neither side is null so the caller falls through to a real +/// comparison. +/// +/// A null bound means "no value seen yet", regardless of whether it is the +/// untyped [`ScalarValue::Null`] or a typed null such as +/// `ScalarValue::Int64(None)`. A typed null is what a partition produces when +/// none of its rows carry the aggregated column, for example a Parquet file +/// written before the column was added. Nulls must be handled here rather +/// than in `partial_cmp`, where `None` orders before `Some(_)` and would win a +/// `MIN` comparison against a real value. fn scalar_cmp_null_short_circuit( v1: &ScalarValue, v2: &ScalarValue, ) -> Option { - if v1.is_null() { - Some(v2.clone()) - } else if v2.is_null() { - Some(v1.clone()) - } else { - None + match (v1.is_null(), v2.is_null()) { + (true, _) => Some(v2.clone()), + (_, true) => Some(v1.clone()), + _ => None, } } @@ -716,13 +719,11 @@ mod tests { Ok(()) } - /// Regression test for . - /// - /// A partition whose input lacks the aggregated column (e.g. a - /// schema-evolved Parquet file) yields a *typed* null bound such as - /// `Int64(None)`. Merging it into the shared bound must not replace a - /// valid value, otherwise the dynamic filter loses its lower/upper bound - /// and can prune files containing the true MIN/MAX. + /// A partition whose input lacks the aggregated column (for example a + /// Parquet file written before the column was added) yields a typed null + /// bound such as `Int64(None)`. Merging it into the shared bound must not + /// replace a valid value, otherwise the dynamic filter loses its bound and + /// can prune files containing the true MIN/MAX. #[test] fn scalar_min_max_ignore_typed_nulls() -> Result<()> { let value = ScalarValue::Int64(Some(100)); diff --git a/datafusion/sqllogictest/test_files/push_down_filter_regression.slt b/datafusion/sqllogictest/test_files/push_down_filter_regression.slt index 16ba9ea43d33c..ba073c2a34565 100644 --- a/datafusion/sqllogictest/test_files/push_down_filter_regression.slt +++ b/datafusion/sqllogictest/test_files/push_down_filter_regression.slt @@ -572,17 +572,15 @@ statement ok drop table agg_filter_pushdown; ######## -# Regression test for https://github.com/apache/datafusion/issues/25147 +# MIN/MAX dynamic filter over a schema-evolved dataset. # -# MIN/MAX dynamic filter with a schema-evolved dataset: one of the files does -# not contain the aggregated column at all. That partition's Partial -# aggregate evaluates to a *typed* null (Int64(NULL)) rather than -# ScalarValue::Null. The shared-bound merge used to only short-circuit on -# the untyped Null, so the typed null fell through to `partial_cmp`, where -# NULL orders before any value and replaced (or blocked) a valid shared MIN. -# The dynamic filter then lost its lower bound and became `latency_ms > 204` -# alone, which pruned the file holding the true minimum and returned 200 -# instead of 100. +# 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 From b0bae2355e0eff2d96ec3472d65e3ada42dc5407 Mon Sep 17 00:00:00 2001 From: udev Date: Thu, 10 Sep 2026 16:17:21 +0530 Subject: [PATCH 3/3] fix: remove comments --- .../src/aggregates/aggregate_stream.rs | 13 ------------- 1 file changed, 13 deletions(-) diff --git a/datafusion/physical-plan/src/aggregates/aggregate_stream.rs b/datafusion/physical-plan/src/aggregates/aggregate_stream.rs index b903a6ed3ebcb..c3f66621c5b09 100644 --- a/datafusion/physical-plan/src/aggregates/aggregate_stream.rs +++ b/datafusion/physical-plan/src/aggregates/aggregate_stream.rs @@ -258,19 +258,6 @@ fn scalar_max(v1: &ScalarValue, v2: &ScalarValue) -> Result { } } -/// Short-circuits `scalar_min` / `scalar_max` when either side is null. -/// -/// Returns the non-null side, or a null when both sides are null. Returns -/// `None` when neither side is null so the caller falls through to a real -/// comparison. -/// -/// A null bound means "no value seen yet", regardless of whether it is the -/// untyped [`ScalarValue::Null`] or a typed null such as -/// `ScalarValue::Int64(None)`. A typed null is what a partition produces when -/// none of its rows carry the aggregated column, for example a Parquet file -/// written before the column was added. Nulls must be handled here rather -/// than in `partial_cmp`, where `None` orders before `Some(_)` and would win a -/// `MIN` comparison against a real value. fn scalar_cmp_null_short_circuit( v1: &ScalarValue, v2: &ScalarValue,