[core] Support Parquet row-group copy fast path for append-only compaction

[hbgstc123]

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 (default false). 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 (default false) keeps ColumnIndex/OffsetIndex on compacted files at the cost of extra reads, and append.compaction.row-group-copy.footer-read.parallelism (default 1, 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.html regenerated, 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.
• Benchmarks double as correctness checks (content equality verification mode).

[JingsongLi]

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.

[JingsongLi]

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

[hbgstc123]

thanks for the review, fixed accordingly.

[JingsongLi]

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.

[JingsongLi]

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

[hbgstc123]

Thanks for catching this. Fixed by making the stats row type's fields nullable in SimpleStatsMerger.merge, mirroring SimpleStatsConverter, which covers both the serializer and the field getters.

Added regressions:

• SimpleStatsMergerTest#testMergeWithNotNullFieldType: NOT NULL INT + null min/max bounds (counts stats mode). Verified it throws the exact NPE you described without the fix.
• ParquetFastPathCompactRewriterTest#testFastPathHitWithNotNullColumnAndCountsStatsMode: NOT NULL primitive + metadata.stats-mode=counts end-to-end, asserting HIT_COUNT == 1 (no silent fallback) plus row content and null min/max bounds on the compacted file.

[JingsongLi]

This PR is too large.

[hbgstc123]

This PR is too large.

Thanks for the review. This PR is now limited to the minimal, default-off fast path.

Still in this PR:

• append.compaction.row-group-copy.enabled (default false)
• serial Parquet row-group copy, with whole-batch fallback to the existing rewrite
• the earlier statistics fixes: unsigned binary min/max, and nullable stats fields for NOT NULL columns under metadata.stats-mode=counts

Not in this PR; they will be a follow-up after this one is merged:

• append.compaction.row-group-copy.preserve-page-index
• append.compaction.row-group-copy.footer-read.parallelism
• the row-group-copy and page-index benchmarks

The branch has been rebased onto current master.

[JingsongLi]

Reviewed head 2e599bd502 end to end: the new opt-in append-table Parquet row-group copy path, eligibility/fallback checks, output footer/statistics, pruning and integration behavior. This is a useful feature and should remain open. The recent scope reduction and shaded-Parquet test changes are improvements.

[P2] Close the fast-path metric group with the writer. BaseAppendFileStoreWrite.withMetricRegistry now always constructs CompactionFastPathMetrics (14 counters), including when append.compaction.row-group-copy.enabled=false, but BaseAppendFileStoreWrite.close() only closes blobFetchMetrics. CompactionFastPathMetrics has no close method. In Flink, createTableMetricGroup registers a child group in the task's metric system; each writer lifecycle leaves this group registered after write.close(). I reproduced this with a tracking MetricRegistry: create a normal append writer, call withMetricRegistry, close the writer, and assert that the compactionFastPath group closed. The focused probe fails (1 test, 1 failure); the adjacent BlobFetchMetrics follows the expected close pattern. Please make this metric holder closeable, close it from the writer, and cover that lifecycle (preferably also avoiding registration when the feature is disabled).

Verification on JDK 8: ParquetFastPathCompactRewriterTest 14/14, SimpleStatsMergerTest 6/6, AppendOnlyTableCompactionTest 11/11, and adjacent Parquet format/stats tests 33/33 passed. These exercise actual copied row groups, row reads, predicate pruning and stats. The temporary lifecycle probe was removed after reproduction; the review checkout is clean.

Release gate: this branch currently reports no GitHub checks. Please run the affected core/format CI on the revised head before merge, including the metric lifecycle fix.

[JingsongLi]

Rechecked current head 7932c9bafd after the two formatting-only commits (7c9dceb63d, 7932c9bafd). They do not change the metric lifecycle finding in my review: CompactionFastPathMetrics is still registered in withMetricRegistry and not closed from BaseAppendFileStoreWrite.close(). The earlier JDK 8 data and format test results remain applicable; affected CI on this head is pending.

[hbgstc123]

Rechecked current head 7932c9bafd after the two formatting-only commits (7c9dceb63d, 7932c9bafd). They do not change the metric lifecycle finding in my review: CompactionFastPathMetrics is still registered in withMetricRegistry and not closed from BaseAppendFileStoreWrite.close(). The earlier JDK 8 data and format test results remain applicable; affected CI on this head is pending.

Thanks for the detailed review. Both points addressed in ef566eecf.

Metric lifecycle — fixed as suggested:

• CompactionFastPathMetrics is now closeable: close() delegates to metricGroup.close(), mirroring the BlobFetchMetrics pattern, and BaseAppendFileStoreWrite.close() closes it right after blobFetchMetrics.
• withMetricRegistry now only constructs and registers the compactionFastPath group when append.compaction.row-group-copy.enabled=true. With the default config, no counters are registered at all.

Added CompactionFastPathMetricsTest (4 tests):

• testMetricRegistration: group name, table variable, and all 14 counters (1 hit + 13 miss reasons).
• testCloseClosesMetricGroup: close() closes the underlying group, using a tracking MetricGroup.
• testWriterClosesMetricGroupWhenEnabled: the probe scenario from your review — normal append writer, withMetricRegistry, close(), asserting the compactionFastPath group is registered and then closed.
• testWriterSkipsMetricGroupWhenDisabled: feature disabled → no group is registered.

Verification on JDK 8: CompactionFastPathMetricsTest 4/4, ParquetFastPathCompactRewriterTest 14/14, SimpleStatsMergerTest 6/6, AppendOnlyTableCompactionTest 11/11 passed; spotless:check and checkstyle:check are clean.

[JingsongLi]

Reviewed head ef566eecf9 end to end. Requirement fit: SUPPORTED. Implementation: CLEAN in the inspected paths.

The opt-in Parquet row-group copy path has a concrete compaction benefit. I checked eligibility and fallback, copied footer and value-stat handling, output cleanup, and the writer metric lifecycle. The previous metric-group leak is fixed: registration is conditional on the option, close propagates to the group, and both behaviors have focused regressions.

Verification on JDK 8 / Maven 3.9.9: mvn -pl paimon-core -am -Pfast-build -DfailIfNoTests=false -DwildcardSuites=none -Dtest=ParquetFastPathCompactRewriterTest,SimpleStatsMergerTest,AppendOnlyTableCompactionTest,CompactionFastPathMetricsTest test passed: 35 tests, 0 failures. This includes actual copied Parquet row groups, table scan and predicate cases, stats merging, fallback, and metric lifecycle. git diff --check passed.

The current CI run has one failure in S3FileIOTest: Testcontainers could not pull the unrelated quay.io/minio/minio:RELEASE.2022-02-07T08-17-33Z image; other jobs remain in progress. Please ensure the affected CI jobs complete and the S3 infrastructure failure is resolved or rerun before merge. I did not independently rerun a production object-store benchmark or fault-injection campaign.