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 cab12de7 [860] Reuse the Delta snapshot across an incremental backlog
(#861)
cab12de7 is described below
commit cab12de7ad24d2b3d46045449f41e361d090264f
Author: AadhiKat <[email protected]>
AuthorDate: Wed Aug 12 03:03:30 2026 +0530
[860] Reuse the Delta snapshot across an incremental backlog (#861)
* Reuse the Delta snapshot across an incremental backlog
getTableChangeForCommit reconstructed the table snapshot (getSnapshotAt) for
every commit, re-reading the Delta checkpoint each time. That is cheap on a
small checkpoint but dominates incremental catch-up on tables with a large
checkpoint.
Cache the snapshot (and the table/file format derived from it) and reload it
only when a commit carries a Metadata or Protocol action. Otherwise the
snapshot is used only for schema, partitioning, file format and base path,
which do not change between commits.
latestCommitTime feeds the sync watermark, so it can't come from the reused
snapshot or the watermark stops advancing. Take it from the commit's own
CommitInfo (already in the incremental actions), falling back to the
snapshot
timestamp only if a commit has no CommitInfo.
Signed-off-by: Aadhithya Hari <[email protected]>
* Cache table metadata across an incremental backlog instead of the snapshot
Addresses review feedback on #861:
- Cache Metadata rather than Snapshot: the baseline comes from one
snapshot read (the snapshot is not retained), and a commit carrying a
metaData action refreshes the cache from the action itself, last one
in the commit wins, mirroring InMemoryLogReplay. Protocol no longer
forces a reload since nothing reads snapshot.protocol(). The actions
converter gains base-path variants (the Snapshot overloads delegate),
matching the kernel converter's shape.
- Take latestCommitTime from the commit file's modification time,
carried by DeltaIncrementalChangesState from getChangeLogFiles.
CommitInfo's timestamp is the writer's clock, and getCommitsBacklog
resolves the persisted watermark against mtimes, so a fast writer
clock could push the watermark past commits that were never
processed.
- Gate the reuse behind xtable.delta.source.reuse_metadata_across_commits,
read from the source table's additionalProperties, default true.
Setting it to false restores per-commit snapshot reconstruction.
- Add a mid-backlog schema evolution test on a single source instance
(append, ALTER TABLE ADD COLUMNS, append) asserting later commits
carry the new schema while getSnapshotAt still runs exactly once, and
a flag-off test pinning one reconstruction per commit.
Signed-off-by: Aadhithya Hari <[email protected]>
* Address review feedback on the Delta metadata reuse
- Default xtable.delta.source.reuse_metadata_across_commits to false
- Move the config into DeltaConversionSourceConfig, mirroring
DeltaConversionTargetConfig, so the constant is public and the parsing
is unit testable
- Clear the metadata cache in resetState and pin non-decreasing version
order, so the cache is scoped to a backlog rather than to the instance
- Gate the commit-timestamp listing on the flag, so turning it off
reverts the snapshot reuse, the watermark source and the extra listing
- Delegate the Snapshot overload of DeltaTableExtractor.table to the
Metadata overload instead of duplicating the builder block
- Drop the duplicate deletion-vector null check
- Correct the stale watermark rationale in the two-run test
- Add TestDeltaConversionSourceConfig covering the property parsing
* Extract the metadata reuse block into resolveMetadataForCommit
---------
Signed-off-by: Aadhithya Hari <[email protected]>
Co-authored-by: Vinish Reddy <[email protected]>
---
.../apache/xtable/delta/DeltaActionsConverter.java | 49 ++++-
.../apache/xtable/delta/DeltaConversionSource.java | 81 ++++++-
.../xtable/delta/DeltaConversionSourceConfig.java | 54 +++++
.../delta/DeltaConversionSourceProvider.java | 3 +
.../xtable/delta/DeltaIncrementalChangesState.java | 37 +++-
.../apache/xtable/delta/DeltaTableExtractor.java | 36 ++-
.../xtable/delta/ITDeltaConversionSource.java | 244 +++++++++++++++++++++
.../xtable/delta/TestDeltaActionsConverter.java | 6 +-
.../delta/TestDeltaConversionSourceConfig.java | 62 ++++++
9 files changed, 552 insertions(+), 20 deletions(-)
diff --git
a/xtable-core/src/main/java/org/apache/xtable/delta/DeltaActionsConverter.java
b/xtable-core/src/main/java/org/apache/xtable/delta/DeltaActionsConverter.java
index 16a320f1..c7d27488 100644
---
a/xtable-core/src/main/java/org/apache/xtable/delta/DeltaActionsConverter.java
+++
b/xtable-core/src/main/java/org/apache/xtable/delta/DeltaActionsConverter.java
@@ -57,12 +57,32 @@ public class DeltaActionsConverter {
boolean includeColumnStats,
DeltaPartitionExtractor partitionExtractor,
DeltaStatsExtractor fileStatsExtractor) {
+ return convertAddActionToInternalDataFile(
+ addFile,
+ tableBasePath(deltaSnapshot),
+ fileFormat,
+ partitionFields,
+ fields,
+ includeColumnStats,
+ partitionExtractor,
+ fileStatsExtractor);
+ }
+
+ public InternalDataFile convertAddActionToInternalDataFile(
+ AddFile addFile,
+ String tableBasePath,
+ FileFormat fileFormat,
+ List<InternalPartitionField> partitionFields,
+ List<InternalField> fields,
+ boolean includeColumnStats,
+ DeltaPartitionExtractor partitionExtractor,
+ DeltaStatsExtractor fileStatsExtractor) {
FileStats fileStats = fileStatsExtractor.getColumnStatsForFile(addFile,
fields);
List<ColumnStat> columnStats =
includeColumnStats ? fileStats.getColumnStats() :
Collections.emptyList();
long recordCount = fileStats.getNumRecords();
return InternalDataFile.builder()
- .physicalPath(getFullPathToFile(deltaSnapshot, addFile.path()))
+ .physicalPath(getFullPathToFile(tableBasePath, addFile.path()))
.fileFormat(fileFormat)
.fileSizeBytes(addFile.size())
.lastModified(addFile.modificationTime())
@@ -79,8 +99,18 @@ public class DeltaActionsConverter {
FileFormat fileFormat,
List<InternalPartitionField> partitionFields,
DeltaPartitionExtractor partitionExtractor) {
+ return convertRemoveActionToInternalDataFile(
+ removeFile, tableBasePath(deltaSnapshot), fileFormat, partitionFields,
partitionExtractor);
+ }
+
+ public InternalDataFile convertRemoveActionToInternalDataFile(
+ RemoveFile removeFile,
+ String tableBasePath,
+ FileFormat fileFormat,
+ List<InternalPartitionField> partitionFields,
+ DeltaPartitionExtractor partitionExtractor) {
return InternalDataFile.builder()
- .physicalPath(getFullPathToFile(deltaSnapshot, removeFile.path()))
+ .physicalPath(getFullPathToFile(tableBasePath, removeFile.path()))
.fileFormat(fileFormat)
.partitionValues(
partitionExtractor.partitionValueExtraction(
@@ -99,13 +129,20 @@ public class DeltaActionsConverter {
}
static String getFullPathToFile(Snapshot snapshot, String dataFilePath) {
- String tableBasePath = snapshot.deltaLog().dataPath().toUri().toString();
+ return getFullPathToFile(tableBasePath(snapshot), dataFilePath);
+ }
+
+ static String getFullPathToFile(String tableBasePath, String dataFilePath) {
if (dataFilePath.startsWith(tableBasePath)) {
return dataFilePath;
}
return tableBasePath + Path.SEPARATOR + dataFilePath;
}
+ private static String tableBasePath(Snapshot snapshot) {
+ return snapshot.deltaLog().dataPath().toUri().toString();
+ }
+
/**
* Extracts the representation of the deletion vector information
corresponding to an AddFile
* action. Currently, this method extracts and returns the path to the data
file for which a
@@ -117,12 +154,16 @@ public class DeltaActionsConverter {
* is present
*/
public String extractDeletionVectorFile(Snapshot snapshot, AddFile addFile) {
+ return extractDeletionVectorFile(tableBasePath(snapshot), addFile);
+ }
+
+ public String extractDeletionVectorFile(String tableBasePath, AddFile
addFile) {
DeletionVectorDescriptor deletionVector = addFile.deletionVector();
if (deletionVector == null) {
return null;
}
String dataFilePath = addFile.path();
- return getFullPathToFile(snapshot, dataFilePath);
+ return getFullPathToFile(tableBasePath, dataFilePath);
}
}
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 7fc8ea7b..fc1d17c6 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
@@ -38,10 +38,13 @@ import org.apache.spark.sql.delta.DeltaLog;
import org.apache.spark.sql.delta.Snapshot;
import org.apache.spark.sql.delta.actions.Action;
import org.apache.spark.sql.delta.actions.AddFile;
+import org.apache.spark.sql.delta.actions.Metadata;
import org.apache.spark.sql.delta.actions.RemoveFile;
import scala.Option;
+import com.google.common.base.Preconditions;
+
import io.delta.tables.DeltaTable;
import org.apache.xtable.exception.ReadException;
@@ -78,6 +81,13 @@ public class DeltaConversionSource implements
ConversionSource<Long> {
private final String tableName;
private final String basePath;
+ // See DeltaConversionSourceConfig#REUSE_METADATA_ACROSS_COMMITS.
+ @Builder.Default private final boolean reuseMetadataAcrossCommits = false;
+
+ // Scoped to a single backlog; cleared by resetState.
+ private Metadata cachedMetadata;
+ private Long lastProcessedVersion;
+
@Override
public InternalTable getTable(Long version) {
return tableExtractor.table(deltaLog, tableName, version);
@@ -103,10 +113,27 @@ public class DeltaConversionSource implements
ConversionSource<Long> {
@Override
public TableChange getTableChangeForCommit(Long versionNumber) {
List<Action> actionsForVersion =
getChangesState().getActionsForVersion(versionNumber);
- Snapshot snapshotAtVersion = deltaLog.getSnapshotAt(versionNumber,
Option.empty());
- InternalTable tableAtVersion = tableExtractor.table(snapshotAtVersion,
tableName);
- FileFormat fileFormat =
-
actionsConverter.convertToFileFormat(snapshotAtVersion.metadata().format().provider());
+ String tableBasePath = deltaLog.dataPath().toUri().toString();
+ InternalTable tableAtVersion;
+ FileFormat fileFormat;
+ if (reuseMetadataAcrossCommits) {
+ Metadata metadataAtVersion = resolveMetadataForCommit(versionNumber,
actionsForVersion);
+ // latestCommitTime is persisted as the sync watermark and matched
against commit-file mtimes
+ // on the next run, so it must come from that clock.
Snapshot.timestamp() is the same mtime;
+ // the changes state carries it without the read.
+ tableAtVersion =
+ tableExtractor.table(
+ metadataAtVersion,
+ deltaLog,
+ tableName,
+ getChangesState().getCommitTimestamp(versionNumber));
+ fileFormat =
actionsConverter.convertToFileFormat(metadataAtVersion.format().provider());
+ } else {
+ Snapshot snapshotAtVersion = deltaLog.getSnapshotAt(versionNumber,
Option.empty());
+ tableAtVersion = tableExtractor.table(snapshotAtVersion, tableName);
+ fileFormat =
+
actionsConverter.convertToFileFormat(snapshotAtVersion.metadata().format().provider());
+ }
// All 3 of the following data structures use data file's absolute path as
the key
Map<String, InternalDataFile> addedFiles = new HashMap<>();
@@ -119,7 +146,7 @@ public class DeltaConversionSource implements
ConversionSource<Long> {
InternalDataFile dataFile =
actionsConverter.convertAddActionToInternalDataFile(
(AddFile) action,
- snapshotAtVersion,
+ tableBasePath,
fileFormat,
tableAtVersion.getPartitioningFields(),
tableAtVersion.getReadSchema().getAllFields(),
@@ -128,7 +155,7 @@ public class DeltaConversionSource implements
ConversionSource<Long> {
DeltaStatsExtractor.getInstance());
addedFiles.put(dataFile.getPhysicalPath(), dataFile);
String deleteVectorPath =
- actionsConverter.extractDeletionVectorFile(snapshotAtVersion,
(AddFile) action);
+ actionsConverter.extractDeletionVectorFile(tableBasePath,
(AddFile) action);
if (deleteVectorPath != null) {
deletionVectors.add(deleteVectorPath);
}
@@ -136,7 +163,7 @@ public class DeltaConversionSource implements
ConversionSource<Long> {
InternalDataFile dataFile =
actionsConverter.convertRemoveActionToInternalDataFile(
(RemoveFile) action,
- snapshotAtVersion,
+ tableBasePath,
fileFormat,
tableAtVersion.getPartitioningFields(),
DeltaPartitionExtractor.getInstance());
@@ -208,17 +235,57 @@ public class DeltaConversionSource implements
ConversionSource<Long> {
return String.valueOf(commit);
}
+ /**
+ * Returns the table metadata as of the given commit, reading a snapshot
only for the first commit
+ * of a backlog. Everything the conversion needs from the snapshot derives
from the table
+ * metadata, so it is carried across the backlog rather than re-read from
the checkpoint per
+ * commit, and refreshed from the commit's own metadata action when the
schema evolves.
+ */
+ private Metadata resolveMetadataForCommit(Long versionNumber, List<Action>
actionsForVersion) {
+ // Walking backwards would convert a commit with metadata the table did
not have yet, and
+ // nothing would correct it, so pin the ordering the SPI does not state.
+ Preconditions.checkArgument(
+ lastProcessedVersion == null || versionNumber >= lastProcessedVersion,
+ String.format(
+ "Version %s is before the last processed version %s. A backlog
must be walked in "
+ + "non-decreasing version order when %s is enabled.",
+ versionNumber,
+ lastProcessedVersion,
+ DeltaConversionSourceConfig.REUSE_METADATA_ACROSS_COMMITS));
+ lastProcessedVersion = versionNumber;
+ Metadata metadataInCommit = null;
+ for (Action action : actionsForVersion) {
+ if (action instanceof Metadata) {
+ // last one in a commit wins, mirroring InMemoryLogReplay
+ metadataInCommit = (Metadata) action;
+ }
+ }
+ if (cachedMetadata == null) {
+ // baseline; the snapshot is read for its metadata and not retained
+ cachedMetadata = deltaLog.getSnapshotAt(versionNumber,
Option.empty()).metadata();
+ } else if (metadataInCommit != null) {
+ cachedMetadata = metadataInCommit;
+ }
+ return cachedMetadata;
+ }
+
private DeltaIncrementalChangesState getChangesState() {
return deltaIncrementalChangesState.orElseThrow(
() -> new IllegalStateException("DeltaIncrementalChangesState is not
initialized"));
}
private void resetState(long versionToStartFrom) {
+ // The cache is scoped to one backlog, not to the source instance: an
embedder holding a source
+ // across syncs can start a later backlog at an earlier version, which
would otherwise convert
+ // those commits with newer metadata. Clearing costs one snapshot read per
backlog.
+ cachedMetadata = null;
+ lastProcessedVersion = null;
deltaIncrementalChangesState =
Optional.of(
DeltaIncrementalChangesState.builder()
.deltaLog(deltaLog)
.versionToStartFrom(versionToStartFrom)
+ .loadCommitTimestamps(reuseMetadataAcrossCommits)
.build());
}
diff --git
a/xtable-core/src/main/java/org/apache/xtable/delta/DeltaConversionSourceConfig.java
b/xtable-core/src/main/java/org/apache/xtable/delta/DeltaConversionSourceConfig.java
new file mode 100644
index 00000000..5f0e735a
--- /dev/null
+++
b/xtable-core/src/main/java/org/apache/xtable/delta/DeltaConversionSourceConfig.java
@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.xtable.delta;
+
+import java.util.Properties;
+
+import lombok.Value;
+
+/**
+ * Configuration of the Delta source format for the sync process.
+ *
+ * <p>Read from the source table's additional properties, mirroring {@link
+ * DeltaConversionTargetConfig} on the target side.
+ */
+@Value
+public class DeltaConversionSourceConfig {
+ /**
+ * When {@code true}, an incremental backlog carries the table metadata
across commits instead of
+ * reconstructing the Delta snapshot for every commit. Defaults to {@code
false}.
+ *
+ * <p>Enabling it also moves {@code latestCommitTime} to the commit file's
modification time
+ * carried by {@link DeltaIncrementalChangesState} (the same value {@code
Snapshot.timestamp()}
+ * returns), at the cost of a second listing of the log per backlog. Setting
it back to {@code
+ * false} reverts all three.
+ */
+ public static final String REUSE_METADATA_ACROSS_COMMITS =
+ "xtable.delta.source.reuse_metadata_across_commits";
+
+ boolean reuseMetadataAcrossCommits;
+
+ public static DeltaConversionSourceConfig fromProperties(Properties
properties) {
+ boolean reuseMetadataAcrossCommits =
+ properties != null
+ && Boolean.parseBoolean(
+ properties.getProperty(REUSE_METADATA_ACROSS_COMMITS,
Boolean.FALSE.toString()));
+ return new DeltaConversionSourceConfig(reuseMetadataAcrossCommits);
+ }
+}
diff --git
a/xtable-core/src/main/java/org/apache/xtable/delta/DeltaConversionSourceProvider.java
b/xtable-core/src/main/java/org/apache/xtable/delta/DeltaConversionSourceProvider.java
index 045e2b72..0048d1a7 100644
---
a/xtable-core/src/main/java/org/apache/xtable/delta/DeltaConversionSourceProvider.java
+++
b/xtable-core/src/main/java/org/apache/xtable/delta/DeltaConversionSourceProvider.java
@@ -31,12 +31,15 @@ public class DeltaConversionSourceProvider extends
ConversionSourceProvider<Long
public DeltaConversionSource getConversionSourceInstance(SourceTable
sourceTable) {
SparkSession sparkSession =
DeltaConversionUtils.buildSparkSession(hadoopConf);
DeltaTable deltaTable = DeltaTable.forPath(sparkSession,
sourceTable.getBasePath());
+ DeltaConversionSourceConfig sourceConfig =
+
DeltaConversionSourceConfig.fromProperties(sourceTable.getAdditionalProperties());
return DeltaConversionSource.builder()
.sparkSession(sparkSession)
.tableName(sourceTable.getName())
.basePath(sourceTable.getBasePath())
.deltaTable(deltaTable)
.deltaLog(deltaTable.deltaLog())
+
.reuseMetadataAcrossCommits(sourceConfig.isReuseMetadataAcrossCommits())
.build();
}
}
diff --git
a/xtable-core/src/main/java/org/apache/xtable/delta/DeltaIncrementalChangesState.java
b/xtable-core/src/main/java/org/apache/xtable/delta/DeltaIncrementalChangesState.java
index 4642e818..30340207 100644
---
a/xtable-core/src/main/java/org/apache/xtable/delta/DeltaIncrementalChangesState.java
+++
b/xtable-core/src/main/java/org/apache/xtable/delta/DeltaIncrementalChangesState.java
@@ -18,6 +18,7 @@
package org.apache.xtable.delta;
+import java.time.Instant;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
@@ -26,6 +27,8 @@ import java.util.Map;
import lombok.Builder;
+import org.apache.hadoop.fs.FileStatus;
+
import org.apache.spark.sql.delta.DeltaLog;
import org.apache.spark.sql.delta.actions.Action;
@@ -41,6 +44,7 @@ public class DeltaIncrementalChangesState {
private final Long endVersion;
private final Map<Long, List<Action>> incrementalChangesByVersion = new
HashMap<>();
+ private final Map<Long, Instant> commitTimestampByVersion = new HashMap<>();
/**
* Reloads the cache store with incremental changes. Intentionally thread
safety is the
@@ -48,9 +52,13 @@ public class DeltaIncrementalChangesState {
*
* @param deltaLog The DeltaLog instance.
* @param versionToStartFrom The version to start from.
+ * @param loadCommitTimestamps whether to also collect each commit file's
modification time for
+ * {@link #getCommitTimestamp(Long)}. Costs a second listing of the log,
so callers that
+ * resolve a snapshot per commit leave it off and read {@code
Snapshot.timestamp()}.
*/
@Builder
- public DeltaIncrementalChangesState(DeltaLog deltaLog, Long
versionToStartFrom) {
+ public DeltaIncrementalChangesState(
+ DeltaLog deltaLog, Long versionToStartFrom, boolean
loadCommitTimestamps) {
List<Tuple2<Long, List<Action>>> changesList =
getChangesList(deltaLog.getChanges(versionToStartFrom, false));
Long maxSeenVersion = null;
@@ -61,6 +69,19 @@ public class DeltaIncrementalChangesState {
maxSeenVersion =
maxSeenVersion == null ? versionNumber : Math.max(maxSeenVersion,
versionNumber);
}
+ if (loadCommitTimestamps) {
+ // The same clock Snapshot.timestamp() reads and getCommitsBacklog
matches the watermark
+ // against, so it holds regardless of the writer's clock. This pass
reads no file contents.
+ Iterator<Tuple2<Object, FileStatus>> logFiles =
+ JavaConverters.asJavaIteratorConverter(
+ deltaLog.getChangeLogFiles(versionToStartFrom, false))
+ .asJava();
+ while (logFiles.hasNext()) {
+ Tuple2<Object, FileStatus> logFile = logFiles.next();
+ commitTimestampByVersion.put(
+ (Long) logFile._1(),
Instant.ofEpochMilli(logFile._2().getModificationTime()));
+ }
+ }
startVersion = versionToStartFrom;
endVersion = maxSeenVersion;
}
@@ -85,6 +106,20 @@ public class DeltaIncrementalChangesState {
return incrementalChangesByVersion.get(version);
}
+ /**
+ * Returns the commit-file modification time of the given version. Only
available when the state
+ * was built with {@code loadCommitTimestamps}.
+ */
+ public Instant getCommitTimestamp(Long version) {
+ Preconditions.checkArgument(
+ commitTimestampByVersion.containsKey(version),
+ String.format(
+ "Version %s not found in the DeltaIncrementalChangesState.
Timestamps are only "
+ + "collected when the state is built with
loadCommitTimestamps.",
+ version));
+ return commitTimestampByVersion.get(version);
+ }
+
private List<Tuple2<Long, List<Action>>> getChangesList(
scala.collection.Iterator<Tuple2<Object, Seq<Action>>> scalaIterator) {
List<Tuple2<Long, List<Action>>> changesList = new ArrayList<>();
diff --git
a/xtable-core/src/main/java/org/apache/xtable/delta/DeltaTableExtractor.java
b/xtable-core/src/main/java/org/apache/xtable/delta/DeltaTableExtractor.java
index 8f1f1d06..a9599652 100644
--- a/xtable-core/src/main/java/org/apache/xtable/delta/DeltaTableExtractor.java
+++ b/xtable-core/src/main/java/org/apache/xtable/delta/DeltaTableExtractor.java
@@ -25,6 +25,7 @@ import lombok.Builder;
import org.apache.spark.sql.delta.DeltaLog;
import org.apache.spark.sql.delta.Snapshot;
+import org.apache.spark.sql.delta.actions.Metadata;
import scala.Option;
@@ -57,10 +58,35 @@ public class DeltaTableExtractor {
*/
public InternalTable table(Snapshot snapshot, String tableName) {
requireMetadata(snapshot, tableName);
- InternalSchema schema =
schemaExtractor.toInternalSchema(snapshot.metadata().schema());
+ return table(
+ snapshot.metadata(),
+ snapshot.deltaLog(),
+ tableName,
+ Instant.ofEpochMilli(snapshot.timestamp()));
+ }
+
+ /**
+ * Builds an {@link InternalTable} from the table's {@link Metadata}
directly, without a resolved
+ * {@link Snapshot}. Everything an InternalTable carries is either constant
for the table (paths)
+ * or derived from the metadata, so callers that already know the metadata
at a version — e.g. an
+ * incremental backlog reusing it across commits — can avoid reconstructing
the snapshot.
+ *
+ * @param metadata the table metadata in effect at the commit
+ * @param deltaLog the delta log, used only for the table and log paths
+ * @param tableName name of the table
+ * @param commitTimestamp the commit-file modification time of the commit,
used as the
+ * incremental-sync watermark
+ * @return the internal representation of the table at the commit
+ */
+ public InternalTable table(
+ Metadata metadata, DeltaLog deltaLog, String tableName, Instant
commitTimestamp) {
+ if (metadata == null || metadata.schema() == null) {
+ throw new IllegalStateException("Missing metadata/schema for table: " +
tableName);
+ }
+ InternalSchema schema =
schemaExtractor.toInternalSchema(metadata.schema());
List<InternalPartitionField> partitionFields =
DeltaPartitionExtractor.getInstance()
- .convertFromDeltaPartitionFormat(schema,
snapshot.metadata().partitionSchema());
+ .convertFromDeltaPartitionFormat(schema,
metadata.partitionSchema());
// Delta follows Hive Style partitioning layout
// (https://delta.io/blog/2023-01-18-add-remove-partition-delta-lake/)
DataLayoutStrategy dataLayoutStrategy =
@@ -69,13 +95,13 @@ public class DeltaTableExtractor {
: DataLayoutStrategy.FLAT;
return InternalTable.builder()
.tableFormat(TableFormat.DELTA)
- .basePath(snapshot.deltaLog().dataPath().toString())
+ .basePath(deltaLog.dataPath().toString())
.name(tableName)
.layoutStrategy(dataLayoutStrategy)
.partitioningFields(partitionFields)
.readSchema(schema)
- .latestCommitTime(Instant.ofEpochMilli(snapshot.timestamp()))
- .latestMetadataPath(snapshot.deltaLog().logPath().toString())
+ .latestCommitTime(commitTimestamp)
+ .latestMetadataPath(deltaLog.logPath().toString())
.build();
}
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 326a98d8..fa44daec 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
@@ -25,6 +25,13 @@ import static
org.apache.xtable.testutil.ITTestUtils.validateTable;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
import java.net.URI;
import java.net.URISyntaxException;
@@ -55,6 +62,8 @@ import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;
+import org.apache.spark.sql.delta.DeltaLog;
+
import io.delta.tables.DeltaTable;
import org.apache.xtable.GenericTable;
@@ -829,4 +838,239 @@ public class ITDeltaConversionSource {
.collect(Collectors.toSet());
return filePathsRemoved.contains(activePath);
}
+
+ @Test
+ public void
getTableChangeForCommitReconstructsSnapshotOnceForAppendOnlyBacklog() {
+ String tableName = GenericTable.getTableName();
+ TestSparkDeltaTable testSparkDeltaTable =
+ new TestSparkDeltaTable(tableName, tempDir, sparkSession, null, false);
+ List<List<String>> allActiveFiles = new ArrayList<>();
+ testSparkDeltaTable.insertRows(50);
+ Long timestamp1 = testSparkDeltaTable.getLastCommitTimestamp();
+ allActiveFiles.add(testSparkDeltaTable.getAllActiveFiles());
+ testSparkDeltaTable.insertRows(50);
+ allActiveFiles.add(testSparkDeltaTable.getAllActiveFiles());
+ testSparkDeltaTable.insertRows(50);
+ allActiveFiles.add(testSparkDeltaTable.getAllActiveFiles());
+ testSparkDeltaTable.insertRows(50);
+ allActiveFiles.add(testSparkDeltaTable.getAllActiveFiles());
+
+ // Spy the DeltaLog so we can count how many times a snapshot is
reconstructed (each
+ // reconstruction reads the table checkpoint from storage).
+ DeltaLog spiedDeltaLog =
+ spy(DeltaLog.forTable(sparkSession,
testSparkDeltaTable.getBasePath()));
+ DeltaConversionSource conversionSource =
+ DeltaConversionSource.builder()
+ .sparkSession(sparkSession)
+ .deltaLog(spiedDeltaLog)
+ .deltaTable(DeltaTable.forPath(sparkSession,
testSparkDeltaTable.getBasePath()))
+ .tableName(tableName)
+ .basePath(testSparkDeltaTable.getBasePath())
+ .reuseMetadataAcrossCommits(true)
+ .build();
+
+ InstantsForIncrementalSync instantsForIncrementalSync =
+ InstantsForIncrementalSync.builder()
+ .lastSyncInstant(Instant.ofEpochMilli(timestamp1))
+ .build();
+ CommitsBacklog<Long> commitsBacklog =
+ conversionSource.getCommitsBacklog(instantsForIncrementalSync);
+ List<TableChange> allTableChanges = new ArrayList<>();
+ for (Long version : commitsBacklog.getCommitsToProcess()) {
+ allTableChanges.add(conversionSource.getTableChangeForCommit(version));
+ }
+
+ // Behaviour is unchanged relative to reconstructing the snapshot per
commit.
+ ValidationTestHelper.validateTableChanges(allActiveFiles, allTableChanges);
+ // The backlog spans multiple commits with no schema change, so the
baseline snapshot is read
+ // exactly once and its metadata reused (previously: one snapshot
reconstruction per commit).
+ assertTrue(
+ commitsBacklog.getCommitsToProcess().size() >= 2,
+ "backlog must span multiple commits for this assertion to be
meaningful");
+ verify(spiedDeltaLog, times(1)).getSnapshotAt(anyLong(), any());
+ }
+
+ @Test
+ public void
schemaChangeMidBacklogRefreshesCachedMetadataWithoutSnapshotReload() {
+ String tableName = GenericTable.getTableName();
+ TestSparkDeltaTable testSparkDeltaTable =
+ new TestSparkDeltaTable(tableName, tempDir, sparkSession, null, false);
+ testSparkDeltaTable.insertRows(50);
+ Long timestamp1 = testSparkDeltaTable.getLastCommitTimestamp();
+ testSparkDeltaTable.insertRows(50);
+ // A metaData action lands mid-backlog in its own commit. Unlike
includeAdditionalColumns
+ // (a creation-time flag), this evolves the schema while a single source
instance holds a
+ // populated metadata cache.
+ sparkSession.sql(
+ String.format(
+ "ALTER TABLE delta.`%s` ADD COLUMNS (street string)",
+ testSparkDeltaTable.getBasePath()));
+ testSparkDeltaTable.insertRows(50);
+
+ DeltaLog spiedDeltaLog =
+ spy(DeltaLog.forTable(sparkSession,
testSparkDeltaTable.getBasePath()));
+ DeltaConversionSource conversionSource =
+ DeltaConversionSource.builder()
+ .sparkSession(sparkSession)
+ .deltaLog(spiedDeltaLog)
+ .deltaTable(DeltaTable.forPath(sparkSession,
testSparkDeltaTable.getBasePath()))
+ .tableName(tableName)
+ .basePath(testSparkDeltaTable.getBasePath())
+ .reuseMetadataAcrossCommits(true)
+ .build();
+
+ CommitsBacklog<Long> commitsBacklog =
+ conversionSource.getCommitsBacklog(
+ InstantsForIncrementalSync.builder()
+ .lastSyncInstant(Instant.ofEpochMilli(timestamp1))
+ .build());
+ List<Long> versions = commitsBacklog.getCommitsToProcess();
+ // insert, ALTER, insert — in ascending version order
+ assertEquals(3, versions.size(), "expected backlog of insert, alter,
insert");
+
+ for (int i = 0; i < versions.size(); i++) {
+ TableChange tableChange =
conversionSource.getTableChangeForCommit(versions.get(i));
+ boolean hasNewColumn =
+
tableChange.getTableAsOfChange().getReadSchema().getAllFields().stream()
+ .anyMatch(field -> "street".equals(field.getName()));
+ if (i == 0) {
+ assertFalse(hasNewColumn, "commit before the schema change must not
carry the new column");
+ } else {
+ assertTrue(
+ hasNewColumn,
+ String.format("commit at index %d must carry the schema added
mid-backlog", i));
+ }
+ }
+ // The refresh comes from the commit's own metaData action, not a snapshot
reload: the only
+ // snapshot read is the baseline at the first commit.
+ verify(spiedDeltaLog, times(1)).getSnapshotAt(anyLong(), any());
+ }
+
+ @Test
+ public void reuseMetadataDisabledReconstructsSnapshotPerCommit() {
+ String tableName = GenericTable.getTableName();
+ TestSparkDeltaTable testSparkDeltaTable =
+ new TestSparkDeltaTable(tableName, tempDir, sparkSession, null, false);
+ List<List<String>> allActiveFiles = new ArrayList<>();
+ testSparkDeltaTable.insertRows(50);
+ Long timestamp1 = testSparkDeltaTable.getLastCommitTimestamp();
+ allActiveFiles.add(testSparkDeltaTable.getAllActiveFiles());
+ testSparkDeltaTable.insertRows(50);
+ allActiveFiles.add(testSparkDeltaTable.getAllActiveFiles());
+ testSparkDeltaTable.insertRows(50);
+ allActiveFiles.add(testSparkDeltaTable.getAllActiveFiles());
+
+ DeltaLog spiedDeltaLog =
+ spy(DeltaLog.forTable(sparkSession,
testSparkDeltaTable.getBasePath()));
+ DeltaConversionSource conversionSource =
+ DeltaConversionSource.builder()
+ .sparkSession(sparkSession)
+ .deltaLog(spiedDeltaLog)
+ .deltaTable(DeltaTable.forPath(sparkSession,
testSparkDeltaTable.getBasePath()))
+ .tableName(tableName)
+ .basePath(testSparkDeltaTable.getBasePath())
+ .reuseMetadataAcrossCommits(false)
+ .build();
+
+ CommitsBacklog<Long> commitsBacklog =
+ conversionSource.getCommitsBacklog(
+ InstantsForIncrementalSync.builder()
+ .lastSyncInstant(Instant.ofEpochMilli(timestamp1))
+ .build());
+ List<TableChange> allTableChanges = new ArrayList<>();
+ for (Long version : commitsBacklog.getCommitsToProcess()) {
+ allTableChanges.add(conversionSource.getTableChangeForCommit(version));
+ }
+
+ ValidationTestHelper.validateTableChanges(allActiveFiles, allTableChanges);
+ // With reuse disabled the pre-optimization behaviour is restored in full:
one snapshot
+ // reconstruction per commit, and no second listing of the log for
commit-file mtimes.
+ verify(spiedDeltaLog, times(commitsBacklog.getCommitsToProcess().size()))
+ .getSnapshotAt(anyLong(), any());
+ verify(spiedDeltaLog, never()).getChangeLogFiles(anyLong(), anyBoolean());
+ }
+
+ @Test
+ public void
incrementalSyncWatermarkAdvancesAndDoesNotReprocessSyncedCommits() {
+ String tableName = GenericTable.getTableName();
+ TestSparkDeltaTable testSparkDeltaTable =
+ new TestSparkDeltaTable(tableName, tempDir, sparkSession, null, false);
+ testSparkDeltaTable.insertRows(50);
+ Long syncStart = testSparkDeltaTable.getLastCommitTimestamp();
+ // First backlog: several append-only commits (no schema change), so the
metadata is cached
+ // and reused across all of them.
+ testSparkDeltaTable.insertRows(50);
+ testSparkDeltaTable.insertRows(50);
+ testSparkDeltaTable.insertRows(50);
+
+ DeltaConversionSource firstRun =
+ DeltaConversionSource.builder()
+ .sparkSession(sparkSession)
+ .deltaLog(DeltaLog.forTable(sparkSession,
testSparkDeltaTable.getBasePath()))
+ .deltaTable(DeltaTable.forPath(sparkSession,
testSparkDeltaTable.getBasePath()))
+ .tableName(tableName)
+ .basePath(testSparkDeltaTable.getBasePath())
+ .reuseMetadataAcrossCommits(true)
+ .build();
+ List<Long> firstVersions =
+ firstRun
+ .getCommitsBacklog(
+ InstantsForIncrementalSync.builder()
+ .lastSyncInstant(Instant.ofEpochMilli(syncStart))
+ .build())
+ .getCommitsToProcess();
+ assertTrue(firstVersions.size() >= 3, "first backlog should span multiple
commits");
+
+ Instant firstCommitTime = null;
+ Instant watermark = null;
+ for (Long version : firstVersions) {
+ Instant commitTime =
+
firstRun.getTableChangeForCommit(version).getTableAsOfChange().getLatestCommitTime();
+ if (firstCommitTime == null) {
+ firstCommitTime = commitTime;
+ }
+ watermark = commitTime;
+ }
+ // Regression guard: latestCommitTime feeds the persisted sync watermark,
so it must advance per
+ // commit even though the metadata is cached. If it came from the baseline
reload version, every
+ // commit would report that version's timestamp and the watermark would
not move.
+ assertTrue(
+ !firstCommitTime.equals(watermark),
+ "latestCommitTime did not advance across the cached backlog (watermark
stuck at first version)");
+
+ // Append more commits, then sync again from the watermark the first run
would have persisted.
+ testSparkDeltaTable.insertRows(50);
+ testSparkDeltaTable.insertRows(50);
+ DeltaConversionSource secondRun =
+ DeltaConversionSource.builder()
+ .sparkSession(sparkSession)
+ .deltaLog(DeltaLog.forTable(sparkSession,
testSparkDeltaTable.getBasePath()))
+ .deltaTable(DeltaTable.forPath(sparkSession,
testSparkDeltaTable.getBasePath()))
+ .tableName(tableName)
+ .basePath(testSparkDeltaTable.getBasePath())
+ .reuseMetadataAcrossCommits(true)
+ .build();
+ List<Long> secondVersions =
+ secondRun
+ .getCommitsBacklog(
+
InstantsForIncrementalSync.builder().lastSyncInstant(watermark).build())
+ .getCommitsToProcess();
+
+ // The second run must not re-process the already-synced tail; the
stuck-watermark bug would
+ // re-include all of it. The bound is <= 1 rather than 0 because the
boundary commit can
+ // legitimately reappear: getActiveCommitAtTime compares against
monotonizeCommitTimestamps
+ // output while we persist the raw mtime, so a commit whose mtime tied
with its predecessor
+ // resolves to an earlier version. Unchanged from main, where
Snapshot.timestamp() is the same
+ // raw mtime. Tightening to isEmpty() would flake on coarse mtime
granularity.
+ List<Long> reprocessed = new ArrayList<>(firstVersions);
+ reprocessed.retainAll(secondVersions);
+ assertTrue(
+ reprocessed.size() <= 1,
+ "second run reprocessed already-synced commits: reprocessed="
+ + reprocessed
+ + " first="
+ + firstVersions
+ + " second="
+ + secondVersions);
+ }
}
diff --git
a/xtable-core/src/test/java/org/apache/xtable/delta/TestDeltaActionsConverter.java
b/xtable-core/src/test/java/org/apache/xtable/delta/TestDeltaActionsConverter.java
index e62e9341..c9b0d0a4 100644
---
a/xtable-core/src/test/java/org/apache/xtable/delta/TestDeltaActionsConverter.java
+++
b/xtable-core/src/test/java/org/apache/xtable/delta/TestDeltaActionsConverter.java
@@ -45,6 +45,9 @@ class TestDeltaActionsConverter {
String filePath =
"https://container.blob.core.windows.net/tablepath/file_path";;
Snapshot snapshot = Mockito.mock(Snapshot.class);
DeltaLog deltaLog = Mockito.mock(DeltaLog.class);
+ Mockito.when(snapshot.deltaLog()).thenReturn(deltaLog);
+ Mockito.when(deltaLog.dataPath())
+ .thenReturn(new
Path("https://container.blob.core.windows.net/tablepath";));
DeletionVectorDescriptor deletionVector = null;
AddFile addFileAction =
@@ -58,9 +61,6 @@ class TestDeltaActionsConverter {
addFileAction =
new AddFile(filePath, null, size, time, dataChange, stats, null,
deletionVector);
- Mockito.when(snapshot.deltaLog()).thenReturn(deltaLog);
- Mockito.when(deltaLog.dataPath())
- .thenReturn(new
Path("https://container.blob.core.windows.net/tablepath";));
Assertions.assertEquals(
filePath, actionsConverter.extractDeletionVectorFile(snapshot,
addFileAction));
}
diff --git
a/xtable-core/src/test/java/org/apache/xtable/delta/TestDeltaConversionSourceConfig.java
b/xtable-core/src/test/java/org/apache/xtable/delta/TestDeltaConversionSourceConfig.java
new file mode 100644
index 00000000..970a4a3f
--- /dev/null
+++
b/xtable-core/src/test/java/org/apache/xtable/delta/TestDeltaConversionSourceConfig.java
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.xtable.delta;
+
+import static
org.apache.xtable.delta.DeltaConversionSourceConfig.REUSE_METADATA_ACROSS_COMMITS;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+
+import java.util.Properties;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+
+public class TestDeltaConversionSourceConfig {
+
+ @Test
+ public void nullPropertiesUsesDefault() {
+
assertFalse(DeltaConversionSourceConfig.fromProperties(null).isReuseMetadataAcrossCommits());
+ }
+
+ @Test
+ public void absentKeyUsesDefault() {
+ assertFalse(
+ DeltaConversionSourceConfig.fromProperties(new Properties())
+ .isReuseMetadataAcrossCommits());
+ }
+
+ @ParameterizedTest
+ @CsvSource({
+ "true, true",
+ "TRUE, true",
+ "True, true",
+ "false, false",
+ "FALSE, false",
+ // Boolean.parseBoolean treats anything that is not "true" as false
+ "yes, false"
+ })
+ public void reuseMetadataAcrossCommitsIsParsedFromProperties(String value,
boolean expected) {
+ Properties properties = new Properties();
+ properties.setProperty(REUSE_METADATA_ACROSS_COMMITS, value);
+ assertEquals(
+ expected,
+
DeltaConversionSourceConfig.fromProperties(properties).isReuseMetadataAcrossCommits());
+ }
+}
