[core] Tolerate the NoSuchObjectException when report the partition s…

…tatistic (#4708)
apache · Dec 15, 2024 · 72e7150 · 72e7150
1 parent d61f3d2
commit 72e7150
Show file tree

Hide file tree

Showing 5 changed files with 26 additions and 12 deletions.
diff --git a/paimon-core/src/main/java/org/apache/paimon/metastore/MetastoreClient.java b/paimon-core/src/main/java/org/apache/paimon/metastore/MetastoreClient.java
@@ -55,7 +55,8 @@ default void addPartitionsSpec(List<LinkedHashMap<String, String>> partitionSpec
     default void alterPartition(
             LinkedHashMap<String, String> partitionSpec,
             Map<String, String> parameters,
-            long modifyTime)
+            long modifyTime,
+            boolean ignoreIfNotExist)
             throws Exception {
         throw new UnsupportedOperationException();
     }

diff --git a/...paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/partition/HmsReporter.java b/...paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/partition/HmsReporter.java
@@ -93,7 +93,7 @@ public void report(String partition, long modifyTime) throws Exception {
             statistic.put(HIVE_LAST_UPDATE_TIME_PROP, String.valueOf(modifyTime / 1000));
 
             LOG.info("alter partition {} with statistic {}.", partitionSpec, statistic);
-            metastoreClient.alterPartition(partitionSpec, statistic, modifyTime);
+            metastoreClient.alterPartition(partitionSpec, statistic, modifyTime, true);
         }
     }
 

diff --git a/...mmon/src/test/java/org/apache/paimon/flink/sink/partition/AddDonePartitionActionTest.java b/...mmon/src/test/java/org/apache/paimon/flink/sink/partition/AddDonePartitionActionTest.java
@@ -66,7 +66,8 @@ public void markDone(LinkedHashMap<String, String> partitionSpec)
                     public void alterPartition(
                             LinkedHashMap<String, String> partitionSpec,
                             Map<String, String> parameters,
-                            long modifyTime)
+                            long modifyTime,
+                            boolean ignoreIfNotExist)
                             throws Exception {
                         throw new UnsupportedOperationException();
                     }

diff --git a/...on-flink-common/src/test/java/org/apache/paimon/flink/sink/partition/HmsReporterTest.java b/...on-flink-common/src/test/java/org/apache/paimon/flink/sink/partition/HmsReporterTest.java
@@ -117,7 +117,8 @@ public void markDone(LinkedHashMap<String, String> partitionSpec)
                     public void alterPartition(
                             LinkedHashMap<String, String> partitionSpec,
                             Map<String, String> parameters,
-                            long modifyTime)
+                            long modifyTime,
+                            boolean ignoreIfNotExist)
                             throws Exception {
                         partitionParams.put(
                                 PartitionPathUtils.generatePartitionPath(partitionSpec),

diff --git a/...on-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveMetastoreClient.java b/...on-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveMetastoreClient.java
@@ -119,17 +119,28 @@ public void addPartitionsSpec(List<LinkedHashMap<String, String>> partitionSpecs
     public void alterPartition(
             LinkedHashMap<String, String> partitionSpec,
             Map<String, String> parameters,
-            long modifyTime)
+            long modifyTime,
+            boolean ignoreIfNotExist)
             throws Exception {
         List<String> partitionValues = new ArrayList<>(partitionSpec.values());
         int currentTime = (int) (modifyTime / 1000);
-        Partition hivePartition =
-                clients.run(
-                        client ->
-                                client.getPartition(
-                                        identifier.getDatabaseName(),
-                                        identifier.getObjectName(),
-                                        partitionValues));
+        Partition hivePartition;
+        try {
+            hivePartition =
+                    clients.run(
+                            client ->
+                                    client.getPartition(
+                                            identifier.getDatabaseName(),
+                                            identifier.getObjectName(),
+                                            partitionValues));
+        } catch (NoSuchObjectException e) {
+            if (ignoreIfNotExist) {
+                return;
+            } else {
+                throw e;
+            }
+        }
+
         hivePartition.setValues(partitionValues);
         hivePartition.setLastAccessTime(currentTime);
         hivePartition.getParameters().putAll(parameters);