feat: Add ArrivalOrder to ArrowScan for bounded-memory concurrent reads

[sumedhsakdeo]

Summary

Addresses #3036 — ArrowScan.to_record_batches() uses executor.map + list() which eagerly materializes all record batches per file into memory, causing OOM on large tables.

This PR adds a new order parameter to to_arrow_batch_reader() with two implementations:

• TaskOrder (default) — preserves existing behavior: batches grouped by file in task submission order, each file fully materialized before proceeding to the next.
• ArrivalOrder — yields batches as they are produced across files without materializing entire files into memory. Accepts three sub-parameters:
  • concurrent_streams: int — number of files to read concurrently (default: 8). A per-scan ThreadPoolExecutor(max_workers=concurrent_streams) bounds concurrency.
  • batch_size: int | None — number of rows per batch passed to PyArrow's ds.Scanner (default: PyArrow's built-in 131,072).
  • max_buffered_batches: int — size of the bounded queue between producers and consumer (default: 16), providing backpressure to cap memory usage.

Problem

The current implementation materializes all batches from each file via list() inside executor.map, which runs up to min(32, cpu_count+4) files in parallel. For large files this means all batches from ~20 files are held in memory simultaneously before any are yielded to the consumer.

Solution

Before: OOM on large tables

batches = table.scan().to_arrow_batch_reader()

After: bounded memory, tunable parallelism

from pyiceberg.table import ArrivalOrder

batches = table.scan().to_arrow_batch_reader(
    order=ArrivalOrder(concurrent_streams=4, batch_size=10000),
)

Default behavior is unchanged — TaskOrder preserves the existing executor.map + list() path for backwards compatibility.

Architecture

When order=ArrivalOrder(...), batches flow through _bounded_concurrent_batches:

1. All file tasks are submitted to a per-scan ThreadPoolExecutor(max_workers=concurrent_streams)
2. Workers push batches into a bounded Queue(maxsize=max_buffered_batches) — when full, workers block (backpressure)
3. The consumer yields batches from the queue via blocking queue.get()
4. A sentinel value signals completion — no timeout-based polling
5. On early termination (consumer stops), a cancel event is set and the queue is drained until the sentinel to unblock all stuck workers
6. The executor context manager handles deterministic shutdown

Refactored to_record_batches into helpers: _prepare_tasks_and_deletes, _iter_batches_arrival, _iter_batches_materialized, _apply_limit.

Ordering semantics

Configuration	File ordering	Within-file ordering
TaskOrder() (default)	Batches grouped by file, in task submission order	Row order
ArrivalOrder(concurrent_streams=1)	Grouped by file, sequential	Row order
ArrivalOrder(concurrent_streams>1)	Interleaved (no grouping guarantee)	Row order within each file

PR Stack

Breakdown of this large PR into smaller PRs:

1. PR 0: batch_size forwarding
2. PR 1: TaskOrder/ArrivalOrder enum — stop materializing entire files
3. PR 2: concurrent_streams — bounded concurrent reads in arrival order
4. PR 3: benchmark + docs guidance

Benchmark results

32 files × 500K rows, 5 columns (int64, float64, string, bool, timestamp), batch_size=131,072 (PyArrow default):

Config	Throughput (rows/s)	TTFR (ms)	Peak Arrow Memory
default (TaskOrder)	190,250,192	73.4	642.2 MB
ArrivalOrder(cs=1)	59,317,085	27.7	10.3 MB
ArrivalOrder(cs=2)	105,414,909	28.8	42.0 MB
ArrivalOrder(cs=4)	175,840,782	28.4	105.5 MB
ArrivalOrder(cs=8)	211,922,538	32.3	271.7 MB
ArrivalOrder(cs=16)	209,011,424	45.0	473.3 MB

TTFR = Time to First Record, cs = concurrent_streams

Note on throughput plateau at cs=8: This benchmark runs against local filesystem where Parquet reads are CPU-bound (decompression + decoding). Throughput plateaus once enough threads saturate available cores. On cloud storage (S3/GCS/ADLS), reads are I/O-bound with 50-200ms per-file latency, so higher concurrent_streams values (16-64+) would continue to show throughput gains until network bandwidth saturates. The optimal concurrent_streams will be higher for remote storage than what this local benchmark suggests.

Positional deletes, row filters, and limit are handled correctly in all modes.

Are these changes tested?

Yes. 25 new unit tests across two test files, plus a micro-benchmark:

• tests/io/test_pyarrow.py (16 tests): batch_size controls rows per batch, arrival order yields all rows correctly, arrival order respects limit, within-file ordering preserved, positional deletes applied correctly in all three modes (task order, arrival order, concurrent), positional deletes with limit, concurrent_streams < 1 raises ValueError
• tests/io/test_bounded_concurrent_batches.py (10 tests): single/multi-file correctness, incremental streaming, backpressure blocks producers when queue is full, error propagation from workers to consumer, early termination cancels workers cleanly, no deadlock when concurrent_streams > max_buffered_batches on early termination, concurrency limit enforced, empty task list, ArrowScan integration with limit
• tests/benchmark/test_read_benchmark.py: read throughput micro-benchmark across 6 configurations measuring rows/sec, TTFR, and peak Arrow memory

Are there any user-facing changes?

Yes. New order parameter on DataScan.to_arrow_batch_reader():

• order: ScanOrder | None — controls batch ordering. Pass TaskOrder() (default) or ArrivalOrder(concurrent_streams=N, batch_size=B, max_buffered_batches=M).

New public classes TaskOrder and ArrivalOrder (subclasses of ScanOrder) exported from pyiceberg.table.

All parameters are optional with backwards-compatible defaults. Existing code is unaffected.

Documentation updated in mkdocs/docs/api.md with usage examples, ordering semantics, and configuration guidance table.

[robreeves]

This could cause a worker to hang when concurrent_files > 1 and max_buffered_batches=1. Here is an example.

Starting state

max_buffered_batches=1, concurrent_files=3. Queue is full (1 item). Workers A, B, and C are all blocked on batch_queue.put().

Timeline

Step	Main thread	Workers A, B, C
1	cancel.set()
2	get_nowait() → removes 1 item. Queue: 0. Internally notifies Worker A	Worker A: woken but hasn't run yet
3	empty() → True (queue IS empty because A hasn't put yet). Exits drain loop.
4	executor.__exit__() → shutdown(wait=True), joins all threads...	Worker A runs, put() completes → Queue: 1. Checks cancel → returns. ✓
5	DEADLOCK — waiting for B and C to finish	Workers B, C: still blocked on put(). Queue is full, nobody will ever drain.

Fix
In the worker use put with a timeout so it can check if the thread is canceled periodically.

[sumedhsakdeo]

Good catch. I did have timeout in the previous version of the code, but it was causing performance regression. Exploring few other alternatives like condition variables, more complex but does not result in the bug.

[sumedhsakdeo]

We have a 15% regression if timeout is added. Will need to find another way.

  │            Config            │ Throughput (rows/s) │ Time  │  TTFR  │ Peak Memory │
  ├──────────────────────────────┼─────────────────────┼───────┼────────┼─────────────┤
  │ default (TASK, all-parallel) │ 199M                │ 0.08s │ 60.9ms │ 609.5 MB    │
  ├──────────────────────────────┼─────────────────────┼───────┼────────┼─────────────┤
  │ arrival-cf1                  │ 56.6M               │ 0.28s │ 29.9ms │ 10.3 MB     │
  ├──────────────────────────────┼─────────────────────┼───────┼────────┼─────────────┤
  │ arrival-cf2                  │ 93.3M               │ 0.17s │ 32.6ms │ 42.6 MB     │
  ├──────────────────────────────┼─────────────────────┼───────┼────────┼─────────────┤
  │ arrival-cf4                  │ 159.7M              │ 0.10s │ 32.9ms │ 114.7 MB    │
  ├──────────────────────────────┼─────────────────────┼───────┼────────┼─────────────┤
  │ arrival-cf8                  │ 186.4M              │ 0.09s │ 37.8ms │ 276.2 MB    │
  ├──────────────────────────────┼─────────────────────┼───────┼────────┼─────────────┤
  │ arrival-cf16                 │ 188.9M              │ 0.09s │ 53.4ms │ 453.2 MB    │
  └──────────────────────────────┴─────────────────────┴───────┴────────┴─────────────┘

[sumedhsakdeo]

Fixed, PTAL.