- replace IPCStreamWriter with IPCFileWriter

- remove test related to apache#14868

remove test

Author: zebsme

Cargo.lock

@@ -87,7 +87,7 @@ ahash = { version = "0.8", default-features = false, features = [
     "runtime-rng",
 ] }
 apache-avro = { version = "0.17", default-features = false }
-arrow = { version = "54.2.1", features = [
+arrow = { version = "54.3.0", features = [
     "prettyprint",
     "chrono-tz",
 ] }

Cargo.toml

@@ -60,6 +60,7 @@ hashbrown = { workspace = true }
 indexmap = { workspace = true }
 itertools = { workspace = true, features = ["use_std"] }
 log = { workspace = true }
+memmap2 = "0.9.5"
 parking_lot = { workspace = true }
 pin-project-lite = "^0.2.7"
 tokio = { workspace = true }

datafusion/physical-plan/Cargo.toml

@@ -24,8 +24,6 @@ use std::any::Any;
 use std::cmp::Ordering;
 use std::collections::{HashMap, VecDeque};
 use std::fmt::Formatter;
-use std::fs::File;
-use std::io::BufReader;
 use std::mem::size_of;
 use std::ops::Range;
 use std::pin::Pin;
@@ -51,8 +49,8 @@ use crate::projection::{
 use crate::spill::spill_manager::SpillManager;
 use crate::{
     metrics, DisplayAs, DisplayFormatType, Distribution, ExecutionPlan,
-    ExecutionPlanProperties, PhysicalExpr, PlanProperties, RecordBatchStream,
-    SendableRecordBatchStream, Statistics,
+    ExecutionPlanProperties, IPCBufferDecoder, PhysicalExpr, PlanProperties,
+    RecordBatchStream, SendableRecordBatchStream, Statistics,
 };
 
 use arrow::array::{types::UInt64Type, *};
@@ -61,7 +59,6 @@ use arrow::compute::{
 };
 use arrow::datatypes::{DataType, SchemaRef, TimeUnit};
 use arrow::error::ArrowError;
-use arrow::ipc::reader::StreamReader;
 use datafusion_common::{
     exec_err, internal_err, not_impl_err, plan_err, DataFusionError, HashSet, JoinSide,
     JoinType, Result,
@@ -2279,15 +2276,13 @@ fn fetch_right_columns_from_batch_by_idxs(
             let mut buffered_cols: Vec<ArrayRef> =
                 Vec::with_capacity(buffered_indices.len());
 
-            let file = BufReader::new(File::open(spill_file.path())?);
-            let reader = StreamReader::try_new(file, None)?;
+            let decoder = IPCBufferDecoder::new(spill_file.path());
 
-            for batch in reader {
-                batch?.columns().iter().for_each(|column| {
+            for i in 0..decoder.num_batches(){
+                decoder.get_batch(i)?.unwrap().columns().iter().for_each(|column| {
                     buffered_cols.extend(take(column, &buffered_indices, None))
                 });
             }
-
                 Ok(buffered_cols)
             }
         // Invalid combination

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

@@ -50,7 +50,8 @@ pub use crate::ordering::InputOrderMode;
 pub use crate::stream::EmptyRecordBatchStream;
 pub use crate::topk::TopK;
 pub use crate::visitor::{accept, visit_execution_plan, ExecutionPlanVisitor};
-
+pub use spill::spill_manager::SpillManager;
+pub use spill::IPCBufferDecoder;
 mod ordering;
 mod render_tree;
 mod topk;

datafusion/physical-plan/src/lib.rs

@@ -24,15 +24,15 @@ use arrow::array::RecordBatch;
 use datafusion_common::exec_datafusion_err;
 use datafusion_execution::disk_manager::RefCountedTempFile;
 
-use super::{spill_manager::SpillManager, IPCStreamWriter};
+use super::{spill_manager::SpillManager, IPCFileWriter};
 
 /// Represents an in-progress spill file used for writing `RecordBatch`es to disk, created by `SpillManager`.
 /// Caller is able to use this struct to incrementally append in-memory batches to
 /// the file, and then finalize the file by calling the `finish` method.
 pub struct InProgressSpillFile {
     pub(crate) spill_writer: Arc<SpillManager>,
     /// Lazily initialized writer
-    writer: Option<IPCStreamWriter>,
+    writer: Option<IPCFileWriter>,
     /// Lazily initialized in-progress file, it will be moved out when the `finish` method is invoked
     in_progress_file: Option<RefCountedTempFile>,
 }
@@ -59,7 +59,7 @@ impl InProgressSpillFile {
         if self.writer.is_none() {
             let schema = batch.schema();
             if let Some(ref in_progress_file) = self.in_progress_file {
-                self.writer = Some(IPCStreamWriter::new(
+                self.writer = Some(IPCFileWriter::new(
                     in_progress_file.path(),
                     schema.as_ref(),
                 )?);