fix: let leaf expressions reach the scan past a RepartitionExec - #27152

…ection

Two regressions in the projection split, both reported in review. Each query
below works on `main` and failed with the split.

Lambda parameters are indexed after the input columns, so widening the input
leaves them pointing at the wrong column:

    SELECT array_transform(arr, x -> x + 1), s['value'], s['label'] FROM t;

`remap_filter_expr` in this file already rebases parameters for the join filter
pushdown, so the node-level shift moves into `rebase_lambda_variable` and both
callers use it rather than carrying two copies.

The symptom depends on where the stale index lands. Out of range it is dropped
and the query happens to be correct; in range it fails `LambdaVariable`'s field
check; in range with a matching name and type it reads the wrong column and
returns wrong values with no error. The last case is covered, since the field
check is a backstop rather than the guarantee.

`ProjectionExec::try_new` derives the output schema from the expressions alone
and drops the field metadata the planner attached through
`try_new_with_schema_metadata`, which the rule's own `schema_check` then
rejects:

    PhysicalOptimizer rule 'ProjectionPushdown' failed. Schema mismatch.

The split now rebuilds the projection with that constructor, taking the
metadata from the projection it replaces.

Tests. The value queries live in `projection_pushdown.slt`; the fixture gains
`arr` and `tags`, appended after `s` so the existing `get_field(s@1, ...)`
assertions do not move. A nested lambda needs two leaf expressions: with one
the split is width-neutral, the parameter does not move, and the query passes
against unfixed code. The metadata test is a Rust test because no SQL syntax
attaches field metadata; it asserts on the physical plan's schema, not the
DataFrame's, and uses Parquet because only a source that absorbs the leaf
projection makes the split fire.

Part of apache#27131

Co-Authored-By: Claude Opus 5 <[email protected]>

…fixes

Review follow-up. Nothing here changes what the split computes.

The split is only committed when the child really absorbs the leaf projection,
but no test held that: removing the check left all 545 SLT files, 50
physical-optimizer unit tests and 669 `core_integration` ones green. A memory
table refuses the lower projection, because `all_alias_free_columns` rejects an
aliased `get_field`, and that is also what any node that does not override
`try_swapping_with_projection` does. An `EXPLAIN` on such a table now pins the
plan; with the check removed it fails on the extra `ProjectionExec`.

Collecting each distinct leaf once was likewise unpinned, and the values cannot
show it: two calls to the same marker agree with each other whether they share a
column or not. The duplicated-marker case gains an `EXPLAIN`, which fails with a
second `__datafusion_leaf_1` in the scan projection if the de-duplication goes.

Three shapes the split fixes had no test either. Each reaches a different path:
a TopK sorts on a column the split produced and runs above the repartition with
`preserve_partitioning=[true]`; a marker inside a lambda body has to be found
through `LambdaExpr`; and CSV is a second scan implementation, while every other
test here uses Parquet. All three error with `is source dependent and cannot be
evaluated directly` without the split.

The cleanup walk this PR had added after the split is removed rather than made
conditional, because it cannot do anything. `remove_unnecessary_projections`
has three paths for a projection: it is not removable, since the split renames
and keeps the original output schema; `overrides_metadata` changes nothing; and
`RepartitionExec` declines it either for not narrowing, when every expression
stayed a plain column, or for holding a computed one, when an expression
consumed more than one leaf and the projection did narrow. Nothing else in the
tree moved, as the split preserves the projection's schema. Instrumenting the
walk confirmed it changed no plan in any of the 19 splits the suite fires.

Handling only `RoundRobinBatch` was unpinned the same way, and no SQL reaches
that shape: the guard is hit 1326 times across the suite, every one of them
round robin. The rule is a plain plan-to-plan function though, so the plan can
be built by hand, which `split_declines_hash_repartition` does, after
`ensure_requirements.rs` for the hash repartition and `test_utils.rs` for the
scan. The scan has to be Parquet rather than `EmptyExec`: only a source that
absorbs the leaf projection lets the split commit once the check is gone.

Also drops a comment reference to a trap number that only exists in the pull
request description, and rewrites the one on `collect_leaf_exprs` that used
`s['a']['b']` as its example: the logical optimizer folds that into a single
call, so the branch it describes never sees it. It now says what stopping at a
leaf avoids instead.

Part of apache#27131

Co-Authored-By: Claude Opus 5 <[email protected]>