Make GenericColumnWriter Send

parquet/src/arrow/arrow_writer/mod.rs

@@ -117,7 +117,7 @@ impl<W: Write> Debug for ArrowWriter<W> {
     }
 }
 
-impl<W: Write> ArrowWriter<W> {
+impl<W: Write + Send> ArrowWriter<W> {
     /// Try to create a new Arrow writer
     ///
     /// The writer will fail if:
@@ -273,7 +273,7 @@ impl<W: Write> ArrowWriter<W> {
     }
 }
 
-impl<W: Write> RecordBatchWriter for ArrowWriter<W> {
+impl<W: Write + Send> RecordBatchWriter for ArrowWriter<W> {
     fn write(&mut self, batch: &RecordBatch) -> Result<(), ArrowError> {
         self.write(batch).map_err(|e| e.into())
     }
@@ -284,7 +284,7 @@ impl<W: Write> RecordBatchWriter for ArrowWriter<W> {
     }
 }
 
-fn write_leaves<W: Write>(
+fn write_leaves<W: Write + Send>(
     row_group_writer: &mut SerializedRowGroupWriter<'_, W>,
     arrays: &[ArrayRef],
     levels: &mut [Vec<LevelInfo>],

parquet/src/column/page.rs

@@ -248,7 +248,7 @@ pub trait PageReader: Iterator<Item = Result<Page>> + Send {
 ///
 /// It is reasonable to assume that all pages will be written in the correct order, e.g.
 /// dictionary page followed by data pages, or a set of data pages, etc.
-pub trait PageWriter {
+pub trait PageWriter: Send {
     /// Writes a page into the output stream/sink.
     /// Returns `PageWriteSpec` that contains information about written page metrics,
     /// including number of bytes, size, number of values, offset, etc.

parquet/src/column/writer/mod.rs

@@ -2174,6 +2174,12 @@ mod tests {
         );
     }
 
+    #[test]
+    fn test_send() {
+        fn test<T: Send>() {}
+        test::<ColumnWriterImpl<Int32Type>>();
+    }
+
     /// Performs write-read roundtrip with randomly generated values and levels.
     /// `max_size` is maximum number of values or levels (if `max_def_level` > 0) to write
     /// for a column.

parquet/src/encodings/encoding/mod.rs

@@ -40,7 +40,7 @@ mod dict_encoder;
 ///
 /// Currently this allocates internal buffers for the encoded values. After done putting
 /// values, caller should call `flush_buffer()` to get an immutable buffer pointer.
-pub trait Encoder<T: DataType> {
+pub trait Encoder<T: DataType>: Send {
     /// Encodes data from `values`.
     fn put(&mut self, values: &[T::T]) -> Result<()>;
 

parquet/src/file/writer.rs

@@ -160,7 +160,7 @@ impl<W: Write> Debug for SerializedFileWriter<W> {
     }
 }
 
-impl<W: Write> SerializedFileWriter<W> {
+impl<W: Write + Send> SerializedFileWriter<W> {
     /// Creates new file writer.
     pub fn new(buf: W, schema: TypePtr, properties: WriterPropertiesPtr) -> Result<Self> {
         let mut buf = TrackedWrite::new(buf);
@@ -406,7 +406,7 @@ pub struct SerializedRowGroupWriter<'a, W: Write> {
     on_close: Option<OnCloseRowGroup<'a>>,
 }
 
-impl<'a, W: Write> SerializedRowGroupWriter<'a, W> {
+impl<'a, W: Write + Send> SerializedRowGroupWriter<'a, W> {
     /// Creates a new `SerializedRowGroupWriter` with:
     ///
     /// - `schema_descr` - the schema to write
@@ -700,7 +700,7 @@ impl<'a, W: Write> SerializedPageWriter<'a, W> {
     }
 }
 
-impl<'a, W: Write> PageWriter for SerializedPageWriter<'a, W> {
+impl<'a, W: Write + Send> PageWriter for SerializedPageWriter<'a, W> {
     fn write_page(&mut self, page: CompressedPage) -> Result<PageWriteSpec> {
         let uncompressed_size = page.uncompressed_size();
         let compressed_size = page.compressed_size();
@@ -1336,7 +1336,7 @@ mod tests {
         compression: Compression,
     ) -> crate::format::FileMetaData
     where
-        W: Write,
+        W: Write + Send,
         R: ChunkReader + From<W> + 'static,
     {
         test_roundtrip::<W, R, Int32Type, _>(
@@ -1356,7 +1356,7 @@ mod tests {
         compression: Compression,
     ) -> crate::format::FileMetaData
     where
-        W: Write,
+        W: Write + Send,
         R: ChunkReader + From<W> + 'static,
         D: DataType,
         F: Fn(Row) -> D::T,

parquet/src/record/record_writer.rs

@@ -21,7 +21,7 @@ use super::super::errors::ParquetError;
 use super::super::file::writer::SerializedRowGroupWriter;
 
 pub trait RecordWriter<T> {
-    fn write_to_row_group<W: std::io::Write>(
+    fn write_to_row_group<W: std::io::Write + Send>(
         &self,
         row_group_writer: &mut SerializedRowGroupWriter<W>,
     ) -> Result<(), ParquetError>;

parquet_derive/src/lib.rs

@@ -96,7 +96,7 @@ pub fn parquet_record_writer(input: proc_macro::TokenStream) -> proc_macro::Toke
 
     (quote! {
     impl #generics ::parquet::record::RecordWriter<#derived_for #generics> for &[#derived_for #generics] {
-      fn write_to_row_group<W: ::std::io::Write>(
+      fn write_to_row_group<W: ::std::io::Write + Send>(
         &self,
         row_group_writer: &mut ::parquet::file::writer::SerializedRowGroupWriter<'_, W>
       ) -> Result<(), ::parquet::errors::ParquetError> {

parquet_derive_test/src/lib.rs

@@ -56,8 +56,7 @@ mod tests {
     use std::{env, fs, io::Write, sync::Arc};
 
     use parquet::{
-        file::{properties::WriterProperties, writer::SerializedFileWriter},
-        record::RecordWriter,
+        file::writer::SerializedFileWriter, record::RecordWriter,
         schema::parser::parse_message_type,
     };