You signed in with another tab or window. Reload to refresh your session.You signed out in another tab or window. Reload to refresh your session.You switched accounts on another tab or window. Reload to refresh your session.Dismiss alert
{{ message }}
Repository navigation
Support Push down expression evaluation in TableProviders #14993
For the scenario of select expensive_thing(col1) from data I would like to pre-process (speed up) expensive_thing(col1).
The easiest way I can think of doing this is by pre-computing the expression and saving it as a column with a specific name like _expensive_thing__col1.
But then I have to hardcode this column into my schema, the hardcoded expressions need to be the same for every file, etc.
To me an ideal solution would be to push down the expression to the file reading level so that I can then check "does _some_other_expensive_expr__col38 exist in the file? if so read that, otherwise read col38 and compute the expression".
The tricky thing is I'd want to do this on a per-file level: depending on the data different expression/column combinations would be pre-computed; it's prohibitive to put them all in the schema that is shared amongst all files.
I am reopening this ticket as I think it covers serveral important usecases (that are all subsets of @adriangb 's example of expensive_thing(col1) above
EXTRACT (minute from "EventDate"). For example, @gatesn mentions that the Vortex format may be able to evaluate this more quickly on the compressed format than extracting the full expression
struct_column["field_name"]: For example, extracting one field from a struct column -- in this case we could potentially update the json or parquet decoders to avoid materializing other fields (we would likely need more arrow-rs support too)
So a query might be
select EXTRACT (minute from"EventDate"), SUM(something)
FROM hits
GROUP BY EXTRACT (minute from"EventDate");
Being able to evlauate the EXTRACT (minute from "EventDate") expression during the scan would be super helpful
/// Returns true for all `Expr` in `expr` that can be directly evaluated by the TableProviderfnsupports_expr_pushdown(&self,expr:&[&Expr],) -> Result<Vec<bool>,DataFusionError>
This information would have to be threaded through to TableProvider::scan as well
(maybe it would be time to make TableProvider::scan_with_args 🤔 )
Being able to evaluate the EXTRACT (minute from "EventDate") expression during the scan would be super helpful
Maybe this is what you meant but my mind it's possible to do even better: instead of evaluating it during the scan, the file might even contain a pre-evaluated version of it. You can imagine something like a column called _computed_<hash of expr's SQL> so that you can read directly from (and use statistics of) a column called _computed_8has07fas98a (made up hash) instead of reading EventDate and computing EXTRACT (minute from "EventDate") on it. This becomes even more useful if you can use the statistics of the pre-computed expressions with a filter (e.g. where extract(minute from ts) = 1.
One note about this use case though: you can mostly achieve it today with some rewrite rules (you rewrite select EXTRACT (minute from "EventDate") from hits to select _computed_8has07fas98a as "EXTRACT (minute from \"EventDate ")" from hits. It's error prone and annoying though:
You have to explicitly list all of these columns as part of your TableProvider::schema which can be super annoying if you have 100s of them.
On top of these issues for a Variant type there is the issue that these columns can't vary file by file.
While this is crucial for a Variant type I think it can still be very helpful for other scenarios: if you are adding pre-computed columns as an optimization it might make sense to only compute them during an optimization / compaction pass. So you end up with some files that have the pre-computed column and some that don't. Blindly rewriting the expression to select _computed_8has07fas98a would by default result in incorrect results because SchemaAdapter would fill in nulls instead of computing the expression in real time. We worked around this by essentially forking SchemaAdapter and giving it the ability to generate columns from other columns instead of always filling them in with nulls, but that's very error prone and wonky.
One possibility here might be add an API to TableProvider similar to TableProvider::supports_filters_pushdown
maybe it would be time to make TableProvider::scan_with_args
Something like that sounds great to me!
A couple things to think about taking the example of select variant_get(json_col, 'key')::int from data:
Who is responsible for evaluating these expressions if they can vary on a per-file basis?
If the TableProvider says "yes, I can evaluate that expression" it is then responsible for doing the compute to evaluate it for every single file. If that data comes from a shredded column that makes sense, it's cheap. But if it has to start deserializing the Variant value column it's going to get expensive. Maybe that's not an issue but I did want to point out that it blurs the lines of where IO happens and where compute happens. If this is a problem I think it would complicate the API substantially.
What expression does the TableProvider get passed? In this example if could be variant_get(json_col, 'key') or variant_get(json_col, 'key')::int. I'm guessing it's the former, and the same rules as TableProvider::supports_filters_pushdown apply.
That said I think the best path forward is likely to prototype something in a PR and go from there 😄
Maybe this is what you meant but my mind it's possible to do even better: instead of evaluating it during the scan, the file might even contain a pre-evaluated version of it.
Yes indeed ! great point
Who is responsible for evaluating these expressions if they can vary on a per-file basis?
If the TableProvider says "yes, I can evaluate that expression" it is then responsible for doing the compute to evaluate it for every single file.
Yes, this is how I would expect it to work.
The table provider would have to figure out the best way to evalute the projection depending on its actual layout
for variants this would mean:
Files that had a field extracted to a shredded column would use that
Files that didn't have the field extracted would need to read the entire variant and pull out the field of interest
Maybe that's not an issue but I did want to point out that it blurs the lines of where IO happens and where compute happens. If this is a problem I think it would complicate the API substantially.
I agree -- implementing this optimally in a TableProvider will be complex
I note that the IO/CPU is already intertwined when implementing something like filter pushdown in parquet, so I am not sure also pusing down expressions makes the problem worse (or better)
Reacted by Adrian Garcia Badaracco and Will Manning
I note that the IO/CPU is already intertwined when implementing something like filter pushdown in parquet, so I am not sure also pusing down expressions makes the problem worse (or better)
I agree here - it's likely to not be a problem in practice.
So it sounds like the main complexity is going to be the TableProvider having to take ownership of applying the expression. It would be interesting to see if there's a way for it to dynamically decide to fall back to DataFusion, I can imagine situations where it wants to handle certain branches / conditions but would rather not re-implement others.
@cetra3 suggested that maybe this can be done with a rewrite of the PhysicalPlan? I guess the main issue would be that you don't know anything about the file except the file path at this point. You'd need to do at least some IO to get the the Parquet metadata to know the file's actual schema.
I guess what I was getting at is that maybe we could use a PhysicalOptimizerRule to do this sort of thing. However currently all the OptimizerRule traits are non async, which makes it a bit hard to do this sort of thing. I don't see why they can't be async
I can imagine situations where it wants to handle certain branches / conditions but would rather not re-implement others.
Given these are DataFusion scalar expressions (rather than any relational algebra), can the implementation not just invoke the expression as a fallback?
--
With Vortex, we've gone one step further and the scan accepts projection: Expr, filter: Option<Expr> where projection can arbitrarily select columns and apply scalar expressions to them. While this may be a little too general for DataFusion, it works well provided the system has good support for struct types and expressions for manipulating them. We have select (to select/exclude fields from a struct), pack (to assemble fields into a new struct), getitem (to extract a field from a struct), and merge (to union multiple structs, although this can be implemented with getitem+pack).
However this does have the downside of pushing disproportionate complexity onto the TableProvider for the simple case of projecting out a few columns.
So it sounds like the main complexity is going to be the TableProvider having to take ownership of applying the expression.
Given these are DataFusion scalar expressions (rather than any relational algebra), can the implementation not just invoke the expression as a fallback?
I agree with @gatesn -- I don't think adding new columns based on expressions has to be all that complex (you can already do it via a SchemaAdapter / SchemaMaper)
(there is a similar usecase for filling in new columns with default values rather than NULL)
With Vortex, we've gone one step further and the scan accepts projection: Expr, filter: Option where projection can arbitrarily select columns and apply scalar expressions to them. While this may be a little too general for DataFusion, it works well provided the system has good support for struct types and expressions for manipulating them.
The normal DataFusion filter pushdown API allows table providers to report which expressions they can handle, which means most providers can ignore filters unless they have code to handle it.
However this does have the downside of pushing disproportionate complexity onto the TableProvider for the simple case of projecting out a few columns.
Again, I think simple TableProviders can use something like SchemaAdapter for this usecase
Reacted by Adrian Garcia Badaracco and Will Manning
Would SchemaAdapter be a good place to implement this functionality? It already has knowledge of the required columns and file schema. We'd need piping around it (changes to TableProvider, Execs?) but at least implementing this as a user of DataFusion could be relatively self contained.
Is plumbing this through the schema overly specific to the pre-computed expression use-case? Or are you suggesting this is the mechanism by which all expression push-down occurs, by faking additional schema columns?
I opened #18627 which I think is the next step and gets us 90% of the way to the finish line. It's a draft PR, I expect a lot of tests to fail (I haven't even really touched the protobuf part yet...) but I would appreciate some eyes on it just to see if we all agree that's the right next step.
Activity
adriangb commentedon Mar 3, 2025
Argh I've asked this before and it's been answered: #7845 (comment)
alamb commentedon Mar 5, 2025
I am reopening this ticket as I think it covers serveral important usecases (that are all subsets of @adriangb 's example of
expensive_thing(col1)aboveEXTRACT (minute from "EventDate"). For example, @gatesn mentions that the Vortex format may be able to evaluate this more quickly on the compressed format than extracting the full expressionstruct_column["field_name"]: For example, extracting one field from a struct column -- in this case we could potentially update the json or parquet decoders to avoid materializing other fields (we would likely need more arrow-rs support too)So a query might be
Being able to evlauate the
EXTRACT (minute from "EventDate")expression during the scan would be super helpfulOne possibility here might be add an API to TableProvider similar to
TableProvider::supports_filters_pushdownmaybe something like
This information would have to be threaded through to
TableProvider::scanas well(maybe it would be time to make
TableProvider::scan_with_args🤔 )[-]Push down expression to files[/-][+]Support Push down expression evaluation in `TableProviders`[/+]alamb commentedon Mar 5, 2025
Related blog from @gatesn
https://blog.spiraldb.com/what-if-we-just-didnt-decompress-it/
adriangb commentedon Mar 5, 2025
Maybe this is what you meant but my mind it's possible to do even better: instead of evaluating it during the scan, the file might even contain a pre-evaluated version of it. You can imagine something like a column called
_computed_<hash of expr's SQL>so that you can read directly from (and use statistics of) a column called_computed_8has07fas98a(made up hash) instead of readingEventDateand computingEXTRACT (minute from "EventDate")on it. This becomes even more useful if you can use the statistics of the pre-computed expressions with a filter (e.g.where extract(minute from ts) = 1.One note about this use case though: you can mostly achieve it today with some rewrite rules (you rewrite
select EXTRACT (minute from "EventDate") from hitstoselect _computed_8has07fas98a as "EXTRACT (minute from \"EventDate ")" from hits. It's error prone and annoying though:TableProvider::schemawhich can be super annoying if you have 100s of them.select _computed_8has07fas98a from hitsand they show up inshow columns from hits. My goal with Support marking columns as system columns via Field's metadata #14362 was to solve this.On top of these issues for a Variant type there is the issue that these columns can't vary file by file.
While this is crucial for a Variant type I think it can still be very helpful for other scenarios: if you are adding pre-computed columns as an optimization it might make sense to only compute them during an optimization / compaction pass. So you end up with some files that have the pre-computed column and some that don't. Blindly rewriting the expression to
select _computed_8has07fas98awould by default result in incorrect results becauseSchemaAdapterwould fill in nulls instead of computing the expression in real time. We worked around this by essentially forkingSchemaAdapterand giving it the ability to generate columns from other columns instead of always filling them in with nulls, but that's very error prone and wonky.Something like that sounds great to me!
A couple things to think about taking the example of
select variant_get(json_col, 'key')::int from data:Who is responsible for evaluating these expressions if they can vary on a per-file basis?
If the
TableProvidersays "yes, I can evaluate that expression" it is then responsible for doing the compute to evaluate it for every single file. If that data comes from a shredded column that makes sense, it's cheap. But if it has to start deserializing the Variant value column it's going to get expensive. Maybe that's not an issue but I did want to point out that it blurs the lines of where IO happens and where compute happens. If this is a problem I think it would complicate the API substantially.What expression does the
TableProviderget passed? In this example if could bevariant_get(json_col, 'key')orvariant_get(json_col, 'key')::int. I'm guessing it's the former, and the same rules as TableProvider::supports_filters_pushdown apply.That said I think the best path forward is likely to prototype something in a PR and go from there 😄
alamb commentedon Mar 5, 2025
Yes indeed ! great point
Yes, this is how I would expect it to work.
The table provider would have to figure out the best way to evalute the projection depending on its actual layout
for variants this would mean:
I agree -- implementing this optimally in a TableProvider will be complex
I note that the IO/CPU is already intertwined when implementing something like filter pushdown in parquet, so I am not sure also pusing down expressions makes the problem worse (or better)
adriangb commentedon Mar 6, 2025
I agree here - it's likely to not be a problem in practice.
So it sounds like the main complexity is going to be the
TableProviderhaving to take ownership of applying the expression. It would be interesting to see if there's a way for it to dynamically decide to fall back to DataFusion, I can imagine situations where it wants to handle certain branches / conditions but would rather not re-implement others.adriangb commentedon Mar 6, 2025
@cetra3 suggested that maybe this can be done with a rewrite of the PhysicalPlan? I guess the main issue would be that you don't know anything about the file except the file path at this point. You'd need to do at least some IO to get the the Parquet metadata to know the file's actual schema.
cetra3 commentedon Mar 6, 2025
I guess what I was getting at is that maybe we could use a
PhysicalOptimizerRuleto do this sort of thing. However currently all the OptimizerRule traits are non async, which makes it a bit hard to do this sort of thing. I don't see why they can't be asyncgatesn commentedon Mar 6, 2025
Given these are DataFusion scalar expressions (rather than any relational algebra), can the implementation not just invoke the expression as a fallback?
--
With Vortex, we've gone one step further and the scan accepts
projection: Expr, filter: Option<Expr>where projection can arbitrarily select columns and apply scalar expressions to them. While this may be a little too general for DataFusion, it works well provided the system has good support for struct types and expressions for manipulating them. We have select (to select/exclude fields from a struct), pack (to assemble fields into a new struct), getitem (to extract a field from a struct), and merge (to union multiple structs, although this can be implemented with getitem+pack).However this does have the downside of pushing disproportionate complexity onto the TableProvider for the simple case of projecting out a few columns.
alamb commentedon Mar 6, 2025
I agree with @gatesn -- I don't think adding new columns based on expressions has to be all that complex (you can already do it via a SchemaAdapter / SchemaMaper)
(there is a similar usecase for filling in new columns with default values rather than NULL)
The normal DataFusion filter pushdown API allows table providers to report which expressions they can handle, which means most providers can ignore filters unless they have code to handle it.
Again, I think simple TableProviders can use something like SchemaAdapter for this usecase
adriangb commentedon Mar 6, 2025
Would SchemaAdapter be a good place to implement this functionality? It already has knowledge of the required columns and file schema. We'd need piping around it (changes to TableProvider, Execs?) but at least implementing this as a user of DataFusion could be relatively self contained.
gatesn commentedon Mar 6, 2025
Is plumbing this through the schema overly specific to the pre-computed expression use-case? Or are you suggesting this is the mechanism by which all expression push-down occurs, by faking additional schema columns?
37 remaining items
Add TableSchema helper to encapsulate file schema + partition fields (a…
Use TableSchema in FileScanConfig (apache#18231)
Let `FileScanConfig` own a list of `ProjectionExpr`s (apache#18253)
Let `FileScanConfig` own a list of `ProjectionExpr`s (apache#18253)
adriangb commentedon Nov 11, 2025
I opened #18627 which I think is the next step and gets us 90% of the way to the finish line. It's a draft PR, I expect a lot of tests to fail (I haven't even really touched the protobuf part yet...) but I would appreciate some eyes on it just to see if we all agree that's the right next step.
Add Projection struct w/ helper methods to manipulate projections (ap…
Add TableSchema helper to encapsulate file schema + partition fields (a…
Use TableSchema in FileScanConfig (apache#18231)
move projection handling into FileSource (#18627)
adriangb commentedon Nov 27, 2025
I've posted a very rough draft of a next step for Parquet in #18966.
Some other tasks that I think may need to happen before or in parallel:
PhysicalExprAdapter.SchemaAdapterfrom predicate pushdown / per row filtering.PhysicalExprAdapter.Cut `Parquet` over to PhysicalExprAdapter, remove `SchemaAdapter` (#1…
adriangb commentedon Dec 7, 2025
I have several a couple small PRs related to this work that need review if anyone here is keen:
#19137
#19129
And a slightly larger one to go after #19129 (diff will be smaller once rebased): #19128
adriangb commentedon Dec 9, 2025
One last one; <100LOC diff (right now)! #19111
Push down projection expressions into ParquetOpener (#19111)