From 0c5a54987b911c57864f9e839d2e5c38bc6c13ef Mon Sep 17 00:00:00 2001 From: peterxcli Date: Fri, 24 Jul 2026 06:05:06 +0800 Subject: [PATCH 1/4] Enable filters for range-partitioned joins --- .../src/distribution_requirements.rs | 4 +- .../physical-plan/src/joins/hash_join/exec.rs | 106 ++++++++- .../src/joins/hash_join/shared_bounds.rs | 210 ++++++++++++++---- datafusion/physical-plan/src/ordering.rs | 99 +++++++++ datafusion/physical-plan/src/topk/mod.rs | 99 +-------- 5 files changed, 371 insertions(+), 147 deletions(-) diff --git a/datafusion/physical-plan/src/distribution_requirements.rs b/datafusion/physical-plan/src/distribution_requirements.rs index 9c7a1336c06a3..80222ce986086 100644 --- a/datafusion/physical-plan/src/distribution_requirements.rs +++ b/datafusion/physical-plan/src/distribution_requirements.rs @@ -19,10 +19,10 @@ use std::sync::Arc; -use datafusion_common::{Result, internal_err}; +use datafusion_common::{Result, ScalarValue, internal_err, validate_range_split_points}; use datafusion_physical_expr::{ Distribution, EquivalenceProperties, Partitioning, PartitioningSatisfaction, - PhysicalExpr, physical_exprs_equal, + PhysicalExpr, RangePartitioning, physical_exprs_equal, }; use crate::execution_plan::{ExecutionPlan, ExecutionPlanProperties, InvariantLevel}; diff --git a/datafusion/physical-plan/src/joins/hash_join/exec.rs b/datafusion/physical-plan/src/joins/hash_join/exec.rs index c5a64da1ea4af..24714c3893dad 100644 --- a/datafusion/physical-plan/src/joins/hash_join/exec.rs +++ b/datafusion/physical-plan/src/joins/hash_join/exec.rs @@ -886,9 +886,6 @@ impl HashJoinExec { if self.mode == PartitionMode::Partitioned && !self.has_partitioned_dynamic_filter_routing() { - // TODO: support partition-routed dynamic filters for compatible - // range co-partitioned joins. - // . return false; } @@ -904,6 +901,14 @@ impl HashJoinExec { Partitioning::Hash(_, left_partition_count), Partitioning::Hash(_, right_partition_count), ) => left_partition_count == right_partition_count, + (Partitioning::Range(_), Partitioning::Range(_)) => { + let children = [self.left.as_ref(), self.right.as_ref()]; + matches!( + self.input_distribution_requirements() + .unsatisfied_co_partitioned_children(self.name(), &children), + Ok(unsatisfied) if unsatisfied.is_empty() + ) + } (left_partitioning, right_partitioning) => { left_partitioning.partition_count() == 1 && right_partitioning.partition_count() == 1 @@ -6766,8 +6771,7 @@ mod tests { } #[test] - fn test_partitioned_dynamic_filter_pushdown_rejects_range_partitioning() -> Result<()> - { + fn test_partitioned_dynamic_filter_pushdown_range_partitioning() -> Result<()> { let (left_schema, right_schema, on) = build_schema_and_on()?; let left_partitioning = Partitioning::Range(RangePartitioning::try_new( [PhysicalSortExpr { @@ -6790,7 +6794,7 @@ mod tests { left_partitioning, )?); let right = Arc::new(PartitionedTestExec::try_new( - right_schema, + Arc::clone(&right_schema), right_partitioning, )?); @@ -6801,8 +6805,35 @@ mod tests { .enable_join_dynamic_filter_pushdown = true; let join = HashJoinExec::try_new( - left, + Arc::clone(&left) as Arc, right, + on.clone(), + None, + &JoinType::Inner, + None, + PartitionMode::Partitioned, + NullEquality::NullEqualsNothing, + false, + )?; + + assert!(join.allow_join_dynamic_filter_pushdown(session_config.options())); + + let mismatched_right_partitioning = + Partitioning::Range(RangePartitioning::try_new( + [PhysicalSortExpr { + expr: Arc::clone(&on[0].1), + options: Default::default(), + }] + .into(), + vec![SplitPoint::new(vec![ScalarValue::Int32(Some(11))])], + )?); + let mismatched_right = Arc::new(PartitionedTestExec::try_new( + right_schema, + mismatched_right_partitioning, + )?); + let mismatched_join = HashJoinExec::try_new( + left, + mismatched_right, on, None, &JoinType::Inner, @@ -6812,6 +6843,67 @@ mod tests { false, )?; + assert!( + !mismatched_join.allow_join_dynamic_filter_pushdown(session_config.options()) + ); + + Ok(()) + } + + #[test] + fn test_partitioned_dynamic_filter_pushdown_rejects_float_zero_split() -> Result<()> { + let left_schema = Arc::new(Schema::new(vec![Field::new( + "left_key", + DataType::Float64, + false, + )])); + let right_schema = Arc::new(Schema::new(vec![Field::new( + "right_key", + DataType::Float64, + false, + )])); + let left_key = Arc::new(Column::new("left_key", 0)) as PhysicalExprRef; + let right_key = Arc::new(Column::new("right_key", 0)) as PhysicalExprRef; + let split_points = vec![SplitPoint::new(vec![ScalarValue::Float64(Some(0.0))])]; + let left = Arc::new(PartitionedTestExec::try_new( + left_schema, + Partitioning::Range(RangePartitioning::try_new( + [PhysicalSortExpr::new( + Arc::clone(&left_key), + Default::default(), + )] + .into(), + split_points.clone(), + )?), + )?); + let right = Arc::new(PartitionedTestExec::try_new( + right_schema, + Partitioning::Range(RangePartitioning::try_new( + [PhysicalSortExpr::new( + Arc::clone(&right_key), + Default::default(), + )] + .into(), + split_points, + )?), + )?); + let join = HashJoinExec::try_new( + left, + right, + vec![(left_key, right_key)], + None, + &JoinType::Inner, + None, + PartitionMode::Partitioned, + NullEquality::NullEqualsNothing, + false, + )?; + let mut session_config = SessionConfig::default(); + session_config + .options_mut() + .optimizer + .enable_join_dynamic_filter_pushdown = true; + assert!(!join.allow_join_dynamic_filter_pushdown(session_config.options())); Ok(()) 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 7146e8dc2ec34..ddb7254c2eb81 100644 --- a/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs +++ b/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs @@ -30,6 +30,7 @@ use crate::joins::hash_join::inlist_builder::build_struct_fields; use crate::joins::hash_join::partitioned_hash_eval::{ HashExpr, HashTableLookupExpr, SeededRandomState, }; +use crate::ordering::build_lexicographic_filter; use arrow::array::ArrayRef; use arrow::datatypes::{DataType, Field, Schema}; use datafusion_common::config::ConfigOptions; @@ -39,7 +40,10 @@ use datafusion_functions::core::r#struct as struct_func; use datafusion_physical_expr::expressions::{ BinaryExpr, CaseExpr, DynamicFilterPhysicalExpr, InListExpr, lit, }; -use datafusion_physical_expr::{PhysicalExpr, PhysicalExprRef, ScalarFunctionExpr}; +use datafusion_physical_expr::{ + PhysicalExpr, PhysicalExprRef, PhysicalSortExpr, RangePartitioning, + ScalarFunctionExpr, +}; use parking_lot::Mutex; use tokio::sync::Notify; @@ -255,6 +259,8 @@ pub(crate) struct SharedBuildAccumulator { repartition_random_state: SeededRandomState, /// Schema of the probe (right) side for evaluating filter expressions probe_schema: Arc, + /// Probe-side Range routing metadata for partitioned dynamic filters. + probe_range_partitioning: Option, } /// Strategy for filter pushdown (decided at collection time) @@ -394,6 +400,15 @@ impl SharedBuildAccumulator { ), }; + let probe_range_partitioning = if partition_mode == PartitionMode::Partitioned { + match right_child.output_partitioning() { + crate::Partitioning::Range(range) => Some(range.clone()), + _ => None, + } + } else { + None + }; + Self { inner: Mutex::new(AccumulatorState { data: mode_data, @@ -404,6 +419,7 @@ impl SharedBuildAccumulator { on_right, repartition_random_state, probe_schema: right_child.schema(), + probe_range_partitioning, } } @@ -595,19 +611,8 @@ impl SharedBuildAccumulator { }, 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 partition_filters = Vec::with_capacity(num_partitions); + let mut real_partition_ids = Vec::new(); let mut empty_partition_ids = Vec::new(); let mut has_canceled_unknown = false; @@ -617,8 +622,10 @@ impl SharedBuildAccumulator { if matches!(partition.pushdown, PushdownStrategy::Empty) => { empty_partition_ids.push(partition_id); + partition_filters.push(lit(false)); } PartitionStatus::Reported(partition) => { + real_partition_ids.push(partition_id); let membership_expr = create_membership_predicate( &self.on_right, partition.pushdown.clone(), @@ -634,13 +641,11 @@ impl SharedBuildAccumulator { bounds_expr, ) .unwrap_or_else(|| lit(true)); - real_branches.push(( - lit(ScalarValue::UInt64(Some(partition_id as u64))), - then_expr, - )); + partition_filters.push(then_expr); } PartitionStatus::CanceledUnknown => { has_canceled_unknown = true; + partition_filters.push(lit(true)); } PartitionStatus::Pending => { return datafusion_common::internal_err!( @@ -650,38 +655,99 @@ impl SharedBuildAccumulator { } } - 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() { + let filter_expr = if has_canceled_unknown + && real_partition_ids.is_empty() + && empty_partition_ids.is_empty() + { + lit(true) + } else if !has_canceled_unknown && real_partition_ids.is_empty() { lit(false) - } else if real_branches.len() == 1 + } else if !has_canceled_unknown + && real_partition_ids.len() == 1 && empty_partition_ids.len() + 1 == num_partitions { - Arc::clone(&real_branches[0].1) + Arc::clone(&partition_filters[real_partition_ids[0]]) + } else if let Some(range_partitioning) = &self.probe_range_partitioning { + // Range partitioning + assert_eq!( + partition_filters.len(), + range_partitioning.partition_count() + ); + assert_eq!(self.on_right.len(), range_partitioning.ordering().len()); + let sort_exprs = self + .on_right + .iter() + .zip(range_partitioning.ordering()) + .map(|(expr, sort_expr)| { + PhysicalSortExpr::new(Arc::clone(expr), sort_expr.options) + }) + .collect::>(); + let else_expr = partition_filters + .pop() + .expect("Range partitioning always has at least one partition"); + let mut when_then_expr = Vec::with_capacity(partition_filters.len()); + // CASE evaluates in order + // + // CASE + // WHEN key } else { + // Hash partitioning + 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 when_then_branches = if has_canceled_unknown { + empty_partition_ids + .into_iter() + .map(|partition_id| { + ( + lit(ScalarValue::UInt64(Some(partition_id as u64))), + lit(false), + ) + }) + .collect::>() + } else { + vec![] + }; + when_then_branches.extend(real_partition_ids.into_iter().map( + |partition_id| { + ( + lit(ScalarValue::UInt64(Some(partition_id as u64))), + Arc::clone(&partition_filters[partition_id]), + ) + }, + )); + Arc::new(CaseExpr::try_new( Some(modulo_expr), - real_branches, - Some(lit(false)), + when_then_branches, + Some(lit(has_canceled_unknown)), )?) as Arc }; @@ -722,6 +788,7 @@ pub(super) fn make_partitioned_accumulator_for_test( on_right: vec![], repartition_random_state: SeededRandomState::with_seed(1), probe_schema, + probe_range_partitioning: None, } } @@ -742,7 +809,9 @@ pub(super) fn completed_partitions_for_test(acc: &SharedBuildAccumulator) -> usi mod tests { use super::*; - use arrow::array::{ArrayRef, Int32Array}; + use arrow::array::{ArrayRef, BooleanArray, Int32Array}; + use arrow::record_batch::RecordBatch; + use datafusion_common::SplitPoint; use datafusion_physical_expr::expressions::{Column, Literal}; fn test_on_right() -> Vec { @@ -778,6 +847,7 @@ mod tests { on_right, repartition_random_state: SeededRandomState::with_seed(1), probe_schema: test_probe_schema(), + probe_range_partitioning: None, } } @@ -981,6 +1051,56 @@ mod tests { ); } + #[test] + fn partitioned_range_dynamic_filter_routes_with_searched_case() -> Result<()> { + let mut acc = make_partitioned_expr_accumulator_for_test(4); + acc.probe_range_partitioning = Some(RangePartitioning::try_new( + [PhysicalSortExpr::new( + Arc::clone(&acc.on_right[0]), + Default::default(), + )] + .into(), + vec![ + SplitPoint::new(vec![ScalarValue::Int32(Some(10))]), + SplitPoint::new(vec![ScalarValue::Int32(Some(20))]), + SplitPoint::new(vec![ScalarValue::Int32(Some(30))]), + ], + )?); + + acc.build_filter(FinalizeInput::Partitioned(vec![ + reported(PushdownStrategy::Empty, no_bounds()), + PartitionStatus::CanceledUnknown, + reported(in_list(&[20, 29]), no_bounds()), + reported(in_list(&[30]), no_bounds()), + ]))?; + + let expr = current_expr(&acc); + let case = case_expr(&expr); + assert!( + case.expr().is_none(), + "Range routing must use searched CASE" + ); + assert_eq!(case.when_then_expr().len(), 3); + + let batch = RecordBatch::try_new( + test_probe_schema(), + vec![Arc::new(Int32Array::from(vec![ + 9, 10, 19, 20, 21, 29, 30, 31, + ]))], + )?; + let result = expr.evaluate(&batch)?.into_array(batch.num_rows())?; + let result = result + .as_any() + .downcast_ref::() + .expect("dynamic filter should evaluate to BooleanArray"); + assert_eq!( + result, + &BooleanArray::from(vec![false, true, true, true, false, true, true, false,]) + ); + + Ok(()) + } + // Regression guard for the build-report lifecycle fix: on `Drop`, a stream // in `BuildReportState::ReportScheduled` still calls `report_canceled_partition` // because it cannot tell whether the coordinator has already observed the diff --git a/datafusion/physical-plan/src/ordering.rs b/datafusion/physical-plan/src/ordering.rs index 8b596b9cb23eb..9c037e44bcdd4 100644 --- a/datafusion/physical-plan/src/ordering.rs +++ b/datafusion/physical-plan/src/ordering.rs @@ -15,6 +15,13 @@ // specific language governing permissions and limitations // under the License. +use std::sync::Arc; + +use datafusion_common::{Result, ScalarValue, assert_or_internal_err}; +use datafusion_expr::Operator; +use datafusion_physical_expr::expressions::{BinaryExpr, is_not_null, is_null, lit}; +use datafusion_physical_expr::{PhysicalExpr, PhysicalSortExpr}; + /// Specifies how the input to an aggregation or window operator is ordered /// relative to their `GROUP BY` or `PARTITION BY` expressions. /// @@ -52,3 +59,95 @@ pub enum InputOrderMode { /// existing ordering. Sorted, } + +/// Build the filter expression with the given thresholds. +/// This is now called outside of any locks to reduce critical section time. +pub(crate) fn build_lexicographic_filter( + sort_exprs: &[PhysicalSortExpr], + thresholds: &[ScalarValue], +) -> Result> { + assert_or_internal_err!(!sort_exprs.is_empty(), "Sort expressions must not be empty"); + assert_or_internal_err!( + sort_exprs.len() == thresholds.len(), + "Sort expressions and thresholds must have the same length" + ); + + // Create filter expressions for each threshold + let mut filters: Vec> = Vec::with_capacity(thresholds.len()); + + let mut prev_sort_expr: Option> = None; + for (sort_expr, value) in sort_exprs.iter().zip(thresholds.iter()) { + // Create the appropriate operator based on sort order + let op = if sort_expr.options.descending { + // For descending sort, we want col > threshold (exclude smaller values) + Operator::Gt + } else { + // For ascending sort, we want col < threshold (exclude larger values) + Operator::Lt + }; + + let value_null = value.is_null(); + + let comparison = Arc::new(BinaryExpr::new( + Arc::clone(&sort_expr.expr), + op, + lit(value.clone()), + )); + + let comparison_with_null = match (sort_expr.options.nulls_first, value_null) { + // For nulls first, transform to (threshold.value is not null) and (threshold.expr is null or comparison) + (true, true) => lit(false), + (true, false) => Arc::new(BinaryExpr::new( + is_null(Arc::clone(&sort_expr.expr))?, + Operator::Or, + comparison, + )), + // For nulls last, transform to (threshold.value is null and threshold.expr is not null) + // or (threshold.value is not null and comparison) + (false, true) => is_not_null(Arc::clone(&sort_expr.expr))?, + (false, false) => comparison, + }; + + let mut eq_expr = Arc::new(BinaryExpr::new( + Arc::clone(&sort_expr.expr), + Operator::Eq, + lit(value.clone()), + )); + + if value_null { + eq_expr = Arc::new(BinaryExpr::new( + is_null(Arc::clone(&sort_expr.expr))?, + Operator::Or, + eq_expr, + )); + } + + // For a query like order by a, b, the filter for column `b` is only applied if + // the condition a = threshold.value (considering null equality) is met. + // Therefore, we add equality predicates for all preceding fields to the filter logic of the current field, + // and include the current field's equality predicate in `prev_sort_expr` for use with subsequent fields. + match prev_sort_expr.take() { + None => { + prev_sort_expr = Some(eq_expr); + filters.push(comparison_with_null); + } + Some(p) => { + filters.push(Arc::new(BinaryExpr::new( + Arc::clone(&p), + Operator::And, + comparison_with_null, + ))); + + prev_sort_expr = + Some(Arc::new(BinaryExpr::new(p, Operator::And, eq_expr))); + } + } + } + + let dynamic_predicate = filters + .into_iter() + .reduce(|a, b| Arc::new(BinaryExpr::new(a, Operator::Or, b))) + .expect("sort expressions are checked non-empty"); + + Ok(dynamic_predicate) +} diff --git a/datafusion/physical-plan/src/topk/mod.rs b/datafusion/physical-plan/src/topk/mod.rs index 1e3efff36b1d8..6d30462219c39 100644 --- a/datafusion/physical-plan/src/topk/mod.rs +++ b/datafusion/physical-plan/src/topk/mod.rs @@ -25,7 +25,7 @@ use arrow::{ }, row::{OwnedRow, RowConverter, Rows, SortField}, }; -use datafusion_expr::{ColumnarValue, Operator}; +use datafusion_expr::ColumnarValue; use std::mem::size_of; use std::sync::atomic::{AtomicUsize, Ordering as AtomicOrdering}; use std::{cmp::Ordering, collections::BinaryHeap, sync::Arc}; @@ -34,6 +34,7 @@ use super::metrics::{ BaselineMetrics, Count, ExecutionPlanMetricsSet, MetricBuilder, MetricCategory, RecordOutput, }; +use crate::ordering::build_lexicographic_filter; use crate::spill::get_record_batch_memory_size; use crate::{SendableRecordBatchStream, stream::RecordBatchStreamAdapter}; @@ -48,7 +49,7 @@ use datafusion_execution::{ }; use datafusion_physical_expr::{ PhysicalExpr, - expressions::{BinaryExpr, DynamicFilterPhysicalExpr, is_not_null, is_null, lit}, + expressions::{DynamicFilterPhysicalExpr, lit}, }; use datafusion_physical_expr_common::sort_expr::{LexOrdering, PhysicalSortExpr}; use parking_lot::RwLock; @@ -563,7 +564,7 @@ impl TopK { let thresholds = boundary.threshold_values(&self.expr)?; // Build the filter expression OUTSIDE any synchronization - let predicate = Self::build_filter_expression(&self.expr, &thresholds)?; + let predicate = build_lexicographic_filter(&self.expr, &thresholds)?; let new_threshold = boundary.threshold(self.encode_topk_common_prefix_row(boundary)?); @@ -583,101 +584,13 @@ impl TopK { filter.shared_threshold = Some(new_threshold); // Update the filter expression - if let Some(pred) = predicate - && !pred.eq(&lit(true)) - { - filter.expr.update(pred)?; + if !predicate.eq(&lit(true)) { + filter.expr.update(predicate)?; } Ok(()) } - /// Build the filter expression with the given thresholds. - /// This is now called outside of any locks to reduce critical section time. - fn build_filter_expression( - sort_exprs: &[PhysicalSortExpr], - thresholds: &[ScalarValue], - ) -> Result>> { - // Create filter expressions for each threshold - let mut filters: Vec> = - Vec::with_capacity(thresholds.len()); - - let mut prev_sort_expr: Option> = None; - for (sort_expr, value) in sort_exprs.iter().zip(thresholds.iter()) { - // Create the appropriate operator based on sort order - let op = if sort_expr.options.descending { - // For descending sort, we want col > threshold (exclude smaller values) - Operator::Gt - } else { - // For ascending sort, we want col < threshold (exclude larger values) - Operator::Lt - }; - - let value_null = value.is_null(); - - let comparison = Arc::new(BinaryExpr::new( - Arc::clone(&sort_expr.expr), - op, - lit(value.clone()), - )); - - let comparison_with_null = match (sort_expr.options.nulls_first, value_null) { - // For nulls first, transform to (threshold.value is not null) and (threshold.expr is null or comparison) - (true, true) => lit(false), - (true, false) => Arc::new(BinaryExpr::new( - is_null(Arc::clone(&sort_expr.expr))?, - Operator::Or, - comparison, - )), - // For nulls last, transform to (threshold.value is null and threshold.expr is not null) - // or (threshold.value is not null and comparison) - (false, true) => is_not_null(Arc::clone(&sort_expr.expr))?, - (false, false) => comparison, - }; - - let mut eq_expr = Arc::new(BinaryExpr::new( - Arc::clone(&sort_expr.expr), - Operator::Eq, - lit(value.clone()), - )); - - if value_null { - eq_expr = Arc::new(BinaryExpr::new( - is_null(Arc::clone(&sort_expr.expr))?, - Operator::Or, - eq_expr, - )); - } - - // For a query like order by a, b, the filter for column `b` is only applied if - // the condition a = threshold.value (considering null equality) is met. - // Therefore, we add equality predicates for all preceding fields to the filter logic of the current field, - // and include the current field's equality predicate in `prev_sort_expr` for use with subsequent fields. - match prev_sort_expr.take() { - None => { - prev_sort_expr = Some(eq_expr); - filters.push(comparison_with_null); - } - Some(p) => { - filters.push(Arc::new(BinaryExpr::new( - Arc::clone(&p), - Operator::And, - comparison_with_null, - ))); - - prev_sort_expr = - Some(Arc::new(BinaryExpr::new(p, Operator::And, eq_expr))); - } - } - } - - let dynamic_predicate = filters - .into_iter() - .reduce(|a, b| Arc::new(BinaryExpr::new(a, Operator::Or, b))); - - Ok(dynamic_predicate) - } - /// If input ordering shares a common sort prefix with the TopK, /// check if the computation can be finished early. /// From a9167cbe6ad187e39d092fd55b4b48ceabfaffa7 Mon Sep 17 00:00:00 2001 From: peterxcli Date: Fri, 24 Jul 2026 06:29:05 +0800 Subject: [PATCH 2/4] revert header change --- .../physical-plan/src/distribution_requirements.rs | 9 ++------- 1 file changed, 2 insertions(+), 7 deletions(-) diff --git a/datafusion/physical-plan/src/distribution_requirements.rs b/datafusion/physical-plan/src/distribution_requirements.rs index dde31f94d32b5..6405b1f121ef7 100644 --- a/datafusion/physical-plan/src/distribution_requirements.rs +++ b/datafusion/physical-plan/src/distribution_requirements.rs @@ -17,13 +17,8 @@ //! Input distribution requirements for physical execution plans. -use std::sync::Arc; - -use datafusion_common::{Result, ScalarValue, internal_err, validate_range_split_points}; -use datafusion_physical_expr::{ - Distribution, EquivalenceProperties, Partitioning, PartitioningSatisfaction, - PhysicalExpr, RangePartitioning, physical_exprs_equal, -}; +use datafusion_common::{Result, internal_err}; +use datafusion_physical_expr::{Distribution, Partitioning, PartitioningSatisfaction}; use crate::execution_plan::{ExecutionPlan, ExecutionPlanProperties, InvariantLevel}; From 4206b36c63ded0121e7b958e5ed4b53b7c47f898 Mon Sep 17 00:00:00 2001 From: peterxcli Date: Fri, 24 Jul 2026 13:57:40 +0800 Subject: [PATCH 3/4] remove float zero test --- .../physical-plan/src/joins/hash_join/exec.rs | 59 ------------------- 1 file changed, 59 deletions(-) diff --git a/datafusion/physical-plan/src/joins/hash_join/exec.rs b/datafusion/physical-plan/src/joins/hash_join/exec.rs index 6e0ead66ff79b..54fcba3659d0a 100644 --- a/datafusion/physical-plan/src/joins/hash_join/exec.rs +++ b/datafusion/physical-plan/src/joins/hash_join/exec.rs @@ -6843,65 +6843,6 @@ mod tests { Ok(()) } - #[test] - fn test_partitioned_dynamic_filter_pushdown_rejects_float_zero_split() -> Result<()> { - let left_schema = Arc::new(Schema::new(vec![Field::new( - "left_key", - DataType::Float64, - false, - )])); - let right_schema = Arc::new(Schema::new(vec![Field::new( - "right_key", - DataType::Float64, - false, - )])); - let left_key = Arc::new(Column::new("left_key", 0)) as PhysicalExprRef; - let right_key = Arc::new(Column::new("right_key", 0)) as PhysicalExprRef; - let split_points = vec![SplitPoint::new(vec![ScalarValue::Float64(Some(0.0))])]; - let left = Arc::new(PartitionedTestExec::try_new( - left_schema, - Partitioning::Range(RangePartitioning::try_new( - [PhysicalSortExpr::new( - Arc::clone(&left_key), - Default::default(), - )] - .into(), - split_points.clone(), - )?), - )?); - let right = Arc::new(PartitionedTestExec::try_new( - right_schema, - Partitioning::Range(RangePartitioning::try_new( - [PhysicalSortExpr::new( - Arc::clone(&right_key), - Default::default(), - )] - .into(), - split_points, - )?), - )?); - let join = HashJoinExec::try_new( - left, - right, - vec![(left_key, right_key)], - None, - &JoinType::Inner, - None, - PartitionMode::Partitioned, - NullEquality::NullEqualsNothing, - false, - )?; - let mut session_config = SessionConfig::default(); - session_config - .options_mut() - .optimizer - .enable_join_dynamic_filter_pushdown = true; - - assert!(!join.allow_join_dynamic_filter_pushdown(session_config.options())); - - Ok(()) - } - #[test] fn test_with_dynamic_filter_rejects_invalid_columns() -> Result<()> { let (_, _, on) = build_schema_and_on()?; From 1a141f3e59942ef41594219e2191e7b623e34a25 Mon Sep 17 00:00:00 2001 From: peterxcli Date: Sat, 25 Jul 2026 03:14:25 +0800 Subject: [PATCH 4/4] address review batch 1 --- .../src/joins/hash_join/shared_bounds.rs | 32 +++++++++---------- 1 file changed, 16 insertions(+), 16 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 9f7b4ee1d9fcb..366fd2b777900 100644 --- a/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs +++ b/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs @@ -23,6 +23,7 @@ use std::sync::Arc; use crate::ExecutionPlan; use crate::ExecutionPlanProperties; +use crate::Partitioning; use crate::joins::Map; use crate::joins::PartitionMode; use crate::joins::hash_join::exec::HASH_JOIN_SEED; @@ -404,14 +405,13 @@ impl SharedBuildAccumulator { ), }; - let probe_range_partitioning = if partition_mode == PartitionMode::Partitioned { - match right_child.output_partitioning() { - crate::Partitioning::Range(range) => Some(range.clone()), + let probe_range_partitioning = + match (partition_mode, right_child.output_partitioning()) { + (PartitionMode::Partitioned, Partitioning::Range(range)) => { + Some(range.clone()) + } _ => None, - } - } else { - None - }; + }; Self { inner: Mutex::new(AccumulatorState { @@ -691,7 +691,6 @@ impl SharedBuildAccumulator { let else_expr = partition_filters .pop() .expect("Range partitioning always has at least one partition"); - let mut when_then_expr = Vec::with_capacity(partition_filters.len()); // CASE evaluates in order // // CASE @@ -700,17 +699,18 @@ impl SharedBuildAccumulator { // ... // ELSE Fn // END - for (split_point, then_expr) in range_partitioning + let when_then_expr = range_partitioning .split_points() .iter() .zip(partition_filters) - { - let when_expr = build_lexicographic_filter( - &sort_exprs, - split_point.values(), - )?; - when_then_expr.push((when_expr, then_expr)); - } + .map(|(split_point, then_expr)| { + let when_expr = build_lexicographic_filter( + &sort_exprs, + split_point.values(), + )?; + Ok((when_expr, then_expr)) + }) + .collect::>>()?; Arc::new(CaseExpr::try_new(None, when_then_expr, Some(else_expr))?) as Arc