private Map<Long, Long> updateCDInBatch(List<Long> cdIds, List<Long> sdIds, Map<Long, Long> sdIdToCdId,
                                          Map<Long, List<FieldSchema>> sdIdToNewColumns) throws MetaException {
    Map<Long, List<Pair<Integer, FieldSchema>>> cdIdToColIdxPair = new HashMap<>();
    Batchable.runBatched(maxBatchSize, cdIds, new Batchable<Long, Void>() {
      @Override
      public List<Void> run(List<Long> input) throws Exception {
        String idLists = MetaStoreDirectSql.getIdListForIn(input);
        String queryText = "select \"CD_ID\", \"COMMENT\", \"COLUMN_NAME\", \"TYPE_NAME\", " +
            "\"INTEGER_IDX\" from \"COLUMNS_V2\" where \"CD_ID\" in (" + idLists + ")";
        try (QueryWrapper query = new QueryWrapper(pm.newQuery("javax.jdo.query.SQL", queryText))) {
          List<Object[]> sqlResult = executeWithArray(query.getInnerQuery(), null, queryText);
          for (Object[] row : sqlResult) {
            Long id = extractSqlLong(row[0]);
            String comment = extractSqlClob(row[1]);
            String name = extractSqlClob(row[2]);
            String type = extractSqlClob(row[3]);
            int index = extractSqlInt(row[4]);
            FieldSchema field = new FieldSchema(name, type, comment);
            cdIdToColIdxPair.computeIfAbsent(id, k -> new ArrayList<>()).add(Pair.of(index, field));
          }
        }
        return null;
      }
    });
    List<Long> newCdIds = new ArrayList<>();
    Map<Long, List<FieldSchema>> newCdIdToCols = new HashMap<>();
    Map<Long, Long> oldCdIdToNewCdId = new HashMap<>();
    Map<Long, Long> sdIdToNewCdId = new HashMap<>();
    // oldCdId -> [(oldIdx, newIdx)], used to update KEY_CONSTRAINTS
    Map<Long, List<Pair<Integer, Integer>>> oldCdIdToColIdxPairs = new HashMap<>();
    for (Long sdId : sdIds) {
      Long cdId = sdIdToCdId.get(sdId);
      List<Pair<Integer, FieldSchema>> cols = cdIdToColIdxPair.get(cdId);
      // Placeholder to avoid IndexOutOfBoundsException.
      List<FieldSchema> oldCols = new ArrayList<>(Collections.nCopies(cols.size(), null));
      cols.forEach(pair -> oldCols.set(pair.getLeft(), pair.getRight()));

      List<FieldSchema> newCols = sdIdToNewColumns.get(sdId);
      // Use the new column descriptor only if the old column descriptor differs from the new one.
      if (oldCols == null || !oldCols.equals(newCols)) {
        if (oldCols != null && newCols != null) {
          Long newCdId = getDataStoreId(MColumnDescriptor.class);
          newCdIds.add(newCdId);
          newCdIdToCols.put(newCdId, newCols);
          oldCdIdToNewCdId.put(cdId, newCdId);
          sdIdToNewCdId.put(sdId, newCdId);
          for (int i = 0; i < oldCols.size(); i++) {
            FieldSchema oldCol = oldCols.get(i);
            int newIdx = newCols.indexOf(oldCol);
            if (newIdx != -1) {
              oldCdIdToColIdxPairs.computeIfAbsent(cdId, k -> new ArrayList<>()).add(Pair.of(i, newIdx));
            }
          }
        }
      }
    }

    insertCDInBatch(newCdIds, newCdIdToCols);
    // TODO: followed the jdo implement now, but it should be an error in such case:
    //       partitions use the default table cd, when changing partition cd with
    //       constraint key mapping, the constraints will be update unexpected.
    updateKeyConstraintsInBatch(oldCdIdToNewCdId, oldCdIdToColIdxPairs);

    return sdIdToNewCdId;
  }
Gets the next CS id from sequence MPartitionColumnStatistics and increment the CS id by numStats.
@return The CD id before update.
standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/DirectSqlUpdatePart.java