HIVE-28956 : Implement DirectSql for alter table add column cascade command.

Author: vikramahuja1001

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

@@ -410,21 +410,44 @@ public List<Void> run(List<Partition> input) throws Exception {
             if (cascade || retainOnColRemoval) {
               parts = msdb.getPartitions(catName, dbname, name, -1);
               for (Partition part : parts) {
-                Partition oldPart = new Partition(part);
                 List<FieldSchema> oldCols = part.getSd().getCols();
                 part.getSd().setCols(newt.getSd().getCols());
                 List<ColumnStatistics> colStats = updateOrGetPartitionColumnStats(msdb, catName, dbname, name,
                     part.getValues(), oldCols, oldt, part, null, null);
                 assert (colStats.isEmpty());
-                Deadline.checkTimeout();
-                if (cascade) {
-                  msdb.alterPartition(
-                    catName, dbname, name, part.getValues(), part, writeIdList);
-                } else {
-                  // update changed properties (stats)
-                  oldPart.setParameters(part.getParameters());
-                  msdb.alterPartition(catName, dbname, name, part.getValues(), oldPart, writeIdList);
-                }
+              }
+              String catalogName = catName;
+              int partitionBatchSize = MetastoreConf.getIntVar(handler.getConf(),
+                      MetastoreConf.ConfVars.BATCH_RETRIEVE_MAX);
+              if (cascade) {
+                Batchable.runBatched(partitionBatchSize, parts, new Batchable<Partition, Void>() {
+                  @Override
+                  public List<Void> run(List<Partition> input) throws Exception {
+                    Deadline.checkTimeout();
+                    msdb.alterPartitions(catalogName, newDbName, newTblName,
+                            input.stream().map(Partition::getValues).collect(Collectors.toList()),
+                            input, newt.getWriteId(), writeIdList);
+                    return Collections.emptyList();
+                  }
+                });
+              } else {
+                Batchable.runBatched(partitionBatchSize, parts, new Batchable<Partition, Void>() {
+                  @Override
+                  public List<Void> run(List<Partition> input) throws Exception {
+                    Deadline.checkTimeout();
+                    List<Partition> oldParts = new ArrayList<>(parts.size());
+                    // update changed properties (stats)
+                    for (Partition part : input) {
+                      Partition oldPart = new Partition(part);
+                      oldPart.setParameters(part.getParameters());
+                      oldParts.add(oldPart);
+                    }
+                    msdb.alterPartitions(catalogName, newDbName, newTblName,
+                            input.stream().map(Partition::getValues).collect(Collectors.toList()),
+                            oldParts, newt.getWriteId(), writeIdList);
+                    return Collections.emptyList();
+                  }
+                });
               }
             } else {
               // clear all column stats to prevent incorract behaviour in case same column is reintroduced