diff --git a/crates/iceberg/src/spec/manifest/metadata.rs b/crates/iceberg/src/spec/manifest/metadata.rs index 25e4ae7e06..9ca3392f67 100644 --- a/crates/iceberg/src/spec/manifest/metadata.rs +++ b/crates/iceberg/src/spec/manifest/metadata.rs @@ -44,21 +44,6 @@ pub struct ManifestMetadata { impl ManifestMetadata { /// Parse from metadata in avro file. pub fn parse(meta: &HashMap>) -> Result { - let schema = Arc::new({ - let bs = meta.get("schema").ok_or_else(|| { - Error::new( - ErrorKind::DataInvalid, - "schema is required in manifest metadata but not found", - ) - })?; - serde_json::from_slice::(bs).map_err(|err| { - Error::new( - ErrorKind::DataInvalid, - "Fail to parse schema in manifest metadata", - ) - .with_source(err) - })? - }); let schema_id: i32 = meta .get("schema-id") .map(|bs| { @@ -72,22 +57,39 @@ impl ManifestMetadata { }) .transpose()? .unwrap_or(0); + let raw_schema = meta.get("schema").ok_or_else(|| { + Error::new( + ErrorKind::DataInvalid, + "schema is required in manifest metadata but not found", + ) + })?; + let partition_fields = { + let bs = meta.get("partition-spec").ok_or_else(|| { + Error::new( + ErrorKind::DataInvalid, + "partition-spec is required in manifest metadata but not found", + ) + })?; + serde_json::from_slice::>(bs).map_err(|err| { + Error::new( + ErrorKind::DataInvalid, + "Fail to parse partition spec in manifest metadata", + ) + .with_source(err) + })? + }; + let schema = Arc::new(parse_manifest_table_schema(raw_schema).or_else(|err| { + if partition_fields.is_empty() { + // Some writers store the manifest-entry schema under the + // `schema` metadata key. For unpartitioned manifests we can + // still plan file scans; column bounds that require the table + // schema are ignored when their field ids cannot be resolved. + Schema::builder().with_schema_id(schema_id).build() + } else { + Err(err) + } + })?); let partition_spec = { - let fields = { - let bs = meta.get("partition-spec").ok_or_else(|| { - Error::new( - ErrorKind::DataInvalid, - "partition-spec is required in manifest metadata but not found", - ) - })?; - serde_json::from_slice::>(bs).map_err(|err| { - Error::new( - ErrorKind::DataInvalid, - "Fail to parse partition spec in manifest metadata", - ) - .with_source(err) - })? - }; let spec_id = meta .get("partition-spec-id") .map(|bs| { @@ -103,7 +105,7 @@ impl ManifestMetadata { .unwrap_or(0); PartitionSpec::builder(schema.clone()) .with_spec_id(spec_id) - .add_unbound_fields(fields.into_iter().map(|f| f.into_unbound()))? + .add_unbound_fields(partition_fields.into_iter().map(|f| f.into_unbound()))? .build()? }; let format_version = if let Some(bs) = meta.get("format-version") { @@ -157,3 +159,13 @@ impl ManifestMetadata { &self.content } } + +fn parse_manifest_table_schema(bs: &[u8]) -> Result { + serde_json::from_slice::(bs).map_err(|err| { + Error::new( + ErrorKind::DataInvalid, + "Fail to parse schema in manifest metadata", + ) + .with_source(err) + }) +} diff --git a/crates/iceberg/src/spec/manifest/mod.rs b/crates/iceberg/src/spec/manifest/mod.rs index 4ba2645f51..6b43c38d8f 100644 --- a/crates/iceberg/src/spec/manifest/mod.rs +++ b/crates/iceberg/src/spec/manifest/mod.rs @@ -169,6 +169,62 @@ mod tests { use crate::io::FileIO; use crate::spec::{Literal, NestedField, PrimitiveType, Struct, Transform, Type}; + #[test] + fn test_manifest_metadata_with_manifest_entry_schema_for_unpartitioned_table() { + let mut meta = HashMap::new(); + meta.insert("schema-id".to_string(), b"0".to_vec()); + meta.insert("partition-spec".to_string(), b"[]".to_vec()); + meta.insert("partition-spec-id".to_string(), b"0".to_vec()); + meta.insert("format-version".to_string(), b"2".to_vec()); + meta.insert("content".to_string(), b"data".to_vec()); + meta.insert( + "schema".to_string(), + br#"{ + "type": "struct", + "schema-id": 0, + "fields": [ + {"id": 0, "name": "status", "required": true, "type": "int"}, + {"id": 1, "name": "snapshot_id", "required": false, "type": "long"}, + { + "id": 2, + "name": "data_file", + "required": true, + "type": { + "type": "struct", + "fields": [ + {"id": 100, "name": "file_path", "required": true, "type": "string"}, + { + "id": 125, + "name": "lower_bounds", + "required": false, + "type": { + "type": "array", + "items": { + "type": "record", + "name": "k126_k127", + "fields": [ + {"name": "key", "type": "int", "id": 126}, + {"name": "value", "type": "binary", "id": 127} + ] + } + } + } + ] + } + } + ] + }"# + .to_vec(), + ); + + let metadata = ManifestMetadata::parse(&meta).unwrap(); + assert_eq!(metadata.schema_id(), 0); + assert!(metadata.schema().as_struct().fields().is_empty()); + assert!(metadata.partition_spec().fields().is_empty()); + assert_eq!(metadata.format_version(), &FormatVersion::V2); + assert_eq!(metadata.content(), &ManifestContentType::Data); + } + #[tokio::test] async fn test_parse_manifest_v2_unpartition() { let schema = Arc::new(