feat(spark): support clustering for LSM tables - #19767
Conversation
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for working on this! This PR enables Spark sort-and-size, single-file, and consistent-hashing bucket rescale clustering on LSM-layout tables while preserving the UTF-8 record-key sorted-run invariant: reads are routed through the LSM file-group reader, LSM-aware Row/RDD partitioners are reused, the loser-tree is extracted into a reusable k-way merge utility, and unsupported ordering strategies are rejected before a clustering instant transitions to inflight. 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 naming and readability suggestions below.
cc @yihua
| override def numPartitions: Int = partitioner.numPartitions | ||
|
|
||
| override def getPartition(key: Any): Int = { | ||
| val bucketRoute = key.asInstanceOf[((String, Int), HoodieUTF8String)]._1 |
There was a problem hiding this comment.
🤖 nit: have you considered a case match instead of asInstanceOf here? Something like case ((route, _), _: HoodieUTF8String) => partitioner.getPartition(route) would make the expected tuple shape self-documenting and give a clear MatchError rather than a ClassCastException deep in the shuffle path if the key type ever changes.
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## master #19767 +/- ##
============================================
+ Coverage 78.29% 78.32% +0.03%
- Complexity 33850 33897 +47
============================================
Files 2541 2543 +2
Lines 141674 141795 +121
Branches 17172 17191 +19
============================================
+ Hits 110917 111063 +146
+ Misses 23048 23024 -24
+ Partials 7709 7708 -1
Flags with carried forward coverage won't be shown. Click here to find out more.
🚀 New features to boost your workflow:
|
dcfc6f5 to
f211eb2
Compare
1fec70e to
63b2590
Compare
github-actions
Bot
added
size:L
ef33b75 to
bda56fe
Compare
bda56fe to
f487a60
Compare
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for working on this! This PR enables Spark sort-and-size / single-file clustering and consistent-hashing bucket split/merge on LSM-layout tables while preserving the record-key (UTF-8) sort invariant — reading inputs through the LSM file-group reader, adding a reusable loser-tree k-way merge for the consistent-hashing merge path, and falling back from binary/stream copy. I traced the loser-tree state machine, the merge vs. split routing and ordering, resource cleanup on the read path, and the reader return-type change, and they look correct for the covered cases. 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 naming nits where the method and variable names no longer match the return type after the refactor to HoodieRecordReader.
cc @yihua
| @@ -142,10 +144,24 @@ protected FileSlice clusteringOperationToFileSlice(String basePath, ClusteringOp | |||
| return fileSlice; | |||
There was a problem hiding this comment.
🤖 nit: now that the return type was widened to HoodieRecordReader, the name getFileGroupReader is a bit misleading — could you rename it to something like getRecordReader to match what it actually returns?
|
|
||
| // instantiate FG reader | ||
| HoodieFileGroupReader<InternalRow> fileGroupReader = getFileGroupReader(metaClient, fileSlice, tableSchemaWithMetaFields, internalSchemaOption, | ||
| HoodieRecordReader<InternalRow> fileGroupReader = getFileGroupReader(metaClient, fileSlice, tableSchemaWithMetaFields, internalSchemaOption, |
There was a problem hiding this comment.
🤖 nit: the local variable fileGroupReader still carries the old name — would recordReader (or just reader) be more accurate now that its type is HoodieRecordReader?
4 participants
Describe the issue this Pull Request addresses
Closes #19440.
Spark clustering and bucket rescale previously could not safely operate on LSM-layout tables. These operations must preserve the LSM sorted-run invariant: every output file has to remain ordered by record key using UTF-8 byte ordering.
Summary and Changelog
Impact
Spark users can run supported clustering and bucket-rescale operations on LSM-layout tables while retaining the ordering invariant required by LSM readers. No configuration defaults or storage format are changed. Unsupported ordering strategies fail early, while binary-copy strategies transparently use the regular ordered rewrite path.
Risk Level
medium. The change affects Spark clustering read/write paths and bucket routing. It is covered by targeted common and Spark-client unit tests plus COW/MOR functional tests that verify sorted output, snapshot correctness, split/merge routing, file IDs, replacement metadata, and subsequent upserts.
Documentation Update
None. This enables existing clustering and bucket-rescale interfaces for LSM tables and adds no new configuration.
Contributor's checklist