fix: split a leaf projection stranded above a limit by a pushed-down OFFSET by cool9850311 · Pull Request #27395 · apache/datafusion

cool9850311 · 2026-11-13T02:09:20Z

Which issue does this PR close?

Part of #27131. Closes the OFFSET row of that issue's table; the two Window
shapes are left for follow-ups.

Rationale for this change

#27152 fixed the RepartitionExec path. The OFFSET row of the issue's own
table still fails, and for a different reason:

SELECT file_row_index() + 1 FROM big OFFSET 5 LIMIT 3   -- still errors on main

There is no RepartitionExec in these plans, and none of the issue's three
controls (batch_size above the row count, target_partitions = 1, a table
smaller than batch_size) changes the outcome. It reproduces on a 5-row table.

Measured rule by rule:

the logical leaf rules do their job — the extracted leaf projection is placed
directly on the TableScan;
a source that supports skip pushdown then wraps its scan in a
GlobalLimitExec for the skip (catalog-listing/src/table.rs:782), which
lands between the scan and that projection. Judging from the gate at
push_down_limit.rs:112 (skip > 0 && scan.source.supports_skip_pushdown())
this arrived with feat: Push down offset/skip to TableScan #25404, though I have not built that commit to confirm it;
when the scan has more than one file group, EnsureRequirements adds a
CoalescePartitionsExec under the limit, so there are two nodes in the
way, not one;
GlobalLimitExec has no try_swapping_with_projection, so the first
ProjectionPushdown cannot get the projection back down;
LimitPushdown removes the now-redundant outer skip=0 limit, leaving the
two projections adjacent, and the second ProjectionPushdown merges them —
hoisting the leaf back above the limit, where file_row_index() is evaluated
directly and errors out, and get_field reads the whole struct.

Base plans (EXPLAIN, indent format):

ProjectionExec: expr=[file_row_index() as r]
  GlobalLimitExec: skip=1, fetch=3
    DataSourceExec: ..., limit=4, file_type=parquet        <-- no rewrite

ProjectionExec: expr=[get_field(s@0, value) as v]
  GlobalLimitExec: skip=1, fetch=3
    CoalescePartitionsExec: fetch=4
      DataSourceExec: ..., projection=[s], limit=4         <-- whole struct

The bare marker alone in its projection fails too, so this is not the issue's
"a KeepInPlace expression shares the projection" condition.

Cost, measured once by hand on the issue's own 20k-row struct table: a bare
s['a'] with an OFFSET 5 reads 178.2 KB on main and 67.6 KB with
this change (2.6x). That is a one-off reading, not an assertion -- what the
tests pin is the projection the scan is left with, which is the thing a plan
can show.

What changes are included in this PR?

try_split_leaf_projection already splits a projection stranded above a
RepartitionExec, pushing only the leaf half down. This is the same shape, so
the change generalises the host from one hardcoded node to a chain of them:
split_host_child accepts a round-robin RepartitionExec, a GlobalLimitExec
or a CoalescePartitionsExec, and split_host_chain walks as many as are
stacked. A single OFFSET on a multi-file table already stacks two, so stopping
at the first would leave that shape broken.

Each arm evaluates no expression of its own, has one child, and passes its
child's schema through — the last property is what lets the outer expressions
keep their column indices across several hosts. It holds by construction, not
by luck: none of the three compute_properties takes a schema, and each clones
its input's equivalence properties, so a host's schema is its child's. A fourth
arm that projects would have to rebase the outer expressions itself, and the
rule sets schema_check, so every unit test compares the plan's schema before
and after and would fail.

Splitting, rather than letting the whole projection swap past the limit, is what
keeps this honest: the rows the limit skips pay only for a field access the
source was going to make anyway, never for the outer expressions.

ProjectionExec: expr=[__datafusion_leaf_0@0 as v, upper(__datafusion_leaf_1@1) as l]
  GlobalLimitExec: skip=1, fetch=3
    DataSourceExec: ..., projection=[get_field(s@1, value) as __datafusion_leaf_0,
                                     get_field(s@1, label) as __datafusion_leaf_1], limit=4

upper(...) stays above the limit; the scan reads two fields instead of the
struct. Every shape without a leaf expression is left byte-identical, including
OFFSET 0, OFFSET with no LIMIT, a projection of only computed expressions,
and a pure column projection.

LocalLimitExec is not a host: no shape in the test corpus strands a leaf
expression above one, because without a skip the source absorbs the projection
before any limit is inserted. There is a test pinning that it stays out.

Are these changes tested?

Yes. Every OFFSET in the new tests sets a skip that differs from the fetch:
the split rebuilds the limit, and with the two numbers equal every assertion
would hold just as well if they were swapped.

The unit tests use OptimizationTest, the harness this test directory already
has, so each one shows the input plan beside the output and, because the rule
sets schema_check, compares the plan's schema before and after.

test	kind	pins
5 value queries, `file_row_index.slt`	behaviour	bare marker, wrapped marker, marker + computed sibling, `input_file_name()`, `arrow_typeof(marker)`. All five error on main
3 `EXPLAIN`s, `file_row_index.slt`	plan	the rewritten marker in the scan; `OFFSET 0` and `OFFSET` with no `LIMIT` as controls that never depended on the fix
2 value + 4 `EXPLAIN`s, `projection_pushdown.slt`	behaviour + plan	pruning for a bare leaf, a leaf + computed sibling, a leaf inside `upper(...)`, and the no-leaf control
multi-file section in both `.slt` files	plan + count	the `GlobalLimitExec` over `CoalescePartitionsExec` shape. Counted rather than listed, because which rows survive the skip depends on partition merge order; `count(r)` not `count(*)`, so the marker is still evaluated
`split_descends_through_global_limit`	unit	the split commits, both of the limit's numbers survive in the right order
`split_descends_through_limit_and_coalesce`	unit	the chain descends past two hosts
`split_descends_through_global_limit_without_fetch`	unit	`fetch = None`, which plain SQL does not reach
`split_declines_global_limit_without_leaf`	unit	no leaf, no split
`split_declines_local_limit`	unit	`LocalLimitExec` stays out of the host set
`split_declines_when_the_child_refuses_the_projection`	unit	a source that will not absorb the lower projection means no split
`split_keeps_the_coalesce_fetch`	unit	a `CoalescePartitionsExec` host keeps its own `fetch` across the rebuild. The `fetch` in the `.slt` plans is added by `LimitPushdown` after the split, so it does not cover this
`split_descends_through_three_hosts`	unit	three hosts, three distinct pairs of numbers, all in the right places
`split_commits_at_a_pass_through_bottom`	unit	`CooperativeExec` and `BufferExec` are not hosts, but as the chain bottom they delegate the swap to their input, so the split commits and the scan narrows. `CoalesceBatchesExec` behaves the same way, but it is deprecated in 52.0.0 and nothing constructs it, so it is not pinned
`split_below_a_filter_narrows_on_the_second_run`	unit	a `FilterExec` bottom takes the lower projection by holding it over its own input, so the scan narrows only on this rule's second scheduled run
`batch_size = 2` pair, `projection_pushdown.slt`	plan + behaviour	a round-robin `RepartitionExec` landing above the limit, so the split acts on the leaf projection underneath it. The one new query whose plan contains a `RepartitionExec`
marker inside a lambda under `OFFSET`, `projection_pushdown.slt`	behaviour	the lambda queries above it have no limit and the limit queries have no lambda; this is also what pins the parameter rebase
`input_file_name()` under `OFFSET` on json, `file_row_index.slt`	behaviour	the defect is not parquet-specific (`supports_skip_pushdown` is unconditional), and every other query in the file is parquet

Every new .slt query but one has no RepartitionExec in its plan, so they
exercise this path and not #27152's. The exception is the batch_size = 2 pair
at the end of the projection_pushdown.slt section, which lowers the batch size
on purpose to put a round-robin repartition above the limit and cover the two
together.

Teeth check — each mutation applied on its own against this branch's head,
then cargo test -p datafusion --test core_integration projection_pushdown (41
tests) and both touched .slt files. Counts are of failing tests and failing
.slt records.

Seven of the unit tests expect a split to commit beneath a GlobalLimitExec;
the table calls them "the seven".

mutation	killed
host set loses `GlobalLimitExec`	the seven, and 16 `.slt` queries: 9 of the 11 in `file_row_index.slt`, where only the two controls survive, and 7 of the 12 in `projection_pushdown.slt`, where the value queries still return the right rows and only the plans change
host set loses `CoalescePartitionsExec`	`split_descends_through_limit_and_coalesce` and `split_keeps_the_coalesce_fetch`, and the 4 multi-file plan assertions
chain walks one host instead of many	those two plus `split_descends_through_three_hosts`, and the same 4 queries
`skip` and `fetch` swapped on the rebuilt limit	the seven, and 20 `.slt` queries -- the widest of any mutation here
`skip` dropped	the seven, and 15 `.slt` queries
`fetch` dropped	six of the seven and nothing else; `split_descends_through_global_limit_without_fetch` already has `fetch = None`, and no `.slt` query notices, plan or value
drop the `leaves.is_empty()` guard	`split_declines_global_limit_without_leaf` and `split_below_a_filter_narrows_on_the_second_run`, whose inner projection has no leaf of its own. No `.slt` query
host set also accepts `LocalLimitExec`	`split_declines_local_limit` only
commit the split even when the child refuses	`split_declines_when_the_child_refuses_the_projection` and the memory-source `EXPLAIN` already in `projection_pushdown.slt`
`ChildrenPropertiesMode::Recompute` -> `Keep`	eight unit tests, `split_keeps_field_metadata` among them, and 21 `projection_pushdown.slt` records, including repartition plans that were already there
a `CoalescePartitionsExec` host rebuilt without its `fetch`	`split_keeps_the_coalesce_fetch` only -- nothing else notices

One semantic note: a leaf expression that moves below the limit is evaluated on
the rows the OFFSET skips. That is not new here — without a limit the logical
rules already push leaf expressions to the source — but a user function that
declares MoveTowardsLeafNodes and errors on some rows will now error on
skipped rows.

Are there any user-facing changes?

Queries that combined a leaf expression with OFFSET and errored now return
rows; ones that silently read a whole struct column now prune to the fields they
use. No API change.

🤖 Generated with Claude Code

…OFFSET A source that supports skip pushdown wraps its scan in a `GlobalLimitExec` for the skip, and when the scan has more than one file group `EnsureRequirements` adds a `CoalescePartitionsExec` under that. Either way the nodes land between the scan and the leaf projection the logical rules placed on it. Neither lets that projection swap back down, so `LimitPushdown` leaves two projections adjacent and the next `ProjectionPushdown` merges them, hoisting the leaf expression back above the limit: `file_row_index()` then errors at execution and `get_field` reads the whole struct. `try_split_leaf_projection` already handles the same shape above a `RepartitionExec`. Generalise its host from one hardcoded node to a chain: `split_host_child` accepts a round-robin `RepartitionExec`, a `GlobalLimitExec` or a `CoalescePartitionsExec`, and `split_host_chain` walks as many as are stacked. A single `OFFSET` on a multi-file table already stacks two, so stopping at the first would leave that shape broken. Each arm evaluates no expression of its own, has one child, and passes its child's schema through. That last property is what lets the outer expressions keep their column indices across several hosts, and it holds by construction rather than by luck: none of the three `compute_properties` takes a schema, and each clones its input's equivalence properties, so a host's schema is its child's. A fourth arm that projects would have to re-establish that, and nothing here will catch it; the doc comment says so. Splitting rather than swapping the whole projection past the limit is what keeps the cost right: only the leaf half descends, so the rows the limit skips pay for a field access the source was going to make anyway and never for the outer expressions. Part of apache#27131 Co-Authored-By: Claude Opus 5 <[email protected]>

…imit Every `OFFSET` in the new sections sets a skip that differs from the fetch, never `OFFSET n LIMIT n`: the split rebuilds the limit, and with the two numbers equal every assertion would hold just as well if they were swapped. `file_row_index.slt` gets five value queries for the marker UDFs, all of which error before the fix, an `EXPLAIN` for the rewritten plan, and two controls that never depended on it: `OFFSET 0`, and `OFFSET` with no `LIMIT`, where the skip is not pushed into the scan at all. It also gets a json table: `ListingTable::supports_skip_pushdown` is unconditional and the `GlobalLimitExec` wrap only looks at `skip`, so the defect is not specific to parquet, and every other query in the file is. `file_row_index()` itself is parquet-only, so that section uses `input_file_name()`. `projection_pushdown.slt` gets two value queries and four `EXPLAIN`s covering a bare leaf, a leaf beside a computed sibling, a leaf nested inside `upper(...)`, and the no-leaf case. The value queries pass before the fix on purpose: they pin the rows, which is the one thing a plan assertion cannot show. A dropped or swapped `skip` turns the plan text red as well, so they are not the only cover for it. The no-leaf `EXPLAIN` is labelled a control rather than a guard, because the scan prunes to `[id]` whether or not the rule bails out early; the unit test is what pins the bail-out. A further value query puts the marker inside a lambda body under an `OFFSET`: the lambda queries above it have no limit and the limit queries have no lambda, so nothing covered both, and it is also what pins the parameter rebase. Both files get a section with two file groups, which is the shape that stacks a `CoalescePartitionsExec` under the limit. Those count the surviving rows instead of listing them, because which rows survive the skip depends on the order the partitions are merged in, and `count(r)` rather than `count(*)`, so `optimize_projections` cannot drop the marker and leave nothing to evaluate. Neither new section contains a `RepartitionExec`, so both exercise this path rather than the one apache#27152 fixed. Ten unit tests hand-build the tree, because plain SQL does not reach these shapes: the split committing with both of the limit's numbers intact, the chain descending past two hosts and past three with a distinct pair of numbers on each, a `CoalescePartitionsExec` host keeping its own `fetch`, a `fetch` of `None`, a projection with no leaf expression left alone, `LocalLimitExec` kept out of the host set, a source that refuses the lower projection, the three nodes that meet the host conditions but are left out of the allowlist appearing instead as the chain bottom, and a `FilterExec` bottom, which takes the lower projection without absorbing it so the scan narrows only on this rule's second scheduled run. Part of apache#27131 Co-Authored-By: Claude Opus 5 <[email protected]>

jayzhan211

@cool9850311 thanks for following up on the OFFSET row! Looks good to me. The three inline comments are optional.

Co-authored-by: Jay Zhan <[email protected]>

alamb

Thanks @cool9850311 and @jayzhan211 -- the code looks ok but I think the comments make this really hard to understand. Can you please cut them down (by reading all comments, and removing anything you don't understand yourself)

I also have some ideas on how to improve the tests

Thank you for the contribution, and sorry for the request, I just think we need to fight to keep DataFusion understandable

alamb · 2026-11-14T13:10:40Z

+#[tokio::test]
+async fn split_descends_through_global_limit() -> Result<()> {
+    let (schema, scan) = limit_split_scan()?;
+    // What a source that pushes `OFFSET` down builds: the skip becomes a


I don't understand these comments -- they are trying to describe in english the plan shape

I think these tests would be much more readable if instead of trying to describe this in textual comments, you instead add a snapshot of the orignal and the optimized plan

Sort of like

assert_snapshot!(input_and_optimized_plan_text(projection)?, @r" --- Input Plan ... --- OutputPlan ProjectionExec: expr=[__datafusion_leaf_0@1 as v, id@0 + 1 as w] GlobalLimitExec: skip=1, fetch=3 DataSourceExec: file_groups={1 group: [[x]]}, projection=[id, get_field(s@1, value) as __datafusion_leaf_0], file_type=parquet "

More background on the rationale here: https://andrew.nerdnetworks.org/blog/2026/11/13/transcript-testing/

alamb · 2026-11-14T13:11:40Z

-        // a plain column, and is then no narrower than the widened child, or it
-        // narrows and holds a computed one. `RepartitionExec` declines both, it
-        // is not removable because it renames, and nothing else moved.
+        // stuck above a chain of the nodes `split_host_child` accepts. The


Can you please update these comments to not refer to implementation details elsewere -- pelase try and rephrase in your own words

alamb · 2026-11-14T13:12:24Z

+/// changed the width would leave both against the wrong one.
+fn split_host_child(node: &Arc<dyn ExecutionPlan>) -> Option<&Arc<dyn ExecutionPlan>> {
+    if let Some(repartition) = node.downcast_ref::<RepartitionExec>() {
+        // Hash and Range partitioning reference columns by index, which would


likewise, these comments is far too verbise -- please rewrite these in your own words rather than leaving this re-explanation of the code in

alamb · 2026-11-14T13:12:43Z

-/// expression sharing it never reaches the source. Splitting leaves the
-/// computation above the repartition, still parallelised.
+/// Two nodes strand a `MoveTowardsLeafNodes` expression above the source this
+/// way, and neither lets the whole projection past:


please reword

alamb · 2026-11-14T13:13:24Z

@@ -314,6 +314,191 @@ SELECT file_row_index() + 1, column1 FROM parquet_table ORDER BY column1
 # `RESET` restores the global default rather than the harness value, which the
 # harness rejects as leftover configuration.

+# A leaf expression must also reach the scan when the source pushes an `OFFSET`


can you please reword these comments to explain the case they are teseting, not details about the code that implements it?

cool9850311 · 2026-11-15T00:28:31Z

Thanks @cool9850311 and @jayzhan211 -- the code looks ok but I think the comments make this really hard to understand. Can you please cut them down (by reading all comments, and removing anything you don't understand yourself)

I also have some ideas on how to improve the tests

Thank you for the contribution, and sorry for the request, I just think we need to fight to keep DataFusion understandable

Sorry for not paying attention on readability of comments and tests, will work on this.

…ments down Addresses the review on the tests and on the comments. Tests: the ten split tests asserted only the optimized plan, through a local `optimized_plan_text` helper. They now use `OptimizationTest`, which this test directory already has (`pushdown_sort.rs` uses it) and which renders the input plan and the output side by side, so a reader sees what the rule changed without opening the test body. It also compares the plan's schema before and after whenever the rule's `schema_check` is true, which `ProjectionPushdown` sets, so each of the ten now carries a schema assertion as well. Two tests needed restructuring to fit: the pass-through-bottom case was a loop over two node types, which inline snapshots cannot be, and is now two sequential assertions; the filter case takes one snapshot per scheduled run of the rule. The same change removes the inline snapshot formatting that the earlier review round asked about. Comments: the first pass measured the surroundings but did not treat the measurement as a ceiling. The second pass asks of every block whether it can be removed rather than whether it can be shortened, keeping one only if a reader would otherwise misread what the code does, or would make a change that breaks something no test catches. What went from the rule: the narration of `optimize()`'s two passes, which the two lines say already; the `split_host_child` and `split_host_chain` doc comments, since the schema pass-through is enforced by `schema_check` and the multi-host chain is stated by `split_descends_through_limit_and_coalesce`; the allowlist note, which `split_declines_local_limit` now carries alone; the `SplitHostChain` type and `child` field docs; the termination guard's note, which claimed non-termination the suite does not demonstrate; and the paragraph on what strands a leaf expression, which is context for the pull request rather than for the function. What stayed is the five facts not inferable from the code: why only round-robin partitioning qualifies, why `CoalescePartitionsExec` is in the chain, that `hosts` runs outermost first, that a `Some` from `try_swapping_with_projection` does not mean the source absorbed the projection, and why the hosts' properties are recomputed. In the tests and the `.slt` files the helper docs and the sentences restating a neighbouring query went; the ones that keep a test from looking redundant, or that explain a deliberately weak assertion, stayed. The termination guard's comment was also the one the review asked to reword. Measuring it first showed the claim was wrong: removing the guard does not hang the suite, it changes one test's output, so the comment went rather than being reworded. Rule: 78 comment lines down to 12, hunk density 64.5% to 21.8%, whole-file 11.9% to 9.4% against a 10.4% baseline, and the longest block this branch adds down from 15 lines to 4 (the file's own maximum is 7). Tests 17.7% to 6.0%, projection_pushdown.slt 18.1% to 10.4%, file_row_index.slt 22.6% to 14.4%. cargo fmt --check clean, clippy clean on datafusion-physical-optimizer --all-targets and on the core_integration test target, 41 projection pushdown tests and 51 physical-optimizer unit tests passing, both touched .slt files green, and the full sqllogictest suite green before the comment pass. Co-Authored-By: Claude Opus 5 <[email protected]>

cool9850311 and others added 2 commits November 12, 2026 23:20

github-actions Bot added optimizer Optimizer rules core Core DataFusion crate sqllogictest SQL Logic Tests (.slt) labels Nov 13, 2026

jayzhan211 approved these changes Nov 14, 2026

View reviewed changes

Comment thread datafusion/core/tests/physical_optimizer/projection_pushdown.rs Resolved

Comment thread datafusion/sqllogictest/test_files/projection_pushdown.slt Resolved

Comment thread datafusion/physical-optimizer/src/projection_pushdown.rs Resolved

Apply batched suggestions from code review

7ff2ba1

Co-authored-by: Jay Zhan <[email protected]>

alamb requested changes Nov 14, 2026

View reviewed changes

Navigation Menu

Repository navigation

fix: split a leaf projection stranded above a limit by a pushed-down OFFSET - #27395

fix: split a leaf projection stranded above a limit by a pushed-down OFFSET#27395
cool9850311 wants to merge 4 commits into
apache:mainfrom
cool9850311:fix-27131-leaf-projection-past-limit

cool9850311 commented Nov 13, 2026 •

edited

Loading

Uh oh!

jayzhan211 left a comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

alamb left a comment

Uh oh!

alamb Nov 14, 2026

Uh oh!

alamb Nov 14, 2026

Uh oh!

alamb Nov 14, 2026

Uh oh!

alamb Nov 14, 2026

Uh oh!

alamb Nov 14, 2026

Uh oh!

cool9850311 commented Nov 15, 2026

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

Conversation

cool9850311 commented Nov 13, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Which issue does this PR close?

Rationale for this change

What changes are included in this PR?

Are these changes tested?

Are there any user-facing changes?

Uh oh!

jayzhan211 left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

alamb left a comment

Choose a reason for hiding this comment

Uh oh!

alamb Nov 14, 2026

Choose a reason for hiding this comment

Uh oh!

alamb Nov 14, 2026

Choose a reason for hiding this comment

Uh oh!

alamb Nov 14, 2026

Choose a reason for hiding this comment

Uh oh!

alamb Nov 14, 2026

Choose a reason for hiding this comment

Uh oh!

alamb Nov 14, 2026

Choose a reason for hiding this comment

Uh oh!

cool9850311 commented Nov 15, 2026

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

cool9850311 commented Nov 13, 2026 •

edited

Loading