[Core] Add snapshot-pinned full-text search with scalar filters

bindings/python/python/pypaimon_rust/datafusion.pyi

@@ -221,6 +221,7 @@ class Table:
     def schema(self) -> TableSchema: ...
     def new_read_builder(self, options: Optional[Dict[str, str]] = None) -> ReadBuilder: ...
     def new_stream_scan(self) -> StreamTableScan: ...
+    def new_full_text_search_builder(self) -> FullTextSearchBuilder: ...
     def new_vector_search_builder(self) -> "VectorSearchBuilder": ...
     def new_batch_vector_search_builder(self) -> "BatchVectorSearchBuilder": ...
     def new_global_index_build_builder(self) -> "GlobalIndexBuildBuilder": ...
@@ -613,3 +614,25 @@ class SearchResultReadBuilder:
     def read(self) -> List[pyarrow.RecordBatch]:
         """Read user columns and scores in relevance order from the search snapshot."""
         ...
+
+class FullTextSearchBuilder:
+    def with_query(self, field_name: str, query: str) -> FullTextSearchBuilder: ...
+    def with_limit(self, limit: int) -> FullTextSearchBuilder: ...
+    def with_filter(self, predicate: Dict[str, Any]) -> 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) -> FullTextSearchResult: ...
+
+class FullTextScan:
+    def scan(self) -> FullTextScanPlan: ...
+
+class FullTextScanPlan:
+    def snapshot_id(self) -> Optional[int]: ...
+
+class FullTextRead:
+    def read_plan(self, plan: FullTextScanPlan) -> FullTextSearchResult: ...
+
+class FullTextSearchResult:
+    def __len__(self) -> int: ...
+    def row_ids(self) -> Dict[int, float]: ...

bindings/python/src/context.rs

@@ -727,6 +727,11 @@ 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::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::full_text_search::PyFullTextSearchResult>()?;
     this.add_class::<crate::vector_search::PyVectorSearchBuilder>()?;
     this.add_class::<crate::vector_search::PyBatchVectorSearchBuilder>()?;
     this.add_class::<crate::vector_search::PyVectorScan>()?;

bindings/python/src/full_text_search.rs

@@ -0,0 +1,192 @@
+// 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.
+
+//! Thin bindings over core full-text Scan -> Plan -> Read.
+
+use crate::error::to_py_err;
+use crate::predicate::dict_to_table_predicate;
+use paimon::spec::Predicate;
+use paimon::table::{FullTextRead, FullTextScan, FullTextScanPlan, Table};
+use paimon_datafusion::runtime::runtime;
+use pyo3::prelude::*;
+use pyo3::types::PyDict;
+use std::collections::HashMap;
+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,
+}
+
+impl PyFullTextSearchBuilder {
+    pub(crate) fn new(table: Arc<Table>) -> Self {
+        Self {
+            table,
+            config: Config::default(),
+        }
+    }
+}
+
+#[pymethods]
+impl PyFullTextSearchBuilder {
+    fn with_query(
+        mut slf: PyRefMut<'_, Self>,
+        field_name: String,
+        query: String,
+    ) -> PyRefMut<'_, Self> {
+        slf.config.query = Some((field_name, query));
+        slf
+    }
+    fn with_limit(mut slf: PyRefMut<'_, Self>, limit: usize) -> PyRefMut<'_, Self> {
+        slf.config.limit = Some(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);
+        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);
+        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)?,
+        })
+    }
+    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)?,
+        })
+    }
+    fn execute_local(&self, py: Python<'_>) -> PyResult<PyFullTextSearchResult> {
+        let builder = self.config.builder(&self.table).map_err(to_py_err)?;
+        py.detach(|| runtime().block_on(builder.execute_scored()))
+            .map(|inner| PyFullTextSearchResult { inner })
+            .map_err(to_py_err)
+    }
+}
+
+#[pyclass(name = "FullTextScan", module = "pypaimon_rust.datafusion")]
+pub struct PyFullTextScan {
+    inner: FullTextScan,
+}
+#[pymethods]
+impl PyFullTextScan {
+    fn scan(&self, py: Python<'_>) -> PyResult<PyFullTextScanPlan> {
+        py.detach(|| runtime().block_on(self.inner.scan()))
+            .map(|inner| PyFullTextScanPlan { inner })
+            .map_err(to_py_err)
+    }
+}
+
+#[pyclass(name = "FullTextScanPlan", module = "pypaimon_rust.datafusion")]
+pub struct PyFullTextScanPlan {
+    inner: FullTextScanPlan,
+}
+#[pymethods]
+impl PyFullTextScanPlan {
+    fn snapshot_id(&self) -> Option<i64> {
+        self.inner.snapshot_id()
+    }
+}
+
+#[pyclass(name = "FullTextRead", module = "pypaimon_rust.datafusion")]
+pub struct PyFullTextRead {
+    inner: FullTextRead,
+}
+#[pymethods]
+impl PyFullTextRead {
+    fn read_plan(
+        &self,
+        py: Python<'_>,
+        plan: &PyFullTextScanPlan,
+    ) -> PyResult<PyFullTextSearchResult> {
+        py.detach(|| runtime().block_on(self.inner.read(plan.inner.clone())))
+            .map(|inner| PyFullTextSearchResult { inner })
+            .map_err(to_py_err)
+    }
+}
+
+#[pyclass(name = "FullTextSearchResult", module = "pypaimon_rust.datafusion")]
+pub struct PyFullTextSearchResult {
+    inner: paimon::full_text::SearchResult,
+}
+#[pymethods]
+impl PyFullTextSearchResult {
+    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

@@ -21,6 +21,7 @@ mod blob;
 mod blob_uri_reader;
 mod context;
 mod error;
+mod full_text_search;
 mod index_build;
 mod merge_into;
 mod oss_cpp;

bindings/python/src/table.rs

@@ -198,6 +198,10 @@ impl PyTable {
         }
     }
 
+    fn new_full_text_search_builder(&self) -> crate::full_text_search::PyFullTextSearchBuilder {
+        crate::full_text_search::PyFullTextSearchBuilder::new(Arc::clone(&self.inner))
+    }
+
     fn new_vector_search_builder(&self) -> crate::vector_search::PyVectorSearchBuilder {
         crate::vector_search::PyVectorSearchBuilder::new(Arc::clone(&self.inner))
     }

crates/paimon/src/table/de_vector_read.rs

@@ -24,11 +24,12 @@ use crate::spec::{
     row_id_data_field, CoreOptions, DataField, DataType, FileKind, GlobalIndexSearchMode,
     IndexFileMeta, IndexManifestEntry, Predicate, ROW_ID_FIELD_NAME,
 };
-use crate::table::de_vector_scan::{matching_row_ids_for_filter, DeVectorScanPlan};
+use crate::table::de_vector_scan::DeVectorScanPlan;
 use crate::table::global_index_scanner::{
     deleted_row_ranges_for_data_evolution_dvs, search_limit_with_deleted_rows,
     unindexed_ranges_for_global_index_entries, GlobalIndexScanner, RowRangeIndex,
 };
+use crate::table::global_index_search_filter::matching_row_ids_for_filter;
 use crate::table::global_index_types::normalize_queryable_global_index_type;
 use crate::table::index_file_path::IndexFileLocation;
 use crate::table::pk_vector_position_read::SEARCH_SCORE_COLUMN;

crates/paimon/src/table/de_vector_scan.rs

@@ -17,14 +17,13 @@
 
 //! Plans global-index vector searches against one snapshot and resolves scalar pre-filters.
 
-use crate::spec::{CoreOptions, IndexManifest, IndexManifestEntry, Predicate, ROW_ID_FIELD_NAME};
+use crate::spec::{CoreOptions, IndexManifest, IndexManifestEntry, Predicate};
 use crate::table::bucket_filter::split_partition_and_data_predicates;
+use crate::table::global_index_search_filter::matching_row_ids_for_filter;
 use crate::table::partition_filter::PartitionFilter;
 use crate::table::vector_scan::Scan;
-use crate::table::{RowRange, Table};
+use crate::table::Table;
 use crate::vindex::vector_search_timing_enabled;
-use arrow_array::{Array, Int64Array};
-use futures::TryStreamExt;
 use roaring::RoaringTreemap;
 use std::sync::Arc;
 use std::time::{Duration, Instant};
@@ -54,65 +53,6 @@ fn same_vector_search_table(left: &Table, right: &Table) -> bool {
         && left.branch() == right.branch()
 }
 
-pub(super) async fn matching_row_ids_for_filter(
-    table: &Table,
-    filter: &Predicate,
-    ranges: Option<Vec<RowRange>>,
-) -> crate::Result<RoaringTreemap> {
-    let mut read_builder = table.new_read_builder();
-    read_builder
-        .with_projection(&[ROW_ID_FIELD_NAME])?
-        .with_filter(filter.clone());
-    if let Some(ranges) = ranges {
-        read_builder.with_row_ranges(ranges);
-    }
-    let plan = read_builder.new_scan().plan().await?;
-    let read = read_builder.new_read()?;
-    let mut stream = read.to_arrow(plan.splits())?;
-    let mut row_ids = RoaringTreemap::new();
-    while let Some(batch) = stream.try_next().await? {
-        let index =
-            batch
-                .schema()
-                .index_of(ROW_ID_FIELD_NAME)
-                .map_err(|_| crate::Error::DataInvalid {
-                    message: format!(
-                        "scalar vector pre-filter read is missing {ROW_ID_FIELD_NAME}"
-                    ),
-                    source: None,
-                })?;
-        let values = batch
-            .column(index)
-            .as_any()
-            .downcast_ref::<Int64Array>()
-            .ok_or_else(|| crate::Error::DataInvalid {
-                message: format!(
-                    "scalar vector pre-filter {ROW_ID_FIELD_NAME} column is not Int64"
-                ),
-                source: None,
-            })?;
-        for row in 0..values.len() {
-            if values.is_null(row) {
-                return Err(crate::Error::DataInvalid {
-                    message: format!(
-                        "scalar vector pre-filter produced a null {ROW_ID_FIELD_NAME}"
-                    ),
-                    source: None,
-                });
-            }
-            let row_id = values.value(row);
-            let row_id = u64::try_from(row_id).map_err(|_| crate::Error::DataInvalid {
-                message: format!(
-                    "scalar vector pre-filter produced a negative {ROW_ID_FIELD_NAME}: {row_id}"
-                ),
-                source: None,
-            })?;
-            row_ids.insert(row_id);
-        }
-    }
-    Ok(row_ids)
-}
-
 impl Table {
     /// Resolve a scalar predicate once and pin all later vector-search/read
     /// stages to the same snapshot.

crates/paimon/src/table/de_vector_scan/tests.rs

@@ -27,6 +27,7 @@ use crate::vector_search::SearchResult;
 use arrow_array::builder::{Float32Builder, ListBuilder};
 use arrow_array::{ArrayRef, Float32Array, Int32Array, RecordBatch};
 use arrow_schema::{DataType as ArrowDataType, Field as ArrowField, Schema as ArrowSchema};
+use futures::TryStreamExt;
 use roaring::RoaringTreemap;
 use std::collections::HashMap;
 use std::sync::Arc;