feat(flink): Extend RocksDBDAO with generic column-family lifecycle operations - #19744
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 extends RocksDBDAO with column-family-per-partition lifecycle methods (create-if-absent, register-as-loaded, drop, and a default-CF registry), backed by unit tests and not yet wired into any operator. The encoding is consistent across put/get/delete/prefixSearch and no existing caller touches the default column family, so the change is self-contained. A couple of edge cases around concurrency and the default-CF exclusivity assumption are worth double-checking in 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. One naming inconsistency worth aligning, otherwise the code is clean and well-documented.
| String columnFamilyName = getColumnFamilyNameForPartition(partitionPath); | ||
| ValidationUtils.checkArgument(managedDescriptorMap.containsKey(columnFamilyName), | ||
| "Column family for partition " + partitionPath + " must be created before it can be registered as loaded"); | ||
| put(DEFAULT_COLUMN_FAMILY_NAME, partitionPath, Boolean.TRUE); |
There was a problem hiding this comment.
🤖 The containsKey check followed by the registry put isn't atomic — if another thread runs dropPartitionColumnFamily for the same partition in between, we could end up registering it as loaded (registry=TRUE) after its column family handle has already been removed from managedHandlesMap. A later isPartitionRegistered would then return true while a read against the partition CF hits a null handle. Is this class intended to be single-writer per partition, or should register/drop be guarded so this interleaving can't leave the registry inconsistent?
There was a problem hiding this comment.
🤖 This looks like a real gap worth the author confirming. The sibling dropColumnFamily uses computeIfPresent to atomically remove the handle from managedHandlesMap, but there's no shared lock between it and the register path, and register spans two structures (the CF map plus the registry write into the default CF), so a single atomic map op wouldn't close it anyway. Existing users like RocksDbBasedFileSystemView rely on external single-threaded access — if the same single-writer-per-partition guarantee holds here it's fine, otherwise a lock around register/drop would be needed to avoid registering a partition whose handle was just dropped.
| */ | ||
| public List<String> listRegisteredPartitions() { | ||
| ValidationUtils.checkArgument(!closed); | ||
| return this.<Boolean>prefixSearch(DEFAULT_COLUMN_FAMILY_NAME, "").map(Pair::getKey).collect(Collectors.toList()); |
There was a problem hiding this comment.
🤖 listRegisteredPartitions treats every key in the default column family as a registered partition path. That works today because no caller writes to the default CF, but it's an implicit exclusivity contract on a shared class. Could you document that the default CF is reserved for the partition registry (or use a dedicated key prefix) so a future caller storing unrelated keys in the default CF doesn't silently corrupt this listing?
| * | ||
| * @param partitionPath Partition path | ||
| */ | ||
| public boolean isPartitionRegistered(String partitionPath) { |
There was a problem hiding this comment.
🤖 nit: registerPartitionAsLoaded / isPartitionRegistered use two different words for the same concept — could you align them? Either isPartitionLoaded or renaming the write side to registerPartition would make the pair easier to scan.
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## master #19744 +/- ##
============================================
+ Coverage 77.96% 77.98% +0.01%
- Complexity 33457 33467 +10
============================================
Files 2539 2539
Lines 140939 140941 +2
Branches 17012 17012
============================================
+ Hits 109886 109907 +21
+ Misses 23397 23374 -23
- Partials 7656 7660 +4
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.
RocksDBDAO should remain a generic column-family abstraction. Partition naming, registration, and loaded-state management are RFC-107 cache policies and should live in the partition-aware Flink backend.
We should limit this PR to basic column-family operations, such as create, drop, existence checks, and listing, and move partition-specific logic to the Flink backend.
…perations Add columnFamilyExists and listColumnFamilies alongside the existing addColumnFamily/dropColumnFamily so callers can check existence and list managed column families; partition-aware naming and completeness tracking belong in the Flink backend instead. Unit tests only; not wired into any operator yet.
f0c11ff to
03876cf
Compare
HuangZhenQiu
changed the title
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 two read accessors to RocksDBDAO — columnFamilyExists and listColumnFamilies — over the managed column-family map, along with unit tests. No issues flagged from this automated pass — a Hudi committer or PMC member can take it from here for a final review.
cc @yihua
| String columnFamilyName = getColumnFamilyNameForPartition(partitionPath); | ||
| ValidationUtils.checkArgument(managedDescriptorMap.containsKey(columnFamilyName), | ||
| "Column family for partition " + partitionPath + " must be created before it can be registered as loaded"); | ||
| put(DEFAULT_COLUMN_FAMILY_NAME, partitionPath, Boolean.TRUE); |
There was a problem hiding this comment.
🤖 This looks like a real gap worth the author confirming. The sibling dropColumnFamily uses computeIfPresent to atomically remove the handle from managedHandlesMap, but there's no shared lock between it and the register path, and register spans two structures (the CF map plus the registry write into the default CF), so a single atomic map op wouldn't close it anyway. Existing users like RocksDbBasedFileSystemView rely on external single-threaded access — if the same single-writer-per-partition guarantee holds here it's fine, otherwise a lock around register/drop would be needed to avoid registering a partition whose handle was just dropped.
Thanks for the suggestions. It is a better design to push the partition management in partition-aware Flink backend. Revised the PR with minimal api changes. |
github-actions
Bot
added
size:S
6 participants
Extend RocksDBDAO with generic column-family lifecycle operations
Describe the issue this Pull Request addresses
Add columnFamilyExists and listColumnFamilies alongside the existing addColumnFamily/dropColumnFamily so callers can check existence and list managed column families; partition-aware naming and completeness tracking
belong in the Flink backend instead. Unit tests only; not wired into any operator yet.
closes #19600
Summary and Changelog
Impact
none
Risk Level
none
Documentation Update
none
Contributor's checklist