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