[hive] Clean up done markers in Hive partition deletion

paimon-core/src/main/java/org/apache/paimon/operation/ChainTablePartitionExpire.java

@@ -352,11 +352,9 @@ private void batchDropPartitions(
                     for (List<Map<String, String>> batch :
                             Lists.partition(partitionSpecs, expireBatchSize)) {
                         partitionModification.dropPartitions(batch);
-                        partitionModification.dropPartitions(toDonePartitions(batch));
                     }
                 } else {
                     partitionModification.dropPartitions(partitionSpecs);
-                    partitionModification.dropPartitions(toDonePartitions(partitionSpecs));
                 }
             } catch (Catalog.TableNotExistException e) {
                 throw new RuntimeException(e);
@@ -373,23 +371,6 @@ private void batchDropPartitions(
         }
     }
 
-    private List<Map<String, String>> toDonePartitions(
-            List<Map<String, String>> expiredPartitions) {
-        List<Map<String, String>> donePartitions = new ArrayList<>(expiredPartitions.size());
-        for (Map<String, String> partition : expiredPartitions) {
-            LinkedHashMap<String, String> donePartition = new LinkedHashMap<>(partition);
-            Map.Entry<String, String> lastEntry = null;
-            for (Map.Entry<String, String> entry : donePartition.entrySet()) {
-                lastEntry = entry;
-            }
-            if (lastEntry != null) {
-                donePartition.put(lastEntry.getKey(), lastEntry.getValue() + ".done");
-                donePartitions.add(donePartition);
-            }
-        }
-        return donePartitions;
-    }
-
     private Map<BinaryRow, List<BinaryRow>> groupByGroupKey(List<PartitionEntry> partitionEntries) {
         Map<BinaryRow, List<BinaryRow>> grouped = new LinkedHashMap<>();
         for (PartitionEntry entry : partitionEntries) {

paimon-core/src/main/java/org/apache/paimon/operation/NormalPartitionExpire.java

@@ -39,7 +39,6 @@
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Comparator;
-import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ThreadLocalRandom;
@@ -198,36 +197,14 @@ private void doBatchExpire(
         if (partitionModification != null) {
             try {
                 partitionModification.dropPartitions(expiredBatchPartitions);
-                // also drop corresponding .done partitions
-                partitionModification.dropPartitions(toDonePartitions(expiredBatchPartitions));
             } catch (Catalog.TableNotExistException e) {
                 throw new RuntimeException(e);
             }
         } else {
-            // .done partitions only exist when partitionModification != null
-            // (metastore.partitioned-table = true), so no need to handle them here
             commit.dropPartitions(expiredBatchPartitions, commitIdentifier);
         }
     }
 
-    private List<Map<String, String>> toDonePartitions(
-            List<Map<String, String>> expiredPartitions) {
-        List<Map<String, String>> donePartitions = new ArrayList<>(expiredPartitions.size());
-        for (Map<String, String> partition : expiredPartitions) {
-            LinkedHashMap<String, String> donePartition = new LinkedHashMap<>(partition);
-            // append .done suffix to the last partition field value
-            Map.Entry<String, String> lastEntry = null;
-            for (Map.Entry<String, String> entry : donePartition.entrySet()) {
-                lastEntry = entry;
-            }
-            if (lastEntry != null) {
-                donePartition.put(lastEntry.getKey(), lastEntry.getValue() + ".done");
-                donePartitions.add(donePartition);
-            }
-        }
-        return donePartitions;
-    }
-
     private List<Map<String, String>> convertToPartitionString(
             List<List<String>> expiredPartValues) {
         // Sort on the joined form, but build the partition from the original values: a value

paimon-core/src/test/java/org/apache/paimon/operation/ChainTablePartitionExpireTest.java

@@ -889,6 +889,16 @@ public void dropPartitions(List<Map<String, String>> partitions)
                 throws Catalog.TableNotExistException {
             for (Map<String, String> partition : partitions) {
                 droppedPartitions.add(new HashMap<>(partition));
+                // Model the Hive metastore's cleanup of the corresponding marker.
+                Map<String, String> donePartition = new HashMap<>(partition);
+                String lastKey = null;
+                for (String key : partition.keySet()) {
+                    lastKey = key;
+                }
+                if (lastKey != null) {
+                    donePartition.put(lastKey, partition.get(lastKey) + ".done");
+                    droppedPartitions.add(donePartition);
+                }
             }
         }
 

paimon-core/src/test/java/org/apache/paimon/operation/PartitionExpireTest.java

@@ -139,12 +139,7 @@ public void createPartitions(List<Map<String, String>> partitions)
                     @Override
                     public void dropPartitions(List<Map<String, String>> partitions)
                             throws Catalog.TableNotExistException {
-                        for (Map<String, String> partition : partitions) {
-                            // only record partitions that were created
-                            if (createdPartitions.remove(partition)) {
-                                deletedPartitions.add(partition);
-                            }
-                        }
+                        dropPartitionMetadata(partitions);
                         try (FileStoreCommit commit =
                                 table.store()
                                         .newCommit(
@@ -153,6 +148,27 @@ public void dropPartitions(List<Map<String, String>> partitions)
                                                 null)) {
                             commit.dropPartitions(partitions, BatchWriteBuilder.COMMIT_IDENTIFIER);
                         }
+                        // Model HiveCatalog's cleanup of markers after deleting ordinary data.
+                        List<Map<String, String>> donePartitions = new ArrayList<>();
+                        String lastKey =
+                                tableSchema
+                                        .partitionKeys()
+                                        .get(tableSchema.partitionKeys().size() - 1);
+                        for (Map<String, String> partition : partitions) {
+                            Map<String, String> donePartition = new HashMap<>(partition);
+                            donePartition.put(lastKey, partition.get(lastKey) + ".done");
+                            donePartitions.add(donePartition);
+                        }
+                        dropPartitionMetadata(donePartitions);
+                    }
+
+                    private void dropPartitionMetadata(List<Map<String, String>> partitions) {
+                        for (Map<String, String> partition : partitions) {
+                            // only record partitions that were created
+                            if (createdPartitions.remove(partition)) {
+                                deletedPartitions.add(partition);
+                            }
+                        }
                     }
 
                     @Override

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

@@ -35,6 +35,7 @@
 import org.apache.paimon.fs.FileIO;
 import org.apache.paimon.fs.Path;
 import org.apache.paimon.hive.pool.CachedClientPool;
+import org.apache.paimon.operation.FileStoreScan;
 import org.apache.paimon.operation.Lock;
 import org.apache.paimon.options.CatalogOptions;
 import org.apache.paimon.options.Options;
@@ -488,35 +489,72 @@ private void createPartitionsUtil(
     public void dropPartitions(Identifier identifier, List<Map<String, String>> partitions)
             throws TableNotExistException {
         TableSchema schema = this.loadTableSchema(identifier);
-        CoreOptions options = CoreOptions.fromMap(schema.options());
-        boolean tagToPart = options.tagToPartitionField() != null;
-        if (metastorePartitioned(schema)) {
-            List<Map<String, String>> metaPartitions =
+        boolean tagToPart = CoreOptions.fromMap(schema.options()).tagToPartitionField() != null;
+        boolean metastorePartitioned = metastorePartitioned(schema);
+        if (metastorePartitioned) {
+            dropPartitionsFromMetastore(
+                    identifier,
                     tagToPart
                             ? partitions
-                            : removePartitionsExistsInOtherBranches(identifier, partitions);
-            for (Map<String, String> part : metaPartitions) {
-                List<String> partitionValues =
-                        new ArrayList<>(
-                                orderByPartitionKeys(schema.partitionKeys(), part).values());
-                try {
-                    clients()
-                            .execute(
-                                    client ->
-                                            client.dropPartition(
-                                                    identifier.getDatabaseName(),
-                                                    identifier.getTableName(),
-                                                    partitionValues,
-                                                    false));
-                } catch (NoSuchObjectException e) {
-                    // do nothing if the partition not exists
-                } catch (Exception e) {
-                    throw new RuntimeException(e);
-                }
-            }
+                            : removePartitionsStillInUse(identifier, partitions, false),
+                    schema);
         }
         if (!tagToPart) {
             super.dropPartitions(identifier, partitions);
+            if (metastorePartitioned) {
+                dropDonePartitions(identifier, partitions, schema);
+            }
+        }
+    }
+
+    private void dropDonePartitions(
+            Identifier identifier, List<Map<String, String>> partitions, TableSchema schema)
+            throws TableNotExistException {
+        List<Map<String, String>> donePartitions = new ArrayList<>(partitions.size());
+        String lastKey = schema.partitionKeys().get(schema.partitionKeys().size() - 1);
+        for (Map<String, String> partition : partitions) {
+            Map<String, String> donePartition = new LinkedHashMap<>(partition);
+            donePartition.put(lastKey, partition.get(lastKey) + ".done");
+            donePartitions.add(donePartition);
+        }
+
+        // HMS partition names are shared by all branches. A marker is still needed if a branch
+        // owns its ordinary partition or a real STRING partition with the same literal spec.
+        Set<Map<String, String>> specsToCheck = new HashSet<>(partitions);
+        specsToCheck.addAll(donePartitions);
+        Set<Map<String, String>> removable =
+                new HashSet<>(
+                        removePartitionsStillInUse(
+                                identifier, new ArrayList<>(specsToCheck), true));
+        List<Map<String, String>> markersToDrop = new ArrayList<>();
+        for (int i = 0; i < partitions.size(); i++) {
+            if (removable.contains(partitions.get(i))
+                    && removable.contains(donePartitions.get(i))) {
+                markersToDrop.add(donePartitions.get(i));
+            }
+        }
+        dropPartitionsFromMetastore(identifier, markersToDrop, schema);
+    }
+
+    private void dropPartitionsFromMetastore(
+            Identifier identifier, List<Map<String, String>> partitions, TableSchema schema) {
+        for (Map<String, String> part : partitions) {
+            List<String> partitionValues =
+                    new ArrayList<>(orderByPartitionKeys(schema.partitionKeys(), part).values());
+            try {
+                clients()
+                        .execute(
+                                client ->
+                                        client.dropPartition(
+                                                identifier.getDatabaseName(),
+                                                identifier.getTableName(),
+                                                partitionValues,
+                                                false));
+            } catch (NoSuchObjectException e) {
+                // do nothing if the partition not exists
+            } catch (Exception e) {
+                throw new RuntimeException(e);
+            }
         }
     }
 
@@ -753,26 +791,23 @@ public List<Partition> listPartitionsFromHms(Identifier identifier)
                                         Short.MAX_VALUE));
     }
 
-    private List<Map<String, String>> removePartitionsExistsInOtherBranches(
-            Identifier identifier, List<Map<String, String>> inputs) throws TableNotExistException {
+    private List<Map<String, String>> removePartitionsStillInUse(
+            Identifier identifier, List<Map<String, String>> inputs, boolean syntheticPartitions)
+            throws TableNotExistException {
         FileStoreTable mainTable =
                 (FileStoreTable)
                         getTable(
                                 new Identifier(
                                         identifier.getDatabaseName(), identifier.getTableName()));
 
-        InternalRowPartitionComputer partitionComputer =
-                new InternalRowPartitionComputer(
-                        mainTable.coreOptions().partitionDefaultName(),
-                        mainTable.rowType().project(mainTable.partitionKeys()),
-                        mainTable.partitionKeys().toArray(new String[0]),
-                        mainTable.coreOptions().legacyPartitionName());
         List<String> branchNames = new ArrayList<>(mainTable.branchManager().branches());
         branchNames.add(DEFAULT_MAIN_BRANCH);
 
         Set<Map<String, String>> inputsToRemove = new HashSet<>(inputs);
         for (String branchName : branchNames) {
-            if (branchName.equals(identifier.getBranchNameOrDefault())) {
+            // Ordinary partitions are checked before their data is dropped. Marker cleanup runs
+            // afterwards and must also protect real, colliding partitions on the current branch.
+            if (!syntheticPartitions && branchName.equals(identifier.getBranchNameOrDefault())) {
                 continue;
             }
 
@@ -782,8 +817,22 @@ private List<Map<String, String>> removePartitionsExistsInOtherBranches(
                 continue;
             }
 
-            branchTableForPartitionExistence(mainTable, branchName).newScan()
-                    .withPartitionsFilter(new ArrayList<>(inputsToRemove)).listPartitions().stream()
+            FileStoreTable branchTable = branchTableForPartitionExistence(mainTable, branchName);
+            InternalRowPartitionComputer partitionComputer =
+                    new InternalRowPartitionComputer(
+                            branchTable.coreOptions().partitionDefaultName(),
+                            branchTable.rowType().project(branchTable.partitionKeys()),
+                            branchTable.partitionKeys().toArray(new String[0]),
+                            branchTable.coreOptions().legacyPartitionName());
+            // Ownership depends on the latest physical snapshot. A table batch scan may instead
+            // follow persisted query selectors such as scan.snapshot-id or scan.tag-name.
+            FileStoreScan scan = branchTable.store().newScan();
+            if (!syntheticPartitions) {
+                scan.withPartitionsFilter(new ArrayList<>(inputsToRemove));
+            }
+            // Synthetic values need not fit the partition type. Compare their literal specs with
+            // actual branch partitions, including real STRING values that collide with markers.
+            scan.listPartitions().stream()
                     .map(partitionComputer::generatePartValues)
                     .forEach(inputsToRemove::remove);
         }