[core] Support scheduled execution of snapshot-scoped search plans

bindings/c/src/vector_scan.rs

@@ -64,7 +64,9 @@ pub unsafe extern "C" fn paimon_vector_scan_plan(
 }
 
 /// Construct a common read plan from already-decoded Java PK bucket splits.
-/// No snapshot or index manifest is read. Handles are borrowed and copied;
+/// Capture the supplied snapshot ID, or the scan's configured snapshot for
+/// unversioned buckets, without replanning the index manifest.
+/// Handles are borrowed and copied;
 /// callers may free them as soon as this returns. Failure leaves inputs intact.
 /// Empty input, DE scans and mixed snapshots are rejected.
 /// # Safety
@@ -111,7 +113,7 @@ pub unsafe extern "C" fn paimon_vector_scan_plan_from_bucket_splits(
         }
         decoded.push((&*((*split).inner as *const BucketVectorSearchSplit)).clone());
     }
-    wrap_plan(scan.plan_from_bucket_splits(decoded))
+    wrap_plan(runtime().block_on(scan.plan_from_bucket_splits(decoded)))
 }
 
 /// Free a vector scan. Null is accepted.

bindings/python/Cargo.toml

@@ -36,7 +36,7 @@ chrono = "0.4.38"
 datafusion = { workspace = true }
 datafusion-ffi = { workspace = true }
 futures = "0.3"
-paimon = { path = "../../crates/paimon", features = ["storage-all"] }
+paimon = { path = "../../crates/paimon", features = ["storage-all", "internal-api"] }
 paimon-datafusion = { path = "../../crates/integrations/datafusion", features = ["fulltext"] }
 paimon-vindex-core = "0.5.0"
 pyo3 = { version = "0.28", features = ["abi3-py310", "chrono"] }

bindings/python/python/pypaimon_rust/datafusion.pyi

@@ -924,7 +924,7 @@ class VectorSearchBuilder:
         ...
     def new_vector_search_scan(self) -> "VectorScan": ...
     def new_vector_search_read(self) -> "VectorRead": ...
-    def execute_local(self) -> "GlobalIndexResult": ...
+    def execute(self) -> "GlobalIndexResult": ...
 
 class BatchVectorSearchBuilder:
     def with_vector_column(self, name: str) -> "BatchVectorSearchBuilder": ...
@@ -936,14 +936,51 @@ 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["GlobalIndexResult"]: ...
+    def execute(self) -> List["GlobalIndexResult"]: ...
 
 class VectorScan:
     def scan(self) -> "VectorScanPlan": ...
 
+class IndexFileMeta:
+    """Immutable index metadata obtained from search source splits."""
+    def index_type(self) -> str: ...
+    def file_name(self) -> str: ...
+    def file_size(self) -> int: ...
+    def row_count(self) -> int: ...
+
+class IndexVectorSearchSplit:
+    def __init__(self, row_range_start: int, row_range_end: int,
+                 vector_index_files: Sequence[IndexFileMeta],
+                 scalar_index_files: Sequence[IndexFileMeta] = ()) -> None: ...
+    def row_range_start(self) -> int: ...
+    def row_range_end(self) -> int: ...
+    def vector_index_files(self) -> List[IndexFileMeta]: ...
+    def scalar_index_files(self) -> List[IndexFileMeta]: ...
+
+class RawVectorSearchSplit:
+    def __init__(self, row_ranges: Sequence[Tuple[int, int]],
+                 scalar_index_files: Sequence[IndexFileMeta] = (),
+                 index_type: Optional[str] = None) -> None: ...
+    def row_ranges(self) -> List[Tuple[int, int]]: ...
+    def scalar_index_files(self) -> List[IndexFileMeta]: ...
+    def index_type(self) -> Optional[str]: ...
+
+class PrimaryKeyVectorSearchSplit:
+    def __init__(self, data_split: Split, payloads: Sequence[IndexFileMeta],
+                 uncovered_data_files: Sequence[str],
+                 row_ranges_by_file: Dict[str, Sequence[Tuple[int, int]]]) -> None: ...
+    def data_split(self) -> Split: ...
+    def payloads(self) -> List[IndexFileMeta]: ...
+    def uncovered_data_files(self) -> List[str]: ...
+    def row_ranges_by_file(self) -> Dict[str, List[Tuple[int, int]]]: ...
+
 class VectorScanPlan:
-    def _metadata(self) -> str: ...
-    def snapshot_id(self) -> Optional[int]: ...
+    def __init__(self, splits: Sequence[Union[IndexVectorSearchSplit, RawVectorSearchSplit,
+                                            PrimaryKeyVectorSearchSplit]],
+                 snapshot: Optional[Snapshot] = None) -> None: ...
+    def snapshot(self) -> Optional[Snapshot]: ...
+    def splits(self) -> List[Union[IndexVectorSearchSplit, RawVectorSearchSplit,
+                                  PrimaryKeyVectorSearchSplit]]: ...
 
 class VectorRead:
     def read_plan(self, plan: VectorScanPlan) -> "GlobalIndexResult": ...
@@ -959,8 +996,8 @@ class GlobalIndexResultReadBuilder:
 
 class HybridSearchBuilder:
     def with_limit(self, limit: int) -> HybridSearchBuilder: ...
-    def with_snapshot(self, snapshot_json: Optional[str]) -> HybridSearchBuilder:
-        """Reuse resolved Java Snapshot JSON; None pins an empty read view."""
+    def with_snapshot(self, snapshot: Optional[Snapshot]) -> HybridSearchBuilder:
+        """Pin the resolved snapshot object; None pins an empty read view."""
         ...
     def with_ranker(self, ranker: Optional[str]) -> HybridSearchBuilder: ...
     def with_rrf_ranker(self) -> HybridSearchBuilder: ...
@@ -971,7 +1008,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) -> GlobalIndexResult: ...
+    def execute(self) -> GlobalIndexResult: ...
 
 class FullTextSearchBuilder:
     def with_query(self, field_name: str, query: str) -> FullTextSearchBuilder: ...
@@ -980,14 +1017,41 @@ 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) -> GlobalIndexResult: ...
+    def execute(self) -> GlobalIndexResult: ...
 
 class FullTextScan:
     def scan(self) -> FullTextScanPlan: ...
 
+class IndexFullTextSearchSplit:
+    def __init__(self, column_name: str, row_range_start: int, row_range_end: int,
+                 full_text_index_files: Sequence[IndexFileMeta],
+                 scalar_index_files: Sequence[IndexFileMeta] = ()) -> None: ...
+    def column_name(self) -> str: ...
+    def row_range_start(self) -> int: ...
+    def row_range_end(self) -> int: ...
+    def full_text_index_files(self) -> List[IndexFileMeta]: ...
+    def scalar_index_files(self) -> List[IndexFileMeta]: ...
+
+class RawFullTextSearchSplit:
+    def __init__(self, row_ranges: Sequence[Tuple[int, int]]) -> None: ...
+    def row_ranges(self) -> List[Tuple[int, int]]: ...
+
+class PrimaryKeyFullTextSearchSplit:
+    def __init__(self, data_split: Split, payloads: Sequence[IndexFileMeta],
+                 uncovered_data_files: Sequence[str],
+                 row_ranges_by_file: Dict[str, Sequence[Tuple[int, int]]]) -> None: ...
+    def data_split(self) -> Split: ...
+    def payloads(self) -> List[IndexFileMeta]: ...
+    def uncovered_data_files(self) -> List[str]: ...
+    def row_ranges_by_file(self) -> Dict[str, List[Tuple[int, int]]]: ...
+
 class FullTextScanPlan:
-    def _metadata(self) -> str: ...
-    def snapshot_id(self) -> Optional[int]: ...
+    def __init__(self, splits: Sequence[Union[IndexFullTextSearchSplit, RawFullTextSearchSplit,
+                                            PrimaryKeyFullTextSearchSplit]],
+                 snapshot: Optional[Snapshot] = None) -> None: ...
+    def snapshot(self) -> Optional[Snapshot]: ...
+    def splits(self) -> List[Union[IndexFullTextSearchSplit, RawFullTextSearchSplit,
+                                  PrimaryKeyFullTextSearchSplit]]: ...
 
 class FullTextRead:
     def read_plan(self, plan: FullTextScanPlan) -> GlobalIndexResult: ...

bindings/python/src/context.rs

@@ -1004,6 +1004,13 @@ pub fn register_module(py: Python<'_>, m: &Bound<'_, PyModule>) -> PyResult<()>
     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::search_plan::PyIndexFileMeta>()?;
+    this.add_class::<crate::search_plan::PyIndexVectorSearchSplit>()?;
+    this.add_class::<crate::search_plan::PyRawVectorSearchSplit>()?;
+    this.add_class::<crate::search_plan::PyPrimaryKeyVectorSearchSplit>()?;
+    this.add_class::<crate::search_plan::PyIndexFullTextSearchSplit>()?;
+    this.add_class::<crate::search_plan::PyRawFullTextSearchSplit>()?;
+    this.add_class::<crate::search_plan::PyPrimaryKeyFullTextSearchSplit>()?;
     this.add_class::<crate::vector_search::PyVectorRead>()?;
     this.add_class::<crate::vector_search::PyBatchVectorRead>()?;
     this.add_class::<crate::global_index_result::PyGlobalIndexResult>()?;

bindings/python/src/full_text_search.rs

@@ -20,11 +20,13 @@
 use crate::error::to_py_err;
 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_datafusion::runtime::runtime;
 use pyo3::prelude::*;
-use pyo3::types::PyDict;
+use pyo3::types::{PyDict, PyTuple};
 use std::sync::Arc;
 
 #[derive(Default)]
@@ -123,7 +125,7 @@ impl PyFullTextSearchBuilder {
                 .map_err(to_py_err)?,
         })
     }
-    fn execute_local(&self, py: Python<'_>) -> PyResult<PyGlobalIndexResult> {
+    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()))
             .map(PyGlobalIndexResult::new)
@@ -150,19 +152,32 @@ pub struct PyFullTextScanPlan {
 }
 #[pymethods]
 impl PyFullTextScanPlan {
-    fn _metadata(&self, py: Python<'_>) -> PyResult<String> {
-        let splits = py
-            .detach(|| runtime().block_on(self.inner.splits()))
-            .map_err(to_py_err)?;
-        crate::vector_search::search_plan_metadata_json(
-            self.inner.snapshot(),
-            self.inner.snapshot_id(),
-            splits,
-        )
+    #[new]
+    #[pyo3(signature = (splits, snapshot=None))]
+    fn new(splits: &Bound<'_, PyAny>, snapshot: Option<PyRef<'_, PySnapshot>>) -> PyResult<Self> {
+        Ok(Self {
+            inner: FullTextScanPlan::new(
+                snapshot.map(|snapshot| snapshot.inner.clone()),
+                full_text_splits_from_python(splits)?,
+            )
+            .map_err(to_py_err)?,
+        })
+    }
+
+    fn snapshot(&self) -> Option<PySnapshot> {
+        self.inner.snapshot().cloned().map(PySnapshot::new)
     }
 
-    fn snapshot_id(&self) -> Option<i64> {
-        self.inner.snapshot_id()
+    fn splits(&self, py: Python<'_>) -> PyResult<Vec<Py<PyAny>>> {
+        full_text_splits_to_python(py, self.inner.splits())
+    }
+
+    fn __reduce__(slf: &Bound<'_, Self>, py: Python<'_>) -> PyResult<(Py<PyAny>, Py<PyTuple>)> {
+        let plan = slf.borrow();
+        let args = (plan.splits(py)?, plan.snapshot())
+            .into_pyobject(py)?
+            .unbind();
+        Ok((slf.get_type().unbind().into_any(), args))
     }
 }
 
@@ -181,4 +196,24 @@ impl PyFullTextRead {
             .map(PyGlobalIndexResult::new)
             .map_err(to_py_err)
     }
+
+    fn _read_plan_with_executor(
+        &self,
+        py: Python<'_>,
+        plan: &PyFullTextScanPlan,
+        executor: Py<PyAny>,
+    ) -> PyResult<PyGlobalIndexResult> {
+        let executor = crate::search_task::PythonSearchTaskExecutor::new(executor);
+        let result = py.detach(|| {
+            runtime().block_on(paimon::table::internal::read_full_text_with_executor(
+                &self.inner,
+                plan.inner.clone(),
+                &executor,
+            ))
+        });
+        if let Some(error) = executor.take_error() {
+            return Err(error);
+        }
+        result.map(PyGlobalIndexResult::new).map_err(to_py_err)
+    }
 }