This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-3631-59946648ef56dcd5b70d46cc7ef90b25a00ed9fb in repository https://gitbox.apache.org/repos/asf/iceberg-rust.git
commit 5370b61638e205d7caf97889affe5da63aea9b16 Author: Matt Butrovich <[email protected]> AuthorDate: Fri Nov 6 18:05:35 2026 +0000 feat(arrow): read manifest lists into Arrow with arrow-avro (#3631) * feat(arrow): read manifest lists into Arrow with arrow-avro * refactor(spec): define manifest field IDs next to their fields. * fix(arrow): reject unknown manifest list format versions --- crates/iceberg/public-api.txt | 1 + crates/iceberg/src/arrow/manifest.rs | 635 ++++++++++++++++++--- crates/iceberg/src/arrow/mod.rs | 2 +- crates/iceberg/src/spec/manifest/entry.rs | 37 +- .../src/spec/manifest_list/_const_schema.rs | 43 +- crates/iceberg/src/spec/manifest_list/mod.rs | 4 + 6 files changed, 627 insertions(+), 95 deletions(-) diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt index 0ba3b6ed7..1c5af7cae 100644 --- a/crates/iceberg/public-api.txt +++ b/crates/iceberg/public-api.txt @@ -132,6 +132,7 @@ pub fn iceberg::arrow::arrow_struct_to_literal(struct_array: &arrow_array::array pub fn iceberg::arrow::arrow_type_to_type(ty: &arrow_schema::datatype::DataType) -> iceberg::Result<iceberg::spec::Type> pub fn iceberg::arrow::datum_to_arrow_type_with_ree(datum: &iceberg::spec::Datum) -> arrow_schema::datatype::DataType pub fn iceberg::arrow::read_manifest_entries(bytes: &[u8], manifest_file: &iceberg::spec::ManifestFile) -> iceberg::Result<arrow_array::record_batch::RecordBatch> +pub fn iceberg::arrow::read_manifest_list(bytes: &[u8]) -> iceberg::Result<arrow_array::record_batch::RecordBatch> pub fn iceberg::arrow::schema_to_arrow_schema(schema: &iceberg::spec::Schema) -> iceberg::Result<arrow_schema::schema::Schema> pub fn iceberg::arrow::strip_metadata_from_schema(schema: &arrow_schema::schema::Schema) -> iceberg::Result<arrow_schema::schema::Schema> pub fn iceberg::arrow::type_to_arrow_type(ty: &iceberg::spec::Type) -> iceberg::Result<arrow_schema::datatype::DataType> diff --git a/crates/iceberg/src/arrow/manifest.rs b/crates/iceberg/src/arrow/manifest.rs index 57bb2402a..08bd9ed4c 100644 --- a/crates/iceberg/src/arrow/manifest.rs +++ b/crates/iceberg/src/arrow/manifest.rs @@ -15,7 +15,7 @@ // specific language governing permissions and limitations // under the License. -//! Reading manifest entries into Arrow. +//! Reading manifests and manifest lists into Arrow. use std::collections::HashMap; use std::sync::Arc; @@ -28,7 +28,7 @@ use arrow_array::{ Array, ArrayRef, BooleanArray, Int32Array, Int64Array, ListArray, MapArray, RecordBatch, StructArray, new_null_array, }; -use arrow_avro::reader::ReaderBuilder; +use arrow_avro::reader::{Reader, ReaderBuilder}; use arrow_avro::schema::AvroSchema; use arrow_cast::{CastOptions, cast_with_options}; use arrow_ord::cmp::eq; @@ -40,22 +40,14 @@ use serde_json::Value as JsonValue; use crate::arrow::schema_to_arrow_schema; use crate::error::invalid_data; use crate::spec::{ - FormatVersion, ManifestContentType, ManifestFile, ManifestMetadata, ManifestStatus, Schema, - manifest_entry_fields, + DATA_FILE_CONTENT_FIELD_ID, DATA_FILE_FIELD_ID, DATA_FILE_PARTITION_FIELD_ID, + FILE_SEQUENCE_NUMBER_FIELD_ID, FormatVersion, MANIFEST_CONTENT_FIELD_ID, + MANIFEST_SEQUENCE_NUMBER_FIELD_ID, MIN_SEQUENCE_NUMBER_FIELD_ID, ManifestContentType, + ManifestFile, ManifestMetadata, ManifestStatus, SEQUENCE_NUMBER_FIELD_ID, Schema, + manifest_entry_fields, manifest_file_fields, }; use crate::{Error, Result}; -/// Field ID of `data_file`. -const DATA_FILE_FIELD_ID: i64 = 2; -/// Field ID of `data_file.partition`. -const PARTITION_FIELD_ID: i64 = 102; -/// Field ID of `data_file.content`. -const CONTENT_FIELD_ID: i32 = 134; -/// Field ID of `sequence_number`. -const SEQUENCE_NUMBER_FIELD_ID: i32 = 3; -/// Field ID of `file_sequence_number`. -const FILE_SEQUENCE_NUMBER_FIELD_ID: i32 = 4; - /// Reads the entries of a manifest file into a [`RecordBatch`] with one row /// per entry. /// @@ -142,30 +134,128 @@ pub fn read_manifest_entries(bytes: &[u8], manifest_file: &ManifestFile) -> Resu .map_err(invalid_manifest)?; } - let Some(batch) = reader.next() else { + let Some(batch) = only_batch(reader, invalid_manifest)? else { return Ok(RecordBatch::new_empty(schema)); }; - let batch = batch.map_err(invalid_manifest)?; - // Taking only the first batch relies on arrow-avro filling it to the batch - // size, so fail instead of dropping entries if that changes. - if reader.next().is_some() { - return Err(invalid_data!( - "arrow-avro split the manifest into more than one batch" - )); - } + // v1 manifests have no content or sequence number columns, which v2+ + // readers read as 0. A v2+ manifest may lack file_sequence_number (Java + // wrote none before 1.1.0), which reads as null and is inherited. + let zero_defaults: &[i32] = match metadata.format_version() { + FormatVersion::V1 => &[ + DATA_FILE_CONTENT_FIELD_ID, + SEQUENCE_NUMBER_FIELD_ID, + FILE_SEQUENCE_NUMBER_FIELD_ID, + ], + _ => &[DATA_FILE_CONTENT_FIELD_ID], + }; let entries = project_struct( &StructArray::from(batch), &avro_schema, schema.fields(), - *metadata.format_version(), + zero_defaults, )?; Ok(RecordBatch::from(inherit(entries, manifest_file)?)) } +/// Reads the entries of a manifest list into a [`RecordBatch`] with one row +/// per manifest. +/// +/// The schema is the spec's `manifest_file` struct for the latest format +/// version, with field IDs in the Arrow field metadata, so manifest lists of +/// every format version read into the same schema. Fields are matched to the +/// manifest list's Avro schema by field ID. The format version comes from the +/// `format-version` key that writers put in the file's metadata, and a +/// manifest list without one is v1. A v1 manifest list reads with the spec's +/// v2+ defaults. Partition summary bounds keep the binary single-value +/// serialization they were written with. +/// +/// # Example +/// +/// ``` +/// use iceberg::arrow::read_manifest_list; +/// +/// let bytes = std::fs::read(concat!( +/// env!("CARGO_MANIFEST_DIR"), +/// "/testdata/manifests_lists/manifest-list-v2-2.avro" +/// ))?; +/// let batch = read_manifest_list(&bytes)?; +/// assert_eq!(batch.num_rows(), 1); +/// # Ok::<(), Box<dyn std::error::Error>>(()) +/// ``` +pub fn read_manifest_list(bytes: &[u8]) -> Result<RecordBatch> { + // usize::MAX decodes the whole manifest list into one batch, which + // `only_batch` checks. + let reader = ReaderBuilder::new() + .with_batch_size(usize::MAX) + .build(bytes) + .map_err(invalid_manifest_list)?; + let header = reader.avro_header(); + let avro_schema: JsonValue = serde_json::from_slice( + header + .get("avro.schema") + .ok_or_else(|| invalid_data!("manifest list has no avro.schema"))?, + )?; + // The spec doesn't define this key, but Java, PyIceberg, and iceberg-rust + // all write it, and rejecting an unknown version keeps a newer manifest + // list from reading as an older one. + let format_version = match header.get("format-version") { + Some(bytes) => serde_json::from_slice::<FormatVersion>(bytes).map_err(|err| { + invalid_data!("failed to parse format version in manifest list metadata") + .with_source(err) + })?, + None => FormatVersion::V1, + }; + let schema = Arc::new(schema_to_arrow_schema( + &Schema::builder() + .with_fields(manifest_file_fields()) + .build()?, + )?); + let Some(batch) = only_batch(reader, invalid_manifest_list)? else { + return Ok(RecordBatch::new_empty(schema)); + }; + // v1 manifest lists have no content or sequence number columns, which v2+ + // readers read as 0. + let zero_defaults: &[i32] = match format_version { + FormatVersion::V1 => &[ + MANIFEST_CONTENT_FIELD_ID, + MANIFEST_SEQUENCE_NUMBER_FIELD_ID, + MIN_SEQUENCE_NUMBER_FIELD_ID, + ], + _ => &[], + }; + let manifests = project_struct( + &StructArray::from(batch), + &avro_schema, + schema.fields(), + zero_defaults, + )?; + Ok(RecordBatch::from(manifests)) +} + +/// Returns the only batch that `reader` decodes, or `None` for a file without +/// records. +fn only_batch( + mut reader: Reader<&[u8]>, + invalid: fn(ArrowError) -> Error, +) -> Result<Option<RecordBatch>> { + let Some(batch) = reader.next() else { + return Ok(None); + }; + let batch = batch.map_err(invalid)?; + // Taking only the first batch relies on arrow-avro filling it to the batch + // size, so fail instead of dropping records if that changes. + if reader.next().is_some() { + return Err(invalid_data!( + "arrow-avro split the file into more than one batch" + )); + } + Ok(Some(batch)) +} + /// Removes `data_file.partition` from a manifest's Avro schema if it has no /// fields, and returns whether it did. fn remove_empty_partition(avro_schema: &mut JsonValue) -> bool { - let has_field_id = |field: &JsonValue, id: i64| field["field-id"].as_i64() == Some(id); + let has_field_id = |field: &JsonValue, id: i32| field["field-id"].as_i64() == Some(id.into()); let Some(data_file_fields) = avro_schema["fields"] .as_array_mut() .and_then(|fields| { @@ -179,7 +269,7 @@ fn remove_empty_partition(avro_schema: &mut JsonValue) -> bool { }; let len = data_file_fields.len(); data_file_fields.retain(|field| { - !(has_field_id(field, PARTITION_FIELD_ID) + !(has_field_id(field, DATA_FILE_PARTITION_FIELD_ID) && field["type"]["fields"] .as_array() .is_some_and(Vec::is_empty)) @@ -191,8 +281,17 @@ fn invalid_manifest(err: ArrowError) -> Error { invalid_data!("failed to read manifest").with_source(err) } +fn invalid_manifest_list(err: ArrowError) -> Error { + invalid_data!("failed to read manifest list").with_source(err) +} + +fn invalid_column(err: ArrowError) -> Error { + invalid_data!("failed to build Arrow column from Avro").with_source(err) +} + /// Returns the field ID of a field in the schema built from -/// `manifest_entry_fields`, which gives every field one. +/// `manifest_entry_fields` or `manifest_file_fields`, which give every field +/// one. fn field_id(field: &Field) -> i32 { field.metadata()[PARQUET_FIELD_ID_META_KEY].parse().unwrap() } @@ -210,16 +309,17 @@ fn non_null_type(avro_type: &JsonValue) -> &JsonValue { } /// Builds each of `fields` from the column of `array` that the Avro record -/// `record` gives the same field ID. +/// `record` gives the same field ID. A field that `record` lacks reads as 0 if +/// its ID is in `zero_defaults`, and as null otherwise. fn project_struct( array: &StructArray, record: &JsonValue, fields: &Fields, - format_version: FormatVersion, + zero_defaults: &[i32], ) -> Result<StructArray> { let avro_fields = record["fields"] .as_array() - .ok_or_else(|| invalid_data!("manifest Avro type {record} is not a record"))?; + .ok_or_else(|| invalid_data!("Avro type {record} is not a record"))?; let mut columns = Vec::with_capacity(fields.len()); for field in fields { let id = field_id(field); @@ -232,34 +332,29 @@ fn project_struct( array.column(index), non_null_type(&avro_fields[index]["type"]), field, - format_version, + zero_defaults, )?, - None => missing_field(field, array.len(), format_version)?, + None => missing_field(field, array.len(), zero_defaults)?, }, ); } StructArray::try_new_with_length(fields.clone(), columns, array.nulls().cloned(), array.len()) - .map_err(invalid_manifest) + .map_err(invalid_column) } -/// Builds the column for a field that the manifest doesn't have. -fn missing_field(field: &Field, len: usize, format_version: FormatVersion) -> Result<ArrayRef> { - // v1 manifests have no content or sequence number columns, which v2+ - // readers read as 0. A v2+ manifest may lack file_sequence_number (Java - // wrote none before 1.1.0), which reads as null and is inherited. - match field_id(field) { - CONTENT_FIELD_ID => Ok(Arc::new(Int32Array::from_value(0, len))), - SEQUENCE_NUMBER_FIELD_ID | FILE_SEQUENCE_NUMBER_FIELD_ID - if format_version == FormatVersion::V1 => - { - Ok(Arc::new(Int64Array::from_value(0, len))) - } +/// Builds the column for a field that the Avro file doesn't have. +fn missing_field(field: &Field, len: usize, zero_defaults: &[i32]) -> Result<ArrayRef> { + let id = field_id(field); + let zero = zero_defaults.contains(&id); + match field.data_type() { + DataType::Int32 if zero => Ok(Arc::new(Int32Array::from_value(0, len))), + DataType::Int64 if zero => Ok(Arc::new(Int64Array::from_value(0, len))), _ if field.is_nullable() => Ok(new_null_array(field.data_type(), len)), - _ if matches!(field.data_type(), DataType::Struct(fields) if fields.is_empty()) => { + DataType::Struct(fields) if fields.is_empty() => { Ok(Arc::new(StructArray::new_empty_fields(len, None))) } - id => Err(invalid_data!( - "manifest has no required field {} with field ID {id}", + _ => Err(invalid_data!( + "Avro file has no required field {} with field ID {id}", field.name() )), } @@ -271,11 +366,11 @@ fn project( array: &ArrayRef, avro_type: &JsonValue, field: &Field, - format_version: FormatVersion, + zero_defaults: &[i32], ) -> Result<ArrayRef> { let type_mismatch = || { invalid_data!( - "manifest field {} has Avro type {avro_type}, which doesn't match {}", + "field {} has Avro type {avro_type}, which doesn't match {}", field.name(), field.data_type() ) @@ -285,7 +380,7 @@ fn project( array.as_struct_opt().ok_or_else(type_mismatch)?, avro_type, fields, - format_version, + zero_defaults, )?), DataType::List(element) => { let list = array.as_list_opt::<i32>().ok_or_else(type_mismatch)?; @@ -293,7 +388,7 @@ fn project( list.values(), non_null_type(&avro_type["items"]), element, - format_version, + zero_defaults, )?; Arc::new( ListArray::try_new( @@ -302,7 +397,7 @@ fn project( values, list.nulls().cloned(), ) - .map_err(invalid_manifest)?, + .map_err(invalid_column)?, ) } // Avro stores maps with int keys as arrays of key-value records. @@ -315,7 +410,7 @@ fn project( list.values().as_struct_opt().ok_or_else(type_mismatch)?, &avro_type["items"], key_value, - format_version, + zero_defaults, )?; Arc::new( MapArray::try_new( @@ -325,7 +420,7 @@ fn project( list.nulls().cloned(), false, ) - .map_err(invalid_manifest)?, + .map_err(invalid_column)?, ) } // arrow-avro gives every timestamp a UTC time zone, so only the type @@ -351,7 +446,7 @@ fn project( safe: false, ..Default::default() }) - .map_err(invalid_manifest)?, + .map_err(invalid_column)?, }) } @@ -477,16 +572,18 @@ mod tests { FixedSizeBinaryArray, Float32Array, Float64Array, Int32Array, Int64Array, StringArray, Time64MicrosecondArray, TimestampMicrosecondArray, TimestampNanosecondArray, }; + use arrow_buffer::{NullBuffer, OffsetBuffer}; use super::*; use crate::ErrorKind; use crate::arrow::arrow_schema_to_schema; use crate::io::FileIO; use crate::spec::{ - DataContentType, DataFile, DataFileBuilder, DataFileFormat, Datum, ListType, Literal, - Manifest, ManifestEntry, ManifestStatus, ManifestWriter, ManifestWriterBuilder, MapType, - NestedField, PartitionSpec, PrimitiveLiteral, PrimitiveType, Schema, Struct, StructType, - Transform, Type, + DataContentType, DataFile, DataFileBuilder, DataFileFormat, Datum, FieldSummary, ListType, + Literal, Manifest, ManifestEntry, ManifestList, ManifestListWriter, ManifestStatus, + ManifestWriter, ManifestWriterBuilder, MapType, NestedField, PartitionSpec, + PrimitiveLiteral, PrimitiveType, Schema, Struct, StructType, Transform, Type, + UNASSIGNED_SEQUENCE_NUMBER, }; fn table_schema() -> Schema { @@ -629,6 +726,26 @@ mod tests { )) } + /// Rewrites an Avro file with the user metadata `key` set to `value`, or + /// without it if `value` is `None`. + fn set_avro_metadata(bytes: &[u8], key: &str, value: Option<&str>) -> Vec<u8> { + let reader = apache_avro::Reader::new(bytes).unwrap(); + let schema = reader.writer_schema().clone(); + let mut metadata = reader.user_metadata().clone(); + match value { + Some(value) => metadata.insert(key.to_string(), value.as_bytes().to_vec()), + None => metadata.remove(key), + }; + let mut writer = apache_avro::Writer::new(&schema, Vec::new()).unwrap(); + for (key, value) in metadata { + writer.add_user_metadata(key, value).unwrap(); + } + for record in reader { + writer.append_value(record.unwrap()).unwrap(); + } + writer.into_inner().unwrap() + } + /// Rewrites a manifest with `codec`, `edit_schema` applied to its Avro /// schema, and `edit_record` applied to each record. fn rewrite_manifest( @@ -698,6 +815,29 @@ mod tests { } } + /// Renames record fields in an Avro file's schema and records, keeping + /// their field IDs. + fn rename_fields(bytes: &[u8], renames: &[(&str, &str)]) -> Vec<u8> { + rewrite_manifest( + bytes, + Codec::Null, + |schema| { + for_each_avro_field(schema, &|field| { + for (from, to) in renames { + if field["name"] == *from { + field["name"] = (*to).into(); + } + } + }) + }, + |record| { + for (from, to) in renames { + rename_avro_field(record, from, to); + } + }, + ) + } + #[tokio::test] async fn test_schema_is_spec_manifest_entry() { let spec = partition_spec(&[("i", Transform::Identity), ("s", Transform::Truncate(2))]); @@ -1304,25 +1444,10 @@ mod tests { |writer| writer.add_file(file, 1).unwrap(), ) .await; - let renames = [("file_path", "path"), ("i_identity", "renamed_partition")]; - let bytes = rewrite_manifest( - &bytes, - Codec::Null, - |schema| { - for_each_avro_field(schema, &|field| { - for (from, to) in renames { - if field["name"] == from { - field["name"] = to.into(); - } - } - }) - }, - |record| { - for (from, to) in renames { - rename_avro_field(record, from, to); - } - }, - ); + let bytes = rename_fields(&bytes, &[ + ("file_path", "path"), + ("i_identity", "renamed_partition"), + ]); let batch = read_manifest_entries(&bytes, &manifest_file).unwrap(); assert_eq!( @@ -1419,4 +1544,354 @@ mod tests { ); } } + + /// A manifest list entry that a manifest list writer for snapshot 1 + /// assigns sequence numbers to. + fn listed_manifest(path: &str, content: ManifestContentType) -> ManifestFile { + ManifestFile { + manifest_path: path.to_string(), + manifest_length: 100, + partition_spec_id: 0, + content, + sequence_number: UNASSIGNED_SEQUENCE_NUMBER, + min_sequence_number: UNASSIGNED_SEQUENCE_NUMBER, + added_snapshot_id: 1, + added_files_count: Some(1), + existing_files_count: Some(2), + deleted_files_count: Some(3), + added_rows_count: Some(10), + existing_rows_count: Some(20), + deleted_rows_count: Some(30), + partitions: None, + key_metadata: None, + first_row_id: None, + } + } + + /// Writes a manifest list for snapshot 1 with sequence number 5 and, in + /// v3, first row ID 100. + async fn write_manifest_list(version: FormatVersion, manifests: Vec<ManifestFile>) -> Vec<u8> { + let io = FileIO::new_with_memory(); + let path = "memory:///manifest-list.avro"; + let output = io.new_output(path).unwrap().writer().await.unwrap(); + let mut writer = match version { + FormatVersion::V1 => ManifestListWriter::v1(output, 1, None), + FormatVersion::V2 => ManifestListWriter::v2(output, 1, None, 5), + FormatVersion::V3 => ManifestListWriter::v3(output, 1, None, 5, Some(100)), + }; + writer.add_manifests(manifests.into_iter()).unwrap(); + writer.close().await.unwrap(); + io.new_input(path).unwrap().read().await.unwrap().to_vec() + } + + #[tokio::test] + async fn test_schema_is_spec_manifest_file() { + let bytes = write_manifest_list(FormatVersion::V3, vec![]).await; + let batch = read_manifest_list(&bytes).unwrap(); + assert_eq!(batch.num_rows(), 0); + + let long = || Type::Primitive(PrimitiveType::Long); + let int = || Type::Primitive(PrimitiveType::Int); + let boolean = || Type::Primitive(PrimitiveType::Boolean); + let binary = || Type::Primitive(PrimitiveType::Binary); + let field_summary = StructType::new(vec![ + NestedField::required(509, "contains_null", boolean()).into(), + NestedField::optional(518, "contains_nan", boolean()).into(), + NestedField::optional(510, "lower_bound", binary()).into(), + NestedField::optional(511, "upper_bound", binary()).into(), + ]); + let expected = Schema::builder() + .with_fields(vec![ + NestedField::required(500, "manifest_path", Type::Primitive(PrimitiveType::String)) + .into(), + NestedField::required(501, "manifest_length", long()).into(), + NestedField::required(502, "partition_spec_id", int()).into(), + NestedField::required(517, "content", int()).into(), + NestedField::required(515, "sequence_number", long()).into(), + NestedField::required(516, "min_sequence_number", long()).into(), + NestedField::required(503, "added_snapshot_id", long()).into(), + NestedField::optional(504, "added_files_count", int()).into(), + NestedField::optional(505, "existing_files_count", int()).into(), + NestedField::optional(506, "deleted_files_count", int()).into(), + NestedField::optional(512, "added_rows_count", long()).into(), + NestedField::optional(513, "existing_rows_count", long()).into(), + NestedField::optional(514, "deleted_rows_count", long()).into(), + NestedField::optional( + 507, + "partitions", + Type::List(ListType::new( + NestedField::list_element(508, Type::Struct(field_summary), true).into(), + )), + ) + .into(), + NestedField::optional(519, "key_metadata", binary()).into(), + NestedField::optional(520, "first_row_id", long()).into(), + ]) + .build() + .unwrap(); + assert_eq!( + arrow_schema_to_schema(&batch.schema()).unwrap().as_struct(), + expected.as_struct() + ); + } + + #[tokio::test] + async fn test_manifest_list_columns() { + let data = ManifestFile { + partitions: Some(vec![FieldSummary { + contains_null: true, + contains_nan: Some(false), + lower_bound: Some(Datum::int(1).to_bytes().unwrap()), + upper_bound: Some(Datum::int(9).to_bytes().unwrap()), + }]), + key_metadata: Some(vec![1, 2]), + ..listed_manifest("s3://b/metadata/data.avro", ManifestContentType::Data) + }; + let existing_deletes = ManifestFile { + partition_spec_id: 1, + sequence_number: 3, + min_sequence_number: 2, + added_snapshot_id: 0, + added_files_count: Some(0), + ..listed_manifest("s3://b/metadata/deletes.avro", ManifestContentType::Deletes) + }; + let bytes = write_manifest_list(FormatVersion::V3, vec![data, existing_deletes]).await; + let batch = read_manifest_list(&bytes).unwrap(); + + let schema = batch.schema(); + let DataType::List(summary) = schema.field_with_name("partitions").unwrap().data_type() + else { + panic!("partitions is not a list"); + }; + let DataType::Struct(summary_fields) = summary.data_type() else { + panic!("partitions element is not a struct"); + }; + let summaries = StructArray::new( + summary_fields.clone(), + vec![ + Arc::new(BooleanArray::from(vec![true])), + Arc::new(BooleanArray::from(vec![Some(false)])), + Arc::new(BinaryArray::from(vec![&1i32.to_le_bytes()[..]])), + Arc::new(BinaryArray::from(vec![&9i32.to_le_bytes()[..]])), + ], + None, + ); + let columns: Vec<ArrayRef> = vec![ + Arc::new(StringArray::from(vec![ + "s3://b/metadata/data.avro", + "s3://b/metadata/deletes.avro", + ])), + Arc::new(Int64Array::from(vec![100, 100])), + Arc::new(Int32Array::from(vec![0, 1])), + Arc::new(Int32Array::from(vec![0, 1])), + Arc::new(Int64Array::from(vec![5, 3])), + Arc::new(Int64Array::from(vec![5, 2])), + Arc::new(Int64Array::from(vec![1, 0])), + Arc::new(Int32Array::from(vec![1, 0])), + Arc::new(Int32Array::from(vec![2, 2])), + Arc::new(Int32Array::from(vec![3, 3])), + Arc::new(Int64Array::from(vec![10, 10])), + Arc::new(Int64Array::from(vec![20, 20])), + Arc::new(Int64Array::from(vec![30, 30])), + Arc::new(ListArray::new( + summary.clone(), + OffsetBuffer::from_lengths([1, 0]), + Arc::new(summaries), + Some(NullBuffer::from(vec![true, false])), + )), + Arc::new(BinaryArray::from(vec![Some(&[1u8, 2][..]), None])), + // The writer assigns the data manifest the snapshot's first row ID. + Arc::new(Int64Array::from(vec![Some(100), None])), + ]; + assert_eq!(batch, RecordBatch::try_new(schema, columns).unwrap()); + } + + #[tokio::test] + async fn test_v1_manifest_list_reads_with_v2_defaults() { + let unknown_counts = ManifestFile { + added_files_count: None, + existing_files_count: None, + deleted_files_count: None, + added_rows_count: None, + existing_rows_count: None, + deleted_rows_count: None, + ..listed_manifest("s3://b/metadata/1.avro", ManifestContentType::Data) + }; + let bytes = write_manifest_list(FormatVersion::V1, vec![ + listed_manifest("s3://b/metadata/0.avro", ManifestContentType::Data), + unknown_counts, + ]) + .await; + // A manifest list without a format version reads as v1. + let without_format_version = set_avro_metadata(&bytes, "format-version", None); + for bytes in [bytes, without_format_version] { + let batch = read_manifest_list(&bytes).unwrap(); + + assert_eq!( + column(&batch, "content").as_ref(), + &Int32Array::from(vec![0, 0]) as &dyn Array + ); + for name in ["sequence_number", "min_sequence_number"] { + assert_eq!( + column(&batch, name).as_ref(), + &Int64Array::from(vec![0, 0]) as &dyn Array, + "{name}" + ); + } + assert_eq!( + column(&batch, "added_files_count").as_ref(), + &Int32Array::from(vec![Some(1), None]) as &dyn Array + ); + assert_eq!( + column(&batch, "deleted_rows_count").as_ref(), + &Int64Array::from(vec![Some(30), None]) as &dyn Array + ); + assert_eq!(column(&batch, "first_row_id").null_count(), 2); + } + } + + #[tokio::test] + async fn test_manifest_list_fields_are_matched_by_field_id() { + let manifest = ManifestFile { + partitions: Some(vec![FieldSummary { + contains_null: true, + contains_nan: None, + lower_bound: None, + upper_bound: None, + }]), + ..listed_manifest("s3://b/metadata/0.avro", ManifestContentType::Data) + }; + let bytes = write_manifest_list(FormatVersion::V2, vec![manifest]).await; + let bytes = rename_fields(&bytes, &[ + ("manifest_path", "path"), + ("added_files_count", "added_data_files_count"), + ("contains_null", "has_null"), + ]); + let batch = read_manifest_list(&bytes).unwrap(); + + assert_eq!( + column(&batch, "manifest_path").as_ref(), + &StringArray::from(vec!["s3://b/metadata/0.avro"]) as &dyn Array + ); + assert_eq!( + column(&batch, "added_files_count").as_ref(), + &Int32Array::from(vec![1]) as &dyn Array + ); + let partitions = column(&batch, "partitions"); + assert_eq!( + partitions + .as_list::<i32>() + .value(0) + .as_struct() + .column(0) + .as_ref(), + &BooleanArray::from(vec![true]) as &dyn Array + ); + } + + #[tokio::test] + async fn test_invalid_manifest_list_is_an_error() { + let err = read_manifest_list(&[]).unwrap_err(); + assert_eq!(err.kind(), ErrorKind::DataInvalid); + + let bytes = write_manifest_list(FormatVersion::V2, vec![listed_manifest( + "s3://b/metadata/0.avro", + ManifestContentType::Data, + )]) + .await; + // A manifest_path under another field ID leaves the required + // manifest_path missing. + let moved_manifest_path = rewrite_manifest( + &bytes, + Codec::Null, + |schema| { + for_each_avro_field(schema, &|field| { + if field["name"] == "manifest_path" { + field["field-id"] = 999.into(); + } + }) + }, + |_| {}, + ); + // Only v1 manifest lists read a missing sequence number as 0. + let without_sequence_number = rewrite_manifest( + &bytes, + Codec::Null, + |schema| { + schema["fields"] + .as_array_mut() + .unwrap() + .retain(|field| field["name"] != "sequence_number") + }, + |record| { + let AvroValue::Record(fields) = record else { + panic!("manifest list entry is not a record"); + }; + fields.retain(|(name, _)| name != "sequence_number"); + }, + ); + let unknown_format_version = set_avro_metadata(&bytes, "format-version", Some("4")); + for bytes in [ + moved_manifest_path, + without_sequence_number, + unknown_format_version, + ] { + let err = read_manifest_list(&bytes).unwrap_err(); + assert_eq!(err.kind(), ErrorKind::DataInvalid); + } + } + + #[test] + fn test_manifest_lists_written_by_java() { + // Written by Java 1.4.0, with the v1 file count names, and 1.5.0. + for name in ["manifest-list-v2-1.avro", "manifest-list-v2-2.avro"] { + let bytes = std::fs::read(format!( + "{}/testdata/manifests_lists/{name}", + env!("CARGO_MANIFEST_DIR") + )) + .unwrap(); + let manifest_list = + ManifestList::parse_with_version(&bytes, FormatVersion::V2).unwrap(); + let batch = read_manifest_list(&bytes).unwrap(); + + assert_eq!(batch.num_rows(), manifest_list.entries().len(), "{name}"); + let paths = column(&batch, "manifest_path"); + let sequence_numbers = column(&batch, "sequence_number"); + let added_files_counts = column(&batch, "added_files_count"); + let partitions = column(&batch, "partitions"); + for (row, manifest) in manifest_list.entries().iter().enumerate() { + assert_eq!( + paths.as_string::<i32>().value(row), + manifest.manifest_path, + "{name}" + ); + assert_eq!( + sequence_numbers.as_primitive::<Int64Type>().value(row), + manifest.sequence_number, + "{name}" + ); + assert_eq!( + added_files_counts.as_primitive::<Int32Type>().value(row), + manifest.added_files_count.unwrap() as i32, + "{name}" + ); + let summaries = partitions.as_list::<i32>().value(row); + let lower_bounds: Vec<_> = summaries + .as_struct() + .column_by_name("lower_bound") + .unwrap() + .as_binary::<i32>() + .iter() + .map(|bound| bound.map(<[u8]>::to_vec)) + .collect(); + let expected: Vec<_> = manifest + .partitions + .iter() + .flatten() + .map(|summary| summary.lower_bound.as_ref().map(|bound| bound.to_vec())) + .collect(); + assert_eq!(lower_bounds, expected, "{name}"); + } + } + } } diff --git a/crates/iceberg/src/arrow/mod.rs b/crates/iceberg/src/arrow/mod.rs index c74b20d7c..d8343ef1e 100644 --- a/crates/iceberg/src/arrow/mod.rs +++ b/crates/iceberg/src/arrow/mod.rs @@ -29,7 +29,7 @@ pub(crate) mod delete_filter; mod int96; mod manifest; -pub use manifest::read_manifest_entries; +pub use manifest::{read_manifest_entries, read_manifest_list}; mod reader; /// RecordBatch projection utilities pub mod record_batch_projector; diff --git a/crates/iceberg/src/spec/manifest/entry.rs b/crates/iceberg/src/spec/manifest/entry.rs index bbd2965d5..c8ed0ea1b 100644 --- a/crates/iceberg/src/spec/manifest/entry.rs +++ b/crates/iceberg/src/spec/manifest/entry.rs @@ -177,6 +177,17 @@ impl TryFrom<i32> for ManifestStatus { use super::DataFileFormat; +/// Field ID of `manifest_entry.data_file`. +pub(crate) const DATA_FILE_FIELD_ID: i32 = 2; +/// Field ID of `manifest_entry.sequence_number`. +pub(crate) const SEQUENCE_NUMBER_FIELD_ID: i32 = 3; +/// Field ID of `manifest_entry.file_sequence_number`. +pub(crate) const FILE_SEQUENCE_NUMBER_FIELD_ID: i32 = 4; +/// Field ID of `data_file.partition`. +pub(crate) const DATA_FILE_PARTITION_FIELD_ID: i32 = 102; +/// Field ID of `data_file.content`. +pub(crate) const DATA_FILE_CONTENT_FIELD_ID: i32 = 134; + static STATUS: Lazy<NestedFieldRef> = { Lazy::new(|| { Arc::new(NestedField::required( @@ -210,7 +221,7 @@ static SNAPSHOT_ID_V2: Lazy<NestedFieldRef> = { static SEQUENCE_NUMBER: Lazy<NestedFieldRef> = { Lazy::new(|| { Arc::new(NestedField::optional( - 3, + SEQUENCE_NUMBER_FIELD_ID, "sequence_number", Type::Primitive(PrimitiveType::Long), )) @@ -220,7 +231,7 @@ static SEQUENCE_NUMBER: Lazy<NestedFieldRef> = { static FILE_SEQUENCE_NUMBER: Lazy<NestedFieldRef> = { Lazy::new(|| { Arc::new(NestedField::optional( - 4, + FILE_SEQUENCE_NUMBER_FIELD_ID, "file_sequence_number", Type::Primitive(PrimitiveType::Long), )) @@ -230,9 +241,13 @@ static FILE_SEQUENCE_NUMBER: Lazy<NestedFieldRef> = { static CONTENT: Lazy<NestedFieldRef> = { Lazy::new(|| { Arc::new( - NestedField::required(134, "content", Type::Primitive(PrimitiveType::Int)) - // 0 refers to DataContentType::DATA - .with_initial_default(Literal::Primitive(PrimitiveLiteral::Int(0))), + NestedField::required( + DATA_FILE_CONTENT_FIELD_ID, + "content", + Type::Primitive(PrimitiveType::Int), + ) + // 0 refers to DataContentType::DATA + .with_initial_default(Literal::Primitive(PrimitiveLiteral::Int(0))), ) }) }; @@ -512,7 +527,7 @@ fn data_file_fields_v3(partition_type: &StructType) -> Vec<NestedFieldRef> { FILE_PATH.clone(), FILE_FORMAT.clone(), Arc::new(NestedField::required( - 102, + DATA_FILE_PARTITION_FIELD_ID, "partition", Type::Struct(partition_type.clone()), )), @@ -546,7 +561,7 @@ pub(crate) fn manifest_entry_fields(partition_type: &StructType) -> Vec<NestedFi SEQUENCE_NUMBER.clone(), FILE_SEQUENCE_NUMBER.clone(), Arc::new(NestedField::required( - 2, + DATA_FILE_FIELD_ID, "data_file", Type::Struct(StructType::new(data_file_fields_v3(partition_type))), )), @@ -566,7 +581,7 @@ fn data_file_fields_v2(partition_type: &StructType) -> Vec<NestedFieldRef> { FILE_PATH.clone(), FILE_FORMAT.clone(), Arc::new(NestedField::required( - 102, + DATA_FILE_PARTITION_FIELD_ID, "partition", Type::Struct(partition_type.clone()), )), @@ -605,7 +620,7 @@ pub(super) fn manifest_schema_v2(partition_type: &StructType) -> Result<AvroSche SEQUENCE_NUMBER.clone(), FILE_SEQUENCE_NUMBER.clone(), Arc::new(NestedField::required( - 2, + DATA_FILE_FIELD_ID, "data_file", Type::Struct(StructType::new(data_file_fields_v2(partition_type))), )), @@ -619,7 +634,7 @@ fn data_file_fields_v1(partition_type: &StructType) -> Vec<NestedFieldRef> { FILE_PATH.clone(), FILE_FORMAT.clone(), Arc::new(NestedField::required( - 102, + DATA_FILE_PARTITION_FIELD_ID, "partition", Type::Struct(partition_type.clone()), )), @@ -650,7 +665,7 @@ pub(super) fn manifest_schema_v1(partition_type: &StructType) -> Result<AvroSche STATUS.clone(), SNAPSHOT_ID_V1.clone(), Arc::new(NestedField::required( - 2, + DATA_FILE_FIELD_ID, "data_file", Type::Struct(StructType::new(data_file_fields_v1(partition_type))), )), diff --git a/crates/iceberg/src/spec/manifest_list/_const_schema.rs b/crates/iceberg/src/spec/manifest_list/_const_schema.rs index d72ead9a4..ae417f336 100644 --- a/crates/iceberg/src/spec/manifest_list/_const_schema.rs +++ b/crates/iceberg/src/spec/manifest_list/_const_schema.rs @@ -24,6 +24,13 @@ use once_cell::sync::Lazy; use crate::avro::schema_to_avro_schema; use crate::spec::{ListType, NestedField, NestedFieldRef, PrimitiveType, Schema, StructType, Type}; +/// Field ID of `manifest_file.sequence_number`. +pub(crate) const MANIFEST_SEQUENCE_NUMBER_FIELD_ID: i32 = 515; +/// Field ID of `manifest_file.min_sequence_number`. +pub(crate) const MIN_SEQUENCE_NUMBER_FIELD_ID: i32 = 516; +/// Field ID of `manifest_file.content`. +pub(crate) const MANIFEST_CONTENT_FIELD_ID: i32 = 517; + static MANIFEST_PATH: Lazy<NestedFieldRef> = { Lazy::new(|| { Arc::new(NestedField::required( @@ -54,7 +61,7 @@ static PARTITION_SPEC_ID: Lazy<NestedFieldRef> = { static CONTENT: Lazy<NestedFieldRef> = { Lazy::new(|| { Arc::new(NestedField::required( - 517, + MANIFEST_CONTENT_FIELD_ID, "content", Type::Primitive(PrimitiveType::Int), )) @@ -63,7 +70,7 @@ static CONTENT: Lazy<NestedFieldRef> = { static SEQUENCE_NUMBER: Lazy<NestedFieldRef> = { Lazy::new(|| { Arc::new(NestedField::required( - 515, + MANIFEST_SEQUENCE_NUMBER_FIELD_ID, "sequence_number", Type::Primitive(PrimitiveType::Long), )) @@ -72,7 +79,7 @@ static SEQUENCE_NUMBER: Lazy<NestedFieldRef> = { static MIN_SEQUENCE_NUMBER: Lazy<NestedFieldRef> = { Lazy::new(|| { Arc::new(NestedField::required( - 516, + MIN_SEQUENCE_NUMBER_FIELD_ID, "min_sequence_number", Type::Primitive(PrimitiveType::Long), )) @@ -318,6 +325,36 @@ static V3_SCHEMA: Lazy<Schema> = { }) }; +/// Fields of the `manifest_file` struct in the latest format version, with the +/// file and row counts optional as in v1, so a manifest list of any version +/// fits them. +pub(crate) fn manifest_file_fields() -> Vec<NestedFieldRef> { + let optional = |field: &NestedFieldRef| { + Arc::new(NestedField { + required: false, + ..field.as_ref().clone() + }) + }; + vec![ + MANIFEST_PATH.clone(), + MANIFEST_LENGTH.clone(), + PARTITION_SPEC_ID.clone(), + CONTENT.clone(), + SEQUENCE_NUMBER.clone(), + MIN_SEQUENCE_NUMBER.clone(), + ADDED_SNAPSHOT_ID.clone(), + optional(&ADDED_FILES_COUNT_V2), + optional(&EXISTING_FILES_COUNT_V2), + optional(&DELETED_FILES_COUNT_V2), + optional(&ADDED_ROWS_COUNT_V2), + optional(&EXISTING_ROWS_COUNT_V2), + optional(&DELETED_ROWS_COUNT_V2), + PARTITIONS.clone(), + KEY_METADATA.clone(), + FIRST_ROW_ID.clone(), + ] +} + pub(super) static MANIFEST_LIST_AVRO_SCHEMA_V1: Lazy<AvroSchema> = Lazy::new(|| schema_to_avro_schema("manifest_file", &V1_SCHEMA).unwrap()); diff --git a/crates/iceberg/src/spec/manifest_list/mod.rs b/crates/iceberg/src/spec/manifest_list/mod.rs index 24a4debf9..8a5db38f3 100644 --- a/crates/iceberg/src/spec/manifest_list/mod.rs +++ b/crates/iceberg/src/spec/manifest_list/mod.rs @@ -31,6 +31,10 @@ pub use serde_bytes::ByteBuf; pub use writer::*; use self::_const_schema::MANIFEST_LIST_AVRO_SCHEMA_V1; +pub(crate) use self::_const_schema::{ + MANIFEST_CONTENT_FIELD_ID, MANIFEST_SEQUENCE_NUMBER_FIELD_ID, MIN_SEQUENCE_NUMBER_FIELD_ID, + manifest_file_fields, +}; use super::FormatVersion; use crate::error::Result;
