jayzhan211 commented on code in PR #27079:
URL: https://github.com/apache/datafusion/pull/27079#discussion_r4512760918
##########
datafusion/datasource/src/file_scan_config/mod.rs:
##########
@@ -1074,6 +1077,15 @@ impl DataSource for FileScanConfig {
filters: Vec<Arc<dyn PhysicalExpr>>,
config: &ConfigOptions,
) -> Result<FilterPushdownPropagation<Arc<dyn DataSource>>> {
+ // The scan applies its limit after any filter it holds, so a filter
+ // from above the limit would let through rows the limit should have
+ // cut.
+ if self.limit.is_some() {
Review Comment:
@zhuqi-lucas thanks, everything from the last round checks out here. One
change, and it reverses my own suggestion, sorry: the two
`self.limit.is_some()` checks make the scan the third node with its own "a
fetch blocks pushdown" guard (after #27060 and #23599), while one
`node.fetch()` check in the `PushdownSort` and `FilterPushdown` rules covers
every node. With the diff below in place of the two checks, the full
sqllogictest suite passes, tests 11–13 included, and two `filter_pushdown.rs`
snapshots change, correctly, since they merged an outer filter into a
`FilterExec` with `fetch=5`.
This also fixes #27194 at default config, so could you add `Closes #27194`
and its three queries, with the option off, to the slt?
<details>
<summary>Rule-level check (replaces both scan checks and
`sort_pushdown_with_limit_unsupported`; clippy clean)</summary>
```diff
--- a/datafusion/physical-optimizer/src/filter_pushdown.rs
+++ b/datafusion/physical-optimizer/src/filter_pushdown.rs
@@ fn push_down_filters(
phase: FilterPushdownPhase,
) -> Result<FilterPushdownPropagation<Arc<dyn ExecutionPlan>>> {
+ // A node with a fetch picks its rows before a filter from above
applies, so
+ // that filter can't move below or into it. The node's own filters
still can.
+ if node.fetch().is_some() && !parent_predicates.is_empty() {
+ let result = push_down_filters(node, vec![], config, phase)?;
+ return Ok(FilterPushdownPropagation {
+ filters: vec![PushedDown::No; parent_predicates.len()],
+ updated_node: result.updated_node,
+ });
+ }
+
--- a/datafusion/physical-optimizer/src/pushdown_sort.rs
+++ b/datafusion/physical-optimizer/src/pushdown_sort.rs
+/// Whether a node below this sort has a fetch. A fetch keeps the first
rows in the
+/// input's current order, so changing that order would change which rows
it keeps.
+fn input_has_fetch(input: &Arc<dyn ExecutionPlan>) -> Result<bool> {
+ input.exists(|node| Ok(node.fetch().is_some()))
+}
@@ (SortPreservingMergeExec arm)
let sort_input = Arc::clone(sort_child.input());
+ if input_has_fetch(&sort_input)? {
+ return Ok(Transformed::no(plan));
+ }
@@ (SortExec arm)
let sort_input = Arc::clone(sort_exec.input());
+ if input_has_fetch(&sort_input)? {
+ return Ok(Transformed::no(plan));
+ }
```
New snapshot for `test_filter_collapse_inner_fetch_preserved` (the
`both_fetch` one changes the same way):
```
- FilterExec: b@1 = bar
- FilterExec: a@0 = foo, fetch=5
- DataSourceExec: ...
```
</details>
--
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]
