Only compute bounds/ dynamic filters if consumer asks for it

[LiaCastaneda]

Describe the bug

Just a follow up to this comment.

Currently, DataFusion computes bounds for all queries that contain a HashJoinExec node whenever the option enable_dynamic_filter_pushdown is set to true (default). It might make sense to compute these bounds only when we explicitly know there is a consumer that will use them.

One way to achieve this could be during physical planning: while traversing the plan, check whether there is any scan/leaf node that is “interested in” or supports dynamic filters (determined by gather_filters_for_pushdown). This might just require adding some logic to the filter pushdown optimization rule itself I think?

Then, only if there is at least one interested consumer that accepts the DynamicFilterPhysicalExpr, set a flag on HashJoinExec to build the bounds accumulator, otherwise, skip bounds computation entirely.

To Reproduce

No response

Expected behavior

No response

Additional context

No response

[rkrishn7]

I think this is intentionally not handled in handle_child_pushdown_result 🤔

// Note that we don't check PushdDownPredicate::discrimnant because even if nothing said
// "yes, I can fully evaluate this filter" things might still use it for statistics -> it's worth updating

cc @adriangb

[adriangb]

I think this is intentionally not handled in handle_child_pushdown_result 🤔

// Note that we don't check PushdDownPredicate::discrimnant because even if nothing said
// "yes, I can fully evaluate this filter" things might still use it for statistics -> it's worth updating

cc @adriangb

Yep that's the case. For example Parquet may not be able to fully evaluate the filter for each row but may still be able to use it for file level and row group stats pruning.

I think we have two options:

1. Expand the pushdown result enum from two yes/no cases to three cases: exact (parent can drop the filter), inexact (parent should still apply the filter but it may be referenced for stats/partial pruning) and unsupported (the filter is not going to be used at all).
2. Build in some notion of "subscribers" to dynamic filters where the thing that's going to use them has to "subscribe".

[LiaCastaneda]

Yep that's the case. For example Parquet may not be able to fully evaluate the filter for each row but may still be able to use it for file level and row group stats pruning.

Ah, I didn't acocunt for that

Expand the pushdown result enum from two yes/no cases to three cases: exact (parent can drop the filter), inexact (parent should still apply the filter but it may be referenced for stats/partial pruning) and unsupported (the filter is not going to be used at all).

maybe we could start with this? although the subscriber option would still be interesting, especially now that we’re considering multiple filter types, a scan node could subscribe to just the type of filter it supports 🤔

[adriangb]

I think the pro of the subscriber option is:

1. Like you say it can be applied to only some filters and not others.
2. It avoids making the general filter pushdown API more complex (although the complexity gets pushed somewhere else).

[LiaCastaneda]

I imagine the "subscribe" action would still be done by the consumer during optimization, and "publish" by has join during execution. If no one else is interested, I can take a stab at this one

[adriangb]

That could be a good opportunity to broadcast updates

[adriangb]

That is something like #17370