diff --git a/crates/iceberg/src/spec/manifest_list/_serde.rs b/crates/iceberg/src/spec/manifest_list/_serde.rs index 3528dec7cf..4f3fcf7329 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; diff --git a/crates/iceberg/src/spec/manifest_list/mod.rs b/crates/iceberg/src/spec/manifest_list/mod.rs index 24a4debf9e..9c599102c3 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. @@ -123,7 +119,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 +155,71 @@ mod test { assert_eq!(manifest_list, parsed_manifest_list); } + 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 +259,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 +377,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), } ]