feat(parquet): stream-encode definition/repetition levels incrementally#9447
Merged
alamb merged 7 commits intoapache:mainfrom Apr 1, 2026
Merged
feat(parquet): stream-encode definition/repetition levels incrementally#9447alamb merged 7 commits intoapache:mainfrom
alamb merged 7 commits intoapache:mainfrom
Conversation
Previously, the column writer accumulated raw definition and repetition levels in `Vec<i16>` sinks (`def_levels_sink` / `rep_levels_sink`) and only RLE-encoded them in bulk at page-flush time. Replace the two sinks with streaming `LevelEncoder` fields. Levels are now encoded incrementally as each `write_batch` call arrives, so only the compact encoded bytes are held in memory at all times. At page flush, the encoder is consumed and its bytes are written directly into the page buffer; a fresh encoder is swapped in for the next page. Signed-off-by: Hippolyte Barraud <hippolyte.barraud@datadoghq.com>
Signed-off-by: Hippolyte Barraud <hippolyte.barraud@datadoghq.com>
brunal
approved these changes
Feb 20, 2026
Contributor
|
run benchmark arrow_writer |
|
Benchmark for this request failed. Last 20 lines of output: Click to expand |
etseidl
approved these changes
Mar 19, 2026
Contributor
etseidl
left a comment
etseidl
left a comment
There was a problem hiding this comment.
Looks sound to me. Thanks!
Comment on lines
-1158
to
-1159
|
|
||
| // Reset state. |
Contributor
There was a problem hiding this comment.
Suggested change
| // Reset state. | |
| // Reset state. |
parquet/src/encodings/levels.rs
Outdated
| /// Computes max buffer size for level encoder/decoder based on encoding, max | ||
| /// repetition/definition level and number of total buffered values (includes null | ||
| /// values). | ||
| #[allow(dead_code)] |
Contributor
There was a problem hiding this comment.
should the dead code be deprecated?
Contributor
There was a problem hiding this comment.
ahh, I didn't notice that the encodings module is experimental...thought this was public. Yes, let's just remove the dead code then :)
Contributor
|
I plan to benchmark on my WS since the bot seems tired |
Contributor
|
run benchmark arrow_writer |
|
Benchmark for this request failed. Last 20 lines of output: Click to expand |
Contributor
|
Ahh, I get it. The benchbot can't handle the remote branch also being named "main". Not much difference either way on my laptop DetailsOn my WS the times are all over the place :( The huge regressions can show up on either branch (could be too much else going on during benchmarking) Details |
alamb
approved these changes
Mar 20, 2026
Contributor
alamb
left a comment
alamb
left a comment
There was a problem hiding this comment.
Thank you @HippoBaro I agree with @etseidl . I think this one is ready to go. I think we can merge it and address follow ons as another PR.
Let us know what you prefer
parquet/src/column/writer/mod.rs
Outdated
| max_rep_level, | ||
| )[..], | ||
| let encoder = mem::replace( | ||
| &mut self.rep_levels_encoder, |
Contributor
There was a problem hiding this comment.
This path recreates the encoders each time (and thus probably allocates, etc)
It seems like the old code path does the same thing (in encode_levels_v1, ..)
@HippoBaro would you be open to exploring adding a "clear" method to the encoder now that you have it encapsulated to save the allocation?
parquet/src/encodings/levels.rs
Outdated
| /// Computes max buffer size for level encoder/decoder based on encoding, max | ||
| /// repetition/definition level and number of total buffered values (includes null | ||
| /// values). | ||
| #[allow(dead_code)] |
Previously, flushing a data page would `mem::replace` each level encoder with a freshly allocated one, consuming the old encoder to get its buffer. This allocated new internal `Vec`s on every page boundary. We now preserve the internal state of the encoder and reuse memory across pages. Signed-off-by: Hippolyte Barraud <hippolyte.barraud@datadoghq.com>
Contributor
Author
|
@alamb @etseidl Thank you both for the review! See 0e28fba, which now preserves internal state across pages. This latest change also adds Your comment suggests we should address those in a follow-up. I’m happy to open another PR after this one. I can also remove them here if that’s more appropriate. Let me know. |
HippoBaro
force-pushed
the
main
branch
3 times, most recently
from
757f11c to
0e28fba
Compare
Contributor
Author
|
Apologies for the force-push there, using my fork's |
Contributor
|
Thanks @HippoBaro, this looks even better now. I'm fine with simply removing the dead code in this PR. This does have me wondering if we should remove the BIT_PACKED encoder. We need a decoder for backwards compatibility, but we shouldn't ever be using it for level encoding. |
Contributor
|
Contributor
|
I made a fake PR to run benchmarks on: Assuming they all look good I will merge this PR |
Contributor
|
🤔 the benchmarks found an issue: #9636 (comment) It doesn't seem to be introduced by this PR though. Looking more carefully |
This was referenced Mar 31, 2026
Contributor
The issue is I made a PR to disable this benchmark temporarly: |
The `v1`, `v2`, and `max_buffer_size` functions required knowing the number of values upfront and pre-allocated buffers. All callers have been migrated to the streaming variants (`v1_streaming`, `v2_streaming`), so remove the dead code and switch the last remaining caller in test utils. Signed-off-by: Hippolyte Barraud <hippolyte.barraud@datadoghq.com>
Contributor
Author
alamb
added a commit
that referenced
this pull request
Apr 1, 2026
# Which issue does this PR close? - Part of #9637 # Rationale for this change I can't benchmark the arrow-writer changes in #9447 due to hitting a panic: - #9637 # What changes are included in this PR? Temporarily disable the cdc benchmarks until the underlying bug is fixed # Are these changes tested? <!-- We typically require tests for all PRs in order to: 1. Prevent the code from being accidentally broken by subsequent changes 2. Serve as another way to document the expected behavior of the code If tests are not included in your PR, please explain why (for example, are they covered by existing tests)? --> # Are there any user-facing changes? <!-- If there are user-facing changes then we may require documentation to be updated before approving the PR. If there are any breaking changes to public APIs, please call them out. -->
Contributor
|
I am looking into this one trying to get CI to pass / clean (and benchmarks run) so we can merge it in. So close |
Contributor
|
The benchmarks on #9636 show no noticeable difference so I think this one is good to go Thank you @HippoBaro for bearing with us and thank you @etseidl for the review |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
5 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Which issue does this PR close?
Rationale for this change
When writing a Parquet column with very sparse data,
GenericColumnWriteraccumulates unbounded memory for definition and repetition levels. The rawi16values are appended intoVec<i16>sinks on everywrite_batchcall and only RLE-encoded in bulk when a data page is flushed. For a column that is almost entirely nulls, the actual RLE-encoded output can be tiny, yet the intermediate buffer grows linearly with the number of rows.What changes are included in this PR?
Replace the two raw-level
Vec<i16>sinks (def_levels_sink/rep_levels_sink) with streamingLevelEncoderfields (def_levels_encoder/rep_levels_encoder). Behavior is the same, but we keep running RLE-encoded state rather than the full list of rows in memory. Existing logic is reused.Are these changes tested?
Yes, all tests passing.
Benchmarks show no regression.
list_primitivebenches improved by 3-5%:Are there any user-facing changes?
None. Some internal symbols are now unused. I added some
#[allow(dead_code)]statements since these were experimental-visible and might be externally relied on.