class PartitionedTableMetadataMirrorReplication implements Replication {

  @Override
  public void replicate() throws CircusTrainException {
    try {
      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, Collections.<String, Object>emptyMap());
      ReplicaLocationManager replicaLocationManager = new MetadataMirrorReplicaLocationManager(sourceLocationManager,
          TableType.PARTITIONED);
      sourceLocationManager.cleanUpLocations();
      if (sourcePartitions.isEmpty()) {
        LOG.debug("Update table {}.{} metadata only", database, table);
        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 {
        replica
            .updateMetadata(eventId, sourceTableAndStatistics, sourcePartitionsAndStatistics, replicaDatabaseName,
                replicaTableName, replicaLocationManager);
        int partitionsCopied = sourcePartitions.size();
        LOG
            .info("Metadata mirrored for {} partitions of table {}.{} (no data copied).", partitionsCopied, database,
                table);
      }
    } catch (Throwable t) {
      throw new CircusTrainException("Unable to replicate", t);
    }
  }

    PartitionedTableMetadataMirrorReplication(
      String database,
      String table,
      PartitionPredicate partitionPredicate,
      Source source,
      Replica replica,
      EventIdFactory eventIdFactory,
      String replicaDatabaseName,
      String replicaTableName);

}