fix: push filters above a pure extraction projection into the scan below it - #26281

Which issue does this PR close?

• Follow-up to fix: decide the precedence between PushDownFilter and PushDownLeafProjections #25455, which names this case as residual risk.

Rationale for this change

Since #25455, a filter on a partition column can stop reaching the scan. Partition pruning is then lost, and the scan reads every file.

This occurs when a filter reaches a pure extraction projection (one with only __datafusion_extracted_N aliases and pass-through columns) after the optimizer pass that created the projection. PushDownFilter no longer moves a filter below such a projection, so the predicate never gets into TableScan::filters. A common way to get this shape is to put a filter on an already optimized plan and optimize it again. Query rewrites that reuse an optimized subtree do this.

Reproduction

Data:

COPY (SELECT * FROM (VALUES
  ('2025-01-01', named_struct('a', 1, 'b', 'x')),
  ('2025-01-02', named_struct('a', 2, 'b', 'y')),
  ('2025-01-03', named_struct('a', 3, 'b', 'z'))) AS t(day, s))
TO 'data/' STORED AS PARQUET PARTITIONED BY (day);

Program:

use datafusion::logical_expr::{LogicalPlanBuilder, col, lit};
use datafusion::prelude::*;

#[tokio::main]
async fn main() -> datafusion::error::Result<()> {
    let ctx = SessionContext::new();
    ctx.sql("CREATE EXTERNAL TABLE t STORED AS PARQUET LOCATION 'data/' PARTITIONED BY (day)")
        .await?;

    // The filter on a struct field keeps a pure extraction projection in the optimized plan.
    let optimized = ctx
        .sql("SELECT s['a'] AS a, day FROM t WHERE s['b'] <> 'q'")
        .await?
        .into_optimized_plan()?;

    // Put a filter on the optimized plan and optimize again.
    let filtered = LogicalPlanBuilder::from(optimized)
        .filter(col("day").eq(lit("2025-01-03")))?
        .build()?;
    let df = DataFrame::new(ctx.state(), filtered);
    println!("{}", df.clone().into_optimized_plan()?.display_indent());
    let physical = df.create_physical_plan().await?;
    println!("{}", datafusion::physical_plan::displayable(physical.as_ref()).indent(false));
    Ok(())
}

Before #25455, the predicate is in full_filters and the scan reads one file:

Projection: __datafusion_extracted_2 AS a, t.day
  Filter: __datafusion_extracted_1 != Utf8("q")
    Projection: get_field(t.s, Utf8("b")) AS __datafusion_extracted_1, t.day, get_field(t.s, Utf8("a")) AS __datafusion_extracted_2
      TableScan: t projection=[s, day], full_filters=[t.day = Dictionary(UInt16, Utf8("2025-01-03"))], partial_filters=[get_field(t.s, Utf8("b")) != Utf8("q")]

DataSourceExec: file_groups={1 group: [[data/day=2025-01-03/....parquet]]}, ...

On main, the predicate stays in the filter above the projection, and the scan reads all three files:

Projection: __datafusion_extracted_2 AS a, t.day
  Filter: __datafusion_extracted_1 != Utf8("q") AND t.day = Dictionary(UInt16, Utf8("2025-01-03"))
    Projection: get_field(t.s, Utf8("b")) AS __datafusion_extracted_1, t.day, get_field(t.s, Utf8("a")) AS __datafusion_extracted_2
      TableScan: t projection=[s, day], partial_filters=[get_field(t.s, Utf8("b")) != Utf8("q")]

DataSourceExec: file_groups={3 groups: [[data/day=2025-01-01/....parquet], [data/day=2025-01-02/....parquet], [data/day=2025-01-03/....parquet]]}, ...

With this PR, the logical plan and the file list are the same as before #25455.

What changes are included in this PR?

• When PushDownFilter finds a pure extraction projection directly above a TableScan, it keeps the filter above the projection, as fix: decide the precedence between PushDownFilter and PushDownLeafProjections #25455 requires. But it offers the scan each predicate that reads only pass-through columns. The scan records the predicates that its source accepts. A predicate that the source applies exactly leaves the filter. Thus PushDownLeafProjections has nothing to undo, and a second pass changes nothing.
• The TableScan arm of PushDownFilter moves into the function push_down_into_table_scan, so both paths use the same code. Its behaviour does not change.
• The module documentation of push_down_filter and extract_leaf_expressions, and the "Rule Precedence" section of the query optimizer guide, no longer say that the earlier pass always sets TableScan::filters. They describe the new step.

No new public API and no new configuration option.

What is the testing strategy for this PR?

Unit tests in push_down_filter.rs:

• filter_above_pure_extraction_projection_reaches_the_scan replaces filter_not_pushed_through_pure_extraction_projection. It shows that a predicate on a pass-through column reaches full_filters and leaves no filter node, with the projection still next to the scan.
• inexact_predicate_above_pure_extraction_projection_reaches_the_scan shows that an inexact source records the predicate in partial_filters, and that the filter keeps it. A predicate on an extracted alias stays in the filter only.
• pushdown_into_scan_below_extraction_projection_is_a_fixed_point applies the rule two times and shows that the second application changes nothing.
• unsupported_predicate_stays_above_pure_extraction_projection shows that the plan does not change when the source does not support the predicate.

The existing filter_and_extraction_projection_reach_fixed_point test still passes.

Commands run:

• cargo test -p datafusion-optimizer: all pass.
• cargo test -p datafusion-sqllogictest --test sqllogictests: all files pass.
• cargo clippy -p datafusion-optimizer --all-targets -- -D warnings: clean.

I could not reproduce the shape with one SQL statement, because in that case PushDownFilter runs before the extraction projection exists. Thus the tests are unit tests, not sqllogictest cases.

Are there any user-facing changes?

Yes. A filter that is put on an optimized plan reaches the scan again, so partition pruning and the provider's filter pushdown work again for such plans. No API changes.