[cool9850311]

Which issue does this PR close?

• Closes file_row_index()/input_file_name() error and get_field() loses pruning when their projection also contains a KeepInPlace expression #27131.

Rationale for this change

A MoveTowardsLeafNodes expression loses leaf pushdown as soon as it shares a
projection with a KeepInPlace expression and a RepartitionExec is inserted.
The two symptom classes are:

-- errors: "file_row_index() is source dependent and cannot be evaluated directly"
SELECT file_row_index(), id + 1 FROM t;

-- no error, but reads the whole struct instead of pruning to one field
SELECT s['value'], id + 1 FROM t;

ProjectionExec::benefits_from_input_partitioning is true as soon as any
expression in the projection is KeepInPlace (the default placement). That one
flag does two opposing things:

• EnsureRequirements inserts a RepartitionExec beneath the projection, and
• RepartitionExec::try_swapping_with_projection then refuses to let that same
  projection descend past it.

So the leaf expression can never reach the source. The leaf call does not have to
be wrapped in anything — an unrelated computed column in the same projection is
enough, because the flag is a property of the whole projection, not of one
expression. get_field is the headline case: nested column pruning is its
documented purpose and it degrades silently.

The trigger needs a RepartitionExec, so the table has to exceed batch_size
(EnsureRequirements uses batch_size as a row-count threshold). That is why
this has not shown up in the existing small-fixture tests.

What changes are included in this PR?

ProjectionPushdown now splits such a projection instead of giving up:

• the leaf expressions move into a lower projection that the source absorbs;
• the remaining computation stays above the repartition, where it is still
  parallelised.

Both guards keep their intent rather than one being traded for the other: the
scan only gains the cheap leaf expression, and the expensive computation is still
run in parallel.

Deliberate limits:

• The split is committed only when child.try_swapping_with_projection actually
  returns Some for the lower projection, i.e. when the source really does
  absorb it. Otherwise the plan is left untouched.
• Only Partitioning::RoundRobinBatch is handled. Hash and range partitioning
  reference columns by index, so changing the schema below them would require
  remapping their expressions; that is left out of this PR.
• Only the columns the outer expressions still reference outside the extracted
  leaves are passed through, so s['a'] + 1 does not end up reading the whole
  struct alongside the extracted field.

What is the testing strategy for this PR?

• datafusion/sqllogictest/test_files/file_row_index.slt — value-based
  regression tests for the error class: bare marker plus a computed sibling,
  + 1, arrow_typeof, two different markers mixed, a duplicated marker, and
  non-regression for Filter / Sort / Aggregate.
• datafusion/sqllogictest/test_files/projection_pushdown.slt — plan assertions
  for the get_field class. Value comparison can never catch that one: without
  pruning the results are still correct, only more data is read, so the test has
  to assert the pruned projection=[...] in the plan.
• datafusion/physical-optimizer unit tests: 50 passed.

Two traps are called out in comments in both test files, because either one makes
the tests pass against unfixed code:

1. the fixture must exceed batch_size, or no RepartitionExec is inserted and
   there is nothing to reproduce;
2. a bare LIMIT n is pushed into the scan and hides the bug again.

Verified by reverting the source file with git checkout main -- <file>: the new
tests then fail with 6 errors and 16 plan-diff lines.

Are there any user-facing changes?

No API changes. Queries that previously failed with "is source dependent and
cannot be evaluated directly" now succeed, and get_field regains nested column
pruning when it shares a projection with other expressions.

One plan-shape note for #25456 (inlining projection columns), which touches
adjacent code: its test pins file_row_index() used twice as not inlined. When
the split here fires, the two identical leaf calls are de-duplicated into a
single lower-projection column. Values are unaffected; only the plan shape
differs. Happy to rebase in whichever order is convenient.

🤖 Generated with Claude Code

[jayzhan211]

@cool9850311 thanks for tracking this down so thoroughly! The split fixes the reported errors and pruning, and scanned bytes drop from 178.2 KB to 67.6 KB in my s['a'] + 1 check. I left two inline comments on shapes that work on main but fail with this PR.

Two queries from the issue's tables still error with this PR, because a different node sits under the projection:

SELECT file_row_index() + 1 FROM big OFFSET 5 LIMIT 3;                                    -- GlobalLimitExec
SELECT file_row_index() AS r, sum(column1) OVER (ORDER BY column1) AS w FROM big;         -- BoundedWindowAggExec

These need different handling, so a follow-up is fine. Could you change Closes #27131 to Part of #27131 so the issue stays open for them?

[jayzhan211]

Lambda parameters are indexed after the input columns, so they go stale when the split widens the input. This query works on main but now fails with Field of physical LambdaVariable with index 0 doesn't match batch field:

SET datafusion.sql_parser.dialect = databricks;
SELECT array_transform(arr, x -> x + 1) AS a, s['value'] AS v, s['label'] AS l FROM t ORDER BY v;

This fixes it, with child_schema.fields().len() passed in as a new old_input_len parameter and let new_input_len = first_leaf + leaves.len(); added next to first_leaf. Could you add the query above to the SLT, plus a nested lambda case?

Suggested change

	Ok(Transformed::no(node))
	// Lambda parameters are appended after the input columns, so they move
	// when the split changes the input width.
	if let Some(variable) = node.downcast_ref::<LambdaVariable>()
	&& old_input_len != new_input_len
	{
	let rebased = LambdaVariable::new(
	variable.index() - old_input_len + new_input_len,
	Arc::clone(variable.field()),
	);
	return Ok(Transformed::yes(Arc::new(rebased) as Arc<dyn PhysicalExpr>));
	}
	Ok(Transformed::no(node))

[jayzhan211]

try_new drops metadata that the planner set through try_new_with_schema_metadata, so this select now fails with PhysicalOptimizer rule 'ProjectionPushdown' failed. Schema mismatch, while it works on main:

ctx.table("t").await?.select(vec![
    get_field(col("s"), "value").alias_with_metadata("v", Some(metadata.into())),
    (col("id") + lit(1i64)).alias("w"),
])?;

Could you apply this and add that select as a test, with batch_size below the row count so the repartition appears, asserting the output field keeps its metadata?

Suggested change

	Ok(Some(Arc::new(ProjectionExec::try_new(
	new_exprs,
	new_repartition,
	)?)))
	Ok(Some(Arc::new(
	ProjectionExec::try_new_with_schema_metadata(
	new_exprs,
	new_repartition,
	projection.schema().as_ref(),
	)?,
	)))