class UnpartitionedTableReplication implements Replication {

  @Override
  public void replicate() throws CircusTrainException {
    try {
      String replicaDatabaseName = tableReplication.getReplicaDatabaseName();
      String replicaTableName = tableReplication.getReplicaTableName();
      replica.validateReplicaTable(replicaDatabaseName, replicaTableName);
      TableAndStatistics sourceTableAndStatistics = source.getTableAndStatistics(database, table);
      Table sourceTable = sourceTableAndStatistics.getTable();
      SourceLocationManager sourceLocationManager = source.getLocationManager(sourceTable, eventId);
      Path sourceLocation = sourceLocationManager.getTableLocation();

      ReplicaLocationManager replicaLocationManager = replica
          .getLocationManager(TableType.UNPARTITIONED, tableReplication.getReplicaTable().getTableLocation(), eventId,
              sourceLocationManager);
      Path replicaLocation = replicaLocationManager.getTableLocation();

      CopierFactory copierFactory = copierFactoryManager
          .getCopierFactory(sourceLocation, replicaLocation, copierOptions);

      CopierContext copierContext = new CopierContext(tableReplication, eventId, sourceLocation, null, replicaLocation,
          copierOptions, sourceTable, null);
      Copier copier = copierFactory.newInstance(copierContext);
      copierListener.copierStart(copier.getClass().getName());
      try {
        metrics = copier.copy();
      } finally {
        copierListener.copierEnd(metrics);
      }
      sourceLocationManager.cleanUpLocations();

      DataManipulatorFactory dataManipulatorFactory = dataManipulatorFactoryManager
          .getFactory(sourceLocation, replicaLocation, copierOptions);
      DataManipulator dataManipulator = dataManipulatorFactory.newInstance(replicaLocation, copierOptions);
      replica.cleanupReplicaTableIfRequired(replicaDatabaseName, replicaTableName, dataManipulator);
      replica
          .updateMetadata(eventId, sourceTableAndStatistics, replicaDatabaseName, replicaTableName,
              replicaLocationManager);
      replicaLocationManager.cleanUpLocations();

      LOG.info("Replicated table {}.{}.", database, table);
    } catch (Throwable t) {
      throw new CircusTrainException("Unable to replicate", t);
    }
  }

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

}