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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
