From 43ccc6bf38512f381d7b0090e060e834bcf729c0 Mon Sep 17 00:00:00 2001 From: Jay Zhan Date: Fri, 23 Oct 2026 19:31:50 +0800 Subject: [PATCH 1/3] refactor: add RowBound for skip/fetch arithmetic in the limit-moving rules --- .../enforce_sorting/mod.rs | 17 +-- .../enforce_sorting/sort_pushdown.rs | 29 ++-- datafusion/physical-optimizer/src/lib.rs | 1 + .../src/limit_pushdown_past_window.rs | 14 +- .../physical-optimizer/src/row_bound.rs | 143 ++++++++++++++++++ 5 files changed, 166 insertions(+), 38 deletions(-) create mode 100644 datafusion/physical-optimizer/src/row_bound.rs diff --git a/datafusion/physical-optimizer/src/ensure_requirements/enforce_sorting/mod.rs b/datafusion/physical-optimizer/src/ensure_requirements/enforce_sorting/mod.rs index 8a4905365b7a..222cd1ae1219 100644 --- a/datafusion/physical-optimizer/src/ensure_requirements/enforce_sorting/mod.rs +++ b/datafusion/physical-optimizer/src/ensure_requirements/enforce_sorting/mod.rs @@ -49,6 +49,7 @@ pub mod sort_pushdown; use std::sync::Arc; use crate::output_requirements::OutputRequirementExec; +use crate::row_bound::{RowBound, limit_node}; use crate::utils::{ add_sort_above, add_sort_above_with_check, is_coalesce_partitions, is_limit, is_repartition, is_sort, is_sort_preserving_merge, is_window, @@ -60,7 +61,7 @@ use datafusion_common::tree_node::Transformed; use datafusion_physical_expr::{Distribution, Partitioning}; use datafusion_physical_expr_common::sort_expr::{LexOrdering, LexRequirement}; use datafusion_physical_plan::coalesce_partitions::CoalescePartitionsExec; -use datafusion_physical_plan::limit::{GlobalLimitExec, LocalLimitExec}; +use datafusion_physical_plan::limit::LocalLimitExec; use datafusion_physical_plan::repartition::RepartitionExec; use datafusion_physical_plan::sorts::partial_sort::PartialSortExec; use datafusion_physical_plan::sorts::sort::SortExec; @@ -512,17 +513,13 @@ fn analyze_immediate_sort_removal( .with_fetch(sort_exec.fetch()), ) as _ } else { - // Remove the sort: + // Remove the sort, keeping its fetch as a limit: node.children = node.children.swap_remove(0).children; - if let Some(fetch) = sort_exec.fetch() { - // If the sort has a fetch, we need to add a limit: - if properties.output_partitioning().partition_count() == 1 { - Arc::new(GlobalLimitExec::new(Arc::clone(sort_input), 0, Some(fetch))) - } else { - Arc::new(LocalLimitExec::new(Arc::clone(sort_input), fetch)) + match sort_exec.fetch() { + Some(fetch) => { + limit_node(Arc::clone(sort_input), RowBound::fetch(Some(fetch))) } - } else { - Arc::clone(sort_input) + None => Arc::clone(sort_input), } }; for child in node.children.iter_mut() { diff --git a/datafusion/physical-optimizer/src/ensure_requirements/enforce_sorting/sort_pushdown.rs b/datafusion/physical-optimizer/src/ensure_requirements/enforce_sorting/sort_pushdown.rs index 2aeddd23aaff..c39687d7c46d 100644 --- a/datafusion/physical-optimizer/src/ensure_requirements/enforce_sorting/sort_pushdown.rs +++ b/datafusion/physical-optimizer/src/ensure_requirements/enforce_sorting/sort_pushdown.rs @@ -51,6 +51,8 @@ use datafusion_physical_plan::sorts::sort_preserving_merge::SortPreservingMergeE use datafusion_physical_plan::tree_node::PlanContext; use datafusion_physical_plan::{ExecutionPlan, ExecutionPlanProperties}; +use crate::row_bound::{RowBound, limit_node}; + /// "Data class" used by sort pushdown (now driven from `EnsureRequirements`) /// to push down [`SortExec`] in the plan. In some cases the total /// computational cost is reduced by pushing down `SortExec`s through certain @@ -94,21 +96,19 @@ pub type SortPushDown = PlanContext; /// fetch is the tighter bound. Using the bare `fetch` would be wrong too, as /// it would turn `LIMIT 10 OFFSET 5` into `TopK(10)` below the limit, i.e. 5 /// result rows. -/// -/// `skip` and `fetch` are independent `usize`s, so their sum can overflow. Like -/// [`combine_limit`], we saturate: `usize::MAX` input rows is never a smaller -/// bound than the real one, so the pushed-down fetch stays correct. -/// -/// [`combine_limit`]: datafusion_common::utils::combine_limit fn input_fetch( plan: &Arc, parent_fetch: Option, ) -> Option { - let fetch = min_fetch(plan.fetch(), parent_fetch)?; let skip = plan .downcast_ref::() .map_or(0, |limit| limit.skip()); - Some(fetch.saturating_add(skip)) + let output = RowBound::fetch(plan.fetch()).min(RowBound::fetch(parent_fetch)); + RowBound { + skip, + fetch: output.fetch, + } + .input_rows() } /// Assigns the ordering requirement of the root node to the its children. @@ -140,12 +140,7 @@ pub fn pushdown_sorts(sort_push_down: SortPushDown) -> Result { } fn min_fetch(f1: Option, f2: Option) -> Option { - match (f1, f2) { - (Some(f1), Some(f2)) => Some(f1.min(f2)), - (Some(_), _) => f1, - (_, Some(_)) => f2, - _ => None, - } + RowBound::fetch(f1).min(RowBound::fetch(f2)).fetch } /// Returns whether a fetch on `plan` can also be applied to each child. @@ -188,11 +183,7 @@ fn preserve_fetch_above_ordered_plan( node.plan.output_partitioning().partition_count() > 1; let input = Arc::clone(&node.plan); - let limit: Arc = if input_has_multiple_partitions { - Arc::new(LocalLimitExec::new(input, fetch)) - } else { - Arc::new(GlobalLimitExec::new(input, 0, Some(fetch))) - }; + let limit = limit_node(input, RowBound::fetch(Some(fetch))); let limit_node = PlanContext::new(limit, ParentRequirements::default(), vec![node]); if input_has_multiple_partitions diff --git a/datafusion/physical-optimizer/src/lib.rs b/datafusion/physical-optimizer/src/lib.rs index 25c945291549..dc9d5d92ebb3 100644 --- a/datafusion/physical-optimizer/src/lib.rs +++ b/datafusion/physical-optimizer/src/lib.rs @@ -45,6 +45,7 @@ pub mod projection_pushdown; pub use datafusion_pruning as pruning; pub mod hash_join_buffering; pub mod pushdown_sort; +mod row_bound; pub mod sanity_checker; pub mod topk_aggregation; pub mod topk_repartition; diff --git a/datafusion/physical-optimizer/src/limit_pushdown_past_window.rs b/datafusion/physical-optimizer/src/limit_pushdown_past_window.rs index 092570b05197..52b5d6e596d7 100644 --- a/datafusion/physical-optimizer/src/limit_pushdown_past_window.rs +++ b/datafusion/physical-optimizer/src/limit_pushdown_past_window.rs @@ -16,6 +16,7 @@ // under the License. use crate::PhysicalOptimizerRule; +use crate::row_bound::RowBound; use datafusion_common::ScalarValue; use datafusion_common::config::ConfigOptions; use datafusion_common::tree_node::{Transformed, TreeNode}; @@ -25,13 +26,11 @@ use datafusion_physical_expr::window::{ StandardWindowFunctionExpr, WindowExpr, }; use datafusion_physical_plan::execution_plan::CardinalityEffect; -use datafusion_physical_plan::limit::{GlobalLimitExec, LocalLimitExec}; use datafusion_physical_plan::repartition::RepartitionExec; use datafusion_physical_plan::sorts::sort::SortExec; use datafusion_physical_plan::sorts::sort_preserving_merge::SortPreservingMergeExec; use datafusion_physical_plan::windows::{BoundedWindowAggExec, WindowUDFExpr}; use datafusion_physical_plan::{ExecutionPlan, ExecutionPlanProperties}; -use std::cmp; use std::sync::Arc; /// This rule inspects [`ExecutionPlan`]'s attempting to find fetch limits that were not pushed @@ -196,20 +195,16 @@ fn apply_limit( }; let fetch = match node.fetch() { None => fetch + ctx.lookahead, - Some(existing) => cmp::min(existing, fetch + ctx.lookahead), + Some(existing) => existing.min(fetch + ctx.lookahead), }; Some(Transformed::complete(node.with_fetch(Some(fetch)).unwrap())) } fn get_limit(node: &Arc, ctx: &mut TraverseState) -> bool { - if let Some(limit) = node.downcast_ref::() { - ctx.reset_limit(limit.fetch().map(|fetch| fetch + limit.skip())); - return true; - } // In distributed execution, GlobalLimitExec becomes LocalLimitExec // per partition. Handle it the same way (LocalLimitExec has no skip). - if let Some(limit) = node.downcast_ref::() { - ctx.reset_limit(Some(limit.fetch())); + if let Some(limit) = RowBound::of_limit_node(node) { + ctx.reset_limit(limit.fetch.map(|fetch| fetch + limit.skip)); return true; } if let Some(limit) = node.downcast_ref::() { @@ -271,6 +266,7 @@ mod tests { use datafusion_physical_expr_common::sort_expr::{LexOrdering, PhysicalSortExpr}; use datafusion_physical_plan::InputOrderMode; use datafusion_physical_plan::displayable; + use datafusion_physical_plan::limit::{GlobalLimitExec, LocalLimitExec}; use datafusion_physical_plan::placeholder_row::PlaceholderRowExec; use datafusion_physical_plan::windows::{ BoundedWindowAggExec, create_udwf_window_expr, diff --git a/datafusion/physical-optimizer/src/row_bound.rs b/datafusion/physical-optimizer/src/row_bound.rs new file mode 100644 index 000000000000..72d930ac4960 --- /dev/null +++ b/datafusion/physical-optimizer/src/row_bound.rs @@ -0,0 +1,143 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! [`RowBound`]: the `OFFSET skip LIMIT fetch` that an optimizer rule wants on +//! an operator's output. +//! +//! A bound is written into a plan in one of two forms: a limit node +//! ([`GlobalLimitExec`], [`LocalLimitExec`]) or an operator's own fetch +//! ([`ExecutionPlan::with_fetch`]). Operators can't skip rows, so a bound with +//! `skip > 0` always keeps a [`GlobalLimitExec`]. + +use std::sync::Arc; + +use datafusion_physical_plan::limit::{GlobalLimitExec, LocalLimitExec}; +use datafusion_physical_plan::{ExecutionPlan, ExecutionPlanProperties}; + +/// `OFFSET skip LIMIT fetch` on the output of one operator. +/// +/// `fetch == None` means no row limit (only `skip`). The default bound, +/// `skip = 0, fetch = None`, keeps every row. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub(crate) struct RowBound { + pub skip: usize, + pub fetch: Option, +} + +impl RowBound { + /// `LIMIT fetch` with no offset. + pub fn fetch(fetch: Option) -> Self { + Self { skip: 0, fetch } + } + + /// The bound a limit node imposes, or `None` if `plan` is not a limit node. + pub fn of_limit_node(plan: &Arc) -> Option { + if let Some(limit) = plan.downcast_ref::() { + Some(Self { + skip: limit.skip(), + fetch: limit.fetch(), + }) + } else { + plan.downcast_ref::() + .map(|limit| Self::fetch(Some(limit.fetch()))) + } + } + + /// The tighter of two bounds on the same output. Both must have `skip == 0`. + pub fn min(self, other: RowBound) -> RowBound { + debug_assert!(self.skip == 0 && other.skip == 0); + let fetch = match (self.fetch, other.fetch) { + (Some(a), Some(b)) => Some(a.min(b)), + (a, b) => a.or(b), + }; + RowBound::fetch(fetch) + } + + /// Rows the input must produce so this bound can be applied on top of it: + /// `skip + fetch`. `None` when there is no row limit. + /// + /// Saturates: `usize::MAX` input rows is never a smaller bound than the + /// real one, so a fetch computed from it stays correct. + pub fn input_rows(self) -> Option { + self.fetch.map(|fetch| fetch.saturating_add(self.skip)) + } +} + +/// A limit node applying `bound` to `plan`: a [`GlobalLimitExec`] when it +/// skips rows or `plan` has one output partition, otherwise a per-partition +/// [`LocalLimitExec`]. +pub(crate) fn limit_node( + plan: Arc, + bound: RowBound, +) -> Arc { + match bound.input_rows() { + Some(fetch) + if bound.skip == 0 && plan.output_partitioning().partition_count() > 1 => + { + Arc::new(LocalLimitExec::new(plan, fetch)) + } + _ => Arc::new(GlobalLimitExec::new(plan, bound.skip, bound.fetch)), + } +} + +#[cfg(test)] +mod tests { + use super::*; + use arrow::datatypes::{DataType, Field, Schema}; + use datafusion_physical_plan::empty::EmptyExec; + + fn bound(skip: usize, fetch: Option) -> RowBound { + RowBound { skip, fetch } + } + + #[test] + fn min_keeps_the_tighter_fetch() { + assert_eq!( + RowBound::fetch(Some(3)).min(RowBound::fetch(None)), + bound(0, Some(3)) + ); + assert_eq!( + RowBound::fetch(Some(3)).min(RowBound::fetch(Some(2))), + bound(0, Some(2)) + ); + } + + #[test] + fn input_rows_saturates() { + assert_eq!(bound(5, Some(10)).input_rows(), Some(15)); + assert_eq!(bound(5, None).input_rows(), None); + assert_eq!(bound(usize::MAX, Some(1)).input_rows(), Some(usize::MAX)); + } + + #[test] + fn limit_node_is_local_only_without_skip_over_many_partitions() { + let schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Int32, true)])); + let one: Arc = Arc::new(EmptyExec::new(Arc::clone(&schema))); + let many: Arc = + Arc::new(EmptyExec::new(schema).with_partitions(4)); + + let node = limit_node(Arc::clone(&many), bound(0, Some(3))); + assert_eq!(node.downcast_ref::().unwrap().fetch(), 3); + + let node = limit_node(Arc::clone(&many), bound(2, Some(3))); + let global = node.downcast_ref::().unwrap(); + assert_eq!((global.skip(), global.fetch()), (2, Some(3))); + + let node = limit_node(one, bound(0, Some(3))); + assert!(node.is::()); + } +} From 04c77e3767bf035965923ad2f7e70804f44bf448 Mon Sep 17 00:00:00 2001 From: Jay Zhan Date: Fri, 23 Oct 2026 19:33:14 +0800 Subject: [PATCH 2/3] refactor: write LimitPushdown as one walk over an explicit bound state --- .../physical-optimizer/src/limit_pushdown.rs | 521 ++++++++---------- .../physical-optimizer/src/row_bound.rs | 36 ++ .../library-user-guide/upgrading/56.0.0.md | 14 +- 3 files changed, 279 insertions(+), 292 deletions(-) diff --git a/datafusion/physical-optimizer/src/limit_pushdown.rs b/datafusion/physical-optimizer/src/limit_pushdown.rs index 55e2a71038bc..b944bf60b522 100644 --- a/datafusion/physical-optimizer/src/limit_pushdown.rs +++ b/datafusion/physical-optimizer/src/limit_pushdown.rs @@ -67,16 +67,15 @@ use crate::PhysicalOptimizerRule; use crate::optimizer::{ ConfigOnlyContext, PhysicalOptimizerContext, with_statistics_context, }; +use crate::row_bound::{RowBound, limit_node, place}; use datafusion_common::config::ConfigOptions; use datafusion_common::error::Result; use datafusion_common::stats::Precision; use datafusion_common::tree_node::{Transformed, TreeNodeRecursion}; -use datafusion_common::utils::combine_limit; use datafusion_physical_plan::coalesce_partitions::CoalescePartitionsExec; use datafusion_physical_plan::empty::EmptyExec; use datafusion_physical_plan::execution_plan::replace_children_if_necessary; -use datafusion_physical_plan::limit::{GlobalLimitExec, LocalLimitExec}; use datafusion_physical_plan::placeholder_row::PlaceholderRowExec; use datafusion_physical_plan::projection::ProjectionExec; use datafusion_physical_plan::sorts::sort_preserving_merge::SortPreservingMergeExec; @@ -87,21 +86,6 @@ 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. -/// -/// For example: If the plan is satisfied with current fetch info, we decide to not add a LocalLimit -/// -/// [`LimitPushdown`]: crate::limit_pushdown::LimitPushdown -#[derive(Default, Clone, Debug)] -pub struct GlobalRequirements { - fetch: Option, - skip: usize, - satisfied: bool, -} - impl LimitPushdown { #[expect(missing_docs)] pub fn new() -> Self { @@ -123,13 +107,8 @@ impl PhysicalOptimizerRule for LimitPushdown { plan: Arc, context: &dyn PhysicalOptimizerContext, ) -> Result> { - let global_state = GlobalRequirements { - fetch: None, - skip: 0, - satisfied: false, - }; with_statistics_context(context, |stats_ctx| { - pushdown_limits(plan, global_state, stats_ctx) + push_limits(plan, Above::Nothing, stats_ctx) }) } @@ -142,29 +121,228 @@ impl PhysicalOptimizerRule for LimitPushdown { } } -struct LimitInfo { - input: Arc, +/// 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), +} + +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) -> Step { + if let Some(bound) = RowBound::of_limit_node(plan) { + Step::Limit(bound) + } else if !plan.supports_limit_pushdown() { + Step::Stop + } else if plan.is::() || plan.is::() + { + Step::Merge + } else { + Step::Pass + } +} + +/// Pushes the bounds of all limit nodes in `plan` down as far as they pass, +/// and writes each one once, at the first node that can enforce it: +/// +/// - a merge, which takes it as its own fetch; +/// - otherwise the first node the bound can't pass, which takes it as its +/// fetch or gets a limit node on top. +/// +/// Below the place a bound is enforced, nodes that can stop early are given +/// it as a fetch too. Operators can't skip rows, so a bound with an `OFFSET` +/// always keeps a [`GlobalLimitExec`] where it is enforced. +/// +/// [`GlobalLimitExec`]: datafusion_physical_plan::limit::GlobalLimitExec +fn push_limits( + mut plan: Arc, + mut above: Above, + stats_ctx: &StatisticsContext, +) -> Result> { + while let Some((input, folded)) = fold_limit(&plan, above, stats_ctx)? { + (plan, above) = (input, folded); + } + let (plan, below) = apply_bound(plan, above); + + let children = plan.children(); + let mut changed = false; + let new_children = children + .into_iter() + .map(|child| { + let new_child = push_limits(Arc::clone(child), below, stats_ctx)?; + changed |= !Arc::ptr_eq(child, &new_child); + Ok(new_child) + }) + .collect::>()?; + if changed { + replace_children_if_necessary(plan, new_children) + } else { + Ok(plan) + } +} + +/// If `plan` is a limit node, returns its input and `above` combined with the +/// limit's bound. Otherwise returns `None`. +fn fold_limit( + plan: &Arc, + above: Above, + stats_ctx: &StatisticsContext, +) -> Result, Above)>> { + let Step::Limit(limit) = step(plan) else { + return Ok(None); + }; + let input = Arc::clone(plan.children()[0]); + let bound = above.bound().over(limit); + // Exact statistics may prove the input already has few enough rows. The + // bound is kept so a nested limit can still be combined with it. + let above = if limit_satisfied_by_input(&input, bound, stats_ctx)? { + Above::Enforced(bound.fetch) + } else { + Above::open(bound) + }; + Ok(Some((input, above))) +} + +/// Applies `above` to `plan`, which is not a limit node. Returns the new plan +/// and the bound its children receive. +fn apply_bound( + plan: Arc, + above: Above, +) -> (Arc, Above) { + let above = match plan.fetch() { + Some(fetch) => above.above_fetching_node(fetch), + None => above, + }; + match (step(&plan), above) { + (_, Above::Nothing | Above::Enforced(None)) => (plan, Above::Nothing), + // Only an OFFSET: it is placed here, operators can't skip. + (_, Above::Open(bound)) if bound.fetch.is_none() => { + (limit_node(plan, bound), Above::Nothing) + } + (Step::Limit(_), _) => unreachable!("limit nodes are folded before apply_bound"), + (Step::Pass, above) => (plan, above), + (Step::Merge, Above::Open(bound)) => match plan.with_fetch(bound.input_rows()) { + Some(_) => (place(plan, bound), Above::Enforced(bound.input_rows())), + // Not reachable in-tree: both merges absorb a fetch. Unlike the + // other branches, the bound below does not add `skip`. + None => (limit_node(plan, bound), Above::Enforced(bound.fetch)), + }, + (Step::Merge, Above::Enforced(fetch)) => ( + plan.with_fetch(fetch).unwrap_or(plan), + Above::Enforced(fetch), + ), + (Step::Stop, Above::Open(bound)) => (place(plan, bound), Above::Nothing), + (Step::Stop, Above::Enforced(fetch)) => { + (plan.with_fetch(fetch).unwrap_or(plan), Above::Nothing) + } + } +} + +/// The state [`pushdown_limit_helper`] carries from one node to the next. +#[deprecated( + since = "56.0.0", + note = "LimitPushdown no longer works one node at a time; run the `LimitPushdown` rule" +)] +#[derive(Default, Clone, Debug)] +pub struct GlobalRequirements { fetch: Option, skip: usize, + satisfied: bool, } -/// This function is the main helper function of the `LimitPushDown` rule. -/// The helper takes an `ExecutionPlan` and a global (algorithm) state which is -/// an instance of `GlobalRequirements` and modifies these parameters while -/// checking if the limits can be pushed down or not. -/// -/// If a limit is encountered, a [`TreeNodeRecursion::Stop`] is returned. Otherwise, -/// return a [`TreeNodeRecursion::Continue`]. +#[expect(deprecated)] +impl From for Above { + fn from(state: GlobalRequirements) -> Self { + let bound = RowBound { + skip: state.skip, + fetch: state.fetch, + }; + if state.satisfied && bound.skip == 0 { + Above::Enforced(bound.fetch) + } else { + Above::open(bound) + } + } +} + +#[expect(deprecated)] +impl From for GlobalRequirements { + fn from(above: Above) -> Self { + match above { + Above::Nothing => GlobalRequirements::default(), + Above::Open(bound) => GlobalRequirements { + fetch: bound.fetch, + skip: bound.skip, + satisfied: false, + }, + Above::Enforced(fetch) => GlobalRequirements { + fetch, + skip: 0, + satisfied: true, + }, + } + } +} + +/// Applies the [`LimitPushdown`] rule to one node. A limit node is removed and +/// [`TreeNodeRecursion::Stop`] is returned; call again on its input. /// /// Computes statistics with a new [`StatisticsContext`] that has no statistics -/// providers. A context built from a statistics registry, as [`LimitPushdown`] -/// uses, also consults the registered providers, so switching to -/// [`pushdown_limit_helper_with_stats`] with such a context can change the -/// result when providers are registered. +/// providers. #[deprecated( since = "56.0.0", - note = "use `pushdown_limit_helper_with_stats` and share one `StatisticsContext` across calls" + note = "LimitPushdown no longer works one node at a time; run the `LimitPushdown` rule" )] +#[expect(deprecated)] pub fn pushdown_limit_helper( pushdown_plan: Arc, global_state: GlobalRequirements, @@ -176,172 +354,43 @@ pub fn pushdown_limit_helper( ) } -/// Like [`pushdown_limit_helper`], but computes statistics with `stats_ctx`, -/// so its cache can be shared with other calls and other rules. +/// Like [`pushdown_limit_helper`], but computes statistics with `stats_ctx`. +#[deprecated( + since = "56.0.0", + note = "LimitPushdown no longer works one node at a time; run the `LimitPushdown` rule" +)] +#[expect(deprecated)] pub fn pushdown_limit_helper_with_stats( - mut pushdown_plan: Arc, - mut global_state: GlobalRequirements, + pushdown_plan: Arc, + global_state: GlobalRequirements, stats_ctx: &StatisticsContext, ) -> Result<(Transformed>, 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( - global_state.skip, - global_state.fetch, - limit_info.skip, - limit_info.fetch, - ); - global_state.skip = skip; - global_state.fetch = fetch; - global_state.satisfied = false; - - if let Some(fetch) = fetch - && limit_satisfied_by_input(&limit_info.input, skip, fetch, stats_ctx)? - { - // 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, - )); - } - - // 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. + let above = Above::from(global_state); + if let Some((input, above)) = fold_limit(&pushdown_plan, above, stats_ctx)? { return Ok(( - Transformed { - data: limit_info.input, - transformed: true, - tnr: TreeNodeRecursion::Stop, - }, - global_state, + Transformed::new(input, true, TreeNodeRecursion::Stop), + above.into(), )); } - - // If we have a non-limit operator with fetch capability, update global - // state as necessary: - if pushdown_plan.fetch().is_some() { - if global_state.skip == 0 { - global_state.satisfied = true; - } - (global_state.skip, global_state.fetch) = combine_limit( - global_state.skip, - global_state.fetch, - 0, - pushdown_plan.fetch(), - ); - } - - let Some(global_fetch) = global_state.fetch else { - // There's no valid fetch information, exit early: - return if global_state.skip > 0 && !global_state.satisfied { - // There might be a case with only offset, if so add a global limit: - global_state.satisfied = true; - Ok(( - Transformed::yes(add_global_limit( - pushdown_plan, - global_state.skip, - None, - )), - global_state, - )) - } else { - // There's no info on offset or fetch, nothing to do: - Ok((Transformed::no(pushdown_plan), global_state)) - }; - }; - - let skip_and_fetch = Some(global_fetch + global_state.skip); - - if pushdown_plan.supports_limit_pushdown() { - if !combines_input_partitions(&pushdown_plan) { - // We have information in the global state and the plan pushes down, - // continue: - Ok((Transformed::no(pushdown_plan), global_state)) - } else if let Some(plan_with_fetch) = pushdown_plan.with_fetch(skip_and_fetch) { - // This plan is combining input partitions, so we need to add the - // fetch info to plan if possible. If not, we must add a limit node - // with the information from the global state. - let mut new_plan = plan_with_fetch; - // Execution plans can't (yet) handle skip, so if we have one, - // we still need to add a global limit - if global_state.skip > 0 { - new_plan = - add_global_limit(new_plan, global_state.skip, global_state.fetch); - } - global_state.fetch = skip_and_fetch; - global_state.skip = 0; - global_state.satisfied = true; - Ok((Transformed::yes(new_plan), global_state)) - } else if global_state.satisfied { - // If the plan is already satisfied, do not add a limit: - Ok((Transformed::no(pushdown_plan), global_state)) - } else { - global_state.satisfied = true; - Ok(( - Transformed::yes(add_limit( - pushdown_plan, - global_state.skip, - global_fetch, - )), - global_state, - )) - } - } else { - // The plan does not support push down and it is not a limit. We will need - // to add a limit or a fetch. If the plan is already satisfied, we will try - // to add the fetch info and return the plan. - - // There's no push down, change fetch & skip to default values: - let global_skip = global_state.skip; - global_state.fetch = None; - global_state.skip = 0; - - let maybe_fetchable = pushdown_plan.with_fetch(skip_and_fetch); - if global_state.satisfied { - if let Some(plan_with_fetch) = maybe_fetchable { - Ok((Transformed::yes(plan_with_fetch), global_state)) - } else { - Ok((Transformed::no(pushdown_plan), global_state)) - } - } else { - global_state.satisfied = true; - pushdown_plan = if let Some(plan_with_fetch) = maybe_fetchable { - if global_skip > 0 { - add_global_limit(plan_with_fetch, global_skip, Some(global_fetch)) - } else { - plan_with_fetch - } - } else { - add_limit(pushdown_plan, global_skip, global_fetch) - }; - Ok((Transformed::yes(pushdown_plan), global_state)) - } - } + let original = Arc::clone(&pushdown_plan); + let (plan, below) = apply_bound(pushdown_plan, above); + let transformed = !Arc::ptr_eq(&original, &plan); + Ok(( + Transformed::new_transformed(plan, transformed), + below.into(), + )) } /// Returns true if exact input statistics prove that applying the limit would /// not remove any rows. fn limit_satisfied_by_input( plan: &Arc, - skip: usize, - fetch: usize, + bound: RowBound, stats_ctx: &StatisticsContext, ) -> Result { - if skip > 0 { + let (0, Some(fetch)) = (bound.skip, bound.fetch) else { return Ok(false); - } + }; if plan.output_partitioning().partition_count() != 1 { return Ok(false); @@ -388,100 +437,4 @@ fn limit_eliminable_exact_num_rows( Ok(None) } -/// Pushes down the limit through the plan. -pub(crate) fn pushdown_limits( - pushdown_plan: Arc, - global_state: GlobalRequirements, - stats_ctx: &StatisticsContext, -) -> Result> { - // Call pushdown_limit_helper. - // This will either extract the limit node (returning the child), or apply the limit pushdown. - let (mut new_node, mut global_state) = - pushdown_limit_helper_with_stats(pushdown_plan, global_state, stats_ctx)?; - - // While limits exist, continue combining the global_state. - while new_node.tnr == TreeNodeRecursion::Stop { - (new_node, global_state) = - pushdown_limit_helper_with_stats(new_node.data, global_state, stats_ctx)?; - } - - // Once a limit has been materialized above the current node, child - // subtrees should not inherit its `skip`. Keep `fetch`, but clear - // `skip` before recursing so child-local limits are not merged with - // an `OFFSET` that has already been applied. - if global_state.satisfied { - global_state.skip = 0; - } - - // Apply pushdown limits in children - let children = new_node.data.children(); - let mut changed = false; - let new_children = children - .into_iter() - .map(|child: &Arc| { - let new_child = pushdown_limits( - Arc::::clone(child), - global_state.clone(), - stats_ctx, - )?; - // Tracking if any of the children changed - changed |= !Arc::ptr_eq(child, &new_child); - Ok(new_child) - }) - .collect::>()?; - - if changed { - replace_children_if_necessary(new_node.data, new_children) - } else { - Ok(new_node.data) - } -} - -/// Extracts limit information from the [`ExecutionPlan`] if it is a -/// [`GlobalLimitExec`] or a [`LocalLimitExec`]. -fn extract_limit(plan: &Arc) -> Option { - if let Some(global_limit) = plan.downcast_ref::() { - Some(LimitInfo { - input: Arc::clone(global_limit.input()), - fetch: global_limit.fetch(), - skip: global_limit.skip(), - }) - } else { - plan.downcast_ref::() - .map(|local_limit| LimitInfo { - input: Arc::clone(local_limit.input()), - fetch: Some(local_limit.fetch()), - skip: 0, - }) - } -} - -/// Checks if the given plan combines input partitions. -fn combines_input_partitions(plan: &Arc) -> bool { - plan.is::() || plan.is::() -} - -/// Adds a limit to the plan, chooses between global and local limits based on -/// skip value and the number of partitions. -fn add_limit( - pushdown_plan: Arc, - skip: usize, - fetch: usize, -) -> Arc { - if skip > 0 || pushdown_plan.output_partitioning().partition_count() == 1 { - add_global_limit(pushdown_plan, skip, Some(fetch)) - } else { - Arc::new(LocalLimitExec::new(pushdown_plan, fetch + skip)) as _ - } -} - -/// Adds a global limit to the plan. -fn add_global_limit( - pushdown_plan: Arc, - skip: usize, - fetch: Option, -) -> Arc { - Arc::new(GlobalLimitExec::new(pushdown_plan, skip, fetch)) as _ -} - // See tests in datafusion/core/tests/physical_optimizer diff --git a/datafusion/physical-optimizer/src/row_bound.rs b/datafusion/physical-optimizer/src/row_bound.rs index 72d930ac4960..6149875fac41 100644 --- a/datafusion/physical-optimizer/src/row_bound.rs +++ b/datafusion/physical-optimizer/src/row_bound.rs @@ -25,6 +25,7 @@ use std::sync::Arc; +use datafusion_common::utils::combine_limit; use datafusion_physical_plan::limit::{GlobalLimitExec, LocalLimitExec}; use datafusion_physical_plan::{ExecutionPlan, ExecutionPlanProperties}; @@ -57,6 +58,12 @@ impl RowBound { } } + /// `self` applied over the output of `inner`: first `inner`, then `self`. + pub fn over(self, inner: RowBound) -> RowBound { + let (skip, fetch) = combine_limit(self.skip, self.fetch, inner.skip, inner.fetch); + RowBound { skip, fetch } + } + /// The tighter of two bounds on the same output. Both must have `skip == 0`. pub fn min(self, other: RowBound) -> RowBound { debug_assert!(self.skip == 0 && other.skip == 0); @@ -77,6 +84,24 @@ impl RowBound { } } +/// Writes `bound` on top of `plan`'s output. +/// +/// `plan` absorbs `bound.input_rows()` as its own fetch when it can; a +/// [`GlobalLimitExec`] on top still applies the skip, if any. When `plan` +/// can't absorb it, a [`limit_node`] is added instead. +pub(crate) fn place( + plan: Arc, + bound: RowBound, +) -> Arc { + match plan.with_fetch(bound.input_rows()) { + Some(absorbed) if bound.skip == 0 => absorbed, + Some(absorbed) => { + Arc::new(GlobalLimitExec::new(absorbed, bound.skip, bound.fetch)) + } + None => limit_node(plan, bound), + } +} + /// A limit node applying `bound` to `plan`: a [`GlobalLimitExec`] when it /// skips rows or `plan` has one output partition, otherwise a per-partition /// [`LocalLimitExec`]. @@ -104,6 +129,17 @@ mod tests { RowBound { skip, fetch } } + #[test] + fn over_applies_inner_first() { + // LIMIT 5 over (OFFSET 2 LIMIT 10): rows 2..7 of the input. + assert_eq!( + bound(0, Some(5)).over(bound(2, Some(10))), + bound(2, Some(5)) + ); + // OFFSET 3 over LIMIT 10: 7 rows remain after the skip. + assert_eq!(bound(3, None).over(bound(0, Some(10))), bound(3, Some(7))); + } + #[test] fn min_keeps_the_tighter_fetch() { assert_eq!( diff --git a/docs/source/library-user-guide/upgrading/56.0.0.md b/docs/source/library-user-guide/upgrading/56.0.0.md index 2ec01c99479c..146ec031f928 100644 --- a/docs/source/library-user-guide/upgrading/56.0.0.md +++ b/docs/source/library-user-guide/upgrading/56.0.0.md @@ -876,14 +876,12 @@ default) are not affected. As for operators, `Exact` statistics from a provider must be guaranteed values: `AggregateStatistics` answers `COUNT(*)`, `MIN` and `MAX` from them, and `LimitPushdown` removes a limit over an `Exact(0)` input. -For the same reason, -`datafusion_physical_optimizer::limit_pushdown::pushdown_limit_helper` is -deprecated in favour of `pushdown_limit_helper_with_stats`, which takes the -`StatisticsContext` from the caller. The deprecated form creates a new context -on every call, without statistics providers. When providers are registered, -switching to `pushdown_limit_helper_with_stats` with a context built from the -statistics registry can change the result, because the providers are then -consulted. +`datafusion_physical_optimizer::limit_pushdown::pushdown_limit_helper`, +`pushdown_limit_helper_with_stats` and `GlobalRequirements` are deprecated. +`LimitPushdown` no longer works one node at a time; run the `LimitPushdown` +rule instead. `pushdown_limit_helper` creates a new `StatisticsContext` on +every call, without statistics providers, so its result can differ from the +rule's when providers are registered. ### `MergeIntoOp` requires the SQL-visible target qualifier From 34bc00854d17b673f9312ca6a1fb335795a11ee8 Mon Sep 17 00:00:00 2001 From: Jay Zhan Date: Fri, 23 Oct 2026 19:37:48 +0800 Subject: [PATCH 3/3] refactor: let operators declare their LimitBehavior instead of supports_limit_pushdown --- .../physical_optimizer/ensure_requirements.rs | 29 +++++ .../enforce_sorting/sort_pushdown.rs | 43 ++---- .../physical-optimizer/src/limit_pushdown.rs | 19 +-- .../src/limit_pushdown_past_window.rs | 122 ++++-------------- .../physical-optimizer/src/row_bound.rs | 13 +- datafusion/physical-plan/src/buffer.rs | 11 ++ .../physical-plan/src/coalesce_partitions.rs | 5 + datafusion/physical-plan/src/coop.rs | 5 + .../physical-plan/src/execution_plan.rs | 77 ++++++++++- .../physical-plan/src/joins/hash_join/exec.rs | 7 - datafusion/physical-plan/src/lib.rs | 9 +- datafusion/physical-plan/src/limit.rs | 15 +++ datafusion/physical-plan/src/projection.rs | 5 + .../src/sorts/sort_preserving_merge.rs | 5 + datafusion/physical-plan/src/union.rs | 5 + .../src/windows/bounded_window_agg_exec.rs | 62 ++++++++- .../library-user-guide/upgrading/56.0.0.md | 39 ++++++ 17 files changed, 308 insertions(+), 163 deletions(-) diff --git a/datafusion/core/tests/physical_optimizer/ensure_requirements.rs b/datafusion/core/tests/physical_optimizer/ensure_requirements.rs index 8b1b4ff011f9..5e0c652b8f3a 100644 --- a/datafusion/core/tests/physical_optimizer/ensure_requirements.rs +++ b/datafusion/core/tests/physical_optimizer/ensure_requirements.rs @@ -510,6 +510,35 @@ fn test_skip_and_fetch_multi_partition() { "); } +/// A merge with its own fetch already selects rows, so the fetch of a +/// stricter TopK above it must stay above it. +#[test] +fn test_fetched_merge_keeps_stricter_topk_above() { + let source = Arc::new(MockMultiPartitionExec::new(4)); + let a = sort_expr_on("a", 0, false, false); + let a_b = LexOrdering::new( + a.iter() + .cloned() + .chain(sort_expr_on("b", 1, false, false).iter().cloned()), + ) + .unwrap(); + let inner = Arc::new( + SortExec::new(a.clone(), source) + .with_fetch(Some(5)) + .with_preserve_partitioning(true), + ); + let merge = Arc::new(SortPreservingMergeExec::new(a, inner).with_fetch(Some(5))); + let topk: Arc = + Arc::new(SortExec::new(a_b, merge).with_fetch(Some(2))); + + assert_ensure_requirements_plan!(topk, @r" + SortExec: TopK(fetch=2), expr=[a@0 ASC NULLS LAST, b@1 ASC NULLS LAST], preserve_partitioning=[false], sort_prefix=[a@0 ASC NULLS LAST] + SortPreservingMergeExec: [a@0 ASC NULLS LAST], fetch=5 + LocalLimitExec: fetch=5 + MockMultiPartitionExec + "); +} + // ======================================================================== // Multiple sort columns test // ======================================================================== diff --git a/datafusion/physical-optimizer/src/ensure_requirements/enforce_sorting/sort_pushdown.rs b/datafusion/physical-optimizer/src/ensure_requirements/enforce_sorting/sort_pushdown.rs index c39687d7c46d..358b0ab46ce9 100644 --- a/datafusion/physical-optimizer/src/ensure_requirements/enforce_sorting/sort_pushdown.rs +++ b/datafusion/physical-optimizer/src/ensure_requirements/enforce_sorting/sort_pushdown.rs @@ -37,7 +37,6 @@ use datafusion_physical_expr_common::sort_expr::{ PhysicalSortRequirement, }; use datafusion_physical_plan::aggregates::AggregateExec; -use datafusion_physical_plan::execution_plan::CardinalityEffect; use datafusion_physical_plan::filter::FilterExec; use datafusion_physical_plan::joins::utils::{ ColumnIndex, calculate_join_output_ordering, @@ -49,7 +48,7 @@ use datafusion_physical_plan::repartition::RepartitionExec; use datafusion_physical_plan::sorts::sort::SortExec; use datafusion_physical_plan::sorts::sort_preserving_merge::SortPreservingMergeExec; use datafusion_physical_plan::tree_node::PlanContext; -use datafusion_physical_plan::{ExecutionPlan, ExecutionPlanProperties}; +use datafusion_physical_plan::{ExecutionPlan, ExecutionPlanProperties, LimitBehavior}; use crate::row_bound::{RowBound, limit_node}; @@ -143,10 +142,12 @@ fn min_fetch(f1: Option, f2: Option) -> Option { RowBound::fetch(f1).min(RowBound::fetch(f2)).fetch } -/// Returns whether a fetch on `plan` can also be applied to each child. +/// Returns whether a fetch on `plan` can move to its input. fn can_push_fetch_through(plan: &Arc) -> bool { - plan.supports_limit_pushdown() - && matches!(plan.cardinality_effect(), CardinalityEffect::Equal) + matches!( + plan.limit_behavior(), + LimitBehavior::MoveToInput | LimitBehavior::Merge { fetch: None } + ) } /// Returns a plan when the fast path for an already-satisfied ordering can also @@ -443,29 +444,10 @@ fn pushdown_requirement_to_children( parent_required: OrderingRequirements, parent_fetch: Option, ) -> Result>>> { - // A parent fetch can be pushed through a plan only if the plan explicitly - // supports limit pushdown and does not change cardinality. For example, if - // LIMIT 2 is applied below a FilterExec that removes half the rows, only one - // row may remain. Applied after the filter, it correctly returns two rows - // when at least two are available. - // - // Checking `supports_limit_pushdown()` alone is not enough. For example: - // - // UnionExec advertises `supports_limit_pushdown() == true` because it can - // forward a LIMIT k to each of its children—i.e. apply “LIMIT k” separately - // on each branch before merging them together. - // - // However, UnionExec’s `cardinality_effect() == GreaterEqual` (it sums up - // all child row counts), so pushing a global TopK/LIMIT through it would - // break the semantics of “take the first k rows of the combined result.” - // - // For example, with two branches A and B and k = 3: - // — Global LIMIT: take the first 3 rows from (A ∪ B) after merging. - // — Pushed down: take 3 from A, 3 from B, then merge → up to 6 rows! - // - // That’s why we also require equal cardinality: even though UnionExec can - // push a LIMIT to its children, its GreaterEqual effect means it cannot - // preserve the global TopK semantics. + // Pushing the sort below `plan` moves the parent fetch with it. For + // example, if LIMIT 2 is applied below a FilterExec that removes half the + // rows, only one row may remain; and a LIMIT 3 pushed into both inputs of + // a UnionExec keeps up to 6 rows. See `LimitBehavior`. if parent_fetch.is_some() && !can_push_fetch_through(plan) { return Ok(None); } @@ -540,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(_) + ) && plan .maintains_input_order() .into_iter() diff --git a/datafusion/physical-optimizer/src/limit_pushdown.rs b/datafusion/physical-optimizer/src/limit_pushdown.rs index b944bf60b522..7f46b9ec1f8c 100644 --- a/datafusion/physical-optimizer/src/limit_pushdown.rs +++ b/datafusion/physical-optimizer/src/limit_pushdown.rs @@ -73,14 +73,12 @@ use datafusion_common::config::ConfigOptions; use datafusion_common::error::Result; use datafusion_common::stats::Precision; use datafusion_common::tree_node::{Transformed, TreeNodeRecursion}; -use datafusion_physical_plan::coalesce_partitions::CoalescePartitionsExec; use datafusion_physical_plan::empty::EmptyExec; use datafusion_physical_plan::execution_plan::replace_children_if_necessary; use datafusion_physical_plan::placeholder_row::PlaceholderRowExec; use datafusion_physical_plan::projection::ProjectionExec; -use datafusion_physical_plan::sorts::sort_preserving_merge::SortPreservingMergeExec; use datafusion_physical_plan::statistics::{StatisticsArgs, StatisticsContext}; -use datafusion_physical_plan::{ExecutionPlan, ExecutionPlanProperties}; +use datafusion_physical_plan::{ExecutionPlan, ExecutionPlanProperties, LimitBehavior}; /// This rule inspects [`ExecutionPlan`]'s and pushes down the fetch limit from /// the parent to the child if applicable. #[derive(Default, Debug)] @@ -179,15 +177,12 @@ enum Step { } fn step(plan: &Arc) -> Step { - if let Some(bound) = RowBound::of_limit_node(plan) { - Step::Limit(bound) - } else if !plan.supports_limit_pushdown() { - Step::Stop - } else if plan.is::() || plan.is::() - { - Step::Merge - } else { - Step::Pass + match plan.limit_behavior() { + LimitBehavior::Limit { skip, fetch } => Step::Limit(RowBound { skip, fetch }), + LimitBehavior::CopyToChildren | LimitBehavior::MoveToInput => Step::Pass, + LimitBehavior::Merge { .. } => Step::Merge, + // LimitPushPastWindows handles windows. + LimitBehavior::Stops | LimitBehavior::Window(_) => Step::Stop, } } diff --git a/datafusion/physical-optimizer/src/limit_pushdown_past_window.rs b/datafusion/physical-optimizer/src/limit_pushdown_past_window.rs index 52b5d6e596d7..25436fb97bee 100644 --- a/datafusion/physical-optimizer/src/limit_pushdown_past_window.rs +++ b/datafusion/physical-optimizer/src/limit_pushdown_past_window.rs @@ -17,26 +17,16 @@ use crate::PhysicalOptimizerRule; use crate::row_bound::RowBound; -use datafusion_common::ScalarValue; use datafusion_common::config::ConfigOptions; use datafusion_common::tree_node::{Transformed, TreeNode}; -use datafusion_expr::{LimitEffect, WindowFrameBound, WindowFrameUnits}; -use datafusion_physical_expr::window::{ - PlainAggregateWindowExpr, SlidingAggregateWindowExpr, StandardWindowExpr, - StandardWindowFunctionExpr, WindowExpr, -}; -use datafusion_physical_plan::execution_plan::CardinalityEffect; -use datafusion_physical_plan::repartition::RepartitionExec; use datafusion_physical_plan::sorts::sort::SortExec; use datafusion_physical_plan::sorts::sort_preserving_merge::SortPreservingMergeExec; -use datafusion_physical_plan::windows::{BoundedWindowAggExec, WindowUDFExpr}; -use datafusion_physical_plan::{ExecutionPlan, ExecutionPlanProperties}; +use datafusion_physical_plan::{ExecutionPlan, LimitBehavior, WindowLookahead}; use std::sync::Arc; /// This rule inspects [`ExecutionPlan`]'s attempting to find fetch limits that were not pushed -/// down by `LimitPushdown` because [BoundedWindowAggExec]s were "in the way". If the window is -/// bounded by [WindowFrameUnits::Rows] then we calculate the adjustment needed to grow the limit -/// and continue pushdown. +/// down by `LimitPushdown` because windows ([`LimitBehavior::Window`]) were "in the way". If a +/// window's lookahead is known, the limit grows by it and pushdown continues. #[derive(Default, Clone, Debug)] pub struct LimitPushPastWindows; @@ -103,9 +93,9 @@ impl PhysicalOptimizerRule for LimitPushPastWindows { } // grow the limit if we hit a window function - if let Some(window) = node.downcast_ref::() { + if let LimitBehavior::Window(lookahead) = node.limit_behavior() { phase = Phase::Apply; - if !grow_limit(window, &mut ctx) { + if !grow_limit(lookahead, &mut ctx) { return reset(node, &mut ctx); } return Ok(Transformed::no(node)); @@ -119,21 +109,14 @@ impl PhysicalOptimizerRule for LimitPushPastWindows { } // nodes along the way - if !node.supports_limit_pushdown() { - return reset(node, &mut ctx); - } - if let Some(part) = node.downcast_ref::() { - let output = part.partitioning().partition_count(); - let input = part.input().output_partitioning().partition_count(); - if output < input { - return reset(node, &mut ctx); - } - } - match node.cardinality_effect() { - CardinalityEffect::Unknown => return reset(node, &mut ctx), - CardinalityEffect::LowerEqual => return reset(node, &mut ctx), - CardinalityEffect::Equal => {} - CardinalityEffect::GreaterEqual => {} + match node.limit_behavior() { + LimitBehavior::CopyToChildren + | LimitBehavior::MoveToInput + | LimitBehavior::Merge { fetch: None } => {} + LimitBehavior::Stops + | LimitBehavior::Limit { .. } + | LimitBehavior::Merge { fetch: Some(_) } + | LimitBehavior::Window(_) => return reset(node, &mut ctx), } Ok(Transformed::no(node)) @@ -150,33 +133,15 @@ impl PhysicalOptimizerRule for LimitPushPastWindows { } } -fn grow_limit(window: &BoundedWindowAggExec, ctx: &mut TraverseState) -> bool { - let mut max_rel = 0; - for expr in window.window_expr().iter() { - // grow based on function requirements - match get_limit_effect(expr) { - LimitEffect::None => {} - LimitEffect::Unknown => return false, - LimitEffect::Relative(rel) => max_rel = max_rel.max(rel), - LimitEffect::Absolute(val) => { - let cur = ctx.limit.unwrap_or(0); - ctx.limit = Some(cur.max(val)) - } - } - - // grow based on frames - let frame = expr.get_window_frame(); - if frame.units != WindowFrameUnits::Rows { - return false; // expression-based limits not statically evaluatable - } - let Some(end_bound) = bound_to_usize(&frame.end_bound) else { - return false; // can't optimize unbounded window expressions - }; - ctx.max_lookahead(end_bound); +fn grow_limit(lookahead: Option, ctx: &mut TraverseState) -> bool { + let Some(lookahead) = lookahead else { + return false; + }; + if let Some(min_rows) = lookahead.min_rows { + ctx.limit = Some(ctx.limit.unwrap_or(0).max(min_rows)); } - - // finish grow - ctx.max_lookahead(ctx.lookahead + max_rel); + ctx.max_lookahead(lookahead.frame_end); + ctx.max_lookahead(ctx.lookahead + lookahead.relative); true } @@ -214,55 +179,16 @@ fn get_limit(node: &Arc, ctx: &mut TraverseState) -> bool { false } -/// Examines the `WindowExpr` and decides: -/// 1. The expression does not change the window size -/// 2. The expression grows it by X amount -/// 3. We don't know -/// -/// # Arguments -/// -/// * `expr` the expression to examine -/// -/// # Returns -/// -/// The effect on the limit -fn get_limit_effect(expr: &Arc) -> LimitEffect { - // White list aggregates - if expr.as_any().is::() - || expr.as_any().is::() - { - return LimitEffect::None; - } - - // Grab the window function - let Some(swe) = expr.as_any().downcast_ref::() else { - return LimitEffect::Unknown; // should be only remaining type - }; - let swfe = swe.get_standard_func_expr(); - let Some(udf) = swfe.as_any().downcast_ref::() else { - return LimitEffect::Unknown; // should be only remaining type - }; - udf.limit_effect() -} - -fn bound_to_usize(bound: &WindowFrameBound) -> Option { - match bound { - WindowFrameBound::Preceding(_) => Some(0), - WindowFrameBound::CurrentRow => Some(0), - WindowFrameBound::Following(ScalarValue::UInt64(Some(scalar))) => { - Some(*scalar as usize) - } - _ => None, - } -} - #[cfg(test)] mod tests { use super::*; use arrow::datatypes::{DataType, Field, Schema}; + use datafusion_common::ScalarValue; use datafusion_expr::WindowFrame; + use datafusion_expr::{WindowFrameBound, WindowFrameUnits}; use datafusion_functions_window::row_number::row_number_udwf; use datafusion_physical_expr::expressions::col; + use datafusion_physical_expr::window::StandardWindowExpr; use datafusion_physical_expr_common::sort_expr::{LexOrdering, PhysicalSortExpr}; use datafusion_physical_plan::InputOrderMode; use datafusion_physical_plan::displayable; diff --git a/datafusion/physical-optimizer/src/row_bound.rs b/datafusion/physical-optimizer/src/row_bound.rs index 6149875fac41..2335a0489694 100644 --- a/datafusion/physical-optimizer/src/row_bound.rs +++ b/datafusion/physical-optimizer/src/row_bound.rs @@ -27,7 +27,7 @@ use std::sync::Arc; use datafusion_common::utils::combine_limit; use datafusion_physical_plan::limit::{GlobalLimitExec, LocalLimitExec}; -use datafusion_physical_plan::{ExecutionPlan, ExecutionPlanProperties}; +use datafusion_physical_plan::{ExecutionPlan, ExecutionPlanProperties, LimitBehavior}; /// `OFFSET skip LIMIT fetch` on the output of one operator. /// @@ -47,14 +47,9 @@ impl RowBound { /// The bound a limit node imposes, or `None` if `plan` is not a limit node. pub fn of_limit_node(plan: &Arc) -> Option { - if let Some(limit) = plan.downcast_ref::() { - Some(Self { - skip: limit.skip(), - fetch: limit.fetch(), - }) - } else { - plan.downcast_ref::() - .map(|limit| Self::fetch(Some(limit.fetch()))) + match plan.limit_behavior() { + LimitBehavior::Limit { skip, fetch } => Some(Self { skip, fetch }), + _ => None, } } diff --git a/datafusion/physical-plan/src/buffer.rs b/datafusion/physical-plan/src/buffer.rs index df87fee7da08..ff54130587df 100644 --- a/datafusion/physical-plan/src/buffer.rs +++ b/datafusion/physical-plan/src/buffer.rs @@ -18,6 +18,7 @@ //! [`BufferExec`] decouples production and consumption on messages by buffering the input in the //! background up to a certain capacity. +use crate::execution_plan::LimitBehavior; use crate::execution_plan::{ CardinalityEffect, EvaluationType, SchedulingType, replace_children_if_necessary, }; @@ -277,9 +278,19 @@ impl ExecutionPlan for BufferExec { } fn supports_limit_pushdown(&self) -> bool { + #[expect(deprecated)] self.input.supports_limit_pushdown() } + fn limit_behavior(&self) -> LimitBehavior { + // Buffering keeps rows 1:1, so a limit moves below it whenever it can + // also go below the input. + match self.input.limit_behavior() { + LimitBehavior::Stops | LimitBehavior::Window(_) => LimitBehavior::Stops, + _ => LimitBehavior::MoveToInput, + } + } + fn cardinality_effect(&self) -> CardinalityEffect { CardinalityEffect::Equal } diff --git a/datafusion/physical-plan/src/coalesce_partitions.rs b/datafusion/physical-plan/src/coalesce_partitions.rs index 7097cc780de0..d17cf9b191db 100644 --- a/datafusion/physical-plan/src/coalesce_partitions.rs +++ b/datafusion/physical-plan/src/coalesce_partitions.rs @@ -28,6 +28,7 @@ use super::{ DisplayAs, ExecutionPlanProperties, PlanProperties, SendableRecordBatchStream, Statistics, }; +use crate::execution_plan::LimitBehavior; use crate::execution_plan::{CardinalityEffect, EvaluationType, SchedulingType}; use crate::filter_pushdown::{FilterDescription, FilterPushdownPhase}; use crate::projection::{ProjectionExec, make_with_child}; @@ -301,6 +302,10 @@ impl ExecutionPlan for CoalescePartitionsExec { true } + fn limit_behavior(&self) -> LimitBehavior { + LimitBehavior::Merge { fetch: self.fetch } + } + fn cardinality_effect(&self) -> CardinalityEffect { if self.fetch.is_none() { CardinalityEffect::Equal diff --git a/datafusion/physical-plan/src/coop.rs b/datafusion/physical-plan/src/coop.rs index 17166e287e6d..b7904bbda693 100644 --- a/datafusion/physical-plan/src/coop.rs +++ b/datafusion/physical-plan/src/coop.rs @@ -80,6 +80,7 @@ use std::sync::Arc; use std::task::{Context, Poll}; use crate::execution_plan::CardinalityEffect::{self, Equal}; +use crate::execution_plan::LimitBehavior; use crate::filter_pushdown::{ ChildPushdownResult, FilterDescription, FilterPushdownPhase, FilterPushdownPropagation, @@ -338,6 +339,10 @@ impl ExecutionPlan for CooperativeExec { true } + fn limit_behavior(&self) -> LimitBehavior { + LimitBehavior::MoveToInput + } + fn cardinality_effect(&self) -> CardinalityEffect { Equal } diff --git a/datafusion/physical-plan/src/execution_plan.rs b/datafusion/physical-plan/src/execution_plan.rs index ee15653f7df5..d60557676d5d 100644 --- a/datafusion/physical-plan/src/execution_plan.rs +++ b/datafusion/physical-plan/src/execution_plan.rs @@ -777,13 +777,37 @@ pub trait ExecutionPlan: Any + Debug + DisplayAs + Send + Sync { /// Returns `true` if a limit can be safely pushed down through this /// `ExecutionPlan` node. /// - /// If this method returns `true`, and the query plan contains a limit at - /// the output of this node, DataFusion will push the limit to the input - /// of this node. + /// DataFusion's own operators still return the same values as before, so + /// a wrapper that forwards this to its input keeps working. + #[deprecated( + since = "56.0.0", + note = "implement `limit_behavior` instead; DataFusion no longer calls this" + )] fn supports_limit_pushdown(&self) -> bool { false } + /// What a row limit (`LIMIT` / `OFFSET`) on this node's output means for + /// its input. The optimizer rules that move limits read only this; see + /// [`LimitBehavior`]. + /// + /// The default keeps the behavior of nodes that still implement the + /// deprecated [`Self::supports_limit_pushdown`]: [`LimitBehavior::Stops`] + /// if it returns `false`, otherwise [`LimitBehavior::MoveToInput`] when + /// [`Self::cardinality_effect`] is [`CardinalityEffect::Equal`] and + /// [`LimitBehavior::CopyToChildren`] when it is not. + fn limit_behavior(&self) -> LimitBehavior { + #[expect(deprecated)] + let supports_limit_pushdown = self.supports_limit_pushdown(); + if !supports_limit_pushdown { + return LimitBehavior::Stops; + } + match self.cardinality_effect() { + CardinalityEffect::Equal => LimitBehavior::MoveToInput, + _ => LimitBehavior::CopyToChildren, + } + } + /// Returns a fetching variant of this `ExecutionPlan` node, if it supports /// fetch limits. Returns `None` otherwise. /// @@ -2075,6 +2099,53 @@ pub fn get_plan_string(plan: &Arc) -> Vec { actual.iter().map(|elem| (*elem).to_string()).collect() } +/// What a row limit on an operator's output means for the operator's input. +/// Returned by [`ExecutionPlan::limit_behavior`]. +/// +/// A limit can be *copied* below a node as an early-stop hint while it also +/// stays above the node, or *moved* below a node, leaving nothing above it. +/// Copying is safe whenever the first `n` output rows need at most the first +/// `n` rows of each child. Moving also needs the node to output exactly the +/// rows of its input. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum LimitBehavior { + /// A limit above this node must stay above it. The default. + Stops, + /// This node applies `OFFSET skip LIMIT fetch` to its single input + /// (`GlobalLimitExec`, `LocalLimitExec`). + Limit { skip: usize, fetch: Option }, + /// A limit may be copied to each child, but may not move: this node can + /// output more rows than any one child (`UnionExec`). + CopyToChildren, + /// This node outputs exactly the rows of its input: a limit may move to + /// its input (`ProjectionExec`). + MoveToInput, + /// This node merges its input partitions, so it sees every row and can + /// enforce a limit with its own fetch (`CoalescePartitionsExec`, + /// `SortPreservingMergeExec`). Without a `fetch` of its own it outputs + /// exactly the rows of its input, like [`Self::MoveToInput`]. With one, + /// it already selects rows, so a limit above it may not move below it. + /// `fetch` must equal the node's [`ExecutionPlan::fetch`]. + Merge { fetch: Option }, + /// A window function node, with how far it reads ahead of each output + /// row; `None` if that is not known before execution. Only + /// `LimitPushPastWindows` copies a limit below a window. + Window(Option), +} + +/// How many input rows a window needs for its first `n` output rows. See +/// [`LimitBehavior::Window`]. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub struct WindowLookahead { + /// At least this many input rows, whatever `n` is. + pub min_rows: Option, + /// Output row `i` reads input rows up to `i + frame_end` (`ROWS` frames). + pub frame_end: usize, + /// The window functions read this many rows past the frame + /// (e.g. `LEAD(x, 2)`). + pub relative: usize, +} + /// Indicates the effect an execution plan operator will have on the cardinality /// of its input stream pub enum CardinalityEffect { diff --git a/datafusion/physical-plan/src/joins/hash_join/exec.rs b/datafusion/physical-plan/src/joins/hash_join/exec.rs index 9f77d782b27a..a5f7d87410cb 100644 --- a/datafusion/physical-plan/src/joins/hash_join/exec.rs +++ b/datafusion/physical-plan/src/joins/hash_join/exec.rs @@ -2227,13 +2227,6 @@ impl ExecutionPlan for HashJoinExec { Ok(result) } - fn supports_limit_pushdown(&self) -> bool { - // Hash join execution plan does not support pushing limit down through to children - // because the children don't know about the join condition and can't - // determine how many rows to produce - false - } - fn fetch(&self) -> Option { self.fetch } diff --git a/datafusion/physical-plan/src/lib.rs b/datafusion/physical-plan/src/lib.rs index cce3f0857274..27fb466ee946 100644 --- a/datafusion/physical-plan/src/lib.rs +++ b/datafusion/physical-plan/src/lib.rs @@ -47,10 +47,11 @@ pub use crate::distribution_requirements::{ #[expect(deprecated)] pub use crate::execution_plan::{ AsPhysicalExprRef, ChildrenPropertiesMode, ExecutionPlan, ExecutionPlanProperties, - PlanProperties, ReplaceChildrenOptions, apply_expression_roots, collect, - collect_partitioned, displayable, execute_input_stream, execute_stream, - execute_stream_partitioned, get_plan_string, plan_contains_expression_id, - replace_children_if_necessary, with_new_children_if_necessary, + LimitBehavior, PlanProperties, ReplaceChildrenOptions, WindowLookahead, + apply_expression_roots, collect, collect_partitioned, displayable, + execute_input_stream, execute_stream, execute_stream_partitioned, get_plan_string, + plan_contains_expression_id, replace_children_if_necessary, + with_new_children_if_necessary, }; pub use crate::metrics::Metric; pub use crate::ordering::InputOrderMode; diff --git a/datafusion/physical-plan/src/limit.rs b/datafusion/physical-plan/src/limit.rs index e2bfefb13ed1..08b1b5a73026 100644 --- a/datafusion/physical-plan/src/limit.rs +++ b/datafusion/physical-plan/src/limit.rs @@ -26,6 +26,7 @@ use super::{ DisplayAs, ExecutionPlanProperties, PlanProperties, RecordBatchStream, SendableRecordBatchStream, Statistics, }; +use crate::execution_plan::LimitBehavior; use crate::execution_plan::{Boundedness, CardinalityEffect}; use crate::statistics::{ChildStats, StatisticsArgs, with_per_partition_fetch}; use crate::{ @@ -287,6 +288,13 @@ impl ExecutionPlan for GlobalLimitExec { true } + fn limit_behavior(&self) -> LimitBehavior { + LimitBehavior::Limit { + skip: self.skip, + fetch: self.fetch, + } + } + #[cfg(feature = "proto")] fn try_to_proto( &self, @@ -565,6 +573,13 @@ impl ExecutionPlan for LocalLimitExec { true } + fn limit_behavior(&self) -> LimitBehavior { + LimitBehavior::Limit { + skip: 0, + fetch: Some(self.fetch), + } + } + fn cardinality_effect(&self) -> CardinalityEffect { CardinalityEffect::LowerEqual } diff --git a/datafusion/physical-plan/src/projection.rs b/datafusion/physical-plan/src/projection.rs index 52a847191327..c44772725a57 100644 --- a/datafusion/physical-plan/src/projection.rs +++ b/datafusion/physical-plan/src/projection.rs @@ -26,6 +26,7 @@ use super::{ DisplayAs, ExecutionPlanProperties, PlanProperties, RecordBatchStream, SendableRecordBatchStream, SortOrderPushdownResult, Statistics, }; +use crate::execution_plan::LimitBehavior; use crate::execution_plan::{CardinalityEffect, replace_children_if_necessary}; use crate::filter_pushdown::{ ChildFilterDescription, ChildPushdownResult, FilterDescription, FilterPushdownPhase, @@ -504,6 +505,10 @@ impl ExecutionPlan for ProjectionExec { true } + fn limit_behavior(&self) -> LimitBehavior { + LimitBehavior::MoveToInput + } + fn cardinality_effect(&self) -> CardinalityEffect { CardinalityEffect::Equal } diff --git a/datafusion/physical-plan/src/sorts/sort_preserving_merge.rs b/datafusion/physical-plan/src/sorts/sort_preserving_merge.rs index e865fef98fee..67226affea60 100644 --- a/datafusion/physical-plan/src/sorts/sort_preserving_merge.rs +++ b/datafusion/physical-plan/src/sorts/sort_preserving_merge.rs @@ -20,6 +20,7 @@ use std::sync::Arc; use crate::common::spawn_buffered; +use crate::execution_plan::LimitBehavior; use crate::limit::LimitStream; use crate::metrics::{BaselineMetrics, ExecutionPlanMetricsSet, MetricsSet}; use crate::projection::{ProjectionExec, make_with_child, update_ordering}; @@ -425,6 +426,10 @@ impl ExecutionPlan for SortPreservingMergeExec { true } + fn limit_behavior(&self) -> LimitBehavior { + LimitBehavior::Merge { fetch: self.fetch } + } + /// Tries to swap the projection with its input [`SortPreservingMergeExec`]. /// If this is possible, it returns the new [`SortPreservingMergeExec`] whose /// child is a projection. Otherwise, it returns None. diff --git a/datafusion/physical-plan/src/union.rs b/datafusion/physical-plan/src/union.rs index bbf8192d2154..6d14ef9e4072 100644 --- a/datafusion/physical-plan/src/union.rs +++ b/datafusion/physical-plan/src/union.rs @@ -31,6 +31,7 @@ use super::{ PlanProperties, RecordBatchStream, SendableRecordBatchStream, Statistics, metrics::{ExecutionPlanMetricsSet, MetricsSet}, }; +use crate::execution_plan::LimitBehavior; use crate::execution_plan::{ CardinalityEffect, InvariantLevel, boundedness_from_children, check_default_invariants, emission_type_from_children, @@ -462,6 +463,10 @@ impl ExecutionPlan for UnionExec { true } + fn limit_behavior(&self) -> LimitBehavior { + LimitBehavior::CopyToChildren + } + /// Tries to push `projection` down through `union`. If possible, performs the /// pushdown and returns a new [`UnionExec`] as the top plan which has projections /// as its children. Otherwise, returns `None`. diff --git a/datafusion/physical-plan/src/windows/bounded_window_agg_exec.rs b/datafusion/physical-plan/src/windows/bounded_window_agg_exec.rs index 2ef1d3dc69d9..eba7403cd93a 100644 --- a/datafusion/physical-plan/src/windows/bounded_window_agg_exec.rs +++ b/datafusion/physical-plan/src/windows/bounded_window_agg_exec.rs @@ -69,8 +69,14 @@ use datafusion_physical_expr_common::sort_expr::{ OrderingRequirements, PhysicalSortExpr, }; -use crate::execution_plan::CardinalityEffect; +use crate::execution_plan::{CardinalityEffect, LimitBehavior, WindowLookahead}; +use crate::windows::WindowUDFExpr; use datafusion_common::hash_utils::RandomState; +use datafusion_expr::{LimitEffect, WindowFrameBound, WindowFrameUnits}; +use datafusion_physical_expr::window::{ + PlainAggregateWindowExpr, SlidingAggregateWindowExpr, StandardWindowExpr, + StandardWindowFunctionExpr, +}; use futures::stream::Stream; use futures::{StreamExt, ready}; use hashbrown::hash_table::HashTable; @@ -548,6 +554,10 @@ impl ExecutionPlan for BoundedWindowAggExec { CardinalityEffect::Equal } + fn limit_behavior(&self) -> LimitBehavior { + LimitBehavior::Window(window_lookahead(&self.window_expr)) + } + #[cfg(feature = "proto")] fn try_to_proto( &self, @@ -1610,6 +1620,56 @@ pub(crate) fn get_last_row_batch(batch: &RecordBatch) -> Result { Ok(batch.slice(batch.num_rows() - 1, 1)) } +/// The [`WindowLookahead`] of a window computing `exprs`, or `None` when an +/// expression's frame or function does not bound it before execution. +fn window_lookahead(exprs: &[Arc]) -> Option { + let mut lookahead = WindowLookahead::default(); + for expr in exprs { + match limit_effect(expr) { + LimitEffect::None => {} + LimitEffect::Unknown => return None, + LimitEffect::Relative(rows) => { + lookahead.relative = lookahead.relative.max(rows) + } + LimitEffect::Absolute(rows) => { + lookahead.min_rows = Some(lookahead.min_rows.unwrap_or(0).max(rows)) + } + } + // Only a ROWS frame with a literal end bounds the lookahead. + let frame = expr.get_window_frame(); + if frame.units != WindowFrameUnits::Rows { + return None; + } + let end = match &frame.end_bound { + WindowFrameBound::Preceding(_) | WindowFrameBound::CurrentRow => 0, + WindowFrameBound::Following(ScalarValue::UInt64(Some(rows))) => { + *rows as usize + } + _ => return None, + }; + lookahead.frame_end = lookahead.frame_end.max(end); + } + Some(lookahead) +} + +/// The [`LimitEffect`] of one window expression. Plain and sliding aggregates +/// never read ahead; window functions declare it. +fn limit_effect(expr: &Arc) -> LimitEffect { + if expr.as_any().is::() + || expr.as_any().is::() + { + return LimitEffect::None; + } + expr.as_any() + .downcast_ref::() + .and_then(|e| { + e.get_standard_func_expr() + .as_any() + .downcast_ref::() + }) + .map_or(LimitEffect::Unknown, |udf| udf.limit_effect()) +} + #[cfg(test)] mod tests { use std::pin::Pin; diff --git a/docs/source/library-user-guide/upgrading/56.0.0.md b/docs/source/library-user-guide/upgrading/56.0.0.md index 146ec031f928..775d7582aec9 100644 --- a/docs/source/library-user-guide/upgrading/56.0.0.md +++ b/docs/source/library-user-guide/upgrading/56.0.0.md @@ -883,6 +883,45 @@ rule instead. `pushdown_limit_helper` creates a new `StatisticsContext` on every call, without statistics providers, so its result can differ from the rule's when providers are registered. +### `ExecutionPlan::supports_limit_pushdown` is deprecated in favour of `limit_behavior` + +Operators now declare what a `LIMIT` on their output means for their input with +`ExecutionPlan::limit_behavior`, which returns a `LimitBehavior`: + +- `Stops` (the default): a limit above the node stays above it. +- `Limit { skip, fetch }`: the node is a limit. +- `CopyToChildren`: a limit may be copied to each child as an early-stop hint, + but must also stay above the node, which can output more rows than one child + (`UnionExec`). +- `MoveToInput`: the node outputs exactly the rows of its input, so a limit may + move below it (`ProjectionExec`). +- `Merge { fetch }`: the node merges its input partitions, so it can enforce a + limit with its own fetch (`CoalescePartitionsExec`, `SortPreservingMergeExec`). + Without a `fetch` of its own it behaves like `MoveToInput`. +- `Window(lookahead)`: a window function node. + +`LimitPushdown`, `LimitPushPastWindows` and the sort pushdown in +`EnsureRequirements` read only `limit_behavior`. Its default is derived from +`supports_limit_pushdown` and `cardinality_effect`, so an operator that +implements `supports_limit_pushdown` keeps working: `false` maps to `Stops`; +`true` maps to `MoveToInput` when `cardinality_effect` is `Equal` and to +`CopyToChildren` otherwise. Implement `limit_behavior` instead: + +```rust,ignore +fn limit_behavior(&self) -> LimitBehavior { + LimitBehavior::MoveToInput +} +``` + +DataFusion's own operators keep returning the same `supports_limit_pushdown` +values until it is removed, so a wrapper that forwards it to its input keeps +working, but new code should call `limit_behavior`. + +One difference for operators that still implement `supports_limit_pushdown`: +a single-input operator that returns `true` with a `cardinality_effect` of +`Unknown` or `LowerEqual` now also lets `LimitPushPastWindows` push a window's +limit below it, as `LimitPushdown` already did for plain limits. + ### `MergeIntoOp` requires the SQL-visible target qualifier `MergeIntoOp` now stores the SQL-visible target qualifier separately from the