Move batch_size parameter to ArrivalOrder for better semantic design

- Add batch_size parameter to ArrivalOrder class with comprehensive documentation
- Include memory formula: Peak memory ≈ concurrent_streams × batch_size × max_buffered_batches × (average row size)
- Update default concurrent_streams from 1 to 8 for better performance out-of-the-box
- Remove batch_size parameter from to_arrow_batch_reader() and to_record_batches() methods
- Simplify API by putting batch_size where it has direct memory impact (streaming orders)
- TaskOrder uses PyArrow defaults, ArrivalOrder provides full memory control

🤖 Generated with [Claude Code](https://claude.ai/code)

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

Author: sumedhsakdeo

pyiceberg/io/pyarrow.py

@@ -144,7 +144,7 @@
     visit,
     visit_with_partner,
 )
-from pyiceberg.table import DOWNCAST_NS_TIMESTAMP_TO_US_ON_WRITE, ArrivalOrder, ScanOrder, TableProperties, TaskOrder
+from pyiceberg.table import DOWNCAST_NS_TIMESTAMP_TO_US_ON_WRITE, ArrivalOrder, ScanOrder, TableProperties
 from pyiceberg.table.locations import load_location_provider
 from pyiceberg.table.metadata import TableMetadata
 from pyiceberg.table.name_mapping import NameMapping, apply_name_mapping
@@ -1837,7 +1837,6 @@ def to_table(self, tasks: Iterable[FileScanTask]) -> pa.Table:
     def to_record_batches(
         self,
         tasks: Iterable[FileScanTask],
-        batch_size: int | None = None,
         order: ScanOrder | None = None,
     ) -> Iterator[pa.RecordBatch]:
         """Scan the Iceberg table and return an Iterator[pa.RecordBatch].
@@ -1853,11 +1852,15 @@ def to_record_batches(
 
         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.
                 TaskOrder() (default) yields batches one file at a time in task order.
-                ArrivalOrder(concurrent_streams=N, max_buffered_batches=M) yields batches
-                as they are produced without materializing entire files into memory.
+                ArrivalOrder(concurrent_streams=N, batch_size=B, max_buffered_batches=M)
+                yields batches as they are produced without materializing entire files
+                into memory. Peak memory ≈ concurrent_streams × batch_size × max_buffered_batches
+                × (average row size in bytes). batch_size is the number of rows per batch.
+                For example (if average row size ≈ 32 bytes):
+                - ArrivalOrder(concurrent_streams=4, batch_size=32768, max_buffered_batches=8)
+                - Peak memory ≈ 4 × 32768 rows × 8 × 32 bytes ≈ ~32 MB (plus Arrow overhead)
 
         Returns:
             An Iterator of PyArrow RecordBatches.
@@ -1868,9 +1871,6 @@ def to_record_batches(
             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 concurrent_streams < 1.
         """
-        if order is None:
-            order = TaskOrder()
-
         if not isinstance(order, ScanOrder):
             raise ValueError(f"Invalid order: {order!r}. Must be a ScanOrder instance (TaskOrder() or ArrivalOrder()).")
 
@@ -1881,11 +1881,11 @@ def to_record_batches(
                 raise ValueError(f"concurrent_streams must be >= 1, got {order.concurrent_streams}")
             return self._apply_limit(
                 self._iter_batches_arrival(
-                    task_list, deletes_per_file, batch_size, order.concurrent_streams, order.max_buffered_batches
+                    task_list, deletes_per_file, order.batch_size, order.concurrent_streams, order.max_buffered_batches
                 )
             )
 
-        return self._apply_limit(self._iter_batches_materialized(task_list, deletes_per_file, batch_size))
+        return self._apply_limit(self._iter_batches_materialized(task_list, deletes_per_file))
 
     def _prepare_tasks_and_deletes(
         self, tasks: Iterable[FileScanTask]
@@ -1914,13 +1914,12 @@ def _iter_batches_materialized(
         self,
         task_list: list[FileScanTask],
         deletes_per_file: dict[str, list[ChunkedArray]],
-        batch_size: int | None,
     ) -> Iterator[pa.RecordBatch]:
         """Yield batches using executor.map with full file materialization."""
         executor = ExecutorFactory.get_or_create()
 
         def batches_for_task(task: FileScanTask) -> list[pa.RecordBatch]:
-            return list(self._record_batches_from_scan_tasks_and_deletes([task], deletes_per_file, batch_size))
+            return list(self._record_batches_from_scan_tasks_and_deletes([task], deletes_per_file))
 
         for batches in executor.map(batches_for_task, task_list):
             yield from batches

pyiceberg/table/__init__.py

@@ -174,9 +174,23 @@ class ArrivalOrder(ScanOrder):
 
     Batches are yielded as they are produced without materializing entire
     files into memory. Supports concurrent processing of multiple files.
+
+    Memory Usage:
+        Peak memory ≈ concurrent_streams × batch_size × max_buffered_batches
+        × (average row size in bytes). batch_size is the number of rows per batch.
+
+        For example (if average row size ≈ 32 bytes):
+        - ArrivalOrder(concurrent_streams=4, batch_size=32768, max_buffered_batches=8)
+        - Peak memory ≈ 4 × 32768 rows × 8 × 32 bytes ≈ ~32 MB (plus Arrow overhead)
+
+    Args:
+        concurrent_streams: Number of files to read concurrently (default: 8)
+        batch_size: Number of rows per batch, controls memory per stream (default: None, uses PyArrow default ~131K)
+        max_buffered_batches: Maximum batches buffered per stream (default: 16)
     """
 
-    concurrent_streams: int = 1
+    concurrent_streams: int = 8
+    batch_size: int | None = None
     max_buffered_batches: int = 16
 
 
@@ -2181,7 +2195,7 @@ 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 | None = None) -> pa.RecordBatchReader:
+    def to_arrow_batch_reader(self, order: ScanOrder | None = None) -> pa.RecordBatchReader:
         """Return an Arrow RecordBatchReader from this DataScan.
 
         For large results, using a RecordBatchReader requires less memory than
@@ -2194,12 +2208,12 @@ def to_arrow_batch_reader(self, batch_size: int | None = None, order: ScanOrder
               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.
-                TaskOrder() (default) yields batches one file at a time in task order.
-                ArrivalOrder(concurrent_streams=N, max_buffered_batches=M) yields batches
-                as they are produced without materializing entire files into memory.
-                concurrent_streams controls parallelism, max_buffered_batches controls memory.
+                TaskOrder() (default) yields batches one file at a time in task order using
+                PyArrow's default batch size.
+                ArrivalOrder(concurrent_streams=N, batch_size=B, max_buffered_batches=M)
+                yields batches as they are produced without materializing entire files
+                into memory. Memory usage ≈ concurrent_streams × batch_size × max_buffered_batches × (average row size in bytes).
 
         Returns:
             pa.RecordBatchReader: Arrow RecordBatchReader from the Iceberg table's DataScan
@@ -2215,7 +2229,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(), order=order)
 
         return pa.RecordBatchReader.from_batches(
             target_schema,