From 84f5b49be213caddd48e88ea009a57e8cce78315 Mon Sep 17 00:00:00 2001 From: Liang-Chi Hsieh Date: Fri, 6 Nov 2026 00:37:52 +0000 Subject: [PATCH] fix: reject a multi-key null-aware join without a NOT IN value key count (#27111) - Closes #26569. leading keys that are `NOT IN` value keys (the rest are correlation scope keys), and defaulted it to `1` in `HashJoinExecBuilder::new`, `Join::try_new` and the `LogicalPlanBuilder` join methods. A caller that builds `(a, b) NOT IN (SELECT x, y ...)` itself and only sets `with_null_aware(true)` (for example an external adaptive planner) silently gets one value key, with `b` read as a scope key. The outer row `(1, NULL)` against the subquery row `(1, 5)` is then kept, although its `NOT IN` is UNKNOWN and it must be dropped. 55.x rejected such a plan ("null_aware anti join only supports single column join key"). DataFusion's own planning always sets the count, so SQL is not affected. - `null_aware_value_keys` stays a `usize`, and `0` now means unset. It is the new default of `Join::try_new`, the `LogicalPlanBuilder` join methods, `HashJoinExecBuilder::new` and `HashJoinExec::try_new`. - An unset count is accepted only for a single-key null-aware join, which is a scalar `NOT IN`. A null-aware join with more keys and no count fails to build, with an error asking for `with_null_aware_value_keys`. There is deliberately no "all keys are value keys" default: that is just as wrong for a correlated subquery. - The join type, partition mode and an empty `on` are checked before the count, so a join that no count can fix reports its own error. - Decorrelation sets the count only on a null-aware join, so every other join keeps it unset, as `Join::try_new` leaves it. - The logical and physical proto decoders drop their `.max(1)`, so an unset count round-trips as unset. A payload predating the field (55.x, which only allowed single-key null-aware joins) decodes it as `0` and still reads as a scalar `NOT IN`. The proto field types are unchanged; only their comments are updated. - The 56.0.0 upgrade guide section is shortened. - `test_null_aware_validation_value_keys` (`hash_join/exec.rs`): a two-key null-aware join with an unset or out-of-range count fails to build, `HashJoinExec::try_new` rejects a two-key null-aware join, a single-key join with an unset count builds as a scalar `NOT IN`, and a join that no count can fix (a two-key `Inner`, a `Partitioned` `RightAnti` or `LeftAnti`, an empty `on`) reports its own error. - `null_aware_value_keys_round_trip` (physical proto): an unset count is rejected on decode with two keys and kept unset with one. - `decorrelated_mark_join_sets_value_keys_only_when_null_aware` (core `sql/joins.rs`): scalar and multi-column `IN` mark joins on non-nullable keys keep the count unset, and their null-aware counterparts carry `1` and `2`. - Existing physical tests that built correlated scalar `NOT IN` joins through `HashJoinExec::try_new` and relied on the old default now state the count through a test helper. - Reverting each fix makes the matching test fail: the unset-multi-key check, the check order, the decoders' `.max(1)` and the decorrelation guard. Yes, for code that builds null-aware joins directly: a multi-key null-aware join must set `with_null_aware_value_keys`; it no longer silently defaults to one value key. A struct literal that sets the field to `1` for a non-null-aware join still works, but `0` is now the default. The field is new in 56, and the upgrade guide is updated. This pull request and its description were written by Isaac. --------- Co-authored-by: Isaac (cherry picked from commit 22aedf0a7155da8063ab313646f20e4723f7f827) --- datafusion/core/tests/sql/joins.rs | 72 ++++++ datafusion/expr/src/logical_plan/builder.rs | 2 +- datafusion/expr/src/logical_plan/plan.rs | 9 +- .../src/decorrelate_predicate_subquery.rs | 9 +- .../optimizer/src/eliminate_cross_join.rs | 6 +- .../physical-plan/src/joins/hash_join/exec.rs | 241 ++++++++++++------ .../src/joins/hash_join/not_in.rs | 60 +++-- .../proto-models/proto/datafusion.proto | 11 +- .../proto-models/src/generated/prost.rs | 11 +- datafusion/proto/src/logical_plan/mod.rs | 19 +- .../library-user-guide/upgrading/56.0.0.md | 40 ++- 11 files changed, 328 insertions(+), 152 deletions(-) diff --git a/datafusion/core/tests/sql/joins.rs b/datafusion/core/tests/sql/joins.rs index dd1f8d1379876..365343ffa079d 100644 --- a/datafusion/core/tests/sql/joins.rs +++ b/datafusion/core/tests/sql/joins.rs @@ -688,3 +688,75 @@ async fn asof_join_unparser_preserves_right_preselection() -> Result<()> { } Ok(()) } + +/// The `NOT IN` value key count of every decorrelated mark join in `sql`'s +/// optimized plan, with whether the join is null-aware. +async fn mark_join_value_keys( + ctx: &SessionContext, + sql: &str, +) -> Result> { + use datafusion::common::tree_node::{TreeNode, TreeNodeRecursion}; + use datafusion::logical_expr::{JoinType, LogicalPlan}; + + let plan = ctx.sql(sql).await?.into_optimized_plan()?; + let mut joins = vec![]; + plan.apply(|plan| { + if let LogicalPlan::Join(join) = plan + && join.join_type == JoinType::LeftMark + { + joins.push((join.null_aware, join.null_aware_value_keys)); + } + Ok(TreeNodeRecursion::Continue) + })?; + Ok(joins) +} + +/// Decorrelation sets the value key count only on a null-aware join, which is +/// the only one that reads it. Any other join keeps it unset (`0`), the same as a +/// join built through `Join::try_new` or `LogicalPlanBuilder`, so equal plans +/// compare equal however they were built. +#[tokio::test] +async fn decorrelated_mark_join_sets_value_keys_only_when_null_aware() -> Result<()> { + let ctx = SessionContext::new(); + for (name, nullable) in [("t", false), ("n", true)] { + let schema = Arc::new(Schema::new(vec![ + Field::new("a", DataType::Int32, nullable), + Field::new("b", DataType::Int32, nullable), + ])); + let batch = RecordBatch::try_new( + Arc::clone(&schema), + vec![ + Arc::new(Int32Array::from(vec![1, 2])), + Arc::new(Int32Array::from(vec![3, 4])), + ], + )?; + ctx.register_table( + name, + Arc::new(MemTable::try_new(schema, vec![vec![batch]])?), + )?; + } + + // Non-nullable keys: neither the scalar nor the multi-column `IN` is + // null-aware. + for sql in [ + "SELECT a IN (SELECT a FROM t AS s) FROM t", + "SELECT (a, b) IN (SELECT a, b FROM t AS s) FROM t", + ] { + assert_eq!( + mark_join_value_keys(&ctx, sql).await?, + [(false, 0)], + "{sql}" + ); + } + // Nullable keys: both are null-aware and state their count. + assert_eq!( + mark_join_value_keys(&ctx, "SELECT a IN (SELECT a FROM n AS s) FROM n").await?, + [(true, 1)] + ); + assert_eq!( + mark_join_value_keys(&ctx, "SELECT (a, b) IN (SELECT a, b FROM n AS s) FROM n") + .await?, + [(true, 2)] + ); + Ok(()) +} diff --git a/datafusion/expr/src/logical_plan/builder.rs b/datafusion/expr/src/logical_plan/builder.rs index c9575e7a556fa..ab0fef512869a 100644 --- a/datafusion/expr/src/logical_plan/builder.rs +++ b/datafusion/expr/src/logical_plan/builder.rs @@ -1440,7 +1440,7 @@ impl LogicalPlanBuilder { schema: DFSchemaRef::new(join_schema), null_equality, null_aware, - null_aware_value_keys: 1, + null_aware_value_keys: 0, }))) } diff --git a/datafusion/expr/src/logical_plan/plan.rs b/datafusion/expr/src/logical_plan/plan.rs index 0fac69e7a0ec4..226a67fd5a88c 100644 --- a/datafusion/expr/src/logical_plan/plan.rs +++ b/datafusion/expr/src/logical_plan/plan.rs @@ -4650,6 +4650,11 @@ pub struct Join { /// after them are correlation scope keys of a correlated subquery. A NULL /// in a value key follows SQL three-valued logic across the whole tuple, /// while the scope keys only select which subquery rows are compared. + /// + /// `0` leaves the count unset, which is the default of [`Self::try_new`] + /// and the `LogicalPlanBuilder` join methods. Physical planning accepts an + /// unset count only for a single-key join, a scalar `NOT IN`, and rejects + /// it for more keys: no default can tell which of them are value keys. pub null_aware_value_keys: usize, } @@ -4878,7 +4883,7 @@ impl Join { schema: Arc::new(join_schema), null_equality, null_aware, - null_aware_value_keys: 1, + null_aware_value_keys: 0, }) } @@ -7228,7 +7233,7 @@ mod tests { schema: Arc::new(left_schema.join(&right_schema)?), null_equality: NullEquality::NullEqualsNothing, null_aware: false, - null_aware_value_keys: 1, + null_aware_value_keys: 0, })) } diff --git a/datafusion/optimizer/src/decorrelate_predicate_subquery.rs b/datafusion/optimizer/src/decorrelate_predicate_subquery.rs index 95fcdac1fe2d1..00d89f4e4b37e 100644 --- a/datafusion/optimizer/src/decorrelate_predicate_subquery.rs +++ b/datafusion/optimizer/src/decorrelate_predicate_subquery.rs @@ -738,7 +738,11 @@ fn subquery_output_expr(subquery: &LogicalPlan, i: usize) -> Result { } /// Sets [`Join::null_aware_value_keys`](datafusion_expr::Join::null_aware_value_keys) -/// on a join built by [`build_join`]. +/// on a null-aware join built by [`build_join`]. Only a null-aware join reads +/// the count, so any other join keeps it unset, as [`Join::try_new`] leaves +/// it, and equal joins compare equal however they were built. +/// +/// [`Join::try_new`]: datafusion_expr::Join::try_new fn with_null_aware_value_keys( plan: LogicalPlan, null_aware_value_keys: usize, @@ -746,6 +750,9 @@ fn with_null_aware_value_keys( let LogicalPlan::Join(join) = plan else { return internal_err!("expected a join, got {}", plan.display()); }; + if !join.null_aware { + return Ok(LogicalPlan::Join(join)); + } Ok(LogicalPlan::Join( join.with_null_aware_value_keys(null_aware_value_keys), )) diff --git a/datafusion/optimizer/src/eliminate_cross_join.rs b/datafusion/optimizer/src/eliminate_cross_join.rs index 9bccdfaac4d15..68497d2cff044 100644 --- a/datafusion/optimizer/src/eliminate_cross_join.rs +++ b/datafusion/optimizer/src/eliminate_cross_join.rs @@ -345,7 +345,7 @@ fn find_inner_join( schema: join_schema, null_equality, null_aware: false, - null_aware_value_keys: 1, + null_aware_value_keys: 0, })); } } @@ -369,7 +369,7 @@ fn find_inner_join( join_constraint: JoinConstraint::On, null_equality, null_aware: false, - null_aware_value_keys: 1, + null_aware_value_keys: 0, })) } @@ -1443,7 +1443,7 @@ mod tests { schema: join_schema, null_equality: NullEquality::NullEqualsNull, // Test preservation null_aware: false, - null_aware_value_keys: 1, + null_aware_value_keys: 0, }); // Apply filter that can create join conditions diff --git a/datafusion/physical-plan/src/joins/hash_join/exec.rs b/datafusion/physical-plan/src/joins/hash_join/exec.rs index 20a7def8f3f80..2ec44a7e13e6d 100644 --- a/datafusion/physical-plan/src/joins/hash_join/exec.rs +++ b/datafusion/physical-plan/src/joins/hash_join/exec.rs @@ -347,7 +347,7 @@ impl HashJoinExecBuilder { column_indices: vec![], null_equality: NullEquality::NullEqualsNothing, null_aware: false, - null_aware_value_keys: 1, + null_aware_value_keys: 0, dynamic_filter: None, // Will be computed at when plan will be built. cache: stub_properties(), @@ -485,6 +485,10 @@ impl HashJoinExecBuilder { filter.build_accumulator = OnceLock::new(); } } + // Before `not_in`, whose value key checks assume some keys. + if exec.on.is_empty() { + return plan_err!("On constraints in HashJoinExec should be non-empty"); + } // Validate null_aware flag exec.not_in()?; @@ -517,9 +521,6 @@ impl HashJoinExecBuilder { let left_schema = left.schema(); let right_schema = right.schema(); - if on.is_empty() { - return plan_err!("On constraints in HashJoinExec should be non-empty"); - } check_join_is_valid(&left_schema, &right_schema, &on)?; let (join_schema, column_indices) = @@ -861,6 +862,11 @@ pub struct HashJoinExec { /// without correlation keys every NULL-valued row is paired with every row /// on the other side to check its other elements. See /// for the follow-up. + /// + /// `0` leaves the count unset, the default of [`HashJoinExecBuilder::new`] + /// and [`Self::try_new`]. An unset count is accepted only for a single-key + /// join, a scalar `NOT IN`; a null-aware join with more keys must set it + /// with [`HashJoinExecBuilder::with_null_aware_value_keys`]. pub null_aware_value_keys: usize, /// Cache holding plan properties like equivalences, output partitioning etc. cache: Arc, @@ -1422,7 +1428,7 @@ impl DisplayAs for HashJoinExec { let display_null_aware = match (self.null_aware, self.null_aware_value_keys) { (false, _) => String::new(), - (true, 1) => ", null_aware".to_string(), + (true, 0 | 1) => ", null_aware".to_string(), (true, n) => format!(", null_aware(value_keys={n})"), }; let display_prepared = self @@ -1476,7 +1482,7 @@ impl DisplayAs for HashJoinExec { } if self.null_aware { - if self.null_aware_value_keys == 1 { + if self.null_aware_value_keys <= 1 { writeln!(f, "null_aware")?; } else { writeln!( @@ -2287,16 +2293,14 @@ impl HashJoinExec { .with_partition_mode(partition_mode) .with_null_equality(null_equality) .with_null_aware(*null_aware) - // Messages predating the field decode it as 0; they can only hold - // scalar `NOT IN` joins. - .with_null_aware_value_keys( - usize_from_wire( - u64::from(*null_aware_value_keys), - "HashJoinExec", - "null_aware_value_keys", - )? - .max(1), - ) + // `0` is an unset count, which `build` accepts only for a single + // key. Messages predating the field decode it as 0; they can only + // hold scalar `NOT IN` joins. + .with_null_aware_value_keys(usize_from_wire( + u64::from(*null_aware_value_keys), + "HashJoinExec", + "null_aware_value_keys", + )?) .with_fetch(fetch) .build()?; @@ -2592,8 +2596,9 @@ mod proto_tests { } /// A multi-column `NOT IN` keeps its value key count through the round - /// trip; a node written before the field existed decodes it as 0, which - /// is read as a scalar `NOT IN`. + /// trip. A count of 0, which is also how a node written before the field + /// existed decodes it, stays unset: accepted for a single key, rejected + /// for more. #[test] fn null_aware_value_keys_round_trip() { let on: Vec<(PhysicalExprRef, PhysicalExprRef)> = vec![ @@ -2621,7 +2626,17 @@ mod proto_tests { assert_eq!(as_join(&decode(node.clone())).null_aware_value_keys, 2); node.null_aware_value_keys = 0; - assert_eq!(as_join(&decode(node)).null_aware_value_keys, 1); + let decoder = StubPlanDecoder::ok(); + let ctx = ExecutionPlanDecodeCtx::new(&decoder); + let err = + HashJoinExec::try_from_proto(&join_node(node.clone()), &ctx).unwrap_err(); + assert!( + err.to_string().contains("with_null_aware_value_keys"), + "{err}" + ); + + node.on.truncate(1); + assert_eq!(as_join(&decode(node)).null_aware_value_keys, 0); } #[test] @@ -10092,6 +10107,25 @@ mod tests { ); } + /// A null-aware join for a scalar `NOT IN` on `on[0]`, correlated by any + /// keys after it: it states one `NOT IN` value key, which a multi-key + /// join must. + fn try_new_scalar_not_in( + left: Arc, + right: Arc, + on: JoinOn, + filter: Option, + join_type: JoinType, + partition_mode: PartitionMode, + ) -> Result { + HashJoinExecBuilder::new(left, right, on, join_type) + .with_filter(filter) + .with_partition_mode(partition_mode) + .with_null_aware(true) + .with_null_aware_value_keys(1) + .build() + } + /// Test that null_aware validation rejects PartitionMode::Partitioned (must be CollectLeft). #[tokio::test] async fn test_null_aware_validation_partitioned() { @@ -10110,16 +10144,13 @@ mod tests { ]; // A correlated (multi-key) LeftMark requires CollectLeft. - let result = HashJoinExec::try_new( + let result = try_new_scalar_not_in( left, right, on, None, - &JoinType::LeftMark, - None, + JoinType::LeftMark, PartitionMode::Partitioned, - NullEquality::NullEqualsNothing, - true, // null_aware = true (invalid with Partitioned) ); assert!(result.is_err()); @@ -10157,16 +10188,13 @@ mod tests { if multi_key { on.push(key("dummy", "dummy")?); } - let join = HashJoinExec::try_new( + let join = try_new_scalar_not_in( Arc::clone(&left), Arc::clone(&right), on, with_filter.then(prepare_join_filter), - &join_type, - None, + join_type, PartitionMode::CollectLeft, - NullEquality::NullEqualsNothing, - true, )?; let scope = NotInScope::Correlated { value_keys: 1 }; let expected = match join_type { @@ -10480,16 +10508,13 @@ mod tests { ), ]; - let join = HashJoinExec::try_new( + let join = try_new_scalar_not_in( left, right, on, None, - &JoinType::LeftMark, - None, + JoinType::LeftMark, PartitionMode::CollectLeft, - NullEquality::NullEqualsNothing, - true, )?; let stream = join.execute(0, task_ctx)?; @@ -10548,16 +10573,13 @@ mod tests { ), ]; - let join = HashJoinExec::try_new( + let join = try_new_scalar_not_in( left, right, on, None, - &JoinType::LeftMark, - None, + JoinType::LeftMark, PartitionMode::CollectLeft, - NullEquality::NullEqualsNothing, - true, )?; let stream = join.execute(0, task_ctx)?; @@ -10727,35 +10749,103 @@ mod tests { Ok(()) } - /// The value key count must name at least one key and no more than the - /// join has. - #[tokio::test] - async fn test_null_aware_validation_value_keys() -> Result<()> { - for value_keys in [0, 3] { - let left = build_table_two_cols(("a", &vec![Some(1)]), ("b", &vec![Some(2)])); - let right = - build_table_two_cols(("x", &vec![Some(1)]), ("y", &vec![Some(2)])); - let on = vec![ - ( - Arc::new(Column::new_with_schema("a", &left.schema())?) as _, - Arc::new(Column::new_with_schema("x", &right.schema())?) as _, - ), - ( - Arc::new(Column::new_with_schema("b", &left.schema())?) as _, - Arc::new(Column::new_with_schema("y", &right.schema())?) as _, - ), - ]; - let err = HashJoinExecBuilder::new(left, right, on, JoinType::LeftAnti) - .with_partition_mode(PartitionMode::CollectLeft) + /// The value key count must name no more keys than the join has. An + /// unset count (`0`) is a scalar `NOT IN` with a single key, and is + /// rejected with more keys rather than guessing which of them are `NOT IN` + /// value keys and which are correlation scope keys. + #[test] + fn test_null_aware_validation_value_keys() -> Result<()> { + let left = build_table_two_cols(("a", &vec![Some(1)]), ("b", &vec![None])); + let right = build_table_two_cols(("x", &vec![Some(1)]), ("y", &vec![Some(5)])); + let key = |l: &str, r: &str| -> Result<(PhysicalExprRef, PhysicalExprRef)> { + Ok(( + Arc::new(Column::new_with_schema(l, &left.schema())?) as _, + Arc::new(Column::new_with_schema(r, &right.schema())?) as _, + )) + }; + let two_keys = vec![key("a", "x")?, key("b", "y")?]; + let build = |on: JoinOn, join_type, partition_mode, value_keys| { + HashJoinExecBuilder::new(Arc::clone(&left), Arc::clone(&right), on, join_type) + .with_partition_mode(partition_mode) .with_null_aware(true) .with_null_aware_value_keys(value_keys) .build() - .unwrap_err(); - assert!( - err.to_string() - .contains("needs between 1 and 2 `NOT IN` value keys"), - "{err}" - ); + }; + + for (value_keys, expected) in [ + ( + 0, + "join with 2 keys needs the number of its `NOT IN` value keys, set it with `with_null_aware_value_keys`", + ), + (3, "needs between 1 and 2 `NOT IN` value keys"), + ] { + let err = build( + two_keys.clone(), + JoinType::LeftAnti, + PartitionMode::CollectLeft, + value_keys, + ) + .unwrap_err(); + assert!(err.to_string().contains(expected), "{err}"); + } + + // `HashJoinExec::try_new` cannot set the count, so it only builds + // single-key null-aware joins, as before value keys existed. + let err = HashJoinExec::try_new( + Arc::clone(&left), + Arc::clone(&right), + two_keys.clone(), + None, + &JoinType::LeftAnti, + None, + PartitionMode::CollectLeft, + NullEquality::NullEqualsNothing, + true, + ) + .unwrap_err(); + assert!( + err.to_string().contains("with_null_aware_value_keys"), + "{err}" + ); + + let join = build( + vec![key("a", "x")?], + JoinType::LeftAnti, + PartitionMode::CollectLeft, + 0, + )?; + assert_eq!(join.not_in()?.map(NotIn::value_keys), Some(1)); + + // A join that no count can make valid reports what is wrong with it, + // not the missing count. + for (on, join_type, partition_mode, expected) in [ + ( + two_keys.clone(), + JoinType::Inner, + PartitionMode::CollectLeft, + "null_aware can only be true for LeftAnti joins", + ), + ( + two_keys.clone(), + JoinType::RightAnti, + PartitionMode::Partitioned, + "null_aware can only be true for LeftAnti joins", + ), + ( + two_keys, + JoinType::LeftAnti, + PartitionMode::Partitioned, + "null_aware joins require PartitionMode::CollectLeft", + ), + ( + vec![], + JoinType::LeftAnti, + PartitionMode::CollectLeft, + "On constraints in HashJoinExec should be non-empty", + ), + ] { + let err = build(on, join_type, partition_mode, 0).unwrap_err(); + assert!(err.to_string().contains(expected), "{join_type}: {err}"); } Ok(()) } @@ -10874,16 +10964,13 @@ mod tests { // The outer side `left` as the build side (`LeftAnti`), and swapped to // the probe side (`RightAnti`), must keep the same rows. - let left_anti = HashJoinExec::try_new( + let left_anti = try_new_scalar_not_in( Arc::clone(&left), Arc::clone(&right), on.clone(), None, - &JoinType::LeftAnti, - None, + JoinType::LeftAnti, PartitionMode::CollectLeft, - NullEquality::NullEqualsNothing, - true, )?; let right_anti = swap_for_separate_execution(&left_anti)?; let batches = @@ -10976,16 +11063,13 @@ mod tests { )) }) .collect::>()?; - let join = HashJoinExec::try_new( + let join = try_new_scalar_not_in( left, right, on, None, - &join_type, - None, + join_type, PartitionMode::CollectLeft, - NullEquality::NullEqualsNothing, - true, )?; let batches = common::collect(join.execute(0, task_ctx)?).await?; results.push(batches_to_sort_string(&batches)); @@ -11069,16 +11153,13 @@ mod tests { }) .collect::>()?; - let join = HashJoinExec::try_new( + let join = try_new_scalar_not_in( left, right, on, None, - &JoinType::LeftAnti, - None, + JoinType::LeftAnti, PartitionMode::CollectLeft, - NullEquality::NullEqualsNothing, - true, )?; let stream = join.execute(0, task_ctx)?; diff --git a/datafusion/physical-plan/src/joins/hash_join/not_in.rs b/datafusion/physical-plan/src/joins/hash_join/not_in.rs index 1a810cdd0f243..120e9faec05a2 100644 --- a/datafusion/physical-plan/src/joins/hash_join/not_in.rs +++ b/datafusion/physical-plan/src/joins/hash_join/not_in.rs @@ -100,30 +100,14 @@ impl NotIn { num_value_keys: usize, has_filter: bool, ) -> Result { - if num_value_keys == 0 || num_value_keys > num_keys { - return plan_err!( - "null_aware {join_type} join needs between 1 and {num_keys} `NOT IN` value keys, got {num_value_keys}" - ); - } - // `num_keys > 1` also covers a multi-column value key without any - // correlation: a NULL in one tuple element leaves the comparison FALSE - // whenever another element is a definite mismatch, so it too must be - // decided per row. - let scope = if num_keys > 1 || has_filter { - NotInScope::Correlated { - value_keys: num_value_keys, - } - } else { - NotInScope::Uncorrelated - }; + // The join type and mode come first: no value key count can make up + // for them. `None` is `RightAnti`, which emits probe rows. let output = match (join_type, partition_mode) { - (JoinType::LeftAnti, _) => NotInOutput::Rows, - (JoinType::LeftMark, _) => NotInOutput::Mark, + (JoinType::LeftAnti, _) => Some(NotInOutput::Rows), + (JoinType::LeftMark, _) => Some(NotInOutput::Mark), // `PartitionMode::CollectLeft` is safe because `RightAnti` is // probe-driven: every probe row sees the whole build side. - (JoinType::RightAnti, PartitionMode::CollectLeft) => { - return Ok(Self::ProbeRows { scope }); - } + (JoinType::RightAnti, PartitionMode::CollectLeft) => None, _ => { return plan_err!( "null_aware can only be true for LeftAnti joins and RightAnti joins with `CollectLeft` `PartitionMode`, or LeftMark joins, got {join_type} with {partition_mode}" @@ -140,7 +124,39 @@ impl NotIn { "null_aware joins require PartitionMode::CollectLeft, got PartitionMode::Partitioned" ); } - Ok(Self::BuildRows { scope, output }) + // `0` is an unset count, which is only unambiguous with a single key: + // a scalar `NOT IN`. With more keys, any default would silently pick + // which of them are `NOT IN` value keys and which are correlation + // scope keys. + let num_value_keys = match num_value_keys { + 0 if num_keys == 1 => 1, + 0 => { + return plan_err!( + "null_aware {join_type} join with {num_keys} keys needs the number of its `NOT IN` value keys, set it with `with_null_aware_value_keys`" + ); + } + n if n > num_keys => { + return plan_err!( + "null_aware {join_type} join needs between 1 and {num_keys} `NOT IN` value keys, got {n}" + ); + } + n => n, + }; + // `num_keys > 1` also covers a multi-column value key without any + // correlation: a NULL in one tuple element leaves the comparison FALSE + // whenever another element is a definite mismatch, so it too must be + // decided per row. + let scope = if num_keys > 1 || has_filter { + NotInScope::Correlated { + value_keys: num_value_keys, + } + } else { + NotInScope::Uncorrelated + }; + Ok(match output { + Some(output) => Self::BuildRows { scope, output }, + None => Self::ProbeRows { scope }, + }) } fn scope(self) -> NotInScope { diff --git a/datafusion/proto-models/proto/datafusion.proto b/datafusion/proto-models/proto/datafusion.proto index 6c4ef8f0a4b2f..df39cccca2371 100644 --- a/datafusion/proto-models/proto/datafusion.proto +++ b/datafusion/proto-models/proto/datafusion.proto @@ -289,8 +289,9 @@ message JoinNode { LogicalExprNode filter = 8; bool null_aware = 9; // Number of `NOT IN` value keys of a null-aware join (see - // `Join::null_aware_value_keys`). Messages predating this field decode it - // as 0, which is read as 1: they can only hold scalar `NOT IN` joins. + // `Join::null_aware_value_keys`). 0 is an unset count, accepted only for a + // single key. Messages predating this field decode it as 0: they can only + // hold scalar `NOT IN` joins. uint32 null_aware_value_keys = 10; } @@ -1479,9 +1480,9 @@ message HashJoinExecNode { // `None`, which is the correct reading of an older message. optional uint64 fetch = 12; // Number of `NOT IN` value keys of a null-aware join (see - // `HashJoinExec::null_aware_value_keys`). Messages predating this field - // decode it as 0, which is read as 1: they can only hold scalar `NOT IN` - // joins. + // `HashJoinExec::null_aware_value_keys`). 0 is an unset count, accepted + // only for a single key. Messages predating this field decode it as 0: + // they can only hold scalar `NOT IN` joins. uint32 null_aware_value_keys = 13; } diff --git a/datafusion/proto-models/src/generated/prost.rs b/datafusion/proto-models/src/generated/prost.rs index 1034d0443dbce..bb0da62d84c71 100644 --- a/datafusion/proto-models/src/generated/prost.rs +++ b/datafusion/proto-models/src/generated/prost.rs @@ -442,8 +442,9 @@ pub struct JoinNode { #[prost(bool, tag = "9")] pub null_aware: bool, /// Number of `NOT IN` value keys of a null-aware join (see - /// `Join::null_aware_value_keys`). Messages predating this field decode it - /// as 0, which is read as 1: they can only hold scalar `NOT IN` joins. + /// `Join::null_aware_value_keys`). 0 is an unset count, accepted only for a + /// single key. Messages predating this field decode it as 0: they can only + /// hold scalar `NOT IN` joins. #[prost(uint32, tag = "10")] pub null_aware_value_keys: u32, } @@ -2235,9 +2236,9 @@ pub struct HashJoinExecNode { #[prost(uint64, optional, tag = "12")] pub fetch: ::core::option::Option, /// Number of `NOT IN` value keys of a null-aware join (see - /// `HashJoinExec::null_aware_value_keys`). Messages predating this field - /// decode it as 0, which is read as 1: they can only hold scalar `NOT IN` - /// joins. + /// `HashJoinExec::null_aware_value_keys`). 0 is an unset count, accepted + /// only for a single key. Messages predating this field decode it as 0: + /// they can only hold scalar `NOT IN` joins. #[prost(uint32, tag = "13")] pub null_aware_value_keys: u32, } diff --git a/datafusion/proto/src/logical_plan/mod.rs b/datafusion/proto/src/logical_plan/mod.rs index 592d56cb441cd..bd4f100c283f1 100644 --- a/datafusion/proto/src/logical_plan/mod.rs +++ b/datafusion/proto/src/logical_plan/mod.rs @@ -1068,16 +1068,15 @@ impl AsLogicalPlan for LogicalPlanNode { NullEquality::from(null_equality), join.null_aware, )? - // Plans encoded before the field existed decode it as 0; they - // could only hold scalar `NOT IN` joins. - .with_null_aware_value_keys( - usize_from_wire( - join.null_aware_value_keys, - "JoinNode", - "null_aware_value_keys", - )? - .max(1), - ), + // `0` is an unset count, which physical planning accepts + // only for a single key. Plans encoded before the field + // existed decode it as 0; they could only hold scalar + // `NOT IN` joins. + .with_null_aware_value_keys(usize_from_wire( + join.null_aware_value_keys, + "JoinNode", + "null_aware_value_keys", + )?), )) } LogicalPlanType::AsOfJoin(join) => { diff --git a/docs/source/library-user-guide/upgrading/56.0.0.md b/docs/source/library-user-guide/upgrading/56.0.0.md index 413933018a496..b05a5e4f3356f 100644 --- a/docs/source/library-user-guide/upgrading/56.0.0.md +++ b/docs/source/library-user-guide/upgrading/56.0.0.md @@ -1076,34 +1076,28 @@ Wire compatibility is directional: ### Null-aware joins record their `NOT IN` value key count Multi-column `(a, b) NOT IN (SELECT x, y ...)` is now planned as a null-aware -join. Its first equi-join keys are the tuple elements, followed by any -correlation keys, so a null-aware join now records how many of its leading keys -are `NOT IN` value keys: +join, which records how many of its leading equi-join keys are `NOT IN` value +keys; the keys after them are correlation scope keys. `Join` and +`HashJoinExec` have a new `null_aware_value_keys: usize` field, set with +`with_null_aware_value_keys`, and so do the generated protobuf `JoinNode` and +`HashJoinExecNode` types. -- `datafusion_expr::logical_plan::Join` has a new `null_aware_value_keys: usize` - field. -- `HashJoinExec` has a new `null_aware_value_keys: usize` field, set with - `HashJoinExecBuilder::with_null_aware_value_keys`. - -The count is only read when `null_aware` is set. `Join::try_new`, the -`LogicalPlanBuilder` join methods and `HashJoinExec::try_new` set it to `1`, -the scalar `NOT IN` of earlier releases. Code constructing `Join` with a struct -literal must initialize the field: +The count defaults to `0`, unset. An unset count is accepted only for a +single-key null-aware join, the scalar `NOT IN` of earlier releases, so +existing code and 55.x plans keep working. Code that builds a multi-key +null-aware join itself must set the count, or the join fails to plan: ```rust,ignore -let join = Join { - // ... - null_aware, - null_aware_value_keys: 1, -}; +let join = HashJoinExecBuilder::new(left, right, on, JoinType::LeftAnti) + .with_partition_mode(PartitionMode::CollectLeft) + .with_null_aware(true) + // `on[..2]` are the `NOT IN` value keys, any later keys scope keys. + .with_null_aware_value_keys(2) + .build()?; ``` -The generated protobuf `JoinNode` and `HashJoinExecNode` types have a new -`null_aware_value_keys` field, which struct literals must provide too. A 56.0 -reader decodes a payload without the field (value `0`) as a scalar `NOT IN`. A -55.0 reader must not consume a 56.0 plan holding a multi-column `NOT IN`: it -ignores the unknown field and would read the extra tuple elements as -correlation keys, returning wrong results. +A 55.x reader ignores the field and rejects a multi-key null-aware join from a +56.0 plan. ### `FunctionalDependence` has a new `null_equality` field