From 0c734428cf1b21e27644ca56a6bc8137e5cb22eb Mon Sep 17 00:00:00 2001 From: Jay Zhan Date: Sun, 6 Sep 2026 10:26:26 +0800 Subject: [PATCH 1/3] fix: account for `skip` when pushing a sort below `GlobalLimitExec` Sort pushdown seeds the `fetch` it carries from the limit's own `fetch()`, ignoring `skip`. A sort pushed below `GlobalLimitExec: skip=5, fetch=10` therefore became `TopK(fetch=10)`, and the limit then skipped 5 of those 10 rows: SELECT * FROM (SELECT * FROM t LIMIT 10 OFFSET 5) ORDER BY a returned 5 rows instead of 10. The same seed is used when a parent's ordering requirement is already satisfied by a `GlobalLimitExec` child. Use `skip + fetch` for `GlobalLimitExec` in both places. --- .../physical_optimizer/ensure_requirements.rs | 29 +++++++++++++++++++ .../enforce_sorting/sort_pushdown.rs | 18 +++++++++++- datafusion/sqllogictest/test_files/limit.slt | 23 +++++++++++++++ 3 files changed, 69 insertions(+), 1 deletion(-) diff --git a/datafusion/core/tests/physical_optimizer/ensure_requirements.rs b/datafusion/core/tests/physical_optimizer/ensure_requirements.rs index 06b57a8630553..fe73518bf17d2 100644 --- a/datafusion/core/tests/physical_optimizer/ensure_requirements.rs +++ b/datafusion/core/tests/physical_optimizer/ensure_requirements.rs @@ -1531,3 +1531,32 @@ fn test_collect_left_join_keeps_hash_partitioned_build_side_coalesce() -> Result Ok(()) } + +// ======================================================================== +// Limits with a `skip` +// ======================================================================== + +/// A sort pushed below `GlobalLimitExec` with a non-zero `skip` must ask its +/// input for `skip + fetch` rows; asking for only `fetch` rows used to leave +/// `LIMIT 10 OFFSET 5` with 5 result rows. +/// +/// This checks a single pass only: the sort is inserted by `pushdown_sorts`, +/// which runs after `parallelize_sorts`, so a second pass would additionally +/// parallelize it into `SortPreservingMergeExec` + partitioned `SortExec`. +#[test] +fn test_sort_pushed_below_limit_with_skip_keeps_skip_rows() -> Result<()> { + let source = Arc::new(MockMultiPartitionExec::new(4)); + let coalesce = Arc::new(CoalescePartitionsExec::new(source)); + let limit = Arc::new(GlobalLimitExec::new(coalesce, 5, Some(10))); + let sort: Arc = + Arc::new(SortExec::new(sort_expr_on("a", 0, true, true), limit)); + + let optimized = optimize_and_sanity_check(sort)?; + assert_snapshot!(plan_string(&optimized), @r" + GlobalLimitExec: skip=5, fetch=10 + SortExec: TopK(fetch=15), expr=[a@0 DESC], preserve_partitioning=[false] + CoalescePartitionsExec + MockMultiPartitionExec + "); + Ok(()) +} 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 19faf8ea9fa43..b80fd16bc8b83 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 @@ -79,6 +79,22 @@ impl Default for ParentRequirements { pub type SortPushDown = PlanContext; +/// Number of input rows `plan` needs from its children in order to produce +/// its own `fetch` rows. This is `plan.fetch()` for every operator except +/// [`GlobalLimitExec`], which discards `skip` rows first and therefore needs +/// `skip + fetch` input rows. Using the bare `fetch` there would turn +/// `LIMIT 10 OFFSET 5` into `TopK(10)` below the limit, i.e. 5 result rows. +/// +/// Note this is distinct from the fetch a parent imposes on `plan`'s *output* +/// (`ParentRequirements::fetch`), for which `plan.fetch()` is the right bound. +fn input_fetch(plan: &Arc) -> Option { + let fetch = plan.fetch()?; + let skip = plan + .downcast_ref::() + .map_or(0, |limit| limit.skip()); + Some(fetch + skip) +} + /// Assigns the ordering requirement of the root node to the its children. pub fn assign_initial_requirements(sort_push_down: &mut SortPushDown) { let reqs = sort_push_down.plan.required_input_ordering(); @@ -364,7 +380,7 @@ fn pushdown_sorts_helper( // For operators that can take a sort pushdown, continue with updated // requirements. If this node already outputs single partition (e.g. SPM), // don't push SinglePartition to children. - let current_fetch = sort_push_down.plan.fetch(); + let current_fetch = input_fetch(&sort_push_down.plan); let dists = sort_push_down .plan .input_distribution_requirements() diff --git a/datafusion/sqllogictest/test_files/limit.slt b/datafusion/sqllogictest/test_files/limit.slt index e17f633bc64ad..a7f6782b49348 100644 --- a/datafusion/sqllogictest/test_files/limit.slt +++ b/datafusion/sqllogictest/test_files/limit.slt @@ -339,6 +339,29 @@ SELECT COUNT(*) FROM (SELECT a FROM t1 LIMIT 3 OFFSET 8); ---- 2 +# A sort above LIMIT ... OFFSET is pushed below the limit as a TopK. The TopK +# has to keep `skip + fetch` rows (3 + 4 = 7) so that the limit can still skip +# 3 rows and return 4; keeping only `fetch` rows returned a single row. +query TT +EXPLAIN SELECT * FROM (SELECT a FROM t1 LIMIT 4 OFFSET 3) ORDER BY a; +---- +logical_plan +01)Sort: t1.a ASC NULLS LAST +02)--Limit: skip=3, fetch=4 +03)----TableScan: t1 projection=[a], fetch=7 +physical_plan +01)GlobalLimitExec: skip=3, fetch=4 +02)--SortExec: TopK(fetch=7), expr=[a@0 ASC NULLS LAST], preserve_partitioning=[false] +03)----DataSourceExec: partitions=1, partition_sizes=[1] + +query I +SELECT * FROM (SELECT a FROM t1 LIMIT 4 OFFSET 3) ORDER BY a; +---- +4 +5 +6 +7 + # The aggregate does not need to be computed because the input statistics are exact and # an OFFSET, but no LIMIT, is specified. query TT From aac5bf9c4ff46b763b0afb8f3d4e51071ff7b33f Mon Sep 17 00:00:00 2001 From: Jay Zhan Date: Mon, 7 Sep 2026 20:09:52 +0800 Subject: [PATCH 2/3] fix: saturate `skip + fetch` when seeding sort pushdown `GlobalLimitExec::new` accepts arbitrary `usize` values for `skip` and `fetch` with no combined bound, so `input_fetch`'s `fetch + skip` can overflow. Debug builds panic during physical optimization; release builds wrap, and the wrapped value becomes the pushed-down fetch -- wrapping to 0 pushes a `TopK(fetch=0)` below the limit and drops every row. Use `saturating_add`, as `combine_limit` already does for the analogous composition. Saturating is sound here rather than merely safe: `input_fetch` is an upper bound on the rows a child must supply, and `usize::MAX` is never a smaller bound than the true one. Claude-Session: https://claude.ai/code/session_01CtWAf4Fj5hp85a731KDdhz --- .../enforce_sorting/sort_pushdown.rs | 31 ++++++++++++++++++- 1 file changed, 30 insertions(+), 1 deletion(-) 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 b80fd16bc8b83..ce5091a8ea827 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 @@ -87,12 +87,18 @@ pub type SortPushDown = PlanContext; /// /// Note this is distinct from the fetch a parent imposes on `plan`'s *output* /// (`ParentRequirements::fetch`), for which `plan.fetch()` is the right bound. +/// +/// `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) -> Option { let fetch = plan.fetch()?; let skip = plan .downcast_ref::() .map_or(0, |limit| limit.skip()); - Some(fetch + skip) + Some(fetch.saturating_add(skip)) } /// Assigns the ordering requirement of the root node to the its children. @@ -1220,6 +1226,7 @@ mod tests { use datafusion_physical_expr::PhysicalExpr; use datafusion_physical_expr::expressions::{BinaryExpr, col}; use datafusion_physical_plan::empty::EmptyExec; + use datafusion_physical_plan::limit::GlobalLimitExec; const DESC: SortOptions = SortOptions { descending: true, @@ -1230,6 +1237,28 @@ mod tests { nulls_first: true, }; + #[test] + fn input_fetch_adds_skip_for_global_limit() { + let input = Arc::new(EmptyExec::new(child_schema())); + let limit: Arc = + Arc::new(GlobalLimitExec::new(input, 5, Some(10))); + + assert_eq!(input_fetch(&limit), Some(15)); + } + + /// `skip` and `fetch` are unrelated `usize`s, so `skip + fetch` can exceed + /// `usize::MAX`. Saturating keeps this at an (unreachable) upper bound + /// instead of panicking in debug builds or wrapping to a too-small fetch -- + /// wrapping to 0 would push down a `TopK(fetch=0)` and drop every row. + #[test] + fn input_fetch_saturates_instead_of_overflowing() { + let input = Arc::new(EmptyExec::new(child_schema())); + let limit: Arc = + Arc::new(GlobalLimitExec::new(input, usize::MAX, Some(1))); + + assert_eq!(input_fetch(&limit), Some(usize::MAX)); + } + /// Child (input) schema fed to the projections under test: `[a, b, c]`. fn child_schema() -> Arc { Arc::new(Schema::new(vec![ From d7686cd74fffdc434e4af77327c31d68e3dc2ae2 Mon Sep 17 00:00:00 2001 From: Jay Zhan Date: Mon, 7 Sep 2026 23:04:30 +0800 Subject: [PATCH 3/3] refactor: fold parent fetch into `input_fetch` so `skip` is added after the minimum The pushed-down fetch for a `GlobalLimitExec` child must be `min(fetch, parent_fetch) + skip`. The previous code computed `min(fetch + skip, parent_fetch)`, which is too small whenever the parent fetch is the tighter bound. It only happened to be correct because `pushdown_requirement_to_children` refuses to push a parent fetch through `GlobalLimitExec`, leaving `parent_fetch` as `None` at that point. Make `input_fetch` take `parent_fetch`, apply the minimum first and add `skip` afterwards, so the arithmetic is correct regardless of that invariant. Add unit tests for the combined cases. --- .../enforce_sorting/sort_pushdown.rs | 59 +++++++++++++++---- 1 file changed, 47 insertions(+), 12 deletions(-) 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 ce5091a8ea827..d7f556b90d9fe 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 @@ -80,21 +80,31 @@ impl Default for ParentRequirements { pub type SortPushDown = PlanContext; /// Number of input rows `plan` needs from its children in order to produce -/// its own `fetch` rows. This is `plan.fetch()` for every operator except -/// [`GlobalLimitExec`], which discards `skip` rows first and therefore needs -/// `skip + fetch` input rows. Using the bare `fetch` there would turn -/// `LIMIT 10 OFFSET 5` into `TopK(10)` below the limit, i.e. 5 result rows. +/// the rows its parent will consume. +/// +/// `parent_fetch` is the fetch a parent imposes on `plan`'s *output* +/// (`ParentRequirements::fetch`). It bounds `plan.fetch()` directly because +/// both count output rows, so the effective output fetch is +/// `min(plan.fetch(), parent_fetch)`. /// -/// Note this is distinct from the fetch a parent imposes on `plan`'s *output* -/// (`ParentRequirements::fetch`), for which `plan.fetch()` is the right bound. +/// The input fetch is that output fetch for every operator except +/// [`GlobalLimitExec`], which discards `skip` rows first and therefore needs +/// `skip` more input rows. `skip` must be added *after* taking the minimum: +/// `min(fetch + skip, parent_fetch)` would be too small whenever the parent +/// 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) -> Option { - let fetch = plan.fetch()?; +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()); @@ -386,7 +396,7 @@ fn pushdown_sorts_helper( // For operators that can take a sort pushdown, continue with updated // requirements. If this node already outputs single partition (e.g. SPM), // don't push SinglePartition to children. - let current_fetch = input_fetch(&sort_push_down.plan); + let current_fetch = input_fetch(&sort_push_down.plan, parent_fetch); let dists = sort_push_down .plan .input_distribution_requirements() @@ -401,7 +411,7 @@ fn pushdown_sorts_helper( sort_push_down.children.iter_mut().zip(adjusted).enumerate() { child.data.ordering_requirement = order; - child.data.fetch = min_fetch(current_fetch, parent_fetch); + child.data.fetch = current_fetch; child.data.distribution_requirement = stronger_distribution( &effective_dist, dists @@ -1243,7 +1253,32 @@ mod tests { let limit: Arc = Arc::new(GlobalLimitExec::new(input, 5, Some(10))); - assert_eq!(input_fetch(&limit), Some(15)); + assert_eq!(input_fetch(&limit, None), Some(15)); + } + + /// A parent fetch bounds the limit's *output*, so it is applied before + /// `skip` is added: `min(10, 3) + 5 = 8`, not `min(10 + 5, 3) = 3`. + #[test] + fn input_fetch_applies_parent_fetch_before_adding_skip() { + let input = Arc::new(EmptyExec::new(child_schema())); + let limit: Arc = + Arc::new(GlobalLimitExec::new(input, 5, Some(10))); + + assert_eq!(input_fetch(&limit, Some(3)), Some(8)); + // A looser parent fetch changes nothing. + assert_eq!(input_fetch(&limit, Some(20)), Some(15)); + } + + /// `OFFSET` without `LIMIT` has no fetch of its own, but a parent fetch + /// still needs `skip` extra input rows. + #[test] + fn input_fetch_adds_skip_to_parent_fetch_without_own_fetch() { + let input = Arc::new(EmptyExec::new(child_schema())); + let limit: Arc = + Arc::new(GlobalLimitExec::new(input, 5, None)); + + assert_eq!(input_fetch(&limit, None), None); + assert_eq!(input_fetch(&limit, Some(3)), Some(8)); } /// `skip` and `fetch` are unrelated `usize`s, so `skip + fetch` can exceed @@ -1256,7 +1291,7 @@ mod tests { let limit: Arc = Arc::new(GlobalLimitExec::new(input, usize::MAX, Some(1))); - assert_eq!(input_fetch(&limit), Some(usize::MAX)); + assert_eq!(input_fetch(&limit, None), Some(usize::MAX)); } /// Child (input) schema fed to the projections under test: `[a, b, c]`.