[core] Expose phased hybrid search and pure score fusion

bindings/python/README.md

@@ -106,6 +106,25 @@ results expose physical positions and Java-compatible scored split bytes;
 `new_read_builder()` reads both result types from their source snapshot and
 returns projected columns plus `__paimon_search_score` in relevance order.
 
+Hybrid searches return the same `GlobalIndexResult` from `execute()`. To run
+routes separately, call `route_builders()` once and configure each route through
+its `search_builder()`:
+
+```python
+hybrid = (table.new_hybrid_search_builder()
+          .add_vector_route("embedding", [1.0, 0.0], 10)
+          .with_limit(5))
+route = hybrid.route_builders()[0]
+route.search_builder().with_query_vector([0.0, 1.0]).with_limit(3)
+result = route.execute()
+fused = hybrid.rank([hybrid.to_route_result(route, result)])
+```
+
+Each route returns the same live search builder on repeated calls. Query,
+limit, option and filter changes affect later `route.execute()` calls. All
+returned routes keep the snapshot captured by `route_builders()`, including
+a captured empty table, even if later writes publish a newer snapshot.
+
 ### Native Read / Write
 
 Beyond SQL, you can use the lower-level read and write APIs directly from Python.

bindings/python/project-description.md

@@ -128,6 +128,47 @@ when_not_matched)` with `WhenNotMatched` objects. The factories compile conditio
 once and return an executor with `execute(Table | RecordBatch) -> Table`.
 The calling engine owns joins, authorization and table writes.
 
+### Hybrid Search Phases
+
+Hybrid queries can execute routes separately and then fuse their results:
+
+```python
+hybrid = (table.new_hybrid_search_builder()
+          .add_vector_route("embedding", [1.0, 0.0], 20)
+          .with_limit(10))
+routes = hybrid.route_builders()
+results = [hybrid.to_route_result(route, route.search_builder().execute())
+           for route in routes]
+ranked = hybrid.rank(results)
+```
+
+`route_builders()` checks read authorization and captures one snapshot for all
+routes, including an empty view. Each route retains its search builder: repeated
+`search_builder()` calls return the same object, and configuring it changes later
+`route.execute()` calls. Returned builders retain their snapshot after later
+commits. `to_route_result()` accepts scored
+global results and normalizes empty unscored selections; nonempty unscored or
+physical primary-key selections cannot participate in global row-ID fusion.
+
+For externally evaluated scores, `HybridSearchWeightedResult` and
+`HybridSearchRanker.RRF`, `.WEIGHTED_SCORE`, or `.MRR` provide the same fusion
+without a table or query:
+
+```python
+from pypaimon_rust.datafusion import (
+    GlobalIndexResult, HybridSearchRanker, HybridSearchWeightedResult,
+)
+
+hits = GlobalIndexResult.from_scored_row_ids({7: 0.9, 8: 0.5})
+ranked = HybridSearchRanker.MRR.rank([HybridSearchWeightedResult(hits, 2.0)], 1)
+assert ranked.row_ids() == {7: 2.0}
+```
+
+Rankers use Java's float32 scores, weights and arithmetic. A shared captured read
+context is retained only when every nonempty input belongs to the same table
+view. Fusing external scores or different views returns scores without attaching
+an unrelated table or snapshot.
+
 ### Multimodal SQL Helpers
 
 `SQLContext` also registers Python scalar UDFs for common BLOB media and vector

bindings/python/python/pypaimon_rust/datafusion.pyi

@@ -18,7 +18,7 @@
 from datetime import datetime
 from os import PathLike
 from typing import (
-    Any, Callable, Dict, Iterable, List, Literal, Mapping, Optional, Sequence, Tuple,
+    Any, Callable, ClassVar, Dict, Iterable, List, Literal, Mapping, Optional, Sequence, Tuple,
     TypeAlias, Union,
 )
 
@@ -1140,6 +1140,52 @@ class GlobalIndexResultReadBuilder:
         """Read user columns and scores in relevance order from the search snapshot."""
         ...
 
+class HybridSearchWeightedResult:
+    def __init__(self, result: GlobalIndexResult, weight: float = 1.0) -> None:
+        """Capture a scored global result and a finite positive fusion weight."""
+        ...
+    def result(self) -> GlobalIndexResult: ...
+    def weight(self) -> float: ...
+
+class HybridSearchRanker:
+    RRF: ClassVar[HybridSearchRanker]
+    WEIGHTED_SCORE: ClassVar[HybridSearchRanker]
+    MRR: ClassVar[HybridSearchRanker]
+    def rank(self, results: Sequence[HybridSearchWeightedResult], limit: int) -> GlobalIndexResult:
+        """Fuse evaluated results without table access or authorization calls."""
+        ...
+
+class HybridSearchRoute:
+    @staticmethod
+    def vector(field_name: str, vector: Sequence[float], limit: int,
+               weight: float = 1.0, options: Optional[Dict[str, str]] = None) -> HybridSearchRoute: ...
+    @staticmethod
+    def full_text(field_name: str, query: str, limit: int,
+                  weight: float = 1.0, options: Optional[Dict[str, str]] = None) -> HybridSearchRoute: ...
+    def kind(self) -> Literal["vector", "full_text"]: ...
+    def field_name(self) -> str: ...
+    def vector_value(self) -> Optional[List[float]]: ...
+    def full_text_query(self) -> Optional[str]: ...
+    def limit(self) -> int: ...
+    def weight(self) -> float: ...
+    def options(self) -> Dict[str, str]: ...
+
+class HybridSearchRouteBuilder:
+    def route(self) -> HybridSearchRoute: ...
+    def snapshot(self) -> Optional[Snapshot]: ...
+    def search_builder(self) -> Union[VectorSearchBuilder, FullTextSearchBuilder]:
+        """Return this route's live builder; mutations affect subsequent route execution.
+
+        Repeated calls return the same object on the captured snapshot.
+        """
+        ...
+    def execute(self) -> GlobalIndexResult: ...
+
+class HybridSearchRouteResult:
+    def __init__(self, route: HybridSearchRoute, result: GlobalIndexResult) -> None: ...
+    def route(self) -> HybridSearchRoute: ...
+    def result(self) -> GlobalIndexResult: ...
+
 class HybridSearchBuilder:
     def with_limit(self, limit: int) -> HybridSearchBuilder: ...
     def with_snapshot(self, snapshot: Optional[Snapshot]) -> HybridSearchBuilder:
@@ -1154,6 +1200,12 @@ class HybridSearchBuilder:
                          weight: float = 1.0, options: Optional[Dict[str, str]] = None) -> HybridSearchBuilder: ...
     def add_full_text_route(self, field_name: str, query: str, limit: int,
                             weight: float = 1.0, options: Optional[Dict[str, str]] = None) -> HybridSearchBuilder: ...
+    def route_builders(self) -> List[HybridSearchRouteBuilder]:
+        """Capture one snapshot and return configured routes with live search builders."""
+        ...
+    def to_route_result(self, route_builder: HybridSearchRouteBuilder,
+                        result: GlobalIndexResult) -> HybridSearchRouteResult: ...
+    def rank(self, route_results: Sequence[HybridSearchRouteResult]) -> GlobalIndexResult: ...
     def execute(self) -> GlobalIndexResult: ...
 
 class FullTextSearchBuilder:

bindings/python/src/context.rs

@@ -1001,6 +1001,11 @@ pub fn register_module(py: Python<'_>, m: &Bound<'_, PyModule>) -> PyResult<()>
     this.add_class::<crate::full_text_search::PyFullTextScanPlan>()?;
     this.add_class::<crate::full_text_search::PyFullTextRead>()?;
     this.add_class::<crate::hybrid_search::PyHybridSearchBuilder>()?;
+    this.add_class::<crate::hybrid_search::PyHybridSearchRoute>()?;
+    this.add_class::<crate::hybrid_search::PyHybridSearchRouteBuilder>()?;
+    this.add_class::<crate::hybrid_search::PyHybridSearchRouteResult>()?;
+    this.add_class::<crate::hybrid_search::PyHybridSearchWeightedResult>()?;
+    this.add_class::<crate::hybrid_search::PyHybridSearchRanker>()?;
     this.add_class::<crate::vector_search::PyVectorSearchBuilder>()?;
     this.add_class::<crate::vindex::PyVindexVectorGlobalIndexReader>()?;
     this.add_class::<crate::vector_search::PyBatchVectorSearchBuilder>()?;

bindings/python/src/full_text_search.rs

@@ -22,51 +22,32 @@ use crate::global_index_result::PyGlobalIndexResult;
 use crate::predicate::dict_to_table_predicate;
 use crate::search_plan::{full_text_splits_from_python, full_text_splits_to_python};
 use crate::snapshot::PySnapshot;
-use paimon::spec::Predicate;
-use paimon::table::{FullTextRead, FullTextScan, FullTextScanPlan, Table};
+use paimon::table::{
+    FullTextRead, FullTextScan, FullTextScanPlan, FullTextSearchBuilder, HybridSearchRouteBuilder,
+    Table,
+};
 use paimon_datafusion::runtime::runtime;
 use pyo3::prelude::*;
 use pyo3::types::{PyDict, PyTuple};
 use std::sync::Arc;
 
-#[derive(Default)]
-struct Config {
-    query: Option<(String, String)>,
-    limit: Option<usize>,
-    filters: Vec<Predicate>,
-}
-
-impl Config {
-    fn builder<'a>(
-        &self,
-        table: &'a Table,
-    ) -> paimon::Result<paimon::table::FullTextSearchBuilder<'a>> {
-        let mut builder = table.new_full_text_search_builder();
-        if let Some((column, query)) = &self.query {
-            builder.with_query(column, query);
-        }
-        if let Some(limit) = self.limit {
-            builder.with_limit(limit);
-        }
-        for filter in &self.filters {
-            builder.with_filter(filter.clone());
-        }
-        Ok(builder)
-    }
-}
-
 #[pyclass(name = "FullTextSearchBuilder", module = "pypaimon_rust.datafusion")]
 pub struct PyFullTextSearchBuilder {
     table: Arc<Table>,
-    config: Config,
+    inner: FullTextSearchBuilder<'static>,
 }
 
 impl PyFullTextSearchBuilder {
     pub(crate) fn new(table: Arc<Table>) -> Self {
-        Self {
-            table,
-            config: Config::default(),
-        }
+        let inner = table.new_full_text_search_builder().into_owned();
+        Self { table, inner }
+    }
+
+    pub(crate) fn from_hybrid_route(route: &HybridSearchRouteBuilder) -> PyResult<Self> {
+        Ok(Self {
+            table: Arc::clone(route.table()),
+            inner: route.full_text_search_builder().map_err(to_py_err)?,
+        })
     }
 }
 
@@ -77,57 +58,41 @@ impl PyFullTextSearchBuilder {
         field_name: String,
         query: String,
     ) -> PyRefMut<'_, Self> {
-        slf.config.query = Some((field_name, query));
+        slf.inner.with_query(&field_name, &query);
         slf
     }
     fn with_limit(mut slf: PyRefMut<'_, Self>, limit: usize) -> PyRefMut<'_, Self> {
-        slf.config.limit = Some(limit);
+        slf.inner.with_limit(limit);
         slf
     }
     fn with_filter<'py>(
         mut slf: PyRefMut<'py, Self>,
         predicate: &Bound<'_, PyDict>,
     ) -> PyResult<PyRefMut<'py, Self>> {
         let filter = dict_to_table_predicate(predicate, slf.table.schema(), true)?;
-        slf.config.filters.push(filter);
+        slf.inner.with_filter(filter);
         Ok(slf)
     }
     fn with_partition_filter<'py>(
         mut slf: PyRefMut<'py, Self>,
         predicate: &Bound<'_, PyDict>,
     ) -> PyResult<PyRefMut<'py, Self>> {
         let filter = dict_to_table_predicate(predicate, slf.table.schema(), true)?;
-        slf.config
-            .builder(&slf.table)
-            .map_err(to_py_err)?
-            .with_partition_filter(filter.clone())
-            .map_err(to_py_err)?;
-        slf.config.filters.push(filter);
+        slf.inner.with_partition_filter(filter).map_err(to_py_err)?;
         Ok(slf)
     }
     fn new_full_text_scan(&self) -> PyResult<PyFullTextScan> {
         Ok(PyFullTextScan {
-            inner: self
-                .config
-                .builder(&self.table)
-                .map_err(to_py_err)?
-                .new_scan()
-                .map_err(to_py_err)?,
+            inner: self.inner.new_scan().map_err(to_py_err)?,
         })
     }
     fn new_full_text_read(&self) -> PyResult<PyFullTextRead> {
         Ok(PyFullTextRead {
-            inner: self
-                .config
-                .builder(&self.table)
-                .map_err(to_py_err)?
-                .new_read()
-                .map_err(to_py_err)?,
+            inner: self.inner.new_read().map_err(to_py_err)?,
         })
     }
     fn execute(&self, py: Python<'_>) -> PyResult<PyGlobalIndexResult> {
-        let builder = self.config.builder(&self.table).map_err(to_py_err)?;
-        py.detach(|| runtime().block_on(builder.execute_scored()))
+        py.detach(|| runtime().block_on(self.inner.execute()))
             .map(PyGlobalIndexResult::new)
             .map_err(to_py_err)
     }