apache
diff --git a/‎benchmarks/src/smj.rs‎
Lines changed: 78 additions & 2 deletions b/‎benchmarks/src/smj.rs‎
Lines changed: 78 additions & 2 deletions
diff --git a/‎datafusion/core/tests/fuzz_cases/join_fuzz.rs‎
Lines changed: 135 additions & 0 deletions b/‎datafusion/core/tests/fuzz_cases/join_fuzz.rs‎
Lines changed: 135 additions & 0 deletions
diff --git a/‎datafusion/physical-plan/src/joins/mod.rs‎
Lines changed: 0 additions & 1 deletion b/‎datafusion/physical-plan/src/joins/mod.rs‎
Lines changed: 0 additions & 1 deletion
diff --git a/‎datafusion/physical-plan/src/joins/semi_anti_sort_merge_join/mod.rs‎
Lines changed: 0 additions & 25 deletions b/‎datafusion/physical-plan/src/joins/semi_anti_sort_merge_join/mod.rs‎
Lines changed: 0 additions & 25 deletions
@@ -39,7 +39,7 @@ use futures::StreamExt;
 #[derive(Debug, Args, Clone)]
 #[command(verbatim_doc_comment)]
 pub struct RunOpt {
-    /// Query number (between 1 and 23). If not specified, runs all queries
+    /// Query number (between 1 and 26). If not specified, runs all queries
     #[arg(short, long)]
     query: Option<usize>,
 
@@ -456,6 +456,72 @@ const SMJ_QUERIES: &[&str] = &[
           ON t1_sorted.key = t2_sorted.key
          AND t1_sorted.data + t2_sorted.data < 10000000
     "#,
+    // Q24: LEFT MARK 1M x 10M | 1:10 | 1%
+    r#"
+        WITH t1_sorted AS (
+            SELECT value % 100000 as key, value as data
+            FROM range(1000000)
+            ORDER BY key, data
+        ),
+        t2_sorted AS (
+            SELECT value % 100000 as key, value as data
+            FROM range(10000000)
+            ORDER BY key, data
+        )
+        SELECT t1_sorted.key, t1_sorted.data
+        FROM t1_sorted
+        WHERE t1_sorted.data < 0
+           OR EXISTS (
+            SELECT 1 FROM t2_sorted
+            WHERE t2_sorted.key = t1_sorted.key
+              AND t2_sorted.data <> t1_sorted.data
+              AND t2_sorted.data % 100 = 0
+        )
+    "#,
+    // Q25: LEFT MARK 1M x 10M | 1:10 | 50%
+    r#"
+        WITH t1_sorted AS (
+            SELECT value % 100000 as key, value as data
+            FROM range(1000000)
+            ORDER BY key, data
+        ),
+        t2_sorted AS (
+            SELECT value % 100000 as key, value as data
+            FROM range(10000000)
+            ORDER BY key, data
+        )
+        SELECT t1_sorted.key, t1_sorted.data
+        FROM t1_sorted
+        WHERE t1_sorted.data < 0
+           OR EXISTS (
+            SELECT 1 FROM t2_sorted
+            WHERE t2_sorted.key = t1_sorted.key
+              AND t2_sorted.data <> t1_sorted.data
+              AND t2_sorted.data % 2 = 0
+        )
+    "#,
+    // Q26: LEFT MARK 1M x 10M | 1:10 | 90%
+    r#"
+        WITH t1_sorted AS (
+            SELECT value % 100000 as key, value as data
+            FROM range(1000000)
+            ORDER BY key, data
+        ),
+        t2_sorted AS (
+            SELECT value % 100000 as key, value as data
+            FROM range(10000000)
+            ORDER BY key, data
+        )
+        SELECT t1_sorted.key, t1_sorted.data
+        FROM t1_sorted
+        WHERE t1_sorted.data < 0
+           OR EXISTS (
+            SELECT 1 FROM t2_sorted
+            WHERE t2_sorted.key = t1_sorted.key
+              AND t2_sorted.data <> t1_sorted.data
+              AND t2_sorted.data % 10 <> 0
+        )
+    "#,
 ];
 
 impl RunOpt {
@@ -489,7 +555,10 @@ impl RunOpt {
 
             let sql = SMJ_QUERIES[query_index];
             benchmark_run.start_new_case(&format!("Query {query_id}"));
-            let query_run = self.benchmark_query(sql, &query_id.to_string(), &ctx).await;
+            let expect_mark = query_id >= 24;
+            let query_run = self
+                .benchmark_query(sql, &query_id.to_string(), expect_mark, &ctx)
+                .await;
             match query_run {
                 Ok(query_results) => {
                     for iter in query_results {
@@ -513,6 +582,7 @@ impl RunOpt {
         &self,
         sql: &str,
         query_name: &str,
+        expect_mark: bool,
         ctx: &SessionContext,
     ) -> Result<Vec<QueryResult>> {
         let mut query_results = vec![];
@@ -528,6 +598,12 @@ impl RunOpt {
             ));
         }
 
+        if expect_mark && !plan_string.contains("LeftMark") {
+            return Err(exec_datafusion_err!(
+                "Query {query_name} expected LeftMark join. Physical plan: {plan_string}"
+            ));
+        }
+
         for i in 0..self.common.iterations {
             let start = Instant::now();
 
 
@@ -38,6 +38,9 @@ use datafusion::physical_plan::joins::{
 };
 use datafusion::prelude::{SessionConfig, SessionContext};
 use datafusion_common::{NullEquality, ScalarValue};
+use datafusion_execution::TaskContext;
+use datafusion_execution::disk_manager::{DiskManagerBuilder, DiskManagerMode};
+use datafusion_execution::runtime_env::RuntimeEnvBuilder;
 use datafusion_physical_expr::PhysicalExprRef;
 use datafusion_physical_expr::expressions::Literal;
 
@@ -1125,6 +1128,138 @@ impl JoinFuzzTestCase {
     }
 }
 
+/// Fuzz test: compare SMJ (with spilling) against HJ (no spill) for filtered
+/// outer joins under memory pressure. This exercises the deferred filtering +
+/// spill read-back path that unit tests can't easily cover with random data.
+#[tokio::test]
+async fn test_filtered_join_spill_fuzz() {
+    let join_types = [JoinType::Left, JoinType::Right, JoinType::Full];
+
+    let runtime_spill = RuntimeEnvBuilder::new()
+        .with_memory_limit(4096, 1.0)
+        .with_disk_manager_builder(
+            DiskManagerBuilder::default().with_mode(DiskManagerMode::OsTmpDirectory),
+        )
+        .build_arc()
+        .unwrap();
+
+    for join_type in &join_types {
+        for (left_extra, right_extra) in [(true, true), (false, true), (true, false)] {
+            let input1 = make_staggered_batches_i32(1000, left_extra);
+            let input2 = make_staggered_batches_i32(1000, right_extra);
+
+            let schema1 = input1[0].schema();
+            let schema2 = input2[0].schema();
+            let filter = col_lt_col_filter(schema1.clone(), schema2.clone());
+
+            let on = vec![
+                (
+                    Arc::new(Column::new_with_schema("a", &schema1).unwrap()) as _,
+                    Arc::new(Column::new_with_schema("a", &schema2).unwrap()) as _,
+                ),
+                (
+                    Arc::new(Column::new_with_schema("b", &schema1).unwrap()) as _,
+                    Arc::new(Column::new_with_schema("b", &schema2).unwrap()) as _,
+                ),
+            ];
+
+            for batch_size in [2, 49, 100] {
+                let session_config = SessionConfig::new().with_batch_size(batch_size);
+
+                // HJ baseline (no memory limit)
+                let left_hj = MemorySourceConfig::try_new_exec(
+                    std::slice::from_ref(&input1),
+                    schema1.clone(),
+                    None,
+                )
+                .unwrap();
+                let right_hj = MemorySourceConfig::try_new_exec(
+                    std::slice::from_ref(&input2),
+                    schema2.clone(),
+                    None,
+                )
+                .unwrap();
+                let hj = Arc::new(
+                    HashJoinExec::try_new(
+                        left_hj,
+                        right_hj,
+                        on.clone(),
+                        Some(filter.clone()),
+                        join_type,
+                        None,
+                        PartitionMode::Partitioned,
+                        NullEquality::NullEqualsNothing,
+                        false,
+                    )
+                    .unwrap(),
+                );
+                let ctx_hj = SessionContext::new_with_config(session_config.clone());
+                let hj_collected = collect(hj, ctx_hj.task_ctx()).await.unwrap();
+
+                // SMJ with spilling
+                let left_smj = MemorySourceConfig::try_new_exec(
+                    std::slice::from_ref(&input1),
+                    schema1.clone(),
+                    None,
+                )
+                .unwrap();
+                let right_smj = MemorySourceConfig::try_new_exec(
+                    std::slice::from_ref(&input2),
+                    schema2.clone(),
+                    None,
+                )
+                .unwrap();
+                let smj = Arc::new(
+                    SortMergeJoinExec::try_new(
+                        left_smj,
+                        right_smj,
+                        on.clone(),
+                        Some(filter.clone()),
+                        *join_type,
+                        vec![SortOptions::default(); on.len()],
+                        NullEquality::NullEqualsNothing,
+                    )
+                    .unwrap(),
+                );
+                let task_ctx_spill = Arc::new(
+                    TaskContext::default()
+                        .with_session_config(session_config)
+                        .with_runtime(Arc::clone(&runtime_spill)),
+                );
+                let smj_collected = collect(smj, task_ctx_spill).await.unwrap();
+
+                let hj_rows: usize = hj_collected.iter().map(|b| b.num_rows()).sum();
+                let smj_rows: usize = smj_collected.iter().map(|b| b.num_rows()).sum();
+
+                assert_eq!(
+                    hj_rows, smj_rows,
+                    "Row count mismatch for {join_type:?} batch_size={batch_size} \
+                     left_extra={left_extra} right_extra={right_extra}: \
+                     HJ={hj_rows} SMJ={smj_rows}"
+                );
+
+                if hj_rows > 0 {
+                    let hj_fmt =
+                        pretty_format_batches(&hj_collected).unwrap().to_string();
+                    let smj_fmt =
+                        pretty_format_batches(&smj_collected).unwrap().to_string();
+
+                    let mut hj_sorted: Vec<&str> = hj_fmt.trim().lines().collect();
+                    hj_sorted.sort_unstable();
+                    let mut smj_sorted: Vec<&str> = smj_fmt.trim().lines().collect();
+                    smj_sorted.sort_unstable();
+
+                    assert_eq!(
+                        hj_sorted, smj_sorted,
+                        "Content mismatch for {join_type:?} batch_size={batch_size} \
+                         left_extra={left_extra} right_extra={right_extra}"
+                    );
+                }
+            }
+        }
+    }
+}
+
 /// Return randomly sized record batches with:
 /// two sorted int32 columns 'a', 'b' ranged from 0..99 as join columns
 /// two random int32 columns 'x', 'y' as other columns
 
@@ -34,7 +34,6 @@ mod cross_join;
 mod hash_join;
 mod nested_loop_join;
 mod piecewise_merge_join;
-pub(crate) mod semi_anti_sort_merge_join;
 mod sort_merge_join;
 mod stream_join_utils;
 mod symmetric_hash_join;