sunchao commented on code in PR #25579:
URL: https://github.com/apache/datafusion/pull/25579#discussion_r4074509468
##########
datafusion/sqllogictest/test_files/push_down_filter_regression.slt:
##########
@@ -443,6 +443,65 @@ drop table agg_dyn_mixed;
statement ok
reset datafusion.execution.batch_size;
+# --- one aggregate has a bound before another sees its first non-NULL value
---
+# Both files establish MIN(a)=1 and MAX(a)=8 in their first row group. Until b
+# has a non-NULL bound, filtering only on a would prune every remaining row and
+# incorrectly leave MIN(b) and MAX(b) NULL, regardless of partition order.
+
+statement ok
+set datafusion.execution.batch_size = 2;
+
+statement ok
+COPY (
+ SELECT * FROM (VALUES (1, NULL), (8, NULL), (2, 5), (3, 7)) AS v(a, b)
+) TO
'test_files/scratch/push_down_filter_regression/agg_dyn_late_bound/file_0.parquet'
+STORED AS PARQUET
+OPTIONS ('format.max_row_group_size' '2');
+
+statement ok
+COPY (
+ SELECT * FROM (VALUES (1, NULL), (8, NULL), (2, 5), (3, 7)) AS v(a, b)
+) TO
'test_files/scratch/push_down_filter_regression/agg_dyn_late_bound/file_1.parquet'
+STORED AS PARQUET
+OPTIONS ('format.max_row_group_size' '2');
+
+statement ok
+CREATE EXTERNAL TABLE agg_dyn_late_bound (a INT, b INT)
+STORED AS PARQUET
+LOCATION 'test_files/scratch/push_down_filter_regression/agg_dyn_late_bound/';
+
+statement ok
+set datafusion.optimizer.enable_aggregate_dynamic_filter_pushdown = false;
+
+query IIII
+SELECT MIN(a), MAX(a), MIN(b), MAX(b) FROM agg_dyn_late_bound;
+----
+1 8 5 7
+
+statement ok
+set datafusion.optimizer.enable_aggregate_dynamic_filter_pushdown = true;
+
+query IIII
+SELECT MIN(a), MAX(a), MIN(b), MAX(b) FROM agg_dyn_late_bound;
+----
+1 8 5 7
+
+# Filtering resumes once every aggregate has seen a non-NULL value.
+query TT
+EXPLAIN ANALYZE SELECT MIN(a), MAX(a), MIN(b), MAX(b) FROM agg_dyn_late_bound;
+----
+Plan with Metrics
+01)AggregateExec: mode=Final, gby=[], aggr=[min(agg_dyn_late_bound.a),
max(agg_dyn_late_bound.a), min(agg_dyn_late_bound.b),
max(agg_dyn_late_bound.b)], metrics=[]
+02)--CoalescePartitionsExec, metrics=[]
+03)----AggregateExec: mode=Partial, gby=[], aggr=[min(agg_dyn_late_bound.a),
max(agg_dyn_late_bound.a), min(agg_dyn_late_bound.b),
max(agg_dyn_late_bound.b)], metrics=[]
+04)------DataSourceExec: file_groups={2 groups:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_regression/agg_dyn_late_bound/file_0.parquet],
[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_regression/agg_dyn_late_bound/file_1.parquet]]},
projection=[a, b], file_type=parquet, predicate=DynamicFilter [ a@0 < 1 OR a@0
> 8 OR b@1 < 5 OR b@1 > 7 ], dynamic_rg_pruning=eligible,
pruning_predicate=a_null_count@1 != row_count@2 AND a_min@0 < 1 OR
a_null_count@1 != row_count@2 AND a_max@3 > 8 OR b_null_count@5 != row_count@2
AND b_min@4 < 5 OR b_null_count@5 != row_count@2 AND b_max@6 > 7,
required_guarantees=[], metrics=[]
Review Comment:
[P2] Make the final dynamic-filter assertion deterministic
This exact snapshot can fail even when the query correctly returns `1 8 5
7`. Giving both files the same extrema does not guarantee the final filter when
the first batch publishes `true`:
1. Partition A processes its first batch and constructs `true`, then is
preempted before `filter.update`.
2. Partition B processes both batches and publishes the complete four-bound
predicate.
3. A resumes and publishes its saved `true`. Its second batch does not
improve any shared bound, so `bounds_changed` stays false and the complete
predicate is never republished.
The partitions run in independent tasks on a multithreaded runtime, and
predicate construction/publication are not covered by one lock. I verified this
interleaving with a deterministic native test using the production helpers:
both partitions produced `[1, 8, 5, 7]`, while the final predicate was `true`.
This is a controlled scheduling witness, not a naturally observed SLT failure.
The publication race predates this PR; the new exact plan assertion introduces
the flaky expectation.
Please retain the enabled/disabled SQL result assertions and verify
typed-null recovery/resumed filtering in a controlled execution test, or
otherwise make this snapshot assertion insensitive to the valid publication
order.
--
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]
