v3 deletion-vector MERGE grows executor memory without bound under high-churn streaming upserts

Apache Iceberg version

1.11.0 (latest release)

Query engine

Spark

Please describe the bug 🐞

v3 deletion-vector MERGE (merge-on-read) grows executor memory without bound under
high-churn streaming upserts until executors OOM; reproduced on vanilla Spark 4.0.2 +
Iceberg 1.11.0 + Glue/S3 with NO compaction (follow-up to #17209 with the OSS repro
requested there)

Spark 4.0.2 (Scala 2.13), official apache/spark:4.0.2 image, Spark on Kubernetes
(Amazon EKS, 6 x m5.2xlarge), GlueCatalog + S3FileIO (plain S3 bucket, SSE-KMS),
s3a:// streaming checkpoints. No compaction of any kind.

Please describe the bug

Summary. A continuous Structured Streaming CDC "mirror" MERGE INTO (upsert keyed
on the bucket column) against a format-version 3 merge-on-read table exhibits
executor memory growth batch over batch until executors are container-OOM-killed
(exit 137) and the job dies. The identical job against a format-version 2 table is
stable for hours (shown at length in #17209 on EMR; being re-confirmed on this stack,
see "control run" below).

Workload (fully scripted, see Reproduction):

Table: PARTITIONED BY (bucket(64, account_id)), MoR
(write.{delete,update,merge}.mode=merge-on-read, write.merge.distribution-mode=hash,
write.spark.fanout.enabled=true), format-version=3 -> deletion vectors.
Feed: DMS-like CDC over Kafka at ~87k msg/s; 80% of changes hit 100k hot keys (of 2M),
~85% updates / ~15% deletes after first insert -> nearly every batch is WHEN MATCHED,
i.e. sustained row-level deletes against already-written data files across all 64 buckets.
Per 60s micro-batch: dedup to latest change per key (row_number() over (partition by account_id order by seq desc)), then
MERGE INTO mirror ON account_id with WHEN MATCHED DELETE / UPDATE, WHEN NOT MATCHED INSERT.
Sizing: 6 executors x 5 cores x 16g heap (+10% overhead -> 17.6 GiB pod limit), 4g driver.

Observed (v3, fresh table, single job on an otherwise idle cluster)

batch	rows	merge time
1-8	3.7-5.2M	26-32s
9	4.4M	59s
10	5.2M	73s
...	growing	150-220s stalls
~batch 25+ (t+27min, ~130M rows merged)	-	executors OOM in a loop, job dies

Driver log at the end (executor IDs reached 19 - i.e. 13 replacements had already been
OOM-killed and respawned before the context gave up):

ERROR TaskSchedulerImpl: Lost executor 13 on 172.31.87.124:
The executor with id 13 exited with exit code 137(SIGKILL, possible container OOM).
...
ERROR MicroBatchExecution: Query streaming-cdc-mirror-accounts_mirror_dv terminated with error
org.apache.spark.SparkException: Job 51 cancelled because SparkContext was shut down

Spark UI symptoms while degrading (same as on EMR in #17209): tasks with very high
task deserialization time on the MERGE write stage, executors freezing >120s (GC) and
being evicted for missed heartbeats, then 137s.

Control run (v2). Same binary, same feed, same sizing, same cluster, only
fv=2: stable indefinitely, batch time flat (~40-55s), zero executor losses, while
carrying ~10x more delete files than the v3 table had when it started dying
(detailed v2/v3 table in #17209; re-run on this OSS stack ongoing, will attach the
event logs of both runs).

Proposed mechanism

This cames from Fable reading the whole source code, I need to study what is proposed below.

SparkPositionDeltaWrite makes the v3 DV write a read-modify-write whose working set
grows with the table, while the v2 path is a bounded blind append:

Driver side, every batch: broadcastRewritableDeletes() calls
scan.rewritableDeletes(useDVs=true) and broadcasts a
Map<String, DeleteFileSet> of every scanned data file that has existing deletes ->
its delete files. With DVs every previously-touched data file carries one, and a
hot-key upsert keyed on the bucket column touches all 64 buckets every batch, so this
map converges to "all data files in the table" and grows monotonically as appends add
files (no compaction to consolidate). A monotonically growing broadcast is consistent
with the growing task-deserialization times we observe.
Executor side: for DVs the writer is PartitioningDVWriter with a
PreviousDeleteLoader: for each data file a task deletes from, it loads that file's
existing DV bitmap, unions the new positions, and keeps the merged
PositionDeleteIndex in memory until the task completes (one live bitmap per touched
data file, fanout-style, no spill). Files-touched-per-batch grows with the table ->
executor heap grows batch over batch -> GC stalls -> heartbeat evictions -> container OOM.
v2 path for comparison: with rewritableDeletes == null the position-delete
writers (ClusteredPositionDeleteWriter / FanoutPositionOnlyDeleteWriter) just
append delete files bounded by the current batch's deletes. No prior state is
loaded, which matches v2 being stable while accumulating far more delete files.

In other words: DVs move work from readers to writers by design (supersede-on-write),
but the Spark implementation holds the entire supersede working set in memory, per
executor, with no bound and no spill, and additionally re-broadcasts the full
file->deletes map every micro-batch. Under a streaming upsert workload whose scan
touches most files, that working set is O(table data-file count), not O(batch).

Expected behavior

v3 DV writes should have memory consumption comparable to (or at least bounded like) v2
positional-delete writes for the same MERGE workload - or expose a knob to bound/spill
the DV merge working set. Doubling executor memory only delays onset (measured on EMR:
56 GB -> 120 GB moved failure from ~20 to ~45 min).

V2 spark ui with expected behaviour.

v3 spark ui ( note the huge spike when all the executors get killed).

Reproduction

Everything is public in https://github.com/aws-samples/iceberg-streaming-examples
a CDC feed generator (KafkaCDCSimulator: unkeyed, monotonic seq, 80/20 hot keys, ~85% U / 15% D) and the streaming MERGE job (SparkStreamingCDCMirror, shared SQL in CdcSql).

Option A - Spark on EKS (what produced the numbers above): scripts/eks/ stands up
everything (or adopts an existing cluster/bucket); then

PRODUCER_CLASS=com.aws.emr.kafka.KafkaCDCSimulator ./scripts/eks/06-run.sh producer rate=100000
JOB_CLASS=com.aws.emr.spark.cdc.SparkStreamingCDCMirror \
EXECUTORS=6 EXECUTOR_CORES=5 EXECUTOR_MEMORY=32g \
  ./scripts/eks/06-run.sh ingest table=mirror_v3 fv=3 startingoffsets=earliest
# control: identical, with table=mirror_v2 fv=2

Option B - single machine: docker compose up -d (Kafka), then
scripts/run-local.sh with the same fv= knob; the memory growth pattern is visible at
smaller scale in the executor metrics.

Iceberg/Kafka/hadoop-aws jars and versions are pinned in scripts/eks/Dockerfile
(iceberg 1.11.0, spark-sql-kafka 4.0.2, kafka-clients 3.9.1, hadoop-aws 3.4.1,
awssdk bundle 2.44.4). Spark event logs of the failing v3 run and the stable v2 control
are preserved and can be attached on request.

Related: #11122 (v3 position-delete design), #15924 (in-memory DV blob handling in
path rewrite - same "whole DV in memory" pattern in a different code path).

Willingness to contribute

I can contribute a fix for this bug independently
I would be willing to contribute a fix for this bug with guidance from the Iceberg community
I cannot contribute a fix for this bug at this time

Have been studying a little bit.

From DV write path.

This needs to be merged, I can collaborate if needed. Core: Stream DV Puffin rewrite in RewriteTablePathUtil #15927 . Although not used in this codepath.... it is going to be useful.
For big blobs maybe we need to think on "Streaming blob APIs" ( there is no api surface for that now).
PuffinReader.rawBlob(BlobMetadata) → an InputStream over the compressed blob bytes, no decompression
PuffinWriter.writeRaw(BlobMetadata, InputStream) → copies bytes through, records metadata for the footer, no re-compression
For the driver thing... we should think on trimming state but ideally we should move all that work to the executors. Will think about it.

There is an open PR for the streaming part: #16960

I think that code is just used in catalog.system.rewrite_table_path procedure, not in the MERGE of the Job.

So MERGE-time DV rewriting = load the previous bitmap, union it with the new deletes, write a fresh Puffin file, and drop the old DV in the same RowDelta commit. In Spark this is driven by SparkPositionDeltaWrite wrapping BaseDVFileWriter. That's the streaming/MERGE path.

I think that code is just used in catalog.system.rewrite_table_path procedure, not in the MERGE of the Job.

I just mentioned this because it solves the same issue as #15927 but still open. Sorry if it was not clear enough.

I have pushed the driver optimization, although does not solve the issue... it is a simple optimization that I think that it is worthy to push ( #17281)

Doing some more testing I see that if I run compaction "n" batches I can go further... will think further on executor optimization.

But any guidance is more than welcome here.

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):

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.

Well finally after a lot of time I got it.

VectorizedArrowReader$RowIdVectorReader.read() allocates a fresh BigIntVector from the shaded-Arrow root allocator on every batch:

BigIntVector rowIds = allocateBigIntVector(ROW_ID_ARROW_FIELD, numValsToRead);

The reuse parameter is ignored, the vector is returned inside a VectorHolder, the reader keeps no reference to it, and RowIdVectorReader.close() is a no-op (// don't close result vectors as they are not owned by readers). Nothing downstream
closes the holder's vector, so its ArrowBuf refcount never reaches zero. The leaked buffers pin netty pooled-arena chunks (4 MiB each), so the JVM's direct-memory accounting grows monotonically. LastUpdatedSeqVectorReader.read() has the identical
pattern for _last_updated_sequence_number. The helper vectors those readers do close per batch (positions, ids, seqNumbers) are handled correctly - only the result vectors leak.

Because _row_id and _last_updated_sequence_number are projected for row-lineage preservation on v3 MERGE/UPDATE/DELETE reads, any long-running v3 row-level workload with vectorized reads leaks until the executor dies. This is invisible on short queries and looks like "v3 needs more memory" on long ones.

Evidence

Cluster: EKS, Spark 4.0.2, 6 executors x 8 cores, 32g heap + 8g overhead (40Gi pod limit), ~90k rows/s CDC feed, MERGE into a bucket(64) MoR v3 table.

1. JVM Native Memory Tracking (-XX:NativeMemoryTracking=summary, sampled every
2 min): the Other category (direct ByteBuffers) grows linearly while every other
category is flat. Heap stays at 32.0g committed throughout, GC 1.31g, metaspace 128MB:

time                  workingSet  anon(GiB)  file(GiB)  NMT-Other(MB)
t+05min                  33.7      33.5       0.18         157
t+20min                  35.0      34.7       0.20         337
t+40min                  36.1      35.8       0.15        1229
t+60min                  37.4      37.1       0.20        2638
t+80min                  38.8      38.4       0.20        4010
t+87min (pod killed)     39.9      39.8       0.00        5446

cgroup v2 memory.stat confirms the growth is anon, not page cache - the kernel
cannot reclaim it, and the pod is OOM-killed at the 40Gi limit (~t+90min, reproduced
4 times; wave period matches the leak rate).

2. Capped repro: with -XX:MaxDirectMemorySize=4g, all six executors fail at the
cap within seconds of each other, ~t+78min. The JDK runs System.gc() + retry before
throwing, so the buffers are provably still reachable (a true leak, not lazy cleanup):

java.lang.OutOfMemoryError: Cannot reserve 4194304 bytes of direct buffer memory
    (allocated: 4292201200, limit: 4294967296)
  at java.base/java.nio.Bits.reserveMemory(Bits.java:178)
  at java.base/java.nio.ByteBuffer.allocateDirect(ByteBuffer.java:332)
  at org.apache.iceberg.shaded.io.netty.buffer.PoolArena$DirectArena.newChunk(PoolArena.java:737)
  ...
  at org.apache.iceberg.shaded.org.apache.arrow.memory.BaseAllocator.buffer(BaseAllocator.java:280)
  at org.apache.iceberg.shaded.org.apache.arrow.vector.BaseValueVector.allocFixedDataAndValidityBufs(BaseValueVector.java:224)
  at org.apache.iceberg.shaded.org.apache.arrow.vector.BaseFixedWidthVector.allocateNew(BaseFixedWidthVector.java:308)
  at org.apache.iceberg.arrow.vectorized.VectorizedArrowReader.allocateBigIntVector(VectorizedArrowReader.java:880)
  at org.apache.iceberg.arrow.vectorized.VectorizedArrowReader$RowIdVectorReader.read(VectorizedArrowReader.java:758)
  at org.apache.iceberg.spark.data.vectorized.ColumnarBatchReader$ColumnBatchLoader.readDataToColumnVectors(ColumnarBatchReader.java:81)
  at org.apache.iceberg.spark.data.vectorized.ColumnarBatchReader.read(ColumnarBatchReader.java:56)
  at org.apache.iceberg.parquet.VectorizedParquetReader$FileIterator.next(VectorizedParquetReader.java:145)
  ...

3. Control: the same job against a v2 table (no row lineage -> RowIdVectorReader
never instantiated) ran 6h46m with flat memory on identical sizing.

4. Fix validation: rerunning the identical workload with a patched runtime in which
the two lineage readers own their result vector (allocate lazily, reuse across batches,
release in close() - the fix proposed below) eliminates the leak entirely:

	unfixed	fixed
NMT `Other` (direct buffers)	4,010 MB at t+80min, climbing	flat 233-242 MB for 5h49m
cgroup `anon`	38.4 GiB at t+80min, climbing	flat 34.93 GiB for 5h49m
executor OOM kills	first wave t+90min	zero (all six originals alive)
run outcome	driver dead t+92min	healthy at 240+ batches, ongoing

Same cluster, same feed, same sizing, same Spark confs (no direct-memory cap, no extra overhead); the only delta is the reader fix. Direct-memory usage decouples completely from run duration, confirming the per-batch result-vector allocation as the sole leak.

Expected behavior

Long-running vectorized reads of v3 tables should have bounded direct-memory usage. The lineage metadata readers should either reuse their result vector across batches (like PositionVectorReader does when passed a non-null reuse) and release it in close(), or close the previous batch's vector before allocating the next.

Workarounds until fixed

-XX:MaxDirectMemorySize=<n> on executors converts the silent container kill into recoverable task failures (Spark replaces the executor; ingest continues), at the cost of periodic executor churn.
Non-vectorized reads avoid the leaking code path.

Fixed in #17296

@RussellSpitzer

As I know you where interested @RussellSpitzer, just take a look, took me a while as I did many experiments to track this down.

#17296 Fixes it.
#17281 Driver optimization, maybe it is not worthy as will add some memory overhead on the driver ( but in long jobs will make sense).
#17287 Optimizations on on-heap memory while writing DV, didn´t solve the issue ( as we had off-heap leak), but I think that is worthy to have ( same throughput in tests).

Right now the job has been running for more than 6 hours without any failure.

Navigation Menu

Repository navigation

v3 deletion-vector MERGE grows executor memory without bound under high-churn streaming upserts #17241

Apache Iceberg version

Query engine

Please describe the bug 🐞

Please describe the bug

Proposed mechanism

Expected behavior

Reproduction

Willingness to contribute

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

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

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

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

Summary table

Evidence

Expected behavior

Workarounds until fixed

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

v3 deletion-vector MERGE grows executor memory without bound under high-churn streaming upserts #17241

Description

Apache Iceberg version

Query engine

Please describe the bug 🐞

Please describe the bug

Proposed mechanism

Expected behavior

Reproduction

Willingness to contribute

Activity

Neuw84 commented on Jul 16, 2026

ldudas-marx commented on Jul 16, 2026

Neuw84 commented on Jul 16, 2026

ldudas-marx commented on Jul 16, 2026

Neuw84 commented on Jul 17, 2026

Neuw84 commented on Jul 18, 2026

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

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

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

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

Summary table

Neuw84 commented on Jul 19, 2026

Evidence

Expected behavior

Workarounds until fixed

Neuw84 commented on Jul 19, 2026

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions