You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@xtable.apache.org by "AadhiKat (via GitHub)" <gi...@apache.org> on 2026/07/22 20:11:26 UTC
[PR] [860] Reuse the Delta snapshot across an incremental backlog [incubator-xtable]
AadhiKat opened a new pull request, #861:
URL: https://github.com/apache/incubator-xtable/pull/861
Closes #860.
getTableChangeForCommit reconstructs the table snapshot (deltaLog.getSnapshotAt) for every commit,
which re-reads the Delta checkpoint each time. It's fine when the checkpoint is small, but on a table
with a large checkpoint it becomes the bottleneck for incremental catch-up (see #860).
This caches the snapshot, along with the table and file format derived from it, and reloads it only
when a commit carries a Metadata or Protocol action. Apart from those, the snapshot is only used for
schema, partitioning, file format, and the table base path, which don't change between commits.
getActionsForVersion returns the unfiltered action list, so a schema/protocol change is visible to
the check and forces a reload for that version onward.
The one thing that varies per commit and comes from the snapshot is InternalTable.latestCommitTime
(snapshot.timestamp()). With reuse, intermediate commits report the timestamp of the last reload
rather than their own. If keeping that exact matters, it can come from the commit's CommitInfo
instead — happy to change it if you'd prefer.
Testing:
- Existing ITDeltaConversionSource passes unchanged (15 tests). testAddColumns is the relevant one:
it adds a column mid-backlog, so if the cache weren't invalidated on the metadata change the later
commits would carry the old schema and it would fail.
- Added getTableChangeForCommitReconstructsSnapshotOnceForAppendOnlyBacklog: runs a multi-commit
append backlog through a source built with a spied DeltaLog, checks the table changes still
validate, and asserts getSnapshotAt is called once rather than once per commit.
On the ~45M-file table from the issue, incremental catch-up went from ~2-3 min/commit to ~3 s/commit
(around 210 commits in ~10 min), with driver memory flat near 5 GB.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [860] Reuse the Delta snapshot across an incremental backlog [incubator-xtable]
Posted by "vinishjail97 (via GitHub)" <gi...@apache.org>.
vinishjail97 commented on code in PR #861:
URL: https://github.com/apache/incubator-xtable/pull/861#discussion_r3761599371
##########
xtable-core/src/main/java/org/apache/xtable/delta/DeltaConversionSource.java:
##########
@@ -78,6 +79,14 @@ public class DeltaConversionSource implements ConversionSource<Long> {
private final String tableName;
private final String basePath;
+ // When enabled (default), the table metadata is carried across an incremental backlog instead
+ // of reconstructing the Delta snapshot (a full checkpoint read) on every commit. The baseline
+ // comes from one snapshot read; a commit carrying a Metadata action refreshes the cache from
+ // the action itself (see getTableChangeForCommit).
+ @Builder.Default private final boolean reuseMetadataAcrossCommits = true;
+
+ private Metadata cachedMetadata;
Review Comment:
Addressed in `cd928ea1`. `resetState` now clears `cachedMetadata` and `lastProcessedVersion` before rebuilding the changes state, so the "one snapshot read supplies the baseline" invariant is per backlog rather than per source instance.
Added the ordering guard too. One judgement call worth your eye: I put the `Preconditions.checkArgument` inside the reuse branch rather than at the top of the method. Now that the flag defaults to `false`, an unconditional check would impose a new ordering constraint on the default path, which is unrelated to this optimization. Happy to hoist it if you'd rather it held everywhere.
##########
xtable-core/src/main/java/org/apache/xtable/delta/DeltaConversionSourceProvider.java:
##########
@@ -27,16 +27,31 @@
/** A concrete implementation of {@link ConversionSourceProvider} for Delta Lake table format. */
public class DeltaConversionSourceProvider extends ConversionSourceProvider<Long> {
+ /**
+ * When true (default), an incremental backlog carries the table metadata across commits instead
+ * of reconstructing the Delta snapshot (a full checkpoint read) for every commit. Set to false to
+ * restore the per-commit snapshot reconstruction.
+ */
+ static final String REUSE_METADATA_ACROSS_COMMITS =
Review Comment:
Both addressed in `cd928ea1`.
The constant now lives in a new `DeltaConversionSourceConfig`, a `@Value` class with `fromProperties`, mirroring `DeltaConversionTargetConfig`. The constant is `public static final`, so an embedder can reference it, and `DeltaConversionSourceProvider` just calls `fromProperties` on the source table's additional properties.
`TestDeltaConversionSourceConfig` covers the parsing: null properties, absent key, `"false"`, `"FALSE"`, mixed case `"True"`, and a non-boolean value. It pins the default, which as you noted is the one thing about this change an operator would actually notice, and which is now `false`.
##########
xtable-core/src/main/java/org/apache/xtable/delta/DeltaIncrementalChangesState.java:
##########
@@ -61,6 +65,19 @@ public DeltaIncrementalChangesState(DeltaLog deltaLog, Long versionToStartFrom)
maxSeenVersion =
maxSeenVersion == null ? versionNumber : Math.max(maxSeenVersion, versionNumber);
}
+ // Carry each commit's file modification time alongside its actions. This is the same clock
+ // Snapshot.timestamp() reads and the same one getCommitsBacklog matches the persisted
+ // watermark against, so it stays consistent regardless of the writer's clock. The log has
+ // already been listed once for getChanges above; this second pass reads no file contents.
+ Iterator<Tuple2<Object, FileStatus>> logFiles =
+ JavaConverters.asJavaIteratorConverter(
+ deltaLog.getChangeLogFiles(versionToStartFrom, false))
Review Comment:
Good catch, and I went with the first option rather than the documentation-only one. Fixed in `cd928ea1`.
`DeltaIncrementalChangesState` now takes a `loadCommitTimestamps` flag, which `resetState` wires to `reuseMetadataAcrossCommits`. With the flag off, `getChangeLogFiles` never runs, and `getTableChangeForCommit` takes a plain `Snapshot` branch, so `latestCommitTime` is `Snapshot.timestamp()` exactly as on main. The escape hatch now reverts all three behaviour changes rather than one.
`reuseMetadataDisabledReconstructsSnapshotPerCommit` asserts `never()).getChangeLogFiles(...)` alongside the existing per-commit reconstruction count, so the revert is pinned rather than just documented.
##########
xtable-core/src/main/java/org/apache/xtable/delta/DeltaTableExtractor.java:
##########
@@ -79,6 +80,44 @@ public InternalTable table(Snapshot snapshot, String tableName) {
.build();
}
+ /**
+ * 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(
Review Comment:
Done in `cd928ea1`, delegating exactly as you sketched. That drops about 20 duplicated lines, and the drift you spotted goes with it: `requireMetadata` still runs first on the snapshot path, so the missing-schema message keeps its version regardless of which entry point you came through.
##########
xtable-core/src/test/java/org/apache/xtable/delta/ITDeltaConversionSource.java:
##########
@@ -829,4 +836,231 @@ private boolean checkIfFileIsRemoved(String activePath, TableChange tableChange)
.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())
+ .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())
+ .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: one snapshot
+ // reconstruction per commit.
+ verify(spiedDeltaLog, times(commitsBacklog.getCommitsToProcess().size()))
+ .getSnapshotAt(anyLong(), any());
+ }
+
+ @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())
+ .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())
+ .build();
+ List<Long> secondVersions =
+ secondRun
+ .getCommitsBacklog(
+ InstantsForIncrementalSync.builder().lastSyncInstant(watermark).build())
+ .getCommitsToProcess();
+
+ // The second run must not re-process the already-synced tail. (At most the boundary commit may
+ // reappear, since CommitInfo timestamps are the writer's wall-clock rather than the commit
Review Comment:
You're right, the rationale was stale. Corrected in `cd928ea1`, and I took your explanation nearly verbatim since it is the accurate one: `getActiveCommitAtTime` compares against `monotonizeCommitTimestamps` output while we persist the raw `FileStatus` mtime, monotonization only ever increases a timestamp, so a commit whose mtime tied with or went behind its predecessor resolves to an earlier version.
Left the `<= 1` bound alone, with a note that tightening it to `isEmpty()` would flake on coarse mtime granularity, and that this is unchanged from main since `Snapshot.timestamp()` is the same raw value.
##########
xtable-core/src/main/java/org/apache/xtable/delta/DeltaActionsConverter.java:
##########
@@ -117,12 +154,21 @@ static String getFullPathToFile(Snapshot snapshot, String dataFilePath) {
* is present
*/
public String extractDeletionVectorFile(Snapshot snapshot, AddFile addFile) {
+ // Check for the deletion vector before resolving the base path, so an add file without one
Review Comment:
Done in `cd928ea1`, the method is now just the delegation and the single null check lives in the `String` overload.
One thing to note since it touched a test: `TestDeltaActionsConverter.extractDeletionVector` was calling this before stubbing `snapshot.deltaLog()`, so it was relying on that early return to avoid an NPE. I moved the two stubbing lines above the first call. Both assertions are unchanged.
##########
xtable-core/src/main/java/org/apache/xtable/delta/DeltaConversionSource.java:
##########
@@ -78,6 +81,13 @@ public class DeltaConversionSource implements ConversionSource<Long> {
private final String tableName;
private final String basePath;
+ // Cache the version-stable snapshot/table/format across an incremental backlog to avoid
+ // reconstructing the Delta snapshot (a full checkpoint read) on every commit. Invalidated
+ // whenever a commit carries a metadata/protocol change (see getTableChangeForCommit).
+ private Snapshot cachedSnapshot;
Review Comment:
All four landed in `48ef6d7`, thanks for turning them around so thoroughly. The follow-ups from my next round are now in `cd928ea1`.
Answering your two notes so they don't stay open: leaving the two listings in `DeltaIncrementalChangesState` is fine by me, a single pass isn't worth reimplementing Delta's action parsing for a cost that measures inside run-to-run noise. And yes please to a separate issue for the same treatment on `DeltaKernelConversionSource`, this PR is big enough already.
Resolving this one.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [860] Reuse the Delta snapshot across an incremental backlog [incubator-xtable]
Posted by "AadhiKat (via GitHub)" <gi...@apache.org>.
AadhiKat commented on code in PR #861:
URL: https://github.com/apache/incubator-xtable/pull/861#discussion_r3641884400
##########
xtable-core/src/main/java/org/apache/xtable/delta/DeltaConversionSource.java:
##########
@@ -103,10 +112,21 @@ public InternalSnapshot getCurrentSnapshot() {
@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());
+ // Reconstructing the snapshot per commit reloads the table checkpoint each time, which is
+ // expensive when the checkpoint is large. Reuse it across the backlog and reload only when a
+ // commit changes schema or protocol.
+ boolean metadataMayHaveChanged =
+ actionsForVersion.stream()
+ .anyMatch(action -> (action instanceof Metadata) || (action instanceof Protocol));
+ if (cachedSnapshot == null || metadataMayHaveChanged) {
+ cachedSnapshot = deltaLog.getSnapshotAt(versionNumber, Option.empty());
+ cachedTable = tableExtractor.table(cachedSnapshot, tableName);
+ cachedFileFormat =
+ actionsConverter.convertToFileFormat(cachedSnapshot.metadata().format().provider());
+ }
+ Snapshot snapshotAtVersion = cachedSnapshot;
+ InternalTable tableAtVersion = cachedTable;
Review Comment:
Thanks @vinishjail97. You're right that with the snapshot cached the
watermark would sit at the reload version and the next run would re-read the tail.
Fixed by reading `latestCommitTime` per commit from the commit's `CommitInfo`
(already in `actionsForVersion`, so no extra checkpoint read), falling back to
`snapshot.timestamp()` only if a commit has no `CommitInfo`. Everything else still
comes from the cached snapshot, since it doesn't change between commits.
One thing worth flagging: `CommitInfo.timestamp` is the writer's time rather than
the commit-file mtime that `snapshot.timestamp()` uses, so at a backlog boundary the
next run could re-touch the final commit. It's idempotent, but if you'd rather keep
it exact I can read the timestamp from `DeltaHistoryManager` instead. Happy either way.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [860] Reuse the Delta snapshot across an incremental backlog [incubator-xtable]
Posted by "vinishjail97 (via GitHub)" <gi...@apache.org>.
vinishjail97 commented on code in PR #861:
URL: https://github.com/apache/incubator-xtable/pull/861#discussion_r3732145149
##########
xtable-core/src/main/java/org/apache/xtable/delta/DeltaConversionSource.java:
##########
@@ -78,6 +81,13 @@ public class DeltaConversionSource implements ConversionSource<Long> {
private final String tableName;
private final String basePath;
+ // Cache the version-stable snapshot/table/format across an incremental backlog to avoid
+ // reconstructing the Delta snapshot (a full checkpoint read) on every commit. Invalidated
+ // whenever a commit carries a metadata/protocol change (see getTableChangeForCommit).
+ private Snapshot cachedSnapshot;
Review Comment:
Nice profiling, the direction is right. One suggestion on where the cache sits: cache the metadata rather than the Snapshot.
```java
private Metadata cachedMetadata; // the only cached state
// in getTableChangeForCommit, before the file loop:
Metadata metadataInCommit = null;
for (Action action : actionsForVersion) {
if (action instanceof Metadata) {
metadataInCommit = (Metadata) action; // mirror InMemoryLogReplay: overwrite, do not break
}
}
if (cachedMetadata == null) {
// baseline, once per backlog; the Snapshot is read for its metadata and not retained
cachedMetadata = deltaLog.getSnapshotAt(versionNumber, Option.empty()).metadata();
} else if (metadataInCommit != null) {
cachedMetadata = metadataInCommit;
}
InternalTable tableAtVersion =
tableExtractor.table(
cachedMetadata, deltaLog, tableName, getChangesState().getCommitTimestamp(versionNumber));
FileFormat fileFormat = actionsConverter.convertToFileFormat(cachedMetadata.format().provider());
```
One field instead of three, since the table and file format are cheap to derive from the metadata. It needs a `Metadata` overload on `DeltaTableExtractor` and the converter taking a base path instead of a `Snapshot`, which `DeltaKernelConversionSource` already does. `Protocol` can leave the guard too, since nothing reads `snapshot.protocol()`.
On the timestamp: main already used the delta file mtime, because `snapshot.timestamp()` is `deltas.last.getModificationTime`. It just paid a full checkpoint read to get it. `getCommitTimestamp` does not do that. It uses `CommitInfo.timestamp`, which is the writer's clock, and only reaches the mtime through another full reconstruction. `getCommitsBacklog` matches the watermark against mtimes, so a writer whose clock runs ahead can push it past a commit that never got processed.
`DeltaIncrementalChangesState` already lists the log once per backlog, so it can carry the timestamps and `getCommitTimestamp` goes away:
```java
private final Map<Long, Instant> commitTimestampByVersion = new HashMap<>();
// in the constructor, alongside the existing getChanges loop:
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()));
}
public Instant getCommitTimestamp(Long version) {
Preconditions.checkArgument(
commitTimestampByVersion.containsKey(version),
String.format("Version %s not found in the DeltaIncrementalChangesState.", version));
return commitTimestampByVersion.get(version);
}
```
Could we also gate this, following `DeltaConversionTargetConfig`: `xtable.delta.source.reuse_metadata_across_commits`, read from `sourceTable.getAdditionalProperties()`, default true.
On tests, `testAddColumns` does not evolve the schema mid-backlog. `includeAdditionalColumns` is a constructor flag, so the table is created with those columns and no `metaData` action follows. `ITConversionController` does evolve the schema, but each `sync()` builds a fresh source, so the cache is null on the first commit and always reloads. Could you add one backlog on one source instance with append, schema change, append, asserting the later commits carry the new schema.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [860] Reuse the Delta snapshot across an incremental backlog [incubator-xtable]
Posted by "AadhiKat (via GitHub)" <gi...@apache.org>.
AadhiKat commented on code in PR #861:
URL: https://github.com/apache/incubator-xtable/pull/861#discussion_r3735173778
##########
xtable-core/src/main/java/org/apache/xtable/delta/DeltaConversionSource.java:
##########
@@ -78,6 +81,13 @@ public class DeltaConversionSource implements ConversionSource<Long> {
private final String tableName;
private final String basePath;
+ // Cache the version-stable snapshot/table/format across an incremental backlog to avoid
+ // reconstructing the Delta snapshot (a full checkpoint read) on every commit. Invalidated
+ // whenever a commit carries a metadata/protocol change (see getTableChangeForCommit).
+ private Snapshot cachedSnapshot;
Review Comment:
Done, all four — pushed as 48ef6d7.
**Metadata cache.** As you sketched it: single `cachedMetadata` field, baseline from one
`getSnapshotAt().metadata()` (snapshot not retained), refreshed from the commit's own `metaData`
action, last-in-commit wins. `Protocol` left the guard. `DeltaTableExtractor` gained the `Metadata`
overload; the actions converter got base-path variants alongside the `Snapshot` ones (which now
delegate), same shape as the kernel converter.
**Timestamp.** You're right, and this was the sharpest catch of the round — thanks. `CommitInfo.timestamp`
is the writer's clock, `getCommitsBacklog` resolves the watermark against mtimes, so a fast writer
clock could push the watermark past commits that were never processed. Skipped data, not
reprocessed data — worse than what the round-1 fix was defending against. `DeltaIncrementalChangesState`
now carries per-version mtimes from `getChangeLogFiles` (populated in the constructor next to the
existing `getChanges` pass) and exposes `getCommitTimestamp(version)`; `DeltaConversionSource`'s
CommitInfo path and its reconstruction fallback are both gone.
**Config.** `xtable.delta.source.reuse_metadata_across_commits`, read from
`sourceTable.getAdditionalProperties()` in the provider, default true. `false` restores per-commit
snapshot reconstruction exactly.
**Tests.** You were right about `testAddColumns` — `includeAdditionalColumns` is a creation-time
flag, so no `metaData` action ever fires mid-backlog, and `ITConversionController` builds a fresh
source per sync so its cache is always cold. Added
`schemaChangeMidBacklogRefreshesCachedMetadataWithoutSnapshotReload`: one source instance, backlog of
append → `ALTER TABLE ADD COLUMNS` → append, asserting the pre-alter commit lacks the column, the
alter and post-alter commits carry it, and — since the refresh comes from the action —
`getSnapshotAt` is still called exactly once. Also added
`reuseMetadataDisabledReconstructsSnapshotPerCommit` pinning the flag-off path to one reconstruction
per commit.
Re-ran the A/B from the earlier comment on the rebuilt jar (same table, same protocol; it has
grown ~300 versions since, so reloads run slightly heavier now):
```
RELOAD (cache cleared per commit) run 1 : 110,760 ms / 6 commits (genuine reloads 21.6-22.5 s)
REUSE (default) : 21,316 ms / 6 commits (baseline 21.2 s, then 3-24 ms)
RELOAD (cache cleared per commit) run 2 : 108,836 ms / 6 commits
```
(The RELOAD arm nulls `cachedMetadata` before each commit, which takes the same branch the config
flag disables — the flag-off path is additionally pinned by its own test.) Same shape as before —
the metadata cache didn't change the win. One new cost to be upfront about: backlog init now does the extra
`getChangeLogFiles` listing for the mtimes. Measured three runs on each build: 22.1-24.5 s init on
this branch vs 17.1-28.1 s on the previous one — the added pass is inside run-to-run noise on GCS
(init is dominated by `getActiveCommitAtTime` plus the `getChanges` read either way), and it's paid
once per sync rather than per commit.
The schema-change case isn't in the table because I won't ALTER this production table for a
benchmark — but it's pinned by the new test: the refresh comes from the commit's own action, so a
schema change mid-backlog that previously cost a full reload (~21 s here) now costs no snapshot
read at all (`getSnapshotAt` asserted to run exactly once across the evolve-mid-backlog run).
Full `mvn clean install` is green (xtable-core: 486 unit / 126 IT, 0 failures). One note if you
re-run the repro I posted earlier: its reflection cleared the old field names — updated it to
handle both layouts.
Two notes on the diff so you don't have to ask:
- `DeltaIncrementalChangesState` now lists the log twice per backlog (`getChanges` + `getChangeLogFiles`), and since `getChanges` is itself built on `getChangeLogFiles`, one pass could gather actions and mtimes together. I kept the two calls because the one-pass version means reimplementing Delta's action parsing on our side, and the second listing measured within run-to-run noise (numbers above) — but if you'd rather have the single pass, it's mechanical and I'm happy to do it here.
- `DeltaKernelConversionSource.getTableChangeForCommit` has the same per-commit reconstruction (`table.getSnapshotAsOfVersion` per commit). I'd rather not grow this PR's scope — if you agree, I'll file a follow-up issue to bring the same metadata reuse to the kernel path and take it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [860] Reuse the Delta snapshot across an incremental backlog [incubator-xtable]
Posted by "AadhiKat (via GitHub)" <gi...@apache.org>.
AadhiKat commented on code in PR #861:
URL: https://github.com/apache/incubator-xtable/pull/861#discussion_r3641885912
##########
xtable-core/src/test/java/org/apache/xtable/delta/ITDeltaConversionSource.java:
##########
@@ -829,4 +836,54 @@ private boolean checkIfFileIsRemoved(String activePath, TableChange tableChange)
.collect(Collectors.toSet());
return filePathsRemoved.contains(activePath);
}
+
+ @Test
+ public void getTableChangeForCommitReconstructsSnapshotOnceForAppendOnlyBacklog() {
Review Comment:
Added the two-run test you suggested (`incrementalSyncWatermarkAdvancesAndDoesNotReprocessSyncedCommits`):
sync a multi-commit backlog, append a couple more, sync again from the watermark, and
assert the second backlog leaves out the already-synced versions. Kept the
`getSnapshotAt` call-count test too.
mvn clean install is green locally.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [860] Reuse the Delta snapshot across an incremental backlog [incubator-xtable]
Posted by "vinishjail97 (via GitHub)" <gi...@apache.org>.
vinishjail97 commented on code in PR #861:
URL: https://github.com/apache/incubator-xtable/pull/861#discussion_r3761651813
##########
xtable-core/src/main/java/org/apache/xtable/delta/DeltaConversionSource.java:
##########
@@ -103,10 +113,52 @@ public InternalSnapshot getCurrentSnapshot() {
@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) {
Review Comment:
Done in `7249dee9`. The block is now `resolveMetadataForCommit(versionNumber, actionsForVersion)`, which holds the ordering guard, the last-in-commit `Metadata` scan and the baseline/refresh of `cachedMetadata`, and returns the metadata. The branch at the call site is down to three lines, so both arms of the `if` read at the same altitude. The cache rationale moved into the method javadoc; the watermark comment stayed at the call site since it explains the timestamp rather than the cache. No behaviour change — `mvn verify` on `xtable-core` is green (133 unit, 19 IT including `ITDeltaConversionSource`, 0 failures).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [860] Reuse the Delta snapshot across an incremental backlog [incubator-xtable]
Posted by "vinishjail97 (via GitHub)" <gi...@apache.org>.
vinishjail97 commented on code in PR #861:
URL: https://github.com/apache/incubator-xtable/pull/861#discussion_r3761622967
##########
xtable-core/src/main/java/org/apache/xtable/delta/DeltaConversionSource.java:
##########
@@ -103,10 +113,52 @@ public InternalSnapshot getCurrentSnapshot() {
@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) {
Review Comment:
Move this block inside reuseMetadataAcrossCommits to it's separate function and call it here?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [860] Reuse the Delta snapshot across an incremental backlog [incubator-xtable]
Posted by "AadhiKat (via GitHub)" <gi...@apache.org>.
AadhiKat commented on code in PR #861:
URL: https://github.com/apache/incubator-xtable/pull/861#discussion_r3641885912
##########
xtable-core/src/test/java/org/apache/xtable/delta/ITDeltaConversionSource.java:
##########
@@ -829,4 +836,54 @@ private boolean checkIfFileIsRemoved(String activePath, TableChange tableChange)
.collect(Collectors.toSet());
return filePathsRemoved.contains(activePath);
}
+
+ @Test
+ public void getTableChangeForCommitReconstructsSnapshotOnceForAppendOnlyBacklog() {
Review Comment:
Added the two-run test you suggested (`incrementalSyncWatermarkAdvancesAndDoesNotReprocessSyncedCommits`):
sync a multi-commit backlog, append a couple more, sync again from the watermark, and
assert the second backlog leaves out the already-synced versions. Kept the
`getSnapshotAt` call-count test too.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [860] Reuse the Delta snapshot across an incremental backlog [incubator-xtable]
Posted by "AadhiKat (via GitHub)" <gi...@apache.org>.
AadhiKat commented on PR #861:
URL: https://github.com/apache/incubator-xtable/pull/861#issuecomment-5201780563
@vinishjail97, I did profiling and it ended up correcting the previous numbers. Repro you can run yourself is at the bottom.
**Where the time goes**
On the table from the issue (16 vCPU / 64 GB, `local[8]`, GCS, Spark 3.4.2 / Delta 2.4.0 from our pom):
```
v281561 getSnapshotAt = 21,133 ms metadata() = 0 ms schema() = 0 ms
v281560 getSnapshotAt = 21,030 ms metadata() = 0 ms schema() = 0 ms
```
`metadata()` showing 0 ms looks like it refutes your reading of 2.4.0, but it doesn't. `_protocol/_metadata` is a lazy val in `Snapshot.scala`, and the constructor ends with `logInfo(s"Created snapshot $this")`, whose toString includes `metadata=$metadata`. With INFO logging enabled (it was here), that forces the P&M reconstruction during construction, inside `getSnapshotAt`. With INFO off it would land on the first `metadata()` call instead. So you were right about the mechanism: the protocol/metadata reconstruction over the checkpoint is the biggest single cost, it just hides inside snapshot construction.
For listing vs read, Delta's own log timestamps the boundary on each reload:
```
05:37:27 Try to find Delta last complete checkpoint before version 281569
05:37:31 Delta checkpoint is found at version 281560 -> 4 s discovery
05:37:35 Loading version 281569 starting from checkpoint 281560 -> 4 s log segment build
05:37:35 Created DeltaLogFileIndex(Parquet, numFilesInSegment: 459, totalFileSize: 6900706829)
05:37:47 Created snapshot -> 12 s checkpoint read
```
Same shape on all 11 reconstructions across the runs below: discovery 4-5 s, segment build 3-5 s, read 11-12 s. So roughly 9 s listing + 12 s checkpoint read per commit. I'd originally assumed the listing was negligible since a flat `listFiles` of the whole 72,238-object dir takes 5.5 s, but `findLastCompleteCheckpointBefore` lists backwards in 1000-version windows (`Checkpoints.scala`), and at 459 parts every 10 commits one window spans ~46k objects. The segment build then lists again from the found checkpoint. Both passes happen per commit on main.
(The issue said 454 parts / ~6.5 GB; it's 459 / 6.90 GB now. Table is live, it grew.)
**A/B on the real table**
Same jar, same machine, same source instance, same 6-commit backlog. Only difference: clearing the cached snapshot before each commit (= main) or not (= this patch). RELOAD run before and after REUSE to rule out page cache:
```
RELOAD 21503 20575 20347 20409 20269 3 ms total 103,106 ms / 6 commits
REUSE 20991 2 14 2 19 2 ms total 21,030 ms / 6 commits
RELOAD 21710 20218 19866 20164 20398 2 ms total 102,358 ms / 6 commits
```
The two RELOAD runs agree within 0.7%. Non-snapshot per-commit work is ~8 ms, so the reconstruction is ~99.96% of the per-commit cost. (The last commit is ~3 ms even under RELOAD because `DeltaLog` caches the current snapshot itself, so `getSnapshotAt(latest)` is free.)
**Repro**
Builds a synthetic Delta table, checkpoints it, appends a backlog, then walks it through `DeltaConversionSource` twice: once clearing the cached snapshot per commit via reflection (main), once not (patch). Local filesystem, no credentials, JDK 11:
```
javac -cp xtable-utilities_2.12-*-bundled.jar DeltaRepro.java
java -cp .:xtable-utilities_2.12-*-bundled.jar DeltaRepro /tmp/repro 160000 5 20 60 120
```
```
### table: 160000 add-files, checkpoint 32 parts / 10,304,893 bytes
backlog RELOAD total REUSE total RELOAD/commit REUSE/commit ratio
5 575 ms 133 ms 115 ms 26 ms 4.3x
20 2521 ms 140 ms 126 ms 7 ms 18.0x
60 7538 ms 117 ms 125 ms 1 ms 64.4x
120 14105 ms 98 ms 117 ms 0 ms 143.9x
```
RELOAD grows linearly with backlog length, REUSE stays flat: the checkpoint gets read once instead of N times. Which also means there's no single speedup number, and I shouldn't have implied there was — the longer the backlog, the bigger the win. To push it further, raise the backlog length, or raise `addFiles` / lower `spark.databricks.delta.checkpoint.partSize` to grow the checkpoint. Note at 160k files the checkpoint is only ~10 MB and Spark job overhead dominates, so this shows the shape, not the magnitude.
<details>
<summary><code>DeltaRepro.java</code></summary>
```java
import io.delta.tables.DeltaTable;
import java.lang.reflect.Field;
import java.time.Instant;
import java.util.ArrayList;
import java.util.List;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.spark.sql.SaveMode;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.delta.DeltaLog;
import org.apache.xtable.delta.DeltaConversionSource;
import org.apache.xtable.model.CommitsBacklog;
import org.apache.xtable.model.InstantsForIncrementalSync;
/**
* Self-contained reproduction for apache/incubator-xtable PR #861.
*
* Builds a synthetic Delta table with a configurable number of add-files, checkpoints it, appends a
* short backlog of commits, then walks that backlog through DeltaConversionSource twice:
*
* RELOAD = today's main : the snapshot is reconstructed per commit
* REUSE = the patch : the snapshot is reconstructed once and reused
*
* Vary the backlog length to see the unpatched arm pay the checkpoint read once PER COMMIT while
* the patched arm pays it once in total. Everything is local -- no object storage, no credentials.
*
* java -cp <xtable-bundled-jar>:. DeltaRepro /tmp/repro 160000 5 20 60 120
*/
public class DeltaRepro {
// Fully-qualified rather than the "delta" short name: the short name needs the
// DataSourceRegister services entry, which the shaded xtable-utilities bundle has lost.
// The FQCN resolves on any classpath, shaded or not.
private static final String DELTA = "org.apache.spark.sql.delta.sources.DeltaDataSource";
private static void clearCache(DeltaConversionSource src) throws Exception {
for (String f : new String[] {"cachedSnapshot", "cachedTable", "cachedFileFormat"}) {
Field field = DeltaConversionSource.class.getDeclaredField(f);
field.setAccessible(true);
field.set(src, null);
}
}
private static DeltaConversionSource newSource(SparkSession spark, DeltaLog log, String path) {
return DeltaConversionSource.builder()
.sparkSession(spark).deltaLog(log).deltaTable((DeltaTable) null)
.tableName("repro").basePath(path).build();
}
/** @return {totalMs, commits} */
private static long[] walkBacklog(SparkSession spark, DeltaLog log, String path,
Instant lastSync, boolean reloadEachCommit) throws Exception {
DeltaConversionSource src = newSource(spark, log, path);
CommitsBacklog<Long> backlog = src.getCommitsBacklog(
InstantsForIncrementalSync.builder().lastSyncInstant(lastSync).build());
long total = 0, count = 0;
for (Long v : backlog.getCommitsToProcess()) {
if (reloadEachCommit) {
clearCache(src);
}
long t = System.nanoTime();
src.getTableChangeForCommit(v);
total += (System.nanoTime() - t) / 1_000_000L;
count++;
}
return new long[] {total, count};
}
public static void main(String[] args) throws Exception {
String baseDir = args[0];
int addFiles = Integer.parseInt(args[1]);
int[] backlogs = new int[args.length - 2];
int maxBacklog = 0;
for (int i = 2; i < args.length; i++) {
backlogs[i - 2] = Integer.parseInt(args[i]);
maxBacklog = Math.max(maxBacklog, backlogs[i - 2]);
}
SparkSession spark = SparkSession.builder()
.appName("xtable-861-repro").master("local[8]")
.config("spark.ui.enabled", "false")
.config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
.config("spark.sql.catalog.spark_catalog",
"org.apache.spark.sql.delta.catalog.DeltaCatalog")
.config("spark.hadoop.fs.file.impl", "org.apache.hadoop.fs.LocalFileSystem")
// small parts so a modest file count still yields a multi-part checkpoint,
// mirroring the shape of the table in the issue
.config("spark.databricks.delta.checkpoint.partSize", "5000")
.getOrCreate();
spark.sparkContext().setLogLevel("ERROR");
String path = baseDir + "/t" + addFiles;
spark.range(addFiles).toDF("id").write().format(DELTA)
.option("maxRecordsPerFile", 1)
.option("delta.checkpointInterval", "100000")
.mode(SaveMode.Overwrite).save(path);
DeltaLog log = DeltaLog.forTable(spark, path);
log.checkpoint(log.update(false, scala.Option.empty()));
for (int k = 0; k < maxBacklog; k++) {
spark.range(1).toDF("id").write().format(DELTA).mode(SaveMode.Append).save(path);
}
FileSystem fs = new Path(path).getFileSystem(spark.sessionState().newHadoopConf());
long parts = 0, bytes = 0;
for (FileStatus st : fs.globStatus(new Path(path + "/_delta_log/*.checkpoint*"))) {
parts++;
bytes += st.getLen();
}
System.out.printf("%n### table: %d add-files, checkpoint %d parts / %d bytes%n",
addFiles, parts, bytes);
System.out.printf("%n%-9s %-14s %-14s %-14s %-14s %-8s%n",
"backlog", "RELOAD total", "REUSE total", "RELOAD/commit", "REUSE/commit", "ratio");
System.out.println("-".repeat(78));
for (int b : backlogs) {
long latest = log.update(false, scala.Option.empty()).version();
Path commitFile =
new Path(String.format("%s/_delta_log/%020d.json", path, latest - b));
Instant lastSync =
Instant.ofEpochMilli(fs.getFileStatus(commitFile).getModificationTime());
long[] reload = walkBacklog(spark, log, path, lastSync, true);
long[] reuse = walkBacklog(spark, log, path, lastSync, false);
System.out.printf("%-9d %-14d %-14d %-14d %-14d %.1fx%n",
reload[1], reload[0], reuse[0],
reload[0] / Math.max(1, reload[1]), reuse[0] / Math.max(1, reuse[1]),
(double) reload[0] / Math.max(1, reuse[0]));
}
spark.stop();
}
}
```
</details>
**On the production-path concern**
The snapshot feeds exactly two things in `getTableChangeForCommit`: `tableExtractor.table(snapshot, tableName)` and `convertToFileFormat(snapshot.metadata().format().provider())`. Everything `DeltaTableExtractor.table` populates is either constant for the table (name, basePath, log path) or derived from `metadata()` (schema, partition fields, layout, file format). Nothing reads the file list or per-version state. So the cache can only go stale on a Metadata/Protocol action, which is what the guard checks — and it checks the unfiltered action list (`DeltaIncrementalChangesState` stores `deltaLog.getChanges()` raw), so those actions can't have been filtered out before the check. The cache lives on one `DeltaConversionSource`, which is built fresh per sync run, so there's no cross-run or cross-table state.
`latestCommitTime` is the one per-commit value, from your round-1 catch: it now comes from the commit's own `CommitInfo`, already present in `actionsForVersion`, no extra IO. It's persisted as the sync watermark, so a stale value there would mean reprocessing the tail on the next run.
One thing you'll see in the diff: if a commit has no `CommitInfo` (or a null timestamp), the fallback is `getSnapshotAt(version).timestamp()`, so that commit pays a full reconstruction. It degrades to main's behavior rather than producing a wrong watermark, but a backlog of CommitInfo-less commits would get no benefit from the patch. It could be O(1) instead: `Snapshot.timestamp` is just the last commit file's mtime (`lastCommitTimestamp = deltas.last.getModificationTime` in `SnapshotManagement.scala`), so a `getFileStatus` on `_delta_log/<version>.json` gives the same value without touching the checkpoint. Can do that in this PR or as a follow-up, your call.
Tests: `testAddColumns` is the real regression guard here (adds a column mid-backlog; if invalidation broke, later commits would carry the old schema and it fails). The new spy test asserts `getSnapshotAt` runs exactly once across an append backlog with output identical to main's. The two-run watermark test covers the reprocessing case. One caveat: the guard is sound for what `DeltaTableExtractor.table` reads today. If someone later derives something version-varying from the snapshot, the cache would serve it stale silently, and the spy test wouldn't catch it. I can add an assertion pinning the snapshot-derived fields if you want that enforced.
If you'd like an operational escape hatch on top: the reuse can sit behind a conf key as a kill switch (ConversionSourceProvider already carries the conf, RunSync already takes --hadoopConfig). I'd default it on since output is identical either way and the users hitting this pathology are the least likely to find an opt-in flag — but whether the project wants that config surface at all is your call.
**Correcting my own numbers**
The PR description says "~2-3 min/commit -> ~3 s/commit". The patched half holds (matches the 210 commits in ~10 min I quoted), but the unpatched half doesn't reproduce: it's ~20.5 s/commit, not 2-3 minutes. That figure came from eyeballing a much messier run and shouldn't have gone in the description. Measured: ~8x end-to-end on a ~200-commit backlog (~80 min -> ~10 min, Iceberg writes dominate the patched path), and 4x to >100x on the isolated code path depending on backlog length, per the repro table. I've updated the description.
**Why your 10 TB table showed nothing**
Two drivers, and a 10 TB table can easily have neither. Checkpoint size is driven by add-file count, not bytes: this table is a tiny-file pathology, 45,638,865 add-files -> 459 parts -> 6.90 GB. And the listing half is driven by `_delta_log` density, which is also part count: the discovery window here contains ~100 checkpoints x 459 parts ≈ 46k objects. A 10 TB table with sane file sizes and normal retention has a small checkpoint and a sparse log, so no per-commit penalty is what I'd expect. The shape that reproduces it is many small files plus a long, frequently-checkpointed history. My numbers are GCS; if you retry on S3 I'd expect the read half to transfer more cleanly than the listing half.
What you asked for:
| | |
|---|---|
| schema width | 256 columns |
| partitioning | `year, month, day, hour` |
| checkpoint cadence | every 10 commits |
| checkpoint | 459 parts, 6.90 GB, 45,638,865 add-files |
| `_delta_log` | 72,238 objects, 594.74 GB |
| writer | Apache-Spark/3.5.4, Delta-Lake/3.2.0 |
| reader | Spark 3.4.2, Delta 2.4.0 |
| commit shape | blind append, ~130-1000 files per commit |
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [860] Reuse the Delta snapshot across an incremental backlog [incubator-xtable]
Posted by "vinishjail97 (via GitHub)" <gi...@apache.org>.
vinishjail97 commented on code in PR #861:
URL: https://github.com/apache/incubator-xtable/pull/861#discussion_r3641364305
##########
xtable-core/src/test/java/org/apache/xtable/delta/ITDeltaConversionSource.java:
##########
@@ -829,4 +836,54 @@ private boolean checkIfFileIsRemoved(String activePath, TableChange tableChange)
.collect(Collectors.toSet());
return filePathsRemoved.contains(activePath);
}
+
+ @Test
+ public void getTableChangeForCommitReconstructsSnapshotOnceForAppendOnlyBacklog() {
Review Comment:
Nice to have the getSnapshotAt call-count assertion. Would it be worth adding a two-run case as well — sync, append more commits, sync again — asserting the second run's backlog doesn't re-include already-synced versions? That would guard the watermark-advancement behavior, which the current diff-validation tests don't exercise.
##########
xtable-core/src/main/java/org/apache/xtable/delta/DeltaConversionSource.java:
##########
@@ -103,10 +112,21 @@ public InternalSnapshot getCurrentSnapshot() {
@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());
+ // Reconstructing the snapshot per commit reloads the table checkpoint each time, which is
+ // expensive when the checkpoint is large. Reuse it across the backlog and reload only when a
+ // commit changes schema or protocol.
+ boolean metadataMayHaveChanged =
+ actionsForVersion.stream()
+ .anyMatch(action -> (action instanceof Metadata) || (action instanceof Protocol));
+ if (cachedSnapshot == null || metadataMayHaveChanged) {
+ cachedSnapshot = deltaLog.getSnapshotAt(versionNumber, Option.empty());
+ cachedTable = tableExtractor.table(cachedSnapshot, tableName);
+ cachedFileFormat =
+ actionsConverter.convertToFileFormat(cachedSnapshot.metadata().format().provider());
+ }
+ Snapshot snapshotAtVersion = cachedSnapshot;
+ InternalTable tableAtVersion = cachedTable;
Review Comment:
Reuse is fine for schema/partitioning/format/base path — those only change on Metadata/Protocol, which the reload check handles. The catch is latestCommitTime: it's the per-commit snapshot.timestamp(), and it gets persisted as the sync watermark (syncMetadata/lastInstantSynced). With the cache, every commit in an append-only backlog reports the reload version's timestamp, so we persist the first version's instant instead of the latest and next run re-processes the tail from there.
Could we cache the snapshot but set latestCommitTime per commit from CommitInfo (already in actionsForVersion) or DeltaHistoryManager?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [860] Reuse the Delta snapshot across an incremental backlog [incubator-xtable]
Posted by "vinishjail97 (via GitHub)" <gi...@apache.org>.
vinishjail97 commented on PR #861:
URL: https://github.com/apache/incubator-xtable/pull/861#issuecomment-5258675696
@AadhiKat thanks for the round-2 work, and sorry for the slow turnaround on my side. The profiling write-ups made this much easier to review.
Rather than send another round of comments, I've pushed `cd928ea1` to the branch directly (maintainer edits were enabled) with the remaining items from my last review. Happy to drop or amend it if you'd rather take them yourself, just say so.
One substantive change beyond what I'd asked for, which I want to flag rather than bury: **the config now defaults to `false`.** This is a core path used in production by everyone syncing from Delta, and the optimization only pays off on the tiny-file plus long-history shape from #860. I'd rather ship it opt-in for a release, let the people hitting the pathology turn it on, and revisit the default once it has mileage. Your numbers make the case for the optimization itself; my hesitation is only about who gets it silently.
The rest, thread by thread:
- `resetState` now clears `cachedMetadata`, so the invariant is per backlog rather than per source instance, plus a `Preconditions` guard on non-decreasing versions.
- Config moved to `DeltaConversionSourceConfig` with `fromProperties`, mirroring `DeltaConversionTargetConfig`, and a unit test covering the parsing.
- The flag now reverts all three behaviour changes, not just the snapshot reuse. With it off, `getChangeLogFiles` never runs and `latestCommitTime` comes from `Snapshot.timestamp()` exactly as on main.
- `DeltaTableExtractor.table(Snapshot, ...)` delegates to the `Metadata` overload.
- Dropped the duplicate deletion-vector null check.
- Corrected the stale rationale comment in the two-run test.
`mvn spotless:check` is clean, xtable-core unit tests are 494 passing, and `ITDeltaConversionSource` is 19 passing.
On your two open questions from the last round: leaving `DeltaIncrementalChangesState` with the two listings is fine by me, the single pass isn't worth reimplementing Delta's action parsing for a cost that measures inside noise. And yes please, a separate issue for the same treatment on `DeltaKernelConversionSource` would be great, this PR is big enough.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [860] Reuse the Delta snapshot across an incremental backlog [incubator-xtable]
Posted by "vinishjail97 (via GitHub)" <gi...@apache.org>.
vinishjail97 commented on PR #861:
URL: https://github.com/apache/incubator-xtable/pull/861#issuecomment-5197807893
@AadhiKat
Before I approve, I tried reproducing this on S3 first and couldn't a ~10 TB delta table showed no per-commit penalty, which I take to mean checkpoint part count drives it rather than bytes. Could you share a bit about the table behind the 454-part / ~6.5 GB checkpoint (schema width, how long the history runs between checkpoints, Delta and Spark versions)?
Mainly I'd like to know whether the 2-3 min per commit went into getLogSegmentForVersion's _delta_log listing or the checkpoint Parquet read after it -- reading 2.4.0, the only expensive call in here looks like snapshot.metadata()?
The direction of the optimization looks right either way; but want to see validate it thoroughly once before approving because it's a change in the xtable core path used by users in production.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [860] Reuse the Delta snapshot across an incremental backlog [incubator-xtable]
Posted by "vinishjail97 (via GitHub)" <gi...@apache.org>.
vinishjail97 commented on code in PR #861:
URL: https://github.com/apache/incubator-xtable/pull/861#discussion_r3760833145
##########
xtable-core/src/main/java/org/apache/xtable/delta/DeltaConversionSourceProvider.java:
##########
@@ -27,16 +27,31 @@
/** A concrete implementation of {@link ConversionSourceProvider} for Delta Lake table format. */
public class DeltaConversionSourceProvider extends ConversionSourceProvider<Long> {
+ /**
+ * When true (default), an incremental backlog carries the table metadata across commits instead
+ * of reconstructing the Delta snapshot (a full checkpoint read) for every commit. Set to false to
+ * restore the per-commit snapshot reconstruction.
+ */
+ static final String REUSE_METADATA_ACROSS_COMMITS =
Review Comment:
Two small things on the config.
**Pattern and visibility.** `DeltaConversionTargetConfig.USE_KERNEL` is `public static final` and its parsing lives in a `@Value` class with `fromProperties`, which `HudiSourceConfig` and `ParquetSourceConfig` also follow. This one is package-private and parsed inline in the provider, so a Java embedder cannot reference the constant, and the parsing is not reachable from a unit test. A small `DeltaConversionSourceConfig` mirroring `DeltaConversionTargetConfig` would keep the two Delta configs symmetric.
**No test covers this path.** The three new ITs set `reuseMetadataAcrossCommits(...)` through the builder, so the property itself is never read in any test: `"false"`, mixed case, an absent key, and a null `additionalProperties` are all uncovered. A four-case unit test is cheap and would catch a default flip, which is the one thing about this change an operator would notice.
##########
xtable-core/src/main/java/org/apache/xtable/delta/DeltaActionsConverter.java:
##########
@@ -117,12 +154,21 @@ static String getFullPathToFile(Snapshot snapshot, String dataFilePath) {
* is present
*/
public String extractDeletionVectorFile(Snapshot snapshot, AddFile addFile) {
+ // Check for the deletion vector before resolving the base path, so an add file without one
Review Comment:
Minor: this null-checks `deletionVector()` and then delegates to a method that null-checks it again. The hot path in `DeltaConversionSource` now uses the `String` overload, so the early return does not save the snapshot access it describes for any caller that matters. Could this just be `return extractDeletionVectorFile(tableBasePath(snapshot), addFile);` and let the single check live in one place.
##########
xtable-core/src/main/java/org/apache/xtable/delta/DeltaConversionSource.java:
##########
@@ -78,6 +79,14 @@ public class DeltaConversionSource implements ConversionSource<Long> {
private final String tableName;
private final String basePath;
+ // When enabled (default), the table metadata is carried across an incremental backlog instead
+ // of reconstructing the Delta snapshot (a full checkpoint read) on every commit. The baseline
+ // comes from one snapshot read; a commit carrying a Metadata action refreshes the cache from
+ // the action itself (see getTableChangeForCommit).
+ @Builder.Default private final boolean reuseMetadataAcrossCommits = true;
+
+ private Metadata cachedMetadata;
Review Comment:
Non-blocking, but worth one line: `resetState()` rebuilds `deltaIncrementalChangesState` and leaves `cachedMetadata` populated, so the "one snapshot read supplies the baseline" invariant holds per source *instance* rather than per *backlog*.
Inside this repo that is fine. `ConversionController.sync()` and `syncTableAcrossCatalogs()` both build a fresh source and close it, so a source only ever serves one backlog. But `ConversionSource` is a public `@Evolving` SPI and `DeltaConversionSource.builder()` is public, so an embedder can hold one instance across syncs. If a later `getCommitsBacklog` resolves to an *earlier* start version (a target that fell behind, a retry after a rollback), every commit from that start until the next `metaData` action would be converted with the newer schema, silently and without a snapshot read to correct it. Clearing it costs exactly one snapshot read per backlog, which is the cost the design already budgets for:
```java
private void resetState(long versionToStartFrom) {
cachedMetadata = null;
deltaIncrementalChangesState = ...
```
Related and cheap: the cache is only correct if `getTableChangeForCommit` is called with non-decreasing versions. That holds today (`ExtractFromSource` maps over `getVersionsInSortedOrder()` in a sequential stream, and `TableFormatSync` drains that one iterator), but the SPI does not state it and it is now load-bearing for correctness rather than just for performance. A `Preconditions.checkArgument(version >= lastVersionSeen, ...)` here would pin both this and the case above.
##########
xtable-core/src/main/java/org/apache/xtable/delta/DeltaIncrementalChangesState.java:
##########
@@ -61,6 +65,19 @@ public DeltaIncrementalChangesState(DeltaLog deltaLog, Long versionToStartFrom)
maxSeenVersion =
maxSeenVersion == null ? versionNumber : Math.max(maxSeenVersion, versionNumber);
}
+ // Carry each commit's file modification time alongside its actions. This is the same clock
+ // Snapshot.timestamp() reads and the same one getCommitsBacklog matches the persisted
+ // watermark against, so it stays consistent regardless of the writer's clock. The log has
+ // already been listed once for getChanges above; this second pass reads no file contents.
+ Iterator<Tuple2<Object, FileStatus>> logFiles =
+ JavaConverters.asJavaIteratorConverter(
+ deltaLog.getChangeLogFiles(versionToStartFrom, false))
Review Comment:
The flag javadoc says setting it to false restores "the per-commit snapshot reconstruction", which is accurate, but the flag does not revert this listing. `getChangeLogFiles` runs unconditionally, so a user who flips the flag off still pays the second log listing per backlog, and `latestCommitTime` still comes from the mtime map rather than `snapshot.timestamp()`.
If the flag is meant as an escape hatch for a production incident, it is more useful if it reverts all three behaviour changes. Otherwise, could the javadoc state explicitly what it does and does not revert, so an operator reaching for it knows what they are getting.
##########
xtable-core/src/main/java/org/apache/xtable/delta/DeltaTableExtractor.java:
##########
@@ -79,6 +80,44 @@ public InternalTable table(Snapshot snapshot, String tableName) {
.build();
}
+ /**
+ * 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(
Review Comment:
Could the `Snapshot` overload delegate to this one instead of duplicating the schema, partition, layout and builder block:
```java
public InternalTable table(Snapshot snapshot, String tableName) {
requireMetadata(snapshot, tableName);
return table(
snapshot.metadata(),
snapshot.deltaLog(),
tableName,
Instant.ofEpochMilli(snapshot.timestamp()));
}
```
The two blocks have already drifted: `requireMetadata` includes the version in its message and the new guard drops it, so the same missing-schema condition reports differently depending on which entry point you came through. Delegating removes about 20 duplicated lines and stops a future field addition from landing in only one path. I did check field by field that the two builders currently set the same eight properties, so this is style rather than a bug.
##########
xtable-core/src/test/java/org/apache/xtable/delta/ITDeltaConversionSource.java:
##########
@@ -829,4 +836,231 @@ private boolean checkIfFileIsRemoved(String activePath, TableChange tableChange)
.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())
+ .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())
+ .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: one snapshot
+ // reconstruction per commit.
+ verify(spiedDeltaLog, times(commitsBacklog.getCommitsToProcess().size()))
+ .getSnapshotAt(anyLong(), any());
+ }
+
+ @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())
+ .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())
+ .build();
+ List<Long> secondVersions =
+ secondRun
+ .getCommitsBacklog(
+ InstantsForIncrementalSync.builder().lastSyncInstant(watermark).build())
+ .getCommitsToProcess();
+
+ // The second run must not re-process the already-synced tail. (At most the boundary commit may
+ // reappear, since CommitInfo timestamps are the writer's wall-clock rather than the commit
Review Comment:
The rationale here is stale after the round-2 change. The watermark no longer comes from `CommitInfo`, it comes from the commit file's mtime, so the writer's clock is not the reason a boundary commit can reappear.
The actual reason is worth writing down, because it is real: `getCommitsBacklog` resolves the watermark through `DeltaHistoryManager.getActiveCommitAtTime`, which compares against timestamps passed through `monotonizeCommitTimestamps`, while what we persist is the raw `FileStatus` mtime. Monotonization only ever increases a timestamp, so if commit N's mtime tied with or went behind its predecessor, `monotonized(N) > raw(N)` and the lookup resolves to a version before N, which puts N and possibly its predecessors back in the backlog.
Reprocessing is safe here since Delta commits are self-describing, and this is unchanged from main because `Snapshot.timestamp()` is also a raw `LogSegment.lastCommitTimestamp`. So no regression. I would leave the `<= 1` bound as it is and only fix the explanation. Tightening it to `isEmpty()` would make the test flaky on filesystems with coarse mtime granularity, which is the same monotonization case.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [860] Reuse the Delta snapshot across an incremental backlog [incubator-xtable]
Posted by "vinishjail97 (via GitHub)" <gi...@apache.org>.
vinishjail97 merged PR #861:
URL: https://github.com/apache/incubator-xtable/pull/861
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org