fix: allow AsyncArrowWriter to be not Send

Author: jpopesculian

parquet/src/arrow/async_writer/mod.rs

@@ -70,53 +70,53 @@ use crate::{
 use arrow_array::RecordBatch;
 use arrow_schema::SchemaRef;
 use bytes::Bytes;
-use futures::future::BoxFuture;
+use futures::future::LocalBoxFuture;
 use futures::FutureExt;
 use std::mem;
 use tokio::io::{AsyncWrite, AsyncWriteExt};
 
 /// The asynchronous interface used by [`AsyncArrowWriter`] to write parquet files.
-pub trait AsyncFileWriter: Send {
+pub trait AsyncFileWriter {
     /// Write the provided bytes to the underlying writer
     ///
     /// The underlying writer CAN decide to buffer the data or write it immediately.
     /// This design allows the writer implementer to control the buffering and I/O scheduling.
     ///
     /// The underlying writer MAY implement retry logic to prevent breaking users write process.
-    fn write(&mut self, bs: Bytes) -> BoxFuture<'_, Result<()>>;
+    fn write(&mut self, bs: Bytes) -> LocalBoxFuture<'_, Result<()>>;
 
     /// Flush any buffered data to the underlying writer and finish writing process.
     ///
     /// After `complete` returns `Ok(())`, caller SHOULD not call write again.
-    fn complete(&mut self) -> BoxFuture<'_, Result<()>>;
+    fn complete(&mut self) -> LocalBoxFuture<'_, Result<()>>;
 }
 
 impl AsyncFileWriter for Box<dyn AsyncFileWriter + '_> {
-    fn write(&mut self, bs: Bytes) -> BoxFuture<'_, Result<()>> {
+    fn write(&mut self, bs: Bytes) -> LocalBoxFuture<'_, Result<()>> {
         self.as_mut().write(bs)
     }
 
-    fn complete(&mut self) -> BoxFuture<'_, Result<()>> {
+    fn complete(&mut self) -> LocalBoxFuture<'_, Result<()>> {
         self.as_mut().complete()
     }
 }
 
-impl<T: AsyncWrite + Unpin + Send> AsyncFileWriter for T {
-    fn write(&mut self, bs: Bytes) -> BoxFuture<'_, Result<()>> {
+impl<T: AsyncWrite + Unpin> AsyncFileWriter for T {
+    fn write(&mut self, bs: Bytes) -> LocalBoxFuture<'_, Result<()>> {
         async move {
             self.write_all(&bs).await?;
             Ok(())
         }
-        .boxed()
+        .boxed_local()
     }
 
-    fn complete(&mut self) -> BoxFuture<'_, Result<()>> {
+    fn complete(&mut self) -> LocalBoxFuture<'_, Result<()>> {
         async move {
             self.flush().await?;
             self.shutdown().await?;
             Ok(())
         }
-        .boxed()
+        .boxed_local()
     }
 }
 

parquet/src/arrow/async_writer/store.rs

@@ -16,7 +16,7 @@
 // under the License.
 
 use bytes::Bytes;
-use futures::future::BoxFuture;
+use futures::future::LocalBoxFuture;
 use std::sync::Arc;
 
 use crate::arrow::async_writer::AsyncFileWriter;
@@ -93,7 +93,7 @@ impl ParquetObjectWriter {
 }
 
 impl AsyncFileWriter for ParquetObjectWriter {
-    fn write(&mut self, bs: Bytes) -> BoxFuture<'_, Result<()>> {
+    fn write(&mut self, bs: Bytes) -> LocalBoxFuture<'_, Result<()>> {
         Box::pin(async {
             self.w
                 .put(bs)
@@ -102,7 +102,7 @@ impl AsyncFileWriter for ParquetObjectWriter {
         })
     }
 
-    fn complete(&mut self) -> BoxFuture<'_, Result<()>> {
+    fn complete(&mut self) -> LocalBoxFuture<'_, Result<()>> {
         Box::pin(async {
             self.w
                 .shutdown()