neilconway commented on code in PR #23800:
URL: https://github.com/apache/datafusion/pull/23800#discussion_r3778262856
##########
datafusion/physical-optimizer/src/limit_pushdown.rs:
##########
@@ -148,58 +152,117 @@ pub fn pushdown_limit_helper(
mut pushdown_plan: Arc<dyn ExecutionPlan>,
mut global_state: GlobalRequirements,
) -> Result<(Transformed<Arc<dyn ExecutionPlan>>, GlobalRequirements)> {
- // Extract limit, if exist, and return child inputs.
- if let Some(limit_info) = extract_limit(&pushdown_plan) {
- // If we have fetch/skip info in the global state already, we need to
- // decide which one to continue with:
- let (skip, fetch) = combine_limit(
+ if global_state.pending == Some(PendingScope::Local)
+ && pushdown_plan.output_partitioning().partition_count() == 1
+ {
+ // Local and global scope are equivalent with one output partition, but
+ // retain the global scope in case recursion later exposes
multi-partition
+ // children, including through extension combiners.
+ global_state.pending = Some(PendingScope::Global);
+ }
+
+ if let Some(global_limit) = pushdown_plan.downcast_ref::<GlobalLimitExec>()
+ && global_limit.skip() == 0
+ && global_limit.fetch().is_none()
+ {
+ // Remove this no-op wrapper without clearing inherited state, which
may
+ // have been promoted from local to global scope at a one-output
boundary.
Review Comment:
I found this comment confusing. What about
```rust
// A GlobalLimitExec with no skip and no fetch enforces nothing, so
// remove it and keep the carried state untouched. Falling through to
// the general limit handling would instead re-derive `pending` from
// this node, which could reopen an already-satisfied requirement.
// The local-to-global promotion above is kept as well: it relied
only
// on this node having one output partition, and it remains correct
// after the node is removed.
```
##########
datafusion/physical-optimizer/src/limit_pushdown.rs:
##########
@@ -83,20 +83,31 @@ use datafusion_physical_plan::{ExecutionPlan,
ExecutionPlanProperties};
#[derive(Default, Debug)]
pub struct LimitPushdown {}
-/// This is a "data class" we use within the [`LimitPushdown`] rule to push
-/// down limits in the plan. GlobalRequirements are hold as a rule-wide state
-/// and holds the fetch and skip information. The struct also has a field named
-/// satisfied which means if the "current" plan is valid in terms of limits or
not.
+/// State carried through [`LimitPushdown`] while it pushes limits down the
plan.
///
-/// For example: If the plan is satisfied with current fetch info, we decide
to not add a LocalLimit
+/// `pending` keeps the semantic requirement's scope separate from its numeric
+/// payload. `Some(PendingScope::Local)` means a per-output-partition cap is
+/// still owed, and `Some(PendingScope::Global)` means a subtree-wide cap is
+/// still owed. When `pending` is `None`, no semantic enforcement remains
+/// outstanding; a retained `fetch` is a descendant early-stop hint only.
///
/// [`LimitPushdown`]: crate::limit_pushdown::LimitPushdown
-#[derive(Default, Clone, Debug)]
+#[derive(Clone, Debug)]
Review Comment:
Why remove `Default`?
##########
datafusion/physical-optimizer/src/limit_pushdown.rs:
##########
@@ -148,58 +152,117 @@ pub fn pushdown_limit_helper(
mut pushdown_plan: Arc<dyn ExecutionPlan>,
mut global_state: GlobalRequirements,
) -> Result<(Transformed<Arc<dyn ExecutionPlan>>, GlobalRequirements)> {
- // Extract limit, if exist, and return child inputs.
- if let Some(limit_info) = extract_limit(&pushdown_plan) {
- // If we have fetch/skip info in the global state already, we need to
- // decide which one to continue with:
- let (skip, fetch) = combine_limit(
+ if global_state.pending == Some(PendingScope::Local)
+ && pushdown_plan.output_partitioning().partition_count() == 1
+ {
+ // Local and global scope are equivalent with one output partition, but
+ // retain the global scope in case recursion later exposes
multi-partition
+ // children, including through extension combiners.
+ global_state.pending = Some(PendingScope::Global);
+ }
+
+ if let Some(global_limit) = pushdown_plan.downcast_ref::<GlobalLimitExec>()
+ && global_limit.skip() == 0
+ && global_limit.fetch().is_none()
+ {
+ // Remove this no-op wrapper without clearing inherited state, which
may
+ // have been promoted from local to global scope at a one-output
boundary.
+ return Ok((
+ Transformed {
+ data: Arc::clone(global_limit.input()),
+ transformed: true,
+ tnr: TreeNodeRecursion::Stop,
+ },
+ global_state,
+ ));
+ }
+
+ if global_state.pending == Some(PendingScope::Global)
+ && pushdown_plan.output_partitioning().partition_count() > 1
+ {
+ // This must precede generic `plan.fetch()` handling: a fetch on
multiple
+ // outputs cannot by itself prove a global cap, because the
`ExecutionPlan`
+ // trait does not formally encode scope. Retain it only as a
per-partition
+ // hint; an existing smaller hint does not imply a smaller global cap.
+ let hint = global_state.fetch.map(|fetch| fetch + global_state.skip);
+ if let Some(hint) = hint {
+ let hint = pushdown_plan.fetch().map_or(hint, |fetch|
fetch.min(hint));
+ if pushdown_plan.fetch() != Some(hint)
+ && let Some(plan_with_fetch) =
pushdown_plan.with_fetch(Some(hint))
+ {
+ pushdown_plan = plan_with_fetch;
+ }
+ }
+
+ let plan = materialize_global_requirement(
+ pushdown_plan,
global_state.skip,
global_state.fetch,
- limit_info.skip,
- limit_info.fetch,
+ global_state.preserve_order,
);
- global_state.skip = skip;
- global_state.fetch = fetch;
- global_state.preserve_order = limit_info.preserve_order;
- global_state.satisfied = false;
+ global_state.fetch = hint;
+ global_state.skip = 0;
+ global_state.pending = None;
+ return Ok((Transformed::yes(plan), global_state));
+ }
- if let Some(fetch) = fetch
- && limit_satisfied_by_input(&limit_info.input, skip, fetch)?
+ if let Some(global_limit) =
pushdown_plan.downcast_ref::<GlobalLimitExec>() {
+ let input = Arc::clone(global_limit.input());
+ let skip = global_limit.skip();
+ let fetch = global_limit.fetch();
+
+ (global_state.skip, global_state.fetch) =
+ combine_limit(global_state.skip, global_state.fetch, skip, fetch);
+ global_state.preserve_order |=
global_limit.required_ordering().is_some();
+ global_state.pending = Some(PendingScope::Global);
+ if let Some(fetch) = global_state.fetch
+ && limit_satisfied_by_input(&input, global_state.skip, fetch)?
{
- // The input already produces at most `fetch` rows, so no new limit
- // node is needed. Mark satisfied so downstream won't re-add one,
- // but preserve skip/fetch so any nested limit nodes (e.g. an inner
- // GlobalLimitExec) can still be merged with the outer constraint.
- global_state.satisfied = true;
-
- return Ok((
- Transformed {
- data: limit_info.input,
- transformed: true,
- tnr: TreeNodeRecursion::Stop,
- },
- global_state,
- ));
+ global_state.pending = None;
}
+ return Ok((
+ Transformed {
+ data: input,
+ transformed: true,
+ tnr: TreeNodeRecursion::Stop,
+ },
+ global_state,
+ ));
+ }
- // Now the global state has the most recent information, we can remove
- // the limit node. We will decide later if we should add it again or
- // not.
+ if let Some(local_limit) = pushdown_plan.downcast_ref::<LocalLimitExec>() {
+ let input = Arc::clone(local_limit.input());
+ (global_state.skip, global_state.fetch) = combine_limit(
+ global_state.skip,
+ global_state.fetch,
+ 0,
+ Some(local_limit.fetch()),
+ );
+ global_state.preserve_order |=
local_limit.required_ordering().is_some();
+ global_state.pending = if
input.output_partitioning().partition_count() == 1 {
+ Some(PendingScope::Global)
+ } else {
+ Some(PendingScope::Local)
+ };
+ if let Some(fetch) = global_state.fetch
+ && limit_satisfied_by_input(&input, global_state.skip, fetch)?
+ {
+ global_state.pending = None;
+ }
return Ok((
Transformed {
- data: limit_info.input,
+ data: input,
transformed: true,
tnr: TreeNodeRecursion::Stop,
},
global_state,
));
}
- // If we have a non-limit operator with fetch capability, update global
- // state as necessary:
+ // Merge a fetch already present on a non-limit operator into global state.
if pushdown_plan.fetch().is_some() {
if global_state.skip == 0 {
- global_state.satisfied = true;
+ global_state.pending = None;
}
(global_state.skip, global_state.fetch) = combine_limit(
global_state.skip,
Review Comment:
Why if we owe `fetch=5` but the fetch on the operator is `> 5`?
We rebuild the operator below via `pushdown_plan.with_fetch(skip_and_fetch)`
below, but some operators might support `fetch` but not `with_fetch` (e.g.,
`PartialSort`).
##########
datafusion/physical-optimizer/src/limit_pushdown.rs:
##########
Review Comment:
Is copying the global state into _every_ child correct, for an operator with
multiple children? e.g., if we have an operator with `n` children and a single
output partition, wouldn't this result in copying the fetch limit into `n`
different operators, so we'd produce too many rows?
##########
datafusion/physical-optimizer/src/limit_pushdown.rs:
##########
@@ -83,20 +83,31 @@ use datafusion_physical_plan::{ExecutionPlan,
ExecutionPlanProperties};
#[derive(Default, Debug)]
pub struct LimitPushdown {}
-/// This is a "data class" we use within the [`LimitPushdown`] rule to push
-/// down limits in the plan. GlobalRequirements are hold as a rule-wide state
-/// and holds the fetch and skip information. The struct also has a field named
-/// satisfied which means if the "current" plan is valid in terms of limits or
not.
+/// State carried through [`LimitPushdown`] while it pushes limits down the
plan.
///
-/// For example: If the plan is satisfied with current fetch info, we decide
to not add a LocalLimit
+/// `pending` keeps the semantic requirement's scope separate from its numeric
+/// payload. `Some(PendingScope::Local)` means a per-output-partition cap is
+/// still owed, and `Some(PendingScope::Global)` means a subtree-wide cap is
+/// still owed. When `pending` is `None`, no semantic enforcement remains
+/// outstanding; a retained `fetch` is a descendant early-stop hint only.
///
/// [`LimitPushdown`]: crate::limit_pushdown::LimitPushdown
-#[derive(Default, Clone, Debug)]
+#[derive(Clone, Debug)]
pub struct GlobalRequirements {
fetch: Option<usize>,
skip: usize,
- satisfied: bool,
preserve_order: bool,
+ pending: Option<PendingScope>,
+}
+
+/// Scope of a semantic cap that remains pending independently of its numeric
+/// `skip` and `fetch` payload.
Review Comment:
I found this comment confusing (e.g., whose `skip` and `fetch` are we
referring to? What is a "semantic cap", and is a "non-semantic cap" a thing?).
Calling it `PendingScope` only makes sense if you know the field in
`GlobalRequirements` is named `pending`, which isn't obvious.
What about:
```rust
/// The scope of a row limit: what the limited row count applies to.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum LimitScope {
/// The limit caps each output partition independently, as enforced by
/// [`LocalLimitExec`].
Local,
/// The limit caps the combined output of all partitions, as enforced by
/// [`GlobalLimitExec`] over a single-partition input. A fetch on a
/// multi-partition operator cannot satisfy this scope; the partitions
/// must first be merged into one stream.
Global,
}
```
--
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]
