Push down projection expressions into ParquetOpener #19111
Merged
+236
−198
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
github-actions
bot
added
physical-expr
adriangb
commented
Dec 5, 2025
adriangb
commented
Dec 5, 2025
adriangb
commented
Dec 5, 2025
| let table_schema = table_schema.into(); | ||
| Self { | ||
| projection: SplitProjection::unprojected(&table_schema), | ||
| projection: ProjectionExprs::from_schema(table_schema.table_schema()), |
Contributor
Author
There was a problem hiding this comment.
The logic here of creating this upfront even if it will often be replaced is that the most important thing is that each time we create a ParquetOpener (once per file) we do the minimum amount of compute. Pre-computing here it and storing it in ParquetSource is the easiest way to do that.
adriangb
commented
Dec 5, 2025
github-merge-queue bot
pushed a commit
that referenced
this pull request
Dec 8, 2025
This allows it to be Arc'ed, have multiple references, etc. It's a backwards compatible change (aside from producing compiler warnings about unnecessarily mutable variables). This will help with #19111 where we'll want to re-use a simplifier instance for predicate and projection.
github-merge-queue bot
pushed a commit
that referenced
this pull request
Dec 9, 2025
This PR does some refactoring of `PhysicalExprAdapter` and `PhysicalExprSimplifier` that I found necessary and/or beneficial while working on #19111. ## Changes made ### Replace `PhysicalExprAdapter::with_partition_values` with `replace_columns_with_literals` This is a nice improvement because it: 1. Makes the `PhysicalExprAdapter` trait that users might need to implement simpler (less boilerplate for users). 2. Decouples these two transformations so that we can replace partition values and then apply a projection without having to also do the schema mapping (it would be from the logical schema to the logical schema, confusing and a waste of compute). I ran into this need in #19111. I think there may be other ways of doing it (e.g. piping in the expected output schema from ParquetSource) but it felt nicer this way and I expect other places may also need the decoupled transformation. 3. I think we can use it in the future to implement #19089 (edit: evidently I was right, see identical function in #19136). 4. It's less lines of code 😄 This will require any users calling `PhysicalExprAdapter` directly to change their code, I can add an entry to the upgrade guide. ### Remove partition pruning logic from `FilePruner` and deprecate now unused `PrunableStatistics` and `CompositePruningStatistics`. Since we replace partition values with literals we no longer need these structures, they get handled like any other literals. This is a good chunk of code / complexity that we can bin off. ### Use `TableSchema` instead of `SchemaRef` + `Vec<FieldRef>` in `ParquetOpener` `TableSchema` is basically `SchemaRef` + `Vec<FieldRef>` already and since `ParquetSource` has a `TableSchema` its less code and less clones to propagate that into `ParquetOpener` --------- Co-authored-by: Andrew Lamb <andrew@nerdnetworks.org>
adriangb
force-pushed
the
parquet-expression-pushdown
branch
from
d002d19 to
6aba2d1
Compare
github-actions
bot
removed
physical-expr
adriangb
commented
Dec 9, 2025
Comment on lines
326
to
344
| if let Some(expr_adapter_factory) = expr_adapter_factory.as_ref() { | ||
| // After rewriting to the file schema, further simplifications may be possible. | ||
| // For example, if `'a' = col_that_is_missing` becomes `'a' = NULL` that can then be simplified to `FALSE` | ||
| // and we can avoid doing any more work on the file (bloom filters, loading the page index, etc.). | ||
| // Additionally, if any casts were inserted we can move casts from the column to the literal side: | ||
| // `CAST(col AS INT) = 5` can become `col = CAST(5 AS <col type>)`, which can be evaluated statically. | ||
| let simplifier = PhysicalExprSimplifier::new(&physical_file_schema); | ||
| let rewriter = expr_adapter_factory.create( | ||
| Arc::clone(&logical_file_schema), | ||
| Arc::clone(&physical_file_schema), | ||
| ); | ||
| predicate = predicate | ||
| .map(|p| { | ||
| let expr = expr_adapter_factory | ||
| .create( | ||
| Arc::clone(&logical_file_schema), | ||
| Arc::clone(&physical_file_schema), | ||
| ) | ||
| .rewrite(p)?; | ||
| // After rewriting to the file schema, further simplifications may be possible. | ||
| // For example, if `'a' = col_that_is_missing` becomes `'a' = NULL` that can then be simplified to `FALSE` | ||
| // and we can avoid doing any more work on the file (bloom filters, loading the page index, etc.). | ||
| PhysicalExprSimplifier::new(&physical_file_schema).simplify(expr) | ||
| }) | ||
| .map(|p| simplifier.simplify(rewriter.rewrite(p)?)) | ||
| .transpose()?; | ||
| predicate_file_schema = Arc::clone(&physical_file_schema); | ||
| // Adapt projections to the physical file schema as well | ||
| projection = projection | ||
| .try_map_exprs(|p| simplifier.simplify(rewriter.rewrite(p)?))?; | ||
| } |
Contributor
Author
There was a problem hiding this comment.
Recommend viewing this in split view. The new pattern of applying the expression adapter and simplifer to both the projection and predicate in a unified manner is
github-actions
bot
added
the
documentation
Contributor
I think this is important (though a tech debt item to be sure) |
Contributor
|
Starting to check this one out |
Contributor
|
run benchmarks |
Contributor
|
run benchmark clickbench_pushdown |
|
🤖 |
alamb
approved these changes
Dec 10, 2025
Contributor
alamb
left a comment
alamb
left a comment
There was a problem hiding this comment.
Thank you @adriangb
I went through this code carefully and it looks really nice. Super well done! It has been quite a road but the result is 🧑🍳 👌
I do think we should work on deprecating SchemaAdapter for this release, to ensure that we have ported over all tests, for example.
| let projection = Arc::clone(&self.projection); | ||
| let logical_file_schema = Arc::clone(self.table_schema.table_schema()); | ||
| // Apply partition column replacement to projection expressions | ||
| let mut projection = self.projection.clone(); |
Contributor
There was a problem hiding this comment.
I found this code now very nice and easy to follow
| projector.project_batch(&b) | ||
| b = projector.project_batch(&b)?; | ||
| if replace_schema { | ||
| // Ensure the output batch has the expected schema. |
Contributor
There was a problem hiding this comment.
Thank you for the comments -- I see now that this is similar, but not the same, as what SchemaAdapter did.
Contributor
Author
|
I will follow up with deprecating SchemaAdapter. We already have a tracking issue so 👍🏻 edit: #16800 |
31 tasks
|
🤖: Benchmark completed Details
|
|
🤖 |
Contributor
Author
Well that's unexpected but welcome 😆 |
Contributor
Author
|
Benchmarks generally look great! |
|
🤖: Benchmark completed Details
|
Contributor
|
I know we are trying to go fast, but I think for some of these slightly less trivial PRs we should consider leaving them open for a while after approval and before merge to allow additional time to review from people who may not be awake, as described in https://datafusion.apache.org/contributor-guide/index.html#major-and-minor-prs The major/minor distinction is definitely subjective, but I do think it would be good to make sure as many people who are interested get the chance to comment if they want |
Contributor
Or maybe noise -- I'll see if it reprodces |
Contributor
|
run benchmark clickbench_partitioned |
Contributor
Author
Noted, thank you for the reminder. Just very excited to close the loop on this change. I’m happy to address any feedback that comes in the next couple days or revert if there are issues |
|
🤖 |
|
🤖: Benchmark completed Details
|
Contributor
Author
|
@alamb hmm I'm going to re-run again. That last run seems quite... interesting. |
Contributor
Author
|
run benchmark clickbench_partitioned |
|
🤖 |
|
🤖: Benchmark completed Details
|
Contributor
Author
|
I think my conclusion here is that the benchmarks are just too noisy to draw any conclusions but this change (as expected) seems neutral on performance. |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Closes #14993
Once this is merged I think we can say we support projection expression pushdown into scans and it is implemented for Parquet.
Remaining TODOs which I think should be tracked in other issues (I'll find them or create them later):
SchemaAdapterwithPhysicalExprAdapterand decide if we want to actually deprecateSchemaAdapter(Plan to replaceSchemaAdapterwithPhysicalExprAdapter#16800)