hudi-agent commented on code in PR #19744:
URL: https://github.com/apache/hudi/pull/19744#discussion_r3859652048
##########
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:
🤖 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?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
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);
+ }
+
+ /**
+ * Returns whether a partition has been registered as fully loaded.
+ *
+ * @param partitionPath Partition path
+ */
+ public boolean isPartitionRegistered(String partitionPath) {
+ ValidationUtils.checkArgument(!closed);
+ return Boolean.TRUE.equals(get(DEFAULT_COLUMN_FAMILY_NAME, partitionPath));
+ }
+
+ /**
+ * Drops the column family backing a partition's state and clears its
registry entry, if any.
+ *
+ * @param partitionPath Partition path
+ */
+ public void dropPartitionColumnFamily(String partitionPath) {
+ ValidationUtils.checkArgument(!closed);
+ dropColumnFamily(getColumnFamilyNameForPartition(partitionPath));
+ delete(DEFAULT_COLUMN_FAMILY_NAME, partitionPath);
+ }
+
+ /**
+ * Lists all partitions currently registered as fully loaded.
+ */
+ public List<String> listRegisteredPartitions() {
+ ValidationUtils.checkArgument(!closed);
+ return this.<Boolean>prefixSearch(DEFAULT_COLUMN_FAMILY_NAME,
"").map(Pair::getKey).collect(Collectors.toList());
Review 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?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
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);
+ }
+
+ /**
+ * Returns whether a partition has been registered as fully loaded.
+ *
+ * @param partitionPath Partition path
+ */
+ public boolean isPartitionRegistered(String partitionPath) {
Review 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.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
