refactor: make index file versions exact

java/lance-jni/src/blocking_dataset.rs

@@ -1958,12 +1958,13 @@ fn inner_get_lance_file_format_version<'local>(
     let version_string = {
         let dataset_guard =
             unsafe { env.get_rust_field::<_, _, BlockingDataset>(java_dataset, NATIVE_DATASET) }?;
-        let version = dataset_guard
+        dataset_guard
             .inner
             .manifest()
             .data_storage_format
-            .lance_file_version()?;
-        version.to_string()
+            .lance_file_format()
+            .to_manifest_string()
+            .to_string()
     };
 
     Ok(env

java/lance-jni/src/file_reader.rs

@@ -261,7 +261,6 @@ pub extern "system" fn Java_org_lance_file_LanceFileReader_readAllNative(
             }
 
             let transformed_schema = projection.to_bare_schema();
-
             let (field_ids, column_indices) =
                 file_versions::data_file_columns(file_version, &base_schema);
             let field_id_to_column_index = field_ids

java/lance-jni/src/file_writer.rs

@@ -113,8 +113,7 @@ fn inner_open<'local>(
             .map(|value| value.parse::<LanceFileVersion>())
             .transpose()?
             .unwrap_or_default()
-            .resolve()
-            .into();
+            .resolve();
         file_versions::create_lazy_writer(version, obj_writer, FileWriterOptions::default())
     })?;
 

python/src/file.rs

@@ -279,8 +279,7 @@ impl LanceFileWriter {
             .transpose()
             .infer_error()?
             .unwrap_or_default()
-            .resolve()
-            .into();
+            .resolve();
         let options = FileWriterOptions {
             data_cache_bytes,
             keep_original_array,

python/src/utils.rs

@@ -25,7 +25,7 @@ use arrow_schema::DataType;
 use lance::Result;
 use lance::datatypes::Schema;
 use lance_arrow::FixedSizeListArrayExt;
-use lance_file::versions::v1::writer::FileWriter as V1FileWriter;
+use lance_file::versions::v1::writer::FileWriter as PreviousFileWriter;
 use lance_index::vector::hnsw::{HNSW, builder::HnswBuildParams};
 use lance_index::vector::kmeans::{
     KMeans as LanceKMeans, KMeansAlgoFloat, KMeansParams, compute_partitions,
@@ -242,7 +242,7 @@ impl Hnsw {
         let mut writer = rt()
             .block_on(
                 Some(py),
-                V1FileWriter::<ManifestDescribing>::try_new(
+                PreviousFileWriter::<ManifestDescribing>::try_new(
                     &object_store,
                     &path,
                     Schema::try_from(HNSW::schema().as_ref())

rust/lance-file/src/version.rs

@@ -29,7 +29,7 @@ pub const fn next_file_version() -> ConcreteFileVersion {
 ///
 /// `Stable` and `Next` are release selectors. They resolve to an exact
 /// [`ConcreteFileVersion`] before file or dataset dispatch and are never persisted.
-#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
+#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, Hash)]
 pub enum LanceFileVersion {
     /// The legacy v1 format.
     Legacy,
@@ -247,18 +247,6 @@ impl Display for ConcreteFileVersion {
     }
 }
 
-impl From<LanceFileVersion> for ConcreteFileVersion {
-    fn from(value: LanceFileVersion) -> Self {
-        value.resolve()
-    }
-}
-
-impl From<ConcreteFileVersion> for LanceFileVersion {
-    fn from(value: ConcreteFileVersion) -> Self {
-        value.to_selector()
-    }
-}
-
 fn unknown_version(value: impl Display) -> Error {
     Error::invalid_input_source(format!("Unknown Lance storage version: {}", value).into())
 }

rust/lance-index/src/scalar/lance_format.rs

@@ -14,9 +14,8 @@ use lance_core::{Error, Result, cache::LanceCache};
 use lance_encoding::decoder::{DecoderPlugins, FilterExpression};
 use lance_file::reader::{FileReader as CurrentFileReader, FileReaderOptions};
 use lance_file::version::ConcreteFileVersion;
-use lance_file::version::LanceFileVersion;
-use lance_file::versions;
 use lance_file::versions::v1::reader::FileReader as V1FileReader;
+use lance_file::versions::{self, OpenedFileReader};
 use lance_file::writer as current_writer;
 use lance_io::scheduler::{ScanScheduler, SchedulerConfig};
 use lance_io::utils::CachedFileSize;
@@ -43,7 +42,7 @@ pub struct LanceIndexStore {
     /// Cached file sizes (filename -> size in bytes)
     /// When set, used to avoid HEAD calls when opening files
     file_sizes: HashMap<String, u64>,
-    format_version: LanceFileVersion,
+    format_version: ConcreteFileVersion,
     /// Base I/O priority for all requests this store submits to `scheduler`.
     io_priority: u64,
 }
@@ -70,7 +69,7 @@ impl LanceIndexStore {
             object_store,
             index_dir,
             metadata_cache,
-            LanceFileVersion::V2_0,
+            ConcreteFileVersion::V2_0,
         )
     }
 
@@ -79,7 +78,7 @@ impl LanceIndexStore {
         object_store: Arc<ObjectStore>,
         index_dir: Path,
         metadata_cache: Arc<LanceCache>,
-        format_version: LanceFileVersion,
+        format_version: ConcreteFileVersion,
     ) -> Self {
         let scheduler = ScanScheduler::new(
             object_store.clone(),
@@ -420,7 +419,7 @@ impl IndexStore for LanceIndexStore {
         let schema = schema.as_ref().try_into()?;
         let writer = self.object_store.create(&path).await?;
         let writer = versions::create_writer(
-            ConcreteFileVersion::from(self.format_version),
+            self.format_version,
             writer,
             schema,
             current_writer::FileWriterOptions::default(),
@@ -452,31 +451,24 @@ impl IndexStore for LanceIndexStore {
             .scheduler
             .open_file_with_priority(&path, self.io_priority, &cached_size)
             .await?;
-        match CurrentFileReader::try_open(
+        match versions::open_self_described_reader(
             file_scheduler,
-            None,
             Arc::<DecoderPlugins>::default(),
             &self.metadata_cache,
             FileReaderOptions::default(),
         )
-        .await
+        .await?
         {
-            Ok(reader) => Ok(Arc::new(CurrentIndexReader(reader))),
-            Err(e) => {
-                // If the error is a version conflict we can try to read the file with v1 reader
-                if let Error::VersionConflict { .. } = e {
-                    let path = self.index_file_path(name)?;
-                    let file_reader = V1FileReader::try_new_self_described(
-                        &self.object_store,
-                        &path,
-                        Some(&self.metadata_cache),
-                    )
-                    .await?;
-                    Ok(Arc::new(V1IndexReader(file_reader)))
-                } else {
-                    Err(e)
-                }
+            OpenedFileReader::V1 { .. } => {
+                let reader = V1FileReader::try_new_self_described(
+                    &self.object_store,
+                    &path,
+                    Some(&self.metadata_cache),
+                )
+                .await?;
+                Ok(Arc::new(V1IndexReader(reader)))
             }
+            OpenedFileReader::Current(reader) => Ok(Arc::new(CurrentIndexReader(reader))),
         }
     }
 

rust/lance-index/src/vector/distributed/index_merger.rs

@@ -41,7 +41,6 @@ use bytes::Bytes;
 use lance_core::datatypes::Schema as LanceSchema;
 use lance_file::reader::{FileReader as V2Reader, FileReaderOptions as V2ReaderOptions};
 use lance_file::version::ConcreteFileVersion;
-use lance_file::version::LanceFileVersion;
 use lance_file::versions;
 use lance_file::writer::{FileWriter as V2Writer, FileWriter, FileWriterOptions};
 use lance_io::scheduler::{ScanScheduler, SchedulerConfig};
@@ -251,7 +250,7 @@ pub async fn init_writer_for_flat(
     d0: usize,
     item_type: &DataType,
     dt: DistanceType,
-    format_version: LanceFileVersion,
+    format_version: ConcreteFileVersion,
 ) -> Result<FileWriter> {
     let arrow_schema = ArrowSchema::new(vec![
         (*ROW_ID_FIELD).clone(),
@@ -266,7 +265,7 @@ pub async fn init_writer_for_flat(
     ]);
     let writer = object_store.create(aux_out).await?;
     let mut w = versions::create_writer(
-        ConcreteFileVersion::from(format_version),
+        format_version,
         writer,
         LanceSchema::try_from(&arrow_schema)?,
         FileWriterOptions::default(),
@@ -285,7 +284,7 @@ pub async fn init_writer_for_pq(
     aux_out: &object_store::path::Path,
     dt: DistanceType,
     pm: &ProductQuantizationMetadata,
-    format_version: LanceFileVersion,
+    format_version: ConcreteFileVersion,
 ) -> Result<FileWriter> {
     let num_bytes = if pm.nbits == 4 {
         pm.num_sub_vectors / 2
@@ -305,7 +304,7 @@ pub async fn init_writer_for_pq(
     ]);
     let writer = object_store.create(aux_out).await?;
     let mut w = versions::create_writer(
-        ConcreteFileVersion::from(format_version),
+        format_version,
         writer,
         LanceSchema::try_from(&arrow_schema)?,
         FileWriterOptions::default(),
@@ -330,7 +329,7 @@ pub async fn init_writer_for_sq(
     aux_out: &object_store::path::Path,
     dt: DistanceType,
     sq_meta: &ScalarQuantizationMetadata,
-    format_version: LanceFileVersion,
+    format_version: ConcreteFileVersion,
 ) -> Result<FileWriter> {
     let d0 = sq_meta.dim;
     let arrow_schema = ArrowSchema::new(vec![
@@ -346,7 +345,7 @@ pub async fn init_writer_for_sq(
     ]);
     let writer = object_store.create(aux_out).await?;
     let mut w = versions::create_writer(
-        ConcreteFileVersion::from(format_version),
+        format_version,
         writer,
         LanceSchema::try_from(&arrow_schema)?,
         FileWriterOptions::default(),
@@ -362,7 +361,7 @@ pub async fn init_writer_for_rq(
     aux_out: &object_store::path::Path,
     dt: DistanceType,
     rq_meta: &RabitQuantizationMetadata,
-    format_version: LanceFileVersion,
+    format_version: ConcreteFileVersion,
 ) -> Result<FileWriter> {
     let mut fields = vec![
         (*ROW_ID_FIELD).clone(),
@@ -381,7 +380,7 @@ pub async fn init_writer_for_rq(
     let arrow_schema = ArrowSchema::new(fields);
     let writer = object_store.create(aux_out).await?;
     let mut w = versions::create_writer(
-        ConcreteFileVersion::from(format_version),
+        format_version,
         writer,
         LanceSchema::try_from(&arrow_schema)?,
         FileWriterOptions::default(),
@@ -788,7 +787,7 @@ pub async fn merge_partial_vector_auxiliary_files(
     let mut dim: Option<usize> = None;
     let mut detected_index_type: Option<SupportedIvfIndexType> = None;
     // Inherit file format version from the first shard (set on first iteration)
-    let mut format_version: Option<LanceFileVersion> = None;
+    let mut format_version: Option<ConcreteFileVersion> = None;
 
     // Prepare output path; we'll create writer once when we know schema
     let aux_out = target_dir.clone().join(INDEX_AUXILIARY_FILE_NAME);
@@ -840,7 +839,7 @@ pub async fn merge_partial_vector_auxiliary_files(
 
         // Inherit format version from the first shard file
         if format_version.is_none() {
-            format_version = Some(meta.version().into());
+            format_version = Some(meta.version());
         }
 
         // Read distance type
@@ -949,8 +948,9 @@ pub async fn merge_partial_vector_auxiliary_files(
         let idx_type = detected_index_type
             .ok_or_else(|| Error::index("Unable to detect index type".to_string()))?;
 
-        // Compute format version once; defaults to V2_0 if no shards processed yet
-        let fv = format_version.unwrap_or(LanceFileVersion::V2_0);
+        let fv = format_version.ok_or_else(|| {
+            Error::internal("first shard did not provide a concrete file version".to_string())
+        })?;
 
         match idx_type {
             SupportedIvfIndexType::IvfSq => {

rust/lance-index/src/vector/ivf/shuffler.rs

@@ -34,7 +34,6 @@ use lance_core::{Error, ROW_ID, Result, datatypes::Schema};
 use lance_encoding::decoder::{DecoderPlugins, FilterExpression};
 use lance_file::reader::{FileReader as Lancev2FileReader, FileReaderOptions};
 use lance_file::version::ConcreteFileVersion;
-use lance_file::version::LanceFileVersion;
 use lance_file::versions;
 use lance_file::versions::v1::reader::FileReader as V1FileReader;
 use lance_file::versions::v1::writer::FileWriter as V1FileWriter;
@@ -408,7 +407,7 @@ pub struct IvfShuffler {
 
     shuffle_output_root_filename: String,
 
-    format_version: LanceFileVersion,
+    format_version: ConcreteFileVersion,
 }
 
 /// Represents a range of batches in a file that should be shuffled
@@ -448,11 +447,11 @@ impl IvfShuffler {
             unsorted_buffers: vec![],
             is_legacy,
             shuffle_output_root_filename,
-            format_version: LanceFileVersion::V2_0,
+            format_version: ConcreteFileVersion::V2_0,
         })
     }
 
-    pub fn with_format_version(mut self, format_version: LanceFileVersion) -> Self {
+    pub fn with_format_version(mut self, format_version: ConcreteFileVersion) -> Self {
         self.format_version = format_version;
         self
     }
@@ -808,7 +807,7 @@ impl IvfShuffler {
                     )]));
                     let lance_schema = Schema::try_from(sorted_file_schema.as_ref())?;
                     let mut file_writer = versions::create_writer(
-                        ConcreteFileVersion::from(this.format_version),
+                        this.format_version,
                         writer,
                         lance_schema,
                         FileWriterOptions::default(),

rust/lance-index/src/vector/v3/shuffler.rs

@@ -22,7 +22,6 @@ use lance_core::{
 use lance_encoding::decoder::{DecoderPlugins, FilterExpression};
 use lance_file::reader::{FileReader, FileReaderOptions};
 use lance_file::version::ConcreteFileVersion;
-use lance_file::version::LanceFileVersion;
 use lance_file::versions;
 use lance_file::writer::FileWriterOptions;
 use lance_io::{
@@ -73,7 +72,7 @@ pub struct IvfShuffler {
     object_store: Arc<ObjectStore>,
     output_dir: Path,
     num_partitions: usize,
-    format_version: LanceFileVersion,
+    format_version: ConcreteFileVersion,
 
     progress: Arc<dyn crate::progress::IndexBuildProgress>,
 }
@@ -84,12 +83,12 @@ impl IvfShuffler {
             object_store: Arc::new(ObjectStore::local()),
             output_dir,
             num_partitions,
-            format_version: LanceFileVersion::V2_0,
+            format_version: ConcreteFileVersion::V2_0,
             progress: crate::progress::noop_progress(),
         }
     }
 
-    pub fn with_format_version(mut self, format_version: LanceFileVersion) -> Self {
+    pub fn with_format_version(mut self, format_version: ConcreteFileVersion) -> Self {
         self.format_version = format_version;
         self
     }
@@ -125,7 +124,7 @@ impl Shuffler for IvfShuffler {
                 async move {
                     let writer = object_store.create(&part_path).await?;
                     let file_writer = versions::create_writer(
-                        ConcreteFileVersion::from(format_version),
+                        format_version,
                         writer,
                         lance_core::datatypes::Schema::try_from(&schema)?,
                         FileWriterOptions::default(),
@@ -315,7 +314,7 @@ impl ShuffleReader for EmptyReader {
 pub fn create_ivf_shuffler(
     output_dir: Path,
     num_partitions: usize,
-    format_version: LanceFileVersion,
+    format_version: ConcreteFileVersion,
     progress: Option<Arc<dyn crate::progress::IndexBuildProgress>>,
 ) -> Box<dyn Shuffler> {
     let use_legacy = std::env::var("LANCE_LEGACY_SHUFFLER")
@@ -478,7 +477,7 @@ impl Shuffler for TwoFileShuffler {
         let mut file_writer = versions::v2_1::create_writer(
             writer,
             lance_core::datatypes::Schema::try_from(&schema)?,
-            Default::default(),
+            FileWriterOptions::default(),
         )?
         .with_page_metadata_spill(self.object_store.clone(), spill_path);
 
@@ -489,7 +488,7 @@ impl Shuffler for TwoFileShuffler {
         let mut offsets_writer = versions::v2_1::create_writer(
             writer,
             lance_core::datatypes::Schema::try_from(offsets_schema.as_ref())?,
-            Default::default(),
+            FileWriterOptions::default(),
         )?
         .with_page_metadata_spill(self.object_store.clone(), spill_path);