perf: sort-merge join (SMJ) batch deferred filtering and move mark joins to bitwise stream. Near-unique LEFT and FULL SMJ 20-50x faster

benchmarks/src/smj.rs

@@ -39,7 +39,7 @@ use futures::StreamExt;
 #[derive(Debug, Args, Clone)]
 #[command(verbatim_doc_comment)]
 pub struct RunOpt {
-    /// Query number (between 1 and 20). If not specified, runs all queries
+    /// Query number (between 1 and 23). If not specified, runs all queries
     #[arg(short, long)]
     query: Option<usize>,
 
@@ -410,6 +410,52 @@ const SMJ_QUERIES: &[&str] = &[
         FROM t1_sorted JOIN t2_sorted ON t1_sorted.key = t2_sorted.key
         GROUP BY t1_sorted.key
     "#,
+    // Q21: INNER 10M x 10M | unique keys (1:1) | 50% join filter
+    r#"
+        WITH t1_sorted AS (
+            SELECT value as key, value as data
+            FROM range(10000000) ORDER BY value
+        ),
+        t2_sorted AS (
+            SELECT value as key, value as data
+            FROM range(10000000) ORDER BY value
+        )
+        SELECT t1_sorted.key, t1_sorted.data as d1, t2_sorted.data as d2
+        FROM t1_sorted JOIN t2_sorted
+          ON t1_sorted.key = t2_sorted.key
+         AND t1_sorted.data + t2_sorted.data < 10000000
+    "#,
+    // Q22: LEFT 10M x 10M | unique keys (1:1) | 50% join filter
+    r#"
+        WITH t1_sorted AS (
+            SELECT value as key, value as data
+            FROM range(10000000) ORDER BY value
+        ),
+        t2_sorted AS (
+            SELECT value as key, value as data
+            FROM range(10000000) ORDER BY value
+        )
+        SELECT t1_sorted.key, t1_sorted.data as d1, t2_sorted.data as d2
+        FROM t1_sorted LEFT JOIN t2_sorted
+          ON t1_sorted.key = t2_sorted.key
+         AND t1_sorted.data + t2_sorted.data < 10000000
+    "#,
+    // Q23: FULL 10M x 10M | unique keys (1:1) | 50% join filter
+    r#"
+        WITH t1_sorted AS (
+            SELECT value as key, value as data
+            FROM range(10000000) ORDER BY value
+        ),
+        t2_sorted AS (
+            SELECT value as key, value as data
+            FROM range(10000000) ORDER BY value
+        )
+        SELECT t1_sorted.key as k1, t1_sorted.data as d1,
+               t2_sorted.key as k2, t2_sorted.data as d2
+        FROM t1_sorted FULL JOIN t2_sorted
+          ON t1_sorted.key = t2_sorted.key
+         AND t1_sorted.data + t2_sorted.data < 10000000
+    "#,
 ];
 
 impl RunOpt {

datafusion/physical-plan/src/joins/mod.rs

@@ -34,7 +34,7 @@ mod cross_join;
 mod hash_join;
 mod nested_loop_join;
 mod piecewise_merge_join;
-pub(crate) mod semi_anti_sort_merge_join;
+pub(crate) mod semi_anti_mark_sort_merge_join;
 mod sort_merge_join;
 mod stream_join_utils;
 mod symmetric_hash_join;

datafusion/physical-plan/src/joins/semi_anti_sort_merge_join/mod.rs → datafusion/physical-plan/src/joins/semi_anti_mark_sort_merge_join/mod.rs

@@ -15,9 +15,9 @@
 // specific language governing permissions and limitations
 // under the License.
 
-//! Specialized Sort Merge Join stream for Semi/Anti joins.
+//! Specialized Sort Merge Join stream for Semi/Anti/Mark joins.
 //!
-//! Used internally by `SortMergeJoinExec` for semi/anti join types.
+//! Used internally by `SortMergeJoinExec` for semi/anti/mark join types.
 
 pub(crate) mod stream;
 

datafusion/physical-plan/src/joins/semi_anti_sort_merge_join/stream.rs → datafusion/physical-plan/src/joins/semi_anti_mark_sort_merge_join/stream.rs

@@ -15,10 +15,11 @@
 // specific language governing permissions and limitations
 // under the License.
 
-//! Sort-merge join stream specialized for semi/anti joins.
+//! Sort-merge join stream specialized for semi/anti/mark joins.
 //!
 //! Instantiated by [`SortMergeJoinExec`](crate::joins::sort_merge_join::SortMergeJoinExec)
-//! when the join type is `LeftSemi`, `LeftAnti`, `RightSemi`, or `RightAnti`.
+//! when the join type is `LeftSemi`, `LeftAnti`, `RightSemi`, `RightAnti`,
+//! `LeftMark`, or `RightMark`.
 //!
 //! # Motivation
 //!
@@ -36,7 +37,8 @@
 //!
 //! For `Left*` join types, left is outer and right is inner.
 //! For `Right*` join types, right is outer and left is inner.
-//! The output schema always equals the outer side's schema.
+//! The output schema always equals the outer side's schema (for semi/anti)
+//! or the outer side's schema plus a boolean mark column (for mark joins).
 //!
 //! # Algorithm
 //!
@@ -75,6 +77,7 @@
 //!   On emit:
 //!     Semi  → filter_record_batch(outer_batch, &matched)
 //!     Anti  → filter_record_batch(outer_batch, &NOT(matched))
+//!     Mark  → outer_batch + matched as boolean column
 //! ```
 //!
 //! ## Batch boundaries
@@ -245,10 +248,13 @@ enum PendingBoundary {
     Filtered { saved_keys: Vec<ArrayRef> },
 }
 
-pub(crate) struct SemiAntiSortMergeJoinStream {
-    // Decomposed from JoinType to avoid matching on 4 variants in hot paths.
+pub(crate) struct SemiAntiMarkSortMergeJoinStream {
+    // Decomposed from JoinType to avoid matching on 6 variants in hot paths.
     // true for semi (emit matched), false for anti (emit unmatched).
+    // Ignored when is_mark is true.
     is_semi: bool,
+    // true for mark joins (emit all rows with match boolean column).
+    is_mark: bool,
 
     // Input streams — in the nested-loop model that sort-merge join
     // implements, "outer" is the driving loop and "inner" is probed for
@@ -330,7 +336,7 @@ pub(crate) struct SemiAntiSortMergeJoinStream {
     batch_emitted: bool,
 }
 
-impl SemiAntiSortMergeJoinStream {
+impl SemiAntiMarkSortMergeJoinStream {
     #[expect(clippy::too_many_arguments)]
     pub fn try_new(
         schema: SchemaRef,
@@ -350,8 +356,24 @@ impl SemiAntiSortMergeJoinStream {
         spill_manager: SpillManager,
         runtime_env: Arc<datafusion_execution::runtime_env::RuntimeEnv>,
     ) -> Result<Self> {
+        debug_assert!(
+            matches!(
+                join_type,
+                JoinType::LeftSemi
+                    | JoinType::RightSemi
+                    | JoinType::LeftAnti
+                    | JoinType::RightAnti
+                    | JoinType::LeftMark
+                    | JoinType::RightMark
+            ),
+            "SemiAntiMarkSortMergeJoinStream does not handle {join_type:?}"
+        );
         let is_semi = matches!(join_type, JoinType::LeftSemi | JoinType::RightSemi);
-        let outer_is_left = matches!(join_type, JoinType::LeftSemi | JoinType::LeftAnti);
+        let is_mark = matches!(join_type, JoinType::LeftMark | JoinType::RightMark);
+        let outer_is_left = matches!(
+            join_type,
+            JoinType::LeftSemi | JoinType::LeftAnti | JoinType::LeftMark
+        );
 
         let join_time = MetricBuilder::new(metrics).subset_time("join_time", partition);
         let input_batches =
@@ -361,6 +383,7 @@ impl SemiAntiSortMergeJoinStream {
 
         Ok(Self {
             is_semi,
+            is_mark,
             outer,
             inner,
             outer_batch: None,
@@ -492,17 +515,33 @@ impl SemiAntiSortMergeJoinStream {
 
         // finish() converts the bit-packed builder directly to a
         // BooleanBuffer — no iteration or repacking needed.
-        let selection = BooleanArray::new(self.matched.finish(), None);
-
-        let selection = if self.is_semi {
-            selection
+        let matched_buf = self.matched.finish();
+
+        if self.is_mark {
+            // Mark joins emit ALL outer rows with a boolean match column appended.
+            debug_assert_eq!(
+                self.schema.fields().len(),
+                batch.num_columns() + 1,
+                "Mark join output schema should be outer schema + 1 mark column"
+            );
+            let mark_col = Arc::new(BooleanArray::new(matched_buf, None)) as ArrayRef;
+            let mut columns = batch.columns().to_vec();
+            columns.push(mark_col);
+            let output = RecordBatch::try_new(Arc::clone(&self.schema), columns)?;
+            self.coalescer.push_batch(output)?;
         } else {
-            not(&selection)?
-        };
+            let selection = BooleanArray::new(matched_buf, None);
+
+            let selection = if self.is_semi {
+                selection
+            } else {
+                not(&selection)?
+            };
 
-        let filtered = filter_record_batch(batch, &selection)?;
-        if filtered.num_rows() > 0 {
-            self.coalescer.push_batch(filtered)?;
+            let filtered = filter_record_batch(batch, &selection)?;
+            if filtered.num_rows() > 0 {
+                self.coalescer.push_batch(filtered)?;
+            }
         }
         Ok(())
     }
@@ -1184,7 +1223,7 @@ fn keys_match(
 
 /// Evaluate the join filter for one inner row against a slice of outer rows.
 ///
-/// Free function (not a method on SemiAntiSortMergeJoinStream) so that Rust
+/// Free function (not a method on SemiAntiMarkSortMergeJoinStream) so that Rust
 /// can split the struct borrow in process_key_match_with_filter: the caller
 /// holds &mut self.matched and &self.inner_key_buffer simultaneously, which
 /// is impossible if this borrows all of &self.
@@ -1257,7 +1296,7 @@ fn evaluate_filter_for_inner_row(
     }
 }
 
-impl Stream for SemiAntiSortMergeJoinStream {
+impl Stream for SemiAntiMarkSortMergeJoinStream {
     type Item = Result<RecordBatch>;
 
     fn poll_next(
@@ -1269,7 +1308,7 @@ impl Stream for SemiAntiSortMergeJoinStream {
     }
 }
 
-impl RecordBatchStream for SemiAntiSortMergeJoinStream {
+impl RecordBatchStream for SemiAntiMarkSortMergeJoinStream {
     fn schema(&self) -> SchemaRef {
         Arc::clone(&self.schema)
     }

datafusion/physical-plan/src/joins/semi_anti_sort_merge_join/tests.rs → datafusion/physical-plan/src/joins/semi_anti_mark_sort_merge_join/tests.rs

@@ -19,7 +19,7 @@ use std::pin::Pin;
 use std::sync::Arc;
 use std::task::{Context, Poll};
 
-use super::stream::SemiAntiSortMergeJoinStream;
+use super::stream::SemiAntiMarkSortMergeJoinStream;
 use crate::ExecutionPlan;
 use crate::RecordBatchStream;
 use crate::common;
@@ -149,8 +149,10 @@ impl RecordBatchStream for PendingStream {
     }
 }
 
-/// Helper: collect all output from a SemiAntiSortMergeJoinStream.
-async fn collect_stream(stream: SemiAntiSortMergeJoinStream) -> Result<Vec<RecordBatch>> {
+/// Helper: collect all output from a SemiAntiMarkSortMergeJoinStream.
+async fn collect_stream(
+    stream: SemiAntiMarkSortMergeJoinStream,
+) -> Result<Vec<RecordBatch>> {
     common::collect(Box::pin(stream)).await
 }
 
@@ -259,7 +261,7 @@ async fn filter_buffer_pending_loses_inner_rows() -> Result<()> {
     let inner_schema = inner.schema();
     let (reservation, peak_mem_used, spill_manager, runtime_env) =
         test_stream_resources(inner_schema, &metrics);
-    let stream = SemiAntiSortMergeJoinStream::try_new(
+    let stream = SemiAntiMarkSortMergeJoinStream::try_new(
         left_schema, // output schema = outer schema for semi
         vec![SortOptions::default()],
         NullEquality::NullEqualsNothing,
@@ -359,7 +361,7 @@ async fn no_filter_boundary_pending_loses_outer_rows() -> Result<()> {
     let inner_schema = inner.schema();
     let (reservation, peak_mem_used, spill_manager, runtime_env) =
         test_stream_resources(inner_schema, &metrics);
-    let stream = SemiAntiSortMergeJoinStream::try_new(
+    let stream = SemiAntiMarkSortMergeJoinStream::try_new(
         left_schema,
         vec![SortOptions::default()],
         NullEquality::NullEqualsNothing,
@@ -473,7 +475,7 @@ async fn filtered_boundary_pending_outer_rows() -> Result<()> {
     let inner_schema = inner.schema();
     let (reservation, peak_mem_used, spill_manager, runtime_env) =
         test_stream_resources(inner_schema, &metrics);
-    let stream = SemiAntiSortMergeJoinStream::try_new(
+    let stream = SemiAntiMarkSortMergeJoinStream::try_new(
         left_schema,
         vec![SortOptions::default()],
         NullEquality::NullEqualsNothing,
@@ -756,7 +758,7 @@ async fn spill_filtered_boundary_loses_outer_rows() -> Result<()> {
             Arc::clone(&right_schema),
         );
 
-        let stream = SemiAntiSortMergeJoinStream::try_new(
+        let stream = SemiAntiMarkSortMergeJoinStream::try_new(
             Arc::clone(&left_schema),
             vec![SortOptions::default()],
             NullEquality::NullEqualsNothing,

datafusion/physical-plan/src/joins/sort_merge_join/exec.rs

@@ -25,7 +25,7 @@ use std::sync::Arc;
 
 use crate::execution_plan::{EmissionType, boundedness_from_children};
 use crate::expressions::PhysicalSortExpr;
-use crate::joins::semi_anti_sort_merge_join::stream::SemiAntiSortMergeJoinStream;
+use crate::joins::semi_anti_mark_sort_merge_join::stream::SemiAntiMarkSortMergeJoinStream;
 use crate::joins::sort_merge_join::metrics::SortMergeJoinMetrics;
 use crate::joins::sort_merge_join::stream::SortMergeJoinStream;
 use crate::joins::utils::{
@@ -337,6 +337,8 @@ impl SortMergeJoinExec {
                 | JoinType::RightSemi
                 | JoinType::LeftAnti
                 | JoinType::RightAnti
+                | JoinType::LeftMark
+                | JoinType::RightMark
         ) {
             Ok(Arc::new(new_join))
         } else {
@@ -534,6 +536,8 @@ impl ExecutionPlan for SortMergeJoinExec {
                 | JoinType::LeftAnti
                 | JoinType::RightSemi
                 | JoinType::RightAnti
+                | JoinType::LeftMark
+                | JoinType::RightMark
         ) {
             let peak_mem_used =
                 MetricBuilder::new(&self.metrics).gauge("peak_mem_used", partition);
@@ -544,7 +548,7 @@ impl ExecutionPlan for SortMergeJoinExec {
             )
             .with_compression_type(context.session_config().spill_compression());
 
-            Ok(Box::pin(SemiAntiSortMergeJoinStream::try_new(
+            Ok(Box::pin(SemiAntiMarkSortMergeJoinStream::try_new(
                 Arc::clone(&self.schema),
                 self.sort_options.clone(),
                 self.null_equality,