You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "liamzwbao (via GitHub)" <gi...@apache.org> on 2025/07/04 19:07:34 UTC
[PR] Add the missing equivalence info for filter pushdown [datafusion]
liamzwbao opened a new pull request, #16686:
URL: https://github.com/apache/datafusion/pull/16686
## 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 #16563.
## 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.
-->
Add the missing equivalence info so that optimizer can pick it up when pruning SortExec
## 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.
-->
Add the equivalence info based on the filters
## Are these changes tested?
<!--
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
If tests are not included in your PR, please explain why (for example, are they covered by existing tests)?
-->
Added in `parquet_filter_pushdown.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, fix the unexpected behavior
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add the missing equivalence info for filter pushdown [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb merged PR #16686:
URL: https://github.com/apache/datafusion/pull/16686
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add the missing equivalence info for filter pushdown [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #16686:
URL: https://github.com/apache/datafusion/pull/16686#discussion_r2193493034
##########
datafusion/core/tests/physical_optimizer/filter_pushdown/mod.rs:
##########
@@ -289,7 +289,7 @@ fn test_no_pushdown_through_aggregates() {
Ok:
- FilterExec: b@1 = bar
- CoalesceBatchesExec: target_batch_size=100
- - AggregateExec: mode=Final, gby=[a@0 as a, b@1 as b], aggr=[cnt]
+ - AggregateExec: mode=Final, gby=[a@0 as a, b@1 as b], aggr=[cnt], ordering_mode=PartiallySorted([0])
Review Comment:
I agree -- this looks good to me (the filter pushdown now lets the gby know that the data is sorted on `a` which makes sense as `a` now has exactly one value (`foo`)
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add the missing equivalence info for filter pushdown [datafusion]
Posted by "liamzwbao (via GitHub)" <gi...@apache.org>.
liamzwbao commented on code in PR #16686:
URL: https://github.com/apache/datafusion/pull/16686#discussion_r2186078102
##########
datafusion/core/tests/physical_optimizer/filter_pushdown/mod.rs:
##########
@@ -289,7 +289,7 @@ fn test_no_pushdown_through_aggregates() {
Ok:
- FilterExec: b@1 = bar
- CoalesceBatchesExec: target_batch_size=100
- - AggregateExec: mode=Final, gby=[a@0 as a, b@1 as b], aggr=[cnt]
+ - AggregateExec: mode=Final, gby=[a@0 as a, b@1 as b], aggr=[cnt], ordering_mode=PartiallySorted([0])
Review Comment:
This is a side effect for the fix, but I think it's okay as it preserves the `ordering_mode` instead of defaulting it to `Linear`
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add the missing equivalence info for filter pushdown [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #16686:
URL: https://github.com/apache/datafusion/pull/16686#issuecomment-3038695163
Thank you @liamzwbao
@xudong963 or @adriangb do you have time to review this PR?
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add the missing equivalence info for filter pushdown [datafusion]
Posted by "liamzwbao (via GitHub)" <gi...@apache.org>.
liamzwbao commented on code in PR #16686:
URL: https://github.com/apache/datafusion/pull/16686#discussion_r2188365661
##########
datafusion/sqllogictest/test_files/parquet_filter_pushdown.slt:
##########
@@ -239,6 +349,23 @@ physical_plan
05)--------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=2
06)----------DataSourceExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_filter_pushdown/parquet_table/1.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_filter_pushdown/parquet_table/2.parquet]]}, projection=[a, b], file_type=parquet, predicate=b@1 > 2, pruning_predicate=b_null_count@1 != row_count@2 AND b_max@0 > 2, required_guarantees=[]
+query T
+select a from t_pushdown where b = 2 ORDER BY b;
+----
+bar
+
+query TT
+EXPLAIN select a from t_pushdown where b = 2 ORDER BY b;
+----
+logical_plan
+01)Projection: t_pushdown.a
+02)--Sort: t_pushdown.b ASC NULLS LAST
+03)----Filter: t_pushdown.b = Int32(2)
+04)------TableScan: t_pushdown projection=[a, b], partial_filters=[t_pushdown.b = Int32(2)]
+physical_plan
+01)CoalescePartitionsExec
+02)--DataSourceExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_filter_pushdown/parquet_table/1.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_filter_pushdown/parquet_table/2.parquet]]}, projection=[a], file_type=parquet, predicate=b@1 = 2, pruning_predicate=b_null_count@2 != row_count@3 AND b_min@0 <= 2 AND 2 <= b_max@1, required_guarantees=[b in (2)]
+
Review Comment:
The [1st commit](https://github.com/apache/datafusion/pull/16686/commits/9ae839fddbb2d3d25e1e8ecb77506c42e7754d70) reproduce the issue, and the [2nd commit](https://github.com/apache/datafusion/pull/16686/commits/4415bc3cb92368dc7ace16a698b03890750f89c1) reflect the fix.
Basically, the SortExec which matches the filter equivalence gets removed after the fix.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add the missing equivalence info for filter pushdown [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on code in PR #16686:
URL: https://github.com/apache/datafusion/pull/16686#discussion_r2187232804
##########
datafusion/datasource/src/source.rs:
##########
@@ -325,6 +328,9 @@ impl ExecutionPlan for DataSourceExec {
new_node.data_source = data_source;
new_node.cache =
Self::compute_properties(Arc::clone(&new_node.data_source));
+ // Add the missing filters' equivalence info when filters pushdown is applied
+ let filter = conjunction(res.filters.collect_supported());
+ new_node = new_node.add_filter_equivalence_info(filter)?;
Review Comment:
What happens if the node already had equivalence info / old filters?
Practically speaking this shouldn't happen, but it is technically possible for a data source to have some filters passed in directly (e.g. in a TableProvider) and some filters passed in during pushdown.
I think this is correct because it's adding the new filters and when we built the new node we did a conjunction.
##########
datafusion/datasource/src/source.rs:
##########
@@ -372,6 +378,20 @@ impl DataSourceExec {
self
}
+ /// Add filters' equivalence info
+ fn add_filter_equivalence_info(
+ mut self,
+ filter: Arc<dyn PhysicalExpr>,
+ ) -> Result<Self> {
+ let (equal_pairs, _) = collect_columns_from_predicate(&filter);
+ for (lhs, rhs) in equal_pairs {
+ self.cache
+ .eq_properties
+ .add_equal_conditions(Arc::clone(lhs), Arc::clone(rhs))?
+ }
+ Ok(self)
+ }
Review Comment:
This part I'm not familiar with and would appreciate if someone else took a look at and confirm is correct.
##########
datafusion/datasource/src/source.rs:
##########
@@ -325,6 +328,9 @@ impl ExecutionPlan for DataSourceExec {
new_node.data_source = data_source;
new_node.cache =
Self::compute_properties(Arc::clone(&new_node.data_source));
+ // Add the missing filters' equivalence info when filters pushdown is applied
Review Comment:
```suggestion
// Recompute equivalence info using new filters
```
##########
datafusion/sqllogictest/test_files/parquet_filter_pushdown.slt:
##########
@@ -239,6 +349,23 @@ physical_plan
05)--------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=2
06)----------DataSourceExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_filter_pushdown/parquet_table/1.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_filter_pushdown/parquet_table/2.parquet]]}, projection=[a, b], file_type=parquet, predicate=b@1 > 2, pruning_predicate=b_null_count@1 != row_count@2 AND b_max@0 > 2, required_guarantees=[]
+query T
+select a from t_pushdown where b = 2 ORDER BY b;
+----
+bar
+
+query TT
+EXPLAIN select a from t_pushdown where b = 2 ORDER BY b;
+----
+logical_plan
+01)Projection: t_pushdown.a
+02)--Sort: t_pushdown.b ASC NULLS LAST
+03)----Filter: t_pushdown.b = Int32(2)
+04)------TableScan: t_pushdown projection=[a, b], partial_filters=[t_pushdown.b = Int32(2)]
+physical_plan
+01)CoalescePartitionsExec
+02)--DataSourceExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_filter_pushdown/parquet_table/1.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_filter_pushdown/parquet_table/2.parquet]]}, projection=[a], file_type=parquet, predicate=b@1 = 2, pruning_predicate=b_null_count@2 != row_count@3 AND b_min@0 <= 2 AND 2 <= b_max@1, required_guarantees=[b in (2)]
+
Review Comment:
@liamzwbao could you help me understand what I should be looking for in this output / what proves that this fix works / what would have been different before this fix?
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add the missing equivalence info for filter pushdown [datafusion]
Posted by "xudong963 (via GitHub)" <gi...@apache.org>.
xudong963 commented on code in PR #16686:
URL: https://github.com/apache/datafusion/pull/16686#discussion_r2188374284
##########
datafusion/datasource/src/source.rs:
##########
@@ -372,6 +378,20 @@ impl DataSourceExec {
self
}
+ /// Add filters' equivalence info
+ fn add_filter_equivalence_info(
+ mut self,
+ filter: Arc<dyn PhysicalExpr>,
+ ) -> Result<Self> {
+ let (equal_pairs, _) = collect_columns_from_predicate(&filter);
+ for (lhs, rhs) in equal_pairs {
+ self.cache
+ .eq_properties
+ .add_equal_conditions(Arc::clone(lhs), Arc::clone(rhs))?
+ }
+ Ok(self)
+ }
Review Comment:
Basically, it's enough, lgtm
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add the missing equivalence info for filter pushdown [datafusion]
Posted by "liamzwbao (via GitHub)" <gi...@apache.org>.
liamzwbao commented on code in PR #16686:
URL: https://github.com/apache/datafusion/pull/16686#discussion_r2188415579
##########
datafusion/datasource/src/source.rs:
##########
@@ -325,6 +328,9 @@ impl ExecutionPlan for DataSourceExec {
new_node.data_source = data_source;
new_node.cache =
Self::compute_properties(Arc::clone(&new_node.data_source));
+ // Add the missing filters' equivalence info when filters pushdown is applied
+ let filter = conjunction(res.filters.collect_supported());
+ new_node = new_node.add_filter_equivalence_info(filter)?;
Review Comment:
I think this will add the eq info from filter pushdown into the existing eq info group from `DataSource`.
Maybe it's better if we have a test case for it, but I'm not sure how to create a `DataSource` with eq info here. Are there any examples we can refer to?
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add the missing equivalence info for filter pushdown [datafusion]
Posted by "xudong963 (via GitHub)" <gi...@apache.org>.
xudong963 commented on PR #16686:
URL: https://github.com/apache/datafusion/pull/16686#issuecomment-3041900763
I'll verify the PR in our fork tomorrow, thanks @liamzwbao
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org