[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 7 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.
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.