Braedon-Wooding-Displayr opened a new pull request, #24441:
URL: https://github.com/apache/datafusion/pull/24441
`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?
<!--
We generally require a GitHub issue to be filed for all bug fixes and
enhancements and this helps us generate change logs for our releases. You can
link an issue to this PR using the GitHub syntax. For example `Closes #123`
indicates that this PR will close issue #123.
-->
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?
<!--
There is no need to duplicate the description in the issue here, but it is
sometimes worth providing a summary of the individual changes in this PR.
-->
New benchmark + new tests + propagating None through scans.
## Are these changes tested?
<!--
We typically require tests for all PRs in order to:
1. Prevent the code from being accidentally broken by subsequent changes
2. Serve as another way to document the expected behavior of the code
If tests are not included in your PR, please explain why (for example, are
they covered by existing tests)?
-->
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
