datafusion-contrib / datafusion-contrib/datafusion-distributed

[dynamic filtering] 4. implement remote dynamic filtering

Open
#532 0 comments 0 reactions 0 assignees View on GitHub

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

  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.

Image

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

  1. Read the whole issue, then the project's contributing guide.
  2. Comment on the issue to say you are picking it up — it saves two people doing the same work.
  3. Fork the repository and make your change on a branch.
  4. 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.