feat(spark): support bucket index for LSM tables - #19727
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 bucket indexes (simple and consistent-hashing) for LSM tables by sorting records within each bucket by full record key using UTF-8 byte ordering, across both the RDD and Dataset Row bulk-insert paths, while preserving existing bucket routing and file-id assignment. I traced the ordering consistency across all four partitioner paths and the base-file writer (all use UTF-8 binary ordering), confirmed bucket grouping is preserved in the new shuffle-key scheme, and verified custom-sort-column rejection is reached for every RDD/Row × simple/consistent combination. 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 naming/design nit on the null-sentinel call pattern.
cc @yihua
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## master #19727 +/- ##
============================================
+ Coverage 77.88% 77.98% +0.09%
- Complexity 33264 33474 +210
============================================
Files 2533 2539 +6
Lines 140342 141010 +668
Branches 16912 17162 +250
============================================
+ Hits 109310 109961 +651
+ Misses 23401 23393 -8
- Partials 7631 7656 +25
Flags with carried forward coverage won't be shown. Click here to find out more.
🚀 New features to boost your workflow:
|
d17fb84 to
bd62bcb
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 extends Spark bucket-index bulk-insert to LSM tables by adding full-record-key UTF-8 ordering within each bucket across the RDD, simple-bucket Row, and rescale paths while preserving existing routing and file-id assignment. I traced the ordering logic and confirmed the three write paths use mutually consistent UTF-8 byte ordering (compareUtf8Bytes / UTF8String binaryCompare / Catalyst binary string sort), that the consistent-hashing LSM path still sorts via requireSortedRecords(), and that the InternalRow key is copied to avoid buffer aliasing. No new issues flagged from this automated pass beyond what other reviewers already raised (per-record comparator branch, rescale-path validation bypass, double validation, null sentinel). Please take a look at any inline comments, and this should be ready for a Hudi committer or PMC member to take it from here. A couple of small naming and API-clarity nits below.
cc @yihua
| } | ||
|
|
||
| Comparator<HoodieKey> comparator = (Comparator<HoodieKey> & Serializable) (t1, t2) -> t1.getRecordKey().compareTo(t2.getRecordKey()); | ||
| boolean lsmTable = table.getMetaClient().getTableConfig().isLSMTreeStorageLayout(); |
There was a problem hiding this comment.
🤖 nit: since this is a boolean, could you prefix it with is — isLsmTable — so it reads as a predicate consistently with the rest of the codebase?
| protected BulkInsertPartitioner<Dataset<Row>> getPartitioner(boolean populateMetaFields, boolean isTablePartitioned) { | ||
| if (populateMetaFields) { | ||
| if (writeConfig.getIndexType() == HoodieIndex.IndexType.BUCKET) { | ||
| BucketSortBulkInsertPartitioner.validateCustomSortColumns(table, null); |
There was a problem hiding this comment.
🤖 nit: passing null explicitly here is a bit surprising — the method has a sortString parameter, but at this call site there's no sort string at all. Would an overload validateCustomSortColumns(HoodieTable table) (delegating to the existing one with null) make the intent clearer?
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for working on this! This PR adds simple and consistent-hashing bucket-index support for Spark LSM tables by sorting records within each bucket by UTF-8 record key while preserving existing bucket routing and file-id assignment. No new issues flagged from this automated pass — a Hudi committer or PMC member can take it from here for a final review. One minor simplification suggestion below; otherwise the code is clean and readable.
cc @yihua
| } | ||
|
|
||
| Comparator<HoodieKey> comparator = (Comparator<HoodieKey> & Serializable) (t1, t2) -> t1.getRecordKey().compareTo(t2.getRecordKey()); | ||
| Comparator<HoodieKey> comparator = table.getMetaClient().getTableConfig().isLSMTreeStorageLayout() |
There was a problem hiding this comment.
🤖 nit: the two ternary branches only differ in the comparison expression — could you extract a single Comparator<HoodieKey> that picks between StringUtils.compareUtf8Bytes and String.compareTo on the record keys to avoid duplicating the cast/lambda boilerplate?
Restore the pre-LSM public constructor overloads and add multi-bucket routing coverage for row and RDD writers. Addresses review comments 3861784555 and 3861784758.
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 simple and consistent-hashing bucket indexes for Spark LSM tables by preserving existing bucket routing/file-id assignment while enforcing UTF-8 record-key ordering within each output bucket across the RDD, Dataset Row, and bucket-rescale paths. I traced the three LSM sort paths and confirmed they produce consistent binary-UTF-8 ordering and preserve file-group routing. No new critical correctness issues flagged from this automated pass — a Hudi committer or PMC member can take it from here for a final review.
cc @yihua
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for working on this! This PR adds simple and consistent-hashing bucket-index support for Spark LSM tables by sorting records within each bucket by the full record key using binary UTF-8 ordering, across both the RDD and Dataset Row bulk-insert paths (including the rescale path), while preserving existing bucket routing and file-id assignment. I traced the three ordering paths (RDD compareUtf8Bytes, simple-bucket Row tuple ordering via HoodieUTF8String, and consistent-hashing Row sortWithinPartitions) and they agree on binary UTF-8 semantics; routing and file-id stability look preserved. No new correctness, concurrency, or resource issues flagged from this automated pass — a Hudi committer or PMC member can take it from here for a final review. A minor simplification suggestion below, otherwise clean.
cc @yihua
| } | ||
|
|
||
| Comparator<HoodieKey> comparator = (Comparator<HoodieKey> & Serializable) (t1, t2) -> t1.getRecordKey().compareTo(t2.getRecordKey()); | ||
| Comparator<HoodieKey> comparator = table.getMetaClient().getTableConfig().isLSMTreeStorageLayout() |
There was a problem hiding this comment.
🤖 nit: the two ternary branches differ only in the string-compare call — could you pull the boolean out and build a single comparator (e.g. select a BiFunction<String,String,Integer> and wrap it once) to avoid duplicating the lambda?
5 participants
Describe the issue this Pull Request addresses
Closes #19439.
Spark bucket-index partitioners own file-group routing and file-id assignment, so LSM tables cannot replace them with the generic LSM bulk-insert partitioners. The existing bucket paths therefore need to preserve their routing behavior while guaranteeing UTF-8 record-key ordering inside every output bucket.
Summary and Changelog
Impact
This enables Spark LSM tables to use simple and consistent-hashing bucket indexes. It introduces no new public API or configuration. LSM Dataset Row simple-bucket writes now include the record key in the shuffle sort key, adding the key copy and comparisons required to maintain the LSM physical ordering invariant; other table layouts retain their existing behavior.
Risk Level
Medium. The change affects Spark bucket-index write partitioning and ordering. The risk is mitigated by 14 passing partitioner tests and 7 passing targeted functional cases covering routing, UTF-8 ordering, schema preservation, custom-sort rejection, file-id stability, and snapshot correctness after upsert. Compilation, the relevant Checkstyle and Scalastyle checks, and
git diff --checkalso completed without violations.Documentation Update
None.
Contributor's checklist