(incubator-xtable) branch main updated: [928] Skip incremental sync when Delta source has no new commits (#930)

[vinish]

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

vinishjail97 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-xtable.git


The following commit(s) were added to refs/heads/main by this push:
     new 692c8756 [928] Skip incremental sync when Delta source has no new 
commits (#930)
692c8756 is described below

commit 692c8756c6de94557510853b949f0043c402ac8d
Author: Arvin <31638790+zhang-ar...@users.noreply.github.com>
AuthorDate: Fri Sep 11 05:43:48 2026 +0800

    [928] Skip incremental sync when Delta source has no new commits (#930)
    
    * [Bug] Skip incremental sync when delta source has no new commits
    
    Fixes #928: getCommitsBacklog computed start = lastSyncVersion + 1 with no
    start <= end guard, so a re-sync with no new commits passed an inverted
    version range [n+1, n] to Delta Kernel's getChanges, which failed with
    "Invalid version range". With no new commits the sync is now a successful
    no-op, matching the Hudi and Parquet sources.
    
    * Add Delta Kernel empty backlog test
    
    ---------
    
    Co-authored-by: Vinish Reddy <vinishreddygunne...@gmail.com>
---
 .../xtable/kernel/DeltaKernelConversionSource.java | 10 +++++++++
 .../kernel/ITDeltaKernelConversionSource.java      | 25 ++++++++++++++++++++++
 2 files changed, 35 insertions(+)

diff --git 
a/xtable-core/src/main/java/org/apache/xtable/kernel/DeltaKernelConversionSource.java
 
b/xtable-core/src/main/java/org/apache/xtable/kernel/DeltaKernelConversionSource.java
index 5c1b4cee..14241f92 100644
--- 
a/xtable-core/src/main/java/org/apache/xtable/kernel/DeltaKernelConversionSource.java
+++ 
b/xtable-core/src/main/java/org/apache/xtable/kernel/DeltaKernelConversionSource.java
@@ -180,6 +180,16 @@ public class DeltaKernelConversionSource implements 
ConversionSource<Long> {
             engine, 
Timestamp.from(instantsForIncrementalSync.getLastSyncInstant()).getTime());
 
     long versionNumberAtLastSyncInstant = snapshot.getVersion();
+    long latestVersion = table.getLatestSnapshot(engine).getVersion();
+    if (versionNumberAtLastSyncInstant >= latestVersion) {
+      log.info(
+          "No new delta commits to sync: last sync instant {} corresponds to 
version {} and the "
+              + "latest version is {}. Nothing to sync.",
+          instantsForIncrementalSync.getLastSyncInstant(),
+          versionNumberAtLastSyncInstant,
+          latestVersion);
+      return 
CommitsBacklog.<Long>builder().commitsToProcess(Collections.emptyList()).build();
+    }
     resetState(versionNumberAtLastSyncInstant + 1, engine, table);
     return CommitsBacklog.<Long>builder()
         .commitsToProcess(getChangesState().getVersionsInSortedOrder())
diff --git 
a/xtable-core/src/test/java/org/apache/xtable/kernel/ITDeltaKernelConversionSource.java
 
b/xtable-core/src/test/java/org/apache/xtable/kernel/ITDeltaKernelConversionSource.java
index 82dc2d6d..f5a83d65 100644
--- 
a/xtable-core/src/test/java/org/apache/xtable/kernel/ITDeltaKernelConversionSource.java
+++ 
b/xtable-core/src/test/java/org/apache/xtable/kernel/ITDeltaKernelConversionSource.java
@@ -412,6 +412,31 @@ public class ITDeltaKernelConversionSource {
     ValidationTestHelper.validateTableChanges(allActiveFiles, allTableChanges);
   }
 
+  @Test
+  void testGetCommitsBacklogWithNoNewCommits() {
+    String tableName = GenericTable.getTableName();
+    TestSparkDeltaTable testSparkDeltaTable =
+        new TestSparkDeltaTable(tableName, tempDir, sparkSession, null, false);
+    testSparkDeltaTable.insertRows(50);
+
+    SourceTable tableConfig =
+        SourceTable.builder()
+            .name(testSparkDeltaTable.getTableName())
+            .basePath(testSparkDeltaTable.getBasePath())
+            .formatName(TableFormat.DELTA)
+            .build();
+    DeltaKernelConversionSource conversionSource =
+        conversionSourceProvider.getConversionSourceInstance(tableConfig);
+
+    CommitsBacklog<Long> commitsBacklog =
+        conversionSource.getCommitsBacklog(
+            InstantsForIncrementalSync.builder()
+                
.lastSyncInstant(Instant.ofEpochMilli(testSparkDeltaTable.getLastCommitTimestamp()))
+                .build());
+
+    assertTrue(commitsBacklog.getCommitsToProcess().isEmpty());
+  }
+
   @Test
   public void testsShowingVacuumHasNoEffectOnIncrementalSync() {
     boolean isPartitioned = true;