feat: plan distributed dynamic filters

[jayshrivastava]

Stack

This stack of PRs implements distributed dynamic filtering #528

1. coordinator: display consumer dynamic filters after execution #623
2. feat: plan distributed dynamic filters #634 <- you are here
3. feat: forward remote dynamic filter updates to coordinator #635
4. coordinator: merge partial dynamic filters  #636
5. coordinator: forward merged dynamic filters to consumers #637
6. [do not review] worker: apply merged dynamic filters during execution #639

Goal

The coordinator should know what dynamic filters exist and where to route updates.

Details

1. Dynamic Filter Registry

  QueryCoordinator
  └── DynamicFilterRegistry
      └── filters: Map<expression_id, PlannedDynamicFilter>

Each PlannedDynamicFilter stores

• the producers and their stage/tasks
• the consumers and their stage/tasks

This will be used in future PRs to store incoming dynamic filter updates from workers and determine how/where to forward the updates.

Implementation

In the StageCoordinator, we send every task to the registry and extract dynamic filters.

2. Network Anchors

In this situation, the hash join does an execute()-time check to determine if it should update its dynamic filter. It checks to see if the filter is used by any children using apply_expressions (before apply_expressions was added upstream, it was an Arc pointer strong count check to see if there were multiple references).

worker 1
HashJoinExec  (dynamic_filter_predicate)
    NetworkShuffleExec

worker 2
DataSourceExec (dynamic_filter_predicate)

The join sees that no plan nodes below it use the filter, so it decides not to update it.

Ideally, the hash join decides at optimization time, before distributed planning. I've opened a discussion here about it: apache/datafusion#18856 (comment). While that issue is being resolved, I propose this workaround:

We create an "anchor" to make it seem like the NetworkShuffleExec uses the filter.

worker 1
HashJoinExec  (dynamic_filter_predicate)
    NetworkShuffleExec  (anchor: dynamic_filter_predicate)

worker 2
DataSourceExec (dynamic_filter_predicate)

Network Anchors Implementation

The implementation adds serialization overhead but is simpler. In static and dynamic planning, we recursively propagate all anchors upwards in the plan to all the network boundaries. We can revisit this implementation in future iterations. This recursive implementation is in inject_network_boundaries.

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

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

stage1:
        DataSourceExec (consumer: filter1, filter2)

This means we serialize 8 filters in total.

However, the minimal anchors you need are like this:

stage3:
    HashJoinExec <- producer #1
        NetworkShuffleExec  (anchors: filter2)

stage2:
    RepartitionExec
        AggregateExec <- producer #2
            NetworkShuffleExec   (anchors: filter2)

stage1:
        DataSourceExec (consumer: filter1, filter2)

In this plan, we would serialize 6 filters.

For 1 dynamic filter, the minimum filters you need to serialize are 1 (producer) + N (consumers) + 1 (network boundary). In this implementation, we serialize 1 (producer) + N (consumers) + M (all network boundaries above the consumer)

Other Notes

See #528. During dynamic planning, the sampler on the probe side of a hash join may overreport rows / cost because dynamic filters aren't being applied yet.