public static Iterator<Row> computeUpdates(
      Iterator<Row> rowIterator, StructType rowType, String[] identifierFields) {
    Iterator<Row> carryoverRemoveIterator = removeCarryovers(rowIterator, rowType);
    ChangelogIterator changelogIterator =
        new ComputeUpdateIterator(carryoverRemoveIterator, rowType, identifierFields);
    return Iterators.filter(changelogIterator, Objects::nonNull);
  }
@Test
  public void testRowsWithNullValue() {
    final List<Row> rowsWithNull =
        Lists.newArrayList(
            new GenericRowWithSchema(new Object[] {2, null, null, DELETE, 0, 0}, null),
            new GenericRowWithSchema(new Object[] {3, null, null, INSERT, 0, 0}, null),
            new GenericRowWithSchema(new Object[] {4, null, null, DELETE, 0, 0}, null),
            new GenericRowWithSchema(new Object[] {4, null, null, INSERT, 0, 0}, null),
            // mixed null and non-null value in non-identifier columns
            new GenericRowWithSchema(new Object[] {5, null, null, DELETE, 0, 0}, null),
            new GenericRowWithSchema(new Object[] {5, null, "data", INSERT, 0, 0}, null),
            // mixed null and non-null value in identifier columns
            new GenericRowWithSchema(new Object[] {6, null, null, DELETE, 0, 0}, null),
            new GenericRowWithSchema(new Object[] {6, "name", null, INSERT, 0, 0}, null));

    Iterator<Row> iterator =
        ChangelogIterator.computeUpdates(rowsWithNull.iterator(), SCHEMA, IDENTIFIER_FIELDS);
    List<Row> result = Lists.newArrayList(iterator);

    assertEquals(
        "Rows should match",
        Lists.newArrayList(
            new Object[] {2, null, null, DELETE, 0, 0},
            new Object[] {3, null, null, INSERT, 0, 0},
            new Object[] {5, null, null, UPDATE_BEFORE, 0, 0},
            new Object[] {5, null, "data", UPDATE_AFTER, 0, 0},
            new Object[] {6, null, null, DELETE, 0, 0},
            new Object[] {6, "name", null, INSERT, 0, 0}),
        rowsToJava(result));
  }