Skip to content

Investigate charging native Iceberg scan memory to the task memory pool #6126

Open

Description

@andygrove

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_ranges merges nearby ranges, so the fetched buffers include gap bytes.

Describe the potential solution

To investigate, not decided:

  1. Where to hook the data pages. In the Parquet scan, Comet owns the AsyncFileReader (EagerPageIndexReaderFactory), so wrapping fetched buffers in a Bytes::from_owner holding a MemoryReservation needed no upstream change. For Iceberg, the AsyncFileReader is iceberg-rust's ArrowFileReader, which Comet can't replace. Options:
    • wrap FileRead::read results in Comet's own Storage/StorageFactory implementation, since the reservation survives slicing of the merged buffers;
    • add a reader or memory hook upstream in iceberg-rust.
  2. Where to hold a decode-buffer estimate. There is no per-file reader lifecycle visible to Comet. A reservation held by the IcebergScanExec stream, sized as dataFileConcurrencyLimit times the per-file estimate, may be enough.
  3. Delete-file memory. Measure it and decide whether it needs its own reservation.
  4. 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 infallible grow when 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

  1. self-assigned this
    on Sep 22, 2026
  2. dwsmith1983 commented on Oct 19, 2026

    @dwsmith1983
    Contributor

    Here are some figures on the native Iceberg scan.

    Setup:

    • TPC-H SF10, Comet main (4a8a949), Iceberg 1.12.0.
    • One 14 GiB container: 4g heap, 4g off-heap, local[8].
    • The same data was also loaded as plain Parquet and as Delta.

    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 across
    the three formats:

    • Parquet and Delta produce identical plans and identical shuffle bytes.
    • Iceberg's plans are the same apart from two small differences: q07 reads nation twice instead of reusing the
      exchange, and q22 has one reused subquery fewer.
    • Iceberg's shuffle bytes are within 6% on the large queries; q17 is the widest at 0.97 against 1.03 GB.

    Anonymous memory in the container, one run each:

    point Parquet Delta Iceberg
    after q05 8.21 GiB 9.09 GiB 10.68 GiB
    after q09 10.55 GiB 10.47 GiB 12.35 GiB
    peak over the suite 12.43 GiB, passes 11.77 GiB, passes 13.93 GiB, killed
    Comet live peak (allocated) 3.75 GiB 3.87 GiB 2.35 GiB

    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_ranges merges ranges closer than 1 MiB (DEFAULT_RANGE_COALESCE_BYTES) and fetches up
      to 10 merged ranges at once. With 128 MB row groups, one merged range can cover most of a row group's projected
      columns.
    • OpenDalReader::read ends in .to_bytes(), which copies the range into one buffer when opendal returns it in more
      than one chunk.

    dataFileConcurrencyLimit was left at its default of 1 throughout.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Metadata

Metadata

Assignees

Labels

area:Icebergarea:memoryMemory pools, reservations, OOM handlingarea:scanParquet scan / data readingenhancementNew feature or request

Type

No type

Projects

No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions