Complete native global-index lifecycle and search planning

bindings/python/Cargo.toml

@@ -38,7 +38,9 @@ datafusion-ffi = { workspace = true }
 futures = "0.3"
 paimon = { path = "../../crates/paimon", features = ["storage-all"] }
 paimon-datafusion = { path = "../../crates/integrations/datafusion", features = ["fulltext"] }
+paimon-vindex-core = "0.5.0"
 pyo3 = { version = "0.28", features = ["abi3-py310", "chrono"] }
+roaring = "0.11"
 serde_json = "1.0"
 tokio = { workspace = true }
 

bindings/python/python/pypaimon_rust/datafusion.pyi

@@ -67,6 +67,20 @@ class Plan:
     def splits(self) -> List[Split]: ...
     def __len__(self) -> int: ...
 
+class VindexVectorGlobalIndexReader:
+    """Rust index search over an input exposing pread_many(ranges)."""
+    def __init__(
+        self, input: Any, file_path: str, file_size: int,
+        options: Optional[Dict[str, str]] = None,
+    ) -> None: ...
+    def metric(self) -> str: ...
+    def search(
+        self, vectors: List[List[float]], limit: int,
+        filter_bytes: Optional[bytes] = None,
+        options: Optional[Dict[str, str]] = None,
+    ) -> List[Optional[Dict[int, float]]]: ...
+    def close(self) -> None: ...
+
 class GlobalIndexResult:
     """Index selection shared by table searches and scans; PK selections retain their snapshot."""
     @staticmethod
@@ -276,6 +290,7 @@ class Table:
     def new_vector_search_builder(self) -> "VectorSearchBuilder": ...
     def new_batch_vector_search_builder(self) -> "BatchVectorSearchBuilder": ...
     def new_global_index_build_builder(self) -> "GlobalIndexBuildBuilder": ...
+    def new_global_index_drop_builder(self) -> "GlobalIndexDropBuilder": ...
     def new_batch_write_builder(self) -> "BatchWriteBuilder": ...
     def new_postpone_fixed_bucket_write_builder(self) -> "PostponeFixedBucketWriteBuilder": ...
     def new_stream_write_builder(self) -> "StreamWriteBuilder": ...
@@ -327,6 +342,21 @@ class GlobalIndexBuildBuilder:
         """Build and publish indexes, rejecting a change to the planned snapshot."""
         ...
 
+class GlobalIndexDropBuilder:
+    @staticmethod
+    def supports_index_type(index_type: str) -> bool: ...
+    def with_index_columns(self, columns: List[str]) -> "GlobalIndexDropBuilder": ...
+    def with_index_column(self, column: str) -> "GlobalIndexDropBuilder": ...
+    def with_index_type(self, index_type: str) -> "GlobalIndexDropBuilder": ...
+    def with_partition_filter(self, predicate: Dict[str, Any]) -> "GlobalIndexDropBuilder": ...
+    def with_dry_run(self, dry_run: bool) -> "GlobalIndexDropBuilder": ...
+    def build(self) -> List["CommitMessage"]:
+        """Prepare metadata deletions; the caller decides whether to commit."""
+        ...
+    def execute(self) -> int:
+        """Drop matched indexes, or report their count for dry runs."""
+        ...
+
 class CommitMessage:
     @staticmethod
     def deserialize(data: bytes, *, version: int = 14) -> "CommitMessage":
@@ -774,6 +804,7 @@ class VectorScan:
     def scan(self) -> "VectorScanPlan": ...
 
 class VectorScanPlan:
+    def _metadata(self) -> str: ...
     def snapshot_id(self) -> Optional[int]: ...
 
 class VectorRead:
@@ -817,6 +848,7 @@ class FullTextScan:
     def scan(self) -> FullTextScanPlan: ...
 
 class FullTextScanPlan:
+    def _metadata(self) -> str: ...
     def snapshot_id(self) -> Optional[int]: ...
 
 class FullTextRead:

bindings/python/src/blob_uri_reader.rs

@@ -63,7 +63,15 @@ pub(crate) fn to_callback_error(error: paimon::Error) -> PyErr {
             ) => source
                 .as_deref()
                 .map(|source| source as &(dyn std::error::Error + 'static)),
-            _ => cause.source(),
+            // std::io::Error::source skips its custom error itself. Inspect
+            // that value too, since positional index I/O wraps Paimon errors.
+            _ => match cause.downcast_ref::<std::io::Error>() {
+                Some(error) => error
+                    .get_ref()
+                    .map(|error| error as &(dyn std::error::Error + 'static))
+                    .or_else(|| cause.source()),
+                None => cause.source(),
+            },
         };
         match next {
             Some(source) => cause = source,

bindings/python/src/context.rs

@@ -797,12 +797,14 @@ pub fn register_module(py: Python<'_>, m: &Bound<'_, PyModule>) -> PyResult<()>
     this.add_class::<PaimonCatalog>()?;
     this.add_class::<crate::table::PyTable>()?;
     this.add_class::<crate::index_build::PyGlobalIndexBuildBuilder>()?;
+    this.add_class::<crate::index_build::PyGlobalIndexDropBuilder>()?;
     this.add_class::<crate::full_text_search::PyFullTextSearchBuilder>()?;
     this.add_class::<crate::full_text_search::PyFullTextScan>()?;
     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::vector_search::PyVectorSearchBuilder>()?;
+    this.add_class::<crate::vindex::PyVindexVectorGlobalIndexReader>()?;
     this.add_class::<crate::vector_search::PyBatchVectorSearchBuilder>()?;
     this.add_class::<crate::vector_search::PyVectorScan>()?;
     this.add_class::<crate::vector_search::PyVectorScanPlan>()?;

bindings/python/src/full_text_search.rs

@@ -150,6 +150,17 @@ 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,
+        )
+    }
+
     fn snapshot_id(&self) -> Option<i64> {
         self.inner.snapshot_id()
     }

bindings/python/src/index_build.rs

@@ -21,7 +21,7 @@ use std::collections::HashMap;
 use std::sync::Arc;
 
 use paimon::spec::Predicate;
-use paimon::table::{GlobalIndexBuildBuilder, Table};
+use paimon::table::{GlobalIndexBuildBuilder, GlobalIndexDropBuilder, Table};
 use paimon_datafusion::runtime::runtime;
 use pyo3::prelude::*;
 use pyo3::types::PyDict;
@@ -76,6 +76,103 @@ impl PyGlobalIndexBuildBuilder {
     }
 }
 
+/// Parameter transport for metadata-only index drops; planning stays in core.
+#[pyclass(name = "GlobalIndexDropBuilder", module = "pypaimon_rust.datafusion")]
+pub struct PyGlobalIndexDropBuilder {
+    table: Arc<Table>,
+    columns: Option<PyIndexColumns>,
+    index_type: String,
+    partition_filters: Vec<Predicate>,
+    dry_run: bool,
+}
+
+impl PyGlobalIndexDropBuilder {
+    pub(crate) fn new(table: Arc<Table>) -> Self {
+        Self {
+            table,
+            columns: None,
+            index_type: "btree".into(),
+            partition_filters: vec![],
+            dry_run: false,
+        }
+    }
+
+    fn builder(&self) -> paimon::Result<GlobalIndexDropBuilder<'_>> {
+        let mut builder = self.table.new_global_index_drop_builder();
+        match &self.columns {
+            Some(PyIndexColumns::Column(column)) => {
+                builder.with_index_column(column);
+            }
+            Some(PyIndexColumns::Columns(columns)) => {
+                builder.with_index_columns(&columns.iter().map(String::as_str).collect::<Vec<_>>());
+            }
+            None => {}
+        }
+        builder
+            .with_index_type(&self.index_type)
+            .with_dry_run(self.dry_run);
+        for filter in &self.partition_filters {
+            builder.with_partition_filter(filter.clone())?;
+        }
+        Ok(builder)
+    }
+}
+
+#[pymethods]
+impl PyGlobalIndexDropBuilder {
+    #[staticmethod]
+    fn supports_index_type(index_type: &str) -> bool {
+        GlobalIndexDropBuilder::supports_index_type(index_type)
+    }
+
+    fn with_index_columns(mut slf: PyRefMut<'_, Self>, columns: Vec<String>) -> PyRefMut<'_, Self> {
+        slf.columns = Some(PyIndexColumns::Columns(columns));
+        slf
+    }
+
+    fn with_index_column(mut slf: PyRefMut<'_, Self>, column: String) -> PyRefMut<'_, Self> {
+        slf.columns = Some(PyIndexColumns::Column(column));
+        slf
+    }
+
+    fn with_index_type(mut slf: PyRefMut<'_, Self>, index_type: String) -> PyRefMut<'_, Self> {
+        slf.index_type = index_type;
+        slf
+    }
+
+    fn with_dry_run(mut slf: PyRefMut<'_, Self>, dry_run: bool) -> PyRefMut<'_, Self> {
+        slf.dry_run = dry_run;
+        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)?;
+        let mut builder = slf.builder().map_err(to_py_err)?;
+        builder
+            .with_partition_filter(filter.clone())
+            .map_err(to_py_err)?;
+        slf.partition_filters.push(filter);
+        Ok(slf)
+    }
+
+    fn build(&self, py: Python<'_>) -> PyResult<Vec<PyCommitMessage>> {
+        let builder = self.builder().map_err(to_py_err)?;
+        let messages = py
+            .detach(|| runtime().block_on(builder.build()))
+            .map_err(to_py_err)?;
+        Ok(messages.into_iter().map(PyCommitMessage::new).collect())
+    }
+
+    fn execute(&self, py: Python<'_>) -> PyResult<usize> {
+        let builder = self.builder().map_err(to_py_err)?;
+        py.detach(|| runtime().block_on(builder.execute()))
+            .map_err(to_py_err)
+    }
+}
+
 #[pymethods]
 impl PyGlobalIndexBuildBuilder {
     #[staticmethod]

bindings/python/src/lib.rs

@@ -38,6 +38,7 @@ mod udf;
 mod update_assignment;
 mod variant;
 mod vector_search;
+mod vindex;
 mod write;
 // ---- #285: observability ----
 mod partition;

bindings/python/src/table.rs

@@ -230,6 +230,10 @@ impl PyTable {
         crate::index_build::PyGlobalIndexBuildBuilder::new(Arc::clone(&self.inner))
     }
 
+    fn new_global_index_drop_builder(&self) -> crate::index_build::PyGlobalIndexDropBuilder {
+        crate::index_build::PyGlobalIndexDropBuilder::new(Arc::clone(&self.inner))
+    }
+
     fn new_batch_write_builder(&self) -> PyBatchWriteBuilder {
         PyBatchWriteBuilder::new(Arc::clone(&self.inner))
     }

bindings/python/src/vector_search.rs

@@ -28,6 +28,39 @@ use crate::error::to_py_err;
 use crate::global_index_result::PyGlobalIndexResult;
 use crate::predicate::dict_to_table_predicate;
 
+pub(crate) fn search_plan_metadata_json(
+    snapshot: Option<&paimon::spec::Snapshot>,
+    snapshot_id: Option<i64>,
+    splits: Vec<paimon::table::SearchScanSplit>,
+) -> PyResult<String> {
+    use paimon::table::SearchScanSplit;
+    use serde_json::json;
+    let files = |files: &[paimon::spec::IndexFileMeta]| {
+        files
+            .iter()
+            .map(|file| file.to_serialized_row_data().map_err(to_py_err))
+            .collect::<PyResult<Vec<_>>>()
+    };
+    let splits = splits.into_iter().map(|split| Ok(match split {
+        SearchScanSplit::Index { column, row_range, index_files, scalar_index_files } => json!({
+            "kind": "index", "column": column, "row_range": row_range,
+            "index_files": files(&index_files)?, "scalar_index_files": files(&scalar_index_files)?,
+        }),
+        SearchScanSplit::Raw { row_ranges, scalar_index_files, index_type } => json!({
+            "kind": "raw", "row_ranges": row_ranges, "scalar_index_files": files(&scalar_index_files)?, "index_type": index_type,
+        }),
+        SearchScanSplit::PrimaryKey { data_split, payloads, uncovered_data_files, row_ranges_by_file } => json!({
+            "kind": "primary-key", "data_split": data_split.serialize_split_v1().map_err(to_py_err)?,
+            "payloads": files(&payloads)?, "uncovered_data_files": uncovered_data_files,
+            "row_ranges_by_file": row_ranges_by_file,
+        }),
+    })).collect::<PyResult<Vec<_>>>()?;
+    serde_json::to_string(
+        &json!({"snapshot": snapshot, "snapshot_id": snapshot_id, "splits": splits}),
+    )
+    .map_err(|error| pyo3::exceptions::PyValueError::new_err(error.to_string()))
+}
+
 #[derive(Clone, Default)]
 struct SearchConfig {
     column: Option<String>,
@@ -155,7 +188,15 @@ impl PyVectorSearchBuilder {
             .map_err(to_py_err)?
             .new_scan()
             .map_err(to_py_err)?;
-        Ok(PyVectorScan { inner })
+        Ok(PyVectorScan {
+            inner,
+            options: self
+                .config
+                .options
+                .iter()
+                .flat_map(|options| options.clone())
+                .collect(),
+        })
     }
     fn new_vector_search_read(&self) -> PyResult<PyVectorRead> {
         let mut builder = self.config.single(&self.table).map_err(to_py_err)?;
@@ -250,7 +291,15 @@ impl PyBatchVectorSearchBuilder {
             .map_err(to_py_err)?
             .new_scan()
             .map_err(to_py_err)?;
-        Ok(PyVectorScan { inner })
+        Ok(PyVectorScan {
+            inner,
+            options: self
+                .config
+                .options
+                .iter()
+                .flat_map(|options| options.clone())
+                .collect(),
+        })
     }
     fn new_batch_vector_search_read(&self) -> PyResult<PyBatchVectorRead> {
         let mut builder = self.config.batch(&self.table).map_err(to_py_err)?;
@@ -275,6 +324,7 @@ impl PyBatchVectorSearchBuilder {
 #[pyclass(name = "VectorScan", module = "pypaimon_rust.datafusion")]
 pub struct PyVectorScan {
     inner: VectorScan,
+    options: HashMap<String, String>,
 }
 
 #[pymethods]
@@ -283,17 +333,28 @@ impl PyVectorScan {
         let inner = py
             .detach(|| runtime().block_on(self.inner.plan()))
             .map_err(to_py_err)?;
-        Ok(PyVectorScanPlan { inner })
+        Ok(PyVectorScanPlan {
+            inner,
+            options: self.options.clone(),
+        })
     }
 }
 
 #[pyclass(name = "VectorScanPlan", module = "pypaimon_rust.datafusion")]
 pub struct PyVectorScanPlan {
     inner: VectorScanPlan,
+    options: HashMap<String, String>,
 }
 
 #[pymethods]
 impl PyVectorScanPlan {
+    fn _metadata(&self, py: Python<'_>) -> PyResult<String> {
+        let splits = py
+            .detach(|| runtime().block_on(self.inner.splits(&self.options)))
+            .map_err(to_py_err)?;
+        search_plan_metadata_json(self.inner.snapshot(), self.inner.snapshot_id(), splits)
+    }
+
     fn snapshot_id(&self) -> Option<i64> {
         self.inner.snapshot_id()
     }