iceberg/spec/manifest/
entry.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::sync::Arc;
19
20use apache_avro::Schema as AvroSchema;
21use once_cell::sync::Lazy;
22use typed_builder::TypedBuilder;
23
24use crate::Error;
25use crate::avro::schema_to_avro_schema;
26use crate::error::{Result, invalid_data};
27use crate::spec::{
28    DataContentType, DataFile, INITIAL_SEQUENCE_NUMBER, ListType, Literal, ManifestFile, MapType,
29    NestedField, NestedFieldRef, PrimitiveLiteral, PrimitiveType, Schema, StructType, Type,
30};
31
32/// Reference to [`ManifestEntry`].
33pub type ManifestEntryRef = Arc<ManifestEntry>;
34
35/// A manifest is an immutable Avro file that lists data files or delete
36/// files, along with each file’s partition data tuple, metrics, and tracking
37/// information.
38#[derive(Debug, PartialEq, Eq, Clone, TypedBuilder)]
39pub struct ManifestEntry {
40    /// field: 0
41    ///
42    /// Used to track additions and deletions.
43    pub status: ManifestStatus,
44    /// field id: 1
45    ///
46    /// Snapshot id where the file was added, or deleted if status is 2.
47    /// Inherited when null.
48    #[builder(default, setter(strip_option(fallback = snapshot_id_opt)))]
49    pub snapshot_id: Option<i64>,
50    /// field id: 3
51    ///
52    /// Data sequence number of the file.
53    /// Inherited when null and status is 1 (added).
54    #[builder(default, setter(strip_option(fallback = sequence_number_opt)))]
55    pub sequence_number: Option<i64>,
56    /// field id: 4
57    ///
58    /// File sequence number indicating when the file was added.
59    /// Inherited when null and status is 1 (added).
60    #[builder(default, setter(strip_option(fallback = file_sequence_number_opt)))]
61    pub file_sequence_number: Option<i64>,
62    /// field id: 2
63    ///
64    /// File path, partition tuple, metrics, …
65    pub data_file: DataFile,
66}
67
68impl ManifestEntry {
69    /// Check if this manifest entry is deleted.
70    pub fn is_alive(&self) -> bool {
71        matches!(
72            self.status,
73            ManifestStatus::Added | ManifestStatus::Existing
74        )
75    }
76
77    /// Status of this manifest entry
78    pub fn status(&self) -> ManifestStatus {
79        self.status
80    }
81
82    /// Content type of this manifest entry.
83    #[inline]
84    pub fn content_type(&self) -> DataContentType {
85        self.data_file.content
86    }
87
88    /// File format of this manifest entry.
89    #[inline]
90    pub fn file_format(&self) -> DataFileFormat {
91        self.data_file.file_format
92    }
93
94    /// Data file path of this manifest entry.
95    #[inline]
96    pub fn file_path(&self) -> &str {
97        &self.data_file.file_path
98    }
99
100    /// Data file record count of the manifest entry.
101    #[inline]
102    pub fn record_count(&self) -> u64 {
103        self.data_file.record_count
104    }
105
106    /// Inherit data from manifest list, such as snapshot id, sequence number.
107    pub(crate) fn inherit_data(&mut self, snapshot_entry: &ManifestFile) {
108        if self.snapshot_id.is_none() {
109            self.snapshot_id = Some(snapshot_entry.added_snapshot_id);
110        }
111
112        if self.sequence_number.is_none()
113            && (self.status == ManifestStatus::Added
114                || snapshot_entry.sequence_number == INITIAL_SEQUENCE_NUMBER)
115        {
116            self.sequence_number = Some(snapshot_entry.sequence_number);
117        }
118
119        if self.file_sequence_number.is_none()
120            && (self.status == ManifestStatus::Added
121                || snapshot_entry.sequence_number == INITIAL_SEQUENCE_NUMBER)
122        {
123            self.file_sequence_number = Some(snapshot_entry.sequence_number);
124        }
125    }
126
127    /// Snapshot id
128    #[inline]
129    pub fn snapshot_id(&self) -> Option<i64> {
130        self.snapshot_id
131    }
132
133    /// Data sequence number.
134    #[inline]
135    pub fn sequence_number(&self) -> Option<i64> {
136        self.sequence_number
137    }
138
139    /// File size in bytes.
140    #[inline]
141    pub fn file_size_in_bytes(&self) -> u64 {
142        self.data_file.file_size_in_bytes
143    }
144
145    /// get a reference to the actual data file
146    #[inline]
147    pub fn data_file(&self) -> &DataFile {
148        &self.data_file
149    }
150}
151
152/// Used to track additions and deletions in ManifestEntry.
153#[derive(Debug, PartialEq, Eq, Clone, Copy)]
154pub enum ManifestStatus {
155    /// Value: 0
156    Existing = 0,
157    /// Value: 1
158    Added = 1,
159    /// Value: 2
160    ///
161    /// Deletes are informational only and not used in scans.
162    Deleted = 2,
163}
164
165impl TryFrom<i32> for ManifestStatus {
166    type Error = Error;
167
168    fn try_from(v: i32) -> Result<ManifestStatus> {
169        match v {
170            0 => Ok(ManifestStatus::Existing),
171            1 => Ok(ManifestStatus::Added),
172            2 => Ok(ManifestStatus::Deleted),
173            _ => Err(invalid_data!("manifest status {v} is invalid")),
174        }
175    }
176}
177
178use super::DataFileFormat;
179
180static STATUS: Lazy<NestedFieldRef> = {
181    Lazy::new(|| {
182        Arc::new(NestedField::required(
183            0,
184            "status",
185            Type::Primitive(PrimitiveType::Int),
186        ))
187    })
188};
189
190static SNAPSHOT_ID_V1: Lazy<NestedFieldRef> = {
191    Lazy::new(|| {
192        Arc::new(NestedField::required(
193            1,
194            "snapshot_id",
195            Type::Primitive(PrimitiveType::Long),
196        ))
197    })
198};
199
200static SNAPSHOT_ID_V2: Lazy<NestedFieldRef> = {
201    Lazy::new(|| {
202        Arc::new(NestedField::optional(
203            1,
204            "snapshot_id",
205            Type::Primitive(PrimitiveType::Long),
206        ))
207    })
208};
209
210static SEQUENCE_NUMBER: Lazy<NestedFieldRef> = {
211    Lazy::new(|| {
212        Arc::new(NestedField::optional(
213            3,
214            "sequence_number",
215            Type::Primitive(PrimitiveType::Long),
216        ))
217    })
218};
219
220static FILE_SEQUENCE_NUMBER: Lazy<NestedFieldRef> = {
221    Lazy::new(|| {
222        Arc::new(NestedField::optional(
223            4,
224            "file_sequence_number",
225            Type::Primitive(PrimitiveType::Long),
226        ))
227    })
228};
229
230static CONTENT: Lazy<NestedFieldRef> = {
231    Lazy::new(|| {
232        Arc::new(
233            NestedField::required(134, "content", Type::Primitive(PrimitiveType::Int))
234                // 0 refers to DataContentType::DATA
235                .with_initial_default(Literal::Primitive(PrimitiveLiteral::Int(0))),
236        )
237    })
238};
239
240static FILE_PATH: Lazy<NestedFieldRef> = {
241    Lazy::new(|| {
242        Arc::new(NestedField::required(
243            100,
244            "file_path",
245            Type::Primitive(PrimitiveType::String),
246        ))
247    })
248};
249
250static FILE_FORMAT: Lazy<NestedFieldRef> = {
251    Lazy::new(|| {
252        Arc::new(NestedField::required(
253            101,
254            "file_format",
255            Type::Primitive(PrimitiveType::String),
256        ))
257    })
258};
259
260static RECORD_COUNT: Lazy<NestedFieldRef> = {
261    Lazy::new(|| {
262        Arc::new(NestedField::required(
263            103,
264            "record_count",
265            Type::Primitive(PrimitiveType::Long),
266        ))
267    })
268};
269
270static FILE_SIZE_IN_BYTES: Lazy<NestedFieldRef> = {
271    Lazy::new(|| {
272        Arc::new(NestedField::required(
273            104,
274            "file_size_in_bytes",
275            Type::Primitive(PrimitiveType::Long),
276        ))
277    })
278};
279
280// Deprecated. Always write a default in v1. Do not write in v2.
281static BLOCK_SIZE_IN_BYTES: Lazy<NestedFieldRef> = {
282    Lazy::new(|| {
283        Arc::new(NestedField::required(
284            105,
285            "block_size_in_bytes",
286            Type::Primitive(PrimitiveType::Long),
287        ))
288    })
289};
290
291static COLUMN_SIZES: Lazy<NestedFieldRef> = {
292    Lazy::new(|| {
293        Arc::new(NestedField::optional(
294            108,
295            "column_sizes",
296            Type::Map(MapType {
297                key_field: Arc::new(NestedField::required(
298                    117,
299                    "key",
300                    Type::Primitive(PrimitiveType::Int),
301                )),
302                value_field: Arc::new(NestedField::required(
303                    118,
304                    "value",
305                    Type::Primitive(PrimitiveType::Long),
306                )),
307            }),
308        ))
309    })
310};
311
312static VALUE_COUNTS: Lazy<NestedFieldRef> = {
313    Lazy::new(|| {
314        Arc::new(NestedField::optional(
315            109,
316            "value_counts",
317            Type::Map(MapType {
318                key_field: Arc::new(NestedField::required(
319                    119,
320                    "key",
321                    Type::Primitive(PrimitiveType::Int),
322                )),
323                value_field: Arc::new(NestedField::required(
324                    120,
325                    "value",
326                    Type::Primitive(PrimitiveType::Long),
327                )),
328            }),
329        ))
330    })
331};
332
333static NULL_VALUE_COUNTS: Lazy<NestedFieldRef> = {
334    Lazy::new(|| {
335        Arc::new(NestedField::optional(
336            110,
337            "null_value_counts",
338            Type::Map(MapType {
339                key_field: Arc::new(NestedField::required(
340                    121,
341                    "key",
342                    Type::Primitive(PrimitiveType::Int),
343                )),
344                value_field: Arc::new(NestedField::required(
345                    122,
346                    "value",
347                    Type::Primitive(PrimitiveType::Long),
348                )),
349            }),
350        ))
351    })
352};
353
354static NAN_VALUE_COUNTS: Lazy<NestedFieldRef> = {
355    Lazy::new(|| {
356        Arc::new(NestedField::optional(
357            137,
358            "nan_value_counts",
359            Type::Map(MapType {
360                key_field: Arc::new(NestedField::required(
361                    138,
362                    "key",
363                    Type::Primitive(PrimitiveType::Int),
364                )),
365                value_field: Arc::new(NestedField::required(
366                    139,
367                    "value",
368                    Type::Primitive(PrimitiveType::Long),
369                )),
370            }),
371        ))
372    })
373};
374
375static LOWER_BOUNDS: Lazy<NestedFieldRef> = {
376    Lazy::new(|| {
377        Arc::new(NestedField::optional(
378            125,
379            "lower_bounds",
380            Type::Map(MapType {
381                key_field: Arc::new(NestedField::required(
382                    126,
383                    "key",
384                    Type::Primitive(PrimitiveType::Int),
385                )),
386                value_field: Arc::new(NestedField::required(
387                    127,
388                    "value",
389                    Type::Primitive(PrimitiveType::Binary),
390                )),
391            }),
392        ))
393    })
394};
395
396static UPPER_BOUNDS: Lazy<NestedFieldRef> = {
397    Lazy::new(|| {
398        Arc::new(NestedField::optional(
399            128,
400            "upper_bounds",
401            Type::Map(MapType {
402                key_field: Arc::new(NestedField::required(
403                    129,
404                    "key",
405                    Type::Primitive(PrimitiveType::Int),
406                )),
407                value_field: Arc::new(NestedField::required(
408                    130,
409                    "value",
410                    Type::Primitive(PrimitiveType::Binary),
411                )),
412            }),
413        ))
414    })
415};
416
417static KEY_METADATA: Lazy<NestedFieldRef> = {
418    Lazy::new(|| {
419        Arc::new(NestedField::optional(
420            131,
421            "key_metadata",
422            Type::Primitive(PrimitiveType::Binary),
423        ))
424    })
425};
426
427static SPLIT_OFFSETS: Lazy<NestedFieldRef> = {
428    Lazy::new(|| {
429        Arc::new(NestedField::optional(
430            132,
431            "split_offsets",
432            Type::List(ListType {
433                element_field: Arc::new(NestedField::required(
434                    133,
435                    "element",
436                    Type::Primitive(PrimitiveType::Long),
437                )),
438            }),
439        ))
440    })
441};
442
443static EQUALITY_IDS: Lazy<NestedFieldRef> = {
444    Lazy::new(|| {
445        Arc::new(NestedField::optional(
446            135,
447            "equality_ids",
448            Type::List(ListType {
449                element_field: Arc::new(NestedField::required(
450                    136,
451                    "element",
452                    Type::Primitive(PrimitiveType::Int),
453                )),
454            }),
455        ))
456    })
457};
458
459static SORT_ORDER_ID: Lazy<NestedFieldRef> = {
460    Lazy::new(|| {
461        Arc::new(NestedField::optional(
462            140,
463            "sort_order_id",
464            Type::Primitive(PrimitiveType::Int),
465        ))
466    })
467};
468
469static FIRST_ROW_ID: Lazy<NestedFieldRef> = {
470    Lazy::new(|| {
471        Arc::new(NestedField::optional(
472            142,
473            "first_row_id",
474            Type::Primitive(PrimitiveType::Long),
475        ))
476    })
477};
478
479static REFERENCE_DATA_FILE: Lazy<NestedFieldRef> = {
480    Lazy::new(|| {
481        Arc::new(NestedField::optional(
482            143,
483            "referenced_data_file",
484            Type::Primitive(PrimitiveType::String),
485        ))
486    })
487};
488
489static CONTENT_OFFSET: Lazy<NestedFieldRef> = {
490    Lazy::new(|| {
491        Arc::new(NestedField::optional(
492            144,
493            "content_offset",
494            Type::Primitive(PrimitiveType::Long),
495        ))
496    })
497};
498
499static CONTENT_SIZE_IN_BYTES: Lazy<NestedFieldRef> = {
500    Lazy::new(|| {
501        Arc::new(NestedField::optional(
502            145,
503            "content_size_in_bytes",
504            Type::Primitive(PrimitiveType::Long),
505        ))
506    })
507};
508
509fn data_file_fields_v3(partition_type: &StructType) -> Vec<NestedFieldRef> {
510    vec![
511        CONTENT.clone(),
512        FILE_PATH.clone(),
513        FILE_FORMAT.clone(),
514        Arc::new(NestedField::required(
515            102,
516            "partition",
517            Type::Struct(partition_type.clone()),
518        )),
519        RECORD_COUNT.clone(),
520        FILE_SIZE_IN_BYTES.clone(),
521        COLUMN_SIZES.clone(),
522        VALUE_COUNTS.clone(),
523        NULL_VALUE_COUNTS.clone(),
524        NAN_VALUE_COUNTS.clone(),
525        LOWER_BOUNDS.clone(),
526        UPPER_BOUNDS.clone(),
527        KEY_METADATA.clone(),
528        SPLIT_OFFSETS.clone(),
529        EQUALITY_IDS.clone(),
530        SORT_ORDER_ID.clone(),
531        FIRST_ROW_ID.clone(),
532        REFERENCE_DATA_FILE.clone(),
533        CONTENT_OFFSET.clone(),
534        CONTENT_SIZE_IN_BYTES.clone(),
535    ]
536}
537
538pub(super) fn data_file_schema_v3(partition_type: &StructType) -> Result<AvroSchema> {
539    let schema = Schema::builder()
540        .with_fields(data_file_fields_v3(partition_type))
541        .build()?;
542    schema_to_avro_schema("data_file", &schema)
543}
544
545fn data_file_fields_v2(partition_type: &StructType) -> Vec<NestedFieldRef> {
546    vec![
547        CONTENT.clone(),
548        FILE_PATH.clone(),
549        FILE_FORMAT.clone(),
550        Arc::new(NestedField::required(
551            102,
552            "partition",
553            Type::Struct(partition_type.clone()),
554        )),
555        RECORD_COUNT.clone(),
556        FILE_SIZE_IN_BYTES.clone(),
557        COLUMN_SIZES.clone(),
558        VALUE_COUNTS.clone(),
559        NULL_VALUE_COUNTS.clone(),
560        NAN_VALUE_COUNTS.clone(),
561        LOWER_BOUNDS.clone(),
562        UPPER_BOUNDS.clone(),
563        KEY_METADATA.clone(),
564        SPLIT_OFFSETS.clone(),
565        EQUALITY_IDS.clone(),
566        SORT_ORDER_ID.clone(),
567        FIRST_ROW_ID.clone(),
568        REFERENCE_DATA_FILE.clone(),
569        // Why are the following two fields here in the existing v2 schema?
570        // In the spec, they are not even listed as optional for v2.
571        CONTENT_OFFSET.clone(),
572        CONTENT_SIZE_IN_BYTES.clone(),
573    ]
574}
575
576pub(super) fn data_file_schema_v2(partition_type: &StructType) -> Result<AvroSchema> {
577    let schema = Schema::builder()
578        .with_fields(data_file_fields_v2(partition_type))
579        .build()?;
580    schema_to_avro_schema("data_file", &schema)
581}
582
583pub(super) fn manifest_schema_v2(partition_type: &StructType) -> Result<AvroSchema> {
584    let fields = vec![
585        STATUS.clone(),
586        SNAPSHOT_ID_V2.clone(),
587        SEQUENCE_NUMBER.clone(),
588        FILE_SEQUENCE_NUMBER.clone(),
589        Arc::new(NestedField::required(
590            2,
591            "data_file",
592            Type::Struct(StructType::new(data_file_fields_v2(partition_type))),
593        )),
594    ];
595    let schema = Schema::builder().with_fields(fields).build()?;
596    schema_to_avro_schema("manifest_entry", &schema)
597}
598
599fn data_file_fields_v1(partition_type: &StructType) -> Vec<NestedFieldRef> {
600    vec![
601        FILE_PATH.clone(),
602        FILE_FORMAT.clone(),
603        Arc::new(NestedField::required(
604            102,
605            "partition",
606            Type::Struct(partition_type.clone()),
607        )),
608        RECORD_COUNT.clone(),
609        FILE_SIZE_IN_BYTES.clone(),
610        BLOCK_SIZE_IN_BYTES.clone(),
611        COLUMN_SIZES.clone(),
612        VALUE_COUNTS.clone(),
613        NULL_VALUE_COUNTS.clone(),
614        NAN_VALUE_COUNTS.clone(),
615        LOWER_BOUNDS.clone(),
616        UPPER_BOUNDS.clone(),
617        KEY_METADATA.clone(),
618        SPLIT_OFFSETS.clone(),
619        SORT_ORDER_ID.clone(),
620    ]
621}
622
623pub(super) fn data_file_schema_v1(partition_type: &StructType) -> Result<AvroSchema> {
624    let schema = Schema::builder()
625        .with_fields(data_file_fields_v1(partition_type))
626        .build()?;
627    schema_to_avro_schema("data_file", &schema)
628}
629
630pub(super) fn manifest_schema_v1(partition_type: &StructType) -> Result<AvroSchema> {
631    let fields = vec![
632        STATUS.clone(),
633        SNAPSHOT_ID_V1.clone(),
634        Arc::new(NestedField::required(
635            2,
636            "data_file",
637            Type::Struct(StructType::new(data_file_fields_v1(partition_type))),
638        )),
639    ];
640    let schema = Schema::builder().with_fields(fields).build()?;
641    schema_to_avro_schema("manifest_entry", &schema)
642}
iceberg/spec/manifest/entry.rs

iceberg/spec/manifest/
entry.rs
