From 71456dbbb4721978a821e53fd557eb2f4653cd8f Mon Sep 17 00:00:00 2001 From: Jay Zhan Date: Sun, 23 Aug 2026 13:01:41 +0800 Subject: [PATCH] fix: hash join dynamic filter with dictionary-encoded join keys - Use logical_null_count() for keys_have_null so a null-equal join's pushed filter keeps probe NULL rows when the build NULL lives in a dictionary value - Gate ArrayMap (perfect hash join) creation on the key type instead of the presence of bounds, which are also collected for dynamic filters on any type - Rename should_collect_min_max_for_perfect_hash -> is_perfect_hash_join_candidate --- .../physical-plan/src/joins/hash_join/exec.rs | 125 +++++++++++++++++- .../src/joins/hash_join/stream.rs | 6 +- 2 files changed, 123 insertions(+), 8 deletions(-) diff --git a/datafusion/physical-plan/src/joins/hash_join/exec.rs b/datafusion/physical-plan/src/joins/hash_join/exec.rs index aa06015be7137..33803a6a7439e 100644 --- a/datafusion/physical-plan/src/joins/hash_join/exec.rs +++ b/datafusion/physical-plan/src/joins/hash_join/exec.rs @@ -118,7 +118,9 @@ fn try_create_array_map( perfect_hash_join_min_key_density: f64, null_equality: NullEquality, ) -> Result)>> { - if on_left.len() != 1 { + // `bounds` are also collected for dynamic filters, on any key type, so the + // key type cannot be inferred from their presence. + if !is_perfect_hash_join_candidate(on_left, schema)? { return Ok(None); } @@ -2524,7 +2526,12 @@ impl BuildSideState { } } -fn should_collect_min_max_for_perfect_hash( +/// Returns whether the build side could be joined with an [`ArrayMap`] +/// (perfect hash join): a single join key of a supported integer type. +/// +/// Only a candidate: the final decision also depends on the observed key +/// range and density, see [`try_create_array_map`]. +fn is_perfect_hash_join_candidate( on_left: &[PhysicalExprRef], schema: &SchemaRef, ) -> Result { @@ -2582,15 +2589,14 @@ async fn collect_left_input( ) -> Result { let schema = left_stream.schema(); - let should_collect_min_max_for_phj = - should_collect_min_max_for_perfect_hash(&on_left, &schema)?; + let is_phj_candidate = is_perfect_hash_join_candidate(&on_left, &schema)?; let initial = BuildSideState::try_new( metrics, reservation, on_left.clone(), &schema, - should_compute_dynamic_filters || should_collect_min_max_for_phj, + should_compute_dynamic_filters || is_phj_candidate, )?; let state = left_stream @@ -2750,7 +2756,7 @@ async fn collect_left_input( } }; - if should_collect_min_max_for_phj && !should_compute_dynamic_filters { + if is_phj_candidate && !should_compute_dynamic_filters { bounds = None; } @@ -7517,6 +7523,113 @@ mod tests { Ok(()) } + /// Regression test for a `NullEqualsNull` join whose build-side join key is + /// a dictionary with NULL stored in the dictionary *values*. + /// + /// Such a key is only logically NULL: `null_count() == 0` because the key + /// bitmap has no physical nulls, but `logical_null_count() > 0`. The hash + /// join must still treat it as a NULL key in two places: + /// + /// 1. The build side must not be turned into an `ArrayMap` (perfect hash + /// join), which only supports plain integer keys. Bounds are collected + /// for the dynamic filter regardless of key type, so the `ArrayMap` + /// path must check the key type itself instead of inferring it from + /// the presence of bounds; otherwise the build fails with + /// `Unsupported type for ArrayMap`. + /// 2. The dynamic filter pushed to the probe side must be widened with + /// `c2 IS NULL OR ...`, so the probe row with a NULL key still reaches + /// the join and null-matches the build NULL. Without it the filter + /// `c2 >= 1 AND c2 <= 2 AND c2 IN (...)` evaluates to NULL for that + /// row and silently drops it. + /// + /// Setup: build keys `[1, NULL, 2]`, probe keys `[1, NULL, 4]`, inner join + /// with a `FilterExec` consuming the join's dynamic filter on the probe side. + /// Expected output: the `1 = 1` row and the `NULL = NULL` row. + #[tokio::test] + async fn test_null_equal_dynamic_filter_keeps_probe_nulls_for_build_logical_null() + -> Result<()> { + let task_ctx = Arc::new(TaskContext::default()); + + // Dictionary values: [1, NULL, 2]; keys: [0, 1, 2] => logical [1, NULL, 2] + let left = build_table_dict_key( + "c1", + vec![Some(1), None, Some(2)], + vec![0, 1, 2], + "d1", + vec![Some(100), Some(200), Some(300)], + ); + let left_key = left + .execute(0, Arc::clone(&task_ctx))? + .next() + .await + .unwrap()?; + assert_eq!(left_key.column(0).null_count(), 0); + assert_eq!(left_key.column(0).logical_null_count(), 1); + + // Dictionary values: [1, NULL, 4]; keys: [0, 1, 2] => logical [1, NULL, 4] + let right = build_table_dict_key( + "c2", + vec![Some(1), None, Some(4)], + vec![0, 1, 2], + "d2", + vec![Some(10), Some(20), Some(40)], + ); + + let on = vec![( + Arc::new(Column::new_with_schema("c1", &left.schema())?) as _, + Arc::new(Column::new_with_schema("c2", &right.schema())?) as _, + )]; + + // Wire the join's dynamic filter into a FilterExec over the probe side so + // that the filter is actually built (bounds are only collected when the + // probe subtree contains a consumer) and applied to the probe rows. + let dynamic_filter = HashJoinExec::create_dynamic_filter(&on); + let consumer: Arc = Arc::clone(&dynamic_filter) as _; + let right = Arc::new(FilterExecBuilder::new(consumer, right).build()?); + let mut join = HashJoinExec::try_new( + left, + right, + on, + None, + &JoinType::Inner, + None, + PartitionMode::CollectLeft, + NullEquality::NullEqualsNull, + false, + )?; + join.dynamic_filter = Some(HashJoinExecDynamicFilter { + filter: Arc::clone(&dynamic_filter), + build_accumulator: OnceLock::new(), + }); + + // (1) Building the hash table must not fail on the dictionary key. + let stream = join.execute(0, task_ctx)?; + let batches = common::collect(stream).await?; + + // (2) The build-side logical NULL must widen the pushed filter with + // `IS NULL`, because the join is NullEqualsNull. + dynamic_filter.wait_complete().await; + let filter = dynamic_filter.current()?.to_string(); + assert!( + filter.contains("c2@0 IS NULL OR"), + "expected an IS NULL disjunct in the pushed filter, got: {filter}" + ); + + // The NULL-NULL row (d1 = 200, d2 = 20) is only produced if the probe + // NULL survived the pushed filter. + allow_duplicates! { + assert_snapshot!(batches_to_sort_string(&batches), @r" + +----+-----+----+----+ + | c1 | d1 | c2 | d2 | + +----+-----+----+----+ + | | 200 | | 20 | + | 1 | 100 | 1 | 10 | + +----+-----+----+----+ + "); + } + Ok(()) + } + /// Null-aware RightAnti must drop outer rows whose dictionary key is only /// logically NULL (key points at a null dictionary value). #[apply(hash_join_exec_configs)] diff --git a/datafusion/physical-plan/src/joins/hash_join/stream.rs b/datafusion/physical-plan/src/joins/hash_join/stream.rs index 63a94e1987d91..fe6eaff9e53a2 100644 --- a/datafusion/physical-plan/src/joins/hash_join/stream.rs +++ b/datafusion/physical-plan/src/joins/hash_join/stream.rs @@ -559,11 +559,13 @@ impl HashJoinStream { .bounds .clone() .unwrap_or_else(|| PartitionBounds::new(vec![])); - // Arrow tracks null counts per array, so this costs no data scan. + // Use the logical null count: a dictionary key whose entry points at a + // NULL dictionary value is a NULL key even though the key bitmap has no + // physical nulls (`null_count() == 0` but `logical_null_count() > 0`). let keys_have_null = left_data .values() .iter() - .any(|array| array.null_count() > 0); + .any(|array| array.logical_null_count() > 0); let build_data = match self.mode { PartitionMode::Partitioned => PartitionBuildData::Partitioned {