You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "jayshrivastava (via GitHub)" <gi...@apache.org> on 2026/07/22 19:18:39 UTC
[I] make dynamic filters in `ExecutionPlan` nodes discoverable [datafusion]
jayshrivastava opened a new issue, #23814:
URL: https://github.com/apache/datafusion/issues/23814
### Is your feature request related to a problem or challenge?
Disclaimer: I'm writing this from the perspective of datafusion-distributed, but I imagine that this problem is common to other projects (ballista, comet) that distribute datafusion queries. I am not familiar with comet or ballista, so I would appreciate any feedback from folks who work on those projects.
More context in the [Dynamic Filtering in DataFusion-Distributed RFC](https://github.com/datafusion-contrib/datafusion-distributed/pull/553)
## Background
Datafusion natively preserves dynamic filter producer<->consumer relationships across network boundaries only when you serialize the producer (`HashJoinExec`) and the consumer (`DataSourceExec`) in serialization run. There's [tests](https://github.com/apache/datafusion/blob/5b825e0d7710ed9228b809ba2e487aaea7854708/datafusion/proto/tests/cases/roundtrip_physical_plan.rs?plain=1#L3310) which assert that this works.
```
┌───────────────────────────────────────────────┐
│ ┌───────────────────────┐ │
│ │ RepartitionExec │ │
│ │ │ │
│ └───────────────────────┘ │
│ ┌───────────────────────┐ │
│ │ HashJoinExec │ │
┌───────────────┐ │ │ mode = Partitioned │ │
│ Machine 1 │ │ └───────────────────────┘ │
└───────────────┘ │ ┌───────────────────┐ ┌───────────────────┐ │
│ │ DataSourceExec │ │ ProjectionExec │ │
│ └───────────────────┘ └───────────────────┘ │
│ ┌───────────────────┐ │
│ │ DataSourceExec │ │
│ └───────────────────┘ │
└───────────────────────────────────────────────┘
│
│ machine 1 serializes the plan
│ machine 2 deserializes the plan and preserves dynamic filter
│ relationships. dynamic filtering works
▼
┌───────────────────────────────────────────────┐
│ ┌───────────────────────┐ │
│ │ RepartitionExec │ │
│ │ │ │
│ └───────────────────────┘ │
│ ┌───────────────────────┐ │
│ │ HashJoinExec │ │
│ │ mode = Partitioned │ │
┌───────────────┐ │ └───────────────────────┘ │
│ Machine 2 │ │ ┌───────────────────┐ ┌───────────────────┐ │
└───────────────┘ │ │ DataSourceExec │ │ ProjectionExec │ │
│ └───────────────────┘ └───────────────────┘ │
│ ┌───────────────────┐ │
│ │ DataSourceExec │ │
│ └───────────────────┘ │
└───────────────────────────────────────────────┘
```
The problem is that datafusion doesn't support the case where you split up the consumer from the producer:
```
┌───────────────────────────────────────────────┐
│ ┌───────────────────────┐ │
│ │ RepartitionExec │ │
│ │ │ │
│ └───────────────────────┘ │
│ ┌───────────────────────┐ │
│ │ HashJoinExec │ │
┌───────────────┐ │ │ mode = Partitioned │ │
│ Machine 1 │ │ └───────────────────────┘ │
└───────────────┘ │ ┌───────────────────┐ ┌───────────────────┐ │
│ │ DataSourceExec │ │ ProjectionExec │ │
│ └───────────────────┘ └───────────────────┘ │
│ ┌───────────────────┐ │
│ │ DataSourceExec │ │
│ └───────────────────┘ │
└───────────────────────────────────────────────┘
│
│
│ machine 1 sends the probe side to machine 2
│ how will the hash join send updates to the data source?
▼
┌───────────────────────────────────────────────┐
│ │
│ │
│ │
│ │
│ │
│ │
│ │
┌───────────────┐ │ │
│ Machine 2 │ │ ┌───────────────────┐ │
└───────────────┘ │ │ ProjectionExec │ │
│ └───────────────────┘ │
│ ┌───────────────────┐ │
│ │ DataSourceExec │ │
│ └───────────────────┘ │
└───────────────────────────────────────────────┘
```
### Describe the solution you'd like
Proposal:
1. Distributed projects (ballista, comet, datafusion-distributed) should be responsible for implementing dynamic filter update routing and propagation during query execution
2. Vanilla datafusion should provide a way to find and extract dynamic filters from `ExecutionPlan` to allow someone to implement (1). The minimal required features are
(a) identify which dynamic filters are producers and consumers in an `ExecutionPlan` tree
(b) get access to all `&DynamicFilterPhysicalExpr` in an `ExecutionPlan` (as long as one is holding a reference to a `DynamicFilterPhysicalExpr`, they can read and write updates to it)
(c) allow easy extension for custom `ExecutionPlan` implementations
#### Option 1 - New `ExecutionPlan` API
```rust
trait ExecutionPlan {
fn dynamic_filter_exprs(&self) -> Vec<DynamicFilterNodeBehavior>
}
enum DynamicFilterNodeBehavior {
Producer(Arc<dyn PhysicalExpr>),
Consumer(Arc<dyn PhysicalExpr>),
}
```
Notes
- Previously, a similar API was reverted in https://github.com/apache/datafusion/pull/22437 because implementing it was complex and there was no use case in vanilla datafusion to assert correctness
- The proposed `dynamic_filter_exprs()` API is simpler to implement, making the breaking change less annoying
Pros:
- Does exactly what we want
Cons:
- Breaking change. It must be implemented by every `ExecutionPlan`. We cannot default to `vec![]` because `ExecutionPlan` nodes may fail to declare their dynamic filters.
- Very specific. We usually treat dynamic filters as any other filter expression. However, this method cuts through the abstraction and brings them to the forefront.
#### Option 2 - Store Dynamic Filters in the `Session` (`SessionState`)
We can try updating the `SessionState` to store which plan nodes are producers and consumers of dynamic filters. This can be populated during the `FilterPushdown` optimization rule.
```
struct FilterPushdownReport {
bindings: HashMap<(u64, u64), Vec<RuntimeFilterBinding>>, // (expression_id, query_id) -> bindings
}
enum RuntimeFilterBinding {
Producer {
node: Arc<dyn ExecutionPlan>,
expression: Arc<dyn PhysicalExpr>,
},
Consumer {
node: Arc<dyn ExecutionPlan>,
expression: Arc<dyn PhysicalExpr>,
usage: FilterUsage, // Pruning or FullyEvaluated
},
}
```
Pros
- Not a breaking change. No new `ExecutionPlan` API.
Cons
- We have to deal with a state which spans multiple queries
- `Arc<dyn PhysicalExpr>` references may grow stale. In vanilla datafusion, `FilterPushdown` is the last mutating optimizer rule, so the pointers wouldn't change. However, `datafusion-distributed` certainly modifies the `ExecutionPlan` tree
- `FilterPushdownPhase::Post` says [it should be run at the end of the optimization process since any changes to the plan may break the dynamic filter’s references](https://github.com/apache/datafusion/blob/f33dcec6dfd0b900ddf89b75b46e547f7923c0e7/datafusion/physical-optimizer/src/optimizer.rs?plain=1#L232-L245).
- We could make it the distributed project's responsibility to mutate / update the `FilterPushdownReport` if it changes any `ExecutionPlan`s, but that would still rely on pointer comparisons.
Open questions
- Is it acceptable to add features in vanilla datafusion which aren't used by vanilla datafusion? Who is responsible for testing these? (Related: https://github.com/apache/datafusion/pull/23282).
- It would be nice if vanilla datafusion committed to a minimal feature (ie. "DynamicFilterPhysicalExpr can be discovered from ExecutionPlan") with minimal tests support use cases like datafusion-distributed, ballista, and comet.
### Describe alternatives you've considered
We could downcast `ExecutionPlan` to concrete types and try to discover dynamic filters like that, however this method is less extensible (how would we support custom `ExecutionPlan` implementations?) and it isn't guaranteed to work because not every concrete type will make their dynamic filters `pub`.
As of writing, these public methods on some execution plan nodes exist, byt they were only added to support serialization and will likely be removed by https://github.com/apache/datafusion/issues/23494.
```rust
HashJoinExec::dynamic_filter_expr()
AggregateExec::dynamic_filter_expr()
SortExec::dynamic_filter_expr()
```
### Additional context
_No response_
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] make dynamic filters in `ExecutionPlan` nodes discoverable [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on issue #23814:
URL: https://github.com/apache/datafusion/issues/23814#issuecomment-5073540228
> ^ That is describing the producer side only, right? We need the consumer side as well.
Yes, that is true - however we could do it today and get real benefit (e.g. have the dynamic filters available on read). If you think it is worth while I can file a ticket
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] make dynamic filters in `ExecutionPlan` nodes discoverable [datafusion]
Posted by "jayshrivastava (via GitHub)" <gi...@apache.org>.
jayshrivastava commented on issue #23814:
URL: https://github.com/apache/datafusion/issues/23814#issuecomment-5096920422
> if you decide to go with the generic function approach (apply_expressions) will you need the write one as well: map_expressions?
I don't think we need `map_expressions` because we should be able to call `apply_expressions`, downcast to `&DynamicFilterPhysicalExpr`, and call `update()`, which is `pub`.
https://github.com/apache/datafusion/blob/3a29d6bd8cc9ac2bf5efee9f070dcdeea9f97b32/datafusion/physical-expr/src/expressions/dynamic_filters/mod.rs?plain=1#L246
We can read updates from `&DynamicFilterPhysicalExpr` via `wait_complete`
https://github.com/apache/datafusion/blob/3a29d6bd8cc9ac2bf5efee9f070dcdeea9f97b32/datafusion/physical-expr/src/expressions/dynamic_filters/mod.rs?plain=1#L320
> IIUC for the ExecutionPlan::dynamic_filter approach, in theory consumers can specify their dynamic filters as well because they are also ExecutionPlan no? or is there some consideration I'm missing? 🤔 I think the only "ugly" thing is that the consumer node will need to either store it's dynamic filters in a separate variable or downcast it's predicates to know which expressions are dynamic filters.
Maybe `ExecutionPlan::dynamic_filters_produced` is more clear? The proposed contract is that consumers don't return them here. You make a good point about producers returning their expressions via `apply_expressions`. They probably should do that because `apply_expressions` is for all expressions. Maybe the caller (datafusion distributed or otherwise) can take the difference: `consumer_dynamic_filter_expressions = ExecutionPlan::apply_expressions() - ExecutionPlan::dynamic_filters_produced() - non_dynamic_filter_expressions`
> I think there might be a few places in DF that could be rewritten to use apply_expressions, possibly some optimizer passes that downcast heavily on each node
+1. It would be a good idea to cover that.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] make dynamic filters in `ExecutionPlan` nodes discoverable [datafusion]
Posted by "jayshrivastava (via GitHub)" <gi...@apache.org>.
jayshrivastava commented on issue #23814:
URL: https://github.com/apache/datafusion/issues/23814#issuecomment-5050841633
> But if the plan is being split up and distributed amongst nodes I imagine it would be necessary or at last helpful to know which producers correspond to which consumers? If not you'd have to broadcast updates from all producers to all consumers.
Agreed. I was looking to use `expression_id` for this. The expression_id helps identify pairs that are linked together, but it doesn't make it clear in which direction the updates need to go.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] make dynamic filters in `ExecutionPlan` nodes discoverable [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb closed issue #23814: make dynamic filters in `ExecutionPlan` nodes discoverable
URL: https://github.com/apache/datafusion/issues/23814
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] make dynamic filters in `ExecutionPlan` nodes discoverable [datafusion]
Posted by "jayshrivastava (via GitHub)" <gi...@apache.org>.
jayshrivastava commented on issue #23814:
URL: https://github.com/apache/datafusion/issues/23814#issuecomment-5069532595
> I think a "consumer" is a consumer simply by nature of evaluating an expression tree that happens to have [DynamicFilterPhysicalExpr](https://github.com/apache/datafusion/blob/c1180022e63f48a3af175d8d73dc21cca9379c65/datafusion/physical-expr/src/expressions/dynamic_filters/mod.rs#L69) somewhere.
> 1. A generic API to walk and modify expressions on ExecutionPlan similar to LogicalPlan::map_expressions and LogicalPlan::apply_expressions (basically what @LiaCastaneda added in https://github.com/apache/datafusion/pull/20337 )
> 2. Add an API to ExecutionPlan to return what dynamic filters it produced (basically promote existing methods like [SortExec::dynamic_filters](https://github.com/apache/datafusion/blob/c1180022e63f48a3af175d8d73dc21cca9379c65/datafusion/physical-plan/src/sorts/sort.rs#L1102) and [HashJoinExec::dynamic_filters](https://github.com/apache/datafusion/blob/c1180022e63f48a3af175d8d73dc21cca9379c65/datafusion/physical-plan/src/joins/hash_join/exec.rs#L957) to the ExecutionPlan trait)
This all sounds right to me. Re. `apply_expressions`, most of the reasons for the [revert PR](https://github.com/apache/datafusion/pull/22437) are still true today
- the method isn't called from vanilla datafusion
- it cannot not have a default implementation, so every ExecutionPlan is forced to implement it
- it's tricky to implement because it involves tree node recursion and the behavior of the method can change often (ex. different expressions will be traversed depending on if its called before or after the filter pushdown optimizer pass)
The main difference today is we have a concrete use-case. Is this enough to add `apply_expressions` back? I think we would just need to revert the revert PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] make dynamic filters in `ExecutionPlan` nodes discoverable [datafusion]
Posted by "jayshrivastava (via GitHub)" <gi...@apache.org>.
jayshrivastava commented on issue #23814:
URL: https://github.com/apache/datafusion/issues/23814#issuecomment-5069648526
^ That is describing the producer side only, right? We need the consumer side as well.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] make dynamic filters in `ExecutionPlan` nodes discoverable [datafusion]
Posted by "jayshrivastava (via GitHub)" <gi...@apache.org>.
jayshrivastava commented on issue #23814:
URL: https://github.com/apache/datafusion/issues/23814#issuecomment-5184214483
Opened https://github.com/apache/datafusion/pull/24068 which adds `ExecutionPlan::dynamic_expressions` so we can find dynamic filter producers.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] make dynamic filters in `ExecutionPlan` nodes discoverable [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on issue #23814:
URL: https://github.com/apache/datafusion/issues/23814#issuecomment-5051454400
I think returning `DynamicFilterNodeBehavior:: Consumer` will require the equivalent of apply_expressions anyways (to visit all expressions in a node).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] make dynamic filters in `ExecutionPlan` nodes discoverable [datafusion]
Posted by "LiaCastaneda (via GitHub)" <gi...@apache.org>.
LiaCastaneda commented on issue #23814:
URL: https://github.com/apache/datafusion/issues/23814#issuecomment-5092218448
@jayshrivastava if you decide to go with the generic function approach ( `apply_expressions`) will you need the write one as well: `map_expressions`?
> ^ That is describing the producer side only, right? We need the consumer side as well.
IIUC for the `ExecutionPlan::dynamic_filter` approach, in theory consumers _can_ specify their dynamic filters as well because they are also `ExecutionPlan` no? or is there some consideration I'm missing? 🤔 I think the only "ugly" thing is that the consumer node will need to either store it's dynamic filters in a separate variable or downcast it's predicates to know which expressions are dynamic filters.
--
> it cannot not have a default implementation, so every ExecutionPlan is forced to implement it
What if we implement `apply_expressions` on all DF nodes (like the reverted PR) but leave the default implementation to return `unimplemented!() `(and keep it optional). So if there is code in DF that actually uses it in the future, the user will know and it wont just fail/lie silently? Although I guess this option is not clean either.
> What I would suggest to make progress is finding some spot in the DataFusion code that could be rewritten to use apply_expressions.
I think there might be a few places in DF that could be rewritten to use `apply_expressions`, possibly some optimizer passes that downcast heavily on each node
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] make dynamic filters in `ExecutionPlan` nodes discoverable [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on issue #23814:
URL: https://github.com/apache/datafusion/issues/23814#issuecomment-5073565010
> This all sounds right to me. Re. `apply_expressions`, most of the reasons for the [revert PR](https://github.com/apache/datafusion/pull/22437) are still true today
>
> * the method isn't called from vanilla datafusion
Yes -- I think we should figure out some way to fix this -- I can't imagine there is no way we can't refactor some of DataFusion to use the API so that it will
> * it cannot not have a default implementation, so every ExecutionPlan is forced to implement it
Yes this is unfortunate but I don't see any way around it. It also moves us towards less downcasing (and thus more generic optimzer rules)
> * it's tricky to implement because it involves tree node recursion and the behavior of the method can change often (ex. different expressions will be traversed depending on if its called before or after the filter pushdown optimizer pass
I think it would make more sense if `apply_expressions` only walks the root of each expression (and any recursive tree walking is left to the caller if they want that)
Perhaps we can handle this with appropriate documentation
What I would suggest to make progress is finding some spot in the DataFusion code that could be rewritten to use apply_expressions
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] make dynamic filters in `ExecutionPlan` nodes discoverable [datafusion]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on issue #23814:
URL: https://github.com/apache/datafusion/issues/23814#issuecomment-5050599772
also cc @phillipleblanc @milenkovicm @avantgardnerio @martin-g from Ballista community
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] make dynamic filters in `ExecutionPlan` nodes discoverable [datafusion]
Posted by "jayshrivastava (via GitHub)" <gi...@apache.org>.
jayshrivastava commented on issue #23814:
URL: https://github.com/apache/datafusion/issues/23814#issuecomment-5050857160
We could infer from context (ex. `HashJoinExec` is always a producer), but then there's the question of custom `ExecutionPlan` implementations. I figured we'd want them to declare themselves to be producers or consumers.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] make dynamic filters in `ExecutionPlan` nodes discoverable [datafusion]
Posted by "gabotechs (via GitHub)" <gi...@apache.org>.
gabotechs commented on issue #23814:
URL: https://github.com/apache/datafusion/issues/23814#issuecomment-5069585251
> Is this enough to add apply_expressions back?
What @alamb suggests [here](https://github.com/apache/datafusion/issues/23814#issuecomment-5051471895) might be a better alternative, as it minimizes downstream impact and should be more than enough. @jayshrivastava Do you think we would be missing something by having `ExecutionPlan::dynamic_filters` instead of `ExecutionPlan::apply_expressions`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] make dynamic filters in `ExecutionPlan` nodes discoverable [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on issue #23814:
URL: https://github.com/apache/datafusion/issues/23814#issuecomment-5050800628
> Breaking change. It must be implemented by every ExecutionPlan. We cannot default to vec![] because ExecutionPlan nodes may fail to declare their dynamic filters.
I think what made the previous implementation tricky was that it required declaring all filter expressions and the API was so generic that a reasonable consumer could rely on it in ways that would break if it was not implemented properly. This more narrow version is at least safer if not safe in that sense: dynamic filters are generally optional performance improvements. Their absence or mishandling doesn't result in incorrect results or errors, just in performance degradation. So maybe defaulting to `vec![]` is a viable option.
For `DynamicFilterNodeBehavior` and the `Producer`/`Consumer` split: is it necessary to know the relationship between nodes? I.e. if I have a `SortExec` and a `HashJoinExec` they will both declare themselves as producers, then there will be 1 or maybe 2 `DataSourceExec` that are consumers. But if the plan is being split up and distributed amongst nodes I imagine it would be necessary or at last helpful to know which producers correspond to which consumers? If not you'd have to broadcast updates from all producers to all consumers.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] make dynamic filters in `ExecutionPlan` nodes discoverable [datafusion]
Posted by "avantgardnerio (via GitHub)" <gi...@apache.org>.
avantgardnerio commented on issue #23814:
URL: https://github.com/apache/datafusion/issues/23814#issuecomment-5051345715
I am not able to give it my full attention yet, and I apologize for that. But I think Impala does this:
```
PLAN-ROOT SINK
|
04:EXCHANGE [UNPARTITIONED]
|
02:HASH JOIN [LEFT SEMI JOIN, BROADCAST]
| hash predicates: year = year
| runtime filters: RF000 <- year <-- producer
|
|--03:EXCHANGE [BROADCAST]
| |
| 01:SCAN HDFS [default.yy2]
| partitions=1/1 files=1 size=620B
|
00:SCAN HDFS [default.yy]
partitions=5/5 files=5 size=1.71KB
runtime filters: RF000 -> year <-- consumer
```
I think the only change we would need to track things is:
1. add 2 methods to `ExecutionPlanNode`: `get_sender_ids() -> Vec<usize>` and `get_receiver_ids()`
2. have the rules that connect 2 operators to increment and assign those counters to ops
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] make dynamic filters in `ExecutionPlan` nodes discoverable [datafusion]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on issue #23814:
URL: https://github.com/apache/datafusion/issues/23814#issuecomment-5051005780
> We could infer from context (ex. `HashJoinExec` is always a producer), but then there's the question of custom `ExecutionPlan` implementations. I figured we'd want them to declare themselves to be producers or consumers.
Exactly. There is a draft PR in Ballista to add a custom hash join operator (with spilling). We would want DataFusion to treat it as a HashJoinExec even though it is a custom implementation.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] make dynamic filters in `ExecutionPlan` nodes discoverable [datafusion]
Posted by "jayshrivastava (via GitHub)" <gi...@apache.org>.
jayshrivastava commented on issue #23814:
URL: https://github.com/apache/datafusion/issues/23814#issuecomment-5050523973
cc @alamb @timsaucer @andygrove @gene-bordegaray @gabotechs @adriangb Regarding our discussion during the community meeting today.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] make dynamic filters in `ExecutionPlan` nodes discoverable [datafusion]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on issue #23814:
URL: https://github.com/apache/datafusion/issues/23814#issuecomment-5050726580
Thanks for creating this issue @jayshrivastava, and thanks for linking to https://github.com/apache/datafusion/pull/23282.
My opinion (which I know may not be shared by others) is that it would be beneficial for "core" DataFusion to offer a choice of execution modes:
1. Eager execution of the whole query tree, streaming results immediately (current model)
2. A **distributed-style** execution, but without the distributed (networking, multi-process) part. Essentially this would model dfd/ballista style execution and allow for development of related capabilities in core. I would be fine with this being an optional crate. This was the idea I was trying to demonstrate in #23282
Expanding further on this, enabling a distributed-style execution path with optional persistence of data between stages could allow single process DF to scale more on certain workloads (even though performance may not be amazing).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] make dynamic filters in `ExecutionPlan` nodes discoverable [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on issue #23814:
URL: https://github.com/apache/datafusion/issues/23814#issuecomment-5051433022
> 1. Distributed projects (ballista, comet, datafusion-distributed) should be responsible for implementing dynamic filter update routing and propagation during query execution
Agree (I also agree with your other points)
I think you have three usecases:
1. A way to discover all dynamic filters in a aplan
2. A way to discover, for each `ExecutionPlan`, which dynamic filter it **produces**
3. A way to discover, for each `ExecutionPlan`, which dynamic filter it **consumes**
I think a "consumer" is a consumer simply by nature of evaluating an expression tree that happens to have [`DynamicFilterPhysicalExpr`](https://github.com/apache/datafusion/blob/c1180022e63f48a3af175d8d73dc21cca9379c65/datafusion/physical-expr/src/expressions/dynamic_filters/mod.rs#L69) somewhere.
There is no explicit way to find out what nodes are "producer"s today from what I can tell -- the only thing that makes an ExecutionPlan a producer is if it pushes dynamic filter expressions down in [`FilterPushdownPhase::post`](https://github.com/apache/datafusion/blob/c1180022e63f48a3af175d8d73dc21cca9379c65/datafusion/physical-plan/src/filter_pushdown.rs#L66-L78)
Note that `DynamicFilterPhysicalExpr` [already implements](https://github.com/apache/datafusion/blob/c1180022e63f48a3af175d8d73dc21cca9379c65/datafusion/physical-expr/src/expressions/dynamic_filters/mod.rs#L593-L595) [`PhysicalExpr::expression_id`](https://github.com/apache/datafusion/blob/c1180022e63f48a3af175d8d73dc21cca9379c65/datafusion/physical-expr-common/src/physical_expr.rs#L464-L479) so there is a unique way to identify each expression
To satisfy the usescase then I think the best set of APIs are:
1. A generic API to walk and modify expressions on ExecutionPlan similar to `LogicalPlan::map_expressions` and `LogicalPlan::apply_expressions` (basically what @LiaCastaneda added in #20337 )
2. Add an API to `ExecutionPlan` to return what dynamic filters it **produced** (basically promote existing methods like [SortExec::dynamic_filters](https://github.com/apache/datafusion/blob/c1180022e63f48a3af175d8d73dc21cca9379c65/datafusion/physical-plan/src/sorts/sort.rs#L1102) and [HashJoinExec::dynamic_filters](https://github.com/apache/datafusion/blob/c1180022e63f48a3af175d8d73dc21cca9379c65/datafusion/physical-plan/src/joins/hash_join/exec.rs#L957) to the `ExecutionPlan` trait)
```rust
trait ExecutionPlan
fn dynamic_filters(&self) -> Vec<Arc<DynamicFilterPhysicalExpr>> {
Vec::new()
}
```
I think adding `ExecutionPlan::dynamic_filters` is better than an enum with producer/consumer as it is clear it is only about what filters are *produced* (not consumed)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] make dynamic filters in `ExecutionPlan` nodes discoverable [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on issue #23814:
URL: https://github.com/apache/datafusion/issues/23814#issuecomment-5051471895
> Add an API to ExecutionPlan to return what dynamic filters it produced (basically promote existing methods like [SortExec::dynamic_filters](https://github.com/apache/datafusion/blob/c1180022e63f48a3af175d8d73dc21cca9379c65/datafusion/physical-plan/src/sorts/sort.rs#L1102) and [HashJoinExec::dynamic_filters](https://github.com/apache/datafusion/blob/c1180022e63f48a3af175d8d73dc21cca9379c65/datafusion/physical-plan/src/joins/hash_join/exec.rs#L957) to the ExecutionPlan trait)
FWIW this would be almost free today (as the method already exists on several of the nodes) and gives you the "producer" side immediately with minimal downstream impact
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] make dynamic filters in `ExecutionPlan` nodes discoverable [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on issue #23814:
URL: https://github.com/apache/datafusion/issues/23814#issuecomment-5253078898
I think this was fixed in
- https://github.com/apache/datafusion/pull/24068
- https://github.com/apache/datafusion/pull/20337
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] make dynamic filters in `ExecutionPlan` nodes discoverable [datafusion]
Posted by "jayshrivastava (via GitHub)" <gi...@apache.org>.
jayshrivastava commented on issue #23814:
URL: https://github.com/apache/datafusion/issues/23814#issuecomment-5137657925
👋🏽 I just opened a PR to re-add `apply_expressions`. I modified the original PR to address some of the concerns we discussed in this issue.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org