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 3c117ac827 [python] Execute full-text and hybrid search with Ray
(#10430)
3c117ac827 is described below
commit 3c117ac827745cb3fce00530a43144bdbfdf3a43
Author: chaoyang <[email protected]>
AuthorDate: Thu Oct 8 15:28:02 2026 +0800
[python] Execute full-text and hybrid search with Ray (#10430)
---
docs/docs/pypaimon/multimodal-search.md | 23 +++++
paimon-python/pypaimon/multimodal/query.py | 46 ++++++++-
paimon-python/pypaimon/ray/full_text_search.py | 105 +++++++++++++++++++++
.../pypaimon/tests/ray_full_text_search_test.py | 95 +++++++++++++++++++
4 files changed, 265 insertions(+), 4 deletions(-)
diff --git a/docs/docs/pypaimon/multimodal-search.md
b/docs/docs/pypaimon/multimodal-search.md
index 1c56a5e04b..47b4f8698f 100644
--- a/docs/docs/pypaimon/multimodal-search.md
+++ b/docs/docs/pypaimon/multimodal-search.md
@@ -271,6 +271,29 @@ earlier shards, so a slow shard can delay further
submissions. Global top-k
selection still uses all merged candidate scores, whose storage grows with the
number of unique candidate rows per query.
+## Full-text and Hybrid queries on Ray
+
+Full-text and Hybrid queries accept `execution="ray"`, `concurrency`, and
+`ray_remote_args` on `to_arrow`, `to_pandas`, and `to_list`, matching vector
queries:
+
+```python
+rows = (docs.search("robot grasp", column="content")
+ .with_score().order_by_score().limit(20)
+ .to_arrow(execution="ray", concurrency=4))
+```
+
+This path supports data-evolution tables and requires `pypaimon[ray,full-text]`
+on every worker (plus `vindex` for native vector routes). Index shards run as
+bounded Ray tasks. Full-text raw fallback runs on one worker with the complete
+uncovered corpus, preserving existing BM25 statistics; that corpus must fit the
+worker's memory. Pre-filters and deletion vectors keep the local ranking
semantics.
+
+Hybrid routes share one resolved snapshot, use Ray for their shard work, and
+reuse the existing RRF, MRR, or weighted-score fusion on the driver. Routes run
+successively so `concurrency` bounds the whole query. Final Arrow/Pandas row
lookup
+remains on the driver. Native resources close on workers; task errors propagate
+and pending read tasks are cancelled.
+
## Search Hybrid
Use `search_hybrid` to combine vector and full-text routes, then rerank the
diff --git a/paimon-python/pypaimon/multimodal/query.py
b/paimon-python/pypaimon/multimodal/query.py
index e743bdf6ae..d4d70267c8 100644
--- a/paimon-python/pypaimon/multimodal/query.py
+++ b/paimon-python/pypaimon/multimodal/query.py
@@ -669,7 +669,29 @@ class VectorQuery(_PreFilterQuery):
return builder
-class TextQuery(_PreFilterQuery):
+class _RayTextQuery(_PreFilterQuery):
+ """Shared output methods for full-text and hybrid Ray execution."""
+
+ def to_arrow(self, *, execution="local", concurrency=None,
ray_remote_args=None):
+ if execution == "local":
+ if concurrency is not None or ray_remote_args is not None:
+ raise ValueError("Ray options require execution='ray'.")
+ return super().to_arrow()
+ if execution != "ray":
+ raise ValueError("execution must be 'local' or 'ray'.")
+ query = self._for_execution()
+ return query._read_global_index_result(query._execute_ray(concurrency,
ray_remote_args))
+
+ def to_pandas(self, *, execution="local", concurrency=None,
ray_remote_args=None):
+ return self.to_arrow(execution=execution, concurrency=concurrency,
+ ray_remote_args=ray_remote_args).to_pandas()
+
+ def to_list(self, *, execution="local", concurrency=None,
ray_remote_args=None):
+ return self.to_arrow(execution=execution, concurrency=concurrency,
+ ray_remote_args=ray_remote_args).to_pylist()
+
+
+class TextQuery(_RayTextQuery):
"""Chainable query wrapper for full-text global-index search."""
def __init__(self, table, text_query, pre_filter=None):
@@ -678,6 +700,14 @@ class TextQuery(_PreFilterQuery):
table, result_factory=self._execute_fts, pre_filter=pre_filter)
def _execute_fts(self, query):
+ return self._full_text_search_builder(query).execute_local()
+
+ def _execute_ray(self, concurrency, ray_remote_args):
+ from pypaimon.ray.full_text_search import _execute_full_text_search
+ return _execute_full_text_search(self._full_text_search_builder(self),
+ concurrency=concurrency,
ray_remote_args=ray_remote_args)
+
+ def _full_text_search_builder(self, query):
limit = query._limit if query._limit is not None else 10
builder = (
query._table.new_full_text_search_builder()
@@ -686,10 +716,10 @@ class TextQuery(_PreFilterQuery):
)
if query._pre_filter is not None:
builder = builder.with_filter(query._pre_filter)
- return builder.execute_local()
+ return builder
-class HybridQuery(_PreFilterQuery):
+class HybridQuery(_RayTextQuery):
"""Chainable query wrapper for hybrid global-index search."""
def __init__(
@@ -712,6 +742,14 @@ class HybridQuery(_PreFilterQuery):
return self
def _execute_hybrid(self, query):
+ return self._hybrid_search_builder(query).execute_local()
+
+ def _execute_ray(self, concurrency, ray_remote_args):
+ from pypaimon.ray.full_text_search import _execute_hybrid_search
+ return _execute_hybrid_search(self._hybrid_search_builder(self),
+ concurrency=concurrency,
ray_remote_args=ray_remote_args)
+
+ def _hybrid_search_builder(self, query):
final_limit = query._limit if query._limit is not None else 10
route_limit = self._route_limit or final_limit
builder = (
@@ -737,7 +775,7 @@ class HybridQuery(_PreFilterQuery):
)
if query._pre_filter is not None:
builder = builder.with_filter(query._pre_filter)
- return builder.execute_local()
+ return builder
class BatchVectorQuery(_PreFilterQuery):
diff --git a/paimon-python/pypaimon/ray/full_text_search.py
b/paimon-python/pypaimon/ray/full_text_search.py
new file mode 100644
index 0000000000..6a068f6fab
--- /dev/null
+++ b/paimon-python/pypaimon/ray/full_text_search.py
@@ -0,0 +1,105 @@
+# 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.
+
+"""Ray full-text shard execution, preserving local corpus and merge
semantics."""
+
+from contextlib import closing
+
+from pypaimon.globalindex.vector_search_result import
DictBasedScoredIndexResult
+from pypaimon.ray.vector_search import _execution_options, _map_tasks,
_require_ray, _scores
+from pypaimon.table.source.full_text_read import DataEvolutionFullTextRead
+from pypaimon.table.source import global_index_live_row_filter
+from pypaimon.utils.roaring_bitmap import RoaringBitmap64
+
+
+def _execute_full_text_search(builder, *, concurrency=None,
ray_remote_args=None):
+ concurrency, remote_args = _execution_options(concurrency, ray_remote_args)
+ reader = builder.new_full_text_read()
+ if (type(reader) is not DataEvolutionFullTextRead
+ or not reader._table.options.data_evolution_enabled()):
+ raise ValueError("Ray full-text search supports only data-evolution
tables.")
+ _require_ray()
+ plan = builder.new_full_text_scan().scan()
+ return _RayFullTextRead(reader, concurrency, remote_args).read_plan(plan)
+
+
+class _RayFullTextRead(DataEvolutionFullTextRead):
+ def __init__(self, reader, concurrency, remote_args):
+ super().__init__(reader._table, reader._limit, reader._text_columns,
reader._query,
+ reader._partition_filter, reader._filter)
+ self._concurrency = concurrency
+ self._remote_args = remote_args
+
+ def _worker_reader(self):
+ return DataEvolutionFullTextRead(
+ self._table, self._limit, self._text_columns, self._query,
+ self._partition_filter, self._filter)
+
+ def _eval_column_query(self, splits_by_column, live_rows):
+ items = []
+ for split in splits_by_column.get(self._text_columns[0].name, []):
+ include = global_index_live_row_filter.for_range(
+ live_rows, split.row_range_start, split.row_range_end)
+ if include is not None and include.is_empty():
+ continue
+ items.append((split, None if include is None else
include.serialize()))
+ merged = {}
+ with closing(_map_tasks(_search_index, self._worker_reader(), items,
+ self._concurrency, self._remote_args, True))
as tasks:
+ for _, scores in tasks:
+ # Duplicate row IDs keep the first planned shard's score.
+ for row_id, score in scores.items():
+ merged.setdefault(row_id, score)
+ return DictBasedScoredIndexResult(merged).top_k(self._limit)
+
+ def _read_raw_search(self, ranges, index_type, include_row_ids=None):
+ if not ranges:
+ return DictBasedScoredIndexResult({})
+ include = None if include_row_ids is None else
include_row_ids.serialize()
+ # Splitting this corpus would change document frequencies and BM25.
+ # Keep the existing local corpus intact, but read/build/search on a
worker.
+ with closing(_map_tasks(_search_raw, self._worker_reader(), [(ranges,
index_type, include)],
+ 1, self._remote_args)) as tasks:
+ for _, scores in tasks:
+ return DictBasedScoredIndexResult(scores)
+
+
+def _search_index(reader, item):
+ split, include = item
+ include = None if include is None else RoaringBitmap64.deserialize(include)
+ return _scores(reader._eval(split.row_range_start, split.row_range_end,
+ split.full_text_index_files, include).result())
+
+
+def _search_raw(reader, item):
+ ranges, index_type, include = item
+ include = None if include is None else RoaringBitmap64.deserialize(include)
+ return _scores(reader._read_raw_search(ranges, index_type, include))
+
+
+def _execute_hybrid_search(builder, *, concurrency=None, ray_remote_args=None):
+ from pypaimon.ray.vector_search import _execute_vector_search
+
+ concurrency, remote_args = _execution_options(concurrency, ray_remote_args)
+ route_results = []
+ # Route builders share the resolved snapshot. Running routes successively
+ # makes concurrency a query-wide bound, rather than multiplying it per
route.
+ for route in builder.route_builders():
+ execute = _execute_vector_search if route.route.is_vector() else
_execute_full_text_search
+ result = execute(route.search_builder, concurrency=concurrency,
ray_remote_args=remote_args)
+ route_results.append(builder.to_route_result(route, result))
+ return builder.rank(route_results)
diff --git a/paimon-python/pypaimon/tests/ray_full_text_search_test.py
b/paimon-python/pypaimon/tests/ray_full_text_search_test.py
new file mode 100644
index 0000000000..057662e6b8
--- /dev/null
+++ b/paimon-python/pypaimon/tests/ray_full_text_search_test.py
@@ -0,0 +1,95 @@
+# 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 unittest.mock import patch
+
+import pytest
+
+pytest.importorskip("ray")
+pytest.importorskip("paimon_ftindex")
+
+import pypaimon.multimodal as pm
+from pypaimon.tests import ray_vector_search_test as ray_fixtures
+from pypaimon.tests import full_text_scalar_filter_test as text_fixtures
+from pypaimon.ray import full_text_search as module
+
+ray_cluster = ray_fixtures.ray_cluster
+docs = text_fixtures.docs
+
+
[email protected]("predicate", [None, "label = 'target'", "pt = 'b'",
"id < 0"])
[email protected]("concurrency", [1, 2])
+def test_index_raw_filters_and_scores(docs, ray_cluster, predicate,
concurrency):
+ text_fixtures.append_rows(docs)
+ docs.delete("id = 1")
+ query = docs.search("paimon", column="text",
pre_filter=predicate).select(["id"]).with_score().order_by_score()
+ expected = query.to_list()
+ dispatched = []
+ original = module._map_tasks
+
+ def record(worker, context, items, *args):
+ dispatched.append((worker.__name__, len(items)))
+ return original(worker, context, items, *args)
+
+ with patch.object(module, "_map_tasks", record):
+ assert query.to_list(execution="ray", concurrency=concurrency) ==
expected
+ if predicate is None:
+ assert ("_search_index", 1) in dispatched
+ assert ("_search_raw", 1) in dispatched
+
+
[email protected]("ranker", ["rrf", "weighted_score", "mrr"])
+def test_hybrid_routes_preserve_ranking(docs, ray_cluster, ranker):
+ text_fixtures.append_rows(docs)
+ query = docs.search_hybrid([
+ pm.vector_route("embedding", [0., 1.], limit=3, weight=2),
+ pm.text_route("paimon", column="text", limit=4),
+ ], ranker=ranker, pre_filter="id >
0").select(["id"]).with_score().order_by_score().limit(3)
+ assert query.to_list(execution="ray", concurrency=2) == query.to_list()
+
+
+def test_snapshot_pinning_and_post_filter(docs, ray_cluster):
+ saved = docs.raw_table.snapshot_manager().get_latest_snapshot().id
+ docs.delete("id = 2")
+ query = docs.search("paimon", column="text", snapshot_id=saved,
+ pre_filter="label =
'target'").select(["id"]).with_score()
+ assert query.to_list(execution="ray") == query.to_list()
+ query.where("id = -1")
+ assert query.to_list(execution="ray") == []
+
+
[email protected]("options", [{"execution": "unknown"}, {"concurrency":
2},
+ {"execution": "ray", "concurrency": 0},
+ {"execution": "ray", "ray_remote_args":
{"num_returns": 2}}])
+def test_execution_validation(docs, options):
+ with pytest.raises(ValueError):
+ docs.search("paimon", column="text").to_arrow(**options)
+
+
+def test_multiple_index_shards_and_worker_failure(docs, ray_cluster):
+ text_fixtures.append_rows(docs)
+ docs.raw_table.copy({"deletion-vectors.enabled":
"false"}).create_global_index("text", "full-text")
+ query = docs.search("paimon",
column="text").select(["id"]).with_score().order_by_score()
+ assert query.to_list(execution="ray", concurrency=2) == query.to_list()
+ with patch.object(module, "_search_index", _fail_index):
+ with pytest.raises(Exception, match="injected full-text failure"):
+ query.to_arrow(execution="ray", concurrency=1)
+ assert query.to_list(execution="ray") == query.to_list()
+
+
+def _fail_index(reader, item):
+ raise RuntimeError("injected full-text failure")
