From e2d03fb7689b7e0e15ffa5335b516dbf2a550d15 Mon Sep 17 00:00:00 2001
From: JingsongLi <jingsonglee0@gmail.com>
Date: Fri, 23 Oct 2026 13:44:33 +0800
Subject: [PATCH 1/3] [Python] Dispatch hybrid search phases to native
 execution

---
 .../global-index/hybrid-search.mdx            |  23 +++
 .../native/read/native_hybrid_search.py       |  67 ++++++-
 .../python/read/search/hybrid_search.py       |  42 ++---
 .../table/source/hybrid_search_builder.py     |  31 ++-
 .../pypaimon/table/source/search_dispatch.py  |  28 ++-
 .../hybrid_search_phase_dispatch_test.py      | 177 ++++++++++++++++++
 .../tests/hybrid_search_ranker_test.py        |  51 +++++
 .../tests/native_hybrid_search_test.py        |  96 +++++++++-
 8 files changed, 471 insertions(+), 44 deletions(-)
 create mode 100644 paimon-python/pypaimon/tests/hybrid_search_phase_dispatch_test.py

diff --git a/docs/docs/multimodal-table/global-index/hybrid-search.mdx b/docs/docs/multimodal-table/global-index/hybrid-search.mdx
index 81bfae13e091..e57f825145b2 100644
--- a/docs/docs/multimodal-table/global-index/hybrid-search.mdx
+++ b/docs/docs/multimodal-table/global-index/hybrid-search.mdx
@@ -265,6 +265,29 @@ 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` remains configurable and 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 660c59593a5e..56230dd9cf13 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.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 b53d7cd2314f..0183c67c6af3 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 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,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 043765dc0ce2..d07ca7d8ef61 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 @@ 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)
@@ -225,15 +227,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 +294,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:
diff --git a/paimon-python/pypaimon/table/source/search_dispatch.py b/paimon-python/pypaimon/table/source/search_dispatch.py
index b17445026494..ba14b8dd38f1 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()
 
 
@@ -141,3 +142,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 000000000000..73e5e4494107
--- /dev/null
+++ b/paimon-python/pypaimon/tests/hybrid_search_phase_dispatch_test.py
@@ -0,0 +1,177 @@
+# 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,
+)
+
+
+@pytest.fixture
+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))
+
+
+@pytest.mark.parametrize('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()
+
+
+@pytest.mark.parametrize('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()
+
+
+@pytest.mark.parametrize('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]
+
+
+def test_abstract_builder_executes_public_hooks_without_private_table_state():
+    result = DictBasedScoredIndexResult({7: 0.5})
+
+    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):
+            return [HybridSearchRouteBuilder(None, SimpleNamespace(execute_local=lambda: result))]
+
+        def to_route_result(self, route, result):
+            return result
+
+        def rank(self, results):
+            return results[0]
+
+    builder = CustomHybridSearch()
+    assert builder.execute_local() is result
+    assert builder.execute() is result
diff --git a/paimon-python/pypaimon/tests/hybrid_search_ranker_test.py b/paimon-python/pypaimon/tests/hybrid_search_ranker_test.py
index 5c34c60c39e5..d98096856f55 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 2ecd100e8168..cffa7fa76e1a 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.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,98 @@ 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))
+
+
+@pytest.mark.parametrize('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 _multimodal_query(table, catalog, **kwargs):

From ab1d2dbdf27200688c602a12c54e0d82aff9887f Mon Sep 17 00:00:00 2001
From: JingsongLi <jingsonglee0@gmail.com>
Date: Fri, 23 Oct 2026 14:24:46 +0800
Subject: [PATCH 2/3] [Python] Verify retained hybrid route builder mutations

---
 .../global-index/hybrid-search.mdx            |  3 +-
 .../table/source/hybrid_search_builder.py     |  6 ++-
 .../tests/native_hybrid_search_test.py        | 53 +++++++++++++++++++
 3 files changed, 60 insertions(+), 2 deletions(-)

diff --git a/docs/docs/multimodal-table/global-index/hybrid-search.mdx b/docs/docs/multimodal-table/global-index/hybrid-search.mdx
index e57f825145b2..94625ab91711 100644
--- a/docs/docs/multimodal-table/global-index/hybrid-search.mdx
+++ b/docs/docs/multimodal-table/global-index/hybrid-search.mdx
@@ -282,7 +282,8 @@ result = builder.rank(route_results)
 ```
 
 Route capture checks authorization and fixes one snapshot for all routes. Each
-`route.search_builder` remains configurable and supports its ordinary scan/read
+`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
diff --git a/paimon-python/pypaimon/table/source/hybrid_search_builder.py b/paimon-python/pypaimon/table/source/hybrid_search_builder.py
index d07ca7d8ef61..00aa492778b0 100644
--- a/paimon-python/pypaimon/table/source/hybrid_search_builder.py
+++ b/paimon-python/pypaimon/table/source/hybrid_search_builder.py
@@ -136,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
diff --git a/paimon-python/pypaimon/tests/native_hybrid_search_test.py b/paimon-python/pypaimon/tests/native_hybrid_search_test.py
index cffa7fa76e1a..b1ad02995546 100644
--- a/paimon-python/pypaimon/tests/native_hybrid_search_test.py
+++ b/paimon-python/pypaimon/tests/native_hybrid_search_test.py
@@ -365,6 +365,59 @@ def test_manual_routes_allow_completing_queries_after_capture(rest_catalog):
     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]
+
+
+@pytest.mark.parametrize('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):
     return (MultimodalTable(catalog, 'default.indexes', table).search_hybrid([
         vector_route('embedding', [0., 0.]), text_route('paimon', column='content')], **kwargs)

From ddb84c4ad1c4395ffd0fd1bac2366b20be2a588f Mon Sep 17 00:00:00 2001
From: JingsongLi <jingsonglee0@gmail.com>
Date: Fri, 23 Oct 2026 14:43:25 +0800
Subject: [PATCH 3/3] [Python] Preserve custom hybrid builder execution hooks

---
 .../pypaimon/table/source/search_dispatch.py  |  5 +++-
 .../hybrid_search_phase_dispatch_test.py      | 29 ++++++++++++++++---
 2 files changed, 29 insertions(+), 5 deletions(-)

diff --git a/paimon-python/pypaimon/table/source/search_dispatch.py b/paimon-python/pypaimon/table/source/search_dispatch.py
index ba14b8dd38f1..166d073b83fa 100644
--- a/paimon-python/pypaimon/table/source/search_dispatch.py
+++ b/paimon-python/pypaimon/table/source/search_dispatch.py
@@ -134,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))
diff --git a/paimon-python/pypaimon/tests/hybrid_search_phase_dispatch_test.py b/paimon-python/pypaimon/tests/hybrid_search_phase_dispatch_test.py
index 73e5e4494107..c9b358eaf49e 100644
--- a/paimon-python/pypaimon/tests/hybrid_search_phase_dispatch_test.py
+++ b/paimon-python/pypaimon/tests/hybrid_search_phase_dispatch_test.py
@@ -154,8 +154,17 @@ def test_python_phases_do_not_load_native(search_table):
     assert list(ranked.results()) == [7]
 
 
-def test_abstract_builder_executes_public_hooks_without_private_table_state():
+@pytest.mark.parametrize('keep_table', [False, True])
+@pytest.mark.parametrize('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):
@@ -164,14 +173,26 @@ def _configure(self, *args, **kwargs):
         with_limit = with_ranker = with_filter = with_partition_filter = add_route = _configure
 
         def route_builders(self):
-            return [HybridSearchRouteBuilder(None, SimpleNamespace(execute_local=lambda: result))]
+            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()
-    assert builder.execute_local() is result
-    assert builder.execute() is result
+    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']