Spark: Broadcast slim rewritable delete metadata in position delta wr…

[Neuw84]

Position delta writes (MERGE/UPDATE/DELETE in merge-on-read mode) broadcast a map of every scanned data file with position deletes to its full DeleteFile metadata.

Executors only need enough of it to locate and read the previous delete content (location, format, sizes, key
metadata, DV offsets) and to identify files back to the driver; partition data, column bounds and counts ride along only to be echoed back for RowDelta.removeDeletes at commit time.

Under streaming upserts against v3 DV tables this map converges to one entry per data file in the table and is rebuilt and re-broadcast every micro-batch, adding driver and executor memory pressure that grows with
table age.

Keep the full map driver-local, broadcast slim copies built via FileMetadata.deleteFileBuilder (with referencedDataFile normalized so file-scoped checks keep working without the file_path column bounds),
and resolve reported rewritten files back to their full metadata at commit time before calling removeDeletes. The resolution step is required for correctness of the slimming: manifest filtering uses the
removed file's spec ID and partition, which slim copies do not carry, so reported rewritten files must never reach the RowDelta directly.

Measured serialized size of the broadcast map (1000 entries, Java serialization - Spark's default broadcast serializer):

  map contents         /               full     /     slim   /      factor
  DV entries (v3)       /              661 B/entry  / 466 B/entry   1.4x
  v2 file-scoped entries with stats / 1069 B/entry  / 581 B/entry /  1.8x

The serialized win is bounded by what slimming cannot remove (the data file location key and the referenced data file location dominate each entry); the deserialized heap saving per executor is larger because the
per-entry object graph (partition data, stats maps, split offsets) is gone. Commit messages shrink correspondingly, since executors echo the slim copies back as rewritten files.

The code can be easily backported to Spark 3.4.

AI Disclosure

Model: Claude Fable
Platform/Tool: Kiro
Human Oversight: reviewed

Relates to #17241

[Neuw84]

Follow-up: extended A/B results (memory, throughput, mitigations) + one correction to the original report

All runs below: same feed (~90k rows/s, ~5.4M rows per 60s micro-batch, 80/20 hot keys over 2M accounts), same sizing (6 executors x 8 cores x 32g heap, 8g driver unless noted), Spark 4.0.2 + Iceberg 1.11.0, GlueCatalog + S3, Spark on EKS (dedicated m5.4xlarge nodes), bucket(64, account_id) MoR mirror, no compaction unless noted.

Memory scaling (v3, no compaction) - confirms "more memory only delays"

pod limit	first synchronized OOM wave	working-set growth
36.0 GiB (32g + 10% default overhead)	34.7 -> 36.0 GiB, monotonic,	~55 MB/min late phase
40 GiB (32g + 8g overhead)	~t+70-80min 13.7 -> 39.1 GiB	~65-95 MB/min

Two details worth calling out:

• Replacement executors die in minutes, not tens of minutes: fresh JVMs reached ~35 GiB working set within 5-7 minutes of start and were killed. The per-batch memory demand is a function of accumulated table state (files carrying DVs x DV cardinality), not process age - this is a growing working set, not a leak.
• Failure waves are synchronized across all executors at batch boundaries (all doing the same DV-merge close() phase).

Operational mitigation validated: delete-threshold compaction bounds the working set

Running rewrite_data_files with options => map('delete-file-threshold','1', 'remove-dangling-deletes','true', 'partial-progress.enabled','true') every 20 batches (inline in foreachBatch, serialized with the MERGE):

• batch-40 rewrite: 1,344 data files -> 64, 1,280 DVs removed, 286 MB rewritten, 0 failures; the compacting batch took 57s vs ~18s baseline and the next batch was immediately back to normal;
• the run sailed past both no-compaction death points with the stream never falling behind (maxOffsetsBehindLatest=0 throughout, even across executor replacements);
• caveats: executors need genuine overhead headroom for the compaction spike (at 10% default overhead, pods sitting at the RSS ceiling got clipped during the rewrite; 8g overhead resolved it), and the driver must be sized for compaction planning (a 4g driver was OOMKilled after 2.5h; 8g was fine). Expect occasional Glue ConcurrentModificationException retries from the rewrite racing the streaming commit - handled by commit retries.

v2 control, long run (same sizing, no compaction): no memory failure, pays in read amplification

6h46min and counting at time of writing:

• zero executor kills, zero replacements; memory flat at 36.4-36.8 GiB the entire run (the same level v3 blew through on its way to OOM in ~75 min);
• but batch time degraded from 20.3s (batch 3) to 106-114s (batches 325-328), with double-size ~10M-row batches - i.e. the stream is at ingest break-even (~90k rows/s processed vs ~90k in) after ~6.5h, purely from accumulated positional-delete read amplification; average cadence over the whole run was ~74.5s per 60s of data.

Summary table

variant	outcome
v3, no compaction, 36 GiB	executor OOM waves from t+33min
v3, no compaction, 40 GiB	executor OOM waves from ~t+75min, dead t+2h39min
v3 + delete-threshold compaction every 20 batches	stable (bounded sawtooth); needs overhead + driver headroom
v2, no compaction, 40 GiB	no memory failure in 6h46min; degrades to ingest break-even at ~6.5h

Driver logs, per-2-minute memory/pod snapshots and Spark event logs (s3) for all runs are preserved and can be attached on request.