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 d7bae9ba15 [python] Add multimodal blob store API (#8432)
d7bae9ba15 is described below
commit d7bae9ba15743e4e2fe22c194118d72c24ec8164
Author: Jingsong Lee <[email protected]>
AuthorDate: Thu Jul 2 23:26:33 2026 +0800
[python] Add multimodal blob store API (#8432)
Adds an S3-like object facade for PyPaimon multimodal BLOB columns. The
API supports object-style writes, reads, listing, deletion, byte ranges,
table-column projection, and Paimon-native descriptor/reference storage.
---
docs/docs/pypaimon/multimodal-api.mdx | 161 ++++-
paimon-python/pypaimon/multimodal/__init__.py | 12 +
paimon-python/pypaimon/multimodal/blob_store.py | 646 +++++++++++++++++++++
paimon-python/pypaimon/multimodal/connection.py | 1 +
paimon-python/pypaimon/multimodal/table.py | 4 +
.../pypaimon/tests/multimodal_table_test.py | 253 ++++++++
6 files changed, 1071 insertions(+), 6 deletions(-)
diff --git a/docs/docs/pypaimon/multimodal-api.mdx
b/docs/docs/pypaimon/multimodal-api.mdx
index 2dca79feaf..b061aa36c0 100644
--- a/docs/docs/pypaimon/multimodal-api.mdx
+++ b/docs/docs/pypaimon/multimodal-api.mdx
@@ -34,12 +34,12 @@ a Paimon catalog plus a default database, and exposes a
compact table API for
scalar, text, vector, and blob data.
Creating a multimodal table enables `row-tracking.enabled`,
-`data-evolution.enabled`, and `deletion-vectors.enabled` by default. Global
-index scans use full fallback coverage through `global-index.search-mode` set
-to `FULL`. Data files use Vortex by default through `file.format = vortex`, and
-vector columns are stored in dedicated Vortex files through
-`vector.file.format` set to `vortex`. Install the Vortex extra when writing or
-reading default multimodal tables from Python:
+`data-evolution.enabled`, `deletion-vectors.enabled`, and
+`blob-as-descriptor` by default. Global index scans use full fallback coverage
+through `global-index.search-mode` set to `FULL`. Data files use Vortex by
+default through `file.format = vortex`, and vector columns are stored in
+dedicated Vortex files through `vector.file.format` set to `vortex`. Install
+the Vortex extra when writing or reading default multimodal tables from Python:
```shell
pip install pypaimon[vortex]
@@ -455,3 +455,152 @@ batch_neighbors = (
.to_list()
)
```
+
+## Blobs
+
+Use `blobs()` to work with a BLOB column through an S3-like object API. A blob
+store maps object keys to a table column such as `key`, `object_key`, or an
+explicit `key_column`.
+
+### Put Objects
+
+`put_object` writes one object. `put_objects` writes a batch of objects in one
+Paimon commit. Both methods upsert by key: existing matching rows are updated,
+and missing keys append new rows. If the same key appears more than once in a
+single `put_objects` batch, the last object wins. BlobStore does not enforce
+object-key uniqueness; that should be modeled with a table-level unique key
when
+available. The object body can be raw bytes, a binary file-like object, or a
+PyPaimon `Blob`. Use `columns` to set the non-key, non-BLOB table columns for
+the object row. For managed BLOB storage, use `head_object` or `get_object` to
+inspect the stored descriptor after the write.
+
+```python
+store = docs.blobs(column="image", key_column="key")
+
+store.put_object(
+ "images/cat.jpg",
+ body=open("cat.jpg", "rb"),
+ columns={"content_type": "image/jpeg"},
+)
+
+store.put_objects([
+ {
+ "key": "images/dog.jpg",
+ "body": open("dog.jpg", "rb"),
+ "columns": {"content_type": "image/jpeg"},
+ },
+ {
+ "key": "images/logo.png",
+ "body": b"...",
+ "columns": {"content_type": "image/png"},
+ },
+])
+```
+
+For Paimon-native reference storage, configure the BLOB column as a
+`blob-descriptor-field` and pass `uri`/`offset`/`length` or `descriptor` to
+`put_object` and `put_objects`. The table stores the external `BlobDescriptor`
+instead of copying the object into Paimon `.blob` files. Without
+`blob-descriptor-field`, descriptor-backed inputs are streamed into managed
+Paimon `.blob` files. External S3 URIs are read with the table's FileIO, so S3
+credentials such as `fs.s3.accessKeyId`, `fs.s3.accessKeySecret`,
+`fs.s3.securityToken`, `fs.s3.endpoint`, and `fs.s3.region` can be supplied in
+the options passed to `connect`. The same options are used for every external
+URI; there is no per-object credential override.
+
+```python
+store.put_objects([
+ {
+ "key": "videos/intro.mp4",
+ "uri": "s3://bucket/videos/intro.mp4",
+ "offset": 0,
+ "length": intro_content_length,
+ "columns": {"content_type": "video/mp4"},
+ },
+ {
+ "key": "videos/demo.mp4",
+ "uri": "s3://bucket/videos/demo.mp4",
+ "offset": 0,
+ "length": demo_content_length,
+ "columns": {"content_type": "video/mp4"},
+ },
+])
+
+store.put_object(
+ "videos/trailer.mp4",
+ uri="s3://bucket/videos/trailer.mp4",
+ offset=0,
+ length=trailer_content_length,
+ columns={"content_type": "video/mp4"},
+)
+```
+
+### Read Objects
+
+`get_object` reads an object by key. Its `range` parameter accepts a single
+S3-style byte range such as `bytes=0-1023`, `bytes=1024-`, or `bytes=-512`.
+`head_object` and `list_objects` return object information without opening the
+object stream. `get_object`, `head_object`, and `list_objects` expose non-key,
+non-BLOB table columns through `columns`. These columns are all returned by
+default. Pass `columns` to return only selected columns, or `[]` to skip them.
+`list_objects` requires a non-negative `limit`; `limit=0` returns an empty
+list.
+
+```python
+obj = store.get_object("images/cat.jpg", range="bytes=0-1023")
+with obj.open() as stream:
+ chunk = stream.read()
+
+info = store.head_object(
+ "images/cat.jpg",
+ columns=["content_type"],
+)
+content_type = info.columns["content_type"]
+
+objects = store.list_objects(prefix="images/", columns=[])
+for obj_info in objects:
+ print(obj_info.key, obj_info.size, obj_info.columns)
+```
+
+### Delete Objects
+
+`delete_object` deletes one object key. `delete_objects` deletes a batch of
keys
+in one Paimon commit. Missing keys are ignored, and repeated keys in the same
+request are folded before deletion. BlobStore does not enforce object-key
+uniqueness; if multiple table rows already match a key, deleting that key
removes
+all matching rows.
+
+```python
+store.delete_object("images/dog.jpg")
+
+store.delete_objects([
+ "images/old-1.jpg",
+ "images/old-2.jpg",
+])
+```
+
+### Update Columns
+
+Amazon S3 updates object metadata by copying the object to the same key with
+replacement metadata. BlobStore exposes the corresponding Paimon-native
+operation directly: `update_object_columns` and `update_objects_columns` update
+non-key, non-BLOB table columns without rewriting the blob data. Missing keys
+raise `NoSuchKey`.
+
+```python
+store.update_object_columns(
+ "images/cat.jpg",
+ {"content_type": "image/webp", "owner": "alice"},
+)
+
+store.update_objects_columns([
+ {
+ "key": "images/dog.jpg",
+ "columns": {"content_type": "image/webp"},
+ },
+ {
+ "key": "images/logo.png",
+ "columns": {"content_type": "image/svg+xml"},
+ },
+])
+```
diff --git a/paimon-python/pypaimon/multimodal/__init__.py
b/paimon-python/pypaimon/multimodal/__init__.py
index 2f957630d4..1b5060ec8f 100644
--- a/paimon-python/pypaimon/multimodal/__init__.py
+++ b/paimon-python/pypaimon/multimodal/__init__.py
@@ -17,6 +17,13 @@
"""High-level APIs for mutable multimodal Paimon tables."""
+from pypaimon.multimodal.blob_store import (
+ BlobObject,
+ BlobStore,
+ NoSuchKey,
+ ObjectInfo,
+ PutObjectResult,
+)
from pypaimon.multimodal.connection import MultimodalConnection, connect
from pypaimon.multimodal.table import (
MultimodalTable,
@@ -32,8 +39,13 @@ from pypaimon.table.data_evolution_merge_into import (
)
__all__ = [
+ "BlobObject",
+ "BlobStore",
"MultimodalConnection",
"MultimodalTable",
+ "NoSuchKey",
+ "ObjectInfo",
+ "PutObjectResult",
"TextRoute",
"VectorRoute",
"connect",
diff --git a/paimon-python/pypaimon/multimodal/blob_store.py
b/paimon-python/pypaimon/multimodal/blob_store.py
new file mode 100644
index 0000000000..e4f4b654b5
--- /dev/null
+++ b/paimon-python/pypaimon/multimodal/blob_store.py
@@ -0,0 +1,646 @@
+# 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 io
+import re
+from dataclasses import dataclass
+from typing import BinaryIO, Dict, Iterable, List, Mapping, Optional, Sequence
+
+from pypaimon.common.options.core_options import CoreOptions
+from pypaimon.common.predicate_builder import PredicateBuilder
+from pypaimon.snapshot.snapshot import BATCH_COMMIT_IDENTIFIER
+from pypaimon.table.row.blob import Blob, BlobData, BlobDescriptor
+from pypaimon.table.row.generic_row import GenericRow
+from pypaimon.table.special_fields import SpecialFields
+from pypaimon.write.table_update_by_row_id import TableUpdateByRowId
+
+
+_RANGE_PATTERN = re.compile(r"^bytes=(\d*)-(\d*)$")
+
+
+class NoSuchKey(KeyError):
+ """Raised when a blob key does not exist in a multimodal blob store."""
+
+
+@dataclass(frozen=True)
+class PutObjectResult:
+ key: object
+ size: Optional[int]
+
+
+@dataclass(frozen=True)
+class ObjectInfo:
+ key: object
+ size: Optional[int]
+ descriptor: Optional[BlobDescriptor]
+ columns: Dict[str, object]
+
+
+@dataclass(frozen=True)
+class BlobObject:
+ key: object
+ descriptor: BlobDescriptor
+ columns: Dict[str, object]
+ file_io: object
+ range_header: Optional[str] = None
+
+ @property
+ def size(self) -> int:
+ return self.descriptor.length
+
+ @property
+ def content_length(self) -> int:
+ descriptor = _descriptor_for_range(self.descriptor, self.range_header)
+ return descriptor.length
+
+ def open(self) -> BinaryIO:
+ descriptor = _descriptor_for_range(self.descriptor, self.range_header)
+ uri_reader = self.file_io.uri_reader_factory.create(descriptor.uri)
+ return Blob.from_descriptor(uri_reader, descriptor).new_input_stream()
+
+ def read(self) -> bytes:
+ with self.open() as stream:
+ return stream.read()
+
+
+class BlobStore:
+ """S3-like object facade over a Paimon multimodal table BLOB column."""
+
+ def __init__(
+ self,
+ table,
+ column: Optional[str] = None,
+ key_column: Optional[str] = None):
+ self._table = table
+ self._raw_table = table.raw_table
+ self.column = column or self._infer_blob_column()
+ self.key_column = key_column or self._infer_key_column()
+ self._validate_columns()
+
+ def put_object(
+ self,
+ key,
+ body=None,
+ *,
+ uri: Optional[str] = None,
+ offset: int = 0,
+ length: int = -1,
+ descriptor: Optional[BlobDescriptor] = None,
+ columns: Optional[Mapping[str, object]] = None) -> PutObjectResult:
+ return self.put_objects([
+ {
+ "key": key,
+ "body": body,
+ "uri": uri,
+ "offset": offset,
+ "length": length,
+ "descriptor": descriptor,
+ "columns": dict(columns or {}),
+ }
+ ])[0]
+
+ def put_objects(self, objects: Iterable[Mapping[str, object]]) ->
List[PutObjectResult]:
+ rows = []
+ for obj in objects:
+ key = _require_key(obj)
+ row = dict(obj.get("columns") or {})
+ row[self.key_column] = key
+ row[self.column] = self._object_to_blob(obj)
+ rows.append(row)
+ if not rows:
+ return []
+ return self._put_rows(self._deduplicate_rows_last_write_wins(rows))
+
+ def update_object_columns(
+ self,
+ key,
+ columns: Mapping[str, object]) -> ObjectInfo:
+ return self.update_objects_columns([
+ {"key": key, "columns": columns}
+ ])[0]
+
+ def update_objects_columns(
+ self,
+ objects: Iterable[Mapping[str, object]]) -> List[ObjectInfo]:
+ updates = []
+ keys = []
+ for obj in objects:
+ key = _require_key(obj)
+ columns = dict(obj.get("columns") or {})
+ if not columns:
+ raise ValueError("columns must not be empty.")
+ self._selected_update_columns(columns.keys())
+ updates.append((key, columns))
+ keys.append(key)
+ if not updates:
+ return []
+ self._validate_unique_keys(keys)
+ self._commit_column_updates(updates)
+ return [
+ self.head_object(key)
+ for key in keys
+ ]
+
+ def get_object(
+ self,
+ key,
+ *,
+ range: Optional[str] = None,
+ columns: Optional[Sequence[str]] = None) -> BlobObject:
+ info = self.head_object(key, columns=columns)
+ if info.descriptor is None:
+ raise NoSuchKey(key)
+ _descriptor_for_range(info.descriptor, range)
+ return BlobObject(
+ key=info.key,
+ descriptor=info.descriptor,
+ columns=info.columns,
+ file_io=self._raw_table.file_io,
+ range_header=range,
+ )
+
+ def head_object(
+ self,
+ key,
+ *,
+ columns: Optional[Sequence[str]] = None) -> ObjectInfo:
+ rows = self._read_rows(
+ keys=[key],
+ include_blob=True,
+ columns=columns,
+ )
+ if not rows:
+ raise NoSuchKey(key)
+ if len(rows) > 1:
+ raise ValueError("Multiple rows found for blob key %r." % key)
+ return self._row_to_info(rows[0])
+
+ def list_objects(
+ self,
+ *,
+ prefix: Optional[str] = None,
+ limit: Optional[int] = None,
+ columns: Optional[Sequence[str]] = None) -> List[ObjectInfo]:
+ if limit is not None:
+ if limit < 0:
+ raise ValueError("limit must be greater than or equal to 0.")
+ if limit == 0:
+ return []
+ rows = self._read_rows(
+ include_blob=True,
+ columns=columns,
+ )
+ objects = []
+ for row in rows:
+ key = row[self.key_column]
+ if prefix is not None and not str(key).startswith(prefix):
+ continue
+ objects.append(self._row_to_info(row))
+ if limit is not None and len(objects) >= limit:
+ break
+ return objects
+
+ def delete_object(self, key) -> None:
+ self.delete_objects([key])
+
+ def delete_objects(self, keys: Sequence[object]) -> None:
+ if not keys:
+ return
+ keys = self._deduplicate_keys(keys)
+ write_builder = self._raw_table.new_batch_write_builder()
+ table_update = write_builder.new_update()
+ table_commit = write_builder.new_commit()
+ try:
+ messages = table_update.delete_by_predicate(
+ self._keys_predicate(keys))
+ table_commit.commit(messages)
+ finally:
+ table_commit.close()
+
+ def _put_rows(self, rows: List[Mapping[str, object]]) ->
List[PutObjectResult]:
+ self._commit_upsert_once(rows)
+ return [
+ self._put_result_from_row(row)
+ for row in rows
+ ]
+
+ def _deduplicate_rows_last_write_wins(
+ self,
+ rows: List[Mapping[str, object]]) -> List[Mapping[str, object]]:
+ key_to_last_index = {}
+ for index, row in enumerate(rows):
+ key_to_last_index[row[self.key_column]] = index
+ if len(key_to_last_index) == len(rows):
+ return rows
+ return [
+ rows[index]
+ for index in sorted(key_to_last_index.values())
+ ]
+
+ def _deduplicate_keys(self, keys: Sequence[object]) -> List[object]:
+ unique_keys = []
+ seen = set()
+ for key in keys:
+ if key in seen:
+ continue
+ seen.add(key)
+ unique_keys.append(key)
+ return unique_keys
+
+ def _commit_upsert_once(self, rows: List[Mapping[str, object]]) -> None:
+ generic_rows = self._rows_to_generic_rows(rows)
+ write_builder = self._raw_table.new_batch_write_builder()
+ table_update = write_builder.new_update()
+ table_commit = write_builder.new_commit()
+ try:
+ messages = table_update.upsert_by_key(
+ generic_rows, [self.key_column])
+ table_commit.commit(messages)
+ finally:
+ table_commit.close()
+
+ def _commit_column_updates(self, updates: Sequence[tuple]) -> None:
+ update_columns = self._column_update_names(updates)
+ targets = self._read_update_targets(
+ [key for key, _ in updates],
+ update_columns,
+ )
+ rows = []
+ row_ids_by_row = []
+ for key, columns in updates:
+ target = targets[key]
+ row = dict(target["columns"])
+ row.update(columns)
+ row[self.key_column] = key
+ rows.append(self._row_to_generic_row(row))
+ row_ids_by_row.append([target["row_id"]])
+
+ write_builder = self._raw_table.new_batch_write_builder()
+ table_commit = write_builder.new_commit()
+ try:
+ messages = TableUpdateByRowId(
+ self._raw_table,
+ write_builder.commit_user,
+ BATCH_COMMIT_IDENTIFIER,
+ ).update_rows_columns(rows, row_ids_by_row, update_columns)
+ if messages:
+ table_commit.commit(messages)
+ finally:
+ table_commit.close()
+
+ def _column_update_names(self, updates: Sequence[tuple]) -> List[str]:
+ requested = set()
+ for _, columns in updates:
+ requested.update(columns)
+ return [
+ name for name in self._raw_table.field_names
+ if name in requested
+ ]
+
+ def _read_update_targets(
+ self,
+ keys: Sequence[object],
+ columns: Sequence[str]) -> Dict[object, dict]:
+ read_builder = self._raw_table.new_read_builder()
+ projection = [self.key_column, SpecialFields.ROW_ID.name]
+ projection.extend(columns)
+ read_builder = read_builder.with_projection(projection)
+ read_builder = read_builder.with_filter(self._keys_predicate(keys))
+ plan = read_builder.new_scan().plan()
+ table = read_builder.new_read().to_arrow(plan.splits())
+
+ targets = {}
+ duplicates = []
+ for row in table.to_pylist():
+ key = row[self.key_column]
+ if key in targets:
+ duplicates.append(key)
+ continue
+ targets[key] = {
+ "row_id": row[SpecialFields.ROW_ID.name],
+ "columns": {
+ name: row.get(name)
+ for name in columns
+ },
+ }
+
+ missing = [key for key in keys if key not in targets]
+ if missing:
+ raise NoSuchKey(missing[0])
+ if duplicates:
+ raise ValueError("Multiple rows found for blob keys: %s" %
duplicates)
+ return targets
+
+ def _put_result_from_row(self, row: Mapping[str, object]) ->
PutObjectResult:
+ blob = row.get(self.column)
+ return PutObjectResult(
+ key=row[self.key_column],
+ size=_blob_size(blob),
+ )
+
+ def _rows_to_generic_rows(self, rows: List[Mapping[str, object]]) ->
List[GenericRow]:
+ return [
+ self._row_to_generic_row(row)
+ for row in rows
+ ]
+
+ def _row_to_generic_row(self, row: Mapping[str, object]) -> GenericRow:
+ return GenericRow(
+ [row.get(field.name) for field in self._raw_table.fields],
+ self._raw_table.fields,
+ )
+
+ def _read_rows(
+ self,
+ keys: Optional[Sequence[object]] = None,
+ include_blob: bool = False,
+ columns: Optional[Sequence[str]] = None) -> List[dict]:
+ read_table = self._raw_table.copy({
+ CoreOptions.BLOB_AS_DESCRIPTOR.key(): "true"
+ }) if include_blob else self._raw_table
+ read_builder = read_table.new_read_builder()
+ projection = self._projection(include_blob, columns)
+ read_builder = read_builder.with_projection(projection)
+ if keys is not None:
+ read_builder = read_builder.with_filter(self._keys_predicate(keys))
+ plan = read_builder.new_scan().plan()
+ table = read_builder.new_read().to_arrow(plan.splits())
+ return table.to_pylist()
+
+ def _projection(
+ self,
+ include_blob: bool,
+ selected_columns: Optional[Sequence[str]]) -> List[str]:
+ projection = [self.key_column]
+ if include_blob:
+ projection.append(self.column)
+ projection.extend(self._selected_columns(selected_columns))
+ return projection
+
+ def _row_to_info(self, row: Mapping[str, object]) -> ObjectInfo:
+ descriptor_bytes = row.get(self.column)
+ descriptor = None
+ if descriptor_bytes is not None:
+ descriptor =
BlobDescriptor.deserialize(_bytes_value(descriptor_bytes))
+ columns = {
+ name: value
+ for name, value in row.items()
+ if name not in (self.key_column, self.column)
+ }
+ return ObjectInfo(
+ key=row[self.key_column],
+ size=descriptor.length if descriptor is not None else None,
+ descriptor=descriptor,
+ columns=columns,
+ )
+
+ def _keys_predicate(self, keys: Sequence[object]):
+ builder = PredicateBuilder(self._raw_table.fields)
+ values = list(keys)
+ if len(values) == 1:
+ return builder.equal(self.key_column, values[0])
+ return builder.is_in(self.key_column, values)
+
+ def _infer_blob_column(self) -> str:
+ blob_columns = [
+ field.name for field in self._raw_table.fields
+ if _is_blob_field(field)
+ ]
+ if len(blob_columns) != 1:
+ raise ValueError(
+ "Blob column is required when table has %d BLOB columns: %s."
+ % (len(blob_columns), blob_columns)
+ )
+ return blob_columns[0]
+
+ def _infer_key_column(self) -> str:
+ for candidate in ("key", "object_key", "path"):
+ if candidate in self._raw_table.field_names:
+ return candidate
+ raise ValueError(
+ "key_column is required; no default key/object_key/path column "
+ "exists in table schema."
+ )
+
+ def _validate_columns(self):
+ if self.key_column not in self._raw_table.field_names:
+ raise ValueError("key_column %r is not in table schema." %
self.key_column)
+ if self.column not in self._raw_table.field_names:
+ raise ValueError("blob column %r is not in table schema." %
self.column)
+ if not _is_blob_field(self._raw_table.field_dict[self.column]):
+ raise ValueError("Column %r is not a BLOB column." % self.column)
+ if self.key_column == self.column:
+ raise ValueError("key_column and blob column must be different.")
+
+ def _object_to_blob(self, obj: Mapping[str, object]) -> Blob:
+ has_body = "body" in obj and obj.get("body") is not None
+ has_reference = obj.get("descriptor") is not None or obj.get("uri") is
not None
+ if has_body and has_reference:
+ raise ValueError(
+ "Blob object spec must use either 'body' or descriptor/uri,
not both."
+ )
+ if has_reference:
+ descriptor = _coerce_descriptor(obj)
+ uri_reader =
self._raw_table.file_io.uri_reader_factory.create(descriptor.uri)
+ return Blob.from_descriptor(uri_reader, descriptor)
+ if not has_body:
+ raise ValueError("Blob object spec requires 'body', 'descriptor',
or 'uri'.")
+ body = obj.get("body")
+ if isinstance(body, Blob):
+ try:
+ descriptor = body.to_descriptor()
+ except RuntimeError:
+ return body
+ uri_reader =
self._raw_table.file_io.uri_reader_factory.create(descriptor.uri)
+ return Blob.from_descriptor(uri_reader, descriptor)
+ return BlobData(_body_to_bytes(body))
+
+ def _selected_columns(
+ self,
+ columns: Optional[Sequence[str]]) -> List[str]:
+ if columns is None:
+ return [
+ name for name in self._raw_table.field_names
+ if name not in (self.key_column, self.column)
+ ]
+ if isinstance(columns, str):
+ names = [columns]
+ else:
+ names = list(columns)
+ self._validate_unique_selected_columns(names)
+ for name in names:
+ if name in (self.key_column, self.column):
+ raise ValueError(
+ "columns must not include key or blob column %r." % name
+ )
+ if name not in self._raw_table.field_names:
+ raise ValueError("Column %r is not in table schema." % name)
+ return names
+
+ def _selected_update_columns(self, columns: Sequence[str]) -> List[str]:
+ names = self._selected_columns(columns)
+ partition_keys = set(self._raw_table.partition_keys)
+ for name in names:
+ if name in partition_keys:
+ raise ValueError(
+ "columns must not include partition column %r." % name
+ )
+ return names
+
+ @staticmethod
+ def _validate_unique_selected_columns(names: Sequence[str]):
+ seen = set()
+ duplicates = set()
+ for name in names:
+ if name in seen:
+ duplicates.add(name)
+ seen.add(name)
+ if duplicates:
+ raise ValueError("Duplicate columns: %s" % sorted(duplicates))
+
+ @staticmethod
+ def _validate_unique_keys(keys: Sequence[object]):
+ seen = set()
+ duplicates = set()
+ for key in keys:
+ if key in seen:
+ duplicates.add(key)
+ seen.add(key)
+ if duplicates:
+ raise ValueError("Duplicate blob keys: %s" % sorted(duplicates))
+
+
+def _require_key(obj: Mapping[str, object]):
+ if "key" not in obj:
+ raise ValueError("Blob object spec requires 'key'.")
+ key = obj["key"]
+ if key is None:
+ raise ValueError("Blob object key must not be None.")
+ return key
+
+
+def _body_to_bytes(body) -> bytes:
+ if isinstance(body, bytes):
+ return body
+ if isinstance(body, bytearray):
+ return bytes(body)
+ if isinstance(body, memoryview):
+ return body.tobytes()
+ if isinstance(body, str):
+ return body.encode("utf-8")
+ if isinstance(body, io.IOBase) or hasattr(body, "read"):
+ data = body.read()
+ if isinstance(data, str):
+ return data.encode("utf-8")
+ return bytes(data)
+ raise ValueError("Unsupported blob body type: %r." % type(body))
+
+
+def _coerce_descriptor(obj: Mapping[str, object]) -> BlobDescriptor:
+ descriptor = obj.get("descriptor")
+ if descriptor is not None:
+ if isinstance(descriptor, BlobDescriptor):
+ return descriptor
+ if isinstance(descriptor, (bytes, bytearray)):
+ return BlobDescriptor.deserialize(bytes(descriptor))
+ raise ValueError("descriptor must be BlobDescriptor or serialized
bytes.")
+
+ uri = obj.get("uri")
+ if not uri:
+ raise ValueError("Reference object spec requires 'uri' or
'descriptor'.")
+ return BlobDescriptor(
+ uri=str(uri),
+ offset=int(obj.get("offset", 0)),
+ length=int(obj.get("length", -1)),
+ )
+
+
+def _descriptor_for_range(
+ descriptor: BlobDescriptor,
+ range_header: Optional[str]) -> BlobDescriptor:
+ if range_header is None:
+ return descriptor
+ start, length = _parse_range(range_header, descriptor.length)
+ return BlobDescriptor(descriptor.uri, descriptor.offset + start, length)
+
+
+def _parse_range(range_header: str, total_length: int):
+ match = _RANGE_PATTERN.match(range_header)
+ if match is None:
+ raise ValueError("Range must use S3 syntax like 'bytes=0-1023'.")
+ start_text, end_text = match.groups()
+ if not start_text and not end_text:
+ raise ValueError("Range must specify a start or end byte.")
+
+ if start_text:
+ start = int(start_text)
+ if total_length >= 0 and start >= total_length:
+ raise ValueError("Range start is past the end of the object.")
+ if end_text:
+ end = int(end_text)
+ if end < start:
+ raise ValueError("Range end must be greater than or equal to
start.")
+ if total_length >= 0:
+ end = min(end, total_length - 1)
+ length = end - start + 1
+ else:
+ if total_length < 0:
+ length = -1
+ else:
+ length = max(total_length - start, 0)
+ else:
+ if total_length < 0:
+ raise ValueError("Suffix ranges require a known object length.")
+ suffix_length = int(end_text)
+ if suffix_length < 0:
+ raise ValueError("Suffix range length must be non-negative.")
+ length = min(suffix_length, total_length)
+ start = total_length - length
+
+ return start, length
+
+
+def _blob_descriptor(blob) -> Optional[BlobDescriptor]:
+ if blob is None:
+ return None
+ try:
+ return blob.to_descriptor()
+ except RuntimeError:
+ return None
+
+
+def _blob_size(blob) -> Optional[int]:
+ descriptor = _blob_descriptor(blob)
+ if descriptor is not None:
+ return descriptor.length if descriptor.length >= 0 else None
+ if isinstance(blob, BlobData):
+ return len(blob.data)
+ return None
+
+
+def _bytes_value(value) -> bytes:
+ if hasattr(value, "as_py"):
+ value = value.as_py()
+ if isinstance(value, bytearray):
+ value = bytes(value)
+ if not isinstance(value, bytes):
+ raise ValueError("Expected serialized BlobDescriptor bytes.")
+ return value
+
+
+def _is_blob_field(field) -> bool:
+ return getattr(field.type, "type", None) == "BLOB"
diff --git a/paimon-python/pypaimon/multimodal/connection.py
b/paimon-python/pypaimon/multimodal/connection.py
index b06940e620..2314eecded 100644
--- a/paimon-python/pypaimon/multimodal/connection.py
+++ b/paimon-python/pypaimon/multimodal/connection.py
@@ -32,6 +32,7 @@ _DEFAULT_OPTIONS = {
"row-tracking.enabled": "true",
"data-evolution.enabled": "true",
"deletion-vectors.enabled": "true",
+ "blob-as-descriptor": "true",
"file.format": "vortex",
"global-index.search-mode": "full",
"vector.file.format": "vortex",
diff --git a/paimon-python/pypaimon/multimodal/table.py
b/paimon-python/pypaimon/multimodal/table.py
index bc032d6644..916938137c 100644
--- a/paimon-python/pypaimon/multimodal/table.py
+++ b/paimon-python/pypaimon/multimodal/table.py
@@ -141,6 +141,10 @@ class MultimodalTable:
def scan(self):
return ScanQuery(self.raw_table)
+ def blobs(self, *, column: Optional[str] = None, key_column: Optional[str]
= None):
+ from pypaimon.multimodal.blob_store import BlobStore
+ return BlobStore(self, column=column, key_column=key_column)
+
def search(
self,
query,
diff --git a/paimon-python/pypaimon/tests/multimodal_table_test.py
b/paimon-python/pypaimon/tests/multimodal_table_test.py
index 1259ea7319..e745932c21 100644
--- a/paimon-python/pypaimon/tests/multimodal_table_test.py
+++ b/paimon-python/pypaimon/tests/multimodal_table_test.py
@@ -15,6 +15,7 @@
# specific language governing permissions and limitations
# under the License.
+import io
import json
import os
import shutil
@@ -103,6 +104,7 @@ class MultimodalTableTest(unittest.TestCase):
self.assertEqual("true", options["row-tracking.enabled"])
self.assertEqual("true", options["data-evolution.enabled"])
self.assertEqual("true", options["deletion-vectors.enabled"])
+ self.assertEqual("true", options["blob-as-descriptor"])
self.assertNotIn("data-evolution.row-sidecar.enabled", options)
self.assertEqual("vortex", options["file.format"])
self.assertEqual("full", options["global-index.search-mode"])
@@ -182,6 +184,257 @@ class MultimodalTableTest(unittest.TestCase):
self.assertEqual("ROW<rank: INT>", types_by_name["meta"])
self.assertEqual("VECTOR<FLOAT, 3>", types_by_name["embedding"])
+ def test_blob_store_put_objects_get_list_and_delete(self):
+ table = self.conn.create_table(
+ "objects",
+ schema=_schema({
+ "key": pa.string(),
+ "image": pa.large_binary(),
+ "content_type": pa.string(),
+ "owner": pa.string(),
+ }),
+ options=_PARQUET_OPTIONS,
+ )
+ store = table.blobs(column="image")
+
+ results = store.put_objects([
+ {
+ "key": "images/cat.jpg",
+ "body": b"cat-image-old",
+ "columns": {"content_type": "image/gif", "owner": "ignored"},
+ },
+ {
+ "key": "images/cat.jpg",
+ "body": b"cat-image-v1",
+ "columns": {"content_type": "image/jpeg", "owner": "alice"},
+ },
+ {
+ "key": "images/dog.jpg",
+ "body": bytearray(b"dog-image"),
+ "columns": {"content_type": "image/jpeg", "owner": "bob"},
+ },
+ ])
+
+ self.assertEqual(["images/cat.jpg", "images/dog.jpg"],
+ [result.key for result in results])
+ self.assertEqual([12, 9], [result.size for result in results])
+
+ cat = store.get_object("images/cat.jpg")
+ self.assertEqual(b"cat-image-v1", cat.read())
+ self.assertEqual(b"at-i", store.get_object(
+ "images/cat.jpg", range="bytes=1-4").read())
+ clipped = store.get_object("images/cat.jpg", range="bytes=10-999")
+ self.assertEqual(2, clipped.content_length)
+ self.assertEqual(b"v1", clipped.read())
+ with self.assertRaisesRegex(ValueError, "Range start"):
+ store.get_object("images/cat.jpg", range="bytes=12-13")
+ self.assertEqual("image/jpeg", cat.columns["content_type"])
+ self.assertEqual("alice", cat.columns["owner"])
+ owner_only = store.get_object(
+ "images/cat.jpg",
+ columns=["owner"],
+ )
+ self.assertEqual({"owner": "alice"}, owner_only.columns)
+
+ listed = store.list_objects(prefix="images/")
+ self.assertEqual(["images/cat.jpg", "images/dog.jpg"],
+ sorted(obj.key for obj in listed))
+ self.assertEqual(
+ {
+ "images/cat.jpg": "image/jpeg",
+ "images/dog.jpg": "image/jpeg",
+ },
+ {obj.key: obj.columns["content_type"] for obj in listed},
+ )
+ listed_without_columns = store.list_objects(
+ prefix="images/",
+ columns=[],
+ )
+ self.assertEqual(
+ {"images/cat.jpg": {}, "images/dog.jpg": {}},
+ {obj.key: obj.columns for obj in listed_without_columns},
+ )
+ self.assertEqual([], store.list_objects(prefix="images/", limit=0))
+ with self.assertRaisesRegex(ValueError, "limit"):
+ store.list_objects(prefix="images/", limit=-1)
+
+ store.put_object(
+ "images/cat.jpg",
+ b"cat-image-v2",
+ columns={"content_type": "image/png", "owner": "alice"},
+ )
+ self.assertEqual(b"cat-image-v2",
store.get_object("images/cat.jpg").read())
+ info = store.head_object("images/cat.jpg")
+ self.assertEqual("image/png", info.columns["content_type"])
+ self.assertEqual(
+ {"content_type": "image/png"},
+ store.head_object(
+ "images/cat.jpg",
+ columns="content_type",
+ ).columns,
+ )
+ self.assertEqual(2, table.scan().to_arrow().num_rows)
+
+ previous_descriptor = info.descriptor
+ updated = store.update_object_columns(
+ "images/cat.jpg",
+ {"content_type": "image/webp"},
+ )
+ self.assertEqual(previous_descriptor, updated.descriptor)
+ self.assertEqual("image/webp", updated.columns["content_type"])
+ self.assertEqual("alice", updated.columns["owner"])
+ self.assertEqual(b"cat-image-v2",
store.get_object("images/cat.jpg").read())
+
+ batch_updates = store.update_objects_columns([
+ {"key": "images/cat.jpg", "columns": {"owner": "carol"}},
+ {"key": "images/dog.jpg", "columns": {"owner": "dave"}},
+ ])
+ self.assertEqual(
+ ["images/cat.jpg", "images/dog.jpg"],
+ [obj.key for obj in batch_updates],
+ )
+ self.assertEqual("carol",
store.head_object("images/cat.jpg").columns["owner"])
+ self.assertEqual("dave",
store.head_object("images/dog.jpg").columns["owner"])
+ self.assertEqual(2, table.scan().to_arrow().num_rows)
+
+ with self.assertRaisesRegex(ValueError, "columns must not be empty"):
+ store.update_object_columns("images/cat.jpg", {})
+ with self.assertRaises(pmm.NoSuchKey):
+ store.update_object_columns("images/missing.jpg", {"owner":
"nobody"})
+ with self.assertRaisesRegex(ValueError, "columns must not include"):
+ store.update_object_columns("images/cat.jpg", {"image": b"new"})
+
+ store.delete_objects(["images/dog.jpg", "images/dog.jpg"])
+ with self.assertRaises(pmm.NoSuchKey):
+ store.head_object("images/dog.jpg")
+ self.assertEqual(["images/cat.jpg"],
+ [obj.key for obj in
store.list_objects(prefix="images/")])
+ store.delete_object("images/cat.jpg")
+ self.assertEqual([], store.list_objects(prefix="images/"))
+
+ def test_blob_store_put_object_accepts_blob_without_materializing(self):
+ from pypaimon.table.row.blob import Blob, BlobDescriptor
+
+ table = self.conn.create_table(
+ "streamed_objects",
+ schema=_schema({
+ "key": pa.string(),
+ "payload": pa.large_binary(),
+ }),
+ options=_PARQUET_OPTIONS,
+ )
+ data = b"streamed-managed-payload"
+ stream_uri = "stream://payloads/1"
+ stream_calls = []
+
+ class StreamOnlyReader:
+
+ def new_input_stream(self, uri):
+ stream_calls.append(("new_input_stream", uri))
+ return io.BytesIO(data)
+
+ class StreamOnlyReaderFactory:
+
+ def create(self, uri):
+ stream_calls.append(("create", uri))
+ return StreamOnlyReader()
+
+ class DescriptorOnlyBlob(Blob):
+
+ def to_data(self):
+ raise AssertionError("put_object should not materialize Blob
data")
+
+ def to_descriptor(self):
+ return BlobDescriptor(stream_uri, 0, len(data))
+
+ def new_input_stream(self):
+ raise AssertionError("put_object should use the URI stream")
+
+ store = table.blobs(column="payload")
+ original_factory = table.raw_table.file_io.uri_reader_factory
+ table.raw_table.file_io.uri_reader_factory = StreamOnlyReaderFactory()
+ try:
+ result = store.put_object("payloads/1", DescriptorOnlyBlob())
+ finally:
+ table.raw_table.file_io.uri_reader_factory = original_factory
+
+ self.assertEqual(len(data), result.size)
+ self.assertEqual([
+ ("create", stream_uri),
+ ("new_input_stream", stream_uri),
+ ], stream_calls)
+ stored = store.get_object("payloads/1")
+ self.assertNotEqual(stream_uri, stored.descriptor.uri)
+ self.assertEqual(data, stored.read())
+
+ def test_blob_store_put_object_reference_preserves_descriptor_uri(self):
+ table = self.conn.create_table(
+ "referenced_objects",
+ schema=_schema({
+ "object_key": pa.string(),
+ "payload": pa.large_binary(),
+ "media_type": pa.string(),
+ }),
+ options=dict(_PARQUET_OPTIONS, **{
+ "blob-descriptor-field": "payload",
+ }),
+ )
+ data = b"external-video-payload"
+ external_path = os.path.join(self.temp_dir, "video.bin")
+ with open(external_path, "wb") as f:
+ f.write(data)
+
+ store = table.blobs(column="payload")
+ result = store.put_object(
+ "videos/1",
+ uri=external_path,
+ length=len(data),
+ columns={"media_type": "video/mp4"},
+ )
+ descriptor = store.head_object("videos/1").descriptor
+ more = store.put_objects([
+ {
+ "key": "videos/2",
+ "descriptor": descriptor,
+ "columns": {"media_type": "video/mp4"},
+ }
+ ])
+
+ self.assertEqual("videos/1", result.key)
+ self.assertEqual(len(data), result.size)
+ self.assertEqual("videos/2", more[0].key)
+ obj = store.get_object("videos/1")
+ self.assertEqual(data, obj.read())
+ self.assertEqual("video/mp4", obj.columns["media_type"])
+ self.assertEqual(external_path,
store.head_object("videos/1").descriptor.uri)
+ self.assertEqual(data, store.get_object("videos/2").read())
+
+ def test_blob_store_put_object_uri_streams_into_managed_blob(self):
+ table = self.conn.create_table(
+ "managed_only_objects",
+ schema=_schema({
+ "key": pa.string(),
+ "payload": pa.large_binary(),
+ }),
+ options=_PARQUET_OPTIONS,
+ )
+ data = b"external-managed-payload"
+ external_path = os.path.join(self.temp_dir, "managed.bin")
+ with open(external_path, "wb") as f:
+ f.write(data)
+
+ store = table.blobs(column="payload")
+ result = store.put_object(
+ "payloads/1",
+ uri=external_path,
+ length=len(data),
+ )
+
+ self.assertEqual(len(data), result.size)
+ stored = store.get_object("payloads/1")
+ self.assertNotEqual(external_path, stored.descriptor.uri)
+ self.assertEqual(data, stored.read())
+
def test_drop_table_can_ignore_missing_table(self):
self.conn.drop_table("missing", ignore_if_not_exists=True)
