[core] Reduce get partition requests in migrating hive table (#5364)

Author: Zouxxyy

paimon-core/src/main/java/org/apache/paimon/migrate/FileMetaUtils.java

@@ -240,7 +240,7 @@ private static DataFileMeta constructFileMeta(
 
     public static BinaryRow writePartitionValue(
             RowType partitionRowType,
-            Map<String, String> partitionValues,
+            List<String> partitionValues,
             List<BinaryWriter.ValueSetter> valueSetters,
             String partitionDefaultName) {
 
@@ -250,7 +250,7 @@ public static BinaryRow writePartitionValue(
         List<DataField> fields = partitionRowType.getFields();
 
         for (int i = 0; i < fields.size(); i++) {
-            String partitionName = partitionValues.get(fields.get(i).name());
+            String partitionName = partitionValues.get(i);
             if (partitionName.equals(partitionDefaultName)) {
                 binaryRowWriter.setNullAt(i);
             } else {

paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/migrate/HiveMigrator.java

@@ -163,25 +163,20 @@ public void executeMigrate() throws Exception {
             FileStoreTable paimonTable = (FileStoreTable) hiveCatalog.getTable(identifier);
             checkPaimonTable(paimonTable);
 
-            List<String> partitionsNames =
-                    client.listPartitionNames(sourceDatabase, sourceTable, Short.MAX_VALUE);
+            List<Partition> partitions =
+                    client.listPartitions(sourceDatabase, sourceTable, Short.MAX_VALUE);
             checkCompatible(sourceHiveTable, paimonTable);
 
             List<MigrateTask> tasks = new ArrayList<>();
             Map<Path, Path> rollBack = new ConcurrentHashMap<>();
-            if (partitionsNames.isEmpty()) {
+            if (partitions.isEmpty()) {
                 tasks.add(
                         importUnPartitionedTableTask(
                                 fileIO, sourceHiveTable, paimonTable, rollBack));
             } else {
                 tasks.addAll(
                         importPartitionedTableTask(
-                                client,
-                                fileIO,
-                                partitionsNames,
-                                sourceHiveTable,
-                                paimonTable,
-                                rollBack));
+                                fileIO, partitions, sourceHiveTable, paimonTable, rollBack));
             }
 
             List<Future<CommitMessage>> futures =
@@ -292,13 +287,11 @@ public Schema from(
     }
 
     private List<MigrateTask> importPartitionedTableTask(
-            IMetaStoreClient client,
             FileIO fileIO,
-            List<String> partitionNames,
+            List<Partition> partitions,
             Table sourceTable,
             FileStoreTable paimonTable,
-            Map<Path, Path> rollback)
-            throws Exception {
+            Map<Path, Path> rollback) {
         List<MigrateTask> migrateTasks = new ArrayList<>();
         List<BinaryWriter.ValueSetter> valueSetters = new ArrayList<>();
 
@@ -309,17 +302,14 @@ private List<MigrateTask> importPartitionedTableTask(
                 .getFieldTypes()
                 .forEach(type -> valueSetters.add(BinaryWriter.createValueSetter(type)));
 
-        for (String partitionName : partitionNames) {
-            Partition partition =
-                    client.getPartition(
-                            sourceTable.getDbName(), sourceTable.getTableName(), partitionName);
-            Map<String, String> values = client.partitionNameToSpec(partitionName);
+        for (Partition partition : partitions) {
+            List<String> partitionValues = partition.getValues();
             String format = parseFormat(partition.getSd().getSerdeInfo().toString());
             String location = partition.getSd().getLocation();
             BinaryRow partitionRow =
                     FileMetaUtils.writePartitionValue(
                             partitionRowType,
-                            values,
+                            partitionValues,
                             valueSetters,
                             coreOptions.partitionDefaultName());
             Path path = paimonTable.store().pathFactory().bucketPath(partitionRow, 0);