You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/07/07 08:12:33 UTC
(datafusion) branch gh-readonly-queue/main/pr-23319-32e11465cd038a5080b9e417443d63d05ba633c6 created (now 9819840fed)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-23319-32e11465cd038a5080b9e417443d63d05ba633c6
in repository https://gitbox.apache.org/repos/asf/datafusion.git
at 9819840fed Add regression tests for hash-join dynamic filter expression policy (#23319)
This branch includes the following new commits:
new 9819840fed Add regression tests for hash-join dynamic filter expression policy (#23319)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org
(datafusion) 01/01: Add regression tests for hash-join dynamic filter expression policy (#23319)
Posted by gi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-23319-32e11465cd038a5080b9e417443d63d05ba633c6
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit 9819840fed5e2966a32ccda5cf42dcc7da517e75
Author: kosiew <ko...@gmail.com>
AuthorDate: Tue Jul 7 16:12:11 2026 +0800
Add regression tests for hash-join dynamic filter expression policy (#23319)
## Which issue does this PR close?
* Part of #22772
## Rationale for this change
This change adds focused regression coverage for the current
dynamic-filter expression assembly performed by
`SharedBuildAccumulator::build_filter`. The intent is to make the
existing expression policy explicit before refactoring, ensuring that
future changes preserve the current pruning semantics for both
`CollectLeft` and `Partitioned` finalize paths.
## What changes are included in this PR?
* Add reusable test helpers for constructing `SharedBuildAccumulator`
instances, partition state, bounds, and pushdown strategies.
* Add unit tests covering `CollectLeft` dynamic-filter expression
assembly for:
* membership-only filters,
* bounds-only filters,
* empty build data that should not update the dynamic filter.
* Add unit tests covering `Partitioned` dynamic-filter expression
assembly for:
* a single real partition with otherwise empty partitions (ensuring no
unnecessary `CASE` expression),
* canceled/unknown partitions (ensuring permissive fallback for unknown
routes).
* Add helper assertions that validate the structural shape of the
generated expressions (for example, `InListExpr`, `BinaryExpr`,
`CaseExpr`, and literal boolean expressions) rather than relying on
brittle textual representations.
## Are these changes tested?
Yes.
This PR adds the following unit tests:
* `collect_left_updates_with_membership_only`
* `collect_left_updates_with_bounds_only`
* `collect_left_empty_build_data_does_not_update_filter`
* `partitioned_one_real_partition_with_rest_empty_skips_case`
*
`partitioned_canceled_unknown_partitions_keep_unknown_routes_permissive`
These tests exercise the dynamic-filter expression policy without
changing production behavior.
## Are there any user-facing changes?
No. This PR only adds regression tests and does not change runtime
behavior.
## LLM-generated code disclosure
This PR includes LLM-generated code and comments. All LLM-generated
content has been manually reviewed.
---
.../src/joins/hash_join/shared_bounds.rs | 197 +++++++++++++++++++++
1 file changed, 197 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 0af4015ff7..5c75e3db1e 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,119 @@ 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<PhysicalExprRef> {
+ vec![Arc::new(Column::new("probe_key", 0))]
+ }
+
+ fn test_probe_schema() -> Arc<Schema> {
+ Arc::new(Schema::new(vec![Field::new(
+ "probe_key",
+ DataType::Int32,
+ false,
+ )]))
+ }
+
+ fn test_dynamic_filter(
+ on_right: &[PhysicalExprRef],
+ ) -> Arc<DynamicFilterPhysicalExpr> {
+ Arc::new(DynamicFilterPhysicalExpr::new(on_right.to_vec(), lit(true)))
+ }
+
+ fn make_accumulator_for_test(
+ data: AccumulatedBuildData,
+ on_right: Vec<PhysicalExprRef>,
+ ) -> SharedBuildAccumulator {
+ let dynamic_filter = test_dynamic_filter(&on_right);
+ SharedBuildAccumulator {
+ inner: Mutex::new(AccumulatorState {
+ data,
+ completion: CompletionState::Pending,
+ }),
+ completion_notify: Notify::new(),
+ 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 {
+ make_accumulator_for_test(
+ AccumulatedBuildData::Partitioned {
+ partitions: vec![PartitionStatus::Pending; num_partitions],
+ completed_partitions: 0,
+ },
+ test_on_right(),
+ )
+ }
+
+ 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()
+ .expect("dynamic filter current expression should be available")
+ }
+
+ fn in_list_expr(expr: &PhysicalExprRef) -> &InListExpr {
+ expr.downcast_ref::<InListExpr>()
+ .expect("expected InListExpr dynamic filter")
+ }
+
+ fn binary_expr(expr: &PhysicalExprRef) -> &BinaryExpr {
+ expr.downcast_ref::<BinaryExpr>()
+ .expect("expected BinaryExpr dynamic filter")
+ }
+
+ fn case_expr(expr: &PhysicalExprRef) -> &CaseExpr {
+ expr.downcast_ref::<CaseExpr>()
+ .expect("expected CaseExpr dynamic filter")
+ }
+
+ fn assert_literal_bool(expr: &PhysicalExprRef, expected: bool) {
+ let literal = expr
+ .downcast_ref::<Literal>()
+ .expect("expected literal bool dynamic filter");
+ assert_eq!(literal.value(), &ScalarValue::Boolean(Some(expected)));
+ }
+
+ fn assert_top_binary_op(expr: &PhysicalExprRef, expected: Operator) {
+ assert_eq!(binary_expr(expr).op(), &expected);
+ }
+
fn partitioned_state(acc: &SharedBuildAccumulator) -> (Vec<PartitionStatus>, usize) {
let guard = acc.inner.lock();
let AccumulatedBuildData::Partitioned {
@@ -754,6 +867,90 @@ 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);
+ in_list_expr(&expr);
+ }
+
+ #[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_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,
+ no_bounds(),
+ )))
+ .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);
+ }
+
+ #[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);
+ in_list_expr(&expr);
+ assert!(expr.downcast_ref::<CaseExpr>().is_none());
+ }
+
+ #[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 = case_expr(&expr);
+ 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
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org