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/08/27 13:51:07 UTC
(datafusion) 01/01: fix: hash join dynamic filter with dictionary-encoded join keys (#24587)
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-24587-2a7a1e3060c42e5ade6ad88b9be667e1b72639bb
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit d66f4b97a5fedd2dc2404edaf3f7a5c8a6b386d2
Author: Jay Zhan <ja...@gmail.com>
AuthorDate: Thu Aug 27 13:50:43 2026 +0000
fix: hash join dynamic filter with dictionary-encoded join keys (#24587)
## Rationale for this change
Hash joins on a dictionary-encoded integer join key misbehave when
dynamic filter pushdown is active. A dictionary array can carry NULL in
its *values* while its key bitmap has no physical nulls (`null_count()
== 0` but `logical_null_count() > 0`), and two places in the hash join
only looked at the physical count:
1. **Wrong results:** for a `NullEqualsNull` join, `keys_have_null` was
computed with `null_count()`, so the pushed dynamic filter was not
widened with `key IS NULL` and pruned the probe-side NULL row that
should have null-matched the build-side NULL.
2. **Query error:** `try_create_array_map` inferred "the key type is
supported by `ArrayMap`" from the presence of min/max bounds. But bounds
are also collected for dynamic filters on *any* key type, so with a
dictionary key and a filter consumer the build side failed with
`Unsupported type for ArrayMap: Dictionary(Int32, Int32)` instead of
falling back to the regular hash map.
## What changes are included in this PR?
- `stream.rs`: compute `keys_have_null` with `logical_null_count()`.
- `exec.rs`: gate `ArrayMap` (perfect hash join) creation on the key
type via `is_perfect_hash_join_candidate` rather than on the presence of
bounds; also use `logical_null_count()` for the `NullEqualsNull` check
in the same function.
- Rename `should_collect_min_max_for_perfect_hash` →
`is_perfect_hash_join_candidate` with a doc comment, since it describes
what is checked rather than a side effect.
## Are these changes tested?
Yes. A new test,
`test_null_equal_dynamic_filter_keeps_probe_nulls_for_build_logical_null`,
builds a `NullEqualsNull` inner join over dictionary keys with a dynamic
filter consumer on the probe side and asserts both that the pushed
filter contains the `IS NULL` disjunct and that the NULL–NULL row is
produced. Before this PR it fails with the `ArrayMap` error; with only
the `ArrayMap` gate it fails on the missing `IS NULL`. Existing hash
join unit tests and the `joins`, `push_down_filter_parquet`, and
`explain_analyze` sqllogictests pass.
## Are there any user-facing changes?
No API changes. Joins on dictionary-encoded integer keys with dynamic
filter pushdown now return correct results instead of erroring /
dropping rows.
---
.../physical-plan/src/joins/hash_join/exec.rs | 237 ++++++++++++++++++++-
.../src/joins/hash_join/shared_bounds.rs | 82 +++++++
.../physical-plan/src/joins/hash_join/stream.rs | 6 +-
3 files changed, 317 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 9a4afe3cdb..440606fb82 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<Option<(ArrayMap, RecordBatch, Vec<ArrayRef>)>> {
- 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);
}
@@ -2542,7 +2544,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<bool> {
@@ -2600,15 +2607,14 @@ async fn collect_left_input(
) -> Result<JoinLeftData> {
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
@@ -2768,7 +2774,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;
}
@@ -7535,6 +7541,225 @@ 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<dyn PhysicalExpr> = 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(())
+ }
+
+ /// The [`PartitionMode::Partitioned`] counterpart of
+ /// [`test_null_equal_dynamic_filter_keeps_probe_nulls_for_build_logical_null`].
+ ///
+ /// Here the build-side logical NULL lives in exactly one build partition, so
+ /// the `keys_have_null` flag has to survive the per-partition aggregation in
+ /// `build_partitioned_filter` and widen the *routed* filter as a whole: the
+ /// probe NULL is routed to a single `CASE` branch, and OR-ing `IS NULL` into
+ /// that one branch would still drop it for every other route.
+ #[tokio::test]
+ async fn test_partitioned_null_equal_dynamic_filter_keeps_probe_nulls_for_build_logical_null()
+ -> Result<()> {
+ let mut session_config = SessionConfig::default();
+ session_config
+ .options_mut()
+ .optimizer
+ .enable_dynamic_filter_pushdown = true;
+ let task_ctx =
+ Arc::new(TaskContext::default().with_session_config(session_config));
+
+ let partition_count = 4;
+
+ // 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)],
+ );
+ // 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 left_key =
+ Arc::new(Column::new_with_schema("c1", &left.schema())?) as PhysicalExprRef;
+ let right_key =
+ Arc::new(Column::new_with_schema("c2", &right.schema())?) as PhysicalExprRef;
+ let on = vec![(Arc::clone(&left_key), Arc::clone(&right_key))];
+
+ // Hash-repartition both sides so the join runs with real partition routing:
+ // the build NULL and the probe NULL land in the same partition, and every
+ // other partition reports a NULL-free build.
+ let left_repartitioned: Arc<dyn ExecutionPlan> =
+ Arc::new(RepartitionExec::try_new(
+ left,
+ Partitioning::Hash(vec![left_key], partition_count),
+ )?);
+ let right_repartitioned: Arc<dyn ExecutionPlan> =
+ Arc::new(RepartitionExec::try_new(
+ right,
+ Partitioning::Hash(vec![Arc::clone(&right_key)], partition_count),
+ )?);
+
+ // The FilterExec sits above the repartition so each probe partition is only
+ // read after the join has waited for every partition's build report, which
+ // makes the finalized (routed and widened) filter the one actually applied.
+ let dynamic_filter = HashJoinExec::create_dynamic_filter(&on);
+ let consumer: Arc<dyn PhysicalExpr> = Arc::clone(&dynamic_filter) as _;
+ let right_filtered: Arc<dyn ExecutionPlan> =
+ Arc::new(FilterExecBuilder::new(consumer, right_repartitioned).build()?);
+
+ let mut join = HashJoinExec::try_new(
+ left_repartitioned,
+ right_filtered,
+ on,
+ None,
+ &JoinType::Inner,
+ None,
+ PartitionMode::Partitioned,
+ NullEquality::NullEqualsNull,
+ false,
+ )?;
+ join.dynamic_filter = Some(HashJoinExecDynamicFilter {
+ filter: Arc::clone(&dynamic_filter),
+ build_accumulator: OnceLock::new(),
+ });
+
+ let batches = crate::execution_plan::collect(Arc::new(join), task_ctx).await?;
+
+ dynamic_filter.wait_complete().await;
+ let filter = dynamic_filter.current()?.to_string();
+ // The aggregated build-side NULL must widen the routed filter once, at the
+ // top: `c2 IS NULL OR CASE hash(c2) % 4 WHEN ... END`.
+ assert!(
+ filter.contains("CASE"),
+ "expected a routed partitioned filter, got: {filter}"
+ );
+ assert!(
+ filter.starts_with("c2@0 IS NULL OR"),
+ "expected the IS NULL disjunct to wrap the whole routed filter, got: {filter}"
+ );
+
+ // The NULL-NULL row (d1 = 200, d2 = 20) is only produced if the probe
+ // NULL survived the pushed filter on its route.
+ 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/shared_bounds.rs b/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs
index 1564ca115e..0cff24332c 100644
--- a/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs
+++ b/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs
@@ -1446,6 +1446,88 @@ mod tests {
)
}
+ fn reported_with_null_keys(
+ pushdown: PushdownStrategy,
+ bounds: PartitionBounds,
+ ) -> PartitionStatus {
+ PartitionStatus::Reported(PartitionData {
+ pushdown,
+ bounds,
+ keys_have_null: true,
+ })
+ }
+
+ /// A partitioned accumulator whose probe key is nullable, so that a build-side
+ /// NULL can actually widen the filter.
+ fn null_equal_partitioned_accumulator(
+ num_partitions: usize,
+ ) -> SharedBuildAccumulator {
+ let mut acc = make_accumulator_for_test(
+ AccumulatedBuildData::Partitioned {
+ partitions: vec![PartitionStatus::Pending; num_partitions],
+ completed_partitions: 0,
+ },
+ test_on_right(),
+ );
+ acc.probe_schema = Arc::new(Schema::new(vec![Field::new(
+ "probe_key",
+ DataType::Int32,
+ true,
+ )]));
+ acc.null_equality = NullEquality::NullEqualsNull;
+ acc
+ }
+
+ /// The IS NULL disjunct must wrap the routed `CASE` as a whole. A probe NULL
+ /// takes exactly one route, so widening only the branch of the partition that
+ /// reported the NULL would still drop it on every other route.
+ #[test]
+ fn partitioned_null_keys_in_one_partition_widen_whole_routed_filter() {
+ let acc = null_equal_partitioned_accumulator(2);
+
+ acc.build_filter(FinalizeInput::Partitioned(vec![
+ reported(in_list(&[1]), no_bounds()),
+ reported_with_null_keys(in_list(&[2]), no_bounds()),
+ ]))
+ .unwrap();
+
+ let expr = current_expr(&acc);
+ assert_top_binary_op(&expr, Operator::Or);
+ let widened = binary_expr(&expr);
+ assert!(
+ widened.left().downcast_ref::<IsNullExpr>().is_some(),
+ "expected the IS NULL disjunct first, got: {expr}"
+ );
+ case_expr(widened.right());
+ assert_eq!(
+ format!("{expr}").matches("IS NULL").count(),
+ 1,
+ "the routed filter must be widened once, not per branch"
+ );
+ }
+
+ /// A canceled partition's build content is unknown, so it may hold a NULL key:
+ /// the aggregated flag must be permissive even when no partition reported one.
+ #[test]
+ fn partitioned_canceled_partition_widens_routed_filter() {
+ let acc = null_equal_partitioned_accumulator(2);
+
+ acc.build_filter(FinalizeInput::Partitioned(vec![
+ PartitionStatus::CanceledUnknown,
+ reported(in_list(&[2]), no_bounds()),
+ ]))
+ .unwrap();
+
+ let expr = current_expr(&acc);
+ assert_top_binary_op(&expr, Operator::Or);
+ assert!(
+ binary_expr(&expr)
+ .left()
+ .downcast_ref::<IsNullExpr>()
+ .is_some()
+ );
+ }
+
#[test]
fn preserve_probe_nulls_only_widens_nullable_keys() {
let probe_schema = Arc::new(Schema::new(vec![
diff --git a/datafusion/physical-plan/src/joins/hash_join/stream.rs b/datafusion/physical-plan/src/joins/hash_join/stream.rs
index 63a94e1987..fe6eaff9e5 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 {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org