diff --git a/rust/lance-index/src/registry.rs b/rust/lance-index/src/registry.rs index 753f32afafd..a0f3cd96df2 100644 --- a/rust/lance-index/src/registry.rs +++ b/rust/lance-index/src/registry.rs @@ -1,6 +1,9 @@ // SPDX-License-Identifier: Apache-2.0 // SPDX-FileCopyrightText: Copyright The Lance Authors -use std::{collections::HashMap, sync::Arc}; +use std::{ + collections::{HashMap, HashSet}, + sync::Arc, +}; use lance_core::{Error, Result}; @@ -16,6 +19,10 @@ use crate::{ }, }; +/// Scalar detail package emitted by Lance 0.36 before the messages moved back +/// to `lance.table` for forward compatibility. +const V036_SCALAR_DETAILS_PACKAGE: &str = "lance.index.pb"; + /// Derive the scalar index plugin name from a details type URL. /// /// Takes the last `.`-separated segment, lowercases it, and strips any trailing @@ -49,6 +56,7 @@ pub fn display_type_from_url(type_url: &str) -> &str { /// A registry of index plugins pub struct IndexPluginRegistry { plugins: HashMap>, + details_type_names: HashSet, } impl IndexPluginRegistry { @@ -75,14 +83,22 @@ impl IndexPluginRegistry { &mut self, ) { let plugin_name = self.get_plugin_name_from_details_name(DetailsType::NAME); + self.details_type_names + .insert(DetailsType::full_name().to_ascii_lowercase()); self.plugins .insert(plugin_name, Box::new(PluginType::default())); } + fn add_details_type_alias(&mut self, package: &str) { + self.details_type_names + .insert(format!("{}.{}", package, DetailsType::NAME).to_ascii_lowercase()); + } + /// Create a registry with the default plugins pub fn with_default_plugins() -> Arc { let mut registry = Self { plugins: HashMap::new(), + details_type_names: HashSet::new(), }; registry.add_plugin::(); registry.add_plugin::(); @@ -96,6 +112,17 @@ impl IndexPluginRegistry { #[cfg(feature = "geo")] registry.add_plugin::(); + // Lance 0.36 released these scalar detail messages in the index package. + // Register only those historical identities, not arbitrary packages + // carrying the same terminal message names. + registry.add_details_type_alias::(V036_SCALAR_DETAILS_PACKAGE); + registry.add_details_type_alias::(V036_SCALAR_DETAILS_PACKAGE); + registry + .add_details_type_alias::(V036_SCALAR_DETAILS_PACKAGE); + registry.add_details_type_alias::(V036_SCALAR_DETAILS_PACKAGE); + registry.add_details_type_alias::(V036_SCALAR_DETAILS_PACKAGE); + registry.add_details_type_alias::(V036_SCALAR_DETAILS_PACKAGE); + let registry = Arc::new(registry); for plugin in registry.plugins.values() { plugin.attach_registry(registry.clone()); @@ -104,6 +131,23 @@ impl IndexPluginRegistry { registry } + /// Returns whether the complete protobuf type name in `details` belongs to + /// a registered scalar index reader. + /// + /// Type URL authorities may vary, so matching uses the fully qualified + /// message name after the final slash. The table format requires index type + /// URL comparisons to be case-insensitive. + pub fn supports_details(&self, details: &prost_types::Any) -> bool { + let Some((_, details_type_name)) = details.type_url.rsplit_once('/') else { + return false; + }; + if details_type_name.is_empty() || details_type_name.starts_with('.') { + return false; + } + self.details_type_names + .contains(&details_type_name.to_ascii_lowercase()) + } + /// Get an index plugin suitable for training an index with the given parameters pub fn get_plugin_by_name(&self, name: &str) -> Result<&dyn ScalarIndexPlugin> { let plugin_name = Self::normalize_plugin_name(name); @@ -173,4 +217,36 @@ mod tests { assert_eq!(plugin.name(), expected_name); } } + + #[test] + fn test_supports_details_matches_complete_type_name_case_insensitively() { + let registry = IndexPluginRegistry::with_default_plugins(); + + for type_url in [ + "/lance.table.BTreeIndexDetails", + "type.googleapis.com/LANCE.TABLE.BTREEINDEXDETAILS", + "/lance.index.pb.BTreeIndexDetails", + "/lance.index.pb.BitmapIndexDetails", + "/lance.index.pb.LabelListIndexDetails", + "/lance.index.pb.NGramIndexDetails", + "/lance.index.pb.ZoneMapIndexDetails", + "/lance.index.pb.InvertedIndexDetails", + ] { + assert!(registry.supports_details(&prost_types::Any { + type_url: type_url.to_string(), + value: Vec::new(), + })); + } + + for type_url in [ + "type.googleapis.com/example.BTreeIndexDetails", + "BTreeIndexDetails", + "/.lance.table.BTreeIndexDetails", + ] { + assert!(!registry.supports_details(&prost_types::Any { + type_url: type_url.to_string(), + value: Vec::new(), + })); + } + } } diff --git a/rust/lance/src/dataset/index.rs b/rust/lance/src/dataset/index.rs index b79d4359917..dd2b0f5073e 100644 --- a/rust/lance/src/dataset/index.rs +++ b/rust/lance/src/dataset/index.rs @@ -32,8 +32,9 @@ pub struct DatasetIndexRemapperOptions {} /// Loads index metadata when compaction has at least one index to remap. /// -/// Returns all index metadata, including system indices, so the remapper uses a -/// consistent snapshot. Returns `None` when there are no non-system indices. +/// Returns all usable index metadata, including system indices, so the remapper +/// uses a consistent snapshot. Returns `None` when there are no usable +/// non-system indices. pub(crate) async fn load_indices_for_remapping( dataset: &Dataset, ) -> Result>>> { @@ -255,6 +256,56 @@ mod tests { assert!(options.create_remapper(&dataset).await.unwrap().is_none()); } + #[tokio::test] + async fn test_remapper_not_created_for_unknown_index_type() { + let reader = lance_datagen::gen_batch() + .col("id", array::step::()) + .into_reader_rows(RowCount::from(1), BatchCount::from(1)); + let mut dataset = Dataset::write(reader, "memory://", None).await.unwrap(); + dataset + .create_index( + &["id"], + IndexType::BTree, + Some("id_idx".to_string()), + &ScalarIndexParams::for_builtin(BuiltinIndexType::BTree), + false, + ) + .await + .unwrap(); + + let current = dataset.load_indices().await.unwrap(); + let unknown = IndexMetadata { + index_details: Some(Arc::new(prost_types::Any { + type_url: "type.googleapis.com/example.ForeignIndexDetails".to_string(), + value: Vec::new(), + })), + fragment_bitmap: None, + ..current[0].clone() + }; + let transaction = Transaction::new( + dataset.manifest.version, + Operation::CreateIndex { + new_indices: vec![unknown], + removed_indices: current.to_vec(), + }, + None, + ); + dataset + .apply_commit(transaction, &Default::default(), &Default::default()) + .await + .unwrap(); + + assert!(dataset.load_indices().await.unwrap().is_empty()); + assert!( + DatasetIndexRemapperOptions::default() + .create_remapper(&dataset) + .await + .unwrap() + .is_none(), + "compaction must not migrate an index type this build cannot open" + ); + } + #[tokio::test] async fn test_remapper_only_touches_segments_with_affected_fragments() { let test_dir = tempfile::tempdir().unwrap(); diff --git a/rust/lance/src/dataset/optimize.rs b/rust/lance/src/dataset/optimize.rs index 387a5ad1f99..afa97dbdef3 100644 --- a/rust/lance/src/dataset/optimize.rs +++ b/rust/lance/src/dataset/optimize.rs @@ -102,9 +102,7 @@ use super::{ use crate::Dataset; use crate::Result; use crate::dataset::utils::CapturedRowIds; -use crate::index::{ - DatasetIndexExt, DatasetIndexInternalExt, load_all_indices, unsupported_index_version, -}; +use crate::index::{DatasetIndexExt, DatasetIndexInternalExt, index_is_usable, load_all_indices}; use crate::io::commit::{DEFAULT_COMMIT_RETRY_TIMEOUT, commit_transaction, migrate_fragments}; use arrow::array::AsArray; use arrow::datatypes::{UInt8Type, UInt32Type, UInt64Type}; @@ -2151,7 +2149,7 @@ async fn index_fragment_coverage( 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() { + if index_is_usable(index) { continue; } coverage.push(( diff --git a/rust/lance/src/dataset/scanner.rs b/rust/lance/src/dataset/scanner.rs index fddd98b3339..e6ebb4bba11 100644 --- a/rust/lance/src/dataset/scanner.rs +++ b/rust/lance/src/dataset/scanner.rs @@ -5326,10 +5326,9 @@ impl Scanner { // from `covering_fields` either -- that is computed from a field older // writers drop, so it would widen to the carried columns exactly when the // declaration is lost. - else if let Some(index) = indices - .iter() - .find(|i| i.fields.first() == Some(&column_id)) - { + else if let Some(index) = indices.iter().find(|i| { + i.fields.first() == Some(&column_id) && crate::index::index_type_is_known(i) + }) { // Try to get metric type from index metadata first (fast path for newer indices) let index_metric = if let Some(metric) = crate::index::vector::details::metric_type_from_index_metadata(index) diff --git a/rust/lance/src/index.rs b/rust/lance/src/index.rs index e7be09dbe9e..5f9f9ec3a32 100644 --- a/rust/lance/src/index.rs +++ b/rust/lance/src/index.rs @@ -847,6 +847,18 @@ fn segment_has_vector_details(segment: &IndexMetadata) -> bool { ) } +/// Whether this build has a reader for the index's declared type. +/// +/// Segments without details predate type URLs and remain readable through the +/// legacy file-based detection in the index open paths. +pub(crate) fn index_type_is_known(index: &IndexMetadata) -> bool { + is_system_index(index) + || index + .index_details + .as_ref() + .is_none_or(|details| IndexDetails(details.clone()).has_reader()) +} + /// Detect FTS / inverted segments from manifest details. /// /// Unlike vector, inverted segment support was added after index details were @@ -1859,16 +1871,13 @@ impl DatasetIndexExt for Dataset { async fn load_indices(&self) -> Result>> { let indices = load_all_indices(self).await?; - if indices - .iter() - .all(|idx| unsupported_index_version(idx).is_none()) - { + if indices.iter().all(index_is_usable) { return Ok(indices); } Ok(Arc::new( indices .iter() - .filter(|idx| unsupported_index_version(idx).is_none()) + .filter(|idx| index_is_usable(idx)) .cloned() .collect(), )) @@ -2291,6 +2300,20 @@ impl DatasetIndexExt for Dataset { let mut new_indices = vec![]; let mut removed_indices = vec![]; for (name, deltas) in name_to_indices.iter() { + if let Some(index) = deltas.iter().find(|idx| !index_type_is_known(idx)) { + let type_url = index + .index_details + .as_ref() + .map(|details| details.type_url.as_str()) + .unwrap_or(""); + log::warn!( + "Skipping optimization of index '{}' because this build does not recognize index type '{}'", + index.name, + type_url + ); + continue; + } + // 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 @@ -2682,12 +2705,8 @@ async fn gather_fragment_statistics( /// 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. +/// Reader availability is checked separately by [`index_is_usable`], because +/// an unknown type has no meaningful maximum version in this build. pub(crate) fn unsupported_index_version(index: &IndexMetadata) -> Option { let max_supported_version = index .index_details @@ -2701,6 +2720,15 @@ pub(crate) fn unsupported_index_version(index: &IndexMetadata) -> Option { (index.index_version > max_supported_version as i32).then_some(max_supported_version) } +/// Whether this build may expose an index through the usable-index view. +/// +/// System indices have dedicated readers rather than scalar plugins. Ordinary +/// indices need both a reader for their exact declared type and a supported +/// format version. +pub(crate) fn index_is_usable(index: &IndexMetadata) -> bool { + index_type_is_known(index) && unsupported_index_version(index).is_none() +} + /// Name the indices this build has no reader for, once per manifest read. /// /// Deliberately not inside the filter in [`DatasetIndexExt::load_indices`]: that @@ -2709,7 +2737,18 @@ pub(crate) fn unsupported_index_version(index: &IndexMetadata) -> Option { /// 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) { + if !index_type_is_known(idx) { + let type_url = idx + .index_details + .as_ref() + .map(|details| details.type_url.as_str()) + .unwrap_or(""); + log::warn!( + "Index {} has unrecognized type {}, ignoring it", + idx.name, + type_url, + ); + } else if let Some(max_supported_version) = unsupported_index_version(idx) { log::warn!( "Index {} has version {}, which is not supported (<={}), ignoring it", idx.name, @@ -4780,6 +4819,106 @@ mod tests { assert_eq!(stats["num_indexed_rows"], 512); } + #[tokio::test] + async fn test_v036_scalar_details_are_still_known() { + let test_dir = copy_test_data_to_tmp("0.36.0/btree_in_index_pkg.lance").unwrap(); + let dataset = Dataset::open(&test_dir.path_str()).await.unwrap(); + let indices = dataset.load_indices().await.unwrap(); + let details = indices[0].index_details.clone().unwrap(); + + assert_eq!(details.type_url, "/lance.index.pb.BTreeIndexDetails"); + assert_eq!(IndexDetails(details).get_plugin().unwrap().name(), "BTree"); + assert!(index_type_is_known(&indices[0])); + } + + #[tokio::test] + async fn test_unknown_index_type_does_not_block_queries_or_optimization() { + let reader = gen_batch() + .col("vector", array::rand_vec::(Dimension::from(8))) + .col("number", array::step::()) + .into_reader_rows(RowCount::from(64), BatchCount::from(1)); + let mut dataset = Dataset::write(reader, "memory://", None).await.unwrap(); + dataset + .create_index( + &["number"], + IndexType::BTree, + Some("number_idx".to_string()), + &ScalarIndexParams::default(), + true, + ) + .await + .unwrap(); + + let appended = gen_batch() + .col("vector", array::rand_vec::(Dimension::from(8))) + .col("number", array::step::()) + .into_reader_rows(RowCount::from(32), BatchCount::from(1)); + dataset.append(appended, None).await.unwrap(); + let stats: serde_json::Value = + serde_json::from_str(&dataset.index_statistics("number_idx").await.unwrap()).unwrap(); + assert_eq!(stats["num_unindexed_rows"], 32); + + let field_id = dataset.schema().field("vector").unwrap().id; + let fragment_ids = dataset + .get_fragments() + .iter() + .map(|fragment| fragment.id() as u32) + .collect::>(); + let mut foreign_segment = write_vector_segment_metadata( + &dataset, + "foreign_idx", + field_id, + Uuid::new_v4(), + fragment_ids, + b"opaque external index", + ) + .await; + foreign_segment.index_details = Some(Arc::new(prost_types::Any { + type_url: "type.googleapis.com/example.MyVectorIndexDetails".to_string(), + value: Vec::new(), + })); + foreign_segment.index_version = 1; + dataset + .commit_existing_index_segments("foreign_idx", "vector", vec![foreign_segment]) + .await + .unwrap(); + + assert!( + dataset + .load_indices_by_name("foreign_idx") + .await + .unwrap() + .is_empty(), + "an index with no reader must not enter the usable-index view" + ); + assert_eq!( + load_all_indices_by_name(&dataset, "foreign_idx") + .await + .unwrap() + .len(), + 1, + "hiding an unusable index must not erase its manifest metadata" + ); + + let query = Float32Array::from(vec![0.5_f32; 8]); + let mut scanner = dataset.scan(); + scanner.nearest("vector", &query, 5).unwrap(); + assert_eq!(scanner.try_into_batch().await.unwrap().num_rows(), 5); + + dataset.optimize_indices(&Default::default()).await.unwrap(); + let stats: serde_json::Value = + serde_json::from_str(&dataset.index_statistics("number_idx").await.unwrap()).unwrap(); + assert_eq!(stats["num_unindexed_rows"], 0); + assert_eq!( + load_all_indices_by_name(&dataset, "foreign_idx") + .await + .unwrap() + .len(), + 1, + "optimizing supported indices must preserve the opaque segment" + ); + } + #[tokio::test] async fn test_optimize_delta_indices() { let dimensions = 16; @@ -10811,18 +10950,34 @@ mod tests { .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) { + #[derive(Debug, Clone, Copy)] + enum UnreadableIndexKind { + NewerVersion, + UnknownType, + } + + /// Make `index_name` unreadable to this build, and give it the full fragment + /// coverage a real index of that name would have. + async fn hide_index_as(dataset: &mut Dataset, index_name: &str, kind: UnreadableIndexKind) { 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 mut unreadable = current.clone(); + match kind { + UnreadableIndexKind::NewerVersion => { + unreadable[0].index_version = current[0].index_version + 1; + } + UnreadableIndexKind::UnknownType => { + unreadable[0].index_details = Some(Arc::new(prost_types::Any { + type_url: "type.googleapis.com/example.ForeignIndexDetails".to_string(), + value: Vec::new(), + })); + } + } + unreadable[0].fragment_bitmap = Some(dataset.fragment_bitmap.as_ref().clone()); let transaction = Transaction::new( dataset.manifest.version, Operation::CreateIndex { - new_indices: from_the_future, + new_indices: unreadable, removed_indices: current, }, None, @@ -10833,12 +10988,16 @@ mod tests { .unwrap(); } + /// Raise `index_name` past the version this build can read. + async fn hide_index_from_this_build(dataset: &mut Dataset, index_name: &str) { + hide_index_as(dataset, index_name, UnreadableIndexKind::NewerVersion).await; + } + /// 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`. + /// A dataset carrying an unreadable BTree index over `id` 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" @@ -10848,7 +11007,11 @@ mod tests { /// 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 { + async fn dataset_with_an_unreadable_index( + uri: &str, + index_name: &str, + kind: UnreadableIndexKind, + ) -> Dataset { let mut dataset = Dataset::write(two_column_reader(), uri, None) .await .unwrap(); @@ -10860,7 +11023,7 @@ mod tests { .train(false) .await .unwrap(); - hide_index_from_this_build(&mut dataset, index_name).await; + hide_index_as(&mut dataset, index_name, kind).await; dataset .create_index_builder(&["payload"], IndexType::BTree, &btree_params) @@ -10871,7 +11034,12 @@ mod tests { dataset } - /// Indices the manifest itself carries, bypassing the version filter. + /// A dataset carrying an index whose version is newer than this build. + async fn dataset_with_an_index_from_a_newer_build(uri: &str, index_name: &str) -> Dataset { + dataset_with_an_unreadable_index(uri, index_name, UnreadableIndexKind::NewerVersion).await + } + + /// Indices the manifest itself carries, bypassing the usable-index filter. async fn raw_manifest_indices(dataset: &Dataset) -> Vec { lance_table::io::manifest::read_manifest_indexes( &dataset.object_store, @@ -10985,11 +11153,16 @@ mod tests { /// `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. + #[rstest] + #[case::newer_version(UnreadableIndexKind::NewerVersion)] + #[case::unknown_type(UnreadableIndexKind::UnknownType)] #[tokio::test] - async fn test_unsupported_index_without_a_bitmap_does_not_fail_later_commits() { + async fn test_unsupported_index_without_a_bitmap_does_not_fail_later_commits( + #[case] kind: UnreadableIndexKind, + ) { 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; + let mut dataset = dataset_with_an_unreadable_index(test_uri, "id_idx", kind).await; // Drop the coverage too, so migration would want to rebuild it. let hidden = manifest_index(&dataset, "id_idx").await; @@ -11642,8 +11815,13 @@ mod tests { /// /// 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. + #[rstest] + #[case::newer_version(UnreadableIndexKind::NewerVersion)] + #[case::unknown_type(UnreadableIndexKind::UnknownType)] #[tokio::test] - async fn test_compaction_defers_fragments_an_unsupported_index_covers() { + async fn test_compaction_defers_fragments_an_unsupported_index_covers( + #[case] kind: UnreadableIndexKind, + ) { let test_dir = tempfile::tempdir().unwrap(); let test_uri = test_dir.path().to_str().unwrap(); @@ -11664,7 +11842,7 @@ mod tests { .train(false) .await .unwrap(); - hide_index_from_this_build(&mut dataset, "id_idx").await; + hide_index_as(&mut dataset, "id_idx", kind).await; let covered = manifest_index(&dataset, "id_idx") .await .fragment_bitmap diff --git a/rust/lance/src/index/api.rs b/rust/lance/src/index/api.rs index a8eaf96d603..c17d821b608 100644 --- a/rust/lance/src/index/api.rs +++ b/rust/lance/src/index/api.rs @@ -280,10 +280,11 @@ pub trait DatasetIndexExt { /// 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. + /// An index whose declared type has no reader in this build, or 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/scalar.rs b/rust/lance/src/index/scalar.rs index fbdd062b2c5..9745cee874b 100644 --- a/rust/lance/src/index/scalar.rs +++ b/rust/lance/src/index/scalar.rs @@ -41,6 +41,7 @@ use lance_core::{Error, ROW_ADDR, ROW_ID, Result}; use lance_datafusion::exec::LanceExecutionOptions; use lance_index::frag_reuse::FragReuseIndexHandle; use lance_index::metrics::{MetricsCollector, NoOpMetricsCollector}; +use lance_index::pb::VectorIndexDetails; use lance_index::pbold::{ BTreeIndexDetails, BitmapIndexDetails, InvertedIndexDetails, LabelListIndexDetails, }; @@ -63,7 +64,7 @@ use lance_index::scalar::{ use lance_index::{IndexCriteria, IndexType}; use lance_table::format::{Fragment, IndexMetadata}; use log::info; -use prost::Message; +use prost::{Message, Name}; use tracing::instrument; // Log an update every TRAINING_UPDATE_FREQ million rows processed @@ -304,6 +305,23 @@ impl IndexDetails { SCALAR_INDEX_PLUGIN_REGISTRY.get_plugin_by_details(self.0.as_ref()) } + /// Returns whether this build has a reader for the complete declared type. + pub(crate) fn has_reader(&self) -> bool { + let Some((_, details_type_name)) = self.0.type_url.rsplit_once('/') else { + return false; + }; + if details_type_name.is_empty() || details_type_name.starts_with('.') { + return false; + } + + details_type_name.eq_ignore_ascii_case(&VectorIndexDetails::full_name()) + // MemWAL flush briefly wrote this pre-`pb` package name. Keep that + // exact historical native identity readable without accepting any + // other message that merely shares the VectorIndexDetails suffix. + || details_type_name.eq_ignore_ascii_case("lance.index.VectorIndexDetails") + || SCALAR_INDEX_PLUGIN_REGISTRY.supports_details(self.0.as_ref()) + } + /// Returns the index version pub fn index_version(&self) -> Result { if self.is_vector() { @@ -892,6 +910,34 @@ mod tests { } } + #[test] + fn test_has_reader_matches_complete_type_name_case_insensitively() { + let has_reader = |type_url: &str| { + IndexDetails(Arc::new(prost_types::Any { + type_url: type_url.to_string(), + value: Vec::new(), + })) + .has_reader() + }; + + for type_url in [ + "/lance.index.pb.VectorIndexDetails", + "type.googleapis.com/LANCE.INDEX.PB.VECTORINDEXDETAILS", + "type.googleapis.com/lance.index.VectorIndexDetails", + "type.googleapis.com/LANCE.TABLE.BTREEINDEXDETAILS", + ] { + assert!(has_reader(type_url), "expected a reader for {type_url}"); + } + + for type_url in [ + "type.googleapis.com/example.MyVectorIndexDetails", + "type.googleapis.com/example.BTreeIndexDetails", + "VectorIndexDetails", + ] { + assert!(!has_reader(type_url), "unexpected reader for {type_url}"); + } + } + #[test] fn test_index_matches_criteria_vector_index() { let index1 = make_index_metadata("vector_index", 1, Some(IndexType::Vector)); diff --git a/rust/lance/src/io/commit.rs b/rust/lance/src/io/commit.rs index 397087c709f..5f4a6d01c7a 100644 --- a/rust/lance/src/io/commit.rs +++ b/rust/lance/src/io/commit.rs @@ -56,7 +56,7 @@ use crate::dataset::{ }; use crate::index::DatasetIndexInternalExt; use crate::index::vector::details::infer_missing_vector_details; -use crate::index::{load_all_indices, unsupported_index_version}; +use crate::index::{index_is_usable, load_all_indices}; use crate::io::deletion::read_dataset_deletion_file; use crate::session::Session; use crate::session::caches::DSMetadataCache; @@ -952,7 +952,7 @@ async fn migrate_indices(dataset: &Dataset, indices: &mut [IndexMetadata]) -> Re // 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() { + if !index_is_usable(index) { continue; } if needs_recalculating.contains(&index.name)