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 8f35469 Implement global index search modes (#446)
8f35469 is described below
commit 8f35469eee7556949b61ca68cd84834e8190a3b6
Author: Jingsong Lee <[email protected]>
AuthorDate: Fri Jul 3 18:47:06 2026 +0800
Implement global index search modes (#446)
---
.../datafusion/src/full_text_search.rs | 13 +-
crates/paimon/src/spec/core_options.rs | 71 +++
crates/paimon/src/table/data_file_reader.rs | 75 ++-
.../paimon/src/table/full_text_search_builder.rs | 420 +++++++++++--
crates/paimon/src/table/global_index_scanner.rs | 548 ++++++++++++++--
crates/paimon/src/table/table_scan.rs | 61 +-
crates/paimon/src/table/vector_search_builder.rs | 690 ++++++++++++++++++---
7 files changed, 1663 insertions(+), 215 deletions(-)
diff --git a/crates/integrations/datafusion/src/full_text_search.rs
b/crates/integrations/datafusion/src/full_text_search.rs
index 20ff38f..7e9dc39 100644
--- a/crates/integrations/datafusion/src/full_text_search.rs
+++ b/crates/integrations/datafusion/src/full_text_search.rs
@@ -32,9 +32,11 @@ use async_trait::async_trait;
use datafusion::arrow::datatypes::SchemaRef as ArrowSchemaRef;
use datafusion::catalog::Session;
use datafusion::catalog::TableFunctionImpl;
+use datafusion::common::project_schema;
use datafusion::datasource::{TableProvider, TableType};
use datafusion::error::Result as DFResult;
use datafusion::logical_expr::{Expr, TableProviderFilterPushDown};
+use datafusion::physical_plan::empty::EmptyExec;
use datafusion::physical_plan::ExecutionPlan;
use datafusion::prelude::SessionContext;
use paimon::catalog::Catalog;
@@ -169,16 +171,17 @@ impl TableProvider for FullTextSearchTableProvider {
})
.await?;
+ if row_ranges.is_empty() {
+ let schema = project_schema(&self.schema(), projection)?;
+ return Ok(Arc::new(EmptyExec::new(schema)));
+ }
+
// Convert search results to row ranges and inject into the scan.
let mut read_builder = table.new_read_builder();
if let Some(limit) = limit {
read_builder.with_limit(limit);
}
- let scan = if row_ranges.is_empty() {
- read_builder.new_scan()
- } else {
- read_builder.new_scan().with_row_ranges(row_ranges)
- };
+ let scan = read_builder.new_scan().with_row_ranges(row_ranges);
let plan = await_with_runtime(scan.plan())
.await
.map_err(to_datafusion_error)?;
diff --git a/crates/paimon/src/spec/core_options.rs
b/crates/paimon/src/spec/core_options.rs
index 80e42e2..f11298f 100644
--- a/crates/paimon/src/spec/core_options.rs
+++ b/crates/paimon/src/spec/core_options.rs
@@ -20,6 +20,7 @@ use std::collections::{HashMap, HashSet};
const DELETION_VECTORS_ENABLED_OPTION: &str = "deletion-vectors.enabled";
const DATA_EVOLUTION_ENABLED_OPTION: &str = "data-evolution.enabled";
const GLOBAL_INDEX_ENABLED_OPTION: &str = "global-index.enabled";
+const GLOBAL_INDEX_SEARCH_MODE_OPTION: &str = "global-index.search-mode";
const GLOBAL_INDEX_ROW_COUNT_PER_SHARD_OPTION: &str =
"global-index.row-count-per-shard";
const GLOBAL_INDEX_COLUMN_UPDATE_ACTION_OPTION: &str =
"global-index.column-update-action";
const SOURCE_SPLIT_TARGET_SIZE_OPTION: &str = "source.split.target-size";
@@ -117,6 +118,19 @@ pub enum GlobalIndexColumnUpdateAction {
DropPartitionIndex,
}
+/// Search mode for global index queries.
+///
+/// Reference: Java `CoreOptions.GlobalIndexSearchMode`.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum GlobalIndexSearchMode {
+ /// Only search indexed data.
+ Fast,
+ /// Use snapshot `next_row_id` and global index coverage to detect missing
row IDs.
+ Full,
+ /// Use actual data-file row ID ranges to detect exact missing row IDs.
+ Detail,
+}
+
/// Bucket function used to map bucket keys to fixed bucket ids.
///
/// Reference: Java `CoreOptions.BucketFunctionType`.
@@ -262,6 +276,23 @@ impl<'a> CoreOptions<'a> {
.unwrap_or(false)
}
+ pub fn global_index_search_mode(&self) ->
crate::Result<GlobalIndexSearchMode> {
+ match self
+ .options
+ .get(GLOBAL_INDEX_SEARCH_MODE_OPTION)
+ .map(|v| v.to_ascii_lowercase())
+ .as_deref()
+ .unwrap_or("fast")
+ {
+ "fast" => Ok(GlobalIndexSearchMode::Fast),
+ "full" => Ok(GlobalIndexSearchMode::Full),
+ "detail" => Ok(GlobalIndexSearchMode::Detail),
+ other => Err(crate::Error::ConfigInvalid {
+ message: format!("Unsupported global-index.search-mode:
{other}"),
+ }),
+ }
+ }
+
pub fn global_index_row_count_per_shard(&self) -> crate::Result<i64> {
let value = self
.parse_i64_option(GLOBAL_INDEX_ROW_COUNT_PER_SHARD_OPTION)?
@@ -657,6 +688,10 @@ mod tests {
core_options.global_index_column_update_action().unwrap(),
GlobalIndexColumnUpdateAction::ThrowError
);
+ assert_eq!(
+ core_options.global_index_search_mode().unwrap(),
+ GlobalIndexSearchMode::Fast
+ );
}
#[test]
@@ -678,6 +713,10 @@ mod tests {
GLOBAL_INDEX_COLUMN_UPDATE_ACTION_OPTION.to_string(),
"DROP_PARTITION_INDEX".to_string(),
),
+ (
+ GLOBAL_INDEX_SEARCH_MODE_OPTION.to_string(),
+ "detail".to_string(),
+ ),
]);
let core_options = CoreOptions::new(&options);
@@ -691,6 +730,38 @@ mod tests {
core_options.global_index_column_update_action().unwrap(),
GlobalIndexColumnUpdateAction::DropPartitionIndex
);
+ assert_eq!(
+ core_options.global_index_search_mode().unwrap(),
+ GlobalIndexSearchMode::Detail
+ );
+ }
+
+ #[test]
+ fn test_global_index_search_mode_values() {
+ for (raw, expected) in [
+ ("fast", GlobalIndexSearchMode::Fast),
+ ("FAST", GlobalIndexSearchMode::Fast),
+ ("full", GlobalIndexSearchMode::Full),
+ ("detail", GlobalIndexSearchMode::Detail),
+ ] {
+ let options =
+ HashMap::from([(GLOBAL_INDEX_SEARCH_MODE_OPTION.to_string(),
raw.to_string())]);
+ let core = CoreOptions::new(&options);
+ assert_eq!(core.global_index_search_mode().unwrap(), expected);
+ }
+ }
+
+ #[test]
+ fn test_global_index_search_mode_rejects_invalid_value() {
+ let options = HashMap::from([(
+ GLOBAL_INDEX_SEARCH_MODE_OPTION.to_string(),
+ "slow".to_string(),
+ )]);
+ let core = CoreOptions::new(&options);
+
+ let err = core.global_index_search_mode().expect_err("invalid mode");
+ assert!(matches!(err, crate::Error::ConfigInvalid { message }
+ if message.contains(GLOBAL_INDEX_SEARCH_MODE_OPTION)));
}
#[test]
diff --git a/crates/paimon/src/table/data_file_reader.rs
b/crates/paimon/src/table/data_file_reader.rs
index 111e23e..60ecdf4 100644
--- a/crates/paimon/src/table/data_file_reader.rs
+++ b/crates/paimon/src/table/data_file_reader.rs
@@ -20,7 +20,7 @@ use crate::arrow::format::create_format_reader;
use crate::arrow::schema_evolution::{create_index_mapping, NULL_FIELD_INDEX};
use crate::deletion_vector::{DeletionVector, DeletionVectorFactory};
use crate::io::FileIO;
-use crate::spec::{DataField, DataFileMeta, Predicate};
+use crate::spec::{DataField, DataFileMeta, Predicate, ROW_ID_FIELD_NAME};
use crate::table::schema_manager::SchemaManager;
use crate::table::ArrowRecordBatchStream;
use crate::table::RowRange;
@@ -174,7 +174,14 @@ impl DataFileReader {
None => (df.clone(), None),
}
} else {
- (read_type.clone(), None)
+ (
+ 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()
@@ -213,6 +220,14 @@ impl DataFileReader {
dv.as_deref(),
local_ranges.as_deref(),
);
+ let selected_row_ids = match (file_meta.first_row_id,
row_selection.as_ref()) {
+ (Some(first_row_id), Some(ranges)) => {
+ Some(expand_local_selected_row_ids(first_row_id, ranges))
+ }
+ _ => None,
+ };
+ let mut row_id_cursor = file_meta.first_row_id.unwrap_or(0);
+ let mut row_id_offset = 0usize;
let mut batch_stream = format_reader.read_batch_stream(
Box::new(file_reader),
@@ -231,6 +246,17 @@ impl DataFileReader {
// Build output columns using index mapping (field-ID-based)
or by name.
let mut columns: Vec<Arc<dyn arrow_array::Array>> =
Vec::with_capacity(target_schema.fields().len());
for (i, target_field) in
target_schema.fields().iter().enumerate() {
+ if target_field.name() == ROW_ID_FIELD_NAME {
+ columns.push(row_id_column_for_batch(
+ file_meta.first_row_id,
+ num_rows,
+ &mut row_id_cursor,
+ selected_row_ids.as_deref(),
+ &mut row_id_offset,
+ )?);
+ continue;
+ }
+
let source_col = if let Some(ref idx_map) = index_mapping {
let data_idx = idx_map[i];
if data_idx == NULL_FIELD_INDEX {
@@ -420,6 +446,51 @@ pub(super) fn expand_selected_row_ids(
ids
}
+fn expand_local_selected_row_ids(first_row_id: i64, local_ranges: &[RowRange])
-> Vec<i64> {
+ let mut ids = Vec::new();
+ for range in local_ranges {
+ for local_id in range.from()..=range.to() {
+ ids.push(first_row_id + local_id);
+ }
+ }
+ ids
+}
+
+fn row_id_column_for_batch(
+ first_row_id: Option<i64>,
+ num_rows: usize,
+ row_id_cursor: &mut i64,
+ selected_row_ids: Option<&[i64]>,
+ row_id_offset: &mut usize,
+) -> crate::Result<Arc<dyn arrow_array::Array>> {
+ let Some(_) = first_row_id else {
+ return Ok(Arc::new(Int64Array::new_null(num_rows)));
+ };
+
+ if let Some(selected_row_ids) = selected_row_ids {
+ let end = *row_id_offset + num_rows;
+ if end > selected_row_ids.len() {
+ return Err(Error::UnexpectedError {
+ message: format!(
+ "Row ID offset out of bounds: need {}..{} but
selected_row_ids has {} entries",
+ *row_id_offset,
+ end,
+ selected_row_ids.len()
+ ),
+ source: None,
+ });
+ }
+ let batch_ids = &selected_row_ids[*row_id_offset..end];
+ *row_id_offset = end;
+ return Ok(Arc::new(Int64Array::from(batch_ids.to_vec())));
+ }
+
+ let start = *row_id_cursor;
+ let end = start + num_rows as i64;
+ *row_id_cursor = end;
+ Ok(Arc::new(Int64Array::from((start..end).collect::<Vec<_>>())))
+}
+
pub(super) fn attach_row_id(
batch: RecordBatch,
row_id_index: usize,
diff --git a/crates/paimon/src/table/full_text_search_builder.rs
b/crates/paimon/src/table/full_text_search_builder.rs
index 41297b7..04ffe21 100644
--- a/crates/paimon/src/table/full_text_search_builder.rs
+++ b/crates/paimon/src/table/full_text_search_builder.rs
@@ -19,11 +19,21 @@
//!
//! Reference:
[FullTextSearchBuilderImpl.java](https://github.com/apache/paimon/blob/master/paimon-core/src/main/java/org/apache/paimon/table/source/FullTextSearchBuilderImpl.java)
-use crate::spec::{DataField, FileKind, IndexManifest};
+use crate::io::{FileIO, FileIOBuilder};
+use crate::spec::{
+ CoreOptions, DataField, FileKind, GlobalIndexSearchMode, IndexFileMeta,
IndexManifest,
+ IndexManifestEntry, ROW_ID_FIELD_NAME,
+};
+use
crate::table::global_index_scanner::unindexed_ranges_for_global_index_entries;
use crate::table::snapshot_manager::SnapshotManager;
-use crate::table::{find_field_id_by_name, RowRange, Table};
+use crate::table::{find_field_id_by_name, merge_row_ranges, RowRange, Table};
use crate::tantivy::full_text_search::{FullTextSearch, SearchResult};
use crate::tantivy::reader::TantivyFullTextReader;
+use crate::tantivy::writer::TantivyFullTextWriter;
+use arrow_array::{Array, Int64Array, LargeStringArray, RecordBatch,
StringArray};
+use futures::TryStreamExt;
+use std::collections::{HashMap, HashSet};
+use uuid::Uuid;
const INDEX_DIR: &str = "index";
const TANTIVY_FULLTEXT_INDEX_TYPE: &str = "tantivy-fulltext";
@@ -111,40 +121,53 @@ impl<'a> FullTextSearchBuilder<'a> {
None => return Ok(Vec::new()),
};
- let index_manifest_name = match snapshot.index_manifest() {
- Some(name) => name.to_string(),
- None => return Ok(Vec::new()),
+ let index_entries = match snapshot.index_manifest() {
+ Some(index_manifest_name) => {
+ let manifest_path = format!(
+ "{}/manifest/{}",
+ self.table.location().trim_end_matches('/'),
+ index_manifest_name
+ );
+ IndexManifest::read(self.table.file_io(),
&manifest_path).await?
+ }
+ None => Vec::new(),
};
- let manifest_path = format!(
- "{}/manifest/{}",
- self.table.location().trim_end_matches('/'),
- index_manifest_name
- );
- let index_entries = IndexManifest::read(self.table.file_io(),
&manifest_path).await?;
-
evaluate_full_text_search(
- self.table.file_io(),
- self.table.location(),
+ FullTextSearchEvaluation {
+ table: Some(self.table),
+ file_io: self.table.file_io(),
+ table_path: self.table.location(),
+ table_options: self.table.schema().options(),
+ schema_fields: self.table.schema().fields(),
+ next_row_id: snapshot.next_row_id(),
+ },
&index_entries,
&search,
- self.table.schema().fields(),
)
.await
}
}
/// Evaluate a full-text search query against Tantivy indexes found in the
index manifest.
+struct FullTextSearchEvaluation<'a> {
+ table: Option<&'a Table>,
+ file_io: &'a FileIO,
+ table_path: &'a str,
+ table_options: &'a HashMap<String, String>,
+ schema_fields: &'a [DataField],
+ next_row_id: Option<i64>,
+}
+
async fn evaluate_full_text_search(
- file_io: &crate::io::FileIO,
- table_path: &str,
- index_entries: &[crate::spec::IndexManifestEntry],
+ evaluation: FullTextSearchEvaluation<'_>,
+ index_entries: &[IndexManifestEntry],
search: &FullTextSearch,
- schema_fields: &[DataField],
) -> crate::Result<Vec<RowRange>> {
- let table_path = table_path.trim_end_matches('/');
+ let table_path = evaluation.table_path.trim_end_matches('/');
+ let search_mode =
CoreOptions::new(evaluation.table_options).global_index_search_mode()?;
- let field_id = match find_field_id_by_name(schema_fields,
&search.field_name) {
+ let field_id = match find_field_id_by_name(evaluation.schema_fields,
&search.field_name) {
Some(id) => id,
None => return Ok(Vec::new()),
};
@@ -162,42 +185,335 @@ async fn evaluate_full_text_search(
})
.collect();
- if fulltext_entries.is_empty() {
+ if fulltext_entries.is_empty() && search_mode ==
GlobalIndexSearchMode::Fast {
return Ok(Vec::new());
}
- let futures: Vec<_> = fulltext_entries
- .into_iter()
- .map(|entry| {
- let global_meta =
entry.index_file.global_index_meta.as_ref().unwrap();
- let path = format!("{table_path}/{INDEX_DIR}/{}",
entry.index_file.file_name);
- let file_name = entry.index_file.file_name.clone();
- let query_text = search.query_text.clone();
- let limit = search.limit;
- let row_range_start = global_meta.row_range_start;
- let input = file_io.new_input(&path);
- async move {
- let input = input?;
- let reader = TantivyFullTextReader::from_input_file(&input)
- .await
- .map_err(|e| crate::Error::UnexpectedError {
- message: format!(
- "Failed to open Tantivy full-text index '{}': {}",
- file_name, e
- ),
- source: None,
- })?;
- let result = reader.search(&query_text, limit)?;
- Ok::<_, crate::Error>(result.offset(row_range_start))
- }
- })
- .collect();
-
- let results = futures::future::try_join_all(futures).await?;
let mut merged = SearchResult::empty();
- for r in &results {
- merged = merged.or(r);
+ if !fulltext_entries.is_empty() {
+ let futures: Vec<_> = fulltext_entries
+ .into_iter()
+ .map(|entry| {
+ let global_meta =
entry.index_file.global_index_meta.as_ref().unwrap();
+ let path = format!("{table_path}/{INDEX_DIR}/{}",
entry.index_file.file_name);
+ let file_name = entry.index_file.file_name.clone();
+ let query_text = search.query_text.clone();
+ let limit = search.limit;
+ let row_range_start = global_meta.row_range_start;
+ let input = evaluation.file_io.new_input(&path);
+ async move {
+ let input = input?;
+ let reader = TantivyFullTextReader::from_input_file(&input)
+ .await
+ .map_err(|e| crate::Error::UnexpectedError {
+ message: format!(
+ "Failed to open Tantivy full-text index '{}':
{}",
+ file_name, e
+ ),
+ source: None,
+ })?;
+ let result = reader.search(&query_text, limit)?;
+ Ok::<_, crate::Error>(result.offset(row_range_start))
+ }
+ })
+ .collect();
+
+ let results = futures::future::try_join_all(futures).await?;
+ for r in &results {
+ merged = merged.or(r);
+ }
+ }
+
+ if search_mode != GlobalIndexSearchMode::Fast {
+ let detail_ranges = if search_mode == GlobalIndexSearchMode::Detail {
+ let table = evaluation.table.ok_or_else(||
crate::Error::DataInvalid {
+ message: "Full-text raw search in detail mode requires table
context".to_string(),
+ source: None,
+ })?;
+ detail_data_ranges_for_table(table).await?
+ } else {
+ Vec::new()
+ };
+ let field_ids = HashSet::from([field_id]);
+ let raw_ranges = unindexed_ranges_for_global_index_entries(
+ index_entries,
+ &field_ids,
+ search_mode,
+ evaluation.next_row_id,
+ &detail_ranges,
+ is_tantivy_fulltext_index_file,
+ );
+ if !raw_ranges.is_empty() {
+ let table = evaluation.table.ok_or_else(||
crate::Error::DataInvalid {
+ message: "Full-text raw search requires table
context".to_string(),
+ source: None,
+ })?;
+ let raw_result = read_raw_full_text_search(table, search,
&raw_ranges).await?;
+ merged = merged.or(&raw_result);
+ }
}
Ok(merged.top_k(search.limit).to_row_ranges())
}
+
+fn is_tantivy_fulltext_index_file(index_file: &IndexFileMeta) -> bool {
+ index_file.index_type == TANTIVY_FULLTEXT_INDEX_TYPE
+}
+
+async fn detail_data_ranges_for_table(table: &Table) ->
crate::Result<Vec<RowRange>> {
+ let plan = table
+ .new_read_builder()
+ .new_scan()
+ .with_scan_all_files()
+ .plan()
+ .await?;
+ let mut ranges = Vec::new();
+ for split in plan.splits() {
+ for file in split.data_files() {
+ if let Some((from, to)) = file.row_id_range() {
+ ranges.push(RowRange::new(from, to));
+ }
+ }
+ }
+ Ok(merge_row_ranges(ranges))
+}
+
+async fn read_raw_full_text_search(
+ table: &Table,
+ search: &FullTextSearch,
+ raw_ranges: &[RowRange],
+) -> crate::Result<SearchResult> {
+ if raw_ranges.is_empty() {
+ return Ok(SearchResult::empty());
+ }
+
+ let mut read_builder = table.new_read_builder();
+ read_builder
+ .with_projection(&[search.field_name.as_str(), ROW_ID_FIELD_NAME])
+ .with_row_ranges(raw_ranges.to_vec());
+ let plan = read_builder.new_scan().plan().await?;
+ if plan.splits().is_empty() {
+ return Ok(SearchResult::empty());
+ }
+ let read = read_builder.new_read()?;
+ let mut stream = read.to_arrow(plan.splits())?;
+
+ let mut writer = TantivyFullTextWriter::new()?;
+ while let Some(batch) = stream.try_next().await? {
+ add_raw_full_text_batch(&batch, search, &mut writer)?;
+ }
+
+ let memory_io = FileIOBuilder::new("memory").build()?;
+ let output = memory_io.new_output(&format!("/raw-fulltext-{}.archive",
Uuid::new_v4()))?;
+ if !writer.finish(&output).await? {
+ return Ok(SearchResult::empty());
+ }
+ let input = output.to_input_file();
+ let reader = TantivyFullTextReader::from_input_file(&input)
+ .await
+ .map_err(|e| crate::Error::UnexpectedError {
+ message: format!("Failed to open raw Tantivy full-text index:
{e}"),
+ source: None,
+ })?;
+ reader.search(&search.query_text, search.limit)
+}
+
+fn add_raw_full_text_batch(
+ batch: &RecordBatch,
+ search: &FullTextSearch,
+ writer: &mut TantivyFullTextWriter,
+) -> crate::Result<()> {
+ let text_index =
+ batch
+ .schema()
+ .index_of(&search.field_name)
+ .map_err(|e| crate::Error::DataInvalid {
+ message: format!(
+ "Full-text column '{}' not found in raw search batch: {}",
+ search.field_name, e
+ ),
+ source: None,
+ })?;
+ let row_id_index =
+ batch
+ .schema()
+ .index_of(ROW_ID_FIELD_NAME)
+ .map_err(|e| crate::Error::DataInvalid {
+ message: format!("_ROW_ID column not found in raw search
batch: {e}"),
+ source: None,
+ })?;
+ let row_ids = batch
+ .column(row_id_index)
+ .as_any()
+ .downcast_ref::<Int64Array>()
+ .ok_or_else(|| crate::Error::DataInvalid {
+ message: "Full-text raw search requires non-null Int64
_ROW_ID".to_string(),
+ source: None,
+ })?;
+ let column = batch.column(text_index);
+
+ if let Some(strings) = column.as_any().downcast_ref::<StringArray>() {
+ for row in 0..batch.num_rows() {
+ add_raw_full_text_row(row_ids, row, get_string_value(strings,
row), writer)?;
+ }
+ return Ok(());
+ }
+
+ if let Some(strings) = column.as_any().downcast_ref::<LargeStringArray>() {
+ for row in 0..batch.num_rows() {
+ add_raw_full_text_row(row_ids, row,
get_large_string_value(strings, row), writer)?;
+ }
+ return Ok(());
+ }
+
+ Err(crate::Error::DataInvalid {
+ message: "Full-text raw search requires Utf8 or LargeUtf8 text
column".to_string(),
+ source: None,
+ })
+}
+
+fn get_string_value(strings: &StringArray, row: usize) -> Option<&str> {
+ if strings.is_null(row) {
+ None
+ } else {
+ Some(strings.value(row))
+ }
+}
+
+fn get_large_string_value(strings: &LargeStringArray, row: usize) ->
Option<&str> {
+ if strings.is_null(row) {
+ None
+ } else {
+ Some(strings.value(row))
+ }
+}
+
+fn add_raw_full_text_row(
+ row_ids: &Int64Array,
+ row: usize,
+ text: Option<&str>,
+ writer: &mut TantivyFullTextWriter,
+) -> crate::Result<()> {
+ if row_ids.is_null(row) {
+ return Err(crate::Error::DataInvalid {
+ message: "Full-text raw search found null _ROW_ID".to_string(),
+ source: None,
+ });
+ }
+ let row_id = u64::try_from(row_ids.value(row)).map_err(|_|
crate::Error::DataInvalid {
+ message: format!(
+ "Negative _ROW_ID {} cannot be used for global index search",
+ row_ids.value(row)
+ ),
+ source: None,
+ })?;
+ writer.add_document(row_id, text)
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::catalog::Identifier;
+ use crate::spec::{DataType, IntType, Schema, TableSchema, VarCharType};
+ use crate::table::table_write::TableWrite;
+ use crate::table::TableCommit;
+ use arrow_array::StringArray;
+ use arrow_schema::{DataType as ArrowDataType, Field as ArrowField, Schema
as ArrowSchema};
+ use std::sync::Arc;
+
+ #[tokio::test]
+ async fn test_evaluate_full_mode_without_fulltext_entries_uses_raw_path() {
+ let file_io = FileIOBuilder::new("memory").build().unwrap();
+ let fields = vec![DataField::new(
+ 1,
+ "body".to_string(),
+ DataType::Int(IntType::default()),
+ )];
+ let search = FullTextSearch::new("hello".to_string(), 10,
"body".to_string()).unwrap();
+ let options = HashMap::from([("global-index.search-mode".to_string(),
"full".to_string())]);
+
+ let err = evaluate_full_text_search(
+ FullTextSearchEvaluation {
+ table: None,
+ file_io: &file_io,
+ table_path: "memory:///test_table",
+ table_options: &options,
+ schema_fields: &fields,
+ next_row_id: Some(10),
+ },
+ &[],
+ &search,
+ )
+ .await
+ .unwrap_err();
+ assert!(
+ err.to_string()
+ .contains("Full-text raw search requires table context"),
+ "unexpected error: {err}"
+ );
+ }
+
+ #[tokio::test]
+ async fn test_execute_full_mode_without_index_manifest_searches_raw_rows()
{
+ let file_io = FileIOBuilder::new("memory").build().unwrap();
+ let table_path = "memory:/full_text_raw_no_manifest";
+ setup_dirs(&file_io, table_path).await;
+ let table = full_text_raw_table(&file_io, table_path);
+
+ let mut table_write = TableWrite::new(&table,
"test-user".to_string()).unwrap();
+ table_write
+ .write_arrow_batch(&text_batch(vec!["hello world", "goodbye"]))
+ .await
+ .unwrap();
+ let messages = table_write.prepare_commit().await.unwrap();
+ TableCommit::new(table.clone(), "test-user".to_string())
+ .commit(messages)
+ .await
+ .unwrap();
+
+ let mut builder = table.new_full_text_search_builder();
+ builder
+ .with_text_column("body")
+ .with_query_text("hello")
+ .with_limit(10);
+ let row_ranges = builder.execute().await.unwrap();
+
+ assert_eq!(row_ranges, vec![RowRange::new(0, 0)]);
+ }
+
+ async fn setup_dirs(file_io: &FileIO, table_path: &str) {
+ file_io
+ .mkdirs(&format!("{table_path}/snapshot/"))
+ .await
+ .unwrap();
+ file_io
+ .mkdirs(&format!("{table_path}/manifest/"))
+ .await
+ .unwrap();
+ }
+
+ fn full_text_raw_table(file_io: &FileIO, table_path: &str) -> Table {
+ let schema = Schema::builder()
+ .column("body", DataType::VarChar(VarCharType::string_type()))
+ .option("row-tracking.enabled", "true")
+ .option("global-index.search-mode", "full")
+ .build()
+ .unwrap();
+ Table::new(
+ file_io.clone(),
+ Identifier::new("default", "full_text_raw_no_manifest"),
+ table_path.to_string(),
+ TableSchema::new(0, &schema),
+ None,
+ )
+ }
+
+ fn text_batch(values: Vec<&str>) -> RecordBatch {
+ let schema = Arc::new(ArrowSchema::new(vec![ArrowField::new(
+ "body",
+ ArrowDataType::Utf8,
+ false,
+ )]));
+ RecordBatch::try_new(schema,
vec![Arc::new(StringArray::from(values))]).unwrap()
+ }
+}
diff --git a/crates/paimon/src/table/global_index_scanner.rs
b/crates/paimon/src/table/global_index_scanner.rs
index dd2ab09..8ed18a9 100644
--- a/crates/paimon/src/table/global_index_scanner.rs
+++ b/crates/paimon/src/table/global_index_scanner.rs
@@ -24,13 +24,14 @@ use crate::btree::query::{extract_between, IndexQuery};
use crate::btree::{make_key_comparator, serialize_datum, BTreeIndexMeta,
BTreeIndexReader};
use crate::io::FileIO;
use crate::spec::{
- DataField, DataType, Datum, FileKind, IndexManifestEntry, Predicate,
PredicateOperator,
+ DataField, DataType, Datum, FileKind, GlobalIndexSearchMode,
IndexFileMeta, IndexManifestEntry,
+ Predicate, PredicateOperator,
};
use crate::table::RowRange;
use crate::Result;
use roaring::RoaringTreemap;
use std::cmp::Ordering;
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
use std::sync::Mutex;
type BoxedCmp = Box<dyn Fn(&[u8], &[u8]) -> Ordering + Send + Sync>;
@@ -54,6 +55,8 @@ pub(crate) struct GlobalIndexScanner {
table_path: String,
/// Global index entries grouped by field_id.
entries_by_field: Vec<(i32, Vec<GlobalIndexEntry>)>,
+ /// Indexed row-id coverage grouped by field_id.
+ coverage_by_field: HashMap<i32, Vec<RowRange>>,
/// Schema fields for field_id lookup.
schema_fields: Vec<DataField>,
/// Cache of opened BTree readers, keyed by file name.
@@ -78,6 +81,7 @@ impl GlobalIndexScanner {
) -> Option<Self> {
let mut entries_by_field: std::collections::HashMap<i32,
Vec<GlobalIndexEntry>> =
std::collections::HashMap::new();
+ let mut coverage_by_field: HashMap<i32, Vec<RowRange>> =
HashMap::new();
for entry in index_entries {
if entry.kind != FileKind::Add {
@@ -103,6 +107,20 @@ impl GlobalIndexScanner {
meta: btree_meta,
};
+ let row_range = RowRange::new(global_meta.row_range_start,
global_meta.row_range_end);
+ coverage_by_field
+ .entry(global_meta.index_field_id)
+ .or_default()
+ .push(row_range.clone());
+ if let Some(extra_field_ids) =
global_meta.extra_field_ids.as_ref() {
+ for extra_field_id in extra_field_ids {
+ coverage_by_field
+ .entry(*extra_field_id)
+ .or_default()
+ .push(row_range.clone());
+ }
+ }
+
entries_by_field
.entry(global_meta.index_field_id)
.or_default()
@@ -117,6 +135,7 @@ impl GlobalIndexScanner {
file_io: file_io.clone(),
table_path: table_path.trim_end_matches('/').to_string(),
entries_by_field: entries_by_field.into_iter().collect(),
+ coverage_by_field,
schema_fields: schema_fields.to_vec(),
reader_cache: Mutex::new(HashMap::new()),
})
@@ -397,6 +416,57 @@ impl GlobalIndexScanner {
.find(|(id, _)| *id == field_id)
.map(|(_, entries)| entries.as_slice())
}
+
+ /// Return row ranges not covered by global indexes for this predicate.
+ ///
+ /// `full` uses `[0, snapshot.next_row_id - 1]`; `detail` uses actual
+ /// data-file row ranges collected by the scan. The caller unions these
+ /// ranges with indexed matches, and the normal read filter evaluates the
+ /// predicate on the raw rows.
+ fn unindexed_ranges(
+ &self,
+ predicate: &Predicate,
+ search_mode: GlobalIndexSearchMode,
+ next_row_id: Option<i64>,
+ data_ranges: &[RowRange],
+ ) -> Result<Vec<RowRange>> {
+ let field_ids = self.collect_field_ids(predicate)?;
+ Ok(unindexed_ranges_for_coverage(
+ &self.coverage_by_field,
+ &field_ids,
+ search_mode,
+ next_row_id,
+ data_ranges,
+ ))
+ }
+
+ fn collect_field_ids(&self, predicate: &Predicate) -> Result<HashSet<i32>>
{
+ let mut field_ids = HashSet::new();
+ self.collect_field_ids_inner(predicate, &mut field_ids)?;
+ Ok(field_ids)
+ }
+
+ fn collect_field_ids_inner(
+ &self,
+ predicate: &Predicate,
+ field_ids: &mut HashSet<i32>,
+ ) -> Result<()> {
+ match predicate {
+ Predicate::Leaf { column, .. } => {
+ if let Some(field_id) = self.find_field_id_by_name(column)? {
+ field_ids.insert(field_id);
+ }
+ }
+ Predicate::And(children) | Predicate::Or(children) => {
+ for child in children {
+ self.collect_field_ids_inner(child, field_ids)?;
+ }
+ }
+ Predicate::Not(inner) => self.collect_field_ids_inner(inner,
field_ids)?,
+ Predicate::AlwaysTrue | Predicate::AlwaysFalse => {}
+ }
+ Ok(())
+ }
}
/// Whether the b-tree global index can evaluate this operator directly.
@@ -455,6 +525,144 @@ fn intersect_sorted_ranges(a: &[RowRange], b:
&[RowRange]) -> Vec<RowRange> {
result
}
+fn exclude_row_ranges(data_ranges: &[RowRange], indexed_ranges: &[RowRange])
-> Vec<RowRange> {
+ let data_ranges = super::merge_row_ranges(data_ranges.to_vec());
+ if data_ranges.is_empty() {
+ return Vec::new();
+ }
+ let indexed_ranges = super::merge_row_ranges(indexed_ranges.to_vec());
+ if indexed_ranges.is_empty() {
+ return data_ranges;
+ }
+
+ let mut result = Vec::new();
+ for data_range in data_ranges {
+ let mut cursor = data_range.from();
+ let mut exhausted = false;
+ for indexed_range in &indexed_ranges {
+ if indexed_range.to() < cursor {
+ continue;
+ }
+ if indexed_range.from() > data_range.to() {
+ break;
+ }
+ if indexed_range.from() > cursor {
+ result.push(RowRange::new(cursor, indexed_range.from() - 1));
+ }
+ if indexed_range.to() >= data_range.to() {
+ exhausted = true;
+ break;
+ }
+ cursor = cursor.max(indexed_range.to() + 1);
+ }
+ if !exhausted && cursor <= data_range.to() {
+ result.push(RowRange::new(cursor, data_range.to()));
+ }
+ }
+ super::merge_row_ranges(result)
+}
+
+fn data_ranges_for_search_mode(
+ search_mode: GlobalIndexSearchMode,
+ next_row_id: Option<i64>,
+ data_ranges: &[RowRange],
+) -> Option<Vec<RowRange>> {
+ match search_mode {
+ GlobalIndexSearchMode::Fast => None,
+ GlobalIndexSearchMode::Full => match next_row_id {
+ Some(next_row_id) if next_row_id > 0 => Some(vec![RowRange::new(0,
next_row_id - 1)]),
+ _ => None,
+ },
+ GlobalIndexSearchMode::Detail => {
+ if data_ranges.is_empty() {
+ None
+ } else {
+ Some(data_ranges.to_vec())
+ }
+ }
+ }
+}
+
+fn indexed_ranges_from_coverage(
+ coverage_by_field: &HashMap<i32, Vec<RowRange>>,
+ field_ids: &HashSet<i32>,
+) -> Vec<RowRange> {
+ let mut ranges: Option<Vec<RowRange>> = None;
+ for field_id in field_ids {
+ let Some(field_ranges) = coverage_by_field.get(field_id) else {
+ return Vec::new();
+ };
+ if field_ranges.is_empty() {
+ return Vec::new();
+ }
+ let field_ranges = super::merge_row_ranges(field_ranges.clone());
+ ranges = Some(match ranges {
+ None => field_ranges,
+ Some(existing) => intersect_sorted_ranges(&existing,
&field_ranges),
+ });
+ }
+ ranges.map(super::merge_row_ranges).unwrap_or_default()
+}
+
+fn unindexed_ranges_for_coverage(
+ coverage_by_field: &HashMap<i32, Vec<RowRange>>,
+ field_ids: &HashSet<i32>,
+ search_mode: GlobalIndexSearchMode,
+ next_row_id: Option<i64>,
+ data_ranges: &[RowRange],
+) -> Vec<RowRange> {
+ let Some(data_ranges) = data_ranges_for_search_mode(search_mode,
next_row_id, data_ranges)
+ else {
+ return Vec::new();
+ };
+ let indexed_ranges = indexed_ranges_from_coverage(coverage_by_field,
field_ids);
+ exclude_row_ranges(&data_ranges, &indexed_ranges)
+}
+
+/// Compute row ranges not covered by a family of global index files.
+///
+/// This mirrors Java `GlobalIndexCoverage`: `full` compares index coverage
+/// against `[0, snapshot.next_row_id - 1]`, while `detail` compares against
+/// exact data-file row ranges supplied by the caller.
+pub(crate) fn unindexed_ranges_for_global_index_entries(
+ index_entries: &[IndexManifestEntry],
+ field_ids: &HashSet<i32>,
+ search_mode: GlobalIndexSearchMode,
+ next_row_id: Option<i64>,
+ data_ranges: &[RowRange],
+ index_file_filter: impl Fn(&IndexFileMeta) -> bool,
+) -> Vec<RowRange> {
+ let mut coverage_by_field: HashMap<i32, Vec<RowRange>> = HashMap::new();
+ for entry in index_entries {
+ if entry.kind != FileKind::Add ||
!index_file_filter(&entry.index_file) {
+ continue;
+ }
+ let Some(global_meta) = entry.index_file.global_index_meta.as_ref()
else {
+ continue;
+ };
+ let row_range = RowRange::new(global_meta.row_range_start,
global_meta.row_range_end);
+ coverage_by_field
+ .entry(global_meta.index_field_id)
+ .or_default()
+ .push(row_range.clone());
+ if let Some(extra_field_ids) = global_meta.extra_field_ids.as_ref() {
+ for extra_field_id in extra_field_ids {
+ coverage_by_field
+ .entry(*extra_field_id)
+ .or_default()
+ .push(row_range.clone());
+ }
+ }
+ }
+ unindexed_ranges_for_coverage(
+ &coverage_by_field,
+ field_ids,
+ search_mode,
+ next_row_id,
+ data_ranges,
+ )
+}
+
/// Index for row ranges. Stores sorted, non-overlapping ranges and supports
/// efficient intersection queries via binary search.
///
@@ -550,22 +758,43 @@ fn lower_bound(sorted: &[i64], target: i64) -> usize {
/// This is the main entry point for the table scan integration.
///
/// Returns `None` if global index is not available or predicates can't be
evaluated.
+pub(crate) struct GlobalIndexEvaluation<'a> {
+ pub(crate) file_io: &'a FileIO,
+ pub(crate) table_path: &'a str,
+ pub(crate) index_entries: &'a [IndexManifestEntry],
+ pub(crate) predicates: &'a [Predicate],
+ pub(crate) schema_fields: &'a [DataField],
+ pub(crate) search_mode: GlobalIndexSearchMode,
+ pub(crate) next_row_id: Option<i64>,
+ pub(crate) data_ranges: &'a [RowRange],
+}
+
pub(crate) async fn evaluate_global_index(
- file_io: &FileIO,
- table_path: &str,
- index_entries: &[IndexManifestEntry],
- predicates: &[Predicate],
- schema_fields: &[DataField],
+ evaluation: GlobalIndexEvaluation<'_>,
) -> Result<Option<Vec<RowRange>>> {
- let scanner =
- match GlobalIndexScanner::create(file_io, table_path, index_entries,
schema_fields) {
- Some(s) => s,
- None => return Ok(None),
- };
-
- let combined = Predicate::and(predicates.to_vec());
-
- scanner.evaluate(&combined).await
+ let scanner = match GlobalIndexScanner::create(
+ evaluation.file_io,
+ evaluation.table_path,
+ evaluation.index_entries,
+ evaluation.schema_fields,
+ ) {
+ Some(s) => s,
+ None => return Ok(None),
+ };
+
+ let combined = Predicate::and(evaluation.predicates.to_vec());
+
+ let mut row_ranges = match scanner.evaluate(&combined).await? {
+ Some(row_ranges) => row_ranges,
+ None => return Ok(None),
+ };
+ row_ranges.extend(scanner.unindexed_ranges(
+ &combined,
+ evaluation.search_mode,
+ evaluation.next_row_id,
+ evaluation.data_ranges,
+ )?);
+ Ok(Some(super::merge_row_ranges(row_ranges)))
}
#[cfg(test)]
@@ -741,6 +970,176 @@ mod tests {
)]
}
+ async fn evaluate_global_index_fast(
+ file_io: &FileIO,
+ table_path: &str,
+ entries: &[IndexManifestEntry],
+ predicates: &[Predicate],
+ fields: &[DataField],
+ ) -> Result<Option<Vec<RowRange>>> {
+ super::evaluate_global_index(super::GlobalIndexEvaluation {
+ file_io,
+ table_path,
+ index_entries: entries,
+ predicates,
+ schema_fields: fields,
+ search_mode: GlobalIndexSearchMode::Fast,
+ next_row_id: None,
+ data_ranges: &[],
+ })
+ .await
+ }
+
+ fn two_field_schema_fields() -> Vec<DataField> {
+ vec![
+ DataField::new(
+ 1,
+ "id".to_string(),
+ DataType::Int(crate::spec::IntType::new()),
+ ),
+ DataField::new(
+ 2,
+ "value".to_string(),
+ DataType::Int(crate::spec::IntType::new()),
+ ),
+ ]
+ }
+
+ fn int_eq(column: &str, index: usize, value: i32) -> Predicate {
+ Predicate::Leaf {
+ column: column.to_string(),
+ index,
+ data_type: DataType::Int(crate::spec::IntType::new()),
+ op: PredicateOperator::Eq,
+ literals: vec![Datum::Int(value)],
+ }
+ }
+
+ #[test]
+ fn test_unindexed_ranges_fast_mode_empty() {
+ let file_io = crate::io::FileIOBuilder::new("memory").build().unwrap();
+ let meta = BTreeIndexMeta::new(None, None, false);
+ let entries = vec![make_global_index_entry("idx", 1, 0, 49, &meta)];
+ let fields = int_schema_fields();
+ let scanner =
+ GlobalIndexScanner::create(&file_io, "memory:/t", &entries,
&fields).expect("scanner");
+
+ let ranges = scanner
+ .unindexed_ranges(
+ &int_eq("id", 0, 7),
+ GlobalIndexSearchMode::Fast,
+ Some(100),
+ &[RowRange::new(50, 99)],
+ )
+ .unwrap();
+ assert!(ranges.is_empty());
+ }
+
+ #[test]
+ fn test_unindexed_ranges_full_uses_snapshot_next_row_id() {
+ let file_io = crate::io::FileIOBuilder::new("memory").build().unwrap();
+ let meta = BTreeIndexMeta::new(None, None, false);
+ let entries = vec![make_global_index_entry("idx", 1, 0, 49, &meta)];
+ let fields = int_schema_fields();
+ let scanner =
+ GlobalIndexScanner::create(&file_io, "memory:/t", &entries,
&fields).expect("scanner");
+
+ let ranges = scanner
+ .unindexed_ranges(
+ &int_eq("id", 0, 7),
+ GlobalIndexSearchMode::Full,
+ Some(100),
+ &[],
+ )
+ .unwrap();
+ assert_eq!(ranges, vec![RowRange::new(50, 99)]);
+ }
+
+ #[test]
+ fn test_unindexed_ranges_detail_uses_data_file_ranges() {
+ let file_io = crate::io::FileIOBuilder::new("memory").build().unwrap();
+ let meta = BTreeIndexMeta::new(None, None, false);
+ let entries = vec![make_global_index_entry("idx", 1, 0, 49, &meta)];
+ let fields = int_schema_fields();
+ let scanner =
+ GlobalIndexScanner::create(&file_io, "memory:/t", &entries,
&fields).expect("scanner");
+
+ let ranges = scanner
+ .unindexed_ranges(
+ &int_eq("id", 0, 7),
+ GlobalIndexSearchMode::Detail,
+ Some(100),
+ &[
+ RowRange::new(0, 10),
+ RowRange::new(40, 60),
+ RowRange::new(80, 90),
+ ],
+ )
+ .unwrap();
+ assert_eq!(ranges, vec![RowRange::new(50, 60), RowRange::new(80, 90)]);
+ }
+
+ #[test]
+ fn test_unindexed_ranges_uses_all_predicate_field_coverage() {
+ let file_io = crate::io::FileIOBuilder::new("memory").build().unwrap();
+ let meta = BTreeIndexMeta::new(None, None, false);
+ let entries = vec![
+ make_global_index_entry("idx_id", 1, 0, 49, &meta),
+ make_global_index_entry("idx_value", 2, 0, 99, &meta),
+ ];
+ let fields = two_field_schema_fields();
+ let scanner =
+ GlobalIndexScanner::create(&file_io, "memory:/t", &entries,
&fields).expect("scanner");
+ let predicate = Predicate::and(vec![int_eq("id", 0, 7),
int_eq("value", 1, 8)]);
+
+ let ranges = scanner
+ .unindexed_ranges(&predicate, GlobalIndexSearchMode::Full,
Some(100), &[])
+ .unwrap();
+ assert_eq!(ranges, vec![RowRange::new(50, 99)]);
+ }
+
+ #[test]
+ fn test_unindexed_ranges_missing_field_coverage_reads_all_data_ranges() {
+ let file_io = crate::io::FileIOBuilder::new("memory").build().unwrap();
+ let meta = BTreeIndexMeta::new(None, None, false);
+ let entries = vec![make_global_index_entry("idx_id", 1, 0, 49, &meta)];
+ let fields = two_field_schema_fields();
+ let scanner =
+ GlobalIndexScanner::create(&file_io, "memory:/t", &entries,
&fields).expect("scanner");
+ let predicate = Predicate::and(vec![int_eq("id", 0, 7),
int_eq("value", 1, 8)]);
+
+ let ranges = scanner
+ .unindexed_ranges(&predicate, GlobalIndexSearchMode::Full,
Some(100), &[])
+ .unwrap();
+ assert_eq!(ranges, vec![RowRange::new(0, 99)]);
+ }
+
+ #[test]
+ fn test_unindexed_ranges_counts_extra_field_coverage() {
+ let file_io = crate::io::FileIOBuilder::new("memory").build().unwrap();
+ let meta = BTreeIndexMeta::new(None, None, false);
+ let mut entry = make_global_index_entry("idx_id_value", 1, 0, 99,
&meta);
+ entry
+ .index_file
+ .global_index_meta
+ .as_mut()
+ .unwrap()
+ .extra_field_ids = Some(vec![2]);
+ let fields = two_field_schema_fields();
+ let scanner =
+ GlobalIndexScanner::create(&file_io, "memory:/t", &[entry],
&fields).expect("scanner");
+
+ let ranges = scanner
+ .unindexed_ranges(
+ &int_eq("value", 1, 8),
+ GlobalIndexSearchMode::Full,
+ Some(100),
+ &[],
+ )
+ .unwrap();
+ assert!(ranges.is_empty());
+ }
+
#[tokio::test]
async fn test_evaluate_global_index_eq() {
let (file_io, table_path, file_name, _tmp) =
@@ -758,13 +1157,75 @@ mod tests {
literals: vec![Datum::Int(50)],
}];
- let result = evaluate_global_index(&file_io, &table_path, &entries,
&predicates, &fields)
- .await
- .unwrap();
+ let result =
+ evaluate_global_index_fast(&file_io, &table_path, &entries,
&predicates, &fields)
+ .await
+ .unwrap();
let ranges = result.unwrap();
assert_eq!(ranges, vec![RowRange::new(25, 25)]);
}
+ #[tokio::test]
+ async fn test_evaluate_global_index_full_mode_includes_unindexed_tail() {
+ let (file_io, table_path, file_name, _tmp) =
+ setup_testdata_table("btree_int_100_no_compress.bin");
+ let meta = BTreeIndexMeta::new(Some(le_int_key(0)),
Some(le_int_key(198)), false);
+ let entries = vec![make_global_index_entry(&file_name, 1, 0, 99,
&meta)];
+ let fields = int_schema_fields();
+ let predicates = vec![int_eq("id", 0, 50)];
+
+ let result = super::evaluate_global_index(super::GlobalIndexEvaluation
{
+ file_io: &file_io,
+ table_path: &table_path,
+ index_entries: &entries,
+ predicates: &predicates,
+ schema_fields: &fields,
+ search_mode: GlobalIndexSearchMode::Full,
+ next_row_id: Some(150),
+ data_ranges: &[],
+ })
+ .await
+ .unwrap();
+
+ assert_eq!(
+ result.unwrap(),
+ vec![RowRange::new(25, 25), RowRange::new(100, 149)]
+ );
+ }
+
+ #[tokio::test]
+ async fn test_evaluate_global_index_detail_mode_uses_data_ranges() {
+ let (file_io, table_path, file_name, _tmp) =
+ setup_testdata_table("btree_int_100_no_compress.bin");
+ let meta = BTreeIndexMeta::new(Some(le_int_key(0)),
Some(le_int_key(198)), false);
+ let entries = vec![make_global_index_entry(&file_name, 1, 0, 99,
&meta)];
+ let fields = int_schema_fields();
+ let predicates = vec![int_eq("id", 0, 50)];
+
+ let data_ranges = [RowRange::new(90, 120), RowRange::new(140, 145)];
+ let result = super::evaluate_global_index(super::GlobalIndexEvaluation
{
+ file_io: &file_io,
+ table_path: &table_path,
+ index_entries: &entries,
+ predicates: &predicates,
+ schema_fields: &fields,
+ search_mode: GlobalIndexSearchMode::Detail,
+ next_row_id: Some(150),
+ data_ranges: &data_ranges,
+ })
+ .await
+ .unwrap();
+
+ assert_eq!(
+ result.unwrap(),
+ vec![
+ RowRange::new(25, 25),
+ RowRange::new(100, 120),
+ RowRange::new(140, 145),
+ ]
+ );
+ }
+
#[tokio::test]
async fn test_evaluate_global_index_range() {
let (file_io, table_path, file_name, _tmp) =
@@ -791,9 +1252,10 @@ mod tests {
},
];
- let result = evaluate_global_index(&file_io, &table_path, &entries,
&predicates, &fields)
- .await
- .unwrap();
+ let result =
+ evaluate_global_index_fast(&file_io, &table_path, &entries,
&predicates, &fields)
+ .await
+ .unwrap();
let ranges = result.unwrap();
assert_eq!(ranges, vec![RowRange::new(5, 10)]);
}
@@ -815,9 +1277,10 @@ mod tests {
literals: vec![Datum::Int(0), Datum::Int(50), Datum::Int(198)],
}];
- let result = evaluate_global_index(&file_io, &table_path, &entries,
&predicates, &fields)
- .await
- .unwrap();
+ let result =
+ evaluate_global_index_fast(&file_io, &table_path, &entries,
&predicates, &fields)
+ .await
+ .unwrap();
let ranges = result.unwrap();
assert_eq!(
ranges,
@@ -846,9 +1309,10 @@ mod tests {
literals: vec![Datum::Int(999)],
}];
- let result = evaluate_global_index(&file_io, &table_path, &entries,
&predicates, &fields)
- .await
- .unwrap();
+ let result =
+ evaluate_global_index_fast(&file_io, &table_path, &entries,
&predicates, &fields)
+ .await
+ .unwrap();
let ranges = result.unwrap();
assert!(ranges.is_empty());
}
@@ -871,9 +1335,10 @@ mod tests {
literals: vec![Datum::Int(50)],
}];
- let result = evaluate_global_index(&file_io, &table_path, &entries,
&predicates, &fields)
- .await
- .unwrap();
+ let result =
+ evaluate_global_index_fast(&file_io, &table_path, &entries,
&predicates, &fields)
+ .await
+ .unwrap();
let ranges = result.unwrap();
assert_eq!(ranges, vec![RowRange::new(1025, 1025)]);
}
@@ -895,9 +1360,10 @@ mod tests {
literals: vec![Datum::Int(50)],
}];
- let result = evaluate_global_index(&file_io, &table_path, &entries,
&predicates, &fields)
- .await
- .unwrap();
+ let result =
+ evaluate_global_index_fast(&file_io, &table_path, &entries,
&predicates, &fields)
+ .await
+ .unwrap();
assert!(result.is_none());
}
@@ -972,9 +1438,10 @@ mod tests {
},
];
- let result = evaluate_global_index(&file_io, &table_path, &entries,
&predicates, &fields)
- .await
- .unwrap();
+ let result =
+ evaluate_global_index_fast(&file_io, &table_path, &entries,
&predicates, &fields)
+ .await
+ .unwrap();
let ranges = result.unwrap();
assert_eq!(ranges, vec![RowRange::new(22, 26)]);
}
@@ -1015,9 +1482,10 @@ mod tests {
},
])];
- let result = evaluate_global_index(&file_io, &table_path, &entries,
&predicates, &fields)
- .await
- .unwrap();
+ let result =
+ evaluate_global_index_fast(&file_io, &table_path, &entries,
&predicates, &fields)
+ .await
+ .unwrap();
let ranges = result.unwrap();
assert!(
ranges.is_empty(),
diff --git a/crates/paimon/src/table/table_scan.rs
b/crates/paimon/src/table/table_scan.rs
index 0c9dc95..409f40f 100644
--- a/crates/paimon/src/table/table_scan.rs
+++ b/crates/paimon/src/table/table_scan.rs
@@ -31,9 +31,10 @@ use super::Table;
use crate::io::FileIO;
use crate::spec::{
avro::SharedSchemaCache, bucket_dir_name, BinaryRow, BucketFunctionType,
CoreOptions,
- DataField, DataFileMeta, FileKind, IndexManifest, ManifestEntry,
PartitionComputer, Predicate,
- Snapshot, ROW_ID_FIELD_ID, ROW_ID_FIELD_NAME, SEQUENCE_NUMBER_FIELD_ID,
- SEQUENCE_NUMBER_FIELD_NAME, VALUE_KIND_FIELD_ID, VALUE_KIND_FIELD_NAME,
+ DataField, DataFileMeta, FileKind, GlobalIndexSearchMode, IndexManifest,
ManifestEntry,
+ PartitionComputer, Predicate, Snapshot, ROW_ID_FIELD_ID, ROW_ID_FIELD_NAME,
+ SEQUENCE_NUMBER_FIELD_ID, SEQUENCE_NUMBER_FIELD_NAME, VALUE_KIND_FIELD_ID,
+ VALUE_KIND_FIELD_NAME,
};
use crate::table::bin_pack::split_for_batch;
use crate::table::merge_tree_split_generator::{
@@ -286,6 +287,20 @@ pub(super) fn can_push_down_limit_hint_for_scan(
data_predicates.is_empty() && row_ranges.is_none()
}
+type BucketDataFileGroups = HashMap<(Vec<u8>, i32), (i32, Vec<DataFileMeta>)>;
+
+fn global_index_detail_data_ranges(groups: &BucketDataFileGroups) ->
Vec<RowRange> {
+ let mut ranges = Vec::new();
+ for (_, data_files) in groups.values() {
+ for file in data_files {
+ if let Some((from, to)) = file.row_id_range() {
+ ranges.push(RowRange::new(from, to));
+ }
+ }
+ }
+ merge_row_ranges(ranges)
+}
+
fn should_skip_level_zero_for_scan(
scan_all_files: bool,
has_primary_keys: bool,
@@ -802,8 +817,7 @@ impl<'a> TableScan<'a> {
}
// Group by (partition, bucket), decomposing entries to avoid cloning
partition.
- let mut groups: HashMap<(Vec<u8>, i32), (i32, Vec<DataFileMeta>)> =
- HashMap::with_capacity(entries.len());
+ let mut groups: BucketDataFileGroups =
HashMap::with_capacity(entries.len());
for e in entries {
let (partition, bucket, total_buckets, file) = e.into_parts();
let entry = groups
@@ -812,6 +826,23 @@ impl<'a> TableScan<'a> {
entry.1.push(file);
}
+ let global_index_search_mode = if data_evolution_enabled
+ && core_options.global_index_enabled()
+ && !self.data_predicates.is_empty()
+ {
+ Some(core_options.global_index_search_mode()?)
+ } else {
+ None
+ };
+ let global_index_detail_data_ranges = if matches!(
+ global_index_search_mode,
+ Some(GlobalIndexSearchMode::Detail)
+ ) {
+ global_index_detail_data_ranges(&groups)
+ } else {
+ Vec::new()
+ };
+
let snapshot_id = snapshot.id();
let base_path = table_path.trim_end_matches('/');
let mut splits = Vec::with_capacity(groups.len());
@@ -856,16 +887,18 @@ impl<'a> TableScan<'a> {
// Use pushed-down row_ranges first; otherwise try global
index.
let row_ranges = if self.row_ranges.is_some() {
self.row_ranges.clone()
- } else if data_evolution_enabled
- && core_options.global_index_enabled()
- && !self.data_predicates.is_empty()
- {
+ } else if let Some(search_mode) = global_index_search_mode {
super::global_index_scanner::evaluate_global_index(
- file_io,
- base_path,
- &index_entries,
- &self.data_predicates,
- self.table.schema().fields(),
+ super::global_index_scanner::GlobalIndexEvaluation {
+ file_io,
+ table_path: base_path,
+ index_entries: &index_entries,
+ predicates: &self.data_predicates,
+ schema_fields: self.table.schema().fields(),
+ search_mode,
+ next_row_id: snapshot.next_row_id(),
+ data_ranges: &global_index_detail_data_ranges,
+ },
)
.await?
} else {
diff --git a/crates/paimon/src/table/vector_search_builder.rs
b/crates/paimon/src/table/vector_search_builder.rs
index 48271aa..1ebd422 100644
--- a/crates/paimon/src/table/vector_search_builder.rs
+++ b/crates/paimon/src/table/vector_search_builder.rs
@@ -15,15 +15,24 @@
// specific language governing permissions and limitations
// under the License.
-use crate::lumina::is_lumina_index_type;
+use crate::io::FileIO;
use crate::lumina::reader::LuminaVectorGlobalIndexReader;
-use crate::spec::{DataField, FileKind, IndexManifest};
+use crate::lumina::{is_lumina_index_type, LuminaIndexMeta, LuminaVectorMetric};
+use crate::spec::{
+ CoreOptions, DataField, FileKind, GlobalIndexSearchMode, IndexFileMeta,
IndexManifest,
+ IndexManifestEntry, ROW_ID_FIELD_NAME,
+};
+use
crate::table::global_index_scanner::unindexed_ranges_for_global_index_entries;
use crate::table::snapshot_manager::SnapshotManager;
-use crate::table::{find_field_id_by_name, RowRange, Table};
+use crate::table::{find_field_id_by_name, merge_row_ranges, RowRange, Table};
use crate::vector_search::{GlobalIndexIOMeta, SearchResult, VectorSearch};
use crate::vindex::is_vindex_index_type;
use crate::vindex::reader::VindexVectorGlobalIndexReader;
-use std::collections::HashMap;
+use arrow_array::{Array, FixedSizeListArray, Float32Array, Int64Array,
ListArray, RecordBatch};
+use futures::TryStreamExt;
+use paimon_vindex_core::distance::MetricType;
+use paimon_vindex_core::index::VectorIndexReader as VIndexReader;
+use std::collections::{HashMap, HashSet};
use std::io::Cursor;
const INDEX_DIR: &str = "index";
@@ -115,41 +124,53 @@ impl<'a> VectorSearchBuilder<'a> {
None => return Ok(Vec::new()),
};
- let index_manifest_name = match snapshot.index_manifest() {
- Some(name) => name.to_string(),
- None => return Ok(Vec::new()),
+ let index_entries = match snapshot.index_manifest() {
+ Some(index_manifest_name) => {
+ let manifest_path = format!(
+ "{}/manifest/{}",
+ self.table.location().trim_end_matches('/'),
+ index_manifest_name
+ );
+ IndexManifest::read(self.table.file_io(),
&manifest_path).await?
+ }
+ None => Vec::new(),
};
- let manifest_path = format!(
- "{}/manifest/{}",
- self.table.location().trim_end_matches('/'),
- index_manifest_name
- );
- let index_entries = IndexManifest::read(self.table.file_io(),
&manifest_path).await?;
-
evaluate_vector_search(
- self.table.file_io(),
- self.table.location(),
- self.table.schema().options(),
+ VectorSearchEvaluation {
+ table: Some(self.table),
+ file_io: self.table.file_io(),
+ table_path: self.table.location(),
+ table_options: self.table.schema().options(),
+ schema_fields: self.table.schema().fields(),
+ next_row_id: snapshot.next_row_id(),
+ },
&index_entries,
&vector_search,
- self.table.schema().fields(),
)
.await
}
}
+struct VectorSearchEvaluation<'a> {
+ table: Option<&'a Table>,
+ file_io: &'a FileIO,
+ table_path: &'a str,
+ table_options: &'a HashMap<String, String>,
+ schema_fields: &'a [DataField],
+ next_row_id: Option<i64>,
+}
+
async fn evaluate_vector_search(
- file_io: &crate::io::FileIO,
- table_path: &str,
- table_options: &HashMap<String, String>,
- index_entries: &[crate::spec::IndexManifestEntry],
+ evaluation: VectorSearchEvaluation<'_>,
+ index_entries: &[IndexManifestEntry],
vector_search: &VectorSearch,
- schema_fields: &[DataField],
) -> crate::Result<Vec<RowRange>> {
- let table_path = table_path.trim_end_matches('/');
+ let table_path = evaluation.table_path.trim_end_matches('/');
+ let search_mode =
CoreOptions::new(evaluation.table_options).global_index_search_mode()?;
- let field_id = match find_field_id_by_name(schema_fields,
&vector_search.field_name) {
+ let field_id = match find_field_id_by_name(evaluation.schema_fields,
&vector_search.field_name)
+ {
Some(id) => id,
None => return Ok(Vec::new()),
};
@@ -166,69 +187,473 @@ async fn evaluate_vector_search(
})
.collect();
- if vector_entries.is_empty() {
+ if vector_entries.is_empty() && search_mode == GlobalIndexSearchMode::Fast
{
return Ok(Vec::new());
}
- let futures: Vec<_> = vector_entries
- .into_iter()
- .map(|entry| {
- let global_meta =
entry.index_file.global_index_meta.as_ref().unwrap();
- let backend =
VectorIndexBackend::from_index_type(&entry.index_file.index_type)
- .expect("filtered vector index type");
- let path = format!("{table_path}/{INDEX_DIR}/{}",
entry.index_file.file_name);
- let file_name = entry.index_file.file_name.clone();
- let file_size = entry.index_file.file_size as u64;
- let index_meta_bytes =
global_meta.index_meta.clone().unwrap_or_default();
- let row_range_start = global_meta.row_range_start;
- let vector_search_clone = vector_search.clone();
- let options = table_options.clone();
- let input = file_io.new_input(&path);
- async move {
- let input = input?;
+ let mut merged = SearchResult::empty();
+ if !vector_entries.is_empty() {
+ let futures: Vec<_> = vector_entries
+ .into_iter()
+ .map(|entry| {
+ let global_meta =
entry.index_file.global_index_meta.as_ref().unwrap();
+ let backend =
VectorIndexBackend::from_index_type(&entry.index_file.index_type)
+ .expect("filtered vector index type");
+ let path = format!("{table_path}/{INDEX_DIR}/{}",
entry.index_file.file_name);
+ let file_name = entry.index_file.file_name.clone();
+ let file_size = entry.index_file.file_size as u64;
+ let index_meta_bytes =
global_meta.index_meta.clone().unwrap_or_default();
+ let row_range_start = global_meta.row_range_start;
+ let vector_search_clone = vector_search.clone();
+ let options = evaluation.table_options.clone();
+ let input = evaluation.file_io.new_input(&path);
+ async move {
+ let input = input?;
+ let bytes = input.read().await.map_err(|e|
crate::Error::DataInvalid {
+ message: format!(
+ "Failed to read {} index file '{}': {}",
+ backend.error_name(),
+ file_name,
+ e
+ ),
+ source: None,
+ })?;
+
+ let io_meta =
+ GlobalIndexIOMeta::new(file_name.clone(), file_size,
index_meta_bytes);
+ let data = bytes.to_vec();
+ let result = match backend {
+ VectorIndexBackend::Lumina => {
+ let mut reader =
LuminaVectorGlobalIndexReader::new(io_meta, options);
+ reader.visit_vector_search(&vector_search_clone,
|_| {
+ Ok(Cursor::new(data))
+ })?
+ }
+ VectorIndexBackend::Vindex => {
+ let mut reader =
VindexVectorGlobalIndexReader::new(io_meta, options);
+ reader.visit_vector_search(&vector_search_clone,
|_| {
+ Ok(Cursor::new(data))
+ })?
+ }
+ };
+
+ match result {
+ Some(scored_map) => Ok::<_, crate::Error>(
+
SearchResult::from_scored_map(scored_map).offset(row_range_start),
+ ),
+ None => Ok(SearchResult::empty()),
+ }
+ }
+ })
+ .collect();
+
+ let results = futures::future::try_join_all(futures).await?;
+ for r in &results {
+ merged = merged.or(r);
+ }
+ }
+
+ if search_mode != GlobalIndexSearchMode::Fast {
+ let detail_ranges = if search_mode == GlobalIndexSearchMode::Detail {
+ let table = evaluation.table.ok_or_else(||
crate::Error::DataInvalid {
+ message: "Vector raw search in detail mode requires table
context".to_string(),
+ source: None,
+ })?;
+ detail_data_ranges_for_table(table).await?
+ } else {
+ Vec::new()
+ };
+ let field_ids = HashSet::from([field_id]);
+ let raw_ranges = unindexed_ranges_for_global_index_entries(
+ index_entries,
+ &field_ids,
+ search_mode,
+ evaluation.next_row_id,
+ &detail_ranges,
+ is_vector_global_index_file,
+ );
+ if !raw_ranges.is_empty() {
+ let table = evaluation.table.ok_or_else(||
crate::Error::DataInvalid {
+ message: "Vector raw search requires table
context".to_string(),
+ source: None,
+ })?;
+ let metric = resolve_raw_vector_metric(
+ evaluation.file_io,
+ table_path,
+ evaluation.table_options,
+ index_entries,
+ field_id,
+ &vector_search.field_name,
+ )
+ .await?;
+ let raw_result =
+ read_raw_vector_search(table, vector_search, &raw_ranges,
metric).await?;
+ merged = merged.or(&raw_result);
+ }
+ }
+
+ merged.top_k(vector_search.limit).to_row_ranges()
+}
+
+fn is_vector_global_index_file(index_file: &IndexFileMeta) -> bool {
+ VectorIndexBackend::from_index_type(&index_file.index_type).is_some()
+}
+
+async fn detail_data_ranges_for_table(table: &Table) ->
crate::Result<Vec<RowRange>> {
+ let plan = table
+ .new_read_builder()
+ .new_scan()
+ .with_scan_all_files()
+ .plan()
+ .await?;
+ let mut ranges = Vec::new();
+ for split in plan.splits() {
+ for file in split.data_files() {
+ if let Some((from, to)) = file.row_id_range() {
+ ranges.push(RowRange::new(from, to));
+ }
+ }
+ }
+ Ok(merge_row_ranges(ranges))
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+enum RawVectorMetric {
+ L2,
+ Cosine,
+ InnerProduct,
+}
+
+impl RawVectorMetric {
+ fn parse(value: &str) -> crate::Result<Self> {
+ Self::parse_normalized(&normalize_metric(value)).ok_or_else(||
crate::Error::DataInvalid {
+ message: format!("Unknown vector search metric: {value}"),
+ source: None,
+ })
+ }
+
+ fn parse_normalized(value: &str) -> Option<Self> {
+ match value {
+ "l2" => Some(Self::L2),
+ "cosine" => Some(Self::Cosine),
+ "inner_product" => Some(Self::InnerProduct),
+ _ => None,
+ }
+ }
+
+ fn from_lumina(metric: LuminaVectorMetric) -> Self {
+ match metric {
+ LuminaVectorMetric::L2 => Self::L2,
+ LuminaVectorMetric::Cosine => Self::Cosine,
+ LuminaVectorMetric::InnerProduct => Self::InnerProduct,
+ }
+ }
+
+ fn from_vindex(metric: MetricType) -> Self {
+ match metric {
+ MetricType::L2 => Self::L2,
+ MetricType::Cosine => Self::Cosine,
+ MetricType::InnerProduct => Self::InnerProduct,
+ }
+ }
+}
+
+fn normalize_metric(metric: &str) -> String {
+ metric.to_ascii_lowercase().replace('-', "_")
+}
+
+async fn resolve_raw_vector_metric(
+ file_io: &FileIO,
+ table_path: &str,
+ table_options: &HashMap<String, String>,
+ index_entries: &[IndexManifestEntry],
+ field_id: i32,
+ field_name: &str,
+) -> crate::Result<RawVectorMetric> {
+ for entry in index_entries {
+ if entry.kind != FileKind::Add {
+ continue;
+ }
+ let Some(global_meta) = entry.index_file.global_index_meta.as_ref()
else {
+ continue;
+ };
+ if global_meta.index_field_id != field_id {
+ continue;
+ }
+ let Some(backend) =
VectorIndexBackend::from_index_type(&entry.index_file.index_type)
+ else {
+ continue;
+ };
+ match backend {
+ VectorIndexBackend::Lumina => {
+ if let Some(index_meta) = global_meta.index_meta.as_ref() {
+ if !index_meta.is_empty() {
+ let metric =
LuminaIndexMeta::deserialize(index_meta)?.metric()?;
+ return Ok(RawVectorMetric::from_lumina(metric));
+ }
+ }
+ }
+ VectorIndexBackend::Vindex => {
+ let path = format!("{table_path}/{INDEX_DIR}/{}",
entry.index_file.file_name);
+ let input = file_io.new_input(&path)?;
let bytes = input.read().await.map_err(|e|
crate::Error::DataInvalid {
message: format!(
- "Failed to read {} index file '{}': {}",
- backend.error_name(),
- file_name,
- e
+ "Failed to read vindex index file '{}' for raw search
metric: {}",
+ entry.index_file.file_name, e
),
source: None,
})?;
-
- let io_meta =
- GlobalIndexIOMeta::new(file_name.clone(), file_size,
index_meta_bytes);
- let data = bytes.to_vec();
- let result = match backend {
- VectorIndexBackend::Lumina => {
- let mut reader =
LuminaVectorGlobalIndexReader::new(io_meta, options);
- reader
- .visit_vector_search(&vector_search_clone, |_|
Ok(Cursor::new(data)))?
+ let reader =
VIndexReader::open(Cursor::new(bytes.to_vec())).map_err(|e| {
+ crate::Error::DataInvalid {
+ message: format!(
+ "Failed to open paimon-vindex-core reader for raw
search metric: {}",
+ e
+ ),
+ source: Some(Box::new(e)),
}
- VectorIndexBackend::Vindex => {
- let mut reader =
VindexVectorGlobalIndexReader::new(io_meta, options);
- reader
- .visit_vector_search(&vector_search_clone, |_|
Ok(Cursor::new(data)))?
- }
- };
+ })?;
+ return
Ok(RawVectorMetric::from_vindex(reader.metadata().metric));
+ }
+ }
+ }
- match result {
- Some(scored_map) => Ok::<_, crate::Error>(
-
SearchResult::from_scored_map(scored_map).offset(row_range_start),
- ),
- None => Ok(SearchResult::empty()),
- }
+ configured_raw_vector_metric(table_options, field_name)
+}
+
+fn configured_raw_vector_metric(
+ options: &HashMap<String, String>,
+ field_name: &str,
+) -> crate::Result<RawVectorMetric> {
+ let direct_keys = [
+ format!("fields.{field_name}.distance.metric"),
+ format!("fields.{field_name}.metric"),
+ "test.vector.metric".to_string(),
+ "lumina.distance.metric".to_string(),
+ "distance.metric".to_string(),
+ "metric".to_string(),
+ ];
+ for key in direct_keys {
+ if let Some(value) = options.get(&key) {
+ return RawVectorMetric::parse(value);
+ }
+ }
+
+ let mut inferred = None;
+ for (key, value) in options {
+ if !(key.ends_with(".distance.metric") || key.ends_with(".metric")) {
+ continue;
+ }
+ let normalized = normalize_metric(value);
+ let Some(metric) = RawVectorMetric::parse_normalized(&normalized) else
{
+ continue;
+ };
+ if let Some(existing) = inferred {
+ if existing != metric {
+ return Ok(RawVectorMetric::L2);
}
- })
- .collect();
+ } else {
+ inferred = Some(metric);
+ }
+ }
+ Ok(inferred.unwrap_or(RawVectorMetric::L2))
+}
- let results = futures::future::try_join_all(futures).await?;
- let mut merged = SearchResult::empty();
- for r in &results {
- merged = merged.or(r);
+async fn read_raw_vector_search(
+ table: &Table,
+ vector_search: &VectorSearch,
+ raw_ranges: &[RowRange],
+ metric: RawVectorMetric,
+) -> crate::Result<SearchResult> {
+ if raw_ranges.is_empty() {
+ return Ok(SearchResult::empty());
}
- merged.top_k(vector_search.limit).to_row_ranges()
+ let mut read_builder = table.new_read_builder();
+ read_builder
+ .with_projection(&[vector_search.field_name.as_str(),
ROW_ID_FIELD_NAME])
+ .with_row_ranges(raw_ranges.to_vec());
+ let plan = read_builder.new_scan().plan().await?;
+ if plan.splits().is_empty() {
+ return Ok(SearchResult::empty());
+ }
+ let read = read_builder.new_read()?;
+ let mut stream = read.to_arrow(plan.splits())?;
+
+ let mut row_ids = Vec::new();
+ let mut scores = Vec::new();
+ while let Some(batch) = stream.try_next().await? {
+ collect_raw_vector_batch(&batch, vector_search, metric, &mut row_ids,
&mut scores)?;
+ }
+
+ Ok(SearchResult::new(row_ids, scores).top_k(vector_search.limit))
+}
+
+fn collect_raw_vector_batch(
+ batch: &RecordBatch,
+ vector_search: &VectorSearch,
+ metric: RawVectorMetric,
+ row_ids_out: &mut Vec<u64>,
+ scores_out: &mut Vec<f32>,
+) -> crate::Result<()> {
+ let vector_index = batch
+ .schema()
+ .index_of(&vector_search.field_name)
+ .map_err(|e| crate::Error::DataInvalid {
+ message: format!(
+ "Vector column '{}' not found in raw search batch: {}",
+ vector_search.field_name, e
+ ),
+ source: None,
+ })?;
+ let row_id_index =
+ batch
+ .schema()
+ .index_of(ROW_ID_FIELD_NAME)
+ .map_err(|e| crate::Error::DataInvalid {
+ message: format!("_ROW_ID column not found in raw search
batch: {e}"),
+ source: None,
+ })?;
+
+ let row_ids = batch
+ .column(row_id_index)
+ .as_any()
+ .downcast_ref::<Int64Array>()
+ .ok_or_else(|| crate::Error::DataInvalid {
+ message: "Vector raw search requires non-null Int64
_ROW_ID".to_string(),
+ source: None,
+ })?;
+
+ let column = batch.column(vector_index);
+ enum VectorLayout<'a> {
+ List(&'a ListArray),
+ Fixed(&'a FixedSizeListArray),
+ }
+ let layout = if let Some(a) = column.as_any().downcast_ref::<ListArray>() {
+ VectorLayout::List(a)
+ } else if let Some(a) =
column.as_any().downcast_ref::<FixedSizeListArray>() {
+ VectorLayout::Fixed(a)
+ } else {
+ return Err(crate::Error::DataInvalid {
+ message: "Vector raw search requires Arrow List<Float32> or
FixedSizeList<Float32>"
+ .to_string(),
+ source: None,
+ });
+ };
+ let values = match layout {
+ VectorLayout::List(a) => a.values(),
+ VectorLayout::Fixed(a) => a.values(),
+ }
+ .as_any()
+ .downcast_ref::<Float32Array>()
+ .ok_or_else(|| crate::Error::DataInvalid {
+ message: "Vector raw search requires Float32 vector
elements".to_string(),
+ source: None,
+ })?;
+
+ for row in 0..batch.num_rows() {
+ if row_ids.is_null(row) {
+ return Err(crate::Error::DataInvalid {
+ message: "Vector raw search found null _ROW_ID".to_string(),
+ source: None,
+ });
+ }
+ let row_id = row_id_to_u64(row_ids.value(row))?;
+ if vector_search
+ .include_row_ids
+ .as_ref()
+ .is_some_and(|include_row_ids| !include_row_ids.contains(row_id))
+ {
+ continue;
+ }
+
+ let is_null = match layout {
+ VectorLayout::List(a) => a.is_null(row),
+ VectorLayout::Fixed(a) => a.is_null(row),
+ };
+ if is_null {
+ continue;
+ }
+
+ let (start, end) = match layout {
+ VectorLayout::List(a) => {
+ let offsets = a.value_offsets();
+ (offsets[row] as usize, offsets[row + 1] as usize)
+ }
+ VectorLayout::Fixed(a) => {
+ let len = a.value_length() as usize;
+ (row * len, (row + 1) * len)
+ }
+ };
+ if end - start != vector_search.vector.len() {
+ return Err(crate::Error::DataInvalid {
+ message: format!(
+ "Query vector dimension mismatch: raw row has {}, but
query has {}",
+ end - start,
+ vector_search.vector.len()
+ ),
+ source: None,
+ });
+ }
+
+ let mut stored = Vec::with_capacity(vector_search.vector.len());
+ for value_index in start..end {
+ if values.is_null(value_index) {
+ return Err(crate::Error::DataInvalid {
+ message: "Vector raw search found null vector
element".to_string(),
+ source: None,
+ });
+ }
+ stored.push(values.value(value_index));
+ }
+ row_ids_out.push(row_id);
+ scores_out.push(compute_raw_vector_score(
+ &vector_search.vector,
+ &stored,
+ metric,
+ ));
+ }
+
+ Ok(())
+}
+
+fn row_id_to_u64(row_id: i64) -> crate::Result<u64> {
+ u64::try_from(row_id).map_err(|_| crate::Error::DataInvalid {
+ message: format!("Negative _ROW_ID {row_id} cannot be used for global
index search"),
+ source: None,
+ })
+}
+
+fn compute_raw_vector_score(query: &[f32], stored: &[f32], metric:
RawVectorMetric) -> f32 {
+ match metric {
+ RawVectorMetric::L2 => {
+ let sum_sq = query
+ .iter()
+ .zip(stored.iter())
+ .map(|(q, s)| {
+ let diff = q - s;
+ diff * diff
+ })
+ .sum::<f32>();
+ 1.0 / (1.0 + sum_sq)
+ }
+ RawVectorMetric::Cosine => {
+ let mut dot = 0.0;
+ let mut norm_a = 0.0;
+ let mut norm_b = 0.0;
+ for (q, s) in query.iter().zip(stored.iter()) {
+ dot += q * s;
+ norm_a += q * q;
+ norm_b += s * s;
+ }
+ let denominator = norm_a.sqrt() * norm_b.sqrt();
+ if denominator == 0.0 {
+ 0.0
+ } else {
+ dot / denominator
+ }
+ }
+ RawVectorMetric::InnerProduct =>
query.iter().zip(stored.iter()).map(|(q, s)| q * s).sum(),
+ }
}
#[cfg(test)]
@@ -242,6 +667,22 @@ mod tests {
DataField::new(id, name.to_string(), DataType::Int(IntType::default()))
}
+ fn eval_context<'a>(
+ file_io: &'a FileIO,
+ options: &'a HashMap<String, String>,
+ fields: &'a [DataField],
+ next_row_id: Option<i64>,
+ ) -> VectorSearchEvaluation<'a> {
+ VectorSearchEvaluation {
+ table: None,
+ file_io,
+ table_path: "memory:///test_table",
+ table_options: options,
+ schema_fields: fields,
+ next_row_id,
+ }
+ }
+
#[test]
fn test_find_field_id_by_name() {
let fields = vec![make_field(1, "id"), make_field(2, "embedding")];
@@ -249,11 +690,50 @@ mod tests {
assert_eq!(find_field_id_by_name(&fields, "nonexistent"), None);
}
+ #[test]
+ fn test_raw_vector_score_matches_java_metric_semantics() {
+ let l2 = compute_raw_vector_score(&[1.0, 2.0], &[1.0, 4.0],
RawVectorMetric::L2);
+ assert!((l2 - 0.2).abs() < 1e-6);
+ assert_eq!(
+ compute_raw_vector_score(&[1.0, 2.0], &[3.0, 4.0],
RawVectorMetric::InnerProduct),
+ 11.0
+ );
+ let cosine = compute_raw_vector_score(&[1.0, 0.0], &[1.0, 1.0],
RawVectorMetric::Cosine);
+ assert!((cosine - std::f32::consts::FRAC_1_SQRT_2).abs() < 1e-6);
+ assert_eq!(
+ compute_raw_vector_score(&[0.0, 0.0], &[1.0, 1.0],
RawVectorMetric::Cosine),
+ 0.0
+ );
+ }
+
+ #[test]
+ fn test_configured_raw_vector_metric_precedence_and_conflict_default() {
+ let mut options = HashMap::new();
+ options.insert(
+ "fields.embedding.distance.metric".to_string(),
+ "inner-product".to_string(),
+ );
+ options.insert("metric".to_string(), "cosine".to_string());
+ assert_eq!(
+ configured_raw_vector_metric(&options, "embedding").unwrap(),
+ RawVectorMetric::InnerProduct
+ );
+
+ options.clear();
+ options.insert("foo.metric".to_string(), "cosine".to_string());
+ options.insert("bar.distance.metric".to_string(), "l2".to_string());
+ assert_eq!(
+ configured_raw_vector_metric(&options, "embedding").unwrap(),
+ RawVectorMetric::L2
+ );
+ }
+
#[tokio::test]
async fn test_evaluate_no_matching_entries() {
let file_io = crate::io::FileIOBuilder::new("memory").build().unwrap();
let fields = vec![make_field(1, "id"), make_field(2, "embedding")];
let vs = VectorSearch::new(vec![1.0, 2.0], 10,
"embedding".to_string()).unwrap();
+ let options = HashMap::new();
let entry = IndexManifestEntry {
kind: FileKind::Add,
@@ -271,12 +751,9 @@ mod tests {
};
let result = evaluate_vector_search(
- &file_io,
- "memory:///test_table",
- &HashMap::new(),
+ eval_context(&file_io, &options, &fields, None),
&[entry],
&vs,
- &fields,
)
.await
.unwrap();
@@ -288,27 +765,47 @@ mod tests {
let file_io = crate::io::FileIOBuilder::new("memory").build().unwrap();
let fields = vec![make_field(2, "embedding")];
let vs = VectorSearch::new(vec![1.0], 10,
"embedding".to_string()).unwrap();
+ let options = HashMap::new();
let entry = make_lumina_entry("test.idx", "btree", FileKind::Add, 2);
let result = evaluate_vector_search(
- &file_io,
- "memory:///test_table",
- &HashMap::new(),
+ eval_context(&file_io, &options, &fields, None),
&[entry],
&vs,
- &fields,
)
.await
.unwrap();
assert!(result.is_empty());
}
+ #[tokio::test]
+ async fn test_evaluate_full_mode_without_vector_entries_uses_raw_path() {
+ let file_io = crate::io::FileIOBuilder::new("memory").build().unwrap();
+ let fields = vec![make_field(2, "embedding")];
+ let vs = VectorSearch::new(vec![1.0], 10,
"embedding".to_string()).unwrap();
+ let options = HashMap::from([("global-index.search-mode".to_string(),
"full".to_string())]);
+
+ let err = evaluate_vector_search(
+ eval_context(&file_io, &options, &fields, Some(10)),
+ &[],
+ &vs,
+ )
+ .await
+ .unwrap_err();
+ assert!(
+ err.to_string()
+ .contains("Vector raw search requires table context"),
+ "unexpected error: {err}"
+ );
+ }
+
#[tokio::test]
async fn test_evaluate_no_matching_field() {
let file_io = crate::io::FileIOBuilder::new("memory").build().unwrap();
let fields = vec![make_field(1, "id")];
let vs = VectorSearch::new(vec![1.0], 10,
"embedding".to_string()).unwrap();
+ let options = HashMap::new();
let entry = make_lumina_entry(
"test.idx",
@@ -318,12 +815,9 @@ mod tests {
);
let result = evaluate_vector_search(
- &file_io,
- "memory:///test_table",
- &HashMap::new(),
+ eval_context(&file_io, &options, &fields, None),
&[entry],
&vs,
- &fields,
)
.await
.unwrap();
@@ -335,6 +829,7 @@ mod tests {
let file_io = crate::io::FileIOBuilder::new("memory").build().unwrap();
let fields = vec![make_field(2, "embedding")];
let vs = VectorSearch::new(vec![1.0], 10,
"embedding".to_string()).unwrap();
+ let options = HashMap::new();
let entry = make_lumina_entry(
"test.idx",
@@ -344,12 +839,9 @@ mod tests {
);
let result = evaluate_vector_search(
- &file_io,
- "memory:///test_table",
- &HashMap::new(),
+ eval_context(&file_io, &options, &fields, None),
&[entry],
&vs,
- &fields,
)
.await
.unwrap();
@@ -361,16 +853,14 @@ mod tests {
let file_io = crate::io::FileIOBuilder::new("memory").build().unwrap();
let fields = vec![make_field(2, "embedding")];
let vs = VectorSearch::new(vec![1.0], 10,
"embedding".to_string()).unwrap();
+ let options = HashMap::new();
let entry = make_lumina_entry("missing.idx", LUMINA_IDENTIFIER,
FileKind::Add, 2);
let err = evaluate_vector_search(
- &file_io,
- "memory:///test_table",
- &HashMap::new(),
+ eval_context(&file_io, &options, &fields, None),
&[entry],
&vs,
- &fields,
)
.await
.unwrap_err();
@@ -386,6 +876,7 @@ mod tests {
let file_io = crate::io::FileIOBuilder::new("memory").build().unwrap();
let fields = vec![make_field(2, "embedding")];
let vs = VectorSearch::new(vec![1.0], 10,
"embedding".to_string()).unwrap();
+ let options = HashMap::new();
let entry = make_lumina_entry(
"missing.idx",
@@ -395,12 +886,9 @@ mod tests {
);
let err = evaluate_vector_search(
- &file_io,
- "memory:///test_table",
- &HashMap::new(),
+ eval_context(&file_io, &options, &fields, None),
&[entry],
&vs,
- &fields,
)
.await
.unwrap_err();
@@ -416,16 +904,14 @@ mod tests {
let file_io = crate::io::FileIOBuilder::new("memory").build().unwrap();
let fields = vec![make_field(2, "embedding")];
let vs = VectorSearch::new(vec![1.0], 10,
"embedding".to_string()).unwrap();
+ let options = HashMap::new();
let entry = make_lumina_entry("missing.idx", IVF_FLAT_IDENTIFIER,
FileKind::Add, 2);
let err = evaluate_vector_search(
- &file_io,
- "memory:///test_table",
- &HashMap::new(),
+ eval_context(&file_io, &options, &fields, None),
&[entry],
&vs,
- &fields,
)
.await
.unwrap_err();
