[860] Reuse the Delta snapshot across an incremental backlog - #861
[860] Reuse the Delta snapshot across an incremental backlog#861AadhiKat wants to merge 1 commit into
Conversation
a504af2 to
446797c
Compare
| actionsConverter.convertToFileFormat(cachedSnapshot.metadata().format().provider()); | ||
| } | ||
| Snapshot snapshotAtVersion = cachedSnapshot; | ||
| InternalTable tableAtVersion = cachedTable; |
There was a problem hiding this 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?
There was a problem hiding this 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.
| } | ||
|
|
||
| @Test | ||
| public void getTableChangeForCommitReconstructsSnapshotOnceForAppendOnlyBacklog() { |
There was a problem hiding this 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.
There was a problem hiding this 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.
getTableChangeForCommit reconstructed the table snapshot (getSnapshotAt) for every commit, re-reading the Delta checkpoint each time. That is cheap on a small checkpoint but dominates incremental catch-up on tables with a large checkpoint. Cache the snapshot (and the table/file format derived from it) and reload it only when a commit carries a Metadata or Protocol action. Otherwise the snapshot is used only for schema, partitioning, file format and base path, which do not change between commits. latestCommitTime feeds the sync watermark, so it can't come from the reused snapshot or the watermark stops advancing. Take it from the commit's own CommitInfo (already in the incremental actions), falling back to the snapshot timestamp only if a commit has no CommitInfo. Signed-off-by: Aadhithya Hari <aadhikat@gmail.com>
446797c to
388bb0f
Compare
3 participants
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:
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.
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.