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 82650686 feat(table): align primary-key vector read to
physical-coordinate model (#544)
82650686 is described below
commit 826506861e8589a4501fd281377c318be238b7b5
Author: Junrui Lee <[email protected]>
AuthorDate: Mon Jul 20 11:21:56 2026 +0800
feat(table): align primary-key vector read to physical-coordinate model
(#544)
---
.licenserc.yaml | 1 +
crates/paimon/src/table/data_file_reader.rs | 230 +++++++++++++
crates/paimon/src/table/pk_vector_position_read.rs | 324 +++++++++---------
crates/paimon/src/table/vector_search_builder.rs | 374 +++++----------------
...-932a1249-f7e0-4a03-8e1f-ab8c85cbb76f-0.parquet | Bin 0 -> 1303 bytes
...-fd398030-457f-4444-8d71-fffc2753689d-0.parquet | Bin 0 -> 1303 bytes
.../index-c2314ed5-0000-4a0d-984f-2b894d6040e8-0 | Bin 0 -> 145 bytes
...manifest-7a487b9a-2053-423a-9379-77cb67346955-0 | Bin 0 -> 1422 bytes
...manifest-1328438e-8b23-4aef-a6ee-e2e2c29d90df-0 | Bin 0 -> 2096 bytes
...manifest-1328438e-8b23-4aef-a6ee-e2e2c29d90df-1 | Bin 0 -> 2139 bytes
...est-list-f5034e53-2d6f-40ae-9665-8778726154b4-0 | Bin 0 -> 1006 bytes
...est-list-f5034e53-2d6f-40ae-9665-8778726154b4-1 | Bin 0 -> 1110 bytes
...est-list-f5034e53-2d6f-40ae-9665-8778726154b4-2 | Bin 0 -> 1110 bytes
...est-list-f5034e53-2d6f-40ae-9665-8778726154b4-3 | Bin 0 -> 1109 bytes
.../pkvector/pk_vector_ivf_flat/schema/schema-0 | 32 ++
.../pkvector/pk_vector_ivf_flat/snapshot/EARLIEST | 1 +
.../pkvector/pk_vector_ivf_flat/snapshot/LATEST | 1 +
.../pk_vector_ivf_flat/snapshot/snapshot-1 | 16 +
.../pk_vector_ivf_flat/snapshot/snapshot-2 | 17 +
crates/paimon/tests/pk_vector_baseline_test.rs | 176 +++++++---
crates/paimon/tests/pk_vector_java_fixture_test.rs | 233 +++++++++++++
21 files changed, 906 insertions(+), 499 deletions(-)
diff --git a/.licenserc.yaml b/.licenserc.yaml
index 47a0ae30..ffbce011 100644
--- a/.licenserc.yaml
+++ b/.licenserc.yaml
@@ -26,6 +26,7 @@ header:
- ".gitattributes"
- ".github/PULL_REQUEST_TEMPLATE.md"
- "crates/paimon/tests/**/*.json"
+ - "crates/paimon/testdata/**"
- "**/go.sum"
- "**/DEPENDENCIES.*.tsv"
- ".devcontainer/devcontainer.json"
diff --git a/crates/paimon/src/table/data_file_reader.rs
b/crates/paimon/src/table/data_file_reader.rs
index f5b2647e..0faff2e9 100644
--- a/crates/paimon/src/table/data_file_reader.rs
+++ b/crates/paimon/src/table/data_file_reader.rs
@@ -419,6 +419,212 @@ impl DataFileReader {
}
.boxed())
}
+
+ /// Read one data file selecting rows by their file-LOCAL 0-based physical
+ /// positions. Unlike [`Self::read_single_file_stream`], the selection is
+ /// interpreted directly in file-local coordinates: it never consults
+ /// `first_row_id` (real primary-key tables never write one) and never
emits
+ /// `_ROW_ID`. A deletion vector, when present, is folded into the
selection so
+ /// any selected-but-deleted position is dropped; surviving rows are
returned in
+ /// ascending physical-position order. `local_positions` must be sorted
+ /// ascending, de-duplicated, and within `[0, file_meta.row_count)`.
+ ///
+ /// Used by `pk_vector_position_read` to recover rows by physical position
on
+ /// primary-key data files that carry no `first_row_id`.
+ pub(super) fn read_single_file_stream_local(
+ &self,
+ split: &DataSplit,
+ file_meta: DataFileMeta,
+ data_fields: Option<Vec<DataField>>,
+ dv: Option<Arc<DeletionVector>>,
+ local_positions: Vec<i64>,
+ ) -> crate::Result<ArrowRecordBatchStream> {
+ // Local-position selection is only sound against a predicate-free
reader: a
+ // row-filtering predicate drops arbitrary selected rows and desyncs
the
+ // caller's position/score cursor. Guard the invariant here (not only
at the
+ // PK-vector caller) so this `pub(super)` entry cannot be misused
within the
+ // module. This path never projects `_ROW_ID`, so a `_ROW_ID`-only
guard
+ // would be a no-op — check for any row-filtering predicate instead.
+ if self.has_row_filtering_predicate() {
+ return Err(crate::Error::DataInvalid {
+ message: "read_single_file_stream_local requires a
predicate-free reader: a \
+ row-filtering predicate would desync local-position selection"
+ .to_string(),
+ source: None,
+ });
+ }
+
+ let read_type = self.read_type.clone();
+ let table_fields = self.table_fields.clone();
+ let predicates = self.predicates.clone();
+ let file_io = self.file_io.clone();
+ let split = split.clone();
+ let blob_as_descriptor = self.blob_as_descriptor;
+
+ let target_schema = build_target_arrow_schema(&read_type)?;
+ let file_fields = data_fields.clone().unwrap_or_else(||
table_fields.clone());
+ let is_row_file = is_row_file(&file_meta);
+
+ // Compute index mapping and determine which columns to read from the
file.
+ let (projected_read_fields, index_mapping) = if let Some(ref df) =
data_fields {
+ let mapping = create_index_mapping(&read_type, df);
+ let fields_to_read = read_data_fields(df, &read_type)?;
+ (fields_to_read, mapping)
+ } else {
+ (
+ read_type
+ .iter()
+ .filter(|field| field.name() != ROW_ID_FIELD_NAME)
+ .cloned()
+ .collect(),
+ None,
+ )
+ };
+ let format_read_fields = if is_row_file {
+ file_fields.clone()
+ } else {
+ projected_read_fields
+ };
+
+ // Remap predicates from table-level to file-level indices.
+ let file_predicates = {
+ let remapped = crate::arrow::filtering::remap_predicates_to_file(
+ &predicates,
+ &table_fields,
+ &file_fields,
+ );
+ if remapped.is_empty() {
+ None
+ } else {
+ Some(crate::arrow::format::FilePredicates {
+ predicates: remapped,
+ file_fields: file_fields.clone(),
+ })
+ }
+ };
+
+ // Interpret `local_positions` directly as file-local ranges (no
+ // `to_local_row_ranges`, no `first_row_id`), then fold the DV in.
+ // `merge_row_selection` intersects the selection with the file's
+ // non-deleted ranges, so the reader emits exactly the selected,
non-deleted
+ // rows in ascending physical order.
+ let local_ranges =
coalesce_positions_to_local_ranges(&local_positions);
+ let row_selection =
+ merge_row_selection(file_meta.row_count, dv.as_deref(),
Some(&local_ranges));
+
+ Ok(try_stream! {
+ let path_to_read = split.data_file_path(&file_meta);
+ let format_reader =
+ create_format_reader(&path_to_read, blob_as_descriptor,
&format_read_fields)?;
+ let input_file = file_io.new_input(&path_to_read)?;
+ let file_reader = input_file.reader().await?;
+
+ let mut batch_stream = format_reader
+ .read_batch_stream(
+ Box::new(file_reader),
+ file_meta.file_size as u64,
+ &format_read_fields,
+ file_predicates.as_ref(),
+ None,
+ row_selection,
+ )
+ .await?;
+
+ while let Some(batch) = batch_stream.next().await {
+ let batch = batch?;
+ let result = project_file_batch(
+ &batch,
+ &target_schema,
+ index_mapping.as_deref(),
+ data_fields.as_deref(),
+ )?;
+ yield result;
+ }
+ }
+ .boxed())
+ }
+}
+
+/// Project one decoded file `batch` onto `target_schema`, resolving each
target
+/// column through the field-ID `index_mapping` (or by name when the file
schema
+/// matches the table schema), casting on type mismatch and null-filling absent
+/// columns. Unlike the inline projection in
+/// [`DataFileReader::read_single_file_stream`], this never materializes
`_ROW_ID`:
+/// the local-selection PK-vector read path never projects it.
+fn project_file_batch(
+ batch: &RecordBatch,
+ target_schema: &Arc<arrow_schema::Schema>,
+ index_mapping: Option<&[i32]>,
+ data_fields: Option<&[DataField]>,
+) -> crate::Result<RecordBatch> {
+ let num_rows = batch.num_rows();
+ let batch_schema = batch.schema();
+ let mut columns: Vec<Arc<dyn Array>> =
Vec::with_capacity(target_schema.fields().len());
+ for (i, target_field) in target_schema.fields().iter().enumerate() {
+ let source_col = if let Some(idx_map) = index_mapping {
+ let data_idx = idx_map[i];
+ if data_idx == NULL_FIELD_INDEX {
+ None
+ } else {
+ let data_field = &data_fields.unwrap()[data_idx as usize];
+ batch_schema
+ .index_of(data_field.name())
+ .ok()
+ .map(|col_idx| batch.column(col_idx))
+ }
+ } else if let Some(df) = data_fields {
+ batch_schema
+ .index_of(df[i].name())
+ .ok()
+ .map(|col_idx| batch.column(col_idx))
+ } else {
+ batch_schema
+ .index_of(target_field.name())
+ .ok()
+ .map(|col_idx| batch.column(col_idx))
+ };
+
+ match source_col {
+ Some(col) => {
+ if col.data_type() == target_field.data_type() {
+ columns.push(col.clone());
+ } else {
+ let casted = cast(col,
target_field.data_type()).map_err(|e| {
+ Error::UnexpectedError {
+ message: format!(
+ "Failed to cast column '{}' from {:?} to {:?}:
{e}",
+ target_field.name(),
+ col.data_type(),
+ target_field.data_type()
+ ),
+ source: Some(Box::new(e)),
+ }
+ })?;
+ columns.push(casted);
+ }
+ }
+ None => {
+ columns.push(arrow_array::new_null_array(
+ target_field.data_type(),
+ num_rows,
+ ));
+ }
+ }
+ }
+
+ if columns.is_empty() {
+ RecordBatch::try_new_with_options(
+ target_schema.clone(),
+ columns,
+
&arrow_array::RecordBatchOptions::new().with_row_count(Some(num_rows)),
+ )
+ } else {
+ RecordBatch::try_new(target_schema.clone(), columns)
+ }
+ .map_err(|e| Error::UnexpectedError {
+ message: format!("Failed to build schema-evolved RecordBatch: {e}"),
+ source: Some(Box::new(e)),
+ })
}
fn read_data_fields(
@@ -510,6 +716,30 @@ fn to_local_row_ranges(
.collect()
}
+/// Coalesce sorted, de-duplicated 0-based physical positions into contiguous
+/// file-LOCAL inclusive `RowRange`s. Unlike a global-range build, there is no
+/// `first_row_id` offset: the positions are already file-local coordinates.
+fn coalesce_positions_to_local_ranges(sorted_positions: &[i64]) ->
Vec<RowRange> {
+ let mut ranges = Vec::new();
+ let mut iter = sorted_positions.iter().copied();
+ let Some(first) = iter.next() else {
+ return ranges;
+ };
+ let mut start = first;
+ let mut end = first;
+ for pos in iter {
+ if end + 1 == pos {
+ end = pos;
+ } else {
+ ranges.push(RowRange::new(start, end));
+ start = pos;
+ end = pos;
+ }
+ }
+ ranges.push(RowRange::new(start, end));
+ ranges
+}
+
/// Merge DV and row_ranges into a unified list of 0-based inclusive RowRanges.
/// Returns `None` if no filtering is needed (no DV and no ranges).
///
diff --git a/crates/paimon/src/table/pk_vector_position_read.rs
b/crates/paimon/src/table/pk_vector_position_read.rs
index 9662c8b4..0c838979 100644
--- a/crates/paimon/src/table/pk_vector_position_read.rs
+++ b/crates/paimon/src/table/pk_vector_position_read.rs
@@ -32,12 +32,10 @@ use arrow_schema::{DataType as ArrowDataType, Field as
ArrowField, Fields, Schem
use futures::StreamExt;
use crate::deletion_vector::DeletionVector;
-use crate::spec::{
- BigIntType, DataField, DataFileMeta, DataType, ROW_ID_FIELD_ID,
ROW_ID_FIELD_NAME,
-};
+use crate::spec::{DataField, DataFileMeta, ROW_ID_FIELD_NAME};
use crate::table::data_file_reader::DataFileReader;
use crate::table::source::DataSplit;
-use crate::table::{ArrowRecordBatchStream, RowRange};
+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";
@@ -49,44 +47,6 @@ fn data_invalid(message: impl Into<String>) -> crate::Error {
}
}
-/// Coalesce sorted, de-duplicated 0-based physical positions into contiguous
-/// global `RowRange`s (each bound offset by `first_row_id`). The read path
-/// converts these back to local ranges via
`to_local_row_ranges(first_row_id)`,
-/// so this round-trips to exactly the requested physical positions.
-fn positions_to_global_ranges(
- sorted_positions: &[i64],
- first_row_id: i64,
-) -> crate::Result<Vec<RowRange>> {
- let mut ranges = Vec::new();
- let mut iter = sorted_positions.iter().copied();
- let Some(first) = iter.next() else {
- return Ok(ranges);
- };
- let mut push_range = |start: i64, end: i64| -> crate::Result<()> {
- let from = start
- .checked_add(first_row_id)
- .ok_or_else(|| data_invalid("vector position offset overflows
i64"))?;
- let to = end
- .checked_add(first_row_id)
- .ok_or_else(|| data_invalid("vector position offset overflows
i64"))?;
- ranges.push(RowRange::new(from, to));
- Ok(())
- };
- let mut start = first;
- let mut end = first;
- for pos in iter {
- if end.checked_add(1) == Some(pos) {
- end = pos;
- } else {
- push_range(start, end)?;
- start = pos;
- end = pos;
- }
- }
- push_range(start, end)?;
- Ok(ranges)
-}
-
/// Reads selected physical rows of one data file, appending position (+
optional
/// score) metadata columns. Rust equivalent of Java
/// `PrimaryKeyVectorPositionReader`.
@@ -168,81 +128,85 @@ impl<'a> PkVectorPositionRead<'a> {
));
}
- // (5) first_row_id guard
- let first_row_id = file_meta.first_row_id.ok_or_else(|| {
- data_invalid("PK vector position read requires a data file with
first_row_id")
- })?;
-
- // Build a reader whose read_type includes _ROW_ID so the lower-level
read
- // emits real global row-ids we can convert to physical positions. A
- // caller-requested _ROW_ID was already rejected in (3), so it is
always
- // absent here and appended fresh.
- let mut inner_read_type = self.reader.read_type().to_vec();
- inner_read_type.push(DataField::new(
- ROW_ID_FIELD_ID,
- ROW_ID_FIELD_NAME.to_string(),
- DataType::BigInt(BigIntType::new()),
- ));
- let inner_reader = self.reader.clone().with_read_type(inner_read_type);
+ // Effective selection: the requested positions minus any the deletion
+ // vector marks deleted, ascending. `read_single_file_stream_local`
folds
+ // the same DV into its row selection, so the reader returns exactly
these
+ // rows in this order; the cursor below maps each returned batch back
onto
+ // its slice of `effective` to recover file-LOCAL positions (no
+ // `first_row_id`, no `_ROW_ID` round-trip).
+ let effective: Vec<i64> = match dv.as_deref() {
+ Some(dv) => sorted
+ .iter()
+ .copied()
+ .filter(|&p| !dv.is_deleted(p as u64))
+ .collect(),
+ None => sorted.clone(),
+ };
- let ranges = positions_to_global_ranges(&sorted, first_row_id)?;
- let inner = inner_reader.read_single_file_stream(
- split,
- file_meta.clone(),
- data_fields,
- dv,
- Some(ranges),
- )?;
+ let inner =
+ self.reader
+ .read_single_file_stream_local(split, file_meta, data_fields,
dv, sorted)?;
let want_score_col = scores.is_some();
let stream = async_stream::try_stream! {
futures::pin_mut!(inner);
+ let mut cursor = 0usize;
while let Some(batch) = inner.next().await {
let batch = batch?;
+ let n = batch.num_rows();
+ let end = cursor + n;
+ if end > effective.len() {
+ let overflow: crate::Result<()> = Err(data_invalid(format!(
+ "PK vector position read returned {end} rows but only
{} positions were \
+ selected",
+ effective.len()
+ )));
+ overflow?;
+ }
let out = append_metadata_columns(
batch,
- first_row_id,
+ &effective[cursor..end],
want_score_col,
scores.as_ref(),
)?;
+ cursor = end;
yield out;
}
+ if cursor != effective.len() {
+ let mismatch: crate::Result<()> = Err(data_invalid(format!(
+ "PK vector position read returned {cursor} rows but {}
positions were selected",
+ effective.len()
+ )));
+ mismatch?;
+ }
};
Ok(Box::pin(stream))
}
}
-/// Split `batch` (which contains a `_ROW_ID` column) into an output batch with
-/// `_ROW_ID` removed and `_PKEY_VECTOR_POSITION` (+ optional
`_PKEY_VECTOR_SCORE`)
-/// appended. Positions are `_ROW_ID - first_row_id`; scores are looked up by
the
-/// returned position. Returns the new batch.
+/// Append `_PKEY_VECTOR_POSITION` (and, when `want_score_col`,
`_PKEY_VECTOR_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
+/// up by position. The batch carries no `_ROW_ID` column, so every existing
+/// column is retained.
fn append_metadata_columns(
batch: RecordBatch,
- first_row_id: i64,
+ positions: &[i64],
want_score_col: bool,
scores: Option<&BTreeMap<i64, f32>>,
) -> crate::Result<RecordBatch> {
+ debug_assert_eq!(
+ batch.num_rows(),
+ positions.len(),
+ "position slice must align with batch rows"
+ );
let schema = batch.schema();
- let row_id_idx = schema
- .index_of(ROW_ID_FIELD_NAME)
- .map_err(|_| data_invalid("internal: _ROW_ID column missing from
position read"))?;
- let row_ids = batch
- .column(row_id_idx)
- .as_any()
- .downcast_ref::<Int64Array>()
- .ok_or_else(|| data_invalid("internal: _ROW_ID column is not Int64"))?;
-
- let mut positions = Vec::with_capacity(row_ids.len());
- let mut score_vals = if want_score_col {
- Some(Vec::with_capacity(row_ids.len()))
- } else {
- None
- };
- for i in 0..row_ids.len() {
- let position = row_ids.value(i) - first_row_id;
- positions.push(position);
- if let Some(sv) = score_vals.as_mut() {
+
+ let score_vals = if want_score_col {
+ let mut sv = Vec::with_capacity(positions.len());
+ for &position in positions {
let score = scores
.and_then(|m| m.get(&position).copied())
.ok_or_else(|| {
@@ -252,24 +216,20 @@ fn append_metadata_columns(
})?;
sv.push(score);
}
- }
+ Some(sv)
+ } else {
+ None
+ };
- // Rebuild columns/fields excluding _ROW_ID, then append metadata columns.
- let mut fields: Vec<ArrowField> = Vec::new();
- let mut columns: Vec<Arc<dyn Array>> = Vec::new();
- for (i, f) in schema.fields().iter().enumerate() {
- if i == row_id_idx {
- continue;
- }
- fields.push(f.as_ref().clone());
- columns.push(batch.column(i).clone());
- }
+ // Retain every existing column (no `_ROW_ID` to strip), then append
metadata.
+ let mut fields: Vec<ArrowField> = schema.fields().iter().map(|f|
f.as_ref().clone()).collect();
+ let mut columns: Vec<Arc<dyn Array>> = batch.columns().to_vec();
fields.push(ArrowField::new(
PKEY_VECTOR_POSITION_COLUMN,
ArrowDataType::Int64,
false,
));
- columns.push(Arc::new(Int64Array::from(positions)));
+ columns.push(Arc::new(Int64Array::from(positions.to_vec())));
if let Some(sv) = score_vals {
fields.push(ArrowField::new(
PKEY_VECTOR_SCORE_COLUMN,
@@ -294,7 +254,8 @@ mod tests {
use crate::io::FileIOBuilder;
use crate::spec::stats::BinaryTableStats;
use crate::spec::{
- DataFileMeta, DataType, Datum, IntType, PredicateBuilder,
ROW_ID_FIELD_NAME,
+ BigIntType, DataFileMeta, DataType, Datum, IntType, PredicateBuilder,
ROW_ID_FIELD_ID,
+ ROW_ID_FIELD_NAME,
};
use crate::table::data_file_reader::DataFileReader;
use crate::table::schema_manager::SchemaManager;
@@ -454,8 +415,9 @@ mod tests {
}
/// Build a `DataFileReader` over an in-memory mosaic file plus the
matching
- /// `DataSplit`. `read_type`/`predicates` override the reader's projection
and
- /// filter; `deleted_rows`, when non-empty, writes a DV into the split.
+ /// `DataSplit`, pinning the data file's `first_row_id = Some(0)`.
+ /// `read_type`/`predicates` override the reader's projection and filter;
+ /// `deleted_rows`, when non-empty, writes a DV into the split.
async fn build_reader_and_split(
table_path: &str,
data: &Bytes,
@@ -463,6 +425,32 @@ mod tests {
read_type: Vec<DataField>,
predicates: Vec<crate::spec::Predicate>,
deleted_rows: &[u32],
+ ) -> (DataFileReader, DataSplit, Option<Arc<DeletionVector>>) {
+ build_reader_and_split_with_first_row_id(
+ table_path,
+ data,
+ row_count,
+ read_type,
+ predicates,
+ deleted_rows,
+ Some(0),
+ )
+ .await
+ }
+
+ /// As `build_reader_and_split`, but the caller controls the data file's
+ /// `first_row_id`. Real Java primary-key tables never write `first_row_id`
+ /// (row-tracking is forbidden for PK tables), so `None` is the shape the
read
+ /// path must handle by keying off file-local physical positions.
+ #[allow(clippy::too_many_arguments)]
+ async fn build_reader_and_split_with_first_row_id(
+ table_path: &str,
+ data: &Bytes,
+ row_count: i64,
+ read_type: Vec<DataField>,
+ predicates: Vec<crate::spec::Predicate>,
+ deleted_rows: &[u32],
+ first_row_id: Option<i64>,
) -> (DataFileReader, DataSplit, Option<Arc<DeletionVector>>) {
let file_io = FileIOBuilder::new("memory").build().unwrap();
let bucket_path = format!("{table_path}/bucket-0");
@@ -486,7 +474,7 @@ mod tests {
data.len() as i64,
row_count,
schema_id,
- Some(0),
+ first_row_id,
)]);
let mut dv = None;
if !deleted_rows.is_empty() {
@@ -772,6 +760,65 @@ mod tests {
);
}
+ #[tokio::test]
+ async fn
pk_position_read_aligns_positions_across_dv_and_batches_without_first_row_id() {
+ // The Java primary-key shape: the data file carries NO first_row_id,
rows
+ // span multiple row groups (batches), a DV deletes a NON-candidate
+ // position, and the candidates sit at non-contiguous local positions.
The
+ // read must recover each row's FILE-LOCAL position (not a global row
id),
+ // skip only the deleted row, and keep id/position/score aligned
best-first.
+ //
+ // Three row groups [10,11] [12,13] [14,15] -> reader yields >1 batch.
+ // Candidates [1,3,4,5] with scores keyed by local position; DV deletes
+ // position 2 (a NON-candidate) -> it must not perturb the surviving
rows.
+ // Expected surviving rows: ids [11,13,14,15], positions [1,3,4,5].
+ let data = write_mosaic_multi_group(&[
+ id_batch(vec![10, 11]),
+ id_batch(vec![12, 13]),
+ id_batch(vec![14, 15]),
+ ]);
+ let (reader, split, dv) = build_reader_and_split_with_first_row_id(
+ "memory:/pkvpr_local_dv_multibatch",
+ &data,
+ 6,
+ id_fields(),
+ Vec::new(),
+ &[2],
+ None,
+ )
+ .await;
+
+ let scores = BTreeMap::from([(1, 0.9f32), (3, 0.5), (4, 0.3), (5,
0.1)]);
+ let batches = PkVectorPositionRead::new(&reader)
+ .read(
+ &split,
+ split.data_files()[0].clone(),
+ None,
+ dv,
+ vec![1, 3, 4, 5],
+ Some(scores),
+ )
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+
+ assert!(
+ batches.len() > 1,
+ "expected multiple batches, got {}",
+ batches.len()
+ );
+ assert_eq!(collect_i32(&batches, "id"), vec![11, 13, 14, 15]);
+ assert_eq!(
+ collect_i64(&batches, PKEY_VECTOR_POSITION_COLUMN),
+ vec![1, 3, 4, 5]
+ );
+ assert_eq!(
+ collect_f32(&batches, PKEY_VECTOR_SCORE_COLUMN),
+ vec![0.9, 0.5, 0.3, 0.1]
+ );
+ }
+
#[tokio::test]
async fn test_empty_positions_is_error() {
let data = write_mosaic_single_group(&id_batch(vec![10, 11, 12]));
@@ -877,44 +924,6 @@ mod tests {
assert!(format!("{err:?}").contains("Scores keys"), "got: {err:?}");
}
- #[tokio::test]
- async fn test_missing_first_row_id_is_error() {
- // first_row_id = None -> Err mentioning first_row_id. Positions must
be
- // valid so validation reaches the first_row_id guard.
- let file_io = FileIOBuilder::new("memory").build().unwrap();
- let schema_manager = SchemaManager::new(file_io.clone(),
"memory:/pkvpr_frid".to_string());
- let reader = DataFileReader::new(
- file_io,
- schema_manager,
- 1,
- id_fields(),
- id_fields(),
- Vec::new(),
- );
- let split = DataSplitBuilder::new()
- .with_snapshot(1)
- .with_partition(crate::spec::BinaryRow::new(0))
- .with_bucket(0)
- .with_bucket_path("memory:/pkvpr_frid/bucket-0".to_string())
- .with_total_buckets(1)
- .with_data_files(vec![data_file("part-0.mosaic", 1, 5, 1, None)])
- .build()
- .unwrap();
-
- let err = PkVectorPositionRead::new(&reader)
- .read(
- &split,
- split.data_files()[0].clone(),
- None,
- None,
- vec![0],
- None,
- )
- .err()
- .expect("missing first_row_id must be an error");
- assert!(format!("{err:?}").contains("first_row_id"), "got: {err:?}");
- }
-
#[tokio::test]
async fn test_predicate_reader_is_rejected() {
// A row-filtering predicate on the reader must be rejected.
@@ -1047,29 +1056,4 @@ mod tests {
.expect("requested _ROW_ID must be an error");
assert!(format!("{err:?}").contains("_ROW_ID"), "got: {err:?}");
}
-
- #[test]
- fn test_positions_to_global_ranges_coalesces_and_offsets() {
- // positions [0,1,2,4,5] with first_row_id 100 -> global ranges
- // [100..=102] and [104..=105] (two coalesced ranges, offset by 100).
- let ranges = positions_to_global_ranges(&[0, 1, 2, 4, 5],
100).unwrap();
- assert_eq!(ranges.len(), 2);
- assert_eq!((ranges[0].from(), ranges[0].to()), (100, 102));
- assert_eq!((ranges[1].from(), ranges[1].to()), (104, 105));
- }
-
- #[test]
- fn test_positions_to_global_ranges_single() {
- let ranges = positions_to_global_ranges(&[3], 0).unwrap();
- assert_eq!(ranges.len(), 1);
- assert_eq!((ranges[0].from(), ranges[0].to()), (3, 3));
- }
-
- #[test]
- fn test_positions_to_global_ranges_overflow_is_error() {
- // position 1 offset by i64::MAX overflows -> Err.
- let err = positions_to_global_ranges(&[1], i64::MAX)
- .expect_err("offset overflow must be an error");
- assert!(format!("{err:?}").contains("overflows"), "got: {err:?}");
- }
}
diff --git a/crates/paimon/src/table/vector_search_builder.rs
b/crates/paimon/src/table/vector_search_builder.rs
index 2ce2f8df..57cde2be 100644
--- a/crates/paimon/src/table/vector_search_builder.rs
+++ b/crates/paimon/src/table/vector_search_builder.rs
@@ -16,13 +16,13 @@
// under the License.
use crate::arrow::format::FilePredicates;
-use crate::arrow::residual::{filter_record_batch_by_predicates,
widen_scan_fields};
+use crate::arrow::residual::{evaluate_predicates_mask, widen_scan_fields};
use crate::io::FileIO;
use crate::lumina::reader::LuminaVectorGlobalIndexReader;
use crate::lumina::{is_lumina_index_type, LuminaIndexMeta, LuminaVectorMetric};
use crate::spec::{
- BigIntType, CoreOptions, DataField, DataType, FileKind,
GlobalIndexSearchMode, IndexFileMeta,
- IndexManifest, IndexManifestEntry, Predicate, ROW_ID_FIELD_ID,
ROW_ID_FIELD_NAME,
+ CoreOptions, DataField, FileKind, GlobalIndexSearchMode, IndexFileMeta,
IndexManifest,
+ IndexManifestEntry, Predicate, ROW_ID_FIELD_NAME,
};
use crate::table::data_file_reader::DataFileReader;
use crate::table::global_index_scanner::{
@@ -32,8 +32,8 @@ use crate::table::global_index_scanner::{
use crate::table::pk_vector_data_file_reader::DataFilePkVectorReaderFactory;
use crate::table::pk_vector_indexed_split_read::PkVectorIndexedSplitRead;
use crate::table::pk_vector_orchestrator::{
- as_split_exact_reader_factory, build_indexed_splits,
validate_row_position, PkVectorCandidate,
- PkVectorOrchestrator, PkVectorSearchSplit,
+ 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,
@@ -196,22 +196,21 @@ impl<'a> VectorSearchBuilder<'a> {
// targets a configured PK-vector column; otherwise fall through to the
// data-evolution (DE) global-index path below.
//
- // Membership is resolved first via the non-erroring columns accessor
so a
- // malformed PK-vector config (e.g. more than one column, or a blank
list)
- // cannot abort an unrelated DE query. The exactly-one-column rule is
- // enforced only once this query is known to target a PK-vector column,
- // keeping fail-loud behavior for a genuinely-broken config on the path
- // where erroring is correct.
+ // Membership is resolved via the non-erroring columns accessor so a
+ // malformed PK-vector config (e.g. a blank list) cannot abort an
unrelated
+ // DE query. A query that does target the PK-vector column fails loud
here:
+ // the PK path produces physical positions, not global row ids, so
scored
+ // search is unsupported and callers must use `execute_read` instead.
if core.primary_key_vector_index_enabled() {
let targets_pk_column = core
.primary_key_vector_index_columns()
.ok()
.is_some_and(|cols| cols.iter().any(|c| c == vector_column));
if targets_pk_column {
- let pk_col = core.primary_key_vector_index_column()?;
- return self
- .execute_primary_key_vector_search(&core, &pk_col,
query_vector, limit)
- .await;
+ return Err(crate::Error::DataInvalid {
+ message: "primary-key vector search does not produce
global row ids; use the materialized read (execute_read) instead".to_string(),
+ source: None,
+ });
}
}
@@ -290,23 +289,6 @@ impl<'a> VectorSearchBuilder<'a> {
})
}
- /// Run the primary-key bucket-local vector search: plan the per-bucket
splits,
- /// build the real vindex ANN scorer and (outside FAST mode) the
exact-fallback
- /// readers, run the orchestrator, and convert the best-first candidates
into a
- /// `SearchResult`. Mirrors Java `PrimaryKeyVectorRead`.
- async fn execute_primary_key_vector_search(
- &self,
- core: &CoreOptions<'_>,
- pk_col: &str,
- query_vector: &[f32],
- limit: usize,
- ) -> crate::Result<SearchResult> {
- let (candidates, plan, metric) = self
- .plan_and_search_pk_candidates(core, pk_col, query_vector, limit)
- .await?;
- candidates_to_search_result(&candidates, &plan.splits, metric)
- }
-
/// Shared PK-vector search core for both the search-only and
search-and-read
/// paths: plan the per-bucket splits, verify the configured metric
against each
/// ANN segment, build the real vindex ANN scorer and (outside FAST mode)
the
@@ -422,9 +404,10 @@ impl<'a> VectorSearchBuilder<'a> {
// per-split allow-list is threaded into the bucket search so the
residual
// folds into recall (best-first order and Top-K are preserved). Built
only
// when a filter is set; otherwise `None` leaves the search
unfiltered. The
- // residual reader projects the predicate columns plus `_ROW_ID` (used
to
- // recover file-local physical positions) and carries no pushdown,
matching
- // `residual_positions_by_file`. A file the allow-list leaves empty is
+ // residual reader projects only the predicate columns and carries no
+ // pushdown; `residual_positions_by_file` recovers each surviving row's
+ // file-local physical position from its ordinal in the unfiltered
scan (no
+ // `_ROW_ID`, no `first_row_id`). A file the allow-list leaves empty is
// skipped by the bucket search without opening an exact reader.
let residual_by_split: Option<Vec<HashMap<String, RoaringTreemap>>> =
match &self.filter {
Some(filter) => {
@@ -432,13 +415,7 @@ impl<'a> VectorSearchBuilder<'a> {
predicates: vec![filter.clone()],
file_fields: self.table.schema().fields().to_vec(),
};
- let row_id_field = DataField::new(
- ROW_ID_FIELD_ID,
- ROW_ID_FIELD_NAME.to_string(),
- DataType::BigInt(BigIntType::new()),
- );
- let residual_read_type =
- widen_scan_fields(std::slice::from_ref(&row_id_field),
Some(&file_predicates));
+ let residual_read_type = widen_scan_fields(&[],
Some(&file_predicates));
let residual_reader = DataFileReader::new(
self.table.file_io().clone(),
self.table.schema_manager().clone(),
@@ -1003,29 +980,27 @@ fn is_vector_global_index_file(index_file:
&IndexFileMeta) -> bool {
VectorIndexBackend::from_index_type(&index_file.index_type).is_some()
}
-/// Compute, per data file in `split`, the set of physical row positions whose
-/// rows satisfy the residual predicate. Mirrors the row-collecting half of
Java
-/// `PrimaryKeyVectorRead`'s `executeFilter`: because
-/// [`DataFileReader::read_single_file_stream`] rejects projecting `_ROW_ID`
-/// alongside a row-filtering predicate (the residual filter would drop rows
-/// before `_ROW_ID` is assigned positionally, desyncing it), the predicate is
-/// NOT pushed down. Instead `reader` projects the residual columns together
with
-/// `_ROW_ID` and carries no pushdown predicate; the residual is applied here
at
-/// the Arrow level, after `_ROW_ID` is materialized, and each surviving row's
-/// `_ROW_ID - first_row_id` is the file-local physical position.
+/// Compute, per data file in `split`, the set of file-LOCAL physical row
+/// positions whose rows satisfy the residual predicate. Mirrors the
+/// row-collecting half of Java `PrimaryKeyVectorRead`'s `executeFilter`: the
+/// predicate is NOT pushed down (a pushed filter would drop rows before their
+/// position could be recovered). Instead `reader` projects only the residual
+/// columns and carries no pushdown predicate; every physical row is scanned in
+/// file order, the residual is evaluated here at the Arrow level, and each
+/// surviving row's file-local 0-based position is its running ordinal in the
scan.
+/// This needs no `_ROW_ID` and no `first_row_id` — real primary-key tables
never
+/// write one.
///
/// Every *active* data file in the split gets an entry, possibly empty. The
/// bucket search treats an absent entry and an empty entry identically (the
file
/// contributes no candidates), so the empty entries only make the map cover
every
/// active file. Non-active files (e.g. level-0 files the bucket search
excludes)
/// are skipped entirely: they are never searched, so re-reading them would be
-/// wasted IO and their possibly-absent `first_row_id` must not fail an
otherwise
-/// valid query.
+/// wasted IO.
///
-/// `reader` must project `_ROW_ID` and be predicate-free;
`residual.file_fields`
-/// are the fields the residual leaf indices point into (resolved by name
against
-/// each emitted batch). A data file without `first_row_id` fails loud,
matching
-/// the position-read guard.
+/// `reader` must be predicate-free and project the residual columns;
+/// `residual.file_fields` are the fields the residual leaf indices point into
+/// (resolved by name against each emitted batch).
async fn residual_positions_by_file(
reader: &DataFileReader,
split: &DataSplit,
@@ -1037,57 +1012,46 @@ async fn residual_positions_by_file(
let mut out: HashMap<String, RoaringTreemap> = HashMap::new();
for file_meta in split.data_files() {
// Only files the bucket search actually recalls from need residual
- // positions; skip everything else so a non-active file cannot trigger
the
- // `first_row_id` guard below or incur a wasted read.
+ // positions; skip everything else to avoid a wasted read.
if !active_names.contains(file_meta.file_name.as_str()) {
continue;
}
- let first_row_id = file_meta
- .first_row_id
- .ok_or_else(|| crate::Error::DataInvalid {
- message: format!(
- "residual position read requires data file '{}' to have
first_row_id",
- file_meta.file_name
- ),
- source: None,
- })?;
let data_fields = reader.derive_data_fields(file_meta).await?;
let mut stream =
reader.read_single_file_stream(split, file_meta.clone(),
data_fields, None, None)?;
// Register the file up front so a file whose rows all fail the
residual
// still appears in the map (empty set).
let positions = out.entry(file_meta.file_name.clone()).or_default();
+ // The scan has no row selection and no DV, so rows arrive in physical
file
+ // order with no gaps: each row's file-local 0-based position is its
running
+ // ordinal `base + row_index`.
+ let mut base: u64 = 0;
while let Some(batch) = stream.try_next().await? {
- let filtered = filter_record_batch_by_predicates(batch, residual,
&scan_fields)?;
- if filtered.num_rows() == 0 {
- continue;
- }
- let row_id_idx =
filtered.schema().index_of(ROW_ID_FIELD_NAME).map_err(|_| {
- crate::Error::DataInvalid {
- message: "residual position read batch is missing the
_ROW_ID column"
- .to_string(),
- source: None,
+ let num_rows = batch.num_rows();
+ let mask = evaluate_predicates_mask(
+ &batch,
+ &residual.predicates,
+ &residual.file_fields,
+ &scan_fields,
+ )?;
+ match mask {
+ Some(mask) => {
+ for row_index in 0..num_rows {
+ // NULL follows the same NULL -> false convention the
Arrow
+ // filter kernel applies, so a null mask slot drops
the row.
+ if mask.is_valid(row_index) && mask.value(row_index) {
+ positions.insert(base + row_index as u64);
+ }
+ }
+ }
+ // No predicate contributed a mask (identity) -> keep every
row.
+ None => {
+ for row_index in 0..num_rows {
+ positions.insert(base + row_index as u64);
+ }
}
- })?;
- let row_ids = filtered
- .column(row_id_idx)
- .as_any()
- .downcast_ref::<Int64Array>()
- .ok_or_else(|| crate::Error::DataInvalid {
- message: "residual position read _ROW_ID column is not
Int64".to_string(),
- source: None,
- })?;
- for i in 0..row_ids.len() {
- let position = row_ids.value(i) - first_row_id;
- let position = u64::try_from(position).map_err(|_|
crate::Error::DataInvalid {
- message: format!(
- "residual position {position} is negative for data
file '{}'",
- file_meta.file_name
- ),
- source: None,
- })?;
- positions.insert(position);
}
+ base += num_rows as u64;
}
}
Ok(out)
@@ -1167,62 +1131,6 @@ fn verify_pk_vector_segment_metrics(
Ok(())
}
-/// candidate order (no re-sort). Each candidate's global row id is
-/// `first_row_id + row_position` of the data file it references; the score is
-/// derived from the raw distance via the metric. A candidate referencing a
file
-/// absent from its split, or a file with no `first_row_id`, fails loud.
-fn candidates_to_search_result(
- candidates: &[PkVectorCandidate],
- splits: &[PkVectorSearchSplit],
- metric: VectorSearchMetric,
-) -> crate::Result<SearchResult> {
- let mut row_ids = Vec::with_capacity(candidates.len());
- let mut scores = Vec::with_capacity(candidates.len());
- for c in candidates {
- let split = splits
- .get(c.split_index)
- .ok_or_else(|| crate::Error::DataInvalid {
- message: format!("candidate split_index {} out of range",
c.split_index),
- source: None,
- })?;
- let file_meta = split
- .data_split
- .data_files()
- .iter()
- .find(|f| f.file_name == c.data_file_name)
- .ok_or_else(|| crate::Error::DataInvalid {
- message: format!(
- "candidate references data file {} not present in its
split",
- c.data_file_name
- ),
- source: None,
- })?;
- let first_row_id = file_meta
- .first_row_id
- .ok_or_else(|| crate::Error::DataInvalid {
- message: format!("data file {} has no first_row_id",
c.data_file_name),
- source: None,
- })?;
- validate_row_position(&c.data_file_name, c.row_position,
file_meta.row_count)?;
- let global =
- first_row_id
- .checked_add(c.row_position)
- .ok_or_else(|| crate::Error::DataInvalid {
- message: "global row id overflows i64".to_string(),
- source: None,
- })?;
- row_ids.push(
- u64::try_from(global).map_err(|_| crate::Error::DataInvalid {
- message: format!("negative global row id {global}"),
- source: None,
- })?,
- );
- scores.push(metric.distance_to_score(c.distance));
- }
- // Order preserved: best-first, as produced by the orchestrator.
- Ok(SearchResult::new(row_ids, scores))
-}
-
/// One materialized row tagged with its best-first `rank` and its
`(batch_index,
/// row_index)` location in the retained materialization batches.
struct RankedRow {
@@ -2733,75 +2641,6 @@ mod tests {
}
}
- fn pk_candidate(
- split_index: usize,
- bucket: i32,
- file: &str,
- pos: i64,
- distance: f32,
- ) -> PkVectorCandidate {
- PkVectorCandidate {
- split_index,
- partition: BinaryRow::new(0),
- bucket,
- data_file_name: file.to_string(),
- row_position: pos,
- distance,
- }
- }
-
- #[test]
- fn candidates_to_search_result_global_row_id_and_best_first_order() {
- // Two files in one split with different first_row_id. The helper is a
pure
- // order-preserving map: the orchestrator already established
best-first
- // order upstream, so the candidate INPUT order here is deliberately
NOT in
- // score order and NOT in (file, position) order. This proves the
helper
- // preserves the given sequence rather than sorting.
- let splits = vec![pk_search_split(
- 0,
- vec![
- pk_data_file("file-a", 100, Some(1000)),
- pk_data_file("file-b", 100, Some(5000)),
- ],
- )];
- // Input sequence (NOT sorted by score, NOT sorted by file/position):
- // c0: file-b pos5 d=2.0 -> WORST distance, appears FIRST
- // c1: file-b pos1 d=1.0 -> tie with c2
- // c2: file-a pos2 d=1.0 -> tie with c1
- // A score-based best-first re-sort would produce [c1, c2, c0] (worst
last);
- // a (file, position) re-sort would produce [c2 (file-a), c1, c0]. Both
- // differ from the input order, so the exact assertion below
discriminates.
- let candidates = vec![
- pk_candidate(0, 0, "file-b", 5, 2.0),
- pk_candidate(0, 0, "file-b", 1, 1.0),
- pk_candidate(0, 0, "file-a", 2, 1.0),
- ];
- let result = candidates_to_search_result(&candidates, &splits,
VectorSearchMetric::L2)
- .expect("conversion succeeds");
- // global_row_id = first_row_id + position; INPUT order preserved (not
sorted).
- assert_eq!(result.row_ids, vec![5005, 5001, 1002]);
- assert_eq!(
- result.scores,
- vec![
- VectorSearchMetric::L2.distance_to_score(2.0),
- VectorSearchMetric::L2.distance_to_score(1.0),
- VectorSearchMetric::L2.distance_to_score(1.0),
- ]
- );
- }
-
- #[test]
- fn candidates_to_search_result_absent_first_row_id_fails_loud() {
- let splits = vec![pk_search_split(0, vec![pk_data_file("file-a", 100,
None)])];
- let candidates = vec![pk_candidate(0, 0, "file-a", 0, 1.0)];
- let err = candidates_to_search_result(&candidates, &splits,
VectorSearchMetric::L2)
- .expect_err("absent first_row_id must fail loud");
- assert!(
- matches!(err, crate::Error::DataInvalid { ref message, .. } if
message.contains("first_row_id")),
- "unexpected error: {err:?}"
- );
- }
-
/// Build a real vindex IVF-flat segment trained with `metric`, returning
the
/// serialized bytes. `nlist = 1` keeps training trivial and
deterministic; the
/// only thing the metric check cares about is the persisted metadata
metric.
@@ -2873,18 +2712,6 @@ mod tests {
);
}
- #[test]
- fn candidates_to_search_result_missing_file_fails_loud() {
- let splits = vec![pk_search_split(
- 0,
- vec![pk_data_file("known", 100, Some(0))],
- )];
- let candidates = vec![pk_candidate(0, 0, "unknown", 0, 1.0)];
- let err = candidates_to_search_result(&candidates, &splits,
VectorSearchMetric::L2)
- .expect_err("missing file must fail loud");
- assert!(matches!(err, crate::Error::DataInvalid { .. }));
- }
-
fn pk_vector_table(options: &[(&str, &str)]) -> Table {
let mut builder = Schema::builder()
.column("id", DataType::Int(IntType::new()))
@@ -2923,22 +2750,29 @@ mod tests {
}
#[tokio::test]
- async fn pk_branch_enabled_empty_plan_returns_empty() {
- // pk-vector.index.columns set, but no snapshot -> empty plan -> empty
result.
+ async fn pk_branch_execute_scored_fails_loud() {
+ // On a PK-vector table `execute_scored` reports global row ids, which
the
+ // PK path cannot produce (physical (file, position) coords, no global
ids).
+ // It must fail loud rather than fabricate ids; callers use
`execute_read`.
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 result = table
+ let err = table
.new_vector_search_builder()
.with_vector_column("embedding")
.with_query_vector(vec![1.0])
.with_limit(5)
.execute_scored()
.await
- .unwrap();
- assert!(result.is_empty());
+ .map(|_| ())
+ .expect_err("execute_scored on a PK-vector column must fail loud");
+ assert!(
+ matches!(err, crate::Error::DataInvalid { ref message, .. }
+ if message.contains("does not produce global row ids")),
+ "unexpected error: {err:?}"
+ );
}
#[tokio::test]
@@ -3006,34 +2840,6 @@ mod tests {
.unwrap()
}
- #[tokio::test]
- async fn pk_branch_filter_without_deletion_vectors_fails_loud() {
- // A residual filter on a PK-vector table that does NOT enable deletion
- // vectors must be rejected (merge-on-read semantics would make
physical
- // -position filtering unsound). Mirrors Java `PrimaryKeyVectorScan`.
- 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 filter = id_gt_filter(&table, 2);
- let err = table
- .new_vector_search_builder()
- .with_vector_column("embedding")
- .with_query_vector(vec![1.0])
- .with_limit(5)
- .with_filter(filter)
- .execute_scored()
- .await
- .map(|_| ())
- .expect_err("filter without deletion vectors must fail loud");
- assert!(
- matches!(err, crate::Error::DataInvalid { ref message, .. }
- if message.contains("deletion vectors without merge-on-read")),
- "unexpected error: {err:?}"
- );
- }
-
#[tokio::test]
async fn execute_read_filter_without_deletion_vectors_fails_loud() {
let table = pk_vector_table(&[
@@ -3085,7 +2891,7 @@ mod tests {
}
#[tokio::test]
- async fn pk_branch_filter_with_merge_on_read_fails_loud() {
+ async fn execute_read_filter_with_merge_on_read_fails_loud() {
// Deletion vectors enabled BUT merge-on-read on: still rejected,
because a
// merge-on-read scan can surface stale key versions that a
physical-row
// filter cannot reconcile.
@@ -3103,7 +2909,7 @@ mod tests {
.with_query_vector(vec![1.0])
.with_limit(5)
.with_filter(filter)
- .execute_scored()
+ .execute_read()
.await
.map(|_| ())
.expect_err("merge-on-read filter must fail loud");
@@ -3115,10 +2921,10 @@ mod tests {
}
#[tokio::test]
- async fn pk_branch_filter_with_deletion_vectors_passes_guard() {
+ async fn execute_read_filter_with_deletion_vectors_passes_guard() {
// Deletion vectors enabled, merge-on-read off (default): the residual
guard
// passes. With no snapshot the plan is empty, so the (guarded) filter
path
- // simply yields an empty result rather than erroring — proving the
guard
+ // simply yields an empty stream rather than erroring — proving the
guard
// admits a legal filtered query.
let table = pk_vector_table(&[
("pk-vector.index.columns", "embedding"),
@@ -3127,16 +2933,16 @@ mod tests {
("deletion-vectors.enabled", "true"),
]);
let filter = id_gt_filter(&table, 2);
- let result = table
+ let mut stream = table
.new_vector_search_builder()
.with_vector_column("embedding")
.with_query_vector(vec![1.0])
.with_limit(5)
.with_filter(filter)
- .execute_scored()
+ .execute_read()
.await
.expect("guarded filter query must be admitted");
- assert!(result.is_empty());
+ assert!(stream.try_next().await.unwrap().is_none());
}
fn make_lumina_entry(
@@ -3438,9 +3244,9 @@ mod tests {
}
/// Tests for [`residual_positions_by_file`]: the residual predicate is
applied at
-/// the Arrow level (no pushdown) against the predicate columns plus `_ROW_ID`,
-/// and each surviving row's `_ROW_ID` is converted back to a file-local
physical
-/// position.
+/// the Arrow level (no pushdown) against the predicate columns, and each
surviving
+/// row's file-local physical position is recovered from its ordinal in the
+/// unfiltered scan (no `_ROW_ID`, no `first_row_id`).
#[cfg(test)]
mod residual_positions_tests {
use super::*;
@@ -3712,12 +3518,15 @@ mod residual_positions_tests {
}
#[tokio::test]
- async fn test_missing_first_row_id_is_error() {
+ async fn test_missing_first_row_id_recovers_local_positions() {
+ // Real primary-key data files carry no `first_row_id`. Positions are
+ // recovered from each row's ordinal in the scan, so the residual still
+ // works: ids [1,2,3] with id > 0 -> all match -> local positions
[0,1,2].
let (reader, split, active) =
build_reader_and_split_no_first_row_id().await;
- let err = residual_positions_by_file(&reader, &split, &active,
&residual_id_gt(0))
+ let map = residual_positions_by_file(&reader, &split, &active,
&residual_id_gt(0))
.await
- .expect_err("missing first_row_id must error");
- assert!(format!("{err:?}").contains("first_row_id"), "got: {err:?}");
+ .expect("missing first_row_id must not fail the residual read");
+ assert_eq!(sorted(&map["part-0.mosaic"]), vec![0, 1, 2]);
}
async fn build_reader_and_split_no_first_row_id(
@@ -3749,7 +3558,8 @@ mod residual_positions_tests {
vec![id_field(), row_id_field()],
Vec::new(),
);
- // The lone file is active, so the `first_row_id` guard applies to it.
+ // The lone file is active and carries no first_row_id, exercising the
+ // ordinal-based position recovery.
let active = vec![BucketActiveFile {
file_name: "part-0.mosaic".to_string(),
row_count: 3,
diff --git
a/crates/paimon/testdata/pkvector/pk_vector_ivf_flat/bucket-0/data-932a1249-f7e0-4a03-8e1f-ab8c85cbb76f-0.parquet
b/crates/paimon/testdata/pkvector/pk_vector_ivf_flat/bucket-0/data-932a1249-f7e0-4a03-8e1f-ab8c85cbb76f-0.parquet
new file mode 100644
index 00000000..5184c7a9
Binary files /dev/null and
b/crates/paimon/testdata/pkvector/pk_vector_ivf_flat/bucket-0/data-932a1249-f7e0-4a03-8e1f-ab8c85cbb76f-0.parquet
differ
diff --git
a/crates/paimon/testdata/pkvector/pk_vector_ivf_flat/bucket-0/data-fd398030-457f-4444-8d71-fffc2753689d-0.parquet
b/crates/paimon/testdata/pkvector/pk_vector_ivf_flat/bucket-0/data-fd398030-457f-4444-8d71-fffc2753689d-0.parquet
new file mode 100644
index 00000000..5184c7a9
Binary files /dev/null and
b/crates/paimon/testdata/pkvector/pk_vector_ivf_flat/bucket-0/data-fd398030-457f-4444-8d71-fffc2753689d-0.parquet
differ
diff --git
a/crates/paimon/testdata/pkvector/pk_vector_ivf_flat/index/index-c2314ed5-0000-4a0d-984f-2b894d6040e8-0
b/crates/paimon/testdata/pkvector/pk_vector_ivf_flat/index/index-c2314ed5-0000-4a0d-984f-2b894d6040e8-0
new file mode 100644
index 00000000..5ebc99c7
Binary files /dev/null and
b/crates/paimon/testdata/pkvector/pk_vector_ivf_flat/index/index-c2314ed5-0000-4a0d-984f-2b894d6040e8-0
differ
diff --git
a/crates/paimon/testdata/pkvector/pk_vector_ivf_flat/manifest/index-manifest-7a487b9a-2053-423a-9379-77cb67346955-0
b/crates/paimon/testdata/pkvector/pk_vector_ivf_flat/manifest/index-manifest-7a487b9a-2053-423a-9379-77cb67346955-0
new file mode 100644
index 00000000..43190db9
Binary files /dev/null and
b/crates/paimon/testdata/pkvector/pk_vector_ivf_flat/manifest/index-manifest-7a487b9a-2053-423a-9379-77cb67346955-0
differ
diff --git
a/crates/paimon/testdata/pkvector/pk_vector_ivf_flat/manifest/manifest-1328438e-8b23-4aef-a6ee-e2e2c29d90df-0
b/crates/paimon/testdata/pkvector/pk_vector_ivf_flat/manifest/manifest-1328438e-8b23-4aef-a6ee-e2e2c29d90df-0
new file mode 100644
index 00000000..28905ecd
Binary files /dev/null and
b/crates/paimon/testdata/pkvector/pk_vector_ivf_flat/manifest/manifest-1328438e-8b23-4aef-a6ee-e2e2c29d90df-0
differ
diff --git
a/crates/paimon/testdata/pkvector/pk_vector_ivf_flat/manifest/manifest-1328438e-8b23-4aef-a6ee-e2e2c29d90df-1
b/crates/paimon/testdata/pkvector/pk_vector_ivf_flat/manifest/manifest-1328438e-8b23-4aef-a6ee-e2e2c29d90df-1
new file mode 100644
index 00000000..bc47a20f
Binary files /dev/null and
b/crates/paimon/testdata/pkvector/pk_vector_ivf_flat/manifest/manifest-1328438e-8b23-4aef-a6ee-e2e2c29d90df-1
differ
diff --git
a/crates/paimon/testdata/pkvector/pk_vector_ivf_flat/manifest/manifest-list-f5034e53-2d6f-40ae-9665-8778726154b4-0
b/crates/paimon/testdata/pkvector/pk_vector_ivf_flat/manifest/manifest-list-f5034e53-2d6f-40ae-9665-8778726154b4-0
new file mode 100644
index 00000000..a4b65b09
Binary files /dev/null and
b/crates/paimon/testdata/pkvector/pk_vector_ivf_flat/manifest/manifest-list-f5034e53-2d6f-40ae-9665-8778726154b4-0
differ
diff --git
a/crates/paimon/testdata/pkvector/pk_vector_ivf_flat/manifest/manifest-list-f5034e53-2d6f-40ae-9665-8778726154b4-1
b/crates/paimon/testdata/pkvector/pk_vector_ivf_flat/manifest/manifest-list-f5034e53-2d6f-40ae-9665-8778726154b4-1
new file mode 100644
index 00000000..b693f695
Binary files /dev/null and
b/crates/paimon/testdata/pkvector/pk_vector_ivf_flat/manifest/manifest-list-f5034e53-2d6f-40ae-9665-8778726154b4-1
differ
diff --git
a/crates/paimon/testdata/pkvector/pk_vector_ivf_flat/manifest/manifest-list-f5034e53-2d6f-40ae-9665-8778726154b4-2
b/crates/paimon/testdata/pkvector/pk_vector_ivf_flat/manifest/manifest-list-f5034e53-2d6f-40ae-9665-8778726154b4-2
new file mode 100644
index 00000000..afb2da6d
Binary files /dev/null and
b/crates/paimon/testdata/pkvector/pk_vector_ivf_flat/manifest/manifest-list-f5034e53-2d6f-40ae-9665-8778726154b4-2
differ
diff --git
a/crates/paimon/testdata/pkvector/pk_vector_ivf_flat/manifest/manifest-list-f5034e53-2d6f-40ae-9665-8778726154b4-3
b/crates/paimon/testdata/pkvector/pk_vector_ivf_flat/manifest/manifest-list-f5034e53-2d6f-40ae-9665-8778726154b4-3
new file mode 100644
index 00000000..b15d4384
Binary files /dev/null and
b/crates/paimon/testdata/pkvector/pk_vector_ivf_flat/manifest/manifest-list-f5034e53-2d6f-40ae-9665-8778726154b4-3
differ
diff --git a/crates/paimon/testdata/pkvector/pk_vector_ivf_flat/schema/schema-0
b/crates/paimon/testdata/pkvector/pk_vector_ivf_flat/schema/schema-0
new file mode 100644
index 00000000..dfaa9639
--- /dev/null
+++ b/crates/paimon/testdata/pkvector/pk_vector_ivf_flat/schema/schema-0
@@ -0,0 +1,32 @@
+{
+ "version" : 3,
+ "id" : 0,
+ "fields" : [ {
+ "id" : 0,
+ "name" : "id",
+ "type" : "INT NOT NULL"
+ }, {
+ "id" : 1,
+ "name" : "embedding",
+ "type" : {
+ "type" : "VECTOR",
+ "element" : "FLOAT",
+ "length" : 2
+ }
+ } ],
+ "highestFieldId" : 1,
+ "partitionKeys" : [ ],
+ "primaryKeys" : [ "id" ],
+ "options" : {
+ "bucket" : "1",
+ "ivf-flat.dimension" : "2",
+ "merge-engine" : "deduplicate",
+ "fields.embedding.pk-vector.distance.metric" : "l2",
+ "ivf-flat.nlist" : "1",
+ "fields.embedding.pk-vector.index.type" : "ivf-flat",
+ "deletion-vectors.enabled" : "true",
+ "pk-vector.index.columns" : "embedding",
+ "ivf-flat.metric" : "l2"
+ },
+ "timeMillis" : 1784359714317
+}
\ No newline at end of file
diff --git
a/crates/paimon/testdata/pkvector/pk_vector_ivf_flat/snapshot/EARLIEST
b/crates/paimon/testdata/pkvector/pk_vector_ivf_flat/snapshot/EARLIEST
new file mode 100644
index 00000000..56a6051c
--- /dev/null
+++ b/crates/paimon/testdata/pkvector/pk_vector_ivf_flat/snapshot/EARLIEST
@@ -0,0 +1 @@
+1
\ No newline at end of file
diff --git a/crates/paimon/testdata/pkvector/pk_vector_ivf_flat/snapshot/LATEST
b/crates/paimon/testdata/pkvector/pk_vector_ivf_flat/snapshot/LATEST
new file mode 100644
index 00000000..d8263ee9
--- /dev/null
+++ b/crates/paimon/testdata/pkvector/pk_vector_ivf_flat/snapshot/LATEST
@@ -0,0 +1 @@
+2
\ No newline at end of file
diff --git
a/crates/paimon/testdata/pkvector/pk_vector_ivf_flat/snapshot/snapshot-1
b/crates/paimon/testdata/pkvector/pk_vector_ivf_flat/snapshot/snapshot-1
new file mode 100644
index 00000000..b4b66746
--- /dev/null
+++ b/crates/paimon/testdata/pkvector/pk_vector_ivf_flat/snapshot/snapshot-1
@@ -0,0 +1,16 @@
+{
+ "version" : 3,
+ "id" : 1,
+ "schemaId" : 0,
+ "baseManifestList" : "manifest-list-f5034e53-2d6f-40ae-9665-8778726154b4-0",
+ "baseManifestListSize" : 1006,
+ "deltaManifestList" : "manifest-list-f5034e53-2d6f-40ae-9665-8778726154b4-1",
+ "deltaManifestListSize" : 1110,
+ "commitUser" : "44bb4893-5f61-4e16-b206-c9223ca5ae83",
+ "commitIdentifier" : 9223372036854775807,
+ "commitKind" : "APPEND",
+ "timeMillis" : 1784359716816,
+ "totalRecordCount" : 5,
+ "deltaRecordCount" : 5,
+ "nextRowId" : 0
+}
\ No newline at end of file
diff --git
a/crates/paimon/testdata/pkvector/pk_vector_ivf_flat/snapshot/snapshot-2
b/crates/paimon/testdata/pkvector/pk_vector_ivf_flat/snapshot/snapshot-2
new file mode 100644
index 00000000..057eb6d6
--- /dev/null
+++ b/crates/paimon/testdata/pkvector/pk_vector_ivf_flat/snapshot/snapshot-2
@@ -0,0 +1,17 @@
+{
+ "version" : 3,
+ "id" : 2,
+ "schemaId" : 0,
+ "baseManifestList" : "manifest-list-f5034e53-2d6f-40ae-9665-8778726154b4-2",
+ "baseManifestListSize" : 1110,
+ "deltaManifestList" : "manifest-list-f5034e53-2d6f-40ae-9665-8778726154b4-3",
+ "deltaManifestListSize" : 1109,
+ "indexManifest" : "index-manifest-7a487b9a-2053-423a-9379-77cb67346955-0",
+ "commitUser" : "44bb4893-5f61-4e16-b206-c9223ca5ae83",
+ "commitIdentifier" : 9223372036854775807,
+ "commitKind" : "COMPACT",
+ "timeMillis" : 1784359716943,
+ "totalRecordCount" : 5,
+ "deltaRecordCount" : 0,
+ "nextRowId" : 0
+}
\ No newline at end of file
diff --git a/crates/paimon/tests/pk_vector_baseline_test.rs
b/crates/paimon/tests/pk_vector_baseline_test.rs
index 55b144a8..18629315 100644
--- a/crates/paimon/tests/pk_vector_baseline_test.rs
+++ b/crates/paimon/tests/pk_vector_baseline_test.rs
@@ -103,9 +103,10 @@ fn analytic_topk(query: &[f32], vectors: &[[f32; DIM]], k:
usize) -> Vec<(u64, f
scored
}
-/// Table options that route searches into the primary-key vector branch
-/// (`VectorSearchBuilder::execute_primary_key_vector_search`). Default search
-/// mode is FAST, so only the ANN segment is consulted (no exact fallback).
+/// Table options that route searches into the primary-key vector branch (the
+/// `VectorSearchBuilder` detects a primary-key table with a PK-vector index
and
+/// takes the bucket-local ANN path). Default search mode is FAST, so only the
ANN
+/// segment is consulted (no exact fallback).
///
/// `deletion-vectors.enabled = true` (and merge-on-read left at its default
/// `false`) is what makes the table expose physical rows directly, the
@@ -324,6 +325,23 @@ async fn build_table(
query: &[f32],
vectors: &[[f32; DIM]],
k: usize,
+) -> (tempfile::TempDir, Table) {
+ // Default fixture: `first_row_id = Some(0)`, so a global row id equals its
+ // physical position. Tests that must decouple the two use
+ // `build_table_with_first_row_id` directly.
+ build_table_with_first_row_id(query, vectors, k, Some(0)).await
+}
+
+/// As `build_table`, but pins the indexed data file's `first_row_id` to the
+/// caller's value. A non-zero (or absent) `first_row_id` breaks the "global
row
+/// id == physical position" coincidence, so the read path must key candidate
+/// selection, position recovery, and score alignment off the file-local
physical
+/// position rather than a global row id.
+async fn build_table_with_first_row_id(
+ query: &[f32],
+ vectors: &[[f32; DIM]],
+ k: usize,
+ first_row_id: Option<i64>,
) -> (tempfile::TempDir, Table) {
let tmp = tempfile::tempdir().expect("create temp dir");
let location = format!("file://{}", tmp.path().display());
@@ -368,12 +386,14 @@ async fn build_table(
// Constraint 1 (PrimaryKeyIndexSourcePolicy.shouldRead): only a compacted,
// non-level-0 file backs the PK-vector index. Clone the real meta and set
- // level > 0 + file_source == COMPACT (1). Pin first_row_id = 0 so the
global
- // row id equals the physical position.
+ // level > 0 + file_source == COMPACT (1). `first_row_id` is
caller-controlled:
+ // real Java PK tables never write it (row-tracking is forbidden), so a
correct
+ // read path must not depend on `first_row_id == 0` for physical-position
+ // recovery.
let indexed_meta = DataFileMeta {
level: 1,
file_source: Some(1),
- first_row_id: Some(0),
+ first_row_id,
..base_meta
};
@@ -564,31 +584,21 @@ async fn
pk_vector_end_to_end_returns_expected_row_ids_and_scores() {
let expected_row_ids: Vec<u64> = expected.iter().map(|(id, _)|
*id).collect();
let expected_scores: Vec<f32> = expected.iter().map(|(_, d)|
l2_score(*d)).collect();
- // Search path: execute_scored() -> row ids + scores.
- let result = table
+ // A primary-key vector table exposes no global row ids, so the search-only
+ // `execute_scored()` path is unsupported and must fail loud, directing
callers
+ // to the materialized `execute_read()` path exercised below.
+ let scored_err = table
.new_vector_search_builder()
.with_vector_column(VECTOR_COLUMN)
.with_query_vector(query.to_vec())
.with_limit(3)
.execute_scored()
.await
- .expect("primary-key vector search failed");
-
- assert_eq!(
- result.row_ids, expected_row_ids,
- "row ids diverge from the analytic expectation"
+ .expect_err("primary-key execute_scored must fail loud");
+ assert!(
+ format!("{scored_err:?}").contains("execute_read"),
+ "primary-key execute_scored should point at execute_read, got:
{scored_err:?}"
);
- assert_eq!(
- result.scores.len(),
- expected_scores.len(),
- "score count diverges from the analytic expectation"
- );
- for (got, want) in result.scores.iter().zip(&expected_scores) {
- assert!(
- (got - want).abs() < 1e-4,
- "score diverges from the analytic expectation: got {got}, want
{want}"
- );
- }
// Search-and-read: execute_read() materializes the matching rows
best-first
// with a `_PKEY_VECTOR_SCORE` column, hiding
`_ROW_ID`/`_PKEY_VECTOR_POSITION`.
@@ -699,6 +709,88 @@ 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`),
+/// 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.
+#[cfg(not(windows))]
+async fn assert_discriminating_local_read(first_row_id: Option<i64>) {
+ let (query, vectors) = fixture_discriminating();
+ let (_tmp, table) = build_table_with_first_row_id(&query, &vectors, 3,
first_row_id).await;
+
+ let expected = analytic_topk(&query, &vectors, 3);
+ let expected_ids: Vec<i32> = expected.iter().map(|(id, _)| *id as
i32).collect();
+ assert_eq!(
+ expected_ids,
+ vec![5, 1, 3],
+ "fixture must produce best-first order distinct from physical position
order"
+ );
+ let expected_scores: Vec<f32> = expected.iter().map(|(_, d)|
l2_score(*d)).collect();
+
+ // Default projection: id + vector column materialize. The read path must
+ // return the correct FILE-LOCAL rows regardless of `first_row_id`.
+ let (ids, scores, batches) = read_id_and_scores(&table, query.to_vec(), 3,
None).await;
+
+ assert_eq!(
+ ids, expected_ids,
+ "materialized `id` column must be file-local best-first [5, 1, 3] and \
+ independent of the data file's first_row_id"
+ );
+
+ // Row content: each emitted row's vector equals the source vector at that
+ // file-local physical position (a global-id offset would fetch the wrong
row).
+ let got_vectors = collect_vectors(&batches);
+ assert_eq!(got_vectors.len(), 3, "three rows expected");
+ for (row_idx, (id, _)) in expected.iter().enumerate() {
+ assert_eq!(
+ got_vectors[row_idx],
+ vectors[*id as usize].to_vec(),
+ "materialized vector for row id {id} diverges from source data"
+ );
+ }
+
+ // Score alignment must also key off the file-local position.
+ assert_eq!(scores.len(), 3);
+ for (got, want) in scores.iter().zip(&expected_scores) {
+ assert!(
+ (got - want).abs() < 1e-4,
+ "materialized score diverges: got {got}, want {want}"
+ );
+ }
+}
+
+/// Physical-coordinate contract: a primary-key vector read must select rows,
+/// recover positions, and align scores by the file-LOCAL physical position,
and
+/// must NOT require the data file to carry a `first_row_id`. Real Java
primary-key
+/// tables never write `first_row_id` (row-tracking is forbidden for PK
tables), so
+/// this fixture pins the indexed data file's `first_row_id = None` — the same
+/// shape the committed Java fixture has, but over a Rust-built table with a
+/// discriminating dataset (best-first [5, 1, 3] != ascending position [1, 3,
5]).
+/// A read path that keys candidate selection or position recovery off a
global row
+/// id (rather than the file-local physical position) cannot satisfy this.
+// Gated off Windows for the same `file://` tempdir reason as the tests above.
+#[cfg(not(windows))]
+#[tokio::test]
+async fn execute_read_without_first_row_id_selects_local_positions() {
+ assert_discriminating_local_read(None).await;
+}
+
+/// Regression pin for the same physical-coordinate contract with a present but
+/// deliberately NON-aligned `first_row_id = Some(100)`: recovering the
file-local
+/// physical position must not be offset by `first_row_id`. This holds on the
+/// current code (the global-range round-trip is symmetric) and must keep
holding
+/// after the read path switches to file-local coordinates, so a fix that reads
+/// local positions yet leaves a stray `first_row_id` offset would surface
here.
+// Gated off Windows for the same `file://` tempdir reason as the tests above.
+#[cfg(not(windows))]
+#[tokio::test]
+async fn execute_read_ignores_nonzero_first_row_id() {
+ assert_discriminating_local_read(Some(100)).await;
+}
+
/// Fixture #3 (residual): the unrestricted nearest neighbours sit at low ids
/// (0, 1, 2), but the residual predicate `id >= 3` excludes exactly those, so
/// the residual result set is disjoint from the unfiltered one. Among the rows
@@ -831,22 +923,23 @@ async fn
pk_vector_residual_filter_excludes_non_matching_rows() {
.greater_or_equal("id", Datum::Int(residual_threshold as i32))
.expect("build residual predicate on id");
- // Search-only: unfiltered vs residual must differ, and every residual hit
- // must satisfy the predicate (id >= 3), disjoint from the unfiltered set.
- let unfiltered_result = table
+ // A primary-key vector table exposes no global row ids, so
`execute_scored()`
+ // is unsupported on this path — with or without a residual filter — and
must
+ // fail loud, directing callers to the materialized `execute_read()` used
below.
+ let unfiltered_err = table
.new_vector_search_builder()
.with_vector_column(VECTOR_COLUMN)
.with_query_vector(query.to_vec())
.with_limit(3)
.execute_scored()
.await
- .expect("unfiltered primary-key vector search failed");
- assert_eq!(
- unfiltered_result.row_ids, unfiltered_ids,
- "unfiltered search must return [0, 1, 2]"
+ .expect_err("primary-key execute_scored must fail loud");
+ assert!(
+ format!("{unfiltered_err:?}").contains("execute_read"),
+ "got: {unfiltered_err:?}"
);
- let residual_result = table
+ let residual_scored_err = table
.new_vector_search_builder()
.with_vector_column(VECTOR_COLUMN)
.with_query_vector(query.to_vec())
@@ -854,22 +947,11 @@ async fn
pk_vector_residual_filter_excludes_non_matching_rows() {
.with_filter(residual.clone())
.execute_scored()
.await
- .expect("residual primary-key vector search failed");
- let residual_row_ids: Vec<u64> = expected_ids.iter().map(|&id| id as
u64).collect();
- assert_eq!(
- residual_result.row_ids, residual_row_ids,
- "residual search must return best-first [4, 5, 3]"
- );
- assert_ne!(
- residual_result.row_ids, unfiltered_result.row_ids,
- "residual must change the result set relative to no filter"
+ .expect_err("primary-key execute_scored must fail loud with a residual
too");
+ assert!(
+ format!("{residual_scored_err:?}").contains("execute_read"),
+ "got: {residual_scored_err:?}"
);
- for &id in &residual_result.row_ids {
- assert!(
- id >= residual_threshold,
- "residual search returned id {id} that fails the predicate id >=
{residual_threshold}"
- );
- }
// Search-and-read with the residual: default projection materializes id +
// vector column, best-first, with an aligned `_PKEY_VECTOR_SCORE`.
diff --git a/crates/paimon/tests/pk_vector_java_fixture_test.rs
b/crates/paimon/tests/pk_vector_java_fixture_test.rs
new file mode 100644
index 00000000..20384ebd
--- /dev/null
+++ b/crates/paimon/tests/pk_vector_java_fixture_test.rs
@@ -0,0 +1,233 @@
+// 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.
+
+//! Cross-language read-back of a primary-key vector table WRITTEN BY JAVA.
+//!
+//! Unlike `pk_vector_baseline_test` (which builds its table entirely on the
Rust
+//! write path), this test opens a table directory produced by Apache Paimon's
+//! Java writer with the production `ivf-flat` primary-key vector indexer, and
+//! asserts Rust reads + searches it correctly. It validates the cross-language
+//! contract at the table-metadata layer (snapshot / manifest / data-file
+//! `level`+`file_source` / `GlobalIndexMeta` / the `PkVectorSourceMeta`
frame);
+//! the ANN segment interior is the same native core on both sides
+//! (`paimon-vindex-core` <-> its JNI wrapper), so a divergence here would be a
+//! real metadata-parse bug, not a fixture artifact.
+//!
+//! Physical-position contract: Java never writes `first_row_id` on a
primary-key
+//! table (row-tracking is forbidden for PK tables), so the committed data
files
+//! below carry NO `first_row_id`. Row identity is therefore physical
+//! `(file, position)`, not a global row id. Two consequences the assertions
pin:
+//! * `execute_scored()` — which reports global row ids — MUST fail on this
+//! table, because a global row id cannot be recovered without
`first_row_id`.
+//! * `execute_read()` — which materializes rows by physical position — MUST
+//! succeed and return the rows best-first.
+//!
+//! Provenance of `testdata/pkvector/pk_vector_ivf_flat` (opaque binary table
+//! directory, regenerate rather than hand-edit):
+//! * Source: Apache Paimon Java, module `paimon-vector`, commit `7234e4c34`.
+//! * Generator: `PkVectorFixtureGenerator`.
+//! * Command: `mvn -pl paimon-vector test -Dtest=PkVectorFixtureGenerator \
+//! -Dgen.pkvector.fixture=true -Drun.e2e.tests=true`.
+//! * Config: primary key `id`, vector column `embedding`, `ivf-flat`,
+//! `nlist = 1` (exact, deterministic single inverted list), `deduplicate`
+//! merge engine, deletion-vectors enabled.
+//! * Rows: `id == row position`, vectors `[0,0] [1,0] [2,0] [3,0] [4,0]`.
+//! * Query `[0, 0]`, squared-L2 distances `[0, 1, 4, 9, 16]`; top-3 -> ids
+//! `[0, 1, 2]`, distances `[0, 1, 4]`, scores `1/(1+d) = [1.0, 0.5, 0.2]`.
+//! * Fixture tree checksum: `f6c21a447fa7be880713c3d1c27791e7dcb1db10`.
+
+use std::path::Path;
+
+use arrow_array::{Array, Float32Array, Int32Array, RecordBatch};
+use futures::TryStreamExt;
+use paimon::catalog::Identifier;
+use paimon::io::{FileIO, FileIOBuilder};
+use paimon::table::{SchemaManager, Table};
+
+const VECTOR_COLUMN: &str = "embedding";
+const FIXTURE: &str = "testdata/pkvector/pk_vector_ivf_flat";
+
+/// The exact vectors the Java generator wrote, `id == row position`. Query is
+/// `[0, 0]`; squared-L2 distances are the squared norms `[0, 1, 4, 9, 16]`.
+const VECTORS: &[[f32; 2]] = &[[0.0, 0.0], [1.0, 0.0], [2.0, 0.0], [3.0, 0.0],
[4.0, 0.0]];
+
+fn l2_score(distance: f32) -> f32 {
+ 1.0 / (1.0 + distance)
+}
+
+/// Brute-force exact squared-L2 top-k over the fixture rows -> `(position,
distance)`
+/// ascending by distance. `position == vector index == id` because the fixture
+/// pins `id == row position`.
+fn analytic_topk(query: &[f32], k: usize) -> Vec<(u64, f32)> {
+ let mut scored: Vec<(u64, f32)> = VECTORS
+ .iter()
+ .enumerate()
+ .map(|(pos, v)| {
+ let dist: f32 = v
+ .iter()
+ .zip(query.iter())
+ .map(|(a, b)| (a - b) * (a - b))
+ .sum();
+ (pos as u64, dist)
+ })
+ .collect();
+ scored.sort_by(|a, b| a.1.total_cmp(&b.1));
+ scored.truncate(k);
+ scored
+}
+
+/// Open the committed Java-written table directory. Copies the fixture into a
+/// fresh temp dir first so the read path has a private `file://` root (mirrors
+/// how `global_index_scanner.rs` fixtures stage into a temp dir), and so a
stray
+/// write during read can never mutate the committed testdata.
+async fn open_java_fixture() -> (tempfile::TempDir, Table) {
+ let manifest_dir = env!("CARGO_MANIFEST_DIR");
+ let src = Path::new(manifest_dir).join(FIXTURE);
+ let tmp = tempfile::tempdir().expect("create temp dir");
+ let dst = tmp.path().join("pk_vector_ivf_flat");
+ copy_dir(&src, &dst);
+
+ let location = format!("file://{}", dst.display());
+ let file_io: FileIO = FileIOBuilder::new("file").build().expect("build fs
FileIO");
+ let schema = SchemaManager::new(file_io.clone(), location.clone())
+ .latest()
+ .await
+ .expect("failed to list schemas")
+ .expect("fixture table has no schema");
+ let table = Table::new(
+ file_io,
+ Identifier::new("default", "pk_vector_ivf_flat"),
+ location,
+ (*schema).clone(),
+ None,
+ );
+ (tmp, table)
+}
+
+fn copy_dir(src: &Path, dst: &Path) {
+ std::fs::create_dir_all(dst).unwrap();
+ for entry in std::fs::read_dir(src).unwrap() {
+ let entry = entry.unwrap();
+ let from = entry.path();
+ let to = dst.join(entry.file_name());
+ if from.is_dir() {
+ copy_dir(&from, &to);
+ } else {
+ std::fs::copy(&from, &to).unwrap();
+ }
+ }
+}
+
+fn batch_i32(batches: &[RecordBatch], col: &str) -> Vec<i32> {
+ batches
+ .iter()
+ .flat_map(|b| {
+ let idx = b.schema().index_of(col).unwrap();
+ b.column(idx)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap()
+ .values()
+ .to_vec()
+ })
+ .collect()
+}
+
+fn batch_f32(batches: &[RecordBatch], col: &str) -> Vec<f32> {
+ batches
+ .iter()
+ .flat_map(|b| {
+ let idx = b.schema().index_of(col).unwrap();
+ b.column(idx)
+ .as_any()
+ .downcast_ref::<Float32Array>()
+ .unwrap()
+ .values()
+ .to_vec()
+ })
+ .collect()
+}
+
+// Gated off Windows: the fixture is opened via a `file://` URL built from a
+// tempdir path, matching how the sibling `pk_vector_baseline_test` and
+// `rest_catalog_test` gate their `file://` tempdir tests.
+#[cfg(not(target_os = "windows"))]
+#[tokio::test]
+async fn reads_back_java_written_pk_vector_table() {
+ let (_tmp, table) = open_java_fixture().await;
+ let query = vec![0.0f32, 0.0];
+ let k = 3;
+
+ let expected = analytic_topk(&query, k);
+ let expected_ids: Vec<i32> = expected.iter().map(|(id, _)| *id as
i32).collect();
+ let expected_scores: Vec<f32> = expected.iter().map(|(_, d)|
l2_score(*d)).collect();
+ // Guard against silent drift of the analytic ground truth.
+ assert_eq!(
+ expected_ids,
+ vec![0, 1, 2],
+ "fixture top-3 ids must be [0, 1, 2]"
+ );
+
+ // execute_scored() reports global row ids. On a Java-written PK table the
data
+ // files carry no `first_row_id`, so a global row id is unrecoverable and
the
+ // scored path MUST fail loudly rather than fabricate ids.
+ let scored = table
+ .new_vector_search_builder()
+ .with_vector_column(VECTOR_COLUMN)
+ .with_query_vector(query.clone())
+ .with_limit(k)
+ .execute_scored()
+ .await;
+ assert!(
+ scored.is_err(),
+ "execute_scored() must fail on a primary-key vector table: global row
ids \
+ are unavailable when the data files carry no first_row_id"
+ );
+
+ // 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`
+ // carries the metric score.
+ let mut builder = table.new_vector_search_builder();
+ builder
+ .with_vector_column(VECTOR_COLUMN)
+ .with_query_vector(query)
+ .with_limit(k)
+ .with_projection(&["id"]);
+ let batches = builder
+ .execute_read()
+ .await
+ .expect("primary-key vector read over the Java fixture failed")
+ .try_collect::<Vec<_>>()
+ .await
+ .expect("collecting read batches failed");
+
+ let ids = batch_i32(&batches, "id");
+ assert_eq!(
+ ids, expected_ids,
+ "materialized `id` column must be best-first and match the analytic
top-k"
+ );
+
+ let scores = batch_f32(&batches, "_PKEY_VECTOR_SCORE");
+ assert_eq!(scores.len(), k);
+ for (got, want) in scores.iter().zip(&expected_scores) {
+ assert!(
+ (got - want).abs() < 1e-4,
+ "materialized score diverges: got {got}, want {want}"
+ );
+ }
+}
