feat: add concurrent_files flag for bounded concurrent streaming

Add _bounded_concurrent_batches() with proper lock discipline:
- Queue backpressure caps memory (scan.max-buffered-batches, default 16)
- Semaphore limits concurrent file reads (concurrent_files param)
- Cancel event with timeouts on all blocking ops (no lock over IO)
- Error propagation and early termination support

When streaming=True and concurrent_files > 1, batches are yielded
as they arrive from parallel file reads. File ordering is not
guaranteed (documented).

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: sumedhsakdeo

mkdocs/docs/api.md

@@ -371,6 +371,22 @@ for buf in tbl.scan().to_arrow_batch_reader(order=ScanOrder.ARRIVAL, batch_size=
     print(f"Buffer contains {len(buf)} rows")
 ```
 
+For maximum throughput, use `concurrent_files` to read multiple files in parallel while streaming. Batches are yielded as they arrive from any file — ordering across files is not guaranteed:
+
+```python
+for buf in tbl.scan().to_arrow_batch_reader(streaming=True, concurrent_files=4, batch_size=1000):
+    print(f"Buffer contains {len(buf)} rows")
+```
+
+**Ordering semantics:**
+
+| Configuration | File ordering | Within-file ordering |
+|---|---|---|
+| Default (`streaming=False`) | Batches grouped by file, in task submission order | Row order |
+| `streaming=True` | Interleaved across files (no grouping guarantee) | Row order within each file |
+
+Within each file, batch ordering always follows row order. The `limit` parameter is enforced correctly regardless of configuration.
+
 To avoid any type inconsistencies during writing, you can convert the Iceberg table schema to Arrow:
 
 ```python
@@ -1655,6 +1671,17 @@ table.scan(
 ).to_arrow_batch_reader(order=ScanOrder.ARRIVAL)
 ```
 
+For concurrent file reads with streaming, use `concurrent_files`. Note that batch ordering across files is not guaranteed:
+
+```python
+table.scan(
+    row_filter=GreaterThanOrEqual("trip_distance", 10.0),
+    selected_fields=("VendorID", "tpep_pickup_datetime", "tpep_dropoff_datetime"),
+).to_arrow_batch_reader(streaming=True, concurrent_files=4)
+```
+
+When using `concurrent_files > 1`, batches from different files may be interleaved. Within each file, batches are always in row order. See the ordering semantics table in the [Apache Arrow section](#apache-arrow) above for details.
+
 ### Pandas
 
 <!-- prettier-ignore-start -->

pyiceberg/io/pyarrow.py

@@ -33,11 +33,13 @@
 import logging
 import operator
 import os
+import queue
 import re
+import threading
 import uuid
 import warnings
 from abc import ABC, abstractmethod
-from collections.abc import Callable, Iterable, Iterator
+from collections.abc import Callable, Generator, Iterable, Iterator
 from copy import copy
 from dataclasses import dataclass
 from enum import Enum
@@ -1682,6 +1684,87 @@ def _read_all_delete_files(io: FileIO, tasks: Iterable[FileScanTask]) -> dict[st
     return deletes_per_file
 
 
+_QUEUE_SENTINEL = object()
+
+
+def _bounded_concurrent_batches(
+    tasks: list[FileScanTask],
+    batch_fn: Callable[[FileScanTask], Iterator[pa.RecordBatch]],
+    concurrent_files: int,
+    max_buffered_batches: int = 16,
+) -> Generator[pa.RecordBatch, None, None]:
+    """Read batches from multiple files concurrently with bounded memory.
+
+    Workers read from files in parallel (up to concurrent_files at a time) and push
+    batches into a shared queue. The consumer yields batches from the queue.
+    A sentinel value signals completion, avoiding timeout-based polling.
+
+    Args:
+        tasks: The file scan tasks to process.
+        batch_fn: A callable that takes a FileScanTask and returns an iterator of RecordBatches.
+        concurrent_files: Maximum number of files to read concurrently.
+        max_buffered_batches: Maximum number of batches to buffer in the queue.
+    """
+    if not tasks:
+        return
+
+    batch_queue: queue.Queue[pa.RecordBatch | BaseException | object] = queue.Queue(maxsize=max_buffered_batches)
+    cancel_event = threading.Event()
+    pending_count = len(tasks)
+    pending_lock = threading.Lock()
+    file_semaphore = threading.Semaphore(concurrent_files)
+
+    def worker(task: FileScanTask) -> None:
+        nonlocal pending_count
+        try:
+            # Blocking acquire — on cancellation, extra permits are released to unblock.
+            file_semaphore.acquire()
+            if cancel_event.is_set():
+                return
+
+            for batch in batch_fn(task):
+                if cancel_event.is_set():
+                    return
+                batch_queue.put(batch)
+        except BaseException as e:
+            if not cancel_event.is_set():
+                batch_queue.put(e)
+        finally:
+            file_semaphore.release()
+            with pending_lock:
+                pending_count -= 1
+                if pending_count == 0:
+                    batch_queue.put(_QUEUE_SENTINEL)
+
+    executor = ExecutorFactory.get_or_create()
+    futures = [executor.submit(worker, task) for task in tasks]
+
+    try:
+        while True:
+            item = batch_queue.get()
+
+            if item is _QUEUE_SENTINEL:
+                break
+
+            if isinstance(item, BaseException):
+                raise item
+
+            yield item
+    finally:
+        cancel_event.set()
+        # Release semaphore permits to unblock any workers waiting on acquire()
+        for _ in range(len(tasks)):
+            file_semaphore.release()
+        # Drain the queue to unblock any workers stuck on put()
+        while not batch_queue.empty():
+            try:
+                batch_queue.get_nowait()
+            except queue.Empty:
+                break
+        for future in futures:
+            future.cancel()
+
+
 class ArrowScan:
     _table_metadata: TableMetadata
     _io: FileIO
@@ -1766,22 +1849,30 @@ def to_record_batches(
         tasks: Iterable[FileScanTask],
         batch_size: int | None = None,
         order: ScanOrder = ScanOrder.TASK,
+        concurrent_files: int = 1,
     ) -> Iterator[pa.RecordBatch]:
         """Scan the Iceberg table and return an Iterator[pa.RecordBatch].
 
         Returns an Iterator of pa.RecordBatch with data from the Iceberg table
         by resolving the right columns that match the current table schema.
         Only data that matches the provided row_filter expression is returned.
 
+        Ordering semantics:
+            - ScanOrder.TASK (default): Batches are grouped by file in task submission order.
+            - ScanOrder.ARRIVAL: Batches may be interleaved across files. Within each file,
+              batch ordering follows row order.
+
         Args:
             tasks: FileScanTasks representing the data files and delete files to read from.
             batch_size: The number of rows per batch. If None, PyArrow's default is used.
             order: Controls the order in which record batches are returned.
                 ScanOrder.TASK (default) returns batches in task order, with each task
                 fully materialized before proceeding to the next. Allows parallel file
                 reads via executor. ScanOrder.ARRIVAL yields batches as they are
-                produced, processing tasks sequentially without materializing entire
-                files into memory.
+                produced without materializing entire files into memory.
+            concurrent_files: Number of files to read concurrently when order=ScanOrder.ARRIVAL.
+                Must be >= 1. When > 1, batches may arrive interleaved across files.
+                Ignored when order=ScanOrder.TASK.
 
         Returns:
             An Iterator of PyArrow RecordBatches.
@@ -1790,18 +1881,32 @@ def to_record_batches(
         Raises:
             ResolveError: When a required field cannot be found in the file
             ValueError: When a field type in the file cannot be projected to the schema type,
-                or when an invalid order value is provided.
+                or when an invalid order value is provided, or when concurrent_files < 1.
         """
         if not isinstance(order, ScanOrder):
             raise ValueError(f"Invalid order: {order!r}. Must be a ScanOrder enum value (ScanOrder.TASK or ScanOrder.ARRIVAL).")
 
+        if concurrent_files < 1:
+            raise ValueError(f"concurrent_files must be >= 1, got {concurrent_files}")
+
         deletes_per_file = _read_all_delete_files(self._io, tasks)
 
         if order == ScanOrder.ARRIVAL:
-            # Arrival order: process all tasks sequentially, yielding batches as produced.
-            # _record_batches_from_scan_tasks_and_deletes handles the limit internally
-            # when called with all tasks, so no outer limit check is needed.
-            yield from self._record_batches_from_scan_tasks_and_deletes(tasks, deletes_per_file, batch_size)
+            # Arrival order: read files with bounded concurrency, yielding batches as produced.
+            # When concurrent_files=1, this is sequential. When >1, batches may interleave across files.
+            task_list = list(tasks)
+
+            def batch_fn(task: FileScanTask) -> Iterator[pa.RecordBatch]:
+                return self._record_batches_from_scan_tasks_and_deletes([task], deletes_per_file, batch_size)
+
+            total_row_count = 0
+            for batch in _bounded_concurrent_batches(task_list, batch_fn, concurrent_files):
+                current_batch_size = len(batch)
+                if self._limit is not None and total_row_count + current_batch_size >= self._limit:
+                    yield batch.slice(0, self._limit - total_row_count)
+                    return
+                yield batch
+                total_row_count += current_batch_size
             return
 
         # Task order: existing behavior with executor.map + list()

pyiceberg/table/__init__.py

@@ -2170,19 +2170,28 @@ def to_arrow(self) -> pa.Table:
             self.table_metadata, self.io, self.projection(), self.row_filter, self.case_sensitive, self.limit
         ).to_table(self.plan_files())
 
-    def to_arrow_batch_reader(self, batch_size: int | None = None, order: ScanOrder = ScanOrder.TASK) -> pa.RecordBatchReader:
+    def to_arrow_batch_reader(
+        self, batch_size: int | None = None, order: ScanOrder = ScanOrder.TASK, concurrent_files: int = 1
+    ) -> pa.RecordBatchReader:
         """Return an Arrow RecordBatchReader from this DataScan.
 
         For large results, using a RecordBatchReader requires less memory than
         loading an Arrow Table for the same DataScan, because a RecordBatch
         is read one at a time.
 
+        Ordering semantics:
+            - ScanOrder.TASK (default): Batches are grouped by file in task submission order.
+            - ScanOrder.ARRIVAL: Batches may be interleaved across files. Within each file,
+              batch ordering follows row order.
+
         Args:
             batch_size: The number of rows per batch. If None, PyArrow's default is used.
             order: Controls the order in which record batches are returned.
                 ScanOrder.TASK (default) returns batches in task order with parallel
-                file reads. ScanOrder.ARRIVAL yields batches as they are produced,
-                processing tasks sequentially.
+                file reads. ScanOrder.ARRIVAL yields batches as they are produced
+                without materializing entire files into memory.
+            concurrent_files: Number of files to read concurrently when order=ScanOrder.ARRIVAL.
+                When > 1, batches may arrive interleaved across files.
 
         Returns:
             pa.RecordBatchReader: Arrow RecordBatchReader from the Iceberg table's DataScan
@@ -2195,7 +2204,7 @@ def to_arrow_batch_reader(self, batch_size: int | None = None, order: ScanOrder
         target_schema = schema_to_pyarrow(self.projection())
         batches = ArrowScan(
             self.table_metadata, self.io, self.projection(), self.row_filter, self.case_sensitive, self.limit
-        ).to_record_batches(self.plan_files(), batch_size=batch_size, order=order)
+        ).to_record_batches(self.plan_files(), batch_size=batch_size, order=order, concurrent_files=concurrent_files)
 
         return pa.RecordBatchReader.from_batches(
             target_schema,