jayzhan211 commented on code in PR #26660:
URL: https://github.com/apache/datafusion/pull/26660#discussion_r4429922276
##########
datafusion/physical-optimizer/src/ensure_requirements/enforce_sorting/sort_pushdown.rs:
##########
@@ -549,7 +522,10 @@ fn pushdown_requirement_to_children(
Ok(None)
}
} else if plan.fetch().is_some()
- && plan.supports_limit_pushdown()
+ && !matches!(
+ plan.limit_behavior(),
+ LimitBehavior::Stops | LimitBehavior::Window(_)
Review Comment:
The sort pushdown match went away in the rebase onto #26118. It now only
asks whether a fetch can move through a node, and any other variant gets "no"
there. The matches that remain (`step` in `limit_pushdown.rs`,
`LimitPushPastWindows`, and `BufferExec::limit_behavior`) list every variant,
so a new variant fails to compile in each of them. I'd rather keep that than
add a method. OK to resolve?
##########
datafusion/physical-optimizer/src/limit_pushdown.rs:
##########
@@ -129,29 +106,225 @@ impl PhysicalOptimizerRule for LimitPushdown {
}
}
-struct LimitInfo {
- input: Arc<dyn ExecutionPlan>,
+/// The row bound a node receives from its ancestors.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum Above {
+ /// No bound.
+ Nothing,
+ /// `OFFSET skip LIMIT fetch` that no ancestor enforces yet: it must be
+ /// placed at or above the first node it can't pass.
+ Open(RowBound),
+ /// A bound an ancestor already enforces, with its `skip` applied. A node
+ /// below may stop after `fetch` rows; nothing else is required of it. A
+ /// limit node further down is still folded in and placed again.
+ Enforced(Option<usize>),
+}
+
+impl Above {
+ fn open(bound: RowBound) -> Self {
+ if bound == RowBound::default() {
+ Above::Nothing
+ } else {
+ Above::Open(bound)
+ }
+ }
+
+ fn bound(self) -> RowBound {
+ match self {
+ Above::Nothing => RowBound::default(),
+ Above::Open(bound) => bound,
+ Above::Enforced(fetch) => RowBound::fetch(fetch),
+ }
+ }
+
+ /// `self` above a node that enforces its own `fetch` (a TopK, a scan with
+ /// a row limit, ...). The node enforces the combined bound unless rows
+ /// still have to be skipped above it.
+ fn above_fetching_node(self, fetch: usize) -> Self {
+ let bound = self.bound().over(RowBound::fetch(Some(fetch)));
+ if bound.skip > 0 {
+ Above::Open(bound)
+ } else {
+ Above::Enforced(bound.fetch)
+ }
+ }
+}
+
+/// What a node does with a bound from its parent.
+enum Step {
+ /// A limit node: its bound is folded in and the node removed.
+ Limit(RowBound),
+ /// Passes the bound to every child.
+ Pass,
+ /// Combines partitions, so it sees every row and can enforce the whole
+ /// bound itself.
+ Merge,
+ /// The bound can't go below this node.
+ Stop,
+}
+
+fn step(plan: &Arc<dyn ExecutionPlan>) -> Step {
+ match plan.limit_behavior() {
+ LimitBehavior::Limit { skip, fetch } => Step::Limit(RowBound { skip,
fetch }),
+ LimitBehavior::CopyToChildren | LimitBehavior::MoveToInput =>
Step::Pass,
Review Comment:
You're right, it moved the bound through any `CopyToChildren` node.
`UnionExec` still passes it: each of its output partitions is one input
partition, and below a node with several partitions the bound is per partition
(`LocalLimitExec`), so the same rows come out. Copying instead would leave a
redundant `LocalLimitExec` above the union (one plan in `union.slt` changes
that way). A `CopyToChildren` or `MoveToInput` node whose children have more
partitions in total than it outputs now takes the bound like a merge: the bound
stays above the node, and each child gets it as an early-stop fetch.
`copy_node_with_several_children_and_one_output_partition_keeps_limit_above`
covers your case. Main had the same problem: it passed a limit through any
operator returning `true` from `supports_limit_pushdown` other than
`CoalescePartitionsExec` and `SortPreservingMergeExec`. Does this cover what
you had in mind?
--
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]
