fix partition values

datafusion_iceberg/src/table.rs

@@ -596,20 +596,14 @@ impl DataSink for IcebergDataSink {
 #[cfg(test)]
 mod tests {
 
-    use datafusion::{
-        arrow::{
-            array::{Float32Array, Int64Array},
-            record_batch::RecordBatch,
-        },
-        prelude::SessionContext,
-    };
+    use datafusion::{arrow::array::Int64Array, prelude::SessionContext};
     use iceberg_rust::spec::{
         partition::{PartitionField, Transform},
         schema::Schema,
         types::{PrimitiveType, StructField, StructType, Type},
     };
     use iceberg_rust::{
-        catalog::{identifier::Identifier, tabular::Tabular, Catalog},
+        catalog::Catalog,
         spec::{
             partition::PartitionSpec,
             view_metadata::{Version, ViewRepresentation},
@@ -618,64 +612,10 @@ mod tests {
         view::View,
     };
     use iceberg_sql_catalog::SqlCatalog;
-    use object_store::{local::LocalFileSystem, memory::InMemory, ObjectStore};
+    use object_store::{memory::InMemory, ObjectStore};
     use std::sync::Arc;
 
-    use crate::{catalog::catalog::IcebergCatalog, error::Error, DataFusionTable};
-
-    #[tokio::test]
-    pub async fn test_datafusion_table_scan() {
-        let object_store: Arc<dyn ObjectStore> =
-            Arc::new(LocalFileSystem::new_with_prefix("../iceberg-tests/nyc_taxis").unwrap());
-
-        let catalog: Arc<dyn Catalog> = Arc::new(
-            SqlCatalog::new("sqlite://", "test", object_store.clone())
-                .await
-                .unwrap(),
-        );
-        let identifier = Identifier::parse("test.table1", None).unwrap();
-
-        catalog.clone().register_table(identifier.clone(), "/home/iceberg/warehouse/nyc/taxis/metadata/fb072c92-a02b-11e9-ae9c-1bb7bc9eca94.metadata.json").await.expect("Failed to register table.");
-
-        let table = if let Tabular::Table(table) = catalog
-            .load_tabular(&identifier)
-            .await
-            .expect("Failed to load table")
-        {
-            Ok(Arc::new(DataFusionTable::from(table)))
-        } else {
-            Err(Error::InvalidFormat(
-                "Entity returned from catalog".to_string(),
-            ))
-        }
-        .unwrap();
-
-        let ctx = SessionContext::new();
-
-        ctx.register_table("nyc_taxis", table).unwrap();
-
-        let df = ctx
-            .sql("SELECT vendor_id, MIN(trip_distance) FROM nyc_taxis GROUP BY vendor_id")
-            .await
-            .unwrap();
-
-        // execute the plan
-        let results: Vec<RecordBatch> = df.collect().await.expect("Failed to execute query plan.");
-
-        let batch = results
-            .into_iter()
-            .find(|batch| batch.num_rows() > 0)
-            .expect("All record batches are empty");
-
-        let values = batch
-            .column(1)
-            .as_any()
-            .downcast_ref::<Float32Array>()
-            .expect("Failed to get values from batch.");
-
-        // Value can either be 0.9 or 1.8
-        assert!(((1.35 - values.value(0)).abs() - 0.45).abs() < 0.001)
-    }
+    use crate::{catalog::catalog::IcebergCatalog, DataFusionTable};
 
     #[tokio::test]
     pub async fn test_datafusion_table_insert() {

iceberg-rust-spec/src/spec/manifest_list.rs

@@ -5,6 +5,7 @@
 use std::{
     io::Read,
     iter::{repeat, Map, Repeat, Zip},
+    sync::OnceLock,
 };
 
 use apache_avro::{types::Value as AvroValue, Reader as AvroReader, Schema as AvroSchema};
@@ -42,12 +43,13 @@ impl<'a, 'metadata, R: Read> Iterator for ManifestListReader<'a, 'metadata, R> {
 
 impl<'a, 'metadata, R: Read> ManifestListReader<'a, 'metadata, R> {
     /// Create a new ManifestFile reader
-    pub fn new(
-        reader: R,
-        table_metadata: &'metadata TableMetadata,
-    ) -> Result<Self, apache_avro::Error> {
+    pub fn new(reader: R, table_metadata: &'metadata TableMetadata) -> Result<Self, Error> {
+        let schema: &AvroSchema = match table_metadata.format_version {
+            FormatVersion::V1 => manifest_list_schema_v1(),
+            FormatVersion::V2 => manifest_list_schema_v2(),
+        };
         Ok(Self {
-            reader: AvroReader::new(reader)?
+            reader: AvroReader::with_schema(&schema, reader)?
                 .zip(repeat(table_metadata))
                 .map(avro_value_to_manifest_file),
         })
@@ -404,11 +406,11 @@ impl FieldSummary {
     }
 }
 
-impl ManifestListEntry {
-    /// Get schema of the manifest list
-    pub fn schema(format_version: &FormatVersion) -> Result<AvroSchema, Error> {
-        let schema = match format_version {
-            FormatVersion::V1 => r#"
+pub fn manifest_list_schema_v1() -> &'static AvroSchema {
+    static MANIFEST_LIST_SCHEMA_V1: OnceLock<AvroSchema> = OnceLock::new();
+    MANIFEST_LIST_SCHEMA_V1.get_or_init(|| {
+        AvroSchema::parse_str(
+            r#"
         {
             "type": "record",
             "name": "manifest_file",
@@ -544,9 +546,16 @@ impl ManifestListEntry {
                 }
             ]
         }
-        "#
-            .to_owned(),
-            &FormatVersion::V2 => r#"
+        "#,
+        )
+        .unwrap()
+    })
+}
+pub fn manifest_list_schema_v2() -> &'static AvroSchema {
+    static MANIFEST_LIST_SCHEMA_V2: OnceLock<AvroSchema> = OnceLock::new();
+    MANIFEST_LIST_SCHEMA_V2.get_or_init(|| {
+        AvroSchema::parse_str(
+            r#"
         {
             "type": "record",
             "name": "manifest_file",
@@ -673,11 +682,10 @@ impl ManifestListEntry {
                 }
             ]
         }
-        "#
-            .to_owned(),
-        };
-        AvroSchema::parse_str(&schema).map_err(Into::into)
-    }
+        "#,
+        )
+        .unwrap()
+    })
 }
 
 /// Convert an avro value to a [ManifestFile] according to the provided format version
@@ -771,7 +779,7 @@ mod tests {
             key_metadata: None,
         };
 
-        let schema = ManifestListEntry::schema(&FormatVersion::V2).unwrap();
+        let schema = manifest_list_schema_v2();
 
         let mut writer = apache_avro::Writer::new(&schema, Vec::new());
 
@@ -851,7 +859,7 @@ mod tests {
             key_metadata: None,
         };
 
-        let schema = ManifestListEntry::schema(&FormatVersion::V1).unwrap();
+        let schema = manifest_list_schema_v1();
 
         let mut writer = apache_avro::Writer::new(&schema, Vec::new());