Make filter pushdown API more precise for their purpose

[2010YOUY01]

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

[alamb]

This is a neat idea

[adriangb]

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?

[adriangb]

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

[LiaCastaneda]

Yeah I think the two issues yield a similar solution.

[LiaCastaneda]

I gave it a try here #18938

[adriangb]

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

[2010YOUY01]

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.

[adriangb]

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.

[jayshrivastava]

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.

[jayshrivastava]

take