Repository navigation
Investigate charging native Iceberg scan memory to the task memory pool #6126
Description
What is the problem the feature request solves?
#6125 charges the native Parquet scan's working set to the task memory pool. The native Iceberg scan (IcebergScanExec, built on iceberg-rust's ArrowReader) has the same gap: nothing it holds is reserved, so the pool, and Spark behind it, cannot see it. Investigate whether the same approach applies there and what it would take.
For the Parquet scan, the gap on TPC-H SF100 was a flat 100 to 150 MiB per executor (8 tasks) that the pool never saw. It was about one row group's compressed projected column chunks per running task, plus decompressed pages and the output batch. The Iceberg scan should have at least the same working set, and some things that could make it larger:
spark.comet.scan.icebergNative.dataFileConcurrencyLimit(default 1, docs suggest 2 to 8) lets a task read several data files at once, multiplying the per-file working set.- Delete files. Positional deletes are loaded into per-file row selections, and equality deletes are loaded into in-memory sets. These scale with the delete files, not the row group, and could be the largest unaccounted consumer on merge-on-read tables.
- Range coalescing in iceberg-rust's
ArrowFileReader::get_byte_rangesmerges nearby ranges, so the fetched buffers include gap bytes.
Describe the potential solution
To investigate, not decided:
- Where to hook the data pages. In the Parquet scan, Comet owns the
AsyncFileReader(EagerPageIndexReaderFactory), so wrapping fetched buffers in aBytes::from_ownerholding aMemoryReservationneeded no upstream change. For Iceberg, theAsyncFileReaderis iceberg-rust'sArrowFileReader, which Comet can't replace. Options:- wrap
FileRead::readresults in Comet's ownStorage/StorageFactoryimplementation, since the reservation survives slicing of the merged buffers; - add a reader or memory hook upstream in iceberg-rust.
- wrap
- Where to hold a decode-buffer estimate. There is no per-file reader lifecycle visible to Comet. A reservation held by the
IcebergScanExecstream, sized asdataFileConcurrencyLimittimes the per-file estimate, may be enough. - Delete-file memory. Measure it and decide whether it needs its own reservation.
- Refusal handling. As in feat: charge native Parquet scan memory to the task memory pool #6125, use
try_grow: both Comet unified pools panic on an infalliblegrowwhen Spark refuses, and a scan cannot spill.
Measure first: a traced run (spark.comet.tracing.enabled, build with jemalloc,alloc-accounting) against an Iceberg copy of TPC-H, comparing native_allocated with comet_memory_reserved_total sample by sample. Include a merge-on-read table with delete files, and more than one dataFileConcurrencyLimit setting.
Additional context
Parquet scan PR: #6125. Its description has the measurement method and the before and after numbers.
Activity
dwsmith1983 commentedon Oct 19, 2026
Here are some figures on the native Iceberg scan.
Setup:
local[8].The OOM kill in #7108 comes mostly from glibc keeping freed memory, which happens on all three formats. Iceberg holds
more resident memory than the other two, though, and it is the only one that gets killed.
To compare the scans more closely I set
spark.sql.autoBroadcastJoinThreshold=-1. Join strategies then match acrossthe three formats:
exchange, and q22 has one reused subquery fewer.
Anonymous memory in the container, one run each:
Parquet and Delta are within run-to-run noise of each other, which is about 0.5 GiB. Iceberg is 1.5 to 2 GiB above
both at every point, even though Comet's live peak is lowest on Iceberg. So the extra resident memory doesn't show up
in what Comet allocates or reserves. That fits the scan holding buffers outside the pool, which is what this issue is
about.
The gap is small on scan-only queries. Running q01 11 times on Iceberg adds about 60 MB per run, while on Parquet the
footprint stays flat. q06 stays flat on both.
I haven't measured where the extra memory goes, and I have no heap profile yet. Reading iceberg-rust at Comet's
pinned rev (1f3bc340), I found two places that could add to the per-task working set:
ArrowFileReader::get_byte_rangesmerges ranges closer than 1 MiB (DEFAULT_RANGE_COALESCE_BYTES) and fetches upto 10 merged ranges at once. With 128 MB row groups, one merged range can cover most of a row group's projected
columns.
OpenDalReader::readends in.to_bytes(), which copies the range into one buffer when opendal returns it in morethan one chunk.
dataFileConcurrencyLimitwas left at its default of 1 throughout.