Revert "refactor: Update SortMergeJoin to use async spill abstractions and remove open_sync"

This reverts commit 086632a.

Author: pantShrey

datafusion/execution/src/disk_manager.rs

@@ -636,6 +636,11 @@ impl SpillFile for RefCountedTempFile {
             current_file_disk_usage: Arc::clone(&self.current_file_disk_usage),
         }))
     }
+
+    fn open_sync_reader(&self) -> Result<Box<dyn std::io::Read + Send>> {
+        let file = std::fs::File::open(self.path())?;
+        Ok(Box::new(file))
+    }
 }
 #[cfg(test)]
 mod tests {

datafusion/execution/src/spill_file.rs

@@ -38,6 +38,19 @@ pub trait SpillFile: Send + Sync {
 
     /// Opens a writer for appending data to this file.
     fn open_writer(&self) -> Result<Box<dyn SpillWriter>>;
+
+    /// Opens a synchronous reader for this file.
+    /// Used by legacy operators (like SortMergeJoin) that haven't been fully migrated to async.
+    ///
+    /// Backends that only support async reads should leave this default implementation,
+    /// which will safely return a NotImplemented error if used in synchronous contexts.
+    fn open_sync_reader(&self) -> Result<Box<dyn std::io::Read + Send>> {
+        datafusion_common::exec_err!(
+            "Synchronous reads are not supported by this spill backend. \
+            This backend cannot be used with synchronous operators like SortMergeJoin \
+            until they are refactored to be fully asynchronous."
+        )
+    }
 }
 
 /// Writer for spill file backends.

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

@@ -119,6 +119,7 @@
 //! factor than the pair-materialization approach.
 
 use std::cmp::Ordering;
+use std::io::BufReader;
 use std::pin::Pin;
 use std::sync::Arc;
 use std::task::{Context, Poll};
@@ -132,6 +133,7 @@ use crate::{EmptyRecordBatchStream, RecordBatchStream};
 use arrow::array::{Array, ArrayRef, BooleanArray, BooleanBufferBuilder, RecordBatch};
 use arrow::compute::{BatchCoalescer, SortOptions, filter_record_batch, not};
 use arrow::datatypes::SchemaRef;
+use arrow::ipc::reader::StreamReader;
 use arrow::util::bit_chunk_iterator::UnalignedBitChunk;
 use arrow::util::bit_util::apply_bitwise_binary_op;
 use datafusion_common::{
@@ -255,11 +257,6 @@ pub(crate) struct BitwiseSortMergeJoinStream {
     inner_key_buffer: Vec<RecordBatch>,
     inner_key_spill: Option<Arc<dyn SpillFile>>,
 
-    //Track the active spill_stream
-    spill_stream: Option<SendableRecordBatchStream>,
-    //Prevents wiping out the buffer if we yield while evaluating the filter
-    inner_group_buffered: bool,
-
     // True when buffer_inner_key_group returned Pending after partially
     // filling inner_key_buffer. On re-entry, buffer_inner_key_group
     // must skip clear() and resume from poll_next_inner_batch (the
@@ -372,8 +369,6 @@ impl BitwiseSortMergeJoinStream {
             matched: BooleanBufferBuilder::new(0),
             inner_key_buffer: vec![],
             inner_key_spill: None,
-            spill_stream: None,
-            inner_group_buffered: false,
             buffering_inner_pending: false,
             pending_boundary: None,
             on_outer,
@@ -471,8 +466,6 @@ impl BitwiseSortMergeJoinStream {
     fn clear_inner_key_group(&mut self) {
         self.inner_key_buffer.clear();
         self.inner_key_spill = None;
-        self.spill_stream = None;
-        self.inner_group_buffered = false;
         self.inner_buffer_size = 0;
     }
 
@@ -754,10 +747,7 @@ impl BitwiseSortMergeJoinStream {
     /// Process a key match with a filter. For each inner row in the buffered
     /// key group, evaluates the filter against the outer key group and ORs
     /// the results into the matched bitset using u64-chunked bitwise ops.
-    fn process_key_match_with_filter(
-        &mut self,
-        cx: &mut Context<'_>,
-    ) -> Poll<Result<()>> {
+    fn process_key_match_with_filter(&mut self) -> Result<()> {
         self.get_outer_self_cmp()?;
         let filter = self.filter.as_ref().unwrap();
         let outer_batch = self.outer_batch.as_ref().unwrap();
@@ -793,40 +783,25 @@ impl BitwiseSortMergeJoinStream {
         )
         .count_ones();
 
-        // Process spilled inner batches first asynchronously.
-        if self.inner_key_spill.is_some() || self.spill_stream.is_some() {
-            if self.spill_stream.is_none()
-                && let Some(spill_file) = &self.inner_key_spill
-            {
-                let stream = self
-                    .spill_manager
-                    .read_spill_as_stream(Arc::clone(spill_file), None)?;
-                self.spill_stream = Some(stream);
-            }
-
-            while matched_count < outer_group_len {
-                let stream = self.spill_stream.as_mut().unwrap();
-                match ready!(stream.poll_next_unpin(cx)) {
-                    Some(Ok(inner_slice)) => {
-                        matched_count = eval_filter_for_inner_slice(
-                            self.outer_is_left,
-                            filter,
-                            &outer_slice,
-                            &inner_slice,
-                            &mut self.matched,
-                            self.outer_offset,
-                            outer_group_len,
-                            matched_count,
-                        )?;
-                    }
-                    Some(Err(e)) => {
-                        self.spill_stream = None;
-                        return Poll::Ready(Err(e));
-                    }
-                    None => {
-                        self.spill_stream = None;
-                        break;
-                    }
+        // Process spilled inner batches first (read back from disk).
+        if let Some(spill_file) = &self.inner_key_spill {
+            let sync_reader = spill_file.open_sync_reader()?;
+            let file = BufReader::new(sync_reader);
+            let reader = StreamReader::try_new(file, None)?;
+            for batch_result in reader {
+                let inner_slice = batch_result?;
+                matched_count = eval_filter_for_inner_slice(
+                    self.outer_is_left,
+                    filter,
+                    &outer_slice,
+                    &inner_slice,
+                    &mut self.matched,
+                    self.outer_offset,
+                    outer_group_len,
+                    matched_count,
+                )?;
+                if matched_count == outer_group_len {
+                    break;
                 }
             }
         }
@@ -854,16 +829,13 @@ impl BitwiseSortMergeJoinStream {
         }
 
         self.outer_offset = outer_group_end;
-
-        self.spill_stream = None;
-
-        Poll::Ready(Ok(()))
+        Ok(())
     }
 
     /// Continue processing an outer key group that spans multiple outer
     /// batches. Returns `true` if this outer batch was fully consumed
     /// by the key group and the caller should load another.
-    fn resume_boundary(&mut self, cx: &mut Context<'_>) -> Poll<Result<bool>> {
+    fn resume_boundary(&mut self) -> Result<bool> {
         debug_assert!(
             self.outer_batch.is_some(),
             "caller must load outer_batch first"
@@ -885,7 +857,7 @@ impl BitwiseSortMergeJoinStream {
                         });
                         self.emit_outer_batch()?;
                         self.outer_batch = None;
-                        return Poll::Ready(Ok(true));
+                        return Ok(true);
                     }
                 }
             }
@@ -901,79 +873,22 @@ impl BitwiseSortMergeJoinStream {
                     self.null_equality,
                 )?;
                 if same_key {
-                    match self.process_key_match_with_filter(cx) {
-                        Poll::Ready(Ok(())) => (),
-                        Poll::Ready(Err(e)) => return Poll::Ready(Err(e)),
-                        Poll::Pending => {
-                            self.pending_boundary =
-                                Some(PendingBoundary::Filtered { saved_keys });
-                            return Poll::Pending;
-                        }
-                    }
+                    self.process_key_match_with_filter()?;
                     let num_outer = self.outer_batch.as_ref().unwrap().num_rows();
                     if self.outer_offset >= num_outer {
                         self.pending_boundary = Some(PendingBoundary::Filtered {
                             saved_keys: slice_keys(&self.outer_key_arrays, num_outer - 1),
                         });
                         self.emit_outer_batch()?;
                         self.outer_batch = None;
-                        return Poll::Ready(Ok(true));
+                        return Ok(true);
                     }
                 }
                 self.clear_inner_key_group();
             }
             None => {}
         }
-        Poll::Ready(Ok(false))
-    }
-
-    /// Helper to process an Equal match across potential outer batch boundaries.
-    fn process_filtered_match_loop(&mut self, cx: &mut Context<'_>) -> Poll<Result<()>> {
-        loop {
-            ready!(self.process_key_match_with_filter(cx))?;
-
-            let outer_batch = self.outer_batch.as_ref().unwrap();
-            if self.outer_offset >= outer_batch.num_rows() {
-                let saved_keys =
-                    slice_keys(&self.outer_key_arrays, outer_batch.num_rows() - 1);
-
-                self.emit_outer_batch()?;
-                self.pending_boundary = Some(PendingBoundary::Filtered { saved_keys });
-
-                // Clear stale batch before polling
-                self.outer_batch = None;
-
-                match ready!(self.poll_next_outer_batch(cx)) {
-                    Err(e) => return Poll::Ready(Err(e)),
-                    Ok(false) => {
-                        self.pending_boundary = None;
-                        break;
-                    }
-                    Ok(true) => {
-                        let Some(PendingBoundary::Filtered { saved_keys }) =
-                            self.pending_boundary.take()
-                        else {
-                            unreachable!()
-                        };
-                        let same = keys_match(
-                            &saved_keys,
-                            &self.outer_key_arrays,
-                            &self.sort_options,
-                            self.null_equality,
-                        )?;
-                        if same {
-                            continue;
-                        }
-                        break;
-                    }
-                }
-            } else {
-                break;
-            }
-        }
-
-        self.clear_inner_key_group(); // This resets inner_group_buffered to false
-        Poll::Ready(Ok(()))
+        Ok(false)
     }
 
     /// Main loop: drive the merge-scan to produce output batches.
@@ -995,21 +910,14 @@ impl BitwiseSortMergeJoinStream {
                         }
                         return Poll::Ready(Ok(None));
                     }
-                    Ok(true) => {} // Loaded batch, move on to checks
+                    Ok(true) => {
+                        if self.resume_boundary()? {
+                            continue;
+                        }
+                    }
                 }
             }
 
-            // Handles pausing while fetching a NEW outer batch.
-            if self.pending_boundary.is_some() && ready!(self.resume_boundary(cx))? {
-                continue;
-            }
-
-            // Handles pausing while reading the disk stream mid-batch.
-            if self.inner_group_buffered {
-                ready!(self.process_filtered_match_loop(cx))?;
-                continue;
-            }
-
             // 2. Ensure we have an inner batch (unless inner is exhausted).
             // Skip this when resuming a pending boundary — inner was already
             // advanced past the key group before the boundary loop started.
@@ -1134,17 +1042,65 @@ impl BitwiseSortMergeJoinStream {
                 }
                 Ordering::Equal => {
                     if self.filter.is_some() {
-                        debug_assert!(!self.inner_group_buffered);
                         // Buffer inner key group (may span batches)
                         match ready!(self.buffer_inner_key_group(cx)) {
                             Err(e) => return Poll::Ready(Err(e)),
-                            Ok(_inner_exhausted) => {
-                                self.inner_group_buffered = true;
-                            }
+                            Ok(_inner_exhausted) => {}
                         }
+
                         // Process outer rows against buffered inner group
                         // (may need to handle outer batch boundary)
-                        ready!(self.process_filtered_match_loop(cx))?;
+                        loop {
+                            self.process_key_match_with_filter()?;
+
+                            let outer_batch = self.outer_batch.as_ref().unwrap();
+                            if self.outer_offset >= outer_batch.num_rows() {
+                                let saved_keys = slice_keys(
+                                    &self.outer_key_arrays,
+                                    outer_batch.num_rows() - 1,
+                                );
+
+                                self.emit_outer_batch()?;
+                                debug_assert!(
+                                    !self.inner_key_buffer.is_empty()
+                                        || self.inner_key_spill.is_some(),
+                                    "Filtered pending boundary requires inner key data in buffer or spill"
+                                );
+                                self.pending_boundary =
+                                    Some(PendingBoundary::Filtered { saved_keys });
+
+                                match ready!(self.poll_next_outer_batch(cx)) {
+                                    Err(e) => return Poll::Ready(Err(e)),
+                                    Ok(false) => {
+                                        self.pending_boundary = None;
+                                        self.outer_batch = None;
+                                        break;
+                                    }
+                                    Ok(true) => {
+                                        let Some(PendingBoundary::Filtered {
+                                            saved_keys,
+                                        }) = self.pending_boundary.take()
+                                        else {
+                                            unreachable!()
+                                        };
+                                        let same = keys_match(
+                                            &saved_keys,
+                                            &self.outer_key_arrays,
+                                            &self.sort_options,
+                                            self.null_equality,
+                                        )?;
+                                        if same {
+                                            continue;
+                                        }
+                                        break;
+                                    }
+                                }
+                            } else {
+                                break;
+                            }
+                        }
+
+                        self.clear_inner_key_group();
                     } else {
                         // No filter: advance inner past key group, then
                         // mark all outer rows with this key as matched.