fix(manifest-list): read V1 file counts written with spec field names

crates/iceberg/src/spec/manifest_list/_serde.rs

@@ -27,37 +27,28 @@ use crate::Error;
 use crate::error::{Result, invalid_data};
 use crate::spec::FieldSummary;
 
+#[cfg(test)]
 #[derive(Debug, Serialize, Deserialize, PartialEq, Eq)]
 #[serde(transparent)]
 pub(crate) struct ManifestListV3 {
     entries: Vec<ManifestFileV3>,
 }
 
+#[cfg(test)]
 #[derive(Debug, Serialize, Deserialize, PartialEq, Eq)]
 #[serde(transparent)]
 pub(crate) struct ManifestListV2 {
     entries: Vec<ManifestFileV2>,
 }
 
+#[cfg(test)]
 #[derive(Debug, Serialize, Deserialize, PartialEq, Eq)]
 #[serde(transparent)]
 pub(crate) struct ManifestListV1 {
     entries: Vec<ManifestFileV1>,
 }
 
-impl ManifestListV3 {
-    /// Converts the [ManifestListV3] into a [ManifestList].
-    pub fn try_into(self) -> Result<super::ManifestList> {
-        Ok(super::ManifestList {
-            entries: self
-                .entries
-                .into_iter()
-                .map(|v| v.try_into())
-                .collect::<Result<Vec<_>>>()?,
-        })
-    }
-}
-
+#[cfg(test)]
 impl TryFrom<super::ManifestList> for ManifestListV3 {
     type Error = Error;
 
@@ -72,19 +63,7 @@ impl TryFrom<super::ManifestList> for ManifestListV3 {
     }
 }
 
-impl ManifestListV2 {
-    /// Converts the [ManifestListV2] into a [ManifestList].
-    pub fn try_into(self) -> Result<super::ManifestList> {
-        Ok(super::ManifestList {
-            entries: self
-                .entries
-                .into_iter()
-                .map(|v| v.try_into())
-                .collect::<Result<Vec<_>>>()?,
-        })
-    }
-}
-
+#[cfg(test)]
 impl TryFrom<super::ManifestList> for ManifestListV2 {
     type Error = Error;
 
@@ -99,19 +78,7 @@ impl TryFrom<super::ManifestList> for ManifestListV2 {
     }
 }
 
-impl ManifestListV1 {
-    /// Converts the [ManifestListV1] into a [ManifestList].
-    pub fn try_into(self) -> Result<super::ManifestList> {
-        Ok(super::ManifestList {
-            entries: self
-                .entries
-                .into_iter()
-                .map(|v| v.try_into())
-                .collect::<Result<Vec<_>>>()?,
-        })
-    }
-}
-
+#[cfg(test)]
 impl TryFrom<super::ManifestList> for ManifestListV1 {
     type Error = Error;
 
@@ -132,8 +99,11 @@ pub(super) struct ManifestFileV1 {
     pub manifest_length: i64,
     pub partition_spec_id: i32,
     pub added_snapshot_id: i64,
+    #[serde(alias = "added_files_count")]
     pub added_data_files_count: Option<i32>,
+    #[serde(alias = "existing_files_count")]
     pub existing_data_files_count: Option<i32>,
+    #[serde(alias = "deleted_files_count")]
     pub deleted_data_files_count: Option<i32>,
     pub added_rows_count: Option<i64>,
     pub existing_rows_count: Option<i64>,

crates/iceberg/src/spec/manifest_list/mod.rs

@@ -23,15 +23,14 @@ mod manifest_file;
 mod reader;
 mod writer;
 
-use apache_avro::types::Value;
-use apache_avro::{Reader, from_value};
+use apache_avro::Reader;
 pub use manifest_file::*;
 pub use reader::*;
 pub use serde_bytes::ByteBuf;
 pub use writer::*;
 
-use self::_const_schema::MANIFEST_LIST_AVRO_SCHEMA_V1;
 use super::FormatVersion;
+use crate::avro::Resolved;
 use crate::error::Result;
 
 /// Placeholder for sequence number. The field with this value must be replaced with the actual sequence number before it write.
@@ -59,25 +58,22 @@ pub struct ManifestList {
 impl ManifestList {
     /// Parse manifest list from bytes.
     pub fn parse_with_version(bs: &[u8], version: FormatVersion) -> Result<ManifestList> {
-        match version {
-            FormatVersion::V1 => {
-                let reader = Reader::builder(bs)
-                    .reader_schema(&MANIFEST_LIST_AVRO_SCHEMA_V1)
-                    .build()?;
-                let values = Value::Array(reader.collect::<std::result::Result<Vec<Value>, _>>()?);
-                from_value::<_serde::ManifestListV1>(&values)?.try_into()
-            }
-            FormatVersion::V2 => {
-                let reader = Reader::new(bs)?;
-                let values = Value::Array(reader.collect::<std::result::Result<Vec<Value>, _>>()?);
-                from_value::<_serde::ManifestListV2>(&values)?.try_into()
-            }
-            FormatVersion::V3 => {
-                let reader = Reader::new(bs)?;
-                let values = Value::Array(reader.collect::<std::result::Result<Vec<Value>, _>>()?);
-                from_value::<_serde::ManifestListV3>(&values)?.try_into()
-            }
-        }
+        let reader = Reader::new(bs)?;
+        let entries = match version {
+            FormatVersion::V1 => reader
+                .into_deser_iter::<Resolved<_serde::ManifestFileV1>>()
+                .map(|entry| entry?.0.try_into())
+                .collect::<Result<Vec<_>>>()?,
+            FormatVersion::V2 => reader
+                .into_deser_iter::<Resolved<_serde::ManifestFileV2>>()
+                .map(|entry| entry?.0.try_into())
+                .collect::<Result<Vec<_>>>()?,
+            FormatVersion::V3 => reader
+                .into_deser_iter::<Resolved<_serde::ManifestFileV3>>()
+                .map(|entry| entry?.0.try_into())
+                .collect::<Result<Vec<_>>>()?,
+        };
+        Ok(ManifestList { entries })
     }
 
     /// Get the entries in the manifest list.
@@ -95,7 +91,8 @@ impl ManifestList {
 mod test {
     use std::fs;
 
-    use apache_avro::{Codec, Writer};
+    use apache_avro::types::Record;
+    use apache_avro::{Codec, Schema as AvroSchema, Writer};
     use tempfile::TempDir;
 
     use super::_const_schema::MANIFEST_LIST_AVRO_SCHEMA_V2;
@@ -123,7 +120,7 @@ mod test {
                     existing_rows_count: Some(0),
                     deleted_rows_count: Some(0),
                     partitions: Some(vec![]),
-                    key_metadata: None,
+                    key_metadata: Some(vec![1, 2, 3]),
                     first_row_id: None,
                 }
             ]
@@ -159,6 +156,130 @@ mod test {
         assert_eq!(manifest_list, parsed_manifest_list);
     }
 
+    #[test]
+    fn test_parse_manifest_list_v1_with_spec_field_names() {
+        let schema = AvroSchema::parse_str(
+            r#"{
+                "type": "record",
+                "name": "manifest_file",
+                "fields": [
+                    {"name": "manifest_path", "type": "string"},
+                    {"name": "manifest_length", "type": "long"},
+                    {"name": "partition_spec_id", "type": "int"},
+                    {"name": "added_snapshot_id", "type": "long"},
+                    {"name": "added_files_count", "type": ["null", "int"], "default": null},
+                    {"name": "existing_files_count", "type": ["null", "int"], "default": null},
+                    {"name": "deleted_files_count", "type": ["null", "int"], "default": null},
+                    {"name": "added_rows_count", "type": ["null", "long"], "default": null},
+                    {"name": "existing_rows_count", "type": ["null", "long"], "default": null},
+                    {"name": "deleted_rows_count", "type": ["null", "long"], "default": null}
+                ]
+            }"#,
+        )
+        .unwrap();
+        let mut record = Record::new(&schema).unwrap();
+        record.put("manifest_path", "s3://test/metadata/manifest.avro");
+        record.put("manifest_length", 1234i64);
+        record.put("partition_spec_id", 0i32);
+        record.put("added_snapshot_id", 42i64);
+        record.put("added_files_count", Some(3i32));
+        record.put("existing_files_count", Some(2i32));
+        record.put("deleted_files_count", Some(1i32));
+        record.put("added_rows_count", Some(30i64));
+        record.put("existing_rows_count", Some(20i64));
+        record.put("deleted_rows_count", Some(10i64));
+        let mut writer = Writer::new(&schema, Vec::new()).unwrap();
+        writer.append_value(record).unwrap();
+        let bs = writer.into_inner().unwrap();
+
+        let parsed_manifest_list =
+            ManifestList::parse_with_version(&bs, FormatVersion::V1).unwrap();
+
+        assert_eq!(parsed_manifest_list.entries(), [ManifestFile {
+            manifest_path: "s3://test/metadata/manifest.avro".to_string(),
+            manifest_length: 1234,
+            partition_spec_id: 0,
+            content: ManifestContentType::Data,
+            sequence_number: 0,
+            min_sequence_number: 0,
+            added_snapshot_id: 42,
+            added_files_count: Some(3),
+            existing_files_count: Some(2),
+            deleted_files_count: Some(1),
+            added_rows_count: Some(30),
+            existing_rows_count: Some(20),
+            deleted_rows_count: Some(10),
+            partitions: None,
+            key_metadata: None,
+            first_row_id: None,
+        }]);
+    }
+
+    const VERSIONS: [FormatVersion; 3] = [FormatVersion::V1, FormatVersion::V2, FormatVersion::V3];
+
+    fn entry(version: FormatVersion) -> ManifestFile {
+        ManifestFile {
+            manifest_path: "s3://test/metadata/manifest.avro".to_string(),
+            manifest_length: 1234,
+            partition_spec_id: 0,
+            content: ManifestContentType::Data,
+            sequence_number: if version == FormatVersion::V1 { 0 } else { 7 },
+            min_sequence_number: if version == FormatVersion::V1 { 0 } else { 3 },
+            added_snapshot_id: 42,
+            added_files_count: Some(1),
+            existing_files_count: Some(0),
+            deleted_files_count: Some(0),
+            added_rows_count: Some(9),
+            existing_rows_count: Some(0),
+            deleted_rows_count: Some(0),
+            partitions: Some(vec![FieldSummary {
+                contains_null: true,
+                contains_nan: None,
+                lower_bound: Some(vec![1, 0, 0, 0].into()),
+                upper_bound: None,
+            }]),
+            key_metadata: Some(vec![1, 2, 3]),
+            first_row_id: (version == FormatVersion::V3).then_some(100),
+        }
+    }
+
+    async fn encode(version: FormatVersion, entries: Vec<ManifestFile>) -> Vec<u8> {
+        let io = FileIO::new_with_memory();
+        let output = io.new_output("memory:///manifest-list.avro").unwrap();
+        let writer = output.writer().await.unwrap();
+        let mut writer = match version {
+            FormatVersion::V1 => ManifestListWriter::v1(writer, 42, None),
+            FormatVersion::V2 => ManifestListWriter::v2(writer, 42, None, 7),
+            FormatVersion::V3 => ManifestListWriter::v3(writer, 42, None, 7, Some(100)),
+        };
+        writer.add_manifests(entries.into_iter()).unwrap();
+        writer.close().await.unwrap();
+        io.new_input("memory:///manifest-list.avro")
+            .unwrap()
+            .read()
+            .await
+            .unwrap()
+            .to_vec()
+    }
+
+    #[tokio::test]
+    async fn test_parse_empty_manifest_list() {
+        for version in VERSIONS {
+            let bs = encode(version, vec![]).await;
+            let parsed = ManifestList::parse_with_version(&bs, version).unwrap();
+            assert!(parsed.entries().is_empty(), "{version:?}");
+        }
+    }
+
+    #[tokio::test]
+    async fn test_parse_truncated_manifest_list() {
+        for version in VERSIONS {
+            let bs = encode(version, vec![entry(version)]).await;
+            let result = ManifestList::parse_with_version(&bs[..bs.len() - 1], version);
+            assert!(result.is_err(), "{version:?}");
+        }
+    }
+
     #[tokio::test]
     async fn test_parse_manifest_list_v2() {
         let manifest_list = ManifestList {
@@ -198,9 +319,9 @@ mod test {
                     existing_rows_count: Some(0),
                     deleted_rows_count: Some(0),
                     partitions: Some(
-                        vec![FieldSummary { contains_null: false, contains_nan: Some(false), lower_bound: Some(Datum::float(1.1_f32).to_bytes().unwrap()), upper_bound: Some(Datum::float(2.1_f32).to_bytes().unwrap())}]
+                        vec![FieldSummary { contains_null: false, contains_nan: None, lower_bound: Some(Datum::float(1.1_f32).to_bytes().unwrap()), upper_bound: None}]
                     ),
-                    key_metadata: None,
+                    key_metadata: Some(vec![1, 2, 3]),
                     first_row_id: None,
                 }
             ]
@@ -316,9 +437,9 @@ mod test {
                     existing_rows_count: Some(0),
                     deleted_rows_count: Some(0),
                     partitions: Some(
-                        vec![FieldSummary { contains_null: false, contains_nan: Some(false), lower_bound: Some(Datum::float(1.1_f32).to_bytes().unwrap()), upper_bound: Some(Datum::float(2.1_f32).to_bytes().unwrap())}]
+                        vec![FieldSummary { contains_null: false, contains_nan: None, lower_bound: Some(Datum::float(1.1_f32).to_bytes().unwrap()), upper_bound: None}]
                     ),
-                    key_metadata: None,
+                    key_metadata: Some(vec![1, 2, 3]),
                     first_row_id: Some(13),
                 }
             ]