andygrove opened a new issue, #6126:
URL: https://github.com/apache/datafusion-comet/issues/6126
### 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 #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.
--
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]
