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;
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        Ok(Manifest::new(metadata, entries))
199    }
200}
201
202/// Field summary for partition field in the spec.
203///
204/// Each field in the list corresponds to a field in the manifest file’s partition spec.
205#[derive(Debug, Serialize, Deserialize, PartialEq, Eq, Clone, Default, Hash)]
206pub struct FieldSummary {
207    /// field: 509
208    ///
209    /// Whether the manifest contains at least one partition with a null
210    /// value for the field
211    pub contains_null: bool,
212    /// field: 518
213    /// Whether the manifest contains at least one partition with a NaN
214    /// value for the field
215    pub contains_nan: Option<bool>,
216    /// field: 510
217    /// The minimum value for the field in the manifests
218    /// partitions.
219    pub lower_bound: Option<ByteBuf>,
220    /// field: 511
221    /// The maximum value for the field in the manifests
222    /// partitions.
223    pub upper_bound: Option<ByteBuf>,
224}
225
226#[cfg(test)]
227mod test {
228    use std::collections::HashMap;
229    use std::sync::Arc;
230
231    use super::{ManifestContentType, ManifestFile};
232    use crate::ErrorKind;
233    use crate::encryption::{EncryptedOutputFile, StandardKeyMetadata};
234    use crate::io::FileIO;
235    use crate::spec::{
236        DataContentType, DataFile, DataFileFormat, ManifestEntry, ManifestStatus,
237        ManifestWriterBuilder, NestedField, PartitionSpec, PrimitiveType, Schema, Struct, Type,
238    };
239
240    #[test]
241    fn test_manifest_content_type_default() {
242        assert_eq!(ManifestContentType::default(), ManifestContentType::Data);
243    }
244
245    #[test]
246    fn test_manifest_content_type_default_value() {
247        assert_eq!(ManifestContentType::default() as i32, 0);
248    }
249
250    /// Writes a single-entry manifest, encrypting it with `key_metadata`, and
251    /// returns the resulting [`ManifestFile`]. The manifest is stored in `io`
252    /// at `path`.
253    async fn write_encrypted_manifest(
254        io: &FileIO,
255        path: &str,
256        key_metadata: StandardKeyMetadata,
257    ) -> ManifestFile {
258        let schema = Arc::new(
259            Schema::builder()
260                .with_fields(vec![Arc::new(NestedField::optional(
261                    1,
262                    "id",
263                    Type::Primitive(PrimitiveType::Long),
264                ))])
265                .build()
266                .unwrap(),
267        );
268
269        let partition_spec = PartitionSpec::builder(schema.clone())
270            .with_spec_id(0)
271            .build()
272            .unwrap();
273
274        let output_file = io.new_output(path).unwrap();
275        let encrypted_output = EncryptedOutputFile::new(output_file, key_metadata);
276
277        let mut writer = ManifestWriterBuilder::new_from_encrypted(
278            encrypted_output,
279            Some(1),
280            schema.clone(),
281            partition_spec.clone(),
282        )
283        .expect("Expected a valid writer")
284        .build_v3_data();
285
286        writer
287            .add_entry(ManifestEntry {
288                status: ManifestStatus::Added,
289                snapshot_id: None,
290                sequence_number: None,
291                file_sequence_number: None,
292                data_file: DataFile {
293                    content: DataContentType::Data,
294                    file_path: "s3://bucket/table/data/00000.parquet".to_string(),
295                    file_format: DataFileFormat::Parquet,
296                    partition: Struct::empty(),
297                    record_count: 100,
298                    file_size_in_bytes: 4096,
299                    column_sizes: HashMap::new(),
300                    value_counts: HashMap::new(),
301                    null_value_counts: HashMap::new(),
302                    nan_value_counts: HashMap::new(),
303                    lower_bounds: HashMap::new(),
304                    upper_bounds: HashMap::new(),
305                    key_metadata: None,
306                    split_offsets: None,
307                    equality_ids: None,
308                    sort_order_id: None,
309                    partition_spec_id: 0,
310                    first_row_id: None,
311                    referenced_data_file: None,
312                    content_offset: None,
313                    content_size_in_bytes: None,
314                },
315            })
316            .unwrap();
317
318        writer.write_manifest_file().await.unwrap()
319    }
320
321    #[tokio::test]
322    async fn test_load_manifest_decrypts_when_key_metadata_present() {
323        let key_metadata = StandardKeyMetadata::try_new(b"0123456789abcdef")
324            .unwrap()
325            .with_aad_prefix(b"test-aad-prefix!");
326        let encoded_key_metadata = key_metadata.encode().unwrap().to_vec();
327
328        let io = FileIO::new_with_memory();
329        let path = "memory:///test/encrypted_manifest.avro";
330        let manifest_file = write_encrypted_manifest(&io, path, key_metadata).await;
331        assert_eq!(manifest_file.key_metadata, Some(encoded_key_metadata));
332
333        let manifest = manifest_file.load_manifest(&io).await.unwrap();
334        assert_eq!(manifest.entries().len(), 1);
335        assert_eq!(
336            manifest.entries()[0].file_path(),
337            "s3://bucket/table/data/00000.parquet"
338        );
339        assert_eq!(manifest.entries()[0].data_file.record_count, 100);
340    }
341
342    #[tokio::test]
343    async fn test_load_manifest_fails_with_wrong_key() {
344        let key_metadata = StandardKeyMetadata::try_new(b"0123456789abcdef")
345            .unwrap()
346            .with_aad_prefix(b"test-aad-prefix!");
347
348        let io = FileIO::new_with_memory();
349        let path = "memory:///test/wrong_key_manifest.avro";
350        let mut manifest_file = write_encrypted_manifest(&io, path, key_metadata).await;
351
352        // Point the manifest file at key metadata carrying a different DEK (but
353        // the same AAD prefix). The bytes on disk were encrypted with the
354        // original key, so GCM authentication must fail rather than silently
355        // returning garbage.
356        let wrong_key_metadata = StandardKeyMetadata::try_new(b"fedcba9876543210")
357            .unwrap()
358            .with_aad_prefix(b"test-aad-prefix!");
359        manifest_file.key_metadata = Some(wrong_key_metadata.encode().unwrap().to_vec());
360
361        let err = manifest_file
362            .load_manifest(&io)
363            .await
364            .expect_err("load_manifest must fail when decrypting with the wrong key");
365        assert_eq!(err.kind(), ErrorKind::Unexpected);
366    }
367
368    #[tokio::test]
369    async fn test_load_manifest_fails_with_wrong_aad() {
370        let key_metadata = StandardKeyMetadata::try_new(b"0123456789abcdef")
371            .unwrap()
372            .with_aad_prefix(b"test-aad-prefix!");
373
374        let io = FileIO::new_with_memory();
375        let path = "memory:///test/wrong_aad_manifest.avro";
376        let mut manifest_file = write_encrypted_manifest(&io, path, key_metadata).await;
377
378        // Point the manifest file at key metadata carrying the correct DEK but a
379        // different AAD prefix. The per-block AAD is `aad_prefix || block_index`,
380        // so GCM authentication must fail even though the key is right.
381        let wrong_aad_metadata = StandardKeyMetadata::try_new(b"0123456789abcdef")
382            .unwrap()
383            .with_aad_prefix(b"wrong-aad-prefix");
384        manifest_file.key_metadata = Some(wrong_aad_metadata.encode().unwrap().to_vec());
385
386        let err = manifest_file
387            .load_manifest(&io)
388            .await
389            .expect_err("load_manifest must fail when decrypting with the wrong AAD prefix");
390        assert_eq!(err.kind(), ErrorKind::Unexpected);
391    }
392}
iceberg/spec/manifest_list/manifest_file.rs

iceberg/spec/manifest_list/
manifest_file.rs
