Support supplied global-index results in batch scans

bindings/c/src/tests.rs

@@ -3118,7 +3118,7 @@ fn test_two_commits_same_builder() {
 //
 // Two storage shapes are exercised end-to-end through the C `execute_read`
 // terminal, each compared against an independent core Rust
-// `SearchResultReadBuilder::read()` reference:
+// `GlobalIndexResultReadBuilder::read()` reference:
 //
 //   * A primary-key vector table backed by a real vindex IVF-flat ANN segment
 //     built in-process (bucket-local ANN search, residual filter supported).

bindings/c/src/vector_search.rs

@@ -31,8 +31,8 @@ use std::collections::HashMap;
 use std::ffi::{c_char, c_void};
 
 use paimon::spec::Predicate;
+use paimon::table::GlobalIndexResult;
 use paimon::table::{ArrowRecordBatchStream, Table, VectorSearchBuilder};
-use paimon::vector_search::SearchResult;
 
 use crate::error::{check_non_null, paimon_error, validate_cstr, PaimonErrorCode};
 use crate::result::{
@@ -320,7 +320,7 @@ pub unsafe extern "C" fn paimon_vector_search_builder_execute_read(
 }
 
 pub(crate) async fn materialize_search_result(
-    result: SearchResult,
+    result: GlobalIndexResult,
     projection: Option<&[String]>,
 ) -> paimon::Result<ArrowRecordBatchStream> {
     let mut reader = result.new_read_builder();

bindings/python/python/pypaimon_rust/datafusion.pyi

@@ -56,8 +56,8 @@ class Split:
         ...
     def row_count(self) -> int: ...
     def is_streaming(self) -> bool: ...
-    # Java SplitSerializer v1 binary: DataSplit v8/v9, or score-free
-    # IndexedSplit when the native split has row ranges.
+    # Java SplitSerializer v1 binary: DataSplit v8/v9, or IndexedSplit with
+    # selected row ranges and optional aligned scores.
     def serialize(self) -> bytes: ...
 
 class Plan:
@@ -67,7 +67,32 @@ class Plan:
     def splits(self) -> List[Split]: ...
     def __len__(self) -> int: ...
 
+class GlobalIndexResult:
+    """Index selection shared by table searches and scans; PK selections retain their snapshot."""
+    @staticmethod
+    def from_splits(snapshot_id: Optional[int], splits: Iterable[Split]) -> "GlobalIndexResult": ...
+    @staticmethod
+    def from_row_ranges(ranges: List[Tuple[int, int]]) -> "GlobalIndexResult": ...
+    @staticmethod
+    def from_scored_row_ids(scores: Dict[int, float]) -> "GlobalIndexResult": ...
+
+    def snapshot_id(self) -> Optional[int]: ...
+    def __len__(self) -> int: ...
+    def new_read_builder(self) -> "GlobalIndexResultReadBuilder": ...
+    def row_ids(self) -> Dict[int, float]:
+        """DE scores keyed by global row ID; rejects PK results, including empty ones."""
+        ...
+    def positions(self) -> List[Tuple[bytes, int, str, int, float]]:
+        """PK partition bytes, bucket, file, position and score in relevance order."""
+        ...
+    def splits(self) -> List[bytes]:
+        """PK physical selections in Java SplitSerializer v1 format, including scores."""
+        ...
+
 class TableScan:
+    def with_global_index_result(self, result: GlobalIndexResult) -> "TableScan":
+        """Reuse an index selection. Authorized scans ignore supplied index selections."""
+        ...
     def with_row_position_slice(self, start: int, end: int) -> "TableScan": ...
     def with_row_position_shard(self, index: int, count: int) -> "TableScan": ...
     def with_chunk_shuffle(self, seed: str, chunk_size: int) -> "TableScan":
@@ -731,7 +756,7 @@ class VectorSearchBuilder:
         ...
     def new_vector_search_scan(self) -> "VectorScan": ...
     def new_vector_search_read(self) -> "VectorRead": ...
-    def execute_local(self) -> "SearchResult": ...
+    def execute_local(self) -> "GlobalIndexResult": ...
 
 class BatchVectorSearchBuilder:
     def with_vector_column(self, name: str) -> "BatchVectorSearchBuilder": ...
@@ -743,7 +768,7 @@ class BatchVectorSearchBuilder:
     def with_partition_filter(self, predicate: Dict[str, Any]) -> "BatchVectorSearchBuilder": ...
     def new_vector_search_scan(self) -> "VectorScan": ...
     def new_batch_vector_search_read(self) -> "BatchVectorRead": ...
-    def execute_batch_local(self) -> List["SearchResult"]: ...
+    def execute_batch_local(self) -> List["GlobalIndexResult"]: ...
 
 class VectorScan:
     def scan(self) -> "VectorScanPlan": ...
@@ -752,29 +777,13 @@ class VectorScanPlan:
     def snapshot_id(self) -> Optional[int]: ...
 
 class VectorRead:
-    def read_plan(self, plan: VectorScanPlan) -> "SearchResult": ...
+    def read_plan(self, plan: VectorScanPlan) -> "GlobalIndexResult": ...
 
 class BatchVectorRead:
-    def read_batch_plan(self, plan: VectorScanPlan) -> List["SearchResult"]: ...
-
-class SearchResult:
-    """Snapshot-scoped global row IDs or primary-key physical positions."""
-    def snapshot_id(self) -> Optional[int]: ...
-    def __len__(self) -> int: ...
-    def new_read_builder(self) -> "SearchResultReadBuilder": ...
-    def row_ids(self) -> Dict[int, float]:
-        """DE scores keyed by global row ID; rejects PK results, including empty ones."""
-        ...
-    def positions(self) -> List[Tuple[bytes, int, str, int, float]]:
-        """PK partition bytes, bucket, file, position and score in relevance order."""
-        ...
-    def splits(self) -> List[bytes]:
-        """PK physical selections in Java SplitSerializer v1 format, including scores."""
-        ...
-
+    def read_batch_plan(self, plan: VectorScanPlan) -> List["GlobalIndexResult"]: ...
 
-class SearchResultReadBuilder:
-    def with_projection(self, columns: List[str]) -> "SearchResultReadBuilder": ...
+class GlobalIndexResultReadBuilder:
+    def with_projection(self, columns: List[str]) -> "GlobalIndexResultReadBuilder": ...
     def read(self) -> List[pyarrow.RecordBatch]:
         """Read user columns and scores in relevance order from the search snapshot."""
         ...
@@ -793,11 +802,7 @@ 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 execute_local(self) -> HybridSearchResult: ...
-
-class HybridSearchResult:
-    def __len__(self) -> int: ...
-    def row_ids(self) -> Dict[int, float]: ...
+    def execute_local(self) -> GlobalIndexResult: ...
 
 class FullTextSearchBuilder:
     def with_query(self, field_name: str, query: str) -> FullTextSearchBuilder: ...
@@ -806,7 +811,7 @@ class FullTextSearchBuilder:
     def with_partition_filter(self, predicate: Dict[str, Any]) -> FullTextSearchBuilder: ...
     def new_full_text_scan(self) -> FullTextScan: ...
     def new_full_text_read(self) -> FullTextRead: ...
-    def execute_local(self) -> SearchResult: ...
+    def execute_local(self) -> GlobalIndexResult: ...
 
 class FullTextScan:
     def scan(self) -> FullTextScanPlan: ...
@@ -815,7 +820,7 @@ class FullTextScanPlan:
     def snapshot_id(self) -> Optional[int]: ...
 
 class FullTextRead:
-    def read_plan(self, plan: FullTextScanPlan) -> SearchResult: ...
+    def read_plan(self, plan: FullTextScanPlan) -> GlobalIndexResult: ...
 
 class ShardTableUpdator:
     def arrow_reader(self) -> RecordBatchReader:

bindings/python/src/context.rs

@@ -802,15 +802,14 @@ 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::PyHybridSearchResult>()?;
     this.add_class::<crate::vector_search::PyVectorSearchBuilder>()?;
     this.add_class::<crate::vector_search::PyBatchVectorSearchBuilder>()?;
     this.add_class::<crate::vector_search::PyVectorScan>()?;
     this.add_class::<crate::vector_search::PyVectorScanPlan>()?;
     this.add_class::<crate::vector_search::PyVectorRead>()?;
     this.add_class::<crate::vector_search::PyBatchVectorRead>()?;
-    this.add_class::<crate::search_result::PySearchResult>()?;
-    this.add_class::<crate::search_result::PySearchResultReadBuilder>()?;
+    this.add_class::<crate::global_index_result::PyGlobalIndexResult>()?;
+    this.add_class::<crate::global_index_result::PyGlobalIndexResultReadBuilder>()?;
     this.add_class::<crate::read::PyReadBuilder>()?;
     this.add_class::<crate::read::PyTableScan>()?;
     this.add_class::<crate::read::PyStreamTableScan>()?;

bindings/python/src/full_text_search.rs

@@ -18,8 +18,8 @@
 //! Thin bindings over core full-text Scan -> Plan -> Read.
 
 use crate::error::to_py_err;
+use crate::global_index_result::PyGlobalIndexResult;
 use crate::predicate::dict_to_table_predicate;
-use crate::search_result::PySearchResult;
 use paimon::spec::Predicate;
 use paimon::table::{FullTextRead, FullTextScan, FullTextScanPlan, Table};
 use paimon_datafusion::runtime::runtime;
@@ -123,10 +123,10 @@ impl PyFullTextSearchBuilder {
                 .map_err(to_py_err)?,
         })
     }
-    fn execute_local(&self, py: Python<'_>) -> PyResult<PySearchResult> {
+    fn execute_local(&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()))
-            .map(PySearchResult::new)
+            .map(PyGlobalIndexResult::new)
             .map_err(to_py_err)
     }
 }
@@ -161,9 +161,13 @@ pub struct PyFullTextRead {
 }
 #[pymethods]
 impl PyFullTextRead {
-    fn read_plan(&self, py: Python<'_>, plan: &PyFullTextScanPlan) -> PyResult<PySearchResult> {
+    fn read_plan(
+        &self,
+        py: Python<'_>,
+        plan: &PyFullTextScanPlan,
+    ) -> PyResult<PyGlobalIndexResult> {
         py.detach(|| runtime().block_on(self.inner.read(plan.inner.clone())))
-            .map(PySearchResult::new)
+            .map(PyGlobalIndexResult::new)
             .map_err(to_py_err)
     }
 }

bindings/python/src/search_result.rs → bindings/python/src/global_index_result.rs

@@ -16,34 +16,69 @@
 // under the License.
 
 use crate::error::to_py_err;
+use crate::read::extract_splits;
 use arrow::pyarrow::ToPyArrow;
 use futures::TryStreamExt;
-use paimon::table::SearchResult;
+use paimon::table::GlobalIndexResult;
+use paimon::table::RowRange;
 use paimon_datafusion::runtime::runtime;
+use pyo3::exceptions::PyValueError;
 use pyo3::prelude::*;
 use pyo3::types::PyBytes;
 use std::collections::HashMap;
 
-#[pyclass(name = "SearchResult", module = "pypaimon_rust.datafusion")]
-pub struct PySearchResult {
-    inner: SearchResult,
+#[pyclass(name = "GlobalIndexResult", module = "pypaimon_rust.datafusion")]
+pub struct PyGlobalIndexResult {
+    pub(crate) inner: GlobalIndexResult,
 }
 
-impl PySearchResult {
-    pub(crate) fn new(inner: SearchResult) -> Self {
+impl PyGlobalIndexResult {
+    pub(crate) fn new(inner: GlobalIndexResult) -> Self {
         Self { inner }
     }
 }
 
 type PySearchPosition = (Py<PyBytes>, i32, String, i64, f32);
 
 #[pymethods]
-impl PySearchResult {
+impl PyGlobalIndexResult {
+    #[staticmethod]
+    fn from_splits(snapshot_id: Option<i64>, splits: &Bound<'_, PyAny>) -> PyResult<Self> {
+        Ok(Self {
+            inner: GlobalIndexResult::from_splits(snapshot_id, extract_splits(splits)?),
+        })
+    }
+
+    #[staticmethod]
+    fn from_row_ranges(ranges: Vec<(i64, i64)>) -> PyResult<Self> {
+        let ranges = ranges
+            .into_iter()
+            .map(|(from, to)| {
+                if from > to {
+                    Err(PyValueError::new_err("row range start must be <= end"))
+                } else {
+                    Ok(RowRange::new(from, to))
+                }
+            })
+            .collect::<PyResult<Vec<_>>>()?;
+        Ok(Self {
+            inner: GlobalIndexResult::from_row_ranges(ranges),
+        })
+    }
+
+    #[staticmethod]
+    fn from_scored_row_ids(scores: HashMap<u64, f32>) -> Self {
+        Self {
+            inner: GlobalIndexResult::from_scored_row_ids(
+                paimon::vector_search::ScoredRowIds::from_scored_map(scores),
+            ),
+        }
+    }
     fn snapshot_id(&self) -> Option<i64> {
         self.inner.snapshot_id()
     }
-    fn new_read_builder(&self) -> PySearchResultReadBuilder {
-        PySearchResultReadBuilder {
+    fn new_read_builder(&self) -> PyGlobalIndexResultReadBuilder {
+        PyGlobalIndexResultReadBuilder {
             result: self.inner.clone(),
             projection: None,
         }
@@ -88,14 +123,17 @@ impl PySearchResult {
     }
 }
 
-#[pyclass(name = "SearchResultReadBuilder", module = "pypaimon_rust.datafusion")]
-pub struct PySearchResultReadBuilder {
-    result: SearchResult,
+#[pyclass(
+    name = "GlobalIndexResultReadBuilder",
+    module = "pypaimon_rust.datafusion"
+)]
+pub struct PyGlobalIndexResultReadBuilder {
+    result: GlobalIndexResult,
     projection: Option<Vec<String>>,
 }
 
 #[pymethods]
-impl PySearchResultReadBuilder {
+impl PyGlobalIndexResultReadBuilder {
     fn with_projection(mut slf: PyRefMut<'_, Self>, columns: Vec<String>) -> PyRefMut<'_, Self> {
         slf.projection = Some(columns);
         slf

bindings/python/src/hybrid_search.rs

@@ -18,13 +18,14 @@
 //! Hybrid configuration forwarding; snapshot, routes and fusion belong to core.
 
 use crate::error::to_py_err;
+use crate::global_index_result::PyGlobalIndexResult;
 use crate::predicate::dict_to_table_predicate;
 use paimon::spec::{Predicate, Snapshot};
 use paimon::table::{HybridSearchRanker, HybridSearchRoute, Table};
-use paimon::vector_search::ScoredRowIds;
 use paimon_datafusion::runtime::runtime;
 use pyo3::prelude::*;
 use pyo3::types::PyDict;
+
 use std::collections::HashMap;
 use std::sync::Arc;
 
@@ -176,31 +177,10 @@ impl PyHybridSearchBuilder {
         Ok(slf)
     }
 
-    fn execute_local(&self, py: Python<'_>) -> PyResult<PyHybridSearchResult> {
+    fn execute_local(&self, py: Python<'_>) -> PyResult<PyGlobalIndexResult> {
         let builder = self.builder();
         py.detach(|| runtime().block_on(builder.execute_scored()))
-            .map(|inner| PyHybridSearchResult { inner })
+            .map(PyGlobalIndexResult::new)
             .map_err(to_py_err)
     }
 }
-
-#[pyclass(name = "HybridSearchResult", module = "pypaimon_rust.datafusion")]
-pub struct PyHybridSearchResult {
-    inner: ScoredRowIds,
-}
-
-#[pymethods]
-impl PyHybridSearchResult {
-    fn __len__(&self) -> usize {
-        self.inner.len()
-    }
-
-    fn row_ids(&self) -> HashMap<u64, f32> {
-        self.inner
-            .row_ids
-            .iter()
-            .copied()
-            .zip(self.inner.scores.iter().copied())
-            .collect()
-    }
-}

bindings/python/src/lib.rs

@@ -24,14 +24,14 @@ mod context;
 mod error;
 mod evolution;
 mod full_text_search;
+mod global_index_result;
 mod hybrid_search;
 mod index_build;
 mod merge_into;
 mod oss_cpp;
 mod predicate;
 mod read;
 mod schema;
-mod search_result;
 mod shard_update;
 mod table;
 mod udf;