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/09/01 14:38:55 UTC
(datafusion) branch gh-readonly-queue/main/pr-21585-221523813a5fa7a371fd3fe4b925ee0a07707d20 created (now bed9dcd548)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-21585-221523813a5fa7a371fd3fe4b925ee0a07707d20
in repository https://gitbox.apache.org/repos/asf/datafusion.git
at bed9dcd548 feat: Support null aware hash mark-joins (#21585)
This branch includes the following new commits:
new bed9dcd548 feat: Support null aware hash mark-joins (#21585)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org
(datafusion) 01/01: feat: Support null aware hash mark-joins (#21585)
Posted by gi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-21585-221523813a5fa7a371fd3fe4b925ee0a07707d20
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit bed9dcd54870289246a5442fc5ffafbc5f1de8cb
Author: Adam Gutglick <ad...@gmail.com>
AuthorDate: Tue Sep 1 14:38:34 2026 +0000
feat: Support null aware hash mark-joins (#21585)
## Which issue does this PR close?
- Closes #21309.
## Rationale for this change
This change is about correctness/sql completeness, but is also a step
towards better subquery de-correlation.
Before this change, mark joins only produced boolean `true` / `false`
results, so queries such as `(id NOT IN (...)) IS NULL` could return
incorrect results, especially for correlated scalar subqueries.
## What changes are included in this PR?
1. Adds support for null-aware mark joins.
2. Make sure queries that joins that require null awareness go through a
join implementation that supports that.
3. Extends decorrelation so scalar `NOT IN` mark joins use null-aware
semantics when the predicate can be represented as hash join keys.
I pulled tests from existing tests, a DuckDB issue I ran into, the
[Story of Joins (in
HyPer)](https://www.cs.cmu.edu/~15721-f24/papers/Story_of_Joins.pdf)
paper and some AI generated cases.
## Are these changes tested?
1. Existing SLT tests that explicitly showed bad results.
2. New dedicated SLT tests.
5. New unit tests.
## Are there any user-facing changes?
This PR changes planning behavior and introduces more public API around
hash joins, the only new functionality is public function I could find
is the new helper function `build_join_schema_with_null_aware`
## AI Usage
*AI was used in the process of developing this PR, mostly around testing
and planning*
---------
Signed-off-by: Adam Gutglick <ad...@gmail.com>
---
datafusion/common/src/join_type.rs | 7 +-
datafusion/core/src/physical_planner.rs | 90 +++-
datafusion/expr/src/logical_plan/builder.rs | 4 +-
datafusion/expr/src/logical_plan/plan.rs | 16 +-
.../src/decorrelate_predicate_subquery.rs | 133 ++++-
.../optimizer/src/extract_equijoin_predicate.rs | 2 +-
.../physical-plan/src/joins/hash_join/exec.rs | 422 +++++++++++++--
.../src/joins/hash_join/shared_bounds.rs | 87 ++-
.../physical-plan/src/joins/hash_join/stream.rs | 255 ++++++++-
.../physical-plan/src/joins/symmetric_hash_join.rs | 2 +
datafusion/physical-plan/src/joins/utils.rs | 77 ++-
.../test_files/null_aware_anti_join.slt | 11 -
.../test_files/null_aware_mark_join.slt | 583 +++++++++++++++++++++
13 files changed, 1584 insertions(+), 105 deletions(-)
diff --git a/datafusion/common/src/join_type.rs b/datafusion/common/src/join_type.rs
index c77a1475ed..de4e4dc84b 100644
--- a/datafusion/common/src/join_type.rs
+++ b/datafusion/common/src/join_type.rs
@@ -61,9 +61,10 @@ pub enum JoinType {
/// [1]. This join type is used to decorrelate EXISTS subqueries used inside disjunctive
/// predicates.
///
- /// Note: This we currently do not implement the full null semantics for the mark join described
- /// in [1] which will be needed if we and ANY subqueries. In our version the mark column will
- /// only be true for had a match and false when no match was found, never null.
+ /// For scalar `NOT IN`, DataFusion can plan a null-aware hash mark join where the
+ /// mark column is nullable: TRUE for a match, NULL for SQL UNKNOWN, and FALSE
+ /// otherwise. Row-valued multi-column `NOT IN` and non-hash residual predicate
+ /// null-aware mark semantics are not implemented.
///
/// [1]: http://btw2017.informatik.uni-stuttgart.de/slidesandpapers/F1-10-37/paper_web.pdf
LeftMark,
diff --git a/datafusion/core/src/physical_planner.rs b/datafusion/core/src/physical_planner.rs
index 3bcbc4d43c..9dde89fb1c 100644
--- a/datafusion/core/src/physical_planner.rs
+++ b/datafusion/core/src/physical_planner.rs
@@ -1592,6 +1592,12 @@ impl DefaultPhysicalPlanner {
let prefer_hash_join =
session_state.config_options().optimizer.prefer_hash_join;
+ // Null-aware joins are pinned to CollectLeft hash joins (see
+ // `HashJoinExec::null_aware`): never repartition them, and
+ // never route them to the sort-merge path below.
+ let can_repartition_join = session_state.config().target_partitions() > 1
+ && session_state.config().repartition_joins()
+ && !*null_aware;
// TODO: Allow PWMJ to deal with residual equijoin conditions
let join: Arc<dyn ExecutionPlan> = if join_on.is_empty() {
@@ -1744,24 +1750,15 @@ impl DefaultPhysicalPlanner {
vec![SortOptions::default(); join_on_len],
*null_equality,
)?)
- } else if session_state.config().target_partitions() > 1
- && session_state.config().repartition_joins()
- && prefer_hash_join
- && !*null_aware
- // Null-aware joins must use CollectLeft
- {
- Arc::new(HashJoinExec::try_new(
- physical_left,
- physical_right,
- join_on,
- join_filter,
- join_type,
- None,
- PartitionMode::Auto,
- *null_equality,
- *null_aware,
- )?)
} else {
+ // Null-aware joins need global probe-side state, so keep
+ // them in CollectLeft mode.
+ let partition_mode = if can_repartition_join {
+ PartitionMode::Auto
+ } else {
+ PartitionMode::CollectLeft
+ };
+
Arc::new(HashJoinExec::try_new(
physical_left,
physical_right,
@@ -1769,7 +1766,7 @@ impl DefaultPhysicalPlanner {
join_filter,
join_type,
None,
- PartitionMode::CollectLeft,
+ partition_mode,
*null_equality,
*null_aware,
)?)
@@ -3991,6 +3988,63 @@ mod tests {
Ok(())
}
+ #[tokio::test]
+ async fn correlated_not_in_is_null_uses_null_aware_hash_mark_join() -> Result<()> {
+ let query = "
+ SELECT value
+ FROM (
+ VALUES
+ (1, 1, 'a'),
+ (3, 1, 'b'),
+ (1, 2, 'c'),
+ (NULL, 1, 'd'),
+ (5, 3, 'e'),
+ (2, 1, 'f'),
+ (NULL, 2, 'g')
+ ) AS outer_corr_table(id, grp, value)
+ WHERE (id NOT IN (
+ SELECT id
+ FROM (
+ VALUES
+ (2, 1),
+ (NULL, 1),
+ (1, 2)
+ ) AS inner_corr_table(id, grp)
+ WHERE inner_corr_table.grp = outer_corr_table.grp
+ )) IS NULL
+ ORDER BY value";
+
+ let config = SessionConfig::new()
+ .with_target_partitions(4)
+ .set_bool("datafusion.optimizer.prefer_hash_join", false);
+ let ctx = SessionContext::new_with_config(config);
+
+ let plan = ctx.sql(query).await?.create_physical_plan().await?;
+ let formatted = displayable(plan.as_ref()).indent(true).to_string();
+ assert_contains!(
+ &formatted,
+ "HashJoinExec: mode=CollectLeft, join_type=LeftMark"
+ );
+ assert!(!formatted.contains("SortMergeJoinExec"), "{formatted}");
+
+ let batches = ctx.sql(query).await?.collect().await?;
+ assert_batches_eq!(
+ &[
+ "+-------+",
+ "| value |",
+ "+-------+",
+ "| a |",
+ "| b |",
+ "| d |",
+ "| g |",
+ "+-------+",
+ ],
+ &batches
+ );
+
+ Ok(())
+ }
+
#[tokio::test]
async fn scalar_subquery_in_projection_and_filter_plans() -> Result<()> {
let plan = plan_sql(
diff --git a/datafusion/expr/src/logical_plan/builder.rs b/datafusion/expr/src/logical_plan/builder.rs
index ef5e496b0d..ceb2c124e2 100644
--- a/datafusion/expr/src/logical_plan/builder.rs
+++ b/datafusion/expr/src/logical_plan/builder.rs
@@ -1666,7 +1666,9 @@ fn mark_field(schema: &DFSchema) -> (Option<TableReference>, Arc<Field>) {
(
table_reference,
- Arc::new(Field::new("mark", DataType::Boolean, false)),
+ // Nullable: null-aware `LeftMark` joins use NULL to represent SQL
+ // UNKNOWN for `NOT IN`. Other mark joins simply never produce a NULL.
+ Arc::new(Field::new("mark", DataType::Boolean, true)),
)
}
diff --git a/datafusion/expr/src/logical_plan/plan.rs b/datafusion/expr/src/logical_plan/plan.rs
index 2c909b6534..4e039e3e77 100644
--- a/datafusion/expr/src/logical_plan/plan.rs
+++ b/datafusion/expr/src/logical_plan/plan.rs
@@ -4333,13 +4333,13 @@ pub struct Join {
pub schema: DFSchemaRef,
/// Defines the null equality for the join.
pub null_equality: NullEquality,
- /// Whether this is a null-aware anti join (for NOT IN semantics).
+ /// Whether this join needs null-aware NOT IN semantics.
///
- /// Only applies to LeftAnti joins. When true, implements SQL NOT IN semantics where:
- /// - If the right side (subquery) contains any NULL in join keys, no rows are output
- /// - Left side rows with NULL in join keys are not output
+ /// For `LeftAnti`, if the right side contains any NULL in join keys, no rows are output and
+ /// left rows with NULL join keys are also excluded.
///
- /// This is required for correct NOT IN subquery behavior with three-valued logic.
+ /// For `LeftMark`, the generated `mark` column becomes nullable so unmatched rows can produce
+ /// `NULL` rather than `false` when SQL three-valued logic requires it.
pub null_aware: bool,
}
@@ -4358,7 +4358,7 @@ impl Join {
/// * `join_type` - Type of join (Inner, Left, Right, etc.)
/// * `join_constraint` - Join constraint (On, Using)
/// * `null_equality` - How to handle nulls in join comparisons
- /// * `null_aware` - Whether this is a null-aware anti join (for NOT IN semantics)
+ /// * `null_aware` - Whether this join needs null-aware NOT IN semantics
///
/// # Returns
///
@@ -6637,7 +6637,9 @@ mod tests {
assert!(!fields[0].is_nullable());
assert!(!fields[1].is_nullable());
- assert!(!fields[2].is_nullable());
+ // The mark column is always nullable: null-aware `LeftMark`
+ // joins use NULL to represent SQL UNKNOWN for `NOT IN`.
+ assert!(fields[2].is_nullable());
}
_ => {
assert_eq!(join.schema.fields().len(), 4);
diff --git a/datafusion/optimizer/src/decorrelate_predicate_subquery.rs b/datafusion/optimizer/src/decorrelate_predicate_subquery.rs
index 66ea0806bd..5f623f1bef 100644
--- a/datafusion/optimizer/src/decorrelate_predicate_subquery.rs
+++ b/datafusion/optimizer/src/decorrelate_predicate_subquery.rs
@@ -21,6 +21,7 @@ use std::ops::Deref;
use std::sync::Arc;
use crate::decorrelate::PullUpCorrelatedExpr;
+use crate::extract_equijoin_predicate::split_eq_and_noneq_join_predicate;
use crate::optimizer::ApplyOrder;
use crate::utils::replace_qualified_name;
use crate::{OptimizerConfig, OptimizerRule};
@@ -448,9 +449,40 @@ fn build_join(
sub_query_alias.clone()
};
- // Mark joins don't use null-aware semantics (they use three-valued logic with mark column)
+ let mark_filter_is_hashable_only =
+ if join_type == JoinType::LeftMark && in_predicate_opt.is_some() {
+ let (_, residual_filter) = split_eq_and_noneq_join_predicate(
+ join_filter.clone(),
+ left.schema(),
+ right_projected.schema(),
+ )?;
+ residual_filter.is_none()
+ } else {
+ false
+ };
+
+ // For scalar NOT IN mark joins, propagate null-aware semantics into the
+ // nullable mark column when the predicate can be implemented by hash keys.
+ // Non-equality correlated filters stay on the legacy path because hash join
+ // execution cannot mark UNKNOWN candidates for residual predicates.
+ let null_aware = join_type == JoinType::LeftMark
+ && in_predicate_opt.is_some()
+ && mark_filter_is_hashable_only
+ && join_keys_may_be_null(
+ &join_filter,
+ left.schema(),
+ right_projected.schema(),
+ )?;
+
let new_plan = LogicalPlanBuilder::from(left.clone())
- .join_on(right_projected, join_type, Some(join_filter))?
+ .join_detailed_with_options(
+ right_projected,
+ join_type,
+ (Vec::<Column>::new(), Vec::<Column>::new()),
+ Some(join_filter),
+ NullEquality::NullEqualsNothing,
+ null_aware,
+ )?
.build()?;
debug!(
@@ -571,6 +603,43 @@ mod tests {
))
}
+ fn nullable_scalar_mark_scan(name: &str) -> Result<LogicalPlan> {
+ let schema = Schema::new(vec![
+ Field::new("id", DataType::Int32, true),
+ Field::new("grp", DataType::Int32, true),
+ ]);
+ table_scan(Some(name), &schema, None)?.build()
+ }
+
+ fn has_null_aware_left_mark_join(plan: &LogicalPlan) -> bool {
+ if let LogicalPlan::Join(join) = plan
+ && join.join_type == JoinType::LeftMark
+ {
+ return join.null_aware;
+ }
+
+ plan.inputs().into_iter().any(has_null_aware_left_mark_join)
+ }
+
+ fn has_non_null_aware_left_mark_join(plan: &LogicalPlan) -> bool {
+ if let LogicalPlan::Join(join) = plan
+ && join.join_type == JoinType::LeftMark
+ {
+ return !join.null_aware;
+ }
+
+ plan.inputs()
+ .into_iter()
+ .any(has_non_null_aware_left_mark_join)
+ }
+
+ fn optimize_with_decorrelate(plan: LogicalPlan) -> Result<LogicalPlan> {
+ let optimizer = crate::Optimizer::with_rules(vec![Arc::new(
+ DecorrelatePredicateSubquery::new(),
+ )]);
+ optimizer.optimize(plan, &crate::OptimizerContext::new(), |_, _| {})
+ }
+
/// Test for several IN subquery expressions
#[test]
fn in_subquery_multiple() -> Result<()> {
@@ -1201,6 +1270,62 @@ mod tests {
)
}
+ #[test]
+ fn correlated_not_in_mark_join_is_null_aware_for_hashable_filter() -> Result<()> {
+ let outer_scan = nullable_scalar_mark_scan("outer_t")?;
+ let inner_scan = nullable_scalar_mark_scan("inner_t")?;
+
+ let subquery = Arc::new(
+ LogicalPlanBuilder::from(inner_scan)
+ .filter(
+ out_ref_col(DataType::Int32, "outer_t.grp").eq(col("inner_t.grp")),
+ )?
+ .project(vec![col("inner_t.id")])?
+ .build()?,
+ );
+
+ let plan = LogicalPlanBuilder::from(outer_scan)
+ .filter(not_in_subquery(col("outer_t.id"), subquery).is_null())?
+ .build()?;
+
+ let optimized = optimize_with_decorrelate(plan)?;
+ assert!(
+ has_null_aware_left_mark_join(&optimized),
+ "{}",
+ optimized.display_indent_schema()
+ );
+
+ Ok(())
+ }
+
+ #[test]
+ fn correlated_not_in_mark_join_is_not_null_aware_for_residual_filter() -> Result<()> {
+ let outer_scan = nullable_scalar_mark_scan("outer_t")?;
+ let inner_scan = nullable_scalar_mark_scan("inner_t")?;
+
+ let subquery = Arc::new(
+ LogicalPlanBuilder::from(inner_scan)
+ .filter(
+ out_ref_col(DataType::Int32, "outer_t.grp").lt(col("inner_t.grp")),
+ )?
+ .project(vec![col("inner_t.id")])?
+ .build()?,
+ );
+
+ let plan = LogicalPlanBuilder::from(outer_scan)
+ .filter(not_in_subquery(col("outer_t.id"), subquery).is_null())?
+ .build()?;
+
+ let optimized = optimize_with_decorrelate(plan)?;
+ assert!(
+ has_non_null_aware_left_mark_join(&optimized),
+ "{}",
+ optimized.display_indent_schema()
+ );
+
+ Ok(())
+ }
+
#[test]
fn in_subquery_both_side_expr() -> Result<()> {
let table_scan = test_table_scan()?;
@@ -1745,8 +1870,8 @@ mod tests {
@r"
Projection: customer.c_custkey [c_custkey:Int64]
Projection: customer.c_custkey, customer.c_name [c_custkey:Int64, c_name:Utf8]
- Filter: __correlated_sq_1.mark OR customer.c_custkey = Int32(1) [c_custkey:Int64, c_name:Utf8, mark:Boolean]
- LeftMark Join: Filter: Boolean(true) [c_custkey:Int64, c_name:Utf8, mark:Boolean]
+ Filter: __correlated_sq_1.mark OR customer.c_custkey = Int32(1) [c_custkey:Int64, c_name:Utf8, mark:Boolean;N]
+ LeftMark Join: Filter: Boolean(true) [c_custkey:Int64, c_name:Utf8, mark:Boolean;N]
TableScan: customer [c_custkey:Int64, c_name:Utf8]
SubqueryAlias: __correlated_sq_1 [o_custkey:Int64]
Projection: orders.o_custkey [o_custkey:Int64]
diff --git a/datafusion/optimizer/src/extract_equijoin_predicate.rs b/datafusion/optimizer/src/extract_equijoin_predicate.rs
index 0a50761e8a..03a5aaa204 100644
--- a/datafusion/optimizer/src/extract_equijoin_predicate.rs
+++ b/datafusion/optimizer/src/extract_equijoin_predicate.rs
@@ -174,7 +174,7 @@ impl OptimizerRule for ExtractEquijoinPredicate {
///
/// According to the above rule, `expr1` is the equijoin predicate, while `expr2` and `expr3` are not.
/// The function returns Ok(\[expr1\], Some(expr2 AND expr3))
-fn split_eq_and_noneq_join_predicate(
+pub(crate) fn split_eq_and_noneq_join_predicate(
filter: Expr,
left_schema: &DFSchema,
right_schema: &DFSchema,
diff --git a/datafusion/physical-plan/src/joins/hash_join/exec.rs b/datafusion/physical-plan/src/joins/hash_join/exec.rs
index 1539fd4e73..cc396da278 100644
--- a/datafusion/physical-plan/src/joins/hash_join/exec.rs
+++ b/datafusion/physical-plan/src/joins/hash_join/exec.rs
@@ -69,7 +69,7 @@ use crate::{
metrics::{ExecutionPlanMetricsSet, MetricsSet},
};
-use arrow::array::{ArrayRef, BooleanBufferBuilder};
+use arrow::array::{Array, ArrayRef, BooleanBufferBuilder, UInt64Array};
use arrow::compute::concat_batches;
use arrow::datatypes::SchemaRef;
use arrow::record_batch::RecordBatch;
@@ -93,7 +93,7 @@ use datafusion_physical_expr::expressions::{Column, DynamicFilterPhysicalExpr, l
use datafusion_physical_expr::projection::{ProjectionRef, combine_projections};
use datafusion_physical_expr::{PhysicalExpr, PhysicalExprRef};
-use datafusion_common::hash_utils::RandomState;
+use datafusion_common::hash_utils::{RandomState, create_hashes};
use datafusion_physical_expr_common::physical_expr::fmt_sql;
use datafusion_physical_expr_common::utils::evaluate_expressions_to_arrays;
use futures::TryStreamExt;
@@ -192,17 +192,50 @@ fn try_create_array_map(
Ok(Some((array_map, batch, left_values)))
}
+/// Correlation-scope hash map over only the build rows whose scalar `NOT IN`
+/// value key is NULL, used by correlated null-aware `LeftMark` joins.
+///
+/// Such rows produce a NULL (UNKNOWN) mark whenever *any* probe row shares
+/// their correlation scope, so every probe row must be tested against them.
+/// Restricting this map to the NULL-valued build rows keeps that lookup
+/// proportional to the number of NULLs instead of enumerating every scope
+/// match of every probe row.
+pub(super) struct NullValueScopeMap {
+ /// Hash table keyed by the correlation scope values of the NULL-valued
+ /// build rows. Stored positions index into `scope_values`/`build_indices`,
+ /// not the full build batch.
+ pub(super) map: Box<dyn JoinHashMapType>,
+ /// Correlation scope key values of the NULL-valued build rows.
+ pub(super) scope_values: Vec<ArrayRef>,
+ /// Maps positions in `map`/`scope_values` back to row indices in the full
+ /// build batch.
+ pub(super) build_indices: UInt64Array,
+}
+
/// HashTable and input data for the left (build side) of a join
pub(super) struct JoinLeftData {
/// The hash table with indices into `batch`
/// Arc is used to allow sharing with SharedBuildAccumulator for hash map pushdown
pub(super) map: Arc<Map>,
+ /// Hash table over correlated scope keys for scalar null-aware mark joins.
+ ///
+ /// For null-aware `LeftMark`, key 0 is the scalar `NOT IN` value key and
+ /// keys 1..N are correlated equality scope keys. This map covers all build
+ /// rows and is probed only with NULL-valued probe rows; the complementary
+ /// direction uses `null_value_scope_map`.
+ null_aware_mark_scope_map: Option<Box<dyn JoinHashMapType>>,
+ /// Scope map restricted to the build rows whose value key is NULL (see
+ /// [`NullValueScopeMap`]). `None` when the build side has no NULL value
+ /// keys.
+ null_value_scope_map: Option<NullValueScopeMap>,
/// The input rows for the build side
batch: RecordBatch,
/// The build side on expressions values
values: Vec<ArrayRef>,
/// Shared bitmap builder for visited left indices
visited_indices_bitmap: SharedBitmapBuilder,
+ /// Shared bitmap builder for null marks
+ null_indices_bitmap: SharedBitmapBuilder,
/// Counter of running probe-threads, potentially
/// able to update `visited_indices_bitmap`
probe_threads_counter: AtomicUsize,
@@ -218,7 +251,7 @@ pub(super) struct JoinLeftData {
/// Membership testing strategy for filter pushdown
/// Contains either InList values for small build sides or hash table reference for large build sides
pub(super) membership: PushdownStrategy,
- /// Shared atomic flag indicating if any probe partition saw data (for null-aware anti joins)
+ /// Shared atomic flag indicating if any probe partition saw data (for null-aware anti/mark joins)
/// This is shared across all probe partitions to provide global knowledge
pub(super) probe_side_non_empty: AtomicBool,
/// Shared atomic flag indicating if any probe partition saw NULL in join keys (for null-aware anti joins)
@@ -234,6 +267,14 @@ impl JoinLeftData {
&self.map
}
+ pub(super) fn null_aware_mark_scope_map(&self) -> Option<&dyn JoinHashMapType> {
+ self.null_aware_mark_scope_map.as_deref()
+ }
+
+ pub(super) fn null_value_scope_map(&self) -> Option<&NullValueScopeMap> {
+ self.null_value_scope_map.as_ref()
+ }
+
/// returns a reference to the build side batch
pub(super) fn batch(&self) -> &RecordBatch {
&self.batch
@@ -266,6 +307,10 @@ impl JoinLeftData {
&self.visited_indices_bitmap
}
+ pub(super) fn null_indices_bitmap(&self) -> &SharedBitmapBuilder {
+ &self.null_indices_bitmap
+ }
+
/// returns a reference to the InList values for filter pushdown
pub(super) fn membership(&self) -> &PushdownStrategy {
&self.membership
@@ -436,18 +481,32 @@ impl HashJoinExecBuilder {
(join_type, partition_mode),
(JoinType::LeftAnti, _)
| (JoinType::RightAnti, PartitionMode::CollectLeft) // `PartitionMode::CollectLeft` is safe because `RightAnti` is probe-driven
+ | (JoinType::LeftMark, _)
) {
return plan_err!(
- "null_aware can only be true for LeftAnti joins and RightAnti joins with `CollectLeft` `PartitionMode`, got {join_type} with {partition_mode}"
+ "null_aware can only be true for LeftAnti joins and RightAnti joins with `CollectLeft` `PartitionMode`, or LeftMark joins, got {join_type} with {partition_mode}"
);
}
let on = exec.on();
- if on.len() != 1 {
+ if *join_type == JoinType::LeftAnti && on.len() != 1 {
return plan_err!(
- "null_aware anti join only supports single column join key, got {} columns",
+ "null_aware LeftAnti joins only support single column join key, got {} columns",
on.len()
);
}
+ if *join_type == JoinType::RightAnti && on.len() != 1 {
+ return plan_err!(
+ "null_aware RightAnti joins only support single column join key, got {} columns",
+ on.len()
+ );
+ }
+ if *join_type == JoinType::LeftMark
+ && matches!(partition_mode, PartitionMode::Partitioned)
+ {
+ return plan_err!(
+ "null_aware joins require PartitionMode::CollectLeft, got PartitionMode::Partitioned"
+ );
+ }
if *join_type == JoinType::RightAnti && exec.filter.is_some() {
return plan_err!(
"null_aware RightAnti join does not support a join filter"
@@ -784,7 +843,19 @@ pub struct HashJoinExec {
column_indices: Vec<ColumnIndex>,
/// The equality null-handling behavior of the join algorithm.
pub null_equality: NullEquality,
- /// Flag to indicate if this is a null-aware anti join
+ /// Flag to indicate if this join uses null-aware equality semantics.
+ ///
+ /// Set for the physical lowering of scalar `NOT IN` subqueries (producing
+ /// `JoinType::LeftAnti` when uncorrelated or `JoinType::LeftMark` when
+ /// correlated). When `true`, NULLs in the join keys follow SQL `NOT IN`
+ /// three-valued logic rather than ordinary equi-join semantics.
+ ///
+ /// Key-ordering convention (relied on positionally, not enforced): for a
+ /// null-aware `LeftMark` join with more than one key, `on[0]` is the scalar
+ /// `NOT IN` value key and `on[1..N]` are the correlated equality scope keys.
+ /// Reordering these keys would silently produce wrong results, which is why
+ /// such joins are pinned to `PartitionMode::CollectLeft` (the only key
+ /// reorderer acts solely on `PartitionMode::Partitioned`).
pub null_aware: bool,
/// Cache holding plan properties like equivalences, output partitioning etc.
cache: Arc<PlanProperties>,
@@ -882,10 +953,15 @@ impl HashJoinExec {
return false;
}
- // A null-aware anti join emits a build-side NULL only when the probe
- // is truly empty. The pushed filter can empty the probe by pruning
- // every row, which would surface that NULL wrongly. A NOT NULL build
- // key cannot produce such a NULL, so the filter stays there.
+ // Null-aware joins tolerate probe pruning only while the build side
+ // has no NULL keys (the pushed filter keeps NULL-valued probe rows via
+ // the `IS NULL` escape in `preserve_probe_nulls`, but prunes others):
+ // - LeftAnti emits a build-side NULL only when the probe is truly
+ // empty, and pruning every probe row would surface that NULL wrongly.
+ // - Null-aware LeftMark marks a NULL-valued build row UNKNOWN when any
+ // probe row shares its correlation scope, and a pruned probe row
+ // would be missed by that scan.
+ // NOT NULL build keys rule out both, so the filter stays there.
if self.null_aware
&& self.on.iter().any(|(build_key, _)| {
build_key.nullable(&self.left.schema()).unwrap_or(true)
@@ -1513,6 +1589,11 @@ impl ExecutionPlan for HashJoinExec {
.flatten()
.flatten();
+ // The extra scope maps + null bitmap are only built for correlated
+ // null-aware LeftMark joins (`on[1..]` are correlation scope keys).
+ let with_null_aware_mark_state =
+ self.null_aware && self.join_type == JoinType::LeftMark && on_left.len() > 1;
+
let left_fut = match self.mode {
PartitionMode::CollectLeft => self.left_fut.try_once(|| {
let left_stream = self.left.execute(0, Arc::clone(&context))?;
@@ -1533,6 +1614,7 @@ impl ExecutionPlan for HashJoinExec {
self.null_equality,
self.null_aware && self.join_type == JoinType::RightAnti,
array_map_created_count,
+ with_null_aware_mark_state,
))
})?,
PartitionMode::Partitioned => {
@@ -1554,6 +1636,9 @@ impl ExecutionPlan for HashJoinExec {
self.null_equality,
false,
array_map_created_count,
+ // Partitioned mode is rejected for null-aware joins (see
+ // `try_new`), so the extra null-aware state is never built.
+ false,
))
}
PartitionMode::Auto => {
@@ -2562,6 +2647,29 @@ fn is_perfect_hash_join_candidate(
Ok(ArrayMap::is_supported_type(&data_type))
}
+fn new_join_hashmap(
+ num_rows: usize,
+ reservation: &mut MemoryReservation,
+ metrics: &BuildProbeJoinMetrics,
+) -> Result<Box<dyn JoinHashMapType>> {
+ let fixed_size_u32 = size_of::<JoinHashMapU32>();
+ let fixed_size_u64 = size_of::<JoinHashMapU64>();
+
+ if num_rows > u32::MAX as usize {
+ let estimated_hashtable_size =
+ estimate_memory_size::<(u64, u64)>(num_rows, fixed_size_u64)?;
+ reservation.try_grow(estimated_hashtable_size)?;
+ metrics.build_mem_used.add(estimated_hashtable_size);
+ Ok(Box::new(JoinHashMapU64::with_capacity(num_rows)))
+ } else {
+ let estimated_hashtable_size =
+ estimate_memory_size::<(u32, u64)>(num_rows, fixed_size_u32)?;
+ reservation.try_grow(estimated_hashtable_size)?;
+ metrics.build_mem_used.add(estimated_hashtable_size);
+ Ok(Box::new(JoinHashMapU32::with_capacity(num_rows)))
+ }
+}
+
/// Collects all batches from the left (build) side stream and creates a hash map for joining.
///
/// This function is responsible for:
@@ -2579,6 +2687,8 @@ fn is_perfect_hash_join_candidate(
/// * `with_visited_indices_bitmap` - Whether to track visited indices (for outer joins)
/// * `probe_threads_count` - Number of threads that will probe this hash table
/// * `should_compute_dynamic_filters` - Whether to compute min/max bounds for dynamic filtering
+/// * `with_null_aware_mark_state` - Whether to build the per-build-row null-indices bitmap
+/// and correlation-scope maps used by correlated null-aware `LeftMark` joins
///
/// # Dynamic Filter Coordination
/// When `should_compute_dynamic_filters` is true, this function computes the min/max bounds
@@ -2590,7 +2700,7 @@ fn is_perfect_hash_join_candidate(
/// # Returns
/// `JoinLeftData` containing the hash map, consolidated batch, join key values,
/// visited indices bitmap, and computed bounds (if requested).
-#[expect(clippy::too_many_arguments)]
+#[expect(clippy::too_many_arguments, clippy::fn_params_excessive_bools)]
async fn collect_left_input(
random_state: RandomState,
left_stream: SendableRecordBatchStream,
@@ -2604,6 +2714,7 @@ async fn collect_left_input(
null_equality: NullEquality,
compute_build_side_has_null: bool,
array_map_created_count: Count,
+ with_null_aware_mark_state: bool,
) -> Result<JoinLeftData> {
let schema = left_stream.schema();
@@ -2682,25 +2793,10 @@ async fn collect_left_input(
} else {
// Estimation of memory size, required for hashtable, prior to allocation.
// Final result can be verified using `RawTable.allocation_info()`
- let fixed_size_u32 = size_of::<JoinHashMapU32>();
- let fixed_size_u64 = size_of::<JoinHashMapU64>();
-
// Use `u32` indices for the JoinHashMap when num_rows ≤ u32::MAX, otherwise use the
// `u64` indice variant
// Arc is used instead of Box to allow sharing with SharedBuildAccumulator for hash map pushdown
- let mut hashmap: Box<dyn JoinHashMapType> = if num_rows > u32::MAX as usize {
- let estimated_hashtable_size =
- estimate_memory_size::<(u64, u64)>(num_rows, fixed_size_u64)?;
- reservation.try_grow(estimated_hashtable_size)?;
- metrics.build_mem_used.add(estimated_hashtable_size);
- Box::new(JoinHashMapU64::with_capacity(num_rows))
- } else {
- let estimated_hashtable_size =
- estimate_memory_size::<(u32, u64)>(num_rows, fixed_size_u32)?;
- reservation.try_grow(estimated_hashtable_size)?;
- metrics.build_mem_used.add(estimated_hashtable_size);
- Box::new(JoinHashMapU32::with_capacity(num_rows))
- };
+ let mut hashmap = new_join_hashmap(num_rows, &mut reservation, &metrics)?;
let mut hashes_buffer = Vec::new();
let mut offset = 0;
@@ -2733,19 +2829,99 @@ async fn collect_left_input(
(Map::HashMap(hashmap), batch, left_values)
};
- // Reserve additional memory for visited indices bitmap and create shared builder
- let visited_indices_bitmap = if with_visited_indices_bitmap {
+ let allocate_bitmap = || -> Result<BooleanBufferBuilder> {
let bitmap_size = bit_util::ceil(batch.num_rows(), 8);
reservation.try_grow(bitmap_size)?;
metrics.build_mem_used.add(bitmap_size);
- let mut bitmap_buffer = BooleanBufferBuilder::new(batch.num_rows());
- bitmap_buffer.append_n(num_rows, false);
- bitmap_buffer
+ let mut bitmap = BooleanBufferBuilder::new(batch.num_rows());
+ bitmap.append_n(num_rows, false);
+ Ok(bitmap)
+ };
+
+ // Reserve additional memory for visited indices bitmap and create shared builder
+ let visited_indices_bitmap = if with_visited_indices_bitmap {
+ allocate_bitmap()?
} else {
BooleanBufferBuilder::new(0)
};
+ let null_indices_bitmap = if with_null_aware_mark_state {
+ allocate_bitmap()?
+ } else {
+ BooleanBufferBuilder::new(0)
+ };
+
+ let (null_aware_mark_scope_map, null_value_scope_map) = if with_null_aware_mark_state
+ {
+ // Null-aware `LeftMark` convention: `on_left[0]` is the value key and
+ // `on_left[1..]` the scope keys, so the scope map needs more than one key.
+ debug_assert!(
+ on_left.len() > 1,
+ "null-aware LeftMark needs on_left[0]=value, on_left[1..]=scope, got {} key(s)",
+ on_left.len()
+ );
+ // Scope-only NULL marking uses a HashMap (the primary join map may use
+ // ArrayMap for full-key matches, but scope keys have arbitrary shape).
+ let mut scope_map = new_join_hashmap(num_rows, &mut reservation, &metrics)?;
+
+ let mut hashes_buffer = vec![0; batch.num_rows()];
+ update_hash(
+ &on_left[1..],
+ &batch,
+ &mut *scope_map,
+ 0,
+ &random_state,
+ &mut hashes_buffer,
+ 0,
+ true,
+ NullEquality::NullEqualsNothing,
+ )?;
+
+ // Build the dedicated scope map over the NULL-valued build rows (see
+ // `NullValueScopeMap`).
+ let value_key = &left_values[0];
+ let null_value_scope_map = if value_key.null_count() > 0 {
+ let null_mask = arrow::compute::is_null(value_key.as_ref())?;
+ let build_indices = UInt64Array::from_iter_values(
+ null_mask.values().set_indices().map(|i| i as u64),
+ );
+ let scope_values = left_values[1..]
+ .iter()
+ .map(|values| Ok(arrow::compute::filter(values.as_ref(), &null_mask)?))
+ .collect::<Result<Vec<_>>>()?;
+
+ // `scope_values` and `build_indices` are retained for the lifetime
+ // of the join, so charge them to the reservation like the maps and
+ // bitmaps above.
+ let retained_size = build_indices.get_array_memory_size()
+ + scope_values
+ .iter()
+ .map(|values| values.get_array_memory_size())
+ .sum::<usize>();
+ reservation.try_grow(retained_size)?;
+ metrics.build_mem_used.add(retained_size);
+
+ let null_rows = build_indices.len();
+ let mut map = new_join_hashmap(null_rows, &mut reservation, &metrics)?;
+ let mut hashes_buffer = vec![0; null_rows];
+ create_hashes(&scope_values, &random_state, &mut hashes_buffer)?;
+ map.update_from_iter(Box::new(hashes_buffer.iter().enumerate().rev()), 0);
+
+ Some(NullValueScopeMap {
+ map,
+ scope_values,
+ build_indices,
+ })
+ } else {
+ None
+ };
+
+ (Some(scope_map), null_value_scope_map)
+ } else {
+ (None, None)
+ };
+
let map = Arc::new(join_hash_map);
let membership = if num_rows == 0 {
@@ -2784,9 +2960,12 @@ async fn collect_left_input(
let data = JoinLeftData {
map,
+ null_aware_mark_scope_map,
+ null_value_scope_map,
batch,
values: left_values,
visited_indices_bitmap: Mutex::new(visited_indices_bitmap),
+ null_indices_bitmap: Mutex::new(null_indices_bitmap),
probe_threads_counter: AtomicUsize::new(probe_threads_count),
_reservation: reservation,
bounds,
@@ -7927,12 +8106,50 @@ mod tests {
true, // null_aware = true (invalid for multi-column)
);
+ assert!(result.is_err());
+ assert!(
+ result.unwrap_err().to_string().contains(
+ "null_aware LeftAnti joins only support single column join key"
+ )
+ );
+ }
+
+ /// Test that null_aware validation rejects PartitionMode::Partitioned (must be CollectLeft).
+ #[tokio::test]
+ async fn test_null_aware_validation_partitioned() {
+ let left = build_table(("a", &vec![1]), ("b", &vec![2]), ("c", &vec![3]));
+ let right = build_table(("x", &vec![1]), ("y", &vec![2]), ("z", &vec![3]));
+
+ let on = vec![
+ (
+ Arc::new(Column::new_with_schema("a", &left.schema()).unwrap()) as _,
+ Arc::new(Column::new_with_schema("x", &right.schema()).unwrap()) as _,
+ ),
+ (
+ Arc::new(Column::new_with_schema("b", &left.schema()).unwrap()) as _,
+ Arc::new(Column::new_with_schema("y", &right.schema()).unwrap()) as _,
+ ),
+ ];
+
+ // LeftMark (not LeftAnti) so the single-key check is skipped and the partition-mode check is reached.
+ let result = HashJoinExec::try_new(
+ left,
+ right,
+ on,
+ None,
+ &JoinType::LeftMark,
+ None,
+ PartitionMode::Partitioned,
+ NullEquality::NullEqualsNothing,
+ true, // null_aware = true (invalid with Partitioned)
+ );
+
assert!(result.is_err());
assert!(
result
.unwrap_err()
.to_string()
- .contains("null_aware anti join only supports single column join key")
+ .contains("null_aware joins require PartitionMode::CollectLeft")
);
}
@@ -8051,6 +8268,143 @@ mod tests {
Ok(())
}
+ /// Scalar correlated null-aware left mark join where only the build side's
+ /// value key contains NULL.
+ ///
+ /// This isolates the NULL-value scope map direction of
+ /// `mark_null_candidates_for_probe_batch`: NULL-valued build rows become
+ /// UNKNOWN as soon as any probe row shares their correlation scope, while
+ /// non-null build rows stay `true`/`false`.
+ #[apply(hash_join_exec_configs)]
+ #[tokio::test]
+ async fn test_null_aware_left_mark_correlated_build_null_only(
+ batch_size: usize,
+ ) -> Result<()> {
+ let task_ctx = prepare_task_ctx(batch_size, false);
+
+ let left = build_table_two_cols(
+ ("id", &vec![Some(1), None, None, Some(3)]),
+ ("grp", &vec![Some(1), Some(1), Some(2), Some(3)]),
+ );
+
+ let right = build_table_two_cols(
+ ("id", &vec![Some(2), Some(1)]),
+ ("grp", &vec![Some(1), Some(2)]),
+ );
+
+ let on = vec![
+ (
+ Arc::new(Column::new_with_schema("id", &left.schema())?) as _,
+ Arc::new(Column::new_with_schema("id", &right.schema())?) as _,
+ ),
+ (
+ Arc::new(Column::new_with_schema("grp", &left.schema())?) as _,
+ Arc::new(Column::new_with_schema("grp", &right.schema())?) as _,
+ ),
+ ];
+
+ let join = HashJoinExec::try_new(
+ left,
+ right,
+ on,
+ None,
+ &JoinType::LeftMark,
+ None,
+ PartitionMode::CollectLeft,
+ NullEquality::NullEqualsNothing,
+ true,
+ )?;
+
+ let stream = join.execute(0, task_ctx)?;
+ let batches = common::collect(stream).await?;
+
+ // `(NULL, 1)` and `(NULL, 2)` have non-empty correlated scopes, so
+ // their marks are UNKNOWN. `(1, 1)` finds no equal value and no NULL in
+ // its scope (false); `(3, 3)` has an empty correlated scope (false).
+ allow_duplicates! {
+ assert_snapshot!(batches_to_sort_string(&batches), @r"
+ +----+-----+-------+
+ | id | grp | mark |
+ +----+-----+-------+
+ | | 1 | |
+ | | 2 | |
+ | 1 | 1 | false |
+ | 3 | 3 | false |
+ +----+-----+-------+
+ ");
+ }
+
+ Ok(())
+ }
+
+ /// Scalar correlated null-aware left mark join where only the probe side's
+ /// value key contains NULL.
+ ///
+ /// This isolates the NULL-valued-probe-row direction of
+ /// `mark_null_candidates_for_probe_batch`: every unmatched build row whose
+ /// correlation scope contains a NULL-valued probe row becomes UNKNOWN.
+ #[apply(hash_join_exec_configs)]
+ #[tokio::test]
+ async fn test_null_aware_left_mark_correlated_probe_null_only(
+ batch_size: usize,
+ ) -> Result<()> {
+ let task_ctx = prepare_task_ctx(batch_size, false);
+
+ let left = build_table_two_cols(
+ ("id", &vec![Some(1), Some(2), Some(5), Some(7)]),
+ ("grp", &vec![Some(1), Some(1), Some(2), Some(3)]),
+ );
+
+ let right = build_table_two_cols(
+ ("id", &vec![Some(2), None, Some(4)]),
+ ("grp", &vec![Some(1), Some(1), Some(2)]),
+ );
+
+ let on = vec![
+ (
+ Arc::new(Column::new_with_schema("id", &left.schema())?) as _,
+ Arc::new(Column::new_with_schema("id", &right.schema())?) as _,
+ ),
+ (
+ Arc::new(Column::new_with_schema("grp", &left.schema())?) as _,
+ Arc::new(Column::new_with_schema("grp", &right.schema())?) as _,
+ ),
+ ];
+
+ let join = HashJoinExec::try_new(
+ left,
+ right,
+ on,
+ None,
+ &JoinType::LeftMark,
+ None,
+ PartitionMode::CollectLeft,
+ NullEquality::NullEqualsNothing,
+ true,
+ )?;
+
+ let stream = join.execute(0, task_ctx)?;
+ let batches = common::collect(stream).await?;
+
+ // `(1, 1)` is unmatched and its scope contains a NULL value (UNKNOWN);
+ // `(2, 1)` matches (true); `(5, 2)`'s scope holds only `4` (false);
+ // `(7, 3)` has an empty correlated scope (false).
+ allow_duplicates! {
+ assert_snapshot!(batches_to_sort_string(&batches), @r"
+ +----+-----+-------+
+ | id | grp | mark |
+ +----+-----+-------+
+ | 1 | 1 | |
+ | 2 | 1 | true |
+ | 5 | 2 | false |
+ | 7 | 3 | false |
+ +----+-----+-------+
+ ");
+ }
+
+ Ok(())
+ }
+
#[test]
fn test_lr_is_preserved() {
assert_eq!(lr_is_preserved(JoinType::Inner), (true, true));
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 0cff24332c..a4b2a86983 100644
--- a/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs
+++ b/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs
@@ -799,11 +799,13 @@ impl SharedBuildAccumulator {
/// Keeps probe rows with a NULL key when the join semantics need them.
///
- /// The build-side predicate drops probe rows whose key is NULL. A null-aware anti join
- /// (`NOT IN`) needs that NULL to reach the join so three-valued logic can collapse the
- /// result, and a null-equal join needs it to match a build-side NULL. OR-ing `key IS NULL`
- /// keeps those rows while preserving the filter's selectivity for the rest; the join refines
- /// whatever the widened filter lets through.
+ /// The build-side predicate drops probe rows whose key is NULL. A null-aware join
+ /// needs a NULL value key to reach the join for `NOT IN` three-valued logic, and a
+ /// null-equal join needs NULL keys to match build-side NULLs.
+ ///
+ /// For a null-aware join, only `on_right[0]` needs the escape: it is the scalar
+ /// `NOT IN` value key, while `on_right[1..]` are correlation scope keys. For a
+ /// null-equal join, every nullable key needs the escape.
fn preserve_probe_nulls(
&self,
filter_expr: Arc<dyn PhysicalExpr>,
@@ -819,11 +821,20 @@ impl SharedBuildAccumulator {
if !needs_probe_nulls {
return Ok(filter_expr);
}
- // Only a key that can actually be NULL needs the disjunct; a NOT NULL key never widens.
- // Null-aware joins are single-key; null-equal joins can be multi-key, so OR every nullable
- // key. If every key is NOT NULL the filter is left untouched, at full selectivity.
+ let keys = if self.null_aware {
+ assert_or_internal_err!(
+ !self.on_right.is_empty(),
+ "null-aware join must have at least one probe key"
+ );
+ &self.on_right[..1]
+ } else {
+ self.on_right.as_slice()
+ };
+
+ // Only a key that can actually be NULL needs the disjunct; a NOT NULL key
+ // never widens the filter.
let mut any_key_is_null: Option<Arc<dyn PhysicalExpr>> = None;
- for key in &self.on_right {
+ for key in keys {
// `nullable` fails only when a key is out of sync with the probe schema. That is
// a construction bug, so surface it instead of widening around it.
if !key.nullable(&self.probe_schema)? {
@@ -1606,4 +1617,62 @@ mod tests {
let widened = acc.preserve_probe_nulls(lit(true), false).unwrap();
assert_eq!(format!("{widened}").matches("IS NULL").count(), 1);
}
+
+ // Correlated null-aware LeftMark joins have multiple probe keys
+ // (`on_right[0]` = scalar NOT IN value key, `on_right[1..]` = correlation
+ // scope keys). The NULL escape must accept that shape and wrap only the
+ // value key.
+ #[test]
+ fn null_aware_multi_key_filter_escapes_value_key_only() {
+ let on_right: Vec<PhysicalExprRef> = vec![
+ Arc::new(Column::new("value_key", 0)),
+ Arc::new(Column::new("scope_key", 1)),
+ ];
+ let dynamic_filter = test_dynamic_filter(&on_right);
+ let acc = SharedBuildAccumulator {
+ inner: Mutex::new(AccumulatorState {
+ data: AccumulatedBuildData::CollectLeft {
+ data: PartitionStatus::Pending,
+ reported_count: 0,
+ expected_reports: 1,
+ },
+ completion: CompletionState::Pending,
+ }),
+ completion_notify: Notify::new(),
+ dynamic_filter,
+ on_right,
+ repartition_random_state: SeededRandomState::with_seed(1),
+ probe_schema: Arc::new(Schema::new(vec![
+ Field::new("value_key", DataType::Int32, true),
+ // Keep the scope key nullable so this test proves it is not widened.
+ Field::new("scope_key", DataType::Int32, true),
+ ])),
+ probe_range_partitioning: None,
+ null_equality: NullEquality::NullEqualsNothing,
+ null_aware: true,
+ };
+
+ let two_key_bounds = PartitionBounds::new(vec![
+ ColumnBounds::new(ScalarValue::Int32(Some(1)), ScalarValue::Int32(Some(5))),
+ ColumnBounds::new(ScalarValue::Int32(Some(10)), ScalarValue::Int32(Some(20))),
+ ]);
+ acc.build_filter(FinalizeInput::CollectLeft(reported(
+ PushdownStrategy::Empty,
+ two_key_bounds,
+ )))
+ .unwrap();
+
+ let expr = current_expr(&acc);
+ let or = binary_expr(&expr);
+ assert_eq!(or.op(), &Operator::Or);
+ let is_null = or
+ .left()
+ .downcast_ref::<IsNullExpr>()
+ .expect("expected IS NULL escape as the left disjunct");
+ let column = is_null
+ .arg()
+ .downcast_ref::<Column>()
+ .expect("expected column under IS NULL");
+ assert_eq!(column.index(), 0, "escape must target the NOT IN value key");
+ }
}
diff --git a/datafusion/physical-plan/src/joins/hash_join/stream.rs b/datafusion/physical-plan/src/joins/hash_join/stream.rs
index e01a751777..cf2db5b037 100644
--- a/datafusion/physical-plan/src/joins/hash_join/stream.rs
+++ b/datafusion/physical-plan/src/joins/hash_join/stream.rs
@@ -39,11 +39,12 @@ use crate::stream::EmptyRecordBatchStream;
use crate::{
RecordBatchStream, SendableRecordBatchStream, handle_state,
hash_utils::create_hashes,
+ joins::SharedBitmapBuilder,
joins::utils::{
BuildProbeJoinMetrics, ColumnIndex, JoinFilter, JoinHashMapType,
StatefulStreamResult, adjust_indices_by_join_type, apply_join_filter_to_indices,
build_batch_empty_build_side, build_batch_from_indices,
- need_produce_result_in_final,
+ build_null_aware_left_mark_column, need_produce_result_in_final,
},
};
@@ -325,6 +326,16 @@ pub(super) struct HashJoinStream {
probe_indices_buffer: Vec<u32>,
/// Scratch space for build indices during hash lookup
build_indices_buffer: Vec<u64>,
+
+ /// Scratch space for scope-key hashes in the null-aware mark pass, reused
+ /// across probe batches. Separate from `hashes_buffer`, which still holds
+ /// the probe batch's full-key hashes during the chunked lookup loop.
+ null_mark_hashes_buffer: Vec<u64>,
+ /// Scratch space for probe indices during null-aware scope lookups
+ null_mark_probe_indices_buffer: Vec<u32>,
+ /// Scratch space for build indices during null-aware scope lookups
+ null_mark_build_indices_buffer: Vec<u64>,
+
/// Specifies whether the right side has an ordering to potentially preserve
right_side_ordered: bool,
/// Owns this partition's build-data report lifecycle.
@@ -334,7 +345,7 @@ pub(super) struct HashJoinStream {
/// Output buffer for coalescing small batches into larger ones with optional fetch limit.
/// Uses `LimitedBatchCoalescer` to efficiently combine batches and absorb limit with 'fetch'
output_buffer: LimitedBatchCoalescer,
- /// Whether this is a null-aware anti join
+ /// Whether this is a null-aware anti or mark join
null_aware: bool,
}
@@ -509,6 +520,11 @@ impl HashJoinStream {
hashes_buffer,
probe_indices_buffer: Vec::with_capacity(batch_size),
build_indices_buffer: Vec::with_capacity(batch_size),
+ // Left unallocated: only correlated null-aware LeftMark joins ever
+ // use these, and they grow them on first use.
+ null_mark_hashes_buffer: Vec::new(),
+ null_mark_probe_indices_buffer: Vec::new(),
+ null_mark_build_indices_buffer: Vec::new(),
right_side_ordered,
build_report: BuildReportHandle::new(partition, mode, build_accumulator),
mode,
@@ -808,10 +824,48 @@ impl HashJoinStream {
return Ok(StatefulStreamResult::Continue);
}
}
+ JoinType::LeftMark => {
+ if state.batch.num_rows() > 0 {
+ build_side
+ .left_data
+ .probe_side_non_empty
+ .store(true, Ordering::Relaxed);
+ }
+
+ let probe_key_column = &state.values[0];
+ if probe_key_column.null_count() > 0 {
+ build_side
+ .left_data
+ .probe_side_has_null
+ .store(true, Ordering::Relaxed);
+ }
+ }
_ => {}
}
}
+ // For correlated null-aware LeftMark (`on[1..]` scope keys, hence
+ // values len > 1), record this batch's UNKNOWN candidates once, before
+ // the first chunked lookup (offset == (0, None)).
+ //
+ // Must precede the empty-build-map return below: an all-NULL-key build
+ // side has an empty full-key map but still needs UNKNOWN marks.
+ if self.null_aware
+ && self.join_type == JoinType::LeftMark
+ && state.values.len() > 1
+ && state.offset == (0, None)
+ {
+ mark_null_candidates_for_probe_batch(
+ build_side,
+ state,
+ &self.random_state,
+ self.batch_size,
+ &mut self.null_mark_hashes_buffer,
+ &mut self.null_mark_probe_indices_buffer,
+ &mut self.null_mark_build_indices_buffer,
+ )?;
+ }
+
let is_empty = !build_side.left_data.has_matchable_build_rows();
if is_empty {
@@ -972,6 +1026,7 @@ impl HashJoinStream {
&self.column_indices,
join_side,
self.join_type,
+ None,
)?;
let push_status = self.output_buffer.push_batch(batch)?;
@@ -1007,6 +1062,7 @@ impl HashJoinStream {
let timer = self.join_metrics.join_time.timer();
if !need_produce_result_in_final(self.join_type) {
+ timer.done();
self.state = HashJoinStreamState::Completed;
return Ok(StatefulStreamResult::Continue);
}
@@ -1016,6 +1072,7 @@ impl HashJoinStream {
// For null-aware anti join, if probe side had NULL, no rows should be output
// Check shared atomic state to get global knowledge across all partitions
if self.null_aware
+ && self.join_type == JoinType::LeftAnti
&& build_side
.left_data
.probe_side_has_null
@@ -1025,7 +1082,9 @@ impl HashJoinStream {
self.state = HashJoinStreamState::Completed;
return Ok(StatefulStreamResult::Continue);
}
+
if !build_side.left_data.report_probe_completed() {
+ timer.done();
self.state = HashJoinStreamState::Completed;
return Ok(StatefulStreamResult::Continue);
}
@@ -1055,32 +1114,56 @@ impl HashJoinStream {
let filtered_indices: Vec<u64> = left_side
.iter()
.filter_map(|idx| {
- let idx_usize = idx.unwrap() as usize;
+ let idx = idx.expect(
+ "LeftAnti final indices should always contain build-side rows",
+ );
+ let idx_usize = idx as usize;
if build_key_column.is_null(idx_usize) {
None // Skip rows with NULL keys
} else {
- Some(idx.unwrap())
+ Some(idx)
}
})
.collect();
left_side = UInt64Array::from(filtered_indices);
-
- // Update right_side to match the new length
- let mut builder = arrow::array::UInt32Builder::with_capacity(left_side.len());
- builder.append_nulls(left_side.len());
- right_side = builder.finish();
+ right_side = UInt32Array::new_null(left_side.len());
}
self.join_metrics.input_batches.add(1);
self.join_metrics.input_rows.add(left_side.len());
- timer.done();
-
- self.state = HashJoinStreamState::Completed;
-
// Push final unmatched indices to output buffer
if !left_side.is_empty() {
+ let mark_column = if self.null_aware && self.join_type == JoinType::LeftMark {
+ let probe_side_has_null = build_side
+ .left_data
+ .probe_side_has_null
+ .load(Ordering::Relaxed);
+ let probe_side_non_empty = build_side
+ .left_data
+ .probe_side_non_empty
+ .load(Ordering::Relaxed);
+ let build_key_column = &build_side.left_data.values()[0];
+ let null_indices_bitmap =
+ if build_side.left_data.null_aware_mark_scope_map().is_some() {
+ Some(build_side.left_data.null_indices_bitmap().lock())
+ } else {
+ None
+ };
+
+ Some(build_null_aware_left_mark_column(
+ &left_side,
+ &right_side,
+ build_key_column.as_ref(),
+ null_indices_bitmap.as_deref(),
+ probe_side_has_null,
+ probe_side_non_empty,
+ ))
+ } else {
+ None
+ };
+
let empty_right_batch = RecordBatch::new_empty(self.right.schema());
let batch = build_batch_from_indices(
&self.schema,
@@ -1091,6 +1174,7 @@ impl HashJoinStream {
&self.column_indices,
JoinSide::Left,
self.join_type,
+ mark_column.as_ref(),
)?;
let push_status = self.output_buffer.push_batch(batch)?;
@@ -1100,10 +1184,155 @@ impl HashJoinStream {
}
}
+ timer.done();
+ self.state = HashJoinStreamState::Completed;
+
Ok(StatefulStreamResult::Continue)
}
}
+/// Records which build rows of a correlated null-aware `LeftMark` join are
+/// UNKNOWN candidates for this probe batch.
+///
+/// Key layout: `on[0]` is the `NOT IN` value key, `on[1..]` the correlation
+/// scope keys (see `HashJoinExec::null_aware`). A build row's mark must be
+/// NULL (SQL UNKNOWN) instead of FALSE when it is unmatched and either:
+/// 1. its value key is NULL and any probe row shares its correlation scope, or
+/// 2. some probe row in its correlation scope has a NULL value key.
+///
+/// Case 1 probes the build-side NULL-value scope map with all probe rows;
+/// case 2 probes the full scope map with only the NULL-valued probe rows.
+fn mark_null_candidates_for_probe_batch(
+ build_side: &BuildSideReadyState,
+ state: &ProcessProbeBatchState,
+ random_state: &RandomState,
+ batch_size: usize,
+ hashes_buffer: &mut Vec<u64>,
+ probe_indices_buffer: &mut Vec<u32>,
+ build_indices_buffer: &mut Vec<u64>,
+) -> Result<()> {
+ let Some(scope_map) = build_side.left_data.null_aware_mark_scope_map() else {
+ return Ok(());
+ };
+
+ debug_assert_eq!(
+ build_side.left_data.values().len(),
+ state.values.len(),
+ "build/probe key counts must match"
+ );
+ debug_assert!(state.values.len() > 1, "keys must be [value, scope..]");
+
+ let probe_value_key = &state.values[0];
+ let build_scope_values = &build_side.left_data.values()[1..];
+ let probe_scope_values = &state.values[1..];
+
+ let null_value_scope_map = build_side.left_data.null_value_scope_map();
+ let probe_has_null_values = probe_value_key.null_count() > 0;
+ if null_value_scope_map.is_none() && !probe_has_null_values {
+ return Ok(());
+ }
+
+ // Case 1: build rows with a NULL value key are UNKNOWN as soon as any
+ // probe row shares their correlation scope.
+ if let Some(null_value_scope_map) = null_value_scope_map {
+ hashes_buffer.clear();
+ hashes_buffer.resize(state.batch.num_rows(), 0);
+ create_hashes(probe_scope_values, random_state, hashes_buffer)?;
+
+ scan_scope_matches_into_bitmap(
+ null_value_scope_map.map.as_ref(),
+ &null_value_scope_map.scope_values,
+ probe_scope_values,
+ hashes_buffer,
+ batch_size,
+ probe_indices_buffer,
+ build_indices_buffer,
+ // The map indexes only the NULL-valued build rows; translate its
+ // positions back to row indices in the full build batch.
+ |position| null_value_scope_map.build_indices.value(position as usize),
+ build_side.left_data.null_indices_bitmap(),
+ )?;
+ }
+
+ // Case 2: NULL-valued probe rows make every build row in their correlation
+ // scope an UNKNOWN candidate.
+ if probe_has_null_values {
+ let null_mask = arrow::compute::is_null(probe_value_key.as_ref())?;
+ let probe_null_scope_values = probe_scope_values
+ .iter()
+ .map(|values| Ok(arrow::compute::filter(values.as_ref(), &null_mask)?))
+ .collect::<Result<Vec<_>>>()?;
+
+ hashes_buffer.clear();
+ hashes_buffer.resize(null_mask.true_count(), 0);
+ create_hashes(&probe_null_scope_values, random_state, hashes_buffer)?;
+
+ scan_scope_matches_into_bitmap(
+ scope_map,
+ build_scope_values,
+ &probe_null_scope_values,
+ hashes_buffer,
+ batch_size,
+ probe_indices_buffer,
+ build_indices_buffer,
+ |position| position,
+ build_side.left_data.null_indices_bitmap(),
+ )?;
+ }
+
+ Ok(())
+}
+
+/// Scans all correlation-scope matches between `build_scope_values` and
+/// `probe_scope_values` and sets the bit of every matched build row in
+/// `null_bitmap`, translating matched map positions through
+/// `map_position_to_build_row`.
+#[expect(clippy::too_many_arguments)]
+fn scan_scope_matches_into_bitmap(
+ scope_map: &dyn JoinHashMapType,
+ build_scope_values: &[ArrayRef],
+ probe_scope_values: &[ArrayRef],
+ hashes_buffer: &[u64],
+ batch_size: usize,
+ probe_indices_buffer: &mut Vec<u32>,
+ build_indices_buffer: &mut Vec<u64>,
+ map_position_to_build_row: impl Fn(u64) -> u64,
+ null_bitmap: &SharedBitmapBuilder,
+) -> Result<()> {
+ let mut offset = (0, None);
+ loop {
+ let (build_indices, _probe_indices, next_offset) = lookup_join_hashmap(
+ scope_map,
+ build_scope_values,
+ probe_scope_values,
+ NullEquality::NullEqualsNothing,
+ hashes_buffer,
+ None,
+ batch_size,
+ offset,
+ probe_indices_buffer,
+ build_indices_buffer,
+ )?;
+
+ if !build_indices.is_empty() {
+ let mut null_bitmap = null_bitmap.lock();
+
+ for build_idx in build_indices.iter() {
+ let build_idx = build_idx
+ .expect("scope lookup should produce non-null build indices");
+ null_bitmap.set_bit(map_position_to_build_row(build_idx) as usize, true);
+ }
+ }
+
+ let Some(next_offset) = next_offset else {
+ break;
+ };
+ offset = next_offset;
+ }
+
+ Ok(())
+}
+
impl Stream for HashJoinStream {
type Item = Result<RecordBatch>;
diff --git a/datafusion/physical-plan/src/joins/symmetric_hash_join.rs b/datafusion/physical-plan/src/joins/symmetric_hash_join.rs
index aec693e839..99a12796c6 100644
--- a/datafusion/physical-plan/src/joins/symmetric_hash_join.rs
+++ b/datafusion/physical-plan/src/joins/symmetric_hash_join.rs
@@ -1240,6 +1240,7 @@ pub(crate) fn build_side_determined_results(
column_indices,
build_hash_joiner.build_side,
join_type,
+ None,
)
.map(|batch| (batch.num_rows() > 0).then_some(batch))
} else {
@@ -1343,6 +1344,7 @@ pub(crate) fn join_with_probe_batch(
column_indices,
build_hash_joiner.build_side,
join_type,
+ None,
)
.map(|batch| (batch.num_rows() > 0).then_some(batch))
}
diff --git a/datafusion/physical-plan/src/joins/utils.rs b/datafusion/physical-plan/src/joins/utils.rs
index 4f98b649e6..798f22b1d5 100644
--- a/datafusion/physical-plan/src/joins/utils.rs
+++ b/datafusion/physical-plan/src/joins/utils.rs
@@ -310,7 +310,8 @@ pub fn build_join_schema(
JoinType::LeftSemi | JoinType::LeftAnti => left_fields().unzip(),
JoinType::LeftMark => {
let right_field = once((
- Field::new("mark", DataType::Boolean, false),
+ // Nullable: null-aware `LeftMark` joins use NULL for SQL UNKNOWN.
+ Field::new("mark", DataType::Boolean, true),
ColumnIndex {
index: 0,
side: JoinSide::None,
@@ -321,7 +322,8 @@ pub fn build_join_schema(
JoinType::RightSemi | JoinType::RightAnti => right_fields().unzip(),
JoinType::RightMark => {
let left_field = once((
- Field::new("mark", DataType::Boolean, false),
+ // Nullable for symmetry with `LeftMark`; never actually NULL.
+ Field::new("mark", DataType::Boolean, true),
ColumnIndex {
index: 0,
side: JoinSide::None,
@@ -1276,6 +1278,7 @@ pub(crate) fn apply_join_filter_to_indices(
filter.column_indices(),
build_side,
join_type,
+ None,
)?;
let filter_result = filter
.expression()
@@ -1298,6 +1301,7 @@ pub(crate) fn apply_join_filter_to_indices(
filter.column_indices(),
build_side,
join_type,
+ None,
)?;
filter
@@ -1339,6 +1343,7 @@ pub(crate) fn build_batch_from_indices(
column_indices: &[ColumnIndex],
build_side: JoinSide,
join_type: JoinType,
+ mark_column: Option<&ArrayRef>,
) -> Result<RecordBatch> {
if schema.fields().is_empty() {
// For RightAnti and RightSemi joins, after `adjust_indices_by_join_type`
@@ -1358,8 +1363,12 @@ pub(crate) fn build_batch_from_indices(
for column_index in column_indices {
let array = if column_index.side == JoinSide::None {
- // For mark joins, the mark column is a true if the indices is not null, otherwise it will be false
- Arc::new(compute::is_not_null(probe_indices)?)
+ // For mark joins, callers can provide a custom mark column. Otherwise,
+ // matched rows are `true` and unmatched rows are `false`.
+ match mark_column {
+ Some(mark_col) => Arc::clone(mark_col),
+ None => Arc::new(compute::is_not_null(probe_indices)?),
+ }
} else if column_index.side == build_side {
let array = build_input_buffer.column(column_index.index);
if array.is_empty() || build_indices.null_count() == build_indices.len() {
@@ -1386,6 +1395,66 @@ pub(crate) fn build_batch_from_indices(
Ok(RecordBatch::try_new(Arc::new(schema.clone()), columns)?)
}
+/// Builds the nullable mark column for a null-aware `LeftMark` join.
+///
+/// This follows the left mark hash join described in Neumann, Leis, and Kemper,
+/// "The Complete Story of Joins (in HyPer)", Section 5.6:
+/// <https://www.cs.cmu.edu/~15721-f24/papers/Story_of_Joins.pdf>
+///
+/// `build_indices` and `probe_indices` are the final aligned indices derived from the
+/// visited bitmap. At this point:
+/// - valid `probe_indices` mean the build row matched at least one probe row, so the mark is `TRUE`
+/// - null `probe_indices` mean the build row was unmatched, so the result depends on SQL
+/// three-valued logic
+///
+/// For the uncorrelated single-key implementation, unmatched rows are classified as follows:
+/// 1. if the build key is `NULL` and the probe side is non-empty, the mark is `NULL`
+/// 2. if the build key is `NULL` and the probe side is empty, the mark is `FALSE`
+/// 3. if the build key is non-null and the probe side contained a `NULL`, the mark is `NULL`
+/// 4. otherwise, the mark is `FALSE`
+///
+/// For correlated scalar `NOT IN`, `null_indices_bitmap` carries the same UNKNOWN
+/// decision per build row, scoped by the correlated equality keys.
+///
+/// This is the helper equivalent of the paper's "null bucket" and `hadNull` handling.
+/// It is intentionally scoped to scalar null-aware mark joins.
+pub(crate) fn build_null_aware_left_mark_column(
+ build_indices: &UInt64Array,
+ probe_indices: &UInt32Array,
+ build_key_column: &dyn Array,
+ null_indices_bitmap: Option<&BooleanBufferBuilder>,
+ probe_side_has_null: bool,
+ probe_side_non_empty: bool,
+) -> ArrayRef {
+ // Whether an unmatched build row's mark is NULL (UNKNOWN) instead of FALSE:
+ // correlated joins precomputed this per row in `null_indices_bitmap`; the
+ // uncorrelated rules are cases 1-4 in the doc above.
+ let unmatched_mark_is_null = |build_idx: usize| match null_indices_bitmap {
+ Some(bitmap) => bitmap.get_bit(build_idx),
+ None if build_key_column.is_null(build_idx) => probe_side_non_empty,
+ None => probe_side_has_null,
+ };
+
+ let marks: BooleanArray = build_indices
+ .iter()
+ .zip(probe_indices.iter())
+ .map(|(build_idx, probe_idx)| {
+ if probe_idx.is_some() {
+ return Some(true);
+ }
+ let build_idx = build_idx
+ .expect("LeftMark final indices should always contain build-side rows")
+ as usize;
+ if unmatched_mark_is_null(build_idx) {
+ None
+ } else {
+ Some(false)
+ }
+ })
+ .collect();
+ Arc::new(marks)
+}
+
/// Returns a new [RecordBatch] for a probe batch when no probe row can find a
/// match: the build-side map is empty, either because the build side has no
/// rows or because none of its rows has a matchable (non-NULL) join key.
diff --git a/datafusion/sqllogictest/test_files/null_aware_anti_join.slt b/datafusion/sqllogictest/test_files/null_aware_anti_join.slt
index 4beab6998d..3276af2c96 100644
--- a/datafusion/sqllogictest/test_files/null_aware_anti_join.slt
+++ b/datafusion/sqllogictest/test_files/null_aware_anti_join.slt
@@ -162,16 +162,6 @@ WHERE id NOT IN (SELECT id FROM inner_table_no_null)
## Test 9: Multiple NOT IN conditions (OR)
#############
-# KNOWN LIMITATION: Mark joins used for OR conditions don't support null-aware semantics.
-# The NULL row is incorrectly returned here. According to SQL semantics:
-# - NULL NOT IN (2, 4) = UNKNOWN
-# - NULL NOT IN (1, 3) = UNKNOWN
-# - UNKNOWN OR UNKNOWN = UNKNOWN (should be filtered out)
-# But mark joins treat NULL keys as non-matching (FALSE), so:
-# - NULL mark column = FALSE
-# - NOT FALSE OR NOT FALSE = TRUE OR TRUE = TRUE (incorrectly included)
-# TODO: Implement null-aware support for mark joins to fix this
-
query IT rowsort
SELECT * FROM outer_table
WHERE id NOT IN (SELECT id FROM inner_table_no_null)
@@ -181,7 +171,6 @@ WHERE id NOT IN (SELECT id FROM inner_table_no_null)
2 b
3 c
4 d
-NULL e
#############
## Test 10: NOT IN with WHERE clause in subquery
diff --git a/datafusion/sqllogictest/test_files/null_aware_mark_join.slt b/datafusion/sqllogictest/test_files/null_aware_mark_join.slt
new file mode 100644
index 0000000000..62c0dd3192
--- /dev/null
+++ b/datafusion/sqllogictest/test_files/null_aware_mark_join.slt
@@ -0,0 +1,583 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+statement ok
+CREATE TABLE outer_table(id INT, value TEXT) AS VALUES
+(1, 'a'),
+(2, 'b'),
+(3, 'c'),
+(4, 'd'),
+(NULL, 'e');
+
+statement ok
+CREATE TABLE inner_table_no_null(id INT) AS VALUES
+(2),
+(4);
+
+statement ok
+CREATE TABLE inner_table_with_null(id INT) AS VALUES
+(2),
+(NULL);
+
+statement ok
+CREATE TABLE outer_corr_table(id INT, grp INT, value TEXT) AS VALUES
+(1, 1, 'a'),
+(3, 1, 'b'),
+(1, 2, 'c'),
+(NULL, 1, 'd'),
+(5, 3, 'e'),
+(2, 1, 'f'),
+(NULL, 2, 'g');
+
+statement ok
+CREATE TABLE inner_corr_table(id INT, grp INT) AS VALUES
+(2, 1),
+(NULL, 1),
+(1, 2);
+
+# Every row has a NULL join key, so the full-key hash map is empty.
+statement ok
+CREATE TABLE outer_corr_all_null_keys(id INT, grp INT, value TEXT) AS VALUES
+(NULL, 1, 'd'),
+(NULL, 2, 'g'),
+(7, NULL, 'z');
+
+statement ok
+CREATE TABLE empty_table(id INT) AS
+SELECT *
+FROM (VALUES (1)) AS seed(id)
+WHERE id < 0;
+
+#############################
+## Hash join null-aware mark
+#############################
+
+statement ok
+set datafusion.execution.target_partitions = 2;
+
+statement ok
+set datafusion.optimizer.repartition_joins = true;
+
+statement ok
+set datafusion.optimizer.prefer_hash_join = true;
+
+query TT
+EXPLAIN SELECT id, value
+FROM outer_table
+WHERE (id NOT IN (SELECT id FROM inner_table_with_null)) IS NULL;
+----
+logical_plan
+01)Projection: outer_table.id, outer_table.value
+02)--Filter: NOT __correlated_sq_1.mark IS NULL
+03)----LeftMark Join: outer_table.id = __correlated_sq_1.id null_aware
+04)------TableScan: outer_table projection=[id, value]
+05)------SubqueryAlias: __correlated_sq_1
+06)--------TableScan: inner_table_with_null projection=[id]
+physical_plan
+01)FilterExec: NOT mark@2 IS NULL, projection=[id@0, value@1]
+02)--RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1
+03)----HashJoinExec: mode=CollectLeft, join_type=LeftMark, on=[(id@0, id@0)], null_aware
+04)------DataSourceExec: partitions=1, partition_sizes=[1]
+05)------DataSourceExec: partitions=1, partition_sizes=[1]
+
+query T rowsort
+SELECT value
+FROM outer_table
+WHERE (id NOT IN (SELECT id FROM inner_table_with_null)) IS NULL;
+----
+a
+c
+d
+e
+
+query TT
+EXPLAIN SELECT value
+FROM outer_table
+WHERE (id NOT IN (SELECT id FROM inner_table_no_null)) IS NULL;
+----
+logical_plan
+01)Projection: outer_table.value
+02)--Filter: NOT __correlated_sq_1.mark IS NULL
+03)----Projection: outer_table.value, __correlated_sq_1.mark
+04)------LeftMark Join: outer_table.id = __correlated_sq_1.id null_aware
+05)--------TableScan: outer_table projection=[id, value]
+06)--------SubqueryAlias: __correlated_sq_1
+07)----------TableScan: inner_table_no_null projection=[id]
+physical_plan
+01)FilterExec: NOT mark@1 IS NULL, projection=[value@0]
+02)--RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1
+03)----HashJoinExec: mode=CollectLeft, join_type=LeftMark, on=[(id@0, id@0)], projection=[value@1, mark@2], null_aware
+04)------DataSourceExec: partitions=1, partition_sizes=[1]
+05)------DataSourceExec: partitions=1, partition_sizes=[1]
+
+query T rowsort
+SELECT value
+FROM outer_table
+WHERE (id NOT IN (SELECT id FROM inner_table_no_null)) IS NULL;
+----
+e
+
+query T rowsort
+SELECT value
+FROM outer_table
+WHERE (id NOT IN (SELECT id FROM empty_table)) IS TRUE;
+----
+a
+b
+c
+d
+e
+
+# Regression test for correlated NOT IN inside a mark join.
+query T rowsort
+SELECT value
+FROM outer_corr_table
+WHERE (id NOT IN (
+ SELECT id
+ FROM inner_corr_table
+ WHERE inner_corr_table.grp = outer_corr_table.grp
+)) IS NULL;
+----
+a
+b
+d
+g
+
+query T rowsort
+SELECT value
+FROM outer_corr_table
+WHERE (id NOT IN (
+ SELECT id
+ FROM inner_corr_table
+ WHERE inner_corr_table.grp = outer_corr_table.grp
+)) IS TRUE;
+----
+e
+
+# Regression test: the empty full-key map must not skip UNKNOWN marking.
+# `z` has a NULL scope key, so its subquery is empty and NOT IN is TRUE.
+query T rowsort
+SELECT value
+FROM outer_corr_all_null_keys
+WHERE (id NOT IN (
+ SELECT id
+ FROM inner_corr_table
+ WHERE inner_corr_table.grp = outer_corr_all_null_keys.grp
+)) IS NULL;
+----
+d
+g
+
+query T rowsort
+SELECT value
+FROM outer_corr_all_null_keys
+WHERE (id NOT IN (
+ SELECT id
+ FROM inner_corr_table
+ WHERE inner_corr_table.grp = outer_corr_all_null_keys.grp
+)) IS TRUE;
+----
+z
+
+query T rowsort
+SELECT value
+FROM outer_corr_all_null_keys
+WHERE (id NOT IN (
+ SELECT id
+ FROM inner_corr_table
+ WHERE inner_corr_table.grp = outer_corr_all_null_keys.grp
+)) IS FALSE;
+----
+
+# WHERE EXISTS in a disjunction, the motivating mark join example from
+# "The Complete Story of Joins (in HyPer)" (Neumann, Leis, Kemper; BTW 2017),
+# Section 3.3. The OR prevents the semi join rewrite, so the EXISTS must run
+# as a mark join. Unlike the null-aware NOT IN marks above, an EXISTS mark is
+# two-valued: NULLs on either side of the comparison produce FALSE, not NULL.
+
+query TT
+EXPLAIN SELECT value
+FROM outer_table
+WHERE EXISTS (
+ SELECT *
+ FROM inner_table_with_null
+ WHERE inner_table_with_null.id = outer_table.id
+) OR value = 'e';
+----
+logical_plan
+01)Projection: outer_table.value
+02)--Filter: __correlated_sq_1.mark OR outer_table.value = Utf8View("e")
+03)----Projection: outer_table.value, __correlated_sq_1.mark
+04)------LeftMark Join: outer_table.id = __correlated_sq_1.id
+05)--------TableScan: outer_table projection=[id, value]
+06)--------SubqueryAlias: __correlated_sq_1
+07)----------TableScan: inner_table_with_null projection=[id]
+physical_plan
+01)FilterExec: mark@1 OR value@0 = e, projection=[value@0]
+02)--RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1
+03)----HashJoinExec: mode=CollectLeft, join_type=RightMark, on=[(id@0, id@0)], projection=[value@1, mark@2]
+04)------DataSourceExec: partitions=1, partition_sizes=[1]
+05)------DataSourceExec: partitions=1, partition_sizes=[1]
+
+query T rowsort
+SELECT value
+FROM outer_table
+WHERE EXISTS (
+ SELECT *
+ FROM inner_table_with_null
+ WHERE inner_table_with_null.id = outer_table.id
+) OR value = 'e';
+----
+b
+e
+
+# If the EXISTS mark were wrongly null-aware, every unmatched outer row would
+# get a NULL mark (the build side contains a NULL), NOT mark would stay NULL,
+# and all rows except 'b' would disappear.
+query TT
+EXPLAIN SELECT value
+FROM outer_table
+WHERE NOT EXISTS (
+ SELECT *
+ FROM inner_table_with_null
+ WHERE inner_table_with_null.id = outer_table.id
+) OR value = 'b';
+----
+logical_plan
+01)Projection: outer_table.value
+02)--Filter: NOT __correlated_sq_1.mark OR outer_table.value = Utf8View("b")
+03)----Projection: outer_table.value, __correlated_sq_1.mark
+04)------LeftMark Join: outer_table.id = __correlated_sq_1.id
+05)--------TableScan: outer_table projection=[id, value]
+06)--------SubqueryAlias: __correlated_sq_1
+07)----------TableScan: inner_table_with_null projection=[id]
+physical_plan
+01)FilterExec: NOT mark@1 OR value@0 = b, projection=[value@0]
+02)--RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1
+03)----HashJoinExec: mode=CollectLeft, join_type=RightMark, on=[(id@0, id@0)], projection=[value@1, mark@2]
+04)------DataSourceExec: partitions=1, partition_sizes=[1]
+05)------DataSourceExec: partitions=1, partition_sizes=[1]
+
+query T rowsort
+SELECT value
+FROM outer_table
+WHERE NOT EXISTS (
+ SELECT *
+ FROM inner_table_with_null
+ WHERE inner_table_with_null.id = outer_table.id
+) OR value = 'b';
+----
+a
+b
+c
+d
+e
+
+# In a filter the paper's = ANY form works today: rewrite_set_comparison
+# expands it to a CASE over EXISTS marks, and IS NULL / IS TRUE then observe
+# all three mark values.
+
+query TT
+EXPLAIN SELECT value
+FROM outer_table
+WHERE (id = ANY (SELECT id FROM inner_table_with_null)) IS NULL;
+----
+logical_plan
+01)Projection: outer_table.value
+02)--Filter: __correlated_sq_1.mark OR __correlated_sq_2.mark AND NOT __correlated_sq_3.mark AND Boolean(NULL) IS NULL
+03)----Projection: outer_table.value, __correlated_sq_1.mark, __correlated_sq_2.mark, __correlated_sq_3.mark
+04)------LeftMark Join: Filter: outer_table.id = __correlated_sq_3.id IS TRUE
+05)--------LeftMark Join: Filter: outer_table.id = __correlated_sq_2.id IS NULL
+06)----------LeftMark Join: Filter: outer_table.id = __correlated_sq_1.id IS TRUE
+07)------------TableScan: outer_table projection=[id, value]
+08)------------SubqueryAlias: __correlated_sq_1
+09)--------------TableScan: inner_table_with_null projection=[id]
+10)----------SubqueryAlias: __correlated_sq_2
+11)------------TableScan: inner_table_with_null projection=[id]
+12)--------SubqueryAlias: __correlated_sq_3
+13)----------TableScan: inner_table_with_null projection=[id]
+physical_plan
+01)FilterExec: mark@1 OR mark@2 AND NOT mark@3 AND NULL IS NULL, projection=[value@0]
+02)--NestedLoopJoinExec: join_type=RightMark, filter=(id@0 = id@1) IS NOT DISTINCT FROM true, projection=[value@1, mark@2, mark@3, mark@4]
+03)----DataSourceExec: partitions=1, partition_sizes=[1]
+04)----NestedLoopJoinExec: join_type=RightMark, filter=id@0 = id@1 IS NULL
+05)------DataSourceExec: partitions=1, partition_sizes=[1]
+06)------RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1
+07)--------NestedLoopJoinExec: join_type=RightMark, filter=(id@0 = id@1) IS NOT DISTINCT FROM true
+08)----------DataSourceExec: partitions=1, partition_sizes=[1]
+09)----------DataSourceExec: partitions=1, partition_sizes=[1]
+
+query T rowsort
+SELECT value
+FROM outer_table
+WHERE (id = ANY (SELECT id FROM inner_table_with_null)) IS NULL;
+----
+a
+c
+d
+e
+
+query TT
+EXPLAIN SELECT value
+FROM outer_table
+WHERE (id = ANY (SELECT id FROM inner_table_no_null)) IS TRUE;
+----
+logical_plan
+01)Projection: outer_table.value
+02)--Filter: __correlated_sq_1.mark OR __correlated_sq_2.mark AND NOT __correlated_sq_3.mark AND Boolean(NULL) IS TRUE
+03)----Projection: outer_table.value, __correlated_sq_1.mark, __correlated_sq_2.mark, __correlated_sq_3.mark
+04)------LeftMark Join: Filter: outer_table.id = __correlated_sq_3.id IS TRUE
+05)--------LeftMark Join: Filter: outer_table.id = __correlated_sq_2.id IS NULL
+06)----------LeftMark Join: Filter: outer_table.id = __correlated_sq_1.id IS TRUE
+07)------------TableScan: outer_table projection=[id, value]
+08)------------SubqueryAlias: __correlated_sq_1
+09)--------------TableScan: inner_table_no_null projection=[id]
+10)----------SubqueryAlias: __correlated_sq_2
+11)------------TableScan: inner_table_no_null projection=[id]
+12)--------SubqueryAlias: __correlated_sq_3
+13)----------TableScan: inner_table_no_null projection=[id]
+physical_plan
+01)FilterExec: (mark@1 OR mark@2 AND NOT mark@3 AND NULL) IS NOT DISTINCT FROM true, projection=[value@0]
+02)--NestedLoopJoinExec: join_type=RightMark, filter=(id@0 = id@1) IS NOT DISTINCT FROM true, projection=[value@1, mark@2, mark@3, mark@4]
+03)----DataSourceExec: partitions=1, partition_sizes=[1]
+04)----NestedLoopJoinExec: join_type=RightMark, filter=id@0 = id@1 IS NULL
+05)------DataSourceExec: partitions=1, partition_sizes=[1]
+06)------RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1
+07)--------NestedLoopJoinExec: join_type=RightMark, filter=(id@0 = id@1) IS NOT DISTINCT FROM true
+08)----------DataSourceExec: partitions=1, partition_sizes=[1]
+09)----------DataSourceExec: partitions=1, partition_sizes=[1]
+
+query T rowsort
+SELECT value
+FROM outer_table
+WHERE (id = ANY (SELECT id FROM inner_table_no_null)) IS TRUE;
+----
+b
+d
+
+# Same plan shape as the `= ANY (... with_null) IS NULL` case above; only the
+# result differs, so assert just the result here.
+query T rowsort
+SELECT value
+FROM outer_table
+WHERE (id = ANY (SELECT id FROM inner_table_no_null)) IS NULL;
+----
+e
+
+# Positive IN goes through InSubquery decorrelation rather than the = ANY
+# CASE rewrite; make sure its mark is null-aware too.
+query TT
+EXPLAIN SELECT value
+FROM outer_table
+WHERE (id IN (SELECT id FROM inner_table_with_null)) IS NULL;
+----
+logical_plan
+01)Projection: outer_table.value
+02)--Filter: __correlated_sq_1.mark IS NULL
+03)----Projection: outer_table.value, __correlated_sq_1.mark
+04)------LeftMark Join: outer_table.id = __correlated_sq_1.id null_aware
+05)--------TableScan: outer_table projection=[id, value]
+06)--------SubqueryAlias: __correlated_sq_1
+07)----------TableScan: inner_table_with_null projection=[id]
+physical_plan
+01)FilterExec: mark@1 IS NULL, projection=[value@0]
+02)--RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1
+03)----HashJoinExec: mode=CollectLeft, join_type=LeftMark, on=[(id@0, id@0)], projection=[value@1, mark@2], null_aware
+04)------DataSourceExec: partitions=1, partition_sizes=[1]
+05)------DataSourceExec: partitions=1, partition_sizes=[1]
+
+query T rowsort
+SELECT value
+FROM outer_table
+WHERE (id IN (SELECT id FROM inner_table_with_null)) IS NULL;
+----
+a
+c
+d
+e
+
+# Null-aware mark joins are pinned to CollectLeft hash joins (mark joins have no
+# sort-merge path); flipping prefer_hash_join off must not drop null awareness.
+
+statement ok
+set datafusion.optimizer.prefer_hash_join = false;
+
+query TT
+EXPLAIN SELECT id, value
+FROM outer_table
+WHERE (id NOT IN (SELECT id FROM inner_table_with_null)) IS NULL;
+----
+logical_plan
+01)Projection: outer_table.id, outer_table.value
+02)--Filter: NOT __correlated_sq_1.mark IS NULL
+03)----LeftMark Join: outer_table.id = __correlated_sq_1.id null_aware
+04)------TableScan: outer_table projection=[id, value]
+05)------SubqueryAlias: __correlated_sq_1
+06)--------TableScan: inner_table_with_null projection=[id]
+physical_plan
+01)FilterExec: NOT mark@2 IS NULL, projection=[id@0, value@1]
+02)--RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1
+03)----HashJoinExec: mode=CollectLeft, join_type=LeftMark, on=[(id@0, id@0)], null_aware
+04)------DataSourceExec: partitions=1, partition_sizes=[1]
+05)------DataSourceExec: partitions=1, partition_sizes=[1]
+
+####################################
+## Nested loop mark join with NULLs
+####################################
+
+statement ok
+set datafusion.execution.target_partitions = 1;
+
+statement ok
+set datafusion.optimizer.prefer_hash_join = true;
+
+query TT
+EXPLAIN SELECT value
+FROM outer_table
+WHERE (EXISTS (
+ SELECT 1
+ FROM inner_table_no_null
+ WHERE outer_table.id < inner_table_no_null.id
+)) IS TRUE;
+----
+logical_plan
+01)Projection: outer_table.value
+02)--Filter: __correlated_sq_1.mark IS TRUE
+03)----Projection: outer_table.value, __correlated_sq_1.mark
+04)------LeftMark Join: Filter: outer_table.id < __correlated_sq_1.id
+05)--------TableScan: outer_table projection=[id, value]
+06)--------SubqueryAlias: __correlated_sq_1
+07)----------TableScan: inner_table_no_null projection=[id]
+physical_plan
+01)FilterExec: mark@1 IS NOT DISTINCT FROM true, projection=[value@0]
+02)--NestedLoopJoinExec: join_type=RightMark, filter=id@0 < id@1, projection=[value@1, mark@2]
+03)----DataSourceExec: partitions=1, partition_sizes=[1]
+04)----DataSourceExec: partitions=1, partition_sizes=[1]
+
+query T rowsort
+SELECT value
+FROM outer_table
+WHERE (EXISTS (
+ SELECT 1
+ FROM inner_table_no_null
+ WHERE outer_table.id < inner_table_no_null.id
+)) IS TRUE;
+----
+a
+b
+c
+
+##########################################################
+## Dynamic filter pushdown into a null-aware mark join
+##########################################################
+
+# Regression test: with NOT NULL build-side keys, join dynamic filter pushdown
+# (on by default) stays enabled for a correlated null-aware mark join, and the
+# pushed filter must keep NULL-valued probe rows (`id IS NULL OR ...`) so
+# UNKNOWN marks survive. The probe side must be parquet-backed: MemTable scans
+# reject pushed filters and never exercise this path.
+
+statement ok
+COPY (SELECT * FROM (VALUES (1, 1, 'a'), (2, 1, 'b'), (3, 2, 'c'), (4, 3, 'd')) AS t(id, grp, value))
+TO 'test_files/scratch/null_aware_mark_join/outer.parquet'
+STORED AS PARQUET;
+
+statement ok
+COPY (SELECT * FROM (VALUES (2, 1), (NULL, 1), (1, 2)) AS t(id, grp))
+TO 'test_files/scratch/null_aware_mark_join/inner.parquet'
+STORED AS PARQUET;
+
+statement ok
+CREATE EXTERNAL TABLE outer_pushdown_table (id BIGINT NOT NULL, grp BIGINT NOT NULL, value VARCHAR)
+STORED AS PARQUET
+LOCATION 'test_files/scratch/null_aware_mark_join/outer.parquet';
+
+statement ok
+CREATE EXTERNAL TABLE inner_pushdown_table (id BIGINT, grp BIGINT)
+STORED AS PARQUET
+LOCATION 'test_files/scratch/null_aware_mark_join/inner.parquet';
+
+query TT
+EXPLAIN SELECT value
+FROM outer_pushdown_table
+WHERE id NOT IN (
+ SELECT id
+ FROM inner_pushdown_table
+ WHERE inner_pushdown_table.grp = outer_pushdown_table.grp
+) OR value = 'zzz';
+----
+logical_plan
+01)Projection: outer_pushdown_table.value
+02)--Filter: NOT __correlated_sq_1.mark OR outer_pushdown_table.value = Utf8View("zzz")
+03)----Projection: outer_pushdown_table.value, __correlated_sq_1.mark
+04)------LeftMark Join: outer_pushdown_table.id = __correlated_sq_1.id, outer_pushdown_table.grp = __correlated_sq_1.grp null_aware
+05)--------TableScan: outer_pushdown_table projection=[id, grp, value]
+06)--------SubqueryAlias: __correlated_sq_1
+07)----------TableScan: inner_pushdown_table projection=[id, grp]
+physical_plan
+01)FilterExec: NOT mark@1 OR value@0 = zzz, projection=[value@0]
+02)--HashJoinExec: mode=CollectLeft, join_type=LeftMark, on=[(id@0, id@0), (grp@1, grp@1)], projection=[value@2, mark@3], null_aware
+03)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/null_aware_mark_join/outer.parquet]]}, projection=[id, grp, value], file_type=parquet
+04)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/null_aware_mark_join/inner.parquet]]}, projection=[id, grp], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible
+
+# `a` (grp 1 = {2, NULL}) is UNKNOWN and `b` matches, so only `c` (no match in
+# grp 2) and `d` (empty grp 3) pass. A dropped NULL probe row would wrongly
+# emit `a`.
+query T rowsort
+SELECT value
+FROM outer_pushdown_table
+WHERE id NOT IN (
+ SELECT id
+ FROM inner_pushdown_table
+ WHERE inner_pushdown_table.grp = outer_pushdown_table.grp
+) OR value = 'zzz';
+----
+c
+d
+
+statement ok
+DROP TABLE outer_pushdown_table;
+
+statement ok
+DROP TABLE inner_pushdown_table;
+
+statement ok
+reset datafusion.optimizer.prefer_hash_join;
+
+statement ok
+reset datafusion.optimizer.repartition_joins;
+
+statement ok
+set datafusion.execution.target_partitions = 4;
+
+statement ok
+DROP TABLE empty_table;
+
+statement ok
+DROP TABLE inner_table_with_null;
+
+statement ok
+DROP TABLE inner_table_no_null;
+
+statement ok
+DROP TABLE inner_corr_table;
+
+statement ok
+DROP TABLE outer_corr_table;
+
+statement ok
+DROP TABLE outer_table;
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org