[dynamic filtering] 4. implement remote dynamic filtering

[jayshrivastava]

Dynamic filters won't work out of the box when the producer is remote from the consumer. We need to add the necessary plumbing to make this work.

I have thought about this quite a bit and the way to do it is something like

1. Identify tasks which produce dynamic filters.
2. Identify tasks which consume those dynamic filters.
3. On execution, producers should send dynamic filters to the coordinator
4. The coordinator should "merge" the dynamic filters. Currently dynamic filters have a CASE expr for each partition id. We need to merge all the cases together into one filter.
5. The coordinator must push the merged filter to all the consumer tasks

This is very similar to how dynamic filters are implemented in vanilla datafusion. The consumers (data sources) don't know which partition certain rows will be in when they reach the producer node, so the CASEs are used to resolve this in the consumers. We have to do something similar in distributed datafusion, except we must be able to "merge" the cases together.

Note the partitioning of each Data Source here is different than (a) the partitioning of the task and (b) the partitioning of the dynamic filter producer (HashJoinExec). There may be several layers of shuffles, repartitions etc. etc between the producers and consumers. Vanilla datafusion solves this problem using the CASE method described above. That seems to be the best approach.