You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2026/07/03 06:14:33 UTC
(paimon) branch codex/multimodal-row-ids created (now 5394d7fbd5)
This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a change to branch codex/multimodal-row-ids
in repository https://gitbox.apache.org/repos/asf/paimon.git
at 5394d7fbd5 [python] Add row-id APIs to multimodal table
This branch includes the following new commits:
new 5394d7fbd5 [python] Add row-id APIs to multimodal table
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(paimon) 01/01: [python] Add row-id APIs to multimodal table
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch codex/multimodal-row-ids
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 5394d7fbd5901745bf2cea6c2c9d13c23da165b4
Author: JingsongLi <ji...@gmail.com>
AuthorDate: Fri Jul 3 14:13:55 2026 +0800
[python] Add row-id APIs to multimodal table
---
docs/docs/pypaimon/multimodal-api.mdx | 109 +++++++++++++++++++++
paimon-python/pypaimon/multimodal/query.py | 22 ++++-
paimon-python/pypaimon/multimodal/table.py | 29 ++++++
.../pypaimon/tests/multimodal_table_test.py | 109 +++++++++++++++++++++
4 files changed, 267 insertions(+), 2 deletions(-)
diff --git a/docs/docs/pypaimon/multimodal-api.mdx b/docs/docs/pypaimon/multimodal-api.mdx
index 83d13ae345..a6761b8440 100644
--- a/docs/docs/pypaimon/multimodal-api.mdx
+++ b/docs/docs/pypaimon/multimodal-api.mdx
@@ -332,6 +332,115 @@ 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()
+```
+
+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)
+ .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")
+ .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())
+ .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):
+ rows = (
+ docs.take_row_ids(row_id_batch)
+ .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.
+- 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..767c5092d4 100644
--- a/paimon-python/pypaimon/multimodal/table.py
+++ b/paimon-python/pypaimon/multimodal/table.py
@@ -35,6 +35,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()
@@ -158,6 +159,18 @@ class MultimodalTable:
def scan(self):
return ScanQuery(self.raw_table)
+ def take_row_ids(self, row_ids):
+ row_ids = _coerce_row_ids(row_ids)
+ read_builder = self.raw_table.new_read_builder().with_projection(
+ [field.name for field in self.raw_table.fields]
+ + [SpecialFields.ROW_ID.name]
+ )
+ predicate = read_builder.new_predicate_builder().is_in(
+ SpecialFields.ROW_ID.name, row_ids)
+ query = ScanQuery(self.raw_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
return BlobStore(self, column=column, key_column=key_column)
@@ -338,6 +351,22 @@ 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 _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..0d6c1c57d2 100644
--- a/paimon-python/pypaimon/tests/multimodal_table_test.py
+++ b/paimon-python/pypaimon/tests/multimodal_table_test.py
@@ -513,6 +513,76 @@ 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_overwrite_replaces_unpartitioned_table(self):
users = self.conn.create_table(
"users",
@@ -1178,6 +1248,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",