diff --git a/crates/iceberg/src/spec/manifest_list/_serde.rs b/crates/iceberg/src/spec/manifest_list/_serde.rs index 3528dec7cf..f830f579de 100644 --- a/crates/iceberg/src/spec/manifest_list/_serde.rs +++ b/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, } +#[cfg(test)] #[derive(Debug, Serialize, Deserialize, PartialEq, Eq)] #[serde(transparent)] pub(crate) struct ManifestListV2 { entries: Vec, } +#[cfg(test)] #[derive(Debug, Serialize, Deserialize, PartialEq, Eq)] #[serde(transparent)] pub(crate) struct ManifestListV1 { entries: Vec, } -impl ManifestListV3 { - /// Converts the [ManifestListV3] into a [ManifestList]. - pub fn try_into(self) -> Result { - Ok(super::ManifestList { - entries: self - .entries - .into_iter() - .map(|v| v.try_into()) - .collect::>>()?, - }) - } -} - +#[cfg(test)] impl TryFrom for ManifestListV3 { type Error = Error; @@ -72,19 +63,7 @@ impl TryFrom for ManifestListV3 { } } -impl ManifestListV2 { - /// Converts the [ManifestListV2] into a [ManifestList]. - pub fn try_into(self) -> Result { - Ok(super::ManifestList { - entries: self - .entries - .into_iter() - .map(|v| v.try_into()) - .collect::>>()?, - }) - } -} - +#[cfg(test)] impl TryFrom for ManifestListV2 { type Error = Error; @@ -99,19 +78,7 @@ impl TryFrom for ManifestListV2 { } } -impl ManifestListV1 { - /// Converts the [ManifestListV1] into a [ManifestList]. - pub fn try_into(self) -> Result { - Ok(super::ManifestList { - entries: self - .entries - .into_iter() - .map(|v| v.try_into()) - .collect::>>()?, - }) - } -} - +#[cfg(test)] impl TryFrom 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, + #[serde(alias = "existing_files_count")] pub existing_data_files_count: Option, + #[serde(alias = "deleted_files_count")] pub deleted_data_files_count: Option, pub added_rows_count: Option, pub existing_rows_count: Option, diff --git a/crates/iceberg/src/spec/manifest_list/mod.rs b/crates/iceberg/src/spec/manifest_list/mod.rs index 24a4debf9e..53d8ddb4e4 100644 --- a/crates/iceberg/src/spec/manifest_list/mod.rs +++ b/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 { - match version { - FormatVersion::V1 => { - let reader = Reader::builder(bs) - .reader_schema(&MANIFEST_LIST_AVRO_SCHEMA_V1) - .build()?; - let values = Value::Array(reader.collect::, _>>()?); - from_value::<_serde::ManifestListV1>(&values)?.try_into() - } - FormatVersion::V2 => { - let reader = Reader::new(bs)?; - let values = Value::Array(reader.collect::, _>>()?); - from_value::<_serde::ManifestListV2>(&values)?.try_into() - } - FormatVersion::V3 => { - let reader = Reader::new(bs)?; - let values = Value::Array(reader.collect::, _>>()?); - from_value::<_serde::ManifestListV3>(&values)?.try_into() - } - } + let reader = Reader::new(bs)?; + let entries = match version { + FormatVersion::V1 => reader + .into_deser_iter::>() + .map(|entry| entry?.0.try_into()) + .collect::>>()?, + FormatVersion::V2 => reader + .into_deser_iter::>() + .map(|entry| entry?.0.try_into()) + .collect::>>()?, + FormatVersion::V3 => reader + .into_deser_iter::>() + .map(|entry| entry?.0.try_into()) + .collect::>>()?, + }; + 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) -> Vec { + 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), } ]