diff --git a/datafusion/physical-plan/src/aggregates/aggregate_stream.rs b/datafusion/physical-plan/src/aggregates/aggregate_stream.rs index 862a44cb20ab..c3f66621c5b0 100644 --- a/datafusion/physical-plan/src/aggregates/aggregate_stream.rs +++ b/datafusion/physical-plan/src/aggregates/aggregate_stream.rs @@ -262,9 +262,9 @@ 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()), + match (v1.is_null(), v2.is_null()) { + (true, _) => Some(v2.clone()), + (_, true) => Some(v1.clone()), _ => None, } } @@ -706,6 +706,49 @@ mod tests { Ok(()) } + /// 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)); + 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 5a038427d0c6..ba073c2a3456 100644 --- a/datafusion/sqllogictest/test_files/push_down_filter_regression.slt +++ b/datafusion/sqllogictest/test_files/push_down_filter_regression.slt @@ -571,6 +571,100 @@ 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