Push dynamic filters across network boundaries

[adriangb]

Background: https://datafusion.apache.org/blog/2025/09/10/dynamic-filters/

Currently dynamic filters don't work with distributed-datafusion. In order to make them work we need upper stages to be able to update the lower stages:

┌─────────────────────────────────────────────────────────────────────────┐                   
│                                Stage 2                                  │                   
│                                                                         │                   
│ ┌─────────────────────────────────────────────────────────────────────┐ │                   
│ │                                                                     │ │                   
│ │                                                                     │ │                   
│ │                             SortExec (TopK)                         │ │                   
│ │                        ┌──────────────────────┐                     │ │                   
│ │                        │DynamicFilter [ col > │─────────────────────┼─┼─────────────┐     
│ └────────────────────────┴──────────────────────┴─────────────────────┘ │             │     
│ ┌─────────────────────────────────────────────────────────────────────┐ │             │     
│ │                                                                     │ │             │     
│ │                                                                     │ │             │     
│ │                            NetworkShuffleExec                       │ │             │     
│ │                                                                     │ │             │     
│ │                                                                     │ │             │     
│ └─────────────────────────────────────────────────────────────────────┘ │             │     
└─────────────────────────────────────────────────────────────────────────┘             │     
                                                                                      Push    
┌─────────────────────────────────────────────────────────────────────────┐          updates  
│                                Stage 1                                  │          across   
│                                                                         │          network  
│ ┌─────────────────────────────────────────────────────────────────────┐ │         boundary  
│ │                                                                     │ │             │     
│ │                                                                     │ │             │     
│ │                            RepartitionExec                          │ │             │     
│ │                                                                     │ │             │     
│ │                                                                     │ │             │     
│ └─────────────────────────────────────────────────────────────────────┘ │             │     
│ ┌─────────────────────────────────────────────────────────────────────┐ │             │     
│ │                                                                     │ │             │     
│ │                                                                     │ │             │     
│ │                            DataSourceExec                           │ │             │     
│ │                        ┌──────────────────────┐                     │ │             │     
│ │                        │DynamicFilter [ true ]│◀────────────────────┼─┼─────────────┘     
│ └────────────────────────┴──────────────────────┴─────────────────────┘ │                   
└─────────────────────────────────────────────────────────────────────────┘                              

There's several moving pieces to this:

1. We need to add some hooks to DataFusion's DynamicFilterPhysicalExpr to get a callback when a filter is updated.
2. We need some way to identify each DynamicFilterPhysicalExpr so that we can relate them across the wire.
3. When we serialize out we need to detect each DynamicFilterPhysicalExpr and kick off some bookkeeping. I imagine this might be something like hook into updates via (1) and then set up some sort of broadcast channel for updates.
4. When we deserialize we need to detect each DynamicFilterPhysicalExpr and subscribe to updates via its broadcast channel.

The main thing that is not clear in my mind is:

1. Is serialization and deserialization the right place to discover all DynamicFilterPhysicalExprs? Do we have enough information to set up the broadcast / network communication there? If not I think we'd have to add APIs to DataFusion something like ExecutionPlan::filters() to discover all of the dynamic filters. Maybe we can re-use ExecutionPlan::gather_filters_for_pushdown? I'm not sure. I'm also not sure where we would do this discovery from (optimizer rule? some new place? NetworkShuffleExec::execute?)
2. How are we going to do the network communication? Some sort of broadcast channel? A gRPC call that just hangs until there is an update?

I also think we should be conscious of how invasive the changes to DataFusion are. It would be great to minimize how much we add APIs and complexity just to support distributed execution.

[adriangb]

@LiaCastaneda @gabotechs any thoughts on these items?

[gabotechs]

🤔 Some idea that comes to mind: at planning time, wrapping all the DynamicFilterPhysicalExprs with DistributedDynamicFilterPhysicalExpr, and do nothing else during planing or protobuf parsing.

During execution, as late and lazily as possible, these DistributedDynamicFilterPhysicalExpr could:

1. if trying to consume the dynamic filters, advertise themselves in the ArrowFlightEndpoint to which they are running, opening a channel ready to listen to updates.
2. if trying to produce new filters, publish the update to the corresponding ArrowFlightEndpoint[s] to which that are either already listening for updates, or that will be listening very soon

All this, "post filters to a remote ArrowFlightEndpoint" and "expect filters to be posted in the current ArrowFlightEndpoint" could be handled lazily during execution by this new DistributedDynamicFilterPhysicalExpr.

This idea is pretty much a brain dump at this point, but at first sight it sounds feasable

[adriangb]

Makes sense. A question and a clarification:

at planning time, wrapping all the DynamicFilterPhysicalExprs with DistributedDynamicFilterPhysicalExpr

When would this happen? I don't think we want to modify the things that produce DynamicFilterPhysicalExprs, they depend on those APIs directly. So would it be an optimizer rule? I don't think there's any good way to discover all PhysicalExprs starting from an ExecutionPlan. We also need to replace are references (e.g. the reference in HashJoinExec and the reference in ParquetSource). It seems hard to me to swap out PhysicalExpr references.

During execution, as late and lazily as possible

I think a key here is to never "poll": you don't want each evaluate() call on each row or even each RecordBatch to result in an HTTP call. The current design focuses on pushing into an RwLock so that reads are very cheap and writes only happen when the filter actually changes. I think this is doable with the endpoints proposal, it would have to be that consumers expose an endpoint that takes in something like (expr_id: i64, value: PhysicalExprNode) and internally holds a reference to every DynamicFilterPhysicalExprs and calls update() on it with the new value. Producers have to keep track of the URLs of their consumers and hook into DynamicFilterPhysicalExprs::update() to push updates to their consumers.

[gabotechs]

So would it be an optimizer rule? I don't think there's any good way to discover all PhysicalExprs starting from an ExecutionPlan.

I was getting the impression that it would be easy given a ExecutionPlan to walk the PhysicalExpr it contains, but you are right that unless we do some heavy downcasting to all possible nodes, this does not seem like an easy task.

I think a key here is to never "poll": you don't want each evaluate() call on each row or even each RecordBatch to result in an HTTP call. The current design focuses on pushing into an RwLock so that reads are very cheap and writes only happen when the filter actually changes. I think this is doable with the endpoints proposal, it would have to be that consumers expose an endpoint that takes in something like (expr_id: i64, value: PhysicalExprNode) and internally holds a reference to every DynamicFilterPhysicalExprs and calls update() on it with the new value. Producers have to keep track of the URLs of their consumers and hook into DynamicFilterPhysicalExprs::update() to push updates to their consumers.

🤔 the networking part does not sound very challenging, what I still don't wrap my head around is:

• how we discover all the DynamicFilterPhysicalExprs in a plan
• how do we handle multiple producers and multiple consumers

For the second point. I imagine there should be already some machinery in place for handling multiple local partitions producing values in the dynamic filter isn't it? if multiple partitions produce different filters, are they just OR-ed together?

[adriangb]

how we discover all the DynamicFilterPhysicalExprs in a plan

My thought was we could do it when we serialize? The serialization already has hooks for "serialize this PhysicalExpr".

if multiple partitions produce different filters, are they just OR-ed together?

They just get replaced by the last one. PhysicalExpr has no idea about partitioning. Hence why there's the whole dance in HashJoinExec where we wait until all build side partitions are done then we push down a combined filter. SortExec uses a single filter across all partitions but it coordinates internally to have a shared heap across all partitions in non-distributed mode which means there is only one filter. So yeah this is going to be gnarly to solve for distributed uses.

[gabotechs]

🤔 one thing that comes to mind that could be a good starting point for kickstarting some work on this front:

How about shipping something initially that does not imply syncing a dynamic filter across the wire?

For example, a CollectLeft hash join that fully waits for the build side to be finished. In this case, whenever we start executing the probe side, the dynamic filter has already all the necessary data, and we are good to send it over the wire as is without further future syncs.

[gabotechs]

Related to this, but under upstream's scope:

apache/datafusion#18296