fix(table): align native vector index configuration semantics

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)] {

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(),

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

@@ -942,6 +942,78 @@ 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<_>>(),
+        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
+        .expect_err("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;

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() {

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()

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<MetricType> {
+    if let Some(index_meta) = index_meta {
         if let Ok(options) = serde_json::from_slice::<HashMap<String, String>>(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<IndexVectorSemantics> {
+    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()

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<String, String>,
     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"