uddhavdave commented on code in PR #25149:
URL: https://github.com/apache/datafusion/pull/25149#discussion_r4092533272
##########
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:
Thanks for pointing this out. Agree, it's better to have a deterministic
coverage. I replaced the scheduling-dependent SLT with a Rust integration test
that forces the interleaving. verified that it fails deterministically on the
old code.
Thank you @alamb for the suggestion. I tried setting target_partitions = 1
but had to remove the `CombinePartialFinalAggregate` optimizer rule to make
sure aggregation mode does not get rewritten to `single`. The test works well.
--
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]
