Repository navigation
store partitioning in dynamic filters and remove CASE hash(expr) #23817
Description
Is your feature request related to a problem or challenge?
If dynamic filters stored (a) the partitioning of the producer and (b) per-partition expressions, then we could avoid using CASE hash(expr) for co-partitioned data sources
Related: #23376 (comment)
To support cases like below where the partitioning of the dynamic filter producer (hash join) is different than the consumer (data source), the hash join uses a CASE hash(expr) to "bake" partitioning in to the join.
DynamicFilterPhysicalExpr:
┌────────────────────────┐
│ HashJoinExec: │ CASE Hash(a@0) % 4
│mode=Partitioned on=a@0 │ WHEN 0: a@0 >= v0 AND a@0 <= v1
└────────────────────────┘ WHEN 1: a@0 IN (SET) ([v2, v3, v4 ...])
▲ ▲ ... (1 case per partition)
│ │
┌────────────────┘ └────────────────┐ │
│ │ Pushed down to data source
┌──────────────────────────┐ ┌──────────────────────────┐
│ DataSourceExec: │ │ RepartitionExec: │ │
│Partitioning=Hash(a, 4) │ │ Partitioning=Hash(a, 4) │ │
└──────────────────────────┘ └──────────────────────────┘ │
│ │
│ │
│ │
│ │
... any number of RepartitionExecs or │
plan nodes. It does not matter │
│ │
│ │
│ ▼
┌──────────────────────────┐
│ DataSourceExec: │ Each partition uses the same DynamicFilterPhysicalExpr, except
│ Partitioning=Unknown(10) │ each row will only hash to one case.
└──────────────────────────┘
What if the partitioning of the data source is the same as the hash join? In this case, the CASE hash(expr) is pure overhead.
┌────────────────────────┐
│ HashJoinExec: │
│mode=CollectLeft on=a@0 │
└────────────────────────┘
▲ ▲
│ │
┌────────────────┘ └────────────────┐
│ │
┌──────────────────────────┐ ┌──────────────────────────┐
│ DataSourceExec: │ │ DataSourceExec: │
│Partitioning=Hash(a, 4) │ │ Partitioning=Hash(4) │
└──────────────────────────┘ └──────────────────────────┘
It would be nice if the DataSourceExec could consume per-partition dynamic filters. I think most DataSourceExec have to downcast PhysicalExpr to DynamicFilterPhysicalExpr to consume them anyways. If the DynamicFilterPhysicalExpr stored expressions per partition, the DataSourceExec could apply per-partition filters rather than use the CASE hash(expr).
Note that the DataSourceExec must verify that its partitioning is equal to the partitioning of the dynamic filter.
Describe the solution you'd like
A very highlevel design
Today dynamic filters store one expression which is updated atomically:
struct Inner {
expression_id: u64,.
generation: u64,
expr: Arc<dyn PhysicalExpr>, // the actual expression
is_complete: bool,
}
impl DynamicFilterPhysicalExpr {
// Atomically update the expression
pub fn update(&self, new_expr: Arc<dyn PhysicalExpr>) -> Result<()>;
// Atomically get the current expression
pub fn current(&self) -> Result<Arc<dyn PhysicalExpr>>;
}This is how we could make it partition-aware:
type Expr = Arc<dyn PhysicalExpr>;
struct Inner {
expression_id: u64,
generation: u64,
expr: LiveFilterExpr,
is_complete: bool,
}
// Either a global expression or a map of per-partition expressions
enum LiveFilterExpr {
Global(Arc<dyn PhysicalExpr>;),
Partitioned(PartitionedFilterExpr),
}
struct PartitionedFilterExpr {
// Store the producer partitioning scheme so we know if the consumer is equivalently partitioned
partitioning: Partitioning,
cases: BTreeMap<usize, Arc<dyn PhysicalExpr>;>,
}
struct PartitionCtx {
partition_id: usize,
partitioning: Partitioning,
}
impl LiveFilterExpr {
fn current(&self) -> Result<Arc<dyn PhysicalExpr>> {
match self {
Self::Global(expr) => Ok(Arc::clone(expr)),
Self::Partitioned(p) => {
// Generate CASE expression over all the partitions
}
}
}
// Useful for combining two predicates. The caller does not have to downcast through a CASE expression etc.
// The expression can internally OR things together.
fn union(&self, other: LiveFilterExpr) -> Result<()> {
// OR the current expression with other or do a partition-wise OR.
}
}
impl DynamicFilterPhysicalExpr {
// Update the entire expression or the expression for a particular partition.
pub fn update(&self, expr: Arc<dyn PhysicalExpr>, ctx: Option<&PartitionCtx>) -> Result<()> {
match (&mut inner.expr, ctx) {
(LiveFilterExpr::Global(current), None) => *current = expr,
(LiveFilterExpr::Partitioned(p), Some(ctx))
if p.partitioning == ctx.partitioning
&& ctx.partition_id < p.partitioning.partition_count() => {
p.cases.insert(ctx.partition_id, expr);
}
_ => return internal_err!("Dynamic-filter partition context mismatch"),
}
inner.generation += 1;
Ok(())
}
// Potentially get an expression for the provided partition id rather than always getting a CASE expression
pub fn current(&self, ctx: Option<&PartitionCtx>) -> Result<Arc<dyn PhysicalExpr>;> {
let inner = self.inner.read();
if let (LiveFilterExpr::Partitioned(p), Some(ctx)) = (&inner.expr, ctx)
&& p.partitioning == ctx.partitioning
&& ctx.partition_id < p.partitioning.partition_count()
{
return Ok(p.cases.get(&ctx.partition_id).cloned().unwrap_or_else(|| lit(true)));
}
inner.expr.current()
}
}We would probably want to refactor datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs (~700 LoC) to migrate the partition specific logic into the DynamicFilterPhysicalExpr itself.
Describe alternatives you've considered
No response
Additional context
No response
Activity
stuhood commentedon Jul 24, 2026
I believe that the general thrust of this design makes sense.
It feels like the analogy is that while data flows up a plan (via
ExecutionPlans), it is partitioned. Likewise, when data flows down a plan (via dynamic filters), is makes sense not to lose that partitioning information either.But the largest problem I see is that the partitioning information will be duplicated in the
ExecutionPlans and in the filters. That means that planning would need to coordinate its phases even more carefully than before to ensure that they're not out of sync. But perhaps that can always be backstopped with asserts on the producing/consuming sides?jayshrivastava commentedon Jul 24, 2026
I'm not 100% sure if that's a problem since dynamic filters already store partitioning, baked into the
CASEstatement. Storing it separately should behave similarly. At execution time, I would expect theDataSourceto take an optional optimization path where, if the partitioning is equivalent, it can optionally call acurrent_partitionedAPI instead ofcurrentlike it does today.The partitioning of the filter is with respect to the producer (ex. hash join) and doesn't change, so it shouldn't be affected by any repartitions in between the producer and consumer.
[-]store partitioning in dynamic filters[/-][+]store partitioning in dynamic filters and remove CASE hash(expr)[/+]stuhood commentedon Jul 28, 2026
The difference (I think) is that a
CASEexpression is a "total" function: it correctly handles all inputs that have been given to it, even if it hasn't been repartitioned correctly/at-all. Whereas each expression inside a partition-aware dynamic filter would be "partial": if you don't correctly choose which expression to evaluate, you produce the wrong result.But as mentioned, I think that asserts could be sufficient.
peterxcli commentedon Jul 31, 2026
This structure looks good, I have two comments:
casescould be simple list, right? because partitions id must be continuous.partition_expr, if want to fit range partitioning in this structure, then it would be sth like:or maybe a
PhysicalExprwhich is likelower_bound.Per #23376 (comment), but now, because we've decoupled the routing expression (
partition_expr) from the per-partition filters (cases), we can brainstorm whether there are any other optimal approaches.stuhood commentedon Aug 12, 2026
@jayshrivastava with a bit more context and after further discussion with @barbarj, I think that isolating filters per-partition is going to be a very important piece. We're interested in helping to push this forward!
peterxcli commentedon Aug 17, 2026
@jayshrivastava I think we could use a vector of optional physical exprs to store the sparse partition filters, to reduce the BTree lookup overhead. We could also add a new fallback for those edge cases like empty partitions or canceled/unknown partitions that we originally handled.
ref: #24417
jayshrivastava commentedon Aug 17, 2026
@peterxcli That seems reasonable!
devdattatalele commentedon Sep 13, 2026
I'd like to help drive this, and I can bring a concrete consumer to the table. I work on ParadeDB's pg_search, where right now every partitioned-mode join dynamic filter contributes zero pruning: our index pushdown only recurses AND chains, so a CASE root extracts no InLists, and our fast-field prefilter rejects the conjunct because CaseExpr is not in its allowlist. The sharp edge is that our range co-partitioned join rewrite is itself what flips eligible joins into Partitioned mode, so our own optimization currently disables our own pruning. With the structured form, a range-partitioned scan that already knows its target partition can select cases[partition] and push exactly that partition's InList and bounds into the index, which is better than what even the CollectLeft encoding offers.
On the total vs partial concern: I think it is structurally defused by keeping current() lowered to the same total CASE the join builds today, and making per-partition evaluation strictly opt-in through current_partitioned(partitioning), returning Some only when the consumer proves equivalence: same variant, equal partition count, partition expressions equivalent after remapping into the consumer schema, identical hash random state, and excluding preserve_file_partitions-derived Hash metadata, since #20195 is precisely the case where dynamic filtering is disabled today because the declared partitioning is not proof that partition indexes follow the hash router. Anything that cannot prove equivalence uses the total lowering, with debug asserts on both sides.
Concretely I would slice it as: (1) LiveFilterExpr plus PartitionedFilterExpr with the shape agreed above, partition-scoped update, and current() lowering that reproduces build_partitioned_filter's fast paths, zero behavior change; (2) producer migration in shared_bounds; (3) consumer opt-in where DataSourceExec::execute knows its partition; (4) proto following the #24618 destructure convention; with benchmarks per #24095. I would defer the distributed union() semantics to a follow-up since RFC #553 is still open, and #24235's InList-collapse idea is worth folding into the lowering rather than competing with it.
@jayshrivastava @peterxcli do either of you intend to implement this? If not, I will take it in those slices, and if you have already started I can pick up whichever pieces are free.
stuhood commentedon Sep 13, 2026
This isn't accurate under
datafusion-distributed, which is the only place where we should have more than 1 partition. So I do not think that this issue is relevant to ParadeDB.But having said that, I still think that it is a very important issue.