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% 74.14% -4.16%
+ Complexity 33871 32157 -1714
============================================
Files 2541 2542 +1
Lines 141708 141810 +102
Branches 17177 17191 +14
============================================
- Hits 110961 105152 -5809
- Misses 23038 29432 +6394
+ Partials 7709 7226 -483
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.
| * entirely, which is the expected fallback for non-temporal (non date-partitioned) tables. | ||
| */ | ||
| @Slf4j | ||
| public class PartitionedRLIBootstrapOperator |
There was a problem hiding this comment.
PartitionedRLIBootstrapOperator sounds like a general bootstrap operator, while this implementation only preloads RLI data for a configured time window. Could we rename it to make the scope and behavior clearer, like TimeBoundedRLIBootstrapOperator?
| * Returns whether the table uses partitioned record level index served by the local RocksDB-based | ||
| * partitioned index cache, i.e. {@link FlinkOptions#INDEX_RLI_BACKEND_TYPE} is configured as {@code rocksdb}. | ||
| */ | ||
| public static boolean isPartitionedRLIWithRocksDBBackend(Configuration conf) { |
There was a problem hiding this comment.
Selecting the RocksDB backend does not necessarily mean this bootstrap is applicable, since it only supports time-partitioned tables with a configured preload window.
Could we change INDEX_RLI_CACHE_ROCKSDB_BOOTSTRAP_DAYS to default to -1, and enable the preload only when the user configures a positive value? This predicate could then be renamed to something like isTimeBoundedRLIBootstrapEnabled and require the partitioned RLI type and BOOTSTRAP_DAYS > 0.
| } | ||
|
|
||
| private void emitIndexRecord(String partitionPath, String recordKey, HoodieRecordGlobalLocation location) { | ||
| output.collect(new StreamRecord<>( |
There was a problem hiding this comment.
Just a reminder that these index records are not currently handled by the downstream DynamicBucketAssignFunction, which treats them as normal data records. If the downstream handling is planned for a follow-up PR, please keep this preload disabled by default until the complete processing path is available.
| return partitionsInWindow; | ||
| } | ||
|
|
||
| private LocalDate parsePartitionDate(String partitionPath, DateTimeFormatter formatter, boolean hiveStylePartitioning) { |
There was a problem hiding this comment.
Can you check if there is any existing utility we can reuse to parse dates from partition paths?
| * Uses round-robin assignment: file group i is assigned to task (i % parallelism). | ||
| */ | ||
| @VisibleForTesting | ||
| boolean shouldLoadBucket(int fileGroupIdx, int parallelism, int taskID) { |
There was a problem hiding this comment.
This may cause a hotspot. Since fileGroupIdx restarts from zero for each partition, for e.g., if each partition has 4 RLI shards, all shards will always be loaded by subtasks 0–3, leaving the remaining subtasks idle.
The distribution should also account for the partition path; you can refer to the shuffle strategy used by the bucket index partitioner via BucketIndexUtil#getPartitionIndexFunc.
5 participants
Describe the issue this Pull Request addresses
add partitioned rli boostrap operator to support
Summary and Changelog
Impact
none
Risk Level
none
Documentation Update
none.
Contributor's checklist