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 cc4cd0dd28 [python] Add row-id APIs to multimodal table (#8442)
cc4cd0dd28 is described below
commit cc4cd0dd28b7cd36fcf26d811b07ba65809359d0
Author: Jingsong Lee <[email protected]>
AuthorDate: Fri Jul 3 16:01:46 2026 +0800
[python] Add row-id APIs to multimodal table (#8442)
Adds row-id manifest support to PyPaimon multimodal APIs so applications
can return `_ROW_ID` from scans/searches and fetch selected rows later
with `take_row_ids`. Persisted manifests can now be consumed against a
pinned source snapshot or tag.
---
docs/docs/pypaimon/multimodal-api.mdx | 123 +++++++++++++++
paimon-python/pypaimon/multimodal/query.py | 22 ++-
paimon-python/pypaimon/multimodal/table.py | 93 +++++++++--
.../pypaimon/tests/multimodal_table_test.py | 170 +++++++++++++++++++++
4 files changed, 394 insertions(+), 14 deletions(-)
diff --git a/docs/docs/pypaimon/multimodal-api.mdx
b/docs/docs/pypaimon/multimodal-api.mdx
index 83d13ae345..ebd1427367 100644
--- a/docs/docs/pypaimon/multimodal-api.mdx
+++ b/docs/docs/pypaimon/multimodal-api.mdx
@@ -332,6 +332,129 @@ result = (
)
```
+## Row IDs
+
+Multimodal tables enable `row-tracking.enabled` by default, so each row has a
+Paimon system column named `_ROW_ID`. Use `_ROW_ID` as an internal coordination
+key for retrieval, reranking, inference, and training jobs. Keep user-visible
+document IDs, object keys, or primary keys in normal columns.
+
+Use `with_row_id()` to include `_ROW_ID` in scan or search results. The method
+appends `_ROW_ID` to the current projection; if no projection is set, it
returns
+all table columns plus `_ROW_ID`.
+
+```python
+candidates = (
+ docs.search([0.1, 0.2, 0.3], column="embedding")
+ .where("category = 'lake'")
+ .select(["id", "content"])
+ .limit(100)
+ .with_row_id()
+ .to_pandas()
+)
+
+row_ids = candidates["_ROW_ID"].tolist()
+```
+
+When a row-id manifest is persisted beyond a short-lived latest-snapshot
+workflow, record the source snapshot or tag and pass it back to the read API.
+`scan`, `search`, `search_vectors`, `search_hybrid`, and `take_row_ids` accept
+`snapshot_id` or `tag_name`. These two options are mutually exclusive.
+
+Use `take_row_ids` to fetch rows selected by an earlier scan, vector search,
+full-text search, hybrid search, sampler, or split manifest. Results are not
+guaranteed to follow the input row-id order. Include `_ROW_ID` and reorder on
+the client if order matters.
+
+```python
+payload = (
+ docs.take_row_ids(row_ids, snapshot_id=source_snapshot_id)
+ .select(["id", "content", "image"])
+ .with_row_id()
+ .to_arrow()
+ .to_pylist()
+)
+
+payload_by_row_id = {row["_ROW_ID"]: row for row in payload}
+ordered_payload = [payload_by_row_id[row_id] for row_id in row_ids]
+```
+
+This pattern keeps broad candidate generation cheap: first search or filter to
+produce a compact row-id manifest, then fetch only the columns needed by the
+next stage.
+
+```python
+# Stage 1: broad retrieval with a narrow projection.
+manifest = (
+ docs.search(
+ query_vector,
+ column="embedding",
+ snapshot_id=source_snapshot_id,
+ )
+ .select(["id"])
+ .limit(500)
+ .with_row_id()
+ .to_pandas()
+)
+
+# Stage 2: expensive reranking payload, fetched only for candidates.
+rerank_payload = (
+ docs.take_row_ids(
+ manifest["_ROW_ID"].tolist(),
+ snapshot_id=source_snapshot_id,
+ )
+ .select(["id", "title", "body"])
+ .with_row_id()
+ .to_list()
+)
+
+payload_by_row_id = {row["_ROW_ID"]: row for row in rerank_payload}
+rerank_inputs = [
+ {
+ "row_id": row_id,
+ "candidate_rank": rank,
+ "doc": payload_by_row_id[row_id],
+ }
+ for rank, row_id in enumerate(manifest["_ROW_ID"])
+]
+```
+
+For offline inference, feature backfills, or training splits, store row IDs in
a
+work queue or manifest table together with the source table snapshot or tag
used
+to produce them. Workers can then read row-id batches and fetch only the
columns
+they need:
+
+```python
+def run_inference_worker(docs, row_id_batch, source_snapshot_id):
+ rows = (
+ docs.take_row_ids(row_id_batch, snapshot_id=source_snapshot_id)
+ .select(["id", "content"])
+ .with_row_id()
+ .to_list()
+ )
+
+ outputs = []
+ for row in rows:
+ outputs.append(
+ {
+ "id": row["id"],
+ "source_row_id": row["_ROW_ID"],
+ "prediction": model_predict(row["content"]),
+ }
+ )
+ return outputs
+```
+
+Best practices:
+
+- Treat `_ROW_ID` as an internal row handle, not a business identifier.
+- Store the source snapshot or tag with persisted row-id manifests.
+- Pass the stored `snapshot_id` or `tag_name` when consuming persisted row-id
+ manifests.
+- Use business keys or application columns when writing inference or training
+ outputs back to a table.
+- Reorder `take_row_ids` results client-side when input order matters.
+
## Create Index
Use `create_index` to create the global indexes used by search APIs. The
diff --git a/paimon-python/pypaimon/multimodal/query.py
b/paimon-python/pypaimon/multimodal/query.py
index be0d4fb065..86012b9608 100644
--- a/paimon-python/pypaimon/multimodal/query.py
+++ b/paimon-python/pypaimon/multimodal/query.py
@@ -18,6 +18,7 @@
from typing import Callable, List, Optional
from pypaimon.common.where_parser import parse_where_clause
+from pypaimon.table.special_fields import SpecialFields
class ScanQuery:
@@ -31,6 +32,7 @@ class ScanQuery:
self._predicate = None
self._projection = None
self._limit = None
+ self._include_row_id = False
self._result_factory = result_factory
def where(self, predicate):
@@ -45,6 +47,10 @@ class ScanQuery:
self._projection = list(columns)
return self
+ def with_row_id(self):
+ self._include_row_id = True
+ return self
+
def limit(self, limit: int):
self._limit = limit
return self
@@ -62,12 +68,24 @@ class ScanQuery:
read_builder = self._table.new_read_builder()
if self._predicate is not None:
read_builder = read_builder.with_filter(self._predicate)
- if self._projection is not None:
- read_builder = read_builder.with_projection(self._projection)
+ projection = self._effective_projection()
+ if projection is not None:
+ read_builder = read_builder.with_projection(projection)
if self._limit is not None:
read_builder = read_builder.with_limit(self._limit)
return read_builder
+ def _effective_projection(self):
+ if self._projection is None:
+ if not self._include_row_id:
+ return None
+ projection = [field.name for field in self._table.fields]
+ else:
+ projection = list(self._projection)
+ if self._include_row_id and SpecialFields.ROW_ID.name not in
projection:
+ projection.append(SpecialFields.ROW_ID.name)
+ return projection
+
def _read_global_index_result(self, result):
read_builder = self._configured_read_builder()
scan = read_builder.new_scan().with_global_index_result(result)
diff --git a/paimon-python/pypaimon/multimodal/table.py
b/paimon-python/pypaimon/multimodal/table.py
index 12de790060..a00f8d1156 100644
--- a/paimon-python/pypaimon/multimodal/table.py
+++ b/paimon-python/pypaimon/multimodal/table.py
@@ -21,6 +21,7 @@ from typing import Dict, Mapping, Optional, Sequence
import pyarrow as pa
+from pypaimon.common.options.core_options import CoreOptions
from pypaimon.globalindex.full_text_query import FullTextQuery
from pypaimon.multimodal.query import (
BatchVectorQuery,
@@ -35,6 +36,7 @@ from pypaimon.table.data_evolution_merge_into import (
WhenNotMatched,
source_col as _source_col,
)
+from pypaimon.table.special_fields import SpecialFields
_ALL_SOURCE_COLUMNS = object()
@@ -155,8 +157,32 @@ class MultimodalTable:
def merge(self, on):
return _MergeBuilder(self, on)
- def scan(self):
- return ScanQuery(self.raw_table)
+ def scan(
+ self,
+ *,
+ snapshot_id: Optional[int] = None,
+ tag_name: Optional[str] = None):
+ return ScanQuery(_time_travel_table(
+ self.raw_table, snapshot_id=snapshot_id, tag_name=tag_name))
+
+ def take_row_ids(
+ self,
+ row_ids,
+ *,
+ snapshot_id: Optional[int] = None,
+ tag_name: Optional[str] = None):
+ row_ids = _coerce_row_ids(row_ids)
+ read_table = _time_travel_table(
+ self.raw_table, snapshot_id=snapshot_id, tag_name=tag_name)
+ read_builder = read_table.new_read_builder().with_projection(
+ [field.name for field in read_table.fields]
+ + [SpecialFields.ROW_ID.name]
+ )
+ predicate = read_builder.new_predicate_builder().is_in(
+ SpecialFields.ROW_ID.name, row_ids)
+ query = ScanQuery(read_table)
+ query._predicate = predicate
+ return query
def blobs(self, *, column: Optional[str] = None, key_column: Optional[str]
= None):
from pypaimon.multimodal.blob_store import BlobStore
@@ -168,11 +194,15 @@ class MultimodalTable:
*,
column: Optional[str] = None,
options: Optional[Dict[str, str]] = None,
- pre_filter=None):
- schema = _target_schema(self.raw_table)
+ pre_filter=None,
+ snapshot_id: Optional[int] = None,
+ tag_name: Optional[str] = None):
+ read_table = _time_travel_table(
+ self.raw_table, snapshot_id=snapshot_id, tag_name=tag_name)
+ schema = _target_schema(read_table)
if isinstance(query, str):
return TextQuery(
- self.raw_table,
+ read_table,
text_query=_coerce_full_text_query(query, "search", schema),
pre_filter=pre_filter,
)
@@ -182,7 +212,7 @@ class MultimodalTable:
"search() accepts a single query; use search_vectors() for "
"multiple vectors.")
return VectorQuery(
- self.raw_table,
+ read_table,
vector=vector,
vector_column=column or _infer_vector_column(schema, "column"),
vector_options=options,
@@ -195,12 +225,16 @@ class MultimodalTable:
*,
column: Optional[str] = None,
options: Optional[Dict[str, str]] = None,
- pre_filter=None):
- schema = _target_schema(self.raw_table)
+ pre_filter=None,
+ snapshot_id: Optional[int] = None,
+ tag_name: Optional[str] = None):
+ read_table = _time_travel_table(
+ self.raw_table, snapshot_id=snapshot_id, tag_name=tag_name)
+ schema = _target_schema(read_table)
vectors = _coerce_vectors(vectors)
vector_column = column or _infer_vector_column(schema, "column")
return BatchVectorQuery(
- self.raw_table,
+ read_table,
vectors=vectors,
vector_column=vector_column,
vector_options=options,
@@ -213,8 +247,12 @@ class MultimodalTable:
*,
ranker: str = "rrf",
route_limit: Optional[int] = None,
- pre_filter=None):
- schema = _target_schema(self.raw_table)
+ pre_filter=None,
+ snapshot_id: Optional[int] = None,
+ tag_name: Optional[str] = None):
+ read_table = _time_travel_table(
+ self.raw_table, snapshot_id=snapshot_id, tag_name=tag_name)
+ schema = _target_schema(read_table)
vector_routes, text_routes = _normalize_hybrid_routes(
schema,
routes=routes,
@@ -224,7 +262,7 @@ class MultimodalTable:
raise ValueError(
"search_hybrid requires at least one route.")
return HybridQuery(
- self.raw_table,
+ read_table,
vector_routes=vector_routes,
text_routes=text_routes,
ranker=ranker,
@@ -338,6 +376,37 @@ def _to_arrow_table(data, target_schema=None):
return _align_to_schema(table, target_schema)
+def _coerce_row_ids(row_ids):
+ if row_ids is None or isinstance(row_ids, (str, bytes)):
+ raise ValueError("row_ids must be an iterable of row id integers.")
+ try:
+ iterator = iter(row_ids)
+ except TypeError:
+ raise ValueError("row_ids must be an iterable of row id integers.")
+
+ coerced = []
+ for row_id in iterator:
+ if hasattr(row_id, "as_py"):
+ row_id = row_id.as_py()
+ coerced.append(int(row_id))
+ return coerced
+
+
+def _time_travel_table(table, snapshot_id=None, tag_name=None):
+ if snapshot_id is not None and tag_name is not None:
+ raise ValueError(
+ "snapshot_id and tag_name cannot be set at the same time")
+ if snapshot_id is not None:
+ return table.copy({
+ CoreOptions.SCAN_SNAPSHOT_ID.key(): str(snapshot_id),
+ })
+ if tag_name is not None:
+ return table.copy({
+ CoreOptions.SCAN_TAG_NAME.key(): tag_name,
+ })
+ return table
+
+
def _target_schema(table):
return PyarrowFieldParser.from_paimon_schema(table.table_schema.fields)
diff --git a/paimon-python/pypaimon/tests/multimodal_table_test.py
b/paimon-python/pypaimon/tests/multimodal_table_test.py
index 14c11e0040..f880c888e0 100644
--- a/paimon-python/pypaimon/tests/multimodal_table_test.py
+++ b/paimon-python/pypaimon/tests/multimodal_table_test.py
@@ -513,6 +513,137 @@ class MultimodalTableTest(unittest.TestCase):
self.assertEqual(1, result.num_rows)
self.assertEqual([1], result["id"].to_pylist())
+ def test_scan_with_row_id_returns_system_column(self):
+ users = self.conn.create_table(
+ "users",
+ data=[
+ {"id": 1, "name": "Alice"},
+ {"id": 2, "name": "Bob"},
+ ],
+ schema=_schema({
+ "id": pa.int32(),
+ "name": pa.string(),
+ }),
+ options=_PARQUET_OPTIONS,
+ )
+
+ result = (
+ users.scan()
+ .with_row_id()
+ .select(["id"])
+ .to_arrow()
+ )
+
+ self.assertEqual(["id", "_ROW_ID"], result.column_names)
+ self.assertEqual([1, 2], result["id"].to_pylist())
+ self.assertEqual([0, 1], result["_ROW_ID"].to_pylist())
+
+ def test_take_row_ids_reads_projected_rows(self):
+ docs = self.conn.create_table(
+ "docs",
+ data=[
+ {"id": 1, "content": "alpha"},
+ {"id": 2, "content": "beta"},
+ {"id": 3, "content": "gamma"},
+ ],
+ schema=_schema({
+ "id": pa.int32(),
+ "content": pa.string(),
+ }),
+ options=_PARQUET_OPTIONS,
+ )
+ manifest = {
+ row["id"]: row["_ROW_ID"]
+ for row in docs.scan().select(["id"]).with_row_id().to_list()
+ }
+
+ rows = sorted(
+ docs.take_row_ids([manifest[3], manifest[1]])
+ .select(["id", "content"])
+ .with_row_id()
+ .to_list(),
+ key=lambda row: row["id"],
+ )
+
+ self.assertEqual(
+ [
+ {"id": 1, "content": "alpha", "_ROW_ID": manifest[1]},
+ {"id": 3, "content": "gamma", "_ROW_ID": manifest[3]},
+ ],
+ rows,
+ )
+
+ def test_take_row_ids_accepts_empty_manifest(self):
+ docs = self.conn.create_table(
+ "docs",
+ data=[{"id": 1}],
+ schema=_schema({"id": pa.int32()}),
+ options=_PARQUET_OPTIONS,
+ )
+
+ self.assertEqual([], docs.take_row_ids([]).select(["id"]).to_list())
+
+ def test_take_row_ids_supports_snapshot_and_tag_time_travel(self):
+ docs = self.conn.create_table(
+ "docs",
+ schema=_schema({
+ "id": pa.int32(),
+ "content": pa.string(),
+ "embedding": _vector(3),
+ }),
+ options=_PARQUET_OPTIONS,
+ )
+ docs.add([
+ {"id": 1, "content": "first", "embedding": [1.0, 0.0, 0.0]},
+ ])
+ snapshot_id =
docs.raw_table.snapshot_manager().get_latest_snapshot().id
+ docs.raw_table.create_tag("v1", snapshot_id=snapshot_id)
+ row_id = (
+ docs.scan(snapshot_id=snapshot_id)
+ .select(["id"])
+ .with_row_id()
+ .to_list()[0]["_ROW_ID"]
+ )
+
+ docs.delete(where="id = 1")
+
+ self.assertEqual(
+ [],
+ docs.take_row_ids([row_id]).select(["id"]).to_list(),
+ )
+ self.assertEqual(
+ [{"id": 1, "_ROW_ID": row_id}],
+ docs.take_row_ids([row_id], snapshot_id=snapshot_id)
+ .select(["id"])
+ .with_row_id()
+ .to_list(),
+ )
+ self.assertEqual(
+ [{"id": 1}],
+ docs.take_row_ids([row_id], tag_name="v1")
+ .select(["id"])
+ .to_list(),
+ )
+ self.assertEqual(
+ snapshot_id,
+ docs.search(
+ [1.0, 0.0, 0.0],
+ column="embedding",
+ snapshot_id=snapshot_id,
+ )._table.options.scan_snapshot_id(),
+ )
+
+ def test_time_travel_rejects_snapshot_id_and_tag_name_together(self):
+ docs = self.conn.create_table(
+ "docs",
+ data=[{"id": 1}],
+ schema=_schema({"id": pa.int32()}),
+ options=_PARQUET_OPTIONS,
+ )
+
+ with self.assertRaisesRegex(ValueError, "cannot be set at the same
time"):
+ docs.take_row_ids([0], snapshot_id=1, tag_name="v1")
+
def test_overwrite_replaces_unpartitioned_table(self):
users = self.conn.create_table(
"users",
@@ -1178,6 +1309,45 @@ class MultimodalTableTest(unittest.TestCase):
self.assertEqual([1.0, 0.0, 0.0], calls["vector"])
self.assertEqual([{"id": 1, "embedding": [1.0, 0.0, 0.0]}], result)
+ def test_search_with_row_id_returns_system_column(self):
+ docs = self.conn.create_table(
+ "docs",
+ schema=_schema({
+ "id": pa.int32(),
+ "embedding": _vector(3),
+ }),
+ options=_PARQUET_OPTIONS,
+ )
+ docs.add([{"id": 1, "embedding": [1.0, 0.0, 0.0]}])
+
+ class FakeVectorBuilder:
+ def with_vector_column(self, column):
+ return self
+
+ def with_query_vector(self, vector):
+ return self
+
+ def with_limit(self, limit):
+ return self
+
+ def with_options(self, options):
+ return self
+
+ def execute_local(self):
+ return GlobalIndexResult.from_range(Range(0, 0))
+
+ docs.raw_table.new_vector_search_builder = lambda: FakeVectorBuilder()
+
+ result = (
+ docs.search([1.0, 0.0, 0.0], column="embedding")
+ .select(["id"])
+ .with_row_id()
+ .limit(1)
+ .to_list()
+ )
+
+ self.assertEqual([{"id": 1, "_ROW_ID": 0}], result)
+
def test_search_rejects_batch_vectors(self):
docs = self.conn.create_table(
"docs",
