1use 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
30pub mod attributes;
32pub mod manifest;
34
35#[path = "./flatbuffers/all_generated.rs"]
37#[allow(clippy::all, warnings)]
38pub mod flatbuffers;
39
40pub mod repo_info;
42pub mod serializers;
44pub mod snapshot;
46pub 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
60pub trait FileTypeTag: sealed::Sealed {}
62
63#[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#[derive(Debug, Hash, Clone, PartialEq, Eq, PartialOrd, Ord)]
72pub struct SnapshotTag;
73
74#[derive(Debug, Hash, Clone, PartialEq, Eq, PartialOrd, Ord)]
76pub struct ManifestTag;
77
78#[derive(Debug, Hash, Clone, PartialEq, Eq, PartialOrd, Ord)]
80pub struct ChunkTag;
81
82#[derive(Debug, Hash, Clone, PartialEq, Eq, PartialOrd, Ord)]
84pub struct AttributesTag;
85
86#[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
101pub type SnapshotId = ObjectId<12, SnapshotTag>;
106pub type ManifestId = ObjectId<12, ManifestTag>;
108pub type ChunkId = ObjectId<12, ChunkTag>;
110pub type AttributesId = ObjectId<12, AttributesTag>;
112
113pub type NodeId = ObjectId<8, NodeTag>;
116
117#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, Hash)]
118pub enum NodeType {
119 Group,
120 Array,
121}
122
123#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, Hash)]
124pub struct Move {
125 pub from: Path,
126 pub to: Path,
127 pub node_id: NodeId,
128 pub node_type: NodeType,
129}
130
131impl From<NodeType> for generated::NodeType {
132 fn from(value: NodeType) -> Self {
133 match value {
134 NodeType::Group => generated::NodeType::Group,
135 NodeType::Array => generated::NodeType::Array,
136 }
137 }
138}
139
140impl NodeType {
141 fn max_supported() -> u8 {
142 generated::NodeType::ENUM_MAX
143 }
144}
145
146impl TryFrom<generated::NodeType> for NodeType {
147 type Error = IcechunkFormatErrorKind;
148
149 fn try_from(value: generated::NodeType) -> Result<Self, Self::Error> {
150 match value {
151 generated::NodeType::Group => Ok(NodeType::Group),
152 generated::NodeType::Array => Ok(NodeType::Array),
153 generated::NodeType(v) => Err(IcechunkFormatErrorKind::InvalidNodeType {
154 found: v,
155 max_supported: Self::max_supported(),
156 }),
157 }
158 }
159}
160
161impl<const SIZE: usize, T: FileTypeTag> ObjectId<SIZE, T> {
162 pub fn random() -> Self {
163 let mut buf = [0u8; SIZE];
164 rng().fill(&mut buf[..]);
165 Self(buf, PhantomData)
166 }
167
168 pub const fn new(buf: [u8; SIZE]) -> Self {
169 Self(buf, PhantomData)
170 }
171
172 pub const FAKE: Self = Self([0; SIZE], PhantomData);
173}
174
175impl<const SIZE: usize, T: FileTypeTag> Debug for ObjectId<SIZE, T> {
176 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
177 write!(f, "{}", String::from(self))
178 }
179}
180
181impl<const SIZE: usize, T: FileTypeTag> TryFrom<&[u8]> for ObjectId<SIZE, T> {
182 type Error = &'static str;
183
184 fn try_from(value: &[u8]) -> Result<Self, Self::Error> {
185 let buf = value.try_into();
186 buf.map(|buf| ObjectId(buf, PhantomData))
187 .map_err(|_| "Invalid ObjectId buffer length")
188 }
189}
190
191impl<const SIZE: usize, T: FileTypeTag> TryFrom<&str> for ObjectId<SIZE, T> {
192 type Error = &'static str;
193
194 fn try_from(value: &str) -> Result<Self, Self::Error> {
195 let bytes = base32::decode(base32::Alphabet::Crockford, value);
196 let Some(bytes) = bytes else { return Err("Invalid ObjectId string") };
197 Self::try_from(bytes.as_slice())
198 }
199}
200
201impl<const SIZE: usize, T: FileTypeTag> From<&ObjectId<SIZE, T>> for String {
202 fn from(value: &ObjectId<SIZE, T>) -> Self {
203 base32::encode(base32::Alphabet::Crockford, &value.0)
204 }
205}
206
207impl<const SIZE: usize, T: FileTypeTag> From<[u8; SIZE]> for ObjectId<SIZE, T> {
208 fn from(value: [u8; SIZE]) -> Self {
209 ObjectId::new(value)
210 }
211}
212
213impl<const SIZE: usize, T: FileTypeTag> TryInto<String> for ObjectId<SIZE, T> {
214 type Error = Infallible;
215
216 fn try_into(self) -> Result<String, Self::Error> {
217 Ok(self.to_string())
218 }
219}
220
221impl<const SIZE: usize, T: FileTypeTag> TryInto<ObjectId<SIZE, T>> for String {
222 type Error = &'static str;
223
224 fn try_into(self) -> Result<ObjectId<SIZE, T>, Self::Error> {
225 self.as_str().try_into()
226 }
227}
228
229impl<const SIZE: usize, T: FileTypeTag> Display for ObjectId<SIZE, T> {
230 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
231 write!(f, "{}", String::from(self))
232 }
233}
234
235#[derive(Clone, Debug, Hash, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
237pub struct ChunkIndices(pub Vec<u32>);
238
239pub type ChunkOffset = u64;
241pub type ChunkLength = u64;
243
244impl<'a> From<generated::ChunkIndices<'a>> for ChunkIndices {
245 fn from(value: generated::ChunkIndices<'a>) -> Self {
246 ChunkIndices(value.coords().iter().collect())
247 }
248}
249
250#[derive(Debug, Clone, PartialEq, Eq, Hash)]
252pub enum ByteRange {
253 Bounded(Range<ChunkOffset>),
255 From(ChunkOffset),
257 Last(ChunkLength),
259 Until(ChunkOffset),
261}
262
263impl From<Range<ChunkOffset>> for ByteRange {
264 fn from(value: Range<ChunkOffset>) -> Self {
265 ByteRange::Bounded(value)
266 }
267}
268
269impl ByteRange {
270 pub fn from_offset(offset: ChunkOffset) -> Self {
271 Self::From(offset)
272 }
273
274 pub fn from_offset_with_length(offset: ChunkOffset, length: ChunkOffset) -> Self {
275 Self::Bounded(offset..offset + length)
276 }
277
278 pub fn to_offset(offset: ChunkOffset) -> Self {
279 Self::Bounded(0..offset)
280 }
281
282 pub fn bounded(start: ChunkOffset, end: ChunkOffset) -> Self {
283 (start..end).into()
284 }
285
286 pub const ALL: Self = Self::From(0);
287
288 pub fn slice(&self, bytes: &Bytes) -> Bytes {
289 match self {
290 ByteRange::Bounded(range) => {
291 bytes.slice(range.start as usize..range.end as usize)
292 }
293 ByteRange::From(from) => bytes.slice(*from as usize..),
294 ByteRange::Last(n) => bytes.slice(bytes.len() - *n as usize..),
295 ByteRange::Until(n) => bytes.slice(0usize..bytes.len() - *n as usize),
296 }
297 }
298}
299
300impl From<(Option<ChunkOffset>, Option<ChunkOffset>)> for ByteRange {
301 fn from((start, end): (Option<ChunkOffset>, Option<ChunkOffset>)) -> Self {
302 match (start, end) {
303 (Some(start), Some(end)) => Self::Bounded(start..end),
304 (Some(start), None) => Self::From(start),
305 (None, Some(end)) => Self::Until(end),
307 (None, None) => Self::ALL,
308 }
309 }
310}
311
312pub type TableOffset = u32;
314
315#[derive(Debug, Error)]
317#[non_exhaustive]
318pub enum IcechunkFormatErrorKind {
319 #[error(transparent)]
320 VirtualReferenceError(#[from] VirtualReferenceErrorKind),
321 #[error("node not found at `{path:?}`")]
322 NodeNotFound { path: Path },
323 #[error("chunk coordinates not found `{coords:?}`")]
324 ChunkCoordinatesNotFound { coords: ChunkIndices },
325 #[error("snapshot id not found `{snapshot_id}`")]
326 SnapshotIdNotFound { snapshot_id: SnapshotId },
327 #[error("branch already exists `{branch} -> {snapshot_id}`")]
328 BranchAlreadyExists { branch: String, snapshot_id: SnapshotId },
329 #[error("branch not found `{branch}`")]
330 BranchNotFound { branch: String },
331 #[error("tag already exists `{tag}`")]
332 TagAlreadyExists { tag: String },
333 #[error("icechunk does not allow tag reuse and tag was already deleted `{tag}`")]
334 TagPreviouslyDeleted { tag: String },
335 #[error("tag not found `{tag}`")]
336 TagNotFound { tag: String },
337 #[error("snapshot id is already present in the repository: `{snapshot_id}`")]
338 DuplicateSnapshotId { snapshot_id: SnapshotId },
339 #[error("manifest information cannot be found in snapshot for id `{manifest_id}`")]
340 ManifestInfoNotFound { manifest_id: ManifestId },
341 #[error("invalid magic numbers in file")]
342 InvalidMagicNumbers, #[error(
344 "invalid icechunk header size, got {found} bytes, expected at least {expected}"
345 )]
346 InvalidIcechunkHeaderSize { found: usize, expected: usize },
347 #[error("invalid node type {found}, max supported value is {max_supported}")]
348 InvalidNodeType { found: u8, max_supported: u8 },
349 #[error(
350 "this repository uses Icechunk format version {found}, but this library only supports up to version {max_supported}. Please upgrade the icechunk library"
351 )]
352 InvalidSpecVersion { found: u8, max_supported: u8 },
353 #[error("this operation is not supported for Icechunk format version {version}")]
354 UnsupportedOperationForVersion { version: u8 },
355 #[error("Icechunk cannot read this file type, expected {expected:?} got {got}")]
356 InvalidFileType { expected: FileTypeBin, got: u8 }, #[error("Icechunk cannot read file, invalid compression algorithm")]
358 InvalidCompressionAlgorithm, #[error("Invalid Icechunk metadata file")]
360 InvalidFlatBuffer(#[from] InvalidFlatbuffer),
361 #[error("error during metadata deserialization")]
362 DeserializationError(#[from] Box<rmp_serde::decode::Error>),
363 #[error("error during metadata serialization")]
364 SerializationError(#[from] Box<rmp_serde::encode::Error>),
365 #[error("error during metadata serialization")]
366 SerializationErrorFlexBuffers(#[from] Box<flexbuffers::SerializationError>),
367 #[error("error during metadata deserialization")]
368 DeserializationErrorFlexBuffers(#[from] Box<flexbuffers::DeserializationError>),
369 #[error("I/O error")]
370 IO(#[from] std::io::Error),
371 #[error("path error")]
372 Path(#[from] PathError),
373 #[error("invalid timestamp in file")]
374 InvalidTimestamp,
375 #[error(
376 "update timestamp is invalid, please verify if the machine clock has drifted: update time: `{new_time}`, latest update time: `{latest_time}`"
377 )]
378 InvalidUpdateTimestamp { latest_time: DateTime<Utc>, new_time: DateTime<Utc> },
379 #[error("invalid feature flag name: {name}")]
380 InvalidFeatureFlagName { name: String },
381 #[error("invalid feature flag id: {id}")]
382 InvalidFeatureFlagId { id: u16 },
383 #[error("{feature_description} is disabled by a feature flag ({feature_flag})")]
384 FeatureFlagDisabled { feature_description: String, feature_flag: String },
385 #[error(
386 "compressed chunk location present but no decompression dictionary available"
387 )]
388 MissingLocationCompressionDictionary,
389 #[error("Invalid array metadata: {0}")]
390 InvalidArrayMetadata(String),
391 #[error("Move operation missing required field '{0}'")]
392 MissingRequiredField(String),
393}
394
395pub type IcechunkFormatError = ICError<IcechunkFormatErrorKind>;
396
397impl From<Infallible> for IcechunkFormatErrorKind {
398 fn from(value: Infallible) -> Self {
399 match value {}
400 }
401}
402
403pub type IcechunkResult<T> = Result<T, IcechunkFormatError>;
404
405pub mod format_constants {
407 use std::sync::LazyLock;
408
409 use serde::{Deserialize, Serialize};
410
411 use super::IcechunkFormatErrorKind;
412
413 #[repr(u8)]
415 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
416 pub enum FileTypeBin {
417 Snapshot = 1u8,
418 Manifest = 2,
419 Attributes = 3,
420 TransactionLog = 4,
421 Chunk = 5,
422 RepoInfo = 6,
423 }
424
425 impl TryFrom<u8> for FileTypeBin {
426 type Error = String;
427
428 fn try_from(value: u8) -> Result<Self, Self::Error> {
429 match value {
430 n if n == FileTypeBin::Snapshot as u8 => Ok(FileTypeBin::Snapshot),
431 n if n == FileTypeBin::Manifest as u8 => Ok(FileTypeBin::Manifest),
432 n if n == FileTypeBin::Attributes as u8 => Ok(FileTypeBin::Attributes),
433 n if n == FileTypeBin::RepoInfo as u8 => Ok(FileTypeBin::RepoInfo),
434 n if n == FileTypeBin::TransactionLog as u8 => {
435 Ok(FileTypeBin::TransactionLog)
436 }
437 n if n == FileTypeBin::Chunk as u8 => Ok(FileTypeBin::Chunk),
438 n => Err(format!("Bad file type code: {n}")),
439 }
440 }
441 }
442
443 #[repr(u8)]
445 #[derive(
446 Debug,
447 Clone,
448 Copy,
449 PartialEq,
450 Eq,
451 Serialize,
452 Deserialize,
453 Default,
454 PartialOrd,
455 Ord,
456 )]
457 pub enum SpecVersionBin {
458 V1 = 1u8,
459 #[default]
460 V2 = 2u8,
461 }
464
465 impl TryFrom<u8> for SpecVersionBin {
466 type Error = IcechunkFormatErrorKind;
467
468 fn try_from(value: u8) -> Result<Self, Self::Error> {
469 match value {
470 n if n == SpecVersionBin::V1 as u8 => Ok(SpecVersionBin::V1),
471 n if n == SpecVersionBin::V2 as u8 => Ok(SpecVersionBin::V2),
472 n => Err(IcechunkFormatErrorKind::InvalidSpecVersion {
473 found: n,
474 max_supported: Self::current() as u8,
475 }),
476 }
477 }
478 }
479
480 impl SpecVersionBin {
481 pub fn current() -> Self {
482 Default::default()
483 }
484 }
485
486 impl std::fmt::Display for SpecVersionBin {
487 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
488 let s = match self {
489 SpecVersionBin::V1 => "1.0",
490 SpecVersionBin::V2 => "2.0",
491 };
492 write!(f, "{s}")
493 }
494 }
495
496 #[repr(u8)]
498 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
499 pub enum CompressionAlgorithmBin {
500 None = 0u8,
501 Zstd = 1u8,
502 }
503
504 impl TryFrom<u8> for CompressionAlgorithmBin {
505 type Error = String;
506
507 fn try_from(value: u8) -> Result<Self, Self::Error> {
508 match value {
509 n if n == CompressionAlgorithmBin::None as u8 => {
510 Ok(CompressionAlgorithmBin::None)
511 }
512 n if n == CompressionAlgorithmBin::Zstd as u8 => {
513 Ok(CompressionAlgorithmBin::Zstd)
514 }
515 n => Err(format!("Bad cmpression algorithm code: {n}")),
516 }
517 }
518 }
519
520 pub const ICECHUNK_FORMAT_MAGIC_BYTES: &[u8] = "ICE🧊CHUNK".as_bytes();
521 const _: () = assert!(ICECHUNK_FORMAT_MAGIC_BYTES.len() == 12);
523
524 pub const ICECHUNK_IMPL_NAME_LEN: usize = 24;
527 pub const ICECHUNK_SPEC_VERSION_OFFSET: usize =
528 ICECHUNK_FORMAT_MAGIC_BYTES.len() + ICECHUNK_IMPL_NAME_LEN;
529 pub const ICECHUNK_FILE_TYPE_OFFSET: usize = ICECHUNK_SPEC_VERSION_OFFSET + 1;
530 pub const ICECHUNK_COMPRESSION_OFFSET: usize = ICECHUNK_FILE_TYPE_OFFSET + 1;
531 pub const ICECHUNK_FILE_HEADER_LEN: usize = ICECHUNK_COMPRESSION_OFFSET + 1;
532
533 pub const LATEST_ICECHUNK_FORMAT_VERSION_METADATA_KEY: &str = "ic_spec_ver";
534
535 pub const ICECHUNK_LIB_VERSION: &str = env!("CARGO_PKG_VERSION");
536
537 pub static ICECHUNK_CLIENT_NAME: LazyLock<String> =
538 LazyLock::new(|| "ic-".to_string() + ICECHUNK_LIB_VERSION);
539 pub const ICECHUNK_CLIENT_NAME_METADATA_KEY: &str = "ic_client";
540
541 pub const ICECHUNK_FILE_TYPE_SNAPSHOT: &str = "snapshot";
542 pub const ICECHUNK_FILE_TYPE_MANIFEST: &str = "manifest";
543 pub const ICECHUNK_FILE_TYPE_TRANSACTION_LOG: &str = "transaction-log";
544 pub const ICECHUNK_FILE_TYPE_REPO_INFO: &str = "repo-info";
545 pub const ICECHUNK_FILE_TYPE_METADATA_KEY: &str = "ic_file_type";
546
547 pub const ICECHUNK_COMPRESSION_METADATA_KEY: &str = "ic_comp_alg";
548 pub const ICECHUNK_COMPRESSION_ZSTD: &str = "zstd";
549}
550
551#[inline(always)]
552#[expect(clippy::needless_pass_by_value)]
553pub fn lookup_index_by_key<'a, T: ::flatbuffers::Follow<'a> + 'a, K: Ord>(
554 v: ::flatbuffers::Vector<'a, T>,
555 key: K,
556 f: fn(&<T as ::flatbuffers::Follow<'a>>::Inner, &K) -> Ordering,
557) -> Option<usize> {
558 if v.is_empty() {
559 return None;
560 }
561
562 let mut left: usize = 0;
563 let mut right = v.len() - 1;
564
565 while left <= right {
566 let mid = (left + right) / 2;
567 let value = v.get(mid);
568 match f(&value, &key) {
569 Ordering::Equal => return Some(mid),
570 Ordering::Less => left = mid + 1,
571 Ordering::Greater => {
572 if mid == 0 {
573 return None;
574 }
575 right = mid - 1;
576 }
577 }
578 }
579
580 None
581}
582
583#[macro_export]
592macro_rules! roundtrip_serialization_tests {
593 ($($test_name: ident - $generator: ident), +) => {
594 $(
595 proptest!{
596 #[icechunk_macros::test]
597 fn $test_name(elem in $generator()) {
598 let bytes = rmp_serde::to_vec(&elem).unwrap();
599 let roundtrip = rmp_serde::from_slice(&bytes).unwrap();
600 assert_eq!(elem, roundtrip);
601 }
602 }
603 )*
604 }
605}
606
607#[cfg(test)]
608pub mod strategies;
609
610#[cfg(test)]
611mod tests {
612 use super::*;
613 use crate::strategies::{attributes_id, spec_version};
614 use pretty_assertions::assert_eq;
615 use proptest::prelude::*;
616
617 roundtrip_serialization_tests!(
618 serialize_and_deserialize_attribute_ids - attributes_id,
619 serialize_and_deserialize_spec_version_bin - spec_version
620 );
621
622 #[icechunk_macros::test]
623 fn test_object_id_serialization() {
624 let sid = SnapshotId::new([0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11]);
625 assert_eq!(
626 serde_json::to_string(&sid).unwrap(),
627 r#"[[0,1,2,3,4,5,6,7,8,9,10,11],null]"#
628 );
629 assert_eq!(String::from(&sid), "000G40R40M30E209185G");
630 assert_eq!(sid, SnapshotId::try_from("000G40R40M30E209185G").unwrap());
631 let sid = SnapshotId::random();
632 assert_eq!(
633 serde_json::from_slice::<SnapshotId>(
634 serde_json::to_vec(&sid).unwrap().as_slice()
635 )
636 .unwrap(),
637 sid,
638 );
639 }
640
641 #[icechunk_macros::test]
642 fn test_unknown_spec_version_gives_nice_error() {
643 use format_constants::SpecVersionBin;
644
645 let future_version: u8 = 3;
646 let result = SpecVersionBin::try_from(future_version);
647 assert!(result.is_err());
648 let err = result.unwrap_err();
649
650 let msg = err.to_string();
651 assert!(
652 msg.contains("format version 3"),
653 "Error should mention the found version: {msg}"
654 );
655 assert!(
656 msg.contains("upgrade the icechunk library"),
657 "Error should suggest upgrading: {msg}"
658 );
659 }
660}