iceberg/puffin/
metadata.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, HashSet};
19
20use bytes::Bytes;
21use serde::{Deserialize, Serialize};
22
23use crate::compression::CompressionCodec;
24use crate::io::FileRead;
25use crate::{Error, ErrorKind, Result};
26
27/// Human-readable identification of the application writing the file, along with its version.
28/// Example: "Trino version 381"
29pub const CREATED_BY_PROPERTY: &str = "created-by";
30
31/// Metadata about a blob.
32/// For more information, see: https://iceberg.apache.org/puffin-spec/#blobmetadata
33#[derive(Debug, PartialEq, Eq, Serialize, Deserialize, Clone)]
34#[serde(rename_all = "kebab-case")]
35pub struct BlobMetadata {
36    pub(crate) r#type: String,
37    pub(crate) fields: Vec<i32>,
38    pub(crate) snapshot_id: i64,
39    pub(crate) sequence_number: i64,
40    pub(crate) offset: u64,
41    pub(crate) length: u64,
42    #[serde(skip_serializing_if = "CompressionCodec::is_none")]
43    #[serde(default)]
44    pub(crate) compression_codec: CompressionCodec,
45    #[serde(skip_serializing_if = "HashMap::is_empty")]
46    #[serde(default)]
47    pub(crate) properties: HashMap<String, String>,
48}
49
50impl BlobMetadata {
51    #[inline]
52    /// See blob types: https://iceberg.apache.org/puffin-spec/#blob-types
53    pub fn blob_type(&self) -> &str {
54        &self.r#type
55    }
56
57    #[inline]
58    /// List of field IDs the blob was computed for; the order of items is used to compute sketches stored in the blob.
59    pub fn fields(&self) -> &[i32] {
60        &self.fields
61    }
62
63    #[inline]
64    /// ID of the Iceberg table's snapshot the blob was computed from
65    pub fn snapshot_id(&self) -> i64 {
66        self.snapshot_id
67    }
68
69    #[inline]
70    /// Sequence number of the Iceberg table's snapshot the blob was computed from
71    pub fn sequence_number(&self) -> i64 {
72        self.sequence_number
73    }
74
75    #[inline]
76    /// The offset in the file where the blob contents start
77    pub fn offset(&self) -> u64 {
78        self.offset
79    }
80
81    #[inline]
82    /// The length of the blob stored in the file (after compression, if compressed)
83    pub fn length(&self) -> u64 {
84        self.length
85    }
86
87    #[inline]
88    /// The compression codec used to compress the data
89    pub fn compression_codec(&self) -> CompressionCodec {
90        self.compression_codec
91    }
92
93    #[inline]
94    /// Arbitrary meta-information about the blob
95    pub fn properties(&self) -> &HashMap<String, String> {
96        &self.properties
97    }
98}
99
100#[derive(Clone, Copy, PartialEq, Eq, Hash, Debug)]
101pub(crate) enum Flag {
102    FooterPayloadCompressed = 0,
103}
104
105impl Flag {
106    pub(crate) fn byte_idx(self) -> u8 {
107        (self as u8) / 8
108    }
109
110    pub(crate) fn bit_idx(self) -> u8 {
111        (self as u8) % 8
112    }
113
114    fn matches(self, byte_idx: u8, bit_idx: u8) -> bool {
115        self.byte_idx() == byte_idx && self.bit_idx() == bit_idx
116    }
117
118    fn from(byte_idx: u8, bit_idx: u8) -> Result<Flag> {
119        if Flag::FooterPayloadCompressed.matches(byte_idx, bit_idx) {
120            Ok(Flag::FooterPayloadCompressed)
121        } else {
122            Err(Error::new(
123                ErrorKind::DataInvalid,
124                format!("Unknown flag byte {byte_idx} and bit {bit_idx} combination"),
125            ))
126        }
127    }
128}
129
130/// Metadata about a puffin file.
131///
132/// For more information, see: https://iceberg.apache.org/puffin-spec/#filemetadata
133#[derive(Debug, PartialEq, Eq, Serialize, Deserialize, Clone)]
134pub struct FileMetadata {
135    pub(crate) blobs: Vec<BlobMetadata>,
136    #[serde(skip_serializing_if = "HashMap::is_empty")]
137    #[serde(default)]
138    pub(crate) properties: HashMap<String, String>,
139}
140
141impl FileMetadata {
142    pub(crate) const MAGIC_LENGTH: u8 = 4;
143    pub(crate) const MAGIC: [u8; FileMetadata::MAGIC_LENGTH as usize] = [0x50, 0x46, 0x41, 0x31];
144
145    /// We use the term FOOTER_STRUCT to refer to the fixed-length portion of the Footer.
146    /// The structure of the Footer specification is illustrated below:
147    ///
148    /// ```text                                             
149    ///        Footer
150    ///        ┌────────────────────┐                 
151    ///        │  Magic (4 bytes)   │                 
152    ///        │                    │                 
153    ///        ├────────────────────┤                 
154    ///        │   FooterPayload    │                 
155    ///        │  (PAYLOAD_LENGTH)  │                 
156    ///        ├────────────────────┤ ◀─┐             
157    ///        │ FooterPayloadSize  │   │             
158    ///        │     (4 bytes)      │   │             
159    ///        ├────────────────────┤                 
160    ///        │  Flags (4 bytes)   │  FOOTER_STRUCT  
161    ///        │                    │                 
162    ///        ├────────────────────┤   │             
163    ///        │  Magic (4 bytes)   │   │             
164    ///        │                    │   │             
165    ///        └────────────────────┘ ◀─┘  
166    /// ```                      
167    const FOOTER_STRUCT_PAYLOAD_LENGTH_OFFSET: u8 = 0;
168    const FOOTER_STRUCT_PAYLOAD_LENGTH_LENGTH: u8 = 4;
169    const FOOTER_STRUCT_FLAGS_OFFSET: u8 = FileMetadata::FOOTER_STRUCT_PAYLOAD_LENGTH_OFFSET
170        + FileMetadata::FOOTER_STRUCT_PAYLOAD_LENGTH_LENGTH;
171    pub(crate) const FOOTER_STRUCT_FLAGS_LENGTH: u8 = 4;
172    const FOOTER_STRUCT_MAGIC_OFFSET: u8 =
173        FileMetadata::FOOTER_STRUCT_FLAGS_OFFSET + FileMetadata::FOOTER_STRUCT_FLAGS_LENGTH;
174    pub(crate) const FOOTER_STRUCT_LENGTH: u8 =
175        FileMetadata::FOOTER_STRUCT_MAGIC_OFFSET + FileMetadata::MAGIC_LENGTH;
176
177    /// Smallest possible Puffin file: the file header magic, followed by a footer
178    /// holding an empty payload (footer magic + FOOTER_STRUCT).
179    const MIN_FILE_LENGTH: u64 =
180        (FileMetadata::MAGIC_LENGTH as u64) * 2 + (FileMetadata::FOOTER_STRUCT_LENGTH as u64);
181
182    /// Constructs new puffin `FileMetadata`
183    pub fn new(blobs: Vec<BlobMetadata>, properties: HashMap<String, String>) -> Self {
184        Self { blobs, properties }
185    }
186
187    fn check_magic(bytes: &[u8]) -> Result<()> {
188        if bytes == FileMetadata::MAGIC {
189            Ok(())
190        } else {
191            Err(Error::new(
192                ErrorKind::DataInvalid,
193                format!(
194                    "Bad magic value: {:?} should be {:?}",
195                    bytes,
196                    FileMetadata::MAGIC
197                ),
198            ))
199        }
200    }
201
202    async fn read_footer_payload_length(
203        file_read: &dyn FileRead,
204        input_file_length: u64,
205    ) -> Result<u32> {
206        let start = input_file_length - FileMetadata::FOOTER_STRUCT_LENGTH as u64;
207        let end = start + FileMetadata::FOOTER_STRUCT_PAYLOAD_LENGTH_LENGTH as u64;
208        let footer_payload_length_bytes = file_read.read(start..end).await?;
209        let mut buf = [0; 4];
210        buf.copy_from_slice(&footer_payload_length_bytes);
211        let footer_payload_length = u32::from_le_bytes(buf);
212        Ok(footer_payload_length)
213    }
214
215    async fn read_footer_bytes(
216        file_read: &dyn FileRead,
217        input_file_length: u64,
218        footer_payload_length: u32,
219    ) -> Result<Bytes> {
220        let footer_length = footer_payload_length as u64
221            + FileMetadata::FOOTER_STRUCT_LENGTH as u64
222            + FileMetadata::MAGIC_LENGTH as u64;
223        let start = input_file_length
224            .checked_sub(footer_length)
225            .ok_or_else(|| {
226                Error::new(
227                    ErrorKind::DataInvalid,
228                    format!(
229                        "Footer length {footer_length} exceeds file length {input_file_length}"
230                    ),
231                )
232            })?;
233        let end = input_file_length;
234        file_read.read(start..end).await
235    }
236
237    fn decode_flags(footer_bytes: &[u8]) -> Result<HashSet<Flag>> {
238        let mut flags = HashSet::new();
239
240        for byte_idx in 0..FileMetadata::FOOTER_STRUCT_FLAGS_LENGTH {
241            let byte_offset = footer_bytes.len()
242                - FileMetadata::MAGIC_LENGTH as usize
243                - FileMetadata::FOOTER_STRUCT_FLAGS_LENGTH as usize
244                + byte_idx as usize;
245
246            let flag_byte = *footer_bytes.get(byte_offset).ok_or_else(|| {
247                Error::new(ErrorKind::DataInvalid, "Index range is out of bounds.")
248            })?;
249
250            for bit_idx in 0..8 {
251                if ((flag_byte >> bit_idx) & 1) != 0 {
252                    let flag = Flag::from(byte_idx, bit_idx)?;
253                    flags.insert(flag);
254                }
255            }
256        }
257
258        Ok(flags)
259    }
260
261    fn extract_footer_payload_as_str(
262        footer_bytes: &[u8],
263        footer_payload_length: u32,
264    ) -> Result<String> {
265        let flags = FileMetadata::decode_flags(footer_bytes)?;
266        let footer_compression_codec = if flags.contains(&Flag::FooterPayloadCompressed) {
267            CompressionCodec::Lz4
268        } else {
269            CompressionCodec::None
270        };
271
272        let start_offset = FileMetadata::MAGIC_LENGTH as usize;
273        let end_offset =
274            FileMetadata::MAGIC_LENGTH as usize + usize::try_from(footer_payload_length)?;
275        let footer_payload_bytes = footer_bytes
276            .get(start_offset..end_offset)
277            .ok_or_else(|| Error::new(ErrorKind::DataInvalid, "Index range is out of bounds."))?;
278        let decompressed_footer_payload_bytes =
279            footer_compression_codec.decompress(footer_payload_bytes.into())?;
280
281        String::from_utf8(decompressed_footer_payload_bytes).map_err(|src| {
282            Error::new(ErrorKind::DataInvalid, "Footer is not a valid UTF-8 string")
283                .with_source(src)
284        })
285    }
286
287    fn from_json_str(string: &str) -> Result<FileMetadata> {
288        serde_json::from_str::<FileMetadata>(string).map_err(|src| {
289            Error::new(ErrorKind::DataInvalid, "Given string is not valid JSON").with_source(src)
290        })
291    }
292
293    /// Returns the file metadata about a Puffin file
294    pub(crate) async fn read(file_read: &dyn FileRead, file_length: u64) -> Result<FileMetadata> {
295        if file_length < FileMetadata::MIN_FILE_LENGTH {
296            return Err(Error::new(
297                ErrorKind::DataInvalid,
298                format!(
299                    "File length {} is too short to be a Puffin file, expected at least {} bytes",
300                    file_length,
301                    FileMetadata::MIN_FILE_LENGTH
302                ),
303            ));
304        }
305
306        let first_four_bytes = file_read.read(0..FileMetadata::MAGIC_LENGTH.into()).await?;
307        FileMetadata::check_magic(&first_four_bytes)?;
308
309        let footer_payload_length =
310            FileMetadata::read_footer_payload_length(file_read, file_length).await?;
311        let footer_bytes =
312            FileMetadata::read_footer_bytes(file_read, file_length, footer_payload_length).await?;
313
314        let magic_length = FileMetadata::MAGIC_LENGTH as usize;
315        // check first four bytes of footer
316        FileMetadata::check_magic(&footer_bytes[..magic_length])?;
317        // check last four bytes of footer
318        FileMetadata::check_magic(&footer_bytes[footer_bytes.len() - magic_length..])?;
319
320        let footer_payload_str =
321            FileMetadata::extract_footer_payload_as_str(&footer_bytes, footer_payload_length)?;
322
323        FileMetadata::from_json_str(&footer_payload_str)
324    }
325
326    /// Reads file_metadata in puffin file with a prefetch hint
327    ///
328    /// `prefetch_hint` is used to try to fetch the entire footer in one read. If
329    /// the entire footer isn't fetched in one read the function will call the regular
330    /// read option.
331    #[allow(dead_code)]
332    pub(crate) async fn read_with_prefetch(
333        file_read: &dyn FileRead,
334        file_length: u64,
335        prefetch_hint: u8,
336    ) -> Result<FileMetadata> {
337        if prefetch_hint > 16 {
338            // Hint cannot be larger than input file
339            if prefetch_hint as u64 > file_length {
340                return FileMetadata::read(file_read, file_length).await;
341            }
342
343            // Validate file header magic
344            let first_four_bytes = file_read.read(0..FileMetadata::MAGIC_LENGTH.into()).await?;
345            FileMetadata::check_magic(&first_four_bytes)?;
346
347            // Read footer based on prefetch hint
348            let start = file_length - prefetch_hint as u64;
349            let end = file_length;
350            let footer_bytes = file_read.read(start..end).await?;
351
352            let payload_length_start =
353                footer_bytes.len() - (FileMetadata::FOOTER_STRUCT_LENGTH as usize);
354            let payload_length_end =
355                payload_length_start + (FileMetadata::FOOTER_STRUCT_PAYLOAD_LENGTH_LENGTH as usize);
356            let payload_length_bytes = &footer_bytes[payload_length_start..payload_length_end];
357
358            let mut buf = [0; 4];
359            buf.copy_from_slice(payload_length_bytes);
360            let footer_payload_length = u32::from_le_bytes(buf);
361
362            // If the (footer payload length + FOOTER_STRUCT_LENGTH + MAGIC_LENGTH) is greater
363            // than the fetched footer then you can have it read regularly from a read with no
364            // prefetch while passing in the footer_payload_length.
365            let footer_length = (footer_payload_length as usize)
366                + FileMetadata::FOOTER_STRUCT_LENGTH as usize
367                + FileMetadata::MAGIC_LENGTH as usize;
368            if footer_length > prefetch_hint as usize {
369                return FileMetadata::read(file_read, file_length).await;
370            }
371
372            // Read footer bytes
373            let footer_start = footer_bytes.len() - footer_length;
374            let footer_end = footer_bytes.len();
375            let footer_bytes = &footer_bytes[footer_start..footer_end];
376
377            let magic_length = FileMetadata::MAGIC_LENGTH as usize;
378            // check first four bytes of footer
379            FileMetadata::check_magic(&footer_bytes[..magic_length])?;
380            // check last four bytes of footer
381            FileMetadata::check_magic(&footer_bytes[footer_bytes.len() - magic_length..])?;
382
383            let footer_payload_str =
384                FileMetadata::extract_footer_payload_as_str(footer_bytes, footer_payload_length)?;
385            return FileMetadata::from_json_str(&footer_payload_str);
386        }
387
388        FileMetadata::read(file_read, file_length).await
389    }
390
391    #[inline]
392    /// Metadata about blobs in file
393    pub fn blobs(&self) -> &[BlobMetadata] {
394        &self.blobs
395    }
396
397    #[inline]
398    /// Arbitrary meta-information, like writer identification/version.
399    pub fn properties(&self) -> &HashMap<String, String> {
400        &self.properties
401    }
402}
403
404#[cfg(test)]
405mod tests {
406    use std::collections::HashMap;
407
408    use bytes::Bytes;
409    use tempfile::TempDir;
410
411    use crate::ErrorKind;
412    use crate::io::{FileIO, InputFile};
413    use crate::puffin::metadata::{BlobMetadata, CompressionCodec, FileMetadata};
414    use crate::puffin::test_utils::{
415        empty_footer_payload, empty_footer_payload_bytes, empty_footer_payload_bytes_length_bytes,
416        java_empty_uncompressed_input_file, java_uncompressed_metric_input_file,
417        java_zstd_compressed_metric_input_file, read_file_metadata,
418        read_file_metadata_with_prefetch, uncompressed_metric_file_metadata,
419        zstd_compressed_metric_file_metadata,
420    };
421
422    const INVALID_MAGIC_VALUE: [u8; 4] = [80, 70, 65, 0];
423
424    async fn input_file_with_bytes(temp_dir: &TempDir, slice: &[u8]) -> InputFile {
425        let file_io = FileIO::new_with_fs();
426
427        let path_buf = temp_dir.path().join("abc.puffin");
428        let temp_path = path_buf.to_str().unwrap();
429        let output_file = file_io.new_output(temp_path).unwrap();
430
431        output_file
432            .write(Bytes::copy_from_slice(slice))
433            .await
434            .unwrap();
435
436        output_file.to_input_file()
437    }
438
439    async fn input_file_with_payload(temp_dir: &TempDir, payload_str: &str) -> InputFile {
440        let payload_bytes = payload_str.as_bytes();
441
442        let mut bytes = vec![];
443        bytes.extend(FileMetadata::MAGIC.to_vec());
444        bytes.extend(FileMetadata::MAGIC.to_vec());
445        bytes.extend(payload_bytes);
446        bytes.extend(u32::to_le_bytes(payload_bytes.len() as u32));
447        bytes.extend(vec![0, 0, 0, 0]);
448        bytes.extend(FileMetadata::MAGIC);
449
450        input_file_with_bytes(temp_dir, &bytes).await
451    }
452
453    #[tokio::test]
454    async fn test_file_starting_with_invalid_magic_returns_error() {
455        let temp_dir = TempDir::new().unwrap();
456
457        let mut bytes = vec![];
458        bytes.extend(INVALID_MAGIC_VALUE.to_vec());
459        bytes.extend(FileMetadata::MAGIC.to_vec());
460        bytes.extend(empty_footer_payload_bytes());
461        bytes.extend(empty_footer_payload_bytes_length_bytes());
462        bytes.extend(vec![0, 0, 0, 0]);
463        bytes.extend(FileMetadata::MAGIC);
464
465        let input_file = input_file_with_bytes(&temp_dir, &bytes).await;
466
467        assert_eq!(
468            read_file_metadata(&input_file)
469                .await
470                .unwrap_err()
471                .to_string(),
472            "DataInvalid => Bad magic value: [80, 70, 65, 0] should be [80, 70, 65, 49]",
473        )
474    }
475
476    #[tokio::test]
477    async fn test_file_with_invalid_magic_at_start_of_footer_returns_error() {
478        let temp_dir = TempDir::new().unwrap();
479
480        let mut bytes = vec![];
481        bytes.extend(FileMetadata::MAGIC.to_vec());
482        bytes.extend(INVALID_MAGIC_VALUE.to_vec());
483        bytes.extend(empty_footer_payload_bytes());
484        bytes.extend(empty_footer_payload_bytes_length_bytes());
485        bytes.extend(vec![0, 0, 0, 0]);
486        bytes.extend(FileMetadata::MAGIC);
487
488        let input_file = input_file_with_bytes(&temp_dir, &bytes).await;
489
490        assert_eq!(
491            read_file_metadata(&input_file)
492                .await
493                .unwrap_err()
494                .to_string(),
495            "DataInvalid => Bad magic value: [80, 70, 65, 0] should be [80, 70, 65, 49]",
496        )
497    }
498
499    #[tokio::test]
500    async fn test_file_ending_with_invalid_magic_returns_error() {
501        let temp_dir = TempDir::new().unwrap();
502
503        let mut bytes = vec![];
504        bytes.extend(FileMetadata::MAGIC.to_vec());
505        bytes.extend(FileMetadata::MAGIC.to_vec());
506        bytes.extend(empty_footer_payload_bytes());
507        bytes.extend(empty_footer_payload_bytes_length_bytes());
508        bytes.extend(vec![0, 0, 0, 0]);
509        bytes.extend(INVALID_MAGIC_VALUE);
510
511        let input_file = input_file_with_bytes(&temp_dir, &bytes).await;
512
513        assert_eq!(
514            read_file_metadata(&input_file)
515                .await
516                .unwrap_err()
517                .to_string(),
518            "DataInvalid => Bad magic value: [80, 70, 65, 0] should be [80, 70, 65, 49]",
519        )
520    }
521
522    #[tokio::test]
523    async fn test_encoded_payload_length_larger_than_actual_payload_length_returns_error() {
524        let temp_dir = TempDir::new().unwrap();
525
526        let mut bytes = vec![];
527        bytes.extend(FileMetadata::MAGIC.to_vec());
528        bytes.extend(FileMetadata::MAGIC.to_vec());
529        bytes.extend(empty_footer_payload_bytes());
530        bytes.extend(u32::to_le_bytes(
531            empty_footer_payload_bytes().len() as u32 + 1,
532        ));
533        bytes.extend(vec![0, 0, 0, 0]);
534        bytes.extend(FileMetadata::MAGIC.to_vec());
535
536        let input_file = input_file_with_bytes(&temp_dir, &bytes).await;
537
538        assert_eq!(
539            read_file_metadata(&input_file)
540                .await
541                .unwrap_err()
542                .to_string(),
543            "DataInvalid => Bad magic value: [49, 80, 70, 65] should be [80, 70, 65, 49]",
544        )
545    }
546
547    #[tokio::test]
548    async fn test_encoded_payload_length_smaller_than_actual_payload_length_returns_error() {
549        let temp_dir = TempDir::new().unwrap();
550
551        let mut bytes = vec![];
552        bytes.extend(FileMetadata::MAGIC.to_vec());
553        bytes.extend(FileMetadata::MAGIC.to_vec());
554        bytes.extend(empty_footer_payload_bytes());
555        bytes.extend(u32::to_le_bytes(
556            empty_footer_payload_bytes().len() as u32 - 1,
557        ));
558        bytes.extend(vec![0, 0, 0, 0]);
559        bytes.extend(FileMetadata::MAGIC.to_vec());
560
561        let input_file = input_file_with_bytes(&temp_dir, &bytes).await;
562
563        assert_eq!(
564            read_file_metadata(&input_file)
565                .await
566                .unwrap_err()
567                .to_string(),
568            "DataInvalid => Bad magic value: [70, 65, 49, 123] should be [80, 70, 65, 49]",
569        )
570    }
571
572    #[tokio::test]
573    async fn test_lz4_compressed_footer_returns_error() {
574        let temp_dir = TempDir::new().unwrap();
575
576        let mut bytes = vec![];
577        bytes.extend(FileMetadata::MAGIC.to_vec());
578        bytes.extend(FileMetadata::MAGIC.to_vec());
579        bytes.extend(empty_footer_payload_bytes());
580        bytes.extend(empty_footer_payload_bytes_length_bytes());
581        bytes.extend(vec![0b00000001, 0, 0, 0]);
582        bytes.extend(FileMetadata::MAGIC.to_vec());
583
584        let input_file = input_file_with_bytes(&temp_dir, &bytes).await;
585
586        assert_eq!(
587            read_file_metadata(&input_file)
588                .await
589                .unwrap_err()
590                .to_string(),
591            "FeatureUnsupported => LZ4 decompression is not supported currently",
592        )
593    }
594
595    #[tokio::test]
596    async fn test_unknown_byte_bit_combination_returns_error() {
597        let temp_dir = TempDir::new().unwrap();
598
599        let mut bytes = vec![];
600        bytes.extend(FileMetadata::MAGIC.to_vec());
601        bytes.extend(FileMetadata::MAGIC.to_vec());
602        bytes.extend(empty_footer_payload_bytes());
603        bytes.extend(empty_footer_payload_bytes_length_bytes());
604        bytes.extend(vec![0b00000010, 0, 0, 0]);
605        bytes.extend(FileMetadata::MAGIC.to_vec());
606
607        let input_file = input_file_with_bytes(&temp_dir, &bytes).await;
608
609        assert_eq!(
610            read_file_metadata(&input_file)
611                .await
612                .unwrap_err()
613                .to_string(),
614            "DataInvalid => Unknown flag byte 0 and bit 1 combination",
615        )
616    }
617
618    #[tokio::test]
619    async fn test_non_utf8_string_payload_returns_error() {
620        let temp_dir = TempDir::new().unwrap();
621
622        let payload_bytes: [u8; 4] = [0, 159, 146, 150];
623        let payload_bytes_length_bytes: [u8; 4] = u32::to_le_bytes(payload_bytes.len() as u32);
624
625        let mut bytes = vec![];
626        bytes.extend(FileMetadata::MAGIC.to_vec());
627        bytes.extend(FileMetadata::MAGIC.to_vec());
628        bytes.extend(payload_bytes);
629        bytes.extend(payload_bytes_length_bytes);
630        bytes.extend(vec![0, 0, 0, 0]);
631        bytes.extend(FileMetadata::MAGIC.to_vec());
632
633        let input_file = input_file_with_bytes(&temp_dir, &bytes).await;
634
635        assert_eq!(
636            read_file_metadata(&input_file)
637                .await
638                .unwrap_err()
639                .to_string(),
640            "DataInvalid => Footer is not a valid UTF-8 string, source: invalid utf-8 sequence of 1 bytes from index 1",
641        )
642    }
643
644    #[tokio::test]
645    async fn test_file_shorter_than_minimum_length_returns_error() {
646        let temp_dir = TempDir::new().unwrap();
647
648        // Only the file header magic, nothing else.
649        let input_file = input_file_with_bytes(&temp_dir, &FileMetadata::MAGIC).await;
650
651        let err = read_file_metadata(&input_file).await.unwrap_err();
652        assert_eq!(err.kind(), ErrorKind::DataInvalid);
653        assert!(
654            err.to_string().contains("too short to be a Puffin file"),
655            "unexpected error: {err}"
656        );
657    }
658
659    #[tokio::test]
660    async fn test_footer_payload_length_larger_than_file_returns_error() {
661        let temp_dir = TempDir::new().unwrap();
662
663        let mut bytes = vec![];
664        bytes.extend(FileMetadata::MAGIC.to_vec());
665        bytes.extend(FileMetadata::MAGIC.to_vec());
666        bytes.extend(empty_footer_payload_bytes());
667        // Declared footer payload length is far larger than the file itself.
668        bytes.extend(u32::to_le_bytes(u32::MAX));
669        bytes.extend(vec![0, 0, 0, 0]);
670        bytes.extend(FileMetadata::MAGIC);
671
672        let input_file = input_file_with_bytes(&temp_dir, &bytes).await;
673
674        let err = read_file_metadata(&input_file).await.unwrap_err();
675        assert_eq!(err.kind(), ErrorKind::DataInvalid);
676        assert!(
677            err.to_string().contains("exceeds file length"),
678            "unexpected error: {err}"
679        );
680    }
681
682    #[tokio::test]
683    async fn test_minimal_valid_file_returns_file_metadata() {
684        let temp_dir = TempDir::new().unwrap();
685
686        let mut bytes = vec![];
687        bytes.extend(FileMetadata::MAGIC.to_vec());
688        bytes.extend(FileMetadata::MAGIC.to_vec());
689        bytes.extend(empty_footer_payload_bytes());
690        bytes.extend(empty_footer_payload_bytes_length_bytes());
691        bytes.extend(vec![0, 0, 0, 0]);
692        bytes.extend(FileMetadata::MAGIC);
693
694        let input_file = input_file_with_bytes(&temp_dir, &bytes).await;
695
696        assert_eq!(
697            read_file_metadata(&input_file).await.unwrap(),
698            FileMetadata {
699                blobs: vec![],
700                properties: HashMap::new(),
701            }
702        )
703    }
704
705    #[tokio::test]
706    async fn test_returns_file_metadata_property() {
707        let temp_dir = TempDir::new().unwrap();
708
709        let input_file = input_file_with_payload(
710            &temp_dir,
711            r#"{
712                "blobs" : [ ],
713                "properties" : {
714                    "a property" : "a property value"
715                }
716            }"#,
717        )
718        .await;
719
720        assert_eq!(
721            read_file_metadata(&input_file).await.unwrap(),
722            FileMetadata {
723                blobs: vec![],
724                properties: {
725                    let mut map = HashMap::new();
726                    map.insert("a property".to_string(), "a property value".to_string());
727                    map
728                },
729            }
730        )
731    }
732
733    #[tokio::test]
734    async fn test_returns_file_metadata_properties() {
735        let temp_dir = TempDir::new().unwrap();
736
737        let input_file = input_file_with_payload(
738            &temp_dir,
739            r#"{
740                "blobs" : [ ],
741                "properties" : {
742                    "a property" : "a property value",
743                    "another one": "also with value"
744                }
745            }"#,
746        )
747        .await;
748
749        assert_eq!(
750            read_file_metadata(&input_file).await.unwrap(),
751            FileMetadata {
752                blobs: vec![],
753                properties: {
754                    let mut map = HashMap::new();
755                    map.insert("a property".to_string(), "a property value".to_string());
756                    map.insert("another one".to_string(), "also with value".to_string());
757                    map
758                },
759            }
760        )
761    }
762
763    #[tokio::test]
764    async fn test_returns_error_if_blobs_field_is_missing() {
765        let temp_dir = TempDir::new().unwrap();
766
767        let input_file = input_file_with_payload(
768            &temp_dir,
769            r#"{
770                "properties" : {}
771            }"#,
772        )
773        .await;
774
775        assert_eq!(
776            read_file_metadata(&input_file)
777                .await
778                .unwrap_err()
779                .to_string(),
780            format!(
781                "DataInvalid => Given string is not valid JSON, source: missing field `blobs` at line 3 column 13"
782            ),
783        )
784    }
785
786    #[tokio::test]
787    async fn test_returns_error_if_blobs_field_is_bad() {
788        let temp_dir = TempDir::new().unwrap();
789
790        let input_file = input_file_with_payload(
791            &temp_dir,
792            r#"{
793                "blobs" : {}
794            }"#,
795        )
796        .await;
797
798        assert_eq!(
799            read_file_metadata(&input_file)
800                .await
801                .unwrap_err()
802                .to_string(),
803            format!(
804                "DataInvalid => Given string is not valid JSON, source: invalid type: map, expected a sequence at line 2 column 26"
805            ),
806        )
807    }
808
809    #[tokio::test]
810    async fn test_returns_blobs_metadatas() {
811        let temp_dir = TempDir::new().unwrap();
812
813        let input_file = input_file_with_payload(
814            &temp_dir,
815            r#"{
816                "blobs" : [
817                    {
818                        "type" : "type-a",
819                        "fields" : [ 1 ],
820                        "snapshot-id" : 14,
821                        "sequence-number" : 3,
822                        "offset" : 4,
823                        "length" : 16
824                    },
825                    {
826                        "type" : "type-bbb",
827                        "fields" : [ 2, 3, 4 ],
828                        "snapshot-id" : 77,
829                        "sequence-number" : 4,
830                        "offset" : 21474836470000,
831                        "length" : 79834
832                    }
833                ]
834            }"#,
835        )
836        .await;
837
838        assert_eq!(
839            read_file_metadata(&input_file).await.unwrap(),
840            FileMetadata {
841                blobs: vec![
842                    BlobMetadata {
843                        r#type: "type-a".to_string(),
844                        fields: vec![1],
845                        snapshot_id: 14,
846                        sequence_number: 3,
847                        offset: 4,
848                        length: 16,
849                        compression_codec: CompressionCodec::None,
850                        properties: HashMap::new(),
851                    },
852                    BlobMetadata {
853                        r#type: "type-bbb".to_string(),
854                        fields: vec![2, 3, 4],
855                        snapshot_id: 77,
856                        sequence_number: 4,
857                        offset: 21474836470000,
858                        length: 79834,
859                        compression_codec: CompressionCodec::None,
860                        properties: HashMap::new(),
861                    },
862                ],
863                properties: HashMap::new(),
864            }
865        )
866    }
867
868    #[tokio::test]
869    async fn test_returns_properties_in_blob_metadata() {
870        let temp_dir = TempDir::new().unwrap();
871
872        let input_file = input_file_with_payload(
873            &temp_dir,
874            r#"{
875                "blobs" : [
876                    {
877                        "type" : "type-a",
878                        "fields" : [ 1 ],
879                        "snapshot-id" : 14,
880                        "sequence-number" : 3,
881                        "offset" : 4,
882                        "length" : 16,
883                        "properties" : {
884                            "some key" : "some value"
885                        }
886                    }
887                ]
888            }"#,
889        )
890        .await;
891
892        assert_eq!(
893            read_file_metadata(&input_file).await.unwrap(),
894            FileMetadata {
895                blobs: vec![BlobMetadata {
896                    r#type: "type-a".to_string(),
897                    fields: vec![1],
898                    snapshot_id: 14,
899                    sequence_number: 3,
900                    offset: 4,
901                    length: 16,
902                    compression_codec: CompressionCodec::None,
903                    properties: {
904                        let mut map = HashMap::new();
905                        map.insert("some key".to_string(), "some value".to_string());
906                        map
907                    },
908                }],
909                properties: HashMap::new(),
910            }
911        )
912    }
913
914    #[tokio::test]
915    async fn test_returns_error_if_blobs_fields_value_is_outside_i32_range() {
916        let temp_dir = TempDir::new().unwrap();
917
918        let out_of_i32_range_number: i64 = i32::MAX as i64 + 1;
919
920        let input_file = input_file_with_payload(
921            &temp_dir,
922            &format!(
923                r#"{{
924                    "blobs" : [
925                        {{
926                            "type" : "type-a",
927                            "fields" : [ {out_of_i32_range_number} ],
928                            "snapshot-id" : 14,
929                            "sequence-number" : 3,
930                            "offset" : 4,
931                            "length" : 16
932                        }}
933                    ]
934                }}"#
935            ),
936        )
937        .await;
938
939        assert_eq!(
940            read_file_metadata(&input_file)
941                .await
942                .unwrap_err()
943                .to_string(),
944            format!(
945                "DataInvalid => Given string is not valid JSON, source: invalid value: integer `{out_of_i32_range_number}`, expected i32 at line 5 column 51"
946            ),
947        )
948    }
949
950    #[tokio::test]
951    async fn test_returns_errors_if_footer_payload_is_not_encoded_in_json_format() {
952        let temp_dir = TempDir::new().unwrap();
953
954        let input_file = input_file_with_payload(&temp_dir, r#""blobs" = []"#).await;
955
956        assert_eq!(
957            read_file_metadata(&input_file)
958                .await
959                .unwrap_err()
960                .to_string(),
961            "DataInvalid => Given string is not valid JSON, source: invalid type: string \"blobs\", expected struct FileMetadata at line 1 column 7",
962        )
963    }
964
965    #[tokio::test]
966    async fn test_read_file_metadata_of_uncompressed_empty_file() {
967        let input_file = java_empty_uncompressed_input_file();
968
969        let file_metadata = read_file_metadata(&input_file).await.unwrap();
970        assert_eq!(file_metadata, empty_footer_payload())
971    }
972
973    #[tokio::test]
974    async fn test_read_file_metadata_of_uncompressed_metric_data() {
975        let input_file = java_uncompressed_metric_input_file();
976
977        let file_metadata = read_file_metadata(&input_file).await.unwrap();
978        assert_eq!(file_metadata, uncompressed_metric_file_metadata())
979    }
980
981    #[tokio::test]
982    async fn test_read_file_metadata_of_zstd_compressed_metric_data() {
983        let input_file = java_zstd_compressed_metric_input_file();
984
985        let file_metadata = read_file_metadata_with_prefetch(&input_file, 64)
986            .await
987            .unwrap();
988        assert_eq!(file_metadata, zstd_compressed_metric_file_metadata())
989    }
990
991    #[tokio::test]
992    async fn test_read_file_metadata_of_empty_file_with_prefetching() {
993        let input_file = java_empty_uncompressed_input_file();
994        let file_metadata = read_file_metadata_with_prefetch(&input_file, 64)
995            .await
996            .unwrap();
997
998        assert_eq!(file_metadata, empty_footer_payload());
999    }
1000
1001    #[tokio::test]
1002    async fn test_read_file_metadata_of_uncompressed_metric_data_with_prefetching() {
1003        let input_file = java_uncompressed_metric_input_file();
1004        let file_metadata = read_file_metadata_with_prefetch(&input_file, 64)
1005            .await
1006            .unwrap();
1007
1008        assert_eq!(file_metadata, uncompressed_metric_file_metadata());
1009    }
1010
1011    #[tokio::test]
1012    async fn test_read_file_metadata_of_zstd_compressed_metric_data_with_prefetching() {
1013        let input_file = java_zstd_compressed_metric_input_file();
1014        let file_metadata = read_file_metadata_with_prefetch(&input_file, 64)
1015            .await
1016            .unwrap();
1017
1018        assert_eq!(file_metadata, zstd_compressed_metric_file_metadata());
1019    }
1020
1021    #[tokio::test]
1022    async fn test_read_with_incorrect_header_magic() {
1023        let temp_dir = TempDir::new().unwrap();
1024
1025        let prefetch_hint: u8 = 64;
1026        let mut bytes = vec![];
1027        // Invalid header magic
1028        bytes.extend([0x00, 0x00, 0x00, 0x00]);
1029        // Intentionally keep file size larger than prefetch_hint.
1030        bytes.extend(vec![0u8; prefetch_hint as usize]);
1031        // Valid footer: magic + payload + footer struct
1032        bytes.extend(FileMetadata::MAGIC);
1033        bytes.extend(empty_footer_payload_bytes());
1034        bytes.extend(empty_footer_payload_bytes_length_bytes());
1035        bytes.extend(vec![0, 0, 0, 0]); // flags
1036        bytes.extend(FileMetadata::MAGIC);
1037
1038        let input_file = input_file_with_bytes(&temp_dir, &bytes).await;
1039
1040        assert_eq!(
1041            read_file_metadata(&input_file).await.unwrap_err().kind(),
1042            ErrorKind::DataInvalid,
1043        );
1044        assert_eq!(
1045            read_file_metadata_with_prefetch(&input_file, prefetch_hint)
1046                .await
1047                .unwrap_err()
1048                .kind(),
1049            ErrorKind::DataInvalid,
1050        );
1051    }
1052
1053    #[tokio::test]
1054    async fn test_gzip_compression_allowed_in_metadata() {
1055        let temp_dir = TempDir::new().unwrap();
1056
1057        // Create a JSON payload with Gzip compression codec
1058        // Metadata should be readable, but accessing the blob will fail
1059        let payload = r#"{
1060            "blobs": [
1061                {
1062                    "type": "test-type",
1063                    "fields": [1],
1064                    "snapshot-id": 1,
1065                    "sequence-number": 1,
1066                    "offset": 4,
1067                    "length": 10,
1068                    "compression-codec": "gzip"
1069                }
1070            ]
1071        }"#;
1072
1073        let input_file = input_file_with_payload(&temp_dir, payload).await;
1074
1075        // Reading metadata should succeed (lazy validation)
1076        let result = read_file_metadata(&input_file).await;
1077        assert!(result.is_ok());
1078        let metadata = result.unwrap();
1079        assert_eq!(metadata.blobs.len(), 1);
1080        assert_eq!(
1081            metadata.blobs[0].compression_codec,
1082            CompressionCodec::gzip_default()
1083        );
1084    }
1085}
iceberg/puffin/metadata.rs

iceberg/puffin/
metadata.rs
