class PartitionedTableMetadataUpdateReplication 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);

      try (CloseableMetaStoreClient client = replica.getMetaStoreClientSupplier().get()) {
        if (sourcePartitions.isEmpty()) {
          ReplicaLocationManager replicaLocationManager = newMetadataUpdateReplicaLocationManager(client,
              targetTableLocation);
          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 {
          String previousLocation = getPreviousLocation(client);
          ReplicaLocationManager replicaLocationManager = newMetadataUpdateReplicaLocationManager(client,
              previousLocation);

          PartitionsAndStatistics sourcePartitionsAndStatisticsThatWereReplicated = filterOnReplicatedPartitions(client,
              sourcePartitionsAndStatistics, sourceTable.getPartitionKeys());
          replica
              .updateMetadata(eventId, sourceTableAndStatistics, sourcePartitionsAndStatisticsThatWereReplicated,
                  replicaDatabaseName, replicaTableName, replicaLocationManager);
          int partitionsCopied = sourcePartitions.size();
          LOG
              .info("Metadata updated for {} partitions of table {}.{}. (no data copied)", partitionsCopied, database,
                  table);
        }
      }

    } catch (Throwable t) {
      throw new CircusTrainException("Unable to replicate", t);
    }
  }

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

}