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: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
