class PartitionedTableReplication implements Replication {

  @Override
  public void replicate() throws CircusTrainException {
    try {
      String replicaDatabaseName = tableReplication.getReplicaDatabaseName();
      String replicaTableName = tableReplication.getReplicaTableName();

      TableAndStatistics sourceTableAndStatistics = source.getTableAndStatistics(database, table);
      Table sourceTable = sourceTableAndStatistics.getTable();

      PartitionsAndStatistics sourcePartitionsAndStatistics = source
          .getPartitions(sourceTable, partitionPredicate.getPartitionPredicate(),
              partitionPredicate.getPartitionPredicateLimit());
      List<Partition> sourcePartitions = sourcePartitionsAndStatistics.getPartitions();

      replica.validateReplicaTable(replicaDatabaseName, replicaTableName);

      // We expect all partitions to be under the table base path
      SourceLocationManager sourceLocationManager = source
          .getLocationManager(sourceTable, sourcePartitions, eventId, copierOptions);
      Path sourceBaseLocation = sourceLocationManager.getTableLocation();
      List<Path> sourceSubLocations = sourceLocationManager.getPartitionLocations();

      ReplicaLocationManager replicaLocationManager = replica
          .getLocationManager(TableType.PARTITIONED, tableReplication.getReplicaTable().getTableLocation(), eventId,
              sourceLocationManager);
      Path replicaPartitionBaseLocation = replicaLocationManager.getPartitionBaseLocation();

      DataManipulatorFactory dataManipulatorFactory = dataManipulatorFactoryManager
          .getFactory(sourceBaseLocation, replicaPartitionBaseLocation, copierOptions);
      DataManipulator dataManipulator = dataManipulatorFactory.newInstance(replicaPartitionBaseLocation, copierOptions);

      if (sourcePartitions.isEmpty()) {
        LOG.debug("Update table {}.{} metadata only", database, table);
        replica.cleanupReplicaTableIfRequired(replicaDatabaseName, replicaTableName, dataManipulator);
        replica
            .updateMetadata(eventId, sourceTableAndStatistics, replicaDatabaseName, replicaTableName,
                replicaLocationManager);
        LOG
            .info("No matching partitions found on table {}.{} with predicate {}."
                + " Table metadata updated, no partitions were updated.", database, table, partitionPredicate);
      } else {
        CopierFactory copierFactory = copierFactoryManager
            .getCopierFactory(sourceBaseLocation, replicaPartitionBaseLocation, copierOptions);
        CopierContext copierContext = new CopierContext(tableReplication, eventId, sourceBaseLocation,
            sourceSubLocations, replicaPartitionBaseLocation, copierOptions, sourceTable, sourcePartitions);
        Copier copier = copierFactory.newInstance(copierContext);
        copierListener.copierStart(copier.getClass().getName());
        try {
          metrics = copier.copy();
        } finally {
          copierListener.copierEnd(metrics);
        }
        sourceLocationManager.cleanUpLocations();

        replica.cleanupReplicaTableIfRequired(replicaDatabaseName, replicaTableName, dataManipulator);
        replica
            .updateMetadata(eventId, sourceTableAndStatistics, sourcePartitionsAndStatistics, replicaDatabaseName,
                replicaTableName, replicaLocationManager);
        replicaLocationManager.cleanUpLocations();

        int partitionsCopied = sourcePartitions.size();
        LOG.info("Replicated {} partitions of table {}.{}.", partitionsCopied, database, table);
      }
    } catch (Throwable t) {
      throw new CircusTrainException("Unable to replicate", t);
    }
  }

    PartitionedTableReplication(
      TableReplication tableReplication,
      PartitionPredicate partitionPredicate,
      Source source,
      Replica replica,
      CopierFactoryManager copierFactoryManager,
      EventIdFactory eventIdFactory,
      Map<String, Object> copierOptions,
      CopierListener copierListener,
      DataManipulatorFactoryManager dataManipulatorFactoryManager);

}