Skip to content

Support Push down expression evaluation in TableProviders #14993

Closed

Description

@adriangb

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.

Activity

  1. adriangb commented on Mar 3, 2025

    @adriangb
    ContributorAuthor

    Argh I've asked this before and it's been answered: #7845 (comment)

  2. reopened this on Mar 5, 2025
  3. alamb commented on Mar 5, 2025

    @alamb
    Contributor

    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

    One possibility here might be add an API to TableProvider similar to TableProvider::supports_filters_pushdown

    maybe something like

    /// Returns true for all `Expr` in `expr` that can be directly evaluated by the TableProvider
    fn supports_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 🤔 )

  4. changed the title [-]Push down expression to files[/-] [+]Support Push down expression evaluation in `TableProviders`[/+] on Mar 5, 2025
  5. alamb commented on Mar 5, 2025

    @alamb
    Contributor
  6. adriangb commented on Mar 5, 2025

    @adriangb
    ContributorAuthor

    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:

    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 😄

  7. alamb commented on Mar 5, 2025

    @alamb
    Contributor

    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:

    1. Files that had a field extracted to a shredded column would use that
    2. 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)

  8. adriangb commented on Mar 6, 2025

    @adriangb
    ContributorAuthor

    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.

  9. adriangb commented on Mar 6, 2025

    @adriangb
    ContributorAuthor

    @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.

  10. cetra3 commented on Mar 6, 2025

    @cetra3
    Contributor

    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

  11. gatesn commented on Mar 6, 2025

    @gatesn
    Contributor

    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.

  12. alamb commented on Mar 6, 2025

    @alamb
    Contributor

    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

  13. adriangb commented on Mar 6, 2025

    @adriangb
    ContributorAuthor

    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.

  14. gatesn commented on Mar 6, 2025

    @gatesn
    Contributor

    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?

  15. 37 remaining items

  16. adriangb commented on Nov 11, 2025

    @adriangb
    ContributorAuthor

    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.

  17. adriangb commented on Nov 27, 2025

    @adriangb
    ContributorAuthor

    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:

  18. adriangb commented on Dec 7, 2025

    @adriangb
    ContributorAuthor

    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

  19. adriangb commented on Dec 9, 2025

    @adriangb
    ContributorAuthor

    One last one; <100LOC diff (right now)! #19111

  20. added a commit that references this issue on Dec 10, 2025
    021188e
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Metadata

Metadata

Assignees

No one assigned

    Labels

    No labels
    No labels

    Type

    No type

    Projects

    No projects

      Milestone

      No milestone

      Relationships

      None yet

      Development

      No branches or pull requests

      Issue actions