Change parquet writers to use standard `std:io::Write` rather custom `ParquetWriter` trait (#1717) (#1163) #1719

tustvold · 2022-05-21T17:41:46Z

Which issue does this PR close?

Part of #1163

Rationale for this change

See tickets, but in short the current write path makes use of a lot of custom IO abstractions which can be hard to use correctly. In particular the use of TryClone can easily lead to races if used to share a file descriptor across threads.

What changes are included in this PR?

This reworks the write path to use std::io::Write and nothing else. Unfortunately to achieve this requires a few changes:

A TrackedWrite that keeps track of how many bytes have been written, allowing removal of Seek
A callback based approach to get metadata from a child writer to its parent
Lifetimes, lots of lifetimes...

This last point becomes a bit obnoxious when it interacts with the RowGroupWriter trait. In order to be object-safe, i.e. possible to construct Box<dyn RowGroupWriter>, RowGroupWriter::close cannot take self by value. This results in explicit scopes, or calls to std::mem::drop in order to truncate its lifetime.

I'm not entirely sure what the purpose of these traits is, but perhaps we could simplify things, and fix this slight lifetime annoyance by removing them? Maybe something for a follow up PR?

Are there any user-facing changes?

Yes, this makes non-trivial changes to the parquet write API.

tustvold · 2022-05-21T17:43:08Z

parquet/src/file/mod.rs

-//! while let Some(mut col_writer) = row_group_writer.next_column().unwrap() {
-//!     // ... write values to a column writer
-//!     row_group_writer.close_column(col_writer).unwrap();
+//! {


This is an example of the somewhat unfortunate manual scoping... I personally think removing the trait indirection would make the code a lot easier to understand...

Which trait indirection are you referring to?

Also, it seems to me like we could change this example to actually run (not sure why it is marked norun -- it just needs to pick a tempfile as target rather than /path/to/sample.parquet)

Which trait indirection are you referring to?

We have FileWriter and RowGroupWriter traits which are then implemented by SerializedFileWriter and SerializedRowGroupWriter.

I'm not sure why you would ever want to override the default implementation, and the need to expose these as object-safe traits prevents FileWriter::close(self) or RowGroupWriter::close(self) which causes lifetime annoyances.

I also find the indirection very confusing in general 😅

Ah, yes, I also find this very confusing -- I think that split was part of the original implementation from @sunchao

Would be possible to create a ParallelizedFileWriter that used multiple threads and buffers internally to write files out with greater concurrency?

It would have been highly non-trivial to do this correctly before, and with the lifetime change in this PR it would actually be impossible, as FileWriter::next_row_group returns a mutable borrow of FileWriter

That being said, nothing would prevent creating a ParallelizedFileWriter, you just wouldn't be able to be generic over parallel and non-parallel implementations. I'm not sure this is actually a problem, especially as any ParallelizedFileWriter is likely to have additional constraints, e.g. Send, async, etc...

I personally think removing these traits as part of the PR would go a long way to making the change easy for users to understand, the lifetime errors are not very obvious unless you know what you are looking for. What do you think?

I am not opposed to removing the FileWriter and RowGroupWriter. I think it would help readability a lot. However I think other reviewers should weigh in.

If we are going to make such an API change, perhaps we should do it in the same release as this change to get it all over with at once.

tustvold · 2022-05-21T17:44:34Z

parquet/src/file/writer.rs

        }

        let metadata = self.row_group_metadata.as_ref().unwrap().clone();
        Ok(metadata)
    }
 }

+/// A wrapper around a [`ColumnWriter`] that invokes a callback on [`Self::close`]


I'm not a massive fan of this, but I was somewhat wary of making changes to the ColumnWriter plumbing as it is used in lots of places

tustvold · 2022-05-21T17:45:37Z

parquet/src/arrow/arrow_writer.rs

@@ -65,7 +67,7 @@ pub struct ArrowWriter<W: ParquetWriter> {
    max_row_group_size: usize,
 }

-impl<W: 'static + ParquetWriter> ArrowWriter<W> {
+impl<W: Write> ArrowWriter<W> {


This seemingly small change means you can pass in &mut std::io::Cursor<Vec<_>> or any other construction which makes this much easier to use

tustvold · 2022-05-21T17:45:51Z

parquet/src/arrow/arrow_writer.rs

@@ -744,15 +748,15 @@ mod tests {
        let expected_batch =
            RecordBatch::try_new(schema.clone(), vec![Arc::new(a), Arc::new(b)]).unwrap();

-        let cursor = InMemoryWriteableCursor::default();
+        let mut cursor = Cursor::new(vec![]);


tustvold · 2022-05-21T17:48:06Z

parquet/src/file/writer.rs

    /// Can be called multiple times. In subsequent calls will result in no-op and return
    /// already created row group metadata.
    fn close(&mut self) -> Result<RowGroupMetaDataPtr>;
 }

+/// Callback invoked on closing a column chunk, arguments are:


We have to move to a callback model as the child writers now mutably borrow from their parents, and so it would be impossible to keep the current interface. I also happen to think this is cleaner, you simply close the ColumnWriter when you are finished, vs passing the ColumnWriter back to the RowGroupWriter.

Also importantly, by keeping a mutable reference it lets the compiler prevent concurrent writes to the same writer, as reported in #1717

codecov-commenter · 2022-05-21T19:25:22Z

Codecov Report

Merging #1719 (63491e5) into master (5cf06bf) will decrease coverage by 0.04%.
The diff coverage is 91.13%.

❗ Current head 63491e5 differs from pull request most recent head d699960. Consider uploading reports for the commit d699960 to get more accurate results

@@            Coverage Diff             @@
##           master    #1719      +/-   ##
==========================================
- Coverage   83.32%   83.27%   -0.05%     
==========================================
  Files         196      195       -1     
  Lines       55961    55896      -65     
==========================================
- Hits        46627    46549      -78     
- Misses       9334     9347      +13

Impacted Files	Coverage Δ
parquet/benches/arrow_writer.rs	`0.00% <0.00%> (ø)`
parquet/src/arrow/array_reader/list_array.rs	`93.41% <ø> (ø)`
parquet/src/arrow/schema.rs	`96.81% <ø> (ø)`
parquet/src/util/cursor.rs	`63.86% <ø> (-13.45%)`	⬇️
parquet/src/util/io.rs	`88.33% <ø> (-1.67%)`	⬇️
parquet_derive/src/lib.rs	`0.00% <ø> (ø)`
parquet_derive/src/parquet_field.rs	`65.98% <0.00%> (ø)`
parquet/src/data_type.rs	`75.84% <50.00%> (ø)`
parquet/src/file/writer.rs	`92.85% <92.17%> (-0.45%)`	⬇️
parquet/src/arrow/arrow_reader.rs	`96.88% <100.00%> (-0.01%)`	⬇️
... and 9 more

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 5cf06bf...d699960. Read the comment docs.

alamb

Thank you @tustvold !

All in all I think this is a great improvement -- It is great to see that the API now reflects the "write only a single row group at a time" restriction in a way that the compiler can enforce. This also seems like a nice step towards tracking information (like bytes written) in a way that I can imagine actually parallelizing the encoding.

I do think it might be worth a little more investment to make the change "less breaking" (by marking some structs as deprecated for a while)

cc @pacman82 @manojkarthick @nevi-me

alamb · 2022-05-23T10:29:34Z

parquet/benches/arrow_writer.rs

@@ -278,8 +276,8 @@ fn _create_nested_bench_batch(
 #[inline]
 fn write_batch(batch: &RecordBatch) -> Result<()> {
    // Write batch to an in-memory writer
-    let cursor = InMemoryWriteableCursor::default();
-    let mut writer = ArrowWriter::try_new(cursor, batch.schema(), None)?;
+    let buffer = vec![];


here is a nice example of the new API in action: use something that does std::io::Write

alamb · 2022-05-23T10:31:24Z

parquet/src/util/cursor.rs

@@ -133,68 +131,6 @@ impl Seek for SliceableCursor {
    }
 }

-/// Use this type to write Parquet to memory rather than a file.
-#[derive(Debug, Default, Clone)]


In terms of easing the transition of this change, what would you think about marking this struct "deprecated" and keeping the code backwards compatible for a few releases (e.g. impl Wite for InMemoryWriteableCursor)?

That might make the API change easier to manage and give users some time to remove it.

alamb · 2022-05-23T10:32:26Z

parquet/src/util/io.rs

@@ -153,47 +153,6 @@ impl<R: ParquetReader> Length for FileSource<R> {
        self.end - self.start
    }
 }
-
-/// Struct that represents `File` output stream with position tracking.
-/// Used as a sink in file writer.


Likewise, I recommend considering leaving this structure in (and marking it as "deprecated") for a release or two to give people a chance to update their code rather over time.

alamb · 2022-05-23T10:34:53Z

parquet_derive_test/src/lib.rs

+            drs.as_slice()
+                .write_to_row_group(row_group.as_mut())
+                .unwrap();
+            row_group.close().unwrap();


I do like this pattern better (close the row group writer rather than passing it back to the parquet file writer to close)

alamb · 2022-05-23T10:35:52Z

parquet/src/file/mod.rs

-//! while let Some(mut col_writer) = row_group_writer.next_column().unwrap() {
-//!     // ... write values to a column writer
-//!     row_group_writer.close_column(col_writer).unwrap();
+//! {


Which trait indirection are you referring to?

alamb · 2022-05-23T10:44:01Z

parquet/src/file/writer.rs

    /// Can be called multiple times. In subsequent calls will result in no-op and return
    /// already created row group metadata.
    fn close(&mut self) -> Result<RowGroupMetaDataPtr>;
 }

+/// Callback invoked on closing a column chunk, arguments are:


Also importantly, by keeping a mutable reference it lets the compiler prevent concurrent writes to the same writer, as reported in #1717

alamb · 2022-05-23T10:46:55Z

parquet/src/file/writer.rs

@@ -196,13 +203,13 @@ impl<W: ParquetWriter> SerializedFileWriter<W> {
        };

        // Write file metadata
-        let start_pos = self.buf.seek(SeekFrom::Current(0))?;
+        let start_pos = self.buf.bytes_written();


alamb · 2022-05-23T10:47:49Z

parquet/src/file/writer.rs

@@ -214,7 +221,7 @@ impl<W: ParquetWriter> SerializedFileWriter<W> {
    }

    #[inline]
-    fn assert_closed(&self) -> Result<()> {
+    fn assert_not_closed(&self) -> Result<()> {


alamb · 2022-05-23T10:49:00Z

parquet/src/file/writer.rs

    pub fn new(
        schema_descr: SchemaDescPtr,
        properties: WriterPropertiesPtr,
-        buf: &W,
+        buf: &'a mut TrackedWrite<W>,
+        on_close: Option<OnCloseRowGroup<'a>>,


why is this Option -- shouldn't it always be required?

So these writers are technically public and I don't really know what people might be using them for, I thought this was a way to make the change less annoying - you can just pass None and move on...

makes sense

alamb · 2022-05-23T10:49:37Z

parquet/src/file/writer.rs

        }
    }

-    /// Checks and finalises current column writer.
-    fn finalise_column_writer(&mut self, writer: ColumnWriter) -> Result<()> {


This code appears to have been inlined into next_column below

parquet/src/arrow/arrow_writer.rs

ahmedriza · 2022-05-23T15:39:32Z

This is a great step forwards, and makes it cleaner as well as let the compiler enforce invariants to avoid things like #1717 happening.

Agree that keeping the current API deprecated will be good to allow users to transition. I've tested our code and the change is minimal from the API perspective.

…riter

tustvold · 2022-05-24T08:03:06Z

I've removed the FileWriter and RowGroupWriter traits as discussed, and deprecated FileSink and InMemoryWriteableCursor.

Assuming CI doesn't find something I've missed I think this is now ready, unless anyone objects I intend to leave it open for another 24 hours and then get it merged.

alamb · 2022-05-24T13:17:05Z

It will be a nice feature in version 15 🎉

nevi-me

I've gone through the changes, and agree that the interface becomes simpler with std::io::Write.

…riter

alamb · 2022-05-25T18:30:58Z

👨‍🍳 👌 -- this will be a very nice improvement in arrow 15 ❤️ -- thank you @tustvold

Rustify parquet writer (apache#1717) (apache#1163)

f0e869a

github-actions bot added the parquet Changes to the parquet crate label May 21, 2022

tustvold commented May 21, 2022

View reviewed changes

tustvold added 2 commits May 21, 2022 19:36

Fix parquet_derive

fe0958c

Fix benches

5d53dc5

github-actions bot added the parquet-derive label May 21, 2022

Fix parquet_derive tests

2620f45

Use raw vec instead of Cursor

c57aa9e

alamb requested a review from sunchao May 23, 2022 10:23

alamb mentioned this pull request May 23, 2022

Trying to write parquet file in parallel results in corrupt file #1717

Closed

alamb changed the title ~~Rustify parquet writer (#1717) (#1163)~~ Change parquet ArrowFileWriter to use standard std:io::Write rather custom ParquetWriter trait (#1717) (#1163) May 23, 2022

alamb approved these changes May 23, 2022

View reviewed changes

alamb mentioned this pull request May 23, 2022

Support writing parquet to stdout #1687

Closed

ahmedriza reviewed May 23, 2022

View reviewed changes

parquet/src/arrow/arrow_writer.rs Outdated Show resolved Hide resolved

tustvold added the api-change Changes to the arrow API label May 23, 2022

tustvold changed the title ~~Change parquet ArrowFileWriter to use standard std:io::Write rather custom ParquetWriter trait (#1717) (#1163)~~ Change parquet writers to use standard std:io::Write rather custom ParquetWriter trait (#1717) (#1163) May 24, 2022

tustvold added 2 commits May 24, 2022 08:32

Merge remote-tracking branch 'upstream/master' into rustify-parquet-w…

e39fa42

…riter

Review feedback

cd5eda8

tustvold force-pushed the rustify-parquet-writer branch from 2716ede to cd5eda8 Compare May 24, 2022 07:58

nevi-me approved these changes May 24, 2022

View reviewed changes

tustvold added 2 commits May 24, 2022 14:40

Fix unnecessary unwrap

196dd76

Merge remote-tracking branch 'upstream/master' into rustify-parquet-w…

d699960

…riter

tustvold merged commit 722fcfc into apache:master May 25, 2022

tustvold mentioned this pull request May 25, 2022

Lack of examples on parquet file write #1745

Open

tustvold mentioned this pull request Jun 2, 2022

MINOR: Remove uses of TryClone for Parquet apache/datafusion#2681

Merged

tustvold mentioned this pull request Jun 30, 2022

close function instead of mutable reference #1969

Closed

This was referenced Aug 8, 2022

Remove deprecated SliceableCursor and InMemoryWriteableCursor #2378

Merged

Remove deprecated ParquetWriter #2380

Merged

tustvold mentioned this pull request Aug 24, 2022

Drop Seek requirement for ParquetWriter #937

Closed

This was referenced Oct 26, 2022

SerializedFileWriter comments about multiple call on consumed self #2935

Closed

Cleanup orphaned doc comments (#2935) #2938

Merged

Allow Position to support arbitrary Cursor type #61

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Change parquet writers to use standard `std:io::Write` rather custom `ParquetWriter` trait (#1717) (#1163) #1719

Change parquet writers to use standard `std:io::Write` rather custom `ParquetWriter` trait (#1717) (#1163) #1719

tustvold commented May 21, 2022

tustvold May 21, 2022

alamb May 23, 2022

alamb May 23, 2022

tustvold May 23, 2022

alamb May 23, 2022

tustvold May 23, 2022 •

edited

Loading

alamb May 23, 2022

tustvold May 21, 2022

tustvold May 21, 2022

tustvold May 21, 2022

tustvold May 21, 2022

alamb May 23, 2022

codecov-commenter commented May 21, 2022 •

edited

Loading

alamb left a comment

alamb May 23, 2022

alamb May 23, 2022

alamb May 23, 2022

alamb May 23, 2022

alamb May 23, 2022

alamb May 23, 2022

alamb May 23, 2022

alamb May 23, 2022

alamb May 23, 2022

tustvold May 23, 2022

alamb May 23, 2022

alamb May 23, 2022

ahmedriza commented May 23, 2022 •

edited

Loading

tustvold commented May 24, 2022

alamb commented May 24, 2022

nevi-me left a comment

alamb commented May 25, 2022

Change parquet writers to use standard std:io::Write rather custom ParquetWriter trait (#1717) (#1163) #1719

Change parquet writers to use standard std:io::Write rather custom ParquetWriter trait (#1717) (#1163) #1719

Conversation

tustvold commented May 21, 2022

Which issue does this PR close?

Rationale for this change

What changes are included in this PR?

Are there any user-facing changes?

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tustvold May 23, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

codecov-commenter commented May 21, 2022 • edited Loading

Codecov Report

alamb left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ahmedriza commented May 23, 2022 • edited Loading

tustvold commented May 24, 2022

alamb commented May 24, 2022

nevi-me left a comment

Choose a reason for hiding this comment

alamb commented May 25, 2022

Change parquet writers to use standard `std:io::Write` rather custom `ParquetWriter` trait (#1717) (#1163) #1719

Change parquet writers to use standard `std:io::Write` rather custom `ParquetWriter` trait (#1717) (#1163) #1719

tustvold May 23, 2022 •

edited

Loading

codecov-commenter commented May 21, 2022 •

edited

Loading

ahmedriza commented May 23, 2022 •

edited

Loading