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

Describe the bug

An expression whose placement is ExpressionPlacement::MoveTowardsLeafNodes is meant to reach the file scan, where the source either supplies it as a column or prunes to it. It never gets there if any other expression in the same projection has KeepInPlace placement and the scan's row estimate exceeds batch_size.

The MoveTowardsLeafNodes expression does not have to be wrapped in anything — a bare call with an unrelated computed column beside it is enough:

SELECT file_row_index(), column1 + 1 FROM big;   -- Execution error, and the marker is bare

All three functions that declare this placement are affected, in two symptom classes:

Execution error: file_row_index() is source dependent and cannot be evaluated directly

Mechanism

ProjectionExec::benefits_from_input_partitioning (physical-plan/src/projection.rs:354) is true when any expression in the projection is KeepInPlace:

fn benefits_from_input_partitioning(&self) -> Vec<bool> {
    let all_simple_exprs = self.projector.projection().as_ref().iter().all(|proj_expr| {
        !matches!(proj_expr.expr.placement(), ExpressionPlacement::KeepInPlace)
    });
    vec![!all_simple_exprs]
}

That one signal then does two opposing things:

1. EnsureRequirements inserts a RepartitionExec beneath the projection — gated on the row estimate exceeding batch_size (ensure_requirements/enforce_distribution.rs:1167).
2. RepartitionExec::try_swapping_with_projection (physical-plan/src/repartition/mod.rs:1949) refuses to push that same projection back below the repartition, because of the same flag:

if projection.benefits_from_input_partitioning()[0] || !all_columns(projection.expr()) {
    return Ok(None);
}

So a projection that mixes a leaf expression with any computed expression is simultaneously judged worth parallelising and ineligible to descend. The leaf expression can never reach the scan. For the two marker UDFs the rewrite in datasource-parquet/src/source.rs:710 therefore never runs and the placeholder is evaluated directly; for get_field the source just reads the whole struct.

Note the source itself is perfectly capable of absorbing these projections, including wrapped forms — file_row_index.slt already pins a plan where file_row_index() + 1 is folded into DataSourceExec on a table small enough that no repartition is inserted. The gap is only that the projection cannot get past an interposed RepartitionExec.

To Reproduce

In datafusion-cli, stock configuration:

COPY (SELECT value AS column1 FROM range(0, 20000))
TO '/tmp/df_leaf_repro/big.parquet' STORED AS PARQUET;

CREATE EXTERNAL TABLE big
STORED AS PARQUET LOCATION '/tmp/df_leaf_repro/big.parquet';

SELECT file_row_index(), column1 + 1 FROM big;   -- Execution error

The trigger is the projection's expression mix, not the shape of the marker call. Each row below was measured, and the presence of a RepartitionExec tracks the failure exactly:

A plain column or a literal is Column / Literal placement, so it keeps the projection "simple" and nothing is inserted. column1 + 1 and abs(column1) are KeepInPlace, so they flip the whole projection — even though they have nothing to do with the marker.

input_file_name() behaves identically (input_file_name() || 'x', upper(input_file_name()), and a bare call beside a computed column all error).

get_field(): same cause, silent performance regression

COPY (SELECT named_struct('a', value, 'b', value * 2, 'c', CAST(value AS VARCHAR)) AS s
      FROM range(0, 20000)) TO '/tmp/df_leaf_repro/st.parquet' STORED AS PARQUET;

CREATE EXTERNAL TABLE t STORED AS PARQUET LOCATION '/tmp/df_leaf_repro/st.parquet';

No error, just reading an entire struct column to use one scalar field — and nested-column pruning is the headline feature of leaf expression pushdown.

Controls

Each of these removes the RepartitionExec and restores correct behaviour on the same table and data:

Raising batch_size above the row count changes only the threshold at enforce_distribution.rs:1167, which is what distinguishes cause from correlation.

datafusion.execution.parquet.pushdown_filters makes no difference, on or off. EXPLAIN on the failing query succeeds; the error appears only once execution starts.

Other node types

The leaf rules extract MoveTowardsLeafNodes sub-expressions from an allowlist of node types (extract_leaf_expressions.rs:259), which is why the same expression mix is fine elsewhere:

In the Filter case the marker is extracted into its own __datafusion_extracted_N projection, which contains no KeepInPlace expression and so is absorbed by the scan before any repartition matters.

Expected behavior

A leaf expression should reach the scan regardless of what else shares its projection, so that SELECT file_row_index(), column1 + 1 FROM big returns rows and SELECT s['a'], s['b'] + 1 FROM t prunes to the a and b leaves.

Failing that, the two marker UDFs should at least be rejected at plan time with an actionable message, rather than accepted, passed by EXPLAIN, and failed mid-execution.

Additional context

This is a concrete failure of goal G3 in #25459 ("Cheap leaf expressions (s['a'], file_row_index()) and filters reach the scan"), which that EPIC does not currently list.

Two fix directions:

1. Split the projection in the logical rules. Extend extraction so a projection that mixes leaf and KeepInPlace expressions is split, putting the leaf expressions in their own projection:

   Projection: __datafusion_extracted_1, column1 + 1
     Projection: file_row_index() AS __datafusion_extracted_1, column1
   

   The inner projection has no KeepInPlace expression, so nothing is inserted beneath it and the source absorbs it; the outer one only references columns. This also covers the bare-marker-with-computed-sibling case. Constraint: the allowlist comment excludes node types whose output schema depends on their expressions, and Projection is one of them, so the split must keep the outer projection's alias list and output field names identical.
2. Make the physical gate leaf-aware. Let RepartitionExec::try_swapping_with_projection accept a projection that contains leaf expressions, or avoid inserting a repartition between a scan and a projection that has them. Smaller, but all_columns / benefits_from_input_partitioning guard real concerns about duplicating expensive expressions across partitions.

Detection should key off ExpressionPlacement rather than per-function downcasts. #23638 added a hardcoded try_downcast_func chain for FileRowIndexFunc/InputFileNameFunc and noted at the time: "If we keep adding this sort of metadata functions, we might want a better API to detect them." That API now exists and all three functions declare it. See also #21306 (struct-aware optimizations hardcoded to GetFieldFunc).

Note for implementers: file_row_index() and input_file_name() are Volatility::Volatile, so no rule may evaluate them at a second site. The split above keeps exactly one evaluation. #25456 (open draft) adds a shared inlining policy with that rule and rewrites much of extract_leaf_expressions.rs, so a fix in direction 1 should be coordinated with it.

Related: #23531 / #23638 are the same "is source dependent and cannot be evaluated directly" error from a different pushdown gap (input_file_name() inside a Parquet RowFilter). file_row_index() was added in #22604, input_file_name() in #22978.

Two further gaps found while investigating, which I believe need their own issues:

1. Join is on the extraction allowlist but still fails. Extraction routes each extracted sub-expression to an input by checking which input's schema contains all of the expression's column references. A marker call has no column references, so it cannot be routed. Pre-extracting it in a subquery works:

   SELECT count(*) FROM (SELECT file_row_index() AS r, column1 FROM big) a
   JOIN big b ON a.column1 = b.column1 AND a.r >= 0;   -- ok

2. file_row_index() does not work on CSV or JSON at all. Even a bare call errors, independently of row count and target_partitions (verified on a 3-row file and with target_partitions = 1); the only production caller of the rewrite is the Parquet source. input_file_name() bare does work on CSV. So file_row_index() is effectively Parquet-only, which the documentation does not state.

Verified on c120619de (48.0.0-rc2-4901-gc120619de), datafusion-cli reporting 57.0.0-dev, macOS arm64, default target_partitions (= core count).

Willingness to contribute

I plan to contribute a fix for this bug independently.