[Neuw84]

Closes #18257.

What

BaseDeleteLoader.loadEqualityDeletes caches each equality delete file's rows, but it builds the merged StructLikeSet again in every scan task. On a table with many equality deletes, those inserts dominate the task, and every task on an executor repeats them for the same delete files.

This change can cache the merged set through the loader's existing canCache / getOrLoad hooks. It is opt-in:

• The switch: BaseDeleteLoader.cacheEqualityDeleteSets(), false by default, so every loader keeps today's one-entry-per-delete-file behavior unless it enables it. Spark 4.2 enables it with spark.sql.iceberg.executor-cache.equality-delete-sets.enabled (default false), read when the executor cache is created, like the other executor-cache settings. Merged sets pay off when many tasks reference the same delete files. They duplicate reads and memory when groups only partially overlap: a delete file in several different groups is held once per group (see the review discussion).
• The key: the projection plus the sorted delete file locations. Tasks with exactly the same delete files share one set. A task whose file list differs in any file (sequence numbers can exclude deletes for newer data files) gets its own entry.
• Reading the files: when the merged set is cached, its files are read directly, not through their per-file entries. A cache load must not start other cache loads, since the delete worker threads would wait on the cache this load holds, and the merged set supersedes those entries anyway.
• When the set isn't cacheable, or the switch is off: nothing changes; it takes the existing path.
• Sizing: the size handed to the cache is the sum of the per-file estimates (the convention estimateEqDeletesSize already uses), so max-entry-size / max-total-size decide as they do today.
• Thread safety: the set isn't mutated after it's built, and StructLikeSet.contains is safe for concurrent readers.

Only Spark 4.2 is changed here; the older Spark versions can follow as backports.

Results

This is a Spark 4.1.3 cold MERGE INTO (CDC: 42.0 M updates, 11.0 M deletes, 4.6 M inserts) into a v2 table with 432 M rows, on 8 × m5.4xlarge, with merged-set caching enabled:

• Timing: 162.1 s on 1.11.0, 74.6 s with this change (2.17×).
• Where it comes from: the two scan stages that apply the equality deletes drop from 118 s and 96 s to 10 s and 32 s.
• Correctness: both runs leave the same 425,665,996 rows.
• Cache limits: the executor cache limits were raised to 2 / 4 GiB on both runs so the merged set fits (see the open question on the issue).

These runs predate the opt-in switch; with the switch on, the code path is the one measured.

Related

On the same workload, the next-largest cost in the scan was building position-delete indexes: Deletes.toPositionIndexes hashed and looked up the data-file path for every row. That was 20.5 % of the target-scan stage in JFR. #17864 has since fixed it on main (not in a release yet). We backported it to 1.11.0 for our measurements, and the two changes are independent: this one is equality deletes only, #17864 is v2 position delete files only.

#16956 proposes a snapshot-wide EqualityDeletes structure, which also covers data files whose delete file groups only partially overlap.

Tests

TestBaseDeleteLoaderEqualityDeleteCache:

• the merged set is built once per group of files;
• different files, or different equality fields, get their own set;
• without caching every call builds its own set;
• a merged set too large for the cache falls back to the per-file entries;
• merged sets aren't cached unless the loader opts in (per-file entries as on main).

TestSparkExecutorCache (Spark 4.2): the new property defaults to false and is read from the session.

TestGenericReaderDeletes passes unchanged.

AI Disclosure

Model: Claude Opus 5.5
Platform/Tool: Claude Code
Human Oversight: reviewed

[Neuw84]

A second, independent measurement on today's code, for both engines. It's the same CDC MERGE INTO as in the description: a v2 table with 432 M rows, a change batch of 42.0 M updates, 11.0 M deletes and 4.6 M inserts, and 425,665,996 rows after. It runs on 8 executors × 14 cores on m5.4xlarge, reading from S3. Stock Apache Iceberg 1.11.0 is compared with the same runtime jar carrying this change (plus a backport of #17864, which the description found independent). The executor cache limits are 2 / 4 GiB on every run.

Engine	Iceberg	MERGE	Target scan (the stage that applies the deletes): wall / task time / CPU / GC
OSS Spark 4.1.3	1.11.0	189.3 s, 164.1 s	138 s, 118 s / 11,655 s, 11,069 s / 4,080 s, 4,051 s / 4,850 s, 4,655 s
OSS Spark 4.1.3	with this PR	85.9 s, 86.5 s	41 s, 41 s / 3,837 s, 3,952 s / 1,914 s, 1,911 s / 224 s, 246 s
Spark with a columnar engine plugin (vecruntime)	1.11.0	157.7 s, 156.6 s	112 s, 115 s / 10,118 s, 10,434 s / 4,142 s, 4,458 s / 3,258 s, 3,294 s
Spark with a columnar engine plugin (vecruntime)	with this PR	80.6 s, 84.5 s	40 s, 40 s / 4,136 s, 4,098 s / 2,113 s, 2,099 s / 267 s, 280 s

The two runs of each combination were interleaved.

• The MERGE is 1.9-2.2x faster in both engines. The table's contents after the merge are the same in every run.
• Where it comes from: the target scan does about half the CPU and a twentieth of the GC. The merged StructLikeSet is built once per executor rather than once per task, so executors stop allocating it again and again.
• The same holds for a different execution engine: the plugin replaces Spark's operators above the scan, but the delete application is Iceberg's reader in both, so the gain doesn't depend on Spark's own operators.

[kinolaev]

Hi @Neuw84, thanks for the PR!

I've run into a similar problem: building the equality delete set dominates the task. But in my case (CDC with https://github.com/memiiso/debezium-server-iceberg), there are many more data and delete files than the 64 and 616 in your benchmark, and the data files reference overlapping sets of delete files. As far as I understand, with this PR applied, a delete file that participates in several sets is loaded and cached once per set. For example, here are the ID ranges of the files from three commits:

deletes-01 1000-1999
   data-01 1000-1999

deletes-02 1000-1799
   data-02 1000-1799

deletes-03 1200-1999
   data-03 1200-1999

data-01 references deletes-02 and deletes-03, and data-02 references deletes-03, so deletes-03 is loaded and cached twice (under the eq-delete-set|deletes-03 and eq-delete-set|deletes-02|deletes-03 keys), while each cached set is used only once. Without this PR, deletes-03 is loaded only once, and each merged set is discarded after it has been applied to its data file.

So caching per location list works well when data files reference the same set of delete files, but it duplicates reads and memory for the shared files when two sets differ even by one file.

Another concern is the key length. Even in your case, with 616 delete files, the key for the full set is probably larger than 64 KB, and keys are built and hashed for every split. This cost is negligible compared to rebuilding the delete set, but it can become significant when a set contains more delete files.

I've been working on a related approach in #16956. I'd be happy if you could take a look and test it in your environment. It introduces an org.apache.iceberg.data.EqualityDeletes structure that can be shared by all equality deletes within a snapshot.

For now, the PR uses it only to share equality deletes within a Spark task, but it might be a good idea to make it thread-safe and cache it per executor under a namespace.table.snapshot key, so that each Spark task only adds the equality deletes that haven't already been loaded by other tasks. It could be a great combination of the two approaches: your per-executor caching, plus sharing deletes between data files that reference different sets of delete files.

---

If it's easier to test with prebuilt artifacts, I publish iceberg-spark-runtime jars and container images built from Iceberg releases with #16956 and a few of my other fixes applied. The links below show exactly what's included on top of each release:

• 1.11.0
• 1.12.0

[Neuw84]

Hi @kinolaev,

This makes #9563 to also cache equality delete files per executor not per task. Note that the test I was testing it is heavy ( millions of position and equality deletes) but the performance gain is more than 2x and I think is a worthy option to have under executor cache locality umbrella.

[kinolaev]

I agree that caching equality deletes on the executor for the entire duration of a job (or longer) is the right long-term goal. However, with the current implementation, performance improves when delete file sets are identical, but degrades when they are only partially overlapping.

Could you consider making this behavior opt-in? Similar but non-identical delete file sets are quite common in CDC scenarios, so having a configuration flag would help prevent potential regressions.

[Neuw84]

Thanks @kinolaev, both points are fair. Pushed 30d281b:

• Opt-in: merged-set caching is now behind BaseDeleteLoader.cacheEqualityDeleteSets() (default false); Spark 4.2 enables it with spark.sql.iceberg.executor-cache.equality-delete-sets.enabled (default false). Without it the loader behaves exactly as on main, one cache entry per delete file, so the partially-overlapping case you describe doesn't regress. The docs entry calls out that trade-off.
• Key length: the key is now a SHA-256 digest of the projection and the sorted locations, so it's constant-size regardless of how many delete files a task references.

[Neuw84]

Small follow-up to my previous comment: I've reverted the hashed key in c78e1d3 and kept the plain one. As you said, building the key is negligible next to building the delete set, and with merged-set caching now opt-in your main concern, partially overlapping delete file sets, is covered without it. That keeps the PR smaller.