datafusion-contrib / datafusion-contrib/datafusion-distributed
[dynamic filtering] 4. implement remote dynamic filtering
Nobody has claimed this yet.
- Dominant language
- Rust
- Stars
- 139
- Forks
- 67
- Avg merge
- 3d 1h
- Merged PRs (30d)
- 35
Description
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
- Identify tasks which produce dynamic filters.
- Identify tasks which consume those dynamic filters.
- On execution, producers should send dynamic filters to the coordinator
- 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.
- 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.
Contributor guide
No contributing guide indexed for this repository
First steps
- Read the whole issue, then the project's contributing guide.
- Comment on the issue to say you are picking it up — it saves two people doing the same work.
- Fork the repository and make your change on a branch.
- Open a pull request that references the issue number.
Research direction
No files, tests, or entry points are named. Start by locating the existing local dynamic-filter implementation and comparing it with vanilla DataFusion's CASE-based approach. Done means remote producers can send filters to the coordinator, the coordinator can merge them, and consumer tasks receive the merged filter across repartitions and shuffles.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- rust
- Domain
- distributed-systems
- Issue type
- Feature
- Difficulty
- 5/5
- Estimated time
- Over a week
- Activity status
- Quiet
- Clarity
- Mostly clear
- Newbie friendliness
- 35/100