[core] Speed up row-id manifest sorting

[leaves12138]

What changed

• add manifest-specific primitive Avro reader and writer APIs under paimon-format/.../avro/primitive
• expose only the ordering, filtering, and identity fields required by row-id manifest sorting
• reuse a plain mutable record view instead of materializing InternalRow or ManifestEntry objects
• discover naturally sorted runs and merge them with a fixed selection tree
• collect minor-compaction DELETE identities during run discovery instead of scanning DELETE manifests separately
• copy eligible encoded records and compressed Avro blocks directly into output manifests
• keep the generic format, stats collector, and rolling-writer frameworks unchanged
• retain the spillable external sorter as a bounded fallback for fragmented inputs
• process up to 25,000 fragmented entries in memory before falling back to external sort
• preserve exact DELETE identifier matching and manifest statistics
• add manifest-sort.run-merge-optimize.enabled as a fallback switch

Why

Full row-id manifest compaction currently materializes every ManifestEntry, feeds all entries through the external sorter, and encodes every surviving entry again. Large data-evolution tables therefore spend substantial CPU time, cumulative heap allocation, and temporary-disk I/O in this path even when their manifests already consist of a small number of sorted runs.

The new manifest-only Avro path reads just the primitive fields needed for classification and ordering, interns partition data, and merges already-sorted runs directly. When an encoded record or compressed Avro block can be reused safely, it is copied without materializing a full entry or encoding it again. Minor compaction gathers DELETE identities during the same discovery scan, avoiding an extra pass over DELETE manifests. Fragmented input still falls back to the existing bounded external sorter.

Performance

Full compaction benchmark: snapshot 7537

The benchmark uses the production snapshot 7537 metadata fixture on Linux x86_64:

• 87 input manifests, 666.2 MiB of manifest data
• 26,100,963 input entries
• six-way manifest read parallelism
• 64 MiB external-sort buffer and 128 file handles
• no -Xmx limit
• identical code and configuration except that the existing path forces external sort while the new path enables natural-run merge

Metric	Existing external sort	Final run merge	Improvement
Sort wall time	180.41 s	11.12 s	16.2x faster
Total allocation during sort	150,607.3 MiB (147.08 GiB)	2,536.2 MiB (2.48 GiB)	59.4x lower
Benchmark-command peak RSS	2.25 GiB	1.19 GiB	47.1% lower
Initial sorter spill	1,080,034,577 B (1.006 GiB), 273 files	0 B	eliminated
Intermediate fan-in rewrite	1,514,367,922 B (1.410 GiB)	0 B	eliminated
Total temporary sorter bytes written	2,594,402,499 B (2.416 GiB)	0 B	eliminated

Total allocation during sort is cumulative allocation, calculated by summing JVM per-thread allocation deltas across the measured sort interval. It is GC-reclaimable allocation volume, not live heap or RSS. Peak RSS is the maximum resident set reported by /usr/bin/time -v for the Maven/Surefire benchmark command.

Spill values count compressed temporary files created by the sorter only; they exclude source-manifest reads and final output-manifest writes. External sort first wrote 1,080,034,577 bytes across 273 runs. Because this exceeded the 128-file-handle limit, two fan-in operations wrote another 506,977,657 and 1,007,390,265 bytes. The final run-merge path created no sorter spill files.

Production compaction snapshots: 8795–8826

The follow-up benchmark replays all 17 manifest-compaction snapshots in the production metadata window from snapshot 8795 through 8826 on Linux x86_64, without an -Xmx limit. Each snapshot contains about 26.6 million entries and 0.65–0.67 GiB of manifest data.

Version	Total sort time for 17 snapshots	Improvement
Existing external sort	1,652.7 s	baseline
Run merge before minor optimization	872.6 s	1.89x faster
Final run merge with minor optimization and 25k fragmented-entry limit	182.8 s	9.04x faster than external sort; 4.77x faster than the earlier run merge

Representative snapshots that previously received little or no benefit now avoid external sort:

Snapshot	External sort	Earlier run merge	Final run merge	Final vs external
8795	68.1 s	63.1 s	4.17 s	16.3x
8800	73.8 s	58.6 s	4.67 s	15.8x
8802	59.8 s	54.9 s	3.06 s	19.5x
8808	61.7 s	55.6 s	3.20 s	19.3x
8810	56.8 s	55.0 s	3.70 s	15.4x
8816	156.1 s	68.0 s	14.49 s	10.8x
8818	152.3 s	66.8 s	12.45 s	12.2x
8826	152.4 s	66.6 s	12.84 s	11.9x

For the 8795–8810 sequence, the final run completed all 11 snapshots in 48.7 s with a 0.98 GiB process peak RSS. In an isolated 8795 comparison, peak RSS decreased from 7.89 GiB for external sort to 0.96 GiB for run merge, an 8.2x reduction.

The bounded fallback remains active. Snapshot 8824 contains multiple fragmented manifests with 38,722 fragmented entries in one rewrite section, exceeding the 25,000-entry in-memory limit, so it intentionally uses external sort.

Result equivalence

Both snapshot 7537 paths produced exactly 26,095,685 output entries. Their logical encoded-record SHA-256 was identical:

afe6fcdaff7e8b2e7d047c199164c19457ced5b3cb2adaadd32c48cf11a1cdf5

Validation

• 3 targeted AvroFileFormatTest cases covering primitive projection, nullable/nested fields, reader reuse, and large Zstd blocks
• targeted ManifestFileMetaTest cases covering row-id order, secondary keys, exact DELETE identifiers, raw identity fields, many partitions, fragmented fallback, read amplification, block stats, configured partition fields, missing row-id stats, minor compaction, and single-pass minor DELETE discovery
• 5 ByteArrayKeyTest cases plus 12 binary-manifest identifier/projection cases
• mvn -pl paimon-format,paimon-core -am -DskipTests compile with checkstyle, Spotless, and enforcer enabled
• 5 focused fragmented/minor run-merge tests after raising the bounded in-memory limit to 25,000 entries
• snapshot 7537 output-count and logical-digest comparison on Linux x86_64
• production metadata benchmark across all 17 compaction snapshots from 8795 through 8826 on Linux x86_64

[JingsongLi]

[P1] Do not copy encoded records across Avro writer schemas

rawBlockCopySupported() prevents whole-block copying when the input writer schema differs from the current manifest schema, but this path still returns the encoded bytes from the source record. writeSelected then passes those bytes to ManifestAvroWriter.writeEncoded, which appends them under the current schema header without transcoding or validation.

An upgraded data-evolution table can have legacy manifests that already contain _FIRST_ROW_ID (so run merge is enabled) but lack a later field such as _WRITE_COLS. Copying such records can produce a malformed manifest; a focused legacy-schema reproduction fails with EOFException when the output is read.

Please fall back to the existing external sorter whenever a source writer schema is not exactly compatible, or fully materialize and re-encode those records with the current schema. Please also add full/minor compaction coverage for a legacy RowID manifest containing fields 0..18 but not _WRITE_COLS. This blocks enabling the optimization in production by default.