icechunk_format/
lib.rs

1//! Icechunk data types and serialization.
2//!
3//! Defines ID types ([`SnapshotId`], [`ManifestId`], [`ChunkId`], [`NodeId`]),
4//! [`Path`] for normalized Zarr paths, and [`ByteRange`]/[`ChunkIndices`] for
5//! chunk addressing. Submodules define snapshots, manifests, and transaction logs.
6
7use core::fmt;
8use std::{
9    cmp::Ordering,
10    convert::Infallible,
11    fmt::{Debug, Display},
12    hash::Hash,
13    marker::PhantomData,
14    ops::Range,
15};
16
17use ::flatbuffers::InvalidFlatbuffer;
18use bytes::Bytes;
19use chrono::{DateTime, Utc};
20use flatbuffers::generated;
21use format_constants::FileTypeBin;
22use manifest::VirtualReferenceErrorKind;
23use rand::{RngExt as _, rng};
24use serde::{Deserialize, Serialize};
25use thiserror::Error;
26
27use icechunk_types::error::ICError;
28use icechunk_types::sealed;
29
30/// User attributes stored on arrays and groups.
31pub mod attributes;
32/// Chunk reference tables.
33pub mod manifest;
34
35/// Generated Flatbuffer types for binary serialization.
36#[path = "./flatbuffers/all_generated.rs"]
37#[allow(clippy::all, warnings)]
38pub mod flatbuffers;
39
40/// Repository metadata (version, properties).
41pub mod repo_info;
42/// Flatbuffer serialization.
43pub mod serializers;
44/// Repository state at a point in time.
45pub mod snapshot;
46/// Change records for commits.
47pub mod transaction_log;
48
49pub const CONFIG_FILE_PATH: &str = "config.yaml";
50pub const REPO_INFO_FILE_PATH: &str = "repo";
51pub const CHUNKS_FILE_PATH: &str = "chunks";
52pub const MANIFESTS_FILE_PATH: &str = "manifests";
53pub const SNAPSHOTS_FILE_PATH: &str = "snapshots";
54pub const TRANSACTION_LOGS_FILE_PATH: &str = "transactions";
55pub const OVERWRITTEN_FILES_PATH: &str = "overwritten";
56pub const V1_REFS_FILE_PATH: &str = "refs";
57
58pub use icechunk_types::{Path, PathError};
59
60/// Marker trait for object ID type tags (sealed).
61pub trait FileTypeTag: sealed::Sealed {}
62
63/// The id of a file in object store
64#[derive(Hash, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
65pub struct ObjectId<const SIZE: usize, T: FileTypeTag>(
66    #[serde(with = "serde_bytes")] pub [u8; SIZE],
67    PhantomData<T>,
68);
69
70/// Type tag for [`SnapshotId`].
71#[derive(Debug, Hash, Clone, PartialEq, Eq, PartialOrd, Ord)]
72pub struct SnapshotTag;
73
74/// Type tag for [`ManifestId`].
75#[derive(Debug, Hash, Clone, PartialEq, Eq, PartialOrd, Ord)]
76pub struct ManifestTag;
77
78/// Type tag for [`ChunkId`].
79#[derive(Debug, Hash, Clone, PartialEq, Eq, PartialOrd, Ord)]
80pub struct ChunkTag;
81
82/// Type tag for [`AttributesId`].
83#[derive(Debug, Hash, Clone, PartialEq, Eq, PartialOrd, Ord)]
84pub struct AttributesTag;
85
86/// Type tag for [`NodeId`].
87#[derive(Debug, Hash, Clone, PartialEq, Eq, PartialOrd, Ord)]
88pub struct NodeTag;
89
90impl sealed::Sealed for SnapshotTag {}
91impl sealed::Sealed for ManifestTag {}
92impl sealed::Sealed for ChunkTag {}
93impl sealed::Sealed for AttributesTag {}
94impl sealed::Sealed for NodeTag {}
95impl FileTypeTag for SnapshotTag {}
96impl FileTypeTag for ManifestTag {}
97impl FileTypeTag for ChunkTag {}
98impl FileTypeTag for AttributesTag {}
99impl FileTypeTag for NodeTag {}
100
101// A 1e-9 conflict probability requires 2^33 ~ 8.5 bn chunks
102// using this site for the calculations: https://www.bdayprob.com/
103
104/// Unique identifier for a snapshot.
105pub type SnapshotId = ObjectId<12, SnapshotTag>;
106/// Unique identifier for a manifest file.
107pub type ManifestId = ObjectId<12, ManifestTag>;
108/// Unique identifier for a chunk.
109pub type ChunkId = ObjectId<12, ChunkTag>;
110/// Unique identifier for an attributes blob.
111pub type AttributesId = ObjectId<12, AttributesTag>;
112
113// A 1e-9 conflict probability requires 2^17.55 ~ 200k nodes
114/// The internal id of an array or group, unique only to a single store version
115pub type NodeId = ObjectId<8, NodeTag>;
116
117impl<const SIZE: usize, T: FileTypeTag> ObjectId<SIZE, T> {
118    pub fn random() -> Self {
119        let mut buf = [0u8; SIZE];
120        rng().fill(&mut buf[..]);
121        Self(buf, PhantomData)
122    }
123
124    pub const fn new(buf: [u8; SIZE]) -> Self {
125        Self(buf, PhantomData)
126    }
127
128    pub const FAKE: Self = Self([0; SIZE], PhantomData);
129}
130
131impl<const SIZE: usize, T: FileTypeTag> Debug for ObjectId<SIZE, T> {
132    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
133        write!(f, "{}", String::from(self))
134    }
135}
136
137impl<const SIZE: usize, T: FileTypeTag> TryFrom<&[u8]> for ObjectId<SIZE, T> {
138    type Error = &'static str;
139
140    fn try_from(value: &[u8]) -> Result<Self, Self::Error> {
141        let buf = value.try_into();
142        buf.map(|buf| ObjectId(buf, PhantomData))
143            .map_err(|_| "Invalid ObjectId buffer length")
144    }
145}
146
147impl<const SIZE: usize, T: FileTypeTag> TryFrom<&str> for ObjectId<SIZE, T> {
148    type Error = &'static str;
149
150    fn try_from(value: &str) -> Result<Self, Self::Error> {
151        let bytes = base32::decode(base32::Alphabet::Crockford, value);
152        let Some(bytes) = bytes else { return Err("Invalid ObjectId string") };
153        Self::try_from(bytes.as_slice())
154    }
155}
156
157impl<const SIZE: usize, T: FileTypeTag> From<&ObjectId<SIZE, T>> for String {
158    fn from(value: &ObjectId<SIZE, T>) -> Self {
159        base32::encode(base32::Alphabet::Crockford, &value.0)
160    }
161}
162
163impl<const SIZE: usize, T: FileTypeTag> From<[u8; SIZE]> for ObjectId<SIZE, T> {
164    fn from(value: [u8; SIZE]) -> Self {
165        ObjectId::new(value)
166    }
167}
168
169impl<const SIZE: usize, T: FileTypeTag> TryInto<String> for ObjectId<SIZE, T> {
170    type Error = Infallible;
171
172    fn try_into(self) -> Result<String, Self::Error> {
173        Ok(self.to_string())
174    }
175}
176
177impl<const SIZE: usize, T: FileTypeTag> TryInto<ObjectId<SIZE, T>> for String {
178    type Error = &'static str;
179
180    fn try_into(self) -> Result<ObjectId<SIZE, T>, Self::Error> {
181        self.as_str().try_into()
182    }
183}
184
185impl<const SIZE: usize, T: FileTypeTag> Display for ObjectId<SIZE, T> {
186    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
187        write!(f, "{}", String::from(self))
188    }
189}
190
191/// An ND index to an element in a chunk grid.
192#[derive(Clone, Debug, Hash, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
193pub struct ChunkIndices(pub Vec<u32>);
194
195/// Byte offset within a chunk.
196pub type ChunkOffset = u64;
197/// Size of a chunk in bytes.
198pub type ChunkLength = u64;
199
200impl<'a> From<generated::ChunkIndices<'a>> for ChunkIndices {
201    fn from(value: generated::ChunkIndices<'a>) -> Self {
202        ChunkIndices(value.coords().iter().collect())
203    }
204}
205
206/// A byte range within a chunk or object.
207#[derive(Debug, Clone, PartialEq, Eq, Hash)]
208pub enum ByteRange {
209    /// The fixed length range represented by the given `Range`
210    Bounded(Range<ChunkOffset>),
211    /// All bytes from the given offset (included) to the end of the object
212    From(ChunkOffset),
213    /// The last n bytes in the object
214    Last(ChunkLength),
215    /// All bytes up to the last n bytes in the object
216    Until(ChunkOffset),
217}
218
219impl From<Range<ChunkOffset>> for ByteRange {
220    fn from(value: Range<ChunkOffset>) -> Self {
221        ByteRange::Bounded(value)
222    }
223}
224
225impl ByteRange {
226    pub fn from_offset(offset: ChunkOffset) -> Self {
227        Self::From(offset)
228    }
229
230    pub fn from_offset_with_length(offset: ChunkOffset, length: ChunkOffset) -> Self {
231        Self::Bounded(offset..offset + length)
232    }
233
234    pub fn to_offset(offset: ChunkOffset) -> Self {
235        Self::Bounded(0..offset)
236    }
237
238    pub fn bounded(start: ChunkOffset, end: ChunkOffset) -> Self {
239        (start..end).into()
240    }
241
242    pub const ALL: Self = Self::From(0);
243
244    pub fn slice(&self, bytes: &Bytes) -> Bytes {
245        match self {
246            ByteRange::Bounded(range) => {
247                bytes.slice(range.start as usize..range.end as usize)
248            }
249            ByteRange::From(from) => bytes.slice(*from as usize..),
250            ByteRange::Last(n) => bytes.slice(bytes.len() - *n as usize..),
251            ByteRange::Until(n) => bytes.slice(0usize..bytes.len() - *n as usize),
252        }
253    }
254}
255
256impl From<(Option<ChunkOffset>, Option<ChunkOffset>)> for ByteRange {
257    fn from((start, end): (Option<ChunkOffset>, Option<ChunkOffset>)) -> Self {
258        match (start, end) {
259            (Some(start), Some(end)) => Self::Bounded(start..end),
260            (Some(start), None) => Self::From(start),
261            // NOTE: This is relied upon by zarr python
262            (None, Some(end)) => Self::Until(end),
263            (None, None) => Self::ALL,
264        }
265    }
266}
267
268/// Offset within a manifest table.
269pub type TableOffset = u32;
270
271/// Format-level error types.
272#[derive(Debug, Error)]
273#[non_exhaustive]
274pub enum IcechunkFormatErrorKind {
275    #[error(transparent)]
276    VirtualReferenceError(#[from] VirtualReferenceErrorKind),
277    #[error("node not found at `{path:?}`")]
278    NodeNotFound { path: Path },
279    #[error("chunk coordinates not found `{coords:?}`")]
280    ChunkCoordinatesNotFound { coords: ChunkIndices },
281    #[error("snapshot id not found `{snapshot_id}`")]
282    SnapshotIdNotFound { snapshot_id: SnapshotId },
283    #[error("branch already exists `{branch} -> {snapshot_id}`")]
284    BranchAlreadyExists { branch: String, snapshot_id: SnapshotId },
285    #[error("branch not found `{branch}`")]
286    BranchNotFound { branch: String },
287    #[error("tag already exists `{tag}`")]
288    TagAlreadyExists { tag: String },
289    #[error("icechunk does not allow tag reuse and tag was already deleted `{tag}`")]
290    TagPreviouslyDeleted { tag: String },
291    #[error("tag not found `{tag}`")]
292    TagNotFound { tag: String },
293    #[error("snapshot id is already present in the repository: `{snapshot_id}`")]
294    DuplicateSnapshotId { snapshot_id: SnapshotId },
295    #[error("manifest information cannot be found in snapshot for id `{manifest_id}`")]
296    ManifestInfoNotFound { manifest_id: ManifestId },
297    #[error("invalid magic numbers in file")]
298    InvalidMagicNumbers, // TODO: add more info
299    #[error(
300        "this repository uses Icechunk format version {found}, but this library only supports up to version {max_supported}. Please upgrade the icechunk library"
301    )]
302    InvalidSpecVersion { found: u8, max_supported: u8 },
303    #[error("this operation is not supported for Icechunk format version {version}")]
304    UnsupportedOperationForVersion { version: u8 },
305    #[error("Icechunk cannot read this file type, expected {expected:?} got {got}")]
306    InvalidFileType { expected: FileTypeBin, got: u8 }, // TODO: add more info
307    #[error("Icechunk cannot read file, invalid compression algorithm")]
308    InvalidCompressionAlgorithm, // TODO: add more info
309    #[error("Invalid Icechunk metadata file")]
310    InvalidFlatBuffer(#[from] InvalidFlatbuffer),
311    #[error("error during metadata deserialization")]
312    DeserializationError(#[from] Box<rmp_serde::decode::Error>),
313    #[error("error during metadata serialization")]
314    SerializationError(#[from] Box<rmp_serde::encode::Error>),
315    #[error("error during metadata serialization")]
316    SerializationErrorFlexBuffers(#[from] Box<flexbuffers::SerializationError>),
317    #[error("error during metadata deserialization")]
318    DeserializationErrorFlexBuffers(#[from] Box<flexbuffers::DeserializationError>),
319    #[error("I/O error")]
320    IO(#[from] std::io::Error),
321    #[error("path error")]
322    Path(#[from] PathError),
323    #[error("invalid timestamp in file")]
324    InvalidTimestamp,
325    #[error(
326        "update timestamp is invalid, please verify if the machine clock has drifted: update time: `{new_time}`, latest update time: `{latest_time}`"
327    )]
328    InvalidUpdateTimestamp { latest_time: DateTime<Utc>, new_time: DateTime<Utc> },
329    #[error("invalid feature flag name: {name}")]
330    InvalidFeatureFlagName { name: String },
331    #[error("invalid feature flag id: {id}")]
332    InvalidFeatureFlagId { id: u16 },
333    #[error("{feature_description} is disabled by a feature flag ({feature_flag})")]
334    FeatureFlagDisabled { feature_description: String, feature_flag: String },
335    #[error(
336        "compressed chunk location present but no decompression dictionary available"
337    )]
338    MissingLocationCompressionDictionary,
339    #[error("Invalid array metadata: {0}")]
340    InvalidArrayMetadata(String),
341}
342
343pub type IcechunkFormatError = ICError<IcechunkFormatErrorKind>;
344
345impl From<Infallible> for IcechunkFormatErrorKind {
346    fn from(value: Infallible) -> Self {
347        match value {}
348    }
349}
350
351pub type IcechunkResult<T> = Result<T, IcechunkFormatError>;
352
353/// Binary format constants (file types, spec versions, compression).
354pub mod format_constants {
355    use std::sync::LazyLock;
356
357    use serde::{Deserialize, Serialize};
358
359    use super::IcechunkFormatErrorKind;
360
361    /// Binary file type identifier in the file header.
362    #[repr(u8)]
363    #[derive(Debug, Clone, Copy, PartialEq, Eq)]
364    pub enum FileTypeBin {
365        Snapshot = 1u8,
366        Manifest = 2,
367        Attributes = 3,
368        TransactionLog = 4,
369        Chunk = 5,
370        RepoInfo = 6,
371    }
372
373    impl TryFrom<u8> for FileTypeBin {
374        type Error = String;
375
376        fn try_from(value: u8) -> Result<Self, Self::Error> {
377            match value {
378                n if n == FileTypeBin::Snapshot as u8 => Ok(FileTypeBin::Snapshot),
379                n if n == FileTypeBin::Manifest as u8 => Ok(FileTypeBin::Manifest),
380                n if n == FileTypeBin::Attributes as u8 => Ok(FileTypeBin::Attributes),
381                n if n == FileTypeBin::RepoInfo as u8 => Ok(FileTypeBin::RepoInfo),
382                n if n == FileTypeBin::TransactionLog as u8 => {
383                    Ok(FileTypeBin::TransactionLog)
384                }
385                n if n == FileTypeBin::Chunk as u8 => Ok(FileTypeBin::Chunk),
386                n => Err(format!("Bad file type code: {n}")),
387            }
388        }
389    }
390
391    /// Icechunk format specification version.
392    #[repr(u8)]
393    #[derive(
394        Debug,
395        Clone,
396        Copy,
397        PartialEq,
398        Eq,
399        Serialize,
400        Deserialize,
401        Default,
402        PartialOrd,
403        Ord,
404    )]
405    pub enum SpecVersionBin {
406        V1 = 1u8,
407        #[default]
408        V2 = 2u8,
409        // When adding new versions here, don't forget to update the
410        // PySpecVersion enum in icechunk-python/src/repository.rs too!
411    }
412
413    impl TryFrom<u8> for SpecVersionBin {
414        type Error = IcechunkFormatErrorKind;
415
416        fn try_from(value: u8) -> Result<Self, Self::Error> {
417            match value {
418                n if n == SpecVersionBin::V1 as u8 => Ok(SpecVersionBin::V1),
419                n if n == SpecVersionBin::V2 as u8 => Ok(SpecVersionBin::V2),
420                n => Err(IcechunkFormatErrorKind::InvalidSpecVersion {
421                    found: n,
422                    max_supported: Self::current() as u8,
423                }),
424            }
425        }
426    }
427
428    impl SpecVersionBin {
429        pub fn current() -> Self {
430            Default::default()
431        }
432    }
433
434    impl std::fmt::Display for SpecVersionBin {
435        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
436            let s = match self {
437                SpecVersionBin::V1 => "1.0",
438                SpecVersionBin::V2 => "2.0",
439            };
440            write!(f, "{s}")
441        }
442    }
443
444    /// Compression algorithm used for metadata files.
445    #[repr(u8)]
446    #[derive(Debug, Clone, Copy, PartialEq, Eq)]
447    pub enum CompressionAlgorithmBin {
448        None = 0u8,
449        Zstd = 1u8,
450    }
451
452    impl TryFrom<u8> for CompressionAlgorithmBin {
453        type Error = String;
454
455        fn try_from(value: u8) -> Result<Self, Self::Error> {
456            match value {
457                n if n == CompressionAlgorithmBin::None as u8 => {
458                    Ok(CompressionAlgorithmBin::None)
459                }
460                n if n == CompressionAlgorithmBin::Zstd as u8 => {
461                    Ok(CompressionAlgorithmBin::Zstd)
462                }
463                n => Err(format!("Bad cmpression algorithm code: {n}")),
464            }
465        }
466    }
467
468    pub const ICECHUNK_FORMAT_MAGIC_BYTES: &[u8] = "ICE🧊CHUNK".as_bytes();
469
470    pub const LATEST_ICECHUNK_FORMAT_VERSION_METADATA_KEY: &str = "ic_spec_ver";
471
472    pub const ICECHUNK_LIB_VERSION: &str = env!("CARGO_PKG_VERSION");
473
474    pub static ICECHUNK_CLIENT_NAME: LazyLock<String> =
475        LazyLock::new(|| "ic-".to_string() + ICECHUNK_LIB_VERSION);
476    pub const ICECHUNK_CLIENT_NAME_METADATA_KEY: &str = "ic_client";
477
478    pub const ICECHUNK_FILE_TYPE_SNAPSHOT: &str = "snapshot";
479    pub const ICECHUNK_FILE_TYPE_MANIFEST: &str = "manifest";
480    pub const ICECHUNK_FILE_TYPE_TRANSACTION_LOG: &str = "transaction-log";
481    pub const ICECHUNK_FILE_TYPE_REPO_INFO: &str = "repo-info";
482    pub const ICECHUNK_FILE_TYPE_METADATA_KEY: &str = "ic_file_type";
483
484    pub const ICECHUNK_COMPRESSION_METADATA_KEY: &str = "ic_comp_alg";
485    pub const ICECHUNK_COMPRESSION_ZSTD: &str = "zstd";
486}
487
488#[inline(always)]
489#[expect(clippy::needless_pass_by_value)]
490pub fn lookup_index_by_key<'a, T: ::flatbuffers::Follow<'a> + 'a, K: Ord>(
491    v: ::flatbuffers::Vector<'a, T>,
492    key: K,
493    f: fn(&<T as ::flatbuffers::Follow<'a>>::Inner, &K) -> Ordering,
494) -> Option<usize> {
495    if v.is_empty() {
496        return None;
497    }
498
499    let mut left: usize = 0;
500    let mut right = v.len() - 1;
501
502    while left <= right {
503        let mid = (left + right) / 2;
504        let value = v.get(mid);
505        match f(&value, &key) {
506            Ordering::Equal => return Some(mid),
507            Ordering::Less => left = mid + 1,
508            Ordering::Greater => {
509                if mid == 0 {
510                    return None;
511                }
512                right = mid - 1;
513            }
514        }
515    }
516
517    None
518}
519
520// This macro is used for creating property tests
521// which check that serializing and deserializing
522// an instance of a type T is equivalent to the
523// identity function
524// Given pairs of test names and arbitraries to be used
525// for the tests, e.g., (n1, a1), (n2, a2),... (nx, ax)
526// the tests can be created by doing
527// roundtrip_serialization_tests!(n1 - a1, n2 - a2, .... nx - ax)
528#[macro_export]
529macro_rules! roundtrip_serialization_tests {
530    ($($test_name: ident - $generator: ident), +) => {
531        $(
532            proptest!{
533                #[icechunk_macros::test]
534                fn $test_name(elem in $generator()) {
535                    let bytes = rmp_serde::to_vec(&elem).unwrap();
536                    let roundtrip = rmp_serde::from_slice(&bytes).unwrap();
537                    assert_eq!(elem, roundtrip);
538                }
539            }
540        )*
541    }
542}
543
544#[cfg(test)]
545pub mod strategies;
546
547#[cfg(test)]
548mod tests {
549    use super::*;
550    use crate::strategies::{attributes_id, spec_version};
551    use pretty_assertions::assert_eq;
552    use proptest::prelude::*;
553
554    roundtrip_serialization_tests!(
555        serialize_and_deserialize_attribute_ids - attributes_id,
556        serialize_and_deserialize_spec_version_bin - spec_version
557    );
558
559    #[icechunk_macros::test]
560    fn test_object_id_serialization() {
561        let sid = SnapshotId::new([0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11]);
562        assert_eq!(
563            serde_json::to_string(&sid).unwrap(),
564            r#"[[0,1,2,3,4,5,6,7,8,9,10,11],null]"#
565        );
566        assert_eq!(String::from(&sid), "000G40R40M30E209185G");
567        assert_eq!(sid, SnapshotId::try_from("000G40R40M30E209185G").unwrap());
568        let sid = SnapshotId::random();
569        assert_eq!(
570            serde_json::from_slice::<SnapshotId>(
571                serde_json::to_vec(&sid).unwrap().as_slice()
572            )
573            .unwrap(),
574            sid,
575        );
576    }
577
578    #[icechunk_macros::test]
579    fn test_unknown_spec_version_gives_nice_error() {
580        use format_constants::SpecVersionBin;
581
582        let future_version: u8 = 3;
583        let result = SpecVersionBin::try_from(future_version);
584        assert!(result.is_err());
585        let err = result.unwrap_err();
586
587        let msg = err.to_string();
588        assert!(
589            msg.contains("format version 3"),
590            "Error should mention the found version: {msg}"
591        );
592        assert!(
593            msg.contains("upgrade the icechunk library"),
594            "Error should suggest upgrading: {msg}"
595        );
596    }
597}
icechunk_format/lib.rs

icechunk_format/
lib.rs
