alamb commented on code in PR #25149:
URL: https://github.com/apache/datafusion/pull/25149#discussion_r4076326561
##########
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
Review Comment:
I think we could do this as a follow on PR as well. Or perhaps set target
partitions to 1?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
