rfc: distributed dynamic filtering - #553
Conversation
jayshrivastava
changed the title
jayshrivastava
changed the title
jayshrivastava
changed the title
gabotechs
left a comment
gabotechs
left a comment
There was a problem hiding this comment.
This is looking very good! I really do not have any major comments, this seems to be going in the right direction.
| The proposed change is to implement dynamic filtering similarly to Trino. | ||
|
|
||
| 1. During execution, the dynamic filter producers (`HashJoinExec`, `SortExec`, `AggregateExec`) will send dynamic filter updates to the coordinator. | ||
| 2. The coordinator will union / merge the dynamic filter updates from all workers. | ||
| 3. The coordinator will send the unioned / merged dynamic filter updates to the consumers (`DataSourceExec`). |
There was a problem hiding this comment.
👍 Sounds good, with the current infrastructure, I don't think it would be terrible difficult to do this
| - Less overhead than ORing | ||
|
|
||
| Cons: | ||
| - Is brittle. What if we have to support non-range and non-IN-LIST expressions? It would be nice if dynamic filters in vanilla datafusion natively implemented a `merge` or `union` operation |
There was a problem hiding this comment.
Why do you think it's brittle? For merging, I imagine you can just rely on Arrow comparison kernels for deduping IN LIST expressions.
There was a problem hiding this comment.
Can you expand on this? It's easy to implement a merge operation if we assume the expressions to merge always look like a@0 >= 0 AND a@0 <= 5 OR IN LIST [10, 11]. We would downcast to InListExpr and RangeExpr and define a merge operation for those.
In the future, what if there's other kinds of expressions in there? Then maybe the merge operation would break.
Maybe @LiaCastaneda has more insights on this. I think she implemented something similar.
There was a problem hiding this comment.
I imagine that not any arbitrary expression can be placed there. Even if it can, you still decide to optimize the most common cases, like IN LISTs or min/max ranges, and if there's anything else besides does two things just leave it OR-ed.
There was a problem hiding this comment.
In the future, what if there's other kinds of expressions in there?
I think that rather than adding other kinds of expressions, there is probably a higher possibility that the overall dynamic filter shape changes from the current CASE statement to something else.
Regarding expression types, I think the other discussed expression that hasn’t been implemented yet is bloom filters pushdown as dynamic filters. In that case, deduping probably becomes awkward, I’m not really sure how you would merge that (or if its possible). But I think it’s a very niche use case, and as Gabriel said, it can be ORed anyway.
|
|
||
| // Read Updates | ||
| pub fn current(&self) -> Result<Arc<dyn PhysicalExpr>> | ||
| pub(crate) fn subscribe(&self) -> DynamicFilterSubscription |
There was a problem hiding this comment.
🤔 subscribe not being public might be a problem.
There was a problem hiding this comment.
True. I think we can just use this instead https://github.com/apache/datafusion/blob/3a29d6bd8cc9ac2bf5efee9f070dcdeea9f97b32/datafusion/physical-expr/src/expressions/dynamic_filters/mod.rs?plain=1#L320 since we would wait for completion anyways.
|
|
||
| #### Options 1 & 2: Support new APIs In Vanilla DataFusion | ||
|
|
||
| See https://github.com/apache/datafusion/issues/23814 |
There was a problem hiding this comment.
apache/datafusion#23814 seems to be evolving in a good direction, and people seem engaged in the discussion, so I'd lean towards using whatever comes out of that.
There was a problem hiding this comment.
Yup I will stay on top of that issue.
3 participants
Rendered View: https://github.com/datafusion-contrib/datafusion-distributed/blob/js/dynamic-filtering-design/rfcs/1-distributed-dynamic-filtering.md
Informs #528