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 9a46844453 [python] Support concurrent blob reads for raw and Daft
APIs (#8406)
9a46844453 is described below
commit 9a4684445343c44f14bdf0a1fd0fbc5b0816c7cf
Author: XiaoHongbo <[email protected]>
AuthorDate: Thu Jul 2 18:15:15 2026 +0800
[python] Support concurrent blob reads for raw and Daft APIs (#8406)
Native daft File.open() reads blobs one at a time and rebuilds a client
per call while holding the GIL, so threads/async give no speedup. Add
pypaimon-backed alternatives that use a shared FileIO with concurrent
ranged reads (pread/read_at release the GIL during network I/O)
---
docs/docs/pypaimon/blob.md | 55 ++++----
docs/docs/pypaimon/daft.md | 71 +++++------
paimon-python/pypaimon/common/file_io.py | 114 +++++++++++++++++
paimon-python/pypaimon/daft/__init__.py | 4 +-
paimon-python/pypaimon/daft/daft_blob_read.py | 99 +++++++++++++++
.../read/reader/blob_descriptor_convert_reader.py | 16 ++-
.../pypaimon/read/reader/format_blob_reader.py | 23 +++-
paimon-python/pypaimon/read/split_read.py | 11 +-
paimon-python/pypaimon/read/table_read.py | 58 +++++++--
paimon-python/pypaimon/tests/blob_test.py | 139 ++++++++++++++++++++-
.../pypaimon/tests/daft/daft_blob_read_test.py | 133 ++++++++++++++++++++
.../pypaimon/tests/reader_parallel_test.py | 4 +-
12 files changed, 642 insertions(+), 85 deletions(-)
diff --git a/docs/docs/pypaimon/blob.md b/docs/docs/pypaimon/blob.md
index 9147a96873..ddb9c40be4 100644
--- a/docs/docs/pypaimon/blob.md
+++ b/docs/docs/pypaimon/blob.md
@@ -78,14 +78,32 @@ writer.close()
## Reading Blob Data
-Use `row.get_blob(pos)` to access blob columns. It returns a `Blob` object
-regardless of how the blob is stored.
+### Batch reading (recommended)
+
+Use `to_arrow_batch_reader` to read blob data in batches. Set
+`blob_parallelism` to enable concurrent blob reads within each batch:
```python
read_builder = table.new_read_builder()
splits = read_builder.new_scan().plan().splits()
read = read_builder.new_read()
+for batch in read.to_arrow_batch_reader(splits, blob_parallelism=16):
+ for i in range(len(batch)):
+ image_bytes = batch['image'][i].as_py()
+```
+
+Or read all data into a single Arrow Table:
+
+```python
+arrow_table = read.to_arrow(splits, blob_parallelism=16)
+```
+
+### Row-by-row reading
+
+Use `row.get_blob(pos)` to access blob columns one row at a time:
+
+```python
for row in read.to_iterator(splits):
blob = row.get_blob(2)
if blob is None:
@@ -93,38 +111,31 @@ for row in read.to_iterator(splits):
data = blob.to_data()
```
-## Streaming for Large Blobs
-
-`blob.new_input_stream()` returns a file-like object. Whether it is
-genuinely lazy depends on how the table is configured:
-
-- Default mode (`blob-as-descriptor=false`): the read path materialises
- the payload before it reaches `row.get_blob(pos)`. `Blob` is a
- `BlobData` and `new_input_stream()` wraps the in-memory bytes — not
- true streaming. For large blobs this can still OOM.
-- Descriptor mode (`blob-as-descriptor=true`): the read path preserves
- the descriptor. `Blob` is a `BlobRef` and `new_input_stream()` opens
- the underlying file on demand.
+### Streaming / partial reads
-This mirrors Java's `BlobFormatReader` semantics.
-
-For genuine on-demand streaming of large blobs (videos, model weights),
-use `table.copy` to set `blob-as-descriptor=true` before reading:
+For true on-demand streaming (large blobs like videos or model weights),
+set `blob-as-descriptor=true` so blob values are kept as lightweight
+references instead of being materialized into memory:
```python
-table = catalog.get_table('my_db.image_table')
table = table.copy({'blob-as-descriptor': 'true'})
read_builder = table.new_read_builder()
splits = read_builder.new_scan().plan().splits()
read = read_builder.new_read()
-# Reads now return BlobRef whose new_input_stream() is lazy.
for row in read.to_iterator(splits):
- with row.get_blob(2).new_input_stream() as stream:
- chunk = stream.read(1024)
+ blob = row.get_blob(2)
+ if blob is None:
+ continue
+ with blob.new_input_stream() as stream:
+ chunk = stream.read(4096)
```
+Without `blob-as-descriptor=true`, blob values are materialized before
+`row.get_blob(...)` returns; `new_input_stream()` then reads from
+in-memory bytes, not from storage.
+
## Lower-level: `Blob.from_bytes`
When you already have raw or descriptor bytes (for example from a custom
diff --git a/docs/docs/pypaimon/daft.md b/docs/docs/pypaimon/daft.md
index c553bc2fbd..0c2c118f0f 100644
--- a/docs/docs/pypaimon/daft.md
+++ b/docs/docs/pypaimon/daft.md
@@ -268,69 +268,58 @@ result = df.with_column("size", file_length(col("image")))
result.show()
```
-To read actual blob content (e.g., decode an image), use `file.open()`.
-Only filtered rows trigger I/O:
+### Reading blob content
-```python
[email protected]
-def decode_image(file: daft.File) -> str:
- with file.open() as f:
- data = f.read()
- return f"decoded {len(data)} bytes"
-
-result = df.with_column("info", decode_image(col("image")))
-result.show()
-```
-
-### Parallel blob reading
-
-By default, `@daft.func` processes rows sequentially. To read blobs
-concurrently within a single worker, use an async UDF with
-`max_concurrency` and `asyncio.to_thread` (because `file.open().read()`
-is synchronous blocking I/O):
+Use `read_blob` to read a blob `File` column to `binary` bytes. It reads blobs
+concurrently, so it is much faster than a per-row `file.open()` loop:
```python
-import asyncio
+from daft import col
+from pypaimon.daft import read_paimon, read_blob
-def _read_blob(file: daft.File | None) -> str | None:
- if file is None:
- return None
- with file.open() as f:
- data = f.read()
- return f"decoded {len(data)} bytes"
+catalog_options = {"warehouse": "oss://my-bucket/warehouse"}
+table = "my_db.image_table"
[email protected](max_concurrency=8)
-async def decode_image(file: daft.File | None) -> str | None:
- return await asyncio.to_thread(_read_blob, file)
+df = read_paimon(table, catalog_options)
+df = df.where(col("id") < 100) # filter before reading bytes
-result = df.with_column("info", decode_image(col("image")))
+result = df.select(
+ col("id"),
+ read_blob(col("image"), catalog_options, table).alias("image_bytes"),
+)
result.show()
```
-When running on Ray, the UDF calls are distributed across Ray workers
-automatically — each worker processes its partition in parallel, giving you
-batch-level concurrency without any extra code.
+Tune concurrency with `max_concurrency` (default 64).
-### Streaming (chunk) reads
+### Streaming / partial reads with `open_blob`
-`file.open()` returns a seekable stream that supports `read(size)`, so you can
-process large blobs in chunks without loading everything into memory:
+For large blobs (videos, model weights) where you don't want to load
+everything into memory, use `open_blob` to get a seekable stream:
```python
+import daft
+from daft import col
+from pypaimon.daft import read_paimon, open_blob
+
+catalog_options = {"warehouse": "oss://my-bucket/warehouse"}
+table = "my_db.image_table"
+
+df = read_paimon(table, catalog_options)
+
@daft.func(return_dtype=daft.DataType.binary())
def first_4k(file: daft.File) -> bytes | None:
if file is None:
return None
- with file.open() as f:
+ with open_blob(file, catalog_options, table) as f:
return f.read(4096)
result = df.with_column("header", first_4k(col("image")))
-result.show()
```
-See [Blob Storage — Streaming for Large
Blobs](./blob#streaming-for-large-blobs)
-for more details on the underlying `OffsetInputStream` API (`read(size)` /
-`seek()` / `tell()`).
+`open_blob` is a drop-in replacement for `file.open()` for streaming an
+individual large blob. It reads one blob per call, so for bulk reads of a
+whole column prefer `read_blob`.
## Catalog Abstraction
diff --git a/paimon-python/pypaimon/common/file_io.py
b/paimon-python/pypaimon/common/file_io.py
index 9f35140108..6978914f07 100644
--- a/paimon-python/pypaimon/common/file_io.py
+++ b/paimon-python/pypaimon/common/file_io.py
@@ -48,6 +48,37 @@ def pread(stream, length: int, offset: int) -> bytes:
return os.pread(stream.fileno(), length, offset)
+# Coalescing bounds: merge same-file ranges whose gap is within GAP, capping a
+# merged read at SPAN so threads stay busy and memory stays bounded.
+_COALESCE_GAP = 1 << 20
+_COALESCE_SPAN = 8 << 20
+
+
+def _coalesce_ranges(items, max_gap, max_span):
+ """Group ``(idx, path, offset, length)`` (length >= 0) into merged spans:
+ ``[(path, span_offset, span_length, [(idx, offset, length), ...])]``."""
+ from collections import defaultdict
+ by_path = defaultdict(list)
+ for it in items:
+ by_path[it[1]].append(it)
+ spans = []
+ for path, group in by_path.items():
+ group.sort(key=lambda x: x[2])
+ cur, start, end = [], None, None
+ for idx, _, off, length in group:
+ stop = off + length
+ if cur and off - end <= max_gap and stop - start <= max_span:
+ cur.append((idx, off, length))
+ end = max(end, stop)
+ else:
+ if cur:
+ spans.append((path, start, end - start, cur))
+ cur, start, end = [(idx, off, length)], off, stop
+ if cur:
+ spans.append((path, start, end - start, cur))
+ return spans
+
+
class FileIO(ABC):
"""
File IO interface to read and write files.
@@ -132,6 +163,89 @@ class FileIO(ABC):
else:
self.mkdirs(path)
+ def read_file_range(self, path, offset, length):
+ """Read a byte range. Thread-safe. ``length < 0`` = read to EOF (pread
+ can't express that, so seek + read)."""
+ stream = self.new_input_stream(path)
+ try:
+ if length >= 0 and supports_pread(stream):
+ return pread(stream, length, offset)
+ stream.seek(offset)
+ return stream.read() if length < 0 else stream.read(length)
+ finally:
+ stream.close()
+
+ def read_ranges_coalesced(self, ranges, parallelism,
+ max_gap=_COALESCE_GAP, max_span=_COALESCE_SPAN):
+ """Read ``ranges`` (each ``None`` or ``(path, offset, length)``),
returning
+ bytes in the same order. Same-file nearby ranges are merged into one
read
+ to cut round trips, then sliced; reads run on a thread pool. Negative
+ length (read to EOF) is read on its own, never merged.
+
+ A failed read propagates and aborts the whole batch (unlike a per-row
+ ``file.open()`` loop that fails one row at a time).
+ """
+ from concurrent.futures import ThreadPoolExecutor
+ # Threads write disjoint results[idx]; safe under the GIL (no list
resize).
+ results: List[Optional[bytes]] = [None] * len(ranges)
+ coalescible, singletons = [], []
+ for i, r in enumerate(ranges):
+ # None path/offset/length => null blob, leave result None.
+ if r is None or r[0] is None or r[1] is None or r[2] is None:
+ continue
+ path, offset, length = r
+ if length < 0: # unknown length => read to EOF, never coalesced
+ singletons.append((i, path, offset, length))
+ else:
+ coalescible.append((i, path, offset, length))
+
+ spans = _coalesce_ranges(coalescible, max_gap, max_span)
+
+ def _run(task):
+ kind, payload = task
+ if kind == "span":
+ path, span_off, span_len, members = payload
+ buf = self.read_file_range(path, span_off, span_len)
+ for idx, off, length in members:
+ s = off - span_off
+ results[idx] = buf[s:s + length]
+ else:
+ idx, path, off, length = payload
+ results[idx] = self.read_file_range(path, off, length)
+
+ tasks = [("span", s) for s in spans] + [("one", g) for g in singletons]
+ if not tasks:
+ return results
+ workers = max(1, min(parallelism, len(tasks)))
+ with ThreadPoolExecutor(workers) as pool:
+ list(pool.map(_run, tasks))
+ return results
+
+ def read_blobs_concurrent(self, blobs, parallelism):
+ """Read a list of Blobs concurrently, coalescing same-file ranged
reads.
+
+ ``BlobRef`` values expose a file range and are coalesced; in-memory
+ ``BlobData`` values are returned directly.
+ """
+ from pypaimon.table.row.blob import BlobRef
+ results: List[Optional[bytes]] = [None] * len(blobs)
+ ranges: List[Optional[tuple]] = [None] * len(blobs)
+ inmem = []
+ for i, b in enumerate(blobs):
+ if b is None:
+ continue
+ if isinstance(b, BlobRef):
+ d = b.to_descriptor()
+ ranges[i] = (d.uri, d.offset, d.length)
+ else:
+ inmem.append((i, b))
+ for i, v in enumerate(self.read_ranges_coalesced(ranges, parallelism)):
+ if v is not None:
+ results[i] = v
+ for idx, b in inmem:
+ results[idx] = b.to_data()
+ return results
+
def read_file_utf8(self, path: str) -> str:
with self.new_input_stream(path) as input_stream:
return input_stream.read().decode('utf-8')
diff --git a/paimon-python/pypaimon/daft/__init__.py
b/paimon-python/pypaimon/daft/__init__.py
index b854830173..5749a6d3e7 100644
--- a/paimon-python/pypaimon/daft/__init__.py
+++ b/paimon-python/pypaimon/daft/__init__.py
@@ -17,8 +17,10 @@
################################################################################
from pypaimon.daft.daft_paimon import explain_paimon_scan, read_paimon,
write_paimon
+from pypaimon.daft.daft_blob_read import open_blob, read_blob
-__all__ = ["explain_paimon_scan", "read_paimon", "write_paimon",
"PaimonCatalog", "PaimonTable"]
+__all__ = ["explain_paimon_scan", "read_paimon", "write_paimon", "read_blob",
+ "open_blob", "PaimonCatalog", "PaimonTable"]
def __getattr__(name):
diff --git a/paimon-python/pypaimon/daft/daft_blob_read.py
b/paimon-python/pypaimon/daft/daft_blob_read.py
new file mode 100644
index 0000000000..f27600b0fb
--- /dev/null
+++ b/paimon-python/pypaimon/daft/daft_blob_read.py
@@ -0,0 +1,99 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+
+"""Blob I/O helpers for the Daft integration."""
+
+from __future__ import annotations
+
+from functools import lru_cache
+from typing import BinaryIO, Dict
+
+
+@lru_cache(maxsize=16)
+def _get_file_io(catalog_key: tuple, table: str):
+ from pypaimon.catalog.catalog_factory import CatalogFactory
+ catalog_options = dict(catalog_key)
+ return CatalogFactory.create(catalog_options).get_table(table).file_io
+
+
+def _cache_key(catalog_options: Dict[str, str]) -> tuple:
+ return tuple(sorted(catalog_options.items()))
+
+
+def _file_range_fields():
+ from pypaimon.daft.daft_compat import file_range_position_field,
file_range_size_field
+ return file_range_position_field(), file_range_size_field()
+
+
+def _range_attr(file, name):
+ # Prefer _inner accessors: they read the embedded range; public File.size
+ # does a network stat (~200ms/blob).
+ inner = getattr(file, "_inner", None)
+ src = inner if inner is not None and hasattr(inner, name) else file
+ v = getattr(src, name)
+ return v() if callable(v) else v
+
+
+def _resolve_file_range(file, pos_field=None, size_field=None):
+ if pos_field is None:
+ pos_field, size_field = _file_range_fields()
+ return (_range_attr(file, "path"),
+ _range_attr(file, pos_field),
+ _range_attr(file, size_field))
+
+
+def open_blob(file, catalog_options: Dict[str, str], table: str) -> BinaryIO:
+ """Open a Daft ``File`` as a seekable stream via pypaimon's FileIO."""
+ from pypaimon.table.row.blob import OffsetInputStream
+ path, offset, length = _resolve_file_range(file)
+ fio = _get_file_io(_cache_key(catalog_options), table)
+ stream = fio.new_input_stream(path)
+ try:
+ return OffsetInputStream(stream, offset, length)
+ except Exception:
+ stream.close()
+ raise
+
+
+def read_blob(column, catalog_options: Dict[str, str], table: str, *,
max_concurrency: int = 64):
+ """Read a blob ``File`` column to ``binary`` bytes via concurrent ranged
reads."""
+ import daft
+ from daft import DataType
+
+ @daft.func.batch(return_dtype=DataType.binary())
+ def _read_blobs(files):
+ pos_field, size_field = _file_range_fields()
+ # Vectorized field extraction: per-File attr access (to_pylist +
+ # f.path/.position/.size) is GIL-bound and serializes the pool. daft
+ # File is an arrow ExtensionArray; .storage holds url/offset/size.
+ try:
+ struct = files.to_arrow()
+ struct = getattr(struct, "storage", struct)
+ ranges = list(zip(struct.field("url").to_pylist(),
+ struct.field(pos_field).to_pylist(),
+ struct.field(size_field).to_pylist()))
+ except (AttributeError, KeyError, TypeError, ValueError):
+ # Unexpected column layout: fall back to per-File resolution
(cheap).
+ ranges = [None if f is None else _resolve_file_range(f, pos_field,
size_field)
+ for f in files.to_pylist()]
+
+ fio = _get_file_io(_cache_key(catalog_options), table)
+ # Coalesce same-file adjacent reads to cut per-request round trips.
+ return fio.read_ranges_coalesced(ranges, max_concurrency)
+
+ return _read_blobs(column)
diff --git
a/paimon-python/pypaimon/read/reader/blob_descriptor_convert_reader.py
b/paimon-python/pypaimon/read/reader/blob_descriptor_convert_reader.py
index 12e975bae5..92b30ebbc4 100644
--- a/paimon-python/pypaimon/read/reader/blob_descriptor_convert_reader.py
+++ b/paimon-python/pypaimon/read/reader/blob_descriptor_convert_reader.py
@@ -41,7 +41,8 @@ class BlobInlineConvertReader(RecordBatchReader):
"""
def __init__(self, inner: RecordBatchReader, table,
- prescan_reader_factory: Optional[Callable[[Set[str]],
RecordBatchReader]] = None):
+ prescan_reader_factory: Optional[Callable[[Set[str]],
RecordBatchReader]] = None,
+ blob_parallelism: int = 1):
"""
Args:
inner: The main data reader (reads all columns).
@@ -50,10 +51,12 @@ class BlobInlineConvertReader(RecordBatchReader):
reader projecting only the specified field names. Used for
prescan to collect BlobViewStructs without reading all columns.
Signature: (field_names: Set[str]) -> RecordBatchReader
+ blob_parallelism: number of threads for concurrent blob reads.
"""
self._inner = inner
self._table = table
self._prescan_reader_factory = prescan_reader_factory
+ self._blob_parallelism = blob_parallelism
self.file_io = inner.file_io
self.blob_field_indices = inner.blob_field_indices
# Preserve original BlobViewStruct bytes when resolve disabled: skip
both
@@ -170,10 +173,13 @@ class BlobInlineConvertReader(RecordBatchReader):
if field_name not in batch.schema.names:
continue
values = [self._normalize_blob_to_bytes(v) for v in
batch.column(field_name).to_pylist()]
- converted_values = []
- for value in values:
- blob = Blob.from_bytes(value, self._table.file_io)
- converted_values.append(blob.to_data() if blob else None)
+ blobs = [Blob.from_bytes(v, self._table.file_io) for v in values]
+
+ if self._blob_parallelism > 1:
+ converted_values = self._table.file_io.read_blobs_concurrent(
+ blobs, self._blob_parallelism)
+ else:
+ converted_values = [b.to_data() if b else None for b in blobs]
column_idx = batch.schema.names.index(field_name)
batch = batch.set_column(
diff --git a/paimon-python/pypaimon/read/reader/format_blob_reader.py
b/paimon-python/pypaimon/read/reader/format_blob_reader.py
index 52f197097f..377d257062 100644
--- a/paimon-python/pypaimon/read/reader/format_blob_reader.py
+++ b/paimon-python/pypaimon/read/reader/format_blob_reader.py
@@ -37,12 +37,14 @@ class FormatBlobReader(RecordBatchReader):
def __init__(self, file_io: FileIO, file_path: str, read_fields: List[str],
full_fields: List[DataField], push_down_predicate: Any,
blob_as_descriptor: bool,
- batch_size: int = 1024, row_indices: Optional[Any] = None):
+ batch_size: int = 1024, row_indices: Optional[Any] = None,
+ blob_parallelism: int = 1):
self._file_io = file_io
self._file_path = file_path
self._push_down_predicate = push_down_predicate
self._blob_as_descriptor = blob_as_descriptor
self._batch_size = batch_size
+ self._blob_parallelism = blob_parallelism
# Initialize the low-level blob format reader
self.file_path = file_path
@@ -57,7 +59,9 @@ class FormatBlobReader(RecordBatchReader):
self._input_stream = file_io.new_input_stream(file_path)
self._read_index()
self._apply_row_indices(row_indices)
- if self._blob_as_descriptor:
+ # Drop the shared stream: descriptor/concurrent reads yield
BlobRefs
+ # that each open their own stream (one stream isn't thread-safe).
+ if self._blob_as_descriptor or self._blob_parallelism > 1:
self._input_stream.close()
self._input_stream = None
@@ -96,6 +100,7 @@ class FormatBlobReader(RecordBatchReader):
# Collect records for this batch
pydict_data = {name: [] for name in self._fields}
records_in_batch = 0
+ blobs_to_resolve = []
try:
while True:
@@ -112,6 +117,10 @@ class FormatBlobReader(RecordBatchReader):
)
elif self._blob_as_descriptor:
pydict_data[field_name].append(blob.to_descriptor().serialize())
+ elif self._blob_parallelism > 1:
+ idx = len(pydict_data[field_name])
+ pydict_data[field_name].append(None)
+ blobs_to_resolve.append((field_name, idx, blob))
else:
pydict_data[field_name].append(blob.to_data())
@@ -120,9 +129,11 @@ class FormatBlobReader(RecordBatchReader):
break
except StopIteration:
- # Stop immediately when StopIteration occurs
pass
+ if blobs_to_resolve:
+ self._resolve_blobs_concurrent(pydict_data, blobs_to_resolve)
+
if records_in_batch == 0:
return None
@@ -145,6 +156,12 @@ class FormatBlobReader(RecordBatchReader):
else:
return None
+ def _resolve_blobs_concurrent(self, pydict_data, blobs_to_resolve):
+ blobs = [item[2] for item in blobs_to_resolve]
+ results = self._file_io.read_blobs_concurrent(blobs,
self._blob_parallelism)
+ for (field_name, idx, _), data in zip(blobs_to_resolve, results):
+ pydict_data[field_name][idx] = data
+
def close(self):
self._blob_iterator = None
if self._input_stream is not None:
diff --git a/paimon-python/pypaimon/read/split_read.py
b/paimon-python/pypaimon/read/split_read.py
index fce5378956..dd9058c55f 100644
--- a/paimon-python/pypaimon/read/split_read.py
+++ b/paimon-python/pypaimon/read/split_read.py
@@ -124,6 +124,7 @@ class SplitRead(ABC):
self.value_arity = len(read_type)
self.nested_name_paths = nested_name_paths
self.limit = limit
+ self._blob_parallelism = 1
# Snapshot the raw value-side schema before _create_key_value_fields
# wraps it, so MergeFileSplitRead can hand per-value-field nullable
# flags to merge functions that enforce NOT-NULL on every add().
@@ -284,10 +285,12 @@ class SplitRead(ABC):
raise NotImplementedError(
"Nested-field projection is not supported on BLOB files")
blob_as_descriptor =
CoreOptions.blob_as_descriptor(self.table.options)
+ blob_parallelism = getattr(self, '_blob_parallelism', 1)
format_reader = FormatBlobReader(self.table.file_io, file_path,
read_file_fields,
self.read_fields,
read_arrow_predicate, blob_as_descriptor,
batch_size=batch_size,
- row_indices=row_indices)
+ row_indices=row_indices,
+ blob_parallelism=blob_parallelism)
elif file_format == CoreOptions.FILE_FORMAT_LANCE:
if has_nested:
raise NotImplementedError(
@@ -1003,9 +1006,11 @@ class DataEvolutionSplitRead(SplitRead):
self.table.options))
or (not CoreOptions.blob_as_descriptor(self.table.options)
and
CoreOptions.blob_descriptor_fields(self.table.options))):
+ blob_parallelism = getattr(self, '_blob_parallelism', 1)
reader = BlobInlineConvertReader(
reader, self.table,
- prescan_reader_factory=lambda names:
self._create_prescan_reader(names))
+ prescan_reader_factory=lambda names:
self._create_prescan_reader(names),
+ blob_parallelism=blob_parallelism)
return reader
@@ -1313,6 +1318,7 @@ class DataEvolutionSplitRead(SplitRead):
return None
file_path = file.external_path if file.external_path else
file.file_path
+ blob_parallelism = getattr(self, '_blob_parallelism', 1)
return FormatBlobReader(
self.table.file_io,
file_path,
@@ -1322,6 +1328,7 @@ class DataEvolutionSplitRead(SplitRead):
CoreOptions.blob_as_descriptor(self.table.options),
batch_size=self.table.options.read_batch_size(),
row_indices=row_indices,
+ blob_parallelism=blob_parallelism,
)
def _split_field_bunches(self, need_merge_files: List[DataFileMeta]) ->
List[FieldBunch]:
diff --git a/paimon-python/pypaimon/read/table_read.py
b/paimon-python/pypaimon/read/table_read.py
index 67159a69c7..b19fff109d 100644
--- a/paimon-python/pypaimon/read/table_read.py
+++ b/paimon-python/pypaimon/read/table_read.py
@@ -72,6 +72,11 @@ class _RemainingRows:
class TableRead:
"""Implementation of TableRead for native Python reading."""
+ # Cap on peak concurrent blob reads across the whole parallel read: split
+ # workers (P) each spin up blob_parallelism (B) blob threads, so peak
+ # connections ~= P*B. Shrink per-split B to keep the product bounded.
+ _MAX_TOTAL_BLOB_WORKERS = 64
+
def __init__(
self,
table,
@@ -120,11 +125,13 @@ class TableRead:
return _record_generator()
- def to_arrow_batch_reader(self, splits: List[Split]) ->
pyarrow.ipc.RecordBatchReader:
+ def to_arrow_batch_reader(self, splits: List[Split],
+ blob_parallelism: Optional[int] = None) ->
pyarrow.ipc.RecordBatchReader:
+ effective_bp = self._resolve_blob_parallelism(blob_parallelism)
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)
+ batch_iterator = self._arrow_batch_generator(splits, schema,
effective_bp)
return pyarrow.ipc.RecordBatchReader.from_batches(schema,
batch_iterator)
@staticmethod
@@ -154,6 +161,7 @@ class TableRead:
self,
splits: List[Split],
parallelism: Optional[int] = None,
+ blob_parallelism: Optional[int] = None,
) -> Optional[pyarrow.Table]:
"""Read ``splits`` into a single arrow ``Table``.
@@ -166,7 +174,16 @@ class TableRead:
``>= 2`` enables a thread pool that reads splits
concurrently and assembles the final table in input order.
Must be ``>= 1``.
+ 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
+ concurrent ranged reads. GIL is released during I/O. On the
+ parallel path, peak blob threads (``parallelism`` *
+ ``blob_parallelism``) are capped at
+ ``_MAX_TOTAL_BLOB_WORKERS``; per-split ``blob_parallelism`` is
+ 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)
schema = PyarrowFieldParser.from_paimon_schema(self.read_type)
@@ -174,9 +191,9 @@ class TableRead:
schema = self._add_row_kind_to_schema(schema)
if self._should_run_parallel(splits, effective):
- return self._to_arrow_parallel(splits, schema, effective)
+ return self._to_arrow_parallel(splits, schema, effective,
effective_bp)
- batch_reader = self.to_arrow_batch_reader(splits)
+ batch_reader = self.to_arrow_batch_reader(splits,
blob_parallelism=effective_bp)
table_list = []
for batch in iter(batch_reader.read_next_batch, None):
@@ -189,7 +206,8 @@ class TableRead:
else:
return pyarrow.Table.from_batches(table_list)
- def _arrow_batch_generator(self, splits: List[Split], schema:
pyarrow.Schema) -> Iterator[pyarrow.RecordBatch]:
+ def _arrow_batch_generator(self, splits: List[Split], schema:
pyarrow.Schema,
+ blob_parallelism: int = 1) ->
Iterator[pyarrow.RecordBatch]:
chunk_size = 65536
# ``remaining`` tracks how many rows we are still allowed to emit
# across all splits. ``None`` means unlimited.
@@ -198,7 +216,7 @@ class TableRead:
for split in splits:
if remaining is not None and remaining <= 0:
break
- reader = self._create_split_read(split).create_reader()
+ reader = self._create_split_read(split,
blob_parallelism).create_reader()
try:
if isinstance(reader, RecordBatchReader):
for batch in iter(reader.read_arrow_batch, None):
@@ -266,6 +284,21 @@ class TableRead:
raise ValueError(f"{source} must be >= 1, got {value}")
return value
+ @staticmethod
+ def _resolve_blob_parallelism(runtime: Optional[int]) -> int:
+ if runtime is None:
+ return 1
+ if runtime < 1:
+ raise ValueError(f"blob_parallelism must be >= 1, got {runtime}")
+ return runtime
+
+ @classmethod
+ def _cap_blob_parallelism(cls, workers: int, blob_parallelism: int) -> int:
+ """Shrink per-split blob_parallelism so workers*B <= cap (peak
reads)."""
+ if blob_parallelism <= 1 or workers * blob_parallelism <=
cls._MAX_TOTAL_BLOB_WORKERS:
+ return blob_parallelism
+ return max(1, cls._MAX_TOTAL_BLOB_WORKERS // workers)
+
def _should_run_parallel(
self,
splits: List[Split],
@@ -284,6 +317,7 @@ class TableRead:
splits: List[Split],
schema: pyarrow.Schema,
effective: int,
+ blob_parallelism: int = 1,
) -> pyarrow.Table:
"""Read ``splits`` concurrently and assemble the result in input order.
@@ -296,6 +330,7 @@ class TableRead:
remaining_state = _RemainingRows(self.limit)
results: List[Optional[List[pyarrow.RecordBatch]]] = [None] *
len(splits)
workers = min(effective, len(splits))
+ blob_parallelism = self._cap_blob_parallelism(workers,
blob_parallelism)
with ThreadPoolExecutor(
max_workers=workers,
thread_name_prefix="pypaimon-read",
@@ -306,6 +341,7 @@ class TableRead:
split,
schema,
remaining_state,
+ blob_parallelism,
): idx
for idx, split in enumerate(splits)
}
@@ -333,6 +369,7 @@ class TableRead:
split: Split,
schema: pyarrow.Schema,
remaining_state: _RemainingRows,
+ blob_parallelism: int = 1,
) -> List[pyarrow.RecordBatch]:
"""Read a single split into arrow batches under soft-stop control.
@@ -342,7 +379,7 @@ class TableRead:
"""
chunk_size = 65536
out: List[pyarrow.RecordBatch] = []
- reader = self._create_split_read(split).create_reader()
+ reader = self._create_split_read(split,
blob_parallelism).create_reader()
try:
if isinstance(reader, RecordBatchReader):
for batch in iter(reader.read_arrow_batch, None):
@@ -583,7 +620,12 @@ class TableRead:
dataset = TorchDataset(self, splits)
return dataset
- def _create_split_read(self, split: Split) -> SplitRead:
+ def _create_split_read(self, split: Split, blob_parallelism: int = 1) ->
SplitRead:
+ sr = self._build_split_read(split)
+ sr._blob_parallelism = blob_parallelism
+ return sr
+
+ def _build_split_read(self, split: Split) -> SplitRead:
if self.table.is_primary_key_table and not split.raw_convertible:
inner_read_type = self.read_type
outer_extract_name_paths: Optional[List[List[str]]] = None
diff --git a/paimon-python/pypaimon/tests/blob_test.py
b/paimon-python/pypaimon/tests/blob_test.py
index 37217f8b7c..ac31c399f3 100644
--- a/paimon-python/pypaimon/tests/blob_test.py
+++ b/paimon-python/pypaimon/tests/blob_test.py
@@ -25,7 +25,7 @@ from pathlib import Path
import pyarrow as pa
-from pypaimon import CatalogFactory
+from pypaimon import CatalogFactory, Schema
from pypaimon.common.file_io import FileIO
from pypaimon.filesystem.local_file_io import LocalFileIO
from pypaimon.common.options import Options
@@ -1489,6 +1489,143 @@ class BlobEndToEndTest(unittest.TestCase):
reader.close()
+class BlobParallelismTest(unittest.TestCase):
+
+ def setUp(self):
+ self.temp_dir = tempfile.mkdtemp()
+ self.catalog = CatalogFactory.create({'warehouse':
os.path.join(self.temp_dir, 'wh')})
+ self.catalog.create_database('default', True)
+ pa_schema = pa.schema([('id', pa.int32()), ('img', pa.large_binary())])
+ self.catalog.create_table('default.bp_test',
Schema.from_pyarrow_schema(
+ pa_schema, options={'row-tracking.enabled': 'true',
'data-evolution.enabled': 'true'}), False)
+ self.payloads = [os.urandom(512) for _ in range(20)]
+ t = self.catalog.get_table('default.bp_test')
+ w = t.new_batch_write_builder().new_write()
+ w.write_arrow(pa.Table.from_pydict(
+ {'id': list(range(20)), 'img': self.payloads}, schema=pa_schema))
+ t.new_batch_write_builder().new_commit().commit(w.prepare_commit())
+ w.close()
+
+ def tearDown(self):
+ shutil.rmtree(self.temp_dir, ignore_errors=True)
+
+ def test_to_arrow_blob_parallelism(self):
+ t = self.catalog.get_table('default.bp_test')
+ rb = t.new_read_builder()
+ splits = rb.new_scan().plan().splits()
+ serial = rb.new_read().to_arrow(splits)
+ parallel = rb.new_read().to_arrow(splits, blob_parallelism=4)
+ self.assertEqual(serial.num_rows, parallel.num_rows)
+ for i in range(serial.num_rows):
+ self.assertEqual(serial['img'][i].as_py(),
parallel['img'][i].as_py())
+
+ def test_to_arrow_batch_reader_blob_parallelism(self):
+ t = self.catalog.get_table('default.bp_test')
+ rb = t.new_read_builder()
+ splits = rb.new_scan().plan().splits()
+ serial = rb.new_read().to_arrow(splits)
+ batches = []
+ for batch in rb.new_read().to_arrow_batch_reader(splits,
blob_parallelism=4):
+ batches.append(batch)
+ parallel = pa.Table.from_batches(batches)
+ self.assertEqual(serial.num_rows, parallel.num_rows)
+ for i in range(serial.num_rows):
+ self.assertEqual(serial['img'][i].as_py(),
parallel['img'][i].as_py())
+
+ def test_blob_parallelism_with_projection(self):
+ t = self.catalog.get_table('default.bp_test')
+ rb = t.new_read_builder()
+ rb = rb.with_projection(['id', 'img'])
+ splits = rb.new_scan().plan().splits()
+ result = rb.new_read().to_arrow(splits, blob_parallelism=4)
+ self.assertEqual(result.column_names, ['id', 'img'])
+ got = dict(zip(result['id'].to_pylist(), result['img'].to_pylist()))
+ for i in range(20):
+ self.assertEqual(got[i], self.payloads[i])
+
+
+class CapBlobParallelismTest(unittest.TestCase):
+ """Peak blob threads on the parallel path (workers * blob_parallelism)
+ must stay within TableRead._MAX_TOTAL_BLOB_WORKERS."""
+
+ def test_cap(self):
+ from pypaimon.read.table_read import TableRead
+ cap = TableRead._MAX_TOTAL_BLOB_WORKERS
+ f = TableRead._cap_blob_parallelism
+ self.assertEqual(f(1, 1), 1) # serial blobs, untouched
+ self.assertEqual(f(16, 1), 1) # B<=1 untouched
+ self.assertEqual(f(4, 8), 8) # 32 <= cap, untouched
+ self.assertEqual(f(16, 16), cap // 16) # 256 -> shrink to cap/workers
+ self.assertEqual(f(cap, 2), 1) # workers==cap -> 1
+ self.assertEqual(f(cap + 100, 2), 1) # workers>cap -> floor to 1
+ for w in (2, 4, 8, 16, 32, 64):
+ self.assertLessEqual(w * f(w, 999), cap)
+
+
+class CoalesceRangesTest(unittest.TestCase):
+ """read_ranges_coalesced merges same-file adjacent reads into fewer
requests
+ (JingsongLi's IO-merging suggestion) while returning identical bytes."""
+
+ def test_coalesce_ranges_grouping(self):
+ from pypaimon.common.file_io import _coalesce_ranges
+ items = [(0, "a", 0, 10), (1, "a", 10, 10), (2, "a", 1000, 10), (3,
"b", 0, 5)]
+ # a:[0,20) merged, a:[1000,1010) split by gap, b:[0,5) separate file
+ spans = _coalesce_ranges(items, max_gap=100, max_span=1 << 30)
+ self.assertEqual(len(spans), 3)
+ self.assertEqual(sorted(i for _, _, _, mem in spans for i, _, _ in
mem), [0, 1, 2, 3])
+ # max_span forces a split even when contiguous
+ self.assertEqual(len(_coalesce_ranges(
+ [(0, "a", 0, 10), (1, "a", 10, 10)], max_gap=100, max_span=15)), 2)
+
+ def test_read_ranges_coalesced(self):
+ from pypaimon.common.file_io import FileIO
+ data = bytes(range(256)) * 4 # 1024 bytes
+ with tempfile.TemporaryDirectory() as tmp_dir:
+ path = os.path.join(tmp_dir, "f.bin")
+ with open(path, 'wb') as f:
+ f.write(data)
+ fio = FileIO.get(f"file://{tmp_dir}", {})
+ ranges = [(path, 0, 10), (path, 10, 10), None, (path, 500, 20),
+ (path, 100, -1), (path, None, None)]
+ got = fio.read_ranges_coalesced(ranges, parallelism=4)
+ self.assertEqual(got[0], data[0:10])
+ self.assertEqual(got[1], data[10:20]) # contiguous with got[0],
merged
+ self.assertIsNone(got[2])
+ self.assertEqual(got[3], data[500:520])
+ self.assertEqual(got[4], data[100:]) # length -1 => read to EOF
+ self.assertIsNone(got[5]) # None offset/length =>
skipped
+
+
+class ReadFileRangeTest(unittest.TestCase):
+ """read_file_range must accept length == -1 (read to EOF) -- the valid
+ unknown-length BlobDescriptor case -- not pass -1 into pread."""
+
+ def test_negative_length_reads_to_eof(self):
+ from pypaimon.common.file_io import FileIO
+ data = bytes(range(64))
+ with tempfile.TemporaryDirectory() as tmp_dir:
+ path = os.path.join(tmp_dir, "blob.bin")
+ with open(path, 'wb') as f:
+ f.write(data)
+ file_io = FileIO.get(f"file://{tmp_dir}", {})
+ # -1: offset to EOF; positive: exact range (pread)
+ self.assertEqual(file_io.read_file_range(path, 0, -1), data)
+ self.assertEqual(file_io.read_file_range(path, 10, -1), data[10:])
+ self.assertEqual(file_io.read_file_range(path, 10, 8), data[10:18])
+
+ def test_descriptor_negative_length_roundtrip(self):
+ from pypaimon.common.file_io import FileIO
+ data = b"actual blob content"
+ with tempfile.TemporaryDirectory() as tmp_dir:
+ path = os.path.join(tmp_dir, "blob.bin")
+ with open(path, 'wb') as f:
+ f.write(data)
+ file_io = FileIO.get(f"file://{tmp_dir}", {})
+ blob = Blob.from_bytes(BlobDescriptor(path, 0, -1).serialize(),
file_io)
+ self.assertIsInstance(blob, BlobRef)
+ self.assertEqual(blob.to_data(), data)
+
+
class OffsetInputStreamTest(unittest.TestCase):
def setUp(self):
diff --git a/paimon-python/pypaimon/tests/daft/daft_blob_read_test.py
b/paimon-python/pypaimon/tests/daft/daft_blob_read_test.py
new file mode 100644
index 0000000000..59c2a37b10
--- /dev/null
+++ b/paimon-python/pypaimon/tests/daft/daft_blob_read_test.py
@@ -0,0 +1,133 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+import os
+import shutil
+import tempfile
+import unittest
+
+import pyarrow as pa
+import pytest
+
+pypaimon = pytest.importorskip("pypaimon")
+daft = pytest.importorskip("daft")
+
+from daft import col
+
+from pypaimon import CatalogFactory, Schema
+from pypaimon.daft import open_blob, read_blob, read_paimon
+from pypaimon.daft.daft_compat import has_file_range_reads
+
+
[email protected](has_file_range_reads(), "installed Daft lacks File range
reads")
+class DaftReadBlobTest(unittest.TestCase):
+ """read_blob reads a blob File column to bytes via pypaimon's FileIO and
must
+ return exactly the same bytes as the native per-row File.open() path."""
+
+ N = 40
+
+ def setUp(self):
+ self.tempdir = tempfile.mkdtemp()
+ self.catalog_options = {"warehouse": os.path.join(self.tempdir, "wh")}
+ self.catalog = CatalogFactory.create(self.catalog_options)
+ self.catalog.create_database("default", True)
+ self.table = "default.blob_read_test"
+ pa_schema = pa.schema([("id", pa.int32()), ("content",
pa.large_binary())])
+ self.catalog.create_table(self.table, Schema.from_pyarrow_schema(
+ pa_schema, options={"row-tracking.enabled": "true",
"data-evolution.enabled": "true"}), False)
+ self.payloads = [bytes([i & 0xFF]) + os.urandom(1023) for i in
range(self.N)]
+ t = self.catalog.get_table(self.table)
+ w = t.new_batch_write_builder().new_write()
+ w.write_arrow(pa.Table.from_pydict(
+ {"id": list(range(self.N)), "content": self.payloads},
schema=pa_schema))
+ t.new_batch_write_builder().new_commit().commit(w.prepare_commit())
+ w.close()
+
+ def tearDown(self):
+ shutil.rmtree(self.tempdir, ignore_errors=True)
+
+ def test_read_blob_matches_file_open(self):
+ df = read_paimon(self.table, self.catalog_options)
+ # content is exposed as a lazy File (default read path unchanged)
+ self.assertEqual(str(df.schema()["content"].dtype), "File[Unknown]")
+
+ # new path: read_blob -> bytes
+ out = df.select(col("id"), read_blob(col("content"),
self.catalog_options, self.table)
+ .alias("img")).collect().to_pydict()
+ got = dict(zip(out["id"], out["img"]))
+ for i in range(self.N):
+ self.assertEqual(got[i], self.payloads[i])
+
+ # original path still works and yields identical bytes
+ for r in read_paimon(self.table,
self.catalog_options).select(col("id"), col("content")).to_pylist():
+ with r["content"].open() as h:
+ self.assertEqual(h.read(), self.payloads[r["id"]])
+
+ def test_open_blob_full_read(self):
+ rows = read_paimon(self.table, self.catalog_options).select(
+ col("id"), col("content")).to_pylist()
+ for r in rows:
+ with open_blob(r["content"], self.catalog_options, self.table) as
f:
+ self.assertEqual(f.read(), self.payloads[r["id"]])
+
+ def test_open_blob_partial_read(self):
+ rows = read_paimon(self.table, self.catalog_options).select(
+ col("id"), col("content")).to_pylist()
+ for r in rows:
+ with open_blob(r["content"], self.catalog_options, self.table) as
f:
+ header = f.read(16)
+ self.assertEqual(header, self.payloads[r["id"]][:16])
+
+ def test_open_blob_seek(self):
+ r = read_paimon(self.table, self.catalog_options).select(
+ col("id"), col("content")).to_pylist()[0]
+ with open_blob(r["content"], self.catalog_options, self.table) as f:
+ f.seek(100)
+ self.assertEqual(f.tell(), 100)
+ chunk = f.read(50)
+ self.assertEqual(chunk, self.payloads[r["id"]][100:150])
+
+ def test_no_file_size_stat(self):
+ """File.size() does a network stat; read_blob and open_blob must
resolve
+ ranges from embedded metadata (file._inner) and never call it."""
+ rows = read_paimon(self.table, self.catalog_options).select(
+ col("id"), col("content")).to_pylist()
+ file_cls = type(rows[0]["content"])
+ calls = {"n": 0}
+ orig = file_cls.size
+
+ def boom(self, *a, **k):
+ calls["n"] += 1
+ raise AssertionError("File.size() should not be called")
+
+ file_cls.size = boom
+ try:
+ with open_blob(rows[0]["content"], self.catalog_options,
self.table) as f:
+ self.assertEqual(f.read(), self.payloads[rows[0]["id"]])
+ out = read_paimon(self.table, self.catalog_options).select(
+ col("id"), read_blob(col("content"), self.catalog_options,
self.table).alias("img")
+ ).collect().to_pydict()
+ got = dict(zip(out["id"], out["img"]))
+ for i in range(self.N):
+ self.assertEqual(got[i], self.payloads[i])
+ finally:
+ file_cls.size = orig
+ self.assertEqual(calls["n"], 0)
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/paimon-python/pypaimon/tests/reader_parallel_test.py
b/paimon-python/pypaimon/tests/reader_parallel_test.py
index 692d1c5b48..adad137341 100644
--- a/paimon-python/pypaimon/tests/reader_parallel_test.py
+++ b/paimon-python/pypaimon/tests/reader_parallel_test.py
@@ -282,13 +282,13 @@ class ParallelReaderAppendOnlyTest(unittest.TestCase):
call_counter = {'n': 0}
lock = threading.Lock()
- def flaky(self_, split):
+ def flaky(self_, split, blob_parallelism=1):
with lock:
call_counter['n'] += 1
idx = call_counter['n']
if idx == 2:
raise RuntimeError("simulated reader failure")
- return original_create(self_, split)
+ return original_create(self_, split, blob_parallelism)
with mock.patch.object(TableRead, '_create_split_read', flaky):
read = rb.new_read()
