leaves12138 commented on code in PR #8807:
URL: https://github.com/apache/paimon/pull/8807#discussion_r3740705677
##########
paimon-python/pypaimon/read/table_read.py:
##########
@@ -150,14 +169,150 @@ def _record_generator():
return _record_generator()
- def to_arrow_batch_reader(self, splits: List[Split],
- blob_parallelism: Optional[int] = None) ->
pyarrow.ipc.RecordBatchReader:
+ def to_arrow_batch_reader(
+ self,
+ splits: List[Split],
+ blob_parallelism: Optional[int] = None,
+ parallelism: Optional[int] = None,
+ ) -> pyarrow.ipc.RecordBatchReader:
+ """Stream Arrow batches while reading multiple splits concurrently.
+
+ Batches preserve input split order. Parallel reads are enabled only by
+ explicit configuration or the size- and storage-aware default.
+ PyArrow runtimes without ``RecordBatchReader.close`` and
+ ``from_stream`` fall back to serial reads.
+ """
effective_bp = self._resolve_blob_parallelism(blob_parallelism)
+ effective = self._resolve_parallelism(parallelism, len(splits))
+ configured = (
+ parallelism is not None or self._read_parallelism is not None
+ )
+ cancellable = _supports_cancellable_arrow_reader()
+ workers = (self._pipeline_workers(splits, effective, configured)
Review Comment:
[P1] Every no-argument caller now opts into auto fan-out.
`RayDatasource._get_read_task` still calls
`worker_table_read.to_arrow_batch_reader(splits)` without `parallelism=1`, so a
Ray read task containing multiple splits launches another local split thread
pool. This multiplies concurrency and remote object-store requests instead of
keeping Ray tasks serial as stated in the PR description. I reproduced one Ray
task with 8 splits entering `_pipelined_arrow_batch_generator(..., workers=4)`.
Please force `parallelism=1` in the Ray worker call and add a regression test.
##########
paimon-python/pypaimon/read/table_read.py:
##########
@@ -150,14 +169,150 @@ def _record_generator():
return _record_generator()
- def to_arrow_batch_reader(self, splits: List[Split],
- blob_parallelism: Optional[int] = None) ->
pyarrow.ipc.RecordBatchReader:
+ def to_arrow_batch_reader(
+ self,
+ splits: List[Split],
+ blob_parallelism: Optional[int] = None,
+ parallelism: Optional[int] = None,
+ ) -> pyarrow.ipc.RecordBatchReader:
+ """Stream Arrow batches while reading multiple splits concurrently.
+
+ Batches preserve input split order. Parallel reads are enabled only by
+ explicit configuration or the size- and storage-aware default.
+ PyArrow runtimes without ``RecordBatchReader.close`` and
+ ``from_stream`` fall back to serial reads.
+ """
effective_bp = self._resolve_blob_parallelism(blob_parallelism)
+ effective = self._resolve_parallelism(parallelism, len(splits))
+ configured = (
+ parallelism is not None or self._read_parallelism is not None
+ )
+ cancellable = _supports_cancellable_arrow_reader()
+ workers = (self._pipeline_workers(splits, effective, configured)
+ if cancellable and self.limit is None else 1)
schema = PyarrowFieldParser.from_paimon_schema(self.read_type)
if self.include_row_kind:
schema = self._add_row_kind_to_schema(schema)
- batch_iterator = self._arrow_batch_generator(splits, schema,
effective_bp)
- return pyarrow.ipc.RecordBatchReader.from_batches(schema,
batch_iterator)
+ if self.limit is not None and self.limit <= 0:
+ return pyarrow.ipc.RecordBatchReader.from_batches(schema, iter(()))
+ if (self.limit is None
+ and self._should_run_parallel(splits, workers)
+ and cancellable):
+ effective_bp = self._cap_blob_parallelism(workers, effective_bp)
+ cancel = threading.Event()
+ batch_iterator = self._pipelined_arrow_batch_generator(
+ splits, schema, effective_bp, workers, cancel)
+ return _create_cancellable_arrow_batch_reader(
+ schema, batch_iterator)
+ batch_iterator = self._arrow_batch_generator(
+ splits, schema, effective_bp)
+ return pyarrow.ipc.RecordBatchReader.from_batches(
+ schema, batch_iterator)
+
+ def _pipelined_arrow_batch_generator(
+ self,
+ splits: List[Split],
+ schema: pyarrow.Schema,
+ blob_parallelism: int,
+ workers: int,
+ cancel: Optional[threading.Event] = None,
+ ) -> Iterator[pyarrow.RecordBatch]:
+ cancel = cancel or threading.Event()
+ workers = min(workers, len(splits))
+ if workers < 2:
+ yield from self._arrow_batch_generator(
+ splits, schema, blob_parallelism)
+ return
+ end = object()
+ stop = object()
+ results = queue.Queue()
+ batch_slots = [threading.Semaphore(1) for _ in splits]
Review Comment:
[P2] This allocates one Python `Semaphore` per input split, and the code
below also enqueues every split tuple eagerly. The supposedly bounded streaming
path therefore adds O(total splits) coordination memory before producing the
first batch. On CPython 3.10, 100,000 semaphores alone consume about 111 MiB.
Please keep only O(workers) active slots/tasks (for example, a sliding
submission window) rather than per-split semaphores plus an unbounded task
queue.
--
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]
