This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-24817-4a1fe81dba3f0e6cec3fa78292d545159b42019a in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit b3cb365dd6e5e0441f3f58817fee55dae1814fdb Author: linfeng <[email protected]> AuthorDate: Thu Sep 10 09:12:53 2026 +0000 fix: aggregate dynamic filtering with unsupported expressions (#24817) ## Which issue does this PR close? <!-- We generally require a GitHub issue to be filed for all bug fixes and enhancements and this helps us generate change logs for our releases. You can link an issue to this PR using the GitHub syntax. For example `Closes #123` indicates that this PR will close issue #123. --> - Closes #24816. ## Rationale for this change <!-- Why are you proposing this change? If this is already explained clearly in the issue then this section is not needed. Explaining clearly why changes are proposed helps reviewers understand your changes and offer better suggestions for fixes. Please explain the problem you are trying to solve in terms of the user-visible behavior, rather than the implementation. For example, "The code in `foo.rs` doesn't handle nulls" is a symptom of the implementation. "COUNT(DISTINCT) returns wrong results when the column contains nulls" is the user-visible problem. --> Aggregate dynamic filtering currently ignores unsupported expressions while still generating a shared scan predicate from supported aggregates. This can prune rows required by the unsupported aggregate and produce incorrect results. ## What changes are included in this PR? <!-- There is no need to duplicate the description in the issue here, but it is sometimes worth providing a summary of the individual changes in this PR. --> - Disable aggregate dynamic filtering when any aggregate expression is unsupported. - Document that aggregate dynamic filtering requires every aggregate expression to be supported. - Update the existing sqllogictest to verify that no dynamic filter is produced for mixed supported and unsupported expressions. ## What is the testing strategy for this PR? <!-- We typically require tests for all PRs in order to: 1. Prevent the code from being accidentally broken by subsequent changes 2. Serve as another way to document the expected behavior of the code Briefly describe how this PR is tested, and point to the specific tests you added. For example: 'This new feature is covered by the `sqllogictest` cases added in `foo.slt`'. If this PR does not add tests, explain why. For example, if the change is already covered by existing tests, please mention it. You should also check the `codecov` bot reply on this PR to confirm the changed code is exercised. --> Yes. Updated the existing sqllogictest case in `push_down_filter_regression.slt`. ## Are there any user-facing changes? <!-- If there are user-facing changes then we may require documentation to be updated before approving the PR. If there are any breaking changes to public APIs, please add the `api change` label. --> Yes. This fixes potentially incorrect query results. No API changes. --- .../src/aggregates/aggregate_stream.rs | 6 ++-- datafusion/physical-plan/src/aggregates/mod.rs | 40 +++++++++++++--------- .../test_files/push_down_filter_regression.slt | 35 +++++++++++++------ 3 files changed, 51 insertions(+), 30 deletions(-) diff --git a/datafusion/physical-plan/src/aggregates/aggregate_stream.rs b/datafusion/physical-plan/src/aggregates/aggregate_stream.rs index 23f74e6352..862a44cb20 100644 --- a/datafusion/physical-plan/src/aggregates/aggregate_stream.rs +++ b/datafusion/physical-plan/src/aggregates/aggregate_stream.rs @@ -109,9 +109,9 @@ impl AggregateStreamInner { }; let mut predicates: Vec<Arc<dyn PhysicalExpr>> = - Vec::with_capacity(filter_state.supported_accumulators_info.len()); + Vec::with_capacity(filter_state.accumulator_dyn_filter_info.len()); - for acc_info in &filter_state.supported_accumulators_info { + for acc_info in &filter_state.accumulator_dyn_filter_info { // Skip if we don't yet have a meaningful bound let bound = { let guard = acc_info.shared_bound.lock(); @@ -171,7 +171,7 @@ impl AggregateStreamInner { let mut bounds_changed = false; - for acc_info in &filter_state.supported_accumulators_info { + for acc_info in &filter_state.accumulator_dyn_filter_info { let acc = self.accumulators .get_mut(acc_info.aggr_index) diff --git a/datafusion/physical-plan/src/aggregates/mod.rs b/datafusion/physical-plan/src/aggregates/mod.rs index 9cab9fa276..fc6a676e95 100644 --- a/datafusion/physical-plan/src/aggregates/mod.rs +++ b/datafusion/physical-plan/src/aggregates/mod.rs @@ -754,10 +754,11 @@ impl From<StreamType> for SendableRecordBatchStream { /// /// ## Enable Condition /// - No grouping (no `GROUP BY` clause in the sql, only a single global group to aggregate) -/// - The aggregate expression must be `min`/`max`, and evaluate directly on columns. -/// Note multiple aggregate expressions that satisfy this requirement are allowed, -/// and a dynamic filter will be constructed combining all applicable expr's -/// states. See more in the following example with dynamic filter on multiple columns. +/// - Every aggregate expression must be `min`/`max`, and evaluate directly on a +/// column. If any aggregate expression is unsupported, dynamic filtering is +/// disabled for the entire [`AggregateExec`]. Multiple supported aggregate +/// expressions are combined into one dynamic filter. See the following example +/// with a dynamic filter on multiple columns. /// /// ## Filter Construction /// The filter is kept in the `DataSourceExec`, and it will gets update during execution, @@ -777,11 +778,11 @@ struct AggrDynFilter { /// The current bounds for the dynamic filter, updates during the execution to /// tighten the bound for more effective pruning. /// - /// Each vector element is for the accumulators that support dynamic filter. - /// e.g. This `AggregateExec` has accumulator: - /// min(a), avg(a), max(b) - /// And this field stores [PerAccumulatorDynFilter(min(a)), PerAccumulatorDynFilter(min(b))] - supported_accumulators_info: Vec<PerAccumulatorDynFilter>, + /// Each vector element corresponds to one aggregate expression. Dynamic filtering + /// is enabled only when every aggregate expression is supported, so this vector + /// contains an entry for every accumulator. For example, `min(a), max(b)` produces + /// entries for `min(a)` and `max(b)`. + accumulator_dyn_filter_info: Vec<PerAccumulatorDynFilter>, } // ---- Aggregate Dynamic Filter Utility Structs ---- @@ -1169,7 +1170,7 @@ impl AggregateExec { }; // Validate that the filter is compatible with the aggregation columns. - let cols = self.cols_for_dynamic_filter(&dyn_filter.supported_accumulators_info); + let cols = self.cols_for_dynamic_filter(&dyn_filter.accumulator_dyn_filter_info); if cols.len() != filter.children().len() { return internal_err!( "Dynamic filter expression is incompatible with aggregate due to mismatched number of columns" @@ -1186,7 +1187,7 @@ impl AggregateExec { // Overwrite our filter self.dynamic_filter = Some(Arc::new(AggrDynFilter { filter, - supported_accumulators_info: dyn_filter.supported_accumulators_info.clone(), + accumulator_dyn_filter_info: dyn_filter.accumulator_dyn_filter_info.clone(), })); Ok(self) } @@ -1827,7 +1828,7 @@ impl AggregateExec { return; } - // Collect supported accumulators + // Collect dynamic filter metadata for every accumulator // It is assumed the order of aggregate expressions are not changed from `AggregateExec` // to `AggregateStream` let mut aggr_dyn_filters = Vec::new(); @@ -1858,23 +1859,28 @@ impl AggregateExec { aggr_index: i, shared_bound: Arc::new(Mutex::new(ScalarValue::Null)), }); + } else { + // An incomplete filter could prune rows that still improve an + // unsupported aggregate, so every aggregate must be represented. + // TODO: Derive safe predicates for expressions such as `min(col + literal)`. + return; } } if !aggr_dyn_filters.is_empty() { self.dynamic_filter = Some(Arc::new(AggrDynFilter { filter: Arc::new(DynamicFilterPhysicalExpr::new(all_cols, lit(true))), - supported_accumulators_info: aggr_dyn_filters, + accumulator_dyn_filter_info: aggr_dyn_filters, })) } } - // Collect column references for the dynamic filter expression from the supported accumulators. + // Collect column references for the dynamic filter expression from the accumulators. fn cols_for_dynamic_filter( &self, - supported_accumulators_info: &[PerAccumulatorDynFilter], + accumulator_dyn_filter_info: &[PerAccumulatorDynFilter], ) -> Vec<Arc<dyn PhysicalExpr>> { - let all_cols: Vec<Arc<dyn PhysicalExpr>> = supported_accumulators_info + let all_cols: Vec<Arc<dyn PhysicalExpr>> = accumulator_dyn_filter_info .iter() .filter_map(|info| { // This should always be true due to how the supported accumulators @@ -1887,7 +1893,7 @@ impl AggregateExec { None }) .collect(); - debug_assert_eq!(all_cols.len(), supported_accumulators_info.len()); + debug_assert_eq!(all_cols.len(), accumulator_dyn_filter_info.len()); all_cols } diff --git a/datafusion/sqllogictest/test_files/push_down_filter_regression.slt b/datafusion/sqllogictest/test_files/push_down_filter_regression.slt index 57509fd039..5a038427d0 100644 --- a/datafusion/sqllogictest/test_files/push_down_filter_regression.slt +++ b/datafusion/sqllogictest/test_files/push_down_filter_regression.slt @@ -393,29 +393,41 @@ statement ok drop table agg_dyn_two_col; # --- mixed expressions: MIN(a), MAX(a), MAX(b), MIN(c+1) --- -# Supported aggregates (MIN(a), MAX(a), MAX(b)) should drive a filter; -# MIN(c+1) is unsupported and must not contribute. -# Every file shares the same per-file min(a)=1, max(a)=8 and max(b)=12 so the -# DynamicFilter content is deterministic regardless of publish order (see #22621). +# MIN(c+1) cannot contribute a dynamic-filter predicate. Ignoring it could prune +# rows that still improve MIN(c+1), so the aggregate must not produce a filter. +# Each file starts with a row group that establishes all supported bounds, +# followed by a row group containing the minimum c. With two rows per batch, +# the old filter prunes the required rows in both files regardless of which +# partition runs first. + +statement ok +set datafusion.execution.batch_size = 2; statement ok COPY ( - SELECT * FROM (VALUES (1, 12, 100), (8, 4, 70)) AS v(a, b, c) + SELECT * FROM (VALUES (1, 12, 100), (8, 4, 100), (1, 6, 70), (8, 12, 110)) AS v(a, b, c) ) TO 'test_files/scratch/push_down_filter_regression/agg_dyn_mixed/file_0.parquet' -STORED AS PARQUET; +STORED AS PARQUET +OPTIONS ('format.max_row_group_size' '2'); statement ok COPY ( - SELECT * FROM (VALUES (1, 6, 90), (8, 12, 110)) AS v(a, b, c) + SELECT * FROM (VALUES (1, 12, 100), (8, 4, 100), (1, 6, 70), (8, 12, 110)) AS v(a, b, c) ) TO 'test_files/scratch/push_down_filter_regression/agg_dyn_mixed/file_1.parquet' -STORED AS PARQUET; +STORED AS PARQUET +OPTIONS ('format.max_row_group_size' '2'); statement ok CREATE EXTERNAL TABLE agg_dyn_mixed (a INT, b INT, c INT) STORED AS PARQUET LOCATION 'test_files/scratch/push_down_filter_regression/agg_dyn_mixed/'; -# -> DynamicFilter [ a < 1 OR a > 8 OR b > 12 ] (MIN(c+1) dropped as unsupported) +query IIII +SELECT MIN(a), MAX(a), MAX(b), MIN(c + 1) FROM agg_dyn_mixed; +---- +1 8 12 71 + +# No dynamic filter because not every aggregate has a safe predicate. query TT EXPLAIN ANALYZE SELECT MIN(a), MAX(a), MAX(b), MIN(c + 1) FROM agg_dyn_mixed; ---- @@ -423,11 +435,14 @@ Plan with Metrics 01)AggregateExec: mode=Final, gby=[], aggr=[min(agg_dyn_mixed.a), max(agg_dyn_mixed.a), max(agg_dyn_mixed.b), min(agg_dyn_mixed.c + Int64(1))], metrics=[] 02)--CoalescePartitionsExec, metrics=[] 03)----AggregateExec: mode=Partial, gby=[], aggr=[min(agg_dyn_mixed.a), max(agg_dyn_mixed.a), max(agg_dyn_mixed.b), min(agg_dyn_mixed.c + Int64(1))], metrics=[] -04)------DataSourceExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_regression/agg_dyn_mixed/file_0.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_regression/agg_dyn_mixed/file_1.parquet]]}, projection=[a, b, c], file_type=parquet, predicate=DynamicFilter [ a@0 < 1 OR a@0 > 8 OR b@1 > 12 ], dynamic_rg_pruning=eligible, pruning_predicate=a_null_count@1 != row_count@2 AND a_min@0 < 1 OR a_null_c [...] +04)------DataSourceExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_regression/agg_dyn_mixed/file_0.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_regression/agg_dyn_mixed/file_1.parquet]]}, projection=[a, b, c], file_type=parquet, metrics=[] statement ok drop table agg_dyn_mixed; +statement ok +reset datafusion.execution.batch_size; + # --- all-NULLs input: filter should stay `true` (no meaningful bound) --- statement ok --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
