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