AadhiKat commented on PR #861:
URL: https://github.com/apache/incubator-xtable/pull/861#issuecomment-5201780563
@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.
<details>
<summary><code>DeltaRepro.java</code></summary>
```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();
}
}
```
</details>
**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 |
--
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]
