From 091253d1f318688dce9f3f48de849c5784b20dbd Mon Sep 17 00:00:00 2001 From: JingsongLi Date: Sun, 1 Nov 2026 20:03:34 +0800 Subject: [PATCH 1/2] fix(table): align native vector index configuration semantics --- .../src/table/global_index_build_builder.rs | 36 +++++++ crates/paimon/src/table/pk_vector_read.rs | 26 ++++- .../paimon/src/table/pk_vector_read/tests.rs | 73 ++++++++++++++ .../src/table/pk_vector_search_params.rs | 3 +- .../index_options_tests.rs | 14 ++- .../paimon/src/table/vector_search_common.rs | 49 ++++++--- crates/paimon/src/vindex/mod.rs | 9 +- .../tests/pk_vector_java_fixture_test.rs | 99 ++++++++++++++++++- 8 files changed, 287 insertions(+), 22 deletions(-) diff --git a/crates/paimon/src/table/global_index_build_builder.rs b/crates/paimon/src/table/global_index_build_builder.rs index d9968f081..8c07ff867 100644 --- a/crates/paimon/src/table/global_index_build_builder.rs +++ b/crates/paimon/src/table/global_index_build_builder.rs @@ -374,6 +374,42 @@ mod tests { ) } + #[tokio::test] + async fn vector_build_ignores_opaque_parameters_and_validates_known_parameters() { + let table = empty_table(); + for kind in ["ivf-flat", "ivf-pq"] { + let mut options = HashMap::from([ + (format!("{kind}.index.dimension"), "2".into()), + (format!("{kind}.ignored-option"), "ignored".into()), + ("fields.vector.metric".into(), "l2".into()), + ("metric".into(), "invalid".into()), + ("dimension".into(), "999".into()), + ]); + let mut builder = table.new_global_index_build_builder(); + builder + .with_index_column("vector") + .with_index_type(kind) + .with_options(options.clone()); + assert!(builder.build().await.unwrap().is_empty()); + options.insert("fields.vector.metric".into(), "invalid".into()); + let error = builder + .with_options(options.clone()) + .build() + .await + .unwrap_err(); + assert!(error.to_string().contains("metric"), "{error}"); + if kind == "ivf-flat" { + options.insert("fields.vector.metric".into(), "l2".into()); + options.insert("ivf-flat.ivf.pq-encoding".into(), "canonical".into()); + let error = builder.with_options(options).build().await.unwrap_err(); + assert!( + error.to_string().contains("ivf-flat.ivf.pq-encoding"), + "{error}" + ); + } + } + } + #[tokio::test] async fn build_eligibility_uses_table_row_tracking_not_read_optimizations() { for (evolution, enabled) in [(false, true), (true, false), (false, false)] { diff --git a/crates/paimon/src/table/pk_vector_read.rs b/crates/paimon/src/table/pk_vector_read.rs index aa3508b84..16d768719 100644 --- a/crates/paimon/src/table/pk_vector_read.rs +++ b/crates/paimon/src/table/pk_vector_read.rs @@ -38,7 +38,7 @@ use crate::table::source::DataSplit; use crate::table::vector_read::Read; use crate::table::vector_search_common::{ collect_ranked_rows, current_tokio_runtime_handle, log_vindex_range_io_stats, - reorder_and_strip_position, vindex_concurrency_limits, RankedRow, + read_vector_index_metric, reorder_and_strip_position, vindex_concurrency_limits, RankedRow, }; use crate::table::{ArrowRecordBatchStream, RowRange, Table}; use crate::vector_search::{GlobalIndexIOMeta, GlobalIndexResult, VectorSearch}; @@ -105,7 +105,7 @@ impl Read for PkVectorRead { } metadata_type = Some(segment.index_type.as_str()); } - let params = PkVectorSearchParams::resolve( + let mut params = PkVectorSearchParams::resolve( &self.table, &self.options, self.filter.as_ref(), @@ -114,6 +114,28 @@ impl Read for PkVectorRead { self.limit, metadata_type, )?; + // The payload's training metric governs ANN scores, raw fallback and + // refinement. Schema/query options cannot reinterpret an existing index. + // One payload establishes the metric; the search scorer validates every + // other payload against it without serializing extra remote header reads. + if let Some(segment) = plan + .splits + .iter() + .flat_map(|split| &split.ann_segments) + .find(|segment| segment.row_count > 0) + { + let metric = VectorSearchMetric::from_vindex( + read_vector_index_metric( + self.table.file_io(), + &segment.path, + segment.file_size, + Some(&segment.index_meta), + ) + .await?, + ); + params.metric = metric; + params.vector_options.metric = metric.as_str().into(); + } let candidates = search_pk_candidates_batch_with_plan( &self.table, self.filter.as_ref(), diff --git a/crates/paimon/src/table/pk_vector_read/tests.rs b/crates/paimon/src/table/pk_vector_read/tests.rs index fffce91f2..7dbe50be7 100644 --- a/crates/paimon/src/table/pk_vector_read/tests.rs +++ b/crates/paimon/src/table/pk_vector_read/tests.rs @@ -942,6 +942,79 @@ fn verify_segment_metric_rejects_mismatched_vindex_metric() { } // ---- Search results retain scored positions and the planned source context. ---- +#[tokio::test] +async fn persisted_metric_skips_empty_segments_and_rejects_mixed_payloads() { + let table = build_committed_pk_vector_table(&[[1.0, 0.0], [0.0, 1.0], [1.0, 1.0]]).await; + let scan = crate::table::pk_vector_scan::PkVectorScan::new(&table, 1, None); + let plan = scan.plan().await.unwrap(); + let read_table = table.copy_with_options(HashMap::from([( + "fields.embedding.distance.metric".into(), + "inner_product".into(), + )])); + let read = PkVectorRead::new( + &read_table, + &HashMap::new(), + None, + "embedding", + &[&[0.0, 1.0]], + 2, + ); + + // The second bucket is an independent source unit; source file names may + // coincide across buckets without overlapping segment ownership. + let mut other = plan.splits[0].clone(); + other.data_split = DataSplitBuilder::new() + .with_snapshot(plan.snapshot_id) + .with_partition(other.data_split.partition().clone()) + .with_bucket(1) + .with_bucket_path("memory:/pk_vector_route_test/bucket-1".into()) + .with_total_buckets(2) + .with_data_files(other.data_split.data_files().to_vec()) + .build() + .unwrap(); + let mut with_empty = plan.clone(); + let mut empty = other.clone(); + let segment = &mut empty.ann_segments[0]; + segment.row_count = 0; + segment.path = "memory:/missing-empty-payload".into(); + with_empty.splits.insert(0, empty); + let results = read.read(with_empty).await.unwrap(); + assert_eq!( + results[0] + .positions() + .unwrap() + .iter() + .map(|p| p.score) + .collect::>(), + vec![1.0, 0.5] + ); + + let mut mixed = plan; + let bytes = build_vindex_segment_bytes("cosine"); + let segment = &mut other.ann_segments[0]; + segment.path = "memory:/mixed-cosine-payload".into(); + segment.file_size = bytes.len() as u64; + table + .file_io() + .new_output(&segment.path) + .unwrap() + .write(bytes::Bytes::from(bytes)) + .await + .unwrap(); + mixed.splits.push(other); + let error = read + .read(mixed) + .await + .err() + .expect("mixed persisted metrics must fail"); + assert!( + error + .to_string() + .contains("does not match configured metric"), + "{error}" + ); +} + #[tokio::test] async fn reader_uses_planned_splits_after_index_manifest_is_removed() { let table = build_committed_pk_vector_table(&[[1.0, 0.0], [0.0, 1.0], [1.0, 1.0]]).await; diff --git a/crates/paimon/src/table/pk_vector_search_params.rs b/crates/paimon/src/table/pk_vector_search_params.rs index 95daf9e72..72a3f2803 100644 --- a/crates/paimon/src/table/pk_vector_search_params.rs +++ b/crates/paimon/src/table/pk_vector_search_params.rs @@ -115,7 +115,8 @@ impl PkVectorSearchParams { true, )?; let mut vector_options = configuration.resolved; - // Like Java, persisted PK payloads are checked against the table metric. + // Raw-only PK plans use the table metric. The reader replaces it with + // the persisted payload metric when the captured plan has ANN segments. vector_options.metric = configuration.table_metric; let metric = VectorSearchMetric::parse(&vector_options.metric)?; let refine_factor = if let Some(kind) = vector_options.index_type.as_deref() { diff --git a/crates/paimon/src/table/vector_search_builder/index_options_tests.rs b/crates/paimon/src/table/vector_search_builder/index_options_tests.rs index 99e2efa0e..b8ae3182d 100644 --- a/crates/paimon/src/table/vector_search_builder/index_options_tests.rs +++ b/crates/paimon/src/table/vector_search_builder/index_options_tests.rs @@ -645,14 +645,22 @@ async fn raw_build_map_preserves_json_precedence_and_serialized_shards() { assert_eq!(meta["metric"], "cosine"); } } - // Qualified typos still fail at the Rust parameter boundary. - builder.with_options(HashMap::from([("ivf-flat.nlsit".into(), "1".into())])); + // Opaque procedure parameters are ignored, while known invalid values fail. + builder.with_options(HashMap::from([( + "ivf-flat.future-option".into(), + "1".into(), + )])); + assert!(!builder.plan().await.unwrap().shards().is_empty()); + builder.with_options(HashMap::from([( + "ivf-flat.dimension".into(), + "invalid".into(), + )])); assert!(builder .plan() .await .unwrap_err() .to_string() - .contains("ivf-flat.nlsit")); + .contains("dimension")); let empty = table.copy_with_pinned_snapshot(None); let plan = empty .new_global_index_build_builder() diff --git a/crates/paimon/src/table/vector_search_common.rs b/crates/paimon/src/table/vector_search_common.rs index ffc66ff94..9f881c814 100644 --- a/crates/paimon/src/table/vector_search_common.rs +++ b/crates/paimon/src/table/vector_search_common.rs @@ -57,18 +57,37 @@ pub(super) async fn read_index_vector_metric( if !is_vindex_index_type(&entry.index_file.index_type) { return Ok(None); } - if let Some(index_meta) = global_meta.index_meta.as_ref() { + let path = IndexFileLocation::Global { table_path }.resolve( + &entry.index_file.file_name, + entry.index_file.external_path.as_deref(), + ); + read_vector_index_metric( + file_io, + &path, + entry.index_file.file_size.max(0) as u64, + global_meta.index_meta.as_deref(), + ) + .await + .map(Some) +} + +/// Resolve the metric of an already selected payload, including bucket-local PK files. +pub(super) async fn read_vector_index_metric( + file_io: &FileIO, + path: &str, + file_size: u64, + index_meta: Option<&[u8]>, +) -> crate::Result { + if let Some(index_meta) = index_meta { if let Ok(options) = serde_json::from_slice::>(index_meta) { if let Some(metric) = metric_from_index_options(&options) { - return Ok(Some(metric)); + return Ok(metric); } } } - Ok(Some( - read_index_vector_header(file_io, table_path, entry) - .await? - .metric, - )) + Ok(read_vector_index_header(file_io, path, file_size) + .await? + .metric) } /// Read the structural definition of an existing run before reusing its coverage. @@ -120,16 +139,24 @@ async fn read_index_vector_header( &entry.index_file.file_name, entry.index_file.external_path.as_deref(), ); - let input = file_io.new_input(&path)?; + read_vector_index_header(file_io, &path, entry.index_file.file_size.max(0) as u64).await +} + +async fn read_vector_index_header( + file_io: &FileIO, + path: &str, + file_size: u64, +) -> crate::Result { + let input = file_io.new_input(path)?; let read_error = |e| crate::Error::DataInvalid { message: format!( "Failed to read vindex index file '{}' for vector metric: {}", - entry.index_file.file_name, e + path, e ), source: Some(Box::new(e)), }; - let header_size = if entry.index_file.file_size > 0 { - (entry.index_file.file_size as u64).min(DISKANN_HEADER_SIZE as u64) + let header_size = if file_size > 0 { + file_size.min(DISKANN_HEADER_SIZE as u64) } else { input .metadata() diff --git a/crates/paimon/src/vindex/mod.rs b/crates/paimon/src/vindex/mod.rs index 592b7feeb..24b7bb3cc 100644 --- a/crates/paimon/src/vindex/mod.rs +++ b/crates/paimon/src/vindex/mod.rs @@ -267,8 +267,8 @@ fn compile_training_config( } /// The generic build API accepts a complete table option map. Native parameters -/// remain scoped to this implementation/column; qualified typos still reach the -/// strict validator, while unrelated table properties stay in the captured plan. +/// remain scoped to this implementation/column. Recognized parameters reach the +/// strict validator; opaque properties stay in the captured plan, as in Java. pub(crate) fn build_parameters( options: &HashMap, kind: &str, @@ -279,8 +279,9 @@ pub(crate) fn build_parameters( options .iter() .filter(|(key, _)| { - key.starts_with(&prefix) - || native_option_key(key).is_some() + key.strip_prefix(&prefix).is_some_and(|suffix| { + native_option_key(suffix).is_some() || suffix == "train.sample-ratio" + }) || native_option_key(key).is_some() || matches!(key.as_str(), "index.type" | "vindex.build.granule.enabled") || key.strip_prefix(&field).is_some_and(|suffix| { native_option_key(suffix).is_some() || suffix == "train.sample-ratio" diff --git a/crates/paimon/tests/pk_vector_java_fixture_test.rs b/crates/paimon/tests/pk_vector_java_fixture_test.rs index 91cfcd9d7..a6d44c6a4 100644 --- a/crates/paimon/tests/pk_vector_java_fixture_test.rs +++ b/crates/paimon/tests/pk_vector_java_fixture_test.rs @@ -33,13 +33,17 @@ //! `(file, position)`. Search returns those positions with scores; a projected //! result read must retrieve the same rows without inventing global row IDs. +use std::collections::HashMap; use std::path::Path; -use arrow_array::{Array, Float32Array, Int32Array, RecordBatch}; +use arrow_array::builder::{FixedSizeListBuilder, Float32Builder}; +use arrow_array::{Array, ArrayRef, Float32Array, Int32Array, RecordBatch}; +use arrow_schema::{DataType, Field, Schema}; use futures::TryStreamExt; use paimon::catalog::Identifier; use paimon::io::{FileIO, FileIOBuilder}; use paimon::table::{SchemaManager, Table}; +use std::sync::Arc; const VECTOR_COLUMN: &str = "embedding"; const FIXTURE: &str = "testdata/pkvector/pk_vector_ivf_flat"; @@ -224,3 +228,96 @@ async fn reads_back_java_written_pk_vector_table() { ); } } + +#[cfg(not(target_os = "windows"))] +#[tokio::test] +async fn persisted_metric_survives_table_and_query_overrides() { + let (_tmp, original) = open_java_fixture().await; + // Leave a level-0 file outside the Java ANN segment, so raw fallback must + // use the same persisted metric as indexed scoring and refinement. + let element = Arc::new(Field::new("element", DataType::Float32, true)); + let mut vector = + FixedSizeListBuilder::new(Float32Builder::new(), 2).with_field(element.clone()); + vector.values().append_slice(&[0.5, 0.0]); + vector.append(true); + let batch = RecordBatch::try_new( + Arc::new(Schema::new(vec![ + Field::new("id", DataType::Int32, false), + Field::new("embedding", DataType::FixedSizeList(element, 2), true), + ])), + vec![ + Arc::new(Int32Array::from(vec![100])) as ArrayRef, + Arc::new(vector.finish()), + ], + ) + .unwrap(); + let write_builder = original.new_write_builder(); + let mut writer = write_builder.new_write().unwrap(); + writer.write_arrow_batch(&batch).await.unwrap(); + write_builder + .new_commit() + .commit(writer.prepare_commit().await.unwrap()) + .await + .unwrap(); + for metric in ["inner_product", "cosine"] { + let table = original.copy_with_options(HashMap::from([ + ( + "fields.embedding.pk-vector.distance.metric".into(), + metric.into(), + ), + ("fields.embedding.distance.metric".into(), metric.into()), + ])); + for refine in [None, Some("2")] { + let mut options = HashMap::from([("ivf-flat.metric".into(), metric.into())]); + if let Some(refine) = refine { + options.insert("ivf.refine_factor".into(), refine.into()); + } + let mut builder = table.new_vector_search_builder(); + builder + .with_vector_column(VECTOR_COLUMN) + .with_query_vector(vec![0.0, 0.0]) + .with_limit(3) + .with_options(options.clone()); + let result = builder.execute().await.unwrap(); + let positions = result.positions().unwrap(); + assert_eq!( + positions.iter().map(|p| p.score).collect::>(), + vec![1.0, 0.8, 0.5] + ); + let batches = result + .new_read_builder() + .with_projection(&["id"]) + .read() + .await + .unwrap() + .try_collect::>() + .await + .unwrap(); + assert_eq!(batch_i32(&batches, "id"), vec![0, 100, 1]); + + let mut batch = table.new_batch_vector_search_builder(); + batch + .with_vector_column(VECTOR_COLUMN) + .with_query_vectors(vec![vec![0.0, 0.0]; 2]) + .with_limit(3) + .with_options(options); + for result in batch.execute().await.unwrap() { + let positions = result.positions().unwrap(); + assert_eq!( + positions.iter().map(|p| p.score).collect::>(), + vec![1.0, 0.8, 0.5] + ); + let batches = result + .new_read_builder() + .with_projection(&["id"]) + .read() + .await + .unwrap() + .try_collect::>() + .await + .unwrap(); + assert_eq!(batch_i32(&batches, "id"), vec![0, 100, 1]); + } + } + } +} From 7a911b0796e7b296f9fa203abe7dddc8b71746b3 Mon Sep 17 00:00:00 2001 From: JingsongLi Date: Sun, 1 Nov 2026 21:50:36 +0800 Subject: [PATCH 2/2] test(table): satisfy Clippy for mixed-metric error assertion --- crates/paimon/src/table/pk_vector_read/tests.rs | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/crates/paimon/src/table/pk_vector_read/tests.rs b/crates/paimon/src/table/pk_vector_read/tests.rs index 7dbe50be7..ffa02348b 100644 --- a/crates/paimon/src/table/pk_vector_read/tests.rs +++ b/crates/paimon/src/table/pk_vector_read/tests.rs @@ -1005,8 +1005,7 @@ async fn persisted_metric_skips_empty_segments_and_rejects_mixed_payloads() { let error = read .read(mixed) .await - .err() - .expect("mixed persisted metrics must fail"); + .expect_err("mixed persisted metrics must fail"); assert!( error .to_string()