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 7befc15f18 [Python] Dispatch hybrid search phases to native execution
(#10769)
7befc15f18 is described below
commit 7befc15f18d81bc406679e14b48aa92cfe863335
Author: Jingsong Lee <[email protected]>
AuthorDate: Fri Oct 23 15:11:59 2026 +0800
[Python] Dispatch hybrid search phases to native execution (#10769)
---
.../global-index/hybrid-search.mdx | 24 +++
.../execution/native/read/native_hybrid_search.py | 67 ++++++-
.../execution/python/read/search/hybrid_search.py | 42 ++---
.../pypaimon/table/source/hybrid_search_builder.py | 37 ++--
.../pypaimon/table/source/search_dispatch.py | 33 +++-
.../tests/hybrid_search_phase_dispatch_test.py | 198 +++++++++++++++++++++
.../pypaimon/tests/hybrid_search_ranker_test.py | 51 ++++++
.../pypaimon/tests/native_hybrid_search_test.py | 149 +++++++++++++++-
8 files changed, 555 insertions(+), 46 deletions(-)
diff --git a/docs/docs/multimodal-table/global-index/hybrid-search.mdx
b/docs/docs/multimodal-table/global-index/hybrid-search.mdx
index 81bfae13e0..94625ab917 100644
--- a/docs/docs/multimodal-table/global-index/hybrid-search.mdx
+++ b/docs/docs/multimodal-table/global-index/hybrid-search.mdx
@@ -265,6 +265,30 @@ try (RecordReader<InternalRow> reader =
readBuilder.newRead().createReader(plan)
}
```
+To execute routes separately, use the same public builders and fuse their
results:
+
+```python
+builder = (
+ table.new_hybrid_search_builder()
+ .add_vector_route("title_embedding", [1.0, 0.0, 0.0], limit=50)
+ .with_limit(10)
+)
+routes = builder.route_builders()
+route_results = [
+ builder.to_route_result(route, route.execute_local())
+ for route in routes
+]
+result = builder.rank(route_results)
+```
+
+Route capture checks authorization and fixes one snapshot for all routes. Each
+`route.search_builder` is the builder retained by that route. Configuring it
changes
+subsequent `route.execute_local()` calls, and it supports its ordinary
scan/read
+methods. Ranking uses the supplied results and the builder's current limit and
+ranker; it does not read table metadata or execute queries again. Native
execution
+supports these phases as well as complete hybrid queries. Both backends use
Java's
+float32 scores, weights, and fusion arithmetic.
+
To search two vector indexes without a full-text route, add two vector routes:
```java
diff --git
a/paimon-python/pypaimon/execution/native/read/native_hybrid_search.py
b/paimon-python/pypaimon/execution/native/read/native_hybrid_search.py
index 660c59593a..56230dd9cf 100644
--- a/paimon-python/pypaimon/execution/native/read/native_hybrid_search.py
+++ b/paimon-python/pypaimon/execution/native/read/native_hybrid_search.py
@@ -15,7 +15,7 @@
# specific language governing permissions and limitations
# under the License.
-"""Transport an existing local hybrid query to one Rust core operation."""
+"""Transport hybrid queries, captured routes and weighted results to Rust
Core."""
from pypaimon.catalog.table_query_auth import reject_search_under_query_auth
from pypaimon.globalindex.full_text_options import NativeFullTextIndexOptions
@@ -27,8 +27,8 @@ from pypaimon.table.file_store_table import FileStoreTable
from pypaimon.execution.native.write.native_commit import
_rest_catalog_supported
-def try_native_hybrid_search(builder, executor=None):
- """Qualify before execution; core owns one snapshot, routes and fusion."""
+def _native_builder(builder, executor=None):
+ """Qualify the complete request before executing or capturing any route."""
table = builder._table
if type(table) is not FileStoreTable:
return None
@@ -64,6 +64,15 @@ def try_native_hybrid_search(builder, executor=None):
except (ValueError, NotImplementedError):
# Qualify strict literal support before executing any route or ranking.
return None
+ return table, native
+
+
+def try_native_hybrid_search(builder, executor=None):
+ """Core owns one snapshot, routes and fusion for the complete query."""
+ configured = _native_builder(builder, executor)
+ if configured is None:
+ return None
+ table, native = configured
# Propagate execution failures: retrying Python can select a newer
snapshot.
if executor is None:
result = native.execute()
@@ -76,3 +85,55 @@ def try_native_hybrid_search(builder, executor=None):
callback.cancel()
callback.raise_if_failed()
return DictBasedScoredIndexResult(result.row_ids())
+
+
+def try_native_hybrid_routes(builder):
+ configured = _native_builder(builder)
+ if configured is None:
+ return None
+ _, native = configured
+ # Core resolves the read view once. Retain ordinary public per-route
+ # builders so callers can still configure them and create scans and reads.
+ # Reconstructing these configurations must not resolve another snapshot.
+ from copy import copy
+ from pypaimon.common.json_util import JSON
+ from pypaimon.snapshot.snapshot import Snapshot
+ from pypaimon.table.source.hybrid_search_builder import
HybridSearchRouteBuilder
+
+ captured = native.route_builders()
+ routes = []
+ for route, native_route in zip(builder._routes, captured):
+ snapshot = native_route.snapshot()
+ snapshot = None if snapshot is None else
JSON.from_json(snapshot._to_json(), Snapshot)
+ execution = copy(builder)
+ execution._table = builder._table._copy_with_snapshot(snapshot)
+ search = (execution._new_vector_search_builder(route) if
route.is_vector() else
+ execution._new_full_text_search_builder(route))
+ routes.append(HybridSearchRouteBuilder(route, search))
+ return routes
+
+
+def rank_hybrid_results(builder, route_results):
+ """Fuse supplied scores without table access or validation of unused
queries."""
+ from pypaimon_rust.datafusion import (
+ GlobalIndexResult, HybridSearchRanker, HybridSearchWeightedResult,
+ )
+ from pypaimon.table.source.hybrid_search_builder import MRR_RANKER,
RRF_RANKER, WEIGHTED_SCORE_RANKER
+
+ ranker = {
+ RRF_RANKER: HybridSearchRanker.RRF,
+ MRR_RANKER: HybridSearchRanker.MRR,
+ WEIGHTED_SCORE_RANKER: HybridSearchRanker.WEIGHTED_SCORE,
+ }[builder._ranker]
+ weighted = []
+ for route_result in route_results:
+ result = route_result.result
+ rows = result.results()
+ if rows.is_empty():
+ scores = {}
+ else:
+ score = result.score_getter()
+ scores = {row_id: score(row_id) or 0.0 for row_id in rows}
+ weighted.append(HybridSearchWeightedResult(
+ GlobalIndexResult.from_scored_row_ids(scores),
route_result.route.weight))
+ return DictBasedScoredIndexResult(ranker.rank(weighted,
builder._limit).row_ids())
diff --git
a/paimon-python/pypaimon/execution/python/read/search/hybrid_search.py
b/paimon-python/pypaimon/execution/python/read/search/hybrid_search.py
index b53d7cd231..0183c67c6a 100644
--- a/paimon-python/pypaimon/execution/python/read/search/hybrid_search.py
+++ b/paimon-python/pypaimon/execution/python/read/search/hybrid_search.py
@@ -23,8 +23,9 @@ from concurrent.futures import FIRST_EXCEPTION,
ThreadPoolExecutor, as_completed
from threading import Event
from typing import List
+import numpy as np
+
from pypaimon.catalog.table_query_auth import reject_search_under_query_auth
-from pypaimon.globalindex.global_index_result import GlobalIndexResult
from pypaimon.globalindex.vector_search_result import
DictBasedScoredIndexResult, ScoredGlobalIndexResult
from pypaimon.table.source.hybrid_search_builder import (
HybridSearchRouteBuilder, HybridSearchRouteResult,
@@ -33,7 +34,7 @@ from pypaimon.table.source.hybrid_search_builder import (
_MAX_ROUTE_WORKERS = 4
-_RRF_K = 60.0
+_RRF_K = np.float32(60.0)
def execute(self, executor=None):
@@ -134,21 +135,6 @@ def route_builders(self) -> List[HybridSearchRouteBuilder]:
return builders
-def to_route_result(
- self,
- route_builder: HybridSearchRouteBuilder,
- result: GlobalIndexResult) -> HybridSearchRouteResult:
- if isinstance(result, ScoredGlobalIndexResult):
- scored = result
- elif result.results().is_empty():
- scored = ScoredGlobalIndexResult.create_empty()
- else:
- raise ValueError(
- "Hybrid search requires scored index results, but got: %s"
- % result.__class__.__name__)
- return HybridSearchRouteResult(route_builder.route, scored)
-
-
def rank(
self,
route_results: List[HybridSearchRouteResult]) ->
ScoredGlobalIndexResult:
@@ -167,18 +153,20 @@ def rank(
def _rrf(self, route_results):
scores = {}
for route_result in route_results:
+ weight = np.float32(route_result.route.weight)
for rank, row_id in enumerate(_ranked_row_ids(route_result.result)):
- contribution = route_result.route.weight / (_RRF_K + rank + 1.0)
- scores[row_id] = scores.get(row_id, 0.0) + contribution
+ contribution = weight / (_RRF_K + np.float32(rank) +
np.float32(1.0))
+ scores[row_id] = scores.get(row_id, np.float32(0.0)) + contribution
return _top_k(scores, self._limit)
def _mrr(self, route_results):
scores = {}
for route_result in route_results:
+ weight = np.float32(route_result.route.weight)
for rank, row_id in enumerate(_ranked_row_ids(route_result.result)):
- contribution = route_result.route.weight / (rank + 1.0)
- scores[row_id] = scores.get(row_id, 0.0) + contribution
+ contribution = weight / (np.float32(rank) + np.float32(1.0))
+ scores[row_id] = scores.get(row_id, np.float32(0.0)) + contribution
return _top_k(scores, self._limit)
@@ -186,7 +174,7 @@ def _weighted_score(self, route_results):
scores = {}
for route_result in route_results:
result = route_result.result
- weight = route_result.route.weight
+ weight = np.float32(route_result.route.weight)
score_getter = result.score_getter()
# Route score scales are heterogeneous (e.g. bounded vector similarity
@@ -195,7 +183,7 @@ def _weighted_score(self, route_results):
# weights -- not a route's numeric magnitude -- decide its influence on
# the fused score. This mirrors the Java HybridSearchRanker.
route_scores = {
- row_id: (score_getter(row_id) or 0.0)
+ row_id: np.float32(score_getter(row_id) or 0.0)
for row_id in result.results()
}
if not route_scores:
@@ -206,8 +194,8 @@ def _weighted_score(self, route_results):
# No spread within the route (single hit or all ties) carries no
# relative signal, so every hit maps to 1.0 rather than zeroed out.
normalized = (
- (raw - min_score) / score_range if score_range > 0.0 else 1.0)
- scores[row_id] = scores.get(row_id, 0.0) + weight * normalized
+ (raw - min_score) / score_range if score_range > 0.0 else
np.float32(1.0))
+ scores[row_id] = scores.get(row_id, np.float32(0.0)) + weight *
normalized
return _top_k(scores, self._limit)
@@ -215,7 +203,7 @@ def _ranked_row_ids(result):
score_getter = result.score_getter()
return sorted(
result.results(),
- key=lambda row_id: (-(score_getter(row_id) or 0.0), row_id))
+ key=lambda row_id: (-np.float32(score_getter(row_id) or 0.0), row_id))
def _top_k(scores, limit):
@@ -228,5 +216,5 @@ def _top_k(scores, limit):
heapq.heappush(heap, item)
elif item > heap[0]:
heapq.heapreplace(heap, item)
- top_scores = {-neg_row_id: score for score, neg_row_id in heap}
+ top_scores = {-neg_row_id: float(score) for score, neg_row_id in heap}
return DictBasedScoredIndexResult(top_scores)
diff --git a/paimon-python/pypaimon/table/source/hybrid_search_builder.py
b/paimon-python/pypaimon/table/source/hybrid_search_builder.py
index 043765dc0c..00aa492778 100644
--- a/paimon-python/pypaimon/table/source/hybrid_search_builder.py
+++ b/paimon-python/pypaimon/table/source/hybrid_search_builder.py
@@ -17,6 +17,7 @@
"""Builder to build hybrid search."""
+import ctypes
import math
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
@@ -75,6 +76,7 @@ class HybridSearchRoute:
"Query cannot be None for full-text route")
if self.limit <= 0:
raise ValueError("Limit must be positive, got: %s" % self.limit)
+ self.weight = ctypes.c_float(float(self.weight)).value
if not math.isfinite(self.weight) or self.weight <= 0:
raise ValueError(
"Weight must be finite and positive, got: %s" % self.weight)
@@ -134,7 +136,11 @@ class HybridSearchRouteResult:
class HybridSearchRouteBuilder:
- """A route plus its configured search builder."""
+ """A route retaining the search builder used by every execute_local() call.
+
+ Reconfigure search_builder to change subsequent executions of this route.
+ The builder keeps the read view captured by route_builders().
+ """
def __init__(self, route, search_builder):
self.route = route
@@ -225,15 +231,15 @@ class HybridSearchBuilder(ABC):
pass
def execute_local(self) -> ScoredGlobalIndexResult:
- from pypaimon.execution.python.read.search.hybrid_search import
execute_local
- return execute_local(self)
+ from pypaimon.table.source.search_dispatch import execute_hybrid_search
+ return execute_hybrid_search(self)
def execute(self, executor=None) -> ScoredGlobalIndexResult:
"""Execute complete routes with an optional search-task scheduler."""
if executor is None:
return self.execute_local()
- from pypaimon.execution.python.read.search.hybrid_search import execute
- return execute(self, executor)
+ from pypaimon.table.source.search_dispatch import execute_hybrid_search
+ return execute_hybrid_search(self, executor)
class HybridSearchBuilderImpl(HybridSearchBuilder):
@@ -292,18 +298,27 @@ class HybridSearchBuilderImpl(HybridSearchBuilder):
return self
def route_builders(self) -> List[HybridSearchRouteBuilder]:
- from pypaimon.execution.python.read.search.hybrid_search import
route_builders
- return route_builders(self)
+ from pypaimon.table.source.search_dispatch import hybrid_route_builders
+ return hybrid_route_builders(self)
def to_route_result(
self, route_builder: HybridSearchRouteBuilder,
result: GlobalIndexResult) -> HybridSearchRouteResult:
- from pypaimon.execution.python.read.search.hybrid_search import
to_route_result
- return to_route_result(self, route_builder, result)
+ # This is public result-type validation, independent of execution. Keep
+ # scored objects intact, including custom score getters and metadata.
+ if isinstance(result, ScoredGlobalIndexResult):
+ scored = result
+ elif result.results().is_empty():
+ scored = ScoredGlobalIndexResult.create_empty()
+ else:
+ raise ValueError(
+ "Hybrid search requires scored index results, but got: %s"
+ % result.__class__.__name__)
+ return HybridSearchRouteResult(route_builder.route, scored)
def rank(self, route_results: List[HybridSearchRouteResult]) ->
ScoredGlobalIndexResult:
- from pypaimon.execution.python.read.search.hybrid_search import rank
- return rank(self, route_results)
+ from pypaimon.table.source.search_dispatch import rank_hybrid_results
+ return rank_hybrid_results(self, route_results)
def _validate_search(self):
if not self._routes:
diff --git a/paimon-python/pypaimon/table/source/search_dispatch.py
b/paimon-python/pypaimon/table/source/search_dispatch.py
index b174450264..166d073b83 100644
--- a/paimon-python/pypaimon/table/source/search_dispatch.py
+++ b/paimon-python/pypaimon/table/source/search_dispatch.py
@@ -72,7 +72,8 @@ def python_search_reader(reader):
def _native_requested(builder):
- enabled = getattr(getattr(builder._table, 'options', None),
'native_read_enabled', None)
+ table = getattr(builder, '_table', None)
+ enabled = getattr(getattr(table, 'options', None), 'native_read_enabled',
None)
return callable(enabled) and enabled()
@@ -133,7 +134,10 @@ def execute_full_text_search(builder):
def execute_hybrid_search(builder, executor=None):
- if _native_requested(builder):
+ from pypaimon.table.source.hybrid_search_builder import
HybridSearchBuilderImpl
+ # Custom implementations execute through their public route/rank hooks;
+ # only the default implementation owns the native configuration protocol.
+ if isinstance(builder, HybridSearchBuilderImpl) and
_native_requested(builder):
from pypaimon.execution.native.read.native_hybrid_search import
try_native_hybrid_search
result = (try_native_hybrid_search(builder) if executor is None else
try_native_hybrid_search(builder, executor=executor))
@@ -141,3 +145,28 @@ def execute_hybrid_search(builder, executor=None):
return result
from pypaimon.execution.python.read.search.hybrid_search import execute
return execute(builder, executor)
+
+
+def hybrid_route_builders(builder):
+ from pypaimon.catalog.table_query_auth import
reject_search_under_query_auth
+ # Authorization precedes runtime discovery and snapshot resolution, as in
+ # Java's routeBuilders(). Selecting a backend must not hide a rejection.
+ reject_search_under_query_auth(builder._table)
+ if _native_requested(builder):
+ from pypaimon.execution.native.read.native_hybrid_search import
try_native_hybrid_routes
+ routes = try_native_hybrid_routes(builder)
+ if routes is not None:
+ return routes
+ from pypaimon.execution.python.read.search.hybrid_search import
route_builders
+ return route_builders(builder)
+
+
+def rank_hybrid_results(builder, route_results):
+ builder._validate_search()
+ if _native_requested(builder):
+ from pypaimon.execution.native.runtime import native_runtime_available
+ if native_runtime_available():
+ from pypaimon.execution.native.read.native_hybrid_search import
rank_hybrid_results as rank
+ return rank(builder, route_results)
+ from pypaimon.execution.python.read.search.hybrid_search import rank
+ return rank(builder, route_results)
diff --git a/paimon-python/pypaimon/tests/hybrid_search_phase_dispatch_test.py
b/paimon-python/pypaimon/tests/hybrid_search_phase_dispatch_test.py
new file mode 100644
index 0000000000..c9b358eaf4
--- /dev/null
+++ b/paimon-python/pypaimon/tests/hybrid_search_phase_dispatch_test.py
@@ -0,0 +1,198 @@
+# 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.
+
+"""Backend selection and public contracts for independently executed hybrid
phases."""
+
+import pickle
+from types import SimpleNamespace
+from unittest.mock import Mock, patch
+
+import pytest
+import pyarrow as pa
+
+from pypaimon import CatalogFactory, Schema
+from pypaimon.common.json_util import JSON
+from pypaimon.globalindex.global_index_result import GlobalIndexResult
+from pypaimon.globalindex.vector_search_result import
DictBasedScoredIndexResult, ScoredGlobalIndexResult
+from pypaimon.snapshot.snapshot import Snapshot
+from pypaimon.table.source.hybrid_search_builder import (
+ HybridSearchBuilder, HybridSearchBuilderImpl, HybridSearchRoute,
HybridSearchRouteBuilder,
+)
+
+
[email protected]
+def search_table(tmp_path):
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path / 'warehouse')})
+ catalog.create_database('default', True)
+ catalog.create_table('default.search', Schema.from_pyarrow_schema(
+ pa.schema([('embedding', pa.list_(pa.float32())), ('body',
pa.string())]),
+ options={'row-tracking.enabled': 'true', 'data-evolution.enabled':
'true',
+ 'read.native.enabled': 'true'}), False)
+ return catalog.get_table('default.search')
+
+
+def _builder(table):
+ return (HybridSearchBuilderImpl(table).with_limit(2)
+ .add_vector_route('embedding', [1.0, 0.0], 3,
options={'ivf.nprobe': '2'})
+ .add_full_text_route('body', 'query', 4, weight=2.0))
+
+
[email protected]('phase', ['routes', 'rank'])
+def test_native_phase_uses_dispatch_without_python_execution(search_table,
phase):
+ builder = _builder(search_table)
+ expected = object()
+ target = 'try_native_hybrid_routes' if phase == 'routes' else
'rank_hybrid_results'
+ with patch('pypaimon.execution.native.read.native_hybrid_search.' +
target, return_value=expected) as native, \
+
patch('pypaimon.execution.native.runtime.native_runtime_available',
return_value=True), \
+
patch('pypaimon.execution.python.read.search.hybrid_search.route_builders',
+ side_effect=AssertionError('Python route planning')), \
+ patch('pypaimon.execution.python.read.search.hybrid_search.rank',
+ side_effect=AssertionError('Python fusion')):
+ actual = builder.route_builders() if phase == 'routes' else
builder.rank([])
+ assert actual is expected
+ native.assert_called_once()
+
+
[email protected]('phase', ['routes', 'rank'])
+def test_native_phase_failure_is_not_replayed_in_python(search_table, phase):
+ failure = RuntimeError('selected Native phase failed')
+ target = 'try_native_hybrid_routes' if phase == 'routes' else
'rank_hybrid_results'
+ builder = _builder(search_table)
+ with patch('pypaimon.execution.native.read.native_hybrid_search.' +
target, side_effect=failure), \
+
patch('pypaimon.execution.native.runtime.native_runtime_available',
return_value=True), \
+
patch('pypaimon.execution.python.read.search.hybrid_search.route_builders') as
routes, \
+ patch('pypaimon.execution.python.read.search.hybrid_search.rank')
as rank:
+ with pytest.raises(RuntimeError) as caught:
+ builder.route_builders() if phase == 'routes' else builder.rank([])
+ assert caught.value is failure
+ routes.assert_not_called()
+ rank.assert_not_called()
+
+
+def
test_query_auth_rejects_route_capture_before_backend_discovery(search_table):
+ builder = _builder(search_table.copy({'query-auth.enabled': 'true'}))
+ with
patch('pypaimon.execution.native.read.native_hybrid_search.try_native_hybrid_routes')
as native, \
+
patch('pypaimon.execution.native.runtime.native_runtime_available') as
available:
+ with pytest.raises(ValueError, match='query-auth'):
+ builder.route_builders()
+ native.assert_not_called()
+ available.assert_not_called()
+
+
[email protected]('captured', [False, True])
+def
test_native_routes_keep_public_builders_and_captured_metadata(search_table,
captured):
+ from pypaimon.execution.native.read.native_hybrid_search import
try_native_hybrid_routes
+
+ builder = _builder(search_table)
+ predicate =
search_table.new_read_builder().new_predicate_builder().equal('body', 'keep')
+ builder.with_filter(predicate)
+ snapshot = Snapshot(3, 7, 0, 'base', 'delta', 2, 2, 'writer', 1, 'APPEND',
0) if captured else None
+ native_snapshot = None if snapshot is None else
SimpleNamespace(_to_json=lambda: JSON.to_json(snapshot))
+ native = Mock()
+ native.route_builders.return_value = [SimpleNamespace(snapshot=lambda:
native_snapshot) for _ in range(2)]
+ with
patch('pypaimon.execution.native.read.native_hybrid_search._native_builder',
+ return_value=(search_table, native)), \
+
patch('pypaimon.snapshot.time_travel_util.TimeTravelUtil.resolve_snapshot',
+ side_effect=AssertionError('Resolved a captured snapshot
again')):
+ routes = try_native_hybrid_routes(builder)
+ assert [route.route for route in routes] == builder._routes
+ assert all(route.search_builder._table._read_snapshot == snapshot for
route in routes)
+ assert routes[0].search_builder._vector_column.name == 'embedding'
+ assert routes[0].search_builder._query_vector == [1.0, 0.0]
+ assert routes[0].search_builder._options == {'ivf.nprobe': '2'}
+ assert routes[0].search_builder._limit == 3
+ assert routes[1].search_builder._limit == 4
+ assert all(route.search_builder._filter == predicate for route in routes)
+ assert callable(routes[0].search_builder.new_vector_search_scan)
+ assert callable(routes[1].search_builder.new_full_text_read)
+ restored = pickle.loads(pickle.dumps(routes))
+ assert [route.search_builder._table._read_snapshot for route in restored]
== [snapshot, snapshot]
+ assert '_read_snapshot' not in vars(search_table)
+ native.route_builders.assert_called_once_with()
+
+
+def test_result_conversion_is_shared_validation_without_selecting_a_backend():
+ builder = HybridSearchBuilderImpl(None) # No routes, valid limit, or
table is required.
+ route = HybridSearchRoute.full_text_route('unused', '', 1)
+ route_builder = HybridSearchRouteBuilder(route, None)
+ result = DictBasedScoredIndexResult({7: 0.5})
+ with patch('pypaimon.execution.native.runtime.native_runtime_available',
+ side_effect=AssertionError('Discovered an executor during
validation')):
+ converted = builder.to_route_result(route_builder, result)
+ empty = builder.to_route_result(route_builder,
GlobalIndexResult.create_empty())
+ with pytest.raises(ValueError, match='requires scored'):
+ builder.to_route_result(route_builder,
GlobalIndexResult.create(result.results()))
+ assert converted.route is route
+ assert converted.result is result
+ assert isinstance(empty.result, ScoredGlobalIndexResult)
+ assert empty.result.results().is_empty()
+
+
+def test_python_phases_do_not_load_native(search_table):
+ builder = _builder(search_table.copy({'read.native.enabled': 'false'}))
+ with
patch('pypaimon.execution.native.read.native_hybrid_search.try_native_hybrid_routes',
+ side_effect=AssertionError('Loaded disabled route backend')), \
+
patch('pypaimon.execution.native.read.native_hybrid_search.rank_hybrid_results',
+ side_effect=AssertionError('Loaded disabled fusion
backend')):
+ routes = builder.route_builders()
+ assert len(routes) == 2
+ ranked = builder.rank([builder.to_route_result(routes[0],
DictBasedScoredIndexResult({7: 1.0}))])
+ assert list(ranked.results()) == [7]
+
+
[email protected]('keep_table', [False, True])
[email protected]('entry_point', ['execute_local', 'execute'])
+def test_abstract_builder_executes_public_hooks(search_table, keep_table,
entry_point):
+ from pypaimon.execution.native.read import native_hybrid_search
+
+ result = DictBasedScoredIndexResult({7: 0.5})
+ calls = []
+
+ def execute_route():
+ calls.append('execute route')
+ return result
+
+ class CustomHybridSearch(HybridSearchBuilder):
+ def _configure(self, *args, **kwargs):
+ return self
+
+ with_limit = with_ranker = with_filter = with_partition_filter =
add_route = _configure
+
+ def route_builders(self):
+ calls.append('build routes')
+ return [HybridSearchRouteBuilder(None,
SimpleNamespace(execute_local=execute_route))]
+
+ def to_route_result(self, route, result):
+ calls.append('convert result')
+ return result
+
+ def rank(self, results):
+ calls.append('rank')
+ return results[0]
+
+ builder = CustomHybridSearch()
+ if keep_table:
+ # Keeping a native-enabled table does not opt a public implementation
+ # into HybridSearchBuilderImpl's private configuration protocol.
+ builder._table = search_table
+ with patch.object(native_hybrid_search, 'native_runtime_available',
return_value=True), \
+ patch.object(native_hybrid_search, '_rest_catalog_supported',
return_value=True), \
+ patch.object(native_hybrid_search, 'try_native_hybrid_search',
+ wraps=native_hybrid_search.try_native_hybrid_search)
as native:
+ assert getattr(builder, entry_point)() is result
+ native.assert_not_called()
+ assert calls == ['build routes', 'execute route', 'convert result', 'rank']
diff --git a/paimon-python/pypaimon/tests/hybrid_search_ranker_test.py
b/paimon-python/pypaimon/tests/hybrid_search_ranker_test.py
index 5c34c60c39..d98096856f 100644
--- a/paimon-python/pypaimon/tests/hybrid_search_ranker_test.py
+++ b/paimon-python/pypaimon/tests/hybrid_search_ranker_test.py
@@ -47,6 +47,57 @@ def _rank(limit, ranker, results):
class HybridSearchRankerTest(unittest.TestCase):
+ def test_rankers_compare_external_scores_as_java_floats(self):
+ # ScoreGetter returns float: these distinct Python values are equal in
+ # Java, so the smaller row ID wins every ranker's Top-K boundary tie.
+ route = _route_result(1.0, {1: 1000000000.0, 2: 1000000001.0})
+ for ranker, expected_score in (
+ (RRF_RANKER, 0.016393441706895828),
+ (WEIGHTED_SCORE_RANKER, 1.0),
+ (MRR_RANKER, 1.0)):
+ with self.subTest(ranker=ranker):
+ ranked = _rank(1, ranker, [route])
+ self.assertEqual({1}, set(ranked.results()))
+ self.assertEqual(expected_score, ranked.score_getter()(1))
+
+ def test_route_validates_weights_after_java_float_conversion(self):
+ for weight in (1e39, 1e-46, 3.4028236e38):
+ for full_text in (False, True):
+ with self.subTest(weight=weight, full_text=full_text):
+ with self.assertRaisesRegex(ValueError, "Weight must be
finite and positive"):
+ if full_text:
+ HybridSearchRoute.full_text_route('text', '{}', 1,
weight)
+ else:
+ HybridSearchRoute.vector_route('embedding', [1.0],
1, weight)
+
+ for weight, expected in (
+ (0.1, 0.10000000149011612),
+ (1e-45, 1.401298464324817e-45),
+ (3.4028235e38, 3.4028234663852886e38)):
+ with self.subTest(weight=weight):
+ vector = HybridSearchRoute.vector_route('embedding', [1.0], 1,
weight)
+ text = HybridSearchRoute.full_text_route('text', '{}', 1,
weight)
+ self.assertEqual(expected, vector.weight)
+ self.assertEqual(expected, text.weight)
+
+ def test_rankers_round_normalization_and_accumulation_as_java_floats(self):
+ routes = [_route_result(0.1, {1: 0.0, 2: 0.1, 3: 0.3}),
+ _route_result(0.2, {1: 0.3, 2: 0.2, 3: 0.1}),
+ _route_result(0.3, {1: 0.1, 2: 0.2, 3: 0.3})]
+ # Java float operations, including each per-route addition, determine
+ # these exact scores; rounding just the final Python sum is different.
+ for ranker, expected in (
+ (RRF_RANKER, {1: 0.009627895429730415, 2: 0.009677419438958168,
+ 3: 0.009731980040669441}),
+ (WEIGHTED_SCORE_RANKER, {1: 0.20000000298023224, 2:
0.28333330154418945,
+ 3: 0.4000000059604645}),
+ (MRR_RANKER, {1: 0.3333333432674408, 2: 0.30000001192092896,
+ 3: 0.46666669845581055})):
+ with self.subTest(ranker=ranker):
+ ranked = _rank(3, ranker, routes)
+ self.assertEqual(expected, {row_id:
ranked.score_getter()(row_id)
+ for row_id in ranked.results()})
+
def test_weighted_score_normalizes_heterogeneous_route_scales(self):
# Vector-like route (bounded ~[0, 1]) and BM25-like route (unbounded,
larger).
vector_route = _route_result(5.0, {1: 0.95, 2: 0.10})
diff --git a/paimon-python/pypaimon/tests/native_hybrid_search_test.py
b/paimon-python/pypaimon/tests/native_hybrid_search_test.py
index 2ecd100e81..b1ad029955 100644
--- a/paimon-python/pypaimon/tests/native_hybrid_search_test.py
+++ b/paimon-python/pypaimon/tests/native_hybrid_search_test.py
@@ -17,7 +17,7 @@
"""Whole-operation native hybrid execution against REST, vector and text
indexes."""
-from contextlib import contextmanager
+from contextlib import ExitStack, contextmanager
from unittest.mock import patch
import pyarrow as pa
@@ -31,6 +31,8 @@ from pypaimon.execution.native.read.builder import
(_predicate_to_native)
from pypaimon.snapshot.table_snapshot import TableSnapshot
from pypaimon.snapshot.time_travel_util import TimeTravelUtil
from pypaimon.table.source.hybrid_search_builder import HybridSearchBuilderImpl
+from pypaimon.globalindex.vector_search_result import
DictBasedScoredIndexResult, ScoredGlobalIndexResult
+from pypaimon.table.source.hybrid_search_builder import HybridSearchRoute,
HybridSearchRouteBuilder
from pypaimon.tests import native_sorted_index_build_test as indexes
from pypaimon.tests.native_full_text_search_test import _scores
from pypaimon.tests.vector_search_filter_test import match_query, boost_query
@@ -269,10 +271,151 @@ def
test_manual_route_and_rank_api_remains_available(rest_catalog, native):
catalog, _ = rest_catalog
table = _table(catalog, {'read.native.enabled': str(native).lower()})
builder = _builder(table)
+ with ExitStack() as stack:
+ if native:
+ for phase in ('route_builders', 'rank'):
+
stack.enter_context(patch('pypaimon.execution.python.read.search.hybrid_search.'
+ phase,
+ side_effect=AssertionError('Python
hybrid ' + phase)))
+ routes = builder.route_builders()
+ results = [builder.to_route_result(route, route.execute_local()) for
route in routes]
+ assert _scores(builder.rank(results)) ==
pytest.approx(_scores(builder.execute_local()), abs=1e-6)
+ assert indexes._ids(table) == list(range(8))
+
+
[email protected]('ranker', ['rrf', 'weighted_score', 'mrr'])
+def
test_rank_external_results_does_not_access_table_or_validate_unused_queries(rest_catalog,
ranker):
+ catalog, _ = rest_catalog
+ table = _table(catalog, indexed=False).copy({'query-auth.enabled': 'true'})
+ builder = _builder(table, ranker=ranker, limit=2)
+ # Route metadata is not a request to execute these queries. Both empty
+ # vectors and empty text queries are valid public route descriptions.
+ routes =
[HybridSearchRouteBuilder(HybridSearchRoute.vector_route('absent', [], 10,
weight=2.0), None),
+
HybridSearchRouteBuilder(HybridSearchRoute.full_text_route('absent', '', 10),
None)]
+ results = [builder.to_route_result(routes[0],
DictBasedScoredIndexResult({1: 10., 2: None, 3: 5.})),
+ builder.to_route_result(routes[1],
DictBasedScoredIndexResult({2: 50., 3: 50.}))]
+ classic = _builder(table.copy({'read.native.enabled': 'false'}),
ranker=ranker, limit=1)
+ with
patch('pypaimon.execution.native.read.native_hybrid_search._native_table',
+ side_effect=AssertionError('Rank accessed table metadata')), \
+ patch('pypaimon.execution.python.read.search.hybrid_search.rank',
+ side_effect=AssertionError('Native rank used Python
fusion')):
+ actual = _scores(builder.rank(results))
+ assert _scores(builder.rank([])) == {}
+ builder.with_limit(1)
+ reduced = _scores(builder.rank([results[1], results[0], results[0]]))
+ expected = classic.with_limit(2).rank(results)
+ assert actual == pytest.approx(_scores(expected), abs=1e-6)
+ classic.with_limit(1)
+ assert reduced == pytest.approx(_scores(classic.rank([results[1],
results[0], results[0]])), abs=1e-6)
+ assert all(isinstance(result.result, ScoredGlobalIndexResult) for result
in results)
+
+
+def
test_manual_routes_pin_one_snapshot_and_keep_public_scan_read_api(rest_catalog):
+ catalog, server = rest_catalog
+ table = _table(catalog)
+ builder = _builder(table)
+ with patch.object(server, '_table_snapshot_handle',
wraps=server._table_snapshot_handle) as latest, \
+
patch('pypaimon.execution.python.read.search.hybrid_search.route_builders',
+ side_effect=AssertionError('Python route capture')):
+ routes = builder.route_builders()
+ assert latest.call_count == 1
+ snapshots = [route.search_builder._table._read_snapshot for route in
routes]
+ assert snapshots[0].id == snapshots[1].id
+ indexes._append(table, _rows(8, 4), schema=SCHEMA)
+ with patch.object(server, '_table_snapshot_handle',
side_effect=AssertionError('Re-resolved captured view')):
+ vector = routes[0].search_builder
+ text = routes[1].search_builder
+ vector_result =
vector.new_vector_search_read().read_plan(vector.new_vector_search_scan().scan())
+ text_result =
text.new_full_text_read().read_plan(text.new_full_text_scan().scan())
+ results = [builder.to_route_result(route, result)
+ for route, result in zip(routes, [vector_result,
text_result])]
+ ranked =
_scores(builder.with_limit(4).with_weighted_score_ranker().rank(results))
+ assert ranked
+ assert set(ranked).issubset(set(range(8)))
+ assert 11 in _scores(_builder(table).execute_local())
+
+
+def test_manual_empty_routes_remain_empty_after_first_commit(rest_catalog):
+ catalog, server = rest_catalog
+ table = indexes._create(catalog, schema=SCHEMA,
options={'read.native.enabled': 'true'})
+ builder = _builder(table)
routes = builder.route_builders()
+ assert all(route.search_builder._table._read_snapshot is None for route in
routes)
+ indexes._append(table, _rows(), schema=SCHEMA)
+ with patch.object(server, '_table_snapshot_handle',
side_effect=AssertionError('Re-resolved empty view')):
+ results = [builder.to_route_result(route, route.execute_local()) for
route in routes]
+ assert _scores(builder.rank(results)) == {}
+ assert _scores(_builder(table).execute_local())
+
+
+def test_manual_routes_allow_completing_queries_after_capture(rest_catalog):
+ catalog, _ = rest_catalog
+ table = _table(catalog)
+ builder = (table.new_hybrid_search_builder().with_limit(16)
+ .add_vector_route('embedding', [], 16)
+ .add_full_text_route('content', '', 16))
+ with
patch('pypaimon.execution.python.read.search.hybrid_search.route_builders',
+ side_effect=AssertionError('Python incomplete route capture')):
+ routes = builder.route_builders()
+ indexes._append(table, _rows(8, 4), schema=SCHEMA)
+ routes[0].search_builder.with_query_vector([0., 0.])
+ routes[1].search_builder.with_query('content', match_query('paimon'))
results = [builder.to_route_result(route, route.execute_local()) for route
in routes]
- assert _scores(builder.rank(results)) ==
pytest.approx(_scores(builder.execute_local()), abs=1e-6)
- assert indexes._ids(table) == list(range(8))
+ actual = _scores(builder.rank(results))
+ assert actual
+ assert set(actual).issubset(set(range(8)))
+
+
+def
test_sdk_route_mutations_affect_route_execute_and_keep_the_captured_view(rest_catalog):
+ catalog, _ = rest_catalog
+ table = _table(catalog, indexed=False)
+ table.create_global_index('content', index_type='full-text')
+ native = _native_table(table).new_hybrid_search_builder()
+ native.add_vector_route('embedding', [0., 0.], 16)
+ native.add_full_text_route('content', match_query('paimon'),
16).with_limit(16)
+ vector_route, text_route = native.route_builders()
+ vector, text = vector_route.search_builder(), text_route.search_builder()
+ assert vector is vector_route.search_builder()
+ assert text is text_route.search_builder()
+
+ # A commit after route capture must not enter either reconfigured query.
+ indexes._append(table, _rows(8, 4), schema=SCHEMA)
+ pb = table.new_read_builder().new_predicate_builder()
+ keep = _predicate_to_native(pb.greater_than('id', 4))
+ vector.with_query_vector([6., 0.]).with_limit(1).with_filter(keep)
+ text.with_query('content',
match_query('lake')).with_limit(1).with_filter(keep)
+ vector_result, text_result = vector_route.execute(), text_route.execute()
+ assert set(vector_result.row_ids()) == {6}
+ assert set(text_result.row_ids()) == {5}
+ assert vector_result.row_ids() == vector.execute().row_ids()
+ assert text_result.row_ids() == text.execute().row_ids()
+ assert vector_result.snapshot_id() == vector_route.snapshot().id()
+ assert text_result.snapshot_id() == text_route.snapshot().id()
+
+ # Repeated getters share the same state, including later reconfiguration.
+ vector_route.search_builder().with_limit(2)
+ assert set(vector_route.execute().row_ids()) == {5, 6}
+ ranked = native.rank([native.to_route_result(vector_route, vector_result),
+ native.to_route_result(text_route, text_result)])
+ batches = ranked.new_read_builder().with_projection(['id']).read()
+ assert sorted(pa.Table.from_batches(batches)['id'].to_pylist()) == [5, 6]
+
+
[email protected]('native', [False, True])
+def test_public_route_uses_mutated_search_builder(rest_catalog, native):
+ catalog, _ = rest_catalog
+ table = _table(catalog, {'read.native.enabled': str(native).lower()},
indexed=False)
+ table.create_global_index('content', index_type='full-text')
+ builder = _builder(table)
+ vector_route, text_route = builder.route_builders()
+ vector, text = vector_route.search_builder, text_route.search_builder
+ indexes._append(table, _rows(8, 4), schema=SCHEMA)
+ pb = table.new_read_builder().new_predicate_builder()
+ vector.with_query_vector([6.,
0.]).with_limit(1).with_filter(pb.greater_than('id', 4))
+ text.with_query('content',
match_query('lake')).with_limit(1).with_filter(pb.greater_than('id', 4))
+ assert set(_scores(vector_route.execute_local())) == {6}
+ assert set(_scores(text_route.execute_local())) == {5}
+ vector.with_limit(2)
+ assert set(_scores(vector_route.execute_local())) == {5, 6}
def _multimodal_query(table, catalog, **kwargs):
