From b39aed164ca8043645e6410644f31d4e4c072875 Mon Sep 17 00:00:00 2001 From: JingsongLi Date: Thu, 15 Oct 2026 23:47:43 +0800 Subject: [PATCH 1/2] feat: support deletion vectors in global index builds --- .../src/table/global_index_build_builder.rs | 2 + .../extraction.rs | 3 + .../tests.rs | 101 ++++++++++++++++ .../validation.rs | 7 -- .../vindex_index_build_builder/extraction.rs | 3 + .../table/vindex_index_build_builder/tests.rs | 114 ++++++++++++++++++ .../vindex_index_build_builder/validation.rs | 7 -- 7 files changed, 223 insertions(+), 14 deletions(-) diff --git a/crates/paimon/src/table/global_index_build_builder.rs b/crates/paimon/src/table/global_index_build_builder.rs index a16e6a14a..eaeac4dc8 100644 --- a/crates/paimon/src/table/global_index_build_builder.rs +++ b/crates/paimon/src/table/global_index_build_builder.rs @@ -38,6 +38,8 @@ use crate::{Error, Result}; /// Build global indexes by type, like Java's `GlobalIndexer.create`. /// Sorting and native writer selection are internal implementation details. +/// 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, 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 69f0fdae6..a31fb165b 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 8f9008a59..5c59a0dbb 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,107 @@ async fn scan_ids(table: &Table, predicate: Predicate) -> Vec { ids } +#[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::>(); + // 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::(), 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::(), + 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 c1e64c1f3..3fd3a24e4 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 @@ -49,13 +49,6 @@ pub(super) fn validate_table_options(table: &Table, core_options: &CoreOptions) 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(()) } 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 b8aabaf7e..49a4351ae 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, ) -> Result { + // 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 998791ef3..617a8dbcd 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, vectors: Vec>) { .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::>(); + 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 f1c5270dc..235df4494 100644 --- a/crates/paimon/src/table/vindex_index_build_builder/validation.rs +++ b/crates/paimon/src/table/vindex_index_build_builder/validation.rs @@ -47,13 +47,6 @@ pub(super) fn validate_table_options(table: &Table, core_options: &CoreOptions) 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(()) } From 48b99acd4e02f5077931f906b58b5873a3bc7fdb Mon Sep 17 00:00:00 2001 From: JingsongLi Date: Fri, 16 Oct 2026 00:44:10 +0800 Subject: [PATCH 2/2] Complete native global-index lifecycle and search planning --- Cargo.lock | 2 + bindings/python/Cargo.toml | 2 + .../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 +++++++--- .../paimon/src/table/de_vector_read/tests.rs | 145 ++++++++++++ .../table/full_text_index_build_builder.rs | 26 +- .../table/full_text_search_builder/scan.rs | 30 +++ .../src/table/global_index_build_builder.rs | 86 +++++++ .../src/table/global_index_build_common.rs | 27 +++ .../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 ++++++++++++++++++ .../sorted_global_index_build_builder.rs | 4 +- .../tests.rs | 39 +++ .../validation.rs | 33 +-- crates/paimon/src/table/vector_read.rs | 2 +- crates/paimon/src/table/vector_scan.rs | 145 +++++++++++- .../paimon/src/table/vector_search_common.rs | 56 ++++- .../src/table/vindex_index_build_builder.rs | 6 +- .../vindex_index_build_builder/validation.rs | 29 +-- crates/paimon/src/vindex/reader.rs | 154 ++++++++---- 31 files changed, 1591 insertions(+), 259 deletions(-) create mode 100644 bindings/python/src/vindex.rs create mode 100644 crates/paimon/src/table/search_scan_split.rs diff --git a/Cargo.lock b/Cargo.lock index e1b2da67d..0efb91445 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 2739e62d8..733b7b6d3 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 5b132e812..3af2afc51 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 95b9b38d0..c5cd2f913 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::() { + 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 d312192a4..edd492055 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::()?; this.add_class::()?; this.add_class::()?; + this.add_class::()?; this.add_class::()?; this.add_class::()?; this.add_class::()?; this.add_class::()?; this.add_class::()?; this.add_class::()?; + this.add_class::()?; this.add_class::()?; this.add_class::()?; this.add_class::()?; diff --git a/bindings/python/src/full_text_search.rs b/bindings/python/src/full_text_search.rs index a3fc6a9bd..acdc76e94 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 { + 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 { self.inner.snapshot_id() } diff --git a/bindings/python/src/index_build.rs b/bindings/python/src/index_build.rs index d81e72528..ff5079c60 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, + columns: Option, + index_type: String, + partition_filters: Vec, + dry_run: bool, +} + +impl PyGlobalIndexDropBuilder { + pub(crate) fn new(table: Arc
) -> Self { + Self { + table, + columns: None, + index_type: "btree".into(), + partition_filters: vec![], + dry_run: false, + } + } + + fn builder(&self) -> paimon::Result> { + 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::>()); + } + 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) -> 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> { + 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> { + 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 { + 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 4ae51342b..9b92ae834 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 6eb496755..14e4298f0 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 4c1173f22..8b7d42904 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, + splits: Vec, +) -> PyResult { + 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::>>() + }; + 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::>>()?; + 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, @@ -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 { 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 { 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, } #[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, } #[pymethods] impl PyVectorScanPlan { + fn _metadata(&self, py: Python<'_>) -> PyResult { + 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 { self.inner.snapshot_id() } diff --git a/bindings/python/src/vindex.rs b/bindings/python/src/vindex.rs new file mode 100644 index 000000000..58cd9d2a5 --- /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); + +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::>(); + let chunks = self + .0 + .bind(py) + .call_method1("pread_many", (requests,))? + .extract::>>()?; + 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>, + 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, + file_path: String, + file_size: u64, + options: Option>, + ) -> PyResult { + 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 { + 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>, + limit: usize, + filter_bytes: Option>, + options: Option>, + ) -> PyResult>>> { + 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::>>() + .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::(&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 6de0f5f3d..a96f2f8be 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( } #[derive(Clone, Copy)] -struct VectorSearchEvaluation<'a> { - table: Option<&'a Table>, - file_io: &'a FileIO, - table_path: &'a str, - table_options: &'a HashMap, - schema_fields: &'a [DataField], - next_row_id: Option, - 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, + pub(super) schema_fields: &'a [DataField], + pub(super) next_row_id: Option, + 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::>() }; + 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> { +pub(super) async fn detail_data_ranges_for_table(table: &Table) -> crate::Result> { 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 { + 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, field_name: &str, + index_type: Option<&str>, ) -> crate::Result> { - 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, + field_name: &str, +) -> crate::Result> { + 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, + field_name: &str, + index_type: Option<&str>, +) -> crate::Result> { + 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 293499f7d..a2718e81e 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 49802a1b5..39fb547fc 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 96e494716..3f7926b43 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> { + 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 { 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 eaeac4dc8..cb2036193 100644 --- a/crates/paimon/src/table/global_index_build_builder.rs +++ b/crates/paimon/src/table/global_index_build_builder.rs @@ -38,6 +38,8 @@ 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> { @@ -301,6 +303,90 @@ mod tests { .is_none()); } + fn with_table_options(table: &Table, options: HashMap) -> 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 8466624e3..be66da97b 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 d0bb2bf59..ddd7ca4e6 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> { + Ok(self.prepare().await?.1) + } + pub async fn execute(&self) -> Result { + 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, Vec)> { // 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, i32), Vec> = 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::>(); @@ -205,14 +234,7 @@ impl<'a> GlobalIndexDropBuilder<'a> { }) .collect::>(); - 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::>(); + 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> { 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 bf59ab569..0608b3316 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 7194d35a6..38210babb 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> = 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>> { + let residual_by_split: Option> = 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 3a17c8078..0f6ad524c 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, + pub payloads: HashMap>, // 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::>::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::>(); 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 000000000..692ae15f8 --- /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, + scalar_index_files: Vec, + }, + Raw { + row_ranges: Vec, + scalar_index_files: Vec, + index_type: Option, + }, + PrimaryKey { + data_split: DataSplit, + payloads: Vec, + uncovered_data_files: Vec, + row_ranges_by_file: HashMap>, + }, +} + +pub(super) async fn global_search_splits( + table: &Table, + entries: &[IndexManifestEntry], + column: &str, + filter: Option<&Predicate>, + options: &HashMap, + full_text: bool, +) -> crate::Result> { + 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::>(); + // 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::>(); + 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::>(); + let mut grouped = BTreeMap::<(i64, i64), Vec>::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::>() + }; + 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::>(); + 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 1c1162359..598f2c2ef 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/tests.rs b/crates/paimon/src/table/sorted_global_index_build_builder/tests.rs index 5c59a0dbb..74ac730d6 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,45 @@ async fn scan_ids(table: &Table, predicate: Predicate) -> Vec { 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] { 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 3fd3a24e4..ae90987ad 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,41 +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(), - }); - } - 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 c8ae740d2..5feef4816 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 396833522..f8411fb0e 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), - PrimaryKey(PkVectorScanPlan), + PrimaryKey(Box), } 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, + ) -> crate::Result> { + 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::>(); + let covered = split + .ann_segments + .iter() + .flat_map(|segment| segment.source_meta.source_files()) + .map(|file| file.file_name()) + .collect::>(); + 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 { 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 { 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 10dcab6f1..9fd80e606 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, + query_options: &HashMap, + column: &str, +) -> Option { + [ + "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::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 ddb4470e5..bfc1ecf69 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/validation.rs b/crates/paimon/src/table/vindex_index_build_builder/validation.rs index 235df4494..298cbc427 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,33 +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(), - }); - } - 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 557c2ee91..a203fe767 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( + /// Open and optimize the index without executing a query. + pub fn load( &mut self, stream_fn: impl FnOnce(&str) -> crate::Result, ) -> 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> { + // 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::() - .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 { + 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::>(); + text.and_then(|text| text.parse::().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 { value .parse::() @@ -864,6 +899,28 @@ fn parse_int_parameter(key: &str, value: &str) -> crate::Result { #[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]