Bumped parquet2

Cargo.toml

@@ -68,7 +68,7 @@ futures = { version = "0.3", optional = true }
 ahash = { version = "0.7", optional = true }
 
 # parquet support
-parquet2 = { version = "0.12", optional = true, default_features = false, features = ["stream"] }
+parquet2 = { version = "0.13", optional = true, default_features = false }
 
 # avro support
 avro-schema = { version = "0.2", optional = true }

benches/write_parquet.rs

@@ -32,7 +32,6 @@ fn write(array: &dyn Array, encoding: Encoding) -> Result<()> {
 
     let mut writer = FileWriter::try_new(writer, schema, options)?;
 
-    writer.start()?;
     for group in row_groups {
         writer.write(group?)?;
     }

examples/parquet_write.rs

@@ -34,7 +34,6 @@ fn write_batch(path: &str, schema: Schema, columns: Chunk<Arc<dyn Array>>) -> Re
 
     let mut writer = FileWriter::try_new(file, schema, options)?;
 
-    writer.start()?;
     for group in row_groups {
         writer.write(group?)?;
     }

src/io/parquet/write/file.rs

@@ -72,11 +72,6 @@ impl<W: Write> FileWriter<W> {
         })
     }
 
-    /// Writes the header of the file
-    pub fn start(&mut self) -> Result<()> {
-        Ok(self.writer.start()?)
-    }
-
     /// Writes a row group to the file.
     pub fn write(&mut self, row_group: RowGroupIter<'_, Error>) -> Result<()> {
         Ok(self.writer.write(row_group)?)

src/io/parquet/write/sink.rs

@@ -82,7 +82,7 @@ where
     ) -> Result<Self, Error> {
         let parquet_schema = crate::io::parquet::write::to_parquet_schema(&schema)?;
         let created_by = Some("Arrow2 - Native Rust implementation of Arrow".to_string());
-        let mut writer = FileStreamer::new(
+        let writer = FileStreamer::new(
             writer,
             parquet_schema.clone(),
             ParquetWriteOptions {
@@ -91,16 +91,9 @@ where
             },
             created_by,
         );
-        let task = Some(
-            async move {
-                writer.start().await?;
-                Ok(Some(writer))
-            }
-            .boxed(),
-        );
         Ok(Self {
-            writer: None,
-            task,
+            writer: Some(writer),
+            task: None,
             options,
             schema,
             encoding,
@@ -196,7 +189,7 @@ where
         match futures::ready!(this.poll_complete(cx)) {
             Ok(()) => {
                 let writer = this.writer.take();
-                if let Some(writer) = writer {
+                if let Some(mut writer) = writer {
                     let meta = std::mem::take(&mut this.metadata);
                     let metadata = if meta.is_empty() {
                         None
@@ -209,13 +202,10 @@ where
                     };
                     let kv_meta = add_arrow_schema(&this.schema, metadata);
 
-                    this.task = Some(
-                        writer
-                            .end(kv_meta)
-                            .map_ok(|_| None)
-                            .map_err(Error::from)
-                            .boxed(),
-                    );
+                    this.task = Some(Box::pin(async move {
+                        writer.end(kv_meta).map_err(Error::from).await?;
+                        Ok(None)
+                    }));
                     this.poll_complete(cx)
                 } else {
                     Poll::Ready(Ok(()))

tests/it/io/parquet/mod.rs

@@ -815,7 +815,6 @@ fn integration_write(schema: &Schema, batches: &[Chunk<Arc<dyn Array>>]) -> Resu
 
     let mut writer = FileWriter::try_new(writer, schema.clone(), options)?;
 
-    writer.start()?;
     for group in row_groups {
         writer.write(group?)?;
     }

tests/it/io/parquet/read_indexes.rs

@@ -94,7 +94,6 @@ fn read_with_indexes(
     let writer = vec![];
     let mut writer = FileWriter::try_new(writer, schema, options)?;
 
-    writer.start()?;
     writer.write(row_group)?;
     writer.end(None)?;
     let data = writer.into_inner();

tests/it/io/parquet/write.rs

@@ -47,7 +47,6 @@ fn round_trip(
     let writer = Cursor::new(vec![]);
     let mut writer = FileWriter::try_new(writer, schema, options)?;
 
-    writer.start()?;
     for group in row_groups {
         writer.write(group?)?;
     }