[Python] Dispatch hybrid search phases to native execution

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

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.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())

paimon-python/pypaimon/execution/python/read/search/hybrid_search.py

@@ -23,8 +23,9 @@
 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 @@
 
 
 _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,26 +153,28 @@ 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)
 
 
 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,16 +194,16 @@ 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)
 
 
 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)

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 @@ def __post_init__(self):
                 "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 @@ def rank(
         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 @@ def add_route(self, route: HybridSearchRoute) -> '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:

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,11 +134,39 @@ 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))
         if result is not 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)