Re: [PR] feat(flink): Extend RocksDBDAO with generic column-family lifecycle operations [hudi]

["hudi-agent (via GitHub)" <gi...@apache.org>]

hudi-agent commented on code in PR #19744:
URL: https://github.com/apache/hudi/pull/19744#discussion_r3862930666


##########
hudi-common/src/main/java/org/apache/hudi/common/util/collection/RocksDBDAO.java:
##########
@@ -469,6 +477,73 @@ public void addColumnFamily(String columnFamilyName) {
     });
   }
 
+  /**
+   * Derives the column family name used to store a given partition's state.
+   *
+   * @param partitionPath Partition path
+   */
+  public String getColumnFamilyNameForPartition(String partitionPath) {
+    return PARTITION_COLUMN_FAMILY_PREFIX + partitionPath;
+  }
+
+  /**
+   * Create-if-absent the column family backing a partition's state. Does not register the
+   * partition as loaded; call {@link #registerPartitionAsLoaded(String)} once all of the
+   * partition's data has been written to the returned column family.
+   *
+   * @param partitionPath Partition path
+   * @return Column family name for the partition
+   */
+  public String createColumnFamilyIfAbsent(String partitionPath) {
+    String columnFamilyName = getColumnFamilyNameForPartition(partitionPath);
+    addColumnFamily(columnFamilyName);
+    return columnFamilyName;
+  }
+
+  /**
+   * Marks a partition as fully loaded by registering it in the default column family. Should only
+   * be called after all of the partition's data has been written to its own column family, so a
+   * partial/interrupted load is never observed as complete.
+   *
+   * @param partitionPath Partition path
+   */
+  public void registerPartitionAsLoaded(String partitionPath) {
+    ValidationUtils.checkArgument(!closed);
+    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);

Review 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.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org