apache / apache/datafusion-comet
Account the native Iceberg writer's buffers in Comet's memory pool
- Dominant language
- Scala
- Stars
- 1.3k
- Forks
- 373
- Avg merge
- 2d 4h
- Merged PRs (30d)
- 198
Description
### What is the problem the feature request solves?
The native Iceberg writer (#5361) buffers rows in parquet-rs before flushing row groups, and the clustered and fanout writers hold per-partition writers open, each with its own buffer. It is not clear whether any of that memory is registered with Comet's memory pool (`spark.comet.exec.memoryPool` / the unified off-heap pool), or whether it is Rust heap allocation that Spark's memory accounting cannot see. A fanout write over many partitions multiplies the unaccounted buffers by the partition count.
The native scan and shuffle went through this audit already (#5212, memory pool and accounting sweep); the writer landed after it.
### Describe the potential solution
- Audit what the writer allocates per open file (parquet-rs column writers, dictionary pages, the manifest serializer) and whether `WriterProperties` sizes (row group size, page size, dictionary page size) bound it.
- Register the writer's buffers with the task's memory reservation, or document why the amount is bounded enough not to need it, with the bound stated in terms of the vetted `write.parquet.*` properties and the partition count.
- For the fanout writer, decide whether to cap open writers or account for them, since the JVM `FanoutDataWriter` has the same shape and users size executors for it.
- Add a test that a fanout write over many partitions with a small `spark.comet.memoryOverhead` fails with a Comet out-of-memory error rather than a process-level one.
### Additional context
Part of the native Iceberg writes epic, #5649. Related: #5361, #5212.
Contributor guide
Research direction
Start with the native Iceberg writer and parquet-rs allocations, then compare the JVM FanoutDataWriter and the prior scan and shuffle audit in #5212. Inspect WriterProperties and the vetted write.parquet.* settings to establish per-file and per-partition bounds. Done means buffers are reserved or a documented bound is provided, and a many-partition fanout test with small spark.comet.memoryOverhead produces a Comet out-of-memory error.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- rust, scala
- Domain
- data-engineering
- Issue type
- Feature
- Difficulty
- 5/5
- Estimated time
- Over a week
- Activity status
- Active
- Clarity
- Mostly clear
- Newbie friendliness
- 45/100