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: [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]
