[Data] Fetch a shared Iceberg delete file once per read task, not once per data file

[moomindani]

Description

An Iceberg read task hands PyIceberg one FileScanTask at a time, so that ArrowScan does not
materialize every file in the chunk before yielding anything. PyIceberg deduplicates delete files only
within the tasks handed to one scan call, so that also collapses the deduplication scope to a single
data file: a delete file shared by k of the task's data files is fetched k times, and _read_deletes
has no cache. The comment already in _get_read_task noted this ("Singleton calls can reread delete
files shared by multiple data files").

The read task now counts how many of its data files reference each delete file and, for the ones
referenced more than once, reads them through a FileIO that keeps their bytes until the last reference
has been served. Two properties are deliberate:

• Data files are still scanned one at a time. The fix does not change how many tasks go into a scan
  call, so the memory behaviour the sequential path exists for is untouched. The alternative --
  grouping the tasks that share a delete file so one call covers them all -- was measured too, and it is
  both slower and heavier, because to_record_batches submits every task it is given to its executor at
  once.
• No delete format is involved. Ray sees a path and bytes; PyIceberg keeps parsing Parquet, ORC and
  Puffin. This is what makes it work for V3 deletion vectors, where every data file gets its own
  DataFile entry pointing into one shared Puffin file and PuffinFile parses every blob on each read,
  so PyIceberg's DataFile-level deduplication cannot collapse them even when a scan call is handed
  every task. Grouping would not fix deletion vectors at all; caching by path does.

The count is over (task, delete file) pairs, which is exact when each scan call gets one task and an
over-estimate when a call gets several, so an entry is never dropped while a reference to it is still
outstanding.

The cache is bounded by a new DataContext.iceberg_config.read_delete_file_cache_max_bytes (64 MiB
default, RAY_DATA_ICEBERG_READ_DELETE_CACHE_MAX_BYTES); 0 restores the previous behaviour, and a
delete file referenced by only one data file is never cached.

Related issues

Closes #66147.

Not a duplicate: no issue or PR mentioned the delete-file re-read before #66147, and no open PR touches
iceberg_datasource.py's read loop (#62749 adds CREATE mode to writes, #61753 adds checkpointing,
#66066 reorganizes the DataSourceV2 contract, #63337 is a Delta/write-abstraction stack). #66144 and
#66146 are mine and change iceberg_datasink.py only, so this branch is independent of both.

Additional information

Effect

24 data files of 400,000 rows, one 19.7 MiB positional delete file that deletes 200,000 rows in every
one of them, so all 24 share it. Read from S3 (us-west-2, driven from a laptop), two rounds against
one fixed table:

	before	after
delete-file fetches	24	1
bytes fetched for deletes	473 MiB	19.7 MiB
time inside _read_deletes	74.3s / 90.0s	5.7s / 5.7s
peak RSS of the read	347 / 340 MiB	400 / 373 MiB
wall	92.1s / 107.4s	22.3s / 21.4s

Same rows out (4,800,000) in every run. The 27-53 MiB of extra peak RSS is the one cached copy of the
delete file. What remains inside _read_deletes is the repeated parse of those cached bytes; removing
that too would need PyIceberg to expose an already-computed delete index, which is private API today.

On a local filesystem the same table shows no wall-clock gain (3.6s before, 3.7s after), because there
the cost is the parse, not the fetch. This change is about the fetch.

For comparison, the two alternatives on the same table and rounds: grouping 4 tasks per scan call gives
6 fetches, 22.5s / 26.5s inside _read_deletes, 435 / 446 MiB peak; read_file_tasks_sequentially=False
gives 1 fetch and 2.6s but ~600 MiB peak.

Tests

test_get_read_task_fetches_a_shared_delete_file_once in
python/ray/data/tests/datasource/test_iceberg.py, parametrized over the cache being off and on. It
builds one positional delete file covering row 0 of all 10 data files of the fixture table, passes a
counting FileIO into _get_read_task, and asserts the delete file is opened 10 times with the cache
disabled and once with it enabled, that the data files are still opened exactly once each, and that the
row count is the same either way.

pytest python/ray/data/tests/datasource/test_iceberg.py -k fetches_a_shared_delete_file_once
  -> 2 passed
pytest python/ray/data/tests/datasource/test_iceberg.py
  -> 85 passed
pytest python/ray/data/tests/test_context.py
  -> 7 passed
pre-commit run --files python/ray/data/_internal/datasource/iceberg_datasource.py \
  python/ray/data/context.py python/ray/data/tests/datasource/test_iceberg.py
  -> all hooks pass

On master the new test errors rather than failing, because delete_file_cache_max_bytes does not exist
there; the behaviour before the change was verified separately with the same counting FileIO against
master's _get_read_task -- 8 opens of one delete file shared by 8 data files, against 1 after.

Run on Python 3.12.11, macOS arm64, pyiceberg==0.11.0, pyarrow==23.0.1.

AI assistance

AI assistance was used for this change: the investigation, the implementation, this description and the
measurements above were produced with Claude Code, and every number quoted here comes from the runs
listed. I have reviewed every changed line and run the tests locally.

[gemini-code-assist]

Code Review

This pull request introduces a caching mechanism (_DeleteFileCache) for shared delete files in the Iceberg datasource to prevent redundant disk/network IO when multiple data files reference the same delete file. It also adds a configurable memory limit (read_delete_file_cache_max_bytes) and corresponding unit tests. The review feedback highlights two critical issues: a potential race condition and inefficiency in _DeleteFileCache.read under concurrent reads, and a missing __len__ delegation in _CachedInputFile which would cause a TypeError when len() is called on the cached input file.

[gemini-code-assist]

The current implementation of _DeleteFileCache.read has a potential race condition and is inefficient under concurrent reads (e.g., when read_file_tasks_sequentially is False). Multiple threads can concurrently experience a cache miss, perform redundant disk/network IO, and then overwrite each other's cache entries. Additionally, the lock is acquired twice for every read, even on cache hits.

We can optimize this by:

1. Checking the cache and updating the reference count in a single lock acquisition on cache hits.
2. Handling concurrent cache misses gracefully by checking if another thread has already populated the cache while the current thread was performing IO.
3. Popping the location from _remaining once remaining <= 0 to avoid unnecessary proxying and memory leaks of keys.

        with self._lock:
            payload = self._entries.get(location)
            if payload is not None:
                remaining = self._remaining.get(location, 0) - 1
                if remaining <= 0:
                    self._remaining.pop(location, None)
                    self._cached_bytes -= len(self._entries.pop(location, b""))
                else:
                    self._remaining[location] = remaining
                return payload

        with input_file.open() as stream:
            payload = stream.read()

        with self._lock:
            cached_payload = self._entries.get(location)
            if cached_payload is not None:
                payload = cached_payload

            remaining = self._remaining.get(location, 0) - 1
            if remaining <= 0:
                self._remaining.pop(location, None)
                self._cached_bytes -= len(self._entries.pop(location, b""))
            else:
                self._remaining[location] = remaining
                if cached_payload is None and self._cached_bytes + len(payload) <= self._max_bytes:
                    self._entries[location] = payload
                    self._cached_bytes += len(payload)
        return payload

[moomindani]

Applied in 31a3fba, with one deviation: the read itself still happens outside the lock. A concurrent
miss then costs a duplicate read, but holding the lock across the read would make every other file in
the task wait behind it, and a duplicate read is the thing the cache exists to make rare rather than
impossible. What changed is the rest of your list -- a hit returns under a single lock, a reader that
finds the entry already populated keeps that copy instead of replacing it, and the location is dropped
from the remaining counts after the last reference, so later opens skip the wrapper entirely.

The accounting moved into a _release helper, which also made the invariant easier to state: the counts
are over (task, delete file) pairs, which is an upper bound on the number of reads, so an entry is
never dropped while a reference is still outstanding.

Added a test that reads one shared delete file through a single scan call over PyIceberg's thread pool
(read_file_tasks_sequentially=False) and asserts it is still fetched exactly once.

[gemini-code-assist]

In Python, special/dunder methods (like __len__) are looked up on the class rather than the instance, meaning they bypass __getattr__ entirely. Since PyIceberg's InputFile abstract base class defines __len__ to return the file size, and other parts of PyIceberg or PyArrow may call len(input_file), _CachedInputFile will raise a TypeError when len() is called on it.

We should explicitly delegate __len__ to the underlying _inner input file to satisfy the InputFile contract.

Suggested change

	def __getattr__(self, name: str) -> Any:
	return getattr(self._inner, name)
	def open(self, *args, **kwargs) -> pa.BufferReader:
	return pa.BufferReader(self._cache.read(self._location, self._inner))
	def __getattr__(self, name: str) -> Any:
	return getattr(self._inner, name)
	def __len__(self) -> int:
	return len(self._inner)
	def open(self, *args, **kwargs) -> pa.BufferReader:
	return pa.BufferReader(self._cache.read(self._location, self._inner))

[moomindani]

Good catch, and it is a real one rather than theoretical: PyIceberg calls len(input_file) itself when
it builds a DataFile from a file (io/pyarrow.py), so the wrapper was not honouring the InputFile
contract. Delegated in 31a3fba.

Added a contract test covering len(), location, exists() and open() on the wrapper. With the
delegation removed again it fails with TypeError: object of type '_CachedInputFile' has no len(), so
the test does hold the line.

[cursor]

Cursor Bugbot has reviewed your changes using default effort and found 1 potential issue.

^{Reviewed by Cursor Bugbot for commit 31a3fba. Configure here.}