[Spark][LSM] Enable format-aware LSM reading and sorted-run ordering

[cshuo]

Task Description

What needs to be done:

Enable Spark to read LSM-layout data tables end to end and make the sorted-run ordering consistent with the physical base-file format.

Why this task is needed:

Spark LSM writes and reads must use the same record-key ordering, i.e., UTF-8 byte order.

Performance Considerations

The record-key comparator is on the hot path of Spark sorting and is invoked approximately O(n log n) times. Using UTF-8 byte ordering for Java String keys is measurably slower than String.compareTo in benchmark testing:

• A comparator based on String.getBytes(UTF_8) performs UTF-8 encoding and allocates byte arrays repeatedly inside the comparison loop.
• String.compareTo compares the String's existing representation directly and benefits from optimized JDK implementations. It does not need to transcode the key.
• The difference is more visible when record keys share a long common prefix: both comparators scan most of the prefix repeatedly, but the UTF-8 comparator also pays the per-character encoding and branching cost before it reaches the differing suffix.

Local pairwise comparator benchmark (Java 11.0.27, macOS ARM64; three independent JVM forks, with the median of five measured trials taken per fork; both implementations allocate 0 B/op):

Scenario	compareUtf8Bytes	String.compareTo	compareTo Speedup
32-char UUID-like ASCII	2.88 ns/op	2.70 ns/op	1.07×
BMP Unicode	2.98 ns/op	2.89 ns/op	1.03×
Mixed BMP and Supplementary Unicode	3.72 ns/op	3.27 ns/op	1.14×
ASCII with 8-char Common Prefix	8.35 ns/op	2.88 ns/op	2.90×
ASCII with 16-char Common Prefix	11.88 ns/op	3.08 ns/op	3.86×
ASCII with 32-char Common Prefix	19.07 ns/op	3.76 ns/op	5.07×

For that reason, Spark data-table sorting should not be globally converted to UTF-8 ordering merely to use one comparator everywhere. UTF-8 ordering is required for HFile because it must match HFile's physical key order. Parquet and ORC should retain String.compareTo ordering to avoid an unnecessary sort-performance regression.

Implementation Scope

• Select HoodieLsmFileGroupReader for Spark LSM data-table snapshot/MOR reads when merge semantics and native log files permit it.
• Keep the metadata table exemption and retain the existing reader fallback for unsupported read modes.
• Add a base-file-format-aware record-key comparator:
  • HFile: unsigned UTF-8 byte order.
  • Parquet/ORC: Java String.compareTo order.
• Apply the comparator consistently to the LSM loser tree, sorted file-group record buffer, create handles, and sorted/LSM merge handles.
• Preserve the base-file-only fast path so physical duplicate keys are returned without merging; once a file slice enters the merge path, records with the same key are merged as one logical record.
• Do not introduce a new RecordKeyOrdering abstraction or a new user-facing configuration.

Acceptance Criteria

• Spark can read base-only and base-plus-native-log LSM file slices.
• Reader and writer ordering agree for each supported base-file format.
• Parquet/ORC sorting does not use the UTF-8 comparator in the hot comparison loop.
• Base-only duplicate keys are preserved, while the LSM merge path merges equal keys.
• Metadata-table behavior remains unchanged.
• Targeted common/Spark reader tests, compile, checkstyle/scalastyle, and git diff --check pass.

Related Issues

Parent feature issue: #19065

RFC: RFC-103

[danny0405]

Thanks for publishing the numbers. The allocation-free comparator makes this a much fairer comparison, and the result clearly shows that String.compareTo is faster—especially for long common prefixes.

I do not think the pairwise comparator benchmark alone is enough to justify a format-dependent persisted LSM ordering, though. My main concern is cross-engine compatibility:

• Flink's LSM bulk path uses BinaryExternalSorter. The input is serialized to BinaryRowData, and the generated string normalized key is based on StringData.toBytes(); the fallback comparison is therefore over the binary UTF-8 representation.
• The Java LSM sorting helpers currently use the UTF-8 comparator as well.
• Parquet defines the canonical STRING sort order as unsigned byte-wise comparison of its UTF-8 representation. Physical row order is not mandated by Parquet, but choosing UTF-16 would create a Hudi/JVM-specific ordering contract.

That means a Parquet LSM run written by Flink can be UTF-8 ordered while the proposed Spark reader/writer expects UTF-16 order. The same physical table format would then have engine-dependent sorted-run semantics.

Before locking this into the LSM format, could we add:

1. The JMH source, JVM flags, and operand-generation details so the microbenchmark is reproducible. Fixed pairwise operands can amplify branch-prediction and JDK-intrinsic effects.
2. A whole-sort benchmark (Arrays.sort and/or Spark repartitionAndSortWithinPartitions) with realistic key distributions, reporting wall time, CPU, allocation, shuffle, and spill.
3. A comparison against a precomputed UTF-8 sort representation—encode/materialize once per record, then compare bytes—rather than deriving UTF-8 order repeatedly inside every comparison.
4. A cross-engine test: Spark write → Flink read/merge and Flink write → Spark read/merge, using U+E000 and U+20000.

The reported worst case is a meaningful comparator-level difference (19.07 ns/op vs 3.76 ns/op), but we still need to establish the end-to-end write impact and weigh it against a permanent engine-specific table-format contract. If the end-to-end regression is material, I would prefer optimizing or caching the sort-key representation while retaining one canonical unsigned UTF-8 order for all LSM runs.