feat(spark): enable format-aware sort ordering and LSM reading for Spark - #19502
Conversation
cshuo
changed the title
94ce9af to
6942f82
Compare
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for working on this! The PR wires up format-aware record-key ordering (UTF-8 for HFile, Java String/UTF-16 for Parquet/ORC) across the write and LSM read paths and enables the LSM file-group reader for eligible Spark snapshot/MOR reads of LSM data tables. I traced the writer/reader ordering consistency, comparator serializability, stream reuse, and reader selection gating, and the change is internally consistent. One non-blocking compatibility question is worth double-checking in the inline comment. Please take a look, and this should be ready for a Hudi committer or PMC member to take it from here. A couple of readability nits below, mostly around duplicated builder calls.
| Comparator<Tuple2<HoodieKey, Option<HoodieRecordLocation>>> comparator = (Comparator<Tuple2<HoodieKey, Option<HoodieRecordLocation>>> & Serializable) (t1, t2) -> { | ||
| HoodieKey key1 = t1._1; | ||
| HoodieKey key2 = t2._1; | ||
| return StringUtils.compareUtf8Bytes(key1.getRecordKey(), key2.getRecordKey()); |
There was a problem hiding this comment.
For Hudi’s new LSM layout, use unsigned lexicographic UTF-8 byte order as the table-level record-key ordering contract.
This is the stronger choice than Java String.compareTo() because:
- LSM runs are persisted as Parquet strings, and Parquet defines
STRINGordering as unsigned byte-wise ordering of UTF-8. [Parquet logical types](https://parquet.apache.org/docs/file-format/types/logicaltypes/) - It is portable across Spark, Flink, Java, and future non-JVM implementations.
- It matches the comparator Hudi already introduced in StringUtils.java.
- The Spark sorted-write path already uses UTF-8 ordering in BaseSparkCommitActionExecutor.java.
requireSortedRecords()now covers both HFile and LSM tables in HoodieTable.java.
The current LSM implementation is inconsistent. These two paths still use Java UTF-16 ordering:
They should use:
Comparator.comparing(
HoodieRecord::getRecordKey,
StringUtils.UTF8_LEXICOGRAPHIC_COMPARATOR)and:
int keyCompare = StringUtils.compareUtf8Bytes(
left.current.getRecordKey(),
right.current.getRecordKey());I would define the invariant as:
Every LSM run is strictly ordered by the unsigned UTF-8 representation of
_hoodie_record_key; all writers, range partitioners, merge readers, compaction paths, and key-range metadata must use that same ordering.
Also audit the Spark bulk-insert partitioners: several still use String.compareTo(), including the bucket-index path. For composite routing, compare (partition path/file group, record key) as a tuple, applying UTF-8 comparison to each string component—don’t sort a concatenated string.
Add an end-to-end test using keys whose orders differ:
U+E000
U+20000
Expected UTF-8/Parquet order:
U+E000 < U+20000
That test should cover writing separate L0 runs, LSM k-way merge, compaction into L1, and reading the compacted result.
There was a problem hiding this comment.
Thanks for the detailed review. Two points:
-
Using UTF-8 consistently would be the ideal ordering contract, especially for cross-engine compatibility. The current format-aware design is a performance trade-off: HFile requires UTF-8 byte ordering, while UTF-8 comparison of Java String record keys is slower than String.compareTo, particularly for long common prefixes. Details and benchmarks are in [Spark][LSM] Enable format-aware LSM reading and sorted-run ordering #19436. If this sorting regression is acceptable, using UTF-8 ordering consistently would be an optimal option.
-
Bulk insert and other LSM write paths are follow-up work. They are tracked by [Spark][LSM] Support standard write operations and compaction #19437, [Spark][LSM] Support bulk insert and Row writer overwrite paths #19438, [Spark][LSM] Support bucket index write paths #19439, and [Spark][LSM] Support clustering and bucket rescale #19440. The relevant partitioners and ordering requirements will be handled in those PRs.
6942f82 to
9988740
Compare
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for the contribution! This PR routes eligible Spark LSM data-table reads through HoodieLsmFileGroupReader, aligns the LSM loser tree and merge handle record-key ordering to unsigned UTF-8 byte order, extracts shouldUseLsmReader into a shared LsmReaderUtils, and propagates the table storage layout on Spark SQL table init. I traced the write and read paths and reader-selection edge cases (skip-merge fallback, metadata table, base-file-only duplicate preservation, incremental filtering, non-splittable MOR); the ordering is consistent end-to-end (UTF-8) and effectively fixes a prior read/write comparator mismatch. No correctness issues found. A few style/readability suggestions in the inline comments. Please take a look, and this should be ready for a Hudi committer or PMC member to take it from here. - a Hudi committer or PMC member can take it from here for a final review.
. A couple of readability nits around duplicated builder blocks in the Spark reader selection paths.
cc @yihua
| .withStart(file.start) | ||
| .withLength(baseFileLength) | ||
| .build() | ||
| } else { |
There was a problem hiding this comment.
🤖 nit: the two builder branches share ~12 identical .withX() calls and only differ in .withShouldUseRecordPosition(...). Could you extract the shared setup into a local helper or a partially-constructed builder, so a future author adding a new withX() option doesn't have to remember to update both branches?
| .withInternalSchemaOpt(HOption.ofNullable(tableSchema.internalSchema.orNull)) | ||
| .build() | ||
| } else { | ||
| HoodieFileGroupReader.builder[InternalRow]() |
There was a problem hiding this comment.
🤖 nit: same pattern as HoodieFileGroupReaderBasedFileFormat — both reader branches duplicate all builder options. It might be worth consolidating the shared lines to make the if/else diff obvious at a glance.
9988740 to
f12ca2c
Compare
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for the contribution! This PR wires Spark snapshot/MOR reads onto the LSM file-group reader for LSM-layout data tables and consolidates record-key ordering on the UTF-8 byte comparator across the write and read paths, while preserving metadata-table behavior and the skip-merge fallback. I traced the write/read ordering contract end to end on the PR branch — because requireSortedRecords() is true for LSM layout, the write path sorts LSM base/log runs with the same UTF-8 comparator the reader's loser tree and sorted buffer now use, so the ordering is internally consistent. Stream reuse (fresh getLogFiles() streams), null-safe mergeType handling, skip-merge routing via props, and preserved MDT behavior all check out. No correctness issues found. A few style/readability suggestions in the inline comments. Please take a look, and this should be ready for a Hudi committer or PMC member to take it from here. One maintainability nit on the duplicated builder chains; everything else looks clean.
cc @yihua
| .withLength(baseFileLength) | ||
| .withShouldUseRecordPosition(shouldUseRecordPosition) | ||
| .build() | ||
| val reader: HoodieRecordReader[InternalRow] = |
There was a problem hiding this comment.
🤖 nit: the two branches of this if/else share ~10 identical .withX() builder calls and diverge only on reader type and withShouldUseRecordPosition. Could you extract the shared builder setup into a small helper (e.g. baseReaderBuilder(readerContext, ...)) so a future option added to one branch isn't silently missed in the other? Same pattern appears in HoodieMergeOnReadRDDV2.scala.
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## master #19502 +/- ##
============================================
- Coverage 76.40% 73.99% -2.42%
+ Complexity 32398 31152 -1246
============================================
Files 2520 2521 +1
Lines 138985 139077 +92
Branches 16695 16750 +55
============================================
- Hits 106189 102906 -3283
- Misses 25166 28419 +3253
- Partials 7630 7752 +122
Flags with carried forward coverage won't be shown. Click here to find out more.
🚀 New features to boost your workflow:
|
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for the continued work on this! This PR wires up LSM file-group-reader selection for Spark snapshot/MOR reads and aligns the read-path record-key ordering (UTF-8 byte order) with the write path. I traced the ordering invariant end to end — requireSortedRecords() returns true for any LSM-layout table, so base and log sorted runs are all written in UTF-8 order, and the loser-tree/record-buffer read paths now compare with compareUtf8Bytes to match. Reader selection, merge-type gating (including skip-merge routing through the RDD path), stream consumption, and the storage-layout propagation all check out. No correctness issues found. A few style/readability suggestions in the inline comments. Please take a look, and this should be ready for a Hudi committer or PMC member to take it from here. A couple of builder-duplication suggestions in the Scala reader-selection paths; otherwise clean.
cc @yihua
6 participants
Describe the issue this Pull Request addresses
Closes #19436.
Spark LSM data-table reads need to use the LSM file-group reader and the same record-key ordering as the physical base-file format. HFile therefore uses unsigned UTF-8 byte ordering to match its physical key order, while Parquet and ORC retain Java
String.compareTo(UTF-16 code-unit) ordering. Applying UTF-8 ordering to every format would add unnecessary comparison overhead, especially for keys with long common prefixes. See #19436 for the detailed analysis and benchmark results.Summary and Changelog
HoodieLsmFileGroupReaderfor eligible Spark snapshot/MOR reads of LSM data tables.String.compareToordering.U+E000andU+20000to distinguish UTF-8 and UTF-16 ordering.Impact
Spark can read eligible LSM-layout data tables end to end. Reader and writer ordering now agree with each base-file format without imposing UTF-8 comparison overhead on Parquet and ORC sorting. No new public API or user-facing configuration is introduced.
Risk Level
Medium. This changes Spark MOR reader selection for LSM-layout data tables and comparator selection on sorted-run paths. The change retains existing fallbacks and is covered by targeted common reader/ordering tests, Spark datasource E2E coverage, Spark client compilation, and Spark 4 packaging verification.
Documentation Update
None. No configuration or public API is added or changed.
Contributor's checklist