From 242ff1550514c8e86794dced8f022592c062b1cb Mon Sep 17 00:00:00 2001 From: Siew Kam Onn Date: Sat, 4 Jul 2026 13:13:46 +0800 Subject: [PATCH 1/5] feat(tests): add tests for CollectLeft and Partitioned in shared_bounds.rs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Added tests for `CollectLeft`: - membership only - bounds only - membership + bounds - empty/no-expression → no update - Added tests for `Partitioned`: - one real + rest empty → no CASE - multiple real → CASE, else false - all empty → false - canceled unknown + empty → CASE, else true --- .../src/joins/hash_join/shared_bounds.rs | 229 ++++++++++++++++++ 1 file changed, 229 insertions(+) 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 0af4015ff7239..ed1d5c71f10f2 100644 --- a/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs +++ b/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs @@ -742,6 +742,103 @@ pub(super) fn completed_partitions_for_test(acc: &SharedBuildAccumulator) -> usi mod tests { use super::*; + use arrow::array::{ArrayRef, Int32Array}; + use datafusion_physical_expr::expressions::{Column, Literal}; + + fn test_on_right() -> Vec { + vec![Arc::new(Column::new("probe_key", 0))] + } + + fn test_probe_schema() -> Arc { + Arc::new(Schema::new(vec![Field::new( + "probe_key", + DataType::Int32, + false, + )])) + } + + fn test_dynamic_filter( + on_right: &[PhysicalExprRef], + ) -> Arc { + Arc::new(DynamicFilterPhysicalExpr::new(on_right.to_vec(), lit(true))) + } + + fn make_collect_left_accumulator_for_test() -> SharedBuildAccumulator { + let on_right = test_on_right(); + SharedBuildAccumulator { + inner: Mutex::new(AccumulatorState { + data: AccumulatedBuildData::CollectLeft { + data: PartitionStatus::Pending, + reported_count: 0, + expected_reports: 1, + }, + completion: CompletionState::Pending, + }), + completion_notify: Notify::new(), + dynamic_filter: test_dynamic_filter(&on_right), + on_right, + repartition_random_state: SeededRandomState::with_seed(1), + probe_schema: test_probe_schema(), + } + } + + fn make_partitioned_expr_accumulator_for_test( + num_partitions: usize, + ) -> SharedBuildAccumulator { + let on_right = test_on_right(); + SharedBuildAccumulator { + inner: Mutex::new(AccumulatorState { + data: AccumulatedBuildData::Partitioned { + partitions: vec![PartitionStatus::Pending; num_partitions], + completed_partitions: 0, + }, + completion: CompletionState::Pending, + }), + completion_notify: Notify::new(), + dynamic_filter: test_dynamic_filter(&on_right), + on_right, + repartition_random_state: SeededRandomState::with_seed(1), + probe_schema: test_probe_schema(), + } + } + + fn in_list(values: &[i32]) -> PushdownStrategy { + PushdownStrategy::InList(Arc::new(Int32Array::from(values.to_vec())) as ArrayRef) + } + + fn bounds(min: i32, max: i32) -> PartitionBounds { + PartitionBounds::new(vec![ColumnBounds::new( + ScalarValue::Int32(Some(min)), + ScalarValue::Int32(Some(max)), + )]) + } + + fn no_bounds() -> PartitionBounds { + PartitionBounds::new(vec![]) + } + + fn reported(pushdown: PushdownStrategy, bounds: PartitionBounds) -> PartitionStatus { + PartitionStatus::Reported(PartitionData { pushdown, bounds }) + } + + fn current_expr(acc: &SharedBuildAccumulator) -> PhysicalExprRef { + acc.dynamic_filter.current().unwrap() + } + + fn assert_literal_bool(expr: &PhysicalExprRef, expected: bool) { + let literal = expr + .downcast_ref::() + .expect("expected literal expression"); + assert_eq!(literal.value(), &ScalarValue::Boolean(Some(expected))); + } + + fn assert_top_binary_op(expr: &PhysicalExprRef, expected: Operator) { + let binary = expr + .downcast_ref::() + .expect("expected binary expression"); + assert_eq!(binary.op(), &expected); + } + fn partitioned_state(acc: &SharedBuildAccumulator) -> (Vec, usize) { let guard = acc.inner.lock(); let AccumulatedBuildData::Partitioned { @@ -754,6 +851,138 @@ mod tests { (partitions.clone(), *completed_partitions) } + #[test] + fn collect_left_updates_with_membership_only() { + let acc = make_collect_left_accumulator_for_test(); + + acc.build_filter(FinalizeInput::CollectLeft(reported( + in_list(&[1, 2, 3]), + no_bounds(), + ))) + .unwrap(); + + let expr = current_expr(&acc); + assert!(expr.downcast_ref::().is_some()); + } + + #[test] + fn collect_left_updates_with_bounds_only() { + let acc = make_collect_left_accumulator_for_test(); + + acc.build_filter(FinalizeInput::CollectLeft(reported( + PushdownStrategy::Empty, + bounds(10, 20), + ))) + .unwrap(); + + let expr = current_expr(&acc); + assert_top_binary_op(&expr, Operator::And); + } + + #[test] + fn collect_left_combines_membership_and_bounds() { + let acc = make_collect_left_accumulator_for_test(); + + acc.build_filter(FinalizeInput::CollectLeft(reported( + in_list(&[10, 20]), + bounds(10, 20), + ))) + .unwrap(); + + let expr = current_expr(&acc); + let conjunction = expr + .downcast_ref::() + .expect("expected membership and bounds conjunction"); + assert_eq!(conjunction.op(), &Operator::And); + assert!(conjunction.left().downcast_ref::().is_some()); + assert!(conjunction.right().downcast_ref::().is_some()); + } + + #[test] + fn collect_left_empty_build_data_does_not_update_filter() { + let acc = make_collect_left_accumulator_for_test(); + + acc.build_filter(FinalizeInput::CollectLeft(reported( + PushdownStrategy::Empty, + no_bounds(), + ))) + .unwrap(); + + let expr = current_expr(&acc); + assert_literal_bool(&expr, true); + } + + #[test] + fn partitioned_one_real_partition_with_rest_empty_skips_case() { + let acc = make_partitioned_expr_accumulator_for_test(3); + + acc.build_filter(FinalizeInput::Partitioned(vec![ + reported(PushdownStrategy::Empty, no_bounds()), + reported(in_list(&[2]), no_bounds()), + reported(PushdownStrategy::Empty, no_bounds()), + ])) + .unwrap(); + + let expr = current_expr(&acc); + assert!(expr.downcast_ref::().is_some()); + assert!(expr.downcast_ref::().is_none()); + } + + #[test] + fn partitioned_multiple_real_partitions_uses_case_with_false_fallback() { + let acc = make_partitioned_expr_accumulator_for_test(2); + + acc.build_filter(FinalizeInput::Partitioned(vec![ + reported(in_list(&[1]), no_bounds()), + reported(in_list(&[2]), no_bounds()), + ])) + .unwrap(); + + let expr = current_expr(&acc); + let case = expr + .downcast_ref::() + .expect("expected partition routing CASE expression"); + assert!(case.expr().is_some()); + assert_eq!(case.when_then_expr().len(), 2); + assert_literal_bool(case.else_expr().expect("expected CASE fallback"), false); + } + + #[test] + fn partitioned_all_empty_partitions_updates_filter_to_false() { + let acc = make_partitioned_expr_accumulator_for_test(2); + + acc.build_filter(FinalizeInput::Partitioned(vec![ + reported(PushdownStrategy::Empty, no_bounds()), + reported(PushdownStrategy::Empty, no_bounds()), + ])) + .unwrap(); + + let expr = current_expr(&acc); + assert_literal_bool(&expr, false); + } + + #[test] + fn partitioned_canceled_unknown_partitions_keep_unknown_routes_permissive() { + let acc = make_partitioned_expr_accumulator_for_test(2); + + acc.build_filter(FinalizeInput::Partitioned(vec![ + PartitionStatus::CanceledUnknown, + reported(PushdownStrategy::Empty, no_bounds()), + ])) + .unwrap(); + + let expr = current_expr(&acc); + let case = expr + .downcast_ref::() + .expect("expected CASE expression for mixed empty and unknown partitions"); + assert_eq!(case.when_then_expr().len(), 1); + assert_literal_bool(&case.when_then_expr()[0].1, false); + assert_literal_bool( + case.else_expr().expect("expected permissive fallback"), + true, + ); + } + // 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 From 86608a285dcbd10f92d5b1398cc3e7b17b0474f5 Mon Sep 17 00:00:00 2001 From: Siew Kam Onn Date: Sat, 4 Jul 2026 13:22:36 +0800 Subject: [PATCH 2/5] feat(tests): deduplicate test accumulator setup and add narrow expression helpers - Deduplicated test accumulator setup using `make_accumulator_for_test`. - Introduced new narrow expression helpers: `in_list_expr`, `binary_expr`, and `case_expr`. - Enhanced panic context for `current_expr` and downcasting. - Maintained existing `pub(super) make_partitioned_accumulator_for_test` without changes. --- .../src/joins/hash_join/shared_bounds.rs | 94 ++++++++++--------- 1 file changed, 52 insertions(+), 42 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 ed1d5c71f10f2..085886838ec38 100644 --- a/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs +++ b/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs @@ -763,43 +763,45 @@ mod tests { Arc::new(DynamicFilterPhysicalExpr::new(on_right.to_vec(), lit(true))) } - fn make_collect_left_accumulator_for_test() -> SharedBuildAccumulator { - let on_right = test_on_right(); + fn make_accumulator_for_test( + data: AccumulatedBuildData, + on_right: Vec, + ) -> SharedBuildAccumulator { + let dynamic_filter = test_dynamic_filter(&on_right); SharedBuildAccumulator { inner: Mutex::new(AccumulatorState { - data: AccumulatedBuildData::CollectLeft { - data: PartitionStatus::Pending, - reported_count: 0, - expected_reports: 1, - }, + data, completion: CompletionState::Pending, }), completion_notify: Notify::new(), - dynamic_filter: test_dynamic_filter(&on_right), + dynamic_filter, on_right, repartition_random_state: SeededRandomState::with_seed(1), probe_schema: test_probe_schema(), } } + fn make_collect_left_accumulator_for_test() -> SharedBuildAccumulator { + make_accumulator_for_test( + AccumulatedBuildData::CollectLeft { + data: PartitionStatus::Pending, + reported_count: 0, + expected_reports: 1, + }, + test_on_right(), + ) + } + fn make_partitioned_expr_accumulator_for_test( num_partitions: usize, ) -> SharedBuildAccumulator { - let on_right = test_on_right(); - SharedBuildAccumulator { - inner: Mutex::new(AccumulatorState { - data: AccumulatedBuildData::Partitioned { - partitions: vec![PartitionStatus::Pending; num_partitions], - completed_partitions: 0, - }, - completion: CompletionState::Pending, - }), - completion_notify: Notify::new(), - dynamic_filter: test_dynamic_filter(&on_right), - on_right, - repartition_random_state: SeededRandomState::with_seed(1), - probe_schema: test_probe_schema(), - } + make_accumulator_for_test( + AccumulatedBuildData::Partitioned { + partitions: vec![PartitionStatus::Pending; num_partitions], + completed_partitions: 0, + }, + test_on_right(), + ) } fn in_list(values: &[i32]) -> PushdownStrategy { @@ -822,21 +824,35 @@ mod tests { } fn current_expr(acc: &SharedBuildAccumulator) -> PhysicalExprRef { - acc.dynamic_filter.current().unwrap() + acc.dynamic_filter + .current() + .expect("dynamic filter current expression should be available") + } + + fn in_list_expr(expr: &PhysicalExprRef) -> &InListExpr { + expr.downcast_ref::() + .expect("expected InListExpr dynamic filter") + } + + fn binary_expr(expr: &PhysicalExprRef) -> &BinaryExpr { + expr.downcast_ref::() + .expect("expected BinaryExpr dynamic filter") + } + + fn case_expr(expr: &PhysicalExprRef) -> &CaseExpr { + expr.downcast_ref::() + .expect("expected CaseExpr dynamic filter") } fn assert_literal_bool(expr: &PhysicalExprRef, expected: bool) { let literal = expr .downcast_ref::() - .expect("expected literal expression"); + .expect("expected literal bool dynamic filter"); assert_eq!(literal.value(), &ScalarValue::Boolean(Some(expected))); } fn assert_top_binary_op(expr: &PhysicalExprRef, expected: Operator) { - let binary = expr - .downcast_ref::() - .expect("expected binary expression"); - assert_eq!(binary.op(), &expected); + assert_eq!(binary_expr(expr).op(), &expected); } fn partitioned_state(acc: &SharedBuildAccumulator) -> (Vec, usize) { @@ -862,7 +878,7 @@ mod tests { .unwrap(); let expr = current_expr(&acc); - assert!(expr.downcast_ref::().is_some()); + in_list_expr(&expr); } #[test] @@ -890,12 +906,10 @@ mod tests { .unwrap(); let expr = current_expr(&acc); - let conjunction = expr - .downcast_ref::() - .expect("expected membership and bounds conjunction"); + let conjunction = binary_expr(&expr); assert_eq!(conjunction.op(), &Operator::And); - assert!(conjunction.left().downcast_ref::().is_some()); - assert!(conjunction.right().downcast_ref::().is_some()); + binary_expr(conjunction.left()); + in_list_expr(conjunction.right()); } #[test] @@ -924,7 +938,7 @@ mod tests { .unwrap(); let expr = current_expr(&acc); - assert!(expr.downcast_ref::().is_some()); + in_list_expr(&expr); assert!(expr.downcast_ref::().is_none()); } @@ -939,9 +953,7 @@ mod tests { .unwrap(); let expr = current_expr(&acc); - let case = expr - .downcast_ref::() - .expect("expected partition routing CASE expression"); + let case = case_expr(&expr); assert!(case.expr().is_some()); assert_eq!(case.when_then_expr().len(), 2); assert_literal_bool(case.else_expr().expect("expected CASE fallback"), false); @@ -972,9 +984,7 @@ mod tests { .unwrap(); let expr = current_expr(&acc); - let case = expr - .downcast_ref::() - .expect("expected CASE expression for mixed empty and unknown partitions"); + let case = case_expr(&expr); assert_eq!(case.when_then_expr().len(), 1); assert_literal_bool(&case.when_then_expr()[0].1, false); assert_literal_bool( From 90890f4a95dc08f2cec22b5294a870d8ac322762 Mon Sep 17 00:00:00 2001 From: Siew Kam Onn Date: Sat, 4 Jul 2026 13:37:17 +0800 Subject: [PATCH 3/5] test: enhance no-expression and partitioned CASE tests - Update CollectLeft no-expression test to assert that dynamic filter generation remains unchanged. - Improve partitioned CASE test to assert: - Routing expression is set to '%' - Branch keys are 0 and 1 - Branch expressions are verified to be InList - Fallback condition remains false --- .../src/joins/hash_join/shared_bounds.rs | 20 ++++++++++++++++++- 1 file changed, 19 insertions(+), 1 deletion(-) 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 085886838ec38..5ac0b13708127 100644 --- a/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs +++ b/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs @@ -851,6 +851,13 @@ mod tests { assert_eq!(literal.value(), &ScalarValue::Boolean(Some(expected))); } + fn assert_literal_u64(expr: &PhysicalExprRef, expected: u64) { + let literal = expr + .downcast_ref::() + .expect("expected literal u64 CASE branch key"); + assert_eq!(literal.value(), &ScalarValue::UInt64(Some(expected))); + } + fn assert_top_binary_op(expr: &PhysicalExprRef, expected: Operator) { assert_eq!(binary_expr(expr).op(), &expected); } @@ -915,6 +922,7 @@ mod tests { #[test] fn collect_left_empty_build_data_does_not_update_filter() { let acc = make_collect_left_accumulator_for_test(); + let initial_generation = acc.dynamic_filter.snapshot_generation(); acc.build_filter(FinalizeInput::CollectLeft(reported( PushdownStrategy::Empty, @@ -922,6 +930,11 @@ mod tests { ))) .unwrap(); + assert_eq!( + acc.dynamic_filter.snapshot_generation(), + initial_generation, + "empty CollectLeft input must not update with a no-op filter" + ); let expr = current_expr(&acc); assert_literal_bool(&expr, true); } @@ -954,8 +967,13 @@ mod tests { let expr = current_expr(&acc); let case = case_expr(&expr); - assert!(case.expr().is_some()); + let routing_expr = case.expr().expect("expected CASE routing expression"); + assert_top_binary_op(routing_expr, Operator::Modulo); assert_eq!(case.when_then_expr().len(), 2); + assert_literal_u64(&case.when_then_expr()[0].0, 0); + in_list_expr(&case.when_then_expr()[0].1); + assert_literal_u64(&case.when_then_expr()[1].0, 1); + in_list_expr(&case.when_then_expr()[1].1); assert_literal_bool(case.else_expr().expect("expected CASE fallback"), false); } From 707646879a8b42caab118166ab1907861719f873 Mon Sep 17 00:00:00 2001 From: Siew Kam Onn Date: Sat, 4 Jul 2026 14:03:06 +0800 Subject: [PATCH 4/5] refactor(tests): remove overlapping tests and unused helper function in hash_join - Removed overlapping tests: - collect_left_combines_membership_and_bounds - partitioned_multiple_real_partitions_uses_case_with_false_fallback - partitioned_all_empty_partitions_updates_filter_to_false - Removed now-unused helper function: assert_literal_u64. --- .../src/joins/hash_join/shared_bounds.rs | 60 ------------------- 1 file changed, 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 5ac0b13708127..5c75e3db1ee7e 100644 --- a/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs +++ b/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs @@ -851,13 +851,6 @@ mod tests { assert_eq!(literal.value(), &ScalarValue::Boolean(Some(expected))); } - fn assert_literal_u64(expr: &PhysicalExprRef, expected: u64) { - let literal = expr - .downcast_ref::() - .expect("expected literal u64 CASE branch key"); - assert_eq!(literal.value(), &ScalarValue::UInt64(Some(expected))); - } - fn assert_top_binary_op(expr: &PhysicalExprRef, expected: Operator) { assert_eq!(binary_expr(expr).op(), &expected); } @@ -902,23 +895,6 @@ mod tests { assert_top_binary_op(&expr, Operator::And); } - #[test] - fn collect_left_combines_membership_and_bounds() { - let acc = make_collect_left_accumulator_for_test(); - - acc.build_filter(FinalizeInput::CollectLeft(reported( - in_list(&[10, 20]), - bounds(10, 20), - ))) - .unwrap(); - - let expr = current_expr(&acc); - let conjunction = binary_expr(&expr); - assert_eq!(conjunction.op(), &Operator::And); - binary_expr(conjunction.left()); - in_list_expr(conjunction.right()); - } - #[test] fn collect_left_empty_build_data_does_not_update_filter() { let acc = make_collect_left_accumulator_for_test(); @@ -955,42 +931,6 @@ mod tests { assert!(expr.downcast_ref::().is_none()); } - #[test] - fn partitioned_multiple_real_partitions_uses_case_with_false_fallback() { - let acc = make_partitioned_expr_accumulator_for_test(2); - - acc.build_filter(FinalizeInput::Partitioned(vec![ - reported(in_list(&[1]), no_bounds()), - reported(in_list(&[2]), no_bounds()), - ])) - .unwrap(); - - let expr = current_expr(&acc); - let case = case_expr(&expr); - let routing_expr = case.expr().expect("expected CASE routing expression"); - assert_top_binary_op(routing_expr, Operator::Modulo); - assert_eq!(case.when_then_expr().len(), 2); - assert_literal_u64(&case.when_then_expr()[0].0, 0); - in_list_expr(&case.when_then_expr()[0].1); - assert_literal_u64(&case.when_then_expr()[1].0, 1); - in_list_expr(&case.when_then_expr()[1].1); - assert_literal_bool(case.else_expr().expect("expected CASE fallback"), false); - } - - #[test] - fn partitioned_all_empty_partitions_updates_filter_to_false() { - let acc = make_partitioned_expr_accumulator_for_test(2); - - acc.build_filter(FinalizeInput::Partitioned(vec![ - reported(PushdownStrategy::Empty, no_bounds()), - reported(PushdownStrategy::Empty, no_bounds()), - ])) - .unwrap(); - - let expr = current_expr(&acc); - assert_literal_bool(&expr, false); - } - #[test] fn partitioned_canceled_unknown_partitions_keep_unknown_routes_permissive() { let acc = make_partitioned_expr_accumulator_for_test(2); From e985a80aac141f2880c14ccab8b1acd158196589 Mon Sep 17 00:00:00 2001 From: Siew Kam Onn Date: Tue, 7 Jul 2026 16:10:39 +0800 Subject: [PATCH 5/5] feat: add assert_in_list_column_values and update collect_left_updates_with_membership_only assertions - Added assert_in_list_column_values function. - Updated collect_left_updates_with_membership_only to assert: - child column = probe_key@0 - in-list values = [1, 2, 3] --- .../src/joins/hash_join/shared_bounds.rs | 32 ++++++++++++++++++- 1 file changed, 31 insertions(+), 1 deletion(-) 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 5c75e3db1ee7e..7146e8dc2ec34 100644 --- a/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs +++ b/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs @@ -834,6 +834,36 @@ mod tests { .expect("expected InListExpr dynamic filter") } + fn assert_in_list_column_values( + expr: &PhysicalExprRef, + expected_column_name: &str, + expected_column_index: usize, + expected_values: &[i32], + ) { + let in_list = in_list_expr(expr); + let column = in_list + .expr() + .downcast_ref::() + .expect("expected InListExpr child column"); + assert_eq!(column.name(), expected_column_name); + assert_eq!(column.index(), expected_column_index); + + let actual_values = in_list + .list() + .iter() + .map(|expr| { + let literal = expr + .downcast_ref::() + .expect("expected InListExpr literal value"); + match literal.value() { + ScalarValue::Int32(Some(value)) => *value, + value => panic!("expected Int32 in-list value, got {value:?}"), + } + }) + .collect::>(); + assert_eq!(actual_values, expected_values); + } + fn binary_expr(expr: &PhysicalExprRef) -> &BinaryExpr { expr.downcast_ref::() .expect("expected BinaryExpr dynamic filter") @@ -878,7 +908,7 @@ mod tests { .unwrap(); let expr = current_expr(&acc); - in_list_expr(&expr); + assert_in_list_column_values(&expr, "probe_key", 0, &[1, 2, 3]); } #[test]