iceberg/spec/manifest/
mod.rs

1// Licensed to the Apache Software Foundation (ASF) under one
2// or more contributor license agreements.  See the NOTICE file
3// distributed with this work for additional information
4// regarding copyright ownership.  The ASF licenses this file
5// to you under the Apache License, Version 2.0 (the
6// "License"); you may not use this file except in compliance
7// with the License.  You may obtain a copy of the License at
8//
9//   http://www.apache.org/licenses/LICENSE-2.0
10//
11// Unless required by applicable law or agreed to in writing,
12// software distributed under the License is distributed on an
13// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14// KIND, either express or implied.  See the License for the
15// specific language governing permissions and limitations
16// under the License.
17
18mod _serde;
19
20mod data_file;
21pub use data_file::*;
22mod entry;
23pub use entry::*;
24mod metadata;
25pub use metadata::*;
26mod writer;
27use std::sync::Arc;
28
29use apache_avro::{Reader as AvroReader, from_value};
30pub use writer::*;
31
32use super::{
33    Datum, FormatVersion, ManifestContentType, PartitionSpec, PrimitiveType, Schema, Struct,
34    UNASSIGNED_SEQUENCE_NUMBER,
35};
36use crate::error::Result;
37use crate::{Error, ErrorKind};
38
39/// A manifest contains metadata and a list of entries.
40#[derive(Debug, PartialEq, Eq, Clone)]
41pub struct Manifest {
42    metadata: ManifestMetadata,
43    entries: Vec<ManifestEntryRef>,
44}
45
46impl Manifest {
47    /// Parse manifest metadata and entries from bytes of avro file.
48    pub(crate) fn try_from_avro_bytes(bs: &[u8]) -> Result<(ManifestMetadata, Vec<ManifestEntry>)> {
49        let reader = AvroReader::new(bs)?;
50
51        // Parse manifest metadata
52        let meta = reader.user_metadata();
53        let metadata = ManifestMetadata::parse(meta)?;
54
55        // Parse manifest entries
56        let partition_type = metadata.partition_spec.partition_type(&metadata.schema)?;
57
58        let entries = match metadata.format_version {
59            FormatVersion::V1 => {
60                let schema = manifest_schema_v1(&partition_type)?;
61                let reader = AvroReader::with_schema(&schema, bs)?;
62                reader
63                    .into_iter()
64                    .map(|value| {
65                        from_value::<_serde::ManifestEntryV1>(&value?)?.try_into(
66                            metadata.partition_spec.spec_id(),
67                            &partition_type,
68                            &metadata.schema,
69                        )
70                    })
71                    .collect::<Result<Vec<_>>>()?
72            }
73            // Manifest Schema & Manifest Entry did not change between V2 and V3
74            FormatVersion::V2 | FormatVersion::V3 => {
75                let schema = manifest_schema_v2(&partition_type)?;
76                let reader = AvroReader::with_schema(&schema, bs)?;
77                reader
78                    .into_iter()
79                    .map(|value| {
80                        from_value::<_serde::ManifestEntryV2>(&value?)?.try_into(
81                            metadata.partition_spec.spec_id(),
82                            &partition_type,
83                            &metadata.schema,
84                        )
85                    })
86                    .collect::<Result<Vec<_>>>()?
87            }
88        };
89
90        Ok((metadata, entries))
91    }
92
93    /// Parse manifest from bytes of avro file.
94    pub fn parse_avro(bs: &[u8]) -> Result<Self> {
95        let (metadata, entries) = Self::try_from_avro_bytes(bs)?;
96        Ok(Self::new(metadata, entries))
97    }
98
99    /// Entries slice.
100    pub fn entries(&self) -> &[ManifestEntryRef] {
101        &self.entries
102    }
103
104    /// Get metadata.
105    pub fn metadata(&self) -> &ManifestMetadata {
106        &self.metadata
107    }
108
109    /// Consume this Manifest, returning its constituent parts
110    pub fn into_parts(self) -> (Vec<ManifestEntryRef>, ManifestMetadata) {
111        let Self { entries, metadata } = self;
112        (entries, metadata)
113    }
114
115    /// Constructor from [`ManifestMetadata`] and [`ManifestEntry`]s.
116    pub fn new(metadata: ManifestMetadata, entries: Vec<ManifestEntry>) -> Self {
117        Self {
118            metadata,
119            entries: entries.into_iter().map(Arc::new).collect(),
120        }
121    }
122}
123
124/// Serialize a DataFile to a JSON string.
125pub fn serialize_data_file_to_json(
126    data_file: DataFile,
127    partition_type: &super::StructType,
128    format_version: FormatVersion,
129) -> Result<String> {
130    let serde = _serde::DataFileSerde::try_from(data_file, partition_type, format_version)?;
131    serde_json::to_string(&serde).map_err(|e| {
132        Error::new(
133            ErrorKind::DataInvalid,
134            "Failed to serialize DataFile to JSON!".to_string(),
135        )
136        .with_source(e)
137    })
138}
139
140/// Deserialize a DataFile from a JSON string.
141pub fn deserialize_data_file_from_json(
142    json: &str,
143    partition_spec_id: i32,
144    partition_type: &super::StructType,
145    schema: &Schema,
146) -> Result<DataFile> {
147    let serde = serde_json::from_str::<_serde::DataFileSerde>(json).map_err(|e| {
148        Error::new(
149            ErrorKind::DataInvalid,
150            "Failed to deserialize JSON to DataFile!".to_string(),
151        )
152        .with_source(e)
153    })?;
154
155    serde.try_into(partition_spec_id, partition_type, schema)
156}
157
158#[cfg(test)]
159mod tests {
160    use std::collections::HashMap;
161    use std::fs;
162    use std::sync::Arc;
163
164    use apache_avro::{Codec, Writer, to_value};
165    use serde_json::{Value, to_vec};
166    use tempfile::TempDir;
167
168    use super::*;
169    use crate::io::FileIO;
170    use crate::spec::{Literal, NestedField, PrimitiveType, Struct, Transform, Type};
171
172    #[tokio::test]
173    async fn test_parse_manifest_v2_unpartition() {
174        let schema = Arc::new(
175            Schema::builder()
176                .with_fields(vec![
177                    // id v_int v_long v_float v_double v_varchar v_bool v_date v_timestamp v_decimal v_ts_ntz
178                    Arc::new(NestedField::optional(
179                        1,
180                        "id",
181                        Type::Primitive(PrimitiveType::Long),
182                    )),
183                    Arc::new(NestedField::optional(
184                        2,
185                        "v_int",
186                        Type::Primitive(PrimitiveType::Int),
187                    )),
188                    Arc::new(NestedField::optional(
189                        3,
190                        "v_long",
191                        Type::Primitive(PrimitiveType::Long),
192                    )),
193                    Arc::new(NestedField::optional(
194                        4,
195                        "v_float",
196                        Type::Primitive(PrimitiveType::Float),
197                    )),
198                    Arc::new(NestedField::optional(
199                        5,
200                        "v_double",
201                        Type::Primitive(PrimitiveType::Double),
202                    )),
203                    Arc::new(NestedField::optional(
204                        6,
205                        "v_varchar",
206                        Type::Primitive(PrimitiveType::String),
207                    )),
208                    Arc::new(NestedField::optional(
209                        7,
210                        "v_bool",
211                        Type::Primitive(PrimitiveType::Boolean),
212                    )),
213                    Arc::new(NestedField::optional(
214                        8,
215                        "v_date",
216                        Type::Primitive(PrimitiveType::Date),
217                    )),
218                    Arc::new(NestedField::optional(
219                        9,
220                        "v_timestamp",
221                        Type::Primitive(PrimitiveType::Timestamptz),
222                    )),
223                    Arc::new(NestedField::optional(
224                        10,
225                        "v_decimal",
226                        Type::Primitive(PrimitiveType::Decimal {
227                            precision: 36,
228                            scale: 10,
229                        }),
230                    )),
231                    Arc::new(NestedField::optional(
232                        11,
233                        "v_ts_ntz",
234                        Type::Primitive(PrimitiveType::Timestamp),
235                    )),
236                    Arc::new(NestedField::optional(
237                        12,
238                        "v_ts_ns_ntz",
239                        Type::Primitive(PrimitiveType::TimestampNs),
240                    )),
241                ])
242                .build()
243                .unwrap(),
244        );
245        let metadata = ManifestMetadata {
246            schema_id: 0,
247            schema: schema.clone(),
248            partition_spec: PartitionSpec::builder(schema)
249                .with_spec_id(0)
250                .build()
251                .unwrap(),
252            content: ManifestContentType::Data,
253            format_version: FormatVersion::V2,
254        };
255        let mut entries = vec![
256                ManifestEntry {
257                    status: ManifestStatus::Added,
258                    snapshot_id: None,
259                    sequence_number: None,
260                    file_sequence_number: None,
261                    data_file: DataFile {content:DataContentType::Data,file_path:"s3a://icebergdata/demo/s1/t1/data/00000-0-ba56fbfa-f2ff-40c9-bb27-565ad6dc2be8-00000.parquet".to_string(),file_format:DataFileFormat::Parquet,partition:Struct::empty(),record_count:1,file_size_in_bytes:5442,column_sizes:HashMap::from([(0,73),(6,34),(2,73),(7,61),(3,61),(5,62),(9,79),(10,73),(1,61),(4,73),(8,73)]),value_counts:HashMap::from([(4,1),(5,1),(2,1),(0,1),(3,1),(6,1),(8,1),(1,1),(10,1),(7,1),(9,1)]),null_value_counts:HashMap::from([(1,0),(6,0),(2,0),(8,0),(0,0),(3,0),(5,0),(9,0),(7,0),(4,0),(10,0)]),nan_value_counts:HashMap::new(),lower_bounds:HashMap::new(),upper_bounds:HashMap::new(),key_metadata:None,split_offsets:Some(vec![4]),equality_ids:Some(Vec::new()),sort_order_id:None, partition_spec_id: 0,first_row_id: None,referenced_data_file: None,content_offset: None,content_size_in_bytes: None }
262                }
263            ];
264
265        // write manifest to file
266        let tmp_dir = TempDir::new().unwrap();
267        let path = tmp_dir.path().join("test_manifest.avro");
268        let io = FileIO::new_with_fs();
269        let output_file = io.new_output(path.to_str().unwrap()).unwrap();
270        let mut writer = ManifestWriterBuilder::new(
271            output_file,
272            Some(1),
273            metadata.schema.clone(),
274            metadata.partition_spec.clone(),
275        )
276        .build_v2_data();
277        for entry in &entries {
278            writer.add_entry(entry.clone()).unwrap();
279        }
280        writer.write_manifest_file().await.unwrap();
281
282        // read back the manifest file and check the content
283        let actual_manifest =
284            Manifest::parse_avro(fs::read(path).expect("read_file must succeed").as_slice())
285                .unwrap();
286        // The snapshot id is assigned when the entry is added to the manifest.
287        entries[0].snapshot_id = Some(1);
288        assert_eq!(actual_manifest, Manifest::new(metadata, entries));
289    }
290
291    #[test]
292    fn test_parse_snappy_manifest_v2() {
293        let schema = Arc::new(
294            Schema::builder()
295                .with_fields(vec![Arc::new(NestedField::optional(
296                    1,
297                    "id",
298                    Type::Primitive(PrimitiveType::Long),
299                ))])
300                .build()
301                .unwrap(),
302        );
303        let partition_spec = PartitionSpec::builder(schema.clone())
304            .with_spec_id(0)
305            .build()
306            .unwrap();
307
308        for (manifest_content, file_content, file_path) in [
309            (
310                ManifestContentType::Data,
311                DataContentType::Data,
312                "s3://bucket/table/data/data.parquet",
313            ),
314            (
315                ManifestContentType::Deletes,
316                DataContentType::PositionDeletes,
317                "s3://bucket/table/data/delete.parquet",
318            ),
319        ] {
320            let metadata = ManifestMetadata {
321                schema_id: 0,
322                schema: schema.clone(),
323                partition_spec: partition_spec.clone(),
324                content: manifest_content,
325                format_version: FormatVersion::V2,
326            };
327            let entry = ManifestEntry {
328                status: ManifestStatus::Added,
329                snapshot_id: Some(1),
330                sequence_number: None,
331                file_sequence_number: None,
332                data_file: DataFile {
333                    content: file_content,
334                    file_path: file_path.to_string(),
335                    file_format: DataFileFormat::Parquet,
336                    partition: Struct::empty(),
337                    record_count: 1,
338                    file_size_in_bytes: 1024,
339                    column_sizes: HashMap::new(),
340                    value_counts: HashMap::new(),
341                    null_value_counts: HashMap::new(),
342                    nan_value_counts: HashMap::new(),
343                    lower_bounds: HashMap::new(),
344                    upper_bounds: HashMap::new(),
345                    key_metadata: None,
346                    split_offsets: None,
347                    equality_ids: None,
348                    sort_order_id: None,
349                    partition_spec_id: 0,
350                    first_row_id: None,
351                    referenced_data_file: None,
352                    content_offset: None,
353                    content_size_in_bytes: None,
354                },
355            };
356
357            let partition_type = metadata
358                .partition_spec
359                .partition_type(&metadata.schema)
360                .unwrap();
361            let avro_schema = manifest_schema_v2(&partition_type).unwrap();
362            let mut writer = Writer::with_codec(&avro_schema, Vec::new(), Codec::Snappy);
363            writer
364                .add_user_metadata("schema".to_string(), to_vec(&metadata.schema).unwrap())
365                .unwrap();
366            writer
367                .add_user_metadata(
368                    "schema-id".to_string(),
369                    metadata.schema.schema_id().to_string(),
370                )
371                .unwrap();
372            writer
373                .add_user_metadata(
374                    "partition-spec".to_string(),
375                    to_vec(&metadata.partition_spec.fields()).unwrap(),
376                )
377                .unwrap();
378            writer
379                .add_user_metadata(
380                    "partition-spec-id".to_string(),
381                    metadata.partition_spec.spec_id().to_string(),
382                )
383                .unwrap();
384            writer
385                .add_user_metadata(
386                    "format-version".to_string(),
387                    (metadata.format_version as u8).to_string(),
388                )
389                .unwrap();
390            writer
391                .add_user_metadata("content".to_string(), metadata.content.to_string())
392                .unwrap();
393            let value = to_value(
394                _serde::ManifestEntryV2::try_from(entry.clone(), &partition_type).unwrap(),
395            )
396            .unwrap()
397            .resolve(&avro_schema)
398            .unwrap();
399            writer.append(value).unwrap();
400            let bs = writer.into_inner().unwrap();
401
402            let parsed_manifest = Manifest::parse_avro(&bs).unwrap();
403
404            assert_eq!(parsed_manifest, Manifest::new(metadata, vec![entry]));
405        }
406    }
407
408    #[tokio::test]
409    async fn test_parse_manifest_v2_partition() {
410        let schema = Arc::new(
411            Schema::builder()
412                .with_fields(vec![
413                    Arc::new(NestedField::optional(
414                        1,
415                        "id",
416                        Type::Primitive(PrimitiveType::Long),
417                    )),
418                    Arc::new(NestedField::optional(
419                        2,
420                        "v_int",
421                        Type::Primitive(PrimitiveType::Int),
422                    )),
423                    Arc::new(NestedField::optional(
424                        3,
425                        "v_long",
426                        Type::Primitive(PrimitiveType::Long),
427                    )),
428                    Arc::new(NestedField::optional(
429                        4,
430                        "v_float",
431                        Type::Primitive(PrimitiveType::Float),
432                    )),
433                    Arc::new(NestedField::optional(
434                        5,
435                        "v_double",
436                        Type::Primitive(PrimitiveType::Double),
437                    )),
438                    Arc::new(NestedField::optional(
439                        6,
440                        "v_varchar",
441                        Type::Primitive(PrimitiveType::String),
442                    )),
443                    Arc::new(NestedField::optional(
444                        7,
445                        "v_bool",
446                        Type::Primitive(PrimitiveType::Boolean),
447                    )),
448                    Arc::new(NestedField::optional(
449                        8,
450                        "v_date",
451                        Type::Primitive(PrimitiveType::Date),
452                    )),
453                    Arc::new(NestedField::optional(
454                        9,
455                        "v_timestamp",
456                        Type::Primitive(PrimitiveType::Timestamptz),
457                    )),
458                    Arc::new(NestedField::optional(
459                        10,
460                        "v_decimal",
461                        Type::Primitive(PrimitiveType::Decimal {
462                            precision: 36,
463                            scale: 10,
464                        }),
465                    )),
466                    Arc::new(NestedField::optional(
467                        11,
468                        "v_ts_ntz",
469                        Type::Primitive(PrimitiveType::Timestamp),
470                    )),
471                    Arc::new(NestedField::optional(
472                        12,
473                        "v_ts_ns_ntz",
474                        Type::Primitive(PrimitiveType::TimestampNs),
475                    )),
476                ])
477                .build()
478                .unwrap(),
479        );
480        let metadata = ManifestMetadata {
481            schema_id: 0,
482            schema: schema.clone(),
483            partition_spec: PartitionSpec::builder(schema)
484                .with_spec_id(0)
485                .add_partition_field("v_int", "v_int", Transform::Identity)
486                .unwrap()
487                .add_partition_field("v_long", "v_long", Transform::Identity)
488                .unwrap()
489                .build()
490                .unwrap(),
491            content: ManifestContentType::Data,
492            format_version: FormatVersion::V2,
493        };
494        let mut entries = vec![ManifestEntry {
495                status: ManifestStatus::Added,
496                snapshot_id: None,
497                sequence_number: None,
498                file_sequence_number: None,
499                data_file: DataFile {
500                    content: DataContentType::Data,
501                    file_format: DataFileFormat::Parquet,
502                    file_path: "s3a://icebergdata/demo/s1/t1/data/00000-0-378b56f5-5c52-4102-a2c2-f05f8a7cbe4a-00000.parquet".to_string(),
503                    partition: Struct::from_iter(
504                        vec![
505                            Some(Literal::int(1)),
506                            Some(Literal::long(1000)),
507                        ]
508                            .into_iter()
509                    ),
510                    record_count: 1,
511                    file_size_in_bytes: 5442,
512                    column_sizes: HashMap::from([
513                        (0, 73),
514                        (6, 34),
515                        (2, 73),
516                        (7, 61),
517                        (3, 61),
518                        (5, 62),
519                        (9, 79),
520                        (10, 73),
521                        (1, 61),
522                        (4, 73),
523                        (8, 73)
524                    ]),
525                    value_counts: HashMap::from([
526                        (4, 1),
527                        (5, 1),
528                        (2, 1),
529                        (0, 1),
530                        (3, 1),
531                        (6, 1),
532                        (8, 1),
533                        (1, 1),
534                        (10, 1),
535                        (7, 1),
536                        (9, 1)
537                    ]),
538                    null_value_counts: HashMap::from([
539                        (1, 0),
540                        (6, 0),
541                        (2, 0),
542                        (8, 0),
543                        (0, 0),
544                        (3, 0),
545                        (5, 0),
546                        (9, 0),
547                        (7, 0),
548                        (4, 0),
549                        (10, 0)
550                    ]),
551                    nan_value_counts: HashMap::new(),
552                    lower_bounds: HashMap::new(),
553                    upper_bounds: HashMap::new(),
554                    key_metadata: None,
555                    split_offsets: Some(vec![4]),
556                    equality_ids: Some(Vec::new()),
557                    sort_order_id: None,
558                    partition_spec_id: 0,
559                    first_row_id: None,
560                    referenced_data_file: None,
561                    content_offset: None,
562                    content_size_in_bytes: None,
563                },
564            }];
565
566        // write manifest to file and check the return manifest file.
567        let tmp_dir = TempDir::new().unwrap();
568        let path = tmp_dir.path().join("test_manifest.avro");
569        let io = FileIO::new_with_fs();
570        let output_file = io.new_output(path.to_str().unwrap()).unwrap();
571        let mut writer = ManifestWriterBuilder::new(
572            output_file,
573            Some(2),
574            metadata.schema.clone(),
575            metadata.partition_spec.clone(),
576        )
577        .build_v2_data();
578        for entry in &entries {
579            writer.add_entry(entry.clone()).unwrap();
580        }
581        let manifest_file = writer.write_manifest_file().await.unwrap();
582        assert_eq!(manifest_file.sequence_number, UNASSIGNED_SEQUENCE_NUMBER);
583        assert_eq!(
584            manifest_file.min_sequence_number,
585            UNASSIGNED_SEQUENCE_NUMBER
586        );
587
588        // read back the manifest file and check the content
589        let actual_manifest =
590            Manifest::parse_avro(fs::read(path).expect("read_file must succeed").as_slice())
591                .unwrap();
592        // The snapshot id is assigned when the entry is added to the manifest.
593        entries[0].snapshot_id = Some(2);
594        assert_eq!(actual_manifest, Manifest::new(metadata, entries));
595    }
596
597    #[tokio::test]
598    async fn test_parse_manifest_v1_unpartition() {
599        let schema = Arc::new(
600            Schema::builder()
601                .with_schema_id(1)
602                .with_fields(vec![
603                    Arc::new(NestedField::optional(
604                        1,
605                        "id",
606                        Type::Primitive(PrimitiveType::Int),
607                    )),
608                    Arc::new(NestedField::optional(
609                        2,
610                        "data",
611                        Type::Primitive(PrimitiveType::String),
612                    )),
613                    Arc::new(NestedField::optional(
614                        3,
615                        "comment",
616                        Type::Primitive(PrimitiveType::String),
617                    )),
618                ])
619                .build()
620                .unwrap(),
621        );
622        let metadata = ManifestMetadata {
623            schema_id: 1,
624            schema: schema.clone(),
625            partition_spec: PartitionSpec::builder(schema)
626                .with_spec_id(0)
627                .build()
628                .unwrap(),
629            content: ManifestContentType::Data,
630            format_version: FormatVersion::V1,
631        };
632        let mut entries = vec![ManifestEntry {
633                status: ManifestStatus::Added,
634                snapshot_id: Some(0),
635                sequence_number: Some(0),
636                file_sequence_number: Some(0),
637                data_file: DataFile {
638                    content: DataContentType::Data,
639                    file_path: "s3://testbucket/iceberg_data/iceberg_ctl/iceberg_db/iceberg_tbl/data/00000-7-45268d71-54eb-476c-b42c-942d880c04a1-00001.parquet".to_string(),
640                    file_format: DataFileFormat::Parquet,
641                    partition: Struct::empty(),
642                    record_count: 1,
643                    file_size_in_bytes: 875,
644                    column_sizes: HashMap::from([(1,47),(2,48),(3,52)]),
645                    value_counts: HashMap::from([(1,1),(2,1),(3,1)]),
646                    null_value_counts: HashMap::from([(1,0),(2,0),(3,0)]),
647                    nan_value_counts: HashMap::new(),
648                    lower_bounds: HashMap::from([(1,Datum::int(1)),(2,Datum::string("a")),(3,Datum::string("AC/DC"))]),
649                    upper_bounds: HashMap::from([(1,Datum::int(1)),(2,Datum::string("a")),(3,Datum::string("AC/DC"))]),
650                    key_metadata: None,
651                    split_offsets: Some(vec![4]),
652                    equality_ids: None,
653                    sort_order_id: Some(0),
654                    partition_spec_id: 0,
655                    first_row_id: None,
656                    referenced_data_file: None,
657                    content_offset: None,
658                    content_size_in_bytes: None,
659                }
660            }];
661
662        // write manifest to file
663        let tmp_dir = TempDir::new().unwrap();
664        let path = tmp_dir.path().join("test_manifest.avro");
665        let io = FileIO::new_with_fs();
666        let output_file = io.new_output(path.to_str().unwrap()).unwrap();
667        let mut writer = ManifestWriterBuilder::new(
668            output_file,
669            Some(3),
670            metadata.schema.clone(),
671            metadata.partition_spec.clone(),
672        )
673        .build_v1();
674        for entry in &entries {
675            writer.add_entry(entry.clone()).unwrap();
676        }
677        writer.write_manifest_file().await.unwrap();
678
679        // read back the manifest file and check the content
680        let actual_manifest =
681            Manifest::parse_avro(fs::read(path).expect("read_file must succeed").as_slice())
682                .unwrap();
683        // The snapshot id is assigned when the entry is added to the manifest.
684        entries[0].snapshot_id = Some(3);
685        assert_eq!(actual_manifest, Manifest::new(metadata, entries));
686    }
687
688    #[tokio::test]
689    async fn test_parse_manifest_v1_partition() {
690        let schema = Arc::new(
691            Schema::builder()
692                .with_fields(vec![
693                    Arc::new(NestedField::optional(
694                        1,
695                        "id",
696                        Type::Primitive(PrimitiveType::Long),
697                    )),
698                    Arc::new(NestedField::optional(
699                        2,
700                        "data",
701                        Type::Primitive(PrimitiveType::String),
702                    )),
703                    Arc::new(NestedField::optional(
704                        3,
705                        "category",
706                        Type::Primitive(PrimitiveType::String),
707                    )),
708                ])
709                .build()
710                .unwrap(),
711        );
712        let metadata = ManifestMetadata {
713            schema_id: 0,
714            schema: schema.clone(),
715            partition_spec: PartitionSpec::builder(schema)
716                .add_partition_field("category", "category", Transform::Identity)
717                .unwrap()
718                .build()
719                .unwrap(),
720            content: ManifestContentType::Data,
721            format_version: FormatVersion::V1,
722        };
723        let mut entries = vec![
724                ManifestEntry {
725                    status: ManifestStatus::Added,
726                    snapshot_id: Some(0),
727                    sequence_number: Some(0),
728                    file_sequence_number: Some(0),
729                    data_file: DataFile {
730                        content: DataContentType::Data,
731                        file_path: "s3://testbucket/prod/db/sample/data/category=x/00010-1-d5c93668-1e52-41ac-92a6-bba590cbf249-00001.parquet".to_string(),
732                        file_format: DataFileFormat::Parquet,
733                        partition: Struct::from_iter(
734                            vec![
735                                Some(
736                                    Literal::string("x"),
737                                ),
738                            ]
739                                .into_iter()
740                        ),
741                        record_count: 1,
742                        file_size_in_bytes: 874,
743                        column_sizes: HashMap::from([(1, 46), (2, 48), (3, 48)]),
744                        value_counts: HashMap::from([(1, 1), (2, 1), (3, 1)]),
745                        null_value_counts: HashMap::from([(1, 0), (2, 0), (3, 0)]),
746                        nan_value_counts: HashMap::new(),
747                        lower_bounds: HashMap::from([
748                        (1, Datum::long(1)),
749                        (2, Datum::string("a")),
750                        (3, Datum::string("x"))
751                        ]),
752                        upper_bounds: HashMap::from([
753                        (1, Datum::long(1)),
754                        (2, Datum::string("a")),
755                        (3, Datum::string("x"))
756                        ]),
757                        key_metadata: None,
758                        split_offsets: Some(vec![4]),
759                        equality_ids: None,
760                        sort_order_id: Some(0),
761                        partition_spec_id: 0,
762                        first_row_id: None,
763                        referenced_data_file: None,
764                        content_offset: None,
765                        content_size_in_bytes: None,
766                    },
767                }
768            ];
769
770        // write manifest to file
771        let tmp_dir = TempDir::new().unwrap();
772        let path = tmp_dir.path().join("test_manifest.avro");
773        let io = FileIO::new_with_fs();
774        let output_file = io.new_output(path.to_str().unwrap()).unwrap();
775        let mut writer = ManifestWriterBuilder::new(
776            output_file,
777            Some(2),
778            metadata.schema.clone(),
779            metadata.partition_spec.clone(),
780        )
781        .build_v1();
782        for entry in &entries {
783            writer.add_entry(entry.clone()).unwrap();
784        }
785        let manifest_file = writer.write_manifest_file().await.unwrap();
786        let partitions = manifest_file.partitions.unwrap();
787        assert_eq!(partitions.len(), 1);
788        assert_eq!(
789            partitions[0].clone().lower_bound.unwrap(),
790            Datum::string("x").to_bytes().unwrap()
791        );
792        assert_eq!(
793            partitions[0].clone().upper_bound.unwrap(),
794            Datum::string("x").to_bytes().unwrap()
795        );
796
797        // read back the manifest file and check the content
798        let actual_manifest =
799            Manifest::parse_avro(fs::read(path).expect("read_file must succeed").as_slice())
800                .unwrap();
801        // The snapshot id is assigned when the entry is added to the manifest.
802        entries[0].snapshot_id = Some(2);
803        assert_eq!(actual_manifest, Manifest::new(metadata, entries));
804    }
805
806    #[tokio::test]
807    async fn test_parse_manifest_with_schema_evolution() {
808        let schema = Arc::new(
809            Schema::builder()
810                .with_fields(vec![
811                    Arc::new(NestedField::optional(
812                        1,
813                        "id",
814                        Type::Primitive(PrimitiveType::Long),
815                    )),
816                    Arc::new(NestedField::optional(
817                        2,
818                        "v_int",
819                        Type::Primitive(PrimitiveType::Int),
820                    )),
821                ])
822                .build()
823                .unwrap(),
824        );
825        let metadata = ManifestMetadata {
826            schema_id: 0,
827            schema: schema.clone(),
828            partition_spec: PartitionSpec::builder(schema)
829                .with_spec_id(0)
830                .build()
831                .unwrap(),
832            content: ManifestContentType::Data,
833            format_version: FormatVersion::V2,
834        };
835        let entries = vec![ManifestEntry {
836                status: ManifestStatus::Added,
837                snapshot_id: None,
838                sequence_number: None,
839                file_sequence_number: None,
840                data_file: DataFile {
841                    content: DataContentType::Data,
842                    file_format: DataFileFormat::Parquet,
843                    file_path: "s3a://icebergdata/demo/s1/t1/data/00000-0-378b56f5-5c52-4102-a2c2-f05f8a7cbe4a-00000.parquet".to_string(),
844                    partition: Struct::empty(),
845                    record_count: 1,
846                    file_size_in_bytes: 5442,
847                    column_sizes: HashMap::from([
848                        (1, 61),
849                        (2, 73),
850                        (3, 61),
851                    ]),
852                    value_counts: HashMap::default(),
853                    null_value_counts: HashMap::default(),
854                    nan_value_counts: HashMap::new(),
855                    lower_bounds: HashMap::from([
856                        (1, Datum::long(1)),
857                        (2, Datum::int(2)),
858                        (3, Datum::string("x"))
859                    ]),
860                    upper_bounds: HashMap::from([
861                        (1, Datum::long(1)),
862                        (2, Datum::int(2)),
863                        (3, Datum::string("x"))
864                    ]),
865                    key_metadata: None,
866                    split_offsets: Some(vec![4]),
867                    equality_ids: None,
868                    sort_order_id: None,
869                    partition_spec_id: 0,
870                    first_row_id: None,
871                    referenced_data_file: None,
872                    content_offset: None,
873                    content_size_in_bytes: None,
874                },
875            }];
876
877        // write manifest to file
878        let tmp_dir = TempDir::new().unwrap();
879        let path = tmp_dir.path().join("test_manifest.avro");
880        let io = FileIO::new_with_fs();
881        let output_file = io.new_output(path.to_str().unwrap()).unwrap();
882        let mut writer = ManifestWriterBuilder::new(
883            output_file,
884            Some(2),
885            metadata.schema.clone(),
886            metadata.partition_spec.clone(),
887        )
888        .build_v2_data();
889        for entry in &entries {
890            writer.add_entry(entry.clone()).unwrap();
891        }
892        writer.write_manifest_file().await.unwrap();
893
894        // read back the manifest file and check the content
895        let actual_manifest =
896            Manifest::parse_avro(fs::read(path).expect("read_file must succeed").as_slice())
897                .unwrap();
898
899        // Compared with original manifest, the lower_bounds and upper_bounds no longer has data for field 3, and
900        // other parts should be same.
901        // The snapshot id is assigned when the entry is added to the manifest.
902        let schema = Arc::new(
903            Schema::builder()
904                .with_fields(vec![
905                    Arc::new(NestedField::optional(
906                        1,
907                        "id",
908                        Type::Primitive(PrimitiveType::Long),
909                    )),
910                    Arc::new(NestedField::optional(
911                        2,
912                        "v_int",
913                        Type::Primitive(PrimitiveType::Int),
914                    )),
915                ])
916                .build()
917                .unwrap(),
918        );
919        let expected_manifest = Manifest {
920            metadata: ManifestMetadata {
921                schema_id: 0,
922                schema: schema.clone(),
923                partition_spec: PartitionSpec::builder(schema).with_spec_id(0).build().unwrap(),
924                content: ManifestContentType::Data,
925                format_version: FormatVersion::V2,
926            },
927            entries: vec![Arc::new(ManifestEntry {
928                status: ManifestStatus::Added,
929                snapshot_id: Some(2),
930                sequence_number: None,
931                file_sequence_number: None,
932                data_file: DataFile {
933                    content: DataContentType::Data,
934                    file_format: DataFileFormat::Parquet,
935                    file_path: "s3a://icebergdata/demo/s1/t1/data/00000-0-378b56f5-5c52-4102-a2c2-f05f8a7cbe4a-00000.parquet".to_string(),
936                    partition: Struct::empty(),
937                    record_count: 1,
938                    file_size_in_bytes: 5442,
939                    column_sizes: HashMap::from([
940                        (1, 61),
941                        (2, 73),
942                        (3, 61),
943                    ]),
944                    value_counts: HashMap::default(),
945                    null_value_counts: HashMap::default(),
946                    nan_value_counts: HashMap::new(),
947                    lower_bounds: HashMap::from([
948                        (1, Datum::long(1)),
949                        (2, Datum::int(2)),
950                    ]),
951                    upper_bounds: HashMap::from([
952                        (1, Datum::long(1)),
953                        (2, Datum::int(2)),
954                    ]),
955                    key_metadata: None,
956                    split_offsets: Some(vec![4]),
957                    equality_ids: None,
958                    sort_order_id: None,
959                    partition_spec_id: 0,
960                    first_row_id: None,
961                    referenced_data_file: None,
962                    content_offset: None,
963                    content_size_in_bytes: None,
964                },
965            })],
966        };
967
968        assert_eq!(actual_manifest, expected_manifest);
969    }
970
971    #[tokio::test]
972    async fn test_manifest_summary() {
973        let schema = Arc::new(
974            Schema::builder()
975                .with_fields(vec![
976                    Arc::new(NestedField::optional(
977                        1,
978                        "time",
979                        Type::Primitive(PrimitiveType::Date),
980                    )),
981                    Arc::new(NestedField::optional(
982                        2,
983                        "v_float",
984                        Type::Primitive(PrimitiveType::Float),
985                    )),
986                    Arc::new(NestedField::optional(
987                        3,
988                        "v_double",
989                        Type::Primitive(PrimitiveType::Double),
990                    )),
991                ])
992                .build()
993                .unwrap(),
994        );
995        let partition_spec = PartitionSpec::builder(schema.clone())
996            .with_spec_id(0)
997            .add_partition_field("time", "year_of_time", Transform::Year)
998            .unwrap()
999            .add_partition_field("v_float", "f", Transform::Identity)
1000            .unwrap()
1001            .add_partition_field("v_double", "d", Transform::Identity)
1002            .unwrap()
1003            .build()
1004            .unwrap();
1005        let metadata = ManifestMetadata {
1006            schema_id: 0,
1007            schema,
1008            partition_spec,
1009            content: ManifestContentType::Data,
1010            format_version: FormatVersion::V2,
1011        };
1012        let entries = vec![
1013            ManifestEntry {
1014                status: ManifestStatus::Added,
1015                snapshot_id: None,
1016                sequence_number: None,
1017                file_sequence_number: None,
1018                data_file: DataFile {
1019                    content: DataContentType::Data,
1020                    file_path: "s3a://icebergdata/demo/s1/t1/data/00000-0-ba56fbfa-f2ff-40c9-bb27-565ad6dc2be8-00000.parquet".to_string(),
1021                    file_format: DataFileFormat::Parquet,
1022                    partition: Struct::from_iter(
1023                        vec![
1024                            Some(Literal::int(2021)),
1025                            Some(Literal::float(1.0)),
1026                            Some(Literal::double(2.0)),
1027                        ]
1028                    ),
1029                    record_count: 1,
1030                    file_size_in_bytes: 5442,
1031                    column_sizes: HashMap::from([(0,73),(6,34),(2,73),(7,61),(3,61),(5,62),(9,79),(10,73),(1,61),(4,73),(8,73)]),
1032                    value_counts: HashMap::from([(4,1),(5,1),(2,1),(0,1),(3,1),(6,1),(8,1),(1,1),(10,1),(7,1),(9,1)]),
1033                    null_value_counts: HashMap::from([(1,0),(6,0),(2,0),(8,0),(0,0),(3,0),(5,0),(9,0),(7,0),(4,0),(10,0)]),
1034                    nan_value_counts: HashMap::new(),
1035                    lower_bounds: HashMap::new(),
1036                    upper_bounds: HashMap::new(),
1037                    key_metadata: None,
1038                    split_offsets: Some(vec![4]),
1039                    equality_ids: None,
1040                    sort_order_id: None,
1041                    partition_spec_id: 0,
1042                    first_row_id: None,
1043                    referenced_data_file: None,
1044                    content_offset: None,
1045                    content_size_in_bytes: None,
1046                }
1047            },
1048                ManifestEntry {
1049                    status: ManifestStatus::Added,
1050                    snapshot_id: None,
1051                    sequence_number: None,
1052                    file_sequence_number: None,
1053                    data_file: DataFile {
1054                        content: DataContentType::Data,
1055                        file_path: "s3a://icebergdata/demo/s1/t1/data/00000-0-ba56fbfa-f2ff-40c9-bb27-565ad6dc2be8-00000.parquet".to_string(),
1056                        file_format: DataFileFormat::Parquet,
1057                        partition: Struct::from_iter(
1058                            vec![
1059                                Some(Literal::int(1111)),
1060                                Some(Literal::float(15.5)),
1061                                Some(Literal::double(25.5)),
1062                            ]
1063                        ),
1064                        record_count: 1,
1065                        file_size_in_bytes: 5442,
1066                        column_sizes: HashMap::from([(0,73),(6,34),(2,73),(7,61),(3,61),(5,62),(9,79),(10,73),(1,61),(4,73),(8,73)]),
1067                        value_counts: HashMap::from([(4,1),(5,1),(2,1),(0,1),(3,1),(6,1),(8,1),(1,1),(10,1),(7,1),(9,1)]),
1068                        null_value_counts: HashMap::from([(1,0),(6,0),(2,0),(8,0),(0,0),(3,0),(5,0),(9,0),(7,0),(4,0),(10,0)]),
1069                        nan_value_counts: HashMap::new(),
1070                        lower_bounds: HashMap::new(),
1071                        upper_bounds: HashMap::new(),
1072                        key_metadata: None,
1073                        split_offsets: Some(vec![4]),
1074                        equality_ids: None,
1075                        sort_order_id: None,
1076                        partition_spec_id: 0,
1077                        first_row_id: None,
1078                        referenced_data_file: None,
1079                        content_offset: None,
1080                        content_size_in_bytes: None,
1081                    }
1082                },
1083                ManifestEntry {
1084                    status: ManifestStatus::Added,
1085                    snapshot_id: None,
1086                    sequence_number: None,
1087                    file_sequence_number: None,
1088                    data_file: DataFile {
1089                        content: DataContentType::Data,
1090                        file_path: "s3a://icebergdata/demo/s1/t1/data/00000-0-ba56fbfa-f2ff-40c9-bb27-565ad6dc2be8-00000.parquet".to_string(),
1091                        file_format: DataFileFormat::Parquet,
1092                        partition: Struct::from_iter(
1093                            vec![
1094                                Some(Literal::int(1211)),
1095                                Some(Literal::float(f32::NAN)),
1096                                Some(Literal::double(1.0)),
1097                            ]
1098                        ),
1099                        record_count: 1,
1100                        file_size_in_bytes: 5442,
1101                        column_sizes: HashMap::from([(0,73),(6,34),(2,73),(7,61),(3,61),(5,62),(9,79),(10,73),(1,61),(4,73),(8,73)]),
1102                        value_counts: HashMap::from([(4,1),(5,1),(2,1),(0,1),(3,1),(6,1),(8,1),(1,1),(10,1),(7,1),(9,1)]),
1103                        null_value_counts: HashMap::from([(1,0),(6,0),(2,0),(8,0),(0,0),(3,0),(5,0),(9,0),(7,0),(4,0),(10,0)]),
1104                        nan_value_counts: HashMap::new(),
1105                        lower_bounds: HashMap::new(),
1106                        upper_bounds: HashMap::new(),
1107                        key_metadata: None,
1108                        split_offsets: Some(vec![4]),
1109                        equality_ids: None,
1110                        sort_order_id: None,
1111                        partition_spec_id: 0,
1112                        first_row_id: None,
1113                        referenced_data_file: None,
1114                        content_offset: None,
1115                        content_size_in_bytes: None,
1116                    }
1117                },
1118                ManifestEntry {
1119                    status: ManifestStatus::Added,
1120                    snapshot_id: None,
1121                    sequence_number: None,
1122                    file_sequence_number: None,
1123                    data_file: DataFile {
1124                        content: DataContentType::Data,
1125                        file_path: "s3a://icebergdata/demo/s1/t1/data/00000-0-ba56fbfa-f2ff-40c9-bb27-565ad6dc2be8-00000.parquet".to_string(),
1126                        file_format: DataFileFormat::Parquet,
1127                        partition: Struct::from_iter(
1128                            vec![
1129                                Some(Literal::int(1111)),
1130                                None,
1131                                Some(Literal::double(11.0)),
1132                            ]
1133                        ),
1134                        record_count: 1,
1135                        file_size_in_bytes: 5442,
1136                        column_sizes: HashMap::from([(0,73),(6,34),(2,73),(7,61),(3,61),(5,62),(9,79),(10,73),(1,61),(4,73),(8,73)]),
1137                        value_counts: HashMap::from([(4,1),(5,1),(2,1),(0,1),(3,1),(6,1),(8,1),(1,1),(10,1),(7,1),(9,1)]),
1138                        null_value_counts: HashMap::from([(1,0),(6,0),(2,0),(8,0),(0,0),(3,0),(5,0),(9,0),(7,0),(4,0),(10,0)]),
1139                        nan_value_counts: HashMap::new(),
1140                        lower_bounds: HashMap::new(),
1141                        upper_bounds: HashMap::new(),
1142                        key_metadata: None,
1143                        split_offsets: Some(vec![4]),
1144                        equality_ids: None,
1145                        sort_order_id: None,
1146                        partition_spec_id: 0,
1147                        first_row_id: None,
1148                        referenced_data_file: None,
1149                        content_offset: None,
1150                        content_size_in_bytes: None,
1151                    }
1152                },
1153        ];
1154
1155        // write manifest to file
1156        let tmp_dir = TempDir::new().unwrap();
1157        let path = tmp_dir.path().join("test_manifest.avro");
1158        let io = FileIO::new_with_fs();
1159        let output_file = io.new_output(path.to_str().unwrap()).unwrap();
1160        let mut writer = ManifestWriterBuilder::new(
1161            output_file,
1162            Some(1),
1163            metadata.schema.clone(),
1164            metadata.partition_spec.clone(),
1165        )
1166        .build_v2_data();
1167        for entry in &entries {
1168            writer.add_entry(entry.clone()).unwrap();
1169        }
1170        let res = writer.write_manifest_file().await.unwrap();
1171
1172        let partitions = res.partitions.unwrap();
1173
1174        assert_eq!(partitions.len(), 3);
1175        assert_eq!(
1176            partitions[0].clone().lower_bound.unwrap(),
1177            Datum::int(1111).to_bytes().unwrap()
1178        );
1179        assert_eq!(
1180            partitions[0].clone().upper_bound.unwrap(),
1181            Datum::int(2021).to_bytes().unwrap()
1182        );
1183        assert!(!partitions[0].clone().contains_null);
1184        assert_eq!(partitions[0].clone().contains_nan, Some(false));
1185
1186        assert_eq!(
1187            partitions[1].clone().lower_bound.unwrap(),
1188            Datum::float(1.0).to_bytes().unwrap()
1189        );
1190        assert_eq!(
1191            partitions[1].clone().upper_bound.unwrap(),
1192            Datum::float(15.5).to_bytes().unwrap()
1193        );
1194        assert!(partitions[1].clone().contains_null);
1195        assert_eq!(partitions[1].clone().contains_nan, Some(true));
1196
1197        assert_eq!(
1198            partitions[2].clone().lower_bound.unwrap(),
1199            Datum::double(1.0).to_bytes().unwrap()
1200        );
1201        assert_eq!(
1202            partitions[2].clone().upper_bound.unwrap(),
1203            Datum::double(25.5).to_bytes().unwrap()
1204        );
1205        assert!(!partitions[2].clone().contains_null);
1206        assert_eq!(partitions[2].clone().contains_nan, Some(false));
1207    }
1208
1209    #[test]
1210    fn test_data_file_serialization() {
1211        // Create a simple schema
1212        let schema = Schema::builder()
1213            .with_schema_id(1)
1214            .with_identifier_field_ids(vec![1])
1215            .with_fields(vec![
1216                NestedField::required(1, "id", Type::Primitive(PrimitiveType::Long)).into(),
1217                NestedField::required(2, "name", Type::Primitive(PrimitiveType::String)).into(),
1218            ])
1219            .build()
1220            .unwrap();
1221
1222        // Create a partition spec
1223        let partition_spec = PartitionSpec::builder(schema.clone())
1224            .with_spec_id(1)
1225            .add_partition_field("id", "id_partition", Transform::Identity)
1226            .unwrap()
1227            .build()
1228            .unwrap();
1229
1230        // Get partition type from the partition spec
1231        let partition_type = partition_spec.partition_type(&schema).unwrap();
1232
1233        // Create a vector of DataFile objects
1234        let data_files = vec![
1235            DataFileBuilder::default()
1236                .content(DataContentType::Data)
1237                .file_format(DataFileFormat::Parquet)
1238                .file_path("path/to/file1.parquet".to_string())
1239                .file_size_in_bytes(1024)
1240                .record_count(100)
1241                .partition_spec_id(1)
1242                .partition(Struct::empty())
1243                .column_sizes(HashMap::from([(1, 512), (2, 1024)]))
1244                .value_counts(HashMap::from([(1, 100), (2, 500)]))
1245                .null_value_counts(HashMap::from([(1, 0), (2, 1)]))
1246                .build()
1247                .unwrap(),
1248            DataFileBuilder::default()
1249                .content(DataContentType::Data)
1250                .file_format(DataFileFormat::Parquet)
1251                .file_path("path/to/file2.parquet".to_string())
1252                .file_size_in_bytes(2048)
1253                .record_count(200)
1254                .partition_spec_id(1)
1255                .partition(Struct::empty())
1256                .column_sizes(HashMap::from([(1, 1024), (2, 2048)]))
1257                .value_counts(HashMap::from([(1, 200), (2, 600)]))
1258                .null_value_counts(HashMap::from([(1, 10), (2, 999)]))
1259                .build()
1260                .unwrap(),
1261        ];
1262
1263        // Serialize the DataFile objects
1264        let serialized_files = data_files
1265            .clone()
1266            .into_iter()
1267            .map(|f| serialize_data_file_to_json(f, &partition_type, FormatVersion::V2).unwrap())
1268            .collect::<Vec<String>>();
1269
1270        // Verify we have the expected serialized files
1271        assert_eq!(serialized_files.len(), 2);
1272        let pretty_json1: Value = serde_json::from_str(serialized_files.first().unwrap()).unwrap();
1273        let pretty_json2: Value = serde_json::from_str(serialized_files.get(1).unwrap()).unwrap();
1274        let expected_serialized_file1 = serde_json::json!({
1275            "content": 0,
1276            "file_path": "path/to/file1.parquet",
1277            "file_format": "PARQUET",
1278            "partition": {},
1279            "record_count": 100,
1280            "file_size_in_bytes": 1024,
1281            "column_sizes": [
1282                { "key": 1, "value": 512 },
1283                { "key": 2, "value": 1024 }
1284            ],
1285            "value_counts": [
1286                { "key": 1, "value": 100 },
1287                { "key": 2, "value": 500 }
1288            ],
1289            "null_value_counts": [
1290                { "key": 1, "value": 0 },
1291                { "key": 2, "value": 1 }
1292            ],
1293            "nan_value_counts": [],
1294            "lower_bounds": [],
1295            "upper_bounds": [],
1296            "key_metadata": null,
1297            "split_offsets": null,
1298            "equality_ids": null,
1299            "sort_order_id": null,
1300            "first_row_id": null,
1301            "referenced_data_file": null,
1302            "content_offset": null,
1303            "content_size_in_bytes": null
1304        });
1305        let expected_serialized_file2 = serde_json::json!({
1306            "content": 0,
1307            "file_path": "path/to/file2.parquet",
1308            "file_format": "PARQUET",
1309            "partition": {},
1310            "record_count": 200,
1311            "file_size_in_bytes": 2048,
1312            "column_sizes": [
1313                { "key": 1, "value": 1024 },
1314                { "key": 2, "value": 2048 }
1315            ],
1316            "value_counts": [
1317                { "key": 1, "value": 200 },
1318                { "key": 2, "value": 600 }
1319            ],
1320            "null_value_counts": [
1321                { "key": 1, "value": 10 },
1322                { "key": 2, "value": 999 }
1323            ],
1324            "nan_value_counts": [],
1325            "lower_bounds": [],
1326            "upper_bounds": [],
1327            "key_metadata": null,
1328            "split_offsets": null,
1329            "equality_ids": null,
1330            "sort_order_id": null,
1331            "first_row_id": null,
1332            "referenced_data_file": null,
1333            "content_offset": null,
1334            "content_size_in_bytes": null
1335        });
1336        assert_eq!(pretty_json1, expected_serialized_file1);
1337        assert_eq!(pretty_json2, expected_serialized_file2);
1338
1339        // Now deserialize the JSON strings back into DataFile objects
1340        let deserialized_files: Vec<DataFile> = serialized_files
1341            .into_iter()
1342            .map(|json| {
1343                deserialize_data_file_from_json(
1344                    &json,
1345                    partition_spec.spec_id(),
1346                    &partition_type,
1347                    &schema,
1348                )
1349                .unwrap()
1350            })
1351            .collect();
1352
1353        // Verify we have the expected number of deserialized files
1354        assert_eq!(deserialized_files.len(), 2);
1355        let deserialized_data_file1 = deserialized_files.first().unwrap();
1356        let deserialized_data_file2 = deserialized_files.get(1).unwrap();
1357        let original_data_file1 = data_files.first().unwrap();
1358        let original_data_file2 = data_files.get(1).unwrap();
1359
1360        assert_eq!(deserialized_data_file1, original_data_file1);
1361        assert_eq!(deserialized_data_file2, original_data_file2);
1362    }
1363}
iceberg/spec/manifest/mod.rs

iceberg/spec/manifest/
mod.rs
