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 <[email protected]> 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",
