iceberg/spec/manifest/data_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::collections::HashMap;
19use std::io::{Read, Write};
20use std::str::FromStr;
21
22use apache_avro::{Reader as AvroReader, Writer as AvroWriter, from_value, to_value};
23use serde_derive::{Deserialize, Serialize};
24use serde_with::{DeserializeFromStr, SerializeDisplay};
25
26use super::_serde::DataFileSerde;
27use super::{
28 Datum, FormatVersion, Schema, data_file_schema_v1, data_file_schema_v2, data_file_schema_v3,
29};
30use crate::Error;
31use crate::error::{Result, invalid_data};
32use crate::spec::{
33 DEFAULT_PARTITION_SPEC_ID, Literal, NestedFieldRef, PrimitiveType, Struct, StructType, Type,
34};
35
36/// Data file carries data file path, partition tuple, metrics, …
37#[derive(Debug, PartialEq, Clone, Eq, Builder)]
38pub struct DataFile {
39 /// field id: 134
40 ///
41 /// Type of content stored by the data file: data, equality deletes,
42 /// or position deletes (all v1 files are data files)
43 pub(crate) content: DataContentType,
44 /// field id: 100
45 ///
46 /// Full URI for the file with FS scheme
47 pub(crate) file_path: String,
48 /// field id: 101
49 ///
50 /// String file format name, `avro`, `orc`, `parquet`, or `puffin`
51 pub(crate) file_format: DataFileFormat,
52 /// field id: 102
53 ///
54 /// Partition data tuple, schema based on the partition spec output using
55 /// partition field ids for the struct field ids
56 #[builder(default = "Struct::empty()")]
57 pub(crate) partition: Struct,
58 /// field id: 103
59 ///
60 /// Number of records in this file, or the cardinality of a deletion vector
61 pub(crate) record_count: u64,
62 /// field id: 104
63 ///
64 /// Total file size in bytes
65 pub(crate) file_size_in_bytes: u64,
66 /// field id: 108
67 /// key field id: 117
68 /// value field id: 118
69 ///
70 /// Map from column id to the total size on disk of all regions that
71 /// store the column. Does not include bytes necessary to read other
72 /// columns, like footers. Leave null for row-oriented formats (Avro)
73 #[builder(default)]
74 pub(crate) column_sizes: HashMap<i32, u64>,
75 /// field id: 109
76 /// key field id: 119
77 /// value field id: 120
78 ///
79 /// Map from column id to number of values in the column (including null
80 /// and NaN values)
81 #[builder(default)]
82 pub(crate) value_counts: HashMap<i32, u64>,
83 /// field id: 110
84 /// key field id: 121
85 /// value field id: 122
86 ///
87 /// Map from column id to number of null values in the column
88 #[builder(default)]
89 pub(crate) null_value_counts: HashMap<i32, u64>,
90 /// field id: 137
91 /// key field id: 138
92 /// value field id: 139
93 ///
94 /// Map from column id to number of NaN values in the column
95 #[builder(default)]
96 pub(crate) nan_value_counts: HashMap<i32, u64>,
97 /// field id: 125
98 /// key field id: 126
99 /// value field id: 127
100 ///
101 /// Map from column id to lower bound in the column serialized as binary.
102 /// Each value must be less than or equal to all non-null, non-NaN values
103 /// in the column for the file.
104 ///
105 /// Reference:
106 ///
107 /// - [Binary single-value serialization](https://iceberg.apache.org/spec/#binary-single-value-serialization)
108 #[builder(default)]
109 pub(crate) lower_bounds: HashMap<i32, Datum>,
110 /// field id: 128
111 /// key field id: 129
112 /// value field id: 130
113 ///
114 /// Map from column id to upper bound in the column serialized as binary.
115 /// Each value must be greater than or equal to all non-null, non-Nan
116 /// values in the column for the file.
117 ///
118 /// Reference:
119 ///
120 /// - [Binary single-value serialization](https://iceberg.apache.org/spec/#binary-single-value-serialization)
121 #[builder(default)]
122 pub(crate) upper_bounds: HashMap<i32, Datum>,
123 /// field id: 131
124 ///
125 /// Implementation-specific key metadata for encryption
126 #[builder(default)]
127 pub(crate) key_metadata: Option<Vec<u8>>,
128 /// field id: 132
129 /// element field id: 133
130 ///
131 /// Split offsets for the data file. For example, all row group offsets
132 /// in a Parquet file. Must be sorted ascending. Optional field that
133 /// should be serialized as null when not present.
134 #[builder(default)]
135 pub(crate) split_offsets: Option<Vec<i64>>,
136 /// field id: 135
137 /// element field id: 136
138 ///
139 /// Field ids used to determine row equality in equality delete files.
140 /// Required when content is EqualityDeletes and should be null
141 /// otherwise. Fields with ids listed in this column must be present
142 /// in the delete file
143 #[builder(default)]
144 pub(crate) equality_ids: Option<Vec<i32>>,
145 /// field id: 140
146 ///
147 /// ID representing sort order for this file.
148 ///
149 /// If sort order ID is missing or unknown, then the order is assumed to
150 /// be unsorted. Only data files and equality delete files should be
151 /// written with a non-null order id. Position deletes are required to be
152 /// sorted by file and position, not a table order, and should set sort
153 /// order id to null. Readers must ignore sort order id for position
154 /// delete files.
155 #[builder(default, setter(strip_option))]
156 pub(crate) sort_order_id: Option<i32>,
157 /// field id: 142
158 ///
159 /// The _row_id for the first row in the data file.
160 /// For more details, refer to <https://github.com/apache/iceberg/blob/main/format/spec.md#first-row-id-inheritance>
161 #[builder(default)]
162 pub(crate) first_row_id: Option<i64>,
163 /// This field is not included in spec. It is just store in memory representation used
164 /// in process.
165 #[builder(default = "DEFAULT_PARTITION_SPEC_ID")]
166 pub(crate) partition_spec_id: i32,
167 /// field id: 143
168 ///
169 /// Fully qualified location (URI with FS scheme) of a data file that all deletes reference.
170 /// Position delete metadata can use `referenced_data_file` when all deletes tracked by the
171 /// entry are in a single data file. Setting the referenced file is required for deletion vectors.
172 #[builder(default)]
173 pub(crate) referenced_data_file: Option<String>,
174 /// field: 144
175 ///
176 /// The offset in the file where the content starts.
177 /// The `content_offset` and `content_size_in_bytes` fields are used to reference a specific blob
178 /// for direct access to a deletion vector. For deletion vectors, these values are required and must
179 /// exactly match the `offset` and `length` stored in the Puffin footer for the deletion vector blob.
180 #[builder(default)]
181 pub(crate) content_offset: Option<i64>,
182 /// field: 145
183 ///
184 /// The length of a referenced content stored in the file; required if `content_offset` is present
185 #[builder(default)]
186 pub(crate) content_size_in_bytes: Option<i64>,
187}
188
189fn promote_partition_value(
190 value: Option<&Literal>,
191 current_field: &NestedFieldRef,
192 written_field: &NestedFieldRef,
193) -> Option<Literal> {
194 match value {
195 Some(Literal::Primitive(primitive)) => {
196 let promoted = match current_field.field_type.as_primitive_type() {
197 // The source column was dropped, so there is no current type to widen to.
198 // Widen a numeric to its promotion ceiling so this value and its deletes
199 // normalize the same way; keep every other type (e.g. a `date`, stored as
200 // an `int`) at its written type so deletes that skip normalization match.
201 Some(PrimitiveType::Unknown) => {
202 match written_field.field_type.as_primitive_type() {
203 Some(PrimitiveType::Int | PrimitiveType::Float) => primitive.widen(),
204 _ => None,
205 }
206 }
207 Some(target) => primitive.promote_to(target),
208 None => None,
209 };
210 promoted.map(Literal::Primitive).or_else(|| value.cloned())
211 }
212 other => other.cloned(),
213 }
214}
215
216impl DataFile {
217 /// Returns this file's partition tuple normalized to the current schema: a field
218 /// promoted (`int` -> `long`, `float` -> `double`) since the manifest was written is
219 /// widened to its `current` type; a dropped numeric source, which has no `current`
220 /// type, is widened to its promotion ceiling by its `written` type; all other values
221 /// are copied through. Column bounds are widened separately, lazily, where they are
222 /// compared.
223 ///
224 /// `current` and `written` are the file's spec resolved against the current schema and
225 /// against the manifest's schema, so they match the stored tuple in length.
226 ///
227 /// # Errors
228 /// Returns `DataInvalid` if the stored tuple and `current` have different lengths; the
229 /// lengths should always match, since both come from the same spec.
230 pub(crate) fn promoted_partition(
231 &self,
232 current: &StructType,
233 written: &StructType,
234 ) -> Result<Struct> {
235 if self.partition.fields().len() != current.fields().len() {
236 return Err(invalid_data!(
237 "Partition tuple has {} values but the spec partition type has {} fields",
238 self.partition.fields().len(),
239 current.fields().len()
240 ));
241 }
242
243 Ok(self
244 .partition
245 .iter()
246 .zip(current.fields())
247 .zip(written.fields())
248 .map(|((value, current_field), written_field)| {
249 promote_partition_value(value, current_field, written_field)
250 })
251 .collect())
252 }
253
254 /// Get the content type of the data file (data, equality deletes, or position deletes)
255 pub fn content_type(&self) -> DataContentType {
256 self.content
257 }
258 /// Get the file path as full URI with FS scheme
259 pub fn file_path(&self) -> &str {
260 &self.file_path
261 }
262 /// Get the file format of the file (avro, orc or parquet).
263 pub fn file_format(&self) -> DataFileFormat {
264 self.file_format
265 }
266 /// Get the partition values of the file.
267 pub fn partition(&self) -> &Struct {
268 &self.partition
269 }
270 /// Get the record count in the data file.
271 pub fn record_count(&self) -> u64 {
272 self.record_count
273 }
274 /// Get the file size in bytes.
275 pub fn file_size_in_bytes(&self) -> u64 {
276 self.file_size_in_bytes
277 }
278 /// Get the column sizes.
279 /// Map from column id to the total size on disk of all regions that
280 /// store the column. Does not include bytes necessary to read other
281 /// columns, like footers. Null for row-oriented formats (Avro)
282 pub fn column_sizes(&self) -> &HashMap<i32, u64> {
283 &self.column_sizes
284 }
285 /// Get the columns value counts for the data file.
286 /// Map from column id to number of values in the column (including null
287 /// and NaN values)
288 pub fn value_counts(&self) -> &HashMap<i32, u64> {
289 &self.value_counts
290 }
291 /// Get the null value counts of the data file.
292 /// Map from column id to number of null values in the column
293 pub fn null_value_counts(&self) -> &HashMap<i32, u64> {
294 &self.null_value_counts
295 }
296 /// Get the nan value counts of the data file.
297 /// Map from column id to number of NaN values in the column
298 pub fn nan_value_counts(&self) -> &HashMap<i32, u64> {
299 &self.nan_value_counts
300 }
301 /// Get the lower bounds of the data file values per column.
302 /// Map from column id to lower bound in the column serialized as binary.
303 pub fn lower_bounds(&self) -> &HashMap<i32, Datum> {
304 &self.lower_bounds
305 }
306 /// Get the upper bounds of the data file values per column.
307 /// Map from column id to upper bound in the column serialized as binary.
308 pub fn upper_bounds(&self) -> &HashMap<i32, Datum> {
309 &self.upper_bounds
310 }
311 /// Get the Implementation-specific key metadata for the data file.
312 pub fn key_metadata(&self) -> Option<&[u8]> {
313 self.key_metadata.as_deref()
314 }
315 /// Get the split offsets of the data file.
316 /// For example, all row group offsets in a Parquet file.
317 /// Returns `None` if no split offsets are present.
318 pub fn split_offsets(&self) -> Option<&[i64]> {
319 self.split_offsets.as_deref()
320 }
321 /// Get the equality ids of the data file.
322 /// Field ids used to determine row equality in equality delete files.
323 /// null when content is not EqualityDeletes.
324 pub fn equality_ids(&self) -> Option<Vec<i32>> {
325 self.equality_ids.clone()
326 }
327 /// Get the first row id in the data file.
328 pub fn first_row_id(&self) -> Option<i64> {
329 self.first_row_id
330 }
331 /// Get the sort order id of the data file.
332 /// Only data files and equality delete files should be
333 /// written with a non-null order id. Position deletes are required to be
334 /// sorted by file and position, not a table order, and should set sort
335 /// order id to null. Readers must ignore sort order id for position
336 /// delete files.
337 pub fn sort_order_id(&self) -> Option<i32> {
338 self.sort_order_id
339 }
340 /// Get the fully qualified referenced location for the corresponding data file.
341 /// Positional delete files could have the field set, and deletion vectors must the field set.
342 pub fn referenced_data_file(&self) -> Option<String> {
343 self.referenced_data_file.clone()
344 }
345 /// Get the offset in the file where the blob content starts.
346 /// Only meaningful for puffin blobs, and required for deletion vectors.
347 pub fn content_offset(&self) -> Option<i64> {
348 self.content_offset
349 }
350 /// Get the length of a puffin blob.
351 /// Only meaningful for puffin blobs, and required for deletion vectors.
352 pub fn content_size_in_bytes(&self) -> Option<i64> {
353 self.content_size_in_bytes
354 }
355}
356
357/// Convert data files to avro bytes and write to writer.
358/// Return the bytes written.
359pub fn write_data_files_to_avro<W: Write>(
360 writer: &mut W,
361 data_files: impl IntoIterator<Item = DataFile>,
362 partition_type: &StructType,
363 version: FormatVersion,
364) -> Result<usize> {
365 let avro_schema = match version {
366 FormatVersion::V1 => data_file_schema_v1(partition_type).unwrap(),
367 FormatVersion::V2 => data_file_schema_v2(partition_type).unwrap(),
368 FormatVersion::V3 => data_file_schema_v3(partition_type).unwrap(),
369 };
370 let mut writer = AvroWriter::new(&avro_schema, writer)?;
371
372 // Wrap once and reuse across files so the field-name lookup is not rebuilt
373 // for each data file.
374 let partition_struct_type = Type::Struct(partition_type.clone());
375 for data_file in data_files {
376 let value = to_value(DataFileSerde::try_from(
377 data_file,
378 &partition_struct_type,
379 FormatVersion::V1,
380 )?)?
381 .resolve(&avro_schema)?;
382 writer.append_value(value)?;
383 }
384
385 Ok(writer.flush()?)
386}
387
388/// Parse data files from avro bytes.
389pub fn read_data_files_from_avro<R: Read>(
390 reader: &mut R,
391 schema: &Schema,
392 partition_spec_id: i32,
393 partition_type: &StructType,
394 version: FormatVersion,
395) -> Result<Vec<DataFile>> {
396 let avro_schema = match version {
397 FormatVersion::V1 => data_file_schema_v1(partition_type).unwrap(),
398 FormatVersion::V2 => data_file_schema_v2(partition_type).unwrap(),
399 FormatVersion::V3 => data_file_schema_v3(partition_type).unwrap(),
400 };
401
402 let reader = AvroReader::builder(reader)
403 .reader_schema(&avro_schema)
404 .build()?;
405 // Wrap once and reuse across files so the field-name lookup is not rebuilt
406 // for each data file.
407 let partition_struct_type = Type::Struct(partition_type.clone());
408 reader
409 .into_iter()
410 .map(|value| {
411 from_value::<DataFileSerde>(&value?)?.try_into(
412 partition_spec_id,
413 &partition_struct_type,
414 schema,
415 )
416 })
417 .collect::<Result<Vec<_>>>()
418}
419
420/// Type of content stored by the data file: data, equality deletes, or
421/// position deletes (all v1 files are data files)
422#[derive(Debug, PartialEq, Eq, Clone, Copy, Serialize, Deserialize, Default)]
423pub enum DataContentType {
424 /// value: 0
425 #[default]
426 Data = 0,
427 /// value: 1
428 PositionDeletes = 1,
429 /// value: 2
430 EqualityDeletes = 2,
431}
432
433impl TryFrom<i32> for DataContentType {
434 type Error = Error;
435
436 fn try_from(v: i32) -> Result<DataContentType> {
437 match v {
438 0 => Ok(DataContentType::Data),
439 1 => Ok(DataContentType::PositionDeletes),
440 2 => Ok(DataContentType::EqualityDeletes),
441 _ => Err(invalid_data!("data content type {v} is invalid")),
442 }
443 }
444}
445
446/// Format of this data.
447#[derive(Debug, PartialEq, Eq, Clone, Copy, SerializeDisplay, DeserializeFromStr)]
448pub enum DataFileFormat {
449 /// Avro file format: <https://avro.apache.org/>
450 Avro,
451 /// Orc file format: <https://orc.apache.org/>
452 Orc,
453 /// Parquet file format: <https://parquet.apache.org/>
454 Parquet,
455 /// Puffin file format: <https://iceberg.apache.org/puffin-spec/>
456 Puffin,
457}
458
459impl FromStr for DataFileFormat {
460 type Err = Error;
461
462 fn from_str(s: &str) -> Result<Self> {
463 match s.to_lowercase().as_str() {
464 "avro" => Ok(Self::Avro),
465 "orc" => Ok(Self::Orc),
466 "parquet" => Ok(Self::Parquet),
467 "puffin" => Ok(Self::Puffin),
468 _ => Err(invalid_data!("Unsupported data file format: {s}")),
469 }
470 }
471}
472
473impl std::fmt::Display for DataFileFormat {
474 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
475 match self {
476 DataFileFormat::Avro => write!(f, "avro"),
477 DataFileFormat::Orc => write!(f, "orc"),
478 DataFileFormat::Parquet => write!(f, "parquet"),
479 DataFileFormat::Puffin => write!(f, "puffin"),
480 }
481 }
482}
483
484#[cfg(test)]
485mod test {
486 use crate::ErrorKind;
487 use crate::spec::{
488 DataContentType, DataFileBuilder, DataFileFormat, Literal, NestedField, PrimitiveType,
489 Struct, StructType, Type,
490 };
491
492 #[test]
493 fn test_data_content_type_default() {
494 assert_eq!(DataContentType::default(), DataContentType::Data);
495 }
496
497 #[test]
498 fn test_data_content_type_default_value() {
499 assert_eq!(DataContentType::default() as i32, 0);
500 }
501
502 #[test]
503 fn promoted_partition_widens_int_and_float() {
504 // Partition columns promoted int -> long and float -> double; a file
505 // written before the promotion still carries the narrower types.
506 let current = StructType::new(vec![
507 NestedField::required(1000, "b", Type::Primitive(PrimitiveType::Long)).into(),
508 NestedField::required(1001, "c", Type::Primitive(PrimitiveType::Double)).into(),
509 ]);
510 let written = StructType::new(vec![
511 NestedField::required(1000, "b", Type::Primitive(PrimitiveType::Int)).into(),
512 NestedField::required(1001, "c", Type::Primitive(PrimitiveType::Float)).into(),
513 ]);
514
515 let data_file = DataFileBuilder::default()
516 .content(DataContentType::Data)
517 .file_path("x.parquet".to_string())
518 .file_format(DataFileFormat::Parquet)
519 .partition(Struct::from_iter([
520 Some(Literal::int(7)),
521 Some(Literal::float(1.5_f32)),
522 ]))
523 .record_count(1)
524 .file_size_in_bytes(1)
525 .partition_spec_id(0)
526 .build()
527 .unwrap();
528
529 assert_eq!(
530 data_file.promoted_partition(¤t, &written).unwrap(),
531 Struct::from_iter([Some(Literal::long(7)), Some(Literal::double(1.5))])
532 );
533 }
534
535 #[test]
536 fn promoted_partition_widens_dropped_numeric_source() {
537 // A dropped partition source resolves to `Unknown`, so there is no current
538 // type; a numeric still widens to its ceiling by its written type, keeping it
539 // aligned with deletes written while the source still existed.
540 let current = StructType::new(vec![
541 NestedField::required(1000, "b", Type::Primitive(PrimitiveType::Unknown)).into(),
542 ]);
543 let written = StructType::new(vec![
544 NestedField::required(1000, "b", Type::Primitive(PrimitiveType::Int)).into(),
545 ]);
546
547 let data_file = DataFileBuilder::default()
548 .content(DataContentType::Data)
549 .file_path("x.parquet".to_string())
550 .file_format(DataFileFormat::Parquet)
551 .partition(Struct::from_iter([Some(Literal::int(7))]))
552 .record_count(1)
553 .file_size_in_bytes(1)
554 .partition_spec_id(0)
555 .build()
556 .unwrap();
557
558 assert_eq!(
559 data_file.promoted_partition(¤t, &written).unwrap(),
560 Struct::from_iter([Some(Literal::long(7))])
561 );
562 }
563
564 #[test]
565 fn promoted_partition_preserves_dropped_date_source() {
566 // A `date` is stored as an `int`, so widening a dropped date the way a dropped
567 // int widens would turn it into a `long` and diverge from delete manifests that
568 // skip normalization. A dropped non-numeric source is copied through unchanged.
569 let current = StructType::new(vec![
570 NestedField::required(1000, "d", Type::Primitive(PrimitiveType::Unknown)).into(),
571 ]);
572 let written = StructType::new(vec![
573 NestedField::required(1000, "d", Type::Primitive(PrimitiveType::Date)).into(),
574 ]);
575
576 let data_file = DataFileBuilder::default()
577 .content(DataContentType::Data)
578 .file_path("x.parquet".to_string())
579 .file_format(DataFileFormat::Parquet)
580 .partition(Struct::from_iter([Some(Literal::date(20_000))]))
581 .record_count(1)
582 .file_size_in_bytes(1)
583 .partition_spec_id(0)
584 .build()
585 .unwrap();
586
587 assert_eq!(
588 data_file.promoted_partition(¤t, &written).unwrap(),
589 data_file.partition
590 );
591 }
592
593 #[test]
594 fn promoted_partition_leaves_matching_types_unchanged() {
595 let partition_type = StructType::new(vec![
596 NestedField::required(1000, "b", Type::Primitive(PrimitiveType::Long)).into(),
597 ]);
598
599 let data_file = DataFileBuilder::default()
600 .content(DataContentType::Data)
601 .file_path("x.parquet".to_string())
602 .file_format(DataFileFormat::Parquet)
603 .partition(Struct::from_iter([Some(Literal::long(7))]))
604 .record_count(1)
605 .file_size_in_bytes(1)
606 .partition_spec_id(0)
607 .build()
608 .unwrap();
609
610 assert_eq!(
611 data_file
612 .promoted_partition(&partition_type, &partition_type)
613 .unwrap(),
614 data_file.partition
615 );
616 }
617
618 #[test]
619 fn promoted_partition_rejects_tuple_length_mismatch() {
620 let data_file = |values: Vec<Option<Literal>>| {
621 DataFileBuilder::default()
622 .content(DataContentType::Data)
623 .file_path("x.parquet".to_string())
624 .file_format(DataFileFormat::Parquet)
625 .partition(Struct::from_iter(values))
626 .record_count(1)
627 .file_size_in_bytes(1)
628 .partition_spec_id(0)
629 .build()
630 .unwrap()
631 };
632 let one_field = StructType::new(vec![
633 NestedField::required(1000, "b", Type::Primitive(PrimitiveType::Long)).into(),
634 ]);
635 let two_fields = StructType::new(vec![
636 NestedField::required(1000, "b", Type::Primitive(PrimitiveType::Long)).into(),
637 NestedField::required(1001, "c", Type::Primitive(PrimitiveType::Long)).into(),
638 ]);
639
640 let longer = data_file(vec![Some(Literal::int(7)), Some(Literal::int(8))]);
641 assert_eq!(
642 longer
643 .promoted_partition(&one_field, &one_field)
644 .unwrap_err()
645 .kind(),
646 ErrorKind::DataInvalid
647 );
648
649 let shorter = data_file(vec![Some(Literal::int(7))]);
650 assert_eq!(
651 shorter
652 .promoted_partition(&two_fields, &two_fields)
653 .unwrap_err()
654 .kind(),
655 ErrorKind::DataInvalid
656 );
657 }
658}