[SUPPORT] Metadata-table compaction OOM on executors: unbounded heap from BitCaskDiskMap.valueMetadataMap, LazyFileIterator sort, and ExternalSpillableMap.keySet()
- Dominant language
- Java
- Stars
- 6.2k
- Forks
- 2.5k
- Avg merge
- 2d 8h
- Merged PRs (30d)
- 111
Description
## Tips before filing an issue
- Have you gone through our [FAQs](https://hudi.apache.org/docs/faq)? Yes — none of them cover this.
- Join the mailing list to engage in conversations and get faster support at dev-subscribe@hudi.apache.org.
- If you have triaged this as a bug, then file an [issue](https://issues.apache.org/jira/projects/HUDI/issues) directly.
## Describe the problem you faced
The Hudi metadata-table (MDT) `record_index` partition compaction OOMs executors that are configured with 9 GB heap, four concurrent compaction tasks, and `spark.memory.fraction=0.3`. Heap analysis with Eclipse MAT shows three structural consumers totalling ~1.6 GB per task that are **not bounded by the configured compaction memory cap** (`hoodie.memory.compaction.fraction`):
| Consumer | What | Total (4 tasks) |
|---|---|---|
| **B** | `BitCaskDiskMap.valueMetadataMap` — 18.3M `ValueMetadata` entries × ~64 B/entry of `Integer`/`Long` boxing overhead (57.2M `java.lang.Long` instances = ~1.28 GB pure boxing) | **~3.0–3.2 GB** |
| **F** | `LazyFileIterator` materialised a full sorted `ArrayList` (4.58M entries × 4 tasks) at iterator-open time because `ConcurrentHashMap` doesn't preserve insertion-order and the iterator needs forward-only disk seeks | **~513 MB** |
| **E** | `ExternalSpillableMap.keySet()` allocated a transient `HashSet` copying all 6.29M keys solely to seed the `PriorityQueue` in `HoodieSortedMergeHandle`'s constructor | **~256 MB peak** |
4-task total demand: ~6.25 GB on a 9 GB executor heap, which crashes under GC pressure.
Notably, **lowering `hoodie.memory.compaction.fraction` makes it worse** — a lower in-memory cap forces more records to spill to disk, which makes B (the unbounded structure) grow larger.
## To Reproduce
Steps to reproduce:
1. MOR table with `record_index` enabled and ~63M rows across ~10 file groups (~6.29M records per RLI file group).
2. Generate ~8 delta-log files per file group via streaming ingest.
3. Trigger inline MDT `record_index` compaction with executor settings: `--executor-memory 9000m`, `spark.executor.cores=4`, `spark.memory.fraction=0.3`, ParallelGC.
4. Observe executor OOM during stage that runs `HoodieSortedMergeHandle` constructor (Phase 2: collect keys into `PriorityQueue`).
## Expected behavior
MDT compaction should fit within the configured executor heap. The structural consumers B, F, and E should not grow unboundedly relative to the user-configured compaction memory cap.
## Environment Description
- Hudi version: 1.x (current `master` and `master-1.x`)
- Spark version: 3.5
- Storage: S3
- Running on Docker: no
## Additional context
Root cause analysis details:
- **B (boxing overhead in `ValueMetadata`)**: `BitCaskDiskMap.ValueMetadata` declares `sizeOfValue`, `offsetOfValue`, `timestamp` as `Integer`/`Long`. Each entry pays ~64 B of boxing across 18.3M instances = 1.18 GB total. Histogram shows 57M `java.lang.Long` instances.
- **F (sorted `ArrayList` in `LazyFileIterator`)**: `BitCaskDiskMap`'s metadata map is a `ConcurrentHashMap`, which doesn't preserve insertion order. `LazyFileIterator` materialises a sorted copy of all metadata entries at iterator-open time to enable forward-only disk seeks. The sort is unnecessary if the metadata map preserves insertion order, which it can — the disk file is append-only and `filePosition` is strictly monotonic under the existing write lock, so insertion order naturally equals disk offset order.
- **E (`keySet()` HashSet spike)**: `ExternalSpillableMap.keySet()` allocates `new HashSet<>(inMemory.size() + disk.size())` and copies all keys when the map has spilled. `HoodieSortedMergeHandle` only iterates keys once to seed a `PriorityQueue`, so a lazy stream would suffice.
### Proposed fix
Three targeted changes:
1. Unbox `ValueMetadata` primitives (`Integer`/`Long` → `int`/`long`). Saves ~880 MB across 4 tasks.
2. Replace `ConcurrentHashMap` with `LinkedHashMap` + `ReentrantReadWriteLock` for `valueMetadataMap`. Drop the sort and the `ArrayList` materialisation in `LazyFileIterator`. Saves ~513 MB across 4 tasks.
3. Add `ExternalSpillableMap.keyStream()` and use it in `HoodieSortedMergeHandle` to seed the `PriorityQueue` via the `Collection` constructor (O(N) heapify). Saves ~256 MB peak per task.
Total: ~1.65 GB per task. 4-task demand drops from ~6.25 GB to ~4.57 GB on a 9 GB executor heap.
## Stacktrace
Not a Java exception — executor goes OOM under GC pressure with the heap layout described above. Eclipse MAT dominator-tree and class-histogram analysis is the authoritative signal.
## Will file a PR
Fix forthcoming.
Contributor guide
No contributing guide indexed for this repository
Research direction
Start with BitCaskDiskMap.ValueMetadata and valueMetadataMap, then inspect LazyFileIterator to understand the metadata ordering and sorting assumptions. Review ExternalSpillableMap.keySet() and HoodieSortedMergeHandle's constructor for the key-copying path; done means the three proposed structures are bounded and the reproduced compaction no longer OOMs under the stated executor settings.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java, spark
- Domain
- data-engineering, distributed-systems, performance
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Quiet
- Clarity
- Clearly specified
- Newbie friendliness
- 35/100