[core] Support Parquet row-group copy fast path for append-only compaction - #9660
[core] Support Parquet row-group copy fast path for append-only compaction#9660hbgstc123 wants to merge 8 commits into
Conversation
…ction Add an opt-in fast path for append-only table compaction on Parquet files: when all eligibility conditions hold, compaction concatenates compressed row groups directly and only rewrites the footer, skipping row decode/re-encode entirely. Any ineligible input (schema/codec mismatch, deletion vectors, row tracking, file index, bloom filter, encryption, Parquet writer v2, partial-column writes, etc.) falls back to the traditional rewrite path. New options (all default off/serial): - append.compaction.row-group-copy.enabled - append.compaction.row-group-copy.preserve-page-index: keep ColumnIndex/OffsetIndex so page-level predicate pruning still works on compacted files, at the cost of reading and rewriting page indexes - append.compaction.row-group-copy.footer-read.parallelism: bounded concurrent footer reads during fast-path prepare Parquet-specific compatibility checks live in paimon-format (ParquetRowGroupCopyChecker); value stats of output files are merged from file-level stats when a file is fully copied and recomputed from row-group metadata for partially copied files.
RowGroupCopyCompactionBenchmark compares REWRITE vs row-group copy (including preserve-page-index and footer-read parallelism variants) across column shapes, codecs, row-group sizes and file sizes, tunable via -DrowGroupCopyBenchmark.* properties. ParquetPageIndexBenchmark measures the read-side effect of dropping vs preserving the page index under point and range predicates. Neither runs in CI by default (class names do not match surefire patterns; trigger explicitly with -Dtest).
…ow-group-copy-upstream # Conflicts: # paimon-core/src/main/java/org/apache/paimon/operation/BaseAppendFileStoreWrite.java # paimon-core/src/test/java/org/apache/paimon/append/AppendOnlyTableCompactionTest.java
JingsongLi
left a comment
JingsongLi
left a comment
There was a problem hiding this comment.
Avoiding decode/re-encode during eligible append compaction has clear end-to-end value, but the new statistics merger can silently remove matching query results after compaction. Details are inline.
All 31 existing selected compaction/statistics tests passed on the reviewed head. Additional real-Parquet probes verified the fast path hit, committed the compaction, and compared a filtered table scan with normal rewriting: the fast path lost the matching row. I did not independently benchmark the claimed throughput gains.
| } | ||
| if (current instanceof Comparable && candidate instanceof Comparable) { | ||
| Comparable<Object> currentComparable = (Comparable<Object>) current; | ||
| return currentComparable.compareTo(candidate) >= 0 ? current : candidate; |
There was a problem hiding this comment.
[P1] Merge binary bounds using the existing unsigned comparator
BINARY/VARBINARY values deserialize as byte[], which is not Comparable, so pickMax retains the first contributor's bound; pickMin has the same problem. I reproduced this with two one-row BYTES Parquet files containing 0x01 and 0x02. The fast path copies both rows but records manifest bounds [0x01,0x01]. After committing that compaction, a table scan for payload = 0x02 returns no rows because AppendOnlyFileStoreScan prunes the file; normal rewriting returns the row. The row-count guard passes and cannot catch the wrong metadata.
Use Paimon's unsigned binary ordering (as FullSimpleColStatsCollector/SortUtil.compareBinary already do) for both bounds. Add a compaction-commit/filtered-scan regression whose match is in a later contributor, including reverse ordering for the minimum bound.
There was a problem hiding this comment.
thanks for the review, fixed accordingly.
byte[] is not Comparable, so row-group-copy compaction kept the first contributor's min/max and could prune matching rows after commit.
JingsongLi
left a comment
JingsongLi
left a comment
There was a problem hiding this comment.
Reviewed 72b85d7. Requirement fit: SUPPORTED. Implementation: FINDINGS.
Copying eligible Parquet row groups can avoid decode/re-encode work in append compaction. The previous binary-bounds wrong-result finding is fixed: both contributor orders now use the same unsigned ordering as ordinary statistics and predicates, and both independent real-Parquet probes pass. I found a different, narrower issue in unknown bounds for NOT NULL fields: a valid counts-stats table always falls back after already copying the output. Details are inline; this preserves rows but defeats the optimization and adds I/O.
The exact-head scoped Maven package run passed 33 tests without fast-build. Additional probes verified the prior binary fix and reproduced the new counts-stats failure; changing only the stats serializer to nullable fields made the failing fast-path HIT assertion pass with all 20 rows preserved. No object-store performance benchmark or injected filesystem-failure campaign was run. No current CI rollup was available in the inspected head metadata.
|
|
||
| RowType statsRowType = valueStatsCols == null ? rowType : rowType.project(valueStatsCols); | ||
| int fieldCount = statsRowType.getFieldCount(); | ||
| InternalRowSerializer serializer = new InternalRowSerializer(statsRowType); |
There was a problem hiding this comment.
[P2] Serialize statistics with nullable field types
Please make the statistics serializer's fields nullable, as SimpleStatsConverter already does. For a valid INT NOT NULL table with metadata.stats-mode=counts, merged min/max are null, but this serializer uses the original NOT NULL getter and throws while unboxing the bound. I reproduced this with two real Parquet inputs: all 20 rows survive via normal fallback, but the fast-path HIT stays 0; making only the serializer fields nullable produces HIT=1.
Because copier.copy has already completed before buildResult invokes this serializer, each such compaction first writes and deletes a full copied output and then re-reads/re-encodes the inputs. Add a NOT NULL primitive + counts-stats regression asserting that the fast path actually hits, rather than only checking the fallback result.
2 participants
Purpose
close #9664
Compaction of append-only tables rewrites every data file, even when inputs are mergeable as-is. Since Parquet row groups are self-contained compressed units, files sharing the same schema and codec can be merged by concatenating row groups and rewriting only the footer, skipping row decode/re-encode entirely.
This PR adds an opt-in row-group copy fast path to append-only compaction, controlled by
append.compaction.row-group-copy.enabled(defaultfalse). Each compaction batch is checked for eligibility (Parquet format, current schema, uniform codec, no deletion vectors / row tracking / file index / encryption, etc.); any ineligible file makes the whole batch fall back to the traditional rewrite path, so it is always safe to enable. Parquet-specific checks live in paimon-format (ParquetRowGroupCopyChecker), keeping paimon-core free of Parquet internals.Two companion options:
append.compaction.row-group-copy.preserve-page-index(defaultfalse) keeps ColumnIndex/OffsetIndex on compacted files at the cost of extra reads, andappend.compaction.row-group-copy.footer-read.parallelism(default1, hard cap 8) bounds concurrent footer reads.Benchmarks (
RowGroupCopyCompactionBenchmark, included): 6.4–6.9× on narrow numeric tables, up to 24–32× on wide string tables (zstd, 8MB row groups); production Flink/Spark compaction jobs saw a stable ~59–68% reduction in kernel task time with zero fallbacks.Documentation:
core_configuration.htmlregenerated, plus a new section in the append-table docs.Tests
ParquetFastPathCompactRewriterTest: fast-path hit, all fallback conditions, partial file copy stats merging, row-count verification.SimpleStatsMergerTest: stats merging across files and row groups.