Spark: Fix changelog table bug for start time older than current snapshot (#11564) (#11613)

Co-authored-by: Ace Haidrey <[email protected]>

Author: bryanck

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

@@ -408,4 +408,52 @@ private List<Row> collect(DataFrameReader reader) {
         .orderBy("_change_ordinal", "_commit_snapshot_id", "_change_type", "id")
         .collectAsList();
   }
+
+  @TestTemplate
+  public void testChangelogViewOutsideTimeRange() {
+    createTableWithDefaultRows();
+
+    // Insert new records
+    sql("INSERT INTO %s VALUES (3, 'c')", tableName);
+    sql("INSERT INTO %s VALUES (4, 'd')", tableName);
+
+    Table table = validationCatalog.loadTable(tableIdent);
+    Snapshot insertSnapshot = table.currentSnapshot();
+
+    // Get timestamp after inserts but before our changelog window
+    long beforeWindowTime = System.currentTimeMillis();
+
+    // Small delay to ensure our timestamps are different
+    try {
+      Thread.sleep(100);
+    } catch (InterruptedException e) {
+      throw new RuntimeException("Test interrupted", e);
+    }
+
+    long startTime = System.currentTimeMillis();
+    long endTime = startTime + 1000; // 1 second window
+
+    // Create changelog view for a time window after our inserts
+    sql(
+        "CALL %s.system.create_changelog_view("
+            + "  table => '%s', "
+            + "  options => map("
+            + "    'start-timestamp', '%d',"
+            + "    'end-timestamp', '%d'"
+            + "  ),"
+            + "  changelog_view => 'test_changelog_view'"
+            + ")",
+        catalogName, tableName, startTime, endTime);
+
+    // Query the changelog view
+    List<Object[]> results =
+        sql(
+            "SELECT * FROM test_changelog_view WHERE _change_type IN ('INSERT', 'DELETE') ORDER BY _change_ordinal");
+
+    // Verify no changes are returned since our window is after the inserts
+    assertThat(results).as("Num records must be zero").isEmpty();
+
+    // Clean up the changelog view
+    sql("DROP VIEW IF EXISTS test_changelog_view");
+  }
 }

spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkScanBuilder.java

@@ -561,6 +561,10 @@ public Scan buildChangelogScan() {
 
     boolean emptyScan = false;
     if (startTimestamp != null) {
+      if (table.currentSnapshot() != null
+          && table.currentSnapshot().timestampMillis() < startTimestamp) {
+        emptyScan = true;
+      }
       startSnapshotId = getStartSnapshotId(startTimestamp);
       if (startSnapshotId == null && endTimestamp == null) {
         emptyScan = true;