You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by si...@apache.org on 2026/02/05 09:50:26 UTC
(hudi) branch branch-0.x updated: [HUDI-9770] During hive/glue sync, ensure drop partition events are generated when partition is present in the metastore (#13794) (#17857)
This is an automated email from the ASF dual-hosted git repository.
sivabalan pushed a commit to branch branch-0.x
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/branch-0.x by this push:
new c507cba8bd3d [HUDI-9770] During hive/glue sync, ensure drop partition events are generated when partition is present in the metastore (#13794) (#17857)
c507cba8bd3d is described below
commit c507cba8bd3db3fca7866cb8c769b2a59e1bbad8
Author: Lin Liu <14...@users.noreply.github.com>
AuthorDate: Thu Feb 5 01:50:19 2026 -0800
[HUDI-9770] During hive/glue sync, ensure drop partition events are generated when partition is present in the metastore (#13794) (#17857)
Co-authored-by: Roushan Kumar <ro...@gmail.com>
---
.../java/org/apache/hudi/hive/HiveSyncTool.java | 1 +
.../org/apache/hudi/hive/TestHiveSyncTool.java | 123 +++++++++++++++++++++
.../apache/hudi/sync/common/HoodieSyncClient.java | 13 ++-
3 files changed, 132 insertions(+), 5 deletions(-)
diff --git a/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/HiveSyncTool.java b/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/HiveSyncTool.java
index ddc6da22d91b..1fd4e16baa43 100644
--- a/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/HiveSyncTool.java
+++ b/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/HiveSyncTool.java
@@ -287,6 +287,7 @@ public class HiveSyncTool extends HoodieSyncTool implements AutoCloseable {
// Sync the partitions if needed
// find dropped partitions, if any, in the latest commit
Set<String> droppedPartitions = syncClient.getDroppedPartitionsSince(lastCommitTimeSynced, lastCommitCompletionTimeSynced);
+ LOG.info("Partitions dropped since last sync: {}", droppedPartitions.size());
partitionsChanged = syncPartitions(tableName, writtenPartitionsSince, droppedPartitions);
}
diff --git a/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/TestHiveSyncTool.java b/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/TestHiveSyncTool.java
index af4baaae4a3b..e696bdd9e93a 100644
--- a/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/TestHiveSyncTool.java
+++ b/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/TestHiveSyncTool.java
@@ -1281,6 +1281,129 @@ public class TestHiveSyncTool {
"The last commit that was synced should be updated in the TBLPROPERTIES");
}
+ @ParameterizedTest
+ @MethodSource("syncModeAndEnablePushDown")
+ void testGetPartitionEvents_droppedStoragePartitionNotPresentInMetastore(
+ String syncMode, String enablePushDown) throws Exception {
+ hiveSyncProps.setProperty(HIVE_SYNC_MODE.key(), syncMode);
+ hiveSyncProps.setProperty(HIVE_SYNC_FILTER_PUSHDOWN_ENABLED.key(), enablePushDown);
+
+ // Create a table with 1 partition
+ String instantTime1 = "100";
+ HiveTestUtil.createCOWTable(instantTime1, 1, true);
+
+ reInitHiveSyncClient();
+ // Sync the table to metastore
+ reSyncHiveTable();
+
+ List<Partition> partitionsInMetastore = hiveClient.getAllPartitions(HiveTestUtil.TABLE_NAME);
+ assertEquals(1, partitionsInMetastore.size(), "Should have 1 partition in metastore");
+
+ // Add a partition to storage but don't sync it to metastore
+ String instantTime2 = "101";
+ String newPartition = "2010/02/01";
+ HiveTestUtil.addCOWPartition(newPartition, true, true, instantTime2);
+
+ // Verify the partition is not in metastore yet
+ partitionsInMetastore = hiveClient.getAllPartitions(HiveTestUtil.TABLE_NAME);
+ assertEquals(1, partitionsInMetastore.size(), "Should have 1 partition in metastore");
+
+ // Delete the partition that was never synced to metastore
+ String instantTime3 = "102";
+ HiveTestUtil.createReplaceCommit(instantTime3, newPartition, WriteOperationType.DELETE_PARTITION, true, true);
+
+ // Add another partition to storage but don't sync to metastore
+ String instantTime4 = "103";
+ String addPartition = "2010/04/01";
+ HiveTestUtil.addCOWPartition(addPartition, true, true, instantTime4);
+
+ reInitHiveSyncClient();
+
+ Set<String> droppedPartitionsOnStorage = hiveClient.getDroppedPartitionsSince(Option.of(instantTime1), Option.of(instantTime1));
+ List<String> writtenPartitionsOnStorage = hiveClient.getWrittenPartitionsSince(Option.of(instantTime1), Option.of(instantTime1));
+ List<PartitionEvent> partitionEvents = hiveClient.getPartitionEvents(
+ partitionsInMetastore, writtenPartitionsOnStorage, droppedPartitionsOnStorage);
+
+ // Verify no DROP event is generated for partition that was never in metastore
+ long dropEvents = partitionEvents.stream()
+ .filter(e -> e.eventType == PartitionEventType.DROP)
+ .count();
+ assertEquals(0, dropEvents,
+ "No DROP partition event should be generated for partition that was never in metastore");
+
+ // Verify ADD event is generated for the new partition that was added to storage
+ List<PartitionEvent> addEvents = partitionEvents.stream()
+ .filter(e -> e.eventType == PartitionEventType.ADD)
+ .collect(Collectors.toList());
+ assertEquals(1, addEvents.size(),
+ "ADD partition event should be generated for new partition added to storage");
+ assertEquals(addPartition, addEvents.get(0).storagePartition);
+ }
+
+ @ParameterizedTest
+ @MethodSource("syncModeAndEnablePushDown")
+ void testGetPartitionEvents_droppedStoragePartitionPresentInMetastore(
+ String syncMode, String enablePushDown) throws Exception {
+ hiveSyncProps.setProperty(HIVE_SYNC_MODE.key(), syncMode);
+ hiveSyncProps.setProperty(HIVE_SYNC_FILTER_PUSHDOWN_ENABLED.key(), enablePushDown);
+
+ // Create a table with 1 partition
+ String instantTime1 = "100";
+ HiveTestUtil.createCOWTable(instantTime1, 1, true);
+
+ reInitHiveSyncClient();
+ // Sync the table to metastore
+ reSyncHiveTable();
+
+ List<Partition> partitionsInMetastore = hiveClient.getAllPartitions(HiveTestUtil.TABLE_NAME);
+ assertEquals(1, partitionsInMetastore.size(), "Should have 1 partition in metastore");
+
+ // Add a partition and sync it to metastore
+ String instantTime2 = "101";
+ String newPartition = "2010/02/01";
+ HiveTestUtil.addCOWPartition(newPartition, true, true, instantTime2);
+
+ reInitHiveSyncClient();
+ // Sync the table to metastore
+ reSyncHiveTable();
+
+ partitionsInMetastore = hiveClient.getAllPartitions(HiveTestUtil.TABLE_NAME);
+ assertEquals(2, partitionsInMetastore.size(), "Should have 2 partitions in metastore");
+
+ // Now delete the partition that exists in metastore
+ String instantTime3 = "102";
+ HiveTestUtil.createReplaceCommit(instantTime3, newPartition, WriteOperationType.DELETE_PARTITION, true, true);
+
+ // Add another partition to storage but don't sync to metastore
+ String instantTime4 = "103";
+ String addPartition = "2010/04/01";
+ HiveTestUtil.addCOWPartition(addPartition, true, true, instantTime4);
+
+ reInitHiveSyncClient();
+
+ // Get partition events
+ Set<String> droppedPartitionsOnStorage = hiveClient.getDroppedPartitionsSince(Option.of(instantTime2), Option.of(instantTime2));
+ List<String> writtenPartitionsOnStorage = hiveClient.getWrittenPartitionsSince(Option.of(instantTime2), Option.of(instantTime2));
+ List<PartitionEvent> partitionEvents = hiveClient.getPartitionEvents(
+ partitionsInMetastore, writtenPartitionsOnStorage, droppedPartitionsOnStorage);
+
+ // Verify DROP event is generated for partition that exists in metastore
+ List<PartitionEvent> dropEvents = partitionEvents.stream()
+ .filter(e -> e.eventType == PartitionEventType.DROP)
+ .collect(Collectors.toList());
+ assertEquals(1, dropEvents.size(),
+ "DROP partition event should be generated for partition that exists in metastore");
+ assertEquals(newPartition, dropEvents.get(0).storagePartition);
+
+ // Verify ADD event is generated for the new partition that was added to storage
+ List<PartitionEvent> addEvents = partitionEvents.stream()
+ .filter(e -> e.eventType == PartitionEventType.ADD)
+ .collect(Collectors.toList());
+ assertEquals(1, addEvents.size(),
+ "ADD partition event should be generated for new partition added to storage");
+ assertEquals(addPartition, addEvents.get(0).storagePartition);
+ }
+
@ParameterizedTest
@MethodSource("syncModeAndEnablePushDown")
public void testNonPartitionedSync(String syncMode, String enablePushDown) throws Exception {
diff --git a/hudi-sync/hudi-sync-common/src/main/java/org/apache/hudi/sync/common/HoodieSyncClient.java b/hudi-sync/hudi-sync-common/src/main/java/org/apache/hudi/sync/common/HoodieSyncClient.java
index d7239e0a4b68..b96f2ad76121 100644
--- a/hudi-sync/hudi-sync-common/src/main/java/org/apache/hudi/sync/common/HoodieSyncClient.java
+++ b/hudi-sync/hudi-sync-common/src/main/java/org/apache/hudi/sync/common/HoodieSyncClient.java
@@ -212,11 +212,14 @@ public abstract class HoodieSyncClient implements HoodieMetaSyncOperations, Auto
// Check if the partition values or if hdfs path is the same
List<String> storagePartitionValues = partitionValueExtractor.extractPartitionValuesInPath(storagePartition);
- if (droppedPartitionsOnStorage.contains(storagePartition)) {
- events.add(PartitionEvent.newPartitionDropEvent(storagePartition));
- } else {
- if (!storagePartitionValues.isEmpty()) {
- String storageValue = String.join(", ", storagePartitionValues);
+ if (!storagePartitionValues.isEmpty()) {
+ String storageValue = String.join(", ", storagePartitionValues);
+ if (droppedPartitionsOnStorage.contains(storagePartition)) {
+ if (paths.containsKey(storageValue)) {
+ // Add partition drop event only if it exists in the metastore
+ events.add(PartitionEvent.newPartitionDropEvent(storagePartition));
+ }
+ } else {
if (!paths.containsKey(storageValue)) {
events.add(PartitionEvent.newPartitionAddEvent(storagePartition));
} else if (!paths.get(storageValue).equals(fullStoragePartitionPath)) {