perf: don't manufacture an identity projection in ParquetSource

[Braedon-Wooding-Displayr]

FileSource::projection() returns Option<&ProjectionExprs>, but every built-in source returned Some(...) unconditionally, and ParquetSource::new manufactured an identity projection over the full table schema. Consumers in FileScanConfig therefore always took the Some branch and did work proportional to the schema width (projected schema, projection mapping, statistics projection) even when the projection selected every column, in order, under its own name.

ParquetSource now stores Option<ProjectionExprs> and leaves it None until a projection is genuinely pushed down. None travels through the morselizer and the per-file prepare stages to DecoderProjection, which installs ProjectionMask::all() and no per-batch transform when the decoder's own output already is the scan's output schema.

A concrete projection is still materialized where one is genuinely needed, and per file rather than per partition: when partition or constant columns have to be substituted as literals, when the file's schema does not match the table's and casts or null fills have to land somewhere, and when the decoder's output schema does not match the scan's for any other reason.

A pushdown that reproduces a source's own output is not a pushdown, and detecting it belongs with the caller rather than with every source. Across the sqllogictest corpus 871 of them were being performed, 663 of those into ParquetSource. projection_is_no_op tests the incoming projection against the source's current output, whether that output is the table schema or an existing projection's aliases, and the callers of FileSource::try_pushdown_projection skip the push when it holds, so csv, arrow, avro and json get the same benefit without being converted.

FileScanConfig::try_swapping_with_projection reports success with the scan unchanged rather than Ok(None), so the caller still drops the redundant ProjectionExec.

FileScanConfig::partition_statistics was not equivalent between its two branches: ProjectionExprs::project_statistics recomputes total_byte_size from the output schema, and the unprojected branch did not. Both branches now recompute it, so a scan reports the same statistics whether or not a projection was pushed.

NOTE: AI was used to help review these changes (in particular fable) as well as help write tests / benchmarks. The core changes themselves were pretty simple though.

Which issue does this PR close?

Happy to raise a bug for this, up to you? Just let me know / if it's a bug or a feature.

Rationale for this change

The new parquet_wide_scan bench measures building the physical scan for an unprojected parquet table, which is the work proportional to the table's width:

scan_construction/unprojected_1000_columns      456 us -> 7.0 us   (-98.5%)
scan_construction/unprojected_10000_columns    4.61 ms -> 77.3 us  (-97.9%)
scan_construction/unprojected_100000_columns   59.0 ms -> 1.26 ms  (-97.8%)

End-to-end there is no measurable change, and the bench's planning and execution groups are controls that show this rather than claim otherwise. Physical planning of SELECT * is dominated by expanding the wildcard and running the optimizer over one expression per column, and a full scan is dominated by decoding, so the removed per-file project_schema, per-leaf mask vector and per-batch Projector do not surface above the noise floor, measured at roughly +/-10% by comparing the baseline binary against its own results.

The reason though is that I have a custom SQL command/node that handles a SELECT * without having to expand the wildcard but this still results in relatively slow queries due to this physical overhead (as you can see 60ms is heavy! And we have upwards of 250k columns).

What changes are included in this PR?

New benchmark + new tests + propagating None through scans.

Are these changes tested?

Yes!

Are there any user-facing changes?

We actually use None as a valid result from a projection from scans, this means that any consumer code of this (like analyzers and such) have to accept this as meaning an identity scan. This functionally doesn't seem like an issue.

[codecov-commenter]

Codecov Report

❌ Patch coverage is 98.72495% with 7 lines in your changes missing coverage. Please review.
✅ Project coverage is 81.61%. Comparing base (7e5561d) to head (cc47000).
⚠️ Report is 11 commits behind head on main.

Files with missing lines	Patch %	Lines
datafusion/datasource-parquet/src/opener/mod.rs	96.93%	0 Missing and 3 partials ⚠️
datafusion/physical-expr/src/projection.rs	98.51%	0 Missing and 2 partials ⚠️
datafusion/datasource/src/file_scan_config/mod.rs	99.27%	0 Missing and 1 partial ⚠️
...atafusion/datasource/src/file_scan_config/proto.rs	80.00%	0 Missing and 1 partial ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main   #24441      +/-   ##
==========================================
- Coverage   81.63%   81.61%   -0.03%     
==========================================
  Files        1123     1123              
  Lines      409869   411411    +1542     
  Branches   409869   411411    +1542     
==========================================
+ Hits       334599   335765    +1166     
- Misses      55596    55897     +301     
- Partials    19674    19749      +75     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[Braedon-Wooding-Displayr]

Pushed some improvements to cover a missing line that we should cover the rest of the lines are just ?

[Braedon-Wooding-Displayr]

Note: my optimisations to SELECT * could apply to this codebase as well (if interested). This is not part of this PR and would be a longer effort to port.

The idea is that in lots of cases we don't need to expand the projection of * until physical planning, this is because it doesn't effect anything.

This means that * doesn't become a performance cost. In my case I have a special command EXPORT * just because it restricts the edgecases to consider (no nesting).

But conceptually an expression like;

SELECT x from (
  SELECT * from foo
)

Still doesn't need to expand * it just needs to ensure that * holds x that is effectively it just needs in a "logical" plan to be something like;

SELECT x from (
  SELECT x, * from foo
)

(I'm saying "something like" because obviously this would duplicate the x column but hopefully you understand what I mean).

This results in massive performance improvements for wide queries because a lot of the cost is in planning and quadratic costs (well that and parquet metadata is terrible for wide formats, but that's something ultimately unfixable without caching / having a custom footer through something like delta/iceberg/... though they also don't handle it great).

In my case I ended up with sublinear scaling which is nice, and physical plan is very cheap. Now the risk here is that there might be some optimisations that rely on * expanding but honestly I can't really imagine that to be true.