Make filter pushdown API more precise for their purpose

Is your feature request related to a problem or challenge?

Original PR discussion: #18644 (comment)

Backgrounds

Backgrounds for dynamic filter: https://datafusion.apache.org/blog/2025/09/10/dynamic-filters/
Backgrounds for parquet filter pushdown: Parquet accepts a pushed-down filter for two purposes: 1) pruning partitions using statistics 2) Do row-level filtering for late materialization
Current APIs TLDR:
- When pushdown predicates, we don't differentiate is it for only stats pruning, or both stats and row-level filters, and the parquet operator will try to apply both (unless turned off by an option) Reference:
  
  datafusion/datafusion/physical-plan/src/execution_plan.rs
  
  Line 563 in f1ecacc
  
  fn gather_filters_for_pushdown(
- When child reply back the pushdown result in the optimizer pass, it said either Yes/No (I think the semantics now is Yes -> stat+row, No->maybe-stat)

Describe the solution you'd like

When one ExecutionPlan tries to pushdown a predicate, it should be able to indicate "this predicate is only for stats pruning", etc. The reason is some predicate might be too expensive and not worth evaluate at row level.

When the child propagate back pushdown result, it should also indicate how do they accept the filter (only for stat, or for both stat and row-level), so that the source node for the predicate can respond accordingly.

Describe alternatives you've considered

No response

Additional context

No response

This is a neat idea

I think the for stats pruning only is perhaps not the right distinction to make: what can and can't be used for stats pruning is going to vary by file format and changes over time.
The distinction that the logical filter pushdown makes, and that maybe we should be making here, is:

I am going to use the filter, but I can't guarantee exact filtering aka Inexact. This usually means it might be used for stats pruning but will not be used for row-level pruning.
I am going to apply the filter exactly as FilterExec would i.e. Exact.
I won't use the filter at all i.e. Unsupported.

But maybe I'm missing something... what would a node do with the information that another operator is going to use a filter "only for stats pruning"? Not produce the filter?

But there is also #17527 which is maybe what we actually want?

Yeah I think the two issues yield a similar solution.

I gave it a try here #18938

Is there any use case for Inexact other than not generating dynamic filters? If not, maybe we should separate these two things and keep Yes/No and add the DynamicFilterPhysicalExpr::is_used method

I think the for stats pruning only is perhaps not the right distinction to make: what can and can't be used for stats pruning is going to vary by file format and changes over time. The distinction that the logical filter pushdown makes, and that maybe we should be making here, is:

I am going to use the filter, but I can't guarantee exact filtering aka Inexact. This usually means it might be used for stats pruning but will not be used for row-level pruning.

I am going to apply the filter exactly as FilterExec would i.e. Exact.

I won't use the filter at all i.e. Unsupported.

But maybe I'm missing something... what would a node do with the information that another operator is going to use a filter "only for stats pruning"? Not produce the filter?

This extra 'stat-only' message is for the forward pass like HashJoinExec --(push down only for stat pruning, not row-by-row filtering)--> ParquetExec, and the backward pass ParquetExec->HashJoinExec should implement something like exact/inexact/no

Here is a example that this extra forward pass info helps:

select *
from
locations join spatial_objects
on distance(locations.loc, spatial_objects.loc) < 10m;

We can calculate a spatial range from the build side and push that to the probe side, and use stat pruning to eliminate some data. This is not worth to do row-level filtering at the scanner, because spatial calculation is very expensive to perform row-wise, it can be wasteful to filter in the scan once, and evaluate again during join probing -- only do so once in the join is better.

This idea is a bit ahead of where we are though, so we should better implement it only when needed, not proactively right now.

Even with HashJoinExec and ParquetSource it’s complicated: an InListExpr can be used for bloom filter pruning even if it won’t be used row-wise.

I'm interested in removing the execution-time check in HashJoinExec to see if a filter was pushed down. I think this should be done at optimization time, not at execution time. This is the problem I'm running into in datafusion-distributed:

worker 1
HashJoinExec  (dynamic_filter_predicate)
    NetworkShuffleExec


worker 2
DataSourceExec (dynamic_filter_predicate)

The hash join doesn't produce a filter because there's no filter used on that worker. However, I would expect that the join to produce dynamic filters because that decision should have been made during optimization time, before I split up the plan.

I was wondering why this check even exists because handle_child_pushdown_result should handle it. It turns out pruning predicates return No indicating they weren't pushed down.

One alternative I considered was artificially pushing filters down to the network nodes by doing a plan rewrite:

worker 1
HashJoinExec  (dynamic_filter_predicate)
    NetworkShuffleExec  (dynamic_filter_predicate)


worker 2
DataSourceExec (dynamic_filter_predicate)

This feels like a smell and I'm not sure if it works in other projects. I think other projects like ballista will run into this because they chop up plans as well. IMO it's worth adding a new state to remove runtime filter pushdown checks. I can try writing a small PR.

take

My pushback on adding a new state is that it's not clear what producers (i.e. HashJoinExec) are supposed to do with the information. Let's consider your proposed change in #24528 (diff): !matches!(filter.discriminant, PushedDown::Unsupported). As far as HashJoinExec is concerned there are 2 states: Unsupported and Exact/Inexact, so it did not benefit from having 3 states. It's not clear what benefit having 3 states yields.

One alternative I considered was artificially pushing filters down to the network nodes by doing a plan rewrite [...] This feels like a smell and I'm not sure if it works in other projects.

Could you elaborate on why this seems wrong or wouldn't work? To me it seems like the right thing to do and it should work.

My mental model, which may be wrong or outdated, is that we start with a plan like:

HashJoinExec
    FilterExec [col > 5]
        DataSourceExec

The distributed optimizer runs on this and produces a plan of the form:

HashJoinExec
    FilterExec [col > 5]
        NetworkShuffleExec
            DataSourceExec

Now we run the filter pushdown optimizer and get:

HashJoinExec 
    NetworkShuffleExec [ dynamic_filter_propagator ]
        DataSourceExec [col > 5, dynamic_filter_receiver ]

It's up to NetworkShuffleExec to decide that:

It will push down the static col > 5 filter as is
It hooks into the dynamic filter updates in it's node and pushes down something that listens for updates.

Then you split the plan up:

stage1:
HashJoinExec 
    NetworkShuffleExec [ dynamic_filter_propagator ]

stage2:
        DataSourceExec [col > 5, dynamic_filter_receiver ]

So NetworkShuffleExec is responsible for replying Yes to filter pushdown i.e. Yes this filter will be consumed downstream.

The workaround / explanation makes sense to me. That's what I have implemented now.

My pushback on adding a new state is that it's not clear what producers (i.e. HashJoinExec) are supposed to do with the information.

The only benefit is that the HashJoinExec can determine during planning that it's filter was pushed down, rather than in the execute() call. Since this isn't a hard blocker, so I'm okay with not adding the new state for now.

The only benefit is that the HashJoinExec can determine during planning that it's filter was pushed down, rather than in the execute() call. Since this isn't a hard blocker, so I'm okay with not adding the new state for now.

Can't it determine that now? HashJoinExec::handle_child_pushdown_result is called at planning time by the filter pushdown operator with Yes (was pushed down) or No (was not pushed down) for each filter.

Can't it determine that now? HashJoinExec::handle_child_pushdown_result is called at planning time by the filter pushdown operator with Yes (was pushed down) or No (was not pushed down) for each filter.

For parquet row-filter pushdown, HashJoinExec::handle_child_pushdown_result is called with PushedDown::No because the filtering is not exact. I think the reason is because PushedDown::Yes would indicate to a FilterExec that it can be removed, which is incorrect so it returns PushedDown::No everywhere while still performing inexact pruning.

For parquet row-filter pushdown, HashJoinExec::handle_child_pushdown_result is called with PushedDown::No because the filtering is not exact.

How the pushdown works today is this runtime check in HashJoinExec::execute():

datafusion/datafusion/physical-plan/src/joins/hash_join/exec.rs

Line 1460 in 7e4188e

.map(|id| plan_contains_expression_id(&self.right, id))

I implemented the propagators / anchors here: datafusion-contrib/datafusion-distributed#634

stage3:
   HashJoinExec <- producer of filter1
       NetworkShuffleExec  (anchors: filter1, filter2)

stage2:
   RepartitionExec
       AggregateExec <- producer of filter2
           NetworkShuffleExec   (anchors: filter1, filter2)

stage1:
       DataSourceExec (consumer: filter1, filter2)

A fair bit of code dedicated to working around the fact that the hash join decides during execution if it should produce filters or not.

@jayshrivastava

@jayshrivastava I think I understand a bit of the friction more now. I forget we are doing checks in execute() and once I saw that I understood where you are coming from and why adding an Inexact option is attractive. We may still end up adding it, but as a first step to make things nicer what do you think of #24601? Would it at least remove some of the complexity you mention in datafusion-contrib/datafusion-distributed#634?

Long term I feel we will make scans actually apply all filters and dissolve the filter_pushdown option, which is also needed to keep issues like #24125 from recurring. At that point scan's won't need to reply No and keep a reference for stats pruning, they can always reply Yes. Ironically this is blocked because actually evaluating hash join dynamic filters is generally a regression in our standard benchmarks.

fix: decide hash join dynamic filter production at planning time (apa…

feat: plan distributed dynamic filters (#634)

docs: clarify what `PushedDown::Yes` and `PushedDown::No` mean (apach…

Navigation Menu

Repository navigation

Make filter pushdown API more precise for their purpose #18856

Is your feature request related to a problem or challenge?

Backgrounds

Describe the solution you'd like

Describe alternatives you've considered

Additional context

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Make filter pushdown API more precise for their purpose #18856

Description

Is your feature request related to a problem or challenge?

Backgrounds

Describe the solution you'd like

Describe alternatives you've considered

Additional context

Activity

alamb commented on Nov 21, 2025

adriangb commented on Nov 26, 2025

adriangb commented on Nov 27, 2025

LiaCastaneda commented on Nov 27, 2025

LiaCastaneda commented on Nov 27, 2025

adriangb commented on Nov 27, 2025

2010YOUY01 commented on Nov 28, 2025

adriangb commented on Nov 28, 2025

jayshrivastava commented on Aug 20, 2026

jayshrivastava commented on Aug 20, 2026

adriangb commented on Aug 20, 2026

jayshrivastava commented on Aug 21, 2026

adriangb commented on Aug 21, 2026

jayshrivastava commented on Aug 21, 2026

jayshrivastava commented on Aug 23, 2026

jayshrivastava commented on Aug 23, 2026

adriangb commented on Aug 24, 2026

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions