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 db7a979832 [python] Add multimodal API for data-evolution tables
(#8421)
db7a979832 is described below
commit db7a9798321d5dd2cdb17ecf101655b03bfe9efb
Author: Jingsong Lee <[email protected]>
AuthorDate: Thu Jul 2 13:08:27 2026 +0800
[python] Add multimodal API for data-evolution tables (#8421)
Add a high-level `pypaimon.multimodal` API for local Python applications
on top of data-evolution tables. The facade wraps a catalog plus default
database and exposes table helpers for ingestion, merge, scan, indexing,
vector search, full-text search, batched vector search, and hybrid
search.
---
docs/docs/pypaimon/multimodal-api.mdx | 428 ++++++++
docs/sidebars.js | 1 +
paimon-python/pypaimon/cli/cli_table.py | 6 +-
.../pypaimon/{cli => common}/where_parser.py | 2 +-
paimon-python/pypaimon/multimodal/__init__.py | 45 +
paimon-python/pypaimon/multimodal/connection.py | 200 ++++
paimon-python/pypaimon/multimodal/query.py | 254 +++++
paimon-python/pypaimon/multimodal/table.py | 682 ++++++++++++
.../pypaimon/tests/multimodal_table_test.py | 1132 ++++++++++++++++++++
paimon-python/pypaimon/tests/where_parser_test.py | 2 +-
10 files changed, 2747 insertions(+), 5 deletions(-)
diff --git a/docs/docs/pypaimon/multimodal-api.mdx
b/docs/docs/pypaimon/multimodal-api.mdx
new file mode 100644
index 0000000000..df91f5d120
--- /dev/null
+++ b/docs/docs/pypaimon/multimodal-api.mdx
@@ -0,0 +1,428 @@
+---
+title: "Multimodal API"
+sidebar_position: 3
+---
+
+import Tabs from '@theme/Tabs';
+import TabItem from '@theme/TabItem';
+
+<!--
+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.
+-->
+
+# Multimodal API
+
+`pypaimon.multimodal` provides a high-level API for local Python applications
+on top of Paimon data-evolution tables. It does not replace the lower-level
+`Catalog`, `Database`, or `Table` APIs. Instead, a multimodal connection wraps
+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:
+
+```shell
+pip install pypaimon[vortex]
+```
+
+## Connect
+
+`connect` creates a multimodal connection from Paimon catalog options. The
+`options` argument is forwarded to `CatalogFactory.create`.
+
+<Tabs groupId="pypaimon-multimodal-connect">
+
+<TabItem value="filesystem" label="Filesystem catalog" default>
+
+```python
+import pypaimon.multimodal as pm
+
+conn = pm.connect(
+ database="default",
+ options={
+ "warehouse": "file:///tmp/warehouse",
+ },
+)
+```
+
+</TabItem>
+
+<TabItem value="rest" label="REST catalog">
+
+```python
+import pypaimon.multimodal as pm
+
+conn = pm.connect(
+ database="default",
+ options={
+ "metastore": "rest",
+ "uri": "http://localhost:8080";,
+ "warehouse": "catalog_name",
+ # Optional authentication and storage options go here.
+ # "token.provider": "...",
+ },
+)
+```
+
+REST catalog options use the same keys as the lower-level PyPaimon catalog API.
+See [REST](../concepts/rest/) for server-side catalog configuration.
+
+</TabItem>
+
+</Tabs>
+
+Table names without a database are resolved against the connection's default
+database. Fully-qualified names such as `"analytics.docs"` are also accepted.
+Use `get_table` to open an existing multimodal table. The table must be a
+data-evolution table without primary keys.
+
+```python
+docs = conn.get_table("docs")
+```
+
+## Create
+
+<Tabs groupId="pypaimon-multimodal-create-table">
+
+<TabItem value="schema" label="Create from schema" default>
+
+```python
+import pyarrow as pa
+
+docs = conn.create_table(
+ "docs",
+ schema=pa.schema([
+ pa.field("id", pa.int64()),
+ pa.field("content", pa.string()),
+ pa.field("embedding", pa.list_(pa.float32(), 3)),
+ pa.field("image", pa.large_binary()),
+ pa.field("category", pa.string()),
+ ]),
+ ignore_if_exists=True,
+)
+```
+
+</TabItem>
+
+<TabItem value="data" label="Create from data">
+
+```python
+docs = conn.create_table(
+ "docs",
+ data=[
+ {
+ "id": 1,
+ "content": "Apache Paimon supports mutable lakehouse tables.",
+ "category": "lake",
+ }
+ ],
+)
+```
+
+</TabItem>
+
+<TabItem value="partitioned" label="Create partitioned table">
+
+```python
+import pyarrow as pa
+
+docs = conn.create_table(
+ "docs_by_day",
+ schema=pa.schema([
+ pa.field("id", pa.int64()),
+ pa.field("content", pa.string()),
+ pa.field("embedding", pa.list_(pa.float32(), 3)),
+ pa.field("category", pa.string()),
+ pa.field("dt", pa.string()),
+ ]),
+ partitioned=["dt"],
+ options={
+ "snapshot.time-retained": "7 d",
+ },
+)
+```
+
+</TabItem>
+
+</Tabs>
+
+Partition keys can be specified with `partitioned`, and table options can be
+specified with `options`. The multimodal API accepts `pyarrow.Schema` objects;
+it does not accept field dictionaries or `pypaimon.Schema`. Use Arrow
fixed-size
+list types for vector columns, and Arrow binary or large-binary types for blob
+columns. When creating from data, pass an explicit schema if you need exact
+vector or blob types.
+
+## Add
+
+`add` accepts `pyarrow.Table`, `pyarrow.RecordBatch`, a list of dictionaries, a
+dictionary of arrays, or a pandas DataFrame. Input columns are aligned and cast
+to the Paimon table schema before writing.
+
+```python
+docs.add([
+ {
+ "id": 2,
+ "content": "Paimon stores mutable lakehouse tables.",
+ "embedding": [0.4, 0.5, 0.6],
+ "category": "lake",
+ }
+])
+```
+
+## Update
+
+`update` modifies rows matched by a SQL-like predicate.
+
+```python
+docs.update(
+ where="id = 2",
+ values={"category": "docs"},
+)
+```
+
+## Merge
+
+Use `merge` for idempotent ingestion. The builder delegates to Paimon's local
+`MERGE INTO` implementation.
+
+`execute` takes the source rows for the merge. It accepts the same input
formats
+as `add`, including `pyarrow.Table`, `pyarrow.RecordBatch`, a list of
+dictionaries, a dictionary of arrays, or a pandas DataFrame. A list of
+dictionaries is convenient for small batches. `when_matched_update()` updates
+the columns present in the source data and leaves omitted target columns
+unchanged. `when_not_matched_insert()` inserts the columns present in the
source
+data and fills omitted target columns with null.
+
+```python
+from pypaimon.multimodal import source_col
+
+docs.merge("id") \
+ .when_matched_update(
+ where="source.category != target.category",
+ ) \
+ .when_not_matched_insert() \
+ .execute([
+ {
+ "id": 2,
+ "content": "Updated text",
+ "embedding": [0.7, 0.8, 0.9],
+ "category": "docs",
+ }
+ ])
+```
+
+Clause predicates use `where`. Refer to source rows with `source.<column>` and
+target rows with `target.<column>`. Install `pypaimon[sql]` when using merge
+clause predicates.
+
+When source and target key names differ, pass a mapping from target column to
+source column:
+
+```python
+docs.merge({"id": "doc_id"}) \
+ .when_matched_update({"category": source_col("new_category")}) \
+ .when_not_matched_insert({
+ "id": source_col("doc_id"),
+ "content": source_col("content"),
+ "category": source_col("new_category"),
+ }) \
+ .execute([
+ {"doc_id": 3, "content": "New doc", "new_category": "search"},
+ ])
+```
+
+## Scan
+
+Use `scan()` for ordinary table reads. `where()` accepts SQL-like predicate
+strings; it does not accept lower-level `Predicate` objects.
+
+```python
+result = (
+ docs.scan()
+ .where("category = 'lake'")
+ .select(["id", "content"])
+ .limit(10)
+ .to_pandas()
+)
+```
+
+## Create Index
+
+Use `create_index` to create the global indexes used by search APIs. The
+`index_type` argument is required; the multimodal API does not choose a default
+index implementation. The `full-text` alias is normalized to
+`tantivy-fulltext`.
+
+```python
+docs.create_index("embedding", index_type="ivf-pq")
+docs.create_index("content", index_type="full-text")
+```
+
+## Search
+
+Use `search` for one vector query or one full-text query.
+
+If the table has exactly one vector column, `column` can be omitted for vector
+search. A string query is shorthand for a full-text match query when the table
+has exactly one text column. To target a specific text column, pass a
+`FullTextQuery` object.
+
+Use `pre_filter` to prune search candidates before ranking. Use `where()` to
+filter the rows read from the search result. Both `pre_filter` and `where()`
+accept SQL-like predicate strings. For full-text search, `pre_filter` must only
+reference partition columns.
+
+```python
+neighbors = (
+ docs.search(
+ [0.1, 0.2, 0.3],
+ column="embedding",
+ pre_filter="category = 'lake'",
+ )
+ .limit(10)
+ .to_arrow()
+)
+
+matches = (
+ docs.search("paimon vector")
+ .limit(10)
+ .to_pandas()
+)
+```
+
+```python
+from pypaimon.globalindex.full_text_query import FullTextQuery
+
+content_query = FullTextQuery.from_dict({
+ "match": {
+ "column": "content",
+ "terms": "paimon vector",
+ "operator": "And",
+ },
+})
+
+matches = docs.search(content_query).limit(10).to_list()
+```
+
+## Search Hybrid
+
+Use `search_hybrid` to combine vector and full-text routes, then rerank the
+merged candidates. Create the indexes required by the routes you use.
+
+Pass route specs to `search_hybrid`. Each route can set its own candidate
+`limit`, `weight`, and vector index `options`. String text routes infer the
+text column when the table has exactly one text column. To target a specific
+text column, pass a `FullTextQuery` object to `pm.text_route`.
+
+`pre_filter` is applied before ranking. It accepts a SQL-like predicate string.
+When a hybrid query has a full-text route, `pre_filter` must only reference
+partition columns.
+
+```python
+# This example assumes the table is partitioned by dt.
+hybrid = (
+ docs.search_hybrid(
+ [
+ pm.vector_route("embedding", query_vector, limit=50),
+ pm.text_route("paimon vector", weight=0.2),
+ ],
+ pre_filter="dt = '2026-07-01'",
+ )
+ .rerank("rrf")
+ .limit(10)
+ .to_list()
+)
+```
+
+For multiple vector fields, add multiple vector routes:
+
+```python
+hybrid = (
+ docs.search_hybrid(
+ [
+ pm.vector_route(
+ "image_embedding",
+ image_vector,
+ weight=0.7,
+ limit=50,
+ options={"nprobe": "8"},
+ ),
+ pm.vector_route(
+ "text_embedding",
+ text_vector,
+ weight=0.3,
+ limit=50,
+ ),
+ pm.text_route("paimon vector", weight=0.2),
+ ],
+ ranker="weighted_score",
+ )
+ .limit(10)
+ .to_list()
+)
+```
+
+Dictionary route specs are also accepted:
+
+```python
+hybrid = docs.search_hybrid(
+ [
+ {
+ "column": "image_embedding",
+ "vector": image_vector,
+ "weight": 0.7,
+ "limit": 50,
+ "options": {"nprobe": "8"},
+ },
+ {
+ "column": "text_embedding",
+ "vector": text_vector,
+ "weight": 0.3,
+ "limit": 50,
+ },
+ ],
+)
+```
+
+Dictionary vector routes also accept `anns_field`, `data`, and `param` aliases.
+
+## Search Vectors
+
+Use `search_vectors` for multiple query vectors against one vector column. It
+returns one result set for each input vector, preserving input order.
+
+```python
+batch_neighbors = (
+ docs.search_vectors(
+ [
+ [0.1, 0.2, 0.3],
+ [0.4, 0.5, 0.6],
+ ],
+ column="embedding",
+ pre_filter="category = 'lake'",
+ )
+ .limit(10)
+ .to_list()
+)
+```
diff --git a/docs/sidebars.js b/docs/sidebars.js
index e0558a3a21..62a0f3ec03 100644
--- a/docs/sidebars.js
+++ b/docs/sidebars.js
@@ -187,6 +187,7 @@ const sidebars = {
},
"items": [
"pypaimon/python-api",
+ "pypaimon/multimodal-api",
"pypaimon/manage-tags",
"pypaimon/ray-data",
"pypaimon/daft",
diff --git a/paimon-python/pypaimon/cli/cli_table.py
b/paimon-python/pypaimon/cli/cli_table.py
index daee51db8c..6b9d633e21 100644
--- a/paimon-python/pypaimon/cli/cli_table.py
+++ b/paimon-python/pypaimon/cli/cli_table.py
@@ -86,7 +86,7 @@ def cmd_table_read(args):
# When both select and where are specified, ensure where-referenced fields
# are included in the projection so the filter can work correctly.
if user_columns and where_clause:
- from pypaimon.cli.where_parser import extract_fields_from_where
+ from pypaimon.common.where_parser import extract_fields_from_where
where_fields = extract_fields_from_where(where_clause,
available_fields)
user_column_set = set(user_columns)
extra_where_columns = [f for f in where_fields if f not in
user_column_set]
@@ -97,7 +97,7 @@ def cmd_table_read(args):
# Apply where filter if specified
if where_clause:
- from pypaimon.cli.where_parser import parse_where_clause
+ from pypaimon.common.where_parser import parse_where_clause
try:
predicate = parse_where_clause(where_clause,
table.table_schema.fields)
if predicate:
@@ -191,7 +191,7 @@ def cmd_table_explain(args):
where_clause = getattr(args, 'where', None)
if where_clause:
- from pypaimon.cli.where_parser import parse_where_clause
+ from pypaimon.common.where_parser import parse_where_clause
try:
predicate = parse_where_clause(where_clause,
table.table_schema.fields)
if predicate:
diff --git a/paimon-python/pypaimon/cli/where_parser.py
b/paimon-python/pypaimon/common/where_parser.py
similarity index 99%
rename from paimon-python/pypaimon/cli/where_parser.py
rename to paimon-python/pypaimon/common/where_parser.py
index 1690f5b16d..3286c81a5c 100644
--- a/paimon-python/pypaimon/cli/where_parser.py
+++ b/paimon-python/pypaimon/common/where_parser.py
@@ -16,7 +16,7 @@
# under the License.
"""
-SQL WHERE clause parser for Paimon CLI.
+SQL WHERE clause parser for PyPaimon.
Parses simple SQL-like WHERE expressions into Predicate objects.
diff --git a/paimon-python/pypaimon/multimodal/__init__.py
b/paimon-python/pypaimon/multimodal/__init__.py
new file mode 100644
index 0000000000..2f957630d4
--- /dev/null
+++ b/paimon-python/pypaimon/multimodal/__init__.py
@@ -0,0 +1,45 @@
+# 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.
+
+"""High-level APIs for mutable multimodal Paimon tables."""
+
+from pypaimon.multimodal.connection import MultimodalConnection, connect
+from pypaimon.multimodal.table import (
+ MultimodalTable,
+ TextRoute,
+ VectorRoute,
+ text_route,
+ vector_route,
+)
+from pypaimon.table.data_evolution_merge_into import (
+ lit,
+ source_col,
+ target_col,
+)
+
+__all__ = [
+ "MultimodalConnection",
+ "MultimodalTable",
+ "TextRoute",
+ "VectorRoute",
+ "connect",
+ "lit",
+ "source_col",
+ "target_col",
+ "text_route",
+ "vector_route",
+]
diff --git a/paimon-python/pypaimon/multimodal/connection.py
b/paimon-python/pypaimon/multimodal/connection.py
new file mode 100644
index 0000000000..b06940e620
--- /dev/null
+++ b/paimon-python/pypaimon/multimodal/connection.py
@@ -0,0 +1,200 @@
+# 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.
+
+from typing import Dict, Iterable, Optional
+
+import pyarrow as pa
+
+from pypaimon import CatalogFactory, Schema as PaimonSchema
+from pypaimon.catalog.catalog import Catalog
+from pypaimon.catalog.catalog_exception import (
+ DatabaseNotExistException,
+ TableAlreadyExistException,
+ TableNotExistException,
+)
+from pypaimon.multimodal.table import MultimodalTable, _to_arrow_table
+
+_DEFAULT_OPTIONS = {
+ "row-tracking.enabled": "true",
+ "data-evolution.enabled": "true",
+ "deletion-vectors.enabled": "true",
+ "file.format": "vortex",
+ "global-index.search-mode": "full",
+ "vector.file.format": "vortex",
+}
+
+_DEFAULT_DATABASE = Catalog.DEFAULT_DATABASE
+
+
+def connect(
+ *,
+ database: str = _DEFAULT_DATABASE,
+ options: Optional[Dict[str, str]] = None):
+ """Connect to a Paimon catalog through the multimodal facade."""
+ return MultimodalConnection(
+ _resolve_catalog(options),
+ database=database,
+ )
+
+
+class MultimodalConnection:
+ """High-level entry point backed by a Paimon catalog and database."""
+
+ def __init__(self, catalog, database: str = _DEFAULT_DATABASE):
+ if not database:
+ raise ValueError("database is required.")
+ self.catalog = catalog
+ self.database = database
+
+ def create_table(
+ self,
+ name: str,
+ data=None,
+ schema=None,
+ options: Optional[Dict[str, str]] = None,
+ partitioned: Optional[Iterable[str]] = None,
+ ignore_if_exists: bool = False):
+ """Create a multimodal table and optionally add initial data."""
+ identifier = self._identifier(name)
+ already_exists = _table_exists(self.catalog, identifier)
+ paimon_schema = _to_paimon_schema(schema, data, options, partitioned)
+
+ self._create_database_for(identifier)
+ try:
+ self.catalog.create_table(
+ identifier, paimon_schema, ignore_if_exists)
+ except TableAlreadyExistException:
+ if not ignore_if_exists:
+ raise
+
+ table = self.get_table(name)
+ if data is not None and not already_exists:
+ table.add(data)
+ return table
+
+ def get_table(self, name: str):
+ identifier = self._identifier(name)
+ raw_table = self.catalog.get_table(identifier)
+ _validate_multimodal_table(raw_table, identifier)
+ return MultimodalTable(
+ self.catalog,
+ identifier,
+ raw_table,
+ )
+
+ def drop_table(self, name: str, ignore_if_not_exists: bool = False):
+ self.catalog.drop_table(
+ self._identifier(name),
+ ignore_if_not_exists=ignore_if_not_exists,
+ )
+
+ def _identifier(self, name: str) -> str:
+ return _resolve_identifier(name, self.database)
+
+ def _create_database_for(self, identifier: str):
+ database_name = _database_name(identifier)
+ if database_name is not None:
+ self.catalog.create_database(database_name, ignore_if_exists=True)
+
+
+def _resolve_catalog(options):
+ resolved_options = {}
+ if options:
+ resolved_options.update(options)
+ if not resolved_options:
+ raise ValueError("options is required.")
+ return CatalogFactory.create(resolved_options)
+
+
+def _table_exists(catalog, identifier: str) -> bool:
+ try:
+ catalog.get_table(identifier)
+ return True
+ except (DatabaseNotExistException, TableNotExistException):
+ return False
+
+
+def _validate_multimodal_table(table, identifier: str):
+ table_schema = table.table_schema
+ options = table_schema.options
+ if str(options.get("data-evolution.enabled", "false")).lower() != "true":
+ raise ValueError(
+ "Table %s is not a data-evolution table; "
+ "data-evolution.enabled must be true." % identifier)
+ if table_schema.primary_keys:
+ raise ValueError(
+ "Table %s has primary keys %s; multimodal tables must not have "
+ "primary keys." % (identifier, table_schema.primary_keys))
+
+
+def _resolve_identifier(name: str, database: str) -> str:
+ if not name:
+ raise ValueError("table name is required.")
+ if "." in name:
+ return name
+ return "%s.%s" % (database, name)
+
+
+def _to_paimon_schema(schema, data, options, partitioned):
+ if schema is None:
+ if data is None:
+ raise ValueError("schema or data is required.")
+ pa_schema, inferred_options = _infer_pyarrow_schema(data)
+ else:
+ pa_schema = _to_pyarrow_schema(schema)
+ inferred_options = {}
+ return PaimonSchema.from_pyarrow_schema(
+ pa_schema,
+ partition_keys=_normalize_partitioned(partitioned),
+ options=_merge_options(inferred_options, options),
+ )
+
+
+def _merge_options(*option_groups):
+ table_options = dict(_DEFAULT_OPTIONS)
+ for options in option_groups:
+ if options:
+ table_options.update({str(k): str(v) for k, v in options.items()})
+ return table_options
+
+
+def _normalize_partitioned(partitioned):
+ if partitioned is None:
+ return []
+ if isinstance(partitioned, str):
+ return [partitioned]
+ return list(partitioned)
+
+
+def _to_pyarrow_schema(schema):
+ if isinstance(schema, pa.Schema):
+ return schema
+ raise ValueError("schema must be a pyarrow.Schema.")
+
+
+def _infer_pyarrow_schema(data):
+ table = _to_arrow_table(data)
+ if table.num_columns == 0:
+ raise ValueError("Cannot infer schema from empty data.")
+ return table.schema, {}
+
+
+def _database_name(identifier: str):
+ parts = identifier.split(".")
+ if len(parts) == 2 and parts[0]:
+ return parts[0]
+ return None
diff --git a/paimon-python/pypaimon/multimodal/query.py
b/paimon-python/pypaimon/multimodal/query.py
new file mode 100644
index 0000000000..be0d4fb065
--- /dev/null
+++ b/paimon-python/pypaimon/multimodal/query.py
@@ -0,0 +1,254 @@
+# 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.
+
+from typing import Callable, List, Optional
+
+from pypaimon.common.where_parser import parse_where_clause
+
+
+class ScanQuery:
+ """Chainable scan wrapper for MultimodalTable."""
+
+ def __init__(
+ self,
+ table,
+ result_factory: Optional[Callable] = None):
+ self._table = table
+ self._predicate = None
+ self._projection = None
+ self._limit = None
+ self._result_factory = result_factory
+
+ def where(self, predicate):
+ predicate = self._coerce_predicate(predicate, "where()")
+ if predicate is not None:
+ self._predicate = self._and_predicate(self._predicate, predicate)
+ return self
+
+ def select(self, columns):
+ if isinstance(columns, str):
+ columns = [columns]
+ self._projection = list(columns)
+ return self
+
+ def limit(self, limit: int):
+ self._limit = limit
+ return self
+
+ def to_arrow(self):
+ if self._result_factory is not None:
+ return self._read_global_index_result(self._result_factory(self))
+
+ read_builder = self._configured_read_builder()
+ scan = read_builder.new_scan()
+ plan = scan.plan()
+ return read_builder.new_read().to_arrow(plan.splits())
+
+ def _configured_read_builder(self):
+ 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)
+ if self._limit is not None:
+ read_builder = read_builder.with_limit(self._limit)
+ return read_builder
+
+ def _read_global_index_result(self, result):
+ read_builder = self._configured_read_builder()
+ scan = read_builder.new_scan().with_global_index_result(result)
+ plan = scan.plan()
+ return read_builder.new_read().to_arrow(plan.splits())
+
+ def to_pandas(self):
+ return self.to_arrow().to_pandas()
+
+ def to_list(self) -> List[dict]:
+ return self.to_arrow().to_pylist()
+
+ def _coerce_predicate(self, predicate, method):
+ if predicate is None:
+ return None
+ if isinstance(predicate, str):
+ return parse_where_clause(predicate, self._table.fields)
+ raise ValueError("%s expects a SQL-like string." % method)
+
+ def _and_predicate(self, left, right):
+ if left is None:
+ return right
+ from pypaimon.common.predicate_builder import PredicateBuilder
+ return PredicateBuilder.and_predicates([left, right])
+
+
+class _PreFilterQuery(ScanQuery):
+
+ def __init__(
+ self,
+ table,
+ result_factory: Optional[Callable] = None,
+ pre_filter=None):
+ self._pre_filter = None
+ super().__init__(table, result_factory=result_factory)
+ if pre_filter is not None:
+ self.pre_filter(pre_filter)
+
+ def pre_filter(self, predicate):
+ predicate = self._coerce_predicate(predicate, "pre_filter()")
+ if predicate is not None:
+ self._pre_filter = self._and_predicate(self._pre_filter, predicate)
+ return self
+
+
+class VectorQuery(_PreFilterQuery):
+ """Chainable query wrapper for vector global-index search."""
+
+ def __init__(
+ self,
+ table,
+ vector,
+ vector_column,
+ vector_options=None,
+ pre_filter=None):
+ self._vector = vector
+ self._vector_column = vector_column
+ self._vector_options = dict(vector_options or {})
+ super().__init__(
+ table, result_factory=self._execute_vector, pre_filter=pre_filter)
+
+ def _execute_vector(self, query):
+ limit = query._limit if query._limit is not None else 10
+ builder = (
+ self._table.new_vector_search_builder()
+ .with_vector_column(self._vector_column)
+ .with_query_vector(self._vector)
+ .with_limit(limit)
+ .with_options(self._vector_options)
+ )
+ if query._pre_filter is not None:
+ builder = builder.with_filter(query._pre_filter)
+ return builder.execute_local()
+
+
+class TextQuery(_PreFilterQuery):
+ """Chainable query wrapper for full-text global-index search."""
+
+ def __init__(self, table, text_query, pre_filter=None):
+ self._text_query = text_query
+ super().__init__(
+ table, result_factory=self._execute_fts, pre_filter=pre_filter)
+
+ def _execute_fts(self, query):
+ limit = query._limit if query._limit is not None else 10
+ builder = (
+ self._table.new_full_text_search_builder()
+ .with_query(self._text_query)
+ .with_limit(limit)
+ )
+ if query._pre_filter is not None:
+ builder = builder.with_partition_filter(query._pre_filter)
+ return builder.execute_local()
+
+
+class HybridQuery(_PreFilterQuery):
+ """Chainable query wrapper for hybrid global-index search."""
+
+ def __init__(
+ self,
+ table,
+ vector_routes=None,
+ text_routes=None,
+ ranker="rrf",
+ route_limit=None,
+ pre_filter=None):
+ self._vector_routes = list(vector_routes or [])
+ self._text_routes = list(text_routes or [])
+ self._ranker = ranker
+ self._route_limit = route_limit
+ super().__init__(
+ table, result_factory=self._execute_hybrid, pre_filter=pre_filter)
+
+ def rerank(self, ranker):
+ self._ranker = ranker
+ return self
+
+ def _execute_hybrid(self, query):
+ final_limit = query._limit if query._limit is not None else 10
+ route_limit = self._route_limit or final_limit
+ builder = (
+ self._table.new_hybrid_search_builder()
+ .with_limit(final_limit)
+ .with_ranker(self._ranker)
+ )
+ for route in self._vector_routes:
+ builder = builder.add_vector_route(
+ route["column"],
+ route["vector"],
+ limit=route.get("limit") or route_limit,
+ weight=route["weight"],
+ options=route["options"],
+ )
+ for route in self._text_routes:
+ builder = builder.add_full_text_route(
+ route["query"].to_json(),
+ limit=route.get("limit") or route_limit,
+ weight=route["weight"],
+ options=route["options"],
+ )
+ if query._pre_filter is not None:
+ builder = builder.with_filter(query._pre_filter)
+ return builder.execute_local()
+
+
+class BatchVectorQuery(_PreFilterQuery):
+ """Chainable query wrapper for batch vector global-index search."""
+
+ def __init__(
+ self,
+ table,
+ vectors,
+ vector_column,
+ vector_options=None,
+ pre_filter=None):
+ self._vectors = vectors
+ self._vector_column = vector_column
+ self._vector_options = dict(vector_options or {})
+ super().__init__(table, pre_filter=pre_filter)
+
+ def to_arrow(self):
+ return [
+ self._read_global_index_result(result)
+ for result in self._execute_batch_vector(self)
+ ]
+
+ def to_pandas(self):
+ return [table.to_pandas() for table in self.to_arrow()]
+
+ def to_list(self) -> List[List[dict]]:
+ return [table.to_pylist() for table in self.to_arrow()]
+
+ def _execute_batch_vector(self, query):
+ limit = query._limit if query._limit is not None else 10
+ builder = (
+ self._table.new_batch_vector_search_builder()
+ .with_vector_column(self._vector_column)
+ .with_query_vectors(self._vectors)
+ .with_limit(limit)
+ .with_options(self._vector_options)
+ )
+ if query._pre_filter is not None:
+ builder = builder.with_filter(query._pre_filter)
+ return builder.execute_batch_local()
diff --git a/paimon-python/pypaimon/multimodal/table.py
b/paimon-python/pypaimon/multimodal/table.py
new file mode 100644
index 0000000000..e88d9af8c0
--- /dev/null
+++ b/paimon-python/pypaimon/multimodal/table.py
@@ -0,0 +1,682 @@
+# 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 re
+from dataclasses import dataclass
+from typing import Dict, Mapping, Optional, Sequence
+
+import pyarrow as pa
+
+from pypaimon.globalindex.full_text_query import FullTextQuery
+from pypaimon.multimodal.query import (
+ BatchVectorQuery,
+ HybridQuery,
+ ScanQuery,
+ TextQuery,
+ VectorQuery,
+)
+from pypaimon.schema.data_types import PyarrowFieldParser
+from pypaimon.table.data_evolution_merge_into import (
+ WhenMatched,
+ WhenNotMatched,
+ source_col as _source_col,
+)
+
+
+_ALL_SOURCE_COLUMNS = object()
+_MERGE_REF_PATTERN = re.compile(r'\b(source|target)\.(\w+)\b')
+_STRING_LITERAL_PATTERN = re.compile(r"'(?:[^']|'')*'")
+
+
+@dataclass(frozen=True)
+class VectorRoute:
+ """Vector route spec for hybrid search."""
+
+ column: Optional[str]
+ vector: object
+ weight: float = 1.0
+ limit: Optional[int] = None
+ options: Optional[Dict[str, str]] = None
+
+
+@dataclass(frozen=True)
+class TextRoute:
+ """Full-text route spec for hybrid search."""
+
+ query: object
+ weight: float = 1.0
+ limit: Optional[int] = None
+ options: Optional[Dict[str, str]] = None
+
+
+def vector_route(column, vector, *, weight: float = 1.0,
+ limit: Optional[int] = None,
+ options: Optional[Dict[str, str]] = None) -> VectorRoute:
+ return VectorRoute(
+ column=column,
+ vector=vector,
+ weight=weight,
+ limit=limit,
+ options=dict(options or {}),
+ )
+
+
+def text_route(query, *, weight: float = 1.0,
+ limit: Optional[int] = None,
+ options: Optional[Dict[str, str]] = None) -> TextRoute:
+ return TextRoute(
+ query=query,
+ weight=weight,
+ limit=limit,
+ options=dict(options or {}),
+ )
+
+
+class MultimodalTable:
+ """High-level table facade for mutable multimodal application data."""
+
+ def __init__(self, catalog, identifier: str, raw_table):
+ self.catalog = catalog
+ self.identifier = identifier
+ self.name = identifier
+ self.raw_table = raw_table
+ self.table = raw_table
+
+ def add(self, data):
+ arrow_table = _to_arrow_table(data, _target_schema(self.raw_table))
+ write_builder = self.raw_table.new_batch_write_builder()
+ table_write = write_builder.new_write()
+ table_commit = write_builder.new_commit()
+ try:
+ table_write.write_arrow(arrow_table)
+ table_commit.commit(table_write.prepare_commit())
+ finally:
+ table_write.close()
+ table_commit.close()
+ return self
+
+ def update(self, where, values):
+ query = self.scan().where(where)
+ predicate = query._predicate
+ 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.update_by_predicate(predicate, values)
+ table_commit.commit(messages)
+ finally:
+ table_commit.close()
+ return self
+
+ def merge(self, on):
+ return _MergeBuilder(self, on)
+
+ def scan(self):
+ return ScanQuery(self.raw_table)
+
+ def search(
+ self,
+ query,
+ *,
+ column: Optional[str] = None,
+ options: Optional[Dict[str, str]] = None,
+ pre_filter=None):
+ schema = _target_schema(self.raw_table)
+ if isinstance(query, str):
+ return TextQuery(
+ self.raw_table,
+ text_query=_coerce_full_text_query(query, "search", schema),
+ pre_filter=pre_filter,
+ )
+ vector = _coerce_vector(query, "search")
+ if _looks_like_batch_vectors(vector):
+ raise ValueError(
+ "search() accepts a single query; use search_vectors() for "
+ "multiple vectors.")
+ return VectorQuery(
+ self.raw_table,
+ vector=vector,
+ vector_column=column or _infer_vector_column(schema, "column"),
+ vector_options=options,
+ pre_filter=pre_filter,
+ )
+
+ def search_vectors(
+ self,
+ vectors,
+ *,
+ column: Optional[str] = None,
+ options: Optional[Dict[str, str]] = None,
+ pre_filter=None):
+ schema = _target_schema(self.raw_table)
+ vectors = _coerce_vectors(vectors)
+ vector_column = column or _infer_vector_column(schema, "column")
+ return BatchVectorQuery(
+ self.raw_table,
+ vectors=vectors,
+ vector_column=vector_column,
+ vector_options=options,
+ pre_filter=pre_filter,
+ )
+
+ def search_hybrid(
+ self,
+ routes,
+ *,
+ ranker: str = "rrf",
+ route_limit: Optional[int] = None,
+ pre_filter=None):
+ schema = _target_schema(self.raw_table)
+ vector_routes, text_routes = _normalize_hybrid_routes(
+ schema,
+ routes=routes,
+ method="search_hybrid",
+ )
+ if not vector_routes and not text_routes:
+ raise ValueError(
+ "search_hybrid requires at least one route.")
+ return HybridQuery(
+ self.raw_table,
+ vector_routes=vector_routes,
+ text_routes=text_routes,
+ ranker=ranker,
+ route_limit=route_limit,
+ pre_filter=pre_filter,
+ )
+
+ def create_index(self, column, index_type, options=None):
+ return self.raw_table.create_global_index(
+ column, index_type=_normalize_index_type(index_type),
options=options)
+
+ def _merge(
+ self,
+ source,
+ on,
+ when_matched: Sequence[WhenMatched],
+ when_not_matched: Sequence[WhenNotMatched],
+ operation: str):
+ target_cols, source_cols = _normalize_merge_on(on, operation)
+ target_schema = _target_schema(self.raw_table)
+ source_table = _to_arrow_table(source)
+ _validate_merge_on(
+ target_schema, source_table, target_cols, source_cols, operation)
+ source_table = _cast_merge_on_columns(
+ source_table, target_schema, target_cols, source_cols)
+ when_matched, when_not_matched = _resolve_merge_clauses(
+ when_matched,
+ when_not_matched,
+ target_schema,
+ source_table,
+ dict(zip(target_cols, source_cols)),
+ )
+
+ 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.merge_into(
+ source_table,
+ on=dict(zip(target_cols, source_cols)),
+ when_matched=list(when_matched),
+ when_not_matched=list(when_not_matched),
+ )
+ table_commit.commit(messages)
+ finally:
+ table_commit.close()
+ return self
+
+
+class _MergeBuilder:
+ """Builder for idempotent merge writes."""
+
+ def __init__(self, table: MultimodalTable, on):
+ self._table = table
+ self._on = on
+ self._when_matched = []
+ self._when_not_matched = []
+
+ def when_matched_update(self, values=None, where: Optional[str] = None):
+ self._when_matched.append(
+ WhenMatched(
+ update=_ALL_SOURCE_COLUMNS if values is None else values,
+ condition=_normalize_merge_where(where),
+ ))
+ return self
+
+ def when_not_matched_insert(self, values=None, where: Optional[str] =
None):
+ self._when_not_matched.append(
+ WhenNotMatched(
+ insert=_ALL_SOURCE_COLUMNS if values is None else values,
+ condition=_normalize_merge_where(where),
+ ))
+ return self
+
+ def execute(self, data):
+ if not self._when_matched and not self._when_not_matched:
+ raise ValueError(
+ "merge requires at least one matched or not-matched clause.")
+ return self._table._merge(
+ data,
+ self._on,
+ when_matched=self._when_matched,
+ when_not_matched=self._when_not_matched,
+ operation="merge",
+ )
+
+
+def _to_arrow_table(data, target_schema=None):
+ if isinstance(data, pa.Table):
+ table = data
+ elif isinstance(data, pa.RecordBatch):
+ table = pa.Table.from_batches([data])
+ elif isinstance(data, list):
+ table = pa.Table.from_pylist(data)
+ elif isinstance(data, dict):
+ table = pa.Table.from_pydict(data)
+ elif hasattr(data, "__dataframe__") or
data.__class__.__module__.startswith("pandas"):
+ table = pa.Table.from_pandas(data, preserve_index=False)
+ else:
+ raise ValueError("Unsupported multimodal data type: %r" % type(data))
+
+ if target_schema is None:
+ return table
+ return _align_to_schema(table, target_schema)
+
+
+def _target_schema(table):
+ return PyarrowFieldParser.from_paimon_schema(table.table_schema.fields)
+
+
+def _align_to_schema(
+ table: pa.Table,
+ schema: pa.Schema,
+ column_mapping: Optional[Dict[str, str]] = None) -> pa.Table:
+ column_mapping = column_mapping or {}
+ arrays = []
+ for field in schema:
+ source_name = column_mapping.get(field.name, field.name)
+ if source_name in table.column_names:
+ array = table[source_name]
+ if array.type != field.type:
+ array = array.cast(field.type)
+ else:
+ array = pa.nulls(table.num_rows, type=field.type)
+ arrays.append(array)
+ return pa.Table.from_arrays(arrays, schema=schema)
+
+
+def _normalize_merge_on(on, operation: str):
+ if isinstance(on, Mapping):
+ target_cols = list(on.keys())
+ source_cols = list(on.values())
+ elif isinstance(on, str):
+ target_cols = [on]
+ source_cols = [on]
+ else:
+ target_cols = list(on)
+ source_cols = list(on)
+ if not target_cols:
+ raise ValueError("%s requires at least one ON column." % operation)
+ if len(target_cols) != len(source_cols):
+ raise ValueError(
+ "%s ON target/source column counts do not match." % operation)
+ return target_cols, source_cols
+
+
+def _validate_merge_on(
+ target_schema,
+ source_table,
+ target_cols,
+ source_cols,
+ operation: str):
+ target_names = set(target_schema.names)
+ source_names = set(source_table.column_names)
+ missing_targets = [col for col in target_cols if col not in target_names]
+ if missing_targets:
+ raise ValueError(
+ "%s ON target columns are not in table schema: %s"
+ % (operation, missing_targets))
+ missing_sources = [col for col in source_cols if col not in source_names]
+ if missing_sources:
+ raise ValueError(
+ "%s ON source columns are not in source data: %s"
+ % (operation, missing_sources))
+
+
+def _cast_merge_on_columns(source_table, target_schema, target_cols,
source_cols):
+ result = source_table
+ for target_col, source_col_name in zip(target_cols, source_cols):
+ target_type = target_schema.field(target_col).type
+ index = result.schema.get_field_index(source_col_name)
+ if index < 0:
+ continue
+ array = result[source_col_name]
+ if array.type == target_type:
+ continue
+ source_field = result.schema.field(index)
+ result = result.set_column(
+ index,
+ pa.field(
+ source_col_name,
+ target_type,
+ nullable=source_field.nullable,
+ metadata=source_field.metadata,
+ ),
+ array.cast(target_type),
+ )
+ return result
+
+
+def _resolve_merge_clauses(
+ when_matched,
+ when_not_matched,
+ target_schema,
+ source_table,
+ on_map):
+ target_names = list(target_schema.names)
+ source_names = set(source_table.column_names)
+ return (
+ [
+ WhenMatched(
+ update=_resolve_merge_set_spec(
+ clause.update, target_names, source_names, on_map),
+ condition=clause.condition,
+ )
+ for clause in when_matched
+ ],
+ [
+ WhenNotMatched(
+ insert=_resolve_merge_set_spec(
+ clause.insert, target_names, source_names, on_map),
+ condition=clause.condition,
+ )
+ for clause in when_not_matched
+ ],
+ )
+
+
+def _resolve_merge_set_spec(spec, target_names, source_names, on_map):
+ if spec is not _ALL_SOURCE_COLUMNS:
+ return spec
+ return {
+ target_name: _source_col(source_name)
+ for target_name in target_names
+ for source_name in [on_map.get(target_name, target_name)]
+ if source_name in source_names
+ }
+
+
+def _normalize_merge_where(where):
+ if where is None:
+ return None
+ if not isinstance(where, str):
+ raise ValueError("merge where must be a string.")
+ parts, last = [], 0
+ for match in _STRING_LITERAL_PATTERN.finditer(where):
+ parts.append(_rewrite_merge_refs(where[last:match.start()]))
+ parts.append(match.group())
+ last = match.end()
+ parts.append(_rewrite_merge_refs(where[last:]))
+ return ''.join(parts)
+
+
+def _rewrite_merge_refs(text):
+ return _MERGE_REF_PATTERN.sub(
+ lambda m: "%s.%s" % (
+ "s" if m.group(1) == "source" else "t",
+ m.group(2),
+ ),
+ text,
+ )
+
+
+def _normalize_index_type(index_type):
+ if not isinstance(index_type, str):
+ return index_type
+ normalized = index_type.strip().lower().replace("_", "-")
+ if normalized in ("full-text", "fulltext"):
+ return "tantivy-fulltext"
+ return index_type
+
+
+def _coerce_vector(value, method):
+ if value is None or isinstance(value, str):
+ raise ValueError("%s requires a vector query." % method)
+ if hasattr(value, "tolist"):
+ value = value.tolist()
+ try:
+ return list(value)
+ except TypeError as e:
+ raise ValueError("%s requires a vector query." % method) from e
+
+
+def _coerce_vectors(values):
+ method = "search_vectors"
+ if values is None or isinstance(values, str):
+ raise ValueError("%s requires query vectors." % method)
+ if hasattr(values, "tolist"):
+ values = values.tolist()
+ try:
+ vectors = list(values)
+ except TypeError as e:
+ raise ValueError("%s requires query vectors." % method) from e
+ if not vectors:
+ raise ValueError("%s requires at least one vector query." % method)
+ return [_coerce_vector(vector, method) for vector in vectors]
+
+
+def _looks_like_batch_vectors(value):
+ if not value:
+ return False
+ first = value[0]
+ return not isinstance(first, (int, float)) and hasattr(first, "__iter__")
+
+
+def _normalize_hybrid_routes(
+ schema,
+ routes,
+ method):
+ vector_routes, text_routes = [], []
+ for route in _route_specs(routes):
+ if _is_text_route_spec(route):
+ text_routes.append(_normalize_text_route(route, method, schema))
+ else:
+ vector_routes.append(
+ _normalize_vector_route(route, method, schema))
+ return vector_routes, text_routes
+
+
+def _route_specs(routes):
+ if routes is None:
+ return []
+ if isinstance(routes, (VectorRoute, TextRoute, Mapping)):
+ return [routes]
+ if _is_column_vector_pair(routes):
+ return [routes]
+ return list(routes)
+
+
+def _is_text_route_spec(route):
+ if isinstance(route, TextRoute):
+ return True
+ if isinstance(route, VectorRoute):
+ return False
+ if isinstance(route, Mapping):
+ route_type = route.get("type") or route.get("route_type")
+ if route_type is not None:
+ return str(route_type).replace("-", "_").lower() in (
+ "text", "full_text", "fulltext")
+ return (
+ "query" in route
+ or "text" in route
+ or "terms" in route
+ or "full_text_query" in route
+ ) and not _has_vector_query(route)
+ return False
+
+
+def _normalize_vector_route(route, method, schema):
+ if isinstance(route, VectorRoute):
+ column = route.column
+ vector = route.vector
+ weight = route.weight
+ limit = route.limit
+ options = route.options
+ elif isinstance(route, Mapping):
+ column = (
+ route.get("column")
+ or route.get("vector_column")
+ or route.get("field")
+ or route.get("anns_field")
+ )
+ vector = _mapping_vector(route)
+ weight = route.get("weight", 1.0)
+ limit = route.get("limit")
+ options = _mapping_options(route)
+ elif _is_column_vector_pair(route):
+ column, vector = route
+ weight = 1.0
+ limit = None
+ options = None
+ else:
+ raise ValueError(
+ "%s vector routes require a route spec with a column and vector."
+ % method)
+ if not column:
+ raise ValueError("%s vector routes require a column." % method)
+ return {
+ "column": column,
+ "vector": _coerce_vector(vector, method),
+ "weight": weight,
+ "limit": limit,
+ "options": dict(options or {}),
+ }
+
+
+def _normalize_text_route(route, method, schema):
+ if isinstance(route, TextRoute):
+ query = route.query
+ weight = route.weight
+ limit = route.limit
+ options = route.options
+ elif isinstance(route, Mapping):
+ column = route.get("column") or route.get("text_column") or
route.get("field")
+ if column is not None:
+ raise ValueError(
+ "%s text routes do not accept a column; use a full-text "
+ "query DSL to target a column." % method)
+ query = (
+ route.get("query")
+ or route.get("text")
+ or route.get("terms")
+ or route.get("full_text_query")
+ )
+ weight = route.get("weight", 1.0)
+ limit = route.get("limit")
+ options = route.get("options")
+ else:
+ raise ValueError(
+ "%s text routes require a route spec with a query."
+ % method)
+ return {
+ "query": _coerce_full_text_query(query, method, schema),
+ "weight": weight,
+ "limit": limit,
+ "options": dict(options or {}),
+ }
+
+
+def _has_vector_query(route):
+ return any(key in route for key in ("vector", "query_vector", "data"))
+
+
+def _mapping_vector(route):
+ if "vector" in route:
+ return route["vector"]
+ if "query_vector" in route:
+ return route["query_vector"]
+ if "data" in route:
+ data = route["data"]
+ if hasattr(data, "tolist"):
+ data = data.tolist()
+ values = list(data)
+ if len(values) == 1 and hasattr(values[0], "__iter__"):
+ return values[0]
+ return values
+ raise ValueError("Vector route requires vector, query_vector, or data.")
+
+
+def _mapping_options(route):
+ if "options" in route:
+ return route["options"]
+ if "param" in route:
+ param = route["param"]
+ if isinstance(param, Mapping) and isinstance(param.get("params"),
Mapping):
+ return param["params"]
+ return param
+ if "params" in route:
+ return route["params"]
+ return None
+
+
+def _is_column_vector_pair(route):
+ return (
+ isinstance(route, (list, tuple))
+ and len(route) == 2
+ and isinstance(route[0], str)
+ )
+
+
+def _coerce_full_text_query(query, method, schema):
+ if isinstance(query, FullTextQuery):
+ return query
+ if isinstance(query, str):
+ return FullTextQuery.from_dict({
+ "match": {
+ "column": _infer_text_column(schema, "text"),
+ "terms": query,
+ },
+ })
+ raise ValueError("%s requires a text string." % method)
+
+
+def _infer_vector_column(schema: pa.Schema, parameter: str = "vector_column"):
+ columns = [
+ field.name
+ for field in schema
+ if pa.types.is_fixed_size_list(field.type)
+ ]
+ return _infer_single_column(columns, "vector", parameter)
+
+
+def _infer_text_column(schema: pa.Schema, parameter: str = "text_column"):
+ columns = [
+ field.name
+ for field in schema
+ if pa.types.is_string(field.type) or
pa.types.is_large_string(field.type)
+ ]
+ return _infer_single_column(columns, "text", parameter)
+
+
+def _infer_single_column(columns, kind, parameter):
+ if len(columns) == 1:
+ return columns[0]
+ if not columns:
+ raise ValueError("No %s column found; pass %s." % (kind, parameter))
+ raise ValueError(
+ "Multiple %s columns found %s; pass %s."
+ % (kind, columns, parameter))
diff --git a/paimon-python/pypaimon/tests/multimodal_table_test.py
b/paimon-python/pypaimon/tests/multimodal_table_test.py
new file mode 100644
index 0000000000..1afe8debea
--- /dev/null
+++ b/paimon-python/pypaimon/tests/multimodal_table_test.py
@@ -0,0 +1,1132 @@
+# 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 json
+import os
+import shutil
+import tempfile
+import unittest
+
+import pyarrow as pa
+import pypaimon.multimodal as pmm
+from pypaimon.multimodal import source_col
+from pypaimon.common.predicate_builder import PredicateBuilder
+from pypaimon import Schema as PaimonSchema
+from pypaimon.globalindex.global_index_result import GlobalIndexResult
+from pypaimon.utils.range import Range
+
+
+_PARQUET_OPTIONS = {
+ "row-tracking.enabled": "true",
+ "data-evolution.enabled": "true",
+ "deletion-vectors.enabled": "true",
+ "file.format": "parquet",
+ "vector.file.format": "parquet",
+}
+
+
+def _schema(fields):
+ return pa.schema([
+ pa.field(name, field_type)
+ for name, field_type in fields.items()
+ ])
+
+
+def _vector(dim):
+ return pa.list_(pa.float32(), dim)
+
+
+def _raw_schema(options=None, primary_keys=None):
+ return PaimonSchema.from_pyarrow_schema(
+ _schema({"id": pa.int32(), "name": pa.string()}),
+ primary_keys=list(primary_keys or []),
+ options=dict(options or {"file.format": "parquet"}),
+ )
+
+
+class MultimodalTableTest(unittest.TestCase):
+
+ def setUp(self):
+ self.temp_dir = tempfile.mkdtemp(prefix="pypaimon_mm_")
+ self.warehouse = os.path.join(self.temp_dir, "warehouse")
+ self.conn = pmm.connect(options={"warehouse": self.warehouse})
+
+ def tearDown(self):
+ shutil.rmtree(self.temp_dir, ignore_errors=True)
+
+ def test_connect_accepts_options(self):
+ conn = pmm.connect(
+ database="analytics",
+ options={
+ "warehouse": os.path.join(self.temp_dir, "warehouse_options"),
+ },
+ )
+
+ table = conn.create_table(
+ "docs",
+ schema=_schema({"id": pa.int32()}),
+ options=_PARQUET_OPTIONS,
+ )
+
+ self.assertEqual("analytics.docs", table.identifier)
+
+ def test_connect_rejects_positional_warehouse(self):
+ with self.assertRaises(TypeError):
+ pmm.connect(self.warehouse)
+
+ def test_create_table_defaults_data_evolution_options(self):
+ table = self.conn.create_table(
+ "docs",
+ schema=_schema({
+ "id": pa.int32(),
+ "content": pa.string(),
+ "embedding": _vector(3),
+ "payload": pa.large_binary(),
+ }),
+ )
+
+ options = table.raw_table.table_schema.options
+ self.assertEqual("true", options["row-tracking.enabled"])
+ self.assertEqual("true", options["data-evolution.enabled"])
+ self.assertEqual("true", options["deletion-vectors.enabled"])
+ self.assertNotIn("data-evolution.row-sidecar.enabled", options)
+ self.assertEqual("vortex", options["file.format"])
+ self.assertEqual("full", options["global-index.search-mode"])
+ self.assertEqual("vortex", options["vector.file.format"])
+ self.assertEqual("default.docs",
self.conn.get_table("docs").identifier)
+ self.assertEqual(["id", "content", "embedding", "payload"],
+ [field.name for field in table.raw_table.fields])
+
+ def test_create_table_uses_options_and_partitioned(self):
+ table = self.conn.create_table(
+ "docs",
+ schema=_schema({
+ "id": pa.int32(),
+ "embedding": _vector(3),
+ "dt": pa.string(),
+ }),
+ options=dict(_PARQUET_OPTIONS, **{
+ "deletion-vectors.enabled": "false",
+ }),
+ partitioned=["dt"],
+ )
+
+ options = table.raw_table.table_schema.options
+ self.assertEqual(["dt"], table.raw_table.table_schema.partition_keys)
+ self.assertEqual("false", options["deletion-vectors.enabled"])
+ self.assertEqual("parquet", options["file.format"])
+ self.assertEqual("parquet", options["vector.file.format"])
+
+ def test_create_table_rejects_non_arrow_schema(self):
+ with self.assertRaisesRegex(ValueError, "pyarrow.Schema"):
+ self.conn.create_table("docs", schema={"id": pa.int32()})
+
+ def test_create_table_accepts_pyarrow_schema_types(self):
+ table = self.conn.create_table(
+ "typed",
+ schema=pa.schema([
+ pa.field("flag", pa.bool_(), nullable=False),
+ pa.field("tiny", pa.int8()),
+ pa.field("small", pa.int16()),
+ pa.field("id", pa.int64()),
+ pa.field("score", pa.float32()),
+ pa.field("ratio", pa.float64()),
+ pa.field("title", pa.string()),
+ pa.field("payload", pa.binary()),
+ pa.field("hash", pa.binary(16)),
+ pa.field("blob", pa.large_binary()),
+ pa.field("amount", pa.decimal128(12, 2)),
+ pa.field("dt", pa.date32()),
+ pa.field("created_at", pa.timestamp("us")),
+ pa.field("created_ltz", pa.timestamp("us", tz="UTC")),
+ pa.field("tags", pa.list_(pa.string())),
+ pa.field("attrs", pa.map_(pa.string(), pa.int32())),
+ pa.field("meta", pa.struct([pa.field("rank", pa.int32())])),
+ pa.field("embedding", _vector(3)),
+ ]),
+ )
+
+ types_by_name = {
+ field.name: str(field.type) for field in table.raw_table.fields
+ }
+ self.assertEqual("BOOLEAN NOT NULL", types_by_name["flag"])
+ self.assertEqual("TINYINT", types_by_name["tiny"])
+ self.assertEqual("SMALLINT", types_by_name["small"])
+ self.assertEqual("BIGINT", types_by_name["id"])
+ self.assertEqual("FLOAT", types_by_name["score"])
+ self.assertEqual("DOUBLE", types_by_name["ratio"])
+ self.assertEqual("STRING", types_by_name["title"])
+ self.assertEqual("BYTES", types_by_name["payload"])
+ self.assertEqual("BINARY(16)", types_by_name["hash"])
+ self.assertEqual("BLOB", types_by_name["blob"])
+ self.assertEqual("DECIMAL(12, 2)", types_by_name["amount"])
+ self.assertEqual("DATE", types_by_name["dt"])
+ self.assertEqual("TIMESTAMP(6)", types_by_name["created_at"])
+ self.assertEqual("TIMESTAMP_LTZ(6)", types_by_name["created_ltz"])
+ self.assertEqual("ARRAY<STRING>", types_by_name["tags"])
+ self.assertEqual("MAP<STRING, INT>", types_by_name["attrs"])
+ self.assertEqual("ROW<rank: INT>", types_by_name["meta"])
+ self.assertEqual("VECTOR<FLOAT, 3>", types_by_name["embedding"])
+
+ def test_drop_table_can_ignore_missing_table(self):
+ self.conn.drop_table("missing", ignore_if_not_exists=True)
+
+ def test_get_table_rejects_non_data_evolution_table(self):
+ self.conn.catalog.create_database("default", ignore_if_exists=True)
+ self.conn.catalog.create_table(
+ "default.raw",
+ _raw_schema(),
+ False,
+ )
+
+ with self.assertRaisesRegex(ValueError, "data-evolution.enabled"):
+ self.conn.get_table("raw")
+
+ def test_get_table_rejects_primary_key_table(self):
+ self.conn.catalog.create_database("default", ignore_if_exists=True)
+ self.conn.catalog.create_table(
+ "default.pk",
+ _raw_schema(
+ options=dict(_PARQUET_OPTIONS, **{"bucket": "1"}),
+ primary_keys=["id"],
+ ),
+ False,
+ )
+
+ with self.assertRaisesRegex(ValueError, "primary keys"):
+ self.conn.get_table("pk")
+
+ def test_create_table_can_add_initial_data_and_get_by_short_name(self):
+ self.conn.create_table(
+ "users",
+ data=[
+ {"id": 1, "name": "Alice", "age": 30},
+ {"id": 2, "name": "Bob", "age": 25},
+ ],
+ schema=_schema({
+ "id": pa.int32(),
+ "name": pa.string(),
+ "age": pa.int32(),
+ }),
+ options=_PARQUET_OPTIONS,
+ )
+
+ users = self.conn.get_table("users")
+ result = users.scan().select(["id", "name"]).to_arrow()
+
+ self.assertEqual(["id", "name"], result.column_names)
+ self.assertEqual([1, 2], result["id"].to_pylist())
+
+ def test_add_scan_where_select_limit(self):
+ users = self.conn.create_table(
+ "users",
+ schema=_schema({
+ "id": pa.int32(),
+ "name": pa.string(),
+ "age": pa.int32(),
+ }),
+ options=_PARQUET_OPTIONS,
+ )
+
+ users.add([
+ {"id": 1, "name": "Alice", "age": 30},
+ {"id": 2, "name": "Bob", "age": 25},
+ {"id": 3, "name": "Carol", "age": 40},
+ ])
+
+ result = (
+ users.scan()
+ .where("age >= 30")
+ .select(["id", "name"])
+ .limit(1)
+ .to_arrow()
+ )
+
+ self.assertEqual(["id", "name"], result.column_names)
+ self.assertEqual(1, result.num_rows)
+ self.assertEqual([1], result["id"].to_pylist())
+
+ def test_scan_does_not_expose_pre_filter(self):
+ users = self.conn.create_table(
+ "users",
+ schema=_schema({"id": pa.int32()}),
+ options=_PARQUET_OPTIONS,
+ )
+
+ self.assertFalse(hasattr(users.scan(), "pre_filter"))
+
+ def test_where_rejects_predicate_object(self):
+ users = self.conn.create_table(
+ "users",
+ schema=_schema({"id": pa.int32()}),
+ options=_PARQUET_OPTIONS,
+ )
+ predicate = PredicateBuilder(users.raw_table.fields).equal("id", 1)
+
+ with self.assertRaisesRegex(ValueError, "SQL-like string"):
+ users.scan().where(predicate)
+
+ def test_update_by_filter(self):
+ users = self.conn.create_table(
+ "users",
+ data=[
+ {"id": 1, "name": "Alice", "age": 30},
+ {"id": 2, "name": "Bob", "age": 25},
+ ],
+ schema=_schema({
+ "id": pa.int32(),
+ "name": pa.string(),
+ "age": pa.int32(),
+ }),
+ options=_PARQUET_OPTIONS,
+ )
+
+ users.update(where="id = 2", values={"age": 26})
+
+ rows = sorted(users.scan().to_list(), key=lambda r: r["id"])
+ self.assertEqual(
+ [
+ {"id": 1, "name": "Alice", "age": 30},
+ {"id": 2, "name": "Bob", "age": 26},
+ ],
+ rows,
+ )
+
+ def test_merge_updates_matches_and_inserts_new_rows(self):
+ users = self.conn.create_table(
+ "users",
+ data=[
+ {"id": 1, "name": "Alice", "age": 30},
+ {"id": 2, "name": "Bob", "age": 25},
+ ],
+ schema=_schema({
+ "id": pa.int32(),
+ "name": pa.string(),
+ "age": pa.int32(),
+ }),
+ options=_PARQUET_OPTIONS,
+ )
+
+ users.merge("id") \
+ .when_matched_update() \
+ .when_not_matched_insert() \
+ .execute([
+ {"id": 2, "name": "Bob_v2", "age": 26},
+ {"id": 3, "name": "Carol", "age": 40},
+ ])
+
+ rows = sorted(users.scan().to_list(), key=lambda r: r["id"])
+ self.assertEqual(
+ [
+ {"id": 1, "name": "Alice", "age": 30},
+ {"id": 2, "name": "Bob_v2", "age": 26},
+ {"id": 3, "name": "Carol", "age": 40},
+ ],
+ rows,
+ )
+
+ def test_merge_all_uses_only_source_columns(self):
+ users = self.conn.create_table(
+ "users",
+ data=[
+ {"id": 1, "name": "Alice", "age": 30},
+ {"id": 2, "name": "Bob", "age": 25},
+ ],
+ schema=_schema({
+ "id": pa.int32(),
+ "name": pa.string(),
+ "age": pa.int32(),
+ }),
+ options=_PARQUET_OPTIONS,
+ )
+
+ users.merge("id") \
+ .when_matched_update() \
+ .when_not_matched_insert() \
+ .execute([
+ {"id": 2, "age": 26},
+ {"id": 3, "age": 40},
+ ])
+
+ rows = sorted(users.scan().to_list(), key=lambda r: r["id"])
+ self.assertEqual(
+ [
+ {"id": 1, "name": "Alice", "age": 30},
+ {"id": 2, "name": "Bob", "age": 26},
+ {"id": 3, "name": None, "age": 40},
+ ],
+ rows,
+ )
+
+ def test_merge_supports_source_key_mapping(self):
+ users = self.conn.create_table(
+ "users",
+ data=[
+ {"id": 1, "name": "Alice", "age": 30},
+ {"id": 2, "name": "Bob", "age": 25},
+ ],
+ schema=_schema({
+ "id": pa.int32(),
+ "name": pa.string(),
+ "age": pa.int32(),
+ }),
+ options=_PARQUET_OPTIONS,
+ )
+
+ users.merge({"id": "source_id"}) \
+ .when_matched_update({"age": source_col("age")}) \
+ .when_not_matched_insert() \
+ .execute([
+ {"source_id": 2, "name": "Bob_v2", "age": 26},
+ {"source_id": 3, "name": "Carol", "age": 40},
+ ])
+
+ rows = sorted(users.scan().to_list(), key=lambda r: r["id"])
+ self.assertEqual(
+ [
+ {"id": 1, "name": "Alice", "age": 30},
+ {"id": 2, "name": "Bob", "age": 26},
+ {"id": 3, "name": "Carol", "age": 40},
+ ],
+ rows,
+ )
+
+ def test_merge_where_uses_source_and_target_aliases(self):
+ users = self.conn.create_table(
+ "users",
+ schema=_schema({
+ "id": pa.int32(),
+ "name": pa.string(),
+ "age": pa.int32(),
+ }),
+ options=_PARQUET_OPTIONS,
+ )
+
+ calls = {}
+
+ class FakeUpdate:
+ def merge_into(
+ self,
+ source,
+ on,
+ when_matched=None,
+ when_not_matched=None):
+ calls["on"] = on
+ calls["matched"] = list(when_matched or [])
+ calls["not_matched"] = list(when_not_matched or [])
+ return []
+
+ class FakeCommit:
+ def commit(self, messages):
+ calls["messages"] = messages
+
+ def close(self):
+ pass
+
+ class FakeWriteBuilder:
+ def new_update(self):
+ return FakeUpdate()
+
+ def new_commit(self):
+ return FakeCommit()
+
+ users.raw_table.new_batch_write_builder = lambda: FakeWriteBuilder()
+
+ (
+ users.merge("id")
+ .when_matched_update(
+ {"age": source_col("age")},
+ where="source.age > target.age and source.name !=
'target.name'",
+ )
+ .when_not_matched_insert(where="source.age > 0")
+ .execute([
+ {"id": 1, "name": "Alice", "age": 31},
+ ])
+ )
+
+ self.assertEqual({"id": "id"}, calls["on"])
+ self.assertEqual(
+ "s.age > t.age and s.name != 'target.name'",
+ calls["matched"][0].condition,
+ )
+ self.assertEqual("s.age > 0", calls["not_matched"][0].condition)
+ self.assertEqual([], calls["messages"])
+
+ def test_merge_validates_on_columns(self):
+ users = self.conn.create_table(
+ "users",
+ schema=_schema({"id": pa.int32(), "name": pa.string()}),
+ options=_PARQUET_OPTIONS,
+ )
+
+ with self.assertRaisesRegex(ValueError, "source columns"):
+ users.merge({"id": "missing"}) \
+ .when_matched_update() \
+ .execute([{"id": 1, "name": "Alice"}])
+
+ def test_create_index_normalizes_full_text_alias(self):
+ docs = self.conn.create_table(
+ "docs",
+ schema=_schema({
+ "id": pa.int32(),
+ "content": pa.string(),
+ }),
+ options=_PARQUET_OPTIONS,
+ )
+
+ calls = []
+
+ def create_global_index(column, index_type, options=None):
+ calls.append((column, index_type, options))
+ return index_type
+
+ docs.raw_table.create_global_index = create_global_index
+
+ options = {"tokenizer": "default"}
+ self.assertEqual(
+ "tantivy-fulltext",
+ docs.create_index("content", index_type="full-text",
+ options=options),
+ )
+ self.assertEqual(
+ "tantivy-fulltext",
+ docs.create_index("content", index_type="full_text"),
+ )
+ self.assertEqual(
+ "tantivy-fulltext",
+ docs.create_index("content", index_type="fulltext"),
+ )
+
+ self.assertEqual(
+ [
+ ("content", "tantivy-fulltext", options),
+ ("content", "tantivy-fulltext", None),
+ ("content", "tantivy-fulltext", None),
+ ],
+ calls,
+ )
+
+ def test_create_index_requires_index_type(self):
+ docs = self.conn.create_table(
+ "docs",
+ schema=_schema({
+ "id": pa.int32(),
+ "content": pa.string(),
+ }),
+ options=_PARQUET_OPTIONS,
+ )
+
+ def create_global_index(column, index_type, options=None):
+ return index_type
+
+ docs.raw_table.create_global_index = create_global_index
+
+ with self.assertRaises(TypeError):
+ docs.create_index("content")
+
+ def test_table_does_not_expose_predicate_builder(self):
+ docs = self.conn.create_table(
+ "docs",
+ schema=_schema({"id": pa.int32()}),
+ options=_PARQUET_OPTIONS,
+ )
+
+ self.assertFalse(hasattr(docs, "predicate_builder"))
+
+ def test_search_reads_vector_matching_rows(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": "a", "embedding": [1.0, 0.0, 0.0]},
+ {"id": 2, "content": "b", "embedding": [0.0, 1.0, 0.0]},
+ {"id": 3, "content": "c", "embedding": [0.0, 0.0, 1.0]},
+ ])
+
+ calls = {}
+
+ class FakeVectorBuilder:
+ def with_vector_column(self, column):
+ calls["column"] = column
+ return self
+
+ def with_query_vector(self, vector):
+ calls["vector"] = vector
+ return self
+
+ def with_limit(self, limit):
+ calls["limit"] = limit
+ return self
+
+ def with_options(self, options):
+ calls["options"] = options
+ return self
+
+ def with_filter(self, predicate):
+ calls["filter"] = predicate
+ return self
+
+ def execute_local(self):
+ return GlobalIndexResult.from_range(Range(1, 1))
+
+ docs.raw_table.new_vector_search_builder = lambda: FakeVectorBuilder()
+
+ result = (
+ docs.search([0.0, 1.0, 0.0])
+ .where("id >= 1")
+ .limit(5)
+ .to_arrow()
+ )
+
+ self.assertEqual("embedding", calls["column"])
+ self.assertEqual([0.0, 1.0, 0.0], calls["vector"])
+ self.assertEqual(5, calls["limit"])
+ self.assertEqual([2], result["id"].to_pylist())
+
+ def test_search_applies_pre_filter_to_vector_builder(self):
+ docs = self.conn.create_table(
+ "docs",
+ schema=_schema({
+ "id": pa.int32(),
+ "category": pa.string(),
+ "embedding": _vector(3),
+ }),
+ options=_PARQUET_OPTIONS,
+ )
+ docs.add([
+ {"id": 1, "category": "lake", "embedding": [1.0, 0.0, 0.0]},
+ {"id": 2, "category": "city", "embedding": [0.0, 1.0, 0.0]},
+ ])
+
+ calls = {}
+
+ 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 with_filter(self, predicate):
+ calls["pre_filter"] = predicate
+ return self
+
+ def execute_local(self):
+ return GlobalIndexResult.from_range(Range(0, 0))
+
+ docs.raw_table.new_vector_search_builder = lambda: FakeVectorBuilder()
+
+ docs.search(
+ [1.0, 0.0, 0.0],
+ pre_filter="category = 'lake'",
+ ).limit(1).to_list()
+
+ self.assertEqual("equal", calls["pre_filter"].method)
+ self.assertEqual("category", calls["pre_filter"].field)
+ self.assertEqual(["lake"], calls["pre_filter"].literals)
+
+ def test_search_pre_filter_rejects_predicate_object(self):
+ docs = self.conn.create_table(
+ "docs",
+ schema=_schema({
+ "id": pa.int32(),
+ "embedding": _vector(3),
+ }),
+ options=_PARQUET_OPTIONS,
+ )
+ predicate = PredicateBuilder(docs.raw_table.fields).equal("id", 1)
+
+ with self.assertRaisesRegex(ValueError, "SQL-like string"):
+ docs.search([1.0, 0.0, 0.0]).pre_filter(predicate)
+
+ def test_search_accepts_generator_vector(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]}])
+
+ calls = {}
+
+ class FakeVectorBuilder:
+ def with_vector_column(self, column):
+ calls["column"] = column
+ return self
+
+ def with_query_vector(self, vector):
+ calls["vector"] = vector
+ return self
+
+ def with_limit(self, limit):
+ calls["limit"] = limit
+ return self
+
+ def with_options(self, options):
+ calls["options"] = 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((v for v in [1.0, 0.0, 0.0])).limit(1).to_list()
+
+ self.assertEqual("embedding", calls["column"])
+ 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_rejects_batch_vectors(self):
+ docs = self.conn.create_table(
+ "docs",
+ schema=_schema({
+ "id": pa.int32(),
+ "embedding": _vector(3),
+ }),
+ options=_PARQUET_OPTIONS,
+ )
+
+ with self.assertRaisesRegex(ValueError, "use search_vectors"):
+ docs.search([[1.0, 0.0, 0.0], [0.0, 1.0, 0.0]])
+
+ def test_search_vectors_reads_one_result_set_per_query_vector(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": "a", "embedding": [1.0, 0.0, 0.0]},
+ {"id": 2, "content": "b", "embedding": [0.0, 1.0, 0.0]},
+ {"id": 3, "content": "c", "embedding": [0.0, 0.0, 1.0]},
+ ])
+
+ calls = {}
+
+ class FakeBatchVectorBuilder:
+ def with_vector_column(self, column):
+ calls["column"] = column
+ return self
+
+ def with_query_vectors(self, vectors):
+ calls["vectors"] = vectors
+ return self
+
+ def with_limit(self, limit):
+ calls["limit"] = limit
+ return self
+
+ def with_options(self, options):
+ calls["options"] = options
+ return self
+
+ def with_filter(self, predicate):
+ calls["filter"] = predicate
+ return self
+
+ def execute_batch_local(self):
+ return [
+ GlobalIndexResult.from_range(Range(0, 0)),
+ GlobalIndexResult.from_range(Range(2, 2)),
+ ]
+
+ docs.raw_table.new_batch_vector_search_builder = (
+ lambda: FakeBatchVectorBuilder())
+
+ result = (
+ docs.search_vectors(
+ [[1.0, 0.0, 0.0], [0.0, 0.0, 1.0]],
+ options={"nprobe": "8"},
+ pre_filter="content = 'a'",
+ )
+ .where("id >= 1")
+ .select(["id"])
+ .limit(2)
+ .to_list()
+ )
+
+ self.assertEqual("embedding", calls["column"])
+ self.assertEqual(
+ [[1.0, 0.0, 0.0], [0.0, 0.0, 1.0]],
+ calls["vectors"],
+ )
+ self.assertEqual(2, calls["limit"])
+ self.assertEqual({"nprobe": "8"}, calls["options"])
+ self.assertEqual("content", calls["filter"].field)
+ self.assertEqual(["a"], calls["filter"].literals)
+ self.assertEqual([[{"id": 1}], [{"id": 3}]], result)
+
+ def test_search_vectors_rejects_text_parameter(self):
+ docs = self.conn.create_table(
+ "docs",
+ schema=_schema({
+ "id": pa.int32(),
+ "content": pa.string(),
+ "embedding": _vector(3),
+ }),
+ options=_PARQUET_OPTIONS,
+ )
+
+ with self.assertRaises(TypeError):
+ docs.search_vectors([[1.0, 0.0, 0.0]], text="paimon")
+
+ def test_search_reads_text_query(self):
+ docs = self.conn.create_table(
+ "docs",
+ schema=_schema({
+ "id": pa.int32(),
+ "content": pa.string(),
+ }),
+ options=_PARQUET_OPTIONS,
+ )
+ docs.add([
+ {"id": 1, "content": "paimon vector"},
+ {"id": 2, "content": "lakehouse"},
+ ])
+
+ calls = {}
+
+ class FakeFullTextBuilder:
+ def with_query(self, query):
+ calls["query"] = query.to_dict()
+ return self
+
+ def with_limit(self, limit):
+ calls["limit"] = limit
+ return self
+
+ def execute_local(self):
+ return GlobalIndexResult.from_range(Range(0, 0))
+
+ docs.raw_table.new_full_text_search_builder = lambda:
FakeFullTextBuilder()
+
+ result = (
+ docs.search("paimon vector")
+ .limit(1)
+ .to_arrow()
+ )
+
+ self.assertEqual(1, calls["limit"])
+ self.assertEqual("content", calls["query"]["match"]["column"])
+ self.assertEqual("paimon vector", calls["query"]["match"]["terms"])
+ self.assertEqual("Or", calls["query"]["match"]["operator"])
+ self.assertEqual([1], result["id"].to_pylist())
+
+ def test_search_string_requires_unambiguous_text_column(self):
+ docs = self.conn.create_table(
+ "docs",
+ schema=_schema({
+ "id": pa.int32(),
+ "title": pa.string(),
+ "content": pa.string(),
+ }),
+ options=_PARQUET_OPTIONS,
+ )
+
+ with self.assertRaisesRegex(ValueError, "Multiple text columns"):
+ docs.search("paimon")
+
+ def test_search_hybrid_rejects_shorthand_arguments(self):
+ docs = self.conn.create_table(
+ "docs",
+ schema=_schema({
+ "id": pa.int32(),
+ "embedding": _vector(3),
+ }),
+ options=_PARQUET_OPTIONS,
+ )
+
+ with self.assertRaises(TypeError):
+ docs.search_hybrid(vector=[1.0, 0.0, 0.0])
+
+ def test_hybrid_route_helpers_use_route_specific_arguments(self):
+ with self.assertRaises(TypeError):
+ pmm.vector_route([1.0, 0.0, 0.0])
+ route = pmm.text_route("paimon")
+ self.assertFalse(hasattr(route, "column"))
+ self.assertEqual("paimon", route.query)
+
+ def test_search_can_build_hybrid_routes(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": "paimon", "embedding": [1.0, 0.0, 0.0]},
+ {"id": 2, "content": "vector", "embedding": [0.0, 1.0, 0.0]},
+ ])
+
+ calls = {}
+
+ class FakeHybridBuilder:
+ def with_limit(self, limit):
+ calls["limit"] = limit
+ return self
+
+ def with_ranker(self, ranker):
+ calls["ranker"] = ranker
+ return self
+
+ def add_vector_route(
+ self, column, vector, limit, weight=1.0, options=None):
+ calls["vector"] = (column, vector, limit, weight, options)
+ return self
+
+ def add_full_text_route(
+ self, query_json, limit, weight=1.0, options=None):
+ calls["text"] = (json.loads(query_json), limit, weight,
options)
+ return self
+
+ def with_filter(self, predicate):
+ calls["filter"] = predicate
+ return self
+
+ def execute_local(self):
+ return GlobalIndexResult.from_range(Range(0, 1))
+
+ docs.raw_table.new_hybrid_search_builder = lambda: FakeHybridBuilder()
+
+ result = (
+ docs.search_hybrid(
+ [
+ pmm.vector_route("embedding", [1.0, 0.0, 0.0]),
+ pmm.text_route("paimon"),
+ ],
+ )
+ .rerank("rrf")
+ .limit(2)
+ .to_arrow()
+ )
+
+ self.assertEqual(2, calls["limit"])
+ self.assertEqual("rrf", calls["ranker"])
+ self.assertEqual(
+ ("embedding", [1.0, 0.0, 0.0], 2, 1.0, {}),
+ calls["vector"],
+ )
+ self.assertEqual("content", calls["text"][0]["match"]["column"])
+ self.assertEqual("paimon", calls["text"][0]["match"]["terms"])
+ self.assertEqual([1, 2], result["id"].to_pylist())
+
+ def test_search_hybrid_applies_pre_filter_to_vector_routes(self):
+ docs = self.conn.create_table(
+ "docs",
+ schema=_schema({
+ "id": pa.int32(),
+ "category": pa.string(),
+ "embedding": _vector(3),
+ }),
+ options=_PARQUET_OPTIONS,
+ )
+ docs.add([
+ {"id": 1, "category": "lake", "embedding": [1.0, 0.0, 0.0]},
+ {"id": 2, "category": "city", "embedding": [0.0, 1.0, 0.0]},
+ ])
+
+ calls = {}
+
+ class FakeHybridBuilder:
+ def with_limit(self, limit):
+ calls["limit"] = limit
+ return self
+
+ def with_ranker(self, ranker):
+ calls["ranker"] = ranker
+ return self
+
+ def add_vector_route(
+ self, column, vector, limit, weight=1.0, options=None):
+ calls["vector"] = (column, vector, limit, weight, options)
+ return self
+
+ def with_filter(self, predicate):
+ calls["pre_filter"] = predicate
+ return self
+
+ def execute_local(self):
+ return GlobalIndexResult.from_range(Range(0, 0))
+
+ docs.raw_table.new_hybrid_search_builder = lambda: FakeHybridBuilder()
+
+ result = (
+ docs.search_hybrid(
+ [pmm.vector_route("embedding", [1.0, 0.0, 0.0])],
+ pre_filter="category = 'lake'",
+ )
+ .limit(1)
+ .to_list()
+ )
+
+ self.assertEqual(1, calls["limit"])
+ self.assertEqual(
+ ("embedding", [1.0, 0.0, 0.0], 1, 1.0, {}),
+ calls["vector"],
+ )
+ self.assertEqual("category", calls["pre_filter"].field)
+ self.assertEqual(["lake"], calls["pre_filter"].literals)
+ self.assertEqual(
+ [{"id": 1, "category": "lake", "embedding": [1.0, 0.0, 0.0]}],
+ result,
+ )
+
+ def test_module_does_not_export_vectors_route(self):
+ self.assertFalse(hasattr(pmm, "vectors_route"))
+ self.assertFalse(hasattr(pmm, "VectorsRoute"))
+
+ def test_search_hybrid_can_build_multiple_vector_routes(self):
+ docs = self.conn.create_table(
+ "docs",
+ schema=_schema({
+ "id": pa.int32(),
+ "content": pa.string(),
+ "image_embedding": _vector(3),
+ "text_embedding": _vector(3),
+ }),
+ options=_PARQUET_OPTIONS,
+ )
+ docs.add([
+ {
+ "id": 1,
+ "content": "paimon",
+ "image_embedding": [1.0, 0.0, 0.0],
+ "text_embedding": [0.0, 1.0, 0.0],
+ },
+ {
+ "id": 2,
+ "content": "vector",
+ "image_embedding": [0.0, 1.0, 0.0],
+ "text_embedding": [1.0, 0.0, 0.0],
+ },
+ ])
+
+ calls = {"vector_routes": [], "text_routes": []}
+
+ class FakeHybridBuilder:
+ def with_limit(self, limit):
+ calls["limit"] = limit
+ return self
+
+ def with_ranker(self, ranker):
+ calls["ranker"] = ranker
+ return self
+
+ def add_vector_route(
+ self, column, vector, limit, weight=1.0, options=None):
+ calls["vector_routes"].append(
+ (column, vector, limit, weight, options))
+ return self
+
+ def add_full_text_route(
+ self, query_json, limit, weight=1.0, options=None):
+ calls["text_routes"].append(
+ (json.loads(query_json), limit, weight, options))
+ return self
+
+ def execute_local(self):
+ return GlobalIndexResult.from_range(Range(0, 0))
+
+ docs.raw_table.new_hybrid_search_builder = lambda: FakeHybridBuilder()
+
+ result = (
+ docs.search_hybrid(
+ [
+ pmm.vector_route(
+ "image_embedding",
+ [1.0, 0.0, 0.0],
+ weight=0.7,
+ limit=6,
+ options={"nprobe": "8"},
+ ),
+ pmm.vector_route(
+ "text_embedding",
+ [0.0, 1.0, 0.0],
+ weight=0.3,
+ limit=4,
+ options={"nprobe": "4"},
+ ),
+ pmm.text_route("paimon", weight=0.2),
+ ],
+ ranker="weighted_score",
+ route_limit=4,
+ )
+ .limit(2)
+ .to_list()
+ )
+
+ self.assertEqual(2, calls["limit"])
+ self.assertEqual("weighted_score", calls["ranker"])
+ self.assertEqual(
+ [
+ (
+ "image_embedding",
+ [1.0, 0.0, 0.0],
+ 6,
+ 0.7,
+ {"nprobe": "8"},
+ ),
+ (
+ "text_embedding",
+ [0.0, 1.0, 0.0],
+ 4,
+ 0.3,
+ {"nprobe": "4"},
+ ),
+ ],
+ calls["vector_routes"],
+ )
+ self.assertEqual("content",
calls["text_routes"][0][0]["match"]["column"])
+ self.assertEqual("paimon",
calls["text_routes"][0][0]["match"]["terms"])
+ self.assertEqual(4, calls["text_routes"][0][1])
+ self.assertEqual(0.2, calls["text_routes"][0][2])
+ self.assertEqual(
+ [{
+ "id": 1,
+ "content": "paimon",
+ "image_embedding": [1.0, 0.0, 0.0],
+ "text_embedding": [0.0, 1.0, 0.0],
+ }],
+ result,
+ )
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/paimon-python/pypaimon/tests/where_parser_test.py
b/paimon-python/pypaimon/tests/where_parser_test.py
index 04385243d9..c5ba552ac7 100644
--- a/paimon-python/pypaimon/tests/where_parser_test.py
+++ b/paimon-python/pypaimon/tests/where_parser_test.py
@@ -17,7 +17,7 @@
import unittest
-from pypaimon.cli.where_parser import parse_where_clause, _tokenize,
_cast_literal
+from pypaimon.common.where_parser import parse_where_clause, _tokenize,
_cast_literal
from pypaimon.schema.data_types import ArrayType, AtomicType, DataField
