iceberg/spec/manifest_list/
manifest_file.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::str::FromStr;
19
20use serde_derive::{Deserialize, Serialize};
21
22use super::ByteBuf;
23use crate::encryption::{EncryptedInputFile, StandardKeyMetadata};
24use crate::error::Result;
25use crate::io::FileIO;
26use crate::spec::{Manifest, ManifestEntry};
27use crate::{Error, ErrorKind};
28
29/// Entry in a manifest list.
30#[derive(Debug, PartialEq, Clone, Eq, Hash)]
31pub struct ManifestFile {
32    /// field: 500
33    ///
34    /// Location of the manifest file
35    pub manifest_path: String,
36    /// field: 501
37    ///
38    /// Length of the manifest file in bytes
39    pub manifest_length: i64,
40    /// field: 502
41    ///
42    /// ID of a partition spec used to write the manifest; must be listed
43    /// in table metadata partition-specs
44    pub partition_spec_id: i32,
45    /// field: 517
46    ///
47    /// The type of files tracked by the manifest, either data or delete
48    /// files; 0 for all v1 manifests
49    pub content: ManifestContentType,
50    /// field: 515
51    ///
52    /// The sequence number when the manifest was added to the table; use 0
53    /// when reading v1 manifest lists
54    pub sequence_number: i64,
55    /// field: 516
56    ///
57    /// The minimum data sequence number of all live data or delete files in
58    /// the manifest; use 0 when reading v1 manifest lists
59    pub min_sequence_number: i64,
60    /// field: 503
61    ///
62    /// ID of the snapshot where the manifest file was added
63    pub added_snapshot_id: i64,
64    /// field: 504
65    ///
66    /// Number of entries in the manifest that have status ADDED, when null
67    /// this is assumed to be non-zero
68    pub added_files_count: Option<u32>,
69    /// field: 505
70    ///
71    /// Number of entries in the manifest that have status EXISTING (0),
72    /// when null this is assumed to be non-zero
73    pub existing_files_count: Option<u32>,
74    /// field: 506
75    ///
76    /// Number of entries in the manifest that have status DELETED (2),
77    /// when null this is assumed to be non-zero
78    pub deleted_files_count: Option<u32>,
79    /// field: 512
80    ///
81    /// Number of rows in all of files in the manifest that have status
82    /// ADDED, when null this is assumed to be non-zero
83    pub added_rows_count: Option<u64>,
84    /// field: 513
85    ///
86    /// Number of rows in all of files in the manifest that have status
87    /// EXISTING, when null this is assumed to be non-zero
88    pub existing_rows_count: Option<u64>,
89    /// field: 514
90    ///
91    /// Number of rows in all of files in the manifest that have status
92    /// DELETED, when null this is assumed to be non-zero
93    pub deleted_rows_count: Option<u64>,
94    /// field: 507
95    /// element_field: 508
96    ///
97    /// A list of field summaries for each partition field in the spec. Each
98    /// field in the list corresponds to a field in the manifest file’s
99    /// partition spec.
100    pub partitions: Option<Vec<FieldSummary>>,
101    /// field: 519
102    ///
103    /// Implementation-specific key metadata for encryption
104    pub key_metadata: Option<Vec<u8>>,
105    /// field 520
106    ///
107    /// The starting _row_id to assign to rows added by ADDED data files
108    pub first_row_id: Option<u64>,
109}
110
111impl ManifestFile {
112    /// Checks if the manifest file has any added files.
113    pub fn has_added_files(&self) -> bool {
114        self.added_files_count.map(|c| c > 0).unwrap_or(true)
115    }
116
117    /// Checks whether this manifest contains entries with DELETED status.
118    pub fn has_deleted_files(&self) -> bool {
119        self.deleted_files_count.map(|c| c > 0).unwrap_or(true)
120    }
121
122    /// Checks if the manifest file has any existed files.
123    pub fn has_existing_files(&self) -> bool {
124        self.existing_files_count.map(|c| c > 0).unwrap_or(true)
125    }
126}
127
128/// The type of files tracked by the manifest, either data or delete files; Data(0) for all v1 manifests
129#[derive(Debug, PartialEq, Clone, Copy, Eq, Hash, Default)]
130pub enum ManifestContentType {
131    /// The manifest content is data.
132    #[default]
133    Data = 0,
134    /// The manifest content is deletes.
135    Deletes = 1,
136}
137
138impl FromStr for ManifestContentType {
139    type Err = Error;
140
141    fn from_str(s: &str) -> Result<Self> {
142        match s {
143            "data" => Ok(ManifestContentType::Data),
144            "deletes" => Ok(ManifestContentType::Deletes),
145            _ => Err(Error::new(
146                ErrorKind::DataInvalid,
147                format!("Invalid manifest content type: {s}"),
148            )),
149        }
150    }
151}
152
153impl std::fmt::Display for ManifestContentType {
154    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
155        match self {
156            ManifestContentType::Data => write!(f, "data"),
157            ManifestContentType::Deletes => write!(f, "deletes"),
158        }
159    }
160}
161
162impl TryFrom<i32> for ManifestContentType {
163    type Error = Error;
164
165    fn try_from(value: i32) -> std::result::Result<Self, Self::Error> {
166        match value {
167            0 => Ok(ManifestContentType::Data),
168            1 => Ok(ManifestContentType::Deletes),
169            _ => Err(Error::new(
170                ErrorKind::DataInvalid,
171                format!("Invalid manifest content type. Expected 0 or 1, got {value}"),
172            )),
173        }
174    }
175}
176
177impl ManifestFile {
178    /// Load [`Manifest`].
179    ///
180    /// This method will also initialize inherited values of [`ManifestEntry`](crate::spec::ManifestEntry), such as `sequence_number`.
181    pub async fn load_manifest(&self, file_io: &FileIO) -> Result<Manifest> {
182        let input = file_io.new_input(&self.manifest_path)?;
183        let avro = match &self.key_metadata {
184            Some(key_metadata_bytes) => {
185                let key_metadata = StandardKeyMetadata::decode(key_metadata_bytes)?;
186                EncryptedInputFile::new(input, key_metadata).read().await?
187            }
188            None => input.read().await?,
189        };
190
191        let (metadata, mut entries) = Manifest::try_from_avro_bytes(&avro)?;
192
193        // Let entries inherit values from the manifest list entry.
194        for entry in &mut entries {
195            entry.inherit_data(self);
196        }
197
198        self.assign_first_row_ids(&mut entries)?;
199
200        Ok(Manifest::new(metadata, entries))
201    }
202
203    /// Assigns `first_row_id` to the live data-file entries, following the
204    /// row-lineage inheritance rules in
205    /// <https://github.com/apache/iceberg/blob/main/format/spec.md#first-row-id-inheritance>.
206    ///
207    /// With a manifest-level `first_row_id`, each live entry lacking one is
208    /// assigned the running id, which then advances by that entry's record
209    /// count; entries that already carry a `first_row_id` keep it and do not
210    /// advance the counter. Without a manifest-level `first_row_id`, any
211    /// inherited per-entry value is cleared so callers never observe a stale id.
212    fn assign_first_row_ids(&self, entries: &mut [ManifestEntry]) -> Result<()> {
213        // A `first_row_id` is only valid on data manifests. Delete files always
214        // have a null `first_row_id`, so there is nothing to assign or clear; a
215        // stray value on a delete manifest is a spec violation by the writer,
216        // which we surface without failing the read.
217        if self.content != ManifestContentType::Data {
218            if let Some(manifest_first_row_id) = self.first_row_id {
219                tracing::warn!(
220                    "Ignoring first_row_id {manifest_first_row_id} on delete manifest {}",
221                    self.manifest_path
222                );
223            }
224
225            return Ok(());
226        }
227
228        let Some(manifest_first_row_id) = self.first_row_id else {
229            // A data manifest with no manifest-level `first_row_id` predates row
230            // lineage; clear any per-entry value inherited from an earlier read.
231            for entry in entries {
232                entry.data_file.first_row_id = None;
233            }
234
235            return Ok(());
236        };
237
238        let mut next_row_id = i64::try_from(manifest_first_row_id).map_err(|_| {
239            Error::new(
240                ErrorKind::DataInvalid,
241                format!("Invalid first_row_id: {manifest_first_row_id} (exceeds i64::MAX)"),
242            )
243        })?;
244
245        for entry in entries {
246            if !entry.is_alive() {
247                continue;
248            }
249
250            if entry.data_file.first_row_id.is_none() {
251                let file_first_row_id = next_row_id;
252                entry.data_file.first_row_id = Some(file_first_row_id);
253                let record_count = entry.data_file.record_count;
254                next_row_id = file_first_row_id.checked_add_unsigned(record_count).ok_or_else(|| {
255                    Error::new(
256                        ErrorKind::DataInvalid,
257                        format!(
258                            "Row ID overflow assigning first_row_id in {}. File first_row_id: {file_first_row_id}, record count: {record_count}",
259                            self.manifest_path
260                        ),
261                    )
262                })?;
263            }
264        }
265
266        Ok(())
267    }
268}
269
270/// Field summary for partition field in the spec.
271///
272/// Each field in the list corresponds to a field in the manifest file’s partition spec.
273#[derive(Debug, Serialize, Deserialize, PartialEq, Eq, Clone, Default, Hash)]
274pub struct FieldSummary {
275    /// field: 509
276    ///
277    /// Whether the manifest contains at least one partition with a null
278    /// value for the field
279    pub contains_null: bool,
280    /// field: 518
281    /// Whether the manifest contains at least one partition with a NaN
282    /// value for the field
283    pub contains_nan: Option<bool>,
284    /// field: 510
285    /// The minimum value for the field in the manifests
286    /// partitions.
287    pub lower_bound: Option<ByteBuf>,
288    /// field: 511
289    /// The maximum value for the field in the manifests
290    /// partitions.
291    pub upper_bound: Option<ByteBuf>,
292}
293
294#[cfg(test)]
295mod test {
296    use std::sync::Arc;
297
298    use super::{ManifestContentType, ManifestFile};
299    use crate::ErrorKind;
300    use crate::encryption::{EncryptedOutputFile, StandardKeyMetadata};
301    use crate::io::FileIO;
302    use crate::spec::{
303        DataContentType, DataFileBuilder, DataFileFormat, ManifestEntry, ManifestStatus,
304        ManifestWriterBuilder, NestedField, PartitionSpec, PrimitiveType, Schema, SchemaRef, Type,
305    };
306
307    #[test]
308    fn test_manifest_content_type_default() {
309        assert_eq!(ManifestContentType::default(), ManifestContentType::Data);
310    }
311
312    #[test]
313    fn test_manifest_content_type_default_value() {
314        assert_eq!(ManifestContentType::default() as i32, 0);
315    }
316
317    /// A single-field schema used by the manifest-writing test helpers.
318    fn test_schema() -> SchemaRef {
319        Arc::new(
320            Schema::builder()
321                .with_fields(vec![Arc::new(NestedField::optional(
322                    1,
323                    "id",
324                    Type::Primitive(PrimitiveType::Long),
325                ))])
326                .build()
327                .unwrap(),
328        )
329    }
330
331    /// Writes a single-entry v3 data manifest to `io` at `path`, without
332    /// encryption, and returns the resulting [`ManifestFile`].
333    async fn write_manifest(io: &FileIO, path: &str) -> ManifestFile {
334        let schema = test_schema();
335        let partition_spec = PartitionSpec::builder(schema.clone())
336            .with_spec_id(0)
337            .build()
338            .unwrap();
339
340        let output_file = io.new_output(path).unwrap();
341        let mut writer = ManifestWriterBuilder::new(output_file, Some(1), schema, partition_spec)
342            .build_v3_data();
343
344        writer
345            .add_entry(data_entry(ManifestStatus::Added, 100, None))
346            .unwrap();
347
348        writer.write_manifest_file().await.unwrap()
349    }
350
351    /// Writes a single-entry v3 data manifest to `io` at `path`, encrypting it
352    /// with `key_metadata`, and returns the resulting [`ManifestFile`].
353    async fn write_encrypted_manifest(
354        io: &FileIO,
355        path: &str,
356        key_metadata: StandardKeyMetadata,
357    ) -> ManifestFile {
358        let schema = test_schema();
359        let partition_spec = PartitionSpec::builder(schema.clone())
360            .with_spec_id(0)
361            .build()
362            .unwrap();
363
364        let output_file = io.new_output(path).unwrap();
365        let encrypted_output = EncryptedOutputFile::new(output_file, key_metadata);
366
367        let mut writer = ManifestWriterBuilder::new_from_encrypted(
368            encrypted_output,
369            Some(1),
370            schema,
371            partition_spec,
372        )
373        .expect("Expected a valid writer")
374        .build_v3_data();
375
376        writer
377            .add_entry(data_entry(ManifestStatus::Added, 100, None))
378            .unwrap();
379
380        writer.write_manifest_file().await.unwrap()
381    }
382
383    #[tokio::test]
384    async fn test_load_manifest_decrypts_when_key_metadata_present() {
385        let key_metadata = StandardKeyMetadata::try_new(b"0123456789abcdef")
386            .unwrap()
387            .with_aad_prefix(b"test-aad-prefix!");
388        let encoded_key_metadata = key_metadata.encode().unwrap().to_vec();
389
390        let io = FileIO::new_with_memory();
391        let path = "memory:///test/encrypted_manifest.avro";
392        let manifest_file = write_encrypted_manifest(&io, path, key_metadata).await;
393        assert_eq!(manifest_file.key_metadata, Some(encoded_key_metadata));
394
395        let manifest = manifest_file.load_manifest(&io).await.unwrap();
396        assert_eq!(manifest.entries().len(), 1);
397        assert_eq!(
398            manifest.entries()[0].file_path(),
399            "s3://bucket/table/data/00000.parquet"
400        );
401        assert_eq!(manifest.entries()[0].data_file.record_count, 100);
402    }
403
404    #[tokio::test]
405    async fn test_load_manifest_fails_with_wrong_key() {
406        let key_metadata = StandardKeyMetadata::try_new(b"0123456789abcdef")
407            .unwrap()
408            .with_aad_prefix(b"test-aad-prefix!");
409
410        let io = FileIO::new_with_memory();
411        let path = "memory:///test/wrong_key_manifest.avro";
412        let mut manifest_file = write_encrypted_manifest(&io, path, key_metadata).await;
413
414        // Point the manifest file at key metadata carrying a different DEK (but
415        // the same AAD prefix). The bytes on disk were encrypted with the
416        // original key, so GCM authentication must fail rather than silently
417        // returning garbage.
418        let wrong_key_metadata = StandardKeyMetadata::try_new(b"fedcba9876543210")
419            .unwrap()
420            .with_aad_prefix(b"test-aad-prefix!");
421        manifest_file.key_metadata = Some(wrong_key_metadata.encode().unwrap().to_vec());
422
423        let err = manifest_file
424            .load_manifest(&io)
425            .await
426            .expect_err("load_manifest must fail when decrypting with the wrong key");
427        assert_eq!(err.kind(), ErrorKind::Unexpected);
428    }
429
430    #[tokio::test]
431    async fn test_load_manifest_fails_with_wrong_aad() {
432        let key_metadata = StandardKeyMetadata::try_new(b"0123456789abcdef")
433            .unwrap()
434            .with_aad_prefix(b"test-aad-prefix!");
435
436        let io = FileIO::new_with_memory();
437        let path = "memory:///test/wrong_aad_manifest.avro";
438        let mut manifest_file = write_encrypted_manifest(&io, path, key_metadata).await;
439
440        // Point the manifest file at key metadata carrying the correct DEK but a
441        // different AAD prefix. The per-block AAD is `aad_prefix || block_index`,
442        // so GCM authentication must fail even though the key is right.
443        let wrong_aad_metadata = StandardKeyMetadata::try_new(b"0123456789abcdef")
444            .unwrap()
445            .with_aad_prefix(b"wrong-aad-prefix");
446        manifest_file.key_metadata = Some(wrong_aad_metadata.encode().unwrap().to_vec());
447
448        let err = manifest_file
449            .load_manifest(&io)
450            .await
451            .expect_err("load_manifest must fail when decrypting with the wrong AAD prefix");
452        assert_eq!(err.kind(), ErrorKind::Unexpected);
453    }
454
455    /// Builds a data-file manifest entry with the given status, record count,
456    /// and pre-existing `first_row_id`.
457    fn data_entry(
458        status: ManifestStatus,
459        record_count: u64,
460        first_row_id: Option<i64>,
461    ) -> ManifestEntry {
462        let data_file = DataFileBuilder::default()
463            .content(DataContentType::Data)
464            .file_path("s3://bucket/table/data/00000.parquet".to_string())
465            .file_format(DataFileFormat::Parquet)
466            .file_size_in_bytes(4096)
467            .record_count(record_count)
468            .first_row_id(first_row_id)
469            .build()
470            .unwrap();
471
472        ManifestEntry::builder()
473            .status(status)
474            .data_file(data_file)
475            .build()
476    }
477
478    /// Builds a manifest file with the given content type and manifest-level
479    /// `first_row_id`. Other fields are irrelevant to row-id assignment.
480    fn manifest_file(content: ManifestContentType, first_row_id: Option<u64>) -> ManifestFile {
481        ManifestFile {
482            manifest_path: "memory:///m.avro".to_string(),
483            manifest_length: 0,
484            partition_spec_id: 0,
485            content,
486            sequence_number: 0,
487            min_sequence_number: 0,
488            added_snapshot_id: 0,
489            added_files_count: None,
490            existing_files_count: None,
491            deleted_files_count: None,
492            added_rows_count: None,
493            existing_rows_count: None,
494            deleted_rows_count: None,
495            partitions: None,
496            key_metadata: None,
497            first_row_id,
498        }
499    }
500
501    #[test]
502    fn test_assign_first_row_ids_interleaved() {
503        let manifest = manifest_file(ManifestContentType::Data, Some(10));
504        let mut entries = vec![
505            data_entry(ManifestStatus::Added, 3, None),
506            // A pre-assigned entry between two assigned ones: it keeps its id and
507            // must not advance the running counter.
508            data_entry(ManifestStatus::Added, 5, Some(100)),
509            // A deleted entry with a pre-set id: it is skipped, so the id is
510            // preserved verbatim and does not advance the counter.
511            data_entry(ManifestStatus::Deleted, 7, Some(999)),
512            data_entry(ManifestStatus::Existing, 2, None),
513        ];
514
515        manifest.assign_first_row_ids(&mut entries).unwrap();
516
517        assert_eq!(entries[0].data_file.first_row_id, Some(10));
518        assert_eq!(entries[1].data_file.first_row_id, Some(100));
519        assert_eq!(entries[2].data_file.first_row_id, Some(999));
520        // 10 + 3 = 13; the preserved and deleted entries in between do not move it.
521        assert_eq!(entries[3].data_file.first_row_id, Some(13));
522    }
523
524    #[test]
525    fn test_assign_first_row_ids_clears_without_manifest_first_row_id() {
526        // A data manifest with no manifest-level first_row_id predates row
527        // lineage: any per-entry value inherited from an earlier read is cleared
528        // so callers never observe a stale id.
529        let manifest = manifest_file(ManifestContentType::Data, None);
530        let mut entries = vec![
531            data_entry(ManifestStatus::Added, 3, None),
532            data_entry(ManifestStatus::Existing, 5, Some(100)),
533        ];
534
535        manifest.assign_first_row_ids(&mut entries).unwrap();
536
537        assert_eq!(entries[0].data_file.first_row_id, None);
538        assert_eq!(entries[1].data_file.first_row_id, None);
539    }
540
541    #[test]
542    fn test_assign_first_row_ids_ignores_delete_manifest() {
543        // A stray first_row_id on a delete manifest is a writer-side spec
544        // violation; the read ignores it rather than failing, and does not
545        // assign ids to the entries.
546        let manifest = manifest_file(ManifestContentType::Deletes, Some(10));
547        let mut entries = vec![data_entry(ManifestStatus::Added, 3, None)];
548
549        manifest.assign_first_row_ids(&mut entries).unwrap();
550
551        assert_eq!(entries[0].data_file.first_row_id, None);
552    }
553
554    #[test]
555    fn test_assign_first_row_ids_rejects_oversized_manifest_first_row_id() {
556        // A manifest-level first_row_id above i64::MAX cannot be represented as the
557        // signed running counter and must be rejected.
558        let manifest = manifest_file(ManifestContentType::Data, Some(i64::MAX as u64 + 1));
559        let mut entries = vec![data_entry(ManifestStatus::Added, 3, None)];
560
561        let err = manifest
562            .assign_first_row_ids(&mut entries)
563            .expect_err("an oversized manifest first_row_id must be rejected");
564        assert_eq!(err.kind(), ErrorKind::DataInvalid);
565        assert!(err.message().contains("Invalid first_row_id"));
566    }
567
568    #[test]
569    fn test_assign_first_row_ids_rejects_counter_overflow() {
570        // Advancing the running counter past i64::MAX must be rejected rather than
571        // wrapping to a negative value that would corrupt subsequent assignments.
572        let manifest = manifest_file(ManifestContentType::Data, Some(i64::MAX as u64));
573        let mut entries = vec![data_entry(ManifestStatus::Added, 1, None)];
574
575        let err = manifest
576            .assign_first_row_ids(&mut entries)
577            .expect_err("counter overflow past i64::MAX must be rejected");
578        assert_eq!(err.kind(), ErrorKind::DataInvalid);
579        assert!(err.message().contains("Row ID overflow"));
580    }
581
582    #[tokio::test]
583    async fn test_load_manifest_reads_written_entries() {
584        let io = FileIO::new_with_memory();
585        let path = "memory:///test/plaintext_manifest.avro";
586        let manifest_file = write_manifest(&io, path).await;
587        assert_eq!(manifest_file.key_metadata, None);
588
589        let manifest = manifest_file.load_manifest(&io).await.unwrap();
590        assert_eq!(manifest.entries().len(), 1);
591        assert_eq!(
592            manifest.entries()[0].file_path(),
593            "s3://bucket/table/data/00000.parquet"
594        );
595        assert_eq!(manifest.entries()[0].data_file.record_count, 100);
596    }
597
598    /// End-to-end: writing a v3 data manifest, stamping a manifest-level
599    /// `first_row_id`, and loading it must assign inherited `first_row_id`s to
600    /// the entries. This exercises the wiring in `load_manifest` and the
601    /// write/read round-trip that leaves per-file `first_row_id` as `None`.
602    #[tokio::test]
603    async fn test_load_manifest_assigns_first_row_ids() {
604        let io = FileIO::new_with_memory();
605        let path = "memory:///test/first_row_id_manifest.avro";
606        let mut manifest_file = write_manifest(&io, path).await;
607
608        // Stamp a manifest-level first_row_id, as the manifest-list writer would.
609        manifest_file.first_row_id = Some(1000);
610
611        let manifest = manifest_file.load_manifest(&io).await.unwrap();
612        assert_eq!(manifest.entries().len(), 1);
613        assert_eq!(manifest.entries()[0].data_file().first_row_id(), Some(1000));
614    }
615}
iceberg/spec/manifest_list/manifest_file.rs

iceberg/spec/manifest_list/
manifest_file.rs
