More tests.

Author: wypoon

spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestChangelogTable.java

@@ -26,6 +26,7 @@
 import static org.apache.iceberg.TableProperties.UPDATE_MODE;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.assertj.core.api.Assumptions.assumeThat;
 
 import java.util.List;
 import org.apache.iceberg.DataOperations;
@@ -79,13 +80,11 @@ public void testDataFilters() {
     sql("INSERT INTO %s VALUES (1, 'c'), (2, 'c'), (3, 'c')", tableName);
 
     Table table = validationCatalog.loadTable(tableIdent);
-
     Snapshot snap1 = table.currentSnapshot();
 
     sql("DELETE FROM %s WHERE id = 3", tableName);
 
     table.refresh();
-
     Snapshot snap2 = table.currentSnapshot();
 
     assertEquals(
@@ -123,13 +122,11 @@ public void testUpdates() {
     createTableWithDefaultRows();
 
     Table table = validationCatalog.loadTable(tableIdent);
-
     Snapshot snap2 = table.currentSnapshot();
 
     sql("UPDATE %s SET id = -2 WHERE data = 'b'", tableName);
 
     table.refresh();
-
     Snapshot snap3 = table.currentSnapshot();
 
     assertEquals(
@@ -295,6 +292,42 @@ public void testManifestRewritesAreIgnored() {
         sql("SELECT id, _change_type FROM %s.changes ORDER BY id", tableName));
   }
 
+  @TestTemplate
+  public void testDataRewritesAreIgnored() {
+    assumeThat(formatVersion).isEqualTo(2);
+
+    createTable();
+    sql("INSERT INTO %s VALUES (1, 'c'), (2, 'c'), (3, 'c')", tableName);
+
+    Table table = validationCatalog.loadTable(tableIdent);
+    Snapshot snap1 = table.currentSnapshot();
+
+    sql("DELETE FROM %s WHERE id = 3", tableName);
+
+    table.refresh();
+    Snapshot snap2 = table.currentSnapshot();
+
+    sql(
+        "CALL %s.system.rewrite_data_files(table => '%s', "
+            + "options => map('delete-file-threshold','1'))",
+        catalogName, tableIdent);
+
+    sql("DELETE FROM %s WHERE id = 1", tableName);
+
+    table.refresh();
+    Snapshot snap3 = table.currentSnapshot();
+
+    assertEquals(
+        "Should have expected rows",
+        ImmutableList.of(
+            row(1, "c", "INSERT", 0, snap1.snapshotId()),
+            row(2, "c", "INSERT", 0, snap1.snapshotId()),
+            row(3, "c", "INSERT", 0, snap1.snapshotId()),
+            row(3, "c", "DELETE", 1, snap2.snapshotId()),
+            row(1, "c", "DELETE", 2, snap3.snapshotId())),
+        sql("SELECT * FROM %s.changes ORDER BY _change_ordinal, id, _change_type", tableName));
+  }
+
   @TestTemplate
   public void testMetadataColumns() {
     createTableWithDefaultRows();

spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/source/TestChangelogReader.java

@@ -342,6 +342,52 @@ public void testMixOfPositionAndEqualityDeletes() throws IOException {
     assertEquals("Should have expected rows", expectedRows, internalRowsToJava(rows));
   }
 
+  @Test
+  public void testAddingAndDeletingInSameCommit() throws IOException {
+    GenericRecord record = GenericRecord.create(table.schema());
+    List<Record> records1b = Lists.newArrayList();
+    records1b.add(record.copy("id", 28, "data", "a"));
+    records1b.add(record.copy("id", 29, "data", "a"));
+    records1b.add(record.copy("id", 43, "data", "b"));
+    records1b.add(record.copy("id", 44, "data", "b"));
+    records1b.add(record.copy("id", 61, "data", "c"));
+    records1b.add(record.copy("id", 89, "data", "d"));
+    DataFile dataFile1b = writeDataFile(records1b);
+
+    List<Pair<CharSequence, Long>> deletes =
+        Lists.newArrayList(
+            Pair.of(dataFile1b.path(), 0L), // id = 28
+            Pair.of(dataFile1b.path(), 3L) // id = 44
+            );
+
+    Pair<DeleteFile, CharSequenceSet> posDeletes =
+        FileHelpers.writeDeleteFile(
+            table,
+            Files.localOutput(File.createTempFile("junit", null, temp.toFile())),
+            TestHelpers.Row.of(0),
+            deletes);
+
+    table
+        .newRowDelta()
+        .addRows(dataFile1b)
+        .addDeletes(posDeletes.first())
+        .validateDataFilesExist(posDeletes.second())
+        .commit();
+    // the resulting records in the table are the same as records1
+    long snapshotId1 = table.currentSnapshot().snapshotId();
+
+    table.newAppend().appendFile(dataFile2).commit();
+    long snapshotId2 = table.currentSnapshot().snapshotId();
+
+    List<InternalRow> rows = getChangelogRows();
+
+    List<Object[]> expectedRows = Lists.newArrayList();
+    addExpectedRows(expectedRows, ChangelogOperation.INSERT, snapshotId1, 0, records1);
+    addExpectedRows(expectedRows, ChangelogOperation.INSERT, snapshotId2, 1, records2);
+
+    assertEquals("Should have expected rows", expectedRows, internalRowsToJava(rows));
+  }
+
   private IncrementalChangelogScan newScan() {
     return table.newIncrementalChangelogScan();
   }