This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-rust.git
The following commit(s) were added to refs/heads/main by this push:
new 33a14cc9 Complete native global-index lifecycle and search planning
(#1136)
33a14cc9 is described below
commit 33a14cc9be88fc26e59aafc624e349994e980289
Author: Jingsong Lee <[email protected]>
AuthorDate: Fri Oct 16 09:35:09 2026 +0800
Complete native global-index lifecycle and search planning (#1136)
---
Cargo.lock | 2 +
bindings/python/Cargo.toml | 2 +
.../python/python/pypaimon_rust/datafusion.pyi | 32 +++
bindings/python/src/blob_uri_reader.rs | 10 +-
bindings/python/src/context.rs | 2 +
bindings/python/src/full_text_search.rs | 11 +
bindings/python/src/index_build.rs | 99 ++++++++-
bindings/python/src/lib.rs | 1 +
bindings/python/src/table.rs | 4 +
bindings/python/src/vector_search.rs | 67 +++++-
bindings/python/src/vindex.rs | 197 ++++++++++++++++++
crates/paimon/src/table/de_vector_read.rs | 120 ++++++++---
crates/paimon/src/table/de_vector_read/tests.rs | 145 +++++++++++++
.../src/table/full_text_index_build_builder.rs | 26 +--
.../src/table/full_text_search_builder/scan.rs | 30 +++
.../paimon/src/table/global_index_build_builder.rs | 88 ++++++++
.../paimon/src/table/global_index_build_common.rs | 27 +++
.../paimon/src/table/global_index_drop_builder.rs | 139 +++++++++++--
crates/paimon/src/table/mod.rs | 2 +
crates/paimon/src/table/pk_vector_read.rs | 137 +++++++------
crates/paimon/src/table/pk_vector_scan.rs | 19 +-
crates/paimon/src/table/search_scan_split.rs | 224 +++++++++++++++++++++
.../src/table/sorted_global_index_build_builder.rs | 4 +-
.../extraction.rs | 3 +
.../sorted_global_index_build_builder/tests.rs | 140 +++++++++++++
.../validation.rs | 40 +---
crates/paimon/src/table/vector_read.rs | 2 +-
crates/paimon/src/table/vector_scan.rs | 145 ++++++++++++-
crates/paimon/src/table/vector_search_common.rs | 56 +++++-
.../paimon/src/table/vindex_index_build_builder.rs | 6 +-
.../table/vindex_index_build_builder/extraction.rs | 3 +
.../src/table/vindex_index_build_builder/tests.rs | 114 +++++++++++
.../table/vindex_index_build_builder/validation.rs | 36 +---
crates/paimon/src/vindex/reader.rs | 154 ++++++++++----
34 files changed, 1814 insertions(+), 273 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index e1b2da67..0efb9144 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -5393,7 +5393,9 @@ dependencies = [
"futures",
"paimon",
"paimon-datafusion",
+ "paimon-vindex-core",
"pyo3",
+ "roaring",
"serde_json",
"tokio",
]
diff --git a/bindings/python/Cargo.toml b/bindings/python/Cargo.toml
index 2739e62d..733b7b6d 100644
--- a/bindings/python/Cargo.toml
+++ b/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 }
diff --git a/bindings/python/python/pypaimon_rust/datafusion.pyi
b/bindings/python/python/pypaimon_rust/datafusion.pyi
index 5b132e81..3af2afc5 100644
--- a/bindings/python/python/pypaimon_rust/datafusion.pyi
+++ b/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:
diff --git a/bindings/python/src/blob_uri_reader.rs
b/bindings/python/src/blob_uri_reader.rs
index 95b9b38d..c5cd2f91 100644
--- a/bindings/python/src/blob_uri_reader.rs
+++ b/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,
diff --git a/bindings/python/src/context.rs b/bindings/python/src/context.rs
index d312192a..edd49205 100644
--- a/bindings/python/src/context.rs
+++ b/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>()?;
diff --git a/bindings/python/src/full_text_search.rs
b/bindings/python/src/full_text_search.rs
index a3fc6a9b..acdc76e9 100644
--- a/bindings/python/src/full_text_search.rs
+++ b/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()
}
diff --git a/bindings/python/src/index_build.rs
b/bindings/python/src/index_build.rs
index d81e7252..ff5079c6 100644
--- a/bindings/python/src/index_build.rs
+++ b/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]
diff --git a/bindings/python/src/lib.rs b/bindings/python/src/lib.rs
index 4ae51342..9b92ae83 100644
--- a/bindings/python/src/lib.rs
+++ b/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;
diff --git a/bindings/python/src/table.rs b/bindings/python/src/table.rs
index 6eb49675..14e4298f 100644
--- a/bindings/python/src/table.rs
+++ b/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))
}
diff --git a/bindings/python/src/vector_search.rs
b/bindings/python/src/vector_search.rs
index 4c1173f2..8b7d4290 100644
--- a/bindings/python/src/vector_search.rs
+++ b/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()
}
diff --git a/bindings/python/src/vindex.rs b/bindings/python/src/vindex.rs
new file mode 100644
index 00000000..58cd9d2a
--- /dev/null
+++ b/bindings/python/src/vindex.rs
@@ -0,0 +1,197 @@
+// 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.
+
+//! Python positional I/O over the reusable Rust core vector-index reader.
+
+use crate::blob_uri_reader::{callback_error, to_callback_error};
+use crate::error::to_py_err;
+use paimon::vector_search::{GlobalIndexIOMeta, VectorSearch};
+use paimon::vindex::reader::VindexVectorGlobalIndexReader;
+use paimon_vindex_core::io::{ReadRequest, SeekRead};
+use pyo3::exceptions::{PyRuntimeError, PyTypeError, PyValueError};
+use pyo3::prelude::*;
+use roaring::RoaringTreemap;
+use std::collections::HashMap;
+use std::io::{self, Cursor};
+use std::sync::Mutex;
+
+struct PythonIndexInput(Py<PyAny>);
+
+impl SeekRead for PythonIndexInput {
+ fn pread(&mut self, ranges: &mut [ReadRequest<'_>]) -> io::Result<()> {
+ Python::attach(|py| {
+ let requests = ranges
+ .iter()
+ .map(|range| (range.pos, range.buf.len()))
+ .collect::<Vec<_>>();
+ let chunks = self
+ .0
+ .bind(py)
+ .call_method1("pread_many", (requests,))?
+ .extract::<Vec<Vec<u8>>>()?;
+ if chunks.len() != ranges.len() {
+ return Err(PyValueError::new_err(
+ "Index input returned a different number of ranges",
+ ));
+ }
+ for (chunk, range) in chunks.iter().zip(ranges) {
+ if chunk.len() != range.buf.len() {
+ return Err(PyValueError::new_err(format!(
+ "Index input short read at {}: expected {} bytes, got
{}",
+ range.pos,
+ range.buf.len(),
+ chunk.len()
+ )));
+ }
+ range.buf.copy_from_slice(chunk);
+ }
+ Ok(())
+ })
+ .map_err(|error| io::Error::other(callback_error(error)))
+ }
+}
+
+/// The input only supplies positional I/O; core owns query and score
semantics.
+#[pyclass(
+ name = "VindexVectorGlobalIndexReader",
+ module = "pypaimon_rust.datafusion"
+)]
+pub struct PyVindexVectorGlobalIndexReader {
+ reader: Mutex<Option<VindexVectorGlobalIndexReader>>,
+ metric: String,
+ process_id: u32,
+}
+
+impl PyVindexVectorGlobalIndexReader {
+ fn ensure_process(&self) -> PyResult<()> {
+ if self.process_id != std::process::id() {
+ return Err(to_py_err(paimon::Error::ProcessForkUnsupported {
+ message: "An inherited vector-index reader cannot be used
after fork; reopen it in the child process".into(),
+ }));
+ }
+ Ok(())
+ }
+}
+
+#[pymethods]
+impl PyVindexVectorGlobalIndexReader {
+ #[new]
+ #[pyo3(signature = (input, file_path, file_size, options=None))]
+ fn new(
+ py: Python<'_>,
+ input: Py<PyAny>,
+ file_path: String,
+ file_size: u64,
+ options: Option<HashMap<String, String>>,
+ ) -> PyResult<Self> {
+ if !input.bind(py).getattr("pread_many")?.is_callable() {
+ return Err(PyTypeError::new_err(
+ "Index input must provide callable pread_many",
+ ));
+ }
+ let mut reader = VindexVectorGlobalIndexReader::new(
+ GlobalIndexIOMeta::new(file_path, file_size, Vec::new()),
+ options.unwrap_or_default(),
+ );
+ py.detach(|| reader.load(|_| Ok(PythonIndexInput(input))))
+ .map_err(to_callback_error)?;
+ let metric =
reader.metadata().map_err(to_py_err)?.metric.as_str().into();
+ Ok(Self {
+ reader: Mutex::new(Some(reader)),
+ metric,
+ process_id: std::process::id(),
+ })
+ }
+
+ /// Returns the native index metric, including before the first query.
+ fn metric(&self, py: Python<'_>) -> PyResult<String> {
+ self.ensure_process()?;
+ py.detach(|| {
+ let reader = self
+ .reader
+ .lock()
+ .map_err(|_| PyRuntimeError::new_err("Vector-index reader lock
was poisoned"))?;
+ if reader.is_none() {
+ return Err(PyValueError::new_err("Vector-index reader is
closed"));
+ }
+ Ok(self.metric.clone())
+ })
+ }
+
+ /// Each query returns its actual hits, which may be fewer than the limit.
+ #[pyo3(signature = (vectors, limit, filter_bytes=None, options=None))]
+ fn search(
+ &self,
+ py: Python<'_>,
+ vectors: Vec<Vec<f32>>,
+ limit: usize,
+ filter_bytes: Option<Vec<u8>>,
+ options: Option<HashMap<String, String>>,
+ ) -> PyResult<Vec<Option<HashMap<u64, f32>>>> {
+ self.ensure_process()?;
+ let filter = filter_bytes
+ .map(|bytes| {
+ let mut input = Cursor::new(&bytes);
+ let filter = RoaringTreemap::deserialize_from(&mut input)
+ .map_err(|error|
PyValueError::new_err(error.to_string()))?;
+ if input.position() != bytes.len() as u64 {
+ return Err(PyValueError::new_err(
+ "Trailing bytes after vector-index row-ID filter",
+ ));
+ }
+ Ok(filter)
+ })
+ .transpose()?;
+ let options = options.unwrap_or_default();
+ let searches = vectors
+ .into_iter()
+ .map(|vector| {
+ let mut search = VectorSearch::new(vector, limit,
"vector".into())?
+ .with_options(options.clone());
+ if let Some(filter) = &filter {
+ search = search.with_include_row_ids(filter.clone());
+ }
+ Ok(search)
+ })
+ .collect::<paimon::Result<Vec<_>>>()
+ .map_err(to_py_err)?;
+ py.detach(|| {
+ let mut reader = self
+ .reader
+ .lock()
+ .map_err(|_| PyRuntimeError::new_err("Vector-index reader lock
was poisoned"))?;
+ reader
+ .as_mut()
+ .ok_or_else(|| PyValueError::new_err("Vector-index reader is
closed"))?
+ .visit_batch_vector_search::<PythonIndexInput>(&searches, |_| {
+ unreachable!("the index was opened by the constructor")
+ })
+ .map_err(to_callback_error)
+ })
+ }
+
+ fn close(&self, py: Python<'_>) -> PyResult<()> {
+ self.ensure_process()?;
+ py.detach(|| {
+ self.reader
+ .lock()
+ .map_err(|_| PyRuntimeError::new_err("Vector-index reader lock
was poisoned"))?
+ .take();
+ Ok(())
+ })
+ }
+}
diff --git a/crates/paimon/src/table/de_vector_read.rs
b/crates/paimon/src/table/de_vector_read.rs
index 6de0f5f3..a96f2f8b 100644
--- a/crates/paimon/src/table/de_vector_read.rs
+++ b/crates/paimon/src/table/de_vector_read.rs
@@ -36,8 +36,9 @@ use
crate::table::pk_vector_position_read::SEARCH_SCORE_COLUMN;
use crate::table::row_id_predicate::intersect_sorted_ranges;
use crate::table::vector_read::Read;
use crate::table::vector_search_common::{
- configured_refine_factor, indexed_search_limit, log_vindex_range_io_stats,
normalize_metric,
- resolve_materialize_read_type, vindex_concurrency_limits,
VectorIndexBackend,
+ configured_refine_factor, configured_vector_index_type,
indexed_search_limit,
+ log_vindex_range_io_stats, normalize_metric, resolve_materialize_read_type,
+ validate_global_vector_index_types, vindex_concurrency_limits,
VectorIndexBackend,
};
use crate::table::{
find_field_id_by_name, merge_row_ranges, ArrowRecordBatchStream, RowRange,
Table,
@@ -312,14 +313,14 @@ async fn execute_vindex_searches<S: SeekRead + 'static,
G: Send + 'static>(
}
#[derive(Clone, Copy)]
-struct VectorSearchEvaluation<'a> {
- table: Option<&'a Table>,
- file_io: &'a FileIO,
- table_path: &'a str,
- table_options: &'a HashMap<String, String>,
- schema_fields: &'a [DataField],
- next_row_id: Option<i64>,
- filter: Option<&'a Predicate>,
+pub(super) struct VectorSearchEvaluation<'a> {
+ pub(super) table: Option<&'a Table>,
+ pub(super) file_io: &'a FileIO,
+ pub(super) table_path: &'a str,
+ pub(super) table_options: &'a HashMap<String, String>,
+ pub(super) schema_fields: &'a [DataField],
+ pub(super) next_row_id: Option<i64>,
+ pub(super) filter: Option<&'a Predicate>,
}
#[derive(Clone)]
@@ -506,6 +507,7 @@ async fn evaluate_batch_vector_search(
Some(id) => id,
None => return Ok(vec![ScoredRowIds::empty(); vector_searches.len()]),
};
+ validate_global_vector_index_types(index_entries, field_id, field_name)?;
let vector_entries: Vec<_> = index_entries
.iter()
@@ -636,9 +638,15 @@ async fn evaluate_batch_vector_search(
.map(|entry| (entry, None))
.collect::<Vec<_>>()
};
+ let configured_index_type =
+ configured_vector_index_type(evaluation.table_options,
&search_options, field_name);
+ let index_type = vector_entries
+ .first()
+ .map(|entry| entry.index_file.index_type.as_str())
+ .or(configured_index_type.as_deref());
let metric_check = VectorMetricCheck {
persisted: Arc::new(std::sync::Mutex::new(None)),
- requested: configured_raw_vector_metric_override(&search_options,
field_name)?,
+ requested: vector_metric_override(&search_options, field_name,
index_type)?,
};
let mut permit_wait = Duration::ZERO;
let mut file_reader_open = Duration::ZERO;
@@ -1067,7 +1075,7 @@ async fn evaluate_batch_vector_search(
/// Predicate coverage requires every referenced field to be covered by a
/// single-field scalar index. Reuse the table scanner's coverage accounting.
-fn scalar_unindexed_ranges(
+pub(super) fn scalar_unindexed_ranges(
evaluation: VectorSearchEvaluation<'_>,
entries: &[IndexManifestEntry],
filter: &Predicate,
@@ -1436,7 +1444,7 @@ fn localize_shared_include_row_ids(
.collect())
}
-async fn detail_data_ranges_for_table(table: &Table) ->
crate::Result<Vec<RowRange>> {
+pub(super) async fn detail_data_ranges_for_table(table: &Table) ->
crate::Result<Vec<RowRange>> {
let plan = table
.new_read_builder()
.new_scan()
@@ -1504,9 +1512,23 @@ async fn resolve_raw_vector_metric(
field_id: i32,
field_name: &str,
) -> crate::Result<RawVectorMetric> {
+ let configured_type = configured_vector_index_type(table_options,
search_options, field_name);
+ let index_type = index_entries
+ .iter()
+ .filter(|entry| entry.kind == FileKind::Add)
+ .find(|entry| {
+
VectorIndexBackend::from_index_type(&entry.index_file.index_type).is_some()
+ && entry
+ .index_file
+ .global_index_meta
+ .as_ref()
+ .is_some_and(|meta| meta.index_field_id == field_id)
+ })
+ .map(|entry| entry.index_file.index_type.as_str())
+ .or(configured_type.as_deref());
let check = VectorMetricCheck {
persisted: Arc::new(std::sync::Mutex::new(None)),
- requested: configured_raw_vector_metric_override(search_options,
field_name)?,
+ requested: vector_metric_override(search_options, field_name,
index_type)?,
};
let mut persisted = None;
for entry in index_entries {
@@ -1530,9 +1552,18 @@ async fn resolve_raw_vector_metric(
if let Some(metric) = check.requested {
return Ok(metric);
}
+ let table_metric = vector_metric_override(table_options, field_name,
index_type)?;
+ if let Some(metric) = table_metric {
+ return Ok(metric);
+ }
+ // Before files exist, Java creates the configured indexer for raw search.
+ // NativeVectorGlobalIndexer and LuminaVectorIndexOptions both default to
IP.
Ok(
- configured_raw_vector_metric_override(table_options, field_name)?
- .unwrap_or(RawVectorMetric::L2),
+ if index_type.is_some_and(|kind|
VectorIndexBackend::from_index_type(kind).is_some()) {
+ RawVectorMetric::InnerProduct
+ } else {
+ RawVectorMetric::L2
+ },
)
}
@@ -1607,24 +1638,27 @@ async fn read_index_vector_metric(
Ok(None)
}
-fn configured_raw_vector_metric_override(
+fn vector_metric_override(
options: &HashMap<String, String>,
field_name: &str,
+ index_type: Option<&str>,
) -> crate::Result<Option<RawVectorMetric>> {
- let direct_keys = [
- format!("fields.{field_name}.distance.metric"),
- format!("fields.{field_name}.metric"),
- "test.vector.metric".to_string(),
- "lumina.distance.metric".to_string(),
- "distance.metric".to_string(),
- "metric".to_string(),
- ];
- for key in direct_keys {
- if let Some(value) = options.get(&key) {
- return RawVectorMetric::parse(value).map(Some);
- }
+ // A known index family determines its own defaults and option namespace.
+ // Only a search without a family may infer a metric from suffixes.
+ if index_type.is_some() {
+ direct_raw_vector_metric_override(options, field_name, index_type)
+ } else {
+ configured_raw_vector_metric_override(options, field_name)
}
+}
+fn configured_raw_vector_metric_override(
+ options: &HashMap<String, String>,
+ field_name: &str,
+) -> crate::Result<Option<RawVectorMetric>> {
+ if let Some(metric) = direct_raw_vector_metric_override(options,
field_name, None)? {
+ return Ok(Some(metric));
+ }
let mut inferred = None;
for (key, value) in options {
if !(key.ends_with(".distance.metric") || key.ends_with(".metric")) {
@@ -1645,6 +1679,34 @@ fn configured_raw_vector_metric_override(
Ok(inferred)
}
+fn direct_raw_vector_metric_override(
+ options: &HashMap<String, String>,
+ field_name: &str,
+ index_type: Option<&str>,
+) -> crate::Result<Option<RawVectorMetric>> {
+ let mut direct_keys = vec![
+ format!("fields.{field_name}.distance.metric"),
+ format!("fields.{field_name}.metric"),
+ ];
+ if let Some(index_type) = index_type {
+ direct_keys.push(format!("{index_type}.distance.metric"));
+ direct_keys.push(format!("{index_type}.metric"));
+ }
+ direct_keys.extend([
+ "test.vector.metric".to_string(),
+ "lumina.distance.metric".to_string(),
+ "distance.metric".to_string(),
+ "metric".to_string(),
+ ]);
+ for key in direct_keys {
+ if let Some(value) = options.get(&key) {
+ return RawVectorMetric::parse(value).map(Some);
+ }
+ }
+
+ Ok(None)
+}
+
#[derive(Default)]
struct RawVectorReadTiming {
plan: Duration,
diff --git a/crates/paimon/src/table/de_vector_read/tests.rs
b/crates/paimon/src/table/de_vector_read/tests.rs
index 293499f7..a2718e81 100644
--- a/crates/paimon/src/table/de_vector_read/tests.rs
+++ b/crates/paimon/src/table/de_vector_read/tests.rs
@@ -446,6 +446,151 @@ fn
test_configured_raw_vector_metric_precedence_and_conflict_default() {
);
}
+#[tokio::test]
+async fn raw_search_without_files_uses_the_configured_indexer_default() {
+ let file_io = FileIOBuilder::new("memory").build().unwrap();
+ for (index_type, expected) in [
+ (" IVF-FLAT ", RawVectorMetric::InnerProduct),
+ ("diskann", RawVectorMetric::InnerProduct),
+ ("lumina", RawVectorMetric::InnerProduct),
+ ("lumina-vector-ann", RawVectorMetric::InnerProduct),
+ ] {
+ let options = HashMap::from([(
+ "fields.embedding.index-type".to_string(),
+ index_type.to_string(),
+ )]);
+ assert_eq!(
+ resolve_raw_vector_metric(
+ &file_io,
+ "memory:///test_table",
+ &options,
+ &HashMap::new(),
+ &[],
+ 2,
+ "embedding",
+ )
+ .await
+ .unwrap(),
+ expected
+ );
+ let override_options = HashMap::from([("metric".to_string(),
"cosine".to_string())]);
+ assert_eq!(
+ resolve_raw_vector_metric(
+ &file_io,
+ "memory:///test_table",
+ &options,
+ &override_options,
+ &[],
+ 2,
+ "embedding",
+ )
+ .await
+ .unwrap(),
+ RawVectorMetric::Cosine
+ );
+ }
+ let mut table_options = HashMap::from([
+ (
+ "fields.embedding.index-type".to_string(),
+ "ivf-flat".to_string(),
+ ),
+ ("fields.other.distance.metric".to_string(), "l2".to_string()),
+ ]);
+ for (family_metric, expected) in [
+ (None, RawVectorMetric::InnerProduct),
+ (Some("cosine"), RawVectorMetric::Cosine),
+ ] {
+ if let Some(metric) = family_metric {
+ table_options.insert("ivf-flat.distance.metric".to_string(),
metric.to_string());
+ }
+ for query_options in [
+ HashMap::new(),
+ HashMap::from([("fields.other.metric".to_string(),
"l2".to_string())]),
+ ] {
+ assert_eq!(
+ resolve_raw_vector_metric(
+ &file_io,
+ "memory:///test_table",
+ &table_options,
+ &query_options,
+ &[],
+ 2,
+ "embedding",
+ )
+ .await
+ .unwrap(),
+ expected
+ );
+ }
+ }
+}
+
+#[tokio::test]
+async fn mixed_vector_formats_fail_before_opening_files() {
+ let file_io = FileIOBuilder::new("memory").build().unwrap();
+ let fields = vec![make_field(2, "embedding")];
+ let entries = [
+ make_lumina_entry("missing-flat", IVF_FLAT_IDENTIFIER, FileKind::Add,
2),
+ make_lumina_entry("missing-diskann", "diskann", FileKind::Add, 2),
+ ];
+ let query = VectorSearch::new(vec![1.0, 0.0], 1,
"embedding".to_string()).unwrap();
+ let error = evaluate_vector_search(
+ eval_context(&file_io, &HashMap::new(), &fields, Some(10)),
+ &entries,
+ &query,
+ )
+ .await
+ .unwrap_err();
+ assert!(error.to_string().contains("multiple index types"));
+ let mut ignored = entries[1].clone();
+ ignored.kind = FileKind::Delete;
+ assert!(
+ validate_global_vector_index_types(&[entries[0].clone(), ignored], 2,
"embedding").is_ok()
+ );
+ assert!(validate_global_vector_index_types(&entries, 99,
"another_column").is_ok());
+}
+
+#[tokio::test]
+async fn indexed_vector_metric_ignores_unrelated_column_options() {
+ let file_io = FileIOBuilder::new("memory").build().unwrap();
+ let index = build_vindex_segment_bytes("inner_product");
+ file_io
+ .new_output("memory:///test_table/index/test.idx")
+ .unwrap()
+ .write(bytes::Bytes::from(index.clone()))
+ .await
+ .unwrap();
+ let mut entry = make_lumina_entry("test.idx", IVF_FLAT_IDENTIFIER,
FileKind::Add, 2);
+ entry.index_file.file_size = index.len() as i64;
+ let fields = vec![make_field(2, "embedding")];
+ let options = HashMap::from([("fields.other.metric".to_string(),
"l2".to_string())]);
+ let query = VectorSearch::new(vec![1.0, 0.0], 2, "embedding".to_string())
+ .unwrap()
+ .with_options(options.clone());
+ let results = evaluate_batch_vector_search(
+ eval_context(&file_io, &HashMap::new(), &fields, Some(10)),
+ std::slice::from_ref(&entry),
+ &[query],
+ )
+ .await
+ .unwrap();
+ assert_eq!(results[0].len(), 2);
+ assert_eq!(
+ resolve_raw_vector_metric(
+ &file_io,
+ "memory:///test_table",
+ &HashMap::new(),
+ &options,
+ &[entry],
+ 2,
+ "embedding",
+ )
+ .await
+ .unwrap(),
+ RawVectorMetric::InnerProduct
+ );
+}
+
#[tokio::test]
async fn test_resolve_raw_vector_metric_uses_vindex_manifest_metadata() {
let file_io = FileIOBuilder::new("memory").build().unwrap();
diff --git a/crates/paimon/src/table/full_text_index_build_builder.rs
b/crates/paimon/src/table/full_text_index_build_builder.rs
index 49802a1b..39fb547f 100644
--- a/crates/paimon/src/table/full_text_index_build_builder.rs
+++ b/crates/paimon/src/table/full_text_index_build_builder.rs
@@ -421,31 +421,7 @@ impl<'a> TextColumn<'a> {
}
fn validate_table_options(table: &Table, core_options: &CoreOptions) ->
Result<()> {
- if !table.schema().primary_keys().is_empty() {
- return Err(Error::Unsupported {
- message: "Full-text index build does not support primary-key
tables; \
- use 'pk-full-text.index.columns' instead"
- .to_string(),
- });
- }
- if !core_options.row_tracking_enabled() {
- return Err(Error::DataInvalid {
- message: "Full-text index build requires 'row-tracking.enabled' =
'true'".to_string(),
- source: None,
- });
- }
- if !core_options.data_evolution_enabled() {
- return Err(Error::DataInvalid {
- message: "Full-text index build requires 'data-evolution.enabled'
= 'true'".to_string(),
- source: None,
- });
- }
- if !core_options.global_index_enabled() {
- return Err(Error::DataInvalid {
- message: "Full-text index build requires 'global-index.enabled' =
'true'".to_string(),
- source: None,
- });
- }
+ super::global_index_build_common::validate_build_table(table)?;
if core_options.deletion_vectors_enabled() {
return Err(Error::Unsupported {
message:
diff --git a/crates/paimon/src/table/full_text_search_builder/scan.rs
b/crates/paimon/src/table/full_text_search_builder/scan.rs
index 96e49471..3f7926b4 100644
--- a/crates/paimon/src/table/full_text_search_builder/scan.rs
+++ b/crates/paimon/src/table/full_text_search_builder/scan.rs
@@ -95,6 +95,36 @@ pub struct FullTextScanPlan {
}
impl FullTextScanPlan {
+ /// The snapshot captured by the scan; accessing it never resolves latest.
+ pub fn snapshot(&self) -> Option<&crate::spec::Snapshot> {
+ self.table.travel_snapshot()
+ }
+
+ /// Source metadata for inspection or transport to another search reader.
+ pub async fn splits(&self) ->
crate::Result<Vec<crate::table::SearchScanSplit>> {
+ if let Some(plan) = &self.primary_key_plan {
+ return Ok(plan
+ .splits
+ .iter()
+ .map(|split| crate::table::SearchScanSplit::PrimaryKey {
+ data_split: split.data_split.clone(),
+ payloads: split.current_payloads.clone(),
+ uncovered_data_files: split.uncovered_data_files.clone(),
+ row_ranges_by_file: Default::default(),
+ })
+ .collect());
+ }
+ super::super::search_scan_split::global_search_splits(
+ &self.table,
+ &self.entries,
+ &self.context.column,
+ self.context.filter.as_ref(),
+ &Default::default(),
+ true,
+ )
+ .await
+ }
+
pub fn snapshot_id(&self) -> Option<i64> {
self.snapshot_id
}
diff --git a/crates/paimon/src/table/global_index_build_builder.rs
b/crates/paimon/src/table/global_index_build_builder.rs
index a16e6a14..cb203619 100644
--- a/crates/paimon/src/table/global_index_build_builder.rs
+++ b/crates/paimon/src/table/global_index_build_builder.rs
@@ -38,6 +38,10 @@ use crate::{Error, Result};
/// Build global indexes by type, like Java's `GlobalIndexer.create`.
/// Sorting and native writer selection are internal implementation details.
+/// Global builds require a row-tracked unaware-bucket table. Data evolution
+/// and the query-side global-index optimization may both be disabled.
+/// Scalar and vindex builds include physical rows deleted by deletion vectors;
+/// queries filter those rows using the snapshot selected for the query.
pub struct GlobalIndexBuildBuilder<'a> {
table: &'a Table,
columns: Option<IndexColumns>,
@@ -299,6 +303,90 @@ mod tests {
.is_none());
}
+ fn with_table_options(table: &Table, options: HashMap<String, String>) ->
Table {
+ Table::new(
+ table.file_io().clone(),
+ table.identifier().clone(),
+ table.location().to_string(),
+ table.schema().copy_with_replaced_options(options),
+ None,
+ )
+ }
+
+ #[tokio::test]
+ async fn
build_eligibility_uses_table_row_tracking_not_read_optimizations() {
+ for (evolution, enabled) in [(false, true), (true, false), (false,
false)] {
+ let original = empty_table();
+ let mut options = original.schema().options().clone();
+ options.insert("data-evolution.enabled".into(),
evolution.to_string());
+ options.insert("global-index.enabled".into(), enabled.to_string());
+ let table = with_table_options(&original, options);
+ for (kind, column) in [
+ ("btree", "id"),
+ ("bitmap", "id"),
+ ("full-text", "text"),
+ ("ivf-flat", "vector"),
+ ("ivf-pq", "vector"),
+ ("ivf-sq", "vector"),
+ ("ivf-rq", "vector"),
+ ("diskann", "vector"),
+ ] {
+ if !GlobalIndexBuildBuilder::supports_index_type(kind) {
+ continue;
+ }
+ let mut build_options =
HashMap::from([(format!("{kind}.dimension"), "2".into())]);
+ if kind == "ivf-pq" {
+ build_options.insert("ivf-pq.pq.m".into(), "1".into());
+ }
+ if ["btree", "bitmap", "full-text"].contains(&kind) {
+ // Procedure options tune index generation, not table
identity.
+ build_options.insert("row-tracking.enabled".into(),
"false".into());
+ build_options.insert("bucket".into(), "5".into());
+ }
+ let mut builder = table.new_global_index_build_builder();
+ builder
+ .with_index_column(column)
+ .with_index_type(kind)
+ .with_options(build_options);
+ assert!(builder.build().await.unwrap().is_empty(), "{kind}");
+ }
+ }
+ }
+
+ #[tokio::test]
+ async fn build_options_cannot_enable_row_tracking_or_change_table_bucket()
{
+ for kind in ["btree", "bitmap", "full-text", "ivf-flat"] {
+ if !GlobalIndexBuildBuilder::supports_index_type(kind) {
+ continue;
+ }
+ for (key, value, error) in [
+ ("row-tracking.enabled", "false", "row-tracking.enabled"),
+ ("bucket", "4", "unaware-bucket"),
+ ] {
+ let original = empty_table();
+ let mut options = original.schema().options().clone();
+ options.insert("data-evolution.enabled".into(),
"false".into());
+ options.insert(key.into(), value.into());
+ let table = with_table_options(&original, options);
+ let result = table
+ .new_global_index_build_builder()
+ .with_index_type(kind)
+ .with_index_column(if kind == "ivf-flat" { "vector" } else
{ "text" })
+ .with_options(HashMap::from([
+ ("row-tracking.enabled".into(), "true".into()),
+ ("bucket".into(), "-1".into()),
+ ("ivf-flat.dimension".into(), "2".into()),
+ ]))
+ .build()
+ .await;
+ assert!(
+ result.unwrap_err().to_string().contains(error),
+ "{kind}: {error}"
+ );
+ }
+ }
+ }
+
#[tokio::test]
async fn column_validation_precedes_any_index_io() {
let table = empty_table();
diff --git a/crates/paimon/src/table/global_index_build_common.rs
b/crates/paimon/src/table/global_index_build_common.rs
index 8466624e..be66da97 100644
--- a/crates/paimon/src/table/global_index_build_common.rs
+++ b/crates/paimon/src/table/global_index_build_common.rs
@@ -38,6 +38,33 @@ pub(crate) mod vector;
const COPY_BUFFER_SIZE: usize = 1024 * 1024;
+/// Validate the table that owns global row IDs. Procedure options only tune
+/// index generation; they cannot enable row tracking or change bucket layout.
+/// Data evolution and `global-index.enabled` govern reads, not index creation.
+pub(crate) fn validate_build_table(table: &Table) -> Result<()> {
+ let options = table.schema().core_options();
+ if !table.schema().primary_keys().is_empty() {
+ return Err(Error::Unsupported {
+ message: "Global index build does not support primary-key tables;
use primary-key file indexes instead".into(),
+ });
+ }
+ if !options.row_tracking_enabled() {
+ return Err(Error::DataInvalid {
+ message: "Global index build requires 'row-tracking.enabled' =
'true'".into(),
+ source: None,
+ });
+ }
+ if options.bucket() != -1 {
+ return Err(Error::Unsupported {
+ message: format!(
+ "Global index build only supports unaware-bucket tables
(bucket = -1), got {}",
+ options.bucket()
+ ),
+ });
+ }
+ Ok(())
+}
+
/// Stream a locally built index file to its table location. Native index
/// builders (Lumina, full-text) serialize to a local temporary file first
/// because their writers need synchronous I/O.
diff --git a/crates/paimon/src/table/global_index_drop_builder.rs
b/crates/paimon/src/table/global_index_drop_builder.rs
index d0bb2bf5..ddd7ca4e 100644
--- a/crates/paimon/src/table/global_index_drop_builder.rs
+++ b/crates/paimon/src/table/global_index_drop_builder.rs
@@ -46,6 +46,11 @@ impl<'a> GlobalIndexDropBuilder<'a> {
}
}
+ /// Whether this index identity can be selected by this dropper.
+ pub fn supports_index_type(index_type: &str) -> bool {
+ normalize_global_index_type(index_type.trim()).is_some()
+ }
+
pub fn with_index_column(&mut self, column: &str) -> &mut Self {
self.index_column = Some(IndexColumns::Column(column.to_string()));
self
@@ -80,19 +85,49 @@ impl<'a> GlobalIndexDropBuilder<'a> {
self
}
+ /// Prepare metadata deletions without publishing a snapshot or removing
files.
+ /// The caller decides whether to commit the returned messages.
+ pub async fn build(&self) -> Result<Vec<CommitMessage>> {
+ Ok(self.prepare().await?.1)
+ }
+
pub async fn execute(&self) -> Result<usize> {
+ let (snapshot_id, messages) = self.prepare().await?;
+ let dropped = messages
+ .iter()
+ .map(|message| message.deleted_index_files.len())
+ .sum();
+ if self.dry_run || dropped == 0 {
+ return Ok(dropped);
+ }
+ TableCommit::new(
+ self.table.clone(),
+ format!(
+ "global-index-{}-drop-{}",
+ self.index_type,
+ uuid::Uuid::new_v4()
+ ),
+ )
+ .commit_if_latest_snapshot(messages, snapshot_id.expect("nonempty drop
has a snapshot"))
+ .await?;
+ // A failed commit may have been published. Never remove index files
here.
+ Ok(dropped)
+ }
+
+ async fn prepare(&self) -> Result<(Option<i64>, Vec<CommitMessage>)> {
// Dropping an index reads the index manifest.
crate::spec::CoreOptions::new(self.table.schema().options()).ensure_read_authorized()?;
self.table.ensure_not_branch_reference_for_write()?;
- let index_type =
- normalize_global_index_type(&self.index_type).ok_or_else(||
Error::Unsupported {
+ let index_type =
normalize_global_index_type(self.index_type.trim()).ok_or_else(|| {
+ Error::Unsupported {
message: format!(
"unsupported global index type '{}'; supported: {}",
self.index_type, SUPPORTED_GLOBAL_INDEX_TYPES_FOR_DROP
),
- })?;
+ }
+ })?;
let index_column = self
.index_column
.as_ref()
@@ -130,10 +165,10 @@ impl<'a> GlobalIndexDropBuilder<'a> {
// through the catalog, like reads and commit validation do.
let snapshot_manager = self.table.snapshot_manager();
let Some(snapshot) = snapshot_manager.get_latest_snapshot().await?
else {
- return Ok(0);
+ return Ok((None, vec![]));
};
let Some(index_manifest_name) = snapshot.index_manifest() else {
- return Ok(0);
+ return Ok((Some(snapshot.id()), vec![]));
};
let index_entries = IndexManifest::read(
@@ -143,7 +178,6 @@ impl<'a> GlobalIndexDropBuilder<'a> {
.await?;
let mut deletions_by_partition_bucket: HashMap<(Vec<u8>, i32),
Vec<IndexFileMeta>> =
HashMap::new();
- let mut dropped = 0;
for entry in index_entries {
if entry.kind != FileKind::Add {
continue;
@@ -176,16 +210,11 @@ impl<'a> GlobalIndexDropBuilder<'a> {
continue;
}
}
- dropped += 1;
deletions_by_partition_bucket
.entry((entry.partition, entry.bucket))
.or_default()
.push(entry.index_file);
}
- if self.dry_run || dropped == 0 {
- return Ok(dropped);
- }
-
let mut groups = deletions_by_partition_bucket
.into_iter()
.collect::<Vec<_>>();
@@ -205,14 +234,7 @@ impl<'a> GlobalIndexDropBuilder<'a> {
})
.collect::<Vec<_>>();
- TableCommit::new(
- self.table.clone(),
- format!("global-index-{}-drop-{}", index_type,
uuid::Uuid::new_v4()),
- )
- .commit_if_latest_snapshot(messages, snapshot.id())
- .await?;
-
- Ok(dropped)
+ Ok((Some(snapshot.id()), messages))
}
}
@@ -581,6 +603,85 @@ mod tests {
.unwrap()
}
+ #[tokio::test]
+ async fn build_prepares_only_metadata_deletions_until_explicit_commit() {
+ for dry_run in [false, true] {
+ let table =
test_table(&format!("memory:/prepared_index_drop_{dry_run}"));
+ setup_dirs(&table).await;
+ let mut message = CommitMessage::new(vec![], 0,
vec![data_file("data.parquet")]);
+ message.new_index_files = vec![
+ global_index_file("btree", "id.index", 0, 0, 9),
+ global_index_file("bitmap", "id-bitmap.index", 0, 0, 9),
+ global_index_file("btree", "name.index", 1, 0, 9),
+ ];
+ TableCommit::new(table.clone(), "prepare-test".into())
+ .commit(vec![message])
+ .await
+ .unwrap();
+ let snapshot_id = table
+ .snapshot_manager()
+ .get_latest_snapshot_id()
+ .await
+ .unwrap();
+ let before = table_files(&table).await;
+ let mut builder = table.new_global_index_drop_builder();
+ builder.with_index_column("id").with_dry_run(dry_run);
+ let messages = builder.build().await.unwrap();
+ assert_eq!(messages.len(), 1);
+ assert!(messages[0].new_files.is_empty());
+ assert!(messages[0].new_index_files.is_empty());
+ assert_eq!(messages[0].deleted_index_files.len(), 1);
+ assert_eq!(messages[0].deleted_index_files[0].file_name,
"id.index");
+ assert_eq!(latest_index_entries(&table).await.len(), 3);
+ assert_eq!(table_files(&table).await, before);
+ assert_eq!(
+ table
+ .snapshot_manager()
+ .get_latest_snapshot_id()
+ .await
+ .unwrap(),
+ snapshot_id
+ );
+ // build() is always preparation, even on a builder configured for
+ // execute()'s dry run; only this explicit call publishes
deletions.
+ TableCommit::new(table.clone(), "explicit-drop".into())
+ .commit(messages)
+ .await
+ .unwrap();
+ let mut names = latest_index_entries(&table)
+ .await
+ .into_iter()
+ .map(|entry| entry.index_file.file_name)
+ .collect::<Vec<_>>();
+ names.sort();
+ assert_eq!(names, vec!["id-bitmap.index", "name.index"]);
+ assert!(builder.build().await.unwrap().is_empty());
+ }
+ }
+
+ #[tokio::test]
+ async fn
build_validates_empty_tables_and_authorization_without_publication() {
+ let table = test_table("memory:/prepared_index_drop_empty");
+ setup_dirs(&table).await;
+ assert!(table.new_global_index_drop_builder().build().await.is_err());
+ assert!(table
+ .new_global_index_drop_builder()
+ .with_index_column("id")
+ .build()
+ .await
+ .unwrap()
+ .is_empty());
+ assert!(crate::table::query_auth_table()
+ .new_global_index_drop_builder()
+ .with_index_column("id")
+ .build()
+ .await
+ .is_err());
+ assert!(GlobalIndexDropBuilder::supports_index_type(" bTrEe "));
+ assert!(GlobalIndexDropBuilder::supports_index_type("FULL-TEXT"));
+ assert!(!GlobalIndexDropBuilder::supports_index_type("hash"));
+ }
+
async fn table_files(table: &Table) -> BTreeMap<String, Vec<u8>> {
let mut files = BTreeMap::new();
for status in table
diff --git a/crates/paimon/src/table/mod.rs b/crates/paimon/src/table/mod.rs
index bf59ab56..0608b331 100644
--- a/crates/paimon/src/table/mod.rs
+++ b/crates/paimon/src/table/mod.rs
@@ -139,6 +139,7 @@ mod row_position_selection;
mod row_sidecar;
mod scan_trace;
pub(crate) mod schema_manager;
+mod search_scan_split;
mod shard_table_updater;
pub(crate) mod snapshot_commit;
mod snapshot_manager;
@@ -225,6 +226,7 @@ pub use read_builder::ReadBuilder;
pub use rest_env::RESTEnv;
pub use scan_trace::ScanTrace;
pub use schema_manager::SchemaManager;
+pub use search_scan_split::SearchScanSplit;
pub use shard_table_updater::ShardTableUpdater;
pub use snapshot_commit::{RESTSnapshotCommit, RenamingSnapshotCommit,
SnapshotCommit};
pub use snapshot_manager::SnapshotManager;
diff --git a/crates/paimon/src/table/pk_vector_read.rs
b/crates/paimon/src/table/pk_vector_read.rs
index 7194d35a..38210bab 100644
--- a/crates/paimon/src/table/pk_vector_read.rs
+++ b/crates/paimon/src/table/pk_vector_read.rs
@@ -417,70 +417,7 @@ async fn search_pk_raw_candidates_batch_with_plan(
// them. Partition-only predicates were already applied by the scan. The
// residual depends on the filter and plan, so its file-local ranges are
// shared by all queries. A file with an empty range list is skipped.
- let residual_by_split: Option<Vec<RowRangesByFile>> = match filter {
- Some(filter) => {
- // The whole filter is pushed into scan planning (`PkVectorScan`),
where
- // partition-only conjuncts already prune partitions/files.
Re-applying
- // them as a per-row residual would be redundant, so keep only the
data
- // conjuncts here — a partition-only filter then needs no residual
at
- // all. Mixed partition/data conjuncts stay whole in
`data_predicates`
- // and evaluate against the materialized partition column
(partition
- // columns are physically present in primary-key data files), so
there
- // is no missing-column case to reject.
- let (_partition_predicate, data_predicates) =
split_partition_and_data_predicates(
- filter.clone(),
- table.schema().fields(),
- table.schema().partition_keys(),
- );
- if data_predicates.is_empty() {
- None
- } else {
- let file_predicates = FilePredicates {
- predicates: data_predicates,
- row_filter_factory: None,
- file_fields: table.schema().fields().to_vec(),
- };
- let residual_read_type = widen_scan_fields(&[],
Some(&file_predicates));
- let residual_reader = DataFileReader::new(
- table.file_io().clone(),
- table.schema_manager().clone(),
- table.schema().id(),
- table.schema().fields().to_vec(),
- residual_read_type,
- Vec::new(),
- )
- .with_table_options(table.schema().options().clone());
- let mut per_split = Vec::with_capacity(plan.splits.len());
- for (index, split) in plan.splits.iter().enumerate() {
- // The plan's selection for this split, so the residual is
- // evaluated over the rows an engine-supplied split allows
rather
- // than over the whole file.
- let allowed_rows = plan
- .physical_row_ranges_by_split
- .as_ref()
- .and_then(|per_split| per_split.get(index));
- per_split.push(
- residual_row_ranges_by_file(
- &residual_reader,
- &split.data_split,
- &split.active_files,
- &file_predicates,
- allowed_rows,
- )
- .await?,
- );
- }
- Some(per_split)
- }
- }
- None => None,
- };
- // Preserve the external plan's ranges as well as the residual restriction.
- let row_ranges_by_split = intersect_row_ranges_by_split(
- plan.physical_row_ranges_by_split.as_deref(),
- residual_by_split,
- plan.splits.len(),
- )?;
+ let row_ranges_by_split = filtered_row_ranges(table, filter, plan).await?;
// Build the exact-fallback search on demand: the kernel calls this only
for a
// file it actually searches (uncovered by ANN, residual-allowed, and only
when
@@ -634,6 +571,78 @@ async fn search_pk_candidates_batch_with_plan(
Ok(per_query_candidates)
}
+/// Resolve the plan's residual physical selection once for any search reader.
+pub(super) async fn filtered_row_ranges(
+ table: &Table,
+ filter: Option<&Predicate>,
+ plan: &PkVectorScanPlan,
+) -> crate::Result<Option<Vec<RowRangesByFile>>> {
+ let residual_by_split: Option<Vec<RowRangesByFile>> = match filter {
+ Some(filter) => {
+ // The whole filter is pushed into scan planning (`PkVectorScan`),
where
+ // partition-only conjuncts already prune partitions/files.
Re-applying
+ // them as a per-row residual would be redundant, so keep only the
data
+ // conjuncts here — a partition-only filter then needs no residual
at
+ // all. Mixed partition/data conjuncts stay whole in
`data_predicates`
+ // and evaluate against the materialized partition column
(partition
+ // columns are physically present in primary-key data files), so
there
+ // is no missing-column case to reject.
+ let (_partition_predicate, data_predicates) =
split_partition_and_data_predicates(
+ filter.clone(),
+ table.schema().fields(),
+ table.schema().partition_keys(),
+ );
+ if data_predicates.is_empty() {
+ None
+ } else {
+ let file_predicates = FilePredicates {
+ predicates: data_predicates,
+ row_filter_factory: None,
+ file_fields: table.schema().fields().to_vec(),
+ };
+ let residual_read_type = widen_scan_fields(&[],
Some(&file_predicates));
+ let residual_reader = DataFileReader::new(
+ table.file_io().clone(),
+ table.schema_manager().clone(),
+ table.schema().id(),
+ table.schema().fields().to_vec(),
+ residual_read_type,
+ Vec::new(),
+ )
+ .with_table_options(table.schema().options().clone());
+ let mut per_split = Vec::with_capacity(plan.splits.len());
+ for (index, split) in plan.splits.iter().enumerate() {
+ // The plan's selection for this split, so the residual is
+ // evaluated over the rows an engine-supplied split allows
rather
+ // than over the whole file.
+ let allowed_rows = plan
+ .physical_row_ranges_by_split
+ .as_ref()
+ .and_then(|per_split| per_split.get(index));
+ per_split.push(
+ residual_row_ranges_by_file(
+ &residual_reader,
+ &split.data_split,
+ &split.active_files,
+ &file_predicates,
+ allowed_rows,
+ )
+ .await?,
+ );
+ }
+ Some(per_split)
+ }
+ }
+ None => None,
+ };
+ // Preserve the external plan's ranges as well as the residual restriction.
+ intersect_row_ranges_by_split(
+ plan.physical_row_ranges_by_split.as_deref(),
+ residual_by_split,
+ plan.splits.len(),
+ )
+}
+
/// Read the plan's allowed physical rows and return merged ranges matching the
/// residual, as in Java `PrimaryKeyVectorRead.residualRowRanges`.
///
diff --git a/crates/paimon/src/table/pk_vector_scan.rs
b/crates/paimon/src/table/pk_vector_scan.rs
index 3a17c807..0f6ad524 100644
--- a/crates/paimon/src/table/pk_vector_scan.rs
+++ b/crates/paimon/src/table/pk_vector_scan.rs
@@ -25,8 +25,8 @@ use std::collections::{BTreeMap, HashMap, HashSet};
use indexmap::IndexMap;
use crate::spec::{
- should_read_pk_index_source, BinaryRow, DataFileMeta, FileKind,
IndexManifest, Predicate,
- PrimaryKeyIndexSourceFile, PrimaryKeyIndexSourceMeta,
+ should_read_pk_index_source, BinaryRow, DataFileMeta, FileKind,
IndexFileMeta, IndexManifest,
+ Predicate, PrimaryKeyIndexSourceFile, PrimaryKeyIndexSourceMeta, Snapshot,
};
use crate::table::bucket_filter::split_partition_and_data_predicates;
use crate::table::index_file_path::IndexFileLocation;
@@ -211,6 +211,8 @@ impl BucketAccumulator {
/// The per-bucket search splits produced by planning.
#[derive(Clone)]
pub(crate) struct PkVectorScanPlan {
+ pub snapshot: Option<Snapshot>,
+ pub payloads: HashMap<BucketKey, Vec<IndexFileMeta>>,
// The snapshot the plan resolved during planning (pinned before the index
// manifest is read). It is authoritative even when planning yields zero
// searchable splits, so a cross-route consistency guard can require one
@@ -340,6 +342,8 @@ impl Scan for PkVectorScan {
}
if incremental {
return Ok(PkVectorScanPlan {
+ snapshot: None,
+ payloads: HashMap::new(),
snapshot_id: 0,
splits: Vec::new(),
physical_row_ranges_by_split: None,
@@ -347,6 +351,8 @@ impl Scan for PkVectorScan {
}
let Some(snapshot) = snapshot else {
return Ok(PkVectorScanPlan {
+ snapshot: None,
+ payloads: HashMap::new(),
snapshot_id: 0,
splits: Vec::new(),
physical_row_ranges_by_split: None,
@@ -357,6 +363,7 @@ impl Scan for PkVectorScan {
// Index-manifest scan into filtered ANN payload tuples.
let table_path = self.table.location().trim_end_matches('/');
let mut entries = Vec::new();
+ let mut payloads = HashMap::<BucketKey, Vec<IndexFileMeta>>::new();
if let Some(name) = snapshot.index_manifest() {
let path = snapshot_manager.manifest_path(name);
for entry in IndexManifest::read(self.table.file_io(),
&path).await? {
@@ -381,6 +388,10 @@ impl Scan for PkVectorScan {
if gim.source_meta.is_none() {
continue;
}
+ payloads
+ .entry((entry.partition.clone(), entry.bucket))
+ .or_default()
+ .push(entry.index_file.clone());
let partition =
BinaryRow::from_serialized_bytes(&entry.partition)?;
let file_size = u64::try_from(entry.index_file.file_size)
.map_err(|_| data_invalid("index file size must not be
negative"))?;
@@ -421,6 +432,8 @@ impl Scan for PkVectorScan {
index_file_in_data_file_dir,
)?;
Ok(PkVectorScanPlan {
+ snapshot: Some(snapshot),
+ payloads,
snapshot_id,
splits,
physical_row_ranges_by_split: None,
@@ -594,6 +607,8 @@ fn plan_from_bucket_splits(
.collect::<Vec<_>>();
Ok(PkVectorScanPlan {
+ snapshot: None,
+ payloads: HashMap::new(),
snapshot_id,
splits,
physical_row_ranges_by_split: Some(physical_row_ranges_by_split),
diff --git a/crates/paimon/src/table/search_scan_split.rs
b/crates/paimon/src/table/search_scan_split.rs
new file mode 100644
index 00000000..692ae15f
--- /dev/null
+++ b/crates/paimon/src/table/search_scan_split.rs
@@ -0,0 +1,224 @@
+// 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.
+
+//! Read-only source work retained by a vector or full-text search plan.
+
+use super::bucket_filter::split_partition_and_data_predicates;
+use super::global_index_scanner::unindexed_ranges_for_global_index_entries;
+use super::global_index_types::normalize_queryable_global_index_type;
+use super::row_id_predicate::intersect_sorted_ranges;
+use super::vector_search_common::{configured_vector_index_type,
VectorIndexBackend};
+use super::{find_field_id_by_name, merge_row_ranges, DataSplit, RowRange,
Table};
+use crate::spec::{
+ CoreOptions, FileKind, GlobalIndexSearchMode, IndexFileMeta,
IndexManifestEntry, Predicate,
+};
+use std::collections::{BTreeMap, HashMap, HashSet};
+
+/// Search source metadata, without live readers or filesystem handles.
+/// The global row-ID and primary-key physical address spaces remain distinct.
+#[derive(Clone, Debug)]
+pub enum SearchScanSplit {
+ Index {
+ column: String,
+ row_range: RowRange,
+ index_files: Vec<IndexFileMeta>,
+ scalar_index_files: Vec<IndexFileMeta>,
+ },
+ Raw {
+ row_ranges: Vec<RowRange>,
+ scalar_index_files: Vec<IndexFileMeta>,
+ index_type: Option<String>,
+ },
+ PrimaryKey {
+ data_split: DataSplit,
+ payloads: Vec<IndexFileMeta>,
+ uncovered_data_files: Vec<String>,
+ row_ranges_by_file: HashMap<String, Vec<RowRange>>,
+ },
+}
+
+pub(super) async fn global_search_splits(
+ table: &Table,
+ entries: &[IndexManifestEntry],
+ column: &str,
+ filter: Option<&Predicate>,
+ options: &HashMap<String, String>,
+ full_text: bool,
+) -> crate::Result<Vec<SearchScanSplit>> {
+ let field_id = find_field_id_by_name(table.schema().fields(),
column).ok_or_else(|| {
+ crate::Error::ConfigInvalid {
+ message: format!("Search column '{column}' does not exist"),
+ }
+ })?;
+ let mut effective_options = table.schema().options().clone();
+ effective_options.extend(options.clone());
+ let core = CoreOptions::new(&effective_options);
+ let mode = if full_text {
+ core.full_text_index_search_mode()?
+ } else {
+ core.vector_index_search_mode()?
+ };
+ let belongs = |file: &IndexFileMeta| {
+ file.global_index_meta
+ .as_ref()
+ .is_some_and(|meta| meta.index_field_id == field_id)
+ && if full_text {
+ file.index_type == "full-text"
+ } else {
+ VectorIndexBackend::from_index_type(&file.index_type).is_some()
+ }
+ };
+ let selected = entries
+ .iter()
+ .filter(|entry| entry.kind == FileKind::Add &&
belongs(&entry.index_file))
+ .collect::<Vec<_>>();
+ // Full-text admits raw fallback only after finding its index definition.
+ if full_text && selected.is_empty() {
+ return Ok(vec![]);
+ }
+ let mut filter_fields = Vec::new();
+ let row_filter = filter.and_then(|filter| {
+ let (_, data) = split_partition_and_data_predicates(
+ filter.clone(),
+ table.schema().fields(),
+ table.schema().partition_keys(),
+ );
+ (!data.is_empty()).then(|| Predicate::and(data))
+ });
+ if let Some(filter) = &row_filter {
+ crate::arrow::residual::collect_predicate_leaf_refs(filter, &mut
filter_fields);
+ }
+ let filter_ids = filter_fields
+ .into_iter()
+ .filter_map(|(name, _)| find_field_id_by_name(table.schema().fields(),
name))
+ .collect::<HashSet<_>>();
+ let scalar_files = entries
+ .iter()
+ .filter(|entry| entry.kind == FileKind::Add)
+ .map(|entry| &entry.index_file)
+ .filter(|file| {
+ normalize_queryable_global_index_type(&file.index_type).is_some()
+ && file.global_index_meta.as_ref().is_some_and(|meta| {
+ filter_ids.contains(&meta.index_field_id)
+ || meta
+ .extra_field_ids
+ .as_ref()
+ .is_some_and(|ids| ids.iter().any(|id|
filter_ids.contains(id)))
+ })
+ })
+ .cloned()
+ .collect::<Vec<_>>();
+ let mut grouped = BTreeMap::<(i64, i64), Vec<IndexFileMeta>>::new();
+ for entry in selected {
+ let meta = entry.index_file.global_index_meta.as_ref().unwrap();
+ grouped
+ .entry((meta.row_range_start, meta.row_range_end))
+ .or_default()
+ .push(entry.index_file.clone());
+ }
+ let intersecting_scalar = |ranges: &[RowRange]| {
+ scalar_files
+ .iter()
+ .filter(|file| {
+ let meta = file.global_index_meta.as_ref().unwrap();
+ ranges
+ .iter()
+ .any(|range|
range.overlaps_inclusive(meta.row_range_start, meta.row_range_end))
+ })
+ .cloned()
+ .collect::<Vec<_>>()
+ };
+ let mut result = grouped
+ .iter()
+ .map(|(&(from, to), files)| {
+ let row_range = RowRange::new(from, to);
+ SearchScanSplit::Index {
+ column: column.into(),
+ scalar_index_files:
intersecting_scalar(std::slice::from_ref(&row_range)),
+ row_range,
+ index_files: files.clone(),
+ }
+ })
+ .collect::<Vec<_>>();
+ let scalar_mode = row_filter
+ .as_ref()
+ .map(|_| core.scalar_index_search_mode())
+ .transpose()?;
+ let detail_ranges = if mode == GlobalIndexSearchMode::Detail
+ || scalar_mode == Some(GlobalIndexSearchMode::Detail)
+ {
+ // These readers use this same helper against the plan's pinned table.
+ super::de_vector_read::detail_data_ranges_for_table(table).await?
+ } else {
+ vec![]
+ };
+ let next_row_id = table
+ .travel_snapshot()
+ .and_then(|snapshot| snapshot.next_row_id());
+ let mut raw_ranges = unindexed_ranges_for_global_index_entries(
+ entries,
+ &HashSet::from([field_id]),
+ mode,
+ next_row_id,
+ &detail_ranges,
+ belongs,
+ );
+ if !full_text {
+ if let (Some(filter), Some(scalar_mode)) = (&row_filter, scalar_mode) {
+ let mut scalar_ranges =
super::de_vector_read::scalar_unindexed_ranges(
+ super::de_vector_read::VectorSearchEvaluation {
+ table: Some(table),
+ file_io: table.file_io(),
+ table_path: table.location(),
+ table_options: table.schema().options(),
+ schema_fields: table.schema().fields(),
+ next_row_id,
+ filter: Some(filter),
+ },
+ entries,
+ filter,
+ scalar_mode,
+ &detail_ranges,
+ );
+ if mode == GlobalIndexSearchMode::Fast {
+ let indexed = merge_row_ranges(
+ grouped
+ .keys()
+ .map(|&(from, to)| RowRange::new(from, to))
+ .collect(),
+ );
+ scalar_ranges = intersect_sorted_ranges(&scalar_ranges,
&indexed);
+ }
+ raw_ranges.extend(scalar_ranges);
+ }
+ }
+ let raw_ranges = merge_row_ranges(raw_ranges);
+ if !raw_ranges.is_empty() {
+ let index_type = grouped
+ .values()
+ .flat_map(|files| files.iter())
+ .next()
+ .map(|file| file.index_type.clone())
+ .or_else(|| configured_vector_index_type(table.schema().options(),
options, column));
+ result.push(SearchScanSplit::Raw {
+ scalar_index_files: intersecting_scalar(&raw_ranges),
+ row_ranges: raw_ranges,
+ index_type,
+ });
+ }
+ Ok(result)
+}
diff --git a/crates/paimon/src/table/sorted_global_index_build_builder.rs
b/crates/paimon/src/table/sorted_global_index_build_builder.rs
index 1c116235..598f2c2e 100644
--- a/crates/paimon/src/table/sorted_global_index_build_builder.rs
+++ b/crates/paimon/src/table/sorted_global_index_build_builder.rs
@@ -147,7 +147,7 @@ impl<'a> SortedGlobalIndexBuildBuilder<'a> {
let mut resolved_options = self.table.schema().options().clone();
resolved_options.extend(self.options.clone());
let core_options = CoreOptions::new(&resolved_options);
- validate_table_options(self.table, &core_options)?;
+ super::global_index_build_common::validate_build_table(self.table)?;
let records_per_range = core_options.sorted_index_records_per_range()?;
let write_options = if index_type == FM_GLOBAL_INDEX_TYPE {
GlobalIndexWriteOptions::FM(FMOptions::from_options(&resolved_options)?.write)
@@ -200,7 +200,7 @@ impl<'a> SortedGlobalIndexBuildBuilder<'a> {
self.table.location(),
self.table.schema().partition_keys(),
self.table.schema().fields(),
- &core_options,
+ &self.table.schema().core_options(),
snapshot.id(),
manifest_entries,
records_per_range,
diff --git
a/crates/paimon/src/table/sorted_global_index_build_builder/extraction.rs
b/crates/paimon/src/table/sorted_global_index_build_builder/extraction.rs
index 69f0fdae..a31fb165 100644
--- a/crates/paimon/src/table/sorted_global_index_build_builder/extraction.rs
+++ b/crates/paimon/src/table/sorted_global_index_build_builder/extraction.rs
@@ -179,6 +179,9 @@ pub(super) fn build_read_splits_for_shard(
return None;
}
let raw_convertible = files.len() == 1;
+ // Like Java's global-index build splits, deliberately omit
deletion
+ // files: indexes cover physical row IDs. Readers apply the query
+ // snapshot's deletion vectors, including deletes after this build.
Some(
DataSplitBuilder::new()
.with_snapshot(shard.snapshot_id)
diff --git a/crates/paimon/src/table/sorted_global_index_build_builder/tests.rs
b/crates/paimon/src/table/sorted_global_index_build_builder/tests.rs
index 8f9008a5..74ac730d 100644
--- a/crates/paimon/src/table/sorted_global_index_build_builder/tests.rs
+++ b/crates/paimon/src/table/sorted_global_index_build_builder/tests.rs
@@ -690,6 +690,146 @@ async fn scan_ids(table: &Table, predicate: Predicate) ->
Vec<i32> {
ids
}
+#[tokio::test]
+async fn build_options_do_not_change_source_data_paths() {
+ for kind in [BTREE_GLOBAL_INDEX_TYPE, BITMAP_GLOBAL_INDEX_TYPE] {
+ let mut options = table_options("10");
+ options.insert("data-file.path-directory".into(),
"actual-data".into());
+ options.insert("data-evolution.enabled".into(), "false".into());
+ let table =
+
test_table_with_path(&format!("memory:/index_build_source_path_{kind}"),
options);
+ setup_dirs(&table).await;
+ let writer = table.new_write_builder();
+ let mut write = writer.new_write().unwrap();
+ write
+ .write_arrow_batch(&data_batch(vec![1, 2], vec!["a", "b"]))
+ .await
+ .unwrap();
+ writer
+ .new_commit()
+ .commit(write.prepare_commit().await.unwrap())
+ .await
+ .unwrap();
+ let count = table
+ .new_global_index_build_builder()
+ .with_index_column("name")
+ .with_index_type(kind)
+ .with_options(HashMap::from([(
+ "data-file.path-directory".into(),
+ "wrong-directory".into(),
+ )]))
+ .execute()
+ .await
+ .unwrap();
+ assert_eq!(count, 1);
+ let predicate = PredicateBuilder::new(table.schema().fields())
+ .equal("name", Datum::String("b".into()))
+ .unwrap();
+ assert_eq!(scan_ids(&table, predicate).await, vec![2]);
+ }
+}
+
+#[tokio::test]
+async fn scalar_build_with_deletion_vectors_preserves_physical_coverage() {
+ for kind in [BTREE_GLOBAL_INDEX_TYPE, BITMAP_GLOBAL_INDEX_TYPE] {
+ for deleted in [vec![0, 2], vec![0, 1, 2, 3]] {
+ let mut options = table_options("2");
+ options.insert("deletion-vectors.enabled".into(), "true".into());
+ options.insert("global-index.search-mode".into(), "full".into());
+ let table = test_table_with_path(
+ &format!("memory:/scalar_build_dv_{kind}_{}", deleted.len()),
+ options,
+ );
+ setup_dirs(&table).await;
+ let writer = table.new_write_builder();
+ let mut write = writer.new_write().unwrap();
+ write
+ .write_arrow_batch(&data_batch(vec![1, 2, 3, 4], vec!["a",
"b", "a", "b"]))
+ .await
+ .unwrap();
+ writer
+ .new_commit()
+ .commit(write.prepare_commit().await.unwrap())
+ .await
+ .unwrap();
+ let delete = writer.new_update().unwrap();
+ writer
+ .new_commit()
+
.commit(delete.delete_by_row_id(deleted.clone()).await.unwrap())
+ .await
+ .unwrap();
+
+ let mut builder = table.new_global_index_build_builder();
+ builder.with_index_column("name").with_index_type(kind);
+ let messages = builder.build().await.unwrap();
+ let files = messages
+ .iter()
+ .flat_map(|m| &m.new_index_files)
+ .collect::<Vec<_>>();
+ // Like Java, index writers consume the physical rows, even when
+ // all rows are deleted. The query snapshot owns live-row
filtering.
+ assert_eq!(files.iter().map(|f| f.row_count).sum::<i64>(), 4);
+ assert_eq!(
+ merge_row_ranges(
+ files
+ .iter()
+ .map(|f| {
+ let meta = f.global_index_meta.as_ref().unwrap();
+ RowRange::new(meta.row_range_start,
meta.row_range_end)
+ })
+ .collect()
+ ),
+ vec![RowRange::new(0, 3)]
+ );
+ writer.new_commit().commit(messages).await.unwrap();
+ let predicates = PredicateBuilder::new(table.schema().fields());
+ let filter = predicates.equal("name",
Datum::String("b".into())).unwrap();
+ assert_eq!(
+ scan_ids(&table, filter.clone()).await,
+ if deleted.len() == 2 {
+ vec![2, 4]
+ } else {
+ vec![]
+ }
+ );
+ assert!(builder.build().await.unwrap().is_empty());
+
+ // A delete after preparation does not invalidate a physical index,
+ // and subsequent appends still receive their original global IDs.
+ let mut write = writer.new_write().unwrap();
+ write
+ .write_arrow_batch(&data_batch(vec![5, 6], vec!["b", "a"]))
+ .await
+ .unwrap();
+ writer
+ .new_commit()
+ .commit(write.prepare_commit().await.unwrap())
+ .await
+ .unwrap();
+ let messages = builder.build().await.unwrap();
+ assert_eq!(
+ messages
+ .iter()
+ .flat_map(|m| &m.new_index_files)
+ .map(|f| f.row_count)
+ .sum::<i64>(),
+ 2
+ );
+ writer
+ .new_commit()
+ .commit(delete.delete_by_row_id(vec![1, 4]).await.unwrap())
+ .await
+ .unwrap();
+ writer.new_commit().commit(messages).await.unwrap();
+ assert_eq!(
+ scan_ids(&table, filter).await,
+ if deleted.len() == 2 { vec![4] } else { vec![] }
+ );
+ assert!(builder.build().await.unwrap().is_empty());
+ }
+ }
+}
+
#[tokio::test]
async fn test_execute_writes_btree_index_manifest_and_file() {
for version in [1, 2] {
diff --git
a/crates/paimon/src/table/sorted_global_index_build_builder/validation.rs
b/crates/paimon/src/table/sorted_global_index_build_builder/validation.rs
index c1e64c1f..ae90987a 100644
--- a/crates/paimon/src/table/sorted_global_index_build_builder/validation.rs
+++ b/crates/paimon/src/table/sorted_global_index_build_builder/validation.rs
@@ -17,48 +17,12 @@
//! Build-option, field-type, and row-range validation.
-use crate::spec::{CoreOptions, DataField, DataType};
+use crate::spec::{DataField, DataType};
use crate::table::global_index_types::{FM_GLOBAL_INDEX_TYPE,
MULTIVALUE_GLOBAL_INDEX_TYPE};
+#[cfg(test)]
use crate::table::Table;
use crate::{Error, Result};
-pub(super) fn validate_table_options(table: &Table, core_options:
&CoreOptions) -> Result<()> {
- if !core_options.row_tracking_enabled() {
- return Err(Error::DataInvalid {
- message: "Sorted global index build requires
'row-tracking.enabled' = 'true'"
- .to_string(),
- source: None,
- });
- }
- if !core_options.data_evolution_enabled() {
- return Err(Error::DataInvalid {
- message: "Sorted global index build requires
'data-evolution.enabled' = 'true'"
- .to_string(),
- source: None,
- });
- }
- if !core_options.global_index_enabled() {
- return Err(Error::DataInvalid {
- message: "Sorted global index build requires
'global-index.enabled' = 'true'"
- .to_string(),
- source: None,
- });
- }
- if !table.schema().primary_keys().is_empty() {
- return Err(Error::Unsupported {
- message: "Sorted global index build does not support primary-key
tables".to_string(),
- });
- }
- if core_options.deletion_vectors_enabled() {
- return Err(Error::Unsupported {
- message:
- "Sorted global index build does not support tables with
deletion-vectors.enabled=true"
- .to_string(),
- });
- }
- Ok(())
-}
-
#[cfg(test)]
pub(super) fn find_index_field<'a>(table: &'a Table, column: &str) ->
Result<&'a DataField> {
table
diff --git a/crates/paimon/src/table/vector_read.rs
b/crates/paimon/src/table/vector_read.rs
index c8ae740d..5feef481 100644
--- a/crates/paimon/src/table/vector_read.rs
+++ b/crates/paimon/src/table/vector_read.rs
@@ -101,7 +101,7 @@ impl BatchVectorRead {
reader.read(*plan).await
}
(VectorReadKind::PrimaryKey(reader),
VectorScanWork::PrimaryKey(plan)) => {
- reader.read(plan).await
+ reader.read(*plan).await
}
_ => Err(crate::Error::DataInvalid {
message: "vector plan and reader use different DE/PK address
spaces".to_string(),
diff --git a/crates/paimon/src/table/vector_scan.rs
b/crates/paimon/src/table/vector_scan.rs
index 39683352..f8411fb0 100644
--- a/crates/paimon/src/table/vector_scan.rs
+++ b/crates/paimon/src/table/vector_scan.rs
@@ -22,11 +22,15 @@ use std::sync::Arc;
use roaring::RoaringTreemap;
-use crate::spec::{CoreOptions, Predicate};
+use crate::spec::{CoreOptions, Predicate, Snapshot};
use crate::table::de_vector_scan::{DeVectorScan, DeVectorScanPlan,
PreparedVectorSearchFilter};
use crate::table::pk_vector_scan::{PkVectorScan, PkVectorScanPlan};
-use crate::table::vector_search_common::targets_primary_key_column;
+use crate::table::vector_search_common::{
+ targets_primary_key_column, validate_global_vector_index_types,
+};
+use crate::table::SearchScanSplit;
use crate::table::{find_field_id_by_name, BucketVectorSearchSplit, Table};
+use std::collections::{HashMap, HashSet};
/// Resolve one snapshot and the work a vector reader will consume.
///
@@ -44,15 +48,123 @@ pub(super) trait Scan: Sync {
pub struct VectorScanPlan {
pub(super) context: PlanContext,
pub(super) work: VectorScanWork,
+ table: Table,
}
#[derive(Clone)]
pub(super) enum VectorScanWork {
DataEvolution(Box<DeVectorScanPlan>),
- PrimaryKey(PkVectorScanPlan),
+ PrimaryKey(Box<PkVectorScanPlan>),
}
impl VectorScanPlan {
+ /// The snapshot captured by planning, including retained tag snapshots.
+ pub fn snapshot(&self) -> Option<&Snapshot> {
+ match &self.work {
+ VectorScanWork::DataEvolution(plan) =>
plan.table.travel_snapshot(),
+ VectorScanWork::PrimaryKey(plan) => plan.snapshot.as_ref(),
+ }
+ }
+
+ /// Inspect source work using the scan's effective search options. This
does
+ /// not resolve a newer snapshot or reread its index manifest. Query
readers
+ /// may still use different query vectors, limits and search options.
+ pub async fn splits(
+ &self,
+ options: &HashMap<String, String>,
+ ) -> crate::Result<Vec<SearchScanSplit>> {
+ match &self.work {
+ VectorScanWork::DataEvolution(plan) => {
+ if plan.skip_search {
+ return Ok(vec![]);
+ }
+ super::search_scan_split::global_search_splits(
+ &plan.table,
+ &plan.index_entries,
+ &self.context.column,
+ self.context.filter.as_ref(),
+ options,
+ false,
+ )
+ .await
+ }
+ VectorScanWork::PrimaryKey(plan) => {
+ if plan.snapshot.is_none()
+ && plan
+ .splits
+ .iter()
+ .any(|split| !split.ann_segments.is_empty())
+ {
+ return Err(crate::Error::Unsupported {
+ message:
+ "Source metadata export requires a
manifest-planned primary-key search"
+ .into(),
+ });
+ }
+ let selected = super::pk_vector_read::filtered_row_ranges(
+ &self.table,
+ self.context.filter.as_ref(),
+ plan,
+ )
+ .await?;
+ plan.splits
+ .iter()
+ .enumerate()
+ .map(|(index, split)| {
+ let payloads = plan
+ .payloads
+ .get(&(
+
split.data_split.partition().to_serialized_bytes(),
+ split.data_split.bucket(),
+ ))
+ .into_iter()
+ .flatten()
+ .filter(|payload| {
+ split.ann_segments.iter().any(|segment| {
+ let location =
+
super::index_file_path::IndexFileLocation::BucketLocal {
+ table_path: &self.context.location,
+ bucket_path:
split.data_split.bucket_path(),
+ index_file_in_data_file_dir: false,
+ };
+ segment.path
+ == location.resolve(
+ &payload.file_name,
+ payload.external_path.as_deref(),
+ )
+ || segment
+ .path
+ .ends_with(&format!("/{}",
payload.file_name))
+ })
+ })
+ .cloned()
+ .collect::<Vec<_>>();
+ let covered = split
+ .ann_segments
+ .iter()
+ .flat_map(|segment|
segment.source_meta.source_files())
+ .map(|file| file.file_name())
+ .collect::<HashSet<_>>();
+ Ok(SearchScanSplit::PrimaryKey {
+ data_split: split.data_split.clone(),
+ payloads,
+ uncovered_data_files: split
+ .active_files
+ .iter()
+ .filter(|file|
!covered.contains(file.file_name.as_str()))
+ .map(|file| file.file_name.clone())
+ .collect(),
+ row_ranges_by_file: selected
+ .as_ref()
+ .map(|ranges| ranges[index].clone())
+ .unwrap_or_default(),
+ })
+ })
+ .collect()
+ }
+ }
+ }
+
pub fn snapshot_id(&self) -> Option<i64> {
match &self.work {
VectorScanWork::DataEvolution(plan) =>
plan.table.travel_snapshot().map(|s| s.id()),
@@ -122,6 +234,7 @@ impl PlanContext {
/// Creates query-independent plans for DE or primary-key vector search.
pub struct VectorScan {
+ table: Table,
context: PlanContext,
scan: VectorScanKind,
}
@@ -164,17 +277,34 @@ impl VectorScan {
prepared,
)))
};
- Ok(Self { context, scan })
+ Ok(Self {
+ table: table.clone(),
+ context,
+ scan,
+ })
}
pub async fn plan(&self) -> crate::Result<VectorScanPlan> {
let work = match &self.scan {
VectorScanKind::DataEvolution(scan) => {
- VectorScanWork::DataEvolution(Box::new(scan.plan().await?))
+ let plan = scan.plan().await?;
+ if let Some(field_id) =
+ find_field_id_by_name(plan.table.schema().fields(),
&self.context.column)
+ {
+ validate_global_vector_index_types(
+ &plan.index_entries,
+ field_id,
+ &self.context.column,
+ )?;
+ }
+ VectorScanWork::DataEvolution(Box::new(plan))
+ }
+ VectorScanKind::PrimaryKey(scan) => {
+ VectorScanWork::PrimaryKey(Box::new(scan.plan().await?))
}
- VectorScanKind::PrimaryKey(scan) =>
VectorScanWork::PrimaryKey(scan.plan().await?),
};
Ok(VectorScanPlan {
+ table: self.table.clone(),
context: self.context.clone(),
work,
})
@@ -194,8 +324,9 @@ impl VectorScan {
});
};
Ok(VectorScanPlan {
+ table: self.table.clone(),
context: self.context.clone(),
- work:
VectorScanWork::PrimaryKey(scan.plan_for_bucket_vector_splits(splits)?),
+ work:
VectorScanWork::PrimaryKey(Box::new(scan.plan_for_bucket_vector_splits(splits)?)),
})
}
}
diff --git a/crates/paimon/src/table/vector_search_common.rs
b/crates/paimon/src/table/vector_search_common.rs
index 10dcab6f..9fd80e60 100644
--- a/crates/paimon/src/table/vector_search_common.rs
+++ b/crates/paimon/src/table/vector_search_common.rs
@@ -18,7 +18,9 @@
//! Shared vector-search option validation, index I/O helpers, and result
ordering.
use crate::lumina::is_lumina_index_type;
-use crate::spec::{CoreOptions, DataField, Predicate, ROW_ID_FIELD_NAME};
+use crate::spec::{
+ CoreOptions, DataField, FileKind, IndexManifestEntry, Predicate,
ROW_ID_FIELD_NAME,
+};
use crate::table::pk_vector_position_read::{PKEY_VECTOR_POSITION_COLUMN,
SEARCH_SCORE_COLUMN};
use crate::table::read_builder::resolve_projected_fields;
use crate::table::Table;
@@ -62,6 +64,58 @@ impl VectorIndexBackend {
}
}
+/// A search column selects one persisted vector format, as in Java's scan.
+/// Different columns, removed entries and scalar indexes do not participate.
+pub(super) fn validate_global_vector_index_types(
+ entries: &[IndexManifestEntry],
+ field_id: i32,
+ column: &str,
+) -> crate::Result<()> {
+ let mut index_type = None;
+ for entry in entries {
+ let file = &entry.index_file;
+ if entry.kind != FileKind::Add
+ || VectorIndexBackend::from_index_type(&file.index_type).is_none()
+ || !file
+ .global_index_meta
+ .as_ref()
+ .is_some_and(|meta| meta.index_field_id == field_id)
+ {
+ continue;
+ }
+ if let Some(previous) = index_type {
+ if previous != file.index_type.as_str() {
+ return Err(crate::Error::DataInvalid {
+ message: format!(
+ "Vector column '{column}' has multiple index types:
{previous} and {}",
+ file.index_type
+ ),
+ source: None,
+ });
+ }
+ } else {
+ index_type = Some(file.index_type.as_str());
+ }
+ }
+ Ok(())
+}
+
+pub(super) fn configured_vector_index_type(
+ table_options: &HashMap<String, String>,
+ query_options: &HashMap<String, String>,
+ column: &str,
+) -> Option<String> {
+ [
+ "index_type".to_string(),
+ "index-type".into(),
+ "vector.index-type".into(),
+ format!("fields.{column}.index-type"),
+ ]
+ .iter()
+ .find_map(|key| query_options.get(key).or_else(|| table_options.get(key)))
+ .map(|value| value.trim().to_ascii_lowercase())
+}
+
pub(super) fn current_tokio_runtime_handle() ->
crate::Result<tokio::runtime::Handle> {
tokio::runtime::Handle::try_current().map_err(|error|
crate::Error::UnexpectedError {
message: "Vector index range reader requires a Tokio
runtime".to_string(),
diff --git a/crates/paimon/src/table/vindex_index_build_builder.rs
b/crates/paimon/src/table/vindex_index_build_builder.rs
index ddb4470e..bfc1ecf6 100644
--- a/crates/paimon/src/table/vindex_index_build_builder.rs
+++ b/crates/paimon/src/table/vindex_index_build_builder.rs
@@ -23,7 +23,7 @@ mod validation;
mod writer;
use planning::plan_vindex_shards;
-use validation::{checked_i32, find_index_field, validate_table_options,
validate_vector_field};
+use validation::{checked_i32, find_index_field, validate_vector_field};
use crate::spec::{CoreOptions, Predicate};
use crate::table::{CommitMessage, RowRange, Table};
@@ -90,7 +90,7 @@ impl<'a> VindexIndexBuildBuilder<'a> {
let mut merged_options = self.table.schema().options().clone();
merged_options.extend(self.options.clone());
let core_options = CoreOptions::new(&merged_options);
- validate_table_options(self.table, &core_options)?;
+ super::global_index_build_common::validate_build_table(self.table)?;
let rows_per_shard = core_options.global_index_row_count_per_shard()?;
let parallelism = core_options.global_index_build_parallelism()?;
@@ -138,7 +138,7 @@ impl<'a> VindexIndexBuildBuilder<'a> {
self.table.location(),
self.table.schema().partition_keys(),
self.table.schema().fields(),
- &core_options,
+ &self.table.schema().core_options(),
snapshot.id(),
manifest_entries,
rows_per_shard,
diff --git a/crates/paimon/src/table/vindex_index_build_builder/extraction.rs
b/crates/paimon/src/table/vindex_index_build_builder/extraction.rs
index b8aabaf7..49a4351a 100644
--- a/crates/paimon/src/table/vindex_index_build_builder/extraction.rs
+++ b/crates/paimon/src/table/vindex_index_build_builder/extraction.rs
@@ -34,6 +34,9 @@ pub(super) fn data_split_for_shard_ranges(
shard: &VindexIndexShard,
row_ranges: Vec<RowRange>,
) -> Result<DataSplit> {
+ // Java builds over stable physical row IDs, including deleted vectors.
+ // Do not attach deletion files: live-row filtering belongs to the query
+ // snapshot, and filtering here would leave holes in the training input.
DataSplitBuilder::new()
.with_snapshot(shard.snapshot_id)
.with_partition(shard.partition.clone())
diff --git a/crates/paimon/src/table/vindex_index_build_builder/tests.rs
b/crates/paimon/src/table/vindex_index_build_builder/tests.rs
index 998791ef..617a8dbc 100644
--- a/crates/paimon/src/table/vindex_index_build_builder/tests.rs
+++ b/crates/paimon/src/table/vindex_index_build_builder/tests.rs
@@ -481,6 +481,120 @@ async fn write_vectors(table: &Table, ids: Vec<i32>,
vectors: Vec<Vec<f32>>) {
.unwrap();
}
+#[tokio::test]
+async fn vector_build_with_deletion_vectors_preserves_physical_coverage() {
+ for granule in [false, true] {
+ for deleted in [vec![0, 2], vec![0, 1, 2, 3]] {
+ let path = format!("memory:/vector_build_dv_{granule}_{}",
deleted.len());
+ let mut options = vindex_e2e_options("10");
+ options.insert("ivf-flat.metric".into(), "l2".into());
+ options.insert("deletion-vectors.enabled".into(), "true".into());
+ options.insert("vindex.build.granule.enabled".into(),
granule.to_string());
+ let table = test_table_with_io(
+ FileIOBuilder::new("memory").build().unwrap(),
+ &path,
+ vindex_schema_builder(options).build().unwrap(),
+ );
+ setup_dirs(table.file_io(), &path).await;
+ write_vectors(
+ &table,
+ vec![1, 2, 3, 4],
+ vec![vec![1., 0.], vec![0., 1.], vec![1., 1.], vec![2., 2.]],
+ )
+ .await;
+ let writer = table.new_write_builder();
+ let delete = writer.new_update().unwrap();
+ writer
+ .new_commit()
+
.commit(delete.delete_by_row_id(deleted.clone()).await.unwrap())
+ .await
+ .unwrap();
+ let mut builder = table.new_global_index_build_builder();
+ builder
+ .with_index_type(IVF_FLAT_IDENTIFIER)
+ .with_index_column("embedding");
+ let messages = builder.build().await.unwrap();
+ let files = messages
+ .iter()
+ .flat_map(|m| &m.new_index_files)
+ .collect::<Vec<_>>();
+ assert_eq!(files.len(), 1);
+ assert_eq!(files[0].row_count, 4);
+ let meta = files[0].global_index_meta.as_ref().unwrap();
+ assert_eq!((meta.row_range_start, meta.row_range_end), (0, 3));
+ writer.new_commit().commit(messages).await.unwrap();
+ let result = table
+ .new_vector_search_builder()
+ .with_vector_column("embedding")
+ .with_query_vector(vec![1., 1.])
+ .with_limit(10)
+ .with_options(HashMap::from([("ivf.nprobe".into(),
"10".into())]))
+ .execute()
+ .await
+ .unwrap();
+ let mut ids = result.row_ids().unwrap().row_ids.clone();
+ ids.sort_unstable();
+ assert_eq!(
+ ids,
+ if deleted.len() == 2 {
+ vec![1, 3]
+ } else {
+ vec![]
+ }
+ );
+ // The closest physical vector (row 2) is deleted. Top-K must
+ // continue to a live candidate rather than return an empty result.
+ let result = table
+ .new_vector_search_builder()
+ .with_vector_column("embedding")
+ .with_query_vector(vec![1., 1.])
+ .with_limit(1)
+ .with_options(HashMap::from([("ivf.nprobe".into(),
"10".into())]))
+ .execute()
+ .await
+ .unwrap();
+ assert_eq!(
+ result.row_ids().unwrap().row_ids,
+ if deleted.len() == 2 { vec![1] } else { vec![] }
+ );
+ assert!(builder.build().await.unwrap().is_empty());
+
+ write_vectors(&table, vec![5, 6], vec![vec![3., 3.], vec![4.,
4.]]).await;
+ let messages = builder.build().await.unwrap();
+ let file = &messages[0].new_index_files[0];
+ assert_eq!(file.row_count, 2);
+ let meta = file.global_index_meta.as_ref().unwrap();
+ assert_eq!((meta.row_range_start, meta.row_range_end), (4, 5));
+ writer
+ .new_commit()
+ .commit(delete.delete_by_row_id(vec![1, 4]).await.unwrap())
+ .await
+ .unwrap();
+ writer.new_commit().commit(messages).await.unwrap();
+ let result = table
+ .new_vector_search_builder()
+ .with_vector_column("embedding")
+ .with_query_vector(vec![1., 1.])
+ .with_limit(10)
+ .with_options(HashMap::from([("ivf.nprobe".into(),
"10".into())]))
+ .execute()
+ .await
+ .unwrap();
+ let mut ids = result.row_ids().unwrap().row_ids.clone();
+ ids.sort_unstable();
+ assert_eq!(
+ ids,
+ if deleted.len() == 2 {
+ vec![3, 5]
+ } else {
+ vec![5]
+ }
+ );
+ assert!(builder.build().await.unwrap().is_empty());
+ }
+ }
+}
+
/// Commit a synthetic vindex `IndexFileMeta` covering `[start, end]` for
/// `field_id` directly into the index manifest, without invoking the native
/// builder. Mirrors the Lumina/btree tests so the incremental gap logic can
diff --git a/crates/paimon/src/table/vindex_index_build_builder/validation.rs
b/crates/paimon/src/table/vindex_index_build_builder/validation.rs
index f1c5270d..298cbc42 100644
--- a/crates/paimon/src/table/vindex_index_build_builder/validation.rs
+++ b/crates/paimon/src/table/vindex_index_build_builder/validation.rs
@@ -15,7 +15,7 @@
// specific language governing permissions and limitations
// under the License.
-use crate::spec::{CoreOptions, DataField};
+use crate::spec::DataField;
use crate::table::global_index_build_common::vector::{
find_index_field as find_vector_index_field,
validate_vector_field as validate_common_vector_field,
@@ -23,40 +23,6 @@ use crate::table::global_index_build_common::vector::{
use crate::table::Table;
use crate::{Error, Result};
-pub(super) fn validate_table_options(table: &Table, core_options:
&CoreOptions) -> Result<()> {
- if !core_options.row_tracking_enabled() {
- return Err(Error::DataInvalid {
- message: "vindex index build requires 'row-tracking.enabled' =
'true'".to_string(),
- source: None,
- });
- }
- if !core_options.data_evolution_enabled() {
- return Err(Error::DataInvalid {
- message: "vindex index build requires 'data-evolution.enabled' =
'true'".to_string(),
- source: None,
- });
- }
- if !core_options.global_index_enabled() {
- return Err(Error::DataInvalid {
- message: "vindex index build requires 'global-index.enabled' =
'true'".to_string(),
- source: None,
- });
- }
- if !table.schema().primary_keys().is_empty() {
- return Err(Error::Unsupported {
- message: "vindex index build does not support primary-key
tables".to_string(),
- });
- }
- if core_options.deletion_vectors_enabled() {
- return Err(Error::Unsupported {
- message:
- "vindex index build does not support tables with
deletion-vectors.enabled=true"
- .to_string(),
- });
- }
- Ok(())
-}
-
pub(super) fn find_index_field<'a>(table: &'a Table, column: &str) ->
Result<&'a DataField> {
find_vector_index_field(table, column)
}
diff --git a/crates/paimon/src/vindex/reader.rs
b/crates/paimon/src/vindex/reader.rs
index 557c2ee9..a203fe76 100644
--- a/crates/paimon/src/vindex/reader.rs
+++ b/crates/paimon/src/vindex/reader.rs
@@ -267,15 +267,15 @@ impl VindexVectorGlobalIndexReader {
Ok(results)
}
- #[cfg(test)]
- pub(crate) fn load<S: SeekRead + 'static>(
+ /// Open and optimize the index without executing a query.
+ pub fn load<S: SeekRead + 'static>(
&mut self,
stream_fn: impl FnOnce(&str) -> crate::Result<S>,
) -> crate::Result<()> {
self.ensure_loaded(stream_fn, |_| Ok(()))
}
- pub(crate) fn metadata(&self) -> crate::Result<&VectorIndexMetadata> {
+ pub fn metadata(&self) -> crate::Result<&VectorIndexMetadata> {
self.metadata
.as_ref()
.ok_or_else(|| crate::Error::DataInvalid {
@@ -454,6 +454,15 @@ fn prepare_search_with_shared_filter(
vector_search: &VectorSearch,
shared_filter_bytes: Option<&Arc<[u8]>>,
) -> crate::Result<Option<PreparedSearch>> {
+ // Query options override the defaults of a reusable index reader.
+ let mut merged_options;
+ let options = if vector_search.options.is_empty() {
+ options
+ } else {
+ merged_options = options.clone();
+ merged_options.extend(vector_search.options.clone());
+ &merged_options
+ };
if vector_search.vector.len() != metadata.dimension {
return Err(crate::Error::DataInvalid {
message: format!(
@@ -470,30 +479,32 @@ fn prepare_search_with_shared_filter(
if top_k == 0 {
return Ok(None);
}
- let mut params = match metadata.index_type {
- IndexType::DiskAnn => match options.get(L_SEARCH_PARAMETER) {
- Some(value) => {
- let l_search = value
- .parse::<usize>()
- .ok()
- .filter(|value| *value > 0)
- .ok_or_else(|| crate::Error::DataInvalid {
- message: format!(
- "Invalid value for '{}': {}. Must be a positive
integer.",
- L_SEARCH_PARAMETER, value
- ),
- source: None,
- })?;
- VectorSearchParams::with_l_search(top_k, l_search)
- }
- None => VectorSearchParams::automatic(top_k),
- },
- _ => match options.get(NPROBE_PARAMETER) {
- Some(value) => {
- VectorSearchParams::new(top_k,
parse_int_parameter(NPROBE_PARAMETER, value)?)
+ let mut params = match (
+ options.get(NPROBE_PARAMETER),
+ options.get(L_SEARCH_PARAMETER),
+ ) {
+ (Some(_), Some(_)) => {
+ return Err(crate::Error::DataInvalid {
+ message: format!(
+ "Cannot set both '{NPROBE_PARAMETER}' and
'{L_SEARCH_PARAMETER}'."
+ ),
+ source: None,
+ });
+ }
+ (Some(value), None) => {
+ VectorSearchParams::new(top_k,
parse_search_width(NPROBE_PARAMETER, value)?)
+ }
+ (None, Some(value)) => {
+ let l_search = parse_search_width(L_SEARCH_PARAMETER, value)?;
+ if l_search == 0 {
+ return Err(crate::Error::DataInvalid {
+ message: format!("'{L_SEARCH_PARAMETER}' must be a
positive integer"),
+ source: None,
+ });
}
- None => VectorSearchParams::automatic(top_k),
- },
+ VectorSearchParams::with_l_search(top_k, l_search)
+ }
+ (None, None) => VectorSearchParams::automatic(top_k),
};
let filter_bytes = if let Some(include_ids) =
vector_search.effective_include_row_ids() {
@@ -849,6 +860,30 @@ fn int_parameter(
}
}
+// Java's NativeVectorGlobalIndexReader uses Integer.parseInt, including BMP
+// decimal digits, with no whitespace/fraction coercion or values beyond i32.
+fn parse_search_width(key: &str, value: &str) -> crate::Result<usize> {
+ let text = value
+ .chars()
+ .enumerate()
+ .map(|(index, ch)| {
+ if index == 0 && matches!(ch, '+' | '-') {
+ Some(ch)
+ } else {
+ crate::common::number::java_decimal_digit(ch).map(|digit|
char::from(b'0' + digit))
+ }
+ })
+ .collect::<Option<String>>();
+ text.and_then(|text| text.parse::<i32>().ok())
+ .and_then(|width| usize::try_from(width).ok())
+ .ok_or_else(|| crate::Error::DataInvalid {
+ message: format!(
+ "Invalid value for '{key}': {value}. Must be a non-negative
32-bit integer."
+ ),
+ source: None,
+ })
+}
+
fn parse_int_parameter(key: &str, value: &str) -> crate::Result<usize> {
value
.parse::<usize>()
@@ -864,6 +899,28 @@ fn parse_int_parameter(key: &str, value: &str) ->
crate::Result<usize> {
#[cfg(test)]
mod tests {
use super::*;
+
+ #[test]
+ fn query_widths_follow_java_integer_text_and_bounds() {
+ for key in [NPROBE_PARAMETER, L_SEARCH_PARAMETER] {
+ for (text, expected) in [
+ ("+12", 12),
+ ("１２", 12),
+ ("٢", 2),
+ ("2147483647", 2147483647),
+ ] {
+ assert_eq!(parse_search_width(key, text).unwrap(), expected);
+ }
+ for text in ["2147483648", " 2", "2 ", "2.0", "1_0", "-1", "𝟙",
"+", ""] {
+ assert!(parse_search_width(key, text).is_err(), "{text}");
+ }
+ }
+ // Reader memory budgets are usize byte counts, not Java search widths.
+ assert_eq!(
+ parse_int_parameter(READER_MEMORY_BUDGET_PARAMETER,
"2147483648").unwrap(),
+ 2147483648
+ );
+ }
use crate::io::FileRead;
use crate::vindex::range_reader::VindexFileReader;
use async_trait::async_trait;
@@ -1051,6 +1108,33 @@ mod tests {
assert!(!result.contains_key(&3));
}
+ #[test]
+ fn reusable_reader_query_options_override_defaults_without_leaking() {
+ let metadata = VectorIndexMetadata {
+ index_type: IndexType::IvfFlat,
+ dimension: TEST_DIMENSION,
+ nlist: 16,
+ metric: MetricType::L2,
+ total_vectors: 8192,
+ pq_m: None,
+ pq_bits: None,
+ rq_bits: None,
+ diskann: None,
+ };
+ let defaults = HashMap::from([(NPROBE_PARAMETER.into(), "1".into())]);
+ let explicit =
query().with_options(HashMap::from([(NPROBE_PARAMETER.into(), "7".into())]));
+ let prepared = prepare_batch_searches(&metadata, &defaults,
&[explicit, query()]).unwrap();
+ assert_eq!(prepared[0].as_ref().unwrap().params.width, 7);
+ assert_eq!(prepared[1].as_ref().unwrap().params.width, 1);
+ let conflicting =
+ query().with_options(HashMap::from([(L_SEARCH_PARAMETER.into(),
"32".into())]));
+ assert!(prepare_search(&metadata, &defaults, &conflicting)
+ .err()
+ .expect("conflicting query widths must be rejected")
+ .to_string()
+ .contains("Cannot set both"));
+ }
+
#[test]
fn native_batch_chunk_size_tracks_working_set_inputs() {
let base_metadata = VectorIndexMetadata {
@@ -1321,25 +1405,11 @@ mod tests {
("ivf.nprobe".to_string(), "4".to_string()),
("diskann.l_search".to_string(), "64".to_string()),
]);
- let diskann = prepare_search(&metadata, &mixed_options, &query())
- .unwrap()
- .unwrap();
- assert_eq!(
- diskann.params.search_width,
- paimon_vindex_core::index::SearchWidth::DiskAnnLSearch
- );
- assert_eq!(diskann.params.width, 64);
+ assert!(prepare_search(&metadata, &mixed_options, &query()).is_err());
let mut ivf_metadata = metadata;
ivf_metadata.index_type = IndexType::IvfFlat;
- let ivf = prepare_search(&ivf_metadata, &mixed_options, &query())
- .unwrap()
- .unwrap();
- assert_eq!(
- ivf.params.search_width,
- paimon_vindex_core::index::SearchWidth::IvfNProbe
- );
- assert_eq!(ivf.params.width, 4);
+ assert!(prepare_search(&ivf_metadata, &mixed_options,
&query()).is_err());
}
#[test]
