class CopyPartitionsOperation {

  public void execute(CloseableMetaStoreClient client, Table oldTable, Table newTable) throws TException {
    int count = 0;
    String databaseName = newTable.getDbName();
    String tableName = newTable.getTableName();
    PartitionIterator partitionIterator = new PartitionIterator(client, oldTable, partitionBatchSize);
    while (partitionIterator.hasNext()) {
      List<Partition> batch = new ArrayList<>();
      for (int i = 0; i < partitionBatchSize && partitionIterator.hasNext(); i++) {
        Partition partition = partitionIterator.next();
        count++;
        Partition copy = new Partition(partition);
        copy.setDbName(databaseName);
        copy.setTableName(tableName);
        StorageDescriptor sd = new StorageDescriptor(partition.getSd());
        sd.setCols(newTable.getSd().getCols());
        copy.setSd(sd);
        batch.add(copy);
      }
      LOG.info("Copying batch of size {} to {}.{}", batch.size(), databaseName, tableName);
      client.add_partitions(batch);
    }
    LOG.info("Copied {} partitions to {}.{}", count, databaseName, tableName);
  }

  public  CopyPartitionsOperation();
  @VisibleForTesting  CopyPartitionsOperation(short partitionBatchSize);

}