Skip to content

feat(table): maintain declared global indexes after writes #1261

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
JingsongLi merged 4 commits into from
Oct 31, 2026
Merged

feat(table): maintain declared global indexes after writes #1261

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
ec23d6f
feat(table): maintain declared global indexes after writes
JingsongLi Oct 31, 2026
b35e095
refactor(table): rename automatic index module to auto_index
JingsongLi Oct 31, 2026
01b6ccf
refactor(table): isolate index replacement planning in auto_index
JingsongLi Oct 31, 2026
f5e01a0
refactor(table): rename auto_compact module to auto_maintenance
JingsongLi Oct 31, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion crates/paimon/src/spec/core_options.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1639,7 +1639,8 @@ impl<'a> CoreOptions<'a> {
.unwrap_or(DEFAULT_TARGET_FILE_SIZE)
}

/// Skip automatic compaction after a data commit. Defaults to false.
/// Skip automatic compaction and declared global-index maintenance after
/// a data commit. Defaults to false.
/// `write.compaction-skip` is the legacy alias; `write-only` takes precedence.
pub fn write_only(&self) -> crate::Result<bool> {
match self
Expand Down
88 changes: 25 additions & 63 deletions crates/paimon/src/table/de_vector_read.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@

//! Reads global vector indexes, scores raw vectors, and materializes global-row-ID results.

use crate::io::{FileIO, FileRead};
use crate::io::FileIO;
use crate::spec::{
row_id_data_field, CoreOptions, DataField, DataType, FileKind, GlobalIndexSearchMode,
IndexFileMeta, IndexManifestEntry, Predicate, ROW_ID_FIELD_NAME,
Expand All @@ -38,8 +38,8 @@ use crate::table::search_task::{
use crate::table::vector_read::Read;
use crate::table::vector_search_common::{
configured_refine_factor, configured_vector_index_type, indexed_search_limit,
log_vindex_range_io_stats, normalize_metric, resolve_materialize_read_type,
validate_global_vector_index_types, vindex_concurrency_limits,
log_vindex_range_io_stats, normalize_metric, read_index_vector_metric,
resolve_materialize_read_type, validate_global_vector_index_types, vindex_concurrency_limits,
};
use crate::table::{
find_field_id_by_name, merge_row_ranges, ArrowRecordBatchStream, RowRange, Table,
Expand All @@ -56,9 +56,7 @@ use arrow_array::{Array, FixedSizeListArray, Float32Array, Int64Array, ListArray
use arrow_select::interleave::interleave_record_batch;
use futures::{stream, TryStreamExt};
use paimon_vindex_core::blas::sgemm_a_bt;
use paimon_vindex_core::diskann_io::DISKANN_HEADER_SIZE;
use paimon_vindex_core::distance::MetricType;
use paimon_vindex_core::index::VectorIndexReader as VIndexReader;
use paimon_vindex_core::io::SeekRead;
use roaring::RoaringTreemap;
use serde::{Deserialize, Serialize};
Expand Down Expand Up @@ -900,7 +898,7 @@ async fn evaluate_batch_vector_search_with_executor(
.collect::<Vec<_>>()
};
let configured_index_type =
configured_vector_index_type(evaluation.table_options, &search_options, field_name);
configured_vector_index_type(evaluation.table_options, &search_options, field_name)?;
let index_type = vector_entries
.first()
.map(|entry| entry.index_file.index_type.as_str())
Expand Down Expand Up @@ -1125,7 +1123,7 @@ async fn evaluate_batch_vector_search_with_executor(
})?;
let metric_start = timing_enabled.then(Instant::now);
let mut metric_options = vector_searches[0].options.clone();
if configured_vector_index_type(evaluation.table_options, &metric_options, field_name)
if configured_vector_index_type(evaluation.table_options, &metric_options, field_name)?
.is_none()
{
if let Some(index_type) =
Expand Down Expand Up @@ -1673,7 +1671,7 @@ async fn resolve_raw_vector_metric(
field_id: i32,
field_name: &str,
) -> crate::Result<RawVectorMetric> {
let configured_type = configured_vector_index_type(table_options, search_options, field_name);
let configured_type = configured_vector_index_type(table_options, search_options, field_name)?;
let index_type = index_entries
.iter()
.filter(|entry| entry.kind == FileKind::Add)
Expand Down Expand Up @@ -1703,6 +1701,7 @@ async fn resolve_raw_vector_metric(
continue;
}
if let Some(metric) = read_index_vector_metric(file_io, table_path, entry).await? {
let metric = RawVectorMetric::from_vindex(metric);
check.check(metric)?;
persisted = Some(metric);
}
Expand All @@ -1713,6 +1712,24 @@ async fn resolve_raw_vector_metric(
if let Some(metric) = check.requested {
return Ok(metric);
}
if let Some(kind) = super::global_index_options::vector_index_type(table_options, field_name)? {
if index_type == Some(kind) {
let options = super::global_index_options::field_index_options(
table_options,
field_name,
"vector",
kind,
)?;
let native = crate::vindex::VindexVectorIndexOptions::resolve_native_options(
table_options,
&options,
&HashMap::new(),
kind,
field_name,
)?;
return RawVectorMetric::parse(&native["metric"]);
}
}
let table_metric = vector_metric_override(table_options, field_name, index_type)?;
if let Some(metric) = table_metric {
return Ok(metric);
Expand All @@ -1726,61 +1743,6 @@ async fn resolve_raw_vector_metric(
})
}

/// Rerouted raw searches still validate each shard's persisted metric.
async fn read_index_vector_metric(
file_io: &FileIO,
table_path: &str,
entry: &IndexManifestEntry,
) -> crate::Result<Option<RawVectorMetric>> {
let Some(global_meta) = entry.index_file.global_index_meta.as_ref() else {
return Ok(None);
};
if !is_vindex_index_type(&entry.index_file.index_type) {
return Ok(None);
}
if let Some(index_meta) = global_meta.index_meta.as_ref() {
if let Ok(options) = serde_json::from_slice::<HashMap<String, String>>(index_meta) {
if let Some(metric) = options.get("metric") {
if let Some(metric) = RawVectorMetric::parse_normalized(&normalize_metric(metric)) {
return Ok(Some(metric));
}
}
}
}
let path = IndexFileLocation::Global { table_path }.resolve(
&entry.index_file.file_name,
entry.index_file.external_path.as_deref(),
);
let input = file_io.new_input(&path)?;
let read_error = |e| crate::Error::DataInvalid {
message: format!(
"Failed to read vindex index file '{}' for raw search metric: {}",
entry.index_file.file_name, e
),
source: Some(Box::new(e)),
};
let header_size = if entry.index_file.file_size > 0 {
(entry.index_file.file_size as u64).min(DISKANN_HEADER_SIZE as u64)
} else {
input
.metadata()
.await
.map_err(&read_error)?
.size
.min(DISKANN_HEADER_SIZE as u64)
};
let file_reader = input.reader().await.map_err(&read_error)?;
let bytes = file_reader.read(0..header_size).await.map_err(read_error)?;
let reader = VIndexReader::open(Cursor::new(bytes)).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)),
})?;
Ok(Some(RawVectorMetric::from_vindex(reader.metadata().metric)))
}

fn vector_metric_override(
options: &HashMap<String, String>,
field_name: &str,
Expand Down
26 changes: 13 additions & 13 deletions crates/paimon/src/table/full_text_index_build_builder.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,6 +53,7 @@ pub(super) struct FullTextIndexBuildBuilder<'a> {
index_column: Option<String>,
options: HashMap<String, String>,
pub(super) partition_filter: Option<Predicate>,
pub(super) view: Option<super::global_index_build_common::GlobalIndexBuildView>,
}

impl<'a> FullTextIndexBuildBuilder<'a> {
Expand All @@ -62,6 +63,7 @@ impl<'a> FullTextIndexBuildBuilder<'a> {
index_column: None,
options: HashMap::new(),
partition_filter: None,
view: None,
}
}

Expand Down Expand Up @@ -110,10 +112,12 @@ impl<'a> FullTextIndexBuildBuilder<'a> {
full_text_config(&native_options)?;
let index_meta = serialize_index_meta(&native_options)?;

// The table's manager resolves a REST-managed table's latest snapshot
// through the catalog, like reads and commit validation do.
let snapshot_manager = self.table.snapshot_manager();
let Some(snapshot) = snapshot_manager.get_latest_snapshot().await? else {
let view = super::global_index_build_common::GlobalIndexBuildView::resolve(
self.table,
self.view.as_ref(),
)
.await?;
let Some(snapshot) = view.snapshot() else {
return Ok((None, vec![]));
};

Expand All @@ -124,16 +128,14 @@ impl<'a> FullTextIndexBuildBuilder<'a> {
let manifest_entries = read_builder
.new_scan()
.with_scan_all_files()
.plan_manifest_entries(&snapshot)
.plan_manifest_entries(snapshot)
.await?;
let indexed = indexed_row_ranges(
self.table,
snapshot.index_manifest(),
view.index_entries(),
FULL_TEXT_GLOBAL_INDEX_TYPE,
index_field.id(),
None,
)
.await?;
);
let shards = plan_vector_index_shards(
self.table.location(),
self.table.schema().partition_keys(),
Expand All @@ -150,17 +152,15 @@ impl<'a> FullTextIndexBuildBuilder<'a> {
}

validate_existing_index_overlap(
self.table,
snapshot.index_manifest(),
view.index_entries(),
FULL_TEXT_GLOBAL_INDEX_TYPE,
index_field.id(),
None,
&shards
.iter()
.map(|shard| RowRange::new(shard.row_range_start, shard.row_range_end))
.collect::<Vec<_>>(),
)
.await?;
)?;

let prepared = super::global_index_build_common::preparation::prepare_shards(
self.table,
Expand Down
16 changes: 12 additions & 4 deletions crates/paimon/src/table/full_text_search_builder.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -697,7 +697,7 @@ async fn read_raw_full_text_splits(
let read = read_builder.new_read()?;
let mut stream = read.to_arrow(splits)?;

let config = raw_full_text_config(table_options)?;
let config = raw_full_text_config(table_options, &search.field_name)?;
let mut writer = tokio::task::spawn_blocking(move || {
FullTextIndexWriter::new(config).map_err(full_text_error)
})
Expand Down Expand Up @@ -908,8 +908,16 @@ fn raw_full_text_document(

fn raw_full_text_config(
table_options: &HashMap<String, String>,
column: &str,
) -> crate::Result<FullTextIndexConfig> {
let native_options: HashMap<String, String> = table_options
let mut options = table_options.clone();
options.extend(super::global_index_options::field_index_options(
table_options,
column,
"full-text",
"full-text",
)?);
let native_options: HashMap<String, String> = options
.iter()
.filter_map(|(key, value)| {
key.strip_prefix("full-text.")
Expand Down Expand Up @@ -1321,7 +1329,7 @@ mod tests {
"native_text,secondary".to_string(),
)]);

let config = raw_full_text_config(&options).unwrap();
let config = raw_full_text_config(&options, "body").unwrap();

assert_eq!(config.text_fields, vec!["native_text", "secondary"]);
assert_eq!(config.default_text_field(), "native_text");
Expand All @@ -1333,7 +1341,7 @@ mod tests {
"full-text.text-field".to_string(),
"native_text".to_string(),
)]);
let config = raw_full_text_config(&options).unwrap();
let config = raw_full_text_config(&options, "body").unwrap();
let search = FullTextSearch::new(
r#"{"match":{"column":"native_text","query":"rare common"}}"#.to_string(),
10,
Expand Down
44 changes: 40 additions & 4 deletions crates/paimon/src/table/global_index_build_builder.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@ use std::collections::HashMap;
use std::time::Instant;

use super::full_text_index_build_builder::FullTextIndexBuildBuilder;
use super::global_index_build_common::{resolve_index_fields, IndexColumns};
use super::global_index_build_common::{resolve_index_fields, GlobalIndexBuildView, IndexColumns};
use super::global_index_build_plan::{GlobalIndexBuildPlan, GlobalIndexBuildShard};
use super::global_index_types::{
normalize_global_index_type, normalize_queryable_global_index_type, BTREE_GLOBAL_INDEX_TYPE,
Expand Down Expand Up @@ -105,6 +105,20 @@ impl<'a> GlobalIndexBuildBuilder<'a> {
self
}

fn configured_table(&self, columns: &[&str], index_type: &str) -> Result<Table> {
if columns.len() != 1 || crate::vindex::is_vindex_index_type(index_type) {
return Ok(self.table.clone());
}
// Native vector builders keep declarations in their own option layer.
let defaults = super::global_index_options::declared_options(
self.table.schema().options(),
columns[0],
index_type,
index_type,
)?;
Ok(self.table.copy_with_options(defaults))
}

/// Restrict the build to partitions. Repeated calls combine filters with AND.
pub fn with_partition_filter(&mut self, filter: Predicate) -> Result<&mut Self> {
super::partition_filter::validate_partition_filter(self.table, &filter)?;
Expand Down Expand Up @@ -197,6 +211,24 @@ impl<'a> GlobalIndexBuildBuilder<'a> {
}

async fn prepare(&self) -> Result<PreparedBuild> {
self.prepare_with_view(None).await
}

/// Check the complete writer configuration without scanning or creating files.
pub(crate) async fn validate(&self) -> Result<()> {
self.prepare_with_view(Some(GlobalIndexBuildView::default()))
.await
.map(|_| ())
}

pub(crate) async fn build_with_view(
&self,
view: GlobalIndexBuildView,
) -> Result<Vec<CommitMessage>> {
Ok(self.prepare_with_view(Some(view)).await?.messages)
}

async fn prepare_with_view(&self, view: Option<GlobalIndexBuildView>) -> Result<PreparedBuild> {
CoreOptions::new(self.table.schema().options()).ensure_read_authorized()?;
self.table.ensure_not_branch_reference_for_write()?;
let index_type = normalize_global_index_type(self.index_type.trim()).ok_or_else(|| {
Expand All @@ -210,8 +242,10 @@ impl<'a> GlobalIndexBuildBuilder<'a> {
})?;
let fields = resolve_index_fields(self.table, columns, index_type)?;
let columns = fields.iter().map(|field| field.name()).collect::<Vec<_>>();
let table = self.configured_table(&columns, index_type)?;
if normalize_queryable_global_index_type(index_type).is_some() {
let mut builder = SortedGlobalIndexBuildBuilder::new(self.table);
let mut builder = SortedGlobalIndexBuildBuilder::new(&table);
builder.view = view;
builder
.with_index_columns(&columns)
.with_index_type(index_type)
Expand All @@ -222,14 +256,16 @@ impl<'a> GlobalIndexBuildBuilder<'a> {
return Ok(builder.prepare().await?.into());
}
if index_type == FULL_TEXT_GLOBAL_INDEX_TYPE {
let mut builder = FullTextIndexBuildBuilder::new(self.table);
let mut builder = FullTextIndexBuildBuilder::new(&table);
builder.view = view;
builder
.with_index_column(columns[0])
.with_options(self.options.clone());
builder.partition_filter = self.partition_filter.clone();
return Ok(builder.prepare().await?.into());
}
let mut builder = VindexIndexBuildBuilder::new(self.table, index_type);
let mut builder = VindexIndexBuildBuilder::new(&table, index_type);
builder.view = view;
builder
.with_index_column(columns[0])
.with_options(self.options.clone());
Expand Down
Loading
Loading