From 5f1625e029c927da5909af015747c2c6ee785d64 Mon Sep 17 00:00:00 2001 From: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com> Date: Wed, 4 Nov 2026 09:15:24 -0300 Subject: [PATCH 1/6] bench: add hj queries for partitioned joins on dense and sparse integer keys No hj query runs a `Partitioned` join on integer keys whose build side is large enough for a hash table or an `ArrayMap` per partition. So the suite does not measure how the build map type changes such a join. This adds two queries over TPC-H `orders`, joined to itself on `o_custkey` in `Partitioned` mode. Q26 builds from the orders of the first 40K customers: dense keys, about 10 rows per key. Q27 builds from customers whose key is a multiple of 37: sparse keys. Both probe 15M rows with about a 2.7% hit rate. The runner now also registers `orders`. Co-Authored-By: Claude Opus 5.5 --- benchmarks/src/hj.rs | 28 +++++++++++++++++++++++++++- 1 file changed, 27 insertions(+), 1 deletion(-) diff --git a/benchmarks/src/hj.rs b/benchmarks/src/hj.rs index 2afd8ce3b85d8..2410e9d975207 100644 --- a/benchmarks/src/hj.rs +++ b/benchmarks/src/hj.rs @@ -505,6 +505,32 @@ const HASH_QUERIES: &[HashJoinQuery] = &[ probe_size: "240K_multi_key_count", isolate_partitioned_join: true, }, + // Q26: Partitioned join on dense integer keys + // Build Side: orders of the first 40K customers (~270K rows, ~27K distinct + // dense keys, ~10 rows per key) | Probe Side: orders (15M rows), ~2.7% hit + // rate. The probe keys are not sorted, so statistics prune nothing. + HashJoinQuery { + sql: r###"SELECT count(*) + FROM (SELECT o_custkey as k FROM orders WHERE o_custkey <= 40000) b + JOIN (SELECT o_custkey as k FROM orders) p ON b.k = p.k"###, + density: 1.0, + prob_hit: 0.027, + build_size: "270K_dense_keys", + probe_size: "15M", + isolate_partitioned_join: true, + }, + // Q27: Q26 with sparse integer keys: a similar build size and hit rate, + // with keys 37 apart. + HashJoinQuery { + sql: r###"SELECT count(*) + FROM (SELECT o_custkey as k FROM orders WHERE o_custkey % 37 = 0) b + JOIN (SELECT o_custkey as k FROM orders) p ON b.k = p.k"###, + density: 0.027, + prob_hit: 0.027, + build_size: "270K_sparse_keys", + probe_size: "15M", + isolate_partitioned_join: true, + }, ]; impl RunOpt { @@ -532,7 +558,7 @@ impl RunOpt { let ctx = SessionContext::new_with_config_rt(config, rt); if let Some(path) = &self.path { - for table in &["lineitem", "supplier", "nation", "customer"] { + for table in &["lineitem", "orders", "supplier", "nation", "customer"] { let table_path = path.join(table); if !table_path.exists() { return exec_err!( From c13331096963c79fda6ea80873be8b736d7b8611 Mon Sep 17 00:00:00 2001 From: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com> Date: Wed, 4 Nov 2026 07:40:17 -0300 Subject: [PATCH 2/6] perf: route probe rows to ArrayMap build partitions in the merged filter A `Partitioned` hash join merges its partitions' build maps into one routed `hash_lookup` filter. The routed lookup accepted only hash tables, so a join whose partitions built an `ArrayMap` (dense integer keys) fell back to a `CASE` over one filter per partition. The routed lookup now probes an `ArrayMap` partition with the key values of the rows routed to it, because an `ArrayMap` is keyed by value, not hash. Every non-empty partition can now merge. The `CASE` fallback remains only for canceled partitions. The routed lookup fuzz test now builds an `ArrayMap` for about half of the partitions. Co-Authored-By: Claude Opus 5.5 --- .../joins/hash_join/partitioned_hash_eval.rs | 67 ++++++++++++------- .../src/joins/hash_join/shared_bounds.rs | 9 +-- 2 files changed, 42 insertions(+), 34 deletions(-) diff --git a/datafusion/physical-plan/src/joins/hash_join/partitioned_hash_eval.rs b/datafusion/physical-plan/src/joins/hash_join/partitioned_hash_eval.rs index 66c7c75f616e6..aac100940873c 100644 --- a/datafusion/physical-plan/src/joins/hash_join/partitioned_hash_eval.rs +++ b/datafusion/physical-plan/src/joins/hash_join/partitioned_hash_eval.rs @@ -7,7 +7,8 @@ use std::{fmt::Display, hash::Hash, sync::Arc}; use arrow::{ - array::{ArrayRef, BooleanArray, BooleanBufferBuilder, UInt64Array}, + array::{ArrayRef, BooleanArray, BooleanBufferBuilder, UInt32Array, UInt64Array}, + compute::take, datatypes::{DataType, Schema}, record_batch::RecordBatch, }; @@ -352,8 +353,7 @@ impl HashTableLookupExpr { /// for pruning only, or `None` /// /// A single non-empty map is probed directly without routing, so it may be - /// any [`Map`] and `router` is ignored. Otherwise `router` is required and - /// every map must be a [`Map::HashMap`]. + /// any [`Map`] and `router` is ignored. Otherwise `router` is required. /// /// # Public Only for Internal Use: /// `datafusion-proto` tests require this constructor, but it is not part of @@ -375,21 +375,10 @@ impl HashTableLookupExpr { "a lookup over anything but one non-empty map needs a partition router" ); } - (Some(router), _) => { - if maps - .iter() - .flatten() - .any(|map| !matches!(map.as_ref(), Map::HashMap(_))) - { - return internal_err!( - "routed lookups only support hash table partitions" - ); - } - Lookup::Routed { - router, - maps: maps.into(), - } - } + (Some(router), _) => Lookup::Routed { + router, + maps: maps.into(), + }, }; Ok(Self { on_columns, @@ -441,16 +430,23 @@ impl HashTableLookupExpr { for (partition, map) in maps.iter().enumerate() { // An empty partition holds no key, so its rows stay unmatched. let Some(map) = map else { continue }; - let Map::HashMap(map) = map.as_ref() else { - return internal_err!( - "routed lookups only support hash table partitions" - ); - }; let run = starts[partition]..starts[partition + 1]; if run.is_empty() { continue; } - let hits = map.contain_hashes(&sorted_hashes[run.clone()]); + let hits = match map.as_ref() { + Map::HashMap(map) => map.contain_hashes(&sorted_hashes[run.clone()]), + // An array map is keyed by value, not hash, so it probes + // the run's own key values. + Map::ArrayMap(map) => { + let indices = UInt32Array::from(rows[run.clone()].to_vec()); + let run_keys = keys + .iter() + .map(|key| take(key.as_ref(), &indices, None)) + .collect::, _>>()?; + map.contain_keys(&run_keys)? + } + }; for hit in hits.values().set_indices() { matched.set_bit(rows[run.start + hit] as usize, true); } @@ -639,7 +635,9 @@ fn evaluate_columns( #[cfg(test)] mod tests { use super::*; + use crate::joins::array_map::ArrayMap; use crate::joins::join_hash_map::JoinHashMapU32; + use arrow::array::AsArray; use datafusion_common::ScalarValue; use datafusion_physical_expr::expressions::Column; use std::collections::hash_map::DefaultHasher; @@ -1111,7 +1109,7 @@ mod tests { )])); let batch = RecordBatch::try_new(schema, vec![Arc::clone(probe)])?; let matched = expr.evaluate(&batch)?.into_array(batch.num_rows())?; - Ok(arrow::array::AsArray::as_boolean(&matched).clone()) + Ok(matched.as_boolean().clone()) } /// A hash table of `keys`, hashed with `random_state` as the build side @@ -1125,6 +1123,18 @@ mod tests { Ok(Arc::new(Map::HashMap(Box::new(map)))) } + /// An array map of `keys`, as a dense build side gets. + fn array_map(keys: &ArrayRef) -> Result> { + let keys = keys.as_primitive::(); + let key = |value: Option| { + ArrayMap::key_to_u64(&ScalarValue::Int32(value)).expect("Int32 key") + }; + let min = key(arrow::compute::min(keys)); + let max = key(arrow::compute::max(keys)); + let keys: ArrayRef = Arc::new(keys.clone()); + Ok(Arc::new(Map::ArrayMap(ArrayMap::try_new(&keys, min, max)?))) + } + /// One randomly generated scenario, see [`routed_lookup_fuzz`]. fn run_routed_lookup_scenario(seed: u64) -> Result<()> { use crate::repartition::{BatchPartitioner, REPARTITION_RANDOM_STATE}; @@ -1166,7 +1176,12 @@ mod tests { .partition( RecordBatch::try_new(schema, vec![Arc::clone(&build)])?, |partition, batch| { - maps[partition] = Some(hash_table(batch.column(0), &random_state)?); + // A dense partition builds an array map, others a hash table. + maps[partition] = Some(if rng.random_bool(0.5) { + array_map(batch.column(0))? + } else { + hash_table(batch.column(0), &random_state)? + }); 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 321ca4b10db59..a4e95c320824b 100644 --- a/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs +++ b/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs @@ -721,14 +721,7 @@ impl SharedBuildAccumulator { }) => { real_partition_ids.push(partition_id); keys_have_null |= partition_keys_have_null; - let mergeable = match &pushdown { - PushdownStrategy::Map(map, _) => { - matches!(map.as_ref(), Map::HashMap(_)) - } - PushdownStrategy::InList(_) => true, - PushdownStrategy::Empty => false, - }; - members = members.filter(|_| mergeable).map(|mut members| { + members = members.map(|mut members| { members.push((partition_id, pushdown.clone(), bounds.clone())); members }); From f29afddb8287a015ef0686c10ae47f38f7b22a61 Mon Sep 17 00:00:00 2001 From: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com> Date: Wed, 4 Nov 2026 07:46:13 -0300 Subject: [PATCH 3/6] test: cover signed zero and composite nullable keys in RangeRouter Two hash join dynamic filter tests checked these range routing cases through the `CASE` branch of a canceled partition. The next commit makes a canceled partition reject its rows, which leaves those tests nothing to route. This adds both cases as `RangeRouter` tests, where they belong: a `0.0` split point separates `-0.0` from `0.0`, and each key of a composite key places its NULLs by its own `nulls_first`. Co-Authored-By: Claude Opus 5.5 --- .../physical-plan/src/repartition/range.rs | 45 +++++++++++++++++++ 1 file changed, 45 insertions(+) diff --git a/datafusion/physical-plan/src/repartition/range.rs b/datafusion/physical-plan/src/repartition/range.rs index 337e3752d6da3..f0f64ee00af0d 100644 --- a/datafusion/physical-plan/src/repartition/range.rs +++ b/datafusion/physical-plan/src/repartition/range.rs @@ -713,6 +713,51 @@ mod tests { ) } + /// `-0.0` sorts before `0.0`, so a `0.0` split point separates them. + #[test] + fn test_float_router_signed_zero() -> Result<()> { + let split_points = make_split_points_1d(vec![ScalarValue::Float64(Some(0.0))]); + let router = RangeRouter::try_new(&[SortOptions::default()], &split_points)?; + let input = Arc::new(Float64Array::from(vec![-0.0, 0.0])) as ArrayRef; + assert_routing(&router, &[input], &[0, 1], None) + } + + /// NULLs in composite keys follow each key's own `nulls_first`. + #[test] + fn test_router_composite_nullable_keys() -> Result<()> { + let split_points = vec![ + SplitPoint::new(vec![ScalarValue::Int32(None), ScalarValue::Int32(Some(10))]), + SplitPoint::new(vec![ScalarValue::Int32(None), ScalarValue::Int32(None)]), + SplitPoint::new(vec![ScalarValue::Int32(Some(10)), ScalarValue::Int32(None)]), + ]; + let sort_options = [ + SortOptions::new(false, true), + SortOptions::new(false, false), + ]; + let router = RangeRouter::try_new(&sort_options, &split_points)?; + let key = Arc::new(Int32Array::from(vec![ + None, + None, + None, + None, + Some(9), + Some(10), + Some(10), + Some(11), + ])) as ArrayRef; + let tie = Arc::new(Int32Array::from(vec![ + Some(9), + Some(10), + Some(11), + None, + None, + Some(9), + None, + None, + ])) as ArrayRef; + assert_routing(&router, &[key, tie], &[0, 1, 1, 2, 2, 2, 3, 3], None) + } + #[test] fn test_row_converter_strings() -> Result<()> { let split_points = make_split_points_1d(vec![ From 583c4448fdebaa4fb1d157f8a4188786c07534d5 Mon Sep 17 00:00:00 2001 From: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com> Date: Wed, 4 Nov 2026 07:46:13 -0300 Subject: [PATCH 4/6] fix: reject probe rows routed to a canceled hash join partition A `Partitioned` hash join marks a partition canceled when its stream is dropped before it reports its build data. The dynamic filter kept every probe row routed to a canceled partition, because the partition's build keys were unknown. That choice was safe but bought nothing: a stream probes only after its report is delivered, so a canceled stream has emitted no row and never will. Every row routed to it is discarded either way. It also cost something. A canceled partition forced the routed `CASE` fallback instead of the merged lookup. Under `NullEqualsNull` it also made every live partition keep probe-side NULL rows. A canceled partition now counts as empty. The filter drops the flag that tracked cancellation and the `ELSE true` it fed. The `CASE` fallback is now unreachable. The next commit removes it. Co-Authored-By: Claude Opus 5.5 --- .../src/joins/hash_join/shared_bounds.rs | 266 +++--------------- 1 file changed, 34 insertions(+), 232 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 a4e95c320824b..29a3262524791 100644 --- a/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs +++ b/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs @@ -692,13 +692,10 @@ impl SharedBuildAccumulator { } /// Builds one routed probe-side filter from finalized partitioned build data. - /// Empty partitions reject their routed rows, while canceled partitions stay - /// permissive because their build contents are unknown. + /// Empty and canceled partitions reject their routed rows. fn build_partitioned_filter(&self, partitions: Vec) -> Result<()> { let mut partition_filters = Vec::with_capacity(partitions.len()); let mut real_partition_ids = Vec::new(); - let mut empty_partition_ids = Vec::new(); - let mut has_canceled_unknown = false; let mut keys_have_null = false; // Every non-empty partition's membership strategy and bounds, merged into // one membership filter below; `None` once a partition cannot be merged. @@ -707,11 +704,14 @@ impl SharedBuildAccumulator { for (partition_id, partition) in partitions.into_iter().enumerate() { match partition { - PartitionStatus::Reported(PartitionData { + // A canceled partition's stream emits no row: it is dropped + // before its report is delivered, and a stream probes only + // after delivery. So no key of its build side can matter. + PartitionStatus::CanceledUnknown + | PartitionStatus::Reported(PartitionData { pushdown: PushdownStrategy::Empty, .. }) => { - empty_partition_ids.push(partition_id); partition_filters.push(lit(false)); } PartitionStatus::Reported(PartitionData { @@ -742,15 +742,6 @@ impl SharedBuildAccumulator { .unwrap_or_else(|| lit(true)); partition_filters.push(then_expr); } - PartitionStatus::CanceledUnknown => { - has_canceled_unknown = true; - partition_filters.push(lit(true)); - // Rows routed here must pass, which a merged filter cannot express. - members = None; - // A canceled partition's build content is unknown, so it - // may hold a NULL key. - keys_have_null = true; - } PartitionStatus::Pending => { return datafusion_common::internal_err!( "attempted to finalize dynamic filter with pending partition" @@ -759,20 +750,10 @@ impl SharedBuildAccumulator { } } - let all_partitions_canceled = has_canceled_unknown - && real_partition_ids.is_empty() - && empty_partition_ids.is_empty(); - let all_partitions_empty = !has_canceled_unknown && real_partition_ids.is_empty(); - let one_non_empty_partition = - !has_canceled_unknown && real_partition_ids.len() == 1; - - let filter_expr = if all_partitions_canceled { - // No build data is known, so filtering any probe row could discard a match. - lit(true) - } else if all_partitions_empty { + let filter_expr = if real_partition_ids.is_empty() { // No build row exists, so no probe row can match. lit(false) - } else if one_non_empty_partition { + } else if real_partition_ids.len() == 1 { // Only one build partition contains rows, so its filter covers every // possible probe match without routing. Arc::clone(&partition_filters[real_partition_ids[0]]) @@ -784,34 +765,25 @@ impl SharedBuildAccumulator { )? } else { // Builds the shared sparse `CASE` for partition filter routing. - // Without cancellation, omitted branches are known empty and safely fall - // through to `ELSE false`. With cancellation, omitted canceled partitions - // have unknown contents and must fall through to `ELSE true`, so known-empty - // partitions are emitted explicitly as false branches. - let mut branches = if has_canceled_unknown { - empty_partition_ids - .iter() - .map(|&partition_id| (lit(partition_id as u64), lit(false))) - .collect::>() - } else { - vec![] - }; - branches.extend(real_partition_ids.iter().map(|&partition_id| { - ( - lit(partition_id as u64), - Arc::clone(&partition_filters[partition_id]), - ) - })); + // Omitted branches are known empty and fall through to `ELSE false`. + let branches = real_partition_ids + .iter() + .map(|&partition_id| { + ( + lit(partition_id as u64), + Arc::clone(&partition_filters[partition_id]), + ) + }) + .collect::>(); let routing_expr = if let Some(range_partitioning) = &self.probe_range_partitioning { // Routes probe rows with the partition id selected by [`RangeExpr`]. // CASE range_partition(keys) - // WHEN empty_partition_id THEN false -- only when cancellation exists // WHEN real_partition_id THEN F(real_partition_id) // ... - // ELSE has_canceled_unknown + // ELSE false // END assert_or_internal_err!( partition_filters.len() == range_partitioning.partition_count(), @@ -828,10 +800,9 @@ impl SharedBuildAccumulator { // Routes probe rows with the same `hash(keys) % partition_count` expression used // by Hash repartitioning. // CASE hash(keys) % partition_count - // WHEN empty_partition_id THEN false -- only when cancellation exists // WHEN real_partition_id THEN F(real_partition_id) // ... - // ELSE has_canceled_unknown + // ELSE false // END let routing_hash_expr = Arc::new(HashExpr::new( self.on_right.clone(), @@ -848,7 +819,7 @@ impl SharedBuildAccumulator { Arc::new(CaseExpr::try_new( Some(routing_expr), branches, - Some(lit(has_canceled_unknown)), + Some(lit(false)), )?) }; @@ -1115,8 +1086,7 @@ mod tests { use crate::joins::hash_join::not_in::{NotInOutput, NotInScope}; use crate::joins::join_hash_map::JoinHashMapU32; - use arrow::array::{ArrayRef, BooleanArray, Float64Array, Int32Array}; - use arrow::compute::SortOptions; + use arrow::array::{ArrayRef, BooleanArray, Int32Array}; use arrow::record_batch::RecordBatch; use datafusion_common::SplitPoint; use datafusion_physical_expr::{ @@ -1257,11 +1227,6 @@ mod tests { .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::() @@ -1501,7 +1466,7 @@ mod tests { } #[test] - fn partitioned_canceled_unknown_partitions_keep_unknown_routes_permissive() { + fn partitioned_canceled_and_empty_partitions_reject_every_row() { let acc = make_partitioned_expr_accumulator_for_test(2); acc.build_filter(FinalizeInput::Partitioned(vec![ @@ -1510,14 +1475,7 @@ mod tests { ])) .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, - ); + assert_literal_bool(¤t_expr(&acc), false); } #[test] @@ -1536,26 +1494,20 @@ mod tests { ], )?); + // No combined IN list, so each probe row is routed to its partition. + acc.in_list_max_values = 0; + acc.build_filter(FinalizeInput::Partitioned(vec![ reported(PushdownStrategy::Empty, no_bounds()), - PartitionStatus::CanceledUnknown, + reported(in_list(&[10, 19]), no_bounds()), 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() - .and_then(|expr| expr.downcast_ref::()) - .is_some(), - "Range routing must use RangeExpr" - ); - assert_eq!(case.when_then_expr().len(), 3); - assert_literal_bool(&case.when_then_expr()[0].1, false); - assert_literal_bool( - case.else_expr().expect("expected permissive fallback"), - true, + format!("{expr:?}").contains("routed by range over 4 partitions"), + "{expr:?}" ); let batch = RecordBatch::try_new( @@ -1577,150 +1529,6 @@ mod tests { Ok(()) } - #[test] - fn partitioned_range_dynamic_filter_routes_compound_nullable_keys() -> Result<()> { - let probe_schema = Arc::new(Schema::new(vec![ - Field::new("probe_key", DataType::Int32, true), - Field::new("probe_tie", DataType::Int32, true), - ])); - let on_right: Vec = vec![ - Arc::new(Column::new("probe_key", 0)), - Arc::new(Column::new("probe_tie", 1)), - ]; - let mut acc = make_accumulator_for_test( - AccumulatedBuildData::Partitioned { - partitions: vec![PartitionStatus::Pending; 4], - completed_partitions: 0, - }, - on_right, - ); - acc.probe_schema = Arc::clone(&probe_schema); - acc.probe_range_partitioning = Some(RangePartitioning::try_new( - [ - PhysicalSortExpr::new( - Arc::clone(&acc.on_right[0]), - SortOptions::new(false, true), - ), - PhysicalSortExpr::new( - Arc::clone(&acc.on_right[1]), - SortOptions::new(false, false), - ), - ] - .into(), - vec![ - SplitPoint::new(vec![ - ScalarValue::Int32(None), - ScalarValue::Int32(Some(10)), - ]), - SplitPoint::new(vec![ScalarValue::Int32(None), ScalarValue::Int32(None)]), - SplitPoint::new(vec![ - ScalarValue::Int32(Some(10)), - ScalarValue::Int32(None), - ]), - ], - )?); - - acc.build_filter(FinalizeInput::Partitioned(vec![ - reported(PushdownStrategy::Empty, no_bounds()), - PartitionStatus::CanceledUnknown, - reported(PushdownStrategy::Empty, no_bounds()), - PartitionStatus::CanceledUnknown, - ]))?; - - let expr = current_expr(&acc); - 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 permissive fallback"), - true, - ); - - let batch = RecordBatch::try_new( - probe_schema, - vec![ - Arc::new(Int32Array::from(vec![ - None, - None, - None, - None, - Some(9), - Some(10), - Some(10), - Some(11), - ])), - Arc::new(Int32Array::from(vec![ - Some(9), - Some(10), - Some(11), - None, - None, - Some(9), - None, - None, - ])), - ], - )?; - 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, false, false, false, true, true,] - ) - ); - - Ok(()) - } - - #[test] - fn partitioned_range_dynamic_filter_preserves_signed_zero_routing() -> Result<()> { - let probe_schema = Arc::new(Schema::new(vec![Field::new( - "probe_key", - DataType::Float64, - false, - )])); - let on_right: Vec = vec![Arc::new(Column::new("probe_key", 0))]; - let mut acc = make_accumulator_for_test( - AccumulatedBuildData::Partitioned { - partitions: vec![PartitionStatus::Pending; 2], - completed_partitions: 0, - }, - on_right, - ); - acc.probe_schema = Arc::clone(&probe_schema); - acc.probe_range_partitioning = Some(RangePartitioning::try_new( - [PhysicalSortExpr::new( - Arc::clone(&acc.on_right[0]), - SortOptions::default(), - )] - .into(), - vec![SplitPoint::new(vec![ScalarValue::Float64(Some(0.0))])], - )?); - - acc.build_filter(FinalizeInput::Partitioned(vec![ - PartitionStatus::CanceledUnknown, - reported(PushdownStrategy::Empty, no_bounds()), - ]))?; - - let expr = current_expr(&acc); - let batch = RecordBatch::try_new( - probe_schema, - vec![Arc::new(Float64Array::from(vec![-0.0, 0.0]))], - )?; - 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![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 @@ -1881,10 +1689,10 @@ mod tests { ); } - /// 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. + /// A canceled partition holds no key, so it cannot add a NULL key that + /// would make a null-equal join keep probe-side NULL rows. #[test] - fn partitioned_canceled_partition_widens_routed_filter() { + fn partitioned_canceled_partition_does_not_widen_filter() { let acc = null_equal_partitioned_accumulator(2); acc.build_filter(FinalizeInput::Partitioned(vec![ @@ -1894,13 +1702,7 @@ mod tests { .unwrap(); let expr = current_expr(&acc); - assert_top_binary_op(&expr, Operator::Or); - assert!( - binary_expr(&expr) - .left() - .downcast_ref::() - .is_some() - ); + assert_eq!(format!("{expr}").matches("IS NULL").count(), 0, "{expr}"); } #[test] From 7ca7019315daf129539f0923604dbda4d4d830f8 Mon Sep 17 00:00:00 2001 From: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com> Date: Wed, 4 Nov 2026 07:48:24 -0300 Subject: [PATCH 5/6] refactor: remove the unreachable CASE fallback from the partitioned filter The partitioned dynamic filter built a `CASE` over one filter per partition when a partition could not merge into the routed lookup. After the previous two commits every partition merges: an `ArrayMap` is routed, and a canceled partition counts as empty. The `CASE` builder, its per-partition filters, and the single-partition branch had no input left to handle. One non-empty partition already becomes a direct lookup in the merged path. This removes that code and the test assertions that a filter is not a `CASE`. No behavior changes. Co-Authored-By: Claude Opus 5.5 --- .../src/joins/hash_join/shared_bounds.rs | 122 ++---------------- 1 file changed, 13 insertions(+), 109 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 29a3262524791..5b4fe395578fa 100644 --- a/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs +++ b/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs @@ -17,7 +17,7 @@ use crate::joins::hash_join::exec::HASH_JOIN_SEED; use crate::joins::hash_join::inlist_builder::build_struct_fields; use crate::joins::hash_join::not_in::NotIn; use crate::joins::hash_join::partitioned_hash_eval::{ - HashExpr, HashTableLookupExpr, PartitionRouter, SeededRandomState, + HashTableLookupExpr, PartitionRouter, SeededRandomState, }; use crate::joins::join_hash_map::{JoinHashMapType, JoinHashMapU32}; use crate::joins::key_range_bitmap::{KeyRangeBitmap, KeyRangePart}; @@ -34,7 +34,7 @@ use datafusion_common::{ use datafusion_expr::Operator; use datafusion_functions::core::r#struct as struct_func; use datafusion_physical_expr::expressions::{ - BinaryExpr, CaseExpr, DynamicFilterPhysicalExpr, InListExpr, IsNullExpr, lit, + BinaryExpr, DynamicFilterPhysicalExpr, InListExpr, IsNullExpr, lit, }; use datafusion_physical_expr::{ PhysicalExpr, PhysicalExprRef, RangePartitioning, ScalarFunctionExpr, @@ -691,16 +691,13 @@ impl SharedBuildAccumulator { } } - /// Builds one routed probe-side filter from finalized partitioned build data. - /// Empty and canceled partitions reject their routed rows. + /// Builds one probe-side filter from finalized partitioned build data. + /// Empty and canceled partitions contribute no key. fn build_partitioned_filter(&self, partitions: Vec) -> Result<()> { - let mut partition_filters = Vec::with_capacity(partitions.len()); - let mut real_partition_ids = Vec::new(); - let mut keys_have_null = false; - // Every non-empty partition's membership strategy and bounds, merged into - // one membership filter below; `None` once a partition cannot be merged. - let mut members = Some(Vec::with_capacity(partitions.len())); + let num_partitions = partitions.len(); + let mut members = Vec::with_capacity(num_partitions); let mut merged_bounds: Option = None; + let mut keys_have_null = false; for (partition_id, partition) in partitions.into_iter().enumerate() { match partition { @@ -711,118 +708,33 @@ impl SharedBuildAccumulator { | PartitionStatus::Reported(PartitionData { pushdown: PushdownStrategy::Empty, .. - }) => { - partition_filters.push(lit(false)); - } + }) => {} PartitionStatus::Reported(PartitionData { bounds, pushdown, keys_have_null: partition_keys_have_null, }) => { - real_partition_ids.push(partition_id); keys_have_null |= partition_keys_have_null; - members = members.map(|mut members| { - members.push((partition_id, pushdown.clone(), bounds.clone())); - members - }); merged_bounds = Some(match merged_bounds { None => bounds.clone(), Some(merged) => merged.union(&bounds), }); - - let membership_expr = create_membership_predicate( - &self.on_right, - pushdown, - &HASH_JOIN_SEED, - self.probe_schema.as_ref(), - )?; - let bounds_expr = create_bounds_predicate(&self.on_right, &bounds); - let then_expr = - combine_membership_and_bounds(membership_expr, bounds_expr) - .unwrap_or_else(|| lit(true)); - partition_filters.push(then_expr); + members.push((partition_id, pushdown, bounds)); } PartitionStatus::Pending => { - return datafusion_common::internal_err!( + return internal_err!( "attempted to finalize dynamic filter with pending partition" ); } } } - let filter_expr = if real_partition_ids.is_empty() { + let filter_expr = if members.is_empty() { // No build row exists, so no probe row can match. lit(false) - } else if real_partition_ids.len() == 1 { - // Only one build partition contains rows, so its filter covers every - // possible probe match without routing. - Arc::clone(&partition_filters[real_partition_ids[0]]) - } else if let Some(members) = members { - self.merged_membership_filter( - &members, - merged_bounds, - partition_filters.len(), - )? } else { - // Builds the shared sparse `CASE` for partition filter routing. - // Omitted branches are known empty and fall through to `ELSE false`. - let branches = real_partition_ids - .iter() - .map(|&partition_id| { - ( - lit(partition_id as u64), - Arc::clone(&partition_filters[partition_id]), - ) - }) - .collect::>(); - - let routing_expr = if let Some(range_partitioning) = - &self.probe_range_partitioning - { - // Routes probe rows with the partition id selected by [`RangeExpr`]. - // CASE range_partition(keys) - // WHEN real_partition_id THEN F(real_partition_id) - // ... - // ELSE false - // END - assert_or_internal_err!( - partition_filters.len() == range_partitioning.partition_count(), - "Dynamic filter partition count {} does not match Range partition count {}", - partition_filters.len(), - range_partitioning.partition_count() - ); - Arc::new(RangeExpr::try_new_with_schema( - self.on_right.clone(), - range_partitioning, - &self.probe_schema, - )?) as Arc - } else { - // Routes probe rows with the same `hash(keys) % partition_count` expression used - // by Hash repartitioning. - // CASE hash(keys) % partition_count - // WHEN real_partition_id THEN F(real_partition_id) - // ... - // ELSE false - // END - let routing_hash_expr = Arc::new(HashExpr::new( - self.on_right.clone(), - self.repartition_random_state.clone(), - "hash_repartition".to_string(), - )) as Arc; - Arc::new(BinaryExpr::new( - routing_hash_expr, - Operator::Modulo, - lit(partition_filters.len() as u64), - )) as Arc - }; - - Arc::new(CaseExpr::try_new( - Some(routing_expr), - branches, - Some(lit(false)), - )?) + self.merged_membership_filter(&members, merged_bounds, num_partitions)? }; - self.dynamic_filter .update(self.preserve_probe_nulls(filter_expr, keys_have_null)?) } @@ -1317,7 +1229,6 @@ mod tests { .unwrap(); let expr = current_expr(&acc); - assert!(expr.downcast_ref::().is_none()); assert_eq!( expr.to_string(), "probe_key@0 >= 1 AND probe_key@0 <= 10 AND hash_lookup" @@ -1394,11 +1305,6 @@ mod tests { ))) .unwrap(); - assert!( - current_expr(&acc).downcast_ref::().is_none(), - "expected a merged lookup, got: {}", - current_expr(&acc) - ); let probe: Vec = (0..40).collect(); let expected: Vec = probe.iter().map(|key| key % 2 == 0).collect(); assert_matches(&acc, &probe, &expected); @@ -1424,8 +1330,7 @@ mod tests { let expr = current_expr(&acc); assert!( - expr.downcast_ref::().is_none() - && expr.downcast_ref::().is_none(), + expr.downcast_ref::().is_none(), "expected a hash lookup, got: {expr}" ); let probe: Vec = (0..20).collect(); @@ -1462,7 +1367,6 @@ mod tests { let expr = current_expr(&acc); in_list_expr(&expr); - assert!(expr.downcast_ref::().is_none()); } #[test] From e51f5ed837c416998b1dc2cd357894b0654b6f70 Mon Sep 17 00:00:00 2001 From: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com> Date: Fri, 6 Nov 2026 10:06:51 -0300 Subject: [PATCH 6/6] fix: keep the own filter of a lone non-empty partition With every partition merged into one routed lookup, a build where only one partition holds rows lost its IN list. The merged IN-list limit counts duplicate rows, so one key with many build rows became a hash_lookup and the probe scan lost bloom filter pruning. This restores the branch that uses that partition's own filter, and adds tests for it and for a canceled partition next to live ones. It also drops the remaining comments and test names that describe the removed CASE, and documents why HashExpr stays. Co-Authored-By: Claude Opus 5.5 --- .../physical-plan/src/joins/hash_join/exec.rs | 4 +- .../joins/hash_join/partitioned_hash_eval.rs | 10 +-- .../src/joins/hash_join/shared_bounds.rs | 83 +++++++++++++++++-- 3 files changed, 81 insertions(+), 16 deletions(-) diff --git a/datafusion/physical-plan/src/joins/hash_join/exec.rs b/datafusion/physical-plan/src/joins/hash_join/exec.rs index e22cf6dc2d8b9..cbc5d355de617 100644 --- a/datafusion/physical-plan/src/joins/hash_join/exec.rs +++ b/datafusion/physical-plan/src/joins/hash_join/exec.rs @@ -10018,8 +10018,8 @@ mod tests { /// 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. + /// probe NULL is routed to a single partition, and OR-ing `IS NULL` into + /// that one partition's check 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<()> { diff --git a/datafusion/physical-plan/src/joins/hash_join/partitioned_hash_eval.rs b/datafusion/physical-plan/src/joins/hash_join/partitioned_hash_eval.rs index 6ac20238f056e..5c5fed370bc19 100644 --- a/datafusion/physical-plan/src/joins/hash_join/partitioned_hash_eval.rs +++ b/datafusion/physical-plan/src/joins/hash_join/partitioned_hash_eval.rs @@ -58,9 +58,8 @@ impl SeededRandomState { /// This expression computes the hash of join key columns using a specific RandomState. /// It returns a UInt64Array containing the hash values. /// -/// This is used for: -/// - Computing routing hashes (with RepartitionExec's 0,0,0,0 seeds) -/// - Computing lookup hashes (with HashJoin's 'J','O','I','N' seeds) +/// The hash join does not build this expression. It exists so that +/// `datafusion-proto` can still decode serialized plans that contain it. pub struct HashExpr { /// Columns to hash on_columns: Vec, @@ -352,8 +351,9 @@ impl HashTableLookupExpr { /// * `pruning_bitmap` - key-range summary covering every partition's keys /// for pruning only, or `None` /// - /// A single non-empty map is probed directly without routing, so it may be - /// any [`Map`] and `router` is ignored. Otherwise `router` is required. + /// When `maps` has exactly one entry, that map is probed directly without + /// routing and `router` is ignored. Otherwise `router` is required, even if + /// only one entry is non-empty. /// /// # Public Only for Internal Use: /// `datafusion-proto` tests require this constructor, but it is not part of 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 a1d698350b94c..70558a4bdec4a 100644 --- a/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs +++ b/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs @@ -222,7 +222,7 @@ fn create_bounds_predicate( /// Combines a membership predicate and a bounds predicate with logical AND. /// /// Returns `None` when neither is available; callers decide the fallback (e.g. -/// skip updating the filter vs. emit a `lit(true)` branch inside a CASE). +/// skip updating the filter or use `lit(true)`). fn combine_membership_and_bounds( membership_expr: Option>, bounds_expr: Option>, @@ -484,7 +484,7 @@ impl SharedBuildAccumulator { /// This unified method handles both CollectLeft and Partitioned modes. When all partitions /// have reported (barrier wait), the leader builds the appropriate filter expression: /// - CollectLeft: Simple conjunction of bounds and membership check - /// - Partitioned: CASE expression routing to per-partition filters + /// - Partitioned: one lookup that routes each probe row to its partition's keys /// /// # Arguments /// * `data` - Build data including hash map, pushdown strategy, and bounds @@ -730,11 +730,24 @@ impl SharedBuildAccumulator { } } - let filter_expr = if members.is_empty() { + let filter_expr = match members.as_slice() { // No build row exists, so no probe row can match. - lit(false) - } else { - self.merged_membership_filter(&members, merged_bounds, num_partitions)? + [] => lit(false), + // Only one build partition holds rows, so its own filter covers every + // probe match without routing, and an IN list keeps its pruning. + [(_, pushdown, bounds)] => combine_membership_and_bounds( + create_membership_predicate( + &self.on_right, + pushdown.clone(), + &HASH_JOIN_SEED, + self.probe_schema.as_ref(), + )?, + create_bounds_predicate(&self.on_right, bounds), + ) + .expect("membership predicate is always present"), + _ => { + self.merged_membership_filter(&members, merged_bounds, num_partitions)? + } }; self.dynamic_filter .update(self.preserve_probe_nulls(filter_expr, keys_have_null)?) @@ -1221,7 +1234,7 @@ mod tests { } #[test] - fn partitioned_hash_map_partitions_route_without_case() { + fn partitioned_hash_map_partitions_merge_into_one_lookup() { let acc = make_partitioned_expr_accumulator_for_test(3); acc.build_filter(FinalizeInput::Partitioned(vec![ @@ -1287,7 +1300,7 @@ mod tests { } /// Partitions that chose an IN list and partitions that chose a hash table - /// merge into one lookup without a `CASE`: a probe row passes exactly when + /// merge into one lookup: a probe row passes exactly when /// the partition its key routes to holds the key. #[test] #[cfg(not(feature = "force_hash_collisions"))] @@ -1359,7 +1372,7 @@ mod tests { } #[test] - fn partitioned_one_real_partition_with_rest_empty_skips_case() { + fn partitioned_one_real_partition_with_rest_empty_keeps_its_in_list() { let acc = make_partitioned_expr_accumulator_for_test(3); acc.build_filter(FinalizeInput::Partitioned(vec![ @@ -1373,6 +1386,58 @@ mod tests { in_list_expr(&expr); } + /// One non-empty partition keeps its own filter: no routing, and an IN list + /// stays an IN list even when its duplicate rows exceed the merged list limit. + #[test] + fn partitioned_one_non_empty_partition_keeps_its_own_filter() { + let empty = || reported(PushdownStrategy::Empty, no_bounds()); + let mut acc = make_partitioned_expr_accumulator_for_test(3); + acc.in_list_max_values = 150; + acc.build_filter(FinalizeInput::Partitioned(vec![ + empty(), + reported(in_list(&[2; 1000]), no_bounds()), + PartitionStatus::CanceledUnknown, + ])) + .unwrap(); + in_list_expr(¤t_expr(&acc)); + + let acc = make_partitioned_expr_accumulator_for_test(3); + acc.build_filter(FinalizeInput::Partitioned(vec![ + empty(), + reported(hash_map(), bounds(1, 7)), + PartitionStatus::CanceledUnknown, + ])) + .unwrap(); + let expr = format!("{:?}", current_expr(&acc)); + assert!(!expr.contains("routed by"), "{expr}"); + } + + /// A canceled partition rejects the rows routed to it, and only those: the + /// live partitions keep their matches. + #[cfg(not(feature = "force_hash_collisions"))] + #[test] + fn partitioned_canceled_partition_rejects_only_its_routed_rows() { + let acc = make_partitioned_expr_accumulator_for_test(4); + let keys: Vec = (0..40).collect(); + let mut partitions = + routed_partitions(&acc, 4, &keys, |_, held| hash_map_of(held)); + partitions[0] = PartitionStatus::CanceledUnknown; + acc.build_filter(FinalizeInput::Partitioned(partitions)) + .unwrap(); + + let array: ArrayRef = Arc::new(Int32Array::from(keys.clone())); + let mut hashes = vec![0; keys.len()]; + create_hashes( + [&array], + acc.repartition_random_state.random_state(), + &mut hashes, + ) + .unwrap(); + let expected: Vec = hashes.iter().map(|hash| *hash % 4 != 0).collect(); + assert!(expected.contains(&true) && expected.contains(&false)); + assert_matches(&acc, &keys, &expected); + } + #[test] fn partitioned_canceled_and_empty_partitions_reject_every_row() { let acc = make_partitioned_expr_accumulator_for_test(2);