Neuw84 commented on issue #17241:
URL: https://github.com/apache/iceberg/issues/17241#issuecomment-5014879755
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**:
```java
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
--
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]
