This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 8dc607336d [python] Support read.parallelism=auto for single-process
reads (#8637)
8dc607336d is described below
commit 8dc607336d1e1494aecd38b06a6d1d32e02305c8
Author: chaoyang <[email protected]>
AuthorDate: Thu Jul 16 22:10:57 2026 +0800
[python] Support read.parallelism=auto for single-process reads (#8637)
---
.../pypaimon/common/options/core_options.py | 13 +--
paimon-python/pypaimon/read/table_read.py | 36 +++++----
.../pypaimon/tests/reader_parallel_test.py | 92 ++++++++++++++++++++--
3 files changed, 112 insertions(+), 29 deletions(-)
diff --git a/paimon-python/pypaimon/common/options/core_options.py
b/paimon-python/pypaimon/common/options/core_options.py
index dfcb3a6861..b21a74a798 100644
--- a/paimon-python/pypaimon/common/options/core_options.py
+++ b/paimon-python/pypaimon/common/options/core_options.py
@@ -854,13 +854,14 @@ class CoreOptions:
READ_PARALLELISM: ConfigOption[int] = (
ConfigOptions.key("read.parallelism")
.int_type()
- .default_value(1)
+ .no_default_value()
.with_description(
"Parallelism for reading splits within a single TableRead call. "
- "The value 1 (default) keeps reads serial. Values >= 2 enable a "
- "thread pool that reads splits concurrently and assembles the "
- "result in input order. Has no effect when fewer than 2 splits "
- "are passed.")
+ "When unset (the default), reads auto-scale to "
+ "min(number of splits, CPU count). Set to 1 to force serial "
+ "reads, or to a specific value >= 1 to cap the thread pool that "
+ "reads splits concurrently and assembles the result in input "
+ "order. Has no effect when fewer than 2 splits are passed.")
)
ADD_COLUMN_BEFORE_PARTITION: ConfigOption[bool] = (
@@ -1349,7 +1350,7 @@ class CoreOptions:
def read_batch_size(self, default=None) -> int:
return self.options.get(CoreOptions.READ_BATCH_SIZE, default or 1024)
- def read_parallelism(self, default=None) -> int:
+ def read_parallelism(self, default=None) -> Optional[int]:
return self.options.get(CoreOptions.READ_PARALLELISM, default)
def add_column_before_partition(self) -> bool:
diff --git a/paimon-python/pypaimon/read/table_read.py
b/paimon-python/pypaimon/read/table_read.py
index d95dfe556b..89ca539b73 100644
--- a/paimon-python/pypaimon/read/table_read.py
+++ b/paimon-python/pypaimon/read/table_read.py
@@ -15,6 +15,7 @@
# specific language governing permissions and limitations
# under the License.
+import os
import threading
from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import Any, Dict, Iterator, List, Optional
@@ -180,11 +181,14 @@ class TableRead:
splits: scan-plan splits returned from a ``TableScan``.
parallelism: optional runtime override of the
``read.parallelism`` table option. ``None`` (default) falls
- back to the table option; a non-None value temporarily
- overrides it for this call. ``1`` keeps reads serial;
- ``>= 2`` enables a thread pool that reads splits
- concurrently and assembles the final table in input order.
- Must be ``>= 1``.
+ back to the table option; when that is also unset the read
+ auto-scales to ``min(number of splits, CPU count)``. ``1``
+ keeps reads serial; ``>= 2`` caps the thread pool that reads
+ splits concurrently and assembles the final table in input
+ order. Must be ``>= 1``. Note that with ``>= 2`` (or auto)
+ and a ``limit`` set, the returned rows are an arbitrary
+ subset of the requested size, since which splits fill the row
+ quota first is non-deterministic.
blob_parallelism: number of threads for concurrent blob reads
within each batch. ``None`` or ``1`` (default) reads blobs
serially; ``>= 2`` uses a thread pool with ``pread`` for
@@ -195,8 +199,7 @@ class TableRead:
shrunk to stay within it.
"""
effective_bp = self._resolve_blob_parallelism(blob_parallelism)
- # TODO: default read.parallelism to min(splits, cpu_count()) once
stable
- effective = self._resolve_parallelism(parallelism)
+ effective = self._resolve_parallelism(parallelism, len(splits))
schema = PyarrowFieldParser.from_paimon_schema(self.read_type)
if self.include_row_kind:
schema = self._add_row_kind_to_schema(schema)
@@ -281,20 +284,23 @@ class TableRead:
finally:
reader.close()
- def _resolve_parallelism(self, runtime: Optional[int]) -> int:
+ def _resolve_parallelism(self, runtime: Optional[int], num_splits: int) ->
int:
"""Pick the effective parallelism and reject illegal values.
Priority: explicit ``parallelism`` argument > ``read.parallelism``
- table option > built-in default of 1. The validation message names
- whichever source produced the offending value, so users know where
- to fix it.
+ table option > auto. When neither the argument nor the option is set
+ the read auto-scales to ``min(num_splits, CPU count)``. A value >= 1
+ caps the thread pool; ``1`` forces serial reads. The validation
+ message names whichever source produced the offending value.
"""
if runtime is not None:
- value = runtime
- source = "parallelism"
+ value, source = runtime, "parallelism"
+ elif self._read_parallelism is not None:
+ value, source = self._read_parallelism, "read.parallelism"
else:
- value = self._read_parallelism
- source = "read.parallelism"
+ # os.cpu_count() may return None on exotic platforms; fall back
+ # to 1. min() with num_splits avoids more workers than work.
+ return max(1, min(num_splits, os.cpu_count() or 1))
if value < 1:
raise ValueError(f"{source} must be >= 1, got {value}")
return value
diff --git a/paimon-python/pypaimon/tests/reader_parallel_test.py
b/paimon-python/pypaimon/tests/reader_parallel_test.py
index 9e9897fb50..ea0c3d8f61 100644
--- a/paimon-python/pypaimon/tests/reader_parallel_test.py
+++ b/paimon-python/pypaimon/tests/reader_parallel_test.py
@@ -19,6 +19,7 @@ import os
import shutil
import tempfile
import threading
+import types
import unittest
from unittest import mock
@@ -28,6 +29,50 @@ from pypaimon import CatalogFactory, Schema
from pypaimon.read.table_read import TableRead, _RemainingRows
+class ResolveParallelismTest(unittest.TestCase):
+ """Pure unit tests for parallelism resolution — no Paimon table needed.
+
+ ``_resolve_parallelism`` only reads ``self._read_parallelism``, so a
+ lightweight stand-in with that attribute is enough to exercise it.
+ """
+
+ @staticmethod
+ def _resolve(runtime, num_splits, option=None):
+ stub = types.SimpleNamespace(_read_parallelism=option)
+ return TableRead._resolve_parallelism(stub, runtime, num_splits)
+
+ def test_runtime_int_passthrough(self):
+ self.assertEqual(self._resolve(4, 10), 4)
+ self.assertEqual(self._resolve(1, 10), 1)
+
+ def test_runtime_overrides_option(self):
+ self.assertEqual(self._resolve(2, 10, option=8), 2)
+
+ def test_option_used_when_no_runtime(self):
+ self.assertEqual(self._resolve(None, 10, option=8), 8)
+
+ def test_none_resolves_to_auto_min_splits_cpu(self):
+ cpu = os.cpu_count() or 1
+ # Neither runtime nor option set => auto.
+ # More splits than CPUs => capped at CPU count.
+ self.assertEqual(self._resolve(None, cpu + 100), cpu)
+ # Fewer splits than CPUs => capped at split count.
+ self.assertEqual(self._resolve(None, 1), 1)
+ # Zero splits never yields a sub-1 worker count.
+ self.assertEqual(self._resolve(None, 0), 1)
+
+ def test_invalid_runtime_raises_with_source(self):
+ with self.assertRaises(ValueError) as ctx:
+ self._resolve(0, 10)
+ self.assertIn("parallelism", str(ctx.exception))
+ self.assertNotIn("read.parallelism", str(ctx.exception))
+
+ def test_invalid_option_raises_with_source(self):
+ with self.assertRaises(ValueError) as ctx:
+ self._resolve(None, 10, option=0)
+ self.assertIn("read.parallelism", str(ctx.exception))
+
+
class RemainingRowsTest(unittest.TestCase):
"""Pure unit tests for the row-quota counter — no Paimon table needed."""
@@ -109,11 +154,15 @@ class ParallelReaderAppendOnlyTest(unittest.TestCase):
'dt': dts,
}, schema=cls.pa_schema)
- # Default table — read.parallelism unset (defaults to 1, i.e. serial).
+ # Default table — read.parallelism unset => auto (parallel when
+ # there are >= 2 splits and >= 2 CPUs).
cls.table = cls._build_table('append_parallel_default', None, data)
# Option-set table — read.parallelism=4 baked into the table schema.
cls.table_opt_4 = cls._build_table(
'append_parallel_opt4', {'read.parallelism': '4'}, data)
+ # Option forcing serial reads.
+ cls.table_opt_1 = cls._build_table(
+ 'append_parallel_opt1', {'read.parallelism': '1'}, data)
@classmethod
def _build_table(cls, name, options, data):
@@ -152,7 +201,7 @@ class ParallelReaderAppendOnlyTest(unittest.TestCase):
rb = self.table.new_read_builder()
splits = self._scan_splits(rb)
read = rb.new_read()
- serial = read.to_arrow(splits)
+ serial = read.to_arrow(splits, parallelism=1)
parallel = read.to_arrow(splits, parallelism=4)
# Same split order preserved => byte-identical tables.
self.assertEqual(serial, parallel)
@@ -169,13 +218,38 @@ class ParallelReaderAppendOnlyTest(unittest.TestCase):
splits_serial = self._scan_splits(rb_serial)
splits_parallel = self._scan_splits(rb_parallel)
- serial_df = rb_serial.new_read().to_pandas(splits_serial) \
+ serial_df = rb_serial.new_read().to_pandas(splits_serial,
parallelism=1) \
.sort_values('user_id').reset_index(drop=True)
# No explicit parallelism — must pick up read.parallelism=4 from the
table option.
parallel_df = rb_parallel.new_read().to_pandas(splits_parallel) \
.sort_values('user_id').reset_index(drop=True)
self.assertTrue(serial_df.equals(parallel_df))
+ def test_default_none_runs_auto_parallel(self):
+ # No runtime arg and no table option => auto. With >= 2 splits on a
+ # multi-core box this takes the parallel fan-out path; the result
+ # must still match a forced-serial read row for row.
+ read = self.table.new_read_builder().new_read()
+ splits = self._scan_splits(self.table.new_read_builder())
+ self.assertGreaterEqual(len(splits), 2)
+ serial = read.to_arrow(splits, parallelism=1)
+ auto = read.to_arrow(splits)
+ self.assertEqual(serial, auto)
+ self.assertEqual(auto.num_rows, self.expected_rows)
+
+ def test_default_none_auto_takes_parallel_path_when_multicore(self):
+ read = self.table.new_read_builder().new_read()
+ splits = self._scan_splits(self.table.new_read_builder())
+ # Only assert fan-out where the box actually has >= 2 CPUs, else
+ # auto legitimately resolves to 1 and stays serial.
+ if (os.cpu_count() or 1) < 2:
+ self.skipTest("single-core runner: auto resolves to serial")
+ with mock.patch.object(
+ read, '_to_arrow_parallel', wraps=read._to_arrow_parallel
+ ) as spy:
+ read.to_arrow(splits)
+ spy.assert_called_once()
+
# ------------------------------------------------------------------
# Priority: method arg > table option > built-in default.
# ------------------------------------------------------------------
@@ -190,9 +264,9 @@ class ParallelReaderAppendOnlyTest(unittest.TestCase):
read.to_arrow(splits, parallelism=1)
def test_method_arg_overrides_option_to_parallel(self):
- # option=1 (default) but caller passes 4: should enable parallelism.
- read = self.table.new_read_builder().new_read()
- splits = self._scan_splits(self.table.new_read_builder())
+ # option=1 (forces serial) but caller passes 4: should enable
parallelism.
+ read = self.table_opt_1.new_read_builder().new_read()
+ splits = self._scan_splits(self.table_opt_1.new_read_builder())
with mock.patch.object(
read, '_to_arrow_parallel', wraps=read._to_arrow_parallel
) as spy:
@@ -204,8 +278,10 @@ class ParallelReaderAppendOnlyTest(unittest.TestCase):
# Boundary / invalid value handling.
# ------------------------------------------------------------------
- def test_parallelism_one_equals_serial(self):
- rb = self.table.new_read_builder()
+ def test_option_one_equals_serial(self):
+ # Table option read.parallelism=1 must behave exactly like an
+ # explicit parallelism=1 (serial) read.
+ rb = self.table_opt_1.new_read_builder()
splits = self._scan_splits(rb)
read = rb.new_read()
self.assertEqual(read.to_arrow(splits),
