Repository: apache/iceberg (language: python)

Issue:
Spark: CDC does not respect when the table is rolled back. 
### Apache Iceberg version

1.4.3

### Query engine

Spark

### Please describe the bug 🐞

We had to rollback our table because it had some broken snapshots. We are turning that table which gets upserts into a changelog stream in the downstream and process it that way. We use time boundaries. The way how it seems to work is that it looks at the history of the table and do some sort of a time travel query to find the recent snapshot id as of the end timestamp we pass down the the CDC procedure. 

But since it only uses the history entries which does not give enough info about if the snapshots are in the link of the main branch reference.

Here is the problematic line which calls the function in the iceberg-core

https://github.com/apache/iceberg/blob/426818bfe7fa93e8c677ebf886638d5c50db597b/spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/source/SparkScanBuilder.java#L530

https://github.com/apache/iceberg/blob/main/core/src/main/java/org/apache/iceberg/util/SnapshotUtil.java#L350-L358

I think it should disregard the snapshots when they are no longer in the main branch link




Discussion / hints:
@javrasya Do you mean when the `endSnapshotId` is no longer an ancestor of the current snapshot, it should be skipped? For example, in the following test, the second change log should only contain `row(1, "a", "INSERT", 0, snap1.snapshotId())`?

```java
  @TestTemplate
  public void testQueryWithRollback() {
    createTable();

    sql("INSERT INTO %s VALUES (1, 'a')", tableName);
    Table table = validationCatalog.loadTable(tableIdent);
    Snapshot snap1 = table.currentSnapshot();
    long rightAfterSnap1 = waitUntilAfter(snap1.timestampMillis());

    sql("INSERT INTO %s VALUES (2, 'b')", tableName);
    table.refresh();
    Snapshot snap2 = table.currentSnapshot();
    long rightAfterSnap2 = waitUntilAfter(snap2.timestampMillis());

    sql("CALL %s.system.rollback_to_snapshot('%s', %d)", catalogName, tableIdent, snap1.snapshotId());
    table.refresh();
    Snapshot snap4 = table.currentSnapshot();
    assertThat(snap4).isEqualTo(snap1);

    sql("INSERT OVERWRITE %s VALUES (-2, 'a')", tableName);
    table.refresh();
    Snapshot snap3 = table.currentSnapshot();
    long rightAfterSnap3 = waitUntilAfter(snap3.timestampMillis());

    assertEquals(
        "Should have expected changed rows up to snapshot 3",
        ImmutableList.of(
            row(1, "a", "INSERT", 0, snap1.snapshotId()),
            row(1, "a", "DELETE", 1, snap3.snapshotId()),
            row(-2, "a", "INSERT", 1, snap3.snapshotId())),
        changelogRecords(null, rightAfterSnap3));

    assertEquals(
        "Should have expected changed rows up to snapshot 2",
        ImmutableList.of(
            row(1, "a", "INSERT", 0, snap1.snapshotId()),
            row(2, "b", "INSERT", 1, snap2.snapshotId())),
        changelogRecords(null, rightAfterSnap2));
```
Exactly @manuzhang . It feels like it should filter that out and this is a bug. Wdyt?

Fix patch:
diff --git a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkScanBuilder.java b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkScanBuilder.java
index 09c09c6caad7..d6f34231ae75 100644
--- a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkScanBuilder.java
+++ b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkScanBuilder.java
@@ -546,7 +546,7 @@ public Scan buildChangelogScan() {
     }
 
     if (endTimestamp != null) {
-      endSnapshotId = SnapshotUtil.nullableSnapshotIdAsOfTime(table, endTimestamp);
+      endSnapshotId = getEndSnapshotId(endTimestamp);
       if ((startSnapshotId == null && endSnapshotId == null)
           || (startSnapshotId != null && startSnapshotId.equals(endSnapshotId))) {
         emptyScan = true;
@@ -589,6 +589,17 @@ private Long getStartSnapshotId(Long startTimestamp) {
     }
   }
 
+  private Long getEndSnapshotId(Long endTimestamp) {
+    Long endSnapshotId = null;
+    for (Snapshot snapshot : SnapshotUtil.currentAncestors(table)) {
+      if (snapshot.timestampMillis() <= endTimestamp) {
+        endSnapshotId = snapshot.snapshotId();
+        break;
+      }
+    }
+    return endSnapshotId;
+  }
+
   public Scan buildMergeOnReadScan() {
     Preconditions.checkArgument(
         readConf.snapshotId() == null && readConf.asOfTimestamp() == null && readConf.tag() == null,


Test patch:
diff --git a/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestChangelogTable.java b/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestChangelogTable.java
index daf45626753b..79c52f25e31f 100644
--- a/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestChangelogTable.java
+++ b/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestChangelogTable.java
@@ -291,6 +291,69 @@ public void testMetadataColumns() {
         rows);
   }
 
+  @TestTemplate
+  public void testQueryWithRollback() {
+    createTable();
+
+    sql("INSERT INTO %s VALUES (1, 'a')", tableName);
+    Table table = validationCatalog.loadTable(tableIdent);
+    Snapshot snap1 = table.currentSnapshot();
+    long rightAfterSnap1 = waitUntilAfter(snap1.timestampMillis());
+
+    sql("INSERT INTO %s VALUES (2, 'b')", tableName);
+    table.refresh();
+    Snapshot snap2 = table.currentSnapshot();
+    long rightAfterSnap2 = waitUntilAfter(snap2.timestampMillis());
+
+    sql(
+        "CALL %s.system.rollback_to_snapshot('%s', %d)",
+        catalogName, tableIdent, snap1.snapshotId());
+    table.refresh();
+    assertThat(table.currentSnapshot()).isEqualTo(snap1);
+
+    sql("INSERT OVERWRITE %s VALUES (-2, 'a')", tableName);
+    table.refresh();
+    Snapshot snap3 = table.currentSnapshot();
+    long rightAfterSnap3 = waitUntilAfter(snap3.timestampMillis());
+
+    assertEquals(
+        "Should have expected changed rows up to snapshot 3",
+        ImmutableList.of(
+            row(1, "a", "INSERT", 0, snap1.snapshotId()),
+            row(1, "a", "DELETE", 1, snap3.snapshotId()),
+            row(-2, "a", "INSERT", 1, snap3.snapshotId())),
+        changelogRecords(null, rightAfterSnap3));
+
+    assertEquals(
+        "Should have expected changed rows up to snapshot 2",
+        ImmutableList.of(row(1, "a", "INSERT", 0, snap1.snapshotId())),
+        changelogRecords(null, rightAfterSnap2));
+
+    assertEquals(
+        "Should have expected changed rows from snapshot 3 only since snapshot 2 is on a different branch.",
+        ImmutableList.of(
+            row(1, "a", "DELETE", 0, snap3.snapshotId()),
+            row(-2, "a", "INSERT", 0, snap3.snapshotId())),
+        changelogRecords(rightAfterSnap1, snap3.timestampMillis()));
+
+    assertEquals(
+        "Should have expected changed rows from snapshot 3",
+        ImmutableList.of(
+            row(1, "a", "DELETE", 0, snap3.snapshotId()),
+            row(-2, "a", "INSERT", 0, snap3.snapshotId())),
+        changelogRecords(rightAfterSnap2, null));
+
+    sql(
+        "CALL %s.system.set_current_snapshot('%s', %d)",
+        catalogName, tableIdent, snap2.snapshotId());
+    table.refresh();
+    assertThat(table.currentSnapshot()).isEqualTo(snap2);
+    assertEquals(
+        "Should have expected changed rows from snapshot 2 only since snapshot 3 is on a different branch.",
+        ImmutableList.of(row(2, "b", "INSERT", 0, snap2.snapshotId())),
+        changelogRecords(rightAfterSnap1, null));
+  }
+
   private void createTableWithDefaultRows() {
     createTable();
     insertDefaultRows();