iceberg/spec/manifest/
writer.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
18use std::cmp::min;
19use std::future::Future;
20use std::pin::Pin;
21
22use apache_avro::{Writer as AvroWriter, to_value};
23use bytes::Bytes;
24use itertools::Itertools;
25use serde_json::to_vec;
26
27use super::{
28    Datum, FormatVersion, ManifestContentType, PartitionSpec, PrimitiveType,
29    UNASSIGNED_SEQUENCE_NUMBER,
30};
31use crate::encryption::EncryptedOutputFile;
32use crate::error::Result;
33use crate::io::{FileWrite, OutputFile};
34use crate::spec::manifest::_serde::{ManifestEntryV1, ManifestEntryV2};
35use crate::spec::manifest::{manifest_schema_v1, manifest_schema_v2};
36use crate::spec::{
37    DataContentType, DataFile, FieldSummary, ManifestEntry, ManifestFile, ManifestMetadata,
38    ManifestStatus, PrimitiveLiteral, SchemaRef, StructType, Type,
39};
40use crate::{Error, ErrorKind};
41
42/// Placeholder for snapshot ID. The field with this value must be replaced
43/// with the actual snapshot ID before it is committed.
44const UNASSIGNED_SNAPSHOT_ID: i64 = -1;
45
46type WriterFuture = Pin<Box<dyn Future<Output = Result<Box<dyn FileWrite>>> + Send>>;
47
48/// The builder used to create a [`ManifestWriter`].
49pub struct ManifestWriterBuilder {
50    writer_future: WriterFuture,
51    location: String,
52    snapshot_id: Option<i64>,
53    key_metadata: Option<Vec<u8>>,
54    schema: SchemaRef,
55    partition_spec: PartitionSpec,
56}
57
58impl ManifestWriterBuilder {
59    /// Create a new builder for unencrypted manifests.
60    pub fn new(
61        output: OutputFile,
62        snapshot_id: Option<i64>,
63        schema: SchemaRef,
64        partition_spec: PartitionSpec,
65    ) -> Self {
66        let location = output.location().to_owned();
67        Self {
68            writer_future: Box::pin(async move { output.writer().await }),
69            location,
70            snapshot_id,
71            key_metadata: None,
72            schema,
73            partition_spec,
74        }
75    }
76
77    /// Create a new builder from an [`EncryptedOutputFile`].
78    ///
79    /// Use this when writing manifests with transparent encryption.
80    pub fn new_from_encrypted(
81        encrypted_output: EncryptedOutputFile,
82        snapshot_id: Option<i64>,
83        schema: SchemaRef,
84        partition_spec: PartitionSpec,
85    ) -> Result<Self> {
86        let location = encrypted_output.location().to_owned();
87        let key_metadata = Some(encrypted_output.key_metadata().encode()?.to_vec());
88        Ok(Self {
89            writer_future: Box::pin(async move { encrypted_output.writer().await }),
90            location,
91            snapshot_id,
92            key_metadata,
93            schema,
94            partition_spec,
95        })
96    }
97
98    /// Build a [`ManifestWriter`] for format version 1.
99    pub fn build_v1(self) -> ManifestWriter {
100        let metadata = ManifestMetadata::builder()
101            .schema_id(self.schema.schema_id())
102            .schema(self.schema)
103            .partition_spec(self.partition_spec)
104            .format_version(FormatVersion::V1)
105            .content(ManifestContentType::Data)
106            .build();
107        ManifestWriter::new(
108            self.writer_future,
109            self.location,
110            self.snapshot_id,
111            self.key_metadata,
112            metadata,
113            None,
114        )
115    }
116
117    /// Build a [`ManifestWriter`] for format version 2, data content.
118    pub fn build_v2_data(self) -> ManifestWriter {
119        let metadata = ManifestMetadata::builder()
120            .schema_id(self.schema.schema_id())
121            .schema(self.schema)
122            .partition_spec(self.partition_spec)
123            .format_version(FormatVersion::V2)
124            .content(ManifestContentType::Data)
125            .build();
126
127        ManifestWriter::new(
128            self.writer_future,
129            self.location,
130            self.snapshot_id,
131            self.key_metadata,
132            metadata,
133            None,
134        )
135    }
136
137    /// Build a [`ManifestWriter`] for format version 2, deletes content.
138    pub fn build_v2_deletes(self) -> ManifestWriter {
139        let metadata = ManifestMetadata::builder()
140            .schema_id(self.schema.schema_id())
141            .schema(self.schema)
142            .partition_spec(self.partition_spec)
143            .format_version(FormatVersion::V2)
144            .content(ManifestContentType::Deletes)
145            .build();
146        ManifestWriter::new(
147            self.writer_future,
148            self.location,
149            self.snapshot_id,
150            self.key_metadata,
151            metadata,
152            None,
153        )
154    }
155
156    /// Build a [`ManifestWriter`] for format version 2, data content.
157    pub fn build_v3_data(self) -> ManifestWriter {
158        let metadata = ManifestMetadata::builder()
159            .schema_id(self.schema.schema_id())
160            .schema(self.schema)
161            .partition_spec(self.partition_spec)
162            .format_version(FormatVersion::V3)
163            .content(ManifestContentType::Data)
164            .build();
165        ManifestWriter::new(
166            self.writer_future,
167            self.location,
168            self.snapshot_id,
169            self.key_metadata,
170            metadata,
171            // First row id is assigned by the [`ManifestListWriter`] when the manifest
172            // is added to the list.
173            None,
174        )
175    }
176
177    /// Build a [`ManifestWriter`] for format version 3, deletes content.
178    pub fn build_v3_deletes(self) -> ManifestWriter {
179        let metadata = ManifestMetadata::builder()
180            .schema_id(self.schema.schema_id())
181            .schema(self.schema)
182            .partition_spec(self.partition_spec)
183            .format_version(FormatVersion::V3)
184            .content(ManifestContentType::Deletes)
185            .build();
186        ManifestWriter::new(
187            self.writer_future,
188            self.location,
189            self.snapshot_id,
190            self.key_metadata,
191            metadata,
192            None,
193        )
194    }
195}
196
197/// A manifest writer.
198pub struct ManifestWriter {
199    writer_future: WriterFuture,
200    location: String,
201
202    snapshot_id: Option<i64>,
203
204    added_files: u32,
205    added_rows: u64,
206    existing_files: u32,
207    existing_rows: u64,
208    deleted_files: u32,
209    deleted_rows: u64,
210    first_row_id: Option<u64>,
211
212    min_seq_num: Option<i64>,
213
214    key_metadata: Option<Vec<u8>>,
215
216    manifest_entries: Vec<ManifestEntry>,
217
218    metadata: ManifestMetadata,
219}
220
221impl ManifestWriter {
222    /// Create a new manifest writer.
223    pub(crate) fn new(
224        writer_future: WriterFuture,
225        location: String,
226        snapshot_id: Option<i64>,
227        key_metadata: Option<Vec<u8>>,
228        metadata: ManifestMetadata,
229        first_row_id: Option<u64>,
230    ) -> Self {
231        Self {
232            writer_future,
233            location,
234            snapshot_id,
235            added_files: 0,
236            added_rows: 0,
237            existing_files: 0,
238            existing_rows: 0,
239            deleted_files: 0,
240            deleted_rows: 0,
241            first_row_id,
242            min_seq_num: None,
243            key_metadata,
244            manifest_entries: Vec::new(),
245            metadata,
246        }
247    }
248
249    fn construct_partition_summaries(
250        &mut self,
251        partition_type: &StructType,
252    ) -> Result<Vec<FieldSummary>> {
253        let mut field_stats: Vec<_> = partition_type
254            .fields()
255            .iter()
256            .map(|f| PartitionFieldStats::new(f.field_type.as_primitive_type().unwrap().clone()))
257            .collect();
258        for partition in self.manifest_entries.iter().map(|e| &e.data_file.partition) {
259            for (literal, stat) in partition.iter().zip_eq(field_stats.iter_mut()) {
260                let primitive_literal = literal.map(|v| v.as_primitive_literal().unwrap());
261                stat.update(primitive_literal)?;
262            }
263        }
264        Ok(field_stats.into_iter().map(|stat| stat.finish()).collect())
265    }
266
267    fn check_data_file(&self, data_file: &DataFile) -> Result<()> {
268        match self.metadata.content {
269            ManifestContentType::Data => {
270                if data_file.content != DataContentType::Data {
271                    return Err(Error::new(
272                        ErrorKind::DataInvalid,
273                        format!(
274                            "Date file at path {} with manifest content type `data`, should have DataContentType `Data`, but has `{:?}`",
275                            data_file.file_path(),
276                            data_file.content
277                        ),
278                    ));
279                }
280            }
281            ManifestContentType::Deletes => {
282                if data_file.content != DataContentType::EqualityDeletes
283                    && data_file.content != DataContentType::PositionDeletes
284                {
285                    return Err(Error::new(
286                        ErrorKind::DataInvalid,
287                        format!(
288                            "Date file at path {} with manifest content type `deletes`, should have DataContentType `Data`, but has `{:?}`",
289                            data_file.file_path(),
290                            data_file.content
291                        ),
292                    ));
293                }
294            }
295        }
296        Ok(())
297    }
298
299    /// Add a new manifest entry. This method will update following status of the entry:
300    /// - Update the entry status to `Added`
301    /// - Set the snapshot id to the current snapshot id
302    /// - Set the sequence number to `None` if it is invalid(smaller than 0)
303    /// - Set the file sequence number to `None`
304    pub(crate) fn add_entry(&mut self, mut entry: ManifestEntry) -> Result<()> {
305        self.check_data_file(&entry.data_file)?;
306        if entry.sequence_number().is_some_and(|n| n >= 0) {
307            entry.status = ManifestStatus::Added;
308            entry.snapshot_id = self.snapshot_id;
309            entry.file_sequence_number = None;
310        } else {
311            entry.status = ManifestStatus::Added;
312            entry.snapshot_id = self.snapshot_id;
313            entry.sequence_number = None;
314            entry.file_sequence_number = None;
315        };
316        self.add_entry_inner(entry)?;
317        Ok(())
318    }
319
320    /// Add file as an added entry with a specific sequence number. The entry's snapshot ID will be this manifest's snapshot ID. The entry's data sequence
321    /// number will be the provided data sequence number. The entry's file sequence number will be
322    /// assigned at commit.
323    pub fn add_file(&mut self, data_file: DataFile, sequence_number: i64) -> Result<()> {
324        self.check_data_file(&data_file)?;
325        let entry = ManifestEntry {
326            status: ManifestStatus::Added,
327            snapshot_id: self.snapshot_id,
328            sequence_number: (sequence_number >= 0).then_some(sequence_number),
329            file_sequence_number: None,
330            data_file,
331        };
332        self.add_entry_inner(entry)?;
333        Ok(())
334    }
335
336    /// Add a delete manifest entry. This method will update following status of the entry:
337    /// - Update the entry status to `Deleted`
338    /// - Set the snapshot id to the current snapshot id
339    ///
340    /// # TODO
341    /// Remove this allow later
342    #[allow(dead_code)]
343    pub(crate) fn add_delete_entry(&mut self, mut entry: ManifestEntry) -> Result<()> {
344        self.check_data_file(&entry.data_file)?;
345        entry.status = ManifestStatus::Deleted;
346        entry.snapshot_id = self.snapshot_id;
347        self.add_entry_inner(entry)?;
348        Ok(())
349    }
350
351    /// Add a file as delete manifest entry. The entry's snapshot ID will be this manifest's snapshot ID.
352    /// However, the original data and file sequence numbers of the file must be preserved when
353    /// the file is marked as deleted.
354    pub fn add_delete_file(
355        &mut self,
356        data_file: DataFile,
357        sequence_number: i64,
358        file_sequence_number: Option<i64>,
359    ) -> Result<()> {
360        self.check_data_file(&data_file)?;
361        let entry = ManifestEntry {
362            status: ManifestStatus::Deleted,
363            snapshot_id: self.snapshot_id,
364            sequence_number: Some(sequence_number),
365            file_sequence_number,
366            data_file,
367        };
368        self.add_entry_inner(entry)?;
369        Ok(())
370    }
371
372    /// Add an existing manifest entry. This method will update following status of the entry:
373    /// - Update the entry status to `Existing`
374    ///
375    /// # TODO
376    /// Remove this allow later
377    #[allow(dead_code)]
378    pub(crate) fn add_existing_entry(&mut self, mut entry: ManifestEntry) -> Result<()> {
379        self.check_data_file(&entry.data_file)?;
380        entry.status = ManifestStatus::Existing;
381        self.add_entry_inner(entry)?;
382        Ok(())
383    }
384
385    /// Add an file as existing manifest entry. The original data and file sequence numbers, snapshot ID,
386    /// which were assigned at commit, must be preserved when adding an existing entry.
387    pub fn add_existing_file(
388        &mut self,
389        data_file: DataFile,
390        snapshot_id: i64,
391        sequence_number: i64,
392        file_sequence_number: Option<i64>,
393    ) -> Result<()> {
394        self.check_data_file(&data_file)?;
395        let entry = ManifestEntry {
396            status: ManifestStatus::Existing,
397            snapshot_id: Some(snapshot_id),
398            sequence_number: Some(sequence_number),
399            file_sequence_number,
400            data_file,
401        };
402        self.add_entry_inner(entry)?;
403        Ok(())
404    }
405
406    fn add_entry_inner(&mut self, entry: ManifestEntry) -> Result<()> {
407        // Check if the entry has sequence number
408        if (entry.status == ManifestStatus::Deleted || entry.status == ManifestStatus::Existing)
409            && (entry.sequence_number.is_none() || entry.file_sequence_number.is_none())
410        {
411            return Err(Error::new(
412                ErrorKind::DataInvalid,
413                "Manifest entry with status Existing or Deleted should have sequence number",
414            ));
415        }
416
417        // Update the statistics
418        match entry.status {
419            ManifestStatus::Added => {
420                self.added_files += 1;
421                self.added_rows += entry.data_file.record_count;
422            }
423            ManifestStatus::Deleted => {
424                self.deleted_files += 1;
425                self.deleted_rows += entry.data_file.record_count;
426            }
427            ManifestStatus::Existing => {
428                self.existing_files += 1;
429                self.existing_rows += entry.data_file.record_count;
430            }
431        }
432        if entry.is_alive()
433            && let Some(seq_num) = entry.sequence_number
434        {
435            self.min_seq_num = Some(self.min_seq_num.map_or(seq_num, |v| min(v, seq_num)));
436        }
437        self.manifest_entries.push(entry);
438        Ok(())
439    }
440
441    /// Write manifest file and return it.
442    pub async fn write_manifest_file(mut self) -> Result<ManifestFile> {
443        // Create the avro writer
444        let partition_type = self
445            .metadata
446            .partition_spec
447            .partition_type(&self.metadata.schema)?;
448        // Wrap once and reuse for every entry so the per-entry conversion does
449        // not rebuild the partition type's field-name lookup on each call.
450        let partition_struct_type = Type::Struct(partition_type.clone());
451        let table_schema = &self.metadata.schema;
452        let avro_schema = match self.metadata.format_version {
453            FormatVersion::V1 => manifest_schema_v1(&partition_type)?,
454            // Manifest schema did not change between V2 and V3
455            FormatVersion::V2 | FormatVersion::V3 => manifest_schema_v2(&partition_type)?,
456        };
457        let mut avro_writer = AvroWriter::new(&avro_schema, Vec::new());
458        avro_writer.add_user_metadata(
459            "schema".to_string(),
460            to_vec(table_schema).map_err(|err| {
461                Error::new(ErrorKind::DataInvalid, "Fail to serialize table schema")
462                    .with_source(err)
463            })?,
464        )?;
465        avro_writer.add_user_metadata(
466            "schema-id".to_string(),
467            table_schema.schema_id().to_string(),
468        )?;
469        avro_writer.add_user_metadata(
470            "partition-spec".to_string(),
471            to_vec(&self.metadata.partition_spec.fields()).map_err(|err| {
472                Error::new(ErrorKind::DataInvalid, "Fail to serialize partition spec")
473                    .with_source(err)
474            })?,
475        )?;
476        avro_writer.add_user_metadata(
477            "partition-spec-id".to_string(),
478            self.metadata.partition_spec.spec_id().to_string(),
479        )?;
480        avro_writer.add_user_metadata(
481            "format-version".to_string(),
482            (self.metadata.format_version as u8).to_string(),
483        )?;
484        match self.metadata.format_version {
485            FormatVersion::V1 => {}
486            FormatVersion::V2 | FormatVersion::V3 => {
487                avro_writer
488                    .add_user_metadata("content".to_string(), self.metadata.content.to_string())?;
489            }
490        }
491
492        let partition_summary = self.construct_partition_summaries(&partition_type)?;
493        // Write manifest entries
494        for entry in std::mem::take(&mut self.manifest_entries) {
495            let value = match self.metadata.format_version {
496                FormatVersion::V1 => {
497                    to_value(ManifestEntryV1::try_from(entry, &partition_struct_type)?)?
498                        .resolve(&avro_schema)?
499                }
500                // Manifest entry format did not change between V2 and V3
501                FormatVersion::V2 | FormatVersion::V3 => {
502                    to_value(ManifestEntryV2::try_from(entry, &partition_struct_type)?)?
503                        .resolve(&avro_schema)?
504                }
505            };
506
507            avro_writer.append(value)?;
508        }
509
510        let content = avro_writer.into_inner()?;
511        let length = content.len();
512        let mut writer = self.writer_future.await?;
513        writer.write(Bytes::from(content)).await?;
514        writer.close().await?;
515
516        Ok(ManifestFile {
517            manifest_path: self.location,
518            manifest_length: length as i64,
519            partition_spec_id: self.metadata.partition_spec.spec_id(),
520            content: self.metadata.content,
521            // sequence_number and min_sequence_number with UNASSIGNED_SEQUENCE_NUMBER will be replace with
522            // real sequence number in `ManifestListWriter`.
523            sequence_number: UNASSIGNED_SEQUENCE_NUMBER,
524            min_sequence_number: self.min_seq_num.unwrap_or(UNASSIGNED_SEQUENCE_NUMBER),
525            added_snapshot_id: self.snapshot_id.unwrap_or(UNASSIGNED_SNAPSHOT_ID),
526            added_files_count: Some(self.added_files),
527            existing_files_count: Some(self.existing_files),
528            deleted_files_count: Some(self.deleted_files),
529            added_rows_count: Some(self.added_rows),
530            existing_rows_count: Some(self.existing_rows),
531            deleted_rows_count: Some(self.deleted_rows),
532            partitions: Some(partition_summary),
533            key_metadata: self.key_metadata,
534            first_row_id: self.first_row_id,
535        })
536    }
537}
538
539struct PartitionFieldStats {
540    partition_type: PrimitiveType,
541
542    contains_null: bool,
543    contains_nan: Option<bool>,
544    lower_bound: Option<Datum>,
545    upper_bound: Option<Datum>,
546}
547
548impl PartitionFieldStats {
549    pub(crate) fn new(partition_type: PrimitiveType) -> Self {
550        Self {
551            partition_type,
552            contains_null: false,
553            contains_nan: Some(false),
554            upper_bound: None,
555            lower_bound: None,
556        }
557    }
558
559    pub(crate) fn update(&mut self, value: Option<PrimitiveLiteral>) -> Result<()> {
560        let Some(value) = value else {
561            self.contains_null = true;
562            return Ok(());
563        };
564        if !self.partition_type.compatible(&value) {
565            return Err(Error::new(
566                ErrorKind::DataInvalid,
567                "value is not compatible with type",
568            ));
569        }
570        let value = Datum::new(self.partition_type.clone(), value);
571
572        if value.is_nan() {
573            self.contains_nan = Some(true);
574            return Ok(());
575        }
576
577        self.lower_bound = Some(self.lower_bound.take().map_or(value.clone(), |original| {
578            if value < original {
579                value.clone()
580            } else {
581                original
582            }
583        }));
584        self.upper_bound = Some(self.upper_bound.take().map_or(value.clone(), |original| {
585            if value > original { value } else { original }
586        }));
587
588        Ok(())
589    }
590
591    pub(crate) fn finish(self) -> FieldSummary {
592        FieldSummary {
593            contains_null: self.contains_null,
594            contains_nan: self.contains_nan,
595            upper_bound: self.upper_bound.map(|v| v.to_bytes().unwrap()),
596            lower_bound: self.lower_bound.map(|v| v.to_bytes().unwrap()),
597        }
598    }
599}
600
601#[cfg(test)]
602mod tests {
603    use std::collections::HashMap;
604    use std::fs;
605    use std::sync::Arc;
606
607    use tempfile::TempDir;
608
609    use super::*;
610    use crate::io::FileIO;
611    use crate::spec::{DataFileFormat, Manifest, NestedField, PrimitiveType, Schema, Struct, Type};
612
613    #[tokio::test]
614    async fn test_add_delete_existing() {
615        let schema = Arc::new(
616            Schema::builder()
617                .with_fields(vec![
618                    Arc::new(NestedField::optional(
619                        1,
620                        "id",
621                        Type::Primitive(PrimitiveType::Int),
622                    )),
623                    Arc::new(NestedField::optional(
624                        2,
625                        "name",
626                        Type::Primitive(PrimitiveType::String),
627                    )),
628                ])
629                .build()
630                .unwrap(),
631        );
632        let metadata = ManifestMetadata {
633            schema_id: 0,
634            schema: schema.clone(),
635            partition_spec: PartitionSpec::builder(schema)
636                .with_spec_id(0)
637                .build()
638                .unwrap(),
639            content: ManifestContentType::Data,
640            format_version: FormatVersion::V2,
641        };
642        let mut entries = vec![
643                ManifestEntry {
644                    status: ManifestStatus::Added,
645                    snapshot_id: None,
646                    sequence_number: Some(1),
647                    file_sequence_number: Some(1),
648                    data_file: DataFile {
649                        content: DataContentType::Data,
650                        file_path: "s3a://icebergdata/demo/s1/t1/data/00000-0-ba56fbfa-f2ff-40c9-bb27-565ad6dc2be8-00000.parquet".to_string(),
651                        file_format: DataFileFormat::Parquet,
652                        partition: Struct::empty(),
653                        record_count: 1,
654                        file_size_in_bytes: 5442,
655                        column_sizes: HashMap::from([(1, 61), (2, 73)]),
656                        value_counts: HashMap::from([(1, 1), (2, 1)]),
657                        null_value_counts: HashMap::from([(1, 0), (2, 0)]),
658                        nan_value_counts: HashMap::new(),
659                        lower_bounds: HashMap::new(),
660                        upper_bounds: HashMap::new(),
661                        key_metadata: Some(Vec::new()),
662                        split_offsets: Some(vec![4]),
663                        equality_ids: None,
664                        sort_order_id: None,
665                        partition_spec_id: 0,
666                        first_row_id: None,
667                        referenced_data_file: None,
668                        content_offset: None,
669                        content_size_in_bytes: None,
670                    },
671                },
672                ManifestEntry {
673                    status: ManifestStatus::Deleted,
674                    snapshot_id: Some(1),
675                    sequence_number: Some(1),
676                    file_sequence_number: Some(1),
677                    data_file: DataFile {
678                        content: DataContentType::Data,
679                        file_path: "s3a://icebergdata/demo/s1/t1/data/00000-0-ba56fbfa-f2ff-40c9-bb27-565ad6dc2be8-00000.parquet".to_string(),
680                        file_format: DataFileFormat::Parquet,
681                        partition: Struct::empty(),
682                        record_count: 1,
683                        file_size_in_bytes: 5442,
684                        column_sizes: HashMap::from([(1, 61), (2, 73)]),
685                        value_counts: HashMap::from([(1, 1), (2, 1)]),
686                        null_value_counts: HashMap::from([(1, 0), (2, 0)]),
687                        nan_value_counts: HashMap::new(),
688                        lower_bounds: HashMap::new(),
689                        upper_bounds: HashMap::new(),
690                        key_metadata: Some(Vec::new()),
691                        split_offsets: Some(vec![4]),
692                        equality_ids: None,
693                        sort_order_id: None,
694                        partition_spec_id: 0,
695                        first_row_id: None,
696                        referenced_data_file: None,
697                        content_offset: None,
698                        content_size_in_bytes: None,
699                    },
700                },
701                ManifestEntry {
702                    status: ManifestStatus::Existing,
703                    snapshot_id: Some(1),
704                    sequence_number: Some(1),
705                    file_sequence_number: Some(1),
706                    data_file: DataFile {
707                        content: DataContentType::Data,
708                        file_path: "s3a://icebergdata/demo/s1/t1/data/00000-0-ba56fbfa-f2ff-40c9-bb27-565ad6dc2be8-00000.parquet".to_string(),
709                        file_format: DataFileFormat::Parquet,
710                        partition: Struct::empty(),
711                        record_count: 1,
712                        file_size_in_bytes: 5442,
713                        column_sizes: HashMap::from([(1, 61), (2, 73)]),
714                        value_counts: HashMap::from([(1, 1), (2, 1)]),
715                        null_value_counts: HashMap::from([(1, 0), (2, 0)]),
716                        nan_value_counts: HashMap::new(),
717                        lower_bounds: HashMap::new(),
718                        upper_bounds: HashMap::new(),
719                        key_metadata: Some(Vec::new()),
720                        split_offsets: Some(vec![4]),
721                        equality_ids: None,
722                        sort_order_id: None,
723                        partition_spec_id: 0,
724                        first_row_id: None,
725                        referenced_data_file: None,
726                        content_offset: None,
727                        content_size_in_bytes: None,
728                    },
729                },
730            ];
731
732        // write manifest to file
733        let tmp_dir = TempDir::new().unwrap();
734        let path = tmp_dir.path().join("test_manifest.avro");
735        let io = FileIO::new_with_fs();
736        let output_file = io.new_output(path.to_str().unwrap()).unwrap();
737        let mut writer = ManifestWriterBuilder::new(
738            output_file,
739            Some(3),
740            metadata.schema.clone(),
741            metadata.partition_spec.clone(),
742        )
743        .build_v2_data();
744        writer.add_entry(entries[0].clone()).unwrap();
745        writer.add_delete_entry(entries[1].clone()).unwrap();
746        writer.add_existing_entry(entries[2].clone()).unwrap();
747        writer.write_manifest_file().await.unwrap();
748
749        // read back the manifest file and check the content
750        let actual_manifest =
751            Manifest::parse_avro(fs::read(path).expect("read_file must succeed").as_slice())
752                .unwrap();
753
754        // The snapshot id is assigned when the entry is added and delete to the manifest. Existing entries are keep original.
755        entries[0].snapshot_id = Some(3);
756        entries[1].snapshot_id = Some(3);
757        // file sequence number is assigned to None when the entry is added and delete to the manifest.
758        entries[0].file_sequence_number = None;
759        assert_eq!(actual_manifest, Manifest::new(metadata, entries));
760    }
761
762    #[tokio::test]
763    async fn test_v3_delete_manifest_delete_file_roundtrip() {
764        let schema = Arc::new(
765            Schema::builder()
766                .with_fields(vec![
767                    Arc::new(NestedField::optional(
768                        1,
769                        "id",
770                        Type::Primitive(PrimitiveType::Long),
771                    )),
772                    Arc::new(NestedField::optional(
773                        2,
774                        "data",
775                        Type::Primitive(PrimitiveType::String),
776                    )),
777                ])
778                .build()
779                .unwrap(),
780        );
781
782        let partition_spec = PartitionSpec::builder(schema.clone())
783            .with_spec_id(0)
784            .build()
785            .unwrap();
786
787        // Create a position delete file entry
788        let delete_entry = ManifestEntry {
789            status: ManifestStatus::Added,
790            snapshot_id: None,
791            sequence_number: None,
792            file_sequence_number: None,
793            data_file: DataFile {
794                content: DataContentType::PositionDeletes,
795                file_path: "s3://bucket/table/data/delete-00000.parquet".to_string(),
796                file_format: DataFileFormat::Parquet,
797                partition: Struct::empty(),
798                record_count: 10,
799                file_size_in_bytes: 1024,
800                column_sizes: HashMap::new(),
801                value_counts: HashMap::new(),
802                null_value_counts: HashMap::new(),
803                nan_value_counts: HashMap::new(),
804                lower_bounds: HashMap::new(),
805                upper_bounds: HashMap::new(),
806                key_metadata: None,
807                split_offsets: None,
808                equality_ids: None,
809                sort_order_id: None,
810                partition_spec_id: 0,
811                first_row_id: None,
812                referenced_data_file: None,
813                content_offset: None,
814                content_size_in_bytes: None,
815            },
816        };
817
818        // Write a V3 delete manifest
819        let tmp_dir = TempDir::new().unwrap();
820        let path = tmp_dir.path().join("v3_delete_manifest.avro");
821        let io = FileIO::new_with_fs();
822        let output_file = io.new_output(path.to_str().unwrap()).unwrap();
823
824        let mut writer = ManifestWriterBuilder::new(
825            output_file,
826            Some(1),
827            schema.clone(),
828            partition_spec.clone(),
829        )
830        .build_v3_deletes();
831
832        writer.add_entry(delete_entry).unwrap();
833        let manifest_file = writer.write_manifest_file().await.unwrap();
834
835        // The returned ManifestFile correctly reports Deletes content
836        assert_eq!(manifest_file.content, ManifestContentType::Deletes);
837
838        // Read back the manifest file
839        let actual_manifest =
840            Manifest::parse_avro(fs::read(&path).expect("read_file must succeed").as_slice())
841                .unwrap();
842
843        // Verify the content type is correctly preserved as Deletes
844        assert_eq!(
845            actual_manifest.metadata().content,
846            ManifestContentType::Deletes,
847        );
848    }
849}
iceberg/spec/manifest/writer.rs

iceberg/spec/manifest/
writer.rs
