PartitionComparator implements Callable<PartitionStats> { public PartitionStats call() { PartitionStats partitionStats = new PartitionStats(); if (source == null || target == null) { logger.error("Skipping partition because one result was null (timeout despite retries)"); partitionStats.skipped = true; return partitionStats; } while (hasNextRow(Type.SOURCE) && hasNextRow(Type.TARGET)) { Row sourceRow = getNextRow(Type.SOURCE); Row targetRow = getNextRow(Type.TARGET); if (!clusteringsEqual(sourceRow, targetRow)) { partitionStats.allClusteringsMatch = false; return partitionStats; } partitionStats.matchedRows++; checkRegularColumnEquality(partitionStats, sourceRow, targetRow); } if (hasNextRow(Type.SOURCE) || hasNextRow(Type.TARGET)) partitionStats.allClusteringsMatch = false; return partitionStats; } PartitionComparator(TableSpec tableSpec,
                               Iterator<Row> source,
                               Iterator<Row> target,
                               RetryStrategyProvider retryStrategyProvider); PartitionStats call();  }
@Test public void sourceIsNull() { TableSpec t = spec("table1", names("c1", "c2"), names("v1", "v2")); PartitionComparator comparator = comparator(t, null, rows(row(t, 0, 1, 2, 3))); PartitionStats stats = comparator.call(); assertStats(stats, true, true, 0, 0, 0); }
@Test public void targetIsNull() { TableSpec t = spec("table1", names("c1", "c2"), names("v1", "v2")); PartitionComparator comparator = comparator(t, rows(row(t, 0, 1, 2, 3)), null); PartitionStats stats = comparator.call(); assertStats(stats, true, true, 0, 0, 0); }
@Test public void sourceIsEmpty() { TableSpec t = spec("table1", names("c1", "c2"), names("v1", "v2")); PartitionComparator comparator = comparator(t, rows(), rows(row(t, 0, 1, 2, 3))); PartitionStats stats = comparator.call(); assertStats(stats, false, false, 0, 0, 0); }
@Test public void targetIsEmpty() { TableSpec t = spec("table1", names("c1", "c2"), names("v1", "v2")); PartitionComparator comparator = comparator(t, rows(row(t, 0, 1, 2, 3)), rows()); PartitionStats stats = comparator.call(); assertStats(stats, false, false, 0, 0, 0); }
@Test public void sourceAndTargetAreEmpty() { TableSpec t = spec("table1", names("c1", "c2"), names("v1", "v2")); PartitionComparator comparator = comparator(t, rows(), rows()); PartitionStats stats = comparator.call(); assertStats(stats, false, true, 0, 0, 0); }
@Test public void sourceContainsExtraRowsAtStart() { TableSpec t = spec("table1", names("c1", "c2"), names("v1", "v2")); PartitionComparator comparator = comparator(t, rows(row(t, 0, 1, 2, 3), row(t, 10, 11, 12, 13)), rows(row(t, 10, 11, 12, 13))); PartitionStats stats = comparator.call(); assertStats(stats, false, false, 0, 0, 0); }
@Test public void targetContainsExtraRowsAtStart() { TableSpec t = spec("table1", names("c1", "c2"), names("v1", "v2")); PartitionComparator comparator = comparator(t, rows(row(t, 10, 11, 12, 13)), rows(row(t, 0, 1, 2, 3), row(t, 10, 11, 12, 13))); PartitionStats stats = comparator.call(); assertStats(stats, false, false, 0, 0, 0); }
@Test public void sourceContainsExtraRowsAtEnd() { TableSpec t = spec("table1", names("c1", "c2"), names("v1", "v2")); PartitionComparator comparator = comparator(t, rows(row(t, 0, 1, 2, 3), row(t, 10, 11, 12, 13)), rows(row(t, 0, 1, 2, 3))); PartitionStats stats = comparator.call(); assertStats(stats, false, false, 1, 2, 0); }
@Test public void targetContainsExtraRowsAtEnd() { TableSpec t = spec("table1", names("c1", "c2"), names("v1", "v2")); PartitionComparator comparator = comparator(t, rows(row(t, 0, 1, 2, 3)), rows(row(t, 0, 1, 2, 3), row(t, 10, 11, 12, 13))); PartitionStats stats = comparator.call(); assertStats(stats, false, false, 1, 2, 0); }
@Test public void withoutClusteringsAllRowsMatching() { TableSpec t = spec("table1", names(), names("v1", "v2")); PartitionComparator comparator = comparator(t, rows(row(t, 0, 1), row(t, 1, 11), row(t, 2, 12)), rows(row(t, 0, 1), row(t, 1, 11), row(t, 2, 12))); PartitionStats stats = comparator.call(); assertStats(stats, false, true, 3, 6, 0); }
@Test public void singleClusteringAllRowsMatching() { TableSpec t = spec("table1", names("c1"), names("v1", "v2")); PartitionComparator comparator = comparator(t, rows(row(t, 0, 1, 2), row(t, 1, 11, 12), row(t, 2, 21, 22)), rows(row(t, 0, 1, 2), row(t, 1, 11, 12), row(t, 2, 21, 22))); PartitionStats stats = comparator.call(); assertStats(stats, false, true, 3, 6, 0); }
@Test public void multipleClusteringAllRowsMatching() { TableSpec t = spec("table1", names("c1", "c2"), names("v1", "v2")); PartitionComparator comparator = comparator(t, rows(row(t, 0, 1, 2, 3), row(t, 1, 11, 12, 13), row(t, 2, 21, 22, 23)), rows(row(t, 0, 1, 2, 3), row(t, 1, 11, 12, 13), row(t, 2, 21, 22, 23))); PartitionStats stats = comparator.call(); assertStats(stats, false, true, 3, 6, 0); }
@Test public void withoutClusteringsWithMismatches() { TableSpec t = spec("table1", names(), names("v1", "v2")); PartitionComparator comparator = comparator(t, rows(row(t, 0, 1), row(t, 1, 11), row(t, 2, 12)), rows(row(t, 0, 1), row(t, 1, 1100), row(t, 2, 1200))); PartitionStats stats = comparator.call(); assertStats(stats, false, true, 3, 4, 2); }
@Test public void singleClusteringWithMismatches() { TableSpec t = spec("table1", names("c1"), names("v1", "v2")); PartitionComparator comparator = comparator(t, rows(row(t, 0, 1, 2), row(t, 1, 11, 21), row(t, 2, 12, 22)), rows(row(t, 0, 1, 20), row(t, 1, 1100, 21), row(t, 2, 12, 1200))); PartitionStats stats = comparator.call(); assertStats(stats, false, true, 3, 3, 3); }