You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@xtable.apache.org by vi...@apache.org on 2026/01/12 16:09:28 UTC
(incubator-xtable) branch fallbackForLogTruncation created (now bbfa13eb)
This is an automated email from the ASF dual-hosted git repository.
vinish pushed a change to branch fallbackForLogTruncation
in repository https://gitbox.apache.org/repos/asf/incubator-xtable.git
at bbfa13eb Add fallback for log truncation issue in Delta source
This branch includes the following new commits:
new bbfa13eb Add fallback for log truncation issue in Delta source
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(incubator-xtable) 01/01: Add fallback for log truncation issue in Delta source
Posted by vi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
vinish pushed a commit to branch fallbackForLogTruncation
in repository https://gitbox.apache.org/repos/asf/incubator-xtable.git
commit bbfa13eb61accf9ad6b3aebd5d8b89a0a5053f7f
Author: Vinish Reddy <vi...@apache.org>
AuthorDate: Mon Jan 12 08:09:19 2026 -0800
Add fallback for log truncation issue in Delta source
---
.../apache/xtable/delta/DeltaConversionSource.java | 60 ++++++++++++--
.../xtable/kernel/DeltaKernelConversionSource.java | 43 +++++++++-
.../xtable/delta/ITDeltaConversionSource.java | 95 +++++++++++++++++++++-
3 files changed, 186 insertions(+), 12 deletions(-)
diff --git a/xtable-core/src/main/java/org/apache/xtable/delta/DeltaConversionSource.java b/xtable-core/src/main/java/org/apache/xtable/delta/DeltaConversionSource.java
index bb40a315..3cc7f091 100644
--- a/xtable-core/src/main/java/org/apache/xtable/delta/DeltaConversionSource.java
+++ b/xtable-core/src/main/java/org/apache/xtable/delta/DeltaConversionSource.java
@@ -41,6 +41,8 @@ import org.apache.spark.sql.delta.actions.AddFile;
import org.apache.spark.sql.delta.actions.RemoveFile;
import scala.Option;
+import scala.Tuple2;
+import scala.collection.Seq;
import io.delta.tables.DeltaTable;
@@ -191,18 +193,58 @@ public class DeltaConversionSource implements ConversionSource<Long> {
}
/*
- * In Delta Lake, each commit is a self-describing one i.e. it contains list of new files while
- * also containing list of files that were deleted. So, vacuum has no special effect on the
- * incremental sync. Hence, existence of commit is the only check required.
+ * Following checks are performed:
+ * 1. Check if a commit exists at or before the provided instant.
+ * 2. Verify that commit files needed for incremental sync are still accessible.
+ *
+ * Delta Lake's VACUUM operation removes old JSON commit files from _delta_log/, which can
+ * break incremental sync even though commits are self-describing. This method attempts to
+ * access the commit chain to ensure files haven't been vacuumed.
*/
@Override
public boolean isIncrementalSyncSafeFrom(Instant instant) {
- DeltaHistoryManager.Commit deltaCommitAtOrBeforeInstant =
- deltaLog.history().getActiveCommitAtTime(Timestamp.from(instant), true, false, true);
- // There is a chance earliest commit of the table is returned if the instant is before the
- // earliest commit of the table, hence the additional check.
- Instant deltaCommitInstant = Instant.ofEpochMilli(deltaCommitAtOrBeforeInstant.getTimestamp());
- return deltaCommitInstant.equals(instant) || deltaCommitInstant.isBefore(instant);
+ try {
+ DeltaHistoryManager.Commit deltaCommitAtOrBeforeInstant =
+ deltaLog.history().getActiveCommitAtTime(Timestamp.from(instant), true, false, true);
+
+ // There is a chance earliest commit of the table is returned if the instant is before the
+ // earliest commit of the table, hence the additional check.
+ Instant deltaCommitInstant = Instant.ofEpochMilli(deltaCommitAtOrBeforeInstant.getTimestamp());
+ if (deltaCommitInstant.isAfter(instant)) {
+ log.info(
+ "No commit found at or before instant {}. Earliest available commit is at {}",
+ instant,
+ deltaCommitInstant);
+ return false;
+ }
+
+ long versionAtInstant = deltaCommitAtOrBeforeInstant.version();
+
+ // Verify that we can actually access commit files from this version onward by attempting
+ // to read the changes. This will fail if VACUUM has removed the necessary commit files.
+ // We only need to verify we can start iterating - we don't need to consume all changes.
+ scala.collection.Iterator<Tuple2<Object, Seq<Action>>> changesIterator =
+ deltaLog.getChanges(versionAtInstant, true);
+
+ // Test if we can access at least the first commit. If commit files are missing due to
+ // VACUUM, this will throw an exception (typically FileNotFoundException or similar).
+ if (changesIterator.hasNext()) {
+ // Successfully verified we can access commit files
+ return true;
+ } else {
+ // No changes available from this version (shouldn't happen for valid commits)
+ log.warn(
+ "No changes available starting from version {} (instant: {})", versionAtInstant, instant);
+ return false;
+ }
+ } catch (Exception e) {
+ // Commit files have been vacuumed or are otherwise inaccessible
+ log.info(
+ "Cannot perform incremental sync from instant {} due to missing or inaccessible commit files: {}",
+ instant,
+ e.getMessage());
+ return false;
+ }
}
@Override
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 fa088f08..631c69e5 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
@@ -24,21 +24,27 @@ import java.time.Instant;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
+import java.util.Set;
import lombok.Builder;
import lombok.extern.slf4j.Slf4j;
import io.delta.kernel.Snapshot;
import io.delta.kernel.Table;
+import io.delta.kernel.data.ColumnarBatch;
import io.delta.kernel.engine.Engine;
+import io.delta.kernel.internal.DeltaLogActionUtils;
import io.delta.kernel.internal.SnapshotImpl;
+import io.delta.kernel.internal.TableImpl;
import io.delta.kernel.internal.actions.AddFile;
import io.delta.kernel.internal.actions.RemoveFile;
import io.delta.kernel.internal.actions.RowBackedAction;
import io.delta.kernel.internal.util.VectorUtils;
+import io.delta.kernel.utils.CloseableIterator;
import org.apache.xtable.exception.ReadException;
import org.apache.xtable.model.CommitsBacklog;
@@ -192,9 +198,42 @@ public class DeltaKernelConversionSource implements ConversionSource<Long> {
// There is a chance earliest commit of the table is returned if the instant is before the
// earliest commit of the table, hence the additional check.
Instant deltaCommitInstant = Instant.ofEpochMilli(snapshot.getTimestamp(engine));
- return deltaCommitInstant.equals(instant) || deltaCommitInstant.isBefore(instant);
+ if (deltaCommitInstant.isAfter(instant)) {
+ log.info(
+ "No commit found at or before instant {}. Earliest available commit is at {}",
+ instant,
+ deltaCommitInstant);
+ return false;
+ }
+
+ long versionAtInstant = snapshot.getVersion();
+ long currentVersion = table.getLatestSnapshot(engine).getVersion();
+
+ // Verify that we can actually access commit files from this version to current version.
+ // This will fail if VACUUM has removed the necessary commit files.
+ Set<DeltaLogActionUtils.DeltaAction> actionSet = new HashSet<>();
+ actionSet.add(DeltaLogActionUtils.DeltaAction.ADD);
+ actionSet.add(DeltaLogActionUtils.DeltaAction.REMOVE);
+
+ TableImpl tableImpl = (TableImpl) table;
+ // Attempt to get changes - this will throw if commit files are missing
+ try (CloseableIterator<ColumnarBatch> changesIterator =
+ tableImpl.getChanges(engine, versionAtInstant, currentVersion, actionSet)) {
+ // Test if we can access at least the first batch
+ if (changesIterator.hasNext()) {
+ // Successfully verified we can access commit files
+ return true;
+ } else {
+ // No changes available (edge case: versionAtInstant == currentVersion)
+ return true;
+ }
+ }
} catch (Exception e) {
- log.error("Error checking if incremental sync is safe from " + instant, e);
+ // Commit files have been vacuumed or are otherwise inaccessible
+ log.info(
+ "Cannot perform incremental sync from instant {} due to missing or inaccessible commit files: {}",
+ instant,
+ e.getMessage());
return false;
}
}
diff --git a/xtable-core/src/test/java/org/apache/xtable/delta/ITDeltaConversionSource.java b/xtable-core/src/test/java/org/apache/xtable/delta/ITDeltaConversionSource.java
index 3a754e27..fc855a59 100644
--- a/xtable-core/src/test/java/org/apache/xtable/delta/ITDeltaConversionSource.java
+++ b/xtable-core/src/test/java/org/apache/xtable/delta/ITDeltaConversionSource.java
@@ -21,8 +21,10 @@ package org.apache.xtable.delta;
import static org.apache.xtable.testutil.ITTestUtils.validateTable;
import static org.junit.jupiter.api.Assertions.*;
+import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
+import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.time.Instant;
@@ -422,7 +424,7 @@ public class ITDeltaConversionSource {
}
@Test
- public void testsShowingVacuumHasNoEffectOnIncrementalSync() {
+ public void testVacuumAffectsIncrementalSyncSafety() {
boolean isPartitioned = true;
String tableName = GenericTable.getTableName();
TestSparkDeltaTable testSparkDeltaTable =
@@ -472,6 +474,97 @@ public class ITDeltaConversionSource {
assertFalse(conversionSource.isIncrementalSyncSafeFrom(instantAsOfHourAgo));
}
+ @Test
+ public void testIncrementalSyncSafeWhenCommitFilesExist() {
+ String tableName = GenericTable.getTableName();
+ TestSparkDeltaTable testSparkDeltaTable =
+ new TestSparkDeltaTable(tableName, tempDir, sparkSession, null, false);
+
+ // Insert initial data
+ testSparkDeltaTable.insertRows(50);
+ Long timestamp1 = testSparkDeltaTable.getLastCommitTimestamp();
+
+ // Add more commits
+ testSparkDeltaTable.insertRows(50);
+
+ SourceTable tableConfig =
+ SourceTable.builder()
+ .name(tableName)
+ .basePath(testSparkDeltaTable.getBasePath())
+ .formatName(TableFormat.DELTA)
+ .build();
+ DeltaConversionSource conversionSource =
+ conversionSourceProvider.getConversionSourceInstance(tableConfig);
+
+ // Should be safe - commit files still exist
+ assertTrue(
+ conversionSource.isIncrementalSyncSafeFrom(Instant.ofEpochMilli(timestamp1)),
+ "Incremental sync should be safe when commit files exist");
+ }
+
+ @Test
+ public void testIncrementalSyncUnsafeForInstantBeforeEarliestCommit() {
+ String tableName = GenericTable.getTableName();
+ TestSparkDeltaTable testSparkDeltaTable =
+ new TestSparkDeltaTable(tableName, tempDir, sparkSession, null, false);
+
+ // Insert data
+ testSparkDeltaTable.insertRows(50);
+
+ SourceTable tableConfig =
+ SourceTable.builder()
+ .name(tableName)
+ .basePath(testSparkDeltaTable.getBasePath())
+ .formatName(TableFormat.DELTA)
+ .build();
+ DeltaConversionSource conversionSource =
+ conversionSourceProvider.getConversionSourceInstance(tableConfig);
+
+ // Try instant from 1 hour ago (before table existed)
+ Instant instantHourAgo = Instant.now().minus(1, ChronoUnit.HOURS);
+ assertFalse(
+ conversionSource.isIncrementalSyncSafeFrom(instantHourAgo),
+ "Incremental sync should be unsafe for instant before earliest commit");
+ }
+
+ @Test
+ public void testIncrementalSyncUnsafeAfterManualCommitFileDeletion() throws IOException {
+ String tableName = GenericTable.getTableName();
+ TestSparkDeltaTable testSparkDeltaTable =
+ new TestSparkDeltaTable(tableName, tempDir, sparkSession, null, false);
+
+ testSparkDeltaTable.insertRows(50);
+ Long timestamp1 = testSparkDeltaTable.getLastCommitTimestamp();
+ testSparkDeltaTable.insertRows(50);
+ testSparkDeltaTable.insertRows(50);
+
+ SourceTable tableConfig =
+ SourceTable.builder()
+ .name(tableName)
+ .basePath(testSparkDeltaTable.getBasePath())
+ .formatName(TableFormat.DELTA)
+ .build();
+
+ DeltaConversionSource conversionSource =
+ conversionSourceProvider.getConversionSourceInstance(tableConfig);
+
+ assertTrue(
+ conversionSource.isIncrementalSyncSafeFrom(Instant.ofEpochMilli(timestamp1)),
+ "Should be safe before deleting commit files");
+
+ // Delete commit file to simulate VACUUM or filesystem corruption
+ Path commitFile = Paths.get(testSparkDeltaTable.getBasePath(), "_delta_log", "00000000000000000000.json");
+ if (Files.exists(commitFile)) {
+ Files.delete(commitFile);
+ }
+
+ conversionSource = conversionSourceProvider.getConversionSourceInstance(tableConfig);
+
+ assertFalse(
+ conversionSource.isIncrementalSyncSafeFrom(Instant.ofEpochMilli(timestamp1)),
+ "Should be unsafe after manually deleting commit files");
+ }
+
@ParameterizedTest
@MethodSource("testWithPartitionToggle")
public void testVacuum(boolean isPartitioned) {