feat(flink): add new RocksDBPartitionedIndexBackend - #19765
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 RocksDB-backed PartitionedIndexBackend that stores each partition's record-key→fileId mapping in its own column family, plus a PartitionedIndexBackendFactory to select between the dummy, MDT-backed, and RocksDB backends via index.rli.backend.type. The core RocksDB mechanics (column-family-per-partition, registry-then-visible ordering, close/cleanup) look sound; the main thing worth double-checking is that the RocksDB backend never consults the metadata table, so selecting it on a non-empty table or after a restart could route existing records as inserts — see the inline comments. 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 minor naming inconsistencies worth tidying up.
| } | ||
| String backendType = conf.get(FlinkOptions.INDEX_RLI_BACKEND_TYPE); | ||
| if (ROCKSDB_BACKEND_TYPE.equalsIgnoreCase(backendType)) { | ||
| return new RocksDBPartitionedIndexBackend(conf.get(FlinkOptions.INDEX_RLI_CACHE_ROCKSDB_BASE_PATH)); |
There was a problem hiding this comment.
🤖 With index.rli.backend.type=rocksdb, get() only sees keys written via update() in the current run (RocksDBDAO wipes its dir on startup) and never consults the MDT RLI. On a non-empty table or after a restart, existing keys miss and get routed as inserts in DynamicBucketAssignFunction (L130-136), which can create a second file group for a key that already exists. The index.rli.cache.rocksdb.base.path description says this cache sits "in front of the metadata table" — is the MDT bootstrap/fallback still TODO before this backend is selectable? @danny0405 might want to weigh in.
| * | ||
| * @param partitionPath the partition path to delete | ||
| */ | ||
| public void deletePartition(String partitionPath) { |
There was a problem hiding this comment.
🤖 deletePartition drops the column family before removing the registry entry — the reverse of the register-after-create ordering update() uses to keep get() safe. If a get() ever observed the intermediate state it would see the partition still registered but the CF handle already removed from RocksDBDAO's managedHandlesMap (null handle passed to RocksDB.get). Harmless while the assign operator is single-threaded, but would it be safer to unregister first, then drop the CF, to match the documented visibility invariant?
| this.rocksDBDAO = new RocksDBDAO("hudi-partitioned-index-backend", rocksDbBasePath, new ConcurrentHashMap<>(), true); | ||
| } | ||
|
|
||
| @Override |
There was a problem hiding this comment.
🤖 nit: the parameter rocksDbBasePath uses lowercase Db while the class name spells it RocksDB (uppercase). Could you align to rocksDBBasePath for consistency?
| public class PartitionedIndexBackendFactory { | ||
| private static final String ROCKSDB_BACKEND_TYPE = "rocksdb"; | ||
|
|
||
| /** |
There was a problem hiding this comment.
🤖 nit: the string "rocksdb" is defined here but the valid values for FlinkOptions.INDEX_RLI_BACKEND_TYPE aren't visible at the call site. It might be clearer to keep this constant (or an enum) in FlinkOptions alongside the config key itself, so the two can't drift independently.
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## master #19765 +/- ##
============================================
+ Coverage 77.97% 78.15% +0.18%
- Complexity 33464 33714 +250
============================================
Files 2539 2542 +3
Lines 140974 141465 +491
Branches 17013 17380 +367
============================================
+ Hits 109928 110567 +639
+ Misses 23382 23200 -182
- Partials 7664 7698 +34
Flags with carried forward coverage won't be shown. Click here to find out more.
🚀 New features to boost your workflow:
|
| return new DummyPartitionedIndexBackend(); | ||
| } | ||
| String backendType = conf.get(FlinkOptions.INDEX_RLI_BACKEND_TYPE); | ||
| if (ROCKSDB_BACKEND_TYPE.equalsIgnoreCase(backendType)) { |
There was a problem hiding this comment.
Selecting rocksdb here bypasses the MDT-backed RecordLevelIndexBackend, but this backend starts empty and neither bootstraps from nor falls back to the authoritative MDT RLI. On a non-empty table or after task/job restart, committed keys therefore miss in get() and DynamicBucketAssignFunction routes them as inserts, potentially assigning a second file group to an existing key.
In addition, the first update() registers the column family after loading only one key, which violates RFC-107's partition-completeness invariant. Please keep this backend non-selectable until partition bootstrap/on-demand MDT loading is implemented, or compose it with the MDT backend and register a partition only after its complete RLI state has been loaded.
There was a problem hiding this comment.
Thanks for calling it out. Make the rocksdb option unselectable now.
|
Is |
|
It is a good question. As we described in the RFC, we probably need to decouple the dynamic partitioned record index load and final lookup into two operators. When a record of old partition comes, it will trigger the partition record index load, these record index should be keyed by and shuttle to all of other task managers. In this way, the RocksDB in each of task managers only needs to store the state it required, and one partition need to load once. The RocksDBPartitionedIndexBackend will be a configurable backend owned by BucketAssignFunction and the new function PreLookup. The RocksDB Dao will be initialized in singleton and shared by both BucketAssignFunction and PreLookup. How do you think? |
5 participants
Describe the issue this Pull Request addresses
add new RocksDBPartitionedIndexBackend that can manages record-key-to-fileId mapping for each partition in column family of RocksDB.
closes #19601
Summary and Changelog
Impact
none
Risk Level
none
Documentation Update
none
Contributor's checklist