leaves12138 commented on code in PR #8807:
URL: https://github.com/apache/paimon/pull/8807#discussion_r3729929506
##########
paimon-python/pypaimon/read/table_read.py:
##########
@@ -333,6 +878,290 @@ def _should_run_parallel(
"""
return effective >= 2 and len(splits) >= 2
+ def _pipeline_workers(
+ self,
+ splits: List[Split],
+ effective: int,
+ configured: bool,
+ ) -> int:
+ maximum = min(
+ effective, len(splits), _MAX_PIPELINE_SPLIT_WORKERS)
+ if configured or maximum < 2:
+ return maximum
+
+ total_bytes = sum(
+ self._estimated_split_file_size(split) for split in splits)
+ local = self._pipeline_reads_are_local(splits)
+ if total_bytes <= 0:
+ return (1 if local else
+ min(maximum, self._PIPELINE_REMOTE_MIN_WORKERS))
+ bytes_per_worker = (self._PIPELINE_LOCAL_BYTES_PER_WORKER
+ if local
+ else self._PIPELINE_REMOTE_BYTES_PER_WORKER)
+ estimated = max(
+ 1, (total_bytes + bytes_per_worker - 1) // bytes_per_worker)
+ if not local:
+ # A few remote requests in parallel hide object-store/HDFS RTT even
+ # when their aggregate bytes are below the throughput threshold.
+ estimated = max(
+ estimated, min(maximum, self._PIPELINE_REMOTE_MIN_WORKERS))
+ return min(maximum, estimated)
+
+ def _initial_batch_reservation(
+ self,
+ split: Split,
+ workers: int,
+ fixed_row_size: Optional[int] = None,
+ batch_size: Optional[int] = None,
+ reserve_full_budget: bool = False,
+ ) -> int:
+ """Reserve decoded bytes before pulling a batch from a split."""
+ if reserve_full_budget:
+ return self._PIPELINE_READAHEAD_BYTES
+
+ if fixed_row_size is not None and batch_size is not None:
+ row_count = max(0, int(getattr(split, "row_count", 0) or 0))
+ batch_rows = (
+ min(row_count, batch_size) if row_count > 0 else batch_size
+ )
+ return max(1, batch_rows * fixed_row_size)
+
+ file_size = self._estimated_split_file_size(split)
+ row_count = max(0, int(getattr(split, "row_count", 0) or 0))
+ read_batch_size = self._configured_read_batch_size()
+ fair_share = self._pipeline_fair_share(workers)
+ if file_size <= 0 or row_count <= 0 or read_batch_size <= 0:
+ # Explicit parallelism may be used with custom splits which do not
+ # expose estimates. Divide the budget fairly across workers rather
+ # than admitting every unknown decode without a reservation.
+ return fair_share
+
+ batch_rows = min(row_count, read_batch_size)
+ encoded_estimate = (
+ file_size * batch_rows + row_count - 1
+ ) // row_count
+ return min(
+ fair_share,
+ max(1, encoded_estimate * self._PIPELINE_DECODE_EXPANSION_FACTOR),
+ )
+
+ def _configured_read_batch_size(self) -> int:
+ table = getattr(self, "table", None)
+ options = getattr(table, "options", None)
+ value = (
+ options.read_batch_size()
+ if options is not None and hasattr(options, "read_batch_size")
+ else 1024
+ )
+ return max(1, value)
+
+ def _pipeline_fair_share(self, workers: int) -> int:
+ lookahead_decodes = workers * self._PIPELINE_LOOKAHEAD_FACTOR
+ return max(
+ 1,
+ self._PIPELINE_READAHEAD_BYTES // max(1, lookahead_decodes),
+ )
+
+ def _pipeline_batch_size(
+ self, fixed_row_size: Optional[int], workers: int
+ ) -> int:
+ configured = self._configured_read_batch_size()
+ if fixed_row_size is None:
+ return configured
+ fair_rows = self._pipeline_fair_share(workers) // fixed_row_size
+ return max(1, min(configured, fair_rows))
+
+ def _pipeline_decoded_schema(
+ self, output_schema: pyarrow.Schema
+ ) -> pyarrow.Schema:
+ """Return the widest known schema materialized before projection."""
+ scan_read_type = getattr(self, "_scan_read_type", None)
Review Comment:
`_scan_read_type` is not always the widest schema materialized before
projection. For a non-raw primary-key split, `MergeFileSplitRead` prepends the
trimmed `_KEY_*` columns plus `_SEQUENCE_NUMBER` and `_VALUE_KIND`, and may
also inject a projected-out `sequence.field`. I reproduced this with a table
whose PK contains a `STRING` column while the requested projection is only an
`INT`: this method returns a fixed-width `payload: int32` schema
(`fixed_row_size == 5`), while the actual merge reader schema contains
`_KEY_pk: string`. The pipeline therefore skips `decode_lock`, reserves only
the projected fixed-width bytes, and lets multiple workers decode arbitrarily
large hidden string keys concurrently; resizing from the final projected batch
cannot account for those temporary buffers. This reopens the pre-decode
memory-bound issue for narrow PK projections. Please derive the classification
from the actual per-split internal read schema, or conservatively serialize
when merge/data-ev
olution readers can materialize hidden variable-width fields, and add a
regression test for a fixed-width projection over a variable-width PK/sequence
field.
--
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]
