From daeea94ac7697f4cc8ea60e8b6ef4f15397f0021 Mon Sep 17 00:00:00 2001 From: Siew Kam Onn Date: Tue, 7 Jul 2026 17:48:01 +0800 Subject: [PATCH 1/4] feat: enhance dynamic filtering support in hash join - Added private components: - DynamicFilterExprComposer - FilterComposition variants (NoUpdate, Update) - Moved CollectLeft and Partitioned expression policies into the composer. - Updated SharedBuildAccumulator::build_filter to be a thin delegate with dynamic_filter.update integration. --- .../src/joins/hash_join/shared_bounds.rs | 293 ++++++++++-------- 1 file changed, 171 insertions(+), 122 deletions(-) diff --git a/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs b/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs index 7146e8dc2ec3..8e7caa6199e1 100644 --- a/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs +++ b/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs @@ -325,6 +325,168 @@ enum FinalizeInput { CollectLeft(PartitionStatus), } +enum FilterComposition { + NoUpdate, + Update(PhysicalExprRef), +} + +struct DynamicFilterExprComposer<'a> { + on_right: &'a [PhysicalExprRef], + repartition_random_state: &'a SeededRandomState, + probe_schema: &'a Schema, +} + +impl<'a> DynamicFilterExprComposer<'a> { + fn new( + on_right: &'a [PhysicalExprRef], + repartition_random_state: &'a SeededRandomState, + probe_schema: &'a Schema, + ) -> Self { + Self { + on_right, + repartition_random_state, + probe_schema, + } + } + + fn compose(&self, finalize_input: FinalizeInput) -> Result { + match finalize_input { + FinalizeInput::CollectLeft(partition) => self.compose_collect_left(partition), + FinalizeInput::Partitioned(partitions) => { + self.compose_partitioned(partitions) + } + } + } + + fn compose_collect_left( + &self, + partition: PartitionStatus, + ) -> Result { + let partition_data = match partition { + PartitionStatus::Reported(partition_data) => partition_data, + PartitionStatus::Pending => { + return datafusion_common::internal_err!( + "attempted to finalize collect-left dynamic filter without reported build data" + ); + } + PartitionStatus::CanceledUnknown => { + return datafusion_common::internal_err!( + "collect-left dynamic filter cannot finalize with canceled build data" + ); + } + }; + + let membership_expr = create_membership_predicate( + self.on_right, + partition_data.pushdown, + &HASH_JOIN_SEED, + self.probe_schema, + )?; + let bounds_expr = create_bounds_predicate(self.on_right, &partition_data.bounds); + + Ok( + match combine_membership_and_bounds(membership_expr, bounds_expr) { + Some(filter_expr) => FilterComposition::Update(filter_expr), + None => FilterComposition::NoUpdate, + }, + ) + } + + fn compose_partitioned( + &self, + partitions: Vec, + ) -> Result { + let num_partitions = partitions.len(); + let routing_hash_expr = Arc::new(HashExpr::new( + self.on_right.to_vec(), + self.repartition_random_state.clone(), + "hash_repartition".to_string(), + )) as Arc; + + let modulo_expr = Arc::new(BinaryExpr::new( + routing_hash_expr, + Operator::Modulo, + lit(ScalarValue::UInt64(Some(num_partitions as u64))), + )) as Arc; + + let mut real_branches = Vec::new(); + let mut empty_partition_ids = Vec::new(); + let mut has_canceled_unknown = false; + + for (partition_id, partition) in partitions.iter().enumerate() { + match partition { + PartitionStatus::Reported(partition) + if matches!(partition.pushdown, PushdownStrategy::Empty) => + { + empty_partition_ids.push(partition_id); + } + PartitionStatus::Reported(partition) => { + let membership_expr = create_membership_predicate( + self.on_right, + partition.pushdown.clone(), + &HASH_JOIN_SEED, + self.probe_schema, + )?; + let bounds_expr = + create_bounds_predicate(self.on_right, &partition.bounds); + let then_expr = + combine_membership_and_bounds(membership_expr, bounds_expr) + .unwrap_or_else(|| lit(true)); + real_branches.push(( + lit(ScalarValue::UInt64(Some(partition_id as u64))), + then_expr, + )); + } + PartitionStatus::CanceledUnknown => { + has_canceled_unknown = true; + } + PartitionStatus::Pending => { + return datafusion_common::internal_err!( + "attempted to finalize dynamic filter with pending partition" + ); + } + } + } + + let filter_expr = if has_canceled_unknown { + let mut when_then_branches = empty_partition_ids + .into_iter() + .map(|partition_id| { + ( + lit(ScalarValue::UInt64(Some(partition_id as u64))), + lit(false), + ) + }) + .collect::>(); + when_then_branches.extend(real_branches); + + if when_then_branches.is_empty() { + lit(true) + } else { + Arc::new(CaseExpr::try_new( + Some(modulo_expr), + when_then_branches, + Some(lit(true)), + )?) as Arc + } + } else if real_branches.is_empty() { + lit(false) + } else if real_branches.len() == 1 + && empty_partition_ids.len() + 1 == num_partitions + { + Arc::clone(&real_branches[0].1) + } else { + Arc::new(CaseExpr::try_new( + Some(modulo_expr), + real_branches, + Some(lit(false)), + )?) as Arc + }; + + Ok(FilterComposition::Update(filter_expr)) + } +} + impl SharedBuildAccumulator { /// Creates a new SharedBuildAccumulator configured for the given partition mode /// @@ -564,129 +726,16 @@ impl SharedBuildAccumulator { } fn build_filter(&self, finalize_input: FinalizeInput) -> Result<()> { - match finalize_input { - FinalizeInput::CollectLeft(partition) => match partition { - PartitionStatus::Reported(partition_data) => { - let membership_expr = create_membership_predicate( - &self.on_right, - partition_data.pushdown.clone(), - &HASH_JOIN_SEED, - self.probe_schema.as_ref(), - )?; - let bounds_expr = - create_bounds_predicate(&self.on_right, &partition_data.bounds); - - if let Some(filter_expr) = - combine_membership_and_bounds(membership_expr, bounds_expr) - { - self.dynamic_filter.update(filter_expr)?; - } - } - PartitionStatus::Pending => { - return datafusion_common::internal_err!( - "attempted to finalize collect-left dynamic filter without reported build data" - ); - } - PartitionStatus::CanceledUnknown => { - return datafusion_common::internal_err!( - "collect-left dynamic filter cannot finalize with canceled build data" - ); - } - }, - FinalizeInput::Partitioned(partitions) => { - let num_partitions = partitions.len(); - let routing_hash_expr = Arc::new(HashExpr::new( - self.on_right.clone(), - self.repartition_random_state.clone(), - "hash_repartition".to_string(), - )) as Arc; - - let modulo_expr = Arc::new(BinaryExpr::new( - routing_hash_expr, - Operator::Modulo, - lit(ScalarValue::UInt64(Some(num_partitions as u64))), - )) as Arc; - - let mut real_branches = Vec::new(); - let mut empty_partition_ids = Vec::new(); - let mut has_canceled_unknown = false; - - for (partition_id, partition) in partitions.iter().enumerate() { - match partition { - PartitionStatus::Reported(partition) - if matches!(partition.pushdown, PushdownStrategy::Empty) => - { - empty_partition_ids.push(partition_id); - } - PartitionStatus::Reported(partition) => { - let membership_expr = create_membership_predicate( - &self.on_right, - partition.pushdown.clone(), - &HASH_JOIN_SEED, - self.probe_schema.as_ref(), - )?; - let bounds_expr = create_bounds_predicate( - &self.on_right, - &partition.bounds, - ); - let then_expr = combine_membership_and_bounds( - membership_expr, - bounds_expr, - ) - .unwrap_or_else(|| lit(true)); - real_branches.push(( - lit(ScalarValue::UInt64(Some(partition_id as u64))), - then_expr, - )); - } - PartitionStatus::CanceledUnknown => { - has_canceled_unknown = true; - } - PartitionStatus::Pending => { - return datafusion_common::internal_err!( - "attempted to finalize dynamic filter with pending partition" - ); - } - } - } + let composer = DynamicFilterExprComposer::new( + &self.on_right, + &self.repartition_random_state, + self.probe_schema.as_ref(), + ); - let filter_expr = if has_canceled_unknown { - let mut when_then_branches = empty_partition_ids - .into_iter() - .map(|partition_id| { - ( - lit(ScalarValue::UInt64(Some(partition_id as u64))), - lit(false), - ) - }) - .collect::>(); - when_then_branches.extend(real_branches); - - if when_then_branches.is_empty() { - lit(true) - } else { - Arc::new(CaseExpr::try_new( - Some(modulo_expr), - when_then_branches, - Some(lit(true)), - )?) as Arc - } - } else if real_branches.is_empty() { - lit(false) - } else if real_branches.len() == 1 - && empty_partition_ids.len() + 1 == num_partitions - { - Arc::clone(&real_branches[0].1) - } else { - Arc::new(CaseExpr::try_new( - Some(modulo_expr), - real_branches, - Some(lit(false)), - )?) as Arc - }; - - self.dynamic_filter.update(filter_expr)?; - } + if let FilterComposition::Update(filter_expr) = + composer.compose(finalize_input)? + { + self.dynamic_filter.update(filter_expr)?; } Ok(()) From b1816712818d54f85644d6a152d63ad3936ff180 Mon Sep 17 00:00:00 2001 From: Siew Kam Onn Date: Tue, 7 Jul 2026 17:55:40 +0800 Subject: [PATCH 2/4] feat(datafusion): implement safe simplifications in hash_join shared_bounds - Built lazy routing modulo expr only when CASE is needed. - Deduplicated CaseExpr construction using partition_case. - Extracted reported-partition branch assembly into compose_reported_partition_branch. - Removed trivial private constructor; utilized struct literal instead. --- .../src/joins/hash_join/shared_bounds.rs | 131 ++++++++++-------- 1 file changed, 71 insertions(+), 60 deletions(-) diff --git a/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs b/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs index 8e7caa6199e1..49c90b23cd27 100644 --- a/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs +++ b/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs @@ -330,6 +330,11 @@ enum FilterComposition { Update(PhysicalExprRef), } +enum PartitionBranch { + Empty, + Real(PhysicalExprRef, PhysicalExprRef), +} + struct DynamicFilterExprComposer<'a> { on_right: &'a [PhysicalExprRef], repartition_random_state: &'a SeededRandomState, @@ -337,18 +342,6 @@ struct DynamicFilterExprComposer<'a> { } impl<'a> DynamicFilterExprComposer<'a> { - fn new( - on_right: &'a [PhysicalExprRef], - repartition_random_state: &'a SeededRandomState, - probe_schema: &'a Schema, - ) -> Self { - Self { - on_right, - repartition_random_state, - probe_schema, - } - } - fn compose(&self, finalize_input: FinalizeInput) -> Result { match finalize_input { FinalizeInput::CollectLeft(partition) => self.compose_collect_left(partition), @@ -397,46 +390,20 @@ impl<'a> DynamicFilterExprComposer<'a> { partitions: Vec, ) -> Result { let num_partitions = partitions.len(); - let routing_hash_expr = Arc::new(HashExpr::new( - self.on_right.to_vec(), - self.repartition_random_state.clone(), - "hash_repartition".to_string(), - )) as Arc; - - let modulo_expr = Arc::new(BinaryExpr::new( - routing_hash_expr, - Operator::Modulo, - lit(ScalarValue::UInt64(Some(num_partitions as u64))), - )) as Arc; - let mut real_branches = Vec::new(); let mut empty_partition_ids = Vec::new(); let mut has_canceled_unknown = false; for (partition_id, partition) in partitions.iter().enumerate() { match partition { - PartitionStatus::Reported(partition) - if matches!(partition.pushdown, PushdownStrategy::Empty) => + PartitionStatus::Reported(partition) => match self + .compose_reported_partition_branch(partition_id, partition)? { - empty_partition_ids.push(partition_id); - } - PartitionStatus::Reported(partition) => { - let membership_expr = create_membership_predicate( - self.on_right, - partition.pushdown.clone(), - &HASH_JOIN_SEED, - self.probe_schema, - )?; - let bounds_expr = - create_bounds_predicate(self.on_right, &partition.bounds); - let then_expr = - combine_membership_and_bounds(membership_expr, bounds_expr) - .unwrap_or_else(|| lit(true)); - real_branches.push(( - lit(ScalarValue::UInt64(Some(partition_id as u64))), - then_expr, - )); - } + PartitionBranch::Empty => empty_partition_ids.push(partition_id), + PartitionBranch::Real(when_expr, then_expr) => { + real_branches.push((when_expr, then_expr)); + } + }, PartitionStatus::CanceledUnknown => { has_canceled_unknown = true; } @@ -463,11 +430,7 @@ impl<'a> DynamicFilterExprComposer<'a> { if when_then_branches.is_empty() { lit(true) } else { - Arc::new(CaseExpr::try_new( - Some(modulo_expr), - when_then_branches, - Some(lit(true)), - )?) as Arc + self.partition_case(num_partitions, when_then_branches, true)? } } else if real_branches.is_empty() { lit(false) @@ -476,15 +439,63 @@ impl<'a> DynamicFilterExprComposer<'a> { { Arc::clone(&real_branches[0].1) } else { - Arc::new(CaseExpr::try_new( - Some(modulo_expr), - real_branches, - Some(lit(false)), - )?) as Arc + self.partition_case(num_partitions, real_branches, false)? }; Ok(FilterComposition::Update(filter_expr)) } + + fn compose_reported_partition_branch( + &self, + partition_id: usize, + partition: &PartitionData, + ) -> Result { + if matches!(partition.pushdown, PushdownStrategy::Empty) { + return Ok(PartitionBranch::Empty); + } + + let membership_expr = create_membership_predicate( + self.on_right, + partition.pushdown.clone(), + &HASH_JOIN_SEED, + self.probe_schema, + )?; + let bounds_expr = create_bounds_predicate(self.on_right, &partition.bounds); + let then_expr = combine_membership_and_bounds(membership_expr, bounds_expr) + .unwrap_or_else(|| lit(true)); + + Ok(PartitionBranch::Real( + lit(ScalarValue::UInt64(Some(partition_id as u64))), + then_expr, + )) + } + + fn partition_case( + &self, + num_partitions: usize, + branches: Vec<(PhysicalExprRef, PhysicalExprRef)>, + else_value: bool, + ) -> Result { + Ok(Arc::new(CaseExpr::try_new( + Some(self.routing_modulo_expr(num_partitions)), + branches, + Some(lit(else_value)), + )?) as Arc) + } + + fn routing_modulo_expr(&self, num_partitions: usize) -> PhysicalExprRef { + let routing_hash_expr = Arc::new(HashExpr::new( + self.on_right.to_vec(), + self.repartition_random_state.clone(), + "hash_repartition".to_string(), + )) as Arc; + + Arc::new(BinaryExpr::new( + routing_hash_expr, + Operator::Modulo, + lit(ScalarValue::UInt64(Some(num_partitions as u64))), + )) as Arc + } } impl SharedBuildAccumulator { @@ -726,11 +737,11 @@ impl SharedBuildAccumulator { } fn build_filter(&self, finalize_input: FinalizeInput) -> Result<()> { - let composer = DynamicFilterExprComposer::new( - &self.on_right, - &self.repartition_random_state, - self.probe_schema.as_ref(), - ); + let composer = DynamicFilterExprComposer { + on_right: &self.on_right, + repartition_random_state: &self.repartition_random_state, + probe_schema: self.probe_schema.as_ref(), + }; if let FilterComposition::Update(filter_expr) = composer.compose(finalize_input)? From 63cb7b6dfba8bab8ae32cd66c0074d6082a08724 Mon Sep 17 00:00:00 2001 From: Siew Kam Onn Date: Tue, 7 Jul 2026 18:02:15 +0800 Subject: [PATCH 3/4] feat: refactor PartitionBranch to use struct variant for Real representation --- .../src/joins/hash_join/shared_bounds.rs | 16 +++++++++++----- 1 file changed, 11 insertions(+), 5 deletions(-) diff --git a/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs b/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs index 49c90b23cd27..0ccc163e8e9f 100644 --- a/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs +++ b/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs @@ -332,7 +332,10 @@ enum FilterComposition { enum PartitionBranch { Empty, - Real(PhysicalExprRef, PhysicalExprRef), + Real { + when_expr: PhysicalExprRef, + then_expr: PhysicalExprRef, + }, } struct DynamicFilterExprComposer<'a> { @@ -400,7 +403,10 @@ impl<'a> DynamicFilterExprComposer<'a> { .compose_reported_partition_branch(partition_id, partition)? { PartitionBranch::Empty => empty_partition_ids.push(partition_id), - PartitionBranch::Real(when_expr, then_expr) => { + PartitionBranch::Real { + when_expr, + then_expr, + } => { real_branches.push((when_expr, then_expr)); } }, @@ -464,10 +470,10 @@ impl<'a> DynamicFilterExprComposer<'a> { let then_expr = combine_membership_and_bounds(membership_expr, bounds_expr) .unwrap_or_else(|| lit(true)); - Ok(PartitionBranch::Real( - lit(ScalarValue::UInt64(Some(partition_id as u64))), + Ok(PartitionBranch::Real { + when_expr: lit(ScalarValue::UInt64(Some(partition_id as u64))), then_expr, - )) + }) } fn partition_case( From be1b878e9baaa240bce0b51650ce478e6d1be366 Mon Sep 17 00:00:00 2001 From: Siew Kam Onn Date: Tue, 7 Jul 2026 18:37:48 +0800 Subject: [PATCH 4/4] fix: update compose_partitioned to accept a slice reference instead of a vector --- datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs b/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs index 0ccc163e8e9f..79de8da3304a 100644 --- a/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs +++ b/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs @@ -349,7 +349,7 @@ impl<'a> DynamicFilterExprComposer<'a> { match finalize_input { FinalizeInput::CollectLeft(partition) => self.compose_collect_left(partition), FinalizeInput::Partitioned(partitions) => { - self.compose_partitioned(partitions) + self.compose_partitioned(&partitions) } } } @@ -390,7 +390,7 @@ impl<'a> DynamicFilterExprComposer<'a> { fn compose_partitioned( &self, - partitions: Vec, + partitions: &[PartitionStatus], ) -> Result { let num_partitions = partitions.len(); let mut real_branches = Vec::new();