Repository navigation
[DISCUSSION] Future of Dynamic Filters Sync #21207
Description
Overview
There are several contributors that have been actively working on dynamic filters and have expressed interest in aligning on the path forward. A few open design questions have emerged across recent PRs:
- First-class
DynamicFiltertype vs. continuing to extendPhysicalExpr(feat: add generic runtime binding for physical expressions #20901 ) - Serialization and deduplication strategy (proto: serialize and dedupe dynamic filters #20416 , Serialize dynamic filters across network boundaries #20418 )
- Cross-network propagation in distributed execution (Push dynamic filters across network boundaries datafusion-contrib/datafusion-distributed#180)
We'd like to schedule one public sync call/calls to talk more about the future of dynamic filters. Please let us know availability or questions.
Tuesday, March 31st at 11:30 AM ET
Link: https://calendar.google.com/calendar/u/0/r/eventedit/copy/MnBzMnMyOHY3dWY2czBxZjRxMzVyMjczODAgZ2VuZS5ib3JkZWdhcmF5QGRhdGFkb2docS5jb20/Z2VuZS5ib3JkZWdhcmF5QGRhdGFkb2docS5jb20
cc @alamb @adriangb @jayshrivastava @gabotechs @liamzwbao @stuhood @geoffreyclaude @NGA-TRAN
Activity
gene-bordegaray commentedon Mar 27, 2026
I am available Tuesday-Thursday 8 AM-11 AM EST and 1 PM-EOD every day
[-]Discussion: Future of Dynamic Filters Sync[/-][+][Discussion] Future of Dynamic Filters Sync[/+][-][Discussion] Future of Dynamic Filters Sync[/-][+][DISCUSSION] Future of Dynamic Filters Sync[/+]stuhood commentedon Mar 27, 2026
I'm available Monday-Friday 8:30am-5:15pm in PT, except for 10:30am on Tuesday and Thursday... obviously there are some exceptions, but I can share a bookable calendar link to see those.
NGA-TRAN commentedon Mar 27, 2026
I am available Mon-Fri, 9:30am-12pm ET, but not next Monday March 30th.
Samyak2 commentedon Mar 28, 2026
I am available anytime before 11pm IST (which is 1:30pm ET / 10:30am PT I believe).
gene-bordegaray commentedon Mar 29, 2026
Ok lets do for Tuesday, March 31st at 11:30 AM ET. Let me know if this is ok with you all
adriangb commentedon Mar 29, 2026
stuhood commentedon Mar 29, 2026
I can do 11:30 ET, thanks. 10:30 ET would be fine as a one-off, but probably not recurring.
jayshrivastava commentedon Mar 29, 2026
11:30am EST on March 31st works for me. I can't make 11am-11:30am.
gene-bordegaray commentedon Mar 29, 2026
ok 11-11:30 AM ET on Tuesday it is
here is the calendar invite: https://calendar.google.com/calendar/event?action=TEMPLATE&tmeid=N2s2NnA1b2ZvZGhhc3J1c2NraDM3ZGlsdGMgZ2VuZS5ib3JkZWdhcmF5QGRhdGFkb2docS5jb20&tmsrc=gene.bordegaray%40datadoghq.com
see you all then 👍
cc: @jayshrivastava @NGA-TRAN @stuhood @adriangb @Samyak2
adriangb commentedon Mar 29, 2026
gene-bordegaray commentedon Mar 30, 2026
I have tried updating it and if not here is the direct zoom link:
https://datadog.zoom.us/j/92443847794?pwd=N0fOajbIGnjybXSBv3jgz85yHdg9eE.1&jst=2
LiaCastaneda commentedon Mar 30, 2026
Hey, I can make it too. Thanks for organizing this!
geoffreyclaude commentedon Mar 30, 2026
Thanks for organizing indeed! I can make it, but I'm not sure I'll be super useful as I haven't followed dynamic filters too closely...
58 remaining items
jayshrivastava commentedon Jul 17, 2026
Sounds good. I'll send a PR. Can you clarify what you mean by option 2?
adriangb commentedon Jul 17, 2026
IIRC currently we have something like
CASE hash(c1) WHEN 0 THEN c1p0min < c1 < c1p0max AND c1 IN (1, 2, 3) WHEN 1 THEN c1p1min < c1 < c1p1max AND c1 IN (4, 5, 6) END. Option 2 would be to change it tomin(c1p0min, c1p1min) < c1 < min(c1p0max, c1p1max) AND (c1p0min < c1 < c1p0max OR c1p1min < c1 < c1p1max) AND CASE hash(c1) WHEN 0 THEN c1 IN (1, 2, 3) WHEN 1 THEN c1 IN (4, 5, 6) END. I think this is interesting because it would let adaptive expression evaluation do things like "hey the global bounds are 99% selective and cheap to evaluate, let's just ignore the other expressions" or "stats pruning from the ranges was so effective that evaluating them again on the rows is pointless, let's disable them but keep the membership expressions being evaluated per row", etc. Basically the range and membership parts tend to have different profiles of applicability and performance so treating them independently could be beneficial.This can be stacked with other ideas i.e. we might change the
CASEpart to beORedinstead ofCASE:min(c1p0min, c1p1min) < c1 < min(c1p0max, c1p1max) AND (c1p0min < c1 < c1p0max OR c1p1min < c1 < c1p1max) AND (c1 IN (1, 2, 3) OR c1 IN (4, 5, 6)),min(c1p0min, c1p1min) < c1 < min(c1p0max, c1p1max) AND (c1p0min < c1 < c1p0max OR c1p1min < c1 < c1p1max) AND c1 IN (1, 2, 3, 4, 5, 6)or some other variation.jayshrivastava commentedon Jul 20, 2026
👋🏽 I ran an expanded benchmark matrix here, comparing roughly the 4 implementations for dynamic filtering mentioned earlier: https://github.com/jayshrivastava/datafusion/blob/js/benchmark-4-dynamic-filtering-alternatives/benchmarks/reports/hash_join_dynamic_filter_expanded_matrix.md
The main takeaway is that I do not think we have strong evidence for keeping the current partition-aware
CASEexpression as the default shape. In the benchmark,CASEdoes not participate in parquet row-group pruning, which matters the most for performance.The best-performing options were generally
global(ie.global_bounds AND global_membership) and something very similar to option 2 you mentioned. Although, the nonCASEparts of option 2 are by far the most important because they can be used for parquet pruning. The global filter implementation also gets an edge because it's simple. It would make dynamic filtering in distributed cases like DFD, ballista etc a bit easier to implement. Notably, trino basically just merges filters across partitions too.adriangb commentedon Jul 20, 2026
The one note w.r.t. a single global membership: the merging itself can be expensive if it implies copying 128 hash tables into 1 hash table (for a plan with 128 partitions). For the smaller
InListexpressions I think the cost can more easily be amortized / we are already copying the data. We tried options that don't involve copying, like hashing once and probing each partition's hash table without rehashing, but that proved to be slower at high partition counts, see #21931.cc @zhuqi-lucas since you're working on this stuff too
jayshrivastava commentedon Jul 20, 2026
Took a brief look at #21931.
The implementation in my PR happens to be
InList:concatthe expressionsMap:ORthe mapsNot sure if the
ORis better or worse thanMultiMapLookupExpr.In the
MultiMapLookupExprwe process all rows in all the maps. I thinkORalso ends up vectorizing the hash lookup for each row for each map. I wonder if it's worth exploring a way to prune the record batch as soon as one row evaluatestruein theOR, skipping subsequent maps.stuhood commentedon Jul 20, 2026
So, FWIW: when we consume
InList(e.g. "small") filters in our scans, we frequently tear them apart in order to evaluate them using an index, rather than usingArrow.If a dynamic filter has a lot of consumers then maybe it might make sense to execute these types of merges eagerly... but otherwise, letting the consumer decide whether it wants to further "optimize" it (concatenating, merging, etc) before applying it would be good for us.
adriangb commentedon Jul 20, 2026
@stuhood would it matter for your usage if you got
CASE partition_routing_hash(col) WHEN 0 THEN col IN <list 1> WHEN 1 THEN col IN <list 2> ... ENDv.s.col IN <list1> OR col IN <list2> ...vscol IN <merged list>?stuhood commentedon Jul 20, 2026
The case expression is more difficult to push into the index, but not impossible. We don't currently handle them.
I suppose that one way to think about it though is that the
CASEexpression exposes strictly more information than theOR'd lists do: it may be more or less cheap for a particular expression evaluator to load the partitioning key vs the in-list column... and depending on the list size, each of those portions of the expression may be more or less selective.So maybe the right answer is to expose as much information as you have, and then let the consumer decide what to do with it? But that sortof hinges on being able to recognize the
CASEpattern as an optimization relative to OR'd lists, so that you can choose to ignore theCASEportion andORthem (EDIT: I suppose that whenever a dynamic filter is best-effort, the consumer can safely choose to ignore some of the conditions, and convert aCASEinto anOR'd list if they want?)adriangb commentedon Jul 20, 2026
I thought we always end up with a
CASEexpression now unless there is only 1 partition?stuhood commentedon Jul 20, 2026
We always use one partition, and get parallelism from parallel workers (via
df-d)... we're weird, so don't overfit too much for us 😅zhuqi-lucas commentedon Jul 21, 2026
Thanks for the ping @adriangb — indeed I've been exploring this exact trade-off in #23701.
Our approach there is a simpler "union bounds" shape for Partitioned mode:
col >= global_min AND col <= global_max(equivalent to what CollectLeft mode already emits). It sacrifices per-partition selectivity for O(1) per-row cost regardless of N.The trade-off matrix as I see it:
Union bounds relies on the downstream HashJoin's hash lookup for exact membership — semantics preserving, and cheap per-row at any N. But it loses per-partition selectivity, so for skewed / low-density joins membership-based filters would prune more.
jayshrivastava commentedon Jul 21, 2026
On this topic, for certain data sources, it's difficult to translate and pushdown
CASE hash(expr). For example, if the data source is a an API endpoint implemented in a different language or a some SQL database, it's thehashfunction is not easily translated. For these cases, I think theCASE hashshould be opt-in or easy removable (ex. we currently have a merge() implementation which merges all per-partition expressions into a range + in list).stuhood commentedon Jul 21, 2026
Yea, possibly. I think that this is another flavor of:
It seems like it's a question of whether you need a setting to disable
CASEexpressions (which might harm unrelated consumers of the filter who can consume a case expression), or whether there is a way that a consumer can "simplify an expression (by making it lossy)" at runtime.i.e.: who chooses to simplify-the-expression/discard-the-extra-information?
jayshrivastava commentedon Jul 22, 2026
Filed a related issue to the above discussion: #23817
I think this optimization is possible but not implemented:
If the partitioning of the producer partitioned hash join matches the data source (range or hash I suppose), then the
CASEtechnically wouldn't be needed. You could push down a separate dynamic filter per partition and have the data source apply those. I think one would have to store partitioningDynamicFilterPhysicalExprto make it work though.stuhood commentedon Jul 24, 2026
@jayshrivastava : Commented on #23817: I think that it could make sense, but the exact details of how it interacts with
EnforceDistribution,RepartitionExec, etc are going to be fidgety. It's only safe to consume a partition-specific-subset of the filter if it has been carefully re-partitioned on the way down the plan (which is a new requirement relative to the current state).