(iceberg) branch main updated: Spark: Fix changelog table bug for start time older than current snapshot (#11564)

[yu...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

yufei pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git


The following commit(s) were added to refs/heads/main by this push:
     new c448a4b870 Spark: Fix changelog table bug for start time older than current snapshot (#11564)
c448a4b870 is described below

commit c448a4b87007ae8aad57456e19d32db45de3a4f6
Author: Ace Haidrey <ac...@gmail.com>
AuthorDate: Wed Nov 20 21:15:44 2024 -0500

    Spark: Fix changelog table bug for start time older than current snapshot (#11564)
---
 .../spark/extensions/TestChangelogTable.java       | 48 ++++++++++++++++++++++
 .../iceberg/spark/source/SparkScanBuilder.java     |  4 ++
 2 files changed, 52 insertions(+)

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 a7ed065cae..d4930250cb 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
@@ -408,4 +408,52 @@ public class TestChangelogTable extends ExtensionsTestBase {
         .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");
+  }
 }
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 d511fefd8a..33a4e032cc 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
@@ -561,6 +561,10 @@ public class SparkScanBuilder
 
     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;