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 865b2c96c2 [python][ray] Integrate Ray for distributed multimodal
table reads (#8455)
865b2c96c2 is described below
commit 865b2c96c2f794e1a743d8a417caf2a654cfb17d
Author: XiaoHongbo <[email protected]>
AuthorDate: Sat Jul 4 21:10:35 2026 +0800
[python][ray] Integrate Ray for distributed multimodal table reads (#8455)
Multimodal tables define a very simple API for reading data, but it only
runs in local mode. This PR adds two small
Ray APIs to distribute multimodal table reads:
- **`scan().to_ray()`** — read filtered rows as a `ray.data.Dataset`;
BLOB columns come back as lightweight descriptors, not payloads.
- **`pypaimon.ray.map_blobs(ds, blob_cols, fn, parallelism=,
batch_size=)`** — fetch BLOB bytes on the workers and hand
`(scalar_batch, blobs)` to `fn`, so payloads are consumed in-place,
never materialised on the driver.
```python
ds = table.scan().where(...).select(cols).to_ray(concurrency=8)
result = map_blobs(ds, blob_cols, process, parallelism=16,
batch_size=512)
```
---
docs/docs/pypaimon/multimodal-api.mdx | 48 ++++++
paimon-python/pypaimon/multimodal/blob_read.py | 61 +++++++
paimon-python/pypaimon/multimodal/query.py | 104 ++++++++----
paimon-python/pypaimon/multimodal/table.py | 19 +++
paimon-python/pypaimon/ray/__init__.py | 3 +-
paimon-python/pypaimon/ray/ray_paimon.py | 144 ++++++++++++++++-
.../pypaimon/read/datasource/ray_datasource.py | 1 +
.../pypaimon/tests/multimodal_table_test.py | 180 +++++++++++++++++++++
8 files changed, 526 insertions(+), 34 deletions(-)
diff --git a/docs/docs/pypaimon/multimodal-api.mdx
b/docs/docs/pypaimon/multimodal-api.mdx
index b2b63ab19f..72cf84f48c 100644
--- a/docs/docs/pypaimon/multimodal-api.mdx
+++ b/docs/docs/pypaimon/multimodal-api.mdx
@@ -383,6 +383,54 @@ Notes:
few large reads); scattered point reads coalesce less.
- Blob reads are available only on `scan()`, not on the `search()` queries.
+### Distributed BLOB processing with Ray
+
+For larger jobs, read descriptors with `to_ray()`, then fetch and process BLOB
+bytes on Ray workers with `map_with_blobs`.
+
+```python
+import ray
+import pyarrow as pa
+import pypaimon.multimodal as pm
+
+ray.init(num_cpus=8)
+
+conn = pm.connect(database="default", options={"warehouse":
"file:///tmp/warehouse"})
+docs = conn.get_table("docs")
+
+ids = ["a", "b", "c"]
+scalar_cols = ["id", "category"]
+blob_cols = ["image"]
+in_clause = ", ".join(f"'{i}'" for i in ids)
+
+
+def process_batch(scalar_batch, blobs):
+ images = blobs["image"]
+ # Decode, infer, write samples, or train here.
+ return pa.table({"rows": [scalar_batch.num_rows]})
+
+
+ds = (
+ docs.scan()
+ .where(f"id IN ({in_clause})")
+ .select(scalar_cols + blob_cols)
+ .to_ray()
+)
+
+# Optional Ray transforms are fine if BLOB descriptor columns remain.
+# ds = ds.filter(lambda row: row["category"] == "lake")
+
+result_ds = docs.map_with_blobs(ds, blob_cols, process_batch)
+```
+
+Notes:
+
+- `process_batch` must return a small Ray-compatible batch; return an empty
+ `pyarrow.Table` for side-effect-only jobs.
+- Avoid returning raw BLOB bytes, which would materialize payloads in Ray's
+ object store.
+- Tune `to_ray(...)` and `map_with_blobs(...)` parameters only when needed.
+
## Row IDs
Multimodal tables enable `row-tracking.enabled` by default, so each row has a
diff --git a/paimon-python/pypaimon/multimodal/blob_read.py
b/paimon-python/pypaimon/multimodal/blob_read.py
new file mode 100644
index 0000000000..2db4621b19
--- /dev/null
+++ b/paimon-python/pypaimon/multimodal/blob_read.py
@@ -0,0 +1,61 @@
+# 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.
+
+"""Shared helpers for materialising multimodal BLOB descriptor columns."""
+
+
+def fetch_blob_bodies(file_io, data, blob_cols, parallelism):
+ """Fetch BLOB payload bytes for descriptor/inline/null cells.
+
+ ``data`` is a ``dict`` mapping each BLOB column name to row-aligned cells.
+ Each cell may be serialized ``BlobDescriptor`` bytes, inline payload bytes,
+ or ``None``. Returned values preserve row order and are grouped per column.
+ """
+ from pypaimon.table.row.blob import BlobDescriptor, BlobViewStruct
+
+ ranges = []
+ inline = {}
+ index = 0
+ for col in blob_cols:
+ for value in data[col]:
+ if value is None:
+ ranges.append(None)
+ else:
+ raw = bytes(value)
+ if BlobViewStruct.is_blob_view_struct(raw):
+ raise ValueError(
+ "read_blobs does not support unresolved blob-view
columns; "
+ "read such a column on its own, or enable blob-view
resolution.")
+ if BlobDescriptor.is_blob_descriptor(raw):
+ descriptor = BlobDescriptor.deserialize(raw)
+ ranges.append((descriptor.uri, descriptor.offset,
descriptor.length))
+ else:
+ ranges.append(None)
+ inline[index] = raw
+ index += 1
+
+ fetched = file_io.read_ranges_coalesced(ranges, parallelism)
+ for index, raw in inline.items():
+ fetched[index] = raw
+
+ bodies = {}
+ offset = 0
+ for col in blob_cols:
+ count = len(data[col])
+ bodies[col] = fetched[offset:offset + count]
+ offset += count
+ return bodies
diff --git a/paimon-python/pypaimon/multimodal/query.py
b/paimon-python/pypaimon/multimodal/query.py
index 0ea45b7df5..e164f419fa 100644
--- a/paimon-python/pypaimon/multimodal/query.py
+++ b/paimon-python/pypaimon/multimodal/query.py
@@ -15,14 +15,19 @@
# specific language governing permissions and limitations
# under the License.
-from typing import Callable, Dict, List, Optional, Tuple
+from typing import Any, Callable, Dict, List, Optional, Tuple
import pyarrow as pa
from pypaimon.common.where_parser import parse_where_clause
+from pypaimon.multimodal.blob_read import fetch_blob_bodies
from pypaimon.table.special_fields import SpecialFields
+def _select_arrow_columns(batch, columns):
+ return batch.select(columns)
+
+
class ScanQuery:
"""Chainable scan wrapper for MultimodalTable."""
@@ -100,6 +105,38 @@ class ScanQuery:
def to_list(self) -> List[dict]:
return self.to_arrow().to_pylist()
+ def to_ray(
+ self,
+ *,
+ ray_remote_args: Optional[Dict[str, Any]] = None,
+ concurrency: Optional[int] = None,
+ override_num_blocks: Optional[int] = None,
+ **read_args):
+ """Read this scan as a Ray Dataset.
+
+ BLOB columns are read as serialized descriptors. Use
+ ``table.map_with_blobs(...)`` to fetch payload bytes on Ray workers.
+ """
+ if self._result_factory is not None:
+ raise TypeError("to_ray is only supported on scan(), not search
queries.")
+
+ read_builder, file_io, visible_columns =
self._blob_descriptor_query_read_builder()
+ plan = read_builder.new_scan().plan()
+ ds = read_builder.new_read().to_ray(
+ plan.splits(),
+ ray_remote_args=ray_remote_args,
+ concurrency=concurrency,
+ override_num_blocks=override_num_blocks,
+ **read_args)
+ if visible_columns is not None:
+ ds = ds.map_batches(
+ _select_arrow_columns,
+ fn_kwargs={"columns": visible_columns},
+ batch_format="pyarrow")
+ setattr(ds, "_paimon_blob_file_io", file_io)
+ setattr(ds, "_paimon_blob_columns", self._all_blob_columns())
+ return ds
+
def read_blobs(
self, columns=None, *, parallelism: int = 64
) -> Tuple[pa.Table, Dict[str, List[Optional[bytes]]]]:
@@ -146,6 +183,36 @@ class ScanQuery:
if hasattr(reader, "close"):
reader.close()
+ def _blob_descriptor_query_read_builder(self):
+ from pypaimon.common.options.core_options import CoreOptions
+ read_table = self._table.copy({
+ CoreOptions.BLOB_AS_DESCRIPTOR.key(): "true"
+ })
+ read_builder = read_table.new_read_builder()
+ if self._predicate is not None:
+ read_builder = read_builder.with_filter(self._predicate)
+ projection = self._effective_projection()
+ if projection is not None:
+ internal_projection = list(projection)
+ for name in self._predicate_fields():
+ if name not in internal_projection:
+ internal_projection.append(name)
+ read_builder = read_builder.with_projection(internal_projection)
+ if self._limit is not None:
+ read_builder = read_builder.with_limit(self._limit)
+ if projection is None:
+ return read_builder, read_table.file_io, None
+ internal_column_names = [field.name for field in
read_builder.read_type()]
+ visible_columns = self._projected_output_columns(read_table,
projection)
+ if visible_columns == internal_column_names:
+ visible_columns = None
+ return read_builder, read_table.file_io, visible_columns
+
+ @staticmethod
+ def _projected_output_columns(table, projection):
+ builder = table.new_read_builder().with_projection(projection)
+ return [field.name for field in builder.read_type()]
+
def _blob_descriptor_read_builder(self, blob_cols: List[str]):
"""Blob-as-descriptor read builder with this query's
filter/projection/limit;
returns ``(read_builder, file_io)``."""
@@ -172,37 +239,7 @@ class ScanQuery:
# ``FileIO.get(uri, catalog_options)`` off the raw options (no merged
token),
# failing with "endpoint should be non-empty" / "Init credential
failed" unless
# the caller also passes fs.oss.* -- which users should not have to.
- from pypaimon.table.row.blob import BlobDescriptor, BlobViewStruct
- ranges = []
- inline = {} # cell index -> blob stored inline (returned as-is, no
ranged read)
- i = 0
- for col in blob_cols:
- for value in data[col]:
- if value is None:
- ranges.append(None)
- else:
- raw = bytes(value)
- if BlobViewStruct.is_blob_view_struct(raw):
- raise ValueError(
- "read_blobs does not support unresolved blob-view
columns; "
- "read such a column on its own, or enable
blob-view resolution.")
- if BlobDescriptor.is_blob_descriptor(raw):
- d = BlobDescriptor.deserialize(raw)
- ranges.append((d.uri, d.offset, d.length))
- else: # blob stored inline: the bytes are the payload
- ranges.append(None)
- inline[i] = raw
- i += 1
- fetched = file_io.read_ranges_coalesced(ranges, parallelism)
- for idx, raw in inline.items():
- fetched[idx] = raw
- bodies = {}
- offset = 0
- for col in blob_cols:
- n = len(data[col])
- bodies[col] = fetched[offset:offset + n]
- offset += n
- return bodies
+ return fetch_blob_bodies(file_io, data, blob_cols, parallelism)
def _all_blob_columns(self) -> List[str]:
return [
@@ -300,6 +337,9 @@ class _PreFilterQuery(ScanQuery):
def stream_blobs(self, *args, **kwargs):
raise TypeError("stream_blobs is only supported on scan(), not search
queries.")
+ def to_ray(self, *args, **kwargs):
+ raise TypeError("to_ray is only supported on scan(), not search
queries.")
+
class VectorQuery(_PreFilterQuery):
"""Chainable query wrapper for vector global-index search."""
diff --git a/paimon-python/pypaimon/multimodal/table.py
b/paimon-python/pypaimon/multimodal/table.py
index a00f8d1156..716d6ef621 100644
--- a/paimon-python/pypaimon/multimodal/table.py
+++ b/paimon-python/pypaimon/multimodal/table.py
@@ -157,6 +157,18 @@ class MultimodalTable:
def merge(self, on):
return _MergeBuilder(self, on)
+ def map_with_blobs(self, dataset, columns, fn, **kwargs):
+ from pypaimon.ray import map_with_blobs
+
+ return map_with_blobs(
+ dataset,
+ columns,
+ fn,
+ file_io=self.raw_table.file_io,
+ all_blob_columns=_blob_columns(self.raw_table),
+ **kwargs,
+ )
+
def scan(
self,
*,
@@ -357,6 +369,13 @@ class _MergeBuilder:
)
+def _blob_columns(table):
+ return tuple(
+ field.name for field in table.fields
+ if getattr(field.type, "type", None) == "BLOB"
+ )
+
+
def _to_arrow_table(data, target_schema=None):
if isinstance(data, pa.Table):
table = data
diff --git a/paimon-python/pypaimon/ray/__init__.py
b/paimon-python/pypaimon/ray/__init__.py
index f2daad3a69..52b4575b5b 100644
--- a/paimon-python/pypaimon/ray/__init__.py
+++ b/paimon-python/pypaimon/ray/__init__.py
@@ -15,7 +15,7 @@
# specific language governing permissions and limitations
# under the License.
-from pypaimon.ray.ray_paimon import read_paimon, write_paimon
+from pypaimon.ray.ray_paimon import map_with_blobs, read_paimon, write_paimon
from pypaimon.ray.bucket_join import bucket_join
from pypaimon.ray.data_evolution_merge_into import (
WhenMatched,
@@ -31,6 +31,7 @@ from pypaimon.ray.update_by_row_id import update_by_row_id
__all__ = [
"read_paimon",
+ "map_with_blobs",
"write_paimon",
"bucket_join",
"merge_into",
diff --git a/paimon-python/pypaimon/ray/ray_paimon.py
b/paimon-python/pypaimon/ray/ray_paimon.py
index d0e7706c83..b375e3950f 100644
--- a/paimon-python/pypaimon/ray/ray_paimon.py
+++ b/paimon-python/pypaimon/ray/ray_paimon.py
@@ -27,7 +27,7 @@ Usage::
"""
import importlib
-from typing import Any, Dict, List, Optional, TYPE_CHECKING
+from typing import Any, Callable, Dict, List, Optional, TYPE_CHECKING
from pypaimon.common.predicate import Predicate
@@ -137,6 +137,148 @@ def read_paimon(
return ds
+def map_with_blobs(
+ dataset: "ray.data.Dataset",
+ columns,
+ fn: Callable,
+ *,
+ file_io=None,
+ all_blob_columns=None,
+ parallelism: int = 64,
+ batch_size: Optional[int] = 1024,
+ fn_kwargs: Optional[Dict[str, Any]] = None,
+ ray_remote_args: Optional[Dict[str, Any]] = None,
+ **map_args,
+) -> "ray.data.Dataset":
+ """Fetch BLOB payloads in Ray batches and call ``fn``.
+
+ ``fn(scalar_batch, blobs, **fn_kwargs)`` receives a ``pyarrow.Table`` of
+ non-BLOB columns and a row-aligned ``dict`` of BLOB bytes. Return a small
+ Ray-compatible batch; for side-effect-only work, return an empty
+ ``pyarrow.Table`` instead of ``None``. Call this directly on
+ ``scan().to_ray()`` output, or pass ``file_io`` and ``all_blob_columns``.
+ Tune ``batch_size`` for BLOB size and worker memory.
+ """
+ _require_ray_data()
+
+ if not callable(fn):
+ raise ValueError("fn must be callable")
+ if isinstance(columns, str):
+ blob_cols = [columns]
+ else:
+ blob_cols = list(dict.fromkeys(columns))
+ if not blob_cols:
+ raise ValueError("columns must contain at least one BLOB column")
+ if parallelism < 1:
+ raise ValueError("parallelism must be at least 1, got
{}".format(parallelism))
+ if batch_size is not None and batch_size < 1:
+ raise ValueError("batch_size must be at least 1, got
{}".format(batch_size))
+
+ resolved_file_io = file_io
+ if resolved_file_io is None:
+ resolved_file_io = getattr(dataset, "_paimon_blob_file_io", None)
+ if resolved_file_io is None:
+ raise ValueError(
+ "map_with_blobs requires a FileIO. Use table.scan().to_ray() or "
+ "pass file_io= explicitly.")
+
+ batch_format = map_args.pop("batch_format", "pyarrow")
+ if batch_format != "pyarrow":
+ raise ValueError("map_with_blobs requires batch_format='pyarrow'")
+
+ kwargs = dict(map_args)
+ kwargs["batch_format"] = "pyarrow"
+ if batch_size is not None:
+ kwargs.setdefault("batch_size", batch_size)
+ if ray_remote_args is not None:
+ _set_map_batches_remote_args(dataset, kwargs, ray_remote_args)
+
+ all_blob_cols = all_blob_columns
+ if all_blob_cols is None:
+ all_blob_cols = getattr(dataset, "_paimon_blob_columns", None)
+ if all_blob_cols is None:
+ raise ValueError(
+ "map_with_blobs requires all_blob_columns when Dataset lacks "
+ "BLOB metadata.")
+
+ all_blob = set(all_blob_cols)
+ invalid = [name for name in blob_cols if name not in all_blob]
+ if invalid:
+ raise ValueError("Column {!r} is not a BLOB
column.".format(invalid[0]))
+
+ return dataset.map_batches(
+ _map_blob_batch,
+ fn_kwargs={
+ "file_io": resolved_file_io,
+ "blob_cols": blob_cols,
+ "all_blob_cols": list(all_blob_cols),
+ "parallelism": parallelism,
+ "fn": fn,
+ "fn_kwargs": dict(fn_kwargs or {}),
+ },
+ **kwargs)
+
+
+def _set_map_batches_remote_args(dataset, kwargs, ray_remote_args):
+ import inspect
+
+ param =
inspect.signature(dataset.map_batches).parameters.get("ray_remote_args")
+ if param is not None and param.kind != inspect.Parameter.VAR_KEYWORD:
+ kwargs["ray_remote_args"] = ray_remote_args
+ else:
+ kwargs.update(ray_remote_args)
+
+
+def _map_blob_batch(
+ batch, file_io, blob_cols, all_blob_cols, parallelism, fn, fn_kwargs):
+ from pypaimon.multimodal.blob_read import fetch_blob_bodies
+
+ missing = [name for name in blob_cols if name not in batch.schema.names]
+ if missing:
+ raise ValueError("BLOB column(s) not found in Ray Dataset: {}".format(
+ ", ".join(missing)))
+
+ all_blob = set(all_blob_cols)
+ scalar_cols = [name for name in batch.schema.names if name not in all_blob]
+ unknown = _unknown_blob_descriptor_columns(batch, scalar_cols)
+ if unknown:
+ raise ValueError(
+ "Column {!r} holds BLOB descriptors this table does not own "
+ "(likely from a joined BLOB table). Fetch it with its own "
+ "table.map_with_blobs() in a separate pass, or drop it before "
+ "mapping.".format(unknown[0]))
+
+ bodies = fetch_blob_bodies(
+ file_io, batch.select(blob_cols).to_pydict(), blob_cols, parallelism)
+ result = fn(batch.select(scalar_cols), bodies, **fn_kwargs)
+ if result is None:
+ raise ValueError(
+ "map_with_blobs UDF must return a Ray-compatible batch, such as a "
+ "pyarrow.Table. For side-effect-only processing, return an empty "
+ "pyarrow.Table instead of None.")
+ return result
+
+
+def _unknown_blob_descriptor_columns(batch, scalar_cols):
+ return [
+ name for name in scalar_cols
+ if _looks_like_blob_descriptor(batch.column(name))]
+
+
+def _looks_like_blob_descriptor(column):
+ import pyarrow as pa
+ from pypaimon.table.row.blob import BlobDescriptor
+
+ if not (pa.types.is_binary(column.type) or
pa.types.is_large_binary(column.type)):
+ return False
+ chunks = getattr(column, "chunks", None) or [column]
+ for chunk in chunks:
+ for value in chunk:
+ if value.is_valid:
+ return BlobDescriptor.is_blob_descriptor(value.as_py())
+ return False
+
+
def write_paimon(
dataset: "ray.data.Dataset",
table_identifier: str,
diff --git a/paimon-python/pypaimon/read/datasource/ray_datasource.py
b/paimon-python/pypaimon/read/datasource/ray_datasource.py
index a5ac10bc10..7a2f726143 100644
--- a/paimon-python/pypaimon/read/datasource/ray_datasource.py
+++ b/paimon-python/pypaimon/read/datasource/ray_datasource.py
@@ -238,6 +238,7 @@ class RayDatasource(Datasource):
metadata = BlockMetadata(**metadata_kwargs)
read_fn = partial(get_read_task, chunk_splits)
+ read_fn.__name__ = "read_paimon_table"
read_task_kwargs = {
'read_fn': read_fn,
'metadata': metadata,
diff --git a/paimon-python/pypaimon/tests/multimodal_table_test.py
b/paimon-python/pypaimon/tests/multimodal_table_test.py
index d6ed03c969..9dfc0be22c 100644
--- a/paimon-python/pypaimon/tests/multimodal_table_test.py
+++ b/paimon-python/pypaimon/tests/multimodal_table_test.py
@@ -30,6 +30,11 @@ from pypaimon import Schema as PaimonSchema
from pypaimon.globalindex.global_index_result import GlobalIndexResult
from pypaimon.utils.range import Range
+try:
+ import ray
+except ImportError:
+ ray = None
+
_PARQUET_OPTIONS = {
"row-tracking.enabled": "true",
@@ -945,6 +950,8 @@ class MultimodalTableTest(unittest.TestCase):
t.search([1.0, 0.0, 0.0], column="emb").read_blobs("img")
with self.assertRaisesRegex(TypeError, "only supported on scan"):
t.search([1.0, 0.0, 0.0], column="emb").stream_blobs("img")
+ with self.assertRaisesRegex(TypeError, "only supported on scan"):
+ t.search([1.0, 0.0, 0.0], column="emb").to_ray()
def test_scan_stream_blobs(self):
obs = self.conn.create_table(
@@ -984,6 +991,179 @@ class MultimodalTableTest(unittest.TestCase):
self.assertEqual(
[], list(obs.scan().where("clip = 'none'").stream_blobs("image")))
+ @unittest.skipIf(ray is None, "ray is not installed")
+ def test_scan_to_ray_map_with_blobs(self):
+ started_ray = False
+ if not ray.is_initialized():
+ ray.init(ignore_reinit_error=True, num_cpus=2)
+ started_ray = True
+ obs = self.conn.create_table(
+ "ray_obs",
+ schema=_schema({
+ "clip": pa.string(),
+ "idx": pa.int32(),
+ "image": pa.large_binary(),
+ "audio": pa.large_binary(),
+ }),
+ options=_PARQUET_OPTIONS,
+ partitioned=["clip"],
+ )
+ obs.add([
+ {"clip": "c1", "idx": 0, "image": b"img-0", "audio": b"aud-0"},
+ {"clip": "c1", "idx": 1, "image": b"img-1", "audio": b"aud-1"},
+ {"clip": "c2", "idx": 0, "image": b"img-x", "audio": b"aud-x"},
+ ])
+
+ def collect_batch(scalar, blobs, prefix):
+ assert isinstance(scalar, pa.Table)
+ assert ["idx"] == scalar.column_names
+ idxs = scalar.column("idx").to_pylist()
+ rows = []
+ for idx, image in zip(idxs, blobs["image"]):
+ rows.append({"idx": idx, "image": prefix + image})
+ return pa.Table.from_pylist(rows)
+
+ try:
+ from pypaimon.ray import map_with_blobs
+
+ ds = (
+ obs.scan()
+ .where("clip = 'c1'")
+ .select(["idx", "image", "audio"])
+ .to_ray(concurrency=1, override_num_blocks=1)
+ )
+ ds = ds.filter(lambda row: row["idx"] >= 0)
+
+ with self.assertRaisesRegex(ValueError, "not a BLOB column"):
+ obs.map_with_blobs(ds, ["idx"], collect_batch)
+ with self.assertRaisesRegex(ValueError, "FileIO"):
+ map_with_blobs(ds, ["image"], collect_batch)
+
+ from pypaimon.table.row.blob import BlobDescriptor
+
+ descriptor = BlobDescriptor("oss://bucket/blob", 0, 1).serialize()
+ foreign_ds = ray.data.from_arrow(pa.table({
+ "idx": [0],
+ "image": [descriptor],
+ "foreign_blob": [descriptor],
+ }))
+ with self.assertRaisesRegex(Exception, "does not own"):
+ map_with_blobs(
+ foreign_ds,
+ ["image"],
+ collect_batch,
+ file_io=obs.raw_table.file_io,
+ all_blob_columns=["image"],
+ batch_size=1,
+ ).take_all()
+
+ result = obs.map_with_blobs(
+ ds,
+ ["image"],
+ collect_batch,
+ parallelism=2,
+ batch_size=1,
+ fn_kwargs={"prefix": b"got-"},
+ ray_remote_args={"num_cpus": 1},
+ )
+ rows = sorted(result.to_pandas().to_dict("records"), key=lambda
row: row["idx"])
+
+ self.assertEqual(
+ [
+ {"idx": 0, "image": b"got-img-0"},
+ {"idx": 1, "image": b"got-img-1"},
+ ],
+ rows,
+ )
+ finally:
+ if started_ray:
+ ray.shutdown()
+
+ @unittest.skipIf(ray is None, "ray is not installed")
+ def test_scan_to_ray_map_with_blobs_guards(self):
+ started_ray = False
+ if not ray.is_initialized():
+ ray.init(ignore_reinit_error=True, num_cpus=2)
+ started_ray = True
+ obs = self.conn.create_table(
+ "ray_obs_guards",
+ schema=_schema({
+ "clip": pa.string(),
+ "idx": pa.int32(),
+ "image": pa.large_binary(),
+ }),
+ options=_PARQUET_OPTIONS,
+ partitioned=["clip"],
+ )
+ obs.add([{"clip": "c1", "idx": 0, "image": b"img-0"}])
+
+ def return_none(scalar, blobs):
+ return None
+
+ try:
+ from pypaimon.ray import map_with_blobs
+
+ ds = (
+ obs.scan()
+ .where("clip = 'c1'")
+ .select(["idx", "image"])
+ .to_ray(concurrency=1, override_num_blocks=1)
+ )
+
+ with self.assertRaisesRegex(ValueError, "all_blob_columns"):
+ map_with_blobs(
+ ray.data.from_arrow(pa.table({"image": [b"inline"]})),
+ ["image"],
+ lambda scalar, blobs: pa.table({"rows":
[scalar.num_rows]}),
+ file_io=obs.raw_table.file_io,
+ )
+
+ with self.assertRaisesRegex(Exception, "must return"):
+ obs.map_with_blobs(
+ ds,
+ ["image"],
+ return_none,
+ batch_size=1,
+ ).take_all()
+
+ empty_ds = (
+ obs.scan()
+ .where("clip = 'none'")
+ .select(["idx", "image"])
+ .to_ray(concurrency=1, override_num_blocks=1)
+ )
+ result = obs.map_with_blobs(
+ empty_ds,
+ ["image"],
+ lambda scalar, blobs: pa.table({"rows": [scalar.num_rows]}),
+ batch_size=1,
+ )
+ self.assertEqual([], result.take_all())
+ finally:
+ if started_ray:
+ ray.shutdown()
+
+ def test_scan_to_ray_nested_projection_output_names(self):
+ obs = self.conn.create_table(
+ "ray_nested_obs",
+ schema=_schema({
+ "id": pa.int32(),
+ "tag": pa.string(),
+ "payload": pa.struct([("a", pa.int64()), ("b", pa.string())]),
+ "image": pa.large_binary(),
+ }),
+ options=_PARQUET_OPTIONS,
+ )
+
+ _, _, visible_columns = (
+ obs.scan()
+ .where("tag = 'keep'")
+ .select(["id", "payload.a"])
+ ._blob_descriptor_query_read_builder()
+ )
+
+ self.assertEqual(["id", "payload_a"], visible_columns)
+
def test_fetch_bodies_decodes_descriptor_inline_and_null(self):
# Cells may be descriptor bytes (incl. -1 read-to-EOF), inline bytes,
or null.
from pypaimon.multimodal.query import ScanQuery
