Repository navigation
Fix PushdownSort dropping LIMIT when eliminating SortExec - #21744
Conversation
xudong963
left a comment
There was a problem hiding this comment.
LGTM, would be better if there are some slt tests
Also, do you want to include it into new minor release?
alamb
left a comment
There was a problem hiding this comment.
Thanks @sgrebnov and @martin-g and @xudong963
@xudong963 — There's an existing SLT test (sort_pushdown.slt Test 1.3) that already covers the Exact pushdown + LIMIT case where the source supports |
|
Thanks everyone. If it turns out we can make some slt tests perhaps we can add them as a follow on PR |
…pache#26660) ## Which issue does this PR close? - No issue. This is a refactor; query results and plans of DataFusion's own operators are unchanged. ## Rationale for this change Four optimizer rules move a `LIMIT`/`OFFSET` (or a fetch) through the physical plan: `LimitPushdown`, `LimitPushPastWindows`, the sort pushdown inside `EnsureRequirements`, and `PushdownSort`. Each rule answers the same three questions with its own code: - **How do `skip` and `fetch` combine?** Each rule adds `skip + fetch` itself. `LimitPushdown` adds them unchecked, while the sort pushdown in `EnsureRequirements` and `LimitPushPastWindows` saturate (the latter since apache#26662). - **Can a limit go below this node?** There are three different checks: - `supports_limit_pushdown()` alone. - The same plus `CardinalityEffect::Equal`. - The same plus `Equal | GreaterEqual`, with a hard-coded type list. They differ because a rule can *copy* a limit below a node as an early-stop hint while keeping it above, or *move* it below and drop it above. Moving needs more from the node than copying. Nothing in the code names that difference. - **What does `LimitPushdown` do at each node?** It removes every limit node and carries `(fetch, skip, satisfied)` down. It decides later whether to re-add the limit, and clears `skip` before recursing. The decision is spread over about 150 lines of nested `if`s. Several recent fixes came from one of these places disagreeing with another: apache#24958 (skip ignored when a sort moved below a limit), apache#22525 (OFFSET applied twice), apache#21744 and apache#24809 (fetch lost when an operator was removed). ## What changes are included in this PR? Three main commits, each of which builds and passes the tests on its own, followed by review fixes: 1. **`RowBound`** (crate-private, `physical-optimizer/src/row_bound.rs`). This is `OFFSET skip LIMIT fetch` as a value: combining two bounds, `skip + fetch` (saturating), the tighter of two fetches, and the one rule for choosing between a `GlobalLimitExec` and a `LocalLimitExec`. The sort pushdown in `EnsureRequirements` uses it instead of its own arithmetic. `LimitPushPastWindows` uses it to read limit nodes. 2. **`LimitPushdown` as one walk.** The bound a node receives from its ancestors is a small enum: - `Nothing`. - `Open(bound)`: no ancestor enforces it yet, so it must be placed at or above the first node it can't pass. - `Enforced(fetch)`: an ancestor enforces it, so nodes below may stop early. Each node is one of `Limit`, `Pass`, `Merge` or `Stop`. The rule is a single `match` over the two (`apply_bound` in `limit_pushdown.rs`), which replaces the `satisfied` flag and the skip-clearing step. `GlobalRequirements`, `pushdown_limit_helper` and `pushdown_limit_helper_with_stats` are deprecated; they now run on the new walk. 3. **`ExecutionPlan::limit_behavior()`.** Operators declare what a limit on their output means for their input: - `Stops`. - `Limit { skip, fetch }`. - `CopyToChildren` (Union). - `MoveToInput` (Projection). - `Merge` (CoalescePartitions, SortPreservingMerge). - `Window(lookahead)`. The copy/move difference is documented once, on the enum. So is the rule that a merge with its own fetch (`ExecutionPlan::fetch()`) already selects rows: a limit above it can't move below it. Before, that rule was only implied by the merge's `cardinality_effect()` returning `LowerEqual` when it has a fetch. The three rules match on it instead of combining `supports_limit_pushdown()` with `cardinality_effect()` and type checks. The window's lookahead (frame end plus `LEAD` offsets) is now computed by `BoundedWindowAggExec` rather than by the rule. `supports_limit_pushdown` is deprecated. The default `limit_behavior()` is derived from it, so operators outside DataFusion keep their current behavior. `LimitPushdown` moves a limit below a `CopyToChildren` or `MoveToInput` node only when each output partition is one input partition. A node whose children have more partitions in total than it outputs takes the limit like a merge: the limit stays above it and is copied below. No DataFusion operator is such a node, so this only fixes custom operators (see below). The fetch handling in `PushdownSort` and the other "a fetched operator is a row-selection boundary" checks in `EnsureRequirements` are unchanged. ## What is the testing strategy for this PR? The goal is no change in results or plans, so the existing tests are the check. All of these pass with **no change to expected output**: - every sqllogictest file - `core_integration` `physical_optimizer::` (663 tests: limit pushdown, sort pushdown, enforce sorting, window) - the `limit_fuzz`, `window_fuzz`, `sort_fuzz`, `sort_query_fuzz` and `topk_filter_pushdown` fuzz tests - `datafusion-physical-optimizer` and `datafusion-physical-plan` unit tests New tests: - `row_bound.rs` unit tests: combining bounds, saturation at `usize::MAX`, and the Global/Local choice. - `test_fetched_merge_keeps_stricter_topk_above` in `core/tests/physical_optimizer/ensure_requirements.rs`: a TopK over a `SortPreservingMergeExec` that has its own fetch keeps its fetch above the merge. It passes on main, and fails if a fetched merge is treated as `MoveToInput`. - Three tests in `core/tests/physical_optimizer/limit_pushdown.rs` with a custom operator that combines partitions into one: - declared `Merge` but can't take a fetch: under `OFFSET 5 LIMIT 3`, its input gets `fetch=8`, not 3; - declared `CopyToChildren` with two inputs, or `MoveToInput` with one 8-partition input: the limit stays above it. Benchmarks: the plans are identical, so only planning time can change. Measured on Apple silicon against the merge base: - EXPLAIN output for LIMIT/OFFSET, window, UNION and TopK queries is identical, and planning them takes the same time (1.01x). - `sql_planner` `physical_plan_*` (TPC-H, TPC-DS, ClickBench; 84 benchmarks): median +0.7% on this branch, while re-running main against itself gave +3.2%, so the difference is within noise. TPC-H q3 and q4, the largest differences, were 3-15% faster on this branch when re-run. ## Are there any user-facing changes? No change to query results or plans for DataFusion's own operators. API (upgrade guide updated): - New `ExecutionPlan::limit_behavior()`, `LimitBehavior` and `WindowLookahead`. - `ExecutionPlan::supports_limit_pushdown()` is deprecated. Only the default `limit_behavior()` reads it. DataFusion's own operators keep returning the same values, so a wrapper that forwards it to its input keeps working. - `limit_pushdown::{GlobalRequirements, pushdown_limit_helper}` are deprecated. They now run on the new walk. A caller that drives them one node at a time no longer applies an `OFFSET` twice when a limit node with only an `OFFSET` sits above another limit. - `limit_pushdown::pushdown_limit_helper_with_stats` is removed. It was added in this release cycle (apache#26094) and is in no release. - Two differences for custom operators that still implement `supports_limit_pushdown()`: - A single-input operator that returns `true` and whose `cardinality_effect()` is `Unknown` or `LowerEqual` can now have a window's grown limit pushed below it. `Unknown` is the default, so this includes a wrapper that forwards `supports_limit_pushdown()` and does not override `cardinality_effect()`. `LimitPushdown` already pushes plain limits through such operators. - An operator that returns `true` and combines its input partitions (for example several inputs into one partition) now keeps a limit above it. On main, `LimitPushdown` moved the limit below it, which kept that many rows of each input partition. - `skip + fetch` now saturates at `usize::MAX` in `LimitPushdown`, as it already does in the other two rules. SQL can't reach the overflow there, because `LIMIT` and `OFFSET` are each at most `i64::MAX`.
Which issue does this PR close?
When
PushdownSortremoves aSortExecbecause a source returnsExact(guaranteeing ordering), anyfetch(LIMIT) on theSortExecis silently dropped if the underlying plan does not supportwith_fetch().For example,
ProjectionExecsupportstry_pushdown_sort(delegating to its child) but does not implementwith_fetch(). A plan likeSortExec(fetch=10) → ProjectionExec → sourcethat gets sort-eliminated loses the limit.What changes are included in this PR?
In the
Exactbranch ofPushdownSort, when the eliminatedSortExeccarried afetch:with_fetch()on the pushed-down source firstwith_fetch()returnsNone, fall back to wrapping withGlobalLimitExecAre these changes tested?
Yes. Three new unit tests:
test_sort_pushdown_exact_no_fetch_no_limit— Exact elimination without fetch: no limit wrapper addedtest_sort_pushdown_exact_preserves_fetch_with_global_limit— Exact elimination with fetch, source does NOT supportwith_fetch():GlobalLimitExecwrapper addedtest_sort_pushdown_exact_preserves_fetch_with_source_support— Exact elimination with fetch, source supportswith_fetch(): limit pushed into source directlyAre there any user-facing changes?
No.