diff --git a/rust/lance/src/dataset.rs b/rust/lance/src/dataset.rs index f23d6adc419..4b6beb9e782 100644 --- a/rust/lance/src/dataset.rs +++ b/rust/lance/src/dataset.rs @@ -14,7 +14,6 @@ use lance_core::deepsize::DeepSizeOf; use crate::dataset::metadata::UpdateFieldMetadataBuilder; use crate::dataset::transaction::translate_schema_metadata_updates; -use crate::index::DatasetIndexExt; use crate::session::caches::{DSMetadataCache, ManifestKey, TransactionKey}; use crate::session::index_caches::DSIndexCache; use itertools::Itertools; @@ -130,7 +129,6 @@ use crate::dataset::cleanup::{CleanupOperation, CleanupPolicy, CleanupPolicyBuil use crate::dataset::refs::{BranchContents, BranchIdentifier, Branches, Tags}; use crate::dataset::sql::SqlQueryBuilder; use crate::datatypes::Schema; -use crate::index::retain_supported_indices; use crate::io::commit::{ DEFAULT_COMMIT_RETRY_TIMEOUT, commit_detached_transaction, commit_new_dataset, commit_transaction, detect_overlapping_fragments, @@ -804,12 +802,16 @@ impl Dataset { LittleEndian::read_u32(&last_block[offset_in_block..offset_in_block + 4]) as usize; let message_data = &last_block[offset_in_block + 4..offset_in_block + 4 + message_len]; let section = lance_table::format::pb::IndexSection::decode(message_data)?; - let mut indices: Vec = section + // Cached unfiltered: this is the same cache the commit path reads + // from, and an index this build cannot decode still has to survive + // into the next manifest. Version filtering happens on the way out, + // in `DatasetIndexExt::load_indices`. + let indices: Vec = section .indices .into_iter() .map(IndexMetadata::try_from) .collect::>>()?; - retain_supported_indices(&mut indices); + crate::index::warn_about_unsupported_indices(&indices); let ds_index_cache = session.index_cache.for_dataset(uri); let metadata_key = crate::session::index_caches::IndexMetadataKey { version: manifest_location.version, @@ -3086,8 +3088,12 @@ impl Dataset { rowids::validate_stable_row_ids(self).await?; - // Validate indices - let indices = self.load_indices().await?; + // Validate indices. Over the complete list: these checks are about what + // the manifest says, not about what this build can use, and duplicate + // uuids or overlapping coverage are no less corrupt for involving an + // index this build has no reader for. `migrate_indices` already runs the + // same overlap check over the complete list on every commit. + let indices = crate::index::load_all_indices(self).await?; self.validate_indices(&indices)?; Ok(()) diff --git a/rust/lance/src/dataset/optimize.rs b/rust/lance/src/dataset/optimize.rs index 5123f389625..c0f19a5aea1 100644 --- a/rust/lance/src/dataset/optimize.rs +++ b/rust/lance/src/dataset/optimize.rs @@ -99,7 +99,9 @@ use super::{WriteMode, WriteParams, cleanup_data_fragments, write_fragments_inte use crate::Dataset; use crate::Result; use crate::dataset::utils::CapturedRowIds; -use crate::index::DatasetIndexExt; +use crate::index::{ + DatasetIndexExt, DatasetIndexInternalExt, load_all_indices, unsupported_index_version, +}; use crate::io::commit::{DEFAULT_COMMIT_RETRY_TIMEOUT, commit_transaction, migrate_fragments}; use arrow::array::AsArray; use arrow::datatypes::{UInt8Type, UInt32Type, UInt64Type}; @@ -125,7 +127,8 @@ use lance_core::utils::tokio::get_num_compute_intensive_cpus; use lance_core::utils::tracing::{DATASET_COMPACTING_EVENT, TRACE_DATASET_EVENTS}; use lance_index::frag_reuse::{FRAG_REUSE_INDEX_NAME, FragReuseGroup}; use lance_index::is_system_index; -use lance_table::format::{Fragment, RowIdMeta}; +use lance_index::metrics::NoOpMetricsCollector; +use lance_table::format::{Fragment, IndexMetadata, RowIdMeta}; use roaring::{RoaringBitmap, RoaringTreemap}; use serde::{Deserialize, Serialize}; use tracing::{info, warn}; @@ -752,10 +755,38 @@ impl CompactionPlanner for DefaultCompactionPlanner { fragments.windows(2).all(|w| w[0].id() < w[1].id()), "fragments in manifest are not sorted" ); + // Without stable row ids a rewrite moves every row address, so the + // indices over the rewritten fragments have to be remapped in the same + // commit. An index this build cannot open cannot be remapped, so its + // fragments join the caller's own exclusions and are left uncompacted - + // taking the same path, which terminates the current bin rather than + // letting the candidates on either side of the gap be planned together. + let mut excluded_fragment_ids = self.excluded_fragment_ids.clone(); + if !dataset.manifest.uses_stable_row_ids() && !self.options.defer_index_remap { + let unremappable = unremappable_index_coverage(dataset) + .await? + .into_iter() + .fold(RoaringBitmap::new(), |mut covered, (_, fragments)| { + covered |= fragments; + covered + }); + if !unremappable.is_empty() { + // Otherwise a compaction that plans nothing looks like a + // compaction that found nothing to do. + log::info!( + "holding {} fragment(s) back from compaction: they are covered by an index \ + this build cannot read, and so cannot be remapped here", + unremappable.len(), + ); + } + excluded_fragment_ids |= unremappable; + } + let excluded_fragment_ids = &excluded_fragment_ids; + let mut fragment_metrics = futures::stream::iter(fragments) - .map(|fragment| async { + .map(|fragment| async move { if u32::try_from(fragment.id()) - .is_ok_and(|fragment_id| self.excluded_fragment_ids.contains(fragment_id)) + .is_ok_and(|fragment_id| excluded_fragment_ids.contains(fragment_id)) { Ok(None) } else { @@ -2064,29 +2095,120 @@ impl CandidateBin { } async fn load_index_fragmaps(dataset: &Dataset) -> Result> { - let indices = dataset.load_indices().await?; + // Coverage, not usability: these bitmaps decide the rewrite groups. Under + // stable row ids `Transaction::recalculate_fragment_bitmap` then rejects any + // group that splits an index's coverage, and it walks every index the new + // manifest carries - including the ones this build cannot read. Binning from + // the filtered view fails that check outright on a dataset holding an index + // written by a newer Lance. The same bitmaps also decide, through + // `any_group_indexed`, whether a deferred compaction writes the + // fragment-reuse index that a build which can read that index needs to + // repair its coverage. + let indices = load_all_indices(dataset).await?; let mut index_fragmaps = Vec::with_capacity(indices.len()); // System indices (fragment-reuse, mem-wal) don't define data coverage and // aren't remapped per rewrite group, so they must not constrain compaction // bins -- otherwise deferred compaction's fragment-reuse index repeatedly // splits the small-fragment run and they never coalesce. for index in indices.iter().filter(|idx| !is_system_index(idx)) { - if let Some(fragment_bitmap) = index.fragment_bitmap.as_ref() { - index_fragmaps.push(fragment_bitmap.clone()); - } else { - let dataset_at_index = dataset.checkout_version(index.dataset_version).await?; - // max_fragment_id is inclusive (the highest id); +1 for an exclusive - // upper bound so the last fragment is covered (None => empty range). - let frags = 0..dataset_at_index - .manifest - .max_fragment_id - .map_or(0, |m| m + 1); - index_fragmaps.push(RoaringBitmap::from_sorted_iter(frags).unwrap()); - } + index_fragmaps.push(index_fragment_coverage(dataset, index).await?); } Ok(index_fragmaps) } +/// The fragments an index segment covers, reconstructing the coverage of a +/// legacy segment that predates the bitmap from the dataset it was written +/// against. +async fn index_fragment_coverage( + dataset: &Dataset, + index: &IndexMetadata, +) -> Result { + if let Some(fragment_bitmap) = index.fragment_bitmap.as_ref() { + return Ok(fragment_bitmap.clone()); + } + let dataset_at_index = dataset.checkout_version(index.dataset_version).await?; + // max_fragment_id is inclusive (the highest id); +1 for an exclusive + // upper bound so the last fragment is covered (None => empty range). + let frags = 0..dataset_at_index + .manifest + .max_fragment_id + .map_or(0, |m| m + 1); + let mut coverage = RoaringBitmap::from_sorted_iter(frags).unwrap(); + // Reconstructed in the id space of the version the index was written + // against, which a later compaction has already moved on from. + // `load_all_indices` puts a stored bitmap into the current space by running + // it through the fragment-reuse index and leaves a `None` one alone, so a + // reconstruction has to take that step itself. Skipping it names the + // fragments a deferred compaction moved these rows out of, which is a set + // no rewrite can intersect - the guards below then wave through the rewrite + // of the fragment the rows actually live in. + if let Some(frag_reuse_index) = dataset.open_frag_reuse_index(&NoOpMetricsCollector).await? { + frag_reuse_index.remap_fragment_bitmap(&mut coverage)?; + } + Ok(coverage) +} + +/// Each index this build has no reader for, by name, and the fragments it covers. +/// +/// A rewrite moves every row address in the fragments it touches, and putting an +/// index back in step means opening it. A build that cannot open one cannot +/// remap it, so compacting the fragments it covers would leave it addressing +/// rows that are gone -- worse than the erase this whole path exists to prevent. +/// They are held out of the plan instead, and the rest of the table still +/// compacts. +/// +/// Only the eager remap path needs this. Stable row ids keep the addresses +/// across a rewrite, and `defer_index_remap` hands the repair to a build that +/// can read the index, through the fragment-reuse index it writes. +async fn unremappable_index_coverage(dataset: &Dataset) -> Result> { + let mut coverage = Vec::new(); + for index in load_all_indices(dataset).await?.iter() { + if is_system_index(index) || unsupported_index_version(index).is_none() { + continue; + } + coverage.push(( + index.name.clone(), + index_fragment_coverage(dataset, index).await?, + )); + } + Ok(coverage) +} + +/// Refuse a plan that rewrites fragments an index this build cannot read covers. +/// +/// [`DefaultCompactionPlanner`] keeps those fragments out of the plan, but +/// nothing forces a caller through it: `compact_files_with_planner` takes any +/// planner, [`CompactionPlan`] is public and serializable, and a distributed +/// driver hands [`commit_compaction`] results planned elsewhere. Committing such +/// a plan strands the index on fragment ids the rewrite deleted, so the commit +/// boundary refuses it rather than the planner alone. +async fn reject_unremappable_rewrite( + dataset: &Dataset, + completed_tasks: &[RewriteResult], +) -> Result<()> { + let rewritten = completed_tasks + .iter() + .flat_map(|task| task.original_fragments.iter()) + .filter_map(|fragment| u32::try_from(fragment.id).ok()) + .collect::(); + + for (name, covered) in unremappable_index_coverage(dataset).await? { + let blocked = covered & &rewritten; + if !blocked.is_empty() { + return Err(Error::invalid_input(format!( + "compaction would rewrite fragment(s) {:?}, which index {:?} covers. This build \ + has no reader for that index, so it cannot be remapped onto the rewritten \ + fragments and the commit would leave it addressing rows that no longer exist. \ + Plan with DefaultCompactionPlanner, which holds those fragments back, set \ + defer_index_remap, or compact from a build that can read the index.", + blocked.iter().collect::>(), + name, + ))); + } + } + Ok(()) +} + pub async fn plan_compaction( dataset: &Dataset, options: &CompactionOptions, @@ -2574,6 +2696,14 @@ pub async fn commit_compaction( return Ok(CompactionMetrics::default()); } + // Before anything is written or committed. The condition is the planner's, + // not `has_address_style`: a dataset whose only index is one this build + // cannot read captures no row addresses at all, which is exactly the plan + // that has to be refused here. + if !dataset.manifest.uses_stable_row_ids() && !options.defer_index_remap { + reject_unremappable_rewrite(dataset, &completed_tasks).await?; + } + let has_address_style = completed_tasks.iter().any(|t| t.row_addrs.is_some()); // Address-style results require immediate index remapping unless it is deferred. let needs_remapping = diff --git a/rust/lance/src/dataset/schema_evolution.rs b/rust/lance/src/dataset/schema_evolution.rs index 3bc9525eb52..81c142a5db3 100644 --- a/rust/lance/src/dataset/schema_evolution.rs +++ b/rust/lance/src/dataset/schema_evolution.rs @@ -12,7 +12,7 @@ use super::{ transaction::{Operation, Transaction}, write::cleanup_data_fragments, }; -use crate::index::DatasetIndexExt; +use crate::index::load_all_indices; use crate::{Error, Result, io::exec::Planner}; use arrow::compute::CastOptions; use arrow::compute::can_cast_types; @@ -798,9 +798,12 @@ pub(super) async fn alter_columns( // affected column(s). The current behavior is to drop such indices without // warning, which has caused production incidents where vector search silently // regressed to brute-force scan. We require users to explicitly drop the - // index before altering the column type, so the action is never silent. + // index before altering the column type, so the action is never silent. That + // includes an index this build has no reader for: the cast reassigns the + // field id, so carrying it forward is impossible and staying quiet about it + // is the silent drop this guard exists to abolish. if !cast_fields.is_empty() { - let indices = dataset.load_indices().await?; + let indices = load_all_indices(dataset).await?; let affected: Vec<&lance_table::format::IndexMetadata> = indices .iter() .filter(|idx| { @@ -1016,6 +1019,8 @@ fn exclude(source: &Schema, other: &Schema, version: &ConcreteFileVersion) -> Re mod test { use std::{collections::HashMap, fs, num::NonZero, path::Path as StdPath, sync::Mutex}; + use crate::index::DatasetIndexExt; + #[test] fn test_merge_introduces_required_field() { let schema = |fields: Vec| Schema::try_from(&ArrowSchema::new(fields)).unwrap(); diff --git a/rust/lance/src/index.rs b/rust/lance/src/index.rs index d10ec624d77..9175c66929a 100644 --- a/rust/lance/src/index.rs +++ b/rust/lance/src/index.rs @@ -1635,7 +1635,9 @@ impl DatasetIndexExt for Dataset { } async fn drop_index(&mut self, name: &str) -> Result<()> { - let indices = self.load_indices_by_name(name).await?; + // Removal never opens the index, so an index this build cannot read is + // still droppable - and has to be, since it is otherwise unremovable. + let indices = load_all_indices_by_name(self, name).await?; if indices.is_empty() { return Err(Error::index_not_found(format!("name={}", name))); } @@ -1800,68 +1802,20 @@ impl DatasetIndexExt for Dataset { } async fn load_indices(&self) -> Result>> { - let metadata_key = IndexMetadataKey { - version: self.version().version, - store_identity: &self.object_store.store_prefix, - }; - let mut indices = self - .index_cache - .get_or_insert_with_key(metadata_key, || async { - let mut loaded_indices = read_manifest_indexes( - &self.object_store, - &self.manifest_location, - &self.manifest, - ) - .await?; - retain_supported_indices(&mut loaded_indices); - Ok(loaded_indices) - }) - .await?; - - // Infer details for legacy vector indices (once per index name, concurrently). - // This may run on indices that were opportunistically cached during Dataset::open - // before the full Dataset was available for inference. - { - let schema = self.schema(); - if indices - .iter() - .any(|idx| needs_vector_details_inference(idx, schema)) - { - let mut updated = indices.as_ref().clone(); - infer_missing_vector_details(self, &mut updated).await; - if updated != *indices { - indices = Arc::new(updated); - self.index_cache - .insert_with_key(&metadata_key, indices.clone()) - .await; - } - } - } - - if let Some(frag_reuse_index_meta) = - indices.iter().find(|idx| idx.name == FRAG_REUSE_INDEX_NAME) + let indices = load_all_indices(self).await?; + if indices + .iter() + .all(|idx| unsupported_index_version(idx).is_none()) { - let fri_key = FragReuseIndexKey { - uuid: &frag_reuse_index_meta.uuid, - }; - let frag_reuse_index = self - .index_cache - .get_or_insert_with_key(fri_key, || async move { - let index_details = - load_frag_reuse_index_details(self, frag_reuse_index_meta).await?; - open_frag_reuse_index(frag_reuse_index_meta.uuid, index_details.as_ref()).await - }) - .await?; - let mut indices = indices.as_ref().clone(); - for idx in indices.iter_mut() { - if let Some(bitmap) = idx.fragment_bitmap.as_mut() { - frag_reuse_index.remap_fragment_bitmap(bitmap)?; - } - } - Ok(Arc::new(indices)) - } else { - Ok(indices) + return Ok(indices); } + Ok(Arc::new( + indices + .iter() + .filter(|idx| unsupported_index_version(idx).is_none()) + .cloned() + .collect(), + )) } async fn merge_existing_index_segments( @@ -2064,7 +2018,7 @@ impl DatasetIndexExt for Dataset { } } - let existing_named_indices = self.load_indices_by_name(index_name).await?; + let existing_named_indices = load_all_indices_by_name(self, index_name).await?; if existing_named_indices.iter().any(|idx| { // Same name-collision rule as `CreateIndexBuilder`'s default-name // loop in create.rs. @@ -2099,6 +2053,11 @@ impl DatasetIndexExt for Dataset { } let is_index_type_change = existing_different_type_url.is_some(); + // What a retained sibling has to agree with. Every incoming segment + // already carries the same pair: `build_index_metadata_from_segments` + // compares them against each other before this point. + let expected_fields = new_indices[0].fields.clone(); + let expected_covering_fields = new_indices[0].covering_fields.clone(); let removed_indices = existing_named_indices .into_iter() .map(|idx| -> Result> { @@ -2127,6 +2086,28 @@ impl DatasetIndexExt for Dataset { } if existing_fragments.is_disjoint(&incoming_fragments) { + // Retained, so its declaration outlives this commit and has + // to match what is being written: `IndexDescriptionImpl::try_new` + // requires `fields` to be identical across the segments of one + // logical index. Nothing above catches a disagreement, because + // `keyed_fields` is the prefix left after the carried ones -- a + // segment carrying columns keys on exactly what a plain one + // keys on, and both pass the keyed-field guard. + if idx.fields != expected_fields + || idx.covering_fields != expected_covering_fields + { + return Err(Error::invalid_input(format!( + "CreateIndex: incoming segments for '{}' declare fields {:?} and covering_fields {:?}, \ + but retained segment {} declares fields {:?} and covering_fields {:?}; \ + a logical index cannot mix declarations - rebuild every segment in one commit", + index_name, + expected_fields, + expected_covering_fields, + idx.uuid, + idx.fields, + idx.covering_fields + ))); + } return Ok(None); } @@ -2230,7 +2211,11 @@ impl DatasetIndexExt for Dataset { async fn optimize_indices(&mut self, options: &OptimizeOptions) -> Result<()> { let dataset = Arc::new(self.clone()); - let indices = self.load_indices().await?; + // Grouped from the complete list so a name's segments are all accounted + // for. A segment this build cannot read is still coverage, and merging + // against a group whose coverage is only partly visible would commit a + // new segment claiming fragments an existing one already holds. + let indices = load_all_indices(self).await?; let indices_to_optimize = options .index_names @@ -2249,7 +2234,7 @@ impl DatasetIndexExt for Dataset { let mut new_indices = vec![]; let mut removed_indices = vec![]; - for deltas in name_to_indices.values() { + for (name, deltas) in name_to_indices.iter() { // Optimizing a covered index would republish its declaration on a // segment rebuilt without the carried values: `scan_vector_fragments` // projects the keyed field and `_rowid` only, and the scalar merges @@ -2286,6 +2271,22 @@ impl DatasetIndexExt for Dataset { ))); } + // Optimizing a name means replacing its segments with one that + // covers their union, which this build cannot compute when it + // cannot read one of them: the merged segment would overlap the + // segment left behind, and `Dataset::validate` calls that + // corruption. Leave the whole name to a build that can read it. + if let Some(max_supported_version) = + deltas.iter().find_map(|idx| unsupported_index_version(idx)) + { + log::warn!( + "Index {} has a segment newer than version {}, which this build cannot read; \ + skipping its optimization", + name, + max_supported_version, + ); + continue; + } // Scalar indices have no rebalance concept, so skip them entirely // when every fragment is already covered and the caller hasn't // asked for retrain or an explicit delta merge. Vector indices @@ -2621,20 +2622,38 @@ async fn gather_fragment_statistics( ))) } -pub(crate) fn retain_supported_indices(indices: &mut Vec) { - indices.retain(|idx| { - let max_supported_version = idx - .index_details - .as_ref() - .map(|details| { - IndexDetails(details.clone()) - .index_version() - // If we don't know how to read the index, it isn't supported - .unwrap_or(i32::MAX as u32) - }) - .unwrap_or_default(); - let is_valid = idx.index_version <= max_supported_version as i32; - if !is_valid { +/// `None` when this build supports the index's version, otherwise the highest +/// version it does support. +/// +/// Only a version bump of a type this build already has a plugin for is caught. +/// An index whose `type_url` resolves to no plugin at all - a wholly new index +/// type, or a built-in behind a Cargo feature this build lacks - falls back to a +/// ceiling of `i32::MAX` and is reported as supported, so it reaches the query +/// planner and fails on open instead. That predates this split and is left +/// as-is: reversing it needs the system indices exempted first, since neither +/// the fragment-reuse nor the mem-wal details resolve to a scalar plugin either. +pub(crate) fn unsupported_index_version(index: &IndexMetadata) -> Option { + let max_supported_version = index + .index_details + .as_ref() + .map(|details| { + IndexDetails(details.clone()) + .index_version() + .unwrap_or(i32::MAX as u32) + }) + .unwrap_or_default(); + (index.index_version > max_supported_version as i32).then_some(max_supported_version) +} + +/// Name the indices this build has no reader for, once per manifest read. +/// +/// Deliberately not inside the filter in [`DatasetIndexExt::load_indices`]: that +/// runs on every call, and `load_indices` sits on the query-planning path and on +/// merge_insert's per-batch path. Warning there would cost an operator one line +/// per hidden index per query for as long as the dataset carries one. +pub(crate) fn warn_about_unsupported_indices(indices: &[IndexMetadata]) { + for idx in indices { + if let Some(max_supported_version) = unsupported_index_version(idx) { log::warn!( "Index {} has version {}, which is not supported (<={}), ignoring it", idx.name, @@ -2642,8 +2661,100 @@ pub(crate) fn retain_supported_indices(indices: &mut Vec) { max_supported_version, ); } - is_valid - }) + } +} + +/// Every index the manifest names, including any this build has no reader for. +/// +/// Separate from [`DatasetIndexExt::load_indices`] because the two answer +/// different questions. A reader asks which indices it may *use*, and an index +/// it cannot decode is rightly absent. Everything that decides what the *next* +/// manifest looks like asks a different question, and there the same omission is +/// not a filter but an erasure. `build_manifest` seeds the new index list from +/// what it is handed, so an index left out disappears from the dataset for every +/// build, including the one that could have read it. Index bookkeeping - name +/// reservation, replace and removal selection, explicit drop - answers that +/// second question too: a name it cannot see is a name it will hand out twice. +pub(crate) async fn load_all_indices(dataset: &Dataset) -> Result>> { + let metadata_key = IndexMetadataKey { + version: dataset.version().version, + store_identity: &dataset.object_store.store_prefix, + }; + let mut indices = dataset + .index_cache + .get_or_insert_with_key(metadata_key, || async { + let loaded = read_manifest_indexes( + &dataset.object_store, + &dataset.manifest_location, + &dataset.manifest, + ) + .await?; + warn_about_unsupported_indices(&loaded); + Ok(loaded) + }) + .await?; + + // Infer details for legacy vector indices (once per index name, concurrently). + // This may run on indices that were opportunistically cached during Dataset::open + // before the full Dataset was available for inference. + { + let schema = dataset.schema(); + if indices + .iter() + .any(|idx| needs_vector_details_inference(idx, schema)) + { + let mut updated = indices.as_ref().clone(); + infer_missing_vector_details(dataset, &mut updated).await; + if updated != *indices { + indices = Arc::new(updated); + dataset + .index_cache + .insert_with_key(&metadata_key, indices.clone()) + .await; + } + } + } + + if let Some(frag_reuse_index_meta) = + indices.iter().find(|idx| idx.name == FRAG_REUSE_INDEX_NAME) + { + let fri_key = FragReuseIndexKey { + uuid: &frag_reuse_index_meta.uuid, + }; + let frag_reuse_index = dataset + .index_cache + .get_or_insert_with_key(fri_key, || async move { + let index_details = + load_frag_reuse_index_details(dataset, frag_reuse_index_meta).await?; + open_frag_reuse_index(frag_reuse_index_meta.uuid, index_details.as_ref()).await + }) + .await?; + let mut indices = indices.as_ref().clone(); + for idx in indices.iter_mut() { + if let Some(bitmap) = idx.fragment_bitmap.as_mut() { + frag_reuse_index.remap_fragment_bitmap(bitmap)?; + } + } + Ok(Arc::new(indices)) + } else { + Ok(indices) + } +} + +/// The segments named `name`, including any this build has no reader for. +/// +/// The bookkeeping counterpart to [`DatasetIndexExt::load_indices_by_name`]. See +/// [`load_all_indices`] for which of the two a call site wants. +pub(crate) async fn load_all_indices_by_name( + dataset: &Dataset, + name: &str, +) -> Result> { + Ok(load_all_indices(dataset) + .await? + .iter() + .filter(|idx| idx.name == name) + .cloned() + .collect()) } /// A trait for internal dataset utilities @@ -10637,6 +10748,1202 @@ mod tests { } } + fn two_column_reader() -> impl arrow_array::RecordBatchReader + Send + 'static { + lance_datagen::gen_batch() + .col("id", array::step::()) + .col("payload", array::step::()) + .into_reader_rows(RowCount::from(10), BatchCount::from(1)) + } + + /// Raise `index_name` past the version this build can read, and give it the + /// full fragment coverage a real index of that name would have. + async fn hide_index_from_this_build(dataset: &mut Dataset, index_name: &str) { + let current = dataset.load_indices_by_name(index_name).await.unwrap(); + assert_eq!(current.len(), 1); + let mut from_the_future = current.clone(); + from_the_future[0].index_version = current[0].index_version + 1; + from_the_future[0].fragment_bitmap = Some(dataset.fragment_bitmap.as_ref().clone()); + let transaction = Transaction::new( + dataset.manifest.version, + Operation::CreateIndex { + new_indices: from_the_future, + removed_indices: current, + }, + None, + ); + dataset + .apply_commit(transaction, &Default::default(), &Default::default()) + .await + .unwrap(); + } + + /// The readable companion every fixture below carries over `payload`. + const READABLE_INDEX: &str = "payload_idx"; + + /// A dataset carrying a BTree index over `id` whose version this build has + /// no reader for - what an index written by a newer Lance looks like from + /// here - beside an ordinary readable BTree index over `payload`. + /// + /// The readable companion is what makes the filter's selectivity visible: + /// with a single entry, "hid the one it cannot read" and "hid everything" + /// produce the same answer to every assertion in this module. + /// + /// Both indices are committed untrained and given their coverage by hand. + /// Nothing here ever reads them, and training one would take a non-spillable + /// 40 MB reservation out of the session's shared 150 MB pool to sort ten + /// rows - three of those in flight at once is all the pool has room for. + async fn dataset_with_an_index_from_a_newer_build(uri: &str, index_name: &str) -> Dataset { + let mut dataset = Dataset::write(two_column_reader(), uri, None) + .await + .unwrap(); + + let btree_params = ScalarIndexParams::for_builtin(BuiltinIndexType::BTree); + dataset + .create_index_builder(&["id"], IndexType::BTree, &btree_params) + .name(index_name.to_string()) + .train(false) + .await + .unwrap(); + hide_index_from_this_build(&mut dataset, index_name).await; + + dataset + .create_index_builder(&["payload"], IndexType::BTree, &btree_params) + .name(READABLE_INDEX.to_string()) + .train(false) + .await + .unwrap(); + dataset + } + + /// Indices the manifest itself carries, bypassing the version filter. + async fn raw_manifest_indices(dataset: &Dataset) -> Vec { + lance_table::io::manifest::read_manifest_indexes( + &dataset.object_store, + &dataset.manifest_location, + &dataset.manifest, + ) + .await + .unwrap() + } + + /// The manifest entry named `name`, whole. Compare these, not names: a + /// carried-forward index that kept its name but lost its coverage or gained + /// a new uuid is exactly the corruption this suite exists to catch. + async fn manifest_index(dataset: &Dataset, name: &str) -> IndexMetadata { + raw_manifest_indices(dataset) + .await + .into_iter() + .find(|idx| idx.name == name) + .unwrap_or_else(|| panic!("no index named {name} in the manifest")) + } + + /// Sorted: a commit that replaces an entry appends the replacement, so + /// comparing in manifest order would break on which operation ran rather + /// than on what it did. Order is not meaningless in general - delta merging + /// selects a suffix of it - but no test here asserts on it. + async fn manifest_index_names(dataset: &Dataset) -> Vec { + let mut names = raw_manifest_indices(dataset) + .await + .into_iter() + .map(|idx| idx.name) + .collect::>(); + names.sort(); + names + } + + /// An index this build cannot read must be hidden, not erased. + /// + /// Every commit rebuilds the index list from what it is handed, so filtering + /// the version there turns "ignore it" into "delete it", and the build that + /// could have read the index never gets the chance. + #[tokio::test] + async fn test_unsupported_index_survives_an_unrelated_commit() { + let test_dir = tempfile::tempdir().unwrap(); + let test_uri = test_dir.path().to_str().unwrap(); + dataset_with_an_index_from_a_newer_build(test_uri, "id_idx").await; + + let dataset = Dataset::open(test_uri).await.unwrap(); + assert_eq!( + dataset + .load_indices() + .await + .unwrap() + .iter() + .map(|idx| idx.name.as_str()) + .collect::>(), + [READABLE_INDEX], + "the filter must hide the index this build cannot read, and only it" + ); + assert_eq!( + manifest_index_names(&dataset).await, + ["id_idx", READABLE_INDEX] + ); + let before = manifest_index(&dataset, "id_idx").await; + + // An unrelated append. Nothing about the index is part of this operation. + let dataset = Dataset::write( + two_column_reader(), + test_uri, + Some(WriteParams { + mode: WriteMode::Append, + ..Default::default() + }), + ) + .await + .unwrap(); + + assert_eq!( + manifest_index(&dataset, "id_idx").await, + before, + "an unrelated append changed an index this build merely could not read" + ); + } + + /// Carrying an unreadable index forward is not the same as keeping it + /// forever: dropping the column it covers still removes it. + #[tokio::test] + async fn test_unsupported_index_is_dropped_with_its_column() { + let test_dir = tempfile::tempdir().unwrap(); + let test_uri = test_dir.path().to_str().unwrap(); + dataset_with_an_index_from_a_newer_build(test_uri, "id_idx").await; + + let mut dataset = Dataset::open(test_uri).await.unwrap(); + let readable_before = manifest_index(&dataset, READABLE_INDEX).await; + + dataset.drop_columns(&["id"]).await.unwrap(); + + assert_eq!( + manifest_index_names(&dataset).await, + [READABLE_INDEX], + "dropping `id` must remove the index over it, and only that one" + ); + assert_eq!( + manifest_index(&dataset, READABLE_INDEX).await, + readable_before, + "the index over the surviving column was rewritten" + ); + } + + /// Carrying it forward must also not drag it through index migration. + /// + /// `migrate_indices` recalculates a missing `fragment_bitmap` by opening the + /// index, which is precisely what this build cannot do - so an unreadable + /// index would fail every later commit instead of riding along. + #[tokio::test] + async fn test_unsupported_index_without_a_bitmap_does_not_fail_later_commits() { + let test_dir = tempfile::tempdir().unwrap(); + let test_uri = test_dir.path().to_str().unwrap(); + let mut dataset = dataset_with_an_index_from_a_newer_build(test_uri, "id_idx").await; + + // Drop the coverage too, so migration would want to rebuild it. + let hidden = manifest_index(&dataset, "id_idx").await; + let without_bitmap = IndexMetadata { + fragment_bitmap: None, + ..hidden.clone() + }; + let transaction = Transaction::new( + dataset.manifest.version, + Operation::CreateIndex { + new_indices: vec![without_bitmap.clone()], + removed_indices: vec![hidden], + }, + None, + ); + dataset + .apply_commit(transaction, &Default::default(), &Default::default()) + .await + .unwrap(); + + let mut dataset = Dataset::open(test_uri).await.unwrap(); + dataset.delete("false").await.unwrap(); + + assert_eq!( + manifest_index(&dataset, "id_idx").await, + without_bitmap, + "a commit rewrote an index it cannot open instead of carrying it through" + ); + } + + /// A name an unreadable index already owns cannot be handed out again. + /// + /// Nothing in the format stops two entries from sharing a name, and the + /// build that can read both would take them for segments of one index. + #[tokio::test] + async fn test_unsupported_index_name_is_still_taken() { + let test_dir = tempfile::tempdir().unwrap(); + let test_uri = test_dir.path().to_str().unwrap(); + dataset_with_an_index_from_a_newer_build(test_uri, "id_idx").await; + + let mut dataset = Dataset::open(test_uri).await.unwrap(); + let btree_params = ScalarIndexParams::for_builtin(BuiltinIndexType::BTree); + let err = dataset + .create_index_builder(&["id"], IndexType::BTree, &btree_params) + .name("id_idx".to_string()) + .await + .expect_err("a name is taken by an index this build cannot read"); + assert!( + err.to_string().contains("already exists"), + "expected a name collision, got: {err}" + ); + assert_eq!( + manifest_index_names(&dataset).await, + ["id_idx", READABLE_INDEX] + ); + } + + /// The generated-name loop reads the same view the collision check does, so + /// a name an unreadable index holds is skipped rather than reused. + #[tokio::test] + async fn test_an_auto_generated_name_skips_an_unsupported_index() { + let test_dir = tempfile::tempdir().unwrap(); + let test_uri = test_dir.path().to_str().unwrap(); + dataset_with_an_index_from_a_newer_build(test_uri, "id_idx").await; + + // A different index kind on the same column: the loop steps past the + // taken name instead of stopping at the collision check. + let mut dataset = Dataset::open(test_uri).await.unwrap(); + let bitmap_params = ScalarIndexParams::for_builtin(BuiltinIndexType::Bitmap); + dataset + .create_index_builder(&["id"], IndexType::Bitmap, &bitmap_params) + .train(false) + .await + .unwrap(); + + assert_eq!( + manifest_index_names(&dataset).await, + ["id_idx", "id_idx_2", READABLE_INDEX], + "the generated name reused one an unreadable index already holds" + ); + } + + /// The multi-segment FM-Index builder reserves names on its own, so it needs + /// the same complete view as the single-segment path. + /// + /// The hidden index is a BTree and the new one an FM index, which is what + /// makes this test specific to the multi-segment builder: its name loop + /// (`index/create.rs`) only steps past a taken name when the *fields* differ, + /// where the single-segment loop also steps past a different index kind. So + /// the single-segment path would quietly settle on `text_idx_2` and succeed; + /// only the multi-segment path keeps `text_idx` and hits the collision. + #[tokio::test] + async fn test_multi_segment_fmindex_respects_an_unsupported_index_name() { + let test_dir = tempfile::tempdir().unwrap(); + let test_uri = test_dir.path().to_str().unwrap(); + + let schema = Arc::new(Schema::new(vec![Field::new("text", DataType::Utf8, false)])); + let batch = RecordBatch::try_new( + schema.clone(), + vec![Arc::new(StringArray::from(vec!["alpha", "beta", "gamma"]))], + ) + .unwrap(); + let mut dataset = Dataset::write( + RecordBatchIterator::new(vec![Ok(batch)], schema.clone()), + test_uri, + None, + ) + .await + .unwrap(); + + let btree_params = ScalarIndexParams::for_builtin(BuiltinIndexType::BTree); + dataset + .create_index_builder(&["text"], IndexType::BTree, &btree_params) + .name("text_idx".to_string()) + .train(false) + .await + .unwrap(); + hide_index_from_this_build(&mut dataset, "text_idx").await; + + let mut dataset = Dataset::open(test_uri).await.unwrap(); + let multi_segment_params = ScalarIndexParams::for_builtin(BuiltinIndexType::Fm) + .with_params(&serde_json::json!({ "num_segments": 2 })); + let err = dataset + .create_index_builder(&["text"], IndexType::Fm, &multi_segment_params) + .train(false) + .await + .expect_err("a name is taken by an index this build cannot read"); + assert!( + err.to_string().contains("already exists"), + "expected a name collision, got: {err}" + ); + assert_eq!(manifest_index_names(&dataset).await, ["text_idx"]); + } + + /// Being unreadable must not make an index unremovable. + #[tokio::test] + async fn test_unsupported_index_can_be_dropped_by_name() { + let test_dir = tempfile::tempdir().unwrap(); + let test_uri = test_dir.path().to_str().unwrap(); + dataset_with_an_index_from_a_newer_build(test_uri, "id_idx").await; + + let mut dataset = Dataset::open(test_uri).await.unwrap(); + dataset.drop_index("id_idx").await.unwrap(); + + assert_eq!( + manifest_index_names(&dataset).await, + [READABLE_INDEX], + "drop_index removed the wrong set of indices" + ); + } + + /// `replace` has to select the index it replaces from the same complete view + /// the name was reserved against, or it adds a twin instead of replacing. + #[tokio::test] + async fn test_replacing_an_unsupported_index_does_not_duplicate_it() { + let test_dir = tempfile::tempdir().unwrap(); + let test_uri = test_dir.path().to_str().unwrap(); + dataset_with_an_index_from_a_newer_build(test_uri, "id_idx").await; + + let mut dataset = Dataset::open(test_uri).await.unwrap(); + let btree_params = ScalarIndexParams::for_builtin(BuiltinIndexType::BTree); + dataset + .create_index_builder(&["id"], IndexType::BTree, &btree_params) + .name("id_idx".to_string()) + .replace(true) + .train(false) + .await + .unwrap(); + + assert_eq!( + manifest_index_names(&dataset).await, + ["id_idx", READABLE_INDEX], + "replace committed a second index under a name already taken" + ); + assert!( + unsupported_index_version(&manifest_index(&dataset, "id_idx").await).is_none(), + "replace kept the unreadable index and discarded the new one" + ); + } + + /// The same selection, through the segment-commit path rather than the + /// builder: full coverage replaces the segments already under that name. + #[tokio::test] + async fn test_committing_a_segment_beside_an_unsupported_index_replaces_it() { + let test_dir = tempfile::tempdir().unwrap(); + let test_uri = test_dir.path().to_str().unwrap(); + dataset_with_an_index_from_a_newer_build(test_uri, "id_idx").await; + + let mut dataset = Dataset::open(test_uri).await.unwrap(); + let btree_params = ScalarIndexParams::for_builtin(BuiltinIndexType::BTree); + let mut segment = dataset + .create_index_builder(&["id"], IndexType::BTree, &btree_params) + .name("id_idx".to_string()) + .replace(true) + .train(false) + .execute_uncommitted() + .await + .unwrap(); + // Full coverage, so the removal decision goes through the fragment + // overlap branch rather than the empty-bitmap shortcut. Set by hand + // because training it would take 40 MB of the shared pool to sort ten + // rows - see `dataset_with_an_index_from_a_newer_build`. + segment.fragment_bitmap = Some(dataset.fragment_bitmap.as_ref().clone()); + dataset + .commit_existing_index_segments("id_idx", "id", vec![segment]) + .await + .unwrap(); + + assert_eq!( + manifest_index_names(&dataset).await, + ["id_idx", READABLE_INDEX], + "the incoming segment was committed beside the unreadable one" + ); + assert!( + unsupported_index_version(&manifest_index(&dataset, "id_idx").await).is_none(), + "the incoming segment did not replace the unreadable one" + ); + } + + /// The retention path itself: a segment on fragments the existing one does + /// not cover is kept beside it, and agreeing declarations are what makes + /// that legal. This is the case the rejection below must not swallow - + /// partial-coverage builds depend on it. + #[tokio::test] + async fn test_committing_a_segment_on_disjoint_fragments_keeps_the_existing_one() { + let test_dir = tempfile::tempdir().unwrap(); + let test_uri = test_dir.path().to_str().unwrap(); + + let mut dataset = Dataset::write(two_column_reader(), test_uri, None) + .await + .unwrap(); + let btree_params = ScalarIndexParams::for_builtin(BuiltinIndexType::BTree); + dataset + .create_index_builder(&["id"], IndexType::BTree, &btree_params) + .name("id_idx".to_string()) + .train(false) + .await + .unwrap(); + // Given by hand: an untrained segment commits with an empty bitmap, which + // takes the zero-coverage removal branch rather than the disjoint one. + // See `dataset_with_an_index_from_a_newer_build` for why nothing here trains. + let untrained = manifest_index(&dataset, "id_idx").await; + let mut existing = untrained.clone(); + existing.fragment_bitmap = Some(dataset.fragment_bitmap.as_ref().clone()); + dataset + .apply_commit( + Transaction::new( + dataset.manifest.version, + Operation::CreateIndex { + new_indices: vec![existing.clone()], + removed_indices: vec![untrained], + }, + None, + ), + &Default::default(), + &Default::default(), + ) + .await + .unwrap(); + let covered = existing.fragment_bitmap.clone().unwrap(); + assert!(!covered.is_empty()); + + let mut dataset = Dataset::write( + two_column_reader(), + test_uri, + Some(WriteParams { + mode: WriteMode::Append, + ..Default::default() + }), + ) + .await + .unwrap(); + let appended = dataset.fragment_bitmap.as_ref() - &covered; + assert!(!appended.is_empty()); + + let mut segment = dataset + .create_index_builder(&["id"], IndexType::BTree, &btree_params) + .name("id_idx".to_string()) + .replace(true) + .fragments(appended.iter().collect()) + .train(false) + .execute_uncommitted() + .await + .unwrap(); + segment.fragment_bitmap = Some(appended); + assert_eq!(segment.fields, existing.fields); + assert_eq!(segment.covering_fields, existing.covering_fields); + + dataset + .commit_existing_index_segments("id_idx", "id", vec![segment]) + .await + .unwrap(); + + let uuids = raw_manifest_indices(&dataset) + .await + .into_iter() + .filter(|idx| idx.name == "id_idx") + .map(|idx| idx.uuid) + .collect::>(); + assert_eq!(uuids.len(), 2, "the disjoint existing segment was dropped"); + assert!(uuids.contains(&existing.uuid)); + } + + /// One logical index needs one declaration, and the complete view is what + /// makes the disagreement reachable: a segment this build cannot read may + /// carry columns, and a plain segment committed beside it on disjoint + /// fragments is retained rather than replaced. Both pass the per-segment + /// rules - `keyed_fields` is the prefix left after the carried ones, so + /// `[id, payload]` carrying `[payload]` keys on `id` exactly as `[id]` does. + /// Committing the pair would leave `describe_indices` erroring on metadata + /// this call just wrote, the same failure + /// `test_build_index_metadata_from_segments_rejects_mixed_covering_declarations` + /// pins for the incoming side. + #[tokio::test] + async fn test_committing_a_plain_segment_beside_a_covered_unsupported_one_is_rejected() { + let test_dir = tempfile::tempdir().unwrap(); + let test_uri = test_dir.path().to_str().unwrap(); + let mut dataset = dataset_with_an_index_from_a_newer_build(test_uri, "id_idx").await; + + // Give the hidden segment a carried column. A newer build is exactly + // where a covering segment would come from. + let hidden = manifest_index(&dataset, "id_idx").await; + let payload_id = dataset.schema().field("payload").unwrap().id; + let mut hidden_covered = hidden.clone(); + hidden_covered.fields.push(payload_id); + hidden_covered.covering_fields = vec![payload_id]; + assert_eq!(hidden_covered.keyed_field(), hidden.keyed_field()); + let covered_fragments = hidden_covered.fragment_bitmap.clone().unwrap(); + dataset + .apply_commit( + Transaction::new( + dataset.manifest.version, + Operation::CreateIndex { + new_indices: vec![hidden_covered], + removed_indices: vec![hidden], + }, + None, + ), + &Default::default(), + &Default::default(), + ) + .await + .unwrap(); + + // Fragments the hidden segment does not cover, so the incoming segment + // takes the disjoint branch and the hidden one is retained. + let mut dataset = Dataset::write( + two_column_reader(), + test_uri, + Some(WriteParams { + mode: WriteMode::Append, + ..Default::default() + }), + ) + .await + .unwrap(); + let appended = dataset.fragment_bitmap.as_ref() - &covered_fragments; + assert!(!appended.is_empty()); + + let btree_params = ScalarIndexParams::for_builtin(BuiltinIndexType::BTree); + let mut plain = dataset + .create_index_builder(&["id"], IndexType::BTree, &btree_params) + .name("id_idx".to_string()) + .replace(true) + .fragments(appended.iter().collect()) + .train(false) + .execute_uncommitted() + .await + .unwrap(); + plain.fragment_bitmap = Some(appended); + assert!(plain.covering_fields.is_empty()); + + let err = dataset + .commit_existing_index_segments("id_idx", "id", vec![plain]) + .await + .expect_err("a logical index cannot mix covered and plain segment declarations"); + assert!( + err.to_string().contains("covering_fields"), + "unexpected error: {err}" + ); + } + + /// A cast reassigns the field id, so no index on that column can be carried + /// forward. The guard that makes that explicit has to see the hidden ones + /// too, or they get exactly the silent drop it exists to abolish. + #[tokio::test] + async fn test_casting_a_column_with_an_unsupported_index_is_rejected() { + use crate::dataset::ColumnAlteration; + + let test_dir = tempfile::tempdir().unwrap(); + let test_uri = test_dir.path().to_str().unwrap(); + dataset_with_an_index_from_a_newer_build(test_uri, "id_idx").await; + + let mut dataset = Dataset::open(test_uri).await.unwrap(); + let err = dataset + .alter_columns(&[ColumnAlteration::new("id".into()).cast_to(DataType::Int64)]) + .await + .expect_err("a cast must not silently erase an index it cannot read"); + assert!( + err.to_string().contains("id_idx"), + "the error should name the index, got: {err}" + ); + assert_eq!( + manifest_index_names(&dataset).await, + ["id_idx", READABLE_INDEX], + "the rejected cast still changed the manifest" + ); + } + + /// A cache entry written under the key's previous meaning must cold-miss. + /// + /// v1 of `lance.index.metadata-key` held only the indices the writing build + /// could read. The key fields are identical, so on a persistent backend + /// shared with such a build nothing but the schema version stops this one + /// from reading that filtered list as the complete one. + #[tokio::test] + async fn test_a_pre_rotation_cache_entry_is_not_consulted() { + use lance_core::cache::{CacheCodec, CacheKey, CacheKeySchema, KeyBuilder}; + use std::borrow::Cow; + + struct PreRotationIndexMetadataKey<'a> { + version: u64, + store_identity: &'a str, + } + + impl CacheKey for PreRotationIndexMetadataKey<'_> { + type ValueType = Vec; + + fn key(&self) -> Cow<'_, str> { + Cow::Owned(format!( + "{}:{}/{}", + self.store_identity.len(), + self.store_identity, + self.version + )) + } + + fn type_name() -> &'static str { + "Vec" + } + + fn schema() -> CacheKeySchema { + CacheKeySchema::new("lance.index.metadata-key", 1) + } + + fn write_key(&self, builder: &mut KeyBuilder) { + builder.write_str(self.store_identity); + builder.write_u64(self.version); + } + + fn codec() -> Option { + Some(lance_table::format::index_metadata_codec()) + } + } + + let test_dir = tempfile::tempdir().unwrap(); + let test_uri = test_dir.path().to_str().unwrap(); + dataset_with_an_index_from_a_newer_build(test_uri, "id_idx").await; + + let mut dataset = Dataset::open(test_uri).await.unwrap(); + let complete = raw_manifest_indices(&dataset).await; + let as_a_released_build_would_cache_it = complete + .iter() + .filter(|idx| unsupported_index_version(idx).is_none()) + .cloned() + .collect::>(); + assert!( + as_a_released_build_would_cache_it.len() < complete.len(), + "the fixture must give the two key versions different values to cache" + ); + dataset + .index_cache + .insert_with_key( + &PreRotationIndexMetadataKey { + version: dataset.version().version, + store_identity: &dataset.object_store.store_prefix, + }, + Arc::new(as_a_released_build_would_cache_it), + ) + .await; + + dataset.delete("false").await.unwrap(); + + assert_eq!( + manifest_index_names(&dataset).await, + ["id_idx", READABLE_INDEX], + "a commit read a cache entry written under the key's previous meaning" + ); + } + + /// `validate` checks the manifest, so it has to see all of it. An index this + /// build cannot read is no less corrupt for being unreadable, and now that + /// such an index is carried forward the corrupt state is durable rather than + /// gone at the next commit. + #[tokio::test] + async fn test_validate_sees_an_unsupported_index() { + let test_dir = tempfile::tempdir().unwrap(); + let test_uri = test_dir.path().to_str().unwrap(); + let mut dataset = dataset_with_an_index_from_a_newer_build(test_uri, "id_idx").await; + dataset.validate().await.unwrap(); + + // A second segment under the same name covering the same fragments. Only + // `detect_overlapping_fragments` over the complete list can see it. + let hidden = manifest_index(&dataset, "id_idx").await; + let overlapping = IndexMetadata { + uuid: Uuid::new_v4(), + ..hidden + }; + let transaction = Transaction::new( + dataset.manifest.version, + Operation::CreateIndex { + new_indices: vec![overlapping], + removed_indices: vec![], + }, + None, + ); + dataset + .apply_commit(transaction, &Default::default(), &Default::default()) + .await + .unwrap(); + + let err = dataset + .validate() + .await + .expect_err("two segments of one name covering the same fragments is corrupt"); + assert!( + err.to_string().contains("id_idx"), + "the error should name the index, got: {err}" + ); + } + + /// A detached commit builds its manifest through its own code path, and it + /// carries the index list forward exactly as an attached one does. + #[tokio::test] + async fn test_a_detached_commit_does_not_erase_an_unsupported_index() { + use crate::dataset::InsertBuilder; + use crate::dataset::write::CommitBuilder; + + let test_dir = tempfile::tempdir().unwrap(); + let test_uri = test_dir.path().to_str().unwrap(); + dataset_with_an_index_from_a_newer_build(test_uri, "id_idx").await; + + let batches = two_column_reader() + .collect::, _>>() + .unwrap(); + let dataset = Arc::new(Dataset::open(test_uri).await.unwrap()); + let before = manifest_index(&dataset, "id_idx").await; + let transaction = InsertBuilder::new(dataset.clone()) + .with_params(&WriteParams { + mode: WriteMode::Append, + ..Default::default() + }) + .execute_uncommitted(batches) + .await + .unwrap(); + let detached = CommitBuilder::new(dataset.clone()) + .with_detached(true) + .execute(transaction) + .await + .unwrap(); + + assert_eq!( + manifest_index_names(&detached).await, + ["id_idx", READABLE_INDEX], + "a detached commit erased an index this build merely could not read" + ); + assert_eq!( + manifest_index(&detached, "id_idx").await, + before, + "a detached commit rewrote an index this build merely could not read" + ); + assert!(lance_table::format::is_detached_version( + detached.manifest.version + )); + assert_eq!(detached.count_rows(None).await.unwrap(), 20); + } + + /// Compaction bins fragments so that no rewrite group splits an index's + /// coverage, and `recalculate_fragment_bitmap` rejects the plan if one does. + /// Both sides therefore have to count the same indices: planning from the + /// filtered view while the commit carries the complete one makes compaction + /// fail outright on a dataset holding an index from a newer build. + #[tokio::test] + async fn test_compaction_survives_an_unsupported_index() { + let test_dir = tempfile::tempdir().unwrap(); + let test_uri = test_dir.path().to_str().unwrap(); + + let write_params = WriteParams { + enable_stable_row_ids: true, + max_rows_per_file: 5, + ..Default::default() + }; + let mut dataset = Dataset::write(two_column_reader(), test_uri, Some(write_params.clone())) + .await + .unwrap(); + assert_eq!(dataset.get_fragments().len(), 2); + + let btree_params = ScalarIndexParams::for_builtin(BuiltinIndexType::BTree); + dataset + .create_index_builder(&["id"], IndexType::BTree, &btree_params) + .name("id_idx".to_string()) + .train(false) + .await + .unwrap(); + hide_index_from_this_build(&mut dataset, "id_idx").await; + + // A fragment the index does not cover, so a bin holding it together with + // the covered ones would split the index's coverage. + let mut dataset = Dataset::write( + two_column_reader(), + test_uri, + Some(WriteParams { + mode: WriteMode::Append, + ..write_params + }), + ) + .await + .unwrap(); + let before = manifest_index(&dataset, "id_idx").await; + + let metrics = compact_files(&mut dataset, CompactionOptions::default(), None) + .await + .unwrap(); + + // Without a real rewrite the assertions below hold vacuously: an empty + // plan commits nothing and re-reads the manifest it started from. + assert_eq!(metrics.fragments_removed, 4); + assert_eq!(dataset.get_fragments().len(), 2); + + let after = manifest_index(&dataset, "id_idx").await; + assert_eq!(after.uuid, before.uuid); + assert_eq!(after.index_version, before.index_version); + let live = dataset + .get_fragments() + .iter() + .map(|f| f.id() as u32) + .collect::(); + assert!( + !after.fragment_bitmap.unwrap().is_disjoint(&live), + "the surviving index covers only fragments the rewrite deleted" + ); + } + + /// Without stable row ids a rewrite moves every row address, so each index + /// over the rewritten fragments has to be remapped - and remapping one means + /// opening it. A build with no reader for an index cannot remap it, so + /// compacting the fragments it covers would leave it addressing rows that no + /// longer exist. Those fragments are held back from the plan instead; the + /// rest of the table still compacts. + /// + /// The stable-row-id case is the test above: there the fragment-reuse index + /// repairs the coverage afterwards, so nothing has to be held back. + #[tokio::test] + async fn test_compaction_defers_fragments_an_unsupported_index_covers() { + let test_dir = tempfile::tempdir().unwrap(); + let test_uri = test_dir.path().to_str().unwrap(); + + let write_params = WriteParams { + enable_stable_row_ids: false, + max_rows_per_file: 5, + ..Default::default() + }; + let mut dataset = Dataset::write(two_column_reader(), test_uri, Some(write_params.clone())) + .await + .unwrap(); + assert_eq!(dataset.get_fragments().len(), 2); + + let btree_params = ScalarIndexParams::for_builtin(BuiltinIndexType::BTree); + dataset + .create_index_builder(&["id"], IndexType::BTree, &btree_params) + .name("id_idx".to_string()) + .train(false) + .await + .unwrap(); + hide_index_from_this_build(&mut dataset, "id_idx").await; + let covered = manifest_index(&dataset, "id_idx") + .await + .fragment_bitmap + .unwrap(); + + // Two more fragments the hidden index does not cover: they are the ones + // compaction is still free to rewrite. + let mut dataset = Dataset::write( + two_column_reader(), + test_uri, + Some(WriteParams { + mode: WriteMode::Append, + ..write_params + }), + ) + .await + .unwrap(); + assert_eq!(dataset.get_fragments().len(), 4); + let before = manifest_index(&dataset, "id_idx").await; + + let metrics = compact_files(&mut dataset, CompactionOptions::default(), None) + .await + .unwrap(); + + // The two uncovered fragments coalesce; the two the hidden index covers + // are left alone. Both halves matter: no rewrite at all would satisfy the + // coverage assertion below for the wrong reason. + assert_eq!(metrics.fragments_removed, 2); + assert_eq!(metrics.fragments_added, 1); + let live = dataset + .get_fragments() + .iter() + .map(|f| f.id() as u32) + .collect::(); + assert!( + covered.is_subset(&live), + "a fragment the unreadable index covers was rewritten: covered {covered:?}, live {live:?}" + ); + + let after = manifest_index(&dataset, "id_idx").await; + assert_eq!(after.uuid, before.uuid); + assert_eq!(after.fragment_bitmap, before.fragment_bitmap); + + // Nothing compactable is left outside the held-back set, so the plan is + // empty. That has to be an ordinary no-op: on a table the index covers + // whole - the usual shape - every compaction takes this path. + let metrics = compact_files(&mut dataset, CompactionOptions::default(), None) + .await + .unwrap(); + assert_eq!(metrics.fragments_removed, 0); + assert_eq!(metrics.fragments_added, 0); + assert_eq!(manifest_index(&dataset, "id_idx").await.uuid, before.uuid); + } + + /// Holding back the fragments an unreadable index covers is an optimization + /// in the planner, not the rule: `compact_files_with_planner` takes any + /// planner, `CompactionPlan` is public and serializable, and a distributed + /// driver hands `commit_compaction` results planned on another machine. This + /// takes that last route, so the refusal is pinned to the commit boundary. + /// + /// The other half - that the boundary does not refuse a rewrite the index + /// does not cover - is the test above, which compacts the uncovered + /// fragments of this same shape through `compact_files`. + #[tokio::test] + async fn test_committing_a_compaction_an_unsupported_index_covers_is_rejected() { + use crate::dataset::index::DatasetIndexRemapperOptions; + use crate::dataset::optimize::{CompactionPlan, TaskData, commit_compaction}; + + let test_dir = tempfile::tempdir().unwrap(); + let test_uri = test_dir.path().to_str().unwrap(); + + let write_params = WriteParams { + enable_stable_row_ids: false, + max_rows_per_file: 5, + ..Default::default() + }; + let mut dataset = Dataset::write(two_column_reader(), test_uri, Some(write_params.clone())) + .await + .unwrap(); + + let btree_params = ScalarIndexParams::for_builtin(BuiltinIndexType::BTree); + dataset + .create_index_builder(&["id"], IndexType::BTree, &btree_params) + .name("id_idx".to_string()) + .train(false) + .await + .unwrap(); + hide_index_from_this_build(&mut dataset, "id_idx").await; + let covered = manifest_index(&dataset, "id_idx") + .await + .fragment_bitmap + .unwrap(); + + // Two more fragments the hidden index does not cover, so the plan below + // is a genuine selection rather than "every fragment there is". + let mut dataset = Dataset::write( + two_column_reader(), + test_uri, + Some(WriteParams { + mode: WriteMode::Append, + ..write_params + }), + ) + .await + .unwrap(); + assert_eq!(dataset.get_fragments().len(), 4); + let before = manifest_index(&dataset, "id_idx").await; + + let plan = CompactionPlan { + tasks: vec![TaskData { + fragments: dataset + .fragments() + .iter() + .filter(|fragment| covered.contains(fragment.id as u32)) + .cloned() + .collect(), + }], + read_version: dataset.version().version, + options: CompactionOptions::default(), + }; + assert_eq!(plan.tasks[0].fragments.len(), 2); + + let mut rewrites = Vec::new(); + for task in plan.compaction_tasks() { + rewrites.push(task.execute(&dataset).await.unwrap()); + } + + let err = commit_compaction( + &mut dataset, + rewrites, + Arc::new(DatasetIndexRemapperOptions::default()), + &plan.options, + ) + .await + .unwrap_err(); + assert!( + err.to_string().contains("id_idx"), + "the refusal has to name the index that blocks the rewrite: {err}" + ); + + // Nothing was committed: the fragments the plan named are still there, + // and the index still covers them. + assert_eq!(dataset.get_fragments().len(), 4); + let after = manifest_index(&dataset, "id_idx").await; + assert_eq!(after.uuid, before.uuid); + assert_eq!(after.fragment_bitmap, before.fragment_bitmap); + } + + /// An index old enough to predate the fragment bitmap has its coverage + /// reconstructed from the version it was written against, so it arrives in + /// that version's fragment-id space. `load_all_indices` moves a *stored* + /// bitmap into the current space through the fragment-reuse index and leaves + /// a `None` one alone, so the reconstruction has to make that move itself. + /// + /// Without it, a deferred compaction is enough to defeat both guards: the + /// coverage still names the fragments the rows moved out of, which no later + /// rewrite can intersect, so the planner stops holding anything back and the + /// commit boundary waves the rewrite through. + #[tokio::test] + async fn test_reconstructed_coverage_follows_a_deferred_compaction() { + use crate::dataset::index::DatasetIndexRemapperOptions; + use crate::dataset::optimize::{CompactionPlan, TaskData, commit_compaction}; + + let test_dir = tempfile::tempdir().unwrap(); + let test_uri = test_dir.path().to_str().unwrap(); + + let write_params = WriteParams { + enable_stable_row_ids: false, + max_rows_per_file: 5, + ..Default::default() + }; + let mut dataset = Dataset::write(two_column_reader(), test_uri, Some(write_params.clone())) + .await + .unwrap(); + assert_eq!(dataset.get_fragments().len(), 2); + + let btree_params = ScalarIndexParams::for_builtin(BuiltinIndexType::BTree); + dataset + .create_index_builder(&["id"], IndexType::BTree, &btree_params) + .name("id_idx".to_string()) + .train(false) + .await + .unwrap(); + hide_index_from_this_build(&mut dataset, "id_idx").await; + + // Drop the bitmap, which is what an index written before it existed + // looks like: coverage has to be reconstructed from `dataset_version`. + let hidden = manifest_index(&dataset, "id_idx").await; + let legacy = IndexMetadata { + fragment_bitmap: None, + ..hidden.clone() + }; + let transaction = Transaction::new( + dataset.manifest.version, + Operation::CreateIndex { + new_indices: vec![legacy], + removed_indices: vec![hidden], + }, + None, + ); + dataset + .apply_commit(transaction, &Default::default(), &Default::default()) + .await + .unwrap(); + + // Deferred remap is the one compaction a build that cannot read the + // index may run: it hands the repair on through a fragment-reuse index. + // Fragments 0 and 1 become fragment 2, and the index still covers those + // rows - now under a different id. + compact_files( + &mut dataset, + CompactionOptions { + defer_index_remap: true, + ..Default::default() + }, + None, + ) + .await + .unwrap(); + let after_defer = dataset + .get_fragments() + .iter() + .map(|fragment| fragment.id() as u32) + .collect::>(); + assert_eq!(after_defer, vec![2]); + + // The commit-boundary half, taken first: the planner half below rewrites + // fragment 2 when the remap is missing, which would leave this nothing + // to ask about and hide whether the guard is sensitive on its own. + let plan = CompactionPlan { + tasks: vec![TaskData { + fragments: dataset.fragments().as_ref().clone(), + }], + read_version: dataset.version().version, + options: CompactionOptions::default(), + }; + let mut rewrites = Vec::new(); + for task in plan.compaction_tasks() { + rewrites.push(task.execute(&dataset).await.unwrap()); + } + let err = commit_compaction( + &mut dataset, + rewrites, + Arc::new(DatasetIndexRemapperOptions::default()), + &plan.options, + ) + .await + .unwrap_err(); + let message = err.to_string(); + assert!( + message.contains("id_idx") && message.contains("[2]"), + "the refusal has to name the index and the fragment in current ids: {message}" + ); + + // Two fragments the index never covered, so the planner half has + // something to compact and cannot pass by finding nothing to do. + let mut dataset = Dataset::write( + two_column_reader(), + test_uri, + Some(WriteParams { + mode: WriteMode::Append, + ..write_params + }), + ) + .await + .unwrap(); + assert_eq!(dataset.get_fragments().len(), 3); + + // The planner half: the appended pair coalesces, fragment 2 is held back. + let metrics = compact_files(&mut dataset, CompactionOptions::default(), None) + .await + .unwrap(); + assert_eq!(metrics.fragments_removed, 2); + assert_eq!(metrics.fragments_added, 1); + assert!( + dataset + .get_fragments() + .iter() + .any(|fragment| fragment.id() == 2), + "the fragment the reconstructed coverage maps to was rewritten" + ); + } + + /// Optimizing a name whose segments this build cannot all read would commit + /// a merged segment overlapping the one it left behind - a state + /// `Dataset::validate` reports as corruption and no later commit can heal. + #[tokio::test] + async fn test_optimize_skips_an_index_with_an_unsupported_segment() { + let test_dir = tempfile::tempdir().unwrap(); + let test_uri = test_dir.path().to_str().unwrap(); + let mut dataset = dataset_with_an_index_from_a_newer_build(test_uri, "id_idx").await; + + // A readable segment beside the hidden one, under the same name, so the + // group is exactly the mixed case: `id_idx` is now partly readable. + let hidden = manifest_index(&dataset, "id_idx").await; + let readable_sibling = IndexMetadata { + uuid: Uuid::new_v4(), + index_version: hidden.index_version - 1, + fragment_bitmap: Some(RoaringBitmap::new()), + ..hidden + }; + let transaction = Transaction::new( + dataset.manifest.version, + Operation::CreateIndex { + new_indices: vec![readable_sibling], + removed_indices: vec![], + }, + None, + ); + dataset + .apply_commit(transaction, &Default::default(), &Default::default()) + .await + .unwrap(); + let segments_of = |indices: Vec| { + indices + .into_iter() + .filter(|idx| idx.name == "id_idx") + .collect::>() + }; + let before = segments_of(raw_manifest_indices(&dataset).await); + assert_eq!(before.len(), 2, "the fixture must build the mixed group"); + + dataset + .optimize_indices(&OptimizeOptions::default()) + .await + .unwrap(); + + assert_eq!( + segments_of(raw_manifest_indices(&dataset).await), + before, + "optimize touched a name carrying a segment this build cannot read" + ); + dataset.validate().await.unwrap(); + } + #[tokio::test] async fn test_optimize_rebuilds_dormant_vector_index_instead_of_merging_stale_rows() { use crate::dataset::UpdateBuilder; diff --git a/rust/lance/src/index/api.rs b/rust/lance/src/index/api.rs index 43dd5b535d0..a8eaf96d603 100644 --- a/rust/lance/src/index/api.rs +++ b/rust/lance/src/index/api.rs @@ -278,7 +278,12 @@ pub trait DatasetIndexExt { )) } - /// Read all indices of this Dataset version. + /// Read the indices of this Dataset version that this build can use. + /// + /// An index whose format version is newer than this build supports is + /// omitted: it is still in the manifest and still belongs to the dataset, + /// but nothing here can decode it. Code deciding what the *next* manifest + /// should say must not use this list - it would drop what it omits. /// /// The indices are lazy loaded and cached in memory within the `Dataset` instance. /// The cache is invalidated when the dataset version (Manifest) is changed. diff --git a/rust/lance/src/index/create.rs b/rust/lance/src/index/create.rs index 8168cd1eb2d..5cc5c9f9163 100644 --- a/rust/lance/src/index/create.rs +++ b/rust/lance/src/index/create.rs @@ -9,7 +9,7 @@ use crate::{ }, index::{ DatasetIndexExt, DatasetIndexInternalExt, IntoIndexSegment, - build_index_metadata_from_segments, + build_index_metadata_from_segments, load_all_indices, scalar::{build_bitmap_index_segment, build_scalar_index}, vector::{ LANCE_VECTOR_INDEX, StageParams, VectorIndexParams, build_distributed_vector_index, @@ -261,8 +261,10 @@ impl<'a> CreateIndexBuilder<'a> { ) .await?; - // Load indices from the disk. - let indices = self.dataset.load_indices().await?; + // Load indices from the disk. Names are reserved against every index the + // manifest carries: one this build cannot read still owns its name, and + // handing that name out again commits two indices under it. + let indices = load_all_indices(self.dataset).await?; let fri = self .dataset .open_frag_reuse_index(&NoOpMetricsCollector) @@ -635,8 +637,7 @@ impl<'a> CreateIndexBuilder<'a> { let new_idx = self.execute_uncommitted().await?; let index_uuid = new_idx.uuid; let removed_indices = if self.replace { - self.dataset - .load_indices() + load_all_indices(self.dataset) .await? .iter() .filter(|idx| idx.name == new_idx.name) @@ -709,7 +710,7 @@ impl<'a> CreateIndexBuilder<'a> { false }; - let indices = self.dataset.load_indices().await?; + let indices = load_all_indices(self.dataset).await?; let index_name = if let Some(name) = self.name.take() { name } else { diff --git a/rust/lance/src/io/commit.rs b/rust/lance/src/io/commit.rs index de69d6e2a11..397087c709f 100644 --- a/rust/lance/src/io/commit.rs +++ b/rust/lance/src/io/commit.rs @@ -54,9 +54,9 @@ use crate::dataset::{ ManifestWriteConfig, NewTransactionResult, TRANSACTIONS_DIR, load_new_transactions, write_manifest_file, }; -use crate::index::DatasetIndexExt; use crate::index::DatasetIndexInternalExt; use crate::index::vector::details::infer_missing_vector_details; +use crate::index::{load_all_indices, unsupported_index_version}; use crate::io::deletion::read_dataset_deletion_file; use crate::session::Session; use crate::session::caches::DSMetadataCache; @@ -946,6 +946,15 @@ async fn migrate_indices(dataset: &Dataset, indices: &mut [IndexMetadata]) -> Re } }; for index in indices.iter_mut() { + // Migration is skipped for an index this build has no reader for: every + // branch below would have to open it to recalculate anything, which is + // exactly what this build cannot do, and failing here would fail an + // unrelated commit. Skipped, not untouched - `load_all_indices` still + // remaps its `fragment_bitmap` through the fragment-reuse index, which + // is what keeps its coverage pointing at the fragments its rows live in. + if unsupported_index_version(index).is_some() { + continue; + } if needs_recalculating.contains(&index.name) || must_recalculate_fragment_bitmap(index, dataset.manifest.writer_version.as_ref()) && !is_system_index(index) @@ -1101,7 +1110,7 @@ pub(crate) async fn do_commit_detached_transaction( } _ => transaction.build_manifest( Some(dataset.manifest.as_ref()), - dataset.load_indices().await?.as_ref().clone(), + load_all_indices(dataset).await?.as_ref().clone(), &transaction_file, &write_config.to_build_config(), )?, @@ -1363,10 +1372,10 @@ pub(crate) async fn commit_transaction( // covering every fragment live here holds every row compaction had copied // in by then. // - // The Arc is kept rather than cloned out: `load_indices` returns shared + // The Arc is kept rather than cloned out: `load_all_indices` returns shared // cached data, so the common case is a cache hit rather than a read. let read_version_dataset = dataset.clone(); - let read_version_indices = read_version_dataset.load_indices().await?; + let read_version_indices = load_all_indices(&read_version_dataset).await?; let read_version_state = Some(crate::dataset::transaction::ReadVersionState { manifest: read_version_dataset.manifest.as_ref(), indices: read_version_indices.as_slice(), @@ -1452,7 +1461,7 @@ pub(crate) async fn commit_transaction( } _ => transaction.build_manifest_with_read_version( Some(dataset.manifest.as_ref()), - dataset.load_indices().await?.as_ref().clone(), + load_all_indices(&dataset).await?.as_ref().clone(), transaction_file, &write_config.to_build_config(), read_version_state, @@ -1680,6 +1689,7 @@ mod tests { use crate::Dataset; use crate::dataset::{WriteMode, WriteParams}; + use crate::index::DatasetIndexExt; use crate::index::vector::VectorIndexParams; use crate::utils::test::{DatagenExt, FragmentCount, FragmentRowCount}; diff --git a/rust/lance/src/session/index_caches.rs b/rust/lance/src/session/index_caches.rs index f4a17918d2e..fbd8f9315af 100644 --- a/rust/lance/src/session/index_caches.rs +++ b/rust/lance/src/session/index_caches.rs @@ -139,7 +139,11 @@ impl CacheKey for IndexMetadataKey<'_> { } fn schema() -> CacheKeySchema { - CacheKeySchema::new("lance.index.metadata-key", 1) + // v2 holds every index the manifest names; v1 held only the ones the + // writing build could read. The fields are identical, so on a persistent + // backend shared with another release nothing but this version stops each + // build from reading the other's entry as its own meaning. + CacheKeySchema::new("lance.index.metadata-key", 2) } fn write_key(&self, builder: &mut KeyBuilder) {