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

[adriangb]

I think the next step is apache/datafusion#19437. Would any of you be willing to write the code, do the reviews or offer stewardship to get this across the line?

The plan would be to complete this refactor then get back to #18192 and implement the deduplication on deserialization there.

[LiaCastaneda]

I can take a look. Just a question: after #19437 is complete, shouldn't we do #19580 first before working on this issue? It would avoid having to refactor the serialization/deserialization later, and make the implementation simpler i think

[adriangb]

I can take a look. Just a question: after #19437 is complete, shouldn't we do #19580 first before working on this issue? It would avoid having to refactor the serialization/deserialization later, and make the implementation simpler i think

Yes I think that makes sense 👍🏻

[adriangb]

Actually I think this issue and apache/datafusion#19580 could be worked on at the same time, they shouldn’t interact, the machinery to send dynamic filter expressions across the wire should be agnostic to what’s in those expressions or how the expressions that surround them are structured

[adriangb]

A quick update here.

I have a PR in DataFusion to deduplicate DynamicFilterPhysicalExps when deserialized from Protobuf. The new APIs in apache/datafusion#19437 also means we can implement a visitor pattern for each deserialized DynamicFilterPhysicalExp. The end result should be that it is possible to place a node in the plan that listens for filter updates across the wire and can call update() on it's reference and have that be reflected in the DataSourceExecs. We still need to implement that node and come up with a scheme to send those updates across the wire.

[gabotechs]

One challenge that comes to mind is that typically not all DynamicFilterPhysicalExp will suffer any serialization/deserialization in a distributed query, take TPC-DS query 20 as an example:

┌───── DistributedExec ── Tasks: t0:[p0] 
│ SortPreservingMergeExec: [i_category@2 ASC, i_class@3 ASC, i_item_id@0 ASC, i_item_desc@1 ASC, revenueratio@6 ASC], fetch=100
│   SortExec: TopK(fetch=100), expr=[i_category@2 ASC, i_class@3 ASC, i_item_id@0 ASC, i_item_desc@1 ASC, revenueratio@6 ASC], preserve_partitioning=[true]
│     ProjectionExec: expr=[i_item_id@0 as i_item_id, i_item_desc@1 as i_item_desc, i_category@2 as i_category, i_class@3 as i_class, i_current_price@4 as i_current_price, sum(catalog_sales.cs_ext_sales_price)@5 as itemrevenue, CAST(sum(catalog_sales.cs_ext_sales_price)@5 AS Float64) * 100 / CAST(sum(sum(catalog_sales.cs_ext_sales_price)) PARTITION BY [item.i_class] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING@6 AS Float64) as revenueratio]
│       WindowAggExec: wdw=[sum(sum(catalog_sales.cs_ext_sales_price)) PARTITION BY [item.i_class] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING: Ok(Field { name: "sum(sum(catalog_sales.cs_ext_sales_price)) PARTITION BY [item.i_class] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING", data_type: Decimal128(27, 2), nullable: true }), frame: WindowFrame { units: Rows, start_bound: Preceding(UInt64(NULL)), end_bound: Following(UInt64(NULL)), is_causal: false }]
│         SortExec: expr=[i_class@3 ASC NULLS LAST], preserve_partitioning=[true]
│           [Stage 4] => NetworkShuffleExec: output_partitions=3, input_tasks=2
└──────────────────────────────────────────────────
  ┌───── Stage 4 ── Tasks: t0:[p0..p2] t1:[p0..p2] 
  │ RepartitionExec: partitioning=Hash([i_class@3], 3), input_partitions=3
  │   AggregateExec: mode=FinalPartitioned, gby=[i_item_id@0 as i_item_id, i_item_desc@1 as i_item_desc, i_category@2 as i_category, i_class@3 as i_class, i_current_price@4 as i_current_price], aggr=[sum(catalog_sales.cs_ext_sales_price)]
  │     [Stage 3] => NetworkShuffleExec: output_partitions=3, input_tasks=3
  └──────────────────────────────────────────────────
    ┌───── Stage 3 ── Tasks: t0:[p0..p5] t1:[p0..p5] t2:[p0..p5] 
    │ RepartitionExec: partitioning=Hash([i_item_id@0, i_item_desc@1, i_category@2, i_class@3, i_current_price@4], 6), input_partitions=2
    │   AggregateExec: mode=Partial, gby=[i_item_id@1 as i_item_id, i_item_desc@2 as i_item_desc, i_category@5 as i_category, i_class@4 as i_class, i_current_price@3 as i_current_price], aggr=[sum(catalog_sales.cs_ext_sales_price)]
    │     HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(CAST(date_dim.d_date_sk AS Float64)@1, cs_sold_date_sk@0)], projection=[cs_ext_sales_price@3, i_item_id@4, i_item_desc@5, i_current_price@6, i_class@7, i_category@8]
    │       CoalescePartitionsExec
    │         [Stage 1] => NetworkBroadcastExec: partitions_per_consumer=3, stage_partitions=9, input_tasks=2
    │       ProjectionExec: expr=[cs_sold_date_sk@5 as cs_sold_date_sk, cs_ext_sales_price@6 as cs_ext_sales_price, i_item_id@0 as i_item_id, i_item_desc@1 as i_item_desc, i_current_price@2 as i_current_price, i_class@3 as i_class, i_category@4 as i_category]
    │         HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(i_item_sk@0, cs_item_sk@1)], projection=[i_item_id@1, i_item_desc@2, i_current_price@3, i_class@4, i_category@5, cs_sold_date_sk@6, cs_ext_sales_price@8]
    │           CoalescePartitionsExec
    │             [Stage 2] => NetworkBroadcastExec: partitions_per_consumer=3, stage_partitions=9, input_tasks=2
    │           PartitionIsolatorExec: t0:[p0,p1,__,__,__,__] t1:[__,__,p0,p1,__,__] t2:[__,__,__,__,p0,p1]
    │             DataSourceExec: file_groups={6 groups: [[/testdata/tpcds/plans_sf1_partitions4/catalog_sales/part-0.parquet:<int>..<int>], [/testdata/tpcds/plans_sf1_partitions4/catalog_sales/part-1.parquet:<int>..<int>], [/testdata/tpcds/plans_sf1_partitions4/catalog_sales/part-1.parquet:<int>..<int>], [/testdata/tpcds/plans_sf1_partitions4/catalog_sales/part-2.parquet:<int>..<int>], [/testdata/tpcds/plans_sf1_partitions4/catalog_sales/part-2.parquet:<int>..<int>], ...]}, projection=[cs_sold_date_sk, cs_item_sk, cs_ext_sales_price], file_type=parquet, predicate=DynamicFilter [ empty ] AND DynamicFilter [ empty ]
    └──────────────────────────────────────────────────
      ┌───── Stage 1 ── Tasks: t0:[p0..p8] t1:[p9..p17] 
      │ BroadcastExec: input_partitions=3, consumer_tasks=3, output_partitions=9
      │   ProjectionExec: expr=[d_date_sk@0 as d_date_sk, CAST(d_date_sk@0 AS Float64) as CAST(date_dim.d_date_sk AS Float64)]
      │     FilterExec: d_date@1 >= 1999-02-22 AND d_date@1 <= 1999-03-24, projection=[d_date_sk@0]
      │       RepartitionExec: partitioning=RoundRobinBatch(3), input_partitions=2
      │         PartitionIsolatorExec: t0:[p0,p1,__,__] t1:[__,__,p0,p1]
      │           DataSourceExec: file_groups={4 groups: [[/testdata/tpcds/plans_sf1_partitions4/date_dim/part-0.parquet], [/testdata/tpcds/plans_sf1_partitions4/date_dim/part-1.parquet], [/testdata/tpcds/plans_sf1_partitions4/date_dim/part-2.parquet], [/testdata/tpcds/plans_sf1_partitions4/date_dim/part-3.parquet]]}, projection=[d_date_sk, d_date], file_type=parquet, predicate=d_date@2 >= 1999-02-22 AND d_date@2 <= 1999-03-24, pruning_predicate=d_date_null_count@1 != row_count@2 AND d_date_max@0 >= 1999-02-22 AND d_date_null_count@1 != row_count@2 AND d_date_min@3 <= 1999-03-24, required_guarantees=[]
      └──────────────────────────────────────────────────
      ┌───── Stage 2 ── Tasks: t0:[p0..p8] t1:[p9..p17] 
      │ BroadcastExec: input_partitions=3, consumer_tasks=3, output_partitions=9
      │   FilterExec: i_category@5 = Sports OR i_category@5 = Books OR i_category@5 = Home
      │     RepartitionExec: partitioning=RoundRobinBatch(3), input_partitions=2
      │       PartitionIsolatorExec: t0:[p0,p1,__,__] t1:[__,__,p0,p1]
      │         DataSourceExec: file_groups={4 groups: [[/testdata/tpcds/plans_sf1_partitions4/item/part-0.parquet], [/testdata/tpcds/plans_sf1_partitions4/item/part-1.parquet], [/testdata/tpcds/plans_sf1_partitions4/item/part-2.parquet], [/testdata/tpcds/plans_sf1_partitions4/item/part-3.parquet]]}, projection=[i_item_sk, i_item_id, i_item_desc, i_current_price, i_class, i_category], file_type=parquet, predicate=i_category@12 = Sports OR i_category@12 = Books OR i_category@12 = Home, pruning_predicate=i_category_null_count@2 != row_count@3 AND i_category_min@0 <= Sports AND Sports <= i_category_max@1 OR i_category_null_count@2 != row_count@3 AND i_category_min@0 <= Books AND Books <= i_category_max@1 OR i_category_null_count@2 != row_count@3 AND i_category_min@0 <= Home AND Home <= i_category_max@1, required_guarantees=[i_category in (Books, Home, Sports)]
      └──────────────────────────────────────────────────

In this case, I imagine the top level SortExec: TopK(fetch=100) node will hold a reference to the producer side of a dynamic filter, however, as that one belongs to the top most stage, it will enter the protobuf world at all, and therefore, no serialization/deserialization hook will be triggered for it.

[adriangb]

Yeah this is a good point. So maybe we need a ExecutionPlan::expressions() -> Iterator<&Arc<dyn PhysicalExpr>> or something?

[gabotechs]

Something like that would work yes. It would be nice to put some thought to it just to be double sure that the decision ages well through the years, but at first sight that sounds like it should work.

I experimented with something similar for some other reason here:

• Hash join buffering on probe side apache/datafusion#19761

In that PR, I drafted how adding an ExecutionPlan::dynamic_filters(&self) -> Vec<Arc<DynamicFilterExpr>> would look like so that, upon executing a node that buffers data (BufferExec), the buffering node itself looks recursively in its children for dynamic filters in order to hold references to them and wait for them to complete.

I bet that a relatively general way of answering the question of "what are the dynamic filters I have in this ExecutionPlan" has good chances to be future proof and to satisfy several needs.