perf(read): feed sorted runs into the PK merge LoserTree

[JunRuiLee]

Purpose

Primary-key merge-on-read currently feeds one input stream per data file into the LoserTree. On fragmented tables, peak merge fan-in therefore grows with file count even when many files have disjoint primary-key ranges.

This change concatenates files with strictly non-overlapping key ranges into sorted runs, so peak LoserTree fan-in follows key-range overlap depth while preserving merge semantics.

This is also what paimon-java already does: MergeFileSplitRead unconditionally runs IntervalPartition and wraps each SortedRun in a ConcatRecordReader before handing it to the merge reader, so its fan-in has always tracked overlap depth rather than file count.

Brief change log

• Decode manifest min/max keys with a composite-key comparator that matches physical row ordering, including unsigned binary-key ordering.
• Partition files by overlapping key range and greedily pack each section into lazily concatenated sorted runs. Files inside a run are opened one at a time — the next file's schema lookup and Parquet reader construction happen only after the previous file is drained — so a run holds one row group at a time no matter how many files it spans.
• Preserve global merge behavior when a read spans multiple splits; safely fall back to one stream per file for missing, malformed, or inverted key ranges.
• Apply the existing merge fan-in guard to planned merge inputs rather than raw file count.
• Add regression coverage for integer, binary, and multi-column keys, overlap boundaries, invalid metadata, read equivalence, cross-split deduplication, fan-in limits, and Parquet budget behavior.

Performance

Measured on a fragmented PK table with production shape: 6,647,251 rows, one ARRAY<FLOAT> embedding column of dim 2048, 16 buckets, write-only=true so it was never compacted — roughly 44 level-0 files per bucket whose key ranges overlap only 2 deep. All columns projected, no filter, 16 reader threads.

Peak LoserTree fan-in per split, from planning alone:

	fan-in
one stream per file	~44 (= file count)
sorted runs	2 (= key-range overlap depth)

Peak read memory and throughput at read.batch-size=1024, with paimon-java on the same table for reference — read through the paimon-core API, no Spark or Flink, same projection and parallelism:

	peak memory	throughput
one stream per file	133.61 GiB	59,563 rows/s
sorted runs	6.66 GiB	53,132 rows/s
paimon-java, -Xmx10g (smallest heap that completes)	10.85 GiB	29,026 rows/s
paimon-java, -Xmx16g	16.24 GiB	33,066 rows/s

Sorted runs cut peak memory by 95.0% against the per-file fan-out, at 10.8% lower throughput. Against Java's smallest completing configuration they use 38.6% less memory at 83.0% higher throughput.

Java's RSS grows to fill whatever -Xmx allows — -Xmx24g reaches 23.72 GiB on this table — so its smallest completing heap is the fair comparison point; -Xmx6g OOMs. Java is measured as RSS and Rust as jemalloc allocated, which tracked resident closely here, so neither side is credited with headroom it did not use.

At read.batch-size=8192, where the Rust reader reaches its best throughput on this table, the change wins on both axes. The per-file fan-out keeps 16 × 44 = 704 Parquet readers open at once, and the page-fault and allocator contention that costs shows up as lost CPU:

	peak memory	throughput
one stream per file	336 GiB	~30,000 rows/s
sorted runs	19.21 GiB	47,250 rows/s
	−94.3%	+57.5%

The mechanism is that memory scales with the number of concurrently open Parquet streams rather than with per-stream cost: each open stream keeps its current row group's projected column chunks resident until that row group is fully consumed. Per-stream cost is comparable between the two implementations — Java's off-heap direct memory stayed at 0.01 GiB throughout, so the gap is not hidden allocation on either side.

These numbers come from a downstream build of the same grouping mechanism rather than from this exact diff, so treat the absolute throughput as indicative; the fan-in reduction and the memory-to-stream-count relationship are structural.

Tests

• cargo fmt --all -- --check
• cargo test -p paimon --lib
• cargo test -p paimon table::merge_tree_split_generator::tests --lib
• cargo test -p paimon table::kv_file_reader::tests --lib
• cargo clippy -p paimon --lib --tests -- -D warnings
• git diff --check

API and Format

• No new table option: sorted-run grouping is unconditional, matching paimon-java, which has no switch for it either.
• No storage-format or manifest-format change.

Documentation

• No documentation change.

[QuakeWang]

LGTM

[JunRuiLee]

LGTM

Thanks again for the LGTM! I made a few follow-up fixes based on additional review feedback, mainly around unsigned Binary ordering, sorted-run semantics, unnecessary split cloning, and regression tests. All tests and clippy pass. Could you please take another quick look? Thanks!

[JingsongLi]

+1