This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-rust.git
The following commit(s) were added to refs/heads/main by this push:
new ddd87921 [c] Support vector search materialized read in C FFI bindings
(#537)
ddd87921 is described below
commit ddd87921d268a420835d2137e6d3847a3db0aa76
Author: Junrui Lee <[email protected]>
AuthorDate: Mon Jul 20 14:48:11 2026 +0800
[c] Support vector search materialized read in C FFI bindings (#537)
---
bindings/c/Cargo.toml | 6 +
bindings/c/src/lib.rs | 1 +
bindings/c/src/result.rs | 6 +
bindings/c/src/table.rs | 3 +-
bindings/c/src/tests.rs | 914 ++++++++++++++++++++-
bindings/c/src/types.rs | 18 +
bindings/c/src/vector_search.rs | 324 ++++++++
.../src/table/pk_vector_indexed_split_read.rs | 12 +-
crates/paimon/src/table/pk_vector_orchestrator.rs | 12 +-
crates/paimon/src/table/pk_vector_position_read.rs | 30 +-
crates/paimon/src/table/vector_search_builder.rs | 779 ++++++++++++++++--
crates/paimon/tests/pk_vector_baseline_test.rs | 16 +-
crates/paimon/tests/pk_vector_java_fixture_test.rs | 4 +-
13 files changed, 2030 insertions(+), 95 deletions(-)
diff --git a/bindings/c/Cargo.toml b/bindings/c/Cargo.toml
index 6eb34c12..41bdd964 100644
--- a/bindings/c/Cargo.toml
+++ b/bindings/c/Cargo.toml
@@ -35,3 +35,9 @@ futures = "0.3"
arrow = { workspace = true }
arrow-array = { workspace = true }
arrow-schema = { workspace = true }
+
+[dev-dependencies]
+# Test-only: the vector-search integration tests build a real primary-key
vindex
+# IVF-flat ANN segment fixture in-process. Versions match crates/paimon.
+bytes = "1.7.1"
+paimon-vindex-core = "0.2.0"
diff --git a/bindings/c/src/lib.rs b/bindings/c/src/lib.rs
index 3e1367ea..27972b5f 100644
--- a/bindings/c/src/lib.rs
+++ b/bindings/c/src/lib.rs
@@ -27,6 +27,7 @@ mod table;
#[cfg(test)]
mod tests;
mod types;
+mod vector_search;
mod write;
use std::sync::OnceLock;
diff --git a/bindings/c/src/result.rs b/bindings/c/src/result.rs
index 216509f5..a833de5e 100644
--- a/bindings/c/src/result.rs
+++ b/bindings/c/src/result.rs
@@ -78,6 +78,12 @@ pub struct paimon_result_next_batch {
pub error: *mut paimon_error,
}
+#[repr(C)]
+pub struct paimon_result_vector_search_builder {
+ pub builder: *mut paimon_vector_search_builder,
+ pub error: *mut paimon_error,
+}
+
// === Write/Commit result types ===
#[repr(C)]
diff --git a/bindings/c/src/table.rs b/bindings/c/src/table.rs
index 4daf61ec..a17ba3fc 100644
--- a/bindings/c/src/table.rs
+++ b/bindings/c/src/table.rs
@@ -658,7 +658,8 @@ pub unsafe extern "C" fn paimon_record_batch_reader_next(
/// Free a paimon_record_batch_reader.
///
/// # Safety
-/// Only call with a reader returned from `paimon_table_read_to_arrow`.
+/// Only call with a reader returned from `paimon_table_read_to_arrow` or
+/// `paimon_vector_search_builder_execute_read`.
#[no_mangle]
pub unsafe extern "C" fn paimon_record_batch_reader_free(reader: *mut
paimon_record_batch_reader) {
if !reader.is_null() {
diff --git a/bindings/c/src/tests.rs b/bindings/c/src/tests.rs
index 72ba5829..2ced3245 100644
--- a/bindings/c/src/tests.rs
+++ b/bindings/c/src/tests.rs
@@ -19,7 +19,8 @@
//!
//! Covers: read path (table, scan, plan, predicates, record batch streaming),
//! write path (write builder, write_arrow_batch, prepare_commit, commit,
-//! overwrite, truncate, abort), and full write->read roundtrip.
+//! overwrite, truncate, abort), full write->read roundtrip, and vector search
+//! materialized reads across primary-key and data-evolution (append) tables.
//!
//! IMPORTANT: C FFI functions internally use `runtime().block_on()`. Tests
//! must NOT wrap C FFI calls inside another `block_on`. Use the global
@@ -43,6 +44,7 @@ use paimon::table::{SnapshotManager, Table};
use crate::error::*;
use crate::table::*;
use crate::types::*;
+use crate::vector_search::*;
use crate::write::*;
// =========================================================================
@@ -1506,3 +1508,913 @@ fn test_two_commits_same_builder() {
unwrap_table(handle);
}
}
+
+// =========================================================================
+// Vector search tests (materialized reads)
+// =========================================================================
+//
+// Two storage shapes are exercised end-to-end through the C `execute_read`
+// terminal, each compared against an independent core Rust
+// `VectorSearchBuilder::execute_read()` reference:
+//
+// * A primary-key vector table backed by a real vindex IVF-flat ANN segment
+// built in-process (bucket-local ANN search, residual filter supported).
+// * A data-evolution (append) vector table whose global index is produced by
+// the public `new_vindex_index_build_builder(...).execute()` path.
+//
+// Both fixtures live entirely on the in-memory FileIO, so no temp dirs or
+// on-disk schema files are needed: the written data file keeps `schema_id ==
0`,
+// matching the table, so the read path never reloads a schema from disk.
+//
+// The materialized stream carries the user table columns plus a unified
+// `__paimon_search_score` Float32 column; row order is best-first.
+
+use std::collections::HashMap;
+
+use arrow_array::builder::{FixedSizeListBuilder, Float32Builder, ListBuilder};
+use arrow_array::{ArrayRef, Float32Array};
+use bytes::Bytes;
+use futures::TryStreamExt;
+use paimon::io::FileIO;
+use paimon::spec::{
+ ArrayType, DataFileMeta, Datum, FloatType, GlobalIndexMeta, IndexFileMeta,
Predicate,
+ PredicateBuilder, VectorType,
+};
+use paimon::table::{CommitMessage, TableCommit};
+
+use paimon_vindex_core::index::{VectorIndexConfig, VectorIndexTrainer,
VectorIndexWriter};
+use paimon_vindex_core::io::PosWriter;
+
+/// Unified score column materialized by `execute_read` (Float32).
+const SCORE_COLUMN: &str = "__paimon_search_score";
+/// Vector dimension for the primary-key fixtures.
+const PK_DIM: usize = 4;
+/// Primary-key vector column name (shared by both storage fixtures).
+const VECTOR_COLUMN: &str = "embedding";
+/// vindex index type used for both the PK ANN segment and the DE global index.
+const INDEX_TYPE: &str = "ivf-flat";
+
+// --- Primary-key vector fixture ------------------------------------------
+
+/// Table options routing searches into the primary-key vector branch. A single
+/// bucket keeps one data file; `deletion-vectors.enabled` satisfies the
residual
+/// guard (`with_filter`) without any row actually being deleted.
+fn pk_vector_options() -> Vec<(String, String)> {
+ vec![
+ ("bucket".to_string(), "1".to_string()),
+ ("deletion-vectors.enabled".to_string(), "true".to_string()),
+ (
+ "pk-vector.index.columns".to_string(),
+ VECTOR_COLUMN.to_string(),
+ ),
+ (
+ format!("fields.{VECTOR_COLUMN}.pk-vector.index.type"),
+ INDEX_TYPE.to_string(),
+ ),
+ (
+ format!("fields.{VECTOR_COLUMN}.pk-vector.distance.metric"),
+ "l2".to_string(),
+ ),
+ ]
+}
+
+/// Primary-key schema `(id INT PRIMARY KEY, embedding VECTOR<FLOAT>)`.
+fn pk_vector_schema() -> TableSchema {
+ let mut builder = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column(
+ VECTOR_COLUMN,
+ DataType::Vector(
+ VectorType::try_new(true, PK_DIM as u32,
DataType::Float(FloatType::new()))
+ .unwrap(),
+ ),
+ )
+ .primary_key(["id"]);
+ for (k, v) in pk_vector_options() {
+ builder = builder.option(k, v);
+ }
+ TableSchema::new(0, &builder.build().unwrap())
+}
+
+/// Arrow batch matching the PK schema: `id` (== physical position) plus a
+/// `FixedSizeList<Float32>` vector column.
+fn pk_data_batch(vectors: &[[f32; PK_DIM]]) -> RecordBatch {
+ let ids: Vec<i32> = (0..vectors.len() as i32).collect();
+ let element_field = Arc::new(ArrowField::new("element",
ArrowDataType::Float32, true));
+ let mut vector_builder = FixedSizeListBuilder::new(Float32Builder::new(),
PK_DIM as i32)
+ .with_field(element_field.clone());
+ for vector in vectors {
+ for &value in vector {
+ vector_builder.values().append_value(value);
+ }
+ vector_builder.append(true);
+ }
+ let schema = Arc::new(ArrowSchema::new(vec![
+ ArrowField::new("id", ArrowDataType::Int32, false),
+ ArrowField::new(
+ VECTOR_COLUMN,
+ ArrowDataType::FixedSizeList(element_field, PK_DIM as i32),
+ true,
+ ),
+ ]));
+ RecordBatch::try_new(
+ schema,
+ vec![
+ Arc::new(Int32Array::from(ids)) as ArrayRef,
+ Arc::new(vector_builder.finish()) as ArrayRef,
+ ],
+ )
+ .unwrap()
+}
+
+/// Encode one Java `DataOutput#writeUTF` value (u16-BE length + modified
UTF-8),
+/// as `PkVectorSourceMeta` expects.
+fn java_write_utf(s: &str) -> Vec<u8> {
+ let mut body = Vec::new();
+ for c in s.encode_utf16() {
+ if (0x0001..=0x007F).contains(&c) {
+ body.push(c as u8);
+ } else if c > 0x07FF {
+ body.push(0xE0 | (c >> 12) as u8);
+ body.push(0x80 | ((c >> 6) & 0x3F) as u8);
+ body.push(0x80 | (c & 0x3F) as u8);
+ } else {
+ body.push(0xC0 | (c >> 6) as u8);
+ body.push(0x80 | (c & 0x3F) as u8);
+ }
+ }
+ let mut out = (body.len() as u16).to_be_bytes().to_vec();
+ out.extend_from_slice(&body);
+ out
+}
+
+/// Assemble the `_SOURCE_META` frame the way Java `PkVectorSourceMeta` writes
it:
+/// `i32-BE version=1`, `i32-BE data_level`, `i32-BE count`, then per source
file a
+/// `writeUTF` name and an `i64-BE` row count.
+fn source_meta_bytes(data_level: i32, files: &[(&str, i64)]) -> Vec<u8> {
+ let mut out = Vec::new();
+ out.extend_from_slice(&1i32.to_be_bytes());
+ out.extend_from_slice(&data_level.to_be_bytes());
+ out.extend_from_slice(&(files.len() as i32).to_be_bytes());
+ for (name, rows) in files {
+ out.extend_from_slice(&java_write_utf(name));
+ out.extend_from_slice(&rows.to_be_bytes());
+ }
+ out
+}
+
+/// Build a real vindex IVF-flat ANN segment over `vectors` (label == physical
+/// position) and write it into `{table}/index/{file_name}`. `nlist = 1` keeps
the
+/// single inverted list exhaustive, so the search is exact.
+async fn write_ann_segment(
+ file_io: &FileIO,
+ table_location: &str,
+ file_name: &str,
+ vectors: &[[f32; PK_DIM]],
+) -> u64 {
+ let n = vectors.len();
+ let flat: Vec<f32> = vectors.iter().flat_map(|v|
v.iter().copied()).collect();
+ let ids: Vec<i64> = (0..n as i64).collect();
+
+ let native_options = HashMap::from([
+ ("index.type".to_string(), "ivf_flat".to_string()),
+ ("dimension".to_string(), PK_DIM.to_string()),
+ ("nlist".to_string(), "1".to_string()),
+ ("metric".to_string(), "l2".to_string()),
+ ]);
+ let config = VectorIndexConfig::from_options(&native_options).unwrap();
+ let training = VectorIndexTrainer::train(config, &flat, n).unwrap();
+ let mut writer = VectorIndexWriter::new(training);
+ writer.add_vectors(&ids, &flat, n).unwrap();
+ let mut bytes = Vec::new();
+ {
+ let mut output = PosWriter::new(&mut bytes);
+ writer.write(&mut output).unwrap();
+ }
+
+ let index_dir = format!("{}/index", table_location.trim_end_matches('/'));
+ file_io.mkdirs(&index_dir).await.unwrap();
+ let index_path = format!("{index_dir}/{file_name}");
+ let file_size = bytes.len() as u64;
+ file_io
+ .new_output(&index_path)
+ .unwrap()
+ .write(Bytes::from(bytes))
+ .await
+ .unwrap();
+ file_size
+}
+
+/// Build a complete, self-contained primary-key vector table over `vectors` on
+/// the in-memory FileIO: write a real data file, apply the two PK-vector
+/// constraints to its meta (compacted, non-level-0; Java source-meta frame),
+/// build+commit a real vindex ANN segment, and return the opened table.
+fn build_pk_vector_table(path: &str, vectors: &[[f32; PK_DIM]]) -> Table {
+ let file_io = memory_file_io();
+ setup_table_dirs(&file_io, path);
+ let table = Table::new(
+ file_io.clone(),
+ Identifier::new("default", "pkvector"),
+ path.to_string(),
+ pk_vector_schema(),
+ None,
+ );
+
+ crate::runtime().block_on(async {
+ // Write a real data file via the public write path to obtain a genuine
+ // DataFileMeta (name, row count, stats, file size).
+ let write_builder = table.new_write_builder();
+ let mut writer = write_builder.new_write().unwrap();
+ writer
+ .write_arrow_batch(&pk_data_batch(vectors))
+ .await
+ .unwrap();
+ let write_messages = writer.prepare_commit().await.unwrap();
+ assert_eq!(write_messages.len(), 1, "single bucket -> one message");
+ let written = &write_messages[0];
+ assert_eq!(written.new_files.len(), 1, "single data file expected");
+ let base_meta = written.new_files[0].clone();
+ let bucket = written.bucket;
+ let partition = written.partition.clone();
+ let data_file_name = base_meta.file_name.clone();
+ let row_count = base_meta.row_count;
+
+ // Constraint 1: only a compacted, non-level-0 file backs the PK-vector
+ // index. Pin first_row_id = 0 so global row id == physical position.
+ let indexed_meta = DataFileMeta {
+ level: 1,
+ file_source: Some(1),
+ first_row_id: Some(0),
+ ..base_meta
+ };
+
+ // Build and persist the real vindex ANN segment.
+ let index_file_name = "vector-ivf-flat-pk-c.index".to_string();
+ let index_file_size = write_ann_segment(&file_io, path,
&index_file_name, vectors).await;
+
+ // Constraint 2: GlobalIndexMeta.source_meta is the Java
PkVectorSourceMeta
+ // frame naming the backing data file in ordinal order.
+ let vector_field_id = table
+ .schema()
+ .fields()
+ .iter()
+ .find(|f| f.name() == VECTOR_COLUMN)
+ .expect("vector field present")
+ .id();
+ let index_file = IndexFileMeta {
+ index_type: INDEX_TYPE.to_string(),
+ file_name: index_file_name,
+ file_size: i32::try_from(index_file_size).unwrap(),
+ row_count: i32::try_from(row_count).unwrap(),
+ deletion_vectors_ranges: None,
+ global_index_meta: Some(GlobalIndexMeta {
+ row_range_start: 0,
+ row_range_end: row_count - 1,
+ index_field_id: vector_field_id,
+ extra_field_ids: None,
+ source_meta: Some(source_meta_bytes(
+ indexed_meta.level,
+ &[(&data_file_name, row_count)],
+ )),
+ index_meta: None,
+ }),
+ };
+
+ // Commit the indexed data file together with the ANN segment.
+ let mut message = CommitMessage::new(partition, bucket,
vec![indexed_meta]);
+ message.new_index_files = vec![index_file];
+ TableCommit::new(table.clone(), "pkvector-c".to_string())
+ .commit(vec![message])
+ .await
+ .unwrap();
+ });
+
+ table
+}
+
+/// Empty primary-key vector table (options set, no data): the PK branch
resolves
+/// but the plan is empty, so a search returns an empty (EOF) stream.
+fn build_pk_vector_table_empty(path: &str) -> Table {
+ let file_io = memory_file_io();
+ setup_table_dirs(&file_io, path);
+ Table::new(
+ file_io,
+ Identifier::new("default", "pkvector_empty"),
+ path.to_string(),
+ pk_vector_schema(),
+ None,
+ )
+}
+
+/// Fixture: distances 1 < 41 < 67 < 181; top-3 = rows 0, 4, 5.
+fn pk_fixture_smoke() -> ([f32; PK_DIM], Vec<[f32; PK_DIM]>) {
+ let query = [9.0, 0.0, 0.0, 0.0];
+ let vectors = vec![
+ [10.0, 0.0, 0.0, 0.0],
+ [0.0, 10.0, 0.0, 0.0],
+ [0.0, 0.0, 10.0, 0.0],
+ [0.0, 0.0, 0.0, 10.0],
+ [5.0, 5.0, 0.0, 0.0],
+ [1.0, 1.0, 1.0, 1.0],
+ ];
+ (query, vectors)
+}
+
+/// Residual fixture: unfiltered top-3 = [0, 1, 2]; with `id >= 3` the top-3
+/// becomes [4, 5, 3], disjoint from the unfiltered set.
+fn pk_fixture_residual() -> ([f32; PK_DIM], Vec<[f32; PK_DIM]>) {
+ let query = [10.0, 0.0, 0.0, 0.0];
+ let vectors = vec![
+ [10.0, 0.0, 0.0, 0.0], // pos 0 -> 0
+ [9.0, 0.0, 0.0, 0.0], // pos 1 -> 1
+ [8.0, 0.0, 0.0, 0.0], // pos 2 -> 4
+ [5.0, 0.0, 0.0, 0.0], // pos 3 -> 25
+ [7.0, 0.0, 0.0, 0.0], // pos 4 -> 9
+ [6.0, 0.0, 0.0, 0.0], // pos 5 -> 16
+ ];
+ (query, vectors)
+}
+
+// --- Data-evolution (append) vector fixture ------------------------------
+
+/// Options enabling the data-evolution global-index vindex build/search path.
+fn append_vector_options() -> HashMap<String, String> {
+ HashMap::from([
+ ("row-tracking.enabled".to_string(), "true".to_string()),
+ ("data-evolution.enabled".to_string(), "true".to_string()),
+ ("global-index.enabled".to_string(), "true".to_string()),
+ (
+ "global-index.row-count-per-shard".to_string(),
+ "10".to_string(),
+ ),
+ ("ivf-flat.dimension".to_string(), "2".to_string()),
+ // Single inverted list keeps the ANN search exhaustive (exact, stable
+ // ordering), so the C result matches the Rust reference
deterministically.
+ ("ivf-flat.nlist".to_string(), "1".to_string()),
+ ])
+}
+
+/// Arrow batch for the DE table: `id` INT plus an `embedding` `List<Float32>`.
+fn append_vector_batch(ids: Vec<i32>, vectors: Vec<[f32; 2]>) -> RecordBatch {
+ let element_field = Arc::new(ArrowField::new("element",
ArrowDataType::Float32, true));
+ let mut vector_builder =
+
ListBuilder::new(Float32Builder::new()).with_field(element_field.clone());
+ for vector in vectors {
+ for value in vector {
+ vector_builder.values().append_value(value);
+ }
+ vector_builder.append(true);
+ }
+ let schema = Arc::new(ArrowSchema::new(vec![
+ ArrowField::new("id", ArrowDataType::Int32, false),
+ ArrowField::new("embedding", ArrowDataType::List(element_field), true),
+ ]));
+ RecordBatch::try_new(
+ schema,
+ vec![
+ Arc::new(Int32Array::from(ids)) as ArrayRef,
+ Arc::new(vector_builder.finish()) as ArrayRef,
+ ],
+ )
+ .unwrap()
+}
+
+/// Build a data-evolution vector table: write vectors via the public write
path,
+/// then build the global vindex index via `new_vindex_index_build_builder`.
+fn build_append_vector_table(path: &str) -> Table {
+ let file_io = memory_file_io();
+ setup_table_dirs(&file_io, path);
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column(
+ "embedding",
+ DataType::Array(ArrayType::new(DataType::Float(FloatType::new()))),
+ )
+ .options(append_vector_options())
+ .build()
+ .unwrap();
+ let table = Table::new(
+ file_io,
+ Identifier::new("default", "devector"),
+ path.to_string(),
+ TableSchema::new(0, &schema),
+ None,
+ );
+
+ crate::runtime().block_on(async {
+ let write_builder = table.new_write_builder();
+ let mut writer = write_builder.new_write().unwrap();
+ writer
+ .write_arrow_batch(&append_vector_batch(
+ vec![0, 1, 2, 3, 4, 5],
+ vec![
+ [1.0, 0.0],
+ [0.0, 1.0],
+ [0.9, 0.1],
+ [0.1, 0.9],
+ [0.8, 0.2],
+ [0.2, 0.8],
+ ],
+ ))
+ .await
+ .unwrap();
+ let messages = writer.prepare_commit().await.unwrap();
+ write_builder.new_commit().commit(messages).await.unwrap();
+
+ let built = table
+ .new_vindex_index_build_builder(INDEX_TYPE)
+ .with_index_column("embedding")
+ .execute()
+ .await
+ .unwrap();
+ assert!(built > 0, "DE fixture must build at least one index shard");
+ });
+
+ table
+}
+
+// --- Shared harness: core Rust reference + C read bridges ----------------
+
+/// Import one `paimon_arrow_batch` (Arrow C Data Interface) into a
RecordBatch,
+/// mirroring `collect_rows`: take ownership of the FFI structs via
`ptr::read`,
+/// hand the array to `from_ffi`, then neutralize the originals so the caller's
+/// `paimon_arrow_batch_free` release is a no-op. The imported schema's memory
is
+/// released when the local `ffi_schema` drops at the end of this call.
+unsafe fn import_batch(batch: &paimon_arrow_batch) -> RecordBatch {
+ let ffi_array = ptr::read(batch.array as *const FFI_ArrowArray);
+ let ffi_schema = ptr::read(batch.schema as *const FFI_ArrowSchema);
+ let data = arrow_array::ffi::from_ffi(ffi_array, &ffi_schema).unwrap();
+ ptr::write(batch.array as *mut FFI_ArrowArray, FFI_ArrowArray::empty());
+ ptr::write(
+ batch.schema as *mut FFI_ArrowSchema,
+ FFI_ArrowSchema::empty(),
+ );
+ RecordBatch::from(StructArray::from(data))
+}
+
+/// `(id INT32, score FLOAT32)` pairs from a materialized search batch. Panics
if
+/// the unified score column is missing, pinning the read contract.
+fn batch_id_score_pairs(batch: &RecordBatch) -> Vec<(i32, f32)> {
+ let ids = batch
+ .column_by_name("id")
+ .expect("id column present")
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .expect("id is Int32");
+ let score_idx = batch
+ .schema()
+ .index_of(SCORE_COLUMN)
+ .expect("unified score column present");
+ let scores = batch
+ .column(score_idx)
+ .as_any()
+ .downcast_ref::<Float32Array>()
+ .expect("score is Float32");
+ (0..batch.num_rows())
+ .map(|i| (ids.value(i), scores.value(i)))
+ .collect()
+}
+
+/// Core Rust reference: run `execute_read` and return (row count, score-column
+/// present) — the shape the C path is compared against. Runs on the global
+/// runtime.
+fn rust_execute_read_rows(
+ table: &Table,
+ column: &str,
+ query: Vec<f32>,
+ limit: usize,
+) -> (usize, bool) {
+ crate::runtime().block_on(async {
+ let mut builder = table.new_vector_search_builder();
+ builder
+ .with_vector_column(column)
+ .with_query_vector(query)
+ .with_limit(limit);
+ let mut stream = builder.execute_read().await.unwrap();
+ let (mut rows, mut has_score) = (0usize, false);
+ while let Some(b) = stream.try_next().await.unwrap() {
+ rows += b.num_rows();
+ has_score |= b.schema().index_of(SCORE_COLUMN).is_ok();
+ }
+ (rows, has_score)
+ })
+}
+
+/// Core Rust reference: sorted `(id, score)` pairs materialized by
`execute_read`.
+fn rust_execute_read_pairs(
+ table: &Table,
+ column: &str,
+ query: Vec<f32>,
+ limit: usize,
+ filter: Option<Predicate>,
+) -> Vec<(i32, f32)> {
+ crate::runtime().block_on(async {
+ let mut builder = table.new_vector_search_builder();
+ builder
+ .with_vector_column(column)
+ .with_query_vector(query)
+ .with_limit(limit);
+ if let Some(f) = filter {
+ builder.with_filter(f);
+ }
+ let mut stream = builder.execute_read().await.unwrap();
+ let mut pairs = Vec::new();
+ while let Some(b) = stream.try_next().await.unwrap() {
+ pairs.extend(batch_id_score_pairs(&b));
+ }
+ pairs.sort_by_key(|p| p.0);
+ pairs
+ })
+}
+
+/// Build a `>=` predicate on an integer column via the public C predicate API.
+unsafe fn build_predicate_ge(
+ table: *const paimon_table,
+ column: &str,
+ int_val: i32,
+) -> *mut paimon_predicate {
+ let col = CString::new(column).unwrap();
+ let datum = paimon_datum {
+ tag: 3,
+ int_val: int_val as i64,
+ double_val: 0.0,
+ str_data: ptr::null(),
+ str_len: 0,
+ int_val2: 0,
+ uint_val: 0,
+ uint_val2: 0,
+ };
+ let result = paimon_predicate_greater_or_equal(table, col.as_ptr(), datum);
+ assert!(result.error.is_null());
+ result.predicate
+}
+
+/// Construct + configure a C vector-search builder (column, query, limit, and
an
+/// optional filter that `with_filter` consumes on success). The caller drives
the
+/// terminal and frees the builder.
+unsafe fn c_vector_builder(
+ handle: *const paimon_table,
+ column: &str,
+ query: &[f32],
+ limit: usize,
+ filter: *mut paimon_predicate,
+) -> *mut paimon_vector_search_builder {
+ let builder_result = paimon_table_new_vector_search_builder(handle);
+ assert!(builder_result.error.is_null());
+ let builder = builder_result.builder;
+
+ let col = CString::new(column).unwrap();
+ assert!(paimon_vector_search_builder_with_vector_column(builder,
col.as_ptr()).is_null());
+ assert!(
+ paimon_vector_search_builder_with_query_vector(builder,
query.as_ptr(), query.len())
+ .is_null()
+ );
+ assert!(paimon_vector_search_builder_with_limit(builder, limit).is_null());
+ if !filter.is_null() {
+ assert!(paimon_vector_search_builder_with_filter(builder,
filter).is_null());
+ }
+ builder
+}
+
+/// C path: run `execute_read` on a configured builder, drain the reader, and
+/// return (row count, score-column present). Frees the builder and reader.
+unsafe fn c_execute_read_rows(builder: *mut paimon_vector_search_builder) ->
(usize, bool) {
+ let result = paimon_vector_search_builder_execute_read(builder);
+ paimon_vector_search_builder_free(builder);
+ assert!(result.error.is_null(), "execute_read should not error");
+ assert!(!result.reader.is_null());
+
+ let mut rows = 0usize;
+ let mut has_score = false;
+ loop {
+ let next = paimon_record_batch_reader_next(result.reader);
+ assert!(next.error.is_null());
+ if next.batch.array.is_null() {
+ break; // EOF
+ }
+ let batch = import_batch(&next.batch);
+ rows += batch.num_rows();
+ has_score |= batch.schema().index_of(SCORE_COLUMN).is_ok();
+ paimon_arrow_batch_free(next.batch);
+ }
+ paimon_record_batch_reader_free(result.reader);
+ (rows, has_score)
+}
+
+/// C path: sorted `(id, score)` pairs materialized by a configured builder's
+/// `execute_read`. Frees the builder and reader.
+unsafe fn c_execute_read_pairs(builder: *mut paimon_vector_search_builder) ->
Vec<(i32, f32)> {
+ let result = paimon_vector_search_builder_execute_read(builder);
+ paimon_vector_search_builder_free(builder);
+ assert!(result.error.is_null(), "execute_read should not error");
+ assert!(!result.reader.is_null());
+
+ let mut pairs = Vec::new();
+ loop {
+ let next = paimon_record_batch_reader_next(result.reader);
+ assert!(next.error.is_null());
+ if next.batch.array.is_null() {
+ break; // EOF
+ }
+ let batch = import_batch(&next.batch);
+ pairs.extend(batch_id_score_pairs(&batch));
+ paimon_arrow_batch_free(next.batch);
+ }
+ paimon_record_batch_reader_free(result.reader);
+ pairs.sort_by_key(|p| p.0);
+ pairs
+}
+
+/// Read a `paimon_error`'s UTF-8 message.
+unsafe fn error_message(err: *mut paimon_error) -> String {
+ let bytes = &(*err).message;
+ let slice = std::slice::from_raw_parts(bytes.data, bytes.len);
+ String::from_utf8_lossy(slice).to_string()
+}
+
+// --- Tests ----------------------------------------------------------------
+
+#[test]
+fn vector_search_pk_table_read_matches_rust() {
+ let path = "memory:/vsearch_pk_read";
+ let (query, vectors) = pk_fixture_smoke();
+ let table = build_pk_vector_table(path, &vectors);
+
+ // Independent core reference: row count + score column presence and the
+ // materialized (id, score) pairs.
+ let (rust_rows, rust_has_score) =
+ rust_execute_read_rows(&table, VECTOR_COLUMN, query.to_vec(), 3);
+ let rust_pairs = rust_execute_read_pairs(&table, VECTOR_COLUMN,
query.to_vec(), 3, None);
+ assert_eq!(rust_rows, 3, "PK fixture top-3 must materialize 3 rows");
+ assert!(
+ rust_has_score,
+ "reference must carry the unified score column"
+ );
+
+ let handle = unsafe { wrap_table(table) };
+ unsafe {
+ let builder = c_vector_builder(handle, VECTOR_COLUMN, &query, 3,
ptr::null_mut());
+ let (c_rows, c_has_score) = c_execute_read_rows(builder);
+ assert_eq!(
+ c_rows, rust_rows,
+ "C row count must match the Rust reference"
+ );
+ assert_eq!(
+ c_has_score, rust_has_score,
+ "score-column presence must match"
+ );
+
+ let builder = c_vector_builder(handle, VECTOR_COLUMN, &query, 3,
ptr::null_mut());
+ let c_pairs = c_execute_read_pairs(builder);
+ assert_eq!(c_pairs.len(), rust_pairs.len());
+ for ((c_id, c_score), (r_id, r_score)) in
c_pairs.iter().zip(&rust_pairs) {
+ assert_eq!(c_id, r_id, "C row ids must match the Rust reference");
+ assert!(
+ (c_score - r_score).abs() < 1e-6,
+ "score diverges: {c_score} vs {r_score}"
+ );
+ }
+ unwrap_table(handle);
+ }
+}
+
+#[test]
+fn vector_search_append_table_read_matches_rust() {
+ let path = "memory:/vsearch_append_read";
+ let table = build_append_vector_table(path);
+ let query = vec![1.0f32, 0.0];
+
+ let (rust_rows, rust_has_score) = rust_execute_read_rows(&table,
"embedding", query.clone(), 3);
+ let rust_pairs = rust_execute_read_pairs(&table, "embedding",
query.clone(), 3, None);
+ assert!(rust_rows > 0, "DE fixture must materialize hits");
+ assert!(
+ rust_has_score,
+ "reference must carry the unified score column"
+ );
+
+ let handle = unsafe { wrap_table(table) };
+ unsafe {
+ let builder = c_vector_builder(handle, "embedding", &query, 3,
ptr::null_mut());
+ let (c_rows, c_has_score) = c_execute_read_rows(builder);
+ assert_eq!(
+ c_rows, rust_rows,
+ "C row count must match the Rust reference"
+ );
+ assert_eq!(
+ c_has_score, rust_has_score,
+ "score-column presence must match"
+ );
+
+ // The data-evolution global-index path does not promise a stable order
+ // across two separate reads, so compare the (id, score) hits as an
+ // id-keyed set rather than over-asserting an order neither read
promises.
+ let builder = c_vector_builder(handle, "embedding", &query, 3,
ptr::null_mut());
+ let c_pairs = c_execute_read_pairs(builder);
+ assert_eq!(c_pairs.len(), rust_pairs.len());
+ for ((c_id, c_score), (r_id, r_score)) in
c_pairs.iter().zip(&rust_pairs) {
+ assert_eq!(c_id, r_id, "C row ids must match the Rust reference
set");
+ assert!(
+ (c_score - r_score).abs() < 1e-6,
+ "score diverges: {c_score} vs {r_score}"
+ );
+ }
+ unwrap_table(handle);
+ }
+}
+
+#[test]
+fn vector_search_pk_filter_excludes_neighbor() {
+ let path = "memory:/vsearch_pk_filter_read";
+ let (query, vectors) = pk_fixture_residual();
+ let table = build_pk_vector_table(path, &vectors);
+
+ // Guard: the nearest neighbor (id 0) is present in the unfiltered top-3,
so a
+ // working residual filter of `id >= 3` must exclude it.
+ let unfiltered = rust_execute_read_pairs(&table, VECTOR_COLUMN,
query.to_vec(), 3, None);
+ let unfiltered_ids: Vec<i32> = unfiltered.iter().map(|(id, _)|
*id).collect();
+ assert!(
+ unfiltered_ids.contains(&0),
+ "fixture guard: id 0 must be an unfiltered neighbor"
+ );
+
+ // Independent filtered reference via the core Rust path.
+ let rust_filter = PredicateBuilder::new(table.schema().fields())
+ .greater_or_equal("id", Datum::Int(3))
+ .unwrap();
+ let rust_pairs =
+ rust_execute_read_pairs(&table, VECTOR_COLUMN, query.to_vec(), 3,
Some(rust_filter));
+
+ let handle = unsafe { wrap_table(table) };
+ unsafe {
+ let predicate = build_predicate_ge(handle, "id", 3);
+ let builder = c_vector_builder(handle, VECTOR_COLUMN, &query, 3,
predicate);
+ let c_pairs = c_execute_read_pairs(builder);
+
+ let c_ids: Vec<i32> = c_pairs.iter().map(|(id, _)| *id).collect();
+ for excluded in [0i32, 1, 2] {
+ assert!(
+ !c_ids.contains(&excluded),
+ "filtered read must exclude neighbor {excluded}"
+ );
+ }
+ assert_eq!(
+ c_pairs, rust_pairs,
+ "filtered pairs must match the reference"
+ );
+ unwrap_table(handle);
+ }
+}
+
+#[test]
+fn vector_search_append_filter_returns_invalid_input() {
+ let path = "memory:/vsearch_append_filter_err";
+ let table = build_append_vector_table(path);
+ let handle = unsafe { wrap_table(table) };
+ unsafe {
+ let predicate = build_predicate_ge(handle, "id", 1);
+ let builder = c_vector_builder(handle, "embedding", &[1.0f32, 0.0], 3,
predicate);
+ let result = paimon_vector_search_builder_execute_read(builder);
+ paimon_vector_search_builder_free(builder);
+
+ assert!(
+ result.reader.is_null(),
+ "errored read must not yield a reader"
+ );
+ assert!(!result.error.is_null(), "DE filter must fail loud");
+ assert_eq!(
+ (*result.error).code,
+ PaimonErrorCode::InvalidInput as i32,
+ "DE filter error must map to InvalidInput"
+ );
+ let message = error_message(result.error);
+ assert!(
+ message.contains("primary-key vector path"),
+ "unexpected error message: {message}"
+ );
+ paimon_error_free(result.error);
+ unwrap_table(handle);
+ }
+}
+
+#[test]
+fn vector_search_unknown_column_returns_invalid_input() {
+ // A typo'd vector column must surface as an input error through the C API,
+ // not a silent empty (EOF) reader.
+ let path = "memory:/vsearch_unknown_col_err";
+ let table = build_append_vector_table(path);
+ let handle = unsafe { wrap_table(table) };
+ unsafe {
+ let builder =
+ c_vector_builder(handle, "does_not_exist", &[1.0f32, 0.0], 3,
ptr::null_mut());
+ let result = paimon_vector_search_builder_execute_read(builder);
+ paimon_vector_search_builder_free(builder);
+
+ assert!(
+ result.reader.is_null(),
+ "errored read must not yield a reader"
+ );
+ assert!(!result.error.is_null(), "unknown column must fail loud");
+ assert_eq!(
+ (*result.error).code,
+ PaimonErrorCode::InvalidInput as i32,
+ "unknown column error must map to InvalidInput"
+ );
+ let message = error_message(result.error);
+ assert!(
+ message.contains("does not exist"),
+ "unexpected error message: {message}"
+ );
+ paimon_error_free(result.error);
+ unwrap_table(handle);
+ }
+}
+
+#[test]
+fn vector_search_scalar_column_returns_invalid_input() {
+ // A scalar (non-vector) column must surface as an input error, not an
empty
+ // reader.
+ let path = "memory:/vsearch_scalar_col_err";
+ let table = build_append_vector_table(path);
+ let handle = unsafe { wrap_table(table) };
+ unsafe {
+ // "id" is a scalar Int column on the append vector table.
+ let builder = c_vector_builder(handle, "id", &[1.0f32, 0.0], 3,
ptr::null_mut());
+ let result = paimon_vector_search_builder_execute_read(builder);
+ paimon_vector_search_builder_free(builder);
+
+ assert!(
+ result.reader.is_null(),
+ "errored read must not yield a reader"
+ );
+ assert!(!result.error.is_null(), "scalar column must fail loud");
+ assert_eq!(
+ (*result.error).code,
+ PaimonErrorCode::InvalidInput as i32,
+ "scalar column error must map to InvalidInput"
+ );
+ let message = error_message(result.error);
+ assert!(
+ message.contains("must be a FLOAT vector column"),
+ "unexpected error message: {message}"
+ );
+ paimon_error_free(result.error);
+ unwrap_table(handle);
+ }
+}
+
+#[test]
+fn vector_search_rejects_invalid_query_vector() {
+ let path = "memory:/vsearch_setter_validation";
+ let table = build_pk_vector_table_empty(path);
+ let handle = unsafe { wrap_table(table) };
+ unsafe {
+ let builder_result = paimon_table_new_vector_search_builder(handle);
+ assert!(builder_result.error.is_null());
+ let builder = builder_result.builder;
+
+ // Null data with a non-zero length is rejected at the setter.
+ let err_null = paimon_vector_search_builder_with_query_vector(builder,
ptr::null(), 5);
+ assert!(!err_null.is_null());
+ assert_eq!((*err_null).code, PaimonErrorCode::InvalidInput as i32);
+ paimon_error_free(err_null);
+
+ // A zero-length query is rejected even with a valid pointer.
+ let data = [1.0f32];
+ let err_empty =
paimon_vector_search_builder_with_query_vector(builder, data.as_ptr(), 0);
+ assert!(!err_empty.is_null());
+ assert_eq!((*err_empty).code, PaimonErrorCode::InvalidInput as i32);
+ paimon_error_free(err_empty);
+
+ paimon_vector_search_builder_free(builder);
+ unwrap_table(handle);
+ }
+}
+
+#[test]
+fn vector_search_empty_result_is_eof_stream() {
+ let path = "memory:/vsearch_empty_read";
+ let table = build_pk_vector_table_empty(path);
+ let handle = unsafe { wrap_table(table) };
+ unsafe {
+ let builder = c_vector_builder(
+ handle,
+ VECTOR_COLUMN,
+ &[1.0f32, 2.0, 3.0, 4.0],
+ 5,
+ ptr::null_mut(),
+ );
+ let result = paimon_vector_search_builder_execute_read(builder);
+ paimon_vector_search_builder_free(builder);
+
+ // An empty table is a normal EOF stream, not an error: a reader is
+ // returned and the first `_next` yields a null batch with no error.
+ assert!(result.error.is_null());
+ assert!(!result.reader.is_null());
+ let next = paimon_record_batch_reader_next(result.reader);
+ assert!(next.error.is_null());
+ assert!(next.batch.array.is_null());
+ assert!(next.batch.schema.is_null());
+ paimon_record_batch_reader_free(result.reader);
+ unwrap_table(handle);
+ }
+}
diff --git a/bindings/c/src/types.rs b/bindings/c/src/types.rs
index 8d3a4666..067353fe 100644
--- a/bindings/c/src/types.rs
+++ b/bindings/c/src/types.rs
@@ -127,6 +127,24 @@ pub struct paimon_predicate {
pub inner: *mut c_void,
}
+/// Opaque wrapper around a vector-search builder.
+#[repr(C)]
+pub struct paimon_vector_search_builder {
+ pub inner: *mut c_void,
+}
+
+/// Internal state for a vector-search builder: the table plus the query
+/// parameters accumulated by the setters before the search is run.
+pub(crate) struct VectorSearchState {
+ // Read by the search terminal that runs the accumulated query.
+ pub table: Table,
+ pub vector_column: Option<String>,
+ pub query_vector: Option<Vec<f32>>,
+ pub limit: Option<usize>,
+ pub options: std::collections::HashMap<String, String>,
+ pub filter: Option<Predicate>,
+}
+
/// A typed literal value for predicate comparison, passed across FFI.
///
/// # Design
diff --git a/bindings/c/src/vector_search.rs b/bindings/c/src/vector_search.rs
new file mode 100644
index 00000000..3da28248
--- /dev/null
+++ b/bindings/c/src/vector_search.rs
@@ -0,0 +1,324 @@
+// 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.
+
+//! C FFI bindings for vector search.
+//!
+//! Wraps the Rust vector-search builder over the C ABI: a
+//! `paimon_vector_search_builder` is created from a table, then configured
with
+//! the query vector, target column, result limit, options, and an optional
+//! scalar filter. The builder is storage-agnostic — it targets both
+//! primary-key and append / data-evolution tables.
+//!
+//! This module provides the builder constructor, its setters, the terminal
+//! that runs the search and returns a streaming Arrow reader, and the free
+//! function.
+
+use std::collections::HashMap;
+use std::ffi::{c_char, c_void};
+
+use paimon::spec::Predicate;
+use paimon::table::Table;
+
+use crate::error::{check_non_null, paimon_error, validate_cstr,
PaimonErrorCode};
+use crate::result::{paimon_result_record_batch_reader,
paimon_result_vector_search_builder};
+use crate::runtime;
+use crate::types::*;
+
+/// Create a new vector-search builder from a Table.
+///
+/// # Safety
+/// `table` must be a valid pointer from `paimon_catalog_get_table`, or null
(returns error).
+#[no_mangle]
+pub unsafe extern "C" fn paimon_table_new_vector_search_builder(
+ table: *const paimon_table,
+) -> paimon_result_vector_search_builder {
+ if let Err(e) = check_non_null(table, "table") {
+ return paimon_result_vector_search_builder {
+ builder: std::ptr::null_mut(),
+ error: e,
+ };
+ }
+ let table_ref = &*((*table).inner as *const Table);
+ let state = VectorSearchState {
+ table: table_ref.clone(),
+ vector_column: None,
+ query_vector: None,
+ limit: None,
+ options: HashMap::new(),
+ filter: None,
+ };
+ let inner = Box::into_raw(Box::new(state)) as *mut c_void;
+ paimon_result_vector_search_builder {
+ builder: Box::into_raw(Box::new(paimon_vector_search_builder { inner
})),
+ error: std::ptr::null_mut(),
+ }
+}
+
+/// Set the target vector column for a vector-search builder.
+///
+/// # Safety
+/// `b` must be a valid pointer from `paimon_table_new_vector_search_builder`,
or
+/// null (returns error). `column` must be a valid C string.
+#[no_mangle]
+pub unsafe extern "C" fn paimon_vector_search_builder_with_vector_column(
+ b: *mut paimon_vector_search_builder,
+ column: *const c_char,
+) -> *mut paimon_error {
+ if let Err(e) = check_non_null(b, "b") {
+ return e;
+ }
+ let col = match validate_cstr(column, "vector column") {
+ Ok(s) => s,
+ Err(e) => return e,
+ };
+ let state = &mut *((*b).inner as *mut VectorSearchState);
+ state.vector_column = Some(col);
+ std::ptr::null_mut()
+}
+
+/// Set the query vector for a vector-search builder.
+///
+/// The `len` floats at `data` are copied into the builder; the caller retains
+/// ownership of `data`. An empty vector (`len == 0`) is rejected.
+///
+/// # Safety
+/// `b` must be a valid pointer from `paimon_table_new_vector_search_builder`,
or
+/// null (returns error). `data` must point to `len` `f32` values when `len >
0`.
+#[no_mangle]
+pub unsafe extern "C" fn paimon_vector_search_builder_with_query_vector(
+ b: *mut paimon_vector_search_builder,
+ data: *const f32,
+ len: usize,
+) -> *mut paimon_error {
+ if let Err(e) = check_non_null(b, "b") {
+ return e;
+ }
+ if len == 0 {
+ return paimon_error::new(
+ PaimonErrorCode::InvalidInput,
+ "query vector must not be empty".to_string(),
+ );
+ }
+ if data.is_null() {
+ return paimon_error::new(
+ PaimonErrorCode::InvalidInput,
+ "null query vector pointer with non-zero length".to_string(),
+ );
+ }
+ let state = &mut *((*b).inner as *mut VectorSearchState);
+ state.query_vector = Some(std::slice::from_raw_parts(data, len).to_vec());
+ std::ptr::null_mut()
+}
+
+/// Set the maximum number of results for a vector-search builder.
+///
+/// # Safety
+/// `b` must be a valid pointer from `paimon_table_new_vector_search_builder`,
or
+/// null (returns error).
+#[no_mangle]
+pub unsafe extern "C" fn paimon_vector_search_builder_with_limit(
+ b: *mut paimon_vector_search_builder,
+ limit: usize,
+) -> *mut paimon_error {
+ if let Err(e) = check_non_null(b, "b") {
+ return e;
+ }
+ let state = &mut *((*b).inner as *mut VectorSearchState);
+ state.limit = Some(limit);
+ std::ptr::null_mut()
+}
+
+/// Set scan/search options for a vector-search builder.
+///
+/// # Safety
+/// `b` must be a valid pointer from `paimon_table_new_vector_search_builder`,
or
+/// null (returns error). `options` must be a valid pointer to `len`
+/// `paimon_option` values, or null when `len` is 0.
+#[no_mangle]
+pub unsafe extern "C" fn paimon_vector_search_builder_with_options(
+ b: *mut paimon_vector_search_builder,
+ options: *const paimon_option,
+ len: usize,
+) -> *mut paimon_error {
+ if let Err(e) = check_non_null(b, "b") {
+ return e;
+ }
+ if options.is_null() && len > 0 {
+ return paimon_error::new(
+ PaimonErrorCode::InvalidInput,
+ "null options pointer with non-zero length".to_string(),
+ );
+ }
+ let mut map = HashMap::with_capacity(len);
+ if len > 0 {
+ let slice = std::slice::from_raw_parts(options, len);
+ for opt in slice {
+ let key = match validate_cstr(opt.key, "option key") {
+ Ok(s) => s,
+ Err(e) => return e,
+ };
+ let value = match validate_cstr(opt.value, "option value") {
+ Ok(s) => s,
+ Err(e) => return e,
+ };
+ map.insert(key, value);
+ }
+ }
+ let state = &mut *((*b).inner as *mut VectorSearchState);
+ state.options = map;
+ std::ptr::null_mut()
+}
+
+/// Set an optional scalar residual filter for a vector-search builder.
+///
+/// The predicate is consumed (ownership transferred to the builder). Pass null
+/// to clear any previously set filter.
+///
+/// # Safety
+/// `b` must be a valid pointer from `paimon_table_new_vector_search_builder`,
or
+/// null (returns error). `predicate` must be a valid pointer from a
+/// `paimon_predicate_*` function, or null.
+#[no_mangle]
+pub unsafe extern "C" fn paimon_vector_search_builder_with_filter(
+ b: *mut paimon_vector_search_builder,
+ predicate: *mut paimon_predicate,
+) -> *mut paimon_error {
+ if let Err(e) = check_non_null(b, "b") {
+ return e;
+ }
+
+ let state = &mut *((*b).inner as *mut VectorSearchState);
+
+ if predicate.is_null() {
+ state.filter = None;
+ return std::ptr::null_mut();
+ }
+
+ let pred_wrapper = Box::from_raw(predicate);
+ let pred = Box::from_raw(pred_wrapper.inner as *mut Predicate);
+ state.filter = Some(*pred);
+ std::ptr::null_mut()
+}
+
+/// Free a paimon_vector_search_builder.
+///
+/// # Safety
+/// Only call with a builder returned from
`paimon_table_new_vector_search_builder`.
+#[no_mangle]
+pub unsafe extern "C" fn paimon_vector_search_builder_free(b: *mut
paimon_vector_search_builder) {
+ if !b.is_null() {
+ let wrapper = Box::from_raw(b);
+ if !wrapper.inner.is_null() {
+ drop(Box::from_raw(wrapper.inner as *mut VectorSearchState));
+ }
+ }
+}
+
+/// Execute the vector search and return a streaming Arrow reader over the
+/// materialized rows (projected user columns plus `__paimon_search_score`).
+/// Works for both primary-key and data-evolution tables. Consume via
+/// `paimon_record_batch_reader_next` and free with
`paimon_record_batch_reader_free`.
+///
+/// # Safety
+/// `b` must be a valid pointer from `paimon_table_new_vector_search_builder`,
or
+/// null (returns an error result).
+#[no_mangle]
+pub unsafe extern "C" fn paimon_vector_search_builder_execute_read(
+ b: *mut paimon_vector_search_builder,
+) -> paimon_result_record_batch_reader {
+ if let Err(e) = check_non_null(b, "b") {
+ return paimon_result_record_batch_reader {
+ reader: std::ptr::null_mut(),
+ error: e,
+ };
+ }
+ let state = &*((*b).inner as *const VectorSearchState);
+
+ let mut builder = state.table.new_vector_search_builder();
+ if let Some(col) = &state.vector_column {
+ builder.with_vector_column(col);
+ }
+ if let Some(v) = &state.query_vector {
+ builder.with_query_vector(v.clone());
+ }
+ if let Some(limit) = state.limit {
+ builder.with_limit(limit);
+ }
+ if !state.options.is_empty() {
+ builder.with_options(state.options.clone());
+ }
+ if let Some(f) = &state.filter {
+ builder.with_filter(f.clone());
+ }
+
+ match runtime().block_on(builder.execute_read()) {
+ Ok(stream) => {
+ let reader = Box::new(stream);
+ let wrapper = Box::new(paimon_record_batch_reader {
+ inner: Box::into_raw(reader) as *mut c_void,
+ });
+ paimon_result_record_batch_reader {
+ reader: Box::into_raw(wrapper),
+ error: std::ptr::null_mut(),
+ }
+ }
+ Err(e) => paimon_result_record_batch_reader {
+ reader: std::ptr::null_mut(),
+ error: paimon_error::from_paimon(e),
+ },
+ }
+}
+
+// --- C ABI signature guards -------------------------------------------------
+//
+// These symbols are called across the FFI boundary with fixed argument counts:
+// bindings prepare a libffi call interface (CIF) per symbol, and external
+// consumers link against the generated headers (e.g. Doris integrations).
+// Adding or reordering a parameter on one of these existing symbols silently
+// breaks every such caller — the extra argument is read from an undefined
+// register/stack slot at the ABI boundary.
+//
+// These compile-time assertions pin the existing signatures. To add behavior,
+// introduce a new symbol instead of changing one of these; touching a
signature
+// here will fail to compile.
+const _: unsafe extern "C" fn(*const paimon_table) ->
paimon_result_vector_search_builder =
+ paimon_table_new_vector_search_builder;
+const _: unsafe extern "C" fn(
+ *mut paimon_vector_search_builder,
+ *const c_char,
+) -> *mut paimon_error = paimon_vector_search_builder_with_vector_column;
+const _: unsafe extern "C" fn(
+ *mut paimon_vector_search_builder,
+ *const f32,
+ usize,
+) -> *mut paimon_error = paimon_vector_search_builder_with_query_vector;
+const _: unsafe extern "C" fn(*mut paimon_vector_search_builder, usize) ->
*mut paimon_error =
+ paimon_vector_search_builder_with_limit;
+const _: unsafe extern "C" fn(
+ *mut paimon_vector_search_builder,
+ *const paimon_option,
+ usize,
+) -> *mut paimon_error = paimon_vector_search_builder_with_options;
+const _: unsafe extern "C" fn(
+ *mut paimon_vector_search_builder,
+ *mut paimon_predicate,
+) -> *mut paimon_error = paimon_vector_search_builder_with_filter;
+const _: unsafe extern "C" fn(*mut paimon_vector_search_builder) =
+ paimon_vector_search_builder_free;
+const _: unsafe extern "C" fn(
+ *mut paimon_vector_search_builder,
+) -> paimon_result_record_batch_reader =
paimon_vector_search_builder_execute_read;
diff --git a/crates/paimon/src/table/pk_vector_indexed_split_read.rs
b/crates/paimon/src/table/pk_vector_indexed_split_read.rs
index 668957a6..2dcb6e7f 100644
--- a/crates/paimon/src/table/pk_vector_indexed_split_read.rs
+++ b/crates/paimon/src/table/pk_vector_indexed_split_read.rs
@@ -367,9 +367,7 @@ mod e2e_tests {
use crate::spec::stats::BinaryTableStats;
use crate::spec::{DataField, DataFileMeta, DataType, IntType};
use crate::table::data_file_reader::DataFileReader;
- use crate::table::pk_vector_position_read::{
- PKEY_VECTOR_POSITION_COLUMN, PKEY_VECTOR_SCORE_COLUMN,
- };
+ use crate::table::pk_vector_position_read::{PKEY_VECTOR_POSITION_COLUMN,
SEARCH_SCORE_COLUMN};
use crate::table::schema_manager::SchemaManager;
use crate::table::source::{DataSplit, DataSplitBuilder, DeletionFile};
use arrow_array::{Array, Float32Array, Int32Array, Int64Array,
RecordBatch};
@@ -632,7 +630,7 @@ mod e2e_tests {
column_by_name(batch, "_ROW_ID").is_none(),
"_ROW_ID must not leak"
);
- assert!(column_by_name(batch, PKEY_VECTOR_SCORE_COLUMN).is_none());
+ assert!(column_by_name(batch, SEARCH_SCORE_COLUMN).is_none());
}
}
@@ -659,7 +657,7 @@ mod e2e_tests {
vec![0, 2, 3]
);
assert_eq!(
- collect_f32(&batches, PKEY_VECTOR_SCORE_COLUMN),
+ collect_f32(&batches, SEARCH_SCORE_COLUMN),
vec![0.9, 0.5, 0.1]
);
}
@@ -689,7 +687,7 @@ mod e2e_tests {
vec![0, 2, 3]
);
assert_eq!(
- collect_f32(&batches, PKEY_VECTOR_SCORE_COLUMN),
+ collect_f32(&batches, SEARCH_SCORE_COLUMN),
vec![0.4, 0.2, 0.1]
);
}
@@ -729,7 +727,7 @@ mod e2e_tests {
vec![1, 2, 4]
);
assert_eq!(
- collect_f32(&batches, PKEY_VECTOR_SCORE_COLUMN),
+ collect_f32(&batches, SEARCH_SCORE_COLUMN),
vec![0.9, 0.5, 0.1]
);
}
diff --git a/crates/paimon/src/table/pk_vector_orchestrator.rs
b/crates/paimon/src/table/pk_vector_orchestrator.rs
index e8da709e..0479781f 100644
--- a/crates/paimon/src/table/pk_vector_orchestrator.rs
+++ b/crates/paimon/src/table/pk_vector_orchestrator.rs
@@ -660,9 +660,7 @@ mod e2e_tests {
DataField, DataFileMeta, DataType, IntType, PkVectorSourceFile,
PkVectorSourceMeta,
};
use crate::table::pk_vector_indexed_split_read::PkVectorIndexedSplitRead;
- use crate::table::pk_vector_position_read::{
- PKEY_VECTOR_POSITION_COLUMN, PKEY_VECTOR_SCORE_COLUMN,
- };
+ use crate::table::pk_vector_position_read::{PKEY_VECTOR_POSITION_COLUMN,
SEARCH_SCORE_COLUMN};
use crate::table::schema_manager::SchemaManager;
use crate::table::source::DeletionFile;
use crate::vindex::pkvector::reader::test_support::ArrayReader;
@@ -1099,7 +1097,7 @@ mod e2e_tests {
vec![1, 0, 1]
);
assert_eq!(
- collect_f32(&batches, PKEY_VECTOR_SCORE_COLUMN),
+ collect_f32(&batches, SEARCH_SCORE_COLUMN),
vec![l2_score(0.25), l2_score(1.0), l2_score(4.0)]
);
for batch in &batches {
@@ -1191,7 +1189,7 @@ mod e2e_tests {
vec![0, 0, 1]
);
assert_eq!(
- collect_f32(&batches, PKEY_VECTOR_SCORE_COLUMN),
+ collect_f32(&batches, SEARCH_SCORE_COLUMN),
vec![l2_score(1.0), l2_score(4.0), l2_score(9.0)]
);
}
@@ -1257,7 +1255,7 @@ mod e2e_tests {
vec![0, 2, 3]
);
assert_eq!(
- collect_f32(&batches, PKEY_VECTOR_SCORE_COLUMN),
+ collect_f32(&batches, SEARCH_SCORE_COLUMN),
vec![l2_score(1.0), l2_score(9.0), l2_score(0.0)]
);
}
@@ -1320,7 +1318,7 @@ mod e2e_tests {
);
// Scores aligned to ascending position: d=9,1,4.
assert_eq!(
- collect_f32(&batches, PKEY_VECTOR_SCORE_COLUMN),
+ collect_f32(&batches, SEARCH_SCORE_COLUMN),
vec![l2_score(9.0), l2_score(1.0), l2_score(4.0)]
);
}
diff --git a/crates/paimon/src/table/pk_vector_position_read.rs
b/crates/paimon/src/table/pk_vector_position_read.rs
index 0c838979..5cb1f2b1 100644
--- a/crates/paimon/src/table/pk_vector_position_read.rs
+++ b/crates/paimon/src/table/pk_vector_position_read.rs
@@ -19,7 +19,7 @@
//! `PrimaryKeyVectorPositionReader`).
//!
//! Materializes the selected physical rows of one data file and appends
-//! `_PKEY_VECTOR_POSITION` (+ optional `_PKEY_VECTOR_SCORE`) metadata columns.
+//! `_PKEY_VECTOR_POSITION` (+ optional `__paimon_search_score`) metadata
columns.
//! This is the lowest layer of the PK-vector read kernel; the sibling
//! `pk_vector_indexed_split_read` and `pk_vector_orchestrator` modules build
the
//! indexed-split contract and cross-bucket merge on top of it.
@@ -38,7 +38,9 @@ use crate::table::source::DataSplit;
use crate::table::ArrowRecordBatchStream;
pub(crate) const PKEY_VECTOR_POSITION_COLUMN: &str = "_PKEY_VECTOR_POSITION";
-pub(crate) const PKEY_VECTOR_SCORE_COLUMN: &str = "_PKEY_VECTOR_SCORE";
+// Unified user-visible vector-search score column (matches the engine metadata
+// column name used by Spark and the DataFusion table function).
+pub(crate) const SEARCH_SCORE_COLUMN: &str = "__paimon_search_score";
fn data_invalid(message: impl Into<String>) -> crate::Error {
crate::Error::DataInvalid {
@@ -104,9 +106,7 @@ impl<'a> PkVectorPositionRead<'a> {
// (3) reserved-column-name check against the requested output fields
for field in self.reader.read_type() {
- if field.name() == PKEY_VECTOR_POSITION_COLUMN
- || field.name() == PKEY_VECTOR_SCORE_COLUMN
- {
+ if field.name() == PKEY_VECTOR_POSITION_COLUMN || field.name() ==
SEARCH_SCORE_COLUMN {
return Err(data_invalid(format!(
"Reserved metadata column name conflicts with a table
column: {}",
field.name()
@@ -185,7 +185,7 @@ impl<'a> PkVectorPositionRead<'a> {
}
}
-/// Append `_PKEY_VECTOR_POSITION` (and, when `want_score_col`,
`_PKEY_VECTOR_SCORE`)
+/// Append `_PKEY_VECTOR_POSITION` (and, when `want_score_col`,
`__paimon_search_score`)
/// to `batch`. `positions` are the file-LOCAL physical positions of the
batch's
/// rows, supplied by the caller's cursor into the effective (DV-filtered)
/// selection, so they align 1:1 with the batch rows in order. Scores are
looked
@@ -232,7 +232,7 @@ fn append_metadata_columns(
columns.push(Arc::new(Int64Array::from(positions.to_vec())));
if let Some(sv) = score_vals {
fields.push(ArrowField::new(
- PKEY_VECTOR_SCORE_COLUMN,
+ SEARCH_SCORE_COLUMN,
ArrowDataType::Float32,
false,
));
@@ -557,7 +557,7 @@ mod tests {
async fn test_reads_selected_positions_with_position_column() {
// rows id=[10,11,12,13,14], first_row_id=0, select positions [0,2,4]
// -> output ids [10,12,14], _PKEY_VECTOR_POSITION [0,2,4], ascending;
- // no _ROW_ID leak and no _PKEY_VECTOR_SCORE column.
+ // no _ROW_ID leak and no __paimon_search_score column.
let data = write_mosaic_single_group(&id_batch(vec![10, 11, 12, 13,
14]));
let (reader, split, _dv) = build_reader_and_split(
"memory:/pkvpr_basic",
@@ -594,7 +594,7 @@ mod tests {
"_ROW_ID must not leak into output"
);
assert!(
- column_by_name(batch, PKEY_VECTOR_SCORE_COLUMN).is_none(),
+ column_by_name(batch, SEARCH_SCORE_COLUMN).is_none(),
"_PKEY_VECTOR_SCORE must be absent when no scores are supplied"
);
}
@@ -603,7 +603,7 @@ mod tests {
#[tokio::test]
async fn test_score_alignment_non_contiguous() {
// select [0,2,5] with scores {0:0.9, 2:0.5, 5:0.1}
- // -> _PKEY_VECTOR_SCORE aligned by returned position: [0.9,0.5,0.1].
+ // -> __paimon_search_score aligned by returned position:
[0.9,0.5,0.1].
let data = write_mosaic_single_group(&id_batch(vec![10, 11, 12, 13,
14, 15]));
let (reader, split, _dv) = build_reader_and_split(
"memory:/pkvpr_scores",
@@ -636,7 +636,7 @@ mod tests {
vec![0, 2, 5]
);
assert_eq!(
- collect_f32(&batches, PKEY_VECTOR_SCORE_COLUMN),
+ collect_f32(&batches, SEARCH_SCORE_COLUMN),
vec![0.9, 0.5, 0.1]
);
}
@@ -669,7 +669,7 @@ mod tests {
.unwrap();
for batch in &batches {
- assert!(column_by_name(batch, PKEY_VECTOR_SCORE_COLUMN).is_none());
+ assert!(column_by_name(batch, SEARCH_SCORE_COLUMN).is_none());
}
}
@@ -704,7 +704,7 @@ mod tests {
vec![0, 2, 3]
);
assert_eq!(
- collect_f32(&batches, PKEY_VECTOR_SCORE_COLUMN),
+ collect_f32(&batches, SEARCH_SCORE_COLUMN),
vec![0.4, 0.2, 0.1]
);
}
@@ -755,7 +755,7 @@ mod tests {
vec![1, 2, 4]
);
assert_eq!(
- collect_f32(&batches, PKEY_VECTOR_SCORE_COLUMN),
+ collect_f32(&batches, SEARCH_SCORE_COLUMN),
vec![0.9, 0.5, 0.1]
);
}
@@ -814,7 +814,7 @@ mod tests {
vec![1, 3, 4, 5]
);
assert_eq!(
- collect_f32(&batches, PKEY_VECTOR_SCORE_COLUMN),
+ collect_f32(&batches, SEARCH_SCORE_COLUMN),
vec![0.9, 0.5, 0.3, 0.1]
);
}
diff --git a/crates/paimon/src/table/vector_search_builder.rs
b/crates/paimon/src/table/vector_search_builder.rs
index e8f1abc9..67579f1f 100644
--- a/crates/paimon/src/table/vector_search_builder.rs
+++ b/crates/paimon/src/table/vector_search_builder.rs
@@ -21,8 +21,8 @@ use crate::io::FileIO;
use crate::lumina::reader::LuminaVectorGlobalIndexReader;
use crate::lumina::{is_lumina_index_type, LuminaIndexMeta, LuminaVectorMetric};
use crate::spec::{
- CoreOptions, DataField, FileKind, GlobalIndexSearchMode, IndexFileMeta,
IndexManifest,
- IndexManifestEntry, Predicate, ROW_ID_FIELD_NAME,
+ BigIntType, CoreOptions, DataField, DataType, FileKind,
GlobalIndexSearchMode, IndexFileMeta,
+ IndexManifest, IndexManifestEntry, Predicate, ROW_ID_FIELD_ID,
ROW_ID_FIELD_NAME,
};
use crate::table::data_file_reader::DataFileReader;
use crate::table::global_index_scanner::{
@@ -35,9 +35,7 @@ use crate::table::pk_vector_orchestrator::{
as_split_exact_reader_factory, build_indexed_splits, PkVectorCandidate,
PkVectorOrchestrator,
PkVectorSearchSplit,
};
-use crate::table::pk_vector_position_read::{
- PKEY_VECTOR_POSITION_COLUMN, PKEY_VECTOR_SCORE_COLUMN,
-};
+use crate::table::pk_vector_position_read::{PKEY_VECTOR_POSITION_COLUMN,
SEARCH_SCORE_COLUMN};
use crate::table::pk_vector_scan::{PkVectorScan, PkVectorScanPlan};
use crate::table::read_builder::resolve_projected_fields;
use crate::table::source::DataSplit;
@@ -59,6 +57,7 @@ use roaring::RoaringTreemap;
use std::cmp::Ordering;
use std::collections::{BinaryHeap, HashMap, HashSet};
use std::io::Cursor;
+use std::sync::Arc;
const INDEX_DIR: &str = "index";
@@ -159,7 +158,7 @@ impl<'a> VectorSearchBuilder<'a> {
}
/// Restrict the columns materialized by
[`execute_read`](Self::execute_read)
- /// to `cols` (plus the always-appended `_PKEY_VECTOR_SCORE`). Without this
+ /// to `cols` (plus the always-appended `__paimon_search_score`). Without
this
/// call `execute_read` materializes every user table column. Only affects
/// `execute_read`; the search-only paths ignore it.
pub fn with_projection(&mut self, cols: &[&str]) -> &mut Self {
@@ -241,11 +240,11 @@ impl<'a> VectorSearchBuilder<'a> {
}
/// Run the vector search and materialize the matching rows as Arrow
batches,
- /// ordered best-first. Only supported for primary-key vector indexes; a
- /// data-evolution table or a query targeting a non-PK-vector column fails
- /// loud. Output columns are the projected user table columns (all user
- /// columns by default, or those set via
- /// [`with_projection`](Self::with_projection)) plus `_PKEY_VECTOR_SCORE`;
+ /// ordered best-first. Supported for both primary-key vector indexes and
+ /// data-evolution (global-index) vector search; a query targeting a column
+ /// that is neither fails loud. Output columns are the projected user table
+ /// columns (all user columns by default, or those set via
+ /// [`with_projection`](Self::with_projection)) plus
`__paimon_search_score`;
/// `_ROW_ID` and `_PKEY_VECTOR_POSITION` are always hidden.
pub async fn execute_read(&self) -> crate::Result<ArrowRecordBatchStream> {
// Fail closed: returns data outside `TableScan`/`TableRead`.
@@ -283,10 +282,99 @@ impl<'a> VectorSearchBuilder<'a> {
}
}
- Err(crate::Error::DataInvalid {
- message: "vector search read is only supported for primary-key
vector indexes".into(),
- source: None,
- })
+ // Data-evolution (global-index) vector search: materialize rows from
the
+ // scored global row-ids and attach the unified score column. A
non-vector
+ // column or a set filter fails loud inside execute_scored below.
+ self.execute_de_vector_read().await
+ }
+
+ /// Materialize the best-first data-evolution vector search hits into Arrow
+ /// rows. The global-index search returns global `_ROW_ID`s and their
scores; a
+ /// subsequent row-range read materializes those rows, and each row's
score is
+ /// joined back by `_ROW_ID`. Output columns are the projected user table
+ /// columns (all user columns by default) plus `__paimon_search_score`;
`_ROW_ID`
+ /// is always hidden. A filter is unsupported here and fails loud inside
+ /// `execute_scored`.
+ async fn execute_de_vector_read(&self) ->
crate::Result<ArrowRecordBatchStream> {
+ // Validate the target column exists and is a vector-bearing type
before any
+ // work. The data-evolution search returns an empty result for an
unknown
+ // field (its scored-path behavior), which would make a typo'd or
scalar
+ // column look like a normal empty read here — violating
`execute_read`'s
+ // fail-loud contract (a C/Doris caller would see EOF, not an input
error).
+ // Reject it up front instead.
+ let vector_column =
+ self.vector_column
+ .as_deref()
+ .ok_or_else(|| crate::Error::ConfigInvalid {
+ message: "Vector column must be set via
with_vector_column()".to_string(),
+ })?;
+ let field = self
+ .table
+ .schema()
+ .fields()
+ .iter()
+ .find(|f| f.name() == vector_column)
+ .ok_or_else(|| crate::Error::DataInvalid {
+ message: format!("vector search column '{vector_column}' does
not exist"),
+ source: None,
+ })?;
+ // Require a FLOAT-element vector column: `ARRAY<FLOAT>` or
`VECTOR<FLOAT>`,
+ // matching the element type the vector index/search operates on. An
+ // `ARRAY<INT>` (or any non-float element) is not a searchable vector
column.
+ let is_float_vector = match field.data_type() {
+ DataType::Vector(t) => matches!(t.element_type(),
DataType::Float(_)),
+ DataType::Array(t) => matches!(t.element_type(),
DataType::Float(_)),
+ _ => false,
+ };
+ if !is_float_vector {
+ return Err(crate::Error::DataInvalid {
+ message: format!(
+ "vector search column '{vector_column}' must be a FLOAT
vector column \
+ (ARRAY<FLOAT> or VECTOR<FLOAT>), got {:?}",
+ field.data_type()
+ ),
+ source: None,
+ });
+ }
+
+ let sr = self.execute_scored().await?;
+
+ // Resolve the projected user columns up front so an invalid projection
+ // fails loud even when the result is empty.
+ let mut read_type = self.resolve_materialize_read_type()?;
+
+ if sr.is_empty() {
+ return Ok(Box::pin(stream::empty()));
+ }
+
+ // rank = ordinal in the best-first scored result; score = the aligned
score.
+ // Build ranges first (validates ids fit in i64::MAX) before
constructing the map.
+ let ranges = sr.to_row_ranges()?;
+ let mut rank_score_of: HashMap<i64, (usize, f32)> = HashMap::new();
+ for (rank, (&id, &score)) in
sr.row_ids.iter().zip(sr.scores.iter()).enumerate() {
+ rank_score_of.insert(id as i64, (rank, score));
+ }
+
+ // Add _ROW_ID as the join key for score alignment; it is stripped
before output.
+ if !read_type.iter().any(|f| f.name() == ROW_ID_FIELD_NAME) {
+ read_type.push(row_id_data_field());
+ }
+
+ let mut read_builder = self.table.new_read_builder();
+ read_builder
+ .with_read_type(read_type)
+ .with_row_ranges(ranges);
+ let scan = read_builder.new_scan();
+ let plan = scan.plan().await?;
+ let table_read = read_builder.new_read()?;
+ let mut stream = table_read.to_arrow(plan.splits())?;
+
+ let mut batches: Vec<RecordBatch> = Vec::new();
+ while let Some(batch) = stream.try_next().await? {
+ batches.push(batch);
+ }
+ let output = attach_scores_by_row_id(&batches, &rank_score_of,
sr.len())?;
+ Ok(Box::pin(stream::iter(output.into_iter().map(Ok))))
}
/// Shared PK-vector search core for both the search-only and
search-and-read
@@ -493,7 +581,7 @@ impl<'a> VectorSearchBuilder<'a> {
///
/// Output columns are the projected user table columns (all user columns
when
/// [`with_projection`](Self::with_projection) was not called) plus
- /// `_PKEY_VECTOR_SCORE`; `_ROW_ID` and `_PKEY_VECTOR_POSITION` are always
+ /// `__paimon_search_score`; `_ROW_ID` and `_PKEY_VECTOR_POSITION` are
always
/// hidden. Rows are emitted best-first (the candidate order), which
differs
/// from the file/position order the orchestrator materializes in.
async fn execute_primary_key_vector_read(
@@ -584,20 +672,27 @@ impl<'a> VectorSearchBuilder<'a> {
/// names resolved via `resolve_projected_fields`. Rejects reserved
metadata
/// names and `_ROW_ID` so a user cannot request a hidden column.
fn resolve_materialize_read_type(&self) -> crate::Result<Vec<DataField>> {
+ let is_reserved = |name: &str| {
+ name == PKEY_VECTOR_POSITION_COLUMN
+ || name == SEARCH_SCORE_COLUMN
+ || name == ROW_ID_FIELD_NAME
+ || name == "_PKEY_VECTOR_SCORE"
+ };
+ let reserved_err = |name: &str| crate::Error::DataInvalid {
+ message: format!(
+ "vector search read projection must not request reserved
column '{name}'"
+ ),
+ source: None,
+ };
let fields = match &self.projection {
None => self.table.schema().fields().to_vec(),
Some(names) => {
+ // Reject a requested reserved name on the raw list first:
these
+ // names are not real table columns, so
`resolve_projected_fields`
+ // would otherwise fail with a confusing "not found" error.
for name in names {
- if name == PKEY_VECTOR_POSITION_COLUMN
- || name == PKEY_VECTOR_SCORE_COLUMN
- || name == ROW_ID_FIELD_NAME
- {
- return Err(crate::Error::DataInvalid {
- message: format!(
- "vector search read projection must not
request reserved column '{name}'"
- ),
- source: None,
- });
+ if is_reserved(name) {
+ return Err(reserved_err(name));
}
}
resolve_projected_fields(
@@ -608,6 +703,17 @@ impl<'a> VectorSearchBuilder<'a> {
)?
}
};
+ // Reject reserved output-column names on the RESOLVED field list too
— this
+ // is what catches the default (no `with_projection`) case where a
user table
+ // column is literally named `__paimon_search_score` (or the legacy
+ // `_PKEY_VECTOR_SCORE` alias): it would otherwise survive and collide
with
+ // the score column appended during materialization, producing two
+ // identically named output columns.
+ for field in &fields {
+ if is_reserved(field.name()) {
+ return Err(reserved_err(field.name()));
+ }
+ }
Ok(fields)
}
}
@@ -1199,7 +1305,7 @@ fn collect_ranked_rows(
/// Reorder the materialized rows into best-first order and drop the internal
/// `_PKEY_VECTOR_POSITION` column, yielding a single output batch (empty input
-/// yields no batches). The projected user columns and `_PKEY_VECTOR_SCORE` are
+/// yields no batches). The projected user columns and `__paimon_search_score`
are
/// retained.
fn reorder_and_strip_position(
batches: &[RecordBatch],
@@ -1221,7 +1327,7 @@ fn reorder_and_strip_position(
})?;
// Drop the internal position column; keep every other column (projected
user
- // columns + _PKEY_VECTOR_SCORE) in order.
+ // columns + __paimon_search_score) in order.
let position_idx = reordered
.schema()
.index_of(PKEY_VECTOR_POSITION_COLUMN)
@@ -1241,6 +1347,132 @@ fn reorder_and_strip_position(
Ok(vec![projected])
}
+/// The `_ROW_ID` field to append to a data-evolution read type so the reader
+/// fills each row's global id. Mirrors the field the `DataEvolutionReader`
+/// recognizes (Int64 / `BigInt`, nullable): a data file lacking `first_row_id`
+/// yields nulls here, which `attach_scores_by_row_id` then fails loud on
rather
+/// than mis-aligning scores.
+fn row_id_data_field() -> DataField {
+ DataField::new(
+ ROW_ID_FIELD_ID,
+ ROW_ID_FIELD_NAME.to_string(),
+ DataType::BigInt(BigIntType::with_nullable(true)),
+ )
+}
+
+/// Collect materialized DE rows, join each row's `(rank, score)` by its global
+/// `_ROW_ID`, reorder to the search rank order, append the
`__paimon_search_score`
+/// column, and drop `_ROW_ID`. Every row must map to a search candidate and
the
+/// total materialized count must equal `expected_len`; a miss or count
mismatch
+/// fails loud rather than silently dropping or NaN-scoring a row. Empty input
+/// yields no batches.
+fn attach_scores_by_row_id(
+ batches: &[RecordBatch],
+ rank_score_of: &HashMap<i64, (usize, f32)>,
+ expected_len: usize,
+) -> crate::Result<Vec<RecordBatch>> {
+ // (rank, batch_index, row_index, score) per materialized row.
+ let mut ranked: Vec<(usize, usize, usize, f32)> = Vec::new();
+ for (batch_index, batch) in batches.iter().enumerate() {
+ let row_id_idx =
+ batch
+ .schema()
+ .index_of(ROW_ID_FIELD_NAME)
+ .map_err(|_| crate::Error::DataInvalid {
+ message: format!("materialized batch missing
{ROW_ID_FIELD_NAME} column"),
+ source: None,
+ })?;
+ let col = batch.column(row_id_idx);
+ let ids =
+ col.as_any()
+ .downcast_ref::<Int64Array>()
+ .ok_or_else(|| crate::Error::DataInvalid {
+ message: format!("{ROW_ID_FIELD_NAME} column is not
Int64"),
+ source: None,
+ })?;
+ for row_index in 0..batch.num_rows() {
+ if ids.is_null(row_index) {
+ return Err(crate::Error::DataInvalid {
+ message: format!(
+ "materialized DE vector row has null
{ROW_ID_FIELD_NAME}; cannot align score"
+ ),
+ source: None,
+ });
+ }
+ let id = ids.value(row_index);
+ let (rank, score) =
+ *rank_score_of
+ .get(&id)
+ .ok_or_else(|| crate::Error::DataInvalid {
+ message: format!(
+ "materialized DE vector row (row id {id}) has no
matching search candidate"
+ ),
+ source: None,
+ })?;
+ ranked.push((rank, batch_index, row_index, score));
+ }
+ }
+
+ if ranked.len() != expected_len {
+ return Err(crate::Error::DataInvalid {
+ message: format!(
+ "DE vector materialization produced {} rows but search
returned {expected_len}",
+ ranked.len()
+ ),
+ source: None,
+ });
+ }
+ if ranked.is_empty() {
+ return Ok(Vec::new());
+ }
+
+ ranked.sort_by_key(|r| r.0);
+ let indices: Vec<(usize, usize)> = ranked.iter().map(|r| (r.1,
r.2)).collect();
+ let refs: Vec<&RecordBatch> = batches.iter().collect();
+ let reordered =
+ interleave_record_batch(&refs, &indices).map_err(|e|
crate::Error::DataInvalid {
+ message: format!("failed to reorder DE vector search rows: {e}"),
+ source: None,
+ })?;
+
+ // Drop _ROW_ID.
+ let row_id_idx = reordered
+ .schema()
+ .index_of(ROW_ID_FIELD_NAME)
+ .map_err(|_| crate::Error::DataInvalid {
+ message: format!("reordered batch missing {ROW_ID_FIELD_NAME}
column"),
+ source: None,
+ })?;
+ let keep: Vec<usize> = (0..reordered.num_columns())
+ .filter(|i| *i != row_id_idx)
+ .collect();
+ let stripped = reordered
+ .project(&keep)
+ .map_err(|e| crate::Error::DataInvalid {
+ message: format!("failed to drop {ROW_ID_FIELD_NAME} column: {e}"),
+ source: None,
+ })?;
+
+ // Append the score column in rank order.
+ let scores: Vec<f32> = ranked.iter().map(|r| r.3).collect();
+ let score_array: Arc<dyn Array> = Arc::new(Float32Array::from(scores));
+ let mut fields: Vec<Arc<arrow_schema::Field>> =
+ stripped.schema().fields().iter().cloned().collect();
+ fields.push(Arc::new(arrow_schema::Field::new(
+ SEARCH_SCORE_COLUMN,
+ arrow_schema::DataType::Float32,
+ false,
+ )));
+ let out_schema = Arc::new(arrow_schema::Schema::new(fields));
+ let mut columns = stripped.columns().to_vec();
+ columns.push(score_array);
+ let out = RecordBatch::try_new(out_schema, columns).map_err(|e|
crate::Error::DataInvalid {
+ message: format!("failed to append DE vector score column: {e}"),
+ source: None,
+ })?;
+ Ok(vec![out])
+}
+
fn indexed_search_limit(limit: usize, refine_factor: usize) ->
crate::Result<usize> {
if refine_factor == 0 {
return Ok(limit);
@@ -2052,8 +2284,9 @@ mod tests {
IndexFileMeta, IndexManifestEntry, IntType, PredicateBuilder, Schema,
TableSchema,
};
use crate::table::source::DataSplitBuilder;
+ use crate::table::{TableCommit, TableWrite};
use crate::vindex::IVF_FLAT_IDENTIFIER;
- use arrow_array::builder::{FixedSizeListBuilder, Float32Builder};
+ use arrow_array::builder::{FixedSizeListBuilder, Float32Builder,
ListBuilder};
use arrow_array::ArrayRef;
use arrow_array::Int32Array;
use arrow_schema::{DataType as ArrowDataType, Field as ArrowField, Schema
as ArrowSchema};
@@ -2786,6 +3019,87 @@ mod tests {
)
}
+ /// A data-evolution (global-index) vector table with a committed IVF-flat
+ /// index over the `embedding` column: row-tracking + data-evolution +
+ /// global-index enabled so committed data files carry `first_row_id` and
the
+ /// search returns global row-ids that `execute_read` can materialize. The
+ /// returned table has one committed batch of `(id, embedding)` rows and a
real
+ /// vindex index built end-to-end.
+ async fn de_vector_table() -> Table {
+ let table_path = "memory:/de_vector_search_test";
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column(
+ "embedding",
+
DataType::Array(ArrayType::new(DataType::Float(FloatType::new()))),
+ )
+ .option("row-tracking.enabled", "true")
+ .option("data-evolution.enabled", "true")
+ .option("global-index.enabled", "true")
+ .option("global-index.row-count-per-shard", "10")
+ .option("ivf-flat.dimension", "2")
+ .option("ivf-flat.nlist", "2")
+ .build()
+ .unwrap();
+ let file_io = FileIOBuilder::new("memory").build().unwrap();
+ let table = Table::new(
+ file_io.clone(),
+ Identifier::new("default", "de_vector_test"),
+ table_path.to_string(),
+ TableSchema::new(0, &schema),
+ None,
+ );
+ file_io
+ .mkdirs(&format!("{table_path}/snapshot/"))
+ .await
+ .unwrap();
+ file_io
+ .mkdirs(&format!("{table_path}/manifest/"))
+ .await
+ .unwrap();
+
+ let ids = vec![1, 2, 3];
+ let vectors = vec![vec![1.0, 0.0], vec![0.0, 1.0], vec![1.0, 1.0]];
+ let element_field = Arc::new(ArrowField::new("element",
ArrowDataType::Float32, true));
+ let mut vector_builder =
+
ListBuilder::new(Float32Builder::new()).with_field(element_field.clone());
+ for vector in vectors {
+ for value in vector {
+ vector_builder.values().append_value(value);
+ }
+ vector_builder.append(true);
+ }
+ let arrow_schema = Arc::new(ArrowSchema::new(vec![
+ ArrowField::new("id", ArrowDataType::Int32, false),
+ ArrowField::new("embedding", ArrowDataType::List(element_field),
true),
+ ]));
+ let batch = RecordBatch::try_new(
+ arrow_schema,
+ vec![
+ Arc::new(Int32Array::from(ids)) as ArrayRef,
+ Arc::new(vector_builder.finish()) as ArrayRef,
+ ],
+ )
+ .unwrap();
+
+ let mut table_write = TableWrite::new(&table,
"test-user".to_string()).unwrap();
+ table_write.write_arrow_batch(&batch).await.unwrap();
+ let messages = table_write.prepare_commit().await.unwrap();
+ TableCommit::new(table.clone(), "test-user".to_string())
+ .commit(messages)
+ .await
+ .unwrap();
+
+ let built = table
+ .new_vindex_index_build_builder(IVF_FLAT_IDENTIFIER)
+ .with_index_column("embedding")
+ .execute()
+ .await
+ .unwrap();
+ assert!(built > 0, "DE fixture must build a global vector index");
+ table
+ }
+
#[tokio::test]
async fn pk_branch_disabled_falls_through_to_de_path() {
// No pk-vector.index.columns: behaves exactly as the DE path. With no
@@ -3031,13 +3345,13 @@ mod tests {
// ---- Task B: search-and-read (`execute_read`) tests ----
/// Build a small materialization batch: user column `id: Int32`, the
internal
- /// `_PKEY_VECTOR_POSITION: Int64`, and `_PKEY_VECTOR_SCORE: Float32`
(mirroring
+ /// `_PKEY_VECTOR_POSITION: Int64`, and `__paimon_search_score: Float32`
(mirroring
/// what `PkVectorIndexedSplitRead` emits for a single file).
fn materialized_batch(rows: &[(i32, i64, f32)]) -> RecordBatch {
let schema = Arc::new(ArrowSchema::new(vec![
ArrowField::new("id", ArrowDataType::Int32, false),
ArrowField::new(PKEY_VECTOR_POSITION_COLUMN, ArrowDataType::Int64,
false),
- ArrowField::new(PKEY_VECTOR_SCORE_COLUMN, ArrowDataType::Float32,
false),
+ ArrowField::new(SEARCH_SCORE_COLUMN, ArrowDataType::Float32,
false),
]));
let ids = Int32Array::from(rows.iter().map(|(id, _, _)|
*id).collect::<Vec<_>>());
let positions = Int64Array::from(rows.iter().map(|(_, pos, _)|
*pos).collect::<Vec<_>>());
@@ -3099,7 +3413,7 @@ mod tests {
assert_eq!(i32_col(out, "id"), vec![41, 42, 40]);
// Score column preserved and aligned to the reordered rows.
assert_eq!(
- f32_col(out, PKEY_VECTOR_SCORE_COLUMN),
+ f32_col(out, SEARCH_SCORE_COLUMN),
vec![l2_score(1.0), l2_score(4.0), l2_score(9.0)]
);
// Position column dropped; _ROW_ID never present.
@@ -3127,7 +3441,7 @@ mod tests {
let out = reorder_and_strip_position(&batches, ranked).unwrap();
assert_eq!(i32_col(&out[0], "id"), vec![20, 11, 10]);
assert_eq!(
- f32_col(&out[0], PKEY_VECTOR_SCORE_COLUMN),
+ f32_col(&out[0], SEARCH_SCORE_COLUMN),
vec![l2_score(0.5), l2_score(1.0), l2_score(9.0)]
);
}
@@ -3154,49 +3468,337 @@ mod tests {
);
}
+ #[test]
+ fn attach_scores_reorders_by_rank_not_score() {
+ use arrow_array::{Int32Array, Int64Array, RecordBatch};
+ use arrow_schema::{DataType, Field, Schema};
+ use std::sync::Arc;
+
+ // Two rows materialized in row-id order [10, 20]; ranks say 20 is
best (rank 0),
+ // 10 is rank 1. Scores tie at 0.5 to prove ordering follows rank, not
score.
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("id", DataType::Int32, false),
+ Field::new(ROW_ID_FIELD_NAME, DataType::Int64, false),
+ ]));
+ let batch = RecordBatch::try_new(
+ schema,
+ vec![
+ Arc::new(Int32Array::from(vec![100, 200])),
+ Arc::new(Int64Array::from(vec![10, 20])),
+ ],
+ )
+ .unwrap();
+ let mut map = HashMap::new();
+ map.insert(20i64, (0usize, 0.5f32));
+ map.insert(10i64, (1usize, 0.5f32));
+
+ let out = attach_scores_by_row_id(&[batch], &map, 2).unwrap();
+ assert_eq!(out.len(), 1);
+ let b = &out[0];
+ // _ROW_ID stripped, score appended.
+ assert!(b.schema().index_of(ROW_ID_FIELD_NAME).is_err());
+ let score_idx = b.schema().index_of("__paimon_search_score").unwrap();
+ assert_eq!(
+ b.schema().field(score_idx).data_type(),
+ &arrow_schema::DataType::Float32
+ );
+ // Row order is rank order: id 200 (rank 0) first, then id 100 (rank
1).
+ let ids = b.column(0).as_any().downcast_ref::<Int32Array>().unwrap();
+ assert_eq!(ids.values(), &[200, 100]);
+ }
+
+ #[test]
+ fn attach_scores_fails_on_unknown_row_id() {
+ use arrow_array::{Int32Array, Int64Array, RecordBatch};
+ use arrow_schema::{DataType, Field, Schema};
+ use std::sync::Arc;
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("id", DataType::Int32, false),
+ Field::new(ROW_ID_FIELD_NAME, DataType::Int64, false),
+ ]));
+ let batch = RecordBatch::try_new(
+ schema,
+ vec![
+ Arc::new(Int32Array::from(vec![1])),
+ Arc::new(Int64Array::from(vec![99])),
+ ],
+ )
+ .unwrap();
+ let map: HashMap<i64, (usize, f32)> = HashMap::new(); // no entry for
99
+ let err = attach_scores_by_row_id(&[batch], &map, 1).unwrap_err();
+ assert!(matches!(err, crate::Error::DataInvalid { .. }));
+ }
+
+ #[test]
+ fn attach_scores_fails_on_count_mismatch() {
+ use arrow_array::{Int32Array, Int64Array, RecordBatch};
+ use arrow_schema::{DataType, Field, Schema};
+ use std::sync::Arc;
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("id", DataType::Int32, false),
+ Field::new(ROW_ID_FIELD_NAME, DataType::Int64, false),
+ ]));
+ let batch = RecordBatch::try_new(
+ schema,
+ vec![
+ Arc::new(Int32Array::from(vec![1])),
+ Arc::new(Int64Array::from(vec![10])),
+ ],
+ )
+ .unwrap();
+ let mut map = HashMap::new();
+ map.insert(10i64, (0usize, 0.5f32));
+ // expected_len 2 but only 1 row materialized.
+ let err = attach_scores_by_row_id(&[batch], &map, 2).unwrap_err();
+ assert!(matches!(err, crate::Error::DataInvalid { .. }));
+ }
+
+ #[test]
+ fn attach_scores_fails_on_null_row_id() {
+ use arrow_array::{Int32Array, Int64Array, RecordBatch};
+ use arrow_schema::{DataType, Field, Schema};
+ use std::sync::Arc;
+ // _ROW_ID column has a NULL at row 1; the map contains the non-null
id, so
+ // the failure is specifically the null (not an unknown id).
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("id", DataType::Int32, false),
+ Field::new(ROW_ID_FIELD_NAME, DataType::Int64, true),
+ ]));
+ let batch = RecordBatch::try_new(
+ schema,
+ vec![
+ Arc::new(Int32Array::from(vec![1, 2])),
+ Arc::new(Int64Array::from(vec![Some(10i64), None])),
+ ],
+ )
+ .unwrap();
+ let mut map = HashMap::new();
+ map.insert(10i64, (0usize, 0.5f32));
+ let err = attach_scores_by_row_id(&[batch], &map, 2).unwrap_err();
+ assert!(matches!(err, crate::Error::DataInvalid { .. }));
+ }
+
+ #[test]
+ fn attach_scores_fails_on_wrong_type_row_id() {
+ use arrow_array::{Int32Array, RecordBatch};
+ use arrow_schema::{DataType, Field, Schema};
+ use std::sync::Arc;
+ // _ROW_ID column is Int32, not Int64: the downcast fails loud.
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("id", DataType::Int32, false),
+ Field::new(ROW_ID_FIELD_NAME, DataType::Int32, false),
+ ]));
+ let batch = RecordBatch::try_new(
+ schema,
+ vec![
+ Arc::new(Int32Array::from(vec![1])),
+ Arc::new(Int32Array::from(vec![10])),
+ ],
+ )
+ .unwrap();
+ let mut map = HashMap::new();
+ map.insert(10i64, (0usize, 0.5f32));
+ let err = attach_scores_by_row_id(&[batch], &map, 1).unwrap_err();
+ assert!(matches!(err, crate::Error::DataInvalid { .. }));
+ }
+
#[tokio::test]
- async fn execute_read_de_table_fails_loud() {
- // No pk-vector index configured: execute_read must fail loud (the DE
path
- // has no row materialization).
+ async fn execute_read_de_table_empty_snapshot_yields_empty_stream() {
+ // No pk-vector index configured and no snapshot: execute_read routes
to the
+ // data-evolution path, whose search finds nothing and returns an empty
+ // stream (not an error).
let table = pk_vector_table(&[]);
- let err = table
+ let mut stream = table
.new_vector_search_builder()
.with_vector_column("embedding")
.with_query_vector(vec![1.0])
.with_limit(5)
.execute_read()
.await
- .map(|_| ())
- .expect_err("DE read must fail loud");
- assert!(
- matches!(err, crate::Error::DataInvalid { ref message, .. }
- if message.contains("only supported for primary-key")),
- "unexpected error: {err:?}"
- );
+ .expect("DE read over an empty table must succeed with no rows");
+ let mut rows = 0usize;
+ while let Some(batch) = stream.try_next().await.unwrap() {
+ rows += batch.num_rows();
+ }
+ assert_eq!(rows, 0, "empty DE table must yield no rows");
}
#[tokio::test]
- async fn execute_read_non_pk_column_fails_loud() {
+ async fn execute_read_unknown_column_fails_loud() {
// pk-vector index configured for "embedding", but the query targets a
- // different column -> read is unsupported.
+ // column that does not exist. The read path must fail loud rather than
+ // fall through to the data-evolution path and return an empty stream
(a
+ // typo must not look like a normal empty read through the C API).
let table = pk_vector_table(&[
("pk-vector.index.columns", "embedding"),
("fields.embedding.pk-vector.index.type", IVF_FLAT_IDENTIFIER),
("fields.embedding.pk-vector.distance.metric", "l2"),
]);
- let err = table
+ let err = match table
.new_vector_search_builder()
.with_vector_column("other")
.with_query_vector(vec![1.0])
.with_limit(5)
.execute_read()
.await
- .map(|_| ())
- .expect_err("non-PK column read must fail loud");
+ {
+ Ok(_) => panic!("unknown vector column must fail loud on
execute_read"),
+ Err(e) => e,
+ };
assert!(
- matches!(err, crate::Error::DataInvalid { ref message, .. }
- if message.contains("only supported for primary-key")),
- "unexpected error: {err:?}"
+ matches!(&err, crate::Error::DataInvalid { message, .. } if
message.contains("does not exist")),
+ "expected a does-not-exist error, got: {err}"
+ );
+ }
+
+ #[tokio::test]
+ async fn execute_read_scalar_column_fails_loud() {
+ // A scalar (non-vector) column targeted by a vector read must fail
loud,
+ // not return an empty data-evolution stream.
+ let table = pk_vector_table(&[
+ ("pk-vector.index.columns", "embedding"),
+ ("fields.embedding.pk-vector.index.type", IVF_FLAT_IDENTIFIER),
+ ("fields.embedding.pk-vector.distance.metric", "l2"),
+ ]);
+ let err = match table
+ .new_vector_search_builder()
+ .with_vector_column("id") // scalar Int column
+ .with_query_vector(vec![1.0])
+ .with_limit(5)
+ .execute_read()
+ .await
+ {
+ Ok(_) => panic!("scalar vector column must fail loud on
execute_read"),
+ Err(e) => e,
+ };
+ assert!(
+ matches!(&err, crate::Error::DataInvalid { message, .. } if
message.contains("must be a FLOAT vector column")),
+ "expected a not-a-vector-column error, got: {err}"
+ );
+ }
+
+ #[tokio::test]
+ async fn execute_read_non_float_vector_column_fails_loud() {
+ // An ARRAY<INT> column is not a searchable vector column (the
index/search
+ // operates on FLOAT elements). It must fail loud rather than fall
through
+ // to the DE path and return an empty stream.
+ use crate::spec::{ArrayType, IntType, Schema, TableSchema};
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column(
+ "embedding",
+ DataType::Array(ArrayType::new(DataType::Int(IntType::new()))),
+ )
+ .build()
+ .unwrap();
+ let table = Table::new(
+ FileIOBuilder::new("memory").build().unwrap(),
+ Identifier::new("default", "de_non_float_vector"),
+ "memory:/de_non_float_vector".to_string(),
+ TableSchema::new(0, &schema),
+ None,
+ );
+ let err = match table
+ .new_vector_search_builder()
+ .with_vector_column("embedding")
+ .with_query_vector(vec![1.0])
+ .with_limit(5)
+ .execute_read()
+ .await
+ {
+ Ok(_) => panic!("ARRAY<INT> vector column must fail loud on
execute_read"),
+ Err(e) => e,
+ };
+ assert!(
+ matches!(&err, crate::Error::DataInvalid { message, .. } if
message.contains("must be a FLOAT vector column")),
+ "expected a FLOAT-vector-column error, got: {err}"
+ );
+ }
+
+ #[tokio::test]
+ async fn
execute_read_default_projection_rejects_reserved_score_column_name() {
+ // A user table column literally named `__paimon_search_score` must be
+ // rejected even under the default (no `with_projection`) projection —
+ // otherwise it survives and collides with the score column appended
during
+ // materialization, producing two identically named output columns.
+ use crate::spec::{FloatType, IntType, Schema, TableSchema, VectorType};
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column(
+ "embedding",
+ DataType::Vector(
+ VectorType::try_new(true, 2,
DataType::Float(FloatType::new())).unwrap(),
+ ),
+ )
+ .column(SEARCH_SCORE_COLUMN, DataType::Float(FloatType::new()))
+ .primary_key(["id"])
+ .option("bucket", "1")
+ .option("pk-vector.index.columns", "embedding")
+ .option("fields.embedding.pk-vector.index.type",
IVF_FLAT_IDENTIFIER)
+ .option("fields.embedding.pk-vector.distance.metric", "l2")
+ .build()
+ .unwrap();
+ let table = Table::new(
+ FileIOBuilder::new("memory").build().unwrap(),
+ Identifier::new("default", "pk_vector_collision"),
+ "memory:/pk_vector_collision".to_string(),
+ TableSchema::new(0, &schema),
+ None,
+ );
+ let err = match table
+ .new_vector_search_builder()
+ .with_vector_column("embedding")
+ .with_query_vector(vec![1.0, 0.0])
+ .with_limit(5)
+ .execute_read()
+ .await
+ {
+ Ok(_) => panic!("default projection over a reserved-named column
must fail loud"),
+ Err(e) => e,
+ };
+ assert!(
+ matches!(&err, crate::Error::DataInvalid { message, .. } if
message.contains("reserved column")),
+ "expected a reserved-column error, got: {err}"
+ );
+ }
+
+ #[tokio::test]
+ async fn
execute_read_de_default_projection_rejects_reserved_score_column_name() {
+ // Same collision, on the data-evolution path: a table with no
PK-vector
+ // index (so the query falls through to DE materialization) but a user
+ // column named `__paimon_search_score` must fail loud under the
default
+ // projection rather than emit two identically named columns.
+ use crate::spec::{ArrayType, FloatType, IntType, Schema, TableSchema};
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column(
+ "embedding",
+
DataType::Array(ArrayType::new(DataType::Float(FloatType::new()))),
+ )
+ .column(SEARCH_SCORE_COLUMN, DataType::Float(FloatType::new()))
+ .build()
+ .unwrap();
+ let table = Table::new(
+ FileIOBuilder::new("memory").build().unwrap(),
+ Identifier::new("default", "de_vector_collision"),
+ "memory:/de_vector_collision".to_string(),
+ TableSchema::new(0, &schema),
+ None,
+ );
+ let err = match table
+ .new_vector_search_builder()
+ .with_vector_column("embedding")
+ .with_query_vector(vec![1.0])
+ .with_limit(5)
+ .execute_read()
+ .await
+ {
+ Ok(_) => panic!("DE default projection over a reserved-named
column must fail loud"),
+ Err(e) => e,
+ };
+ assert!(
+ matches!(&err, crate::Error::DataInvalid { message, .. } if
message.contains("reserved column")),
+ "expected a reserved-column error, got: {err}"
);
}
@@ -3215,7 +3817,8 @@ mod tests {
for reserved in [
ROW_ID_FIELD_NAME,
PKEY_VECTOR_POSITION_COLUMN,
- PKEY_VECTOR_SCORE_COLUMN,
+ SEARCH_SCORE_COLUMN,
+ "_PKEY_VECTOR_SCORE",
] {
let mut builder = table.new_vector_search_builder();
builder
@@ -3249,7 +3852,8 @@ mod tests {
for reserved in [
ROW_ID_FIELD_NAME,
PKEY_VECTOR_POSITION_COLUMN,
- PKEY_VECTOR_SCORE_COLUMN,
+ SEARCH_SCORE_COLUMN,
+ "_PKEY_VECTOR_SCORE",
] {
let mut builder = table.new_vector_search_builder();
builder
@@ -3295,6 +3899,73 @@ mod tests {
let names: Vec<&str> = fields.iter().map(|f| f.name()).collect();
assert_eq!(names, vec!["id"]);
}
+
+ #[tokio::test]
+ async fn de_execute_read_materializes_rows_with_score() {
+ // A data-evolution vector table with a committed global index:
execute_read
+ // must materialize one row per scored hit and carry the unified score
+ // column, in best-first rank order.
+ let table = de_vector_table().await;
+ let query = vec![1.0, 0.0];
+
+ let scored = table
+ .new_vector_search_builder()
+ .with_vector_column("embedding")
+ .with_query_vector(query.clone())
+ .with_limit(3)
+ .execute_scored()
+ .await
+ .unwrap();
+ assert!(!scored.is_empty(), "DE search must return hits");
+
+ let mut stream = table
+ .new_vector_search_builder()
+ .with_vector_column("embedding")
+ .with_query_vector(query)
+ .with_limit(3)
+ .execute_read()
+ .await
+ .unwrap();
+
+ let mut rows = 0usize;
+ let mut saw_score = false;
+ while let Some(batch) = stream.try_next().await.unwrap() {
+ rows += batch.num_rows();
+ saw_score |= batch.schema().index_of(SEARCH_SCORE_COLUMN).is_ok();
+ }
+ assert_eq!(
+ rows,
+ scored.len(),
+ "DE read must emit exactly the scored result count"
+ );
+ assert!(
+ saw_score,
+ "DE read output must carry the search score column"
+ );
+ }
+
+ #[tokio::test]
+ async fn de_execute_read_with_filter_fails_loud() {
+ // A filter on the data-evolution path is unsupported (the DE path
never
+ // reads physical rows), so execute_read must fail loud rather than
drop the
+ // predicate. The guard lives in execute_scored.
+ let table = de_vector_table().await;
+ let filter = id_gt_filter(&table, 1);
+ let err = table
+ .new_vector_search_builder()
+ .with_vector_column("embedding")
+ .with_query_vector(vec![1.0, 0.0])
+ .with_limit(3)
+ .with_filter(filter)
+ .execute_read()
+ .await
+ .map(|_| ())
+ .expect_err("DE read with a filter must fail loud");
+ assert!(
+ matches!(err, crate::Error::DataInvalid { .. }),
+ "unexpected error: {err:?}"
+ );
+ }
}
/// Tests for [`residual_positions_by_file`]: the residual predicate is
applied at
diff --git a/crates/paimon/tests/pk_vector_baseline_test.rs
b/crates/paimon/tests/pk_vector_baseline_test.rs
index 18629315..8ec2fa62 100644
--- a/crates/paimon/tests/pk_vector_baseline_test.rs
+++ b/crates/paimon/tests/pk_vector_baseline_test.rs
@@ -22,7 +22,7 @@
//! the snapshot/manifest/index-manifest metadata — then reads it back through
the
//! public `new_vector_search_builder()` API and asserts both the search result
//! (`execute_scored()` -> `row_ids`/`scores`) and the materialized rows
-//! (`execute_read()` -> Arrow batches, best-first order,
`_PKEY_VECTOR_SCORE`).
+//! (`execute_read()` -> Arrow batches, best-first order,
`__paimon_search_score`).
//!
//! Why Rust-built rather than a committed cross-language fixture: the Java
//! primary-key vector ANN segment is an opaque native Lumina format that
cannot
@@ -491,7 +491,7 @@ async fn read_id_and_scores(
let scores: Vec<f32> = batches
.iter()
.flat_map(|b| {
- let idx = b.schema().index_of("_PKEY_VECTOR_SCORE").unwrap();
+ let idx = b.schema().index_of("__paimon_search_score").unwrap();
b.column(idx)
.as_any()
.downcast_ref::<Float32Array>()
@@ -601,7 +601,7 @@ async fn
pk_vector_end_to_end_returns_expected_row_ids_and_scores() {
);
// Search-and-read: execute_read() materializes the matching rows
best-first
- // with a `_PKEY_VECTOR_SCORE` column, hiding
`_ROW_ID`/`_PKEY_VECTOR_POSITION`.
+ // with a `__paimon_search_score` column, hiding
`_ROW_ID`/`_PKEY_VECTOR_POSITION`.
// Projection ['id'] excludes the vector column.
let (ids, scores, batches) = read_id_and_scores(&table, query.to_vec(), 3,
Some(&["id"])).await;
@@ -634,7 +634,7 @@ async fn
pk_vector_end_to_end_returns_expected_row_ids_and_scores() {
/// back a fixture whose nearest neighbours are at physical positions 5, 1, 3
(in
/// that order) and asserts the materialized output is emitted best-first
/// [5, 1, 3], not in ascending physical position [1, 3, 5]. Also asserts the
full
-/// row content (id + vector values) and the aligned `_PKEY_VECTOR_SCORE`,
with no
+/// row content (id + vector values) and the aligned `__paimon_search_score`,
with no
/// `_ROW_ID`/`_PKEY_VECTOR_POSITION` leaking.
// Gated off Windows for the same `file://` tempdir reason as the test above.
#[cfg(not(windows))]
@@ -680,7 +680,7 @@ async fn
pk_vector_read_orders_rows_best_first_not_by_position() {
);
}
- // Score alignment: `_PKEY_VECTOR_SCORE` matches metric.distance_to_score
for
+ // Score alignment: `__paimon_search_score` matches
metric.distance_to_score for
// each emitted row, in best-first order.
assert_eq!(scores.len(), 3);
for (got, (_, distance)) in scores.iter().zip(&expected) {
@@ -712,7 +712,7 @@ async fn
pk_vector_read_orders_rows_best_first_not_by_position() {
/// Shared body for the two physical-coordinate contract tests below. Builds
the
/// discriminating fixture with the caller's `first_row_id`, reads it back
with the
/// default projection, and asserts the materialized rows are the file-LOCAL
-/// best-first top-k (id column, vector content, aligned `_PKEY_VECTOR_SCORE`),
+/// best-first top-k (id column, vector content, aligned
`__paimon_search_score`),
/// invariant to `first_row_id`. The discriminating fixture makes best-first
order
/// [5, 1, 3] distinct from ascending physical position [1, 3, 5], so a
/// position/global-id confusion cannot pass by coincidence.
@@ -862,7 +862,7 @@ async fn read_id_and_scores_filtered(
let scores: Vec<f32> = batches
.iter()
.flat_map(|b| {
- let idx = b.schema().index_of("_PKEY_VECTOR_SCORE").unwrap();
+ let idx = b.schema().index_of("__paimon_search_score").unwrap();
b.column(idx)
.as_any()
.downcast_ref::<Float32Array>()
@@ -954,7 +954,7 @@ async fn
pk_vector_residual_filter_excludes_non_matching_rows() {
);
// Search-and-read with the residual: default projection materializes id +
- // vector column, best-first, with an aligned `_PKEY_VECTOR_SCORE`.
+ // vector column, best-first, with an aligned `__paimon_search_score`.
let (ids, scores, batches) =
read_id_and_scores_filtered(&table, query.to_vec(), 3, residual).await;
diff --git a/crates/paimon/tests/pk_vector_java_fixture_test.rs
b/crates/paimon/tests/pk_vector_java_fixture_test.rs
index 20384ebd..579a58eb 100644
--- a/crates/paimon/tests/pk_vector_java_fixture_test.rs
+++ b/crates/paimon/tests/pk_vector_java_fixture_test.rs
@@ -200,7 +200,7 @@ async fn reads_back_java_written_pk_vector_table() {
// execute_read() materializes rows by physical position, so it MUST
succeed
// and emit the top-k best-first. The `id` column cross-checks the
- // position->id mapping (the fixture pins them equal) and
`_PKEY_VECTOR_SCORE`
+ // position->id mapping (the fixture pins them equal) and
`__paimon_search_score`
// carries the metric score.
let mut builder = table.new_vector_search_builder();
builder
@@ -222,7 +222,7 @@ async fn reads_back_java_written_pk_vector_table() {
"materialized `id` column must be best-first and match the analytic
top-k"
);
- let scores = batch_f32(&batches, "_PKEY_VECTOR_SCORE");
+ let scores = batch_f32(&batches, "__paimon_search_score");
assert_eq!(scores.len(), k);
for (got, want) in scores.iter().zip(&expected_scores) {
assert!(
