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

iceberg/spec/manifest/
mod.rs
