cool9850311 opened a new pull request, #27395:
URL: https://github.com/apache/datafusion/pull/27395
## 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:
```sql
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:
1. the logical leaf rules do their job — the extracted leaf projection is
placed
directly on the `TableScan`;
2. 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 #25404, though I have not built that commit to confirm
it;
3. 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;
4. `GlobalLimitExec` has no `try_swapping_with_projection`, so the first
`ProjectionPushdown` cannot get the projection back down;
5. `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 re-establish that, and nothing in the
function
will catch it; the doc comment says so.
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 is `OFFSET 1 LIMIT 3` rather than
`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.
| 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` meet the host conditions but are left out of the allowlist; 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 |
| 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 |
The new `.slt` sections were checked to contain no `RepartitionExec`, so they
exercise this path and not #27152's.
Teeth check — each mutation applied on its own, with what it killed:
| mutation | killed |
|---|---|
| host set loses `GlobalLimitExec` | 3 unit tests + all 13 new `OFFSET`
queries |
| host set loses `CoalescePartitionsExec` |
`split_descends_through_limit_and_coalesce` + the 4 multi-file queries |
| chain walks one host instead of many | the same set |
| `skip` and `fetch` swapped on the rebuilt limit | 3 unit tests + 16
queries |
| `skip` dropped | the unit tests, the value queries and two `EXPLAIN`s |
| `fetch` dropped | the unit tests only -- no `.slt` query notices, plan or
value |
| drop the `leaves.is_empty()` guard |
`split_declines_global_limit_without_leaf` |
| host set also accepts `LocalLimitExec` | `split_declines_local_limit` |
| commit the split even when the child refuses |
`split_declines_when_the_child_refuses_the_projection` |
| `ChildrenPropertiesMode::Recompute` -> `Keep` | the unit tests,
`split_keeps_field_metadata`, and the existing repartition `.slt` plans |
| 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](https://claude.com/claude-code)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
