[860] Reuse the Delta snapshot across an incremental backlog

[AadhiKat]

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.

[rangareddy]

LGTM

[vinishjail97]

@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.

[AadhiKat]

@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.

DeltaRepro.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();
    }
}

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