Repository navigation
Conversation
github-actions
Bot
added
documentation
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #25041 +/- ##
========================================
Coverage 82.63% 82.64%
========================================
Files 1147 1147
Lines 445392 445596 +204
Branches 445392 445596 +204
========================================
+ Hits 368070 368249 +179
- Misses 54973 54989 +16
- Partials 22349 22358 +9 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
| } | ||
|
|
||
| #[tokio::test(flavor = "multi_thread", worker_threads = 4)] | ||
| async fn byte_limit_finishes_a_group_before_more_input_arrives() { |
There was a problem hiding this comment.
Could we add an end-to-end test with a failing AsyncWrite for the parallel path while max_row_group_bytes is set? The direct drop(group) test covers worker cleanup well, but it would also be useful to exercise an object-store write failure where output_single_parquet_file_parallelized exits while the dispatcher or finalizer may still be blocked.
It would be good to assert that the error returns promptly and that all pool reservations are released. This is non-blocking since SpawnedTask drop cancellation currently makes this safe, but having the test would lock down the top-level failure lifecycle.
| /// estimated. Subsequent batches are split using the observed average row size, | ||
| /// so this is not a hard byte or memory limit. The parallel writer synchronizes |
There was a problem hiding this comment.
I found this pretty wordy and hard to read. I think the old text is clearer as it focuses on the user visible behavior (the memory usage for wide schemas) while this description describes a lot of internal implementation detail I don't think is particularly helpful to users
For example " Subsequent batches are split using the observed average row size, so this is not a hard byte or memory limit." is a lot of words, and I think we could simplify it by just adding "best effort" to the summary if you wanted to emphasize this point
"/// (writing) Target maximum estimated encoded size of each row group in bytes (best effort)"
| use std::task::{Context, Poll}; | ||
| use std::time::Duration; | ||
|
|
||
| struct TestOutput(SharedBuffer); |
There was a problem hiding this comment.
These tests seem very verbose to me and they aren't testing the public API surface (aka these are all basically unit tests)
Like @kosiew suggests, I think we should have an "end to end" type test -- namely create a SQL (or DataFrame test) that sets the row group limit and then writes into multiple parquet files and verifies the written row group limit is honored.
I don't think this kind of low level / orchestration test is super useful
There was a problem hiding this comment.
Or put another way, what additional coverage is this adding that is not already covered by the slt test?
| /// Once the channel is exhausted, returns the ArrowColumnWriter. | ||
| /// A leaf array and its number of root records. Nested leaf arrays can have a | ||
| /// different length, so the dispatcher supplies the record count explicitly. | ||
| struct ColumnInput { |
There was a problem hiding this comment.
Why do we need this? Isn't the row count the same across all leaves?
28c48da to
558e83f
Compare
This comment was marked as outdated.
This comment was marked as outdated.
Sorry, something went wrong.
alamb
left a comment
alamb
left a comment
There was a problem hiding this comment.
Thanks @Satyr09 and @kosea -- this code looks much improved and easier to understand.
However, I am concerned about its impact on performance.
I did a brief ad-hoc benchmark locally and I think this change slows down writing substantially when the max row bytes are set (my adhoc testing looked like the binary keeps the cores less busy):
Specifically, I when I wrote it like this it ran 41 seconds on this branch, but 28 seconds on main 🤔
-- parallel writer, byte target set: the new code path
COPY hits TO 'out_parallel_bytes.parquet' STORED AS PARQUET
OPTIONS ('format.max_row_group_bytes' 134217728);Here is the test script (in write_bench.sql)
Details
CREATE EXTERNAL TABLE hits STORED AS PARQUET
LOCATION '/Users/andrewlamb/Software/datafusion/benchmarks/data/hits_partitioned';
-- parallel writer, option unset: should be unchanged by the PR
COPY hits TO 'out_parallel_nobytes.parquet' STORED AS PARQUET;
-- parallel writer, byte target set: the new code path
COPY hits TO 'out_parallel_bytes.parquet' STORED AS PARQUET
OPTIONS ('format.max_row_group_bytes' 134217728);
-- Use smaller table for single threaded test
CREATE EXTERNAL TABLE hits0 STORED AS PARQUET
LOCATION '/Users/andrewlamb/Software/datafusion/benchmarks/data/hits_partitioned/hits_0.parquet';
-- serial writer, with byte target
COPY hits0 TO 'out_serial_bytes.parquet' STORED AS PARQUET
OPTIONS ('format.max_row_group_bytes' 134217728, 'format.allow_single_file_parallelism' false);Then I ran it like datafusion-cli -f write_bench.sql
Raw data for main
Details
andrewlamb@Andrews-MacBook-Pro-3:~/Downloads$ datafusion-cli -f write_bench.sql
DataFusion CLI v55.1.0
0 row(s) fetched.
Elapsed 0.035 seconds.
+----------+
| count |
+----------+
| 99997497 |
+----------+
1 row(s) fetched.
Elapsed 25.003 seconds.
+----------+
| count |
+----------+
| 99997497 |
+----------+
1 row(s) fetched.
Elapsed 25.851 seconds.
0 row(s) fetched.
Elapsed 0.013 seconds.
+---------+
| count |
+---------+
| 1000000 |
+---------+
1 row(s) fetched.
Elapsed 1.679 seconds.Raw data for this branch (note the second query takes 40.9s vs 25.8s on main)
Details
./datafusion-cli-daipayan_parallel-max-row-group-bytes-barrier-corrected -f write_bench.sql
DataFusion CLI v55.1.0
0 row(s) fetched.
Elapsed 0.038 seconds.
+----------+
| count |
+----------+
| 99997497 |
+----------+
1 row(s) fetched.
Elapsed 26.424 seconds.
+----------+
| count |
+----------+
| 99997497 |
+----------+
1 row(s) fetched.
Elapsed 40.912 seconds.
0 row(s) fetched.
Elapsed 0.001 seconds.
+---------+
| count |
+---------+
| 1000000 |
+---------+
1 row(s) fetched.
Elapsed 1.651 seconds.| task.join_unwind() | ||
| .await | ||
| .map_err(|e| DataFusionError::ExecutionJoin(Box::new(e)))??; | ||
| Err(internal_datafusion_err!( |
There was a problem hiding this comment.
I think this should be an execution error (rather than an internal error)
|
Also, it woud be nice if you could respond to any of the comments you have addressed (I think you have addrssed them all, but it would be good if you could note which you think you have covered so we can double check) |
4 participants
Which issue does this PR close?
Closes #22982.
Rationale for this change
max_row_group_bytesis ignored by the parallel Parquet writer, which is enabled by default. This change lets users apply the byte target while keeping parallel column encoding.What changes are included in this PR?
When a byte target is configured, column workers report their estimated encoded sizes. The writer waits for each input slice to finish before choosing the next row-group boundary. With the option unset, no feedback channels are created.
The work is split into three commits, in merge order:
The prerequisites are on separate fork branches. Until they merge into Apache main, GitHub's full diff includes them. Review just the byte-target change.
What is the testing strategy for this PR?
Public DataFrame tests write multiple files, compare serial and parallel row-group layouts, and read back every column. They cover nested data, growing and shrinking string widths, both size limits, and memory cleanup after column and storage failures. SQL tests cover the configuration through COPY.
Each stage passed formatting, strict full-workspace Clippy, and extended workspace tests on Linux. The feature also passed its SQL and documentation checks, ten focused public regressions, and all 255 Parquet integration tests. Validation logs are available on the fork; the final refactor suite passed in a follow-up run.
The existing SQL benchmark workloads passed all 108 output readback checks. Byte-enabled layouts matched serial writing, and disabled-option layouts matched upstream. Timings varied between rounds, so these are preliminary measurements. Peak tracked memory for changing-width strings remained higher: 38.7 MiB versus 14.4 MiB on upstream.
Are there any user-facing changes?
The byte target now applies to parallel writing. It is a best-effort estimate, can reduce write throughput, and is not a hard memory limit. The default remains unset.