HIVE-29042: Performance degradation for add columns cascade (#5895)

Author: deniskuzZ

standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/DirectSqlUpdatePart.java

@@ -744,7 +744,7 @@ private void updateStorageDescriptorInBatch(Map<Long, StorageDescriptor> idToSd)
       throws MetaException {
     Map<Long, Long> sdIdToCdId = new HashMap<>();
     Map<Long, Long> sdIdToSerdeId = new HashMap<>();
-    List<Long> cdIds = new ArrayList<>();
+    Set<Long> cdIds = new HashSet<>();
     List<Long> validSdIds = filterIdsByNonNullValue(new ArrayList<>(idToSd.keySet()), idToSd);
     Batchable.runBatched(maxBatchSize, validSdIds, new Batchable<Long, Void>() {
       @Override
@@ -793,7 +793,7 @@ public List<Void> run(List<Long> input) throws Exception {
     updateBucketColsInBatch(idToBucketCols, validSdIds);
     updateSortColsInBatch(idToSortCols, validSdIds);
     updateSkewedInfoInBatch(idToSkewedInfo, validSdIds);
-    Map<Long, Long> sdIdToNewCdId = updateCDInBatch(cdIds, validSdIds, sdIdToCdId, sdIdToNewColumns);
+    Map<Long, Long> sdIdToNewCdId = updateCDInBatch(cdIds.stream().toList(), validSdIds, sdIdToCdId, sdIdToNewColumns);
     updateSerdeInBatch(serdeIds, serdeIdToSerde);
     updateParamTableInBatch("\"SERDE_PARAMS\"", "\"SERDE_ID\"", serdeIds, serdeParamsOpt);
 
@@ -1158,6 +1158,10 @@ public List<Void> run(List<Long> input) throws Exception {
     Map<Long, List<Pair<Integer, Integer>>> oldCdIdToColIdxPairs = new HashMap<>();
     for (Long sdId : sdIds) {
       Long cdId = sdIdToCdId.get(sdId);
+      if (oldCdIdToNewCdId.containsKey(cdId)) {
+        sdIdToNewCdId.put(sdId, oldCdIdToNewCdId.get(cdId));
+        continue;
+      }
       List<Pair<Integer, FieldSchema>> cols = cdIdToColIdxPair.get(cdId);
       // Placeholder to avoid IndexOutOfBoundsException.
       List<FieldSchema> oldCols = new ArrayList<>(Collections.nCopies(cols.size(), null));

standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/HiveAlterHandler.java

@@ -420,35 +420,30 @@ public List<Void> run(List<Partition> input) throws Exception {
               req.setCatName(catName);
               req.setMaxParts((short) -1);
               parts = handler.get_partitions_req(req).getPartitions();
-              Table finalOldt = oldt;
+              Table table = oldt;
               int partitionBatchSize = MetastoreConf.getIntVar(handler.getConf(),
-                      MetastoreConf.ConfVars.BATCH_RETRIEVE_MAX);
+                  MetastoreConf.ConfVars.BATCH_RETRIEVE_MAX);
               Batchable.runBatched(partitionBatchSize, parts, new Batchable<Partition, Void>() {
                 @Override
                 public List<Void> run(List<Partition> input) throws Exception {
                   List<Partition> oldParts = new ArrayList<>(input.size());
                   List<List<String>> partVals = input.stream().map(Partition::getValues).collect(Collectors.toList());
-                  // update changed properties (stats)
                   for (Partition part : input) {
                     Partition oldPart = new Partition(part);
                     List<FieldSchema> oldCols = part.getSd().getCols();
                     part.getSd().setCols(newt.getSd().getCols());
                     List<ColumnStatistics> colStats = updateOrGetPartitionColumnStats(msdb, catalogName, databaseName,
-                            tableName, part.getValues(), oldCols, finalOldt, part, null, null);
+                        tableName, part.getValues(), oldCols, table, part, null, null);
                     assert (colStats.isEmpty());
                     if (!cascade) {
+                      // update changed properties (stats)
                       oldPart.setParameters(part.getParameters());
                       oldParts.add(oldPart);
                     }
                   }
                   Deadline.checkTimeout();
-                  if (cascade) {
-                    msdb.alterPartitions(catalogName, databaseName, tableName, partVals, input, newt.getWriteId(),
-                            writeIdList);
-                  } else {
-                    msdb.alterPartitions(catalogName, newDbName, newTblName, partVals, oldParts, newt.getWriteId(),
-                            writeIdList);
-                  }
+                  msdb.alterPartitions(catalogName, databaseName, tableName,
+                      partVals, (cascade) ? input : oldParts, newt.getWriteId(), writeIdList);
                   return Collections.emptyList();
                 }
               });