feat(flink): add partitioned rli boostrap operator - #19805
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 adds a PartitionedRLIBootstrapOperator that eagerly preloads record-level-index data for date partitions within a configurable recency window, and wires it into the bootstrap pipeline behind the RocksDB partitioned-RLI backend config. The logic is well-tested and the window/parse handling looks solid; one parallelism concern around per-partition file-slice distribution is worth double-checking in the inline comment. 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. One minor naming nit below.
| closeMetadataTable(); | ||
| } | ||
|
|
||
| private void preLoadPartition(String partitionPath, List<FileSlice> fileSlices, int taskID, int parallelism) { |
There was a problem hiding this comment.
🤖 The round-robin (i % parallelism == taskID) is applied to the file-slice index within each partition here, whereas RLIBootstrapOperator applies it once over the global file-slice list. For a date-partitioned table where each partition typically has only 1-2 RLI file groups, every single-file-slice partition maps to index 0 and lands entirely on task 0, leaving the other bootstrap tasks idle. Have you considered a global counter across partitions (or hashing on fileId) so the load spreads across all index_bootstrap tasks?
| * determine whether it falls inside the window. Partitions outside the window, and partitions whose | ||
| * path cannot be parsed as a date, are skipped here and are expected to be loaded on demand later. | ||
| * | ||
| * <p>Setting {@link FlinkOptions#INDEX_RLI_CACHE_ROCKSDB_BOOTSTRAP_DAYS} to {@code 0} disables preloading |
There was a problem hiding this comment.
🤖 nit: could you rename loadedCnt to loadedCount? Cnt is an uncommon abbreviation that adds a small mental speed-bump, especially since the field name surfaces directly in log messages.
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## master #19805 +/- ##
==========================================
Coverage 78.30% 78.31%
- Complexity 33871 33925 +54
==========================================
Files 2541 2545 +4
Lines 141708 141869 +161
Branches 17177 17269 +92
==========================================
+ Hits 110961 111098 +137
- Misses 23038 23065 +27
+ Partials 7709 7706 -3
Flags with carried forward coverage won't be shown. Click here to find out more.
🚀 New features to boost your workflow:
|
cshuo
left a comment
cshuo
left a comment
There was a problem hiding this comment.
Thks for the contribution, left some comments.
|
@cshuo |
683285e to
13a81de
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 adds a TimeBoundedRLIBootstrapOperator that preloads the partitioned record-level index for a configurable recent-day window, wires it into the Flink bootstrap pipeline behind the rocksdb RLI backend, and adds a parsePartitionDate helper. The operator's window math, per-task file-group assignment, and partition-path handling all check out against the existing RLI paths. One thing to double-check in the inline comments: the new -1 default for index.rli.cache.rocksdb.bootstrap.days appears to make two of the newly added tests fail, since they assert the feature is enabled without setting a positive day window. 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 doc/readability suggestions below, otherwise the code is clean and well-tested.
| assertFalse(OptionsResolver.isTimeBoundedRLIBootstrapEnabled(conf)); | ||
|
|
||
| conf.set(FlinkOptions.INDEX_RLI_BACKEND_TYPE, "rocksdb"); | ||
| assertTrue(OptionsResolver.isTimeBoundedRLIBootstrapEnabled(conf)); |
There was a problem hiding this comment.
🤖 With index.rli.cache.rocksdb.bootstrap.days now defaulting to -1, isTimeBoundedRLIBootstrapEnabled requires days > 0, but this test never sets it — so conf.get(...) returns -1 and this assertTrue would actually fail. Could you conf.set(FlinkOptions.INDEX_RLI_CACHE_ROCKSDB_BOOTSTRAP_DAYS, <positive>) before asserting enabled?
| Pipelines.bootstrap(conf, TestConfigurations.ROW_TYPE, input, false, false); | ||
|
|
||
| assertEquals("index_bootstrap", streaming.getTransformation().getName()); | ||
| assertInstanceOf(TimeBoundedRLIBootstrapOperator.class, bootstrapOperator(streaming)); |
There was a problem hiding this comment.
🤖 Same root cause as the OptionsResolver test: defaultConf() doesn't set index.rli.cache.rocksdb.bootstrap.days, so it defaults to -1, isTimeBoundedRLIBootstrapEnabled returns false, and streamBootstrap selects BootstrapOperator instead — so this assertInstanceOf(TimeBoundedRLIBootstrapOperator.class, ...) would fail. Setting a positive bootstrap-days on conf should fix it.
| private int parallelism; | ||
| private int taskID; | ||
| /** | ||
| * Functions for calculating the task partition to dispatch. |
There was a problem hiding this comment.
🤖 nit: the javadoc says setting this to 0 disables preloading, but the default is now -1 and the check is <= 0. Could you reword to say any non-positive value disables it to match the code?
| preLoadPartition(partitionPath, partitionedFileGroups.get(partitionPath), taskID, parallelism); | ||
| } | ||
| long costMs = System.currentTimeMillis() - startTime; | ||
| log.info("Finish preloading partitioned RLI records, total records: {}, cost: {} ms, taskId = {}", loadedCnt, costMs, taskID); |
There was a problem hiding this comment.
🤖 nit: preLoadPartition takes taskID and parallelism as parameters, but both are already instance fields set in initializeState. Could you drop the params and read the fields directly to avoid the confusion of two sources of truth?
13a81de to
5928cf0
Compare
cshuo
left a comment
cshuo
left a comment
There was a problem hiding this comment.
lgtm, only one minor comment.
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 a TimeBoundedRLIBootstrapOperator that pre-warms the partitioned record-level index for partitions within a configurable day window, wires it into the Flink write pipeline behind the rocksdb RLI backend config, and flips the bootstrap-days default to -1 (off). I traced the record-emission coverage/disjointness (each RLI file slice is owned by exactly one subtask via getPartitionIndexFunc, so no duplicate/dropped index records), the waitForBootstrapReady barrier paths, the config default change, and the date-window filtering — no new correctness issues surfaced in this round. The most significant open concern (the emitted index records flowing into DynamicBucketAssignFunction, which doesn't check isIndexRecord() and would treat them as null-rowData data records) has already been raised by @cshuo and @danny0405 and remains the key thing to resolve. 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 naming and simplification suggestions below.
| private transient HoodieBackedTableMetadata tableMetadata; | ||
| private transient long loadedCnt; | ||
| private int parallelism; | ||
| private int taskID; |
There was a problem hiding this comment.
🤖 nit: could you spell this out as loadedCount? loadedCnt is a small abbreviation but it reads slightly awkwardly next to the surrounding unabbreviated field names.
|
|
||
| private void preLoadPartition(String partitionPath, List<FileSlice> fileSlices, int taskID, int parallelism) { | ||
| List<FileSlice> filteredFileSlices = new ArrayList<>(); | ||
| for (int i = 0; i < fileSlices.size(); i++) { |
There was a problem hiding this comment.
🤖 nit: parallelism is passed in here but never used inside the method body — might be worth dropping it from the signature and letting shouldLoadBucket and the log statement read off the captured taskID only.
| OptionsResolver.getConflictResolutionStrategy(conf)); | ||
| } | ||
|
|
||
| @Test |
There was a problem hiding this comment.
🤖 nit: testPartitionedRLIWithRocksDBBackend only exercises cases where the method returns false — it might be worth adding one assertTrue with INDEX_TYPE=RECORD_LEVEL_INDEX, INDEX_RLI_BACKEND_TYPE=rocksdb, and INDEX_RLI_CACHE_ROCKSDB_BOOTSTRAP_DAYS > 0 to confirm the happy path.
|
@cshuo @danny0405 |
6 participants
Describe the issue this Pull Request addresses
add partitioned rli boostrap operator to support
closes #19603
Summary and Changelog
Impact
none
Risk Level
none
Documentation Update
none.
Contributor's checklist