Spark 3.3: Output the net changes across snapshots in CDC

spark/v3.3/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java

@@ -18,6 +18,8 @@
  */
 package org.apache.iceberg.spark.extensions;
 
+import static org.junit.Assert.assertThrows;
+
 import java.util.List;
 import java.util.Map;
 import org.apache.iceberg.ChangelogOperation;
@@ -411,6 +413,54 @@ public void testRemoveCarryOversWithoutUpdatedRows() {
         sql("select * from %s order by _change_ordinal, id, data", viewName));
   }
 
+  @Test
+  public void testNetChangesWithRemoveCarryOvers() {
+    // partitioned by id
+    createTableWith3Columns();
+
+    // insert rows: (1, 'a', 12) (2, 'b', 11) (2, 'e', 12)
+    sql("INSERT INTO %s VALUES (1, 'a', 12), (2, 'b', 11), (2, 'e', 12)", tableName);
+    Table table = validationCatalog.loadTable(tableIdent);
+    Snapshot snap1 = table.currentSnapshot();
+
+    // delete rows: (2, 'b', 11) (2, 'e', 12), insert rows: (3, 'c', 13) (2, 'd', 11) (2, 'e', 12)
+    sql("INSERT OVERWRITE %s VALUES (3, 'c', 13), (2, 'd', 11), (2, 'e', 12)", tableName);
+    table.refresh();
+    Snapshot snap2 = table.currentSnapshot();
+
+    // delete rows: (2, 'd', 11) (2, 'e', 12) (3, 'c', 13), insert rows: (3, 'c', 15) (2, 'e', 12)
+    sql("INSERT OVERWRITE %s VALUES (3, 'c', 15), (2, 'e', 12)", tableName);
+    table.refresh();
+    Snapshot snap3 = table.currentSnapshot();
+
+    List<Object[]> returns =
+        sql(
+            "CALL %s.system.create_changelog_view(table => '%s', net_changes => true)",
+            catalogName, tableName);
+
+    String viewName = (String) returns.get(0)[0];
+
+    assertEquals(
+        "Rows should match",
+        ImmutableList.of(
+            row(1, "a", 12, INSERT, 0, snap1.snapshotId()),
+            row(3, "c", 15, INSERT, 2, snap3.snapshotId()),
+            row(2, "e", 12, INSERT, 2, snap3.snapshotId())),
+        sql("select * from %s order by _change_ordinal, data", viewName));
+  }
+
+  @Test
+  public void testNetChangesWithComputeUpdates() {
+    createTableWith2Columns();
+    assertThrows(
+        "Should fail because net_changes is not supported with computing updates",
+        IllegalArgumentException.class,
+        () ->
+            sql(
+                "CALL %s.system.create_changelog_view(table => '%s', identifier_columns => array('id'), net_changes => true)",
+                catalogName, tableName));
+  }
+
   @Test
   public void testNotRemoveCarryOvers() {
     createTableWith3Columns();

spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/ChangelogIterator.java

@@ -79,6 +79,11 @@ public static Iterator<Row> removeCarryovers(Iterator<Row> rowIterator, StructTy
     return Iterators.filter(changelogIterator, Objects::nonNull);
   }
 
+  public static Iterator<Row> removeNetCarryovers(Iterator<Row> rowIterator, StructType rowType) {
+    ChangelogIterator changelogIterator = new RemoveNetCarryoverIterator(rowIterator, rowType);
+    return Iterators.filter(changelogIterator, Objects::nonNull);
+  }
+
   protected boolean isDifferentValue(Row currentRow, Row nextRow, int idx) {
     return !Objects.equals(nextRow.get(idx), currentRow.get(idx));
   }

spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/RemoveCarryoverIterator.java

@@ -48,14 +48,16 @@
  */
 class RemoveCarryoverIterator extends ChangelogIterator {
   private final int[] indicesToIdentifySameRow;
+  private final StructType rowType;
 
   private Row cachedDeletedRow = null;
   private long deletedRowCount = 0;
   private Row cachedNextRecord = null;
 
   RemoveCarryoverIterator(Iterator<Row> rowIterator, StructType rowType) {
     super(rowIterator, rowType);
-    this.indicesToIdentifySameRow = generateIndicesToIdentifySameRow(rowType.size());
+    this.rowType = rowType;
+    this.indicesToIdentifySameRow = generateIndicesToIdentifySameRow();
   }
 
   @Override
@@ -139,8 +141,8 @@ private boolean hasCachedDeleteRow() {
     return cachedDeletedRow != null;
   }
 
-  private int[] generateIndicesToIdentifySameRow(int columnSize) {
-    int[] indices = new int[columnSize - 1];
+  private int[] generateIndicesToIdentifySameRow() {
+    int[] indices = new int[rowType.size() - 1];
     for (int i = 0; i < indices.length; i++) {
       if (i < changeTypeIndex()) {
         indices[i] = i;
@@ -151,13 +153,21 @@ private int[] generateIndicesToIdentifySameRow(int columnSize) {
     return indices;
   }
 
-  private boolean isSameRecord(Row currentRow, Row nextRow) {
-    for (int idx : indicesToIdentifySameRow) {
+  protected boolean isSameRecord(Row currentRow, Row nextRow) {
+    for (int idx : indicesToIdentifySameRow()) {
       if (isDifferentValue(currentRow, nextRow, idx)) {
         return false;
       }
     }
 
     return true;
   }
+
+  protected StructType rowType() {
+    return rowType;
+  }
+
+  protected int[] indicesToIdentifySameRow() {
+    return indicesToIdentifySameRow;
+  }
 }

spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/RemoveNetCarryoverIterator.java

@@ -0,0 +1,145 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.spark;
+
+import java.util.Iterator;
+import java.util.List;
+import org.apache.iceberg.MetadataColumns;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.types.StructType;
+
+/**
+ * This class computes the net changes across multiple snapshots. It takes a row iterator, and
+ * assumes the following:
+ *
+ * <ul>
+ *   <li>The row iterator is partitioned by all columns.
+ *   <li>The row iterator is sorted by all columns, change order, and change type. The change order
+ *       is 1-to-1 mapping to snapshot id.
+ * </ul>
+ */
+public class RemoveNetCarryoverIterator extends RemoveCarryoverIterator {
+
+  private final List<Row> cachedRows = Lists.newArrayList();
+  private final int[] indicesToIdentifySameRow;
+
+  private Row cachedNextRow = null;
+
+  protected RemoveNetCarryoverIterator(Iterator<Row> rowIterator, StructType rowType) {
+    super(rowIterator, rowType);
+    this.indicesToIdentifySameRow = generateIndicesToIdentifySameRow();
+  }
+
+  @Override
+  public boolean hasNext() {
+    if (!cachedRows.isEmpty()) {
+      return true;
+    }
+
+    if (cachedNextRow != null) {
+      return true;
+    }
+
+    return rowIterator().hasNext();
+  }
+
+  @Override
+  public Row next() {
+    // if there are cached rows, return one of them from the beginning
+    if (!cachedRows.isEmpty()) {
+      return cachedRows.remove(0);
+    }
+
+    Row currentRow = getCurrentRow();
+
+    // if there is no next row, return the current row
+    if (!rowIterator().hasNext()) {
+      return currentRow;
+    }
+
+    Row nextRow = rowIterator().next();
+
+    cachedRows.add(currentRow);
+    // if they are the same row, remove the cached row or stack to the cached row
+    while (isSameRecord(currentRow, nextRow)) {
+      if (matched(currentRow, nextRow)) {
+        // if matched, remove both rows, remove the last row in the cache
+        cachedRows.remove(cachedRows.size() - 1);
+        nextRow = null;
+      } else {
+        // stack the row into the cache
+        cachedRows.add(nextRow);
+        nextRow = null;
+      }
+
+      // break the loop if there is no next row or the cache is empty
+      if (cachedRows.isEmpty() || !rowIterator().hasNext()) {
+        break;
+      }
+
+      // get the current row from the cache, and get the next row from the iterator for the next
+      // loop
+      currentRow = cachedRows.get(cachedRows.size() - 1);
+      nextRow = rowIterator().next();
+    }
+
+    // if they are different rows, hit the boundary, cache the next row
+    cachedNextRow = nextRow;
+    return null;
+  }
+
+  private Row getCurrentRow() {
+    Row currentRow;
+    if (cachedNextRow != null) {
+      currentRow = cachedNextRow;
+      cachedNextRow = null;
+    } else {
+      currentRow = rowIterator().next();
+    }
+    return currentRow;
+  }
+
+  private boolean matched(Row currentRow, Row nextRow) {
+    return (nextRow.getString(changeTypeIndex()).equals(INSERT)
+            && currentRow.getString(changeTypeIndex()).equals(DELETE))
+        || (nextRow.getString(changeTypeIndex()).equals(DELETE)
+            && currentRow.getString(changeTypeIndex()).equals(INSERT));
+  }
+
+  private int[] generateIndicesToIdentifySameRow() {
+    int changeOrdinalIndex = rowType().fieldIndex(MetadataColumns.CHANGE_ORDINAL.name());
+    int snapshotIdIndex = rowType().fieldIndex(MetadataColumns.COMMIT_SNAPSHOT_ID.name());
+
+    int[] indices = new int[rowType().size() - 3];
+
+    for (int i = 0, j = 0; i < indices.length; i++) {
+      if (i != changeTypeIndex() && i != changeOrdinalIndex && i != snapshotIdIndex) {
+        indices[j] = i;
+        j++;
+      }
+    }
+    return indices;
+  }
+
+  @Override
+  protected int[] indicesToIdentifySameRow() {
+    return indicesToIdentifySameRow;
+  }
+}