[hive]Fix hive catalog create partitions using table location.

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

@@ -350,7 +350,6 @@ public void createPartitions(Identifier identifier, List<Map<String, String>> pa
         Identifier tableIdentifier =
                 Identifier.create(identifier.getDatabaseName(), identifier.getTableName());
         Table hmsTable = getHmsTable(tableIdentifier);
-        Path location = getTableLocation(tableIdentifier, hmsTable);
         TableSchema schema = loadTableSchema(tableIdentifier, hmsTable);
 
         if (!metastorePartitioned(schema)) {
@@ -359,27 +358,19 @@ public void createPartitions(Identifier identifier, List<Map<String, String>> pa
 
         int currentTime = (int) (System.currentTimeMillis() / 1000);
         StorageDescriptor sd = hmsTable.getSd();
-        String dataFilePath =
-                hmsTable.getParameters().containsKey(DATA_FILE_PATH_DIRECTORY.key())
-                        ? sd.getLocation()
-                                + "/"
-                                + hmsTable.getParameters().get(DATA_FILE_PATH_DIRECTORY.key())
-                        : sd.getLocation();
+        String dataFilePath = getDataFilePath(tableIdentifier, hmsTable);
         List<Partition> hivePartitions = new ArrayList<>();
         for (Map<String, String> partitionSpec : partitions) {
             Partition hivePartition = new Partition();
             StorageDescriptor newSd = new StorageDescriptor(sd);
-            newSd.setLocation(
-                    dataFilePath
-                            + "/"
-                            + PartitionPathUtils.generatePartitionPath(
-                                    new LinkedHashMap<>(partitionSpec)));
             hivePartition.setDbName(identifier.getDatabaseName());
             hivePartition.setTableName(identifier.getTableName());
             hivePartition.setValues(new ArrayList<>(partitionSpec.values()));
             hivePartition.setSd(newSd);
             hivePartition.setCreateTime(currentTime);
             hivePartition.setLastAccessTime(currentTime);
+            String partitionLocation = getPartitionLocation(dataFilePath, partitionSpec);
+            locationHelper.specifyPartitionLocation(hivePartition, partitionLocation);
             hivePartitions.add(hivePartition);
         }
         try {
@@ -400,6 +391,9 @@ public void dropPartitions(Identifier identifier, List<Map<String, String>> part
                     tagToPart
                             ? partitions
                             : removePartitionsExistsInOtherBranches(identifier, partitions);
+            Table hmsTable = getHmsTable(identifier);
+            boolean externalTable = isExternalTable(hmsTable);
+            String dataFilePath = getDataFilePath(identifier, hmsTable);
             for (Map<String, String> part : metaPartitions) {
                 List<String> partitionValues = new ArrayList<>(part.values());
                 try {
@@ -410,6 +404,11 @@ public void dropPartitions(Identifier identifier, List<Map<String, String>> part
                                             identifier.getTableName(),
                                             partitionValues,
                                             false));
+
+                    if (!externalTable) {
+                        String partitionLocation = getPartitionLocation(dataFilePath, part);
+                        locationHelper.dropPathIfRequired(new Path(partitionLocation), fileIO);
+                    }
                 } catch (NoSuchObjectException e) {
                     // do nothing if the partition not exists
                 } catch (Exception e) {
@@ -422,6 +421,19 @@ public void dropPartitions(Identifier identifier, List<Map<String, String>> part
         }
     }
 
+    private String getDataFilePath(Identifier tableIdentifier, Table hmsTable) {
+        String tableLocation = getTableLocation(tableIdentifier, hmsTable).toUri().toString();
+        return hmsTable.getParameters().containsKey(DATA_FILE_PATH_DIRECTORY.key())
+                ? tableLocation + "/" + hmsTable.getParameters().get(DATA_FILE_PATH_DIRECTORY.key())
+                : tableLocation;
+    }
+
+    private String getPartitionLocation(String dataFilePath, Map<String, String> partitionSpec) {
+        return dataFilePath
+                + "/"
+                + PartitionPathUtils.generatePartitionPath(new LinkedHashMap<>(partitionSpec));
+    }
+
     @Override
     public void alterPartitions(
             Identifier identifier, List<org.apache.paimon.partition.Partition> partitions)

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

@@ -22,6 +22,7 @@
 import org.apache.paimon.fs.Path;
 
 import org.apache.hadoop.hive.metastore.api.Database;
+import org.apache.hadoop.hive.metastore.api.Partition;
 import org.apache.hadoop.hive.metastore.api.Table;
 
 import java.io.IOException;
@@ -40,4 +41,8 @@ public interface LocationHelper {
     void specifyDatabaseLocation(Path path, Database database);
 
     String getDatabaseLocation(Database database);
+
+    void specifyPartitionLocation(Partition partition, String location);
+
+    String getPartitionLocation(Partition partition);
 }

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

@@ -22,6 +22,7 @@
 import org.apache.paimon.fs.Path;
 
 import org.apache.hadoop.hive.metastore.api.Database;
+import org.apache.hadoop.hive.metastore.api.Partition;
 import org.apache.hadoop.hive.metastore.api.Table;
 
 /** Helper for Setting Location in Hive Table Storage. */
@@ -58,4 +59,14 @@ public void specifyDatabaseLocation(Path path, Database database) {
     public String getDatabaseLocation(Database database) {
         return database.getLocationUri();
     }
+
+    @Override
+    public void specifyPartitionLocation(Partition partition, String location) {
+        partition.getSd().setLocation(location);
+    }
+
+    @Override
+    public String getPartitionLocation(Partition partition) {
+        return partition.getSd().getLocation();
+    }
 }

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

@@ -22,6 +22,7 @@
 import org.apache.paimon.fs.Path;
 
 import org.apache.hadoop.hive.metastore.api.Database;
+import org.apache.hadoop.hive.metastore.api.Partition;
 import org.apache.hadoop.hive.metastore.api.Table;
 
 import java.io.IOException;
@@ -82,4 +83,20 @@ public String getDatabaseLocation(Database database) {
 
         return database.getLocationUri();
     }
+
+    @Override
+    public void specifyPartitionLocation(Partition partition, String location) {
+        partition.putToParameters(LocationKeyExtractor.TBPROPERTIES_LOCATION_KEY, location);
+    }
+
+    @Override
+    public String getPartitionLocation(Partition partition) {
+        String location =
+                partition.getParameters().get(LocationKeyExtractor.TBPROPERTIES_LOCATION_KEY);
+        if (location != null) {
+            return location;
+        }
+
+        return partition.getSd().getLocation();
+    }
 }