diff --git a/datafusion/core/src/physical_planner.rs b/datafusion/core/src/physical_planner.rs
index 198e5bb7412b2..5ef99206f533a 100644
--- a/datafusion/core/src/physical_planner.rs
+++ b/datafusion/core/src/physical_planner.rs
@@ -65,7 +65,7 @@ use datafusion_common::{
     internal_err, not_impl_err, plan_err,
 };
 use datafusion_common::{
-    TableReference, assert_eq_or_internal_err, assert_or_internal_err,
+    NullEquality, TableReference, assert_eq_or_internal_err, assert_or_internal_err,
 };
 use datafusion_datasource::file_groups::FileGroup;
 use datafusion_datasource::memory::MemorySourceConfig;
@@ -1721,17 +1721,18 @@ impl DefaultPhysicalPlanner {
 
                 let prefer_hash_join =
                     session.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.
+                // Null-aware hash joins require CollectLeft for global NULL tracking,
+                // so they must not be hash-repartitioned.
+                // Supported null-aware LeftAnti joins can instead use the
+                // single-partition sort-merge path below.
                 let can_repartition_join = session.config().target_partitions() > 1
                     && session.config().repartition_joins()
                     && !*null_aware;
 
-                // Only `HashJoinExec` implements null-aware semantics, and it
-                // needs equi-join keys to do so. Without them the join would be
-                // planned as a nested loop (or piecewise merge) join, which
-                // silently ignores the flag and returns wrong results for
+                // Only `HashJoinExec` and the null-aware `LeftAnti` path in `SortMergeJoinExec`
+                // implements null-aware semantics, and those need equi-join keys to do so.
+                // Without them the join would be planned as a nested loop (or piecewise merge)
+                // join, which silently ignores the flag and returns wrong results for
                 // `NOT IN` over a nullable subquery. Fail loudly instead.
                 if *null_aware && join_on.is_empty() {
                     return plan_err!(
@@ -1863,16 +1864,44 @@ impl DefaultPhysicalPlanner {
                             None,
                         )?)
                     }
+                } else if *null_aware
+                    && !prefer_hash_join
+                    && *join_type == JoinType::LeftAnti
+                    && join_on.len() == 1
+                    && join_filter.is_none()
+                    && *null_equality == NullEquality::NullEqualsNothing
+                {
+                    // Use null-aware sort-merge join for single-key LeftAnti joins
+                    // without a filter. Single-partition inputs and NULLS FIRST ordering
+                    // let the stream determine whether the entire RHS is empty or
+                    // contains NULL before emitting any rows.
+                    Arc::new(
+                        SortMergeJoinExec::try_new(
+                            physical_left,
+                            physical_right,
+                            join_on,
+                            join_filter,
+                            *join_type,
+                            vec![SortOptions::new(false, true)], // ASC NULLS FIRST
+                            *null_equality,
+                        )?
+                        .with_null_aware(true)?,
+                    )
                 } else if session.config().target_partitions() > 1
                     && session.config().repartition_joins()
                     && !prefer_hash_join
                     && !*null_aware
-                // Null-aware joins (e.g. `NOT IN` with a nullable subquery) must
-                // use the CollectLeft HashJoin below: SortMergeJoinExec does not
-                // implement null-aware anti-join semantics and would return wrong
-                // results when the right side contains a null join key.
+                // Null-aware joins (e.g. `NOT IN` with a nullable subquery) that do not
+                // satisfy the conditions above must use the CollectLeft HashJoin below.
+                // SortMergeJoinExec currently supports null-aware semantics only for
+                // single-key LeftAnti joins without a filter, using NullEqualsNothing.
+                // Routing other null-aware joins through the ordinary sort-merge path
+                // would not preserve NOT IN semantics, including when the subquery
+                // contains a NULL key.
                 {
-                    // Use SortMergeJoin if hash join is not preferred
+                    // Use ordinary sort-merge join when hash join is not preferred.
+                    // Null-aware joins not handled above fall back to CollectLeft
+                    // HashJoin because their semantics are not supported by sort-merge join.
                     let join_on_len = join_on.len();
                     Arc::new(SortMergeJoinExec::try_new(
                         physical_left,
@@ -1884,7 +1913,7 @@ impl DefaultPhysicalPlanner {
                         *null_equality,
                     )?)
                 } else {
-                    // Null-aware joins need global probe-side state, so keep
+                    // Remaining null-aware joins need global probe-side state, so keep
                     // them in CollectLeft mode.
                     let partition_mode = if can_repartition_join {
                         PartitionMode::Auto
diff --git a/datafusion/core/tests/physical_optimizer/test_utils.rs b/datafusion/core/tests/physical_optimizer/test_utils.rs
index b93281def20fb..146e4a0d7c54a 100644
--- a/datafusion/core/tests/physical_optimizer/test_utils.rs
+++ b/datafusion/core/tests/physical_optimizer/test_utils.rs
@@ -176,6 +176,7 @@ pub fn create_test_schema3() -> Result<SchemaRef> {
     Ok(schema)
 }
 
+/// Creates a sort-merge join with null-aware semantics disabled by default.
 pub fn sort_merge_join_exec(
     left: Arc<dyn ExecutionPlan>,
     right: Arc<dyn ExecutionPlan>,
diff --git a/datafusion/physical-optimizer/src/ensure_requirements/enforce_distribution.rs b/datafusion/physical-optimizer/src/ensure_requirements/enforce_distribution.rs
index b847e195c07b3..5e176e7140fb6 100644
--- a/datafusion/physical-optimizer/src/ensure_requirements/enforce_distribution.rs
+++ b/datafusion/physical-optimizer/src/ensure_requirements/enforce_distribution.rs
@@ -184,17 +184,17 @@ pub fn adjust_input_keys_ordering(
         requirements.children[1].data =
             shift_right_required(&requirements.data, left_columns_len)
                 .unwrap_or_default();
-    } else if let Some(SortMergeJoinExec {
-        left,
-        right,
-        on,
-        filter,
-        join_type,
-        sort_options,
-        null_equality,
-        ..
-    }) = plan.downcast_ref::<SortMergeJoinExec>()
-    {
+    } else if let Some(origin_exec) = plan.downcast_ref::<SortMergeJoinExec>() {
+        let SortMergeJoinExec {
+            left,
+            right,
+            on,
+            filter,
+            join_type,
+            sort_options,
+            null_equality,
+            ..
+        } = origin_exec;
         let join_constructor = |new_conditions: (
             Vec<(PhysicalExprRef, PhysicalExprRef)>,
             Vec<SortOptions>,
@@ -208,6 +208,7 @@ pub fn adjust_input_keys_ordering(
                 new_conditions.1,
                 *null_equality,
             )
+            .and_then(|join| join.with_null_aware(origin_exec.null_aware()))
             .map(|e| Arc::new(e) as _)
         };
         return reorder_partitioned_join_keys(
@@ -469,17 +470,17 @@ pub fn reorder_join_keys_to_inputs(
                     .build_exec();
             }
         }
-    } else if let Some(SortMergeJoinExec {
-        left,
-        right,
-        on,
-        filter,
-        join_type,
-        sort_options,
-        null_equality,
-        ..
-    }) = plan.downcast_ref::<SortMergeJoinExec>()
-    {
+    } else if let Some(origin_exec) = plan.downcast_ref::<SortMergeJoinExec>() {
+        let SortMergeJoinExec {
+            left,
+            right,
+            on,
+            filter,
+            join_type,
+            sort_options,
+            null_equality,
+            ..
+        } = origin_exec;
         let (join_keys, positions) = reorder_current_join_keys(
             extract_join_keys(on),
             Some(left.output_partitioning()),
@@ -507,6 +508,7 @@ pub fn reorder_join_keys_to_inputs(
                 new_sort_options,
                 *null_equality,
             )
+            .and_then(|join| join.with_null_aware(origin_exec.null_aware()))
             .map(|smj| Arc::new(smj) as _);
         }
     }
@@ -687,6 +689,7 @@ fn reorder_join_keys_to_range_inputs(
             new_sort_options,
             join.null_equality,
         )?
+        .with_null_aware(join.null_aware())?
         .with_projection(join.projection.clone())?;
         return Ok(Arc::new(reordered) as _);
     }
diff --git a/datafusion/physical-plan/src/joins/sort_merge_join/bitwise_stream.rs b/datafusion/physical-plan/src/joins/sort_merge_join/bitwise_stream.rs
index 5c1dd58d51c5f..e313594d6b07c 100644
--- a/datafusion/physical-plan/src/joins/sort_merge_join/bitwise_stream.rs
+++ b/datafusion/physical-plan/src/joins/sort_merge_join/bitwise_stream.rs
@@ -110,6 +110,7 @@ use std::cmp::Ordering;
 use std::sync::Arc;
 
 use crate::EmptyRecordBatchStream;
+use crate::joins::sort_merge_join::exec::NullAwareMode;
 use crate::joins::utils::{JoinFilter, JoinKeyComparator};
 use crate::metrics::{
     BaselineMetrics, Count, ExecutionPlanMetricsSet, Gauge, MetricBuilder, Time,
@@ -120,7 +121,7 @@ use crate::stream::{ObservedStream, RecordBatchStreamAdapter};
 use arrow::array::{
     Array, ArrayRef, BooleanArray, BooleanBufferBuilder, RecordBatch, RecordBatchOptions,
 };
-use arrow::compute::{BatchCoalescer, SortOptions, filter_record_batch, not};
+use arrow::compute::{BatchCoalescer, SortOptions, and, filter_record_batch, not};
 use arrow::datatypes::SchemaRef;
 use arrow::util::bit_chunk_iterator::UnalignedBitChunk;
 use arrow::util::bit_util::apply_bitwise_binary_op;
@@ -187,6 +188,48 @@ fn find_key_group_end(cmp: &JoinKeyComparator, from: usize, len: usize) -> usize
     lo
 }
 
+/// Applies SQL NOT IN semantics to the unmatched outer rows.
+fn null_aware_left_anti_selection(
+    unmatched: BooleanArray,
+    outer_key: &dyn Array,
+    subquery_state: SubqueryState,
+) -> Result<BooleanArray> {
+    match subquery_state {
+        SubqueryState::Empty => Ok(unmatched),
+        SubqueryState::NonEmptyNoNull => match outer_key.logical_nulls() {
+            // All outer keys are non-NULL, so unmatched rows can be kept.
+            None => Ok(unmatched),
+            Some(nulls) => {
+                // With a nonempty RHS containing no NULL keys, NOT IN keeps only
+                // unmatched outer rows whose keys are non-NULL. Combine the
+                // unmatched mask with the outer key validity bitmap to exclude
+                // NULL keys, for which NOT IN evaluates to UNKNOWN.
+                let valids = BooleanArray::new(nulls.inner().clone(), None);
+                and(&unmatched, &valids).map_err(DataFusionError::from)
+            }
+        },
+        SubqueryState::Pending | SubqueryState::HasNull => {
+            internal_err!("Cannot emit null-aware anti rows before RHS validation")
+        }
+    }
+}
+
+/// State of the right-side subquery used by a null-aware `LeftAnti` join.
+///
+/// Determined before emitting left rows, using a single right partition
+/// sorted with NULLS FIRST.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum SubqueryState {
+    /// The right input has not been inspected yet.
+    Pending,
+    /// The right input contains no rows.
+    Empty,
+    /// The right input contains rows but no NULL keys.
+    NonEmptyNoNull,
+    /// The right input contains at least one NULL key.
+    HasNull,
+}
+
 /// Sort-Merge join stream for Semi/Anti/Mark joins.
 ///
 /// Named "bitwise" because it tracks outer-row matches via a per-batch
@@ -235,6 +278,8 @@ pub(crate) struct BitwiseSortMergeJoinStream {
     filter: Option<JoinFilter>,
     sort_options: Vec<SortOptions>,
     null_equality: NullEquality,
+    null_aware: Option<NullAwareMode>,
+    subquery_state: SubqueryState,
     // Decomposed from JoinType: when RightSemi/RightAnti, outer=right,
     // inner=left, so we swap sides when building the filter batch.
     outer_is_left: bool,
@@ -274,11 +319,20 @@ pub(crate) struct BitwiseSortMergeJoinStream {
 }
 
 impl BitwiseSortMergeJoinStream {
+    /// Creates a sort-merge join stream for semi, anti, and mark joins.
+    ///
+    /// # Preconditions
+    ///
+    /// If `null_aware` is `Some`, the mode and join configuration must satisfy
+    /// the requirements of [`NullAwareMode::try_new`].
+    /// `SortMergeJoinExec::execute` performs this validation before creating
+    /// the stream.
     #[expect(clippy::too_many_arguments)]
     pub fn try_new(
         schema: SchemaRef,
         sort_options: Vec<SortOptions>,
         null_equality: NullEquality,
+        null_aware: Option<NullAwareMode>,
         outer: SendableRecordBatchStream,
         inner: SendableRecordBatchStream,
         on_outer: Vec<PhysicalExprRef>,
@@ -334,6 +388,8 @@ impl BitwiseSortMergeJoinStream {
             filter,
             sort_options,
             null_equality,
+            null_aware,
+            subquery_state: SubqueryState::Pending,
             outer_is_left,
             coalescer: BatchCoalescer::new(Arc::clone(&schema), batch_size)
                 .with_biggest_coalesce_batch_size(Some(batch_size / 2)),
@@ -575,7 +631,15 @@ impl BitwiseSortMergeJoinStream {
                 }
             }
             JoinType::LeftAnti | JoinType::RightAnti => {
-                let selection = not(&BooleanArray::new(matched_buf, None))?;
+                let unmatched = not(&BooleanArray::new(matched_buf, None))?;
+                let selection = match self.null_aware {
+                    None => unmatched,
+                    Some(NullAwareMode::LeftAnti) => null_aware_left_anti_selection(
+                        unmatched,
+                        self.outer_key_arrays[0].as_ref(),
+                        self.subquery_state,
+                    )?,
+                };
                 let filtered = filter_record_batch(&batch, &selection)?;
                 if filtered.num_rows() > 0 {
                     self.coalescer.push_batch(filtered)?;
@@ -1092,12 +1156,56 @@ impl BitwiseSortMergeJoinStream {
         }
     }
 
+    /// Initializes the inner input's empty/NULL summary for null-aware joins.
+    async fn initialize_null_aware(&mut self) -> Result<()> {
+        if self.subquery_state != SubqueryState::Pending {
+            return Ok(());
+        }
+
+        match self.null_aware {
+            None => return Ok(()),
+            Some(NullAwareMode::LeftAnti) => {}
+        }
+
+        if !self.next_inner_batch().await? {
+            self.subquery_state = SubqueryState::Empty;
+            return Ok(());
+        }
+
+        // Keep the first nonempty inner batch at offset 0 for the merge scan.
+        // The complete right input must be sorted with NULLS FIRST.
+        let first_key_is_null = self.inner_key_arrays[0]
+            .logical_nulls()
+            .is_some_and(|null_keys| null_keys.is_null(0));
+
+        self.subquery_state = if first_key_is_null {
+            SubqueryState::HasNull
+        } else {
+            SubqueryState::NonEmptyNoNull
+        };
+
+        Ok(())
+    }
+
     /// Main loop: a classic merge-scan over the two sorted inputs, emitting
     /// output batches as they complete.
     async fn join(
         &mut self,
         emitter: &mut TryEmitter<RecordBatch, DataFusionError>,
     ) -> Result<()> {
+        self.initialize_null_aware().await?;
+        match self.null_aware {
+            None => {}
+            Some(NullAwareMode::LeftAnti)
+                if self.subquery_state == SubqueryState::HasNull =>
+            {
+                // NOT IN cannot be TRUE for any outer row when the inner
+                // input contains a NULL key. (following NOT IN semantics)
+                return Ok(());
+            }
+            Some(NullAwareMode::LeftAnti) => {}
+        }
+
         // The `has_current_*` / `has_completed_batch` fast paths keep async
         // state machinery out of the per-key-group hot path; the awaiting
         // helpers are only entered at batch boundaries.
diff --git a/datafusion/physical-plan/src/joins/sort_merge_join/exec.rs b/datafusion/physical-plan/src/joins/sort_merge_join/exec.rs
index 6b345c86d903c..1cdb506b562ca 100644
--- a/datafusion/physical-plan/src/joins/sort_merge_join/exec.rs
+++ b/datafusion/physical-plan/src/joins/sort_merge_join/exec.rs
@@ -51,6 +51,61 @@ use datafusion_physical_expr_common::physical_expr::{PhysicalExprRef, fmt_sql};
 use datafusion_physical_expr_common::sort_expr::{LexOrdering, OrderingRequirements};
 use futures::StreamExt;
 
+/// Null-aware (`NOT IN`) semantics of a sort merge join, derived from
+/// [`SortMergeJoinExec::null_aware`] and the join type.
+///
+/// Only these combinations are legal (see [`Self::try_new`]), so the
+/// stream matches on this instead of re-checking `null_aware && join_type == ..`.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) enum NullAwareMode {
+    /// `left.key NOT IN (right.key)`: emits left rows.
+    LeftAnti,
+}
+
+impl NullAwareMode {
+    /// Validates that `null_aware` may be set for this join and returns its mode.
+    pub(crate) fn try_new(
+        join_type: JoinType,
+        num_keys: usize,
+        has_filter: bool,
+        null_equality: NullEquality,
+        sort_options: &[SortOptions],
+    ) -> Result<Self> {
+        if num_keys != 1 || has_filter {
+            return plan_err!("null-aware SortMergeJoin requires one key and no filter");
+        }
+
+        // NULL keys must not match each other.
+        // The null-aware execution path handles UNKNOWN results separately.
+        // UNKNOWN happens when `WHERE x NOT IN (NULL)`
+        if null_equality != NullEquality::NullEqualsNothing {
+            return plan_err!("null-aware SortMergeJoin requires NullEqualsNothing");
+        }
+
+        let mode = match join_type {
+            JoinType::LeftAnti => Self::LeftAnti,
+            _ => {
+                return plan_err!(
+                    "null-aware SortMergeJoin only supports LeftAnti, got {join_type}"
+                );
+            }
+        };
+
+        match mode {
+            Self::LeftAnti => {
+                // With a single right-side partition and NULLS FIRST,
+                // the first row reveals whether the subquery contains NULL.
+                if sort_options.len() != 1 || !sort_options[0].nulls_first {
+                    return plan_err!(
+                        "null-aware {join_type} SortMergeJoin requires NULLS FIRST ordering"
+                    );
+                }
+                Ok(mode)
+            }
+        }
+    }
+}
+
 /// Join execution plan that executes equi-join predicates on multiple partitions using Sort-Merge
 /// join algorithm and applies an optional filter post join. Can be used to join arbitrarily large
 /// inputs where one or both of the inputs don't fit in the available memory.
@@ -121,6 +176,19 @@ pub struct SortMergeJoinExec {
     pub sort_options: Vec<SortOptions>,
     /// Defines the null equality for the join.
     pub null_equality: NullEquality,
+    /// Flag to indicate if this join uses null-aware equality semantics.
+    ///
+    /// Currently supported only for uncorrelated `LeftAnti` joins with one
+    /// join key, no join filter, and [`NullEquality::NullEqualsNothing`].
+    ///
+    /// Both inputs must have a single partition and be sorted with NULLS FIRST.
+    /// This lets the join determine whether the entire right input is empty
+    /// or contains a NULL key before emitting any left rows.
+    ///
+    /// An empty right input preserves all left rows, including NULL keys.
+    /// A right input containing a NULL key produces no rows.
+    /// Otherwise, only unmatched left rows with non-NULL keys are emitted.
+    pub null_aware: bool,
     /// The columns of `schema` to emit, in order. `None` emits all of them.
     pub projection: Option<Vec<usize>>,
     /// Cache holding plan properties like equivalences, output partitioning etc.
@@ -130,6 +198,9 @@ pub struct SortMergeJoinExec {
 impl SortMergeJoinExec {
     /// Tries to create a new [SortMergeJoinExec].
     /// The inputs are sorted using `sort_options` are applied to the columns in the `on`
+    ///
+    /// Null-aware semantics are disabled by default. Enable them with
+    /// [`Self::with_null_aware`].
     /// # Error
     /// This function errors when it is not possible to join the left and right sides on keys `on`.
     pub fn try_new(
@@ -195,11 +266,23 @@ impl SortMergeJoinExec {
             right_sort_exprs,
             sort_options,
             null_equality,
+            null_aware: false,
             projection: None,
             cache: Arc::new(cache),
         })
     }
 
+    /// Enables or disables null-aware (`NOT IN`) semantics.
+    ///
+    /// Enabling requires an uncorrelated `LeftAnti` join with one key, no
+    /// filter, [`NullEquality::NullEqualsNothing`], and NULLS FIRST ordering.
+    /// Both inputs must have a single partition when executed.
+    pub fn with_null_aware(mut self, null_aware: bool) -> Result<Self> {
+        self.null_aware = null_aware;
+        self.null_aware_mode()?;
+        Ok(self)
+    }
+
     /// Returns this join emitting only the columns in `projection`, in that order.
     /// The indices address the join's own schema, before any projection.
     pub fn with_projection(&self, projection: Option<Vec<usize>>) -> Result<Self> {
@@ -293,6 +376,11 @@ impl SortMergeJoinExec {
         self.null_equality
     }
 
+    /// Return whether this join uses null-aware (NOT IN) semantics
+    pub fn null_aware(&self) -> bool {
+        self.null_aware
+    }
+
     /// This function creates the cache object that stores the plan properties such as schema, equivalence properties, ordering, partitioning, etc.
     fn compute_properties(
         left: &Arc<dyn ExecutionPlan>,
@@ -364,6 +452,7 @@ impl SortMergeJoinExec {
             self.sort_options.clone(),
             self.null_equality,
         )?
+        .with_null_aware(self.null_aware)?
         .with_projection(swap_join_projection(
             left.schema().fields().len(),
             right.schema().fields().len(),
@@ -388,6 +477,23 @@ impl SortMergeJoinExec {
             reorder_output_after_swap(Arc::new(new_join), &left.schema(), &right.schema())
         }
     }
+
+    /// The null-aware semantics of this join, if [`Self::null_aware`] is set.
+    ///
+    /// Errors if `null_aware` is set on a join that does not support it.
+    pub(super) fn null_aware_mode(&self) -> Result<Option<NullAwareMode>> {
+        self.null_aware
+            .then(|| {
+                NullAwareMode::try_new(
+                    self.join_type,
+                    self.on.len(),
+                    self.filter.is_some(),
+                    self.null_equality(),
+                    self.sort_options(),
+                )
+            })
+            .transpose()
+    }
 }
 
 impl EmbeddedProjection for SortMergeJoinExec {
@@ -412,6 +518,11 @@ impl DisplayAs for SortMergeJoinExec {
                     } else {
                         ""
                     };
+                let display_null_aware = if self.null_aware() {
+                    ", null_aware"
+                } else {
+                    ""
+                };
                 let display_projections = match &self.projection {
                     Some(projection) => format!(
                         ", projection=[{}]",
@@ -429,7 +540,7 @@ impl DisplayAs for SortMergeJoinExec {
                 };
                 write!(
                     f,
-                    "{}: join_type={:?}, on=[{}]{}{}{}",
+                    "{}: join_type={:?}, on=[{}]{}{}{}{}",
                     Self::static_name(),
                     self.join_type,
                     on,
@@ -438,6 +549,7 @@ impl DisplayAs for SortMergeJoinExec {
                         |f| format!(", filter={}", f.expression())
                     ),
                     display_null_equality,
+                    display_null_aware,
                     display_projections,
                 )
             }
@@ -460,6 +572,10 @@ impl DisplayAs for SortMergeJoinExec {
                     writeln!(f, "NullsEqual: true")?;
                 }
 
+                if self.null_aware() {
+                    writeln!(f, "null_aware")?;
+                }
+
                 Ok(())
             }
         }
@@ -480,6 +596,18 @@ impl ExecutionPlan for SortMergeJoinExec {
     }
 
     fn input_distribution_requirements(&self) -> InputDistributionRequirements {
+        // Null-aware NOT IN needs facts about the entire right input.
+        // Require a single partition on both sides to use the existing
+        // sort-merge join's partition pairing without cross-partition coordination.
+        //
+        // TODO: Allow multiple left partitions by sharing the right-side summary
+        // and providing each left partition with the right input for matching.
+        if self.null_aware {
+            return InputDistributionRequirements::new(vec![
+                Distribution::SinglePartition,
+                Distribution::SinglePartition,
+            ]);
+        }
         let (left_expr, right_expr) = self
             .on
             .iter()
@@ -543,6 +671,7 @@ impl ExecutionPlan for SortMergeJoinExec {
                         self.sort_options.clone(),
                         self.null_equality,
                     )?
+                    .with_null_aware(self.null_aware)?
                     .with_projection(self.projection.clone())?,
                 )),
                 _ => internal_err!("SortMergeJoin wrong number of children"),
@@ -583,6 +712,20 @@ impl ExecutionPlan for SortMergeJoinExec {
             "Invalid SortMergeJoinExec, partition count mismatch {left_partitions}!={right_partitions},\
                  consider using RepartitionExec"
         );
+
+        let null_aware = self.null_aware_mode()?;
+
+        // Null-aware execution requires a single partition on both sides.
+        // Validate this at execution time because callers may bypass the
+        // optimizer or modify the public null_aware flag after planning.
+        if null_aware.is_some()
+            && (partition != 0 || left_partitions != 1 || right_partitions != 1)
+        {
+            return internal_err!(
+                "null-aware SortMergeJoin requires partition 0 and single-partition inputs"
+            );
+        }
+
         // execute children plans
         let left = self.left.execute(partition, Arc::clone(&context))?;
         let right = self.right.execute(partition, Arc::clone(&context))?;
@@ -593,6 +736,7 @@ impl ExecutionPlan for SortMergeJoinExec {
                 schema: Arc::clone(&self.schema),
                 sort_options: self.sort_options.clone(),
                 null_equality: self.null_equality,
+                null_aware,
                 left,
                 right,
                 on_left,
@@ -686,15 +830,18 @@ impl ExecutionPlan for SortMergeJoinExec {
             self.filter().as_ref(),
             &column_indices,
         )? {
-            Ok(Some(Arc::new(SortMergeJoinExec::try_new(
-                Arc::new(projected_left_child),
-                Arc::new(projected_right_child),
-                join_on,
-                join_filter,
-                self.join_type,
-                self.sort_options.clone(),
-                self.null_equality,
-            )?)))
+            Ok(Some(Arc::new(
+                SortMergeJoinExec::try_new(
+                    Arc::new(projected_left_child),
+                    Arc::new(projected_right_child),
+                    join_on,
+                    join_filter,
+                    self.join_type,
+                    self.sort_options.clone(),
+                    self.null_equality,
+                )?
+                .with_null_aware(self.null_aware)?,
+            )))
         } else {
             try_embed_projection(projection, self)
         }
@@ -717,6 +864,7 @@ impl ExecutionPlan for SortMergeJoinExec {
             join_type,
             sort_options,
             null_equality,
+            null_aware,
             projection,
             // derived from the children's schemas by `try_new` on decode
             schema: _,
@@ -768,6 +916,7 @@ impl ExecutionPlan for SortMergeJoinExec {
                         filter,
                         sort_options,
                         null_equality: null_equality.into(),
+                        null_aware: *null_aware,
                         // Proto3 `repeated` cannot distinguish `None` from
                         // `Some(vec![])`. `Some(vec![])` (reachable via
                         // `try_embed_projection` for e.g. `SELECT count(1) … JOIN …`)
@@ -817,6 +966,7 @@ impl SortMergeJoinExec {
             filter,
             sort_options,
             null_equality,
+            null_aware,
             projection,
         } = &**sort_join;
 
@@ -851,6 +1001,7 @@ impl SortMergeJoinExec {
             *null_equality,
             "SortMergeJoinExec",
         )?;
+
         let filter = filter
             .as_ref()
             .map(|filter| {
@@ -886,6 +1037,7 @@ impl SortMergeJoinExec {
                 sort_options,
                 null_equality,
             )?
+            .with_null_aware(*null_aware)?
             .with_projection(projection)?,
         ))
     }
@@ -899,6 +1051,7 @@ pub(crate) struct SortMergeJoinInputs {
     /// Sort options of the join keys, one per key, that both inputs are sorted with
     pub(crate) sort_options: Vec<SortOptions>,
     pub(crate) null_equality: NullEquality,
+    pub(crate) null_aware: Option<NullAwareMode>,
     /// Left input, sorted on `on_left` with `sort_options`
     pub(crate) left: SendableRecordBatchStream,
     /// Right input, sorted on `on_right` with `sort_options`
@@ -927,6 +1080,7 @@ pub(crate) fn sort_merge_join_stream(
         schema,
         sort_options,
         null_equality,
+        null_aware,
         left,
         right,
         on_left,
@@ -970,6 +1124,7 @@ pub(crate) fn sort_merge_join_stream(
             schema,
             sort_options,
             null_equality,
+            null_aware,
             streamed,
             buffered,
             on_streamed,
diff --git a/datafusion/physical-plan/src/joins/sort_merge_join/tests.rs b/datafusion/physical-plan/src/joins/sort_merge_join/tests.rs
index 207113a7338db..babc9851a985c 100644
--- a/datafusion/physical-plan/src/joins/sort_merge_join/tests.rs
+++ b/datafusion/physical-plan/src/joins/sort_merge_join/tests.rs
@@ -24,19 +24,19 @@ use crate::projection::{ProjectionExec, ProjectionExpr};
 use crate::sorts::sort::SortExec;
 use crate::spill::spill_manager::SpillManager;
 use crate::test::TestMemoryExec;
+use crate::test::build_table_i32;
 use crate::test::exec::{BarrierExec, PanicExec};
-use crate::test::{build_table_i32, build_table_i32_two_cols};
 use crate::{ExecutionPlan, RecordBatchStream, common};
 use crate::{
     expressions::Column, joins::sort_merge_join::filter::get_corrected_filter_mask,
     joins::sort_merge_join::materializing_stream::JoinedRecordBatches,
 };
 use arrow::array::{
-    BinaryArray, BooleanArray, Date32Array, Date64Array, FixedSizeBinaryArray,
-    Int32Array, RecordBatch, UInt64Array,
+    ArrayRef, BinaryArray, BooleanArray, Date32Array, Date64Array, DictionaryArray,
+    FixedSizeBinaryArray, Int32Array, RecordBatch, UInt64Array,
 };
 use arrow::compute::{BatchCoalescer, SortOptions, filter_record_batch};
-use arrow::datatypes::{DataType, Field, Schema};
+use arrow::datatypes::{DataType, Field, Int32Type, Schema};
 use arrow_ord::sort::SortColumn;
 use arrow_schema::SchemaRef;
 use bytes::Bytes;
@@ -209,11 +209,47 @@ pub fn build_table_i32_nullable(
     TestMemoryExec::try_new_exec(&[vec![batch]], schema, None).unwrap()
 }
 
-pub fn build_table_two_cols(
-    a: (&str, &Vec<i32>),
-    b: (&str, &Vec<i32>),
+/// Builds a record batch with two nullable columns.
+fn build_batch_two_cols(a: (&str, ArrayRef), b: (&str, ArrayRef)) -> Result<RecordBatch> {
+    let schema = Arc::new(Schema::new(vec![
+        Field::new(a.0, a.1.data_type().clone(), true),
+        Field::new(b.0, b.1.data_type().clone(), true),
+    ]));
+    Ok(RecordBatch::try_new(schema, vec![a.1, b.1])?)
+}
+
+/// Builds a two-column batch whose join key is a dictionary.
+///
+/// All dictionary indices are physically valid. NULL dictionary values
+/// exercise logical NULL handling.
+fn build_batch_dict_key(
+    key_name: &str,
+    dict_values: Vec<Option<i32>>,
+    keys: Vec<i32>,
+    row_id_name: &str,
+    row_ids: Vec<i32>,
+) -> Result<RecordBatch> {
+    let dict = DictionaryArray::<Int32Type>::try_new(
+        Int32Array::from(keys),
+        Arc::new(Int32Array::from(dict_values)),
+    )?;
+
+    build_batch_two_cols(
+        (key_name, Arc::new(dict)),
+        (row_id_name, Arc::new(Int32Array::from(row_ids))),
+    )
+}
+
+/// Builds a single-batch table with two nullable Int32 columns.
+fn build_table_two_cols(
+    a: (&str, &[Option<i32>]),
+    b: (&str, &[Option<i32>]),
 ) -> Arc<dyn ExecutionPlan> {
-    let batch = build_table_i32_two_cols(a, b);
+    let batch = build_batch_two_cols(
+        (a.0, Arc::new(Int32Array::from(a.1.to_vec()))),
+        (b.0, Arc::new(Int32Array::from(b.1.to_vec()))),
+    )
+    .unwrap();
     let schema = batch.schema();
     TestMemoryExec::try_new_exec(&[vec![batch]], schema, None).unwrap()
 }
@@ -1246,8 +1282,8 @@ async fn join_right_different_columns_count_with_filter() -> Result<()> {
     );
 
     let right = build_table_two_cols(
-        ("a2", &vec![10, 20, 30]),
-        ("b1", &vec![4, 5, 6]), // 6 does not exist on the left
+        ("a2", &[Some(10), Some(20), Some(30)]),
+        ("b1", &[Some(4), Some(5), Some(6)]), // 6 does not exist on the left
     );
 
     let on = vec![(
@@ -1298,8 +1334,8 @@ async fn join_left_different_columns_count_with_filter() -> Result<()> {
     // left join t1 on t2.b1 = t1.b1 and t2.a2 > t1.a1
 
     let left = build_table_two_cols(
-        ("a2", &vec![10, 20, 30]),
-        ("b1", &vec![4, 5, 6]), // 6 does not exist on the right
+        ("a2", &[Some(10), Some(20), Some(30)]),
+        ("b1", &[Some(4), Some(5), Some(6)]), // 6 does not exist on the right
     );
 
     let right = build_table(
@@ -1358,8 +1394,8 @@ async fn join_left_with_filter_columns_right_before_left() -> Result<()> {
     // left join t1 on t2.b1 = t1.b1 and t2.a2 > t1.a1
 
     let left = build_table_two_cols(
-        ("a2", &vec![10, 20, 30]),
-        ("b1", &vec![4, 5, 6]), // 6 does not exist on the right
+        ("a2", &[Some(10), Some(20), Some(30)]),
+        ("b1", &[Some(4), Some(5), Some(6)]), // 6 does not exist on the right
     );
 
     let right = build_table(
@@ -1418,7 +1454,10 @@ async fn join_right_with_filter_columns_right_before_left() -> Result<()> {
     // from t2
     // right join t1 on t2.b1 = t1.b1 and t2.a2 > t1.a1
 
-    let left = build_table_two_cols(("a2", &vec![10, 20, 30]), ("b1", &vec![4, 5, 6]));
+    let left = build_table_two_cols(
+        ("a2", &[Some(10), Some(20), Some(30)]),
+        ("b1", &[Some(4), Some(5), Some(6)]),
+    );
 
     let right = build_table(
         ("a1", &vec![1, 21, 3]), // 20(t2.a2) > 21(t1.a1) is false
@@ -1550,8 +1589,8 @@ async fn join_left_mark_different_columns_count_with_filter() -> Result<()> {
     // left mark join t1 on t2.b1 = t1.b1 and t2.a2 > t1.a1
 
     let left = build_table_two_cols(
-        ("a2", &vec![10, 20, 30]),
-        ("b1", &vec![4, 5, 6]), // 6 does not exist on the right
+        ("a2", &[Some(10), Some(20), Some(30)]),
+        ("b1", &[Some(4), Some(5), Some(6)]), // 6 does not exist on the right
     );
 
     let right = build_table(
@@ -1652,8 +1691,8 @@ async fn join_right_mark_different_columns_count_with_filter() -> Result<()> {
     );
 
     let right = build_table_two_cols(
-        ("a2", &vec![10, 20, 30]),
-        ("b1", &vec![4, 5, 6]), // 6 does not exist on the left
+        ("a2", &[Some(10), Some(20), Some(30)]),
+        ("b1", &[Some(4), Some(5), Some(6)]), // 6 does not exist on the left
     );
 
     let on = vec![(
@@ -1761,6 +1800,313 @@ async fn join_left_anti() -> Result<()> {
     Ok(())
 }
 
+mod null_aware_tests {
+    use super::*;
+    use arrow::array::Array;
+
+    /// Runs a null-aware LeftAnti join and collects the surviving left-side row IDs.
+    /// Expects `key` as the first column and `row_id` as the second column.
+    async fn collect_left_anti_ids(
+        left_batches: Vec<RecordBatch>,
+        right_batches: Vec<RecordBatch>,
+        batch_size: usize,
+    ) -> Result<Vec<i32>> {
+        let left = build_table_from_batches(left_batches);
+        // Unlike build_table_from_batches, this must also accept zero batches.
+        let right = TestMemoryExec::try_new_exec(&[right_batches], left.schema(), None)?;
+        let on: JoinOn = vec![(
+            Arc::new(Column::new("key", 0)),
+            Arc::new(Column::new("key", 0)),
+        )];
+        // Direct execution does not insert sorts. Fixtures are globally sorted
+        // ASC NULLS FIRST, including across batch boundaries.
+        let join = SortMergeJoinExec::try_new(
+            left,
+            right,
+            on,
+            None,
+            LeftAnti,
+            vec![SortOptions::new(false, true)],
+            NullEquality::NullEqualsNothing,
+        )?
+        .with_null_aware(true)?;
+        let context = Arc::new(
+            TaskContext::default()
+                .with_session_config(SessionConfig::new().with_batch_size(batch_size)),
+        );
+        let batches = common::collect(join.execute(0, context)?).await?;
+        let mut row_ids = Vec::new();
+        for batch in batches {
+            let ids = batch
+                .column(1)
+                .as_any()
+                .downcast_ref::<Int32Array>()
+                .expect("Int32 row IDs");
+            row_ids.extend(ids.values().iter().copied());
+        }
+        Ok(row_ids)
+    }
+
+    #[tokio::test]
+    async fn empty_rhs_keeps_nulls_and_duplicates() -> Result<()> {
+        let left = vec![
+            build_batch_two_cols(
+                (
+                    "key",
+                    Arc::new(Int32Array::from(vec![None, Some(1), Some(2)])),
+                ),
+                ("row_id", Arc::new(Int32Array::from(vec![0, 1, 2]))),
+            )?,
+            build_batch_two_cols(
+                ("key", Arc::new(Int32Array::from(vec![Some(3), Some(3)]))),
+                ("row_id", Arc::new(Int32Array::from(vec![3, 4]))),
+            )?,
+        ];
+        // EOF without any batch and EOF after empty batches are both empty RHS.
+        for right in [
+            vec![],
+            vec![
+                build_batch_two_cols(
+                    ("key", Arc::new(Int32Array::from(Vec::<Option<i32>>::new()))),
+                    ("row_id", Arc::new(Int32Array::from(Vec::<i32>::new()))),
+                )?,
+                build_batch_two_cols(
+                    ("key", Arc::new(Int32Array::from(Vec::<Option<i32>>::new()))),
+                    ("row_id", Arc::new(Int32Array::from(Vec::<i32>::new()))),
+                )?,
+            ],
+        ] {
+            let actual = collect_left_anti_ids(left.clone(), right, 2).await?;
+            assert_eq!(actual, vec![0, 1, 2, 3, 4]);
+        }
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn rhs_null_after_empty_batch_with_null_lhs_returns_no_rows() -> Result<()> {
+        let left = vec![build_batch_two_cols(
+            (
+                "key",
+                Arc::new(Int32Array::from(vec![None, Some(1), Some(2)])),
+            ),
+            ("row_id", Arc::new(Int32Array::from(vec![0, 1, 2]))),
+        )?];
+        // Cover an all-NULL RHS and a NULL followed by a non-NULL batch.
+        for right in [
+            vec![
+                build_batch_two_cols(
+                    ("key", Arc::new(Int32Array::from(Vec::<Option<i32>>::new()))),
+                    ("row_id", Arc::new(Int32Array::from(Vec::<i32>::new()))),
+                )?,
+                build_batch_two_cols(
+                    ("key", Arc::new(Int32Array::from(vec![None]))),
+                    ("row_id", Arc::new(Int32Array::from(vec![10]))),
+                )?,
+            ],
+            vec![
+                build_batch_two_cols(
+                    ("key", Arc::new(Int32Array::from(Vec::<Option<i32>>::new()))),
+                    ("row_id", Arc::new(Int32Array::from(Vec::<i32>::new()))),
+                )?,
+                build_batch_two_cols(
+                    ("key", Arc::new(Int32Array::from(vec![None]))),
+                    ("row_id", Arc::new(Int32Array::from(vec![10]))),
+                )?,
+                build_batch_two_cols(
+                    ("key", Arc::new(Int32Array::from(vec![Some(2)]))),
+                    ("row_id", Arc::new(Int32Array::from(vec![11]))),
+                )?,
+            ],
+        ] {
+            let actual = collect_left_anti_ids(left.clone(), right, 2).await?;
+            assert!(actual.is_empty(), "unexpected rows: {actual:?}");
+        }
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn rhs_null_after_empty_batch_without_null_lhs_returns_no_rows() -> Result<()> {
+        let left = vec![build_batch_two_cols(
+            ("key", Arc::new(Int32Array::from(vec![Some(1), Some(2)]))),
+            ("row_id", Arc::new(Int32Array::from(vec![0, 1]))),
+        )?];
+        // Cover an all-NULL RHS and a NULL followed by a non-NULL batch.
+        for right in [
+            vec![
+                build_batch_two_cols(
+                    ("key", Arc::new(Int32Array::from(Vec::<Option<i32>>::new()))),
+                    ("row_id", Arc::new(Int32Array::from(Vec::<i32>::new()))),
+                )?,
+                build_batch_two_cols(
+                    ("key", Arc::new(Int32Array::from(vec![None]))),
+                    ("row_id", Arc::new(Int32Array::from(vec![10]))),
+                )?,
+            ],
+            vec![
+                build_batch_two_cols(
+                    ("key", Arc::new(Int32Array::from(Vec::<Option<i32>>::new()))),
+                    ("row_id", Arc::new(Int32Array::from(Vec::<i32>::new()))),
+                )?,
+                build_batch_two_cols(
+                    ("key", Arc::new(Int32Array::from(vec![None]))),
+                    ("row_id", Arc::new(Int32Array::from(vec![10]))),
+                )?,
+                build_batch_two_cols(
+                    ("key", Arc::new(Int32Array::from(vec![Some(2)]))),
+                    ("row_id", Arc::new(Int32Array::from(vec![11]))),
+                )?,
+            ],
+        ] {
+            let actual = collect_left_anti_ids(left.clone(), right, 2).await?;
+            assert!(actual.is_empty(), "unexpected rows: {actual:?}");
+        }
+        Ok(())
+    }
+
+    // Initialization reads the first nonempty RHS batch to check for NULL keys.
+    // The merge scan must reuse that batch: skipping its sole key (2) would
+    // incorrectly retain both left rows with key 2.
+    //
+    // Left key 2 and right key 4 span batch boundaries, with empty batches
+    // interspersed to exercise skipping them without treating them as EOF.
+    // Both left rows with key 3 must survive. Run with batch sizes 1 and 2 to
+    // verify the same result when output is emitted in small batches.
+    //
+    // Also covers a nonempty RHS without NULL keys: retain only unmatched,
+    // non-NULL left keys, preserving duplicate rows.
+    #[tokio::test]
+    async fn startup_preserves_first_rhs_batch_across_key_groups() -> Result<()> {
+        let left = vec![
+            build_batch_two_cols(
+                (
+                    "key",
+                    Arc::new(Int32Array::from(vec![None, Some(1), Some(2)])),
+                ),
+                ("row_id", Arc::new(Int32Array::from(vec![0, 1, 2]))),
+            )?,
+            build_batch_two_cols(
+                ("key", Arc::new(Int32Array::from(Vec::<Option<i32>>::new()))),
+                ("row_id", Arc::new(Int32Array::from(Vec::<i32>::new()))),
+            )?,
+            build_batch_two_cols(
+                (
+                    "key",
+                    Arc::new(Int32Array::from(vec![Some(2), Some(3), Some(3)])),
+                ),
+                ("row_id", Arc::new(Int32Array::from(vec![3, 4, 5]))),
+            )?,
+            build_batch_two_cols(
+                ("key", Arc::new(Int32Array::from(vec![Some(4), Some(5)]))),
+                ("row_id", Arc::new(Int32Array::from(vec![6, 7]))),
+            )?,
+        ];
+        let right = vec![
+            build_batch_two_cols(
+                ("key", Arc::new(Int32Array::from(Vec::<Option<i32>>::new()))),
+                ("row_id", Arc::new(Int32Array::from(Vec::<i32>::new()))),
+            )?,
+            build_batch_two_cols(
+                ("key", Arc::new(Int32Array::from(vec![Some(2)]))),
+                ("row_id", Arc::new(Int32Array::from(vec![10]))),
+            )?,
+            build_batch_two_cols(
+                ("key", Arc::new(Int32Array::from(Vec::<Option<i32>>::new()))),
+                ("row_id", Arc::new(Int32Array::from(Vec::<i32>::new()))),
+            )?,
+            build_batch_two_cols(
+                ("key", Arc::new(Int32Array::from(vec![Some(4)]))),
+                ("row_id", Arc::new(Int32Array::from(vec![11]))),
+            )?,
+            build_batch_two_cols(
+                ("key", Arc::new(Int32Array::from(vec![Some(4)]))),
+                ("row_id", Arc::new(Int32Array::from(vec![12]))),
+            )?,
+        ];
+        for batch_size in [1, 2] {
+            let actual =
+                collect_left_anti_ids(left.clone(), right.clone(), batch_size).await?;
+            // Keep 1, 3, 3, 5; both occurrences of 2 must match the first RHS batch.
+            assert_eq!(actual, vec![1, 4, 5, 7], "batch size {batch_size}");
+        }
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn exhausted_rhs_drains_remaining_left_batches() -> Result<()> {
+        let left = vec![
+            build_batch_two_cols(
+                ("key", Arc::new(Int32Array::from(vec![None, Some(1)]))),
+                ("row_id", Arc::new(Int32Array::from(vec![0, 1]))),
+            )?,
+            build_batch_two_cols(
+                ("key", Arc::new(Int32Array::from(vec![Some(2)]))),
+                ("row_id", Arc::new(Int32Array::from(vec![2]))),
+            )?,
+            build_batch_two_cols(
+                ("key", Arc::new(Int32Array::from(vec![Some(3), Some(3)]))),
+                ("row_id", Arc::new(Int32Array::from(vec![3, 4]))),
+            )?,
+            build_batch_two_cols(
+                ("key", Arc::new(Int32Array::from(vec![Some(4), Some(5)]))),
+                ("row_id", Arc::new(Int32Array::from(vec![5, 6]))),
+            )?,
+        ];
+        let right = vec![build_batch_two_cols(
+            ("key", Arc::new(Int32Array::from(vec![Some(2)]))),
+            ("row_id", Arc::new(Int32Array::from(vec![10]))),
+        )?];
+        let actual = collect_left_anti_ids(left, right, 2).await?;
+        // Exclude the initial NULL and retain unmatched rows after RHS EOF,
+        // including duplicates and the final partial output batch.
+        assert_eq!(actual, vec![1, 3, 4, 5, 6]);
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn dictionary_value_nulls_follow_not_in_semantics() -> Result<()> {
+        let left = build_batch_dict_key(
+            "key",
+            vec![None, Some(1), Some(2), Some(3)],
+            vec![0, 1, 2, 3, 3],
+            "row_id",
+            vec![0, 1, 2, 3, 4],
+        )?;
+        assert_eq!(left.column(0).null_count(), 0);
+        assert_eq!(left.column(0).logical_null_count(), 1);
+
+        // The same logical key (2) has a different dictionary index on each side.
+        let rhs_with_null = build_batch_dict_key(
+            "key",
+            vec![None, Some(2)],
+            vec![0, 1],
+            "row_id",
+            vec![10, 11],
+        )?;
+        assert_eq!(rhs_with_null.column(0).null_count(), 0);
+        assert_eq!(rhs_with_null.column(0).logical_null_count(), 1);
+
+        let cases = [
+            (vec![], vec![0, 1, 2, 3, 4]),
+            (
+                vec![build_batch_dict_key(
+                    "key",
+                    vec![Some(2)],
+                    vec![0],
+                    "row_id",
+                    vec![10],
+                )?],
+                vec![1, 3, 4],
+            ),
+            (vec![rhs_with_null], vec![]),
+        ];
+        for (right, expected) in cases {
+            let actual = collect_left_anti_ids(vec![left.clone()], right, 2).await?;
+            assert_eq!(actual, expected);
+        }
+        Ok(())
+    }
+}
+
 #[tokio::test]
 async fn join_right_anti_one_one() -> Result<()> {
     let left = build_table(
@@ -1768,7 +2114,10 @@ async fn join_right_anti_one_one() -> Result<()> {
         ("b1", &vec![4, 5, 5]),
         ("c1", &vec![7, 8, 8]),
     );
-    let right = build_table_two_cols(("a2", &vec![10, 20, 30]), ("b1", &vec![4, 5, 6]));
+    let right = build_table_two_cols(
+        ("a2", &[Some(10), Some(20), Some(30)]),
+        ("b1", &[Some(4), Some(5), Some(6)]),
+    );
     let on = vec![(
         Arc::new(Column::new_with_schema("b1", &left.schema())?) as _,
         Arc::new(Column::new_with_schema("b1", &right.schema())?) as _,
@@ -1820,7 +2169,10 @@ async fn join_right_anti_two_two() -> Result<()> {
         ("b1", &vec![4, 5, 5]),
         ("c1", &vec![7, 8, 8]),
     );
-    let right = build_table_two_cols(("a2", &vec![10, 20, 30]), ("b1", &vec![4, 5, 6]));
+    let right = build_table_two_cols(
+        ("a2", &[Some(10), Some(20), Some(30)]),
+        ("b1", &[Some(4), Some(5), Some(6)]),
+    );
     let on = vec![
         (
             Arc::new(Column::new_with_schema("a1", &left.schema())?) as _,
@@ -1931,7 +2283,10 @@ async fn join_right_anti_two_with_filter() -> Result<()> {
 
 #[tokio::test]
 async fn join_right_anti_filtered_with_mismatched_columns() -> Result<()> {
-    let left = build_table_two_cols(("a1", &vec![31, 31]), ("b1", &vec![32, 33]));
+    let left = build_table_two_cols(
+        ("a1", &[Some(31), Some(31)]),
+        ("b1", &[Some(32), Some(33)]),
+    );
     let right = build_table(
         ("a2", &vec![31, 31]),
         ("b2", &vec![32, 35]),
@@ -2867,8 +3222,14 @@ async fn join_full_multiple_batches() -> Result<()> {
 async fn join_full_null_filter_result() -> Result<()> {
     // Left: (a, b) all non-null, sorted on a
     let left = build_table_two_cols(
-        ("a1", &vec![1, 1, 2, 2, 3, 3]),
-        ("b1", &vec![1, 2, 1, 2, 1, 2]),
+        (
+            "a1",
+            &[Some(1), Some(1), Some(2), Some(2), Some(3), Some(3)],
+        ),
+        (
+            "b1",
+            &[Some(1), Some(2), Some(1), Some(2), Some(1), Some(2)],
+        ),
     );
 
     // Right: (a, b) with b nullable, sorted on a
@@ -5316,7 +5677,7 @@ async fn join_key_type_batch_boundaries(
     #[values(1, 2)] input_batch_size: usize,
 ) -> Result<()> {
     // Single-digit keys so Utf8 dictionary values sort like the integers.
-    let keys = |unmatched_key| -> Result<arrow::array::ArrayRef> {
+    let keys = |unmatched_key| -> Result<ArrayRef> {
         let keys = arrow::array::Int64Array::from(vec![
             None,
             Some(1),
@@ -5350,15 +5711,15 @@ async fn join_key_type_batch_boundaries(
 /// batches. Row 0 must be null, row 4 must differ between the sides, and all
 /// remaining rows must match.
 async fn assert_key_batch_boundaries(
-    left_keys: arrow::array::ArrayRef,
-    right_keys: arrow::array::ArrayRef,
+    left_keys: ArrayRef,
+    right_keys: ArrayRef,
     join_type: JoinType,
     null_equality: NullEquality,
     descending: bool,
     input_batch_size: usize,
 ) -> Result<()> {
     let key_type = left_keys.data_type().clone();
-    let input = |mut keys: arrow::array::ArrayRef| -> Result<Arc<dyn ExecutionPlan>> {
+    let input = |mut keys: ArrayRef| -> Result<Arc<dyn ExecutionPlan>> {
         let mut ids: Vec<i32> = (0..keys.len() as i32).collect();
         if descending {
             ids.reverse();
@@ -5371,10 +5732,7 @@ async fn assert_key_batch_boundaries(
             )?;
         }
         let batch = RecordBatch::try_from_iter(vec![
-            (
-                "id",
-                Arc::new(Int32Array::from(ids)) as arrow::array::ArrayRef,
-            ),
+            ("id", Arc::new(Int32Array::from(ids)) as ArrayRef),
             ("key", keys),
         ])?;
         Ok(build_table_from_batches(
@@ -5601,6 +5959,7 @@ async fn bitwise_emits_completed_batches_before_pending_outer(
         Arc::clone(&schema),
         vec![SortOptions::default()],
         NullEquality::NullEqualsNothing,
+        None,
         outer,
         inner,
         vec![Arc::new(Column::new("key", 1))],
@@ -5700,6 +6059,7 @@ fn join_time_test_join(
         outer_schema,
         vec![SortOptions::default()],
         NullEquality::NullEqualsNothing,
+        None,
         outer,
         inner,
         vec![Arc::new(Column::new("b1", 1)) as PhysicalExprRef],
@@ -6085,6 +6445,7 @@ async fn filter_buffer_pending_loses_inner_rows() -> Result<()> {
         left_schema, // output schema = outer schema for semi
         vec![SortOptions::default()],
         NullEquality::NullEqualsNothing,
+        None,
         outer,
         inner,
         on_outer,
@@ -6179,6 +6540,7 @@ async fn no_filter_boundary_pending_loses_outer_rows() -> Result<()> {
         left_schema,
         vec![SortOptions::default()],
         NullEquality::NullEqualsNothing,
+        None,
         outer,
         inner,
         on_outer,
@@ -6270,6 +6632,7 @@ async fn no_filter_boundary_pending_with_unmatched_prefix() -> Result<()> {
             Arc::clone(&left_schema),
             vec![SortOptions::default()],
             NullEquality::NullEqualsNothing,
+            None,
             outer,
             inner,
             on_outer.clone(),
@@ -6390,6 +6753,7 @@ async fn filtered_boundary_pending_outer_rows() -> Result<()> {
         left_schema,
         vec![SortOptions::default()],
         NullEquality::NullEqualsNothing,
+        None,
         outer,
         inner,
         on_outer,
@@ -7135,6 +7499,7 @@ async fn spill_filtered_boundary_loses_outer_rows() -> Result<()> {
             Arc::clone(&left_schema),
             vec![SortOptions::default()],
             NullEquality::NullEqualsNothing,
+            None,
             outer,
             inner,
             on_outer.clone(),
diff --git a/datafusion/proto-models/proto/datafusion.proto b/datafusion/proto-models/proto/datafusion.proto
index 6c4ef8f0a4b2f..3fde7c45b4f28 100644
--- a/datafusion/proto-models/proto/datafusion.proto
+++ b/datafusion/proto-models/proto/datafusion.proto
@@ -1835,6 +1835,12 @@ message SortMergeJoinExecNode {
   repeated SortExprNode sort_options = 6;
   datafusion_common.NullEquality null_equality = 7;
   repeated uint32 projection = 8;
+  // Enables null-aware (`NOT IN`) semantics.
+  // (See `SortMergeJoinExec::null_aware`)
+  //
+  // Currently supported only for uncorrelated `LeftAnti` joins with one
+  // join key and no join filter.
+  bool null_aware = 9;
 }
 
 message PiecewiseMergeJoinExecNode {
diff --git a/datafusion/proto-models/src/generated/pbjson.rs b/datafusion/proto-models/src/generated/pbjson.rs
index 725cceeb75970..34f2a293ae816 100644
--- a/datafusion/proto-models/src/generated/pbjson.rs
+++ b/datafusion/proto-models/src/generated/pbjson.rs
@@ -27666,6 +27666,9 @@ impl serde::Serialize for SortMergeJoinExecNode {
         if !self.projection.is_empty() {
             len += 1;
         }
+        if self.null_aware {
+            len += 1;
+        }
         let mut struct_ser = serializer.serialize_struct("datafusion.SortMergeJoinExecNode", len)?;
         if let Some(v) = self.left.as_ref() {
             struct_ser.serialize_field("left", v)?;
@@ -27695,6 +27698,9 @@ impl serde::Serialize for SortMergeJoinExecNode {
         if !self.projection.is_empty() {
             struct_ser.serialize_field("projection", &self.projection)?;
         }
+        if self.null_aware {
+            struct_ser.serialize_field("nullAware", &self.null_aware)?;
+        }
         struct_ser.end()
     }
 }
@@ -27716,6 +27722,8 @@ impl<'de> serde::Deserialize<'de> for SortMergeJoinExecNode {
             "null_equality",
             "nullEquality",
             "projection",
+            "null_aware",
+            "nullAware",
         ];
 
         #[allow(clippy::enum_variant_names)]
@@ -27728,6 +27736,7 @@ impl<'de> serde::Deserialize<'de> for SortMergeJoinExecNode {
             SortOptions,
             NullEquality,
             Projection,
+            NullAware,
         }
         impl<'de> serde::Deserialize<'de> for GeneratedField {
             fn deserialize<D>(deserializer: D) -> std::result::Result<GeneratedField, D::Error>
@@ -27757,6 +27766,7 @@ impl<'de> serde::Deserialize<'de> for SortMergeJoinExecNode {
                             "sortOptions" | "sort_options" => Ok(GeneratedField::SortOptions),
                             "nullEquality" | "null_equality" => Ok(GeneratedField::NullEquality),
                             "projection" => Ok(GeneratedField::Projection),
+                            "nullAware" | "null_aware" => Ok(GeneratedField::NullAware),
                             _ => Err(serde::de::Error::unknown_field(value, FIELDS)),
                         }
                     }
@@ -27784,6 +27794,7 @@ impl<'de> serde::Deserialize<'de> for SortMergeJoinExecNode {
                 let mut sort_options__ = None;
                 let mut null_equality__ = None;
                 let mut projection__ = None;
+                let mut null_aware__ = None;
                 while let Some(k) = map_.next_key()? {
                     match k {
                         GeneratedField::Left => {
@@ -27837,6 +27848,12 @@ impl<'de> serde::Deserialize<'de> for SortMergeJoinExecNode {
                                     .into_iter().map(|x| x.0).collect())
                             ;
                         }
+                        GeneratedField::NullAware => {
+                            if null_aware__.is_some() {
+                                return Err(serde::de::Error::duplicate_field("nullAware"));
+                            }
+                            null_aware__ = Some(map_.next_value()?);
+                        }
                     }
                 }
                 Ok(SortMergeJoinExecNode {
@@ -27848,6 +27865,7 @@ impl<'de> serde::Deserialize<'de> for SortMergeJoinExecNode {
                     sort_options: sort_options__.unwrap_or_default(),
                     null_equality: null_equality__.unwrap_or_default(),
                     projection: projection__.unwrap_or_default(),
+                    null_aware: null_aware__.unwrap_or_default(),
                 })
             }
         }
diff --git a/datafusion/proto-models/src/generated/prost.rs b/datafusion/proto-models/src/generated/prost.rs
index 1034d0443dbce..142e560baea77 100644
--- a/datafusion/proto-models/src/generated/prost.rs
+++ b/datafusion/proto-models/src/generated/prost.rs
@@ -2760,6 +2760,13 @@ pub struct SortMergeJoinExecNode {
     pub null_equality: i32,
     #[prost(uint32, repeated, tag = "8")]
     pub projection: ::prost::alloc::vec::Vec<u32>,
+    /// Enables null-aware (`NOT IN`) semantics.
+    /// (See `SortMergeJoinExec::null_aware`)
+    ///
+    /// Currently supported only for uncorrelated `LeftAnti` joins with one
+    /// join key and no join filter.
+    #[prost(bool, tag = "9")]
+    pub null_aware: bool,
 }
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct PiecewiseMergeJoinExecNode {
diff --git a/datafusion/proto/tests/cases/plans/joins.rs b/datafusion/proto/tests/cases/plans/joins.rs
index 98397e85b0b3c..db0fb6b85dd72 100644
--- a/datafusion/proto/tests/cases/plans/joins.rs
+++ b/datafusion/proto/tests/cases/plans/joins.rs
@@ -429,6 +429,113 @@ fn roundtrip_sort_merge_join_with_projection() -> Result<()> {
     Ok(())
 }
 
+#[test]
+fn roundtrip_sort_merge_join_null_aware() -> Result<()> {
+    let ctx = SessionContext::new();
+    let codec = DefaultPhysicalExtensionCodec {};
+    let proto_converter = DefaultPhysicalProtoConverter {};
+    let schema = Arc::new(Schema::new(vec![Field::new("key", DataType::Int32, true)]));
+    let on = vec![(
+        Arc::new(Column::new("key", 0)) as _,
+        Arc::new(Column::new("key", 0)) as _,
+    )];
+
+    for null_aware in [false, true] {
+        // Preserve the flag together with absent, empty, and nonempty projections.
+        for projection in [None, Some(vec![]), Some(vec![0])] {
+            let plan: Arc<dyn ExecutionPlan> = Arc::new(
+                SortMergeJoinExec::try_new(
+                    Arc::new(EmptyExec::new(Arc::clone(&schema))),
+                    Arc::new(EmptyExec::new(Arc::clone(&schema))),
+                    on.clone(),
+                    None,
+                    JoinType::LeftAnti,
+                    vec![SortOptions::new(false, true)],
+                    NullEquality::NullEqualsNothing,
+                )?
+                .with_null_aware(null_aware)?
+                .with_projection(projection.clone())?,
+            );
+            let decoded = roundtrip_test_and_return(
+                Arc::clone(&plan),
+                &ctx,
+                &codec,
+                &proto_converter,
+            )?;
+            let decoded = decoded.downcast_ref::<SortMergeJoinExec>().unwrap();
+            assert_eq!(decoded.null_aware(), null_aware);
+            assert_eq!(decoded.projection, projection);
+
+            #[cfg(feature = "json")]
+            {
+                let decoded = super::roundtrip_test_json_and_return(plan, &ctx)?;
+                let decoded = decoded.downcast_ref::<SortMergeJoinExec>().unwrap();
+                assert_eq!(decoded.null_aware(), null_aware);
+                assert_eq!(decoded.projection, projection);
+            }
+        }
+    }
+    Ok(())
+}
+
+/// Null-aware SortMergeJoinExec currently supports only LeftAnti joins.
+/// Checks that decoding rejects an unsupported Inner join whose protobuf flag
+/// enables null-aware semantics.
+/// The flag is changed after serialization to simulate invalid input from a
+/// producer. This guards against bypassing builder validation during decoding
+/// now that `try_new` no longer takes the `null_aware` argument.
+#[test]
+fn sort_merge_join_decode_validates_null_aware() -> Result<()> {
+    let ctx = SessionContext::new();
+    let codec = DefaultPhysicalExtensionCodec {};
+    let proto_converter = DefaultPhysicalProtoConverter {};
+    let schema = Arc::new(Schema::new(vec![Field::new("key", DataType::Int32, true)]));
+
+    // join with no null_aware
+    let plan = Arc::new(SortMergeJoinExec::try_new(
+        Arc::new(EmptyExec::new(Arc::clone(&schema))),
+        Arc::new(EmptyExec::new(schema)),
+        vec![(
+            Arc::new(Column::new("key", 0)) as _,
+            Arc::new(Column::new("key", 0)) as _,
+        )],
+        None,
+        JoinType::Inner,
+        vec![SortOptions::new(false, true)],
+        NullEquality::NullEqualsNothing,
+    )?);
+
+    // serialize
+    let bytes =
+        physical_plan_to_bytes_with_proto_converter(plan, &codec, &proto_converter)?;
+
+    // unserialize should be able to decode back to PhysicalPlanNode
+    let mut node = protobuf::PhysicalPlanNode::decode(bytes.as_ref())
+        .expect("encoded physical plan must decode");
+    let Some(PhysicalPlanType::SortMergeJoin(join)) = node.physical_plan_type.as_mut()
+    else {
+        panic!("expected a SortMergeJoin node");
+    };
+    // This makes the protobuf describe an JoinType::Inner join with null_aware = true,
+    // which is invalid because null-aware SortMergeJoinExec currently supports
+    // only JoinType::LeftAnti joins.
+    // With the other settings used here, LeftAnti + null_aware = true would be
+    // valid; keeping JoinType::Inner + null_aware = false would also be valid.
+    join.null_aware = true;
+    let err = physical_plan_from_bytes_with_proto_converter(
+        &node.encode_to_vec(),
+        ctx.task_ctx().as_ref(),
+        &codec,
+        &proto_converter,
+    )
+    .expect_err("decoding must validate null-aware join restrictions");
+    assert!(
+        err.to_string()
+            .contains("null-aware SortMergeJoin only supports LeftAnti")
+    );
+    Ok(())
+}
+
 #[test]
 fn roundtrip_sort_merge_join() -> Result<()> {
     let ctx = SessionContext::new();
@@ -502,6 +609,7 @@ fn roundtrip_sort_merge_join() -> Result<()> {
                 let result = result.downcast_ref::<SortMergeJoinExec>().unwrap();
                 assert_eq!(result.join_type(), join_type);
                 assert_eq!(result.null_equality(), null_equality);
+                assert!(!result.null_aware());
                 assert_eq!(result.sort_options(), sort_options);
                 assert_eq!(
                     result.filter().as_ref().map(|f| f.column_indices()),
diff --git a/datafusion/sqllogictest/test_files/null_aware_anti_join.slt b/datafusion/sqllogictest/test_files/null_aware_anti_join.slt
index f28d0022f91fc..3acc55fa76fdf 100644
--- a/datafusion/sqllogictest/test_files/null_aware_anti_join.slt
+++ b/datafusion/sqllogictest/test_files/null_aware_anti_join.slt
@@ -611,8 +611,8 @@ query I
 SELECT id FROM nai_outer WHERE id > 5 AND id NOT IN (SELECT id FROM nai_inner) ORDER BY id;
 ----
 
-# Same query under SortMergeJoin + multiple partitions: null-aware joins must
-# be planned as a CollectLeft HashJoin, not a plain anti SortMergeJoin.
+# When hash joins are not preferred, the supported uncorrelated case uses a null-aware
+# SortMergeJoin, which must still preserve the subquery NULL.
 statement ok
 SET datafusion.optimizer.prefer_hash_join = false;
 
@@ -771,6 +771,22 @@ NULL
 # The projected value column must be the FIRST equi-join key: a null-aware hash
 # join reads `on[0]` as the `NOT IN` value key and `on[1..]` as correlation
 # scope keys, positionally. Giving the correlation that slot returns wrong rows.
+# Correlated null-aware joins must keep the CollectLeft HashJoin fallback even
+# when hash joins are not preferred: the single-key SMJ path cannot handle scopes.
+statement ok
+SET datafusion.optimizer.prefer_hash_join = false;
+
+query I
+SELECT z FROM naconst_corr_t3 WHERE 1 NOT IN (SELECT naconst_corr_t4.id FROM naconst_corr_t4 WHERE naconst_corr_t4.z = naconst_corr_t3.z) ORDER BY z;
+----
+7
+NULL
+
+query I
+SELECT z FROM naconst_corr_t3 WHERE 2 NOT IN (SELECT naconst_corr_t4.id FROM naconst_corr_t4 WHERE naconst_corr_t4.z = naconst_corr_t3.z) ORDER BY z;
+----
+NULL
+
 query TT
 EXPLAIN SELECT z FROM naconst_corr_t3 WHERE 1 NOT IN (SELECT naconst_corr_t4.id FROM naconst_corr_t4 WHERE naconst_corr_t4.z = naconst_corr_t3.z);
 ----
@@ -789,6 +805,9 @@ physical_plan
 04)--ProjectionExec: expr=[CAST(id@0 AS Int64) as naconst_corr_t4.id, z@1 as z]
 05)----DataSourceExec: partitions=1, partition_sizes=[1]
 
+statement ok
+SET datafusion.optimizer.prefer_hash_join = true;
+
 statement ok
 DROP TABLE naconst_corr_t3;
 
diff --git a/datafusion/sqllogictest/test_files/null_aware_mark_join.slt b/datafusion/sqllogictest/test_files/null_aware_mark_join.slt
index fcbc929948fec..d657ffa60289e 100644
--- a/datafusion/sqllogictest/test_files/null_aware_mark_join.slt
+++ b/datafusion/sqllogictest/test_files/null_aware_mark_join.slt
@@ -387,8 +387,9 @@ 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.
+# Null-aware mark joins are pinned to CollectLeft hash joins because their
+# null-aware semantics have no sort-merge path. Flipping prefer_hash_join off
+# must not drop null awareness.
 
 statement ok
 set datafusion.optimizer.prefer_hash_join = false;
@@ -412,6 +413,18 @@ physical_plan
 04)------DataSourceExec: partitions=1, partition_sizes=[1]
 05)------DataSourceExec: partitions=1, partition_sizes=[1]
 
+# Execute under the same setting: matching keys are FALSE for NOT IN, while
+# unmatched keys and the NULL outer key are UNKNOWN because the RHS contains NULL.
+query IT rowsort
+SELECT id, value
+FROM outer_table
+WHERE (id NOT IN (SELECT id FROM inner_table_with_null)) IS NULL;
+----
+1 a
+3 c
+4 d
+NULL e
+
 ####################################
 ## Nested loop mark join with NULLs
 ####################################
diff --git a/datafusion/sqllogictest/test_files/subquery.slt b/datafusion/sqllogictest/test_files/subquery.slt
index 2ec240ac67f24..4b553cc9aa64e 100644
--- a/datafusion/sqllogictest/test_files/subquery.slt
+++ b/datafusion/sqllogictest/test_files/subquery.slt
@@ -3187,8 +3187,8 @@ DROP TABLE sq_count_orders;
 # NULL the predicate is never TRUE, so the query must return zero rows. This must
 # hold regardless of the chosen physical join operator. Previously, with
 # prefer_hash_join = false and multiple partitions, the planner routed the
-# null-aware anti join to SortMergeJoin (which is not null-aware) and returned
-# wrong results; null-aware anti joins must use the CollectLeft HashJoin.
+# null-aware anti join to an ordinary SortMergeJoin and returned wrong results.
+# The null-aware SortMergeJoin path must preserve the same NOT IN semantics.
 
 statement ok
 set datafusion.optimizer.prefer_hash_join = false;
@@ -3196,6 +3196,12 @@ set datafusion.optimizer.prefer_hash_join = false;
 statement ok
 CREATE TABLE nia_left(x INT) AS VALUES (1), (2), (3), (4);
 
+statement ok
+CREATE TABLE nia_left_with_null(x INT) AS VALUES (NULL), (1), (2), (3), (3), (4);
+
+statement ok
+CREATE TABLE nia_right_with_empty(y INT);
+
 statement ok
 CREATE TABLE nia_right_with_null(y INT) AS VALUES (2), (NULL);
 
@@ -3207,9 +3213,48 @@ query I
 SELECT x FROM nia_left WHERE x NOT IN (SELECT y FROM nia_right_with_null) ORDER BY x;
 ----
 
-# The null-aware anti join must be planned as a CollectLeft HashJoinExec even with
-# prefer_hash_join = false: SortMergeJoinExec is not null-aware and must not be used.
-# This test has no usable size statistics, so JoinSelection keeps LeftAnti.
+# Subquery is empty -> NOT IN keeps every left row.
+query I
+SELECT x FROM nia_left WHERE x NOT IN (SELECT y FROM nia_right_with_empty) ORDER BY x;
+----
+1
+2
+3
+4
+
+# Subquery has no NULL -> NOT IN behaves like a normal anti join.
+query I
+SELECT x FROM nia_left WHERE x NOT IN (SELECT y FROM nia_right_no_null) ORDER BY x;
+----
+1
+3
+
+# A NULL in the RHS also suppresses a NULL left key.
+query I
+SELECT x FROM nia_left_with_null WHERE x NOT IN (SELECT y FROM nia_right_with_null) ORDER BY x;
+----
+
+# An empty RHS preserves NULL left keys and duplicate rows.
+query I
+SELECT x FROM nia_left_with_null WHERE x NOT IN (SELECT y FROM nia_right_with_empty) ORDER BY x NULLS FIRST;
+----
+NULL
+1
+2
+3
+3
+4
+
+# A nonempty RHS without NULL excludes NULL left keys but preserves duplicates.
+query I
+SELECT x FROM nia_left_with_null WHERE x NOT IN (SELECT y FROM nia_right_no_null) ORDER BY x;
+----
+1
+3
+3
+
+# With prefer_hash_join = false, this single-key, unfiltered null-aware LeftAnti
+# join uses SortMergeJoinExec with sorted, single-partition inputs.
 query TT
 EXPLAIN SELECT x FROM nia_left WHERE x NOT IN (SELECT y FROM nia_right_with_null);
 ----
@@ -3219,20 +3264,48 @@ logical_plan
 03)--SubqueryAlias: __correlated_sq_1
 04)----TableScan: nia_right_with_null projection=[y]
 physical_plan
-01)HashJoinExec: mode=CollectLeft, join_type=LeftAnti, on=[(x@0, y@0)], null_aware
-02)--DataSourceExec: partitions=1, partition_sizes=[1]
-03)--DataSourceExec: partitions=1, partition_sizes=[1]
+01)SortMergeJoinExec: join_type=LeftAnti, on=[(x@0, y@0)], null_aware
+02)--SortExec: expr=[x@0 ASC], preserve_partitioning=[false]
+03)----DataSourceExec: partitions=1, partition_sizes=[1]
+04)--SortExec: expr=[y@0 ASC], preserve_partitioning=[false]
+05)----DataSourceExec: partitions=1, partition_sizes=[1]
 
-# Subquery has no NULL -> NOT IN behaves like a normal anti join.
-query I
-SELECT x FROM nia_left WHERE x NOT IN (SELECT y FROM nia_right_no_null) ORDER BY x;
+# Tree-formatted plans must also distinguish null-aware joins from ordinary anti joins.
+query TT
+EXPLAIN FORMAT TREE SELECT x FROM nia_left WHERE x NOT IN (SELECT y FROM nia_right_with_null);
 ----
-1
-3
+physical_plan
+01)┌───────────────────────────┐
+02)│     SortMergeJoinExec     │
+03)│    --------------------   │
+04)│    join_type: LeftAnti    │
+05)│                           ├──────────────┐
+06)│         null_aware        │              │
+07)│                           │              │
+08)│        on: (x = y)        │              │
+09)└─────────────┬─────────────┘              │
+10)┌─────────────┴─────────────┐┌─────────────┴─────────────┐
+11)│          SortExec         ││          SortExec         │
+12)│    --------------------   ││    --------------------   │
+13)│          x@0 ASC          ││          y@0 ASC          │
+14)└─────────────┬─────────────┘└─────────────┬─────────────┘
+15)┌─────────────┴─────────────┐┌─────────────┴─────────────┐
+16)│       DataSourceExec      ││       DataSourceExec      │
+17)│    --------------------   ││    --------------------   │
+18)│         bytes: 112        ││         bytes: 176        │
+19)│       format: memory      ││       format: memory      │
+20)│          rows: 4          ││          rows: 2          │
+21)└───────────────────────────┘└───────────────────────────┘
 
 statement ok
 DROP TABLE nia_left;
 
+statement ok
+DROP TABLE nia_left_with_null;
+
+statement ok
+DROP TABLE nia_right_with_empty;
+
 statement ok
 DROP TABLE nia_right_with_null;
 
@@ -3242,6 +3315,141 @@ DROP TABLE nia_right_no_null;
 statement ok
 reset datafusion.optimizer.prefer_hash_join;
 
+# A NULL in one RHS partition must affect every outer row.
+
+statement ok
+SET datafusion.optimizer.prefer_hash_join = false;
+
+statement ok
+SET datafusion.execution.target_partitions = 4;
+
+statement ok
+SET datafusion.optimizer.repartition_joins = true;
+
+statement ok
+CREATE TABLE nia_mp_left(x INT) AS
+VALUES (NULL), (1), (2), (3), (3);
+
+# Each UNION ALL branch produces one partition.
+# CTAS preserves the partitioned result in the memory table.
+statement ok
+CREATE TABLE nia_mp_right AS
+SELECT CAST(10 AS INT) AS y
+UNION ALL
+SELECT CAST(20 AS INT) AS y
+UNION ALL
+SELECT CAST(30 AS INT) AS y
+UNION ALL
+SELECT CAST(NULL AS INT) AS y;
+
+# Verify that the RHS really has four nonempty partitions.
+# partition_sizes counts batches, not rows.
+query TT
+EXPLAIN SELECT y FROM nia_mp_right;
+----
+logical_plan TableScan: nia_mp_right projection=[y]
+physical_plan DataSourceExec: partitions=4, partition_sizes=[1, 1, 1, 1]
+
+# No left key matches 10, 20, or 30.
+# The NULL in the remaining RHS partition must still suppress every row.
+query I
+SELECT x
+FROM nia_mp_left
+WHERE x NOT IN (SELECT y FROM nia_mp_right)
+ORDER BY x;
+----
+
+# The optimizer must bring all RHS partitions into the null-aware join.
+query TT
+EXPLAIN SELECT x FROM nia_mp_left WHERE x NOT IN (SELECT y FROM nia_mp_right);
+----
+logical_plan
+01)LeftAnti Join: nia_mp_left.x = __correlated_sq_1.y null_aware
+02)--TableScan: nia_mp_left projection=[x]
+03)--SubqueryAlias: __correlated_sq_1
+04)----TableScan: nia_mp_right projection=[y]
+physical_plan
+01)SortMergeJoinExec: join_type=LeftAnti, on=[(x@0, y@0)], null_aware
+02)--SortExec: expr=[x@0 ASC], preserve_partitioning=[false]
+03)----DataSourceExec: partitions=1, partition_sizes=[1]
+04)--SortPreservingMergeExec: [y@0 ASC]
+05)----SortExec: expr=[y@0 ASC], preserve_partitioning=[true]
+06)------DataSourceExec: partitions=4, partition_sizes=[1, 1, 1, 1]
+
+# Disabling join repartitioning must preserve the same semantics.
+statement ok
+SET datafusion.optimizer.repartition_joins = false;
+
+query I
+SELECT x
+FROM nia_mp_left
+WHERE x NOT IN (SELECT y FROM nia_mp_right)
+ORDER BY x;
+----
+
+# Disabling hash repartitioning must not bypass the single-partition requirement.
+query TT
+EXPLAIN SELECT x FROM nia_mp_left WHERE x NOT IN (SELECT y FROM nia_mp_right);
+----
+logical_plan
+01)LeftAnti Join: nia_mp_left.x = __correlated_sq_1.y null_aware
+02)--TableScan: nia_mp_left projection=[x]
+03)--SubqueryAlias: __correlated_sq_1
+04)----TableScan: nia_mp_right projection=[y]
+physical_plan
+01)SortMergeJoinExec: join_type=LeftAnti, on=[(x@0, y@0)], null_aware
+02)--SortExec: expr=[x@0 ASC], preserve_partitioning=[false]
+03)----DataSourceExec: partitions=1, partition_sizes=[1]
+04)--SortPreservingMergeExec: [y@0 ASC]
+05)----SortExec: expr=[y@0 ASC], preserve_partitioning=[true]
+06)------DataSourceExec: partitions=4, partition_sizes=[1, 1, 1, 1]
+
+# Requesting one target partition must also preserve the result.
+statement ok
+SET datafusion.execution.target_partitions = 1;
+
+query I
+SELECT x
+FROM nia_mp_left
+WHERE x NOT IN (SELECT y FROM nia_mp_right)
+ORDER BY x;
+----
+
+query TT
+EXPLAIN
+SELECT x
+FROM nia_mp_left
+WHERE x NOT IN (SELECT y FROM nia_mp_right);
+----
+logical_plan
+01)LeftAnti Join: nia_mp_left.x = __correlated_sq_1.y null_aware
+02)--TableScan: nia_mp_left projection=[x]
+03)--SubqueryAlias: __correlated_sq_1
+04)----TableScan: nia_mp_right projection=[y]
+physical_plan
+01)SortMergeJoinExec: join_type=LeftAnti, on=[(x@0, y@0)], null_aware
+02)--SortExec: expr=[x@0 ASC], preserve_partitioning=[false]
+03)----DataSourceExec: partitions=1, partition_sizes=[1]
+04)--SortPreservingMergeExec: [y@0 ASC]
+05)----SortExec: expr=[y@0 ASC], preserve_partitioning=[true]
+06)------DataSourceExec: partitions=4, partition_sizes=[1, 1, 1, 1]
+
+statement ok
+DROP TABLE nia_mp_left;
+
+statement ok
+DROP TABLE nia_mp_right;
+
+statement ok
+RESET datafusion.optimizer.prefer_hash_join;
+
+statement ok
+RESET datafusion.optimizer.repartition_joins;
+
+# Restore the SLT runner's target partition count.
+statement ok
+SET datafusion.execution.target_partitions = 4;
+
 # Regression test for #24139: CTE inside a correlated scalar subquery causing a schema error
 statement ok
 CREATE TABLE metrics (host VARCHAR, bytes_sent DOUBLE, ts BIGINT) AS VALUES
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 2b11283bd09a9..b2912abfeda82 100644
--- a/docs/source/library-user-guide/upgrading/56.0.0.md
+++ b/docs/source/library-user-guide/upgrading/56.0.0.md
@@ -1044,6 +1044,54 @@ reader decodes a payload without the field (value `0`) as a scalar `NOT IN`. A
 ignores the unknown field and would read the extra tuple elements as
 correlation keys, returning wrong results.
 
+### Null-aware sort-merge joins add a protobuf field
+
+`SortMergeJoinExec` now supports null-aware (`NOT IN`) semantics for
+uncorrelated, single-key `LeftAnti` joins without a join filter.
+The existing seven-argument `SortMergeJoinExec::try_new` continues
+to create a non-null-aware join. Enable null-aware semantics with
+`.with_null_aware(true)?`:
+
+```rust,ignore
+let exec = SortMergeJoinExec::try_new(
+    left,
+    right,
+    on,
+    None,
+    JoinType::LeftAnti,
+    vec![SortOptions::new(false, true)], // ASC NULLS FIRST
+    NullEquality::NullEqualsNothing,
+)?
+.with_null_aware(true)?;
+```
+
+Both inputs must have a single partition when executed.
+
+The generated protobuf `SortMergeJoinExecNode` type has a new
+`null_aware` field. Exhaustive struct literals must initialize it;
+`false` preserves the previous behavior:
+
+```rust,ignore
+let node = SortMergeJoinExecNode {
+    left,
+    right,
+    on,
+    join_type,
+    filter,
+    sort_options,
+    null_equality,
+    projection,
+    null_aware: false,
+};
+```
+
+For binary protobuf plans, an older sort-merge join payload without this
+field decodes with `null_aware = false`. A 55.x reader must not consume
+a new sort-merge join plan with `null_aware = true`: it ignores the
+unknown field and executes a plain anti join, which can return incorrect
+results. Upgrade all plan readers before sending them null-aware
+sort-merge join plans.
+
 ### `FunctionalDependence` has a new `null_equality` field
 
 `FunctionalDependence` now records the NULL semantics its dependency holds under.