public List<ColumnMatchResult<?>> getMismatchedColumns(List<Column> columns, ChecksumResult controlChecksum, ChecksumResult testChecksum)
    {
        return columns.stream()
                .flatMap(column -> columnValidators.get(column.getCategory()).get().validate(column, controlChecksum, testChecksum).stream())
                .filter(columnMatchResult -> !columnMatchResult.isMatched())
                .collect(toImmutableList());
    }
@Test
    public void testFloatingPointArray()
    {
        List<Column> columns = ImmutableList.of(DOUBLE_ARRAY_COLUMN, REAL_ARRAY_COLUMN);

        Map<String, Object> stableCounts = new HashMap<>(10);
        stableCounts.put("double_array$nan_count", 2L);
        stableCounts.put("double_array$pos_inf_count", 3L);
        stableCounts.put("double_array$neg_inf_count", 4L);
        stableCounts.put("double_array$cardinality_checksum", new SqlVarbinary(new byte[] {0xb}));
        stableCounts.put("double_array$cardinality_sum", 10L);
        stableCounts.put("real_array$nan_count", 2L);
        stableCounts.put("real_array$pos_inf_count", 3L);
        stableCounts.put("real_array$neg_inf_count", 4L);
        stableCounts.put("real_array$cardinality_checksum", new SqlVarbinary(new byte[] {0xd}));
        stableCounts.put("real_array$cardinality_sum", 10L);

        Map<String, Object> emptyTableCounts = new HashMap<>(12);
        emptyTableCounts.put("double_array$nan_count", null);
        emptyTableCounts.put("double_array$pos_inf_count", null);
        emptyTableCounts.put("double_array$neg_inf_count", null);
        emptyTableCounts.put("double_array$cardinality_checksum", null);
        emptyTableCounts.put("double_array$cardinality_sum", null);
        emptyTableCounts.put("double_array$sum", null);
        emptyTableCounts.put("real_array$nan_count", null);
        emptyTableCounts.put("real_array$pos_inf_count", null);
        emptyTableCounts.put("real_array$neg_inf_count", null);
        emptyTableCounts.put("real_array$cardinality_checksum", null);
        emptyTableCounts.put("real_array$cardinality_sum", null);
        emptyTableCounts.put("real_array$sum", null);

        final ChecksumResult controlChecksum = new ChecksumResult(
                5,
                ImmutableMap.<String, Object>builder()
                        .putAll(stableCounts)
                        .put("double_array$sum", 1.0)
                        .put("real_array$sum", 1.0)
                        .build());

        // Matched
        ChecksumResult testChecksum = new ChecksumResult(
                5,
                ImmutableMap.<String, Object>builder()
                        .putAll(stableCounts)
                        .put("double_array$sum", 1 + RELATIVE_ERROR_MARGIN)
                        .put("real_array$sum", 1 - RELATIVE_ERROR_MARGIN + RELATIVE_ERROR_MARGIN * RELATIVE_ERROR_MARGIN)
                        .build());
        assertTrue(checksumValidator.getMismatchedColumns(columns, controlChecksum, testChecksum).isEmpty());

        // Matched two empty tables
        final ChecksumResult testEmptyChecksum = new ChecksumResult(0, emptyTableCounts);
        assertTrue(checksumValidator.getMismatchedColumns(columns, testEmptyChecksum, testEmptyChecksum).isEmpty());

        // Mismatched empty and non-empty (number of rows differs).
        assertThatThrownBy(() -> checksumValidator.getMismatchedColumns(columns, controlChecksum, testEmptyChecksum))
                .isInstanceOf(IllegalArgumentException.class);

        // Mismatched
        testChecksum = new ChecksumResult(
                5,
                ImmutableMap.<String, Object>builder()
                        .put("double_array$sum", 1.0)
                        .put("double_array$nan_count", 0L)
                        .put("double_array$pos_inf_count", 3L)
                        .put("double_array$neg_inf_count", 4L)
                        .put("double_array$cardinality_checksum", new SqlVarbinary(new byte[] {0xb}))
                        .put("double_array$cardinality_sum", 10L)
                        .put("real_array$sum", 1.0)
                        .put("real_array$nan_count", 2L)
                        .put("real_array$pos_inf_count", 0L)
                        .put("real_array$neg_inf_count", 4L)
                        .put("real_array$cardinality_checksum", new SqlVarbinary(new byte[] {0xd}))
                        .put("real_array$cardinality_sum", 10L)
                        .build());
        assertMismatchedColumns(columns, controlChecksum, testChecksum, DOUBLE_ARRAY_COLUMN, REAL_ARRAY_COLUMN);

        testChecksum = new ChecksumResult(
                5,
                ImmutableMap.<String, Object>builder()
                        .put("double_array$sum", 1.0)
                        .put("double_array$nan_count", 2L)
                        .put("double_array$pos_inf_count", 3L)
                        .put("double_array$neg_inf_count", 0L)
                        .put("double_array$cardinality_checksum", new SqlVarbinary(new byte[] {0xb}))
                        .put("double_array$cardinality_sum", 10L)
                        .put("real_array$sum", 1 - RELATIVE_ERROR_MARGIN)
                        .put("real_array$nan_count", 2L)
                        .put("real_array$pos_inf_count", 3L)
                        .put("real_array$neg_inf_count", 4L)
                        .put("real_array$cardinality_checksum", new SqlVarbinary(new byte[] {0xd}))
                        .put("real_array$cardinality_sum", 10L)
                        .build());
        List<ColumnMatchResult<?>> mismatchedColumns = assertMismatchedColumns(columns, controlChecksum, testChecksum, DOUBLE_ARRAY_COLUMN, REAL_ARRAY_COLUMN);
        assertEquals(mismatchedColumns.get(1).getMessage(), Optional.of("relative error: 1.0000500025000149E-4"));
    }