From 9884594c1462ba3436aa859a94330c31ebc5aaa1 Mon Sep 17 00:00:00 2001
From: doki <tunyum94@gmail.com>
Date: Fri, 25 Feb 2022 17:42:25 +0800
Subject: [PATCH] remove sync constraint of SendableRecordBatchStream

---
 ballista/rust/core/src/execution_plans/shuffle_reader.rs | 2 +-
 ballista/rust/core/src/utils.rs                          | 8 ++++----
 datafusion/src/physical_plan/mod.rs                      | 2 +-
 3 files changed, 6 insertions(+), 6 deletions(-)

diff --git a/ballista/rust/core/src/execution_plans/shuffle_reader.rs b/ballista/rust/core/src/execution_plans/shuffle_reader.rs
index 7482c1843ed31..f87a3106f73c7 100644
--- a/ballista/rust/core/src/execution_plans/shuffle_reader.rs
+++ b/ballista/rust/core/src/execution_plans/shuffle_reader.rs
@@ -205,7 +205,7 @@ fn stats_for_partitions(
 
 async fn fetch_partition(
     location: &PartitionLocation,
-) -> Result<Pin<Box<dyn RecordBatchStream + Send + Sync>>> {
+) -> Result<Pin<Box<dyn RecordBatchStream + Send>>> {
     let metadata = &location.executor_meta;
     let partition_id = &location.partition_id;
     let mut ballista_client =
diff --git a/ballista/rust/core/src/utils.rs b/ballista/rust/core/src/utils.rs
index fae3826c32d1b..fb0c01ebb39d8 100644
--- a/ballista/rust/core/src/utils.rs
+++ b/ballista/rust/core/src/utils.rs
@@ -71,7 +71,7 @@ use std::time::Instant;
 /// Stream data to disk in Arrow IPC format
 
 pub async fn write_stream_to_disk(
-    stream: &mut Pin<Box<dyn RecordBatchStream + Send + Sync>>,
+    stream: &mut Pin<Box<dyn RecordBatchStream + Send>>,
     path: &str,
     disk_write_metric: &metrics::Time,
 ) -> Result<PartitionStats> {
@@ -110,7 +110,7 @@ pub async fn write_stream_to_disk(
 }
 
 pub async fn collect_stream(
-    stream: &mut Pin<Box<dyn RecordBatchStream + Send + Sync>>,
+    stream: &mut Pin<Box<dyn RecordBatchStream + Send>>,
 ) -> Result<Vec<RecordBatch>> {
     let mut batches = vec![];
     while let Some(batch) = stream.next().await {
@@ -322,13 +322,13 @@ impl<T: 'static + AsLogicalPlan> QueryPlanner for BallistaQueryPlanner<T> {
 }
 
 pub struct WrappedStream {
-    stream: Pin<Box<dyn Stream<Item = ArrowResult<RecordBatch>> + Send + Sync>>,
+    stream: Pin<Box<dyn Stream<Item = ArrowResult<RecordBatch>> + Send>>,
     schema: SchemaRef,
 }
 
 impl WrappedStream {
     pub fn new(
-        stream: Pin<Box<dyn Stream<Item = ArrowResult<RecordBatch>> + Send + Sync>>,
+        stream: Pin<Box<dyn Stream<Item = ArrowResult<RecordBatch>> + Send>>,
         schema: SchemaRef,
     ) -> Self {
         Self { stream, schema }
diff --git a/datafusion/src/physical_plan/mod.rs b/datafusion/src/physical_plan/mod.rs
index 32f2d5b53087a..a4d4f5c5859db 100644
--- a/datafusion/src/physical_plan/mod.rs
+++ b/datafusion/src/physical_plan/mod.rs
@@ -51,7 +51,7 @@ pub trait RecordBatchStream: Stream<Item = ArrowResult<RecordBatch>> {
 }
 
 /// Trait for a stream of record batches.
-pub type SendableRecordBatchStream = Pin<Box<dyn RecordBatchStream + Send + Sync>>;
+pub type SendableRecordBatchStream = Pin<Box<dyn RecordBatchStream + Send>>;
 
 /// EmptyRecordBatchStream can be used to create a RecordBatchStream
 /// that will produce no results