Add the parameter: metastore.drop-partition-ignore-nonexistent to ignore exceptions when a Hive partition does not exist.

.gitignore

@@ -50,3 +50,5 @@ Cargo.lock
 *libvortex_jni*
 ### Tantivy lib ###
 *libtantivy_jni*
+
+.cursor-svg-check/

paimon-api/src/main/java/org/apache/paimon/CoreOptions.java

@@ -1673,6 +1673,20 @@ public InlineElement getDescription() {
                                     + "you need to create this table as a partitioned table in Hive metastore.\n"
                                     + "This config option does not affect the default filesystem metastore.");
 
+    public static final ConfigOption<Boolean> METASTORE_DROP_PARTITION_IGNORE_NONEXISTENT =
+            key("metastore.drop-partition-ignore-nonexistent")
+                    .booleanType()
+                    .defaultValue(false)
+                    .withDescription(
+                            "Whether to ignore HMS drop-partition failures when the partition "
+                                    + "is confirmed not to exist in Hive Metastore. "
+                                    + "Useful when HMS has been manually cleaned up or returns "
+                                    + "non-standard exceptions (e.g., MetaException caused by "
+                                    + "NullPointerException in some HMS versions) instead of "
+                                    + "NoSuchObjectException for missing partitions. "
+                                    + "When enabled, Paimon will verify the partition's existence "
+                                    + "before deciding to ignore the error.");
+
     public static final ConfigOption<String> METASTORE_TAG_TO_PARTITION =
             key("metastore.tag-to-partition")
                     .stringType()
@@ -3426,6 +3440,10 @@ public boolean partitionedTableInMetastore() {
         return options.get(METASTORE_PARTITIONED_TABLE);
     }
 
+    public boolean metastoreDropPartitionIgnoreNonexistent() {
+        return options.get(METASTORE_DROP_PARTITION_IGNORE_NONEXISTENT);
+    }
+
     @Nullable
     public String tagToPartitionField() {
         return options.get(METASTORE_TAG_TO_PARTITION);

paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalog.java

@@ -65,6 +65,7 @@
 import org.apache.hadoop.hive.metastore.TableType;
 import org.apache.hadoop.hive.metastore.api.Database;
 import org.apache.hadoop.hive.metastore.api.FieldSchema;
+import org.apache.hadoop.hive.metastore.api.MetaException;
 import org.apache.hadoop.hive.metastore.api.NoSuchObjectException;
 import org.apache.hadoop.hive.metastore.api.Partition;
 import org.apache.hadoop.hive.metastore.api.PartitionEventType;
@@ -443,6 +444,7 @@ public void dropPartitions(Identifier identifier, List<Map<String, String>> part
         TableSchema schema = this.loadTableSchema(identifier);
         CoreOptions options = CoreOptions.fromMap(schema.options());
         boolean tagToPart = options.tagToPartitionField() != null;
+        boolean ignorePartitionNonexistent = options.metastoreDropPartitionIgnoreNonexistent();
         if (metastorePartitioned(schema)) {
             List<Map<String, String>> metaPartitions =
                     tagToPart
@@ -461,6 +463,20 @@ public void dropPartitions(Identifier identifier, List<Map<String, String>> part
                                                     false));
                 } catch (NoSuchObjectException e) {
                     // do nothing if the partition not exists
+                } catch (MetaException e) {
+                    if (ignorePartitionNonexistent
+                            && !partitionExistsInHms(identifier, partitionValues)) {
+                        LOG.warn(
+                                "Drop partition {} of table {} returned MetaException, but "
+                                        + "the partition is confirmed not to exist in HMS. "
+                                        + "Ignoring the error and continuing with Paimon-side drop. "
+                                        + "Original exception: {}",
+                                part,
+                                identifier,
+                                e.getMessage());
+                    } else {
+                        throw new RuntimeException(e);
+                    }
                 } catch (Exception e) {
                     throw new RuntimeException(e);
                 }
@@ -471,6 +487,31 @@ public void dropPartitions(Identifier identifier, List<Map<String, String>> part
         }
     }
 
+    private boolean partitionExistsInHms(Identifier identifier, List<String> partitionValues) {
+        try {
+            clients.run(
+                    client ->
+                            client.getPartition(
+                                    identifier.getDatabaseName(),
+                                    identifier.getTableName(),
+                                    partitionValues));
+            return true;
+        } catch (NoSuchObjectException e) {
+            return false;
+        } catch (Exception e) {
+            // In case of HMS connection or other unknown errors, conservatively assume the
+            // partition may exist and redirect the original exception upwards to avoid accidental
+            // deletion.
+            LOG.warn(
+                    "Failed to verify partition existence in HMS for table {}, "
+                            + "partition values {}: {}",
+                    identifier,
+                    partitionValues,
+                    e.getMessage());
+            return true;
+        }
+    }
+
     private String getDataFilePath(Identifier tableIdentifier, Table hmsTable) {
         String tableLocation = getTableLocation(tableIdentifier, hmsTable).toString();
         return hmsTable.getParameters().containsKey(DATA_FILE_PATH_DIRECTORY.key())