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 5ab393d feat(core): incremental global index build for btree, lumina,
and vindex (#479)
5ab393d is described below
commit 5ab393d80a20084cd20d7cbad2617401ba8b69b7
Author: Junrui Lee <[email protected]>
AuthorDate: Wed Jul 8 09:21:32 2026 +0800
feat(core): incremental global index build for btree, lumina, and vindex
(#479)
---
.../src/table/btree_global_index_build_builder.rs | 567 ++++++++++++++++---
.../paimon/src/table/global_index_build_common.rs | 129 +++++
crates/paimon/src/table/global_index_scanner.rs | 39 +-
.../paimon/src/table/lumina_index_build_builder.rs | 515 ++++++++++++++---
crates/paimon/src/table/mod.rs | 1 +
crates/paimon/src/table/source.rs | 139 +++++
crates/paimon/src/table/table_commit.rs | 137 +++++
.../paimon/src/table/vindex_index_build_builder.rs | 630 ++++++++++++++++++---
8 files changed, 1909 insertions(+), 248 deletions(-)
diff --git a/crates/paimon/src/table/btree_global_index_build_builder.rs
b/crates/paimon/src/table/btree_global_index_build_builder.rs
index 2f242ca..5725179 100644
--- a/crates/paimon/src/table/btree_global_index_build_builder.rs
+++ b/crates/paimon/src/table/btree_global_index_build_builder.rs
@@ -22,8 +22,9 @@ use super::global_index_types::{
use crate::btree::{make_key_comparator, serialize_datum, BTreeIndexWriter,
BlockCompressionType};
use crate::spec::{
bucket_dir_name, extract_datum_from_arrow, BinaryRow, CoreOptions,
DataField, DataFileMeta,
- DataType, FileKind, GlobalIndexMeta, IndexFileMeta, IndexManifest,
ROW_ID_FIELD_NAME,
+ DataType, FileKind, GlobalIndexMeta, IndexFileMeta, ROW_ID_FIELD_NAME,
};
+use crate::table::source::exclude_row_ranges;
use crate::table::source::is_data_evolution_normal_file;
use crate::table::stats_filter::group_by_overlapping_row_id;
use crate::table::{
@@ -109,6 +110,15 @@ impl<'a> BTreeGlobalIndexBuildBuilder<'a> {
.with_scan_all_files()
.plan_manifest_entries(&snapshot)
.await?;
+ let indexed =
crate::table::global_index_build_common::indexed_row_ranges(
+ self.table,
+ snapshot.index_manifest(),
+ BTREE_INDEX_TYPE,
+ index_field.id(),
+ None, // single-column build; no extra fields today
+ )
+ .await?;
+
let shards = plan_btree_shards(
self.table.location(),
self.table.schema().partition_keys(),
@@ -117,16 +127,22 @@ impl<'a> BTreeGlobalIndexBuildBuilder<'a> {
snapshot.id(),
manifest_entries,
records_per_range,
+ &indexed,
)?;
if shards.is_empty() {
return Ok(0);
}
- validate_existing_index_overlap(
+
crate::table::global_index_build_common::validate_existing_index_overlap(
self.table,
snapshot.index_manifest(),
+ BTREE_INDEX_TYPE,
index_field.id(),
- &shards,
+ None,
+ &shards
+ .iter()
+ .map(|shard| RowRange::new(shard.row_range_start,
shard.row_range_end))
+ .collect::<Vec<_>>(),
)
.await?;
@@ -369,6 +385,7 @@ fn is_btree_supported_data_type(data_type: &DataType) ->
bool {
)
}
+#[allow(clippy::too_many_arguments)]
fn plan_btree_shards(
table_location: &str,
partition_keys: &[String],
@@ -377,6 +394,7 @@ fn plan_btree_shards(
snapshot_id: i64,
entries: Vec<crate::spec::ManifestEntry>,
records_per_range: i64,
+ indexed: &[RowRange],
) -> Result<Vec<BTreeGlobalIndexShard>> {
if records_per_range <= 0 {
return Err(Error::DataInvalid {
@@ -426,24 +444,30 @@ fn plan_btree_shards(
let normal_groups = group_normal_file_ranges(files)?;
for group in normal_groups {
let (coverage_start, coverage_end) =
normal_coverage_range(&group.files)?;
- let start_range = coverage_start / records_per_range;
- let end_range = coverage_end / records_per_range;
- for range_id in start_range..=end_range {
- let range_start = range_id * records_per_range;
- let range_end = range_start + records_per_range - 1;
- let row_range_start = coverage_start.max(range_start);
- let row_range_end = coverage_end.min(range_end);
- result.push(BTreeGlobalIndexShard {
- partition: partition.clone(),
- partition_bytes: partition_bytes.clone(),
- files: group.files.clone(),
- row_range_start,
- row_range_end,
- snapshot_id,
- source_bucket,
- total_buckets,
- bucket_path: bucket_path.clone(),
- });
+ let build_segments =
+ exclude_row_ranges(&[RowRange::new(coverage_start,
coverage_end)], indexed);
+ for seg in build_segments {
+ let seg_start = seg.from();
+ let seg_end = seg.to();
+ let start_range = seg_start / records_per_range;
+ let end_range = seg_end / records_per_range;
+ for range_id in start_range..=end_range {
+ let range_start = range_id * records_per_range;
+ let range_end = range_start + records_per_range - 1;
+ let row_range_start = seg_start.max(range_start);
+ let row_range_end = seg_end.min(range_end);
+ result.push(BTreeGlobalIndexShard {
+ partition: partition.clone(),
+ partition_bytes: partition_bytes.clone(),
+ files: group.files.clone(),
+ row_range_start,
+ row_range_end,
+ snapshot_id,
+ source_bucket,
+ total_buckets,
+ bucket_path: bucket_path.clone(),
+ });
+ }
}
}
}
@@ -565,51 +589,6 @@ fn bucket_path(
))
}
-async fn validate_existing_index_overlap(
- table: &Table,
- index_manifest_name: Option<&str>,
- index_field_id: i32,
- shards: &[BTreeGlobalIndexShard],
-) -> Result<()> {
- let Some(index_manifest_name) = index_manifest_name else {
- return Ok(());
- };
- let path = format!(
- "{}/manifest/{}",
- table.location().trim_end_matches('/'),
- index_manifest_name
- );
- let entries = IndexManifest::read(table.file_io(), &path).await?;
- for entry in entries {
- if entry.kind != FileKind::Add {
- continue;
- }
- let Some(meta) = entry.index_file.global_index_meta else {
- continue;
- };
- if meta.index_field_id != index_field_id {
- continue;
- }
- if shards.iter().any(|shard| {
- ranges_overlap(
- meta.row_range_start,
- meta.row_range_end,
- shard.row_range_start,
- shard.row_range_end,
- )
- }) {
- return Err(Error::DataInvalid {
- message: format!(
- "Existing global index file '{}' overlaps requested row
range for field {}",
- entry.index_file.file_name, index_field_id
- ),
- source: None,
- });
- }
- }
- Ok(())
-}
-
async fn extract_index_rows(
table: &Table,
shard: &BTreeGlobalIndexShard,
@@ -786,11 +765,11 @@ mod tests {
use crate::io::FileIOBuilder;
use crate::spec::stats::BinaryTableStats;
use crate::spec::{
- BinaryType, GlobalIndexSearchMode, IntType, ManifestEntry,
PredicateBuilder, Schema,
- TableSchema, VarBinaryType, VarCharType,
+ BinaryType, GlobalIndexSearchMode, IndexManifest, IntType,
ManifestEntry, PredicateBuilder,
+ Schema, TableSchema, VarBinaryType, VarCharType,
};
use crate::table::global_index_scanner::{evaluate_global_index,
GlobalIndexEvaluation};
- use crate::table::{SnapshotManager, TableCommit, TableWrite};
+ use crate::table::{merge_row_ranges, SnapshotManager, TableCommit,
TableWrite};
use arrow_array::{ArrayRef, Int32Array, Int64Array, StringArray};
use arrow_schema::{DataType as ArrowDataType, Field as ArrowField, Schema
as ArrowSchema};
use chrono::{DateTime, Utc};
@@ -889,6 +868,7 @@ mod tests {
1,
entries,
records_per_range,
+ &[],
)
}
@@ -1360,15 +1340,74 @@ mod tests {
assert_eq!(row_ranges, vec![RowRange::new(0, 0), RowRange::new(2, 2)]);
}
+ async fn latest_btree_index_files(table: &Table) -> Vec<IndexFileMeta> {
+ let snapshot_manager =
+ SnapshotManager::new(table.file_io().clone(),
table.location().to_string());
+ let snapshot = snapshot_manager
+ .get_latest_snapshot()
+ .await
+ .unwrap()
+ .unwrap();
+ let Some(index_manifest_name) = snapshot.index_manifest() else {
+ return Vec::new();
+ };
+ IndexManifest::read(
+ table.file_io(),
+ &snapshot_manager.manifest_path(index_manifest_name),
+ )
+ .await
+ .unwrap()
+ .into_iter()
+ .filter(|entry| {
+ entry.kind == FileKind::Add && entry.index_file.index_type ==
BTREE_INDEX_TYPE
+ })
+ .map(|entry| entry.index_file)
+ .collect()
+ }
+
+ /// Row-id coverage of the committed data files, read back from the data
+ /// manifest (never hard-coded) and merged into contiguous ranges. Mirrors
+ /// how `execute` gathers `manifest_entries` so tests observe the exact
+ /// row-ids the writer assigned.
+ async fn data_row_id_coverage(table: &Table) -> Vec<RowRange> {
+ let snapshot_manager =
+ SnapshotManager::new(table.file_io().clone(),
table.location().to_string());
+ let snapshot = snapshot_manager
+ .get_latest_snapshot()
+ .await
+ .unwrap()
+ .unwrap();
+ let entries = table
+ .new_read_builder()
+ .new_scan()
+ .with_scan_all_files()
+ .plan_manifest_entries(&snapshot)
+ .await
+ .unwrap();
+ let ranges = entries
+ .iter()
+ .filter(|entry| *entry.kind() == FileKind::Add)
+ .filter_map(|entry| {
+ entry
+ .file()
+ .row_id_range()
+ .map(|(start, end)| RowRange::new(start, end))
+ })
+ .collect::<Vec<_>>();
+ merge_row_ranges(ranges)
+ }
+
+ /// Second build with no new data must be a clean no-op (returns 0), not an
+ /// overlap error. This is the core bug fix: today the second call errors.
#[tokio::test]
- async fn test_execute_rejects_existing_overlap_before_commit() {
- let table_path = "memory:/test_btree_global_index_builder_overlap";
+ async fn second_build_without_new_data_is_noop() {
+ let table_path = "memory:/test_btree_global_index_second_build_noop";
let table = test_table_with_path(table_path, table_options("10"));
setup_dirs(&table).await;
let mut table_write = TableWrite::new(&table,
"test-user".to_string()).unwrap();
table_write
- .write_arrow_batch(&data_batch(vec![1, 2], vec!["alice", "bob"]))
+ .write_arrow_batch(&data_batch(vec![1, 2, 3], vec!["alice", "bob",
"carol"]))
.await
.unwrap();
let messages = table_write.prepare_commit().await.unwrap();
@@ -1377,21 +1416,397 @@ mod tests {
.await
.unwrap();
- table
+ let first_built = table
+ .new_btree_global_index_build_builder()
+ .with_index_column("name")
+ .execute()
+ .await
+ .unwrap();
+ assert!(first_built > 0, "first build must index the initial rows");
+
+ let files_after_first = latest_btree_index_files(&table).await;
+ assert!(!files_after_first.is_empty());
+
+ let built = table
+ .new_btree_global_index_build_builder()
+ .with_index_column("name")
+ .execute()
+ .await
+ .unwrap();
+ assert_eq!(built, 0, "fully-indexed table must build nothing on
re-run");
+
+ let files_after_second = latest_btree_index_files(&table).await;
+ let names_first = files_after_first
+ .iter()
+ .map(|f| f.file_name.clone())
+ .collect::<std::collections::BTreeSet<_>>();
+ let names_second = files_after_second
+ .iter()
+ .map(|f| f.file_name.clone())
+ .collect::<std::collections::BTreeSet<_>>();
+ assert_eq!(
+ names_first, names_second,
+ "re-run must not add or remove index manifest entries"
+ );
+ }
+
+ /// Build, append new rows, build again -> only the appended row range is
+ /// indexed; the first build's index files are retained untouched
(append-only).
+ #[tokio::test]
+ async fn incremental_build_indexes_only_new_rows() {
+ let table_path = "memory:/test_btree_global_index_incremental";
+ let table = test_table_with_path(table_path, table_options("10"));
+ setup_dirs(&table).await;
+
+ // Build #1 over rows [0..3).
+ let mut table_write = TableWrite::new(&table,
"test-user".to_string()).unwrap();
+ table_write
+ .write_arrow_batch(&data_batch(vec![1, 2, 3], vec!["alice", "bob",
"carol"]))
+ .await
+ .unwrap();
+ let messages = table_write.prepare_commit().await.unwrap();
+ TableCommit::new(table.clone(), "test-user".to_string())
+ .commit(messages)
+ .await
+ .unwrap();
+
+ let first_built = table
+ .new_btree_global_index_build_builder()
+ .with_index_column("name")
+ .execute()
+ .await
+ .unwrap();
+ assert!(first_built > 0);
+
+ let first_files = latest_btree_index_files(&table).await;
+ let first_names = first_files
+ .iter()
+ .map(|f| f.file_name.clone())
+ .collect::<std::collections::BTreeSet<_>>();
+ let n: i64 = 3;
+
+ // Append a second batch (new row-ids [3..6)).
+ let mut table_write = TableWrite::new(&table,
"test-user".to_string()).unwrap();
+ table_write
+ .write_arrow_batch(&data_batch(vec![4, 5, 6], vec!["dave", "erin",
"frank"]))
+ .await
+ .unwrap();
+ let messages = table_write.prepare_commit().await.unwrap();
+ TableCommit::new(table.clone(), "test-user".to_string())
+ .commit(messages)
+ .await
+ .unwrap();
+
+ let second_built = table
+ .new_btree_global_index_build_builder()
+ .with_index_column("name")
+ .execute()
+ .await
+ .unwrap();
+ assert!(second_built > 0, "appended rows must be indexed");
+
+ let all_files = latest_btree_index_files(&table).await;
+ let all_names = all_files
+ .iter()
+ .map(|f| f.file_name.clone())
+ .collect::<std::collections::BTreeSet<_>>();
+
+ // Every build-#1 file is still present (append-only, no
rewrite/delete).
+ assert!(
+ first_names.iter().all(|name| all_names.contains(name)),
+ "build #1 index files must be retained untouched"
+ );
+
+ // Every build-#2 file covers only the appended range [N, ..].
+ let new_files = all_files
+ .iter()
+ .filter(|f| !first_names.contains(&f.file_name))
+ .collect::<Vec<_>>();
+ assert!(!new_files.is_empty(), "build #2 must add new index files");
+ for file in new_files {
+ let meta = file
+ .global_index_meta
+ .as_ref()
+ .expect("global index meta on new btree file");
+ assert!(
+ meta.row_range_start >= n,
+ "new index file range must start at or after {}, got [{}, {}]",
+ n,
+ meta.row_range_start,
+ meta.row_range_end
+ );
+ }
+ }
+
+ /// Regression: first build (no existing index) must equal the pre-change
+ /// full build -- subtraction with empty `indexed` = full coverage.
+ #[tokio::test]
+ async fn first_build_indexes_full_coverage() {
+ let table_path = "memory:/test_btree_global_index_first_full_coverage";
+ let table = test_table_with_path(table_path, table_options("10"));
+ setup_dirs(&table).await;
+
+ let mut table_write = TableWrite::new(&table,
"test-user".to_string()).unwrap();
+ table_write
+ .write_arrow_batch(&data_batch(vec![1, 2, 3], vec!["alice", "bob",
"carol"]))
+ .await
+ .unwrap();
+ let messages = table_write.prepare_commit().await.unwrap();
+ TableCommit::new(table.clone(), "test-user".to_string())
+ .commit(messages)
+ .await
+ .unwrap();
+
+ let built = table
.new_btree_global_index_build_builder()
.with_index_column("name")
.execute()
.await
.unwrap();
+ assert_eq!(
+ built, 1,
+ "first build must index the full coverage in one shard"
+ );
- let err = table
+ let files = latest_btree_index_files(&table).await;
+ assert_eq!(files.len(), 1);
+ let meta = files[0]
+ .global_index_meta
+ .as_ref()
+ .expect("global index meta");
+ assert_eq!(meta.row_range_start, 0);
+ assert_eq!(meta.row_range_end, 2);
+ }
+
+ /// Grid boundary (spec edge 4): with `records-per-range = 4`, an appended
+ /// gap that spans several grid cells must be split so each new index
file's
+ /// range stays inside one cell, the ranges are contiguous, and together
+ /// they exactly cover the gap. Row-ids are read back from the manifests,
+ /// never hard-coded.
+ #[tokio::test]
+ async fn incremental_build_splits_gap_across_records_per_range_grid() {
+ const RPR: i64 = 4;
+ let table_path = "memory:/test_btree_global_index_grid_boundary";
+ let table = test_table_with_path(table_path, table_options("4"));
+ setup_dirs(&table).await;
+
+ // Build #1 over an initial batch (row-ids the writer assigns).
+ let mut table_write = TableWrite::new(&table,
"test-user".to_string()).unwrap();
+ table_write
+ .write_arrow_batch(&data_batch(vec![1, 2, 3], vec!["alice", "bob",
"carol"]))
+ .await
+ .unwrap();
+ let messages = table_write.prepare_commit().await.unwrap();
+ TableCommit::new(table.clone(), "test-user".to_string())
+ .commit(messages)
+ .await
+ .unwrap();
+
+ let first_built = table
.new_btree_global_index_build_builder()
.with_index_column("name")
.execute()
.await
- .expect_err("second build should overlap existing index");
+ .unwrap();
+ assert!(first_built > 0, "first build must index the initial rows");
+
+ // Row range already covered by build #1 (read back, not hard-coded).
+ let first_index_files = latest_btree_index_files(&table).await;
+ let indexed_before = merge_row_ranges(
+ first_index_files
+ .iter()
+ .filter_map(|f| f.global_index_meta.as_ref())
+ .map(|m| RowRange::new(m.row_range_start, m.row_range_end))
+ .collect(),
+ );
+ assert_eq!(
+ indexed_before.len(),
+ 1,
+ "build #1 should cover one contiguous range"
+ );
+ let gap_start = indexed_before[0].to() + 1;
+ let before_names = first_index_files
+ .iter()
+ .map(|f| f.file_name.clone())
+ .collect::<std::collections::BTreeSet<_>>();
+
+ // Append rows so the new gap crosses records_per_range (=4)
boundaries.
+ let mut table_write = TableWrite::new(&table,
"test-user".to_string()).unwrap();
+ table_write
+ .write_arrow_batch(&data_batch(
+ vec![4, 5, 6, 7, 8, 9, 10],
+ vec!["d", "e", "f", "g", "h", "i", "j"],
+ ))
+ .await
+ .unwrap();
+ let messages = table_write.prepare_commit().await.unwrap();
+ TableCommit::new(table.clone(), "test-user".to_string())
+ .commit(messages)
+ .await
+ .unwrap();
+
+ // Total data coverage read back from the data manifest.
+ let coverage = data_row_id_coverage(&table).await;
+ assert_eq!(
+ coverage.len(),
+ 1,
+ "appended data must be contiguous with build #1"
+ );
+ let gap_end = coverage[0].to();
assert!(
- matches!(err, Error::DataInvalid { message, .. } if
message.contains("overlaps requested row range"))
+ gap_end - gap_start + 1 > RPR,
+ "gap [{gap_start}, {gap_end}] must span more than one
records_per_range cell"
+ );
+
+ let second_built = table
+ .new_btree_global_index_build_builder()
+ .with_index_column("name")
+ .execute()
+ .await
+ .unwrap();
+ assert!(second_built > 0, "appended rows must be indexed");
+
+ // Only the newly written index files (build #1 files are retained).
+ let mut new_metas = latest_btree_index_files(&table)
+ .await
+ .into_iter()
+ .filter(|f| !before_names.contains(&f.file_name))
+ .filter_map(|f| f.global_index_meta)
+ .map(|m| (m.row_range_start, m.row_range_end))
+ .collect::<Vec<_>>();
+ new_metas.sort();
+ assert!(!new_metas.is_empty(), "build #2 must add new index files");
+
+ // (a) Each range lies within a single grid cell: no multiple of RPR is
+ // strictly interior, i.e. start and end share the same cell index.
+ for (start, end) in &new_metas {
+ assert!(end >= start, "range must be non-empty: [{start}, {end}]");
+ assert_eq!(
+ start / RPR,
+ end / RPR,
+ "range [{start}, {end}] straddles a records_per_range boundary"
+ );
+ }
+ // (b) Contiguous with no gaps or overlaps.
+ for pair in new_metas.windows(2) {
+ assert_eq!(
+ pair[1].0,
+ pair[0].1 + 1,
+ "ranges must be contiguous: {:?} then {:?}",
+ pair[0],
+ pair[1]
+ );
+ }
+ // (c) Together they exactly cover the appended gap [gap_start,
gap_end].
+ assert_eq!(
+ new_metas.first().unwrap().0,
+ gap_start,
+ "coverage must start at the gap start"
+ );
+ assert_eq!(
+ new_metas.last().unwrap().1,
+ gap_end,
+ "coverage must end at the gap end"
);
}
+
+ /// Hole splitting (spec edge 5) at build level: a mid-coverage indexed
range
+ /// (constructed directly, as the drop-builder tests build
`GlobalIndexMeta`
+ /// entries) must carve the data coverage into two build segments, one on
+ /// each side, and the hole itself must not be re-indexed.
+ #[tokio::test]
+ async fn incremental_build_splits_gap_around_mid_coverage_indexed_hole() {
+ let table_path = "memory:/test_btree_global_index_mid_hole";
+ // records-per-range large so the grid never splits: the only split is
+ // the hole itself.
+ let table = test_table_with_path(table_path, table_options("100"));
+ setup_dirs(&table).await;
+
+ // Real data spanning row-ids [0, 9].
+ let mut table_write = TableWrite::new(&table,
"test-user".to_string()).unwrap();
+ table_write
+ .write_arrow_batch(&data_batch(
+ (1..=10).collect(),
+ vec!["a", "b", "c", "d", "e", "f", "g", "h", "i", "j"],
+ ))
+ .await
+ .unwrap();
+ let messages = table_write.prepare_commit().await.unwrap();
+ TableCommit::new(table.clone(), "test-user".to_string())
+ .commit(messages)
+ .await
+ .unwrap();
+
+ let coverage = data_row_id_coverage(&table).await;
+ assert_eq!(coverage.len(), 1, "data must be one contiguous range");
+ assert_eq!(coverage[0].from(), 0);
+ let last_row = coverage[0].to();
+ assert!(last_row >= 9, "need at least 10 rows for a mid hole");
+
+ // Inject a mid-coverage indexed range [hole_start, hole_end] for the
+ // `name` field directly into the index manifest.
+ let name_field_id = find_index_field(&table, "name").unwrap().id();
+ let hole_start = 4;
+ let hole_end = 6;
+ let synthetic = IndexFileMeta {
+ index_type: BTREE_INDEX_TYPE.to_string(),
+ file_name: "btree-synthetic-hole.index".to_string(),
+ file_size: 1,
+ row_count: (hole_end - hole_start + 1) as i32,
+ deletion_vectors_ranges: None,
+ global_index_meta: Some(GlobalIndexMeta {
+ row_range_start: hole_start,
+ row_range_end: hole_end,
+ index_field_id: name_field_id,
+ extra_field_ids: None,
+ index_meta: None,
+ }),
+ };
+ let mut message =
CommitMessage::new(BinaryRow::new(0).to_serialized_bytes(), 0, vec![]);
+ message.new_index_files = vec![synthetic];
+ TableCommit::new(table.clone(), "test-user".to_string())
+ .commit(vec![message])
+ .await
+ .unwrap();
+
+ let before_names = latest_btree_index_files(&table)
+ .await
+ .into_iter()
+ .map(|f| f.file_name)
+ .collect::<std::collections::BTreeSet<_>>();
+
+ // Build: gap = coverage minus the hole = [0, hole_start-1] and
+ // [hole_end+1, last_row]; two shards since the grid does not split
here.
+ let built = table
+ .new_btree_global_index_build_builder()
+ .with_index_column("name")
+ .execute()
+ .await
+ .unwrap();
+ assert_eq!(
+ built, 2,
+ "mid-coverage hole must split the gap into two shards"
+ );
+
+ let mut new_metas = latest_btree_index_files(&table)
+ .await
+ .into_iter()
+ .filter(|f| !before_names.contains(&f.file_name))
+ .filter_map(|f| f.global_index_meta)
+ .map(|m| (m.row_range_start, m.row_range_end))
+ .collect::<Vec<_>>();
+ new_metas.sort();
+
+ assert_eq!(
+ new_metas,
+ vec![(0, hole_start - 1), (hole_end + 1, last_row)],
+ "new shards must fill the coverage on both sides of the indexed
hole"
+ );
+ for (start, end) in &new_metas {
+ assert!(
+ *end < hole_start || *start > hole_end,
+ "new shard [{start}, {end}] must not overlap indexed hole
[{hole_start}, {hole_end}]"
+ );
+ }
+ }
}
diff --git a/crates/paimon/src/table/global_index_build_common.rs
b/crates/paimon/src/table/global_index_build_common.rs
new file mode 100644
index 0000000..355a56e
--- /dev/null
+++ b/crates/paimon/src/table/global_index_build_common.rs
@@ -0,0 +1,129 @@
+// 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.
+
+//! Shared helpers for the global-index build builders (btree, lumina, vindex).
+//!
+//! Mirrors Java `GlobalIndexBuilderUtils`, which exposes a single
+//! `indexedRowRanges` used by both the sorted and vector index builders. The
+//! gap computation (what is already indexed for a given field, so a build only
+//! covers the new rows) is identical across index types apart from the
+//! `index_type` string, so it lives here once rather than being copied into
+//! each builder.
+
+use crate::spec::{FileKind, IndexManifest};
+use crate::table::{merge_row_ranges, RowRange, Table};
+use crate::{Error, Result};
+
+/// Java `sameExtraFieldIds`: null/empty are equal; otherwise exact ordered
equality.
+pub(crate) fn same_extra_field_ids(a: Option<&[i32]>, b: Option<&[i32]>) ->
bool {
+ let a = a.unwrap_or(&[]);
+ let b = b.unwrap_or(&[]);
+ a == b
+}
+
+/// Inclusive `[start, end]` range overlap.
+fn ranges_overlap(a_start: i64, a_end: i64, b_start: i64, b_end: i64) -> bool {
+ a_start <= b_end && b_start <= a_end
+}
+
+/// Row ranges already covered by `index_type` global-index files for
+/// `index_field_id` (and matching `extra_field_ids`). Mirrors Java
+/// `GlobalIndexBuilderUtils.indexedRowRanges`.
+pub(crate) async fn indexed_row_ranges(
+ table: &Table,
+ index_manifest_name: Option<&str>,
+ index_type: &str,
+ index_field_id: i32,
+ extra_field_ids: Option<&[i32]>,
+) -> Result<Vec<RowRange>> {
+ let Some(index_manifest_name) = index_manifest_name else {
+ return Ok(Vec::new());
+ };
+ let path = format!(
+ "{}/manifest/{}",
+ table.location().trim_end_matches('/'),
+ index_manifest_name
+ );
+ let entries = IndexManifest::read(table.file_io(), &path).await?;
+ let mut ranges = Vec::new();
+ for entry in entries {
+ if entry.kind != FileKind::Add || entry.index_file.index_type !=
index_type {
+ continue;
+ }
+ let Some(meta) = entry.index_file.global_index_meta else {
+ continue;
+ };
+ if meta.index_field_id != index_field_id
+ || !same_extra_field_ids(meta.extra_field_ids.as_deref(),
extra_field_ids)
+ {
+ continue;
+ }
+ ranges.push(RowRange::new(meta.row_range_start, meta.row_range_end));
+ }
+ Ok(merge_row_ranges(ranges))
+}
+
+/// Guard against building a global index over a row range that an existing
index
+/// file of the SAME identity already covers. Identity is the full
+/// `(index_type, index_field_id, extra_field_ids)` tuple -- matching
+/// `indexed_row_ranges` and Java `GlobalIndexIdentifier` -- so a different
index
+/// type (or different `extra_field_ids`) on the same field coexists. Two files
+/// of the same identity with overlapping ranges are still rejected.
+pub(crate) async fn validate_existing_index_overlap(
+ table: &Table,
+ index_manifest_name: Option<&str>,
+ index_type: &str,
+ index_field_id: i32,
+ extra_field_ids: Option<&[i32]>,
+ planned: &[RowRange],
+) -> Result<()> {
+ let Some(index_manifest_name) = index_manifest_name else {
+ return Ok(());
+ };
+ let path = format!(
+ "{}/manifest/{}",
+ table.location().trim_end_matches('/'),
+ index_manifest_name
+ );
+ let entries = IndexManifest::read(table.file_io(), &path).await?;
+ for entry in entries {
+ if entry.kind != FileKind::Add || entry.index_file.index_type !=
index_type {
+ continue;
+ }
+ let Some(meta) = entry.index_file.global_index_meta else {
+ continue;
+ };
+ if meta.index_field_id != index_field_id
+ || !same_extra_field_ids(meta.extra_field_ids.as_deref(),
extra_field_ids)
+ {
+ continue;
+ }
+ if planned
+ .iter()
+ .any(|r| ranges_overlap(meta.row_range_start, meta.row_range_end,
r.from(), r.to()))
+ {
+ return Err(Error::DataInvalid {
+ message: format!(
+ "Existing global index file '{}' overlaps requested row
range for field {}",
+ entry.index_file.file_name, index_field_id
+ ),
+ source: None,
+ });
+ }
+ }
+ Ok(())
+}
diff --git a/crates/paimon/src/table/global_index_scanner.rs
b/crates/paimon/src/table/global_index_scanner.rs
index 07b4858..57cce1c 100644
--- a/crates/paimon/src/table/global_index_scanner.rs
+++ b/crates/paimon/src/table/global_index_scanner.rs
@@ -850,43 +850,6 @@ 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>,
@@ -941,7 +904,7 @@ fn unindexed_ranges_for_coverage(
return Vec::new();
};
let indexed_ranges = indexed_ranges_from_coverage(coverage_by_field,
field_ids);
- exclude_row_ranges(&data_ranges, &indexed_ranges)
+ super::source::exclude_row_ranges(&data_ranges, &indexed_ranges)
}
/// Compute row ranges not covered by a family of global index files.
diff --git a/crates/paimon/src/table/lumina_index_build_builder.rs
b/crates/paimon/src/table/lumina_index_build_builder.rs
index b6c06c1..5ead6de 100644
--- a/crates/paimon/src/table/lumina_index_build_builder.rs
+++ b/crates/paimon/src/table/lumina_index_build_builder.rs
@@ -21,8 +21,9 @@ use crate::lumina::{
};
use crate::spec::{
bucket_dir_name, BinaryRow, CoreOptions, DataField, DataFileMeta,
DataType, FileKind,
- GlobalIndexMeta, IndexFileMeta, IndexManifest, ROW_ID_FIELD_NAME,
+ GlobalIndexMeta, IndexFileMeta, ROW_ID_FIELD_NAME,
};
+use crate::table::source::exclude_row_ranges;
use crate::table::{
CommitMessage, DataSplitBuilder, RowRange, SnapshotManager, Table,
TableCommit,
};
@@ -118,6 +119,14 @@ impl<'a> LuminaIndexBuildBuilder<'a> {
.with_scan_all_files()
.plan_manifest_entries(&snapshot)
.await?;
+ let indexed =
crate::table::global_index_build_common::indexed_row_ranges(
+ self.table,
+ snapshot.index_manifest(),
+ LUMINA_IDENTIFIER,
+ index_field.id(),
+ None, // single-column build; no extra fields today
+ )
+ .await?;
let shards = plan_lumina_shards(
self.table.location(),
self.table.schema().partition_keys(),
@@ -126,16 +135,22 @@ impl<'a> LuminaIndexBuildBuilder<'a> {
snapshot.id(),
manifest_entries,
rows_per_shard,
+ &indexed,
)?;
if shards.is_empty() {
return Ok(0);
}
- validate_existing_index_overlap(
+
crate::table::global_index_build_common::validate_existing_index_overlap(
self.table,
snapshot.index_manifest(),
+ LUMINA_IDENTIFIER,
index_field.id(),
- &shards,
+ None,
+ &shards
+ .iter()
+ .map(|shard| RowRange::new(shard.row_range_start,
shard.row_range_end))
+ .collect::<Vec<_>>(),
)
.await?;
@@ -356,6 +371,7 @@ fn resolve_lumina_options(
Ok(options)
}
+#[allow(clippy::too_many_arguments)]
fn plan_lumina_shards(
table_location: &str,
partition_keys: &[String],
@@ -364,6 +380,7 @@ fn plan_lumina_shards(
snapshot_id: i64,
entries: Vec<crate::spec::ManifestEntry>,
rows_per_shard: i64,
+ indexed: &[RowRange],
) -> Result<Vec<LuminaIndexShard>> {
if rows_per_shard <= 0 {
return Err(Error::DataInvalid {
@@ -446,19 +463,29 @@ fn plan_lumina_shards(
.map(|file| file.row_id_range().unwrap().1)
.max()
.unwrap();
- let row_range_start = group_start.max(shard_start);
- let row_range_end = group_end.min(shard_end);
- result.push(LuminaIndexShard {
- partition: partition.clone(),
- partition_bytes: partition_bytes.clone(),
- files: group,
- row_range_start,
- row_range_end,
- snapshot_id,
- source_bucket,
- total_buckets,
- bucket_path: bucket_path.clone(),
- });
+ // Coverage of this group clamped to the current shard cell.
Then
+ // subtract the already-indexed ranges so the build only covers
+ // the gap. Because grid-clamp and gap-subtraction are both
range
+ // intersections, applying the gap here is equivalent to
btree's
+ // "exclude then split" -- and each surviving segment stays
inside
+ // one shard cell, preserving per-shard row-id contiguity.
+ let coverage_start = group_start.max(shard_start);
+ let coverage_end = group_end.min(shard_end);
+ let build_segments =
+ exclude_row_ranges(&[RowRange::new(coverage_start,
coverage_end)], indexed);
+ for seg in build_segments {
+ result.push(LuminaIndexShard {
+ partition: partition.clone(),
+ partition_bytes: partition_bytes.clone(),
+ files: group.clone(),
+ row_range_start: seg.from(),
+ row_range_end: seg.to(),
+ snapshot_id,
+ source_bucket,
+ total_buckets,
+ bucket_path: bucket_path.clone(),
+ });
+ }
}
}
}
@@ -535,51 +562,6 @@ fn bucket_path(
))
}
-async fn validate_existing_index_overlap(
- table: &Table,
- index_manifest_name: Option<&str>,
- index_field_id: i32,
- shards: &[LuminaIndexShard],
-) -> Result<()> {
- let Some(index_manifest_name) = index_manifest_name else {
- return Ok(());
- };
- let path = format!(
- "{}/manifest/{}",
- table.location().trim_end_matches('/'),
- index_manifest_name
- );
- let entries = IndexManifest::read(table.file_io(), &path).await?;
- for entry in entries {
- if entry.kind != FileKind::Add {
- continue;
- }
- let Some(meta) = entry.index_file.global_index_meta else {
- continue;
- };
- if meta.index_field_id != index_field_id {
- continue;
- }
- if shards.iter().any(|shard| {
- ranges_overlap(
- meta.row_range_start,
- meta.row_range_end,
- shard.row_range_start,
- shard.row_range_end,
- )
- }) {
- return Err(Error::DataInvalid {
- message: format!(
- "Existing global index file '{}' overlaps requested row
range for field {}",
- entry.index_file.file_name, index_field_id
- ),
- source: None,
- });
- }
- }
- Ok(())
-}
-
async fn extract_vectors(
table: &Table,
shard: &LuminaIndexShard,
@@ -873,10 +855,6 @@ async fn copy_local_file_to_output(
writer.close().await
}
-fn ranges_overlap(left_start: i64, left_end: i64, right_start: i64, right_end:
i64) -> bool {
- left_start <= right_end && right_start <= left_end
-}
-
#[cfg(test)]
mod tests {
use super::*;
@@ -886,7 +864,8 @@ mod tests {
use crate::lumina::LUMINA_DIMENSION_OPTION;
use crate::spec::stats::BinaryTableStats;
use crate::spec::{
- ArrayType, DoubleType, FloatType, IntType, ManifestEntry, Schema,
TableSchema, VectorType,
+ ArrayType, DoubleType, FloatType, IndexManifest, IntType,
ManifestEntry, Schema,
+ TableSchema, VectorType,
};
use crate::table::TableWrite;
use arrow_array::builder::{FixedSizeListBuilder, Float32Builder,
Int64Builder, ListBuilder};
@@ -993,6 +972,14 @@ mod tests {
}
fn plan(entries: Vec<ManifestEntry>, rows_per_shard: i64) ->
Result<Vec<LuminaIndexShard>> {
+ plan_with_indexed(entries, rows_per_shard, &[])
+ }
+
+ fn plan_with_indexed(
+ entries: Vec<ManifestEntry>,
+ rows_per_shard: i64,
+ indexed: &[RowRange],
+ ) -> Result<Vec<LuminaIndexShard>> {
let table = test_table(table_options(&rows_per_shard.to_string()));
let core = CoreOptions::new(table.schema().options());
plan_lumina_shards(
@@ -1003,6 +990,7 @@ mod tests {
1,
entries,
rows_per_shard,
+ indexed,
)
}
@@ -1660,4 +1648,399 @@ mod tests {
let status = file_io.get_status(&index_path).await.unwrap();
assert_eq!(index_file.file_size as u64, status.size);
}
+
+ fn lumina_e2e_options(rows_per_shard: &str) -> HashMap<String, String> {
+ let mut options = table_options(rows_per_shard);
+ options.insert("lumina.index.dimension".to_string(), "2".to_string());
+ options.insert("lumina.encoding.type".to_string(),
"rawf32".to_string());
+ options
+ }
+
+ fn lumina_e2e_table(table_path: &str, rows_per_shard: &str) -> Table {
+ test_table_with_io(
+ FileIOBuilder::new("memory").build().unwrap(),
+ table_path,
+ vector_schema_builder(lumina_e2e_options(rows_per_shard))
+ .build()
+ .unwrap(),
+ )
+ }
+
+ async fn write_vectors(table: &Table, ids: Vec<i32>, vectors:
Vec<Vec<f32>>) {
+ let mut table_write = TableWrite::new(table,
"test-user".to_string()).unwrap();
+ table_write
+ .write_arrow_batch(&build_vector_batch(ids, vectors))
+ .await
+ .unwrap();
+ let messages = table_write.prepare_commit().await.unwrap();
+ TableCommit::new(table.clone(), "test-user".to_string())
+ .commit(messages)
+ .await
+ .unwrap();
+ }
+
+ /// Commit a synthetic Lumina `IndexFileMeta` covering `[start, end]` for
+ /// `field_id` directly into the index manifest, without invoking the
native
+ /// builder. Mirrors the btree mid-hole test so the incremental gap logic
can
+ /// be exercised in CI where the native Lumina library is unavailable.
+ async fn commit_synthetic_lumina_index(table: &Table, field_id: i32,
start: i64, end: i64) {
+ let synthetic = IndexFileMeta {
+ index_type: LUMINA_IDENTIFIER.to_string(),
+ file_name: format!("lumina-synthetic-{start}-{end}.index"),
+ file_size: 1,
+ row_count: (end - start + 1) as i32,
+ deletion_vectors_ranges: None,
+ global_index_meta: Some(GlobalIndexMeta {
+ row_range_start: start,
+ row_range_end: end,
+ index_field_id: field_id,
+ extra_field_ids: None,
+ index_meta: None,
+ }),
+ };
+ let mut message =
CommitMessage::new(BinaryRow::new(0).to_serialized_bytes(), 0, vec![]);
+ message.new_index_files = vec![synthetic];
+ TableCommit::new(table.clone(), "test-user".to_string())
+ .commit(vec![message])
+ .await
+ .unwrap();
+ }
+
+ async fn latest_lumina_index_files(table: &Table) -> Vec<IndexFileMeta> {
+ let snapshot_manager =
+ SnapshotManager::new(table.file_io().clone(),
table.location().to_string());
+ let snapshot = snapshot_manager
+ .get_latest_snapshot()
+ .await
+ .unwrap()
+ .unwrap();
+ let Some(index_manifest_name) = snapshot.index_manifest() else {
+ return Vec::new();
+ };
+ IndexManifest::read(
+ table.file_io(),
+ &snapshot_manager.manifest_path(index_manifest_name),
+ )
+ .await
+ .unwrap()
+ .into_iter()
+ .filter(|entry| {
+ entry.kind == FileKind::Add && entry.index_file.index_type ==
LUMINA_IDENTIFIER
+ })
+ .map(|entry| entry.index_file)
+ .collect()
+ }
+
+ /// Row-id coverage of the committed data files, read back from the data
+ /// manifest (never hard-coded) and merged into contiguous ranges. Mirrors
+ /// how `execute` gathers `manifest_entries`.
+ async fn data_row_id_coverage(table: &Table) -> Vec<RowRange> {
+ let snapshot_manager =
+ SnapshotManager::new(table.file_io().clone(),
table.location().to_string());
+ let snapshot = snapshot_manager
+ .get_latest_snapshot()
+ .await
+ .unwrap()
+ .unwrap();
+ let entries = table
+ .new_read_builder()
+ .new_scan()
+ .with_scan_all_files()
+ .plan_manifest_entries(&snapshot)
+ .await
+ .unwrap();
+ let ranges = entries
+ .iter()
+ .filter(|entry| *entry.kind() == FileKind::Add)
+ .filter_map(|entry| {
+ entry
+ .file()
+ .row_id_range()
+ .map(|(start, end)| RowRange::new(start, end))
+ })
+ .collect::<Vec<_>>();
+ crate::table::merge_row_ranges(ranges)
+ }
+
+ /// Second build with the whole coverage already indexed must be a clean
+ /// no-op (returns 0), not an overlap error. Reaches `Ok(0)` before the
+ /// native build, so it runs in CI without the Lumina library. This is the
+ /// core bug fix: today the second call errors with the overlap message.
+ #[tokio::test]
+ async fn lumina_second_build_without_new_data_is_noop() {
+ let table_path = "memory:/test_lumina_second_build_noop";
+ let table = lumina_e2e_table(table_path, "10");
+ setup_dirs(table.file_io(), table_path).await;
+
+ write_vectors(
+ &table,
+ vec![1, 2, 3],
+ vec![vec![1.0, 0.0], vec![0.0, 1.0], vec![1.0, 1.0]],
+ )
+ .await;
+
+ // Fully index the coverage via a synthetic manifest entry.
+ let coverage = data_row_id_coverage(&table).await;
+ assert_eq!(coverage.len(), 1, "data must be one contiguous range");
+ let field_id = find_index_field(&table, "embedding").unwrap().id();
+ commit_synthetic_lumina_index(&table, field_id, coverage[0].from(),
coverage[0].to()).await;
+
+ let names_before = latest_lumina_index_files(&table)
+ .await
+ .iter()
+ .map(|f| f.file_name.clone())
+ .collect::<std::collections::BTreeSet<_>>();
+ assert!(!names_before.is_empty());
+
+ let built = table
+ .new_lumina_index_build_builder()
+ .with_index_column("embedding")
+ .execute()
+ .await
+ .unwrap();
+ assert_eq!(built, 0, "fully-indexed table must build nothing on
re-run");
+
+ let names_after = latest_lumina_index_files(&table)
+ .await
+ .iter()
+ .map(|f| f.file_name.clone())
+ .collect::<std::collections::BTreeSet<_>>();
+ assert_eq!(
+ names_before, names_after,
+ "re-run must not add or remove index manifest entries"
+ );
+ }
+
+ /// Build over an already-indexed prefix, then append new rows: the second
+ /// build must target only the appended gap and must NOT fail with the
+ /// overlap error. Without the native Lumina library the gap build
surfaces a
+ /// library-load error (not the overlap error); with it present it
succeeds.
+ #[tokio::test]
+ async fn lumina_incremental_build_indexes_only_new_rows() {
+ let table_path = "memory:/test_lumina_incremental";
+ let table = lumina_e2e_table(table_path, "10");
+ setup_dirs(table.file_io(), table_path).await;
+
+ // Initial batch, then mark it fully indexed via a synthetic entry.
+ write_vectors(
+ &table,
+ vec![1, 2, 3],
+ vec![vec![1.0, 0.0], vec![0.0, 1.0], vec![1.0, 1.0]],
+ )
+ .await;
+ let indexed_coverage = data_row_id_coverage(&table).await;
+ assert_eq!(indexed_coverage.len(), 1);
+ let n = indexed_coverage[0].to() + 1;
+ let field_id = find_index_field(&table, "embedding").unwrap().id();
+ commit_synthetic_lumina_index(
+ &table,
+ field_id,
+ indexed_coverage[0].from(),
+ indexed_coverage[0].to(),
+ )
+ .await;
+
+ // Append a second batch (new row-ids [n..]).
+ write_vectors(
+ &table,
+ vec![4, 5, 6],
+ vec![vec![2.0, 0.0], vec![0.0, 2.0], vec![2.0, 2.0]],
+ )
+ .await;
+
+ // White-box: fed the real indexed ranges from the manifest, the
planner
+ // must target only the appended gap [n, ..], never the already-indexed
+ // prefix. Computed before `execute` so it is independent of whether
the
+ // native build (which needs the Lumina library) runs.
+ let snapshot_manager =
+ SnapshotManager::new(table.file_io().clone(),
table.location().to_string());
+ let snapshot = snapshot_manager
+ .get_latest_snapshot()
+ .await
+ .unwrap()
+ .unwrap();
+ let manifest_entries = table
+ .new_read_builder()
+ .new_scan()
+ .with_scan_all_files()
+ .plan_manifest_entries(&snapshot)
+ .await
+ .unwrap();
+ let indexed =
crate::table::global_index_build_common::indexed_row_ranges(
+ &table,
+ snapshot.index_manifest(),
+ LUMINA_IDENTIFIER,
+ field_id,
+ None,
+ )
+ .await
+ .unwrap();
+ let core = CoreOptions::new(table.schema().options());
+ let shards = plan_lumina_shards(
+ table.location(),
+ table.schema().partition_keys(),
+ table.schema().fields(),
+ &core,
+ snapshot.id(),
+ manifest_entries,
+ 10,
+ &indexed,
+ )
+ .unwrap();
+ assert!(!shards.is_empty(), "appended gap must produce build shards");
+ for shard in &shards {
+ assert!(
+ shard.row_range_start >= n,
+ "shard [{}, {}] must start at or after the indexed prefix end
{n}",
+ shard.row_range_start,
+ shard.row_range_end
+ );
+ }
+
+ // End-to-end: the incremental build must no longer fail with the
overlap
+ // error. Without the native Lumina library the gap build surfaces a
+ // library-load error instead; with it present it succeeds.
+ let result = table
+ .new_lumina_index_build_builder()
+ .with_index_column("embedding")
+ .execute()
+ .await;
+ match result {
+ Ok(_) => {}
+ Err(Error::DataInvalid { message, .. }) => {
+ assert!(
+ !message.contains("overlaps requested row range"),
+ "incremental build must not fail with the overlap error;
got: {message}"
+ );
+ }
+ Err(other) => panic!("unexpected error from incremental build:
{other:?}"),
+ }
+ }
+
+ /// Regression: a first build (no existing index) must equal the pre-change
+ /// full build -- subtracting an empty `indexed` yields full coverage.
+ #[test]
+ fn lumina_first_build_indexes_full_coverage() {
+ let full = plan(vec![manifest_entry(data_file("a", Some(0), 25))],
10).unwrap();
+ let gapped =
+ plan_with_indexed(vec![manifest_entry(data_file("a", Some(0),
25))], 10, &[]).unwrap();
+ // Empty `indexed` must not alter the shard layout.
+ assert_eq!(
+ full.iter()
+ .map(|s| (s.row_range_start, s.row_range_end))
+ .collect::<Vec<_>>(),
+ gapped
+ .iter()
+ .map(|s| (s.row_range_start, s.row_range_end))
+ .collect::<Vec<_>>()
+ );
+ assert_eq!(
+ full.iter()
+ .map(|s| (s.row_range_start, s.row_range_end))
+ .collect::<Vec<_>>(),
+ vec![(0, 9), (10, 19), (20, 24)],
+ "first build must cover the full row range across shards"
+ );
+ }
+
+ /// Planner-level mid-coverage hole, mirroring btree's
+ /// `incremental_build_splits_gap_around_mid_coverage_indexed_hole`: with a
+ /// single shard cell (rows_per_shard large enough to hold all data) the
grid
+ /// never splits, so the only split is the indexed hole itself. An indexed
+ /// range strictly inside the data coverage must carve the build into
exactly
+ /// the two contiguous segments on either side of the hole -- both bounds
+ /// pinned, and neither segment may span or touch the hole.
+ #[test]
+ fn lumina_plan_splits_gap_around_mid_coverage_indexed_hole() {
+ // Data row-ids [0, 9]; one shard cell [0, 99] so the grid never
splits.
+ let n = 9;
+ let hole_start = 4;
+ let hole_end = 6;
+ let shards = plan_with_indexed(
+ vec![manifest_entry(data_file("a", Some(0), n + 1))],
+ 100,
+ &[RowRange::new(hole_start, hole_end)],
+ )
+ .unwrap();
+
+ let ranges = shards
+ .iter()
+ .map(|s| (s.row_range_start, s.row_range_end))
+ .collect::<Vec<_>>();
+ // Exactly the two contiguous segments around the hole.
+ assert_eq!(
+ ranges,
+ vec![(0, hole_start - 1), (hole_end + 1, n)],
+ "mid-coverage hole must split into exactly the two segments around
it"
+ );
+ // Every emitted range is contiguous and none spans or touches the
hole.
+ for (start, end) in &ranges {
+ assert!(end >= start, "range must be non-empty: [{start}, {end}]");
+ assert!(
+ *end < hole_start || *start > hole_end,
+ "shard [{start}, {end}] must not overlap indexed hole
[{hole_start}, {hole_end}]"
+ );
+ }
+ // Together the shards cover exactly coverage - indexed.
+ let expected = exclude_row_ranges(
+ &[RowRange::new(0, n)],
+ &[RowRange::new(hole_start, hole_end)],
+ )
+ .into_iter()
+ .map(|r| (r.from(), r.to()))
+ .collect::<Vec<_>>();
+ assert_eq!(
+ ranges, expected,
+ "shards must cover exactly coverage minus the indexed hole"
+ );
+ }
+
+ /// Planner-level incremental prefix. Strengthens
+ /// `lumina_incremental_build_indexes_only_new_rows`, which asserted only a
+ /// one-sided lower bound (`row_range_start >= n`): an indexed prefix [0,
k]
+ /// must leave EXACTLY the suffix [k+1, N] on both bounds, split along the
+ /// shard grid, with nothing re-indexed inside the prefix.
+ #[test]
+ fn lumina_plan_incremental_prefix_leaves_suffix() {
+ // Data row-ids [0, 24], rows_per_shard = 10 -> cells
[0,9],[10,19],[20,29].
+ // Indexed prefix [0, 9] fully fills the first cell, so the build must
be
+ // exactly [10, 19] and [20, 24] (the suffix split along the grid).
+ let n = 24;
+ let k = 9; // prefix [0, k] == the first full shard cell
+ let shards = plan_with_indexed(
+ vec![manifest_entry(data_file("a", Some(0), n + 1))],
+ 10,
+ &[RowRange::new(0, k)],
+ )
+ .unwrap();
+
+ let ranges = shards
+ .iter()
+ .map(|s| (s.row_range_start, s.row_range_end))
+ .collect::<Vec<_>>();
+ assert_eq!(
+ ranges,
+ vec![(k + 1, 19), (20, n)],
+ "indexed prefix must leave exactly the suffix, split along the
shard grid"
+ );
+ // Both bounds pinned (this is what the one-sided existing check
omits).
+ assert_eq!(ranges.first().unwrap().0, k + 1, "suffix must start at
k+1");
+ assert_eq!(ranges.last().unwrap().1, n, "suffix must end at N");
+ // Contiguous, and no shard reaches back into the indexed prefix.
+ for pair in ranges.windows(2) {
+ assert_eq!(
+ pair[1].0,
+ pair[0].1 + 1,
+ "ranges must be contiguous: {:?} then {:?}",
+ pair[0],
+ pair[1]
+ );
+ }
+ for (start, end) in &ranges {
+ assert!(
+ *start > k,
+ "shard [{start}, {end}] must not re-index the prefix [0, {k}]"
+ );
+ }
+ }
}
diff --git a/crates/paimon/src/table/mod.rs b/crates/paimon/src/table/mod.rs
index 373efa5..70eef2d 100644
--- a/crates/paimon/src/table/mod.rs
+++ b/crates/paimon/src/table/mod.rs
@@ -39,6 +39,7 @@ mod data_file_reader;
mod data_file_writer;
#[cfg(feature = "fulltext")]
mod full_text_search_builder;
+pub(crate) mod global_index_build_common;
pub(crate) mod global_index_scanner;
mod global_index_types;
mod hybrid_search_builder;
diff --git a/crates/paimon/src/table/source.rs
b/crates/paimon/src/table/source.rs
index 7c85d6d..88ccfd0 100644
--- a/crates/paimon/src/table/source.rs
+++ b/crates/paimon/src/table/source.rs
@@ -130,6 +130,44 @@ pub fn merge_row_ranges(mut ranges: Vec<RowRange>) ->
Vec<RowRange> {
merged
}
+/// Subtract `subtract` from `base`, returning the remaining sorted,
non-overlapping
+/// inclusive ranges. Mirrors Java `Range.exclude`. Inputs need not be sorted
or
+/// merged; the result is normalized (sorted, merged-adjacent, non-empty
ranges).
+pub(crate) fn exclude_row_ranges(base: &[RowRange], subtract: &[RowRange]) ->
Vec<RowRange> {
+ let base = merge_row_ranges(base.to_vec());
+ let subtract = merge_row_ranges(subtract.to_vec());
+ let mut result = Vec::new();
+ for seg in base {
+ // Current uncovered cursor within [seg.from, seg.to].
+ let mut cursor = seg.from();
+ // Set once a cut covers through seg.to(); the trailing suffix must
then
+ // be suppressed. Distinguishing this from `cursor > seg.to()` is what
+ // keeps the i64::MAX edge correct: at i64::MAX, `saturating_add(1)`
+ // cannot push the cursor past seg.to(), so we rely on this flag
instead.
+ let mut exhausted = false;
+ for cut in &subtract {
+ if cut.to() < cursor || cut.from() > seg.to() {
+ continue; // cut entirely before the cursor or beyond this
segment
+ }
+ if cut.from() > cursor {
+ // Emit the gap before this cut.
+ result.push(RowRange::new(cursor, cut.from() - 1));
+ }
+ if cut.to() >= seg.to() {
+ // The cut reaches or passes the segment end; nothing remains.
+ exhausted = true;
+ break;
+ }
+ // Advance the cursor past the cut (never move it backwards).
+ cursor = cursor.max(cut.to() + 1);
+ }
+ if !exhausted && cursor <= seg.to() {
+ result.push(RowRange::new(cursor, seg.to()));
+ }
+ }
+ result
+}
+
#[cfg(test)]
mod row_range_tests {
use super::*;
@@ -905,4 +943,105 @@ mod tests {
.unwrap();
assert_eq!(unknown.merged_row_count(), None);
}
+
+ fn r(from: i64, to: i64) -> RowRange {
+ RowRange::new(from, to)
+ }
+
+ fn pairs(ranges: &[RowRange]) -> Vec<(i64, i64)> {
+ ranges.iter().map(|x| (x.from(), x.to())).collect()
+ }
+
+ #[test]
+ fn exclude_empty_subtract_returns_base_merged() {
+ // No subtraction -> base, normalized (sorted, merged-adjacent).
+ let base = vec![r(0, 4), r(5, 9)];
+ assert_eq!(pairs(&exclude_row_ranges(&base, &[])), vec![(0, 9)]);
+ }
+
+ #[test]
+ fn exclude_empty_base_returns_empty() {
+ assert!(exclude_row_ranges(&[], &[r(0, 9)]).is_empty());
+ }
+
+ #[test]
+ fn exclude_no_overlap_returns_base() {
+ let base = vec![r(10, 19)];
+ let sub = vec![r(0, 9), r(20, 29)];
+ assert_eq!(pairs(&exclude_row_ranges(&base, &sub)), vec![(10, 19)]);
+ }
+
+ #[test]
+ fn exclude_prefix_leaves_suffix() {
+ // The core incremental case: indexed [0,2], data [0,9] -> build [3,9].
+ let base = vec![r(0, 9)];
+ let sub = vec![r(0, 2)];
+ assert_eq!(pairs(&exclude_row_ranges(&base, &sub)), vec![(3, 9)]);
+ }
+
+ #[test]
+ fn exclude_middle_hole_splits_range() {
+ let base = vec![r(0, 9)];
+ let sub = vec![r(3, 5)];
+ assert_eq!(
+ pairs(&exclude_row_ranges(&base, &sub)),
+ vec![(0, 2), (6, 9)]
+ );
+ }
+
+ #[test]
+ fn exclude_full_cover_returns_empty() {
+ let base = vec![r(0, 9)];
+ let sub = vec![r(0, 9)];
+ assert!(exclude_row_ranges(&base, &sub).is_empty());
+ }
+
+ #[test]
+ fn exclude_superset_cover_returns_empty() {
+ let base = vec![r(2, 7)];
+ let sub = vec![r(0, 100)];
+ assert!(exclude_row_ranges(&base, &sub).is_empty());
+ }
+
+ #[test]
+ fn exclude_multiple_subtract_ranges_and_boundaries() {
+ // Adjacent subtract ranges must not leave a spurious 1-row gap.
+ let base = vec![r(0, 20)];
+ let sub = vec![r(0, 4), r(5, 9), r(15, 15)];
+ assert_eq!(
+ pairs(&exclude_row_ranges(&base, &sub)),
+ vec![(10, 14), (16, 20)]
+ );
+ }
+
+ #[test]
+ fn exclude_multi_base_segments() {
+ let base = vec![r(0, 4), r(10, 14)];
+ let sub = vec![r(2, 11)];
+ assert_eq!(
+ pairs(&exclude_row_ranges(&base, &sub)),
+ vec![(0, 1), (12, 14)]
+ );
+ }
+
+ #[test]
+ fn exclude_saturates_at_i64_max_boundary() {
+ // A cut covering through i64::MAX must exhaust the segment, not emit a
+ // spurious (i64::MAX, i64::MAX). Guards the shared primitive against
the
+ // i64::MAX saturation edge (read + write paths both rely on this fn).
+ let base = vec![RowRange::new(5, i64::MAX)];
+ let sub = vec![RowRange::new(5, i64::MAX)];
+ assert!(exclude_row_ranges(&base, &sub).is_empty());
+
+ // Partial cut up to i64::MAX from the middle leaves the prefix only.
+ let base = vec![RowRange::new(0, i64::MAX)];
+ let sub = vec![RowRange::new(10, i64::MAX)];
+ assert_eq!(
+ exclude_row_ranges(&base, &sub)
+ .iter()
+ .map(|r| (r.from(), r.to()))
+ .collect::<Vec<_>>(),
+ vec![(0, 9)]
+ );
+ }
}
diff --git a/crates/paimon/src/table/table_commit.rs
b/crates/paimon/src/table/table_commit.rs
index 872bea1..453d8aa 100644
--- a/crates/paimon/src/table/table_commit.rs
+++ b/crates/paimon/src/table/table_commit.rs
@@ -30,6 +30,7 @@ use crate::spec::{
PartitionStatistics, Predicate, Snapshot, EMPTY_SERIALIZED_ROW,
MANIFEST_ENTRY_SCHEMA,
};
use crate::table::commit_message::CommitMessage;
+use crate::table::global_index_build_common::same_extra_field_ids;
use crate::table::partition_filter::PartitionFilter;
use crate::table::snapshot_commit::SnapshotCommit;
use crate::table::{SnapshotManager, Table, TableScan};
@@ -1384,6 +1385,11 @@ impl TableCommit {
continue;
};
if retained_meta.index_field_id == added_meta.index_field_id
+ && retained.index_file.index_type ==
added.index_file.index_type
+ && same_extra_field_ids(
+ retained_meta.extra_field_ids.as_deref(),
+ added_meta.extra_field_ids.as_deref(),
+ )
&& ranges_overlap(
retained_meta.row_range_start,
retained_meta.row_range_end,
@@ -1419,6 +1425,11 @@ impl TableCommit {
continue;
};
if left_meta.index_field_id == right_meta.index_field_id
+ && left.index_file.index_type ==
right.index_file.index_type
+ && same_extra_field_ids(
+ left_meta.extra_field_ids.as_deref(),
+ right_meta.extra_field_ids.as_deref(),
+ )
&& ranges_overlap(
left_meta.row_range_start,
left_meta.row_range_end,
@@ -3322,6 +3333,132 @@ mod tests {
assert_eq!(index_entries.len(), 2);
}
+ fn add_index_entry(file: IndexFileMeta) -> IndexManifestEntry {
+ IndexManifestEntry {
+ kind: FileKind::Add,
+ partition: vec![],
+ bucket: 0,
+ index_file: file,
+ version: 1,
+ }
+ }
+
+ #[test]
+ fn test_global_index_overlap_allows_different_index_type() {
+ // Same field id (0) and overlapping row ranges, but the retained
entry is
+ // a `lumina` index and the added one is `btree`: distinct identities
that
+ // must coexist rather than trip the overlap guard.
+ let lumina = test_global_index_file("lumina-0.index", 0, 0, 9);
+ let mut btree = test_global_index_file("btree-0.index", 0, 5, 14);
+ btree.index_type = "btree".to_string();
+ let retained = vec![add_index_entry(lumina)];
+ let added = vec![add_index_entry(btree)];
+ TableCommit::validate_global_index_overlap(&retained, &added)
+ .expect("different index types on the same field must coexist");
+ }
+
+ #[test]
+ fn test_global_index_overlap_allows_different_extra_field_ids() {
+ // Same index type and field id, overlapping ranges, but different
+ // `extra_field_ids` -> distinct identities, must coexist.
+ let retained =
vec![add_index_entry(test_global_index_file_with_extra_fields(
+ "lumina-0.index",
+ 0,
+ vec![1],
+ 0,
+ 9,
+ ))];
+ let added =
vec![add_index_entry(test_global_index_file_with_extra_fields(
+ "lumina-1.index",
+ 0,
+ vec![2],
+ 5,
+ 14,
+ ))];
+ TableCommit::validate_global_index_overlap(&retained, &added)
+ .expect("same field but different extra_field_ids must coexist");
+ }
+
+ #[test]
+ fn test_global_index_overlap_rejects_same_identity() {
+ // Identical identity (same type + field + extra) with overlapping
ranges
+ // must still be rejected.
+ let retained = vec![add_index_entry(test_global_index_file(
+ "lumina-0.index",
+ 0,
+ 0,
+ 9,
+ ))];
+ let added = vec![add_index_entry(test_global_index_file(
+ "lumina-1.index",
+ 0,
+ 5,
+ 14,
+ ))];
+ let result = TableCommit::validate_global_index_overlap(&retained,
&added);
+ let err_msg = result.expect_err("same-identity overlap must be
rejected");
+ let err_msg = err_msg.to_string();
+ assert!(
+ err_msg.contains("overlapping row range"),
+ "expected overlap error, got: {err_msg}"
+ );
+ }
+
+ #[test]
+ fn test_added_global_index_overlap_allows_different_index_type() {
+ // Two entries added within the same commit: same field id (0) and
+ // overlapping row ranges, but distinct index types (`lumina` vs
`btree`).
+ // These are distinct global-index identities that must coexist.
+ let lumina = test_global_index_file("lumina-0.index", 0, 0, 9);
+ let mut btree = test_global_index_file("btree-0.index", 0, 5, 14);
+ btree.index_type = "btree".to_string();
+ let added = vec![add_index_entry(lumina), add_index_entry(btree)];
+ TableCommit::validate_added_global_index_overlap(&added)
+ .expect("different index types on the same field must coexist
within a commit");
+ }
+
+ #[test]
+ fn test_added_global_index_overlap_allows_different_extra_field_ids() {
+ // Two entries added within the same commit: same index type and field
id,
+ // overlapping ranges, but different `extra_field_ids` -> distinct
+ // identities, must coexist.
+ let added = vec![
+ add_index_entry(test_global_index_file_with_extra_fields(
+ "lumina-0.index",
+ 0,
+ vec![1],
+ 0,
+ 9,
+ )),
+ add_index_entry(test_global_index_file_with_extra_fields(
+ "lumina-1.index",
+ 0,
+ vec![2],
+ 5,
+ 14,
+ )),
+ ];
+ TableCommit::validate_added_global_index_overlap(&added)
+ .expect("same field but different extra_field_ids must coexist
within a commit");
+ }
+
+ #[test]
+ fn test_added_global_index_overlap_rejects_same_identity() {
+ // Two entries added within the same commit with identical identity
+ // (same type + field + extra) and overlapping ranges must be rejected.
+ let added = vec![
+ add_index_entry(test_global_index_file("lumina-0.index", 0, 0, 9)),
+ add_index_entry(test_global_index_file("lumina-1.index", 0, 5,
14)),
+ ];
+ let result = TableCommit::validate_added_global_index_overlap(&added);
+ let err_msg = result.expect_err("same-identity overlap must be
rejected");
+ let err_msg = err_msg.to_string();
+ assert!(
+ err_msg.contains("overlapping row range"),
+ "expected overlap error, got: {err_msg}"
+ );
+ }
+
#[tokio::test]
async fn test_index_delete_removes_previous_index_manifest_entry() {
let file_io = test_file_io();
diff --git a/crates/paimon/src/table/vindex_index_build_builder.rs
b/crates/paimon/src/table/vindex_index_build_builder.rs
index 7985411..7adb48a 100644
--- a/crates/paimon/src/table/vindex_index_build_builder.rs
+++ b/crates/paimon/src/table/vindex_index_build_builder.rs
@@ -17,8 +17,9 @@
use crate::spec::{
bucket_dir_name, BinaryRow, CoreOptions, DataField, DataFileMeta,
DataType, FileKind,
- GlobalIndexMeta, IndexFileMeta, IndexManifest, ROW_ID_FIELD_NAME,
+ GlobalIndexMeta, IndexFileMeta, ROW_ID_FIELD_NAME,
};
+use crate::table::source::exclude_row_ranges;
use crate::table::{
CommitMessage, DataSplitBuilder, RowRange, SnapshotManager, Table,
TableCommit,
};
@@ -117,6 +118,14 @@ impl<'a> VindexIndexBuildBuilder<'a> {
.with_scan_all_files()
.plan_manifest_entries(&snapshot)
.await?;
+ let indexed =
crate::table::global_index_build_common::indexed_row_ranges(
+ self.table,
+ snapshot.index_manifest(),
+ &self.index_type,
+ index_field.id(),
+ None, // single-column build; no extra fields today
+ )
+ .await?;
let shards = plan_vindex_shards(
self.table.location(),
self.table.schema().partition_keys(),
@@ -125,16 +134,22 @@ impl<'a> VindexIndexBuildBuilder<'a> {
snapshot.id(),
manifest_entries,
rows_per_shard,
+ &indexed,
)?;
if shards.is_empty() {
return Ok(0);
}
- validate_existing_index_overlap(
+
crate::table::global_index_build_common::validate_existing_index_overlap(
self.table,
snapshot.index_manifest(),
+ &self.index_type,
index_field.id(),
- &shards,
+ None,
+ &shards
+ .iter()
+ .map(|shard| RowRange::new(shard.row_range_start,
shard.row_range_end))
+ .collect::<Vec<_>>(),
)
.await?;
@@ -336,6 +351,7 @@ fn validate_vector_field(field: &DataField) -> Result<()> {
Ok(())
}
+#[allow(clippy::too_many_arguments)]
fn plan_vindex_shards(
table_location: &str,
partition_keys: &[String],
@@ -344,6 +360,7 @@ fn plan_vindex_shards(
snapshot_id: i64,
entries: Vec<crate::spec::ManifestEntry>,
rows_per_shard: i64,
+ indexed: &[RowRange],
) -> Result<Vec<VindexIndexShard>> {
if rows_per_shard <= 0 {
return Err(Error::DataInvalid {
@@ -426,19 +443,30 @@ fn plan_vindex_shards(
.map(|file| file.row_id_range().unwrap().1)
.max()
.unwrap();
- let row_range_start = group_start.max(shard_start);
- let row_range_end = group_end.min(shard_end);
- result.push(VindexIndexShard {
- partition: partition.clone(),
- partition_bytes: partition_bytes.clone(),
- files: group,
- row_range_start,
- row_range_end,
- snapshot_id,
- source_bucket,
- total_buckets,
- bucket_path: bucket_path.clone(),
- });
+ // Coverage of this group clamped to the current shard cell.
Then
+ // subtract the already-indexed ranges so the build only covers
+ // the gap. Because grid-clamp and gap-subtraction are both
range
+ // intersections, applying the gap here is equivalent to
btree's
+ // "exclude then split" -- and each surviving segment stays
inside
+ // one shard cell, preserving per-shard row-id contiguity (the
+ // reader errors on a row-id gap within a shard).
+ let coverage_start = group_start.max(shard_start);
+ let coverage_end = group_end.min(shard_end);
+ let build_segments =
+ exclude_row_ranges(&[RowRange::new(coverage_start,
coverage_end)], indexed);
+ for seg in build_segments {
+ result.push(VindexIndexShard {
+ partition: partition.clone(),
+ partition_bytes: partition_bytes.clone(),
+ files: group.clone(),
+ row_range_start: seg.from(),
+ row_range_end: seg.to(),
+ snapshot_id,
+ source_bucket,
+ total_buckets,
+ bucket_path: bucket_path.clone(),
+ });
+ }
}
}
}
@@ -515,51 +543,6 @@ fn bucket_path(
))
}
-async fn validate_existing_index_overlap(
- table: &Table,
- index_manifest_name: Option<&str>,
- index_field_id: i32,
- shards: &[VindexIndexShard],
-) -> Result<()> {
- let Some(index_manifest_name) = index_manifest_name else {
- return Ok(());
- };
- let path = format!(
- "{}/manifest/{}",
- table.location().trim_end_matches('/'),
- index_manifest_name
- );
- let entries = IndexManifest::read(table.file_io(), &path).await?;
- for entry in entries {
- if entry.kind != FileKind::Add {
- continue;
- }
- let Some(meta) = entry.index_file.global_index_meta else {
- continue;
- };
- if meta.index_field_id != index_field_id {
- continue;
- }
- if shards.iter().any(|shard| {
- ranges_overlap(
- meta.row_range_start,
- meta.row_range_end,
- shard.row_range_start,
- shard.row_range_end,
- )
- }) {
- return Err(Error::DataInvalid {
- message: format!(
- "Existing global index file '{}' overlaps requested row
range for field {}",
- entry.index_file.file_name, index_field_id
- ),
- source: None,
- });
- }
- }
- Ok(())
-}
-
async fn extract_vectors(
table: &Table,
shard: &VindexIndexShard,
@@ -794,19 +777,19 @@ fn validate_vector_buffer(vectors: &[f32], row_count:
i32, dimension: i32) -> Re
Ok(())
}
-fn ranges_overlap(left_start: i64, left_end: i64, right_start: i64, right_end:
i64) -> bool {
- left_start <= right_end && right_start <= left_end
-}
-
#[cfg(test)]
mod tests {
use super::*;
use crate::catalog::Identifier;
- use crate::io::FileIOBuilder;
+ use crate::io::{FileIO, FileIOBuilder};
use crate::spec::stats::BinaryTableStats;
- use crate::spec::{ArrayType, FloatType, IntType, ManifestEntry, Schema,
TableSchema};
+ use crate::spec::{
+ ArrayType, FloatType, IndexManifest, IntType, ManifestEntry, Schema,
TableSchema,
+ };
+ use crate::table::TableWrite;
+ use crate::vindex::IVF_FLAT_IDENTIFIER;
use arrow_array::builder::{Float32Builder, Int64Builder, ListBuilder};
- use arrow_array::ArrayRef;
+ use arrow_array::{ArrayRef, Int32Array};
use arrow_schema::{DataType as ArrowDataType, Field as ArrowField, Schema
as ArrowSchema};
use chrono::{DateTime, Utc};
use std::sync::Arc;
@@ -876,6 +859,14 @@ mod tests {
}
fn plan(entries: Vec<ManifestEntry>, rows_per_shard: i64) ->
Result<Vec<VindexIndexShard>> {
+ plan_with_indexed(entries, rows_per_shard, &[])
+ }
+
+ fn plan_with_indexed(
+ entries: Vec<ManifestEntry>,
+ rows_per_shard: i64,
+ indexed: &[RowRange],
+ ) -> Result<Vec<VindexIndexShard>> {
let table = test_table(table_options(&rows_per_shard.to_string()));
let core = CoreOptions::new(table.schema().options());
plan_vindex_shards(
@@ -886,6 +877,7 @@ mod tests {
1,
entries,
rows_per_shard,
+ indexed,
)
}
@@ -992,4 +984,506 @@ mod tests {
matches!(err, Error::DataInvalid { message, .. } if
message.contains("dimension mismatch"))
);
}
+
+ fn test_table_with_io(file_io: FileIO, table_path: &str, schema: Schema)
-> Table {
+ Table::new(
+ file_io,
+ Identifier::new("default", "test_table"),
+ table_path.to_string(),
+ TableSchema::new(0, &schema),
+ None,
+ )
+ }
+
+ fn vindex_schema_builder(options: HashMap<String, String>) ->
crate::spec::SchemaBuilder {
+ Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column(
+ "embedding",
+
DataType::Array(ArrayType::new(DataType::Float(FloatType::new()))),
+ )
+ .options(options)
+ }
+
+ fn vindex_e2e_options(rows_per_shard: &str) -> HashMap<String, String> {
+ let mut options = table_options(rows_per_shard);
+ // A small, valid IVF config so the (optional) native build can run;
the
+ // no-op/incremental fix is exercised before or independently of it.
+ options.insert("ivf-flat.dimension".to_string(), "2".to_string());
+ options.insert("ivf-flat.nlist".to_string(), "2".to_string());
+ options
+ }
+
+ fn vindex_e2e_table(table_path: &str, rows_per_shard: &str) -> Table {
+ test_table_with_io(
+ FileIOBuilder::new("memory").build().unwrap(),
+ table_path,
+ vindex_schema_builder(vindex_e2e_options(rows_per_shard))
+ .build()
+ .unwrap(),
+ )
+ }
+
+ 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 build_vector_batch(ids: Vec<i32>, vectors: Vec<Vec<f32>>) ->
RecordBatch {
+ let element_field = Arc::new(ArrowField::new("element",
ArrowDataType::Float32, true));
+ let mut vector_builder =
+
ListBuilder::new(Float32Builder::new()).with_field(element_field.clone());
+ for vector in vectors {
+ for value in vector {
+ vector_builder.values().append_value(value);
+ }
+ vector_builder.append(true);
+ }
+ let schema = Arc::new(ArrowSchema::new(vec![
+ ArrowField::new("id", ArrowDataType::Int32, false),
+ ArrowField::new("embedding", ArrowDataType::List(element_field),
true),
+ ]));
+ RecordBatch::try_new(
+ schema,
+ vec![
+ Arc::new(Int32Array::from(ids)) as ArrayRef,
+ Arc::new(vector_builder.finish()) as ArrayRef,
+ ],
+ )
+ .unwrap()
+ }
+
+ async fn write_vectors(table: &Table, ids: Vec<i32>, vectors:
Vec<Vec<f32>>) {
+ let mut table_write = TableWrite::new(table,
"test-user".to_string()).unwrap();
+ table_write
+ .write_arrow_batch(&build_vector_batch(ids, vectors))
+ .await
+ .unwrap();
+ let messages = table_write.prepare_commit().await.unwrap();
+ TableCommit::new(table.clone(), "test-user".to_string())
+ .commit(messages)
+ .await
+ .unwrap();
+ }
+
+ /// Commit a synthetic vindex `IndexFileMeta` covering `[start, end]` for
+ /// `field_id` directly into the index manifest, without invoking the
native
+ /// builder. Mirrors the Lumina/btree tests so the incremental gap logic
can
+ /// be exercised without a trained vector index. Writes the same
`index_type`
+ /// (`ivf-flat`) the builder-under-test uses, so the gap helper matches it.
+ async fn commit_synthetic_vindex_index(table: &Table, field_id: i32,
start: i64, end: i64) {
+ let synthetic = IndexFileMeta {
+ index_type: IVF_FLAT_IDENTIFIER.to_string(),
+ file_name:
format!("vector-ivf-flat-synthetic-{start}-{end}.index"),
+ file_size: 1,
+ row_count: (end - start + 1) as i32,
+ deletion_vectors_ranges: None,
+ global_index_meta: Some(GlobalIndexMeta {
+ row_range_start: start,
+ row_range_end: end,
+ index_field_id: field_id,
+ extra_field_ids: None,
+ index_meta: None,
+ }),
+ };
+ let mut message =
CommitMessage::new(BinaryRow::new(0).to_serialized_bytes(), 0, vec![]);
+ message.new_index_files = vec![synthetic];
+ TableCommit::new(table.clone(), "test-user".to_string())
+ .commit(vec![message])
+ .await
+ .unwrap();
+ }
+
+ async fn latest_vindex_index_files(table: &Table) -> Vec<IndexFileMeta> {
+ let snapshot_manager =
+ SnapshotManager::new(table.file_io().clone(),
table.location().to_string());
+ let snapshot = snapshot_manager
+ .get_latest_snapshot()
+ .await
+ .unwrap()
+ .unwrap();
+ let Some(index_manifest_name) = snapshot.index_manifest() else {
+ return Vec::new();
+ };
+ IndexManifest::read(
+ table.file_io(),
+ &snapshot_manager.manifest_path(index_manifest_name),
+ )
+ .await
+ .unwrap()
+ .into_iter()
+ .filter(|entry| {
+ entry.kind == FileKind::Add && entry.index_file.index_type ==
IVF_FLAT_IDENTIFIER
+ })
+ .map(|entry| entry.index_file)
+ .collect()
+ }
+
+ /// Row-id coverage of the committed data files, read back from the data
+ /// manifest (never hard-coded) and merged into contiguous ranges. Mirrors
+ /// how `execute` gathers `manifest_entries`.
+ async fn data_row_id_coverage(table: &Table) -> Vec<RowRange> {
+ let snapshot_manager =
+ SnapshotManager::new(table.file_io().clone(),
table.location().to_string());
+ let snapshot = snapshot_manager
+ .get_latest_snapshot()
+ .await
+ .unwrap()
+ .unwrap();
+ let entries = table
+ .new_read_builder()
+ .new_scan()
+ .with_scan_all_files()
+ .plan_manifest_entries(&snapshot)
+ .await
+ .unwrap();
+ let ranges = entries
+ .iter()
+ .filter(|entry| *entry.kind() == FileKind::Add)
+ .filter_map(|entry| {
+ entry
+ .file()
+ .row_id_range()
+ .map(|(start, end)| RowRange::new(start, end))
+ })
+ .collect::<Vec<_>>();
+ crate::table::merge_row_ranges(ranges)
+ }
+
+ /// Second build with the whole coverage already indexed must be a clean
+ /// no-op (returns 0), not an overlap error. Reaches `Ok(0)` before the
+ /// native build, so it runs in CI without a trained index. This is the
core
+ /// bug fix: today the second call errors with the overlap message.
+ #[tokio::test]
+ async fn vindex_second_build_without_new_data_is_noop() {
+ let table_path = "memory:/test_vindex_second_build_noop";
+ let table = vindex_e2e_table(table_path, "10");
+ setup_dirs(table.file_io(), table_path).await;
+
+ write_vectors(
+ &table,
+ vec![1, 2, 3],
+ vec![vec![1.0, 0.0], vec![0.0, 1.0], vec![1.0, 1.0]],
+ )
+ .await;
+
+ // Fully index the coverage via a synthetic manifest entry.
+ let coverage = data_row_id_coverage(&table).await;
+ assert_eq!(coverage.len(), 1, "data must be one contiguous range");
+ let field_id = find_index_field(&table, "embedding").unwrap().id();
+ commit_synthetic_vindex_index(&table, field_id, coverage[0].from(),
coverage[0].to()).await;
+
+ let names_before = latest_vindex_index_files(&table)
+ .await
+ .iter()
+ .map(|f| f.file_name.clone())
+ .collect::<std::collections::BTreeSet<_>>();
+ assert!(!names_before.is_empty());
+
+ let built = table
+ .new_vindex_index_build_builder(IVF_FLAT_IDENTIFIER)
+ .with_index_column("embedding")
+ .execute()
+ .await
+ .unwrap();
+ assert_eq!(built, 0, "fully-indexed table must build nothing on
re-run");
+
+ let names_after = latest_vindex_index_files(&table)
+ .await
+ .iter()
+ .map(|f| f.file_name.clone())
+ .collect::<std::collections::BTreeSet<_>>();
+ assert_eq!(
+ names_before, names_after,
+ "re-run must not add or remove index manifest entries"
+ );
+ }
+
+ /// Real end-to-end incremental build. `paimon-vindex-core` is pure Rust
and
+ /// trains/serializes an IVF-flat index in CI without a native lib, so this
+ /// asserts SUCCESS end-to-end (mirroring btree's incremental test): build
#1
+ /// indexes the initial rows, an appended batch is indexed by build #2,
every
+ /// new index file's row range lies entirely in the appended gap `[n, ..]`
+ /// (`n` derived from the manifest, never hard-coded), and build-#1's index
+ /// files are retained untouched (append-only). No overlap error, no
tolerated
+ /// native-build failure -- the build must actually succeed.
+ #[tokio::test]
+ async fn vindex_incremental_build_indexes_only_new_rows() {
+ let table_path = "memory:/test_vindex_incremental";
+ let table = vindex_e2e_table(table_path, "10");
+ setup_dirs(table.file_io(), table_path).await;
+
+ // Build #1 over the initial batch via a real end-to-end build.
+ write_vectors(
+ &table,
+ vec![1, 2, 3],
+ vec![vec![1.0, 0.0], vec![0.0, 1.0], vec![1.0, 1.0]],
+ )
+ .await;
+ let first_built = table
+ .new_vindex_index_build_builder(IVF_FLAT_IDENTIFIER)
+ .with_index_column("embedding")
+ .execute()
+ .await
+ .unwrap();
+ assert!(first_built > 0, "first build must index the initial rows");
+
+ // First appended row-id, derived from the data manifest (never
hard-coded).
+ let indexed_coverage = data_row_id_coverage(&table).await;
+ assert_eq!(indexed_coverage.len(), 1);
+ let n = indexed_coverage[0].to() + 1;
+
+ let first_names = latest_vindex_index_files(&table)
+ .await
+ .iter()
+ .map(|f| f.file_name.clone())
+ .collect::<std::collections::BTreeSet<_>>();
+ assert!(!first_names.is_empty(), "build #1 must write index files");
+
+ // Append a second batch (new row-ids [n..]).
+ write_vectors(
+ &table,
+ vec![4, 5, 6],
+ vec![vec![2.0, 0.0], vec![0.0, 2.0], vec![2.0, 2.0]],
+ )
+ .await;
+
+ // End-to-end: build #2 must SUCCEED and index the appended rows.
+ let second_built = table
+ .new_vindex_index_build_builder(IVF_FLAT_IDENTIFIER)
+ .with_index_column("embedding")
+ .execute()
+ .await
+ .unwrap();
+ assert!(second_built > 0, "appended rows must be indexed");
+
+ let all_files = latest_vindex_index_files(&table).await;
+ let all_names = all_files
+ .iter()
+ .map(|f| f.file_name.clone())
+ .collect::<std::collections::BTreeSet<_>>();
+
+ // Every build-#1 file is still present (append-only, no
rewrite/delete).
+ assert!(
+ first_names.iter().all(|name| all_names.contains(name)),
+ "build #1 index files must be retained untouched"
+ );
+
+ // Every build-#2 file covers only the appended gap [n, ..], never the
+ // already-indexed prefix.
+ let new_files = all_files
+ .iter()
+ .filter(|f| !first_names.contains(&f.file_name))
+ .collect::<Vec<_>>();
+ assert!(!new_files.is_empty(), "build #2 must add new index files");
+ for file in new_files {
+ let meta = file
+ .global_index_meta
+ .as_ref()
+ .expect("global index meta on new vindex file");
+ assert!(
+ meta.row_range_start >= n,
+ "new index file range must start at or after {n}, got [{},
{}]",
+ meta.row_range_start,
+ meta.row_range_end
+ );
+ }
+ }
+
+ /// A field that already carries a DIFFERENT index type (`lumina`) over an
+ /// overlapping row range must not block a vindex (`ivf-flat`) build on the
+ /// same field: the two indexes have distinct identities and coexist.
Before
+ /// the full-identity fix, the overlap guard keyed only on field id + range
+ /// and spuriously rejected this build with the "overlaps requested row
+ /// range" error.
+ #[tokio::test]
+ async fn vindex_build_coexists_with_different_index_type_on_same_field() {
+ let table_path = "memory:/test_vindex_coexist_diff_type";
+ let table = vindex_e2e_table(table_path, "10");
+ setup_dirs(table.file_io(), table_path).await;
+
+ write_vectors(
+ &table,
+ vec![1, 2, 3],
+ vec![vec![1.0, 0.0], vec![0.0, 1.0], vec![1.0, 1.0]],
+ )
+ .await;
+
+ // Pre-existing `lumina` index covering the full data range on the SAME
+ // field the vindex build will target.
+ let coverage = data_row_id_coverage(&table).await;
+ assert_eq!(coverage.len(), 1, "data must be one contiguous range");
+ let field_id = find_index_field(&table, "embedding").unwrap().id();
+ let lumina = IndexFileMeta {
+ index_type: "lumina".to_string(),
+ file_name: "lumina-synthetic-0.index".to_string(),
+ file_size: 1,
+ row_count: (coverage[0].to() - coverage[0].from() + 1) as i32,
+ deletion_vectors_ranges: None,
+ global_index_meta: Some(GlobalIndexMeta {
+ row_range_start: coverage[0].from(),
+ row_range_end: coverage[0].to(),
+ index_field_id: field_id,
+ extra_field_ids: None,
+ index_meta: None,
+ }),
+ };
+ let mut message =
CommitMessage::new(BinaryRow::new(0).to_serialized_bytes(), 0, vec![]);
+ message.new_index_files = vec![lumina];
+ TableCommit::new(table.clone(), "test-user".to_string())
+ .commit(vec![message])
+ .await
+ .unwrap();
+
+ // Building `ivf-flat` on the same field must NOT trip the overlap
guard.
+ // A native-build failure over the tiny synthetic dataset is tolerated;
+ // only the overlap error is forbidden.
+ let result = table
+ .new_vindex_index_build_builder(IVF_FLAT_IDENTIFIER)
+ .with_index_column("embedding")
+ .execute()
+ .await;
+ match result {
+ Ok(_) => {}
+ Err(Error::DataInvalid { message, .. }) => {
+ assert!(
+ !message.contains("overlaps requested row range"),
+ "vindex build must coexist with a different-type index on
the same field; got: {message}"
+ );
+ }
+ Err(other) => panic!("unexpected error from vindex build:
{other:?}"),
+ }
+ }
+
+ /// Regression: a first build (no existing index) must equal the pre-change
+ /// full build -- subtracting an empty `indexed` yields full coverage.
+ #[test]
+ fn vindex_first_build_indexes_full_coverage() {
+ let full = plan(vec![manifest_entry(data_file("a", Some(0), 25))],
10).unwrap();
+ let gapped =
+ plan_with_indexed(vec![manifest_entry(data_file("a", Some(0),
25))], 10, &[]).unwrap();
+ // Empty `indexed` must not alter the shard layout.
+ assert_eq!(
+ full.iter()
+ .map(|s| (s.row_range_start, s.row_range_end))
+ .collect::<Vec<_>>(),
+ gapped
+ .iter()
+ .map(|s| (s.row_range_start, s.row_range_end))
+ .collect::<Vec<_>>()
+ );
+ assert_eq!(
+ full.iter()
+ .map(|s| (s.row_range_start, s.row_range_end))
+ .collect::<Vec<_>>(),
+ vec![(0, 9), (10, 19), (20, 24)],
+ "first build must cover the full row range across shards"
+ );
+ }
+
+ /// Planner-level mid-coverage hole, mirroring btree/lumina: with a single
+ /// shard cell (rows_per_shard large enough to hold all data) the grid
never
+ /// splits, so the only split is the indexed hole itself. An indexed range
+ /// strictly inside the data coverage must carve the build into exactly the
+ /// two contiguous segments on either side of the hole -- both bounds
pinned,
+ /// and neither segment may span or touch the hole.
+ #[test]
+ fn vindex_plan_splits_gap_around_mid_coverage_indexed_hole() {
+ // Data row-ids [0, 9]; one shard cell [0, 99] so the grid never
splits.
+ let n = 9;
+ let hole_start = 4;
+ let hole_end = 6;
+ let shards = plan_with_indexed(
+ vec![manifest_entry(data_file("a", Some(0), n + 1))],
+ 100,
+ &[RowRange::new(hole_start, hole_end)],
+ )
+ .unwrap();
+
+ let ranges = shards
+ .iter()
+ .map(|s| (s.row_range_start, s.row_range_end))
+ .collect::<Vec<_>>();
+ // Exactly the two contiguous segments around the hole.
+ assert_eq!(
+ ranges,
+ vec![(0, hole_start - 1), (hole_end + 1, n)],
+ "mid-coverage hole must split into exactly the two segments around
it"
+ );
+ // Every emitted range is contiguous and none spans or touches the
hole.
+ for (start, end) in &ranges {
+ assert!(end >= start, "range must be non-empty: [{start}, {end}]");
+ assert!(
+ *end < hole_start || *start > hole_end,
+ "shard [{start}, {end}] must not overlap indexed hole
[{hole_start}, {hole_end}]"
+ );
+ }
+ // Together the shards cover exactly coverage - indexed.
+ let expected = exclude_row_ranges(
+ &[RowRange::new(0, n)],
+ &[RowRange::new(hole_start, hole_end)],
+ )
+ .into_iter()
+ .map(|r| (r.from(), r.to()))
+ .collect::<Vec<_>>();
+ assert_eq!(
+ ranges, expected,
+ "shards must cover exactly coverage minus the indexed hole"
+ );
+ }
+
+ /// Planner-level incremental prefix. Strengthens
+ /// `vindex_incremental_build_indexes_only_new_rows`, which asserts only a
+ /// one-sided lower bound (`row_range_start >= n`): an indexed prefix [0,
k]
+ /// must leave EXACTLY the suffix [k+1, N] on both bounds, split along the
+ /// shard grid, with nothing re-indexed inside the prefix.
+ #[test]
+ fn vindex_plan_incremental_prefix_leaves_suffix() {
+ // Data row-ids [0, 24], rows_per_shard = 10 -> cells
[0,9],[10,19],[20,29].
+ // Indexed prefix [0, 9] fully fills the first cell, so the build must
be
+ // exactly [10, 19] and [20, 24] (the suffix split along the grid).
+ let n = 24;
+ let k = 9; // prefix [0, k] == the first full shard cell
+ let shards = plan_with_indexed(
+ vec![manifest_entry(data_file("a", Some(0), n + 1))],
+ 10,
+ &[RowRange::new(0, k)],
+ )
+ .unwrap();
+
+ let ranges = shards
+ .iter()
+ .map(|s| (s.row_range_start, s.row_range_end))
+ .collect::<Vec<_>>();
+ assert_eq!(
+ ranges,
+ vec![(k + 1, 19), (20, n)],
+ "indexed prefix must leave exactly the suffix, split along the
shard grid"
+ );
+ // Both bounds pinned (this is what the one-sided existing check
omits).
+ assert_eq!(ranges.first().unwrap().0, k + 1, "suffix must start at
k+1");
+ assert_eq!(ranges.last().unwrap().1, n, "suffix must end at N");
+ // Contiguous, and no shard reaches back into the indexed prefix.
+ for pair in ranges.windows(2) {
+ assert_eq!(
+ pair[1].0,
+ pair[0].1 + 1,
+ "ranges must be contiguous: {:?} then {:?}",
+ pair[0],
+ pair[1]
+ );
+ }
+ for (start, end) in &ranges {
+ assert!(
+ *start > k,
+ "shard [{start}, {end}] must not re-index the prefix [0, {k}]"
+ );
+ }
+ }
}
