Repository navigation
Conversation
ce16f93 to
c0d7629
Compare
73a59fa to
006ea5b
Compare
Add opt-in content defined chunking (CDC) to the writer. Data page boundaries are chosen by a rolling gear hash over each column's (definition level, repetition level, value) stream instead of by position, so an insert or delete only changes the pages around it and files deduplicate in content addressable storage. The chunker is a port of Arrow C++ (parquet/chunker_internal) and arrow-rs (column/chunker/cdc.rs): same gear tables, mask derivation, defaults and cut rules. Each column keeps one chunker for the whole file, as arrow-rs does, so chunk boundaries continue across row groups instead of restarting in each. Arrow C++ builds a new chunker for every row group, contrary to its own documentation, so its chunks match ours in the first row group but generally not after it. - CdcOptions: min/max chunk size and normalization level, defaulting to 256 KiB / 1 MiB / 0 as in Arrow. - ParquetProperties and ParquetWriter.Builder: withContentDefinedChunkingEnabled(boolean) and withContentDefinedChunking(CdcOptions). - ParquetOutputFormat: parquet.page.content-defined-chunking.* keys. - ChunkingColumnWriter wraps the column writer only when chunking is enabled. It ends pages at chunk boundaries and, as Arrow C++ does, applies the page size and row count limits counted from the page start, so they cut a chunk in the same places after an edit. - The writer keeps each column's chunker for the whole file, in CdcChunkers, which makes every row group's store. - With chunking on, a column falls back from dictionary encoding only when its dictionary outgrows its size limit, as in Arrow C++, not on its first page, which chunking cuts by content. Pages of nulls alone get an empty dictionary page, as Arrow C++ writes one. - CdcWriteBenchmarks measures the write cost. Off by default, and with it off the writer behaves as before.
03f588f to
d5bc4e7
Compare
c44ab39 to
ecf5dfe
Compare
- Chunker state lives per file: InternalParquetRecordWriter creates the chunkers of the file up front, one per leaf column, only when chunking is enabled, as Arrow C++ and arrow-rs do, and passes them to every row group's store through a new ParquetProperties.newColumnWriteStore overload. With chunking enabled, a store made without them fails fast rather than silently restarting the chunking. CdcOptions are validated when built. - Content defined chunking is configured through one setter, withContentDefinedChunking(CdcOptions), where null disables it, and named as Arrow names it: CdcOptions, ContentDefinedChunker. - ChunkingColumnWriter wraps each column writer once, when the store is built. It rolls every triplet into the column's chunker and, at record starts, ends the page where needNewChunk() says, as Arrow C++ does, or where a page limit is reached, counted from the page start; cached nulls are flushed on the unchunked schedule's maximum. - Binary values are hashed through one path. - The tests fail when chunking is switched off or broken, cover the cases the Arrow C++ and arrow-rs suites do (V2 pages, dictionary encoding, edits across row groups, a chunker per column, the record writer path), and follow the project's test guidelines. The benchmark writes one mixed dataset.
ColumnWriterBase carries its column's ContentDefinedChunker, passed in by the store as the bloom filter writer is, instead of being wrapped by ChunkingColumnWriter. Without content defined chunking it gets a no-op chunker, as it gets no-op statistics, which the JIT inlines away, so the write path has no added branch. ContentDefinedChunker follows Arrow C++: the column writer rolls each triplet and asks needNewPage() at record starts, built on Roll-like primitives and needNewChunk(). ContentDefinedChunker.forColumns takes the writer properties, for the page limits. Written files are unchanged.
The column writer asks needNewPage(value, repetitionLevel, definitionLevel, writer) before writing each triplet and ends the page when it says so, instead of rolling the value itself and then asking. Rolling is the content defined chunker's own business, so another chunker can answer the same question its own way, and the no-op chunker inlines to a single false.
|
I'm not sure who to ping about reviews, but @sunchao you are probably familiar with the CDC logic we have in the C++/Rust implementations. Could you please take a preliminary look and/or ping other maintainers? I essentially tried to port the libarrow implementation :) |
|
@wgtmac would appreciate if you could take a look as well |
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
Reviewed e4199d8a48ea19e6f56889e13ce161a46c8c6b94 with five focused agent reviews, followed by local validation. I found no actionable correctness issues in this revision.
This looks useful for publishing successive versions of Parquet datasets that share most of their data: for example, correcting labels, inserting records, or appending batches. Choosing page boundaries from the content helps unchanged runs of values land in the same pages after an edit. Doing this before encoding and compression gives a deduplicating storage layer such as Xet more opportunities to recognize identical byte chunks. The uploader can then send missing chunks and the metadata needed to reconstruct the new file, reusing pieces already stored. The writer still produces a replacement Parquet file; the potential savings are in upload bandwidth and storage across versions.
This brings the Java writer an approach already implemented in Arrow C++/PyArrow and arrow-rs, both merged. Following that design and testing against Arrow-generated examples is sensible. Keeping the feature experimental and disabled by default, while producing ordinary Parquet files that existing readers can read, is appropriate.
The review covered:
- Algorithm: primitive byte representations, floating-point raw bits, binary slices/direct buffers, definition/repetition levels, minimum-size skipping, pending matches, maximum-size cuts, and mask validation. All 2,048 gear-table entries matched Arrow's constants.
- Writer integration: row-aligned page boundaries, page size/value/row limits, cached nulls, statistics and counters, V1/V2 page/index handling, dictionaries, failure propagation, and continuation of chunker state across row groups.
- API and compatibility: immutable options, properties copying, explicit null-disable behavior, Hadoop configuration, existing constructor paths, the disabled path, and the requirement for direct integrations to retain chunkers across row groups for the whole file.
- Test quality: Arrow golden vectors, edit-locality assertions, actual file round trips, configuration/fail-fast cases, and the distinction between independent expectations and helper tests derived from the implementation.
Local validation on the reviewed head passed with JDK 17:
| Test class | Tests | Failures / errors / skips |
|---|---|---|
| TestContentDefinedChunker | 29 | 0 / 0 / 0 |
| TestColumnWriterChunking | 24 | 0 / 0 / 0 |
| TestContentDefinedChunking | 12 | 0 / 0 / 0 |
| TestParquetProperties | 16 | 0 / 0 / 0 |
That is 81 tests passed, with no skips.
A supplemental probe passed 128 full/filtered readbacks across 16 writer combinations: V1/V2, uncompressed/GZIP, dictionary enabled/disabled, and encrypted/plain files. It exercised nested and sparse data, oversized values/repeated records, and multiple row groups. The probe checked that page indexes actually reduced candidate rows, that encrypted cases encrypted all columns, and that dictionary-enabled cases actually used a dictionary.
A separate standalone Java probe passed 900,000 mask-arithmetic checks against an independent BigInteger calculation, including extreme size envelopes and normalization levels. This was an arithmetic check, separate from the actual writer/readback tests.
The full JDK 17 CI job also executed the four test classes successfully. Its checkout was a synthetic merge containing this head and an older base commit.
Validation limits and practical qualifications:
- The local build disabled the optional
non-aarch64profile because its Brotli runtime dependency was unavailable from the configured artifact mirror. Local compression coverage here was uncompressed/GZIP. - I did not independently reproduce the reported 10–30% write overhead or measure end-to-end storage savings.
- Reuse depends on the workload, data ordering, writer/encoding/compression settings, and downstream deduplication. Dictionary changes can also affect byte reuse. Cross-language boundary equality requires matching the bytes fed to the chunker; equal logical values alone are insufficient.
- I did not test a merge with the current target-base commit
db9e61050651c2cb808abc05297a0033552850bc.
Within that scope, the design and implementation look sound to me.
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
LGTM. Approving reviewed revision e4199d8a48ea19e6f56889e13ce161a46c8c6b94 following the comprehensive review and validation. No actionable correctness issues found. The review documents the use case, existing C++/Rust implementations, and validation limits.
Rationale for this change
Content defined chunking (CDC) ends data pages at boundaries derived from the values rather than from page size, so files that share runs of values share byte-identical pages, which content-addressable storage (e.g. Hugging Face Xet) can deduplicate. Arrow C++ and arrow-rs already write such files; Iceberg needs this to honor its
write.parquet.content-defined-chunking.*table properties.What changes are included in this PR?
Opt-in and EXPERIMENTAL, off by default. Output is ordinary Parquet; readers are unchanged.
ContentDefinedChunker: a port of the Arrow C++ / arrow-rs chunker (same gear tables, mask and cut rules), so page boundaries match Arrow's for the same physical values.withContentDefinedChunking(CdcOptions)onParquetProperties.BuilderandParquetWriter.Builder(nulldisables), andparquet.page.content-defined-chunking.{enabled,min.size,max.size,norm.level}inParquetOutputFormat.ColumnWriterBaseasks its column's chunker before each value whether to end the page. Page size, value count and row count limits still apply, counted from the page start. With CDC off the writer gets a no-op chunker that the JIT inlines away; output is byte-identical to master.InternalParquetRecordWritercreates it and passes it to each row group through a newParquetProperties.newColumnWriteStore(..., chunkers)overload; with CDC on, a store built without chunkers fails fast. Integrations that build column write stores themselves, such as Iceberg, must create the chunkers once per file.ContentDefinedChunkingWriteBenchmarks: CDC adds roughly 10–30% write time depending on the data.Changes to existing code are limited to
ColumnWriterBase(one call per value),ColumnWriteStoreBase/V1/V2(passing chunkers to writers, fail-fast, no store-level size check under CDC) and the configuration plumbing.Not included: dictionary fallback is unchanged. parquet-java still decides on the first page, where Arrow uses the dictionary size alone; that could be a follow-up.
Are these changes tested?
Yes:
Are there any user-facing changes?
New EXPERIMENTAL writer options and configuration keys (documented in
parquet-hadoop/README.md), and new public API:CdcOptions,ContentDefinedChunker.forColumns, thenewColumnWriteStoreoverload and theColumnWriteStoreV1/V2constructors that take chunkers.Closes #3817
Generated with Claude Code