refactor!: compose exact current-format readers

java/lance-jni/src/file_reader.rs

@@ -23,8 +23,8 @@ use lance::io::ObjectStore;
 use lance_core::cache::LanceCache;
 use lance_core::datatypes::{BlobHandling, OnMissing, Projection, Schema};
 use lance_encoding::decoder::{DecoderPlugins, FilterExpression};
-use lance_encoding::version::LanceFileVersion;
 use lance_file::reader::{FileReader, FileReaderOptions, ReaderProjection};
+use lance_file::versions as file_versions;
 use lance_io::object_store::{ObjectStoreParams, ObjectStoreRegistry};
 use lance_io::{
     ReadBatchParams,
@@ -262,18 +262,17 @@ pub extern "system" fn Java_org_lance_file_LanceFileReader_readAllNative(
 
             let transformed_schema = projection.to_bare_schema();
 
-            let field_id_to_column_index = base_schema
-                .fields_pre_order()
-                .filter(|field| {
-                    file_version < LanceFileVersion::V2_1
-                        || field.is_leaf()
-                        || field.is_packed_struct()
+            let (field_ids, column_indices) =
+                file_versions::data_file_columns(file_version, &base_schema);
+            let field_id_to_column_index = field_ids
+                .into_iter()
+                .zip(column_indices)
+                .filter_map(|(field_id, column_index)| {
+                    (column_index >= 0).then_some((field_id as u32, column_index as u32))
                 })
-                .enumerate()
-                .map(|(idx, field)| (field.id as u32, idx as u32))
                 .collect::<BTreeMap<_, _>>();
 
-            Some(ReaderProjection::from_field_ids(
+            Some(file_versions::reader_projection_from_field_ids(
                 file_version,
                 &transformed_schema,
                 &field_id_to_column_index,

python/src/file.rs

@@ -26,7 +26,6 @@ use lance_core::utils::path::LancePathExt;
 use lance_encoding::decoder::{DecoderPlugins, FilterExpression};
 use lance_file::reader::{
     BufferDescriptor, CachedFileMetadata, FileReader, FileReaderOptions, FileStatistics,
-    ReaderProjection,
 };
 use lance_file::writer::{FileWriter, FileWriterOptions};
 use lance_file::{LanceEncodingsIo, version::LanceFileVersion, versions as file_versions};
@@ -847,7 +846,7 @@ impl LanceFileReader {
         let mut base_projection = None;
         if let Some(columns) = columns {
             base_projection = Some(
-                ReaderProjection::from_column_names(
+                file_versions::reader_projection_from_column_names(
                     file_metadata.version(),
                     &file_metadata.file_schema,
                     &columns.iter().map(|s| s.as_str()).collect::<Vec<&str>>(),

rust/lance-file/src/compatibility_tests.rs

@@ -25,7 +25,7 @@ use tokio::io::AsyncWriteExt;
 
 use crate::reader::{EncodedBatchReaderExt, FileReader, FileReaderOptions};
 use crate::testing::FsFixture;
-use crate::version::{ConcreteFileVersion, LanceFileVersion};
+use crate::version::ConcreteFileVersion;
 use crate::versions;
 use crate::versions::v1::reader::FileReader as V1Reader;
 use crate::versions::v1::writer::{
@@ -338,10 +338,7 @@ async fn assert_current_reader_roundtrip(
     )
     .await
     .unwrap();
-    assert_eq!(
-        ConcreteFileVersion::from(reader.metadata().version()),
-        version
-    );
+    assert_eq!(reader.metadata().version(), version);
     assert!(
         reader
             .metadata()
@@ -421,7 +418,6 @@ async fn v2_0_embedded_writer_and_reader_are_wire_compatible() {
     assert_eq!(footer_version(expected_self_described), expected_footer);
     assert_eq!(footer_version(expected_mini), expected_footer);
 
-    let version = LanceFileVersion::from(ConcreteFileVersion::V2_0);
     let self_described =
         EncodedBatch::try_from_self_described_lance(Bytes::from_static(expected_self_described))
             .unwrap();
@@ -438,8 +434,7 @@ async fn v2_0_embedded_writer_and_reader_are_wire_compatible() {
     assert_record_batch_eq(&decoded, &batch);
 
     let mini =
-        EncodedBatch::try_from_mini_lance(Bytes::from_static(expected_mini), &schema, version)
-            .unwrap();
+        EncodedBatch::try_from_mini_lance(Bytes::from_static(expected_mini), &schema).unwrap();
     let decoded = decode_batch(
         &mini,
         &FilterExpression::no_filter(),

rust/lance-file/src/lib.rs

@@ -19,15 +19,17 @@ use format::MAGIC;
 use lance_core::{Error, Result};
 use lance_io::object_store::ObjectStore;
 use object_store::path::Path;
-use version::{ConcreteFileVersion, LanceFileVersion};
+use version::ConcreteFileVersion;
 
 pub async fn determine_file_version(
     store: &ObjectStore,
     path: &Path,
     known_size: Option<usize>,
-) -> Result<LanceFileVersion> {
+) -> Result<ConcreteFileVersion> {
     let size = match known_size {
-        None => store.size(path).await.unwrap() as usize,
+        None => usize::try_from(store.size(path).await?).map_err(|_| {
+            Error::invalid_input(format!("file {} is too large for this platform", path))
+        })?,
         Some(size) => size,
     };
     if size < 8 {
@@ -53,5 +55,5 @@ pub async fn determine_file_version(
     let major_version = u16::from_le_bytes([footer[0], footer[1]]);
     let minor_version = u16::from_le_bytes([footer[2], footer[3]]);
 
-    ConcreteFileVersion::from_footer_numbers(major_version, minor_version).map(Into::into)
+    ConcreteFileVersion::from_footer_numbers(major_version, minor_version)
 }