feat: ignore indices created in newer format (#3867)

BubbleCal · web-flow · commit c38f53848c69 · 2025-05-31T11:04:59.000+08:00
diff --git a/protos/table.proto b/protos/table.proto
@@ -194,6 +194,9 @@ message IndexMetadata {
   /// Indices should avoid putting large amounts of information in this field, as it will
   /// bloat the manifest.
   google.protobuf.Any index_details = 6;
+
+  /// The minimum lance version that this index is compatible with.
+  optional int32 index_version = 7;
 }
 
 // Index Section, containing a list of index metadata for one dataset version.
diff --git a/python/python/lance/dataset.py b/python/python/lance/dataset.py
@@ -3099,6 +3099,7 @@ class CreateIndex(BaseOperation):
         fields: List[int]
         dataset_version: int
         fragment_ids: Set[int]
+        index_version: int
 
     @dataclass
     class DataReplacementGroup:
diff --git a/python/python/tests/test_commit_index.py b/python/python/tests/test_commit_index.py
@@ -62,6 +62,7 @@ def test_commit_index(dataset_with_index, test_table, tmp_path):
         [field_idx],
         dataset_without_index.version,
         set([f.fragment_id for f in dataset_without_index.get_fragments()]),
+        0,
     )
     dataset_without_index = lance.LanceDataset.commit(
         dataset_without_index.uri,
diff --git a/python/src/transaction.rs b/python/src/transaction.rs
@@ -128,7 +128,7 @@ impl FromPyObject<'_> for PyLance<Operation> {
                 let name = ob.getattr("name")?.extract()?;
                 let fields = ob.getattr("fields")?.extract()?;
                 let dataset_version = ob.getattr("dataset_version")?.extract()?;
-
+                let index_version = ob.getattr("index_version")?.extract()?;
                 let fragment_ids = ob.getattr("fragment_ids")?;
                 let fragment_ids_ref: &Bound<'_, PySet> = fragment_ids.downcast()?;
                 let fragment_ids = fragment_ids_ref
@@ -147,6 +147,7 @@ impl FromPyObject<'_> for PyLance<Operation> {
                     // TODO: we should use lance::dataset::Dataset::commit_existing_index once
                     // we have a way to determine index details from an existing index.
                     index_details: None,
+                    index_version,
                 }];
 
                 let op = Operation::CreateIndex {
diff --git a/rust/lance-index/src/lib.rs b/rust/lance-index/src/lib.rs
@@ -168,6 +168,31 @@ impl IndexType {
                 | Self::IvfSq
         )
     }
+
+    /// Returns the current format version of the index type,
+    /// bump this when the index format changes.
+    /// Indices which higher version than these will be ignored for compatibility,
+    /// This would happen when creating index in a newer version of Lance,
+    /// but then opening the index in older version of Lance
+    pub fn version(&self) -> i32 {
+        match self {
+            Self::Scalar => 0,
+            Self::BTree => 0,
+            Self::Bitmap => 0,
+            Self::LabelList => 0,
+            Self::Inverted => 0,
+            Self::NGram => 0,
+
+            // for now all vector indices are built by the same builder,
+            // so they share the same version.
+            Self::Vector
+            | Self::IvfFlat
+            | Self::IvfSq
+            | Self::IvfPq
+            | Self::IvfHnswSq
+            | Self::IvfHnswPq => 0,
+        }
+    }
 }
 
 pub trait IndexParams: Send + Sync {
diff --git a/rust/lance-index/src/scalar.rs b/rust/lance-index/src/scalar.rs
@@ -67,6 +67,18 @@ impl TryFrom<IndexType> for ScalarIndexType {
     }
 }
 
+impl From<ScalarIndexType> for IndexType {
+    fn from(val: ScalarIndexType) -> Self {
+        match val {
+            ScalarIndexType::BTree => Self::BTree,
+            ScalarIndexType::Bitmap => Self::Bitmap,
+            ScalarIndexType::LabelList => Self::LabelList,
+            ScalarIndexType::NGram => Self::NGram,
+            ScalarIndexType::Inverted => Self::Inverted,
+        }
+    }
+}
+
 #[derive(Default)]
 pub struct ScalarIndexParams {
     /// If set then always use the given index type and skip auto-detection
diff --git a/rust/lance-table/src/format/index.rs b/rust/lance-table/src/format/index.rs
@@ -10,7 +10,6 @@ use uuid::Uuid;
 
 use super::pb;
 use lance_core::{Error, Result};
-
 /// Index metadata
 #[derive(Debug, Clone, PartialEq)]
 pub struct Index {
@@ -36,6 +35,9 @@ pub struct Index {
     /// This is an Option because older versions of Lance may not have this defined.  However, it should always
     /// be present in newer versions.
     pub index_details: Option<prost_types::Any>,
+
+    /// The index version.
+    pub index_version: i32,
 }
 
 impl DeepSizeOf for Index {
@@ -76,6 +78,7 @@ impl TryFrom<pb::IndexMetadata> for Index {
             dataset_version: proto.dataset_version,
             fragment_bitmap,
             index_details: proto.index_details,
+            index_version: proto.index_version.unwrap_or_default(),
         })
     }
 }
@@ -99,6 +102,7 @@ impl From<&Index> for pb::IndexMetadata {
             dataset_version: idx.dataset_version,
             fragment_bitmap,
             index_details: idx.index_details.clone(),
+            index_version: Some(idx.index_version),
         }
     }
 }
diff --git a/rust/lance/src/dataset/transaction.rs b/rust/lance/src/dataset/transaction.rs
@@ -2510,6 +2510,7 @@ mod tests {
             dataset_version: 1,
             fragment_bitmap: None,
             index_details: None,
+            index_version: 0,
         };
         let fragment0 = Fragment::new(0);
         let fragment1 = Fragment::new(1);
diff --git a/rust/lance/src/index.rs b/rust/lance/src/index.rs
@@ -56,8 +56,8 @@ use lance_table::format::{Fragment, SelfDescribingFileReader};
 use lance_table::io::manifest::read_manifest_indexes;
 use roaring::RoaringBitmap;
 use scalar::{
-    build_inverted_index, detect_scalar_index_type, index_matches_criteria, inverted_index_details,
-    TrainingRequest,
+    build_inverted_index, detect_scalar_index_type, index_matches_criteria, infer_index_type,
+    inverted_index_details, TrainingRequest,
 };
 use serde_json::json;
 use snafu::location;
@@ -281,7 +281,7 @@ impl DatasetIndexExt for Dataset {
         }
 
         let index_id = Uuid::new_v4();
-        let index_details: prost_types::Any = match (index_type, params.index_name()) {
+        let index_details = match (index_type, params.index_name()) {
             (
                 IndexType::Bitmap
                 | IndexType::BTree
@@ -388,6 +388,7 @@ impl DatasetIndexExt for Dataset {
             dataset_version: self.manifest.version,
             fragment_bitmap: Some(self.get_fragments().iter().map(|f| f.id() as u32).collect()),
             index_details: Some(index_details),
+            index_version: index_type.version(),
         };
         let transaction = Transaction::new(
             self.manifest.version,
@@ -456,10 +457,27 @@ impl DatasetIndexExt for Dataset {
             return Ok(indices);
         }
 
-        let loaded_indices: Arc<Vec<IndexMetadata>> =
+        let loaded_indices: Vec<IndexMetadata> =
             read_manifest_indexes(&self.object_store, &self.manifest_location, &self.manifest)
                 .await?
-                .into();
+                .into_iter()
+                .filter(|idx| {
+                    let max_valid_version = infer_index_type(idx)
+                        .map(|t| t.version())
+                        .unwrap_or_default();
+                    let is_valid = idx.index_version <= max_valid_version;
+                    if !is_valid {
+                        log::warn!(
+                            "Index {} has version {}, which is not supported (<={}), ignoring it",
+                            idx.name,
+                            idx.index_version,
+                            max_valid_version,
+                        );
+                    }
+                    is_valid
+                })
+                .collect();
+        let loaded_indices = Arc::new(loaded_indices);
 
         self.session.index_cache.insert_metadata(
             self.base.as_ref(),
@@ -492,6 +510,7 @@ impl DatasetIndexExt for Dataset {
             dataset_version: self.manifest.version,
             fragment_bitmap: Some(self.get_fragments().iter().map(|f| f.id() as u32).collect()),
             index_details: None,
+            index_version: 0,
         };
 
         let transaction = Transaction::new(
@@ -574,28 +593,25 @@ impl DatasetIndexExt for Dataset {
         let mut new_indices = vec![];
         let mut removed_indices = vec![];
         for deltas in name_to_indices.values() {
-            let Some((new_id, removed, mut new_frag_ids)) =
-                merge_indices(dataset.clone(), deltas.as_slice(), options).await?
+            let Some(res) = merge_indices(dataset.clone(), deltas.as_slice(), options).await?
             else {
                 continue;
             };
-            for removed_idx in removed.iter() {
-                new_frag_ids |= removed_idx.fragment_bitmap.as_ref().unwrap();
-            }
 
             let last_idx = deltas.last().expect("Delta indices should not be empty");
             let new_idx = IndexMetadata {
-                uuid: new_id,
+                uuid: res.new_uuid,
                 name: last_idx.name.clone(), // Keep the same name
                 fields: last_idx.fields.clone(),
                 dataset_version: self.manifest.version,
-                fragment_bitmap: Some(new_frag_ids),
+                fragment_bitmap: Some(res.new_fragment_bitmap),
                 index_details: last_idx.index_details.clone(),
+                index_version: res.new_index_version,
             };
-            removed_indices.extend(removed.iter().map(|&idx| idx.clone()));
-            if deltas.len() > removed.len() {
+            removed_indices.extend(res.removed_indices.iter().map(|&idx| idx.clone()));
+            if deltas.len() > removed_indices.len() {
                 new_indices.extend(
-                    deltas[0..(deltas.len() - removed.len())]
+                    deltas[0..(deltas.len() - res.removed_indices.len())]
                         .iter()
                         .map(|&idx| idx.clone()),
                 );
@@ -1104,7 +1120,8 @@ impl DatasetIndexInternalExt for Dataset {
             let is_vector_index = idx_schema
                 .fields
                 .iter()
-                .any(|f| matches!(f.data_type(), DataType::FixedSizeList(_, _)));
+                .any(|f| is_vector_field(f.data_type()));
+
             idx.fields.len() == 1 && !is_vector_index
         }) {
             let field = index.fields[0];
@@ -1208,6 +1225,17 @@ impl DatasetIndexInternalExt for Dataset {
     }
 }
 
+fn is_vector_field(data_type: DataType) -> bool {
+    match data_type {
+        DataType::FixedSizeList(_, _) => true,
+        DataType::List(inner) => {
+            // If the inner type is a fixed size list, then it is a multivector field
+            matches!(inner.data_type(), DataType::FixedSizeList(_, _))
+        }
+        _ => false,
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use crate::dataset::builder::DatasetBuilder;
diff --git a/rust/lance/src/index/append.rs b/rust/lance/src/index/append.rs
@@ -19,6 +19,13 @@ use crate::dataset::index::LanceIndexStoreExt;
 use crate::dataset::scanner::ColumnOrdering;
 use crate::dataset::Dataset;
 
+pub struct IndexMergeResults<'a> {
+    pub new_uuid: Uuid,
+    pub removed_indices: Vec<&'a IndexMetadata>,
+    pub new_fragment_bitmap: RoaringBitmap,
+    pub new_index_version: i32,
+}
+
 /// Merge in-inflight unindexed data, with a specific number of previous indices
 /// into a new index, to improve the query performance.
 ///
@@ -33,7 +40,7 @@ pub async fn merge_indices<'a>(
     dataset: Arc<Dataset>,
     old_indices: &[&'a IndexMetadata],
     options: &OptimizeOptions,
-) -> Result<Option<(Uuid, Vec<&'a IndexMetadata>, RoaringBitmap)>> {
+) -> Result<Option<IndexMergeResults<'a>>> {
     if old_indices.is_empty() {
         return Err(Error::Index {
             message: "Append index: no previous index found".to_string(),
@@ -76,7 +83,8 @@ pub async fn merge_indices<'a>(
         frag_bitmap.insert(frag.id as u32);
     });
 
-    let (new_uuid, indices_merged) = match indices[0].index_type() {
+    let index_type = indices[0].index_type();
+    let (new_uuid, indices_merged) = match index_type {
         it if it.is_scalar() => {
             // There are no delta indices for scalar, so adding all indexed
             // fragments to the new index.
@@ -171,11 +179,17 @@ pub async fn merge_indices<'a>(
         }),
     }?;
 
-    Ok(Some((
+    let removed_indices = old_indices[old_indices.len() - indices_merged..].to_vec();
+    for removed in removed_indices.iter() {
+        frag_bitmap |= removed.fragment_bitmap.as_ref().unwrap();
+    }
+
+    Ok(Some(IndexMergeResults {
         new_uuid,
-        old_indices[old_indices.len() - indices_merged..].to_vec(),
-        frag_bitmap,
-    )))
+        removed_indices,
+        new_fragment_bitmap: frag_bitmap,
+        new_index_version: index_type.version(),
+    }))
 }
 
 #[cfg(test)]
diff --git a/rust/lance/src/index/scalar.rs b/rust/lance/src/index/scalar.rs
diff --git a/rust/lance/src/index/vector/ivf.rs b/rust/lance/src/index/vector/ivf.rs

Original file line number	Diff line number	Diff line change
`@@ -194,6 +194,9 @@ message IndexMetadata {`
`194`	`194`	`/// Indices should avoid putting large amounts of information in this field, as it will`
`195`	`195`	`/// bloat the manifest.`
`196`	`196`	`google.protobuf.Any index_details = 6;`
	`197`	`+`
	`198`	`+ /// The minimum lance version that this index is compatible with.`
	`199`	`+ optional int32 index_version = 7;`
`197`	`200`	`}`
`198`	`201`
`199`	`202`	`// Index Section, containing a list of index metadata for one dataset version.`
Original file line number	Diff line number	Diff line change
`@@ -62,6 +62,7 @@ def test_commit_index(dataset_with_index, test_table, tmp_path):`
`62`	`62`	`[field_idx],`
`63`	`63`	`dataset_without_index.version,`
`64`	`64`	`set([f.fragment_id for f in dataset_without_index.get_fragments()]),`
	`65`	`+ 0,`
`65`	`66`	`)`
`66`	`67`	`dataset_without_index = lance.LanceDataset.commit(`
`67`	`68`	`dataset_without_index.uri,`
Original file line number	Diff line number	Diff line change
`@@ -10,7 +10,6 @@ use uuid::Uuid;`
`10`	`10`
`11`	`11`	`use super::pb;`
`12`	`12`	`use lance_core::{Error, Result};`
`13`		`-`
`14`	`13`	`/// Index metadata`
`15`	`14`	`#[derive(Debug, Clone, PartialEq)]`
`16`	`15`	`pub struct Index {`
`@@ -36,6 +35,9 @@ pub struct Index {`
`36`	`35`	`/// This is an Option because older versions of Lance may not have this defined. However, it should always`
`37`	`36`	`/// be present in newer versions.`
`38`	`37`	`pub index_details: Option<prost_types::Any>,`
	`38`	`+`
	`39`	`+ /// The index version.`
	`40`	`+ pub index_version: i32,`
`39`	`41`	`}`
`40`	`42`
`41`	`43`	`impl DeepSizeOf for Index {`
`@@ -76,6 +78,7 @@ impl TryFrom<pb::IndexMetadata> for Index {`
`76`	`78`	`dataset_version: proto.dataset_version,`
`77`	`79`	`fragment_bitmap,`
`78`	`80`	`index_details: proto.index_details,`
	`81`	`+ index_version: proto.index_version.unwrap_or_default(),`
`79`	`82`	`})`
`80`	`83`	`}`
`81`	`84`	`}`
`@@ -99,6 +102,7 @@ impl From<&Index> for pb::IndexMetadata {`
`99`	`102`	`dataset_version: idx.dataset_version,`
`100`	`103`	`fragment_bitmap,`
`101`	`104`	`index_details: idx.index_details.clone(),`
	`105`	`+ index_version: Some(idx.index_version),`
`102`	`106`	`}`
`103`	`107`	`}`
`104`	`108`	`}`