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<>();
    List<Long> validSdIds = filterIdsByNonNullValue(new ArrayList<>(idToSd.keySet()), idToSd);
    Batchable.runBatched(maxBatchSize, validSdIds, new Batchable<Long, Void>() {
      @Override
      public List<Void> run(List<Long> input) throws Exception {
        String idLists = MetaStoreDirectSql.getIdListForIn(input);
        String queryText = "select \"SD_ID\", \"CD_ID\", \"SERDE_ID\" from \"SDS\" " +
            "where \"SD_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 sdId = extractSqlLong(row[0]);
            Long cdId = extractSqlLong(row[1]);
            Long serdeId = extractSqlLong(row[2]);
            sdIdToCdId.put(sdId, cdId);
            sdIdToSerdeId.put(sdId, serdeId);
            cdIds.add(cdId);
          }
        }
        return null;
      }
    });

    Map<Long, Optional<Map<String, String>>> sdParamsOpt = new HashMap<>();
    Map<Long, List<String>> idToBucketCols = new HashMap<>();
    Map<Long, List<Order>> idToSortCols = new HashMap<>();
    Map<Long, SkewedInfo> idToSkewedInfo = new HashMap<>();
    Map<Long, List<FieldSchema>> sdIdToNewColumns = new HashMap<>();
    List<Long> serdeIds = new ArrayList<>();
    Map<Long, SerDeInfo> serdeIdToSerde = new HashMap<>();
    Map<Long, Optional<Map<String, String>>> serdeParamsOpt = new HashMap<>();
    for (Long sdId : validSdIds) {
      StorageDescriptor sd = idToSd.get(sdId);
      sdParamsOpt.put(sdId, Optional.ofNullable(sd.getParameters()));
      idToBucketCols.put(sdId, sd.getBucketCols());
      idToSortCols.put(sdId, sd.getSortCols());
      idToSkewedInfo.put(sdId, sd.getSkewedInfo());
      sdIdToNewColumns.put(sdId, sd.getCols());

      Long serdeId = sdIdToSerdeId.get(sdId);
      serdeIds.add(serdeId);
      serdeIdToSerde.put(serdeId, sd.getSerdeInfo());
      serdeParamsOpt.put(serdeId, Optional.ofNullable(sd.getSerdeInfo().getParameters()));
    }

    updateParamTableInBatch("\"SD_PARAMS\"", "\"SD_ID\"", validSdIds, sdParamsOpt);
    updateBucketColsInBatch(idToBucketCols, validSdIds);
    updateSortColsInBatch(idToSortCols, validSdIds);
    updateSkewedInfoInBatch(idToSkewedInfo, validSdIds);
    Map<Long, Long> sdIdToNewCdId = updateCDInBatch(cdIds, validSdIds, sdIdToCdId, sdIdToNewColumns);
    updateSerdeInBatch(serdeIds, serdeIdToSerde);
    updateParamTableInBatch("\"SERDE_PARAMS\"", "\"SERDE_ID\"", serdeIds, serdeParamsOpt);

    List<Long> cdIdsMayDelete = sdIdToCdId.entrySet().stream()
        .filter(entry -> sdIdToNewCdId.containsKey(entry.getKey()))
        .map(entry -> entry.getValue())
        .collect(Collectors.toList());

    // Update SDS table after CDS to get the freshest CD_ID values.
    sdIdToCdId.replaceAll((sdId, cdId) ->
        sdIdToNewCdId.containsKey(sdId) ? sdIdToNewCdId.get(sdId) : cdId);
    updateSDInBatch(validSdIds, idToSd, sdIdToCdId);

    List<Long> usedIds = Batchable.runBatched(maxBatchSize, cdIdsMayDelete,
        new Batchable<Long, Long>() {
          @Override
          public List<Long> run(List<Long> input) throws Exception {
            String idLists = MetaStoreDirectSql.getIdListForIn(input);
            String queryText = "select \"CD_ID\" from \"SDS\" where \"CD_ID\" in ( " + idLists + ")";
            try (QueryWrapper query = new QueryWrapper(pm.newQuery("javax.jdo.query.SQL", queryText))) {
              List<Long> sqlResult = executeWithArray(query.getInnerQuery(), null, queryText);
              return new ArrayList<>(sqlResult);
            }
          }
    });
    List<Long> unusedCdIds = cdIdsMayDelete.stream().filter(id -> !usedIds.contains(id)).collect(Collectors.toList());

    deleteCDInBatch(unusedCdIds);
  }
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