Neuw84 opened a new issue, #17209:
URL: https://github.com/apache/iceberg/issues/17209
### Apache Iceberg version
1.10.1
### Query engine
Spark
### Please describe the bug 🐞
### Apache Iceberg version
1.10.1 (Amazon EMR build `1.10.1-amzn-0`)
### Query engine
Spark 4.0.2 (Scala 2.13.16), Amazon EMR Serverless `emr-spark-8.0.0`,
catalog = Amazon S3 Tables (`S3TablesCatalog`)
### Please describe the bug
**Summary**
Running the same continuous (Structured Streaming) CDC "mirror" `MERGE
INTO` (upsert keyed on the bucket
column) against a **format-version 3** merge-on-read table repeatedly
OOM-kills
executors, while the **exact same job** against a **format-version 2** table
is
completely stable. The only difference between the two runs is
`format-version`
(2 vs 3). same code, same Kafka feed, same executor sizing, same table
layout.
On v3 this means deletion vectors; on v2, positional delete files.
**Workload**
- Table: bucketed `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`.
- Per micro-batch (1-minute trigger): dedup CDC changes to latest-per-key,
then
`MERGE INTO target t USING batch s ON t.account_id = s.account_id`
with `WHEN MATCHED ... DELETE / UPDATE` and `WHEN NOT MATCHED ... INSERT`.
- Feed: ~150k msg/s, ~85% updates/deletes concentrated on hot keys
(100k hot of 2M total keys) → almost every batch is `WHEN MATCHED`, i.e. a
steady stream of row-level deletes on already-written data files.
- Executors: 5 × 16 vCPU / 120 GB (also reproduced at 5 × 8 vCPU / 56 GB).
**Observed behavior (same feed, ~same runtime)**
| | v2 (positional deletes) | v3 (deletion vectors) |
|---|---|---|
| executor OOM (exit 137) | 0 | many (25+ per ~45 min) |
| "killed by driver" (no-heartbeat eviction) | 0 | yes (executors freeze
>120s) |
| batch latency | ~55s, stable | 15–20s then repeated 150–220s stalls |
| delete files at time of OOM | ~30k (stable, no failure) | ~4k and OOMing |
| total-records | ~546M | ~76M |
| outcome | runs fine indefinitely | executors die continuously |
Note the v2 table carries **~10x more delete files** than v3 at the point v3
starts OOMing, yet v2 never has memory trouble.
**Log evidence (v3)**
`WARN ExecutorContainerStoreImpl: Executor(s) 1 exited unexpectedly with
exit code 137. ERROR TaskSchedulerImpl: Lost executor 1 ...: Unknown executor
exit code (137) (died from signal 9?)
`
and, for the "killed by driver" stalls:
`WARN HeartbeatReceiver: Removing executor 3 with no recent heartbeats:
154711 ms exceeds timeout 120000 ms INFO EmrServerlessClusterSchedulerBackend:
Requesting to kill executor(s) 3 INFO TaskSchedulerImpl: Executor 3 ... killed
by driver.
`
i.e. an executor GC-freezes for ~154s (matching the 150–220s batch stalls),
then
the driver evicts it for missing heartbeats.
**What we ruled out**
- Not manifest merging: reproduced with `commit.manifest-merge.enabled` both
`true` and `false`.
- Not compaction: OOM happens during deletion-vector *accumulation*, with no
concurrent rewrite/`replace` snapshot at the time.
- Not delete volume per se: v2 handles ~10x more delete files without issue.
- Not (solely) heap size: doubling executor memory 56 GB → 120 GB only
delayed
onset (~45 min), it did not prevent it.
**Hypothesis (please confirm/correct)**
Our reading is that a v3 data file has a single deletion vector that must be
*superseded* on each commit touching that file, i.e. the writer
read-modify-writes
the DV (union new positions into the existing bitmap and rewrite it) for
every
data file the batch touches. With hot keys spread across all 64 buckets, each
1-minute batch touches many data files, so many DVs are rebuilt per commit,
and
this grows as the table accumulates. The v2 path only *appends* a new
positional
delete file per commit (bounded to the current batch's deletes), which stays
memory-cheap. `write.spark.fanout.enabled=true` (many open writers, no sort)
likely compounds the v3 write-side memory.
Is this expected for high-frequency DV rewrites, or is there an
inefficiency/opportunity to spill/bound memory in the Spark DV write path?
**Expected**
v3 deletion-vector writes should not require dramatically more executor
memory
than v2 positional deletes for the same MERGE workload (ideally comparable,
given
DVs are meant to improve read/compaction efficiency).
**Reproduction**
1. Create two identical bucketed MoR tables, one `format-version=2`, one
`=3`.
2. Stream a high-churn CDC feed (mostly updates/deletes on hot keys) and
`MERGE INTO` latest-per-key each micro-batch.
3. v3 executors OOM / get heartbeat-evicted under sustained load; v2 does
not.
**Caveats (not yet isolated)**
Observed on EMR's Iceberg 1.10.1 build with the S3 Tables catalog; we have
not
yet reproduced on vanilla Apache Iceberg with a non-S3-Tables catalog. The DV
write path is in the Spark integration and appears orthogonal to the
catalog, but
flagging in case it matters.
### 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
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
