class HiveDifferences {

  public void run() {
    int partitionsChanged = 0;
    TableAndMetadata source = CLEAN_TABLE_FUNCTION.apply(sourceTableToTableAndMetadata(sourceTable));
    Optional<TableAndMetadata> replica = Optional.absent();
    if (replicaTable.isPresent()) {
      replica = Optional.of(CLEAN_TABLE_FUNCTION.apply(replicaTableToTableAndMetadata(replicaTable.get())));
    }
    diffListener.onDiffStart(source, replica);
    List<Diff<Object, Object>> tableDiffs = comparator(TableAndMetadata.class).compare(source, replica.orNull());
    if (!tableDiffs.isEmpty()) {
      diffListener.onChangedTable(tableDiffs);
    }

    while (sourcePartitionIterator.hasNext() && partitionsChanged < partitionLimit) {
      Partition sourcePartition = sourcePartitionIterator.next();
      String sourcePartitionName = partitionName(source.getTable(), sourcePartition);
      PartitionAndMetadata sourcePartitionAndMetadata = CLEAN_PARTITION_FUNCTION
          .apply(sourcePartitionToPartitionAndMetadata(sourcePartition));

      Partition replicaPartition = null;
      try {
        if (replica.isPresent()) {
          replicaPartition = replicaPartitionFetcher.get().fetch(sourcePartitionName);
        }
      } catch (PartitionNotFoundException e) {
        // Ignore ...
      }
      if (replicaPartition == null) {
        diffListener.onNewPartition(sourcePartitionName, sourcePartition);
        partitionsChanged++;
        continue;
      }

      PartitionAndMetadata replicaPartitionAndMetadata = CLEAN_PARTITION_FUNCTION
          .apply(replicaPartitionToPartitionAndMetadata(replicaPartition));
      List<Diff<Object, Object>> partitionDiffs = comparator(PartitionAndMetadata.class)
          .compare(sourcePartitionAndMetadata, replicaPartitionAndMetadata);
      if (!partitionDiffs.isEmpty()) {
        diffListener.onChangedPartition(sourcePartitionName, sourcePartition, partitionDiffs);
        partitionsChanged++;
        continue;
      }

      String sourceChecksum = checksumFunction.apply(new Path(sourcePartitionAndMetadata.getSourceLocation()));
      String replicaChecksum = replicaPartition
          .getParameters()
          .get(CircusTrainTableParameter.PARTITION_CHECKSUM.parameterName());
      if (replicaChecksum == null || !sourceChecksum.equals(replicaChecksum)) {
        diffListener.onDataChanged(sourcePartitionName, sourcePartition);
        partitionsChanged++;
      }

      // Partition remains unchanged
    }
    diffListener.onDiffEnd();
  }

  private  HiveDifferences(
      ComparatorRegistry comparatorRegistry,
      DiffListener diffListener,
      Table sourceTable,
      Iterator<Partition> sourcePartitionIterator,
      Optional<Table> replicaTable,
      Optional<? extends PartitionFetcher> replicaPartitionFetcher,
      Function<Path, String> checksumFunction,
      int partitionLimit);

}