feat(spark): support standard writes and compaction for LSM tables - #19576
feat(spark): support standard writes and compaction for LSM tables#19576cshuo wants to merge 1 commit into
Conversation
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## master #19576 +/- ##
============================================
+ Coverage 77.09% 77.50% +0.41%
- Complexity 32490 32817 +327
============================================
Files 2522 2527 +5
Lines 139112 139196 +84
Branches 16714 16723 +9
============================================
+ Hits 107243 107889 +646
+ Misses 24291 23740 -551
+ Partials 7578 7567 -11
Flags with carried forward coverage won't be shown. Click here to find out more.
🚀 New features to boost your workflow:
|
8eb64dc to
0a965dc
Compare
0a965dc to
82b59e8
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 LSM-tree table support on Spark to standard writes and compaction, adds table-aware LSM bulk-insert partitioners (HoodieRecord and Row paths) with UTF-8 record-key ordering, and defaults LSM bulk insert to PARTITION_SORT. The partitioner ordering and the RDD/Row consistency look solid on close reading. A couple of cross-engine semantics questions are worth double-checking in the inline comments — particularly whether INSERT should be allowed on LSM at all. 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 few small naming and readability nits below.
| HoodieWriteMetadata result = new HoodieWriteMetadata(); | ||
|
|
||
| boolean isLsmTable = table.getMetaClient().getTableConfig().isLSMTreeStorageLayout(); | ||
| checkArgument(!isLsmTable || userDefinedBulkInsertPartitioner.isEmpty(), |
There was a problem hiding this comment.
🤖 This guards the partitioner choice for LSM, but not duplicate record keys. Flink's HoodieTableFactory.checkStorageLayout hard-rejects INSERT on LSM "because they allow duplicate record keys", whereas this PR enables INSERT/bulk_insert on Spark LSM with no equivalent guard. The updated TestMORDataSource shows dup keys written via INSERT survive only until the first merge, then silently collapse to one row. Is allowing INSERT here intentional, and if so how do we avoid that silent data loss? @danny0405
| return new RDDSimpleBucketBulkInsertPartitioner(table); | ||
| } | ||
| } | ||
| if (table.getMetaClient().getTableConfig().isLSMTreeStorageLayout()) { |
There was a problem hiding this comment.
🤖 The BUCKET-index branch above returns before this LSM check, so an LSM table configured with hoodie.index.type=BUCKET would get the bucket partitioner and skip the LSM UTF-8 ordering entirely. Is that combination prevented at table creation, or could it silently produce base files that aren't record-key sorted and break the LSM reader? Same ordering applies in the WithRows factory.
| } | ||
| return super.repartitionRecords(rows, outputSparkPartitions) | ||
| .sortWithinPartitions(functions.col(HoodieRecord.PARTITION_PATH_METADATA_FIELD), functions.col(HoodieRecord.RECORD_KEY_METADATA_FIELD)); | ||
| } |
There was a problem hiding this comment.
🤖 nit: this line is quite long — could you break the two functions.col(...) arguments onto separate lines to keep it within the usual column limit?
| String expectedPartition) throws IOException { | ||
| HoodieInstant instant = findCompletedInstant(metaClient, instantTime); | ||
| assertEquals(HoodieTimeline.REPLACE_COMMIT_ACTION, instant.getAction()); | ||
| HoodieReplaceCommitMetadata commitMetadata = metaClient.getActiveTimeline().readReplaceCommitMetadata(instant); |
There was a problem hiding this comment.
🤖 nit: the inner class name LsmTableTestContext uses mixed-case "Lsm" while everywhere else in this PR — and in the outer class name — the acronym is all-caps "LSM". Could you rename it to LSMTableTestContext for consistency?
| HoodieWriteConfig config = createWriteConfig(sortMode, true); | ||
| assertEquals(expectedPartitionerClass, | ||
| BulkInsertInternalPartitionerFactory.get(lsmTable, config).getClass()); | ||
| assertEquals(expectedPartitionerClass, |
There was a problem hiding this comment.
🤖 nit: getRowPartitioner and getHoodieRecordPartitioner look unused — assertRowPartitionerSelection and assertHoodieRecordPartitionerSelection call the factories directly. Could you remove them to avoid dead code?
4 participants
Describe the issue this Pull Request addresses
Closes #19437.
Closes #19438.
Spark LSM tables require every base file and native log block to be ordered by partition path and record key using UTF-8 byte ordering. Standard write operations and compaction needed end-to-end coverage for that invariant, while bulk insert could still select partitioners that either used Java string ordering or did not guarantee ordering at all. The HoodieRecord and Dataset Row paths also needed consistent, table-aware partitioner selection so an LSM table could not bypass the required layout.
Summary and Changelog
INSERT,INSERT_PREPPED,UPSERT,UPSERT_PREPPED,DELETE,DELETE_PREPPED,INSERT_OVERWRITE,INSERT_OVERWRITE_TABLE, andDELETE_PARTITIONthrough the Spark RDD client for both COW and MOR LSM tables, including commit and replace-commit metadata validation.GLOBAL_SORT,PARTITION_SORT, andPARTITION_PATH_REPARTITION_AND_SORT. Global and repartition-and-sort modes use Spark sorting with a serializable UTF-8 comparator; partition sort preserves the existing coalesce plus in-memory per-partition sorting model while using the LSM ordering.PARTITION_SORTwhen no sort mode is configured. ExplicitNONEandPARTITION_PATH_REPARTITIONmodes, custom partitioners whose ordering cannot be verified, and the direct Dataset Row path without meta fields fail with clear errors; custom partitioners are rejected before the instant transitions to inflight.Impact
LSM Spark writes now consistently preserve the physical ordering required by the table layout across standard writes, overwrite operations, bulk insert, compaction, and log compaction.
HoodieWriteConfigaddsisLSMTreeStorageLayout(), and an LSM table whose bulk-insert sort mode is unset now defaults toPARTITION_SORT; an explicitly configured mode is still honored and rejected if it cannot guarantee ordering.The sorting cost depends on the selected mode: global sort and partition-path repartition-and-sort introduce their expected shuffle, while partition sort coalesces and materializes each Spark partition for local sorting, with memory usage proportional to the largest resulting partition. Non-LSM partitioner behavior is unchanged.
Risk Level
Medium. The change touches Spark write partitioner selection and ordering for both HoodieRecord and Dataset Row paths. The risk is mitigated by unit coverage for factory selection, supported and rejected sort modes, meta-field validation, and UTF-8 ordering, plus end-to-end COW/MOR coverage for snapshots and physical LSM file layout. Targeted local runs completed for
TestLSMBulkInsertPartitioner(13 tests),TestLSMDataSource, andTestInsertWithLSMLayout(3 tests);git diff --checkalso passes.Documentation Update
None. LSM storage layout support is still under development, this change introduces no new configuration key, and the safe default is derived from the existing bulk-insert sort-mode configuration.
Contributor's checklist