From 1f67204eb9bc850b59568e30d2cf4fb4f340443f Mon Sep 17 00:00:00 2001 From: Aarav Date: Wed, 14 Oct 2026 17:35:58 +0530 Subject: [PATCH] fix(spec): drop v3-only data file fields from the v2 manifest schema The v2 manifest Avro schema declared first_row_id (142), content_offset (144) and content_size_in_bytes (145), which the spec defines only for v3, so v2 manifests could carry v3 values. The v2 schema now matches Java's V2Metadata.fileType, and v3 manifests are written with their own schema, which keeps those fields. Reading is unchanged: v2 manifests written before this change still parse with the writer schema. Closes #3399 Co-Authored-By: Claude Opus 5.5 (1M context) --- crates/iceberg/src/spec/manifest/entry.rs | 21 ++++- crates/iceberg/src/spec/manifest/mod.rs | 28 ++++-- crates/iceberg/src/spec/manifest/writer.rs | 103 ++++++++++++++++++++- 3 files changed, 133 insertions(+), 19 deletions(-) diff --git a/crates/iceberg/src/spec/manifest/entry.rs b/crates/iceberg/src/spec/manifest/entry.rs index 555d86a25c..4a8424f625 100644 --- a/crates/iceberg/src/spec/manifest/entry.rs +++ b/crates/iceberg/src/spec/manifest/entry.rs @@ -564,15 +564,26 @@ fn data_file_fields_v2(partition_type: &StructType) -> Vec { SPLIT_OFFSETS.clone(), EQUALITY_IDS.clone(), SORT_ORDER_ID.clone(), - FIRST_ROW_ID.clone(), REFERENCE_DATA_FILE.clone(), - // Why are the following two fields here in the existing v2 schema? - // In the spec, they are not even listed as optional for v2. - CONTENT_OFFSET.clone(), - CONTENT_SIZE_IN_BYTES.clone(), ] } +pub(super) fn manifest_schema_v3(partition_type: &StructType) -> Result { + let fields = vec![ + STATUS.clone(), + SNAPSHOT_ID_V2.clone(), + SEQUENCE_NUMBER.clone(), + FILE_SEQUENCE_NUMBER.clone(), + Arc::new(NestedField::required( + 2, + "data_file", + Type::Struct(StructType::new(data_file_fields_v3(partition_type))), + )), + ]; + let schema = Schema::builder().with_fields(fields).build()?; + schema_to_avro_schema("manifest_entry", &schema) +} + pub(super) fn data_file_schema_v2(partition_type: &StructType) -> Result { let schema = Schema::builder() .with_fields(data_file_fields_v2(partition_type)) diff --git a/crates/iceberg/src/spec/manifest/mod.rs b/crates/iceberg/src/spec/manifest/mod.rs index f4269638a2..235e104c20 100644 --- a/crates/iceberg/src/spec/manifest/mod.rs +++ b/crates/iceberg/src/spec/manifest/mod.rs @@ -122,7 +122,8 @@ impl Manifest { ) }) .collect::>>()?, - // Manifest Schema & Manifest Entry did not change between V2 and V3 + // V2 and V3 entries read into the same struct; the v3-only data file + // fields are `None` when the writer schema omits them. FormatVersion::V2 | FormatVersion::V3 => reader .into_deser_iter::>() .map(|entry| { @@ -1662,7 +1663,7 @@ mod tests { .partition_type(&metadata.schema) .unwrap(); let full_schema = - serde_json::to_value(manifest_schema_v2(&partition_type).unwrap()).unwrap(); + serde_json::to_value(manifest_schema_v3(&partition_type).unwrap()).unwrap(); let mut full_entry = expected_entry(Struct::from_iter([ Some(Literal::long(5)), Some(Literal::string("a")), @@ -1754,14 +1755,21 @@ mod tests { &["data_file", "record_count"], &["data_file", "file_size_in_bytes"], ]; - assert_reads_without_each_field( - &metadata, - &full_schema, - &full_value, - &full_entry, - &optional, - &required, - ); + // Older iceberg-rust versions wrote v2 manifests with the v3 data file fields. + for format_version in [FormatVersion::V2, FormatVersion::V3] { + let metadata = ManifestMetadata { + format_version, + ..metadata.clone() + }; + assert_reads_without_each_field( + &metadata, + &full_schema, + &full_value, + &full_entry, + &optional, + &required, + ); + } } #[test] diff --git a/crates/iceberg/src/spec/manifest/writer.rs b/crates/iceberg/src/spec/manifest/writer.rs index a79312dfab..d1bfb3cbe3 100644 --- a/crates/iceberg/src/spec/manifest/writer.rs +++ b/crates/iceberg/src/spec/manifest/writer.rs @@ -30,7 +30,7 @@ use crate::encryption::EncryptedOutputFile; use crate::error::{Result, invalid_data}; use crate::io::{FileMetadata, FileWrite, OutputFile}; use crate::spec::manifest::_serde::{ManifestEntryV1, ManifestEntryV2}; -use crate::spec::manifest::{manifest_schema_v1, manifest_schema_v2}; +use crate::spec::manifest::{manifest_schema_v1, manifest_schema_v2, manifest_schema_v3}; use crate::spec::{ DataContentType, DataFile, FieldSummary, ManifestEntry, ManifestFile, ManifestMetadata, ManifestStatus, PrimitiveLiteral, SchemaRef, StructType, Type, @@ -429,8 +429,8 @@ impl ManifestWriter { let table_schema = &self.metadata.schema; let avro_schema = match self.metadata.format_version { FormatVersion::V1 => manifest_schema_v1(&partition_type)?, - // Manifest schema did not change between V2 and V3 - FormatVersion::V2 | FormatVersion::V3 => manifest_schema_v2(&partition_type)?, + FormatVersion::V2 => manifest_schema_v2(&partition_type)?, + FormatVersion::V3 => manifest_schema_v3(&partition_type)?, }; let mut avro_writer = AvroWriter::new(&avro_schema, Vec::new())?; avro_writer.add_user_metadata( @@ -472,7 +472,7 @@ impl ManifestWriter { to_value(ManifestEntryV1::try_from(entry, &partition_struct_type)?)? .resolve(&avro_schema)? } - // Manifest entry format did not change between V2 and V3 + // Resolving against the V2 schema drops the v3-only data file fields. FormatVersion::V2 | FormatVersion::V3 => { to_value(ManifestEntryV2::try_from(entry, &partition_struct_type)?)? .resolve(&avro_schema)? @@ -824,6 +824,101 @@ mod tests { ); } + fn data_file_field_ids(bs: &[u8]) -> Vec { + let reader = apache_avro::Reader::new(bs).unwrap(); + let apache_avro::Schema::Record(entry) = reader.writer_schema() else { + panic!("manifest entry schema is not a record"); + }; + let data_file = entry.fields.iter().find(|f| f.name == "data_file").unwrap(); + let apache_avro::Schema::Record(data_file) = &data_file.schema else { + panic!("data_file schema is not a record"); + }; + data_file + .fields + .iter() + .map(|f| f.custom_attributes["field-id"].as_i64().unwrap()) + .collect() + } + + #[tokio::test] + async fn test_write_manifest_data_file_fields_by_format_version() { + let schema = Arc::new( + Schema::builder() + .with_fields(vec![ + NestedField::optional(1, "id", Type::Primitive(PrimitiveType::Long)).into(), + ]) + .build() + .unwrap(), + ); + let partition_spec = PartitionSpec::builder(schema.clone()) + .with_spec_id(0) + .build() + .unwrap(); + let data_file = DataFileBuilder::default() + .content(DataContentType::Data) + .file_path("s3://bucket/table/data/a.parquet".to_string()) + .file_format(DataFileFormat::Parquet) + .partition_spec_id(0) + .record_count(1) + .file_size_in_bytes(10) + .first_row_id(Some(1000)) + .referenced_data_file(Some("s3://bucket/table/data/b.parquet".to_string())) + .content_offset(Some(4)) + .content_size_in_bytes(Some(8)) + .build() + .unwrap(); + let v2_data_file = DataFile { + first_row_id: None, + content_offset: None, + content_size_in_bytes: None, + ..data_file.clone() + }; + let v2_field_ids = vec![ + 134, 100, 101, 102, 103, 104, 108, 109, 110, 137, 125, 128, 131, 132, 135, 140, 143, + ]; + let v3_field_ids = vec![ + 134, 100, 101, 102, 103, 104, 108, 109, 110, 137, 125, 128, 131, 132, 135, 140, 142, + 143, 144, 145, + ]; + + for (format_version, expected_field_ids, expected_data_file) in [ + (FormatVersion::V2, v2_field_ids, v2_data_file), + (FormatVersion::V3, v3_field_ids, data_file.clone()), + ] { + let tmp_dir = TempDir::new().unwrap(); + let path = tmp_dir.path().join("manifest.avro"); + let output_file = FileIO::new_with_fs() + .new_output(path.to_str().unwrap()) + .unwrap(); + let builder = ManifestWriterBuilder::new( + output_file, + Some(1), + schema.clone(), + partition_spec.clone(), + ); + let mut writer = match format_version { + FormatVersion::V2 => builder.build_v2_data(), + _ => builder.build_v3_data(), + }; + writer.add_file(data_file.clone(), 1).unwrap(); + writer.write_manifest_file().await.unwrap(); + let bs = fs::read(&path).unwrap(); + + let manifest = Manifest::parse_avro(&bs).unwrap(); + + assert_eq!( + data_file_field_ids(&bs), + expected_field_ids, + "{format_version}" + ); + assert_eq!( + *manifest.entries()[0].data_file(), + expected_data_file, + "{format_version}" + ); + } + } + /// Writes a manifest partitioned by identity on two columns of `field_type`, /// reads it back, and returns the manifest file. async fn roundtrip_two_partition_fields_of_type(