Repository: jorgecarleitao/arrow2
Pull request #1035 (2022-05-31 18:01:06)

Bumped parquet2
This adds a couple of improvements, including control of compression level to gzip and brotli.

# Backward incompatible change

The method `start` no longer exists nor is necessary - just remove it and all is good.

codecov[bot]: # [Codecov](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1035?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao) Report
> Merging [#1035](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1035?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao) (01ed472) into [main](https://codecov.io/gh/jorgecarleitao/arrow2/commit/47d5bd2a6deb1b57fe171249e1be0b9fe8ce92d9?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao) (47d5bd2) will **decrease** coverage by `0.00%`.
> The diff coverage is `100.00%`.

```diff
@@            Coverage Diff             @@
##             main    #1035      +/-   ##
==========================================
- Coverage   81.43%   81.42%   -0.01%     
==========================================
  Files         360      360              
  Lines       34333    34320      -13     
==========================================
- Hits        27958    27946      -12     
+ Misses       6375     6374       -1     
```


| [Impacted Files](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1035?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao) | Coverage Δ | |
|---|---|---|
| [src/io/parquet/write/file.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1035/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2lvL3BhcnF1ZXQvd3JpdGUvZmlsZS5ycw==) | `81.25% <ø> (-1.11%)` | :arrow_down: |
| [src/io/parquet/write/sink.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1035/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2lvL3BhcnF1ZXQvd3JpdGUvc2luay5ycw==) | `75.96% <100.00%> (-1.24%)` | :arrow_down: |
| [src/io/avro/read\_async/metadata.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1035/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2lvL2F2cm8vcmVhZF9hc3luYy9tZXRhZGF0YS5ycw==) | `96.77% <0.00%> (-3.23%)` | :arrow_down: |
| [src/io/ipc/read/stream\_async.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1035/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2lvL2lwYy9yZWFkL3N0cmVhbV9hc3luYy5ycw==) | `74.38% <0.00%> (+0.82%)` | :arrow_up: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1035?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1035?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao). Last update [47d5bd2...01ed472](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1035?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao).


Patch:
diff --git a/Cargo.toml b/Cargo.toml
index 397b7dabc3..09fae9293a 100644
--- a/Cargo.toml
+++ b/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 }
diff --git a/benches/write_parquet.rs b/benches/write_parquet.rs
index 5bd2848c2e..b1f5ecfd5e 100644
--- a/benches/write_parquet.rs
+++ b/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?)?;
     }
diff --git a/examples/parquet_write.rs b/examples/parquet_write.rs
index ecbbe486f2..b3baf3bf8f 100644
--- a/examples/parquet_write.rs
+++ b/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?)?;
     }
diff --git a/examples/parquet_write_parallel/src/main.rs b/examples/parquet_write_parallel/src/main.rs
index a313196419..56230b4a4b 100644
--- a/examples/parquet_write_parallel/src/main.rs
+++ b/examples/parquet_write_parallel/src/main.rs
@@ -97,7 +97,6 @@ fn parallel_write(path: &str, schema: &Schema, batches: &[Chunk]) -> Result<()>
     let mut writer = FileWriter::try_new(file, schema, options)?;
 
     // Write the file.
-    writer.start()?;
     for group in row_groups {
         writer.write(group?)?;
     }
diff --git a/src/doc/lib.md b/src/doc/lib.md
index e27e0ff2c3..dbb013cfbd 100644
--- a/src/doc/lib.md
+++ b/src/doc/lib.md
@@ -60,7 +60,6 @@ fn main() -> Result<()> {
     let mut writer = FileWriter::try_new(file, schema, options)?;
 
     // Write the file.
-    writer.start()?;
     for group in row_groups {
         writer.write(group?)?;
     }
diff --git a/src/io/parquet/write/file.rs b/src/io/parquet/write/file.rs
index 21bfe55d5d..42153a93a2 100644
--- a/src/io/parquet/write/file.rs
+++ b/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)?)
diff --git a/src/io/parquet/write/sink.rs b/src/io/parquet/write/sink.rs
index 8a953162aa..1f3e44fa64 100644
--- a/src/io/parquet/write/sink.rs
+++ b/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(()))


Test patch:
diff --git a/arrow-parquet-integration-testing/src/main.rs b/arrow-parquet-integration-testing/src/main.rs
index eef4acc9cb..887ba77d45 100644
--- a/arrow-parquet-integration-testing/src/main.rs
+++ b/arrow-parquet-integration-testing/src/main.rs
@@ -194,7 +194,6 @@ fn main() -> Result<()> {
 
     let mut writer = FileWriter::try_new(writer, schema, options)?;
 
-    writer.start()?;
     for group in row_groups {
         writer.write(group?)?;
     }
diff --git a/tests/it/io/parquet/mod.rs b/tests/it/io/parquet/mod.rs
index 68a0ea43e0..79951951e2 100644
--- a/tests/it/io/parquet/mod.rs
+++ b/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?)?;
     }
diff --git a/tests/it/io/parquet/read_indexes.rs b/tests/it/io/parquet/read_indexes.rs
index d6e4970775..6dd88618ae 100644
--- a/tests/it/io/parquet/read_indexes.rs
+++ b/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();
diff --git a/tests/it/io/parquet/write.rs b/tests/it/io/parquet/write.rs
index 2c41aac5fa..471b9d72af 100644
--- a/tests/it/io/parquet/write.rs
+++ b/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?)?;
     }