fix(index): skip unrecognized index types

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<String, Box<dyn ScalarIndexPlugin>>,
+    details_type_names: HashSet<String>,
 }
 
 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<DetailsType: prost::Name>(&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<Self> {
         let mut registry = Self {
             plugins: HashMap::new(),
+            details_type_names: HashSet::new(),
         };
         registry.add_plugin::<pbold::BTreeIndexDetails, BTreeIndexPlugin>();
         registry.add_plugin::<pbold::BitmapIndexDetails, BitmapIndexPlugin>();
@@ -96,6 +112,17 @@ impl IndexPluginRegistry {
         #[cfg(feature = "geo")]
         registry.add_plugin::<pb::RTreeIndexDetails, RTreeIndexPlugin>();
 
+        // 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::<pbold::BTreeIndexDetails>(V036_SCALAR_DETAILS_PACKAGE);
+        registry.add_details_type_alias::<pbold::BitmapIndexDetails>(V036_SCALAR_DETAILS_PACKAGE);
+        registry
+            .add_details_type_alias::<pbold::LabelListIndexDetails>(V036_SCALAR_DETAILS_PACKAGE);
+        registry.add_details_type_alias::<pbold::NGramIndexDetails>(V036_SCALAR_DETAILS_PACKAGE);
+        registry.add_details_type_alias::<pbold::ZoneMapIndexDetails>(V036_SCALAR_DETAILS_PACKAGE);
+        registry.add_details_type_alias::<pbold::InvertedIndexDetails>(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(),
+            }));
+        }
+    }
 }

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<Option<Arc<Vec<IndexMetadata>>>> {
@@ -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::<arrow_array::types::Int32Type>())
+            .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();

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<Vec<(String, RoaringBitmap)>> {
     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((

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)