Skip to main content

icechunk_format/
snapshot.rs

1//! Repository state at a point in time (arrays, groups, and manifest references).
2
3use std::{borrow::Cow, collections::BTreeMap, ops::Range, sync::Arc};
4
5use bytes::Bytes;
6use chrono::{DateTime, Utc};
7use flatbuffers::{
8    FlatBufferBuilder, ForwardsUOffset, UnionWIPOffset, Vector, VerifierOptions,
9    WIPOffset,
10};
11use itertools::{Either, Itertools as _};
12use quick_cache::sync::{Cache, GuardResult};
13use serde::{Deserialize, Serialize};
14use serde_json::Value;
15
16pub use crate::NodeType;
17use crate::{
18    AttributesId, ChunkIndices, IcechunkFormatError, IcechunkFormatErrorKind,
19    IcechunkResult, ManifestId, NodeId, Path, SnapshotId,
20    flatbuffers::generated,
21    format_constants::SpecVersionBin,
22    lookup_index_by_key,
23    manifest::{Manifest, ManifestExtents, ManifestRef},
24};
25use icechunk_types::{ICResultExt as _, error::ICResultCtxExt as _};
26
27#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
28pub struct DimensionShape {
29    dim_length: u64,
30    num_chunks: u32,
31}
32
33impl DimensionShape {
34    pub fn new(array_length: u64, num_chunks: u32) -> Self {
35        Self { dim_length: array_length, num_chunks }
36    }
37    pub fn array_length(&self) -> u64 {
38        self.dim_length
39    }
40    pub fn num_chunks(&self) -> u32 {
41        self.num_chunks
42    }
43}
44
45#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
46pub struct ArrayShape(Vec<DimensionShape>);
47
48impl ArrayShape {
49    pub fn len(&self) -> usize {
50        self.0.len()
51    }
52    pub fn is_empty(&self) -> bool {
53        self.0.is_empty()
54    }
55
56    pub fn get(&self, ax: usize) -> Option<DimensionShape> {
57        if ax > self.len() - 1 { None } else { Some(self.0[ax].clone()) }
58    }
59
60    pub fn iter(&self) -> impl Iterator<Item = &DimensionShape> {
61        self.0.iter()
62    }
63
64    pub fn num_chunks(&self) -> impl Iterator<Item = u32> {
65        self.0.iter().map(|x| x.num_chunks())
66    }
67
68    pub fn new<I>(it: I) -> Option<Self>
69    where
70        I: IntoIterator<Item = (u64, u32)>,
71    {
72        let v = it.into_iter().map(|(al, nc)| Some(DimensionShape::new(al, nc)));
73        v.collect::<Option<Vec<_>>>().map(Self)
74    }
75
76    /// Validates the provided chunk coordinates for the array.
77    ///
78    /// This function checks if the provided chunk indices are valid for the array.
79    ///
80    /// # Arguments
81    ///
82    /// * `coord` - The chunk indices to validate.
83    ///
84    /// # Returns
85    ///
86    /// An bool indicating whether the chunk coordinates are valid.
87    ///
88    /// # Errors
89    ///
90    /// Returns false if the chunk coordinates are invalid.
91    pub fn valid_chunk_coord(&self, coord: &ChunkIndices) -> bool {
92        coord
93            .0
94            .iter()
95            .zip(self.num_chunks())
96            .all(|(index, index_permitted)| *index <= (index_permitted.max(1) - 1))
97    }
98}
99
100#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
101pub enum DimensionName {
102    NotSpecified,
103    Name(String),
104}
105
106impl From<Option<&str>> for DimensionName {
107    fn from(value: Option<&str>) -> Self {
108        match value {
109            Some(s) => s.into(),
110            None => DimensionName::NotSpecified,
111        }
112    }
113}
114
115impl From<DimensionName> for Option<String> {
116    fn from(value: DimensionName) -> Option<String> {
117        match value {
118            DimensionName::NotSpecified => None,
119            DimensionName::Name(name) => Some(name),
120        }
121    }
122}
123
124impl From<&str> for DimensionName {
125    fn from(value: &str) -> Self {
126        if value.is_empty() {
127            DimensionName::NotSpecified
128        } else {
129            DimensionName::Name(value.to_string())
130        }
131    }
132}
133
134#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
135pub enum NodeData {
136    Array {
137        shape: ArrayShape,
138        dimension_names: Option<Vec<DimensionName>>,
139        manifests: Vec<ManifestRef>,
140    },
141    Group,
142}
143
144#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
145pub struct NodeSnapshot {
146    pub id: NodeId,
147    pub path: Path,
148    pub user_data: Bytes,
149    pub node_data: NodeData,
150}
151
152impl NodeSnapshot {
153    pub fn node_type(&self) -> NodeType {
154        match &self.node_data {
155            NodeData::Group => NodeType::Group,
156            NodeData::Array { .. } => NodeType::Array,
157        }
158    }
159}
160
161impl From<&generated::ObjectId8> for NodeId {
162    fn from(value: &generated::ObjectId8) -> Self {
163        NodeId::new(value.0)
164    }
165}
166
167impl From<&generated::ObjectId12> for ManifestId {
168    fn from(value: &generated::ObjectId12) -> Self {
169        ManifestId::new(value.0)
170    }
171}
172
173impl From<&generated::ObjectId12> for AttributesId {
174    fn from(value: &generated::ObjectId12) -> Self {
175        AttributesId::new(value.0)
176    }
177}
178
179impl<'a> From<generated::ManifestRef<'a>> for ManifestRef {
180    fn from(value: generated::ManifestRef<'a>) -> Self {
181        let extents = ManifestExtents::from_ranges_iter(
182            value
183                .extents()
184                .iter()
185                .map(|range| Range { start: range.from(), end: range.to() }),
186        );
187        ManifestRef { object_id: value.object_id().into(), extents }
188    }
189}
190
191impl TryFrom<&generated::DimensionShape> for DimensionShape {
192    type Error = IcechunkFormatError;
193
194    fn try_from(value: &generated::DimensionShape) -> Result<Self, Self::Error> {
195        if value.chunk_length() == 0 && value.array_length() != 0 {
196            return Err(IcechunkFormatErrorKind::InvalidArrayMetadata(format!(
197                "Array metadata has chunk_length = 0 while array_length={:?}",
198                value.array_length()
199            )))
200            .capture();
201        }
202        let num_chunks = if value.chunk_length() == 0 {
203            0
204        } else {
205            value.array_length().div_ceil(value.chunk_length()) as u32
206        };
207        Ok(DimensionShape { dim_length: value.array_length(), num_chunks })
208    }
209}
210
211impl<'a> From<&generated::DimensionShapeV2<'a>> for DimensionShape {
212    fn from(value: &generated::DimensionShapeV2<'a>) -> Self {
213        DimensionShape {
214            dim_length: value.array_length(),
215            num_chunks: value.num_chunks(),
216        }
217    }
218}
219
220impl<'a> TryFrom<generated::ArrayNodeData<'a>> for NodeData {
221    type Error = IcechunkFormatError;
222
223    fn try_from(value: generated::ArrayNodeData<'a>) -> Result<Self, Self::Error> {
224        let dimension_names = value
225            .dimension_names()
226            .map(|dn| dn.iter().map(|name| name.name().into()).collect());
227        // In our flatbuffers the V1 `shape` is required and will be an empty `[]` for V2 repos
228        // Note that `shape` is *also* `[]` for scalars in V1 repos.
229        // So we branch on `shape_v2` which is optional, and thus None, on V1 repos.
230        let shape = ArrayShape(match value.shape_v2() {
231            None => value
232                .shape()
233                .iter()
234                .map(|dim| dim.try_into())
235                .collect::<Result<Vec<_>, IcechunkFormatError>>()?,
236            Some(x) => x.iter().map(|dim| (&dim).into()).collect(),
237        });
238        let manifests = value.manifests().iter().map(|m| m.into()).collect();
239        Ok(Self::Array { shape, dimension_names, manifests })
240    }
241}
242
243impl<'a> From<generated::GroupNodeData<'a>> for NodeData {
244    fn from(_: generated::GroupNodeData<'a>) -> Self {
245        Self::Group
246    }
247}
248
249impl<'a> TryFrom<generated::NodeSnapshot<'a>> for NodeSnapshot {
250    type Error = IcechunkFormatError;
251
252    fn try_from(value: generated::NodeSnapshot<'a>) -> Result<Self, Self::Error> {
253        #[expect(clippy::expect_used, clippy::panic)]
254        let node_data: NodeData = match value.node_data_type() {
255            generated::NodeData::Array => value
256                .node_data_as_array()
257                .expect("Bug in flatbuffers library")
258                .try_into()?,
259            generated::NodeData::Group => {
260                value.node_data_as_group().expect("Bug in flatbuffers library").into()
261            }
262            x => panic!("Invalid node data type in flatbuffers file {x:?}"),
263        };
264        let res = NodeSnapshot {
265            id: value.id().into(),
266            path: Path::from_trusted(value.path()),
267            node_data,
268            user_data: Bytes::copy_from_slice(value.user_data().bytes()),
269        };
270        Ok(res)
271    }
272}
273
274impl From<&generated::ManifestFileInfo> for ManifestFileInfo {
275    fn from(value: &generated::ManifestFileInfo) -> Self {
276        Self {
277            id: value.id().into(),
278            size_bytes: value.size_bytes(),
279            num_chunk_refs: value.num_chunk_refs(),
280        }
281    }
282}
283
284impl TryFrom<&generated::ManifestFileInfoV2<'_>> for ManifestFileInfo {
285    type Error = IcechunkFormatError;
286
287    fn try_from(value: &generated::ManifestFileInfoV2<'_>) -> Result<Self, Self::Error> {
288        let id = value.id().map(|id| id.into()).ok_or_else(|| {
289            IcechunkFormatError::capture(IcechunkFormatErrorKind::InvalidFlatBuffer(
290                flatbuffers::InvalidFlatbuffer::MissingRequiredField {
291                    required: Cow::Borrowed("id"),
292                    error_trace: Default::default(),
293                },
294            ))
295        })?;
296        Ok(Self {
297            id,
298            size_bytes: value.size_bytes(),
299            num_chunk_refs: value.num_chunk_refs(),
300        })
301    }
302}
303
304pub type SnapshotProperties = BTreeMap<String, Value>;
305
306/// Insert a key-value pair into the `__icechunk` namespace object within snapshot properties.
307/// Creates the `__icechunk` object if it doesn't exist.
308pub fn inject_icechunk_metadata(
309    properties: &mut SnapshotProperties,
310    key: &str,
311    value: Value,
312) {
313    match properties.get_mut("__icechunk") {
314        Some(Value::Object(map)) => {
315            map.insert(key.to_string(), value);
316        }
317        _ => {
318            properties
319                .insert("__icechunk".to_string(), serde_json::json!({ key: value }));
320        }
321    }
322}
323
324#[derive(Debug, PartialEq, Serialize, Deserialize, Clone, Eq, Hash)]
325pub struct ManifestFileInfo {
326    pub id: ManifestId,
327    pub size_bytes: u64,
328    pub num_chunk_refs: u32,
329}
330
331impl ManifestFileInfo {
332    pub fn new(manifest: &Manifest, size_bytes: u64) -> Self {
333        Self {
334            id: manifest.id().clone(),
335            num_chunk_refs: manifest.len() as u32,
336            size_bytes,
337        }
338    }
339}
340
341// It is _extremely_ common to set/get chunks in large batches sequentially.
342// Without a cache we incur the cost of repeatedly decoding the flatbuffer.
343// This cache is very small (size 2 currently), because we could cache a large number of
344// snapshots. The size of 2 is fine for typically workloads that iterate sequentially
345// through nodes.
346// We cannot choose 1 because of a bug in the `quick_cache` dependency
347// where a size-1 cache is effectively no cache
348// https://github.com/arthurprs/quick-cache/issues/105
349const SNAPSHOT_NODE_CACHE_SIZE: usize = 2;
350
351pub struct Snapshot {
352    buffer: Vec<u8>,
353    spec_version: SpecVersionBin,
354    node_cache: Cache<Path, Arc<NodeSnapshot>>,
355}
356
357impl std::fmt::Debug for Snapshot {
358    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
359        let nodes =
360            self.iter().map(|n| n.map(|n| n.path.to_string())).collect::<Vec<_>>();
361        #[expect(deprecated)]
362        f.debug_struct("Snapshot")
363            .field("id", &self.id())
364            .field("parent_id", &self.parent_id())
365            .field("flushed_at", &self.flushed_at())
366            .field("nodes", &nodes)
367            .field(
368                "manifests",
369                &self.manifest_files().collect::<IcechunkResult<Vec<_>>>(),
370            )
371            .field("message", &self.message())
372            .field("metadata", &self.metadata())
373            .finish_non_exhaustive()
374    }
375}
376
377#[derive(Debug, Clone, PartialEq, Eq, Hash)]
378pub struct SnapshotInfo {
379    pub id: SnapshotId,
380    pub parent_id: Option<SnapshotId>,
381    pub flushed_at: DateTime<Utc>,
382    pub message: String,
383    pub metadata: SnapshotProperties,
384    /// Tx-log ids of ancestor commits expiration removed, oldest first.
385    /// Empty unless this snapshot was edited by expiration.
386    /// Notice: icechunk V1 and older versions of icechunk V2
387    /// didn't generate this field, so those repositories will
388    /// have [] until expiration is ran using a version > 2.0.6
389    pub pruned_ancestor_tx_logs: Vec<SnapshotId>,
390}
391
392impl SnapshotInfo {
393    /// Partial constructor: the snapshot file doesn't store
394    /// `pruned_ancestor_tx_logs` (it lives on `RepoInfo`), so it's left empty.
395    /// Only valid for V1 or fresh snapshots; V2 reads must use
396    /// `RepoInfo::find_snapshot` instead.
397    pub fn from_snapshot_file(snapshot: &Snapshot) -> Result<Self, IcechunkFormatError> {
398        #[expect(deprecated)]
399        Ok(Self {
400            id: snapshot.id().clone(),
401            parent_id: snapshot.parent_id().clone(),
402            flushed_at: snapshot.flushed_at()?,
403            message: snapshot.message().clone(),
404            metadata: snapshot.metadata()?.clone(),
405            // Left empty on purpose: not stored in the snapshot file.
406            pruned_ancestor_tx_logs: Vec::new(),
407        })
408    }
409
410    pub fn is_initial(&self) -> bool {
411        self.id == Snapshot::INITIAL_SNAPSHOT_ID
412    }
413}
414
415impl SnapshotId {
416    pub fn is_initial(&self) -> bool {
417        *self == Snapshot::INITIAL_SNAPSHOT_ID
418    }
419}
420
421static ROOT_OPTIONS: VerifierOptions = VerifierOptions {
422    max_depth: 64,
423    max_tables: 50_000_000,
424    max_apparent_size: 1 << 31, // taken from the default
425    ignore_missing_null_terminator: true,
426};
427
428impl Snapshot {
429    pub const INITIAL_COMMIT_MESSAGE: &'static str = "Repository initialized";
430    pub const INITIAL_SNAPSHOT_ID: SnapshotId = SnapshotId::new([
431        0x0b, 0x1c, 0xc8, 0xd6, 0x78, 0x75, 0x80, 0xf0, 0xe3, 0x3a, 0x65,
432        0x34, // Decodes as 1CECHNKREP0F1RSTCMT0
433    ]);
434
435    pub fn from_buffer(
436        spec_version: SpecVersionBin,
437        buffer: Vec<u8>,
438    ) -> IcechunkResult<Snapshot> {
439        let _ = flatbuffers::root_with_opts::<generated::Snapshot<'_>>(
440            &ROOT_OPTIONS,
441            buffer.as_slice(),
442        )
443        .capture()?;
444        Ok(Snapshot {
445            buffer,
446            spec_version,
447            // this number is low because we cache a very large number of snapshot nodes.
448            node_cache: Cache::new(SNAPSHOT_NODE_CACHE_SIZE),
449        })
450    }
451
452    pub fn bytes(&self) -> &[u8] {
453        self.buffer.as_slice()
454    }
455
456    #[expect(clippy::too_many_arguments)]
457    pub fn from_iter<I>(
458        id: Option<SnapshotId>,
459        parent_id: Option<SnapshotId>,
460        spec_version: SpecVersionBin,
461        message: &str,
462        properties: Option<SnapshotProperties>,
463        mut manifest_files: Vec<ManifestFileInfo>,
464        flushed_at: Option<DateTime<Utc>>,
465        sorted_iter: I,
466    ) -> IcechunkResult<Self>
467    where
468        I: IntoIterator<Item = IcechunkResult<NodeSnapshot>>,
469    {
470        // TODO: what's a good capacity?
471        let mut builder = FlatBufferBuilder::with_capacity(4_096);
472
473        manifest_files.sort_by(|a, b| a.id.cmp(&b.id));
474        let (manifest_files_v1, manifest_files_v2) = match spec_version {
475            SpecVersionBin::V1 => {
476                let ms1 = manifest_files
477                    .iter()
478                    .map(|mfi| {
479                        let id = generated::ObjectId12::new(&mfi.id.0);
480                        generated::ManifestFileInfo::new(
481                            &id,
482                            mfi.size_bytes,
483                            mfi.num_chunk_refs,
484                        )
485                    })
486                    .collect::<Vec<_>>();
487                let ms1 = builder.create_vector(&ms1);
488                let ms2 = None;
489                (ms1, ms2)
490            }
491            SpecVersionBin::V2 => {
492                let ms2 = manifest_files
493                    .iter()
494                    .map(|mfi| {
495                        let id = generated::ObjectId12::new(&mfi.id.0);
496
497                        let args = generated::ManifestFileInfoV2Args {
498                            id: Some(&id),
499                            size_bytes: mfi.size_bytes,
500                            num_chunk_refs: mfi.num_chunk_refs,
501                            extra: None,
502                        };
503                        generated::ManifestFileInfoV2::create(&mut builder, &args)
504                    })
505                    .collect::<Vec<_>>();
506                let ms2 = builder.create_vector(&ms2);
507                let ms1 = builder.create_vector::<generated::ManifestFileInfo>(&[]);
508                (ms1, Some(ms2))
509            }
510        };
511
512        let metadata_items: Vec<_> = properties
513            .unwrap_or_default()
514            .iter()
515            .map(|(k, v)| {
516                let name = builder.create_shared_string(k.as_str());
517                let serialized = if spec_version == SpecVersionBin::V1 {
518                    rmp_serde::to_vec(v).map_err(Box::new).capture()?
519                } else {
520                    flexbuffers::to_vec(v).map_err(Box::new).capture()?
521                };
522
523                let value = builder.create_vector(serialized.as_slice());
524                Ok::<_, IcechunkFormatError>(generated::MetadataItem::create(
525                    &mut builder,
526                    &generated::MetadataItemArgs { name: Some(name), value: Some(value) },
527                ))
528            })
529            .try_collect()?;
530        let metadata_items = builder.create_vector(metadata_items.as_slice());
531
532        let message = builder.create_string(message);
533        // Icechunk 2.0 no longer uses this field
534        let parent_id = parent_id.map(|oid| generated::ObjectId12::new(&oid.0));
535        let flushed_at = flushed_at.unwrap_or_else(Utc::now).timestamp_micros() as u64;
536        let id = generated::ObjectId12::new(&id.unwrap_or_else(SnapshotId::random).0);
537
538        let nodes: Vec<_> = sorted_iter
539            .into_iter()
540            .map(|node| node.and_then(|node| mk_node(&mut builder, &node, spec_version)))
541            .try_collect()?;
542        let nodes = builder.create_vector(&nodes);
543
544        let snap = generated::Snapshot::create(
545            &mut builder,
546            &generated::SnapshotArgs {
547                id: Some(&id),
548                parent_id: parent_id.as_ref(),
549                nodes: Some(nodes),
550                flushed_at,
551                message: Some(message),
552                metadata: Some(metadata_items),
553                manifest_files: Some(manifest_files_v1),
554                manifest_files_v2,
555                ..Default::default()
556            },
557        );
558
559        builder.finish(snap, Some("Ichk"));
560        let (mut buffer, offset) = builder.collapse();
561        buffer.drain(0..offset);
562        buffer.shrink_to_fit();
563        Ok(Snapshot {
564            buffer,
565            spec_version,
566            // this number is low because we cache a very large number of snapshot nodes.
567            node_cache: Cache::new(SNAPSHOT_NODE_CACHE_SIZE),
568        })
569    }
570
571    pub fn initial(spec_version: SpecVersionBin) -> IcechunkResult<Self> {
572        let mut properties = SnapshotProperties::default();
573        inject_icechunk_metadata(&mut properties, "is_root", Value::from(true));
574        let nodes = Vec::<IcechunkResult<NodeSnapshot>>::new();
575        Self::from_iter(
576            Some(Self::INITIAL_SNAPSHOT_ID),
577            None,
578            spec_version,
579            Self::INITIAL_COMMIT_MESSAGE,
580            Some(properties),
581            Default::default(),
582            None,
583            nodes,
584        )
585    }
586
587    #[expect(unsafe_code)]
588    fn root(&self) -> generated::Snapshot<'_> {
589        // SAFETY: self.buffer was serialized by our own flatbuffers serialization code.
590        // We skip validation for performance; a corrupt buffer here indicates
591        // file corruption or a bad Icechunk implementation, not a caller error.
592        unsafe { flatbuffers::root_unchecked::<generated::Snapshot<'_>>(&self.buffer) }
593    }
594
595    pub fn id(&self) -> SnapshotId {
596        SnapshotId::new(self.root().id().0)
597    }
598
599    #[deprecated(
600        since = "2.0.0",
601        note = "New versions of icechunk don't use this field and initialize it to None"
602    )]
603    pub fn parent_id(&self) -> Option<SnapshotId> {
604        self.root().parent_id().map(|pid| SnapshotId::new(pid.0))
605    }
606
607    pub fn metadata(&self) -> IcechunkResult<SnapshotProperties> {
608        self.root()
609            .metadata()
610            .iter()
611            .map(|item| {
612                let key = item.name().to_string();
613                let value = if self.spec_version == SpecVersionBin::V1 {
614                    rmp_serde::from_slice(item.value().bytes())
615                        .map_err(Box::new)
616                        .capture()?
617                } else {
618                    flexbuffers::from_slice(item.value().bytes())
619                        .map_err(Box::new)
620                        .capture()?
621                };
622                Ok((key, value))
623            })
624            .try_collect()
625    }
626
627    pub fn flushed_at(&self) -> IcechunkResult<DateTime<Utc>> {
628        let ts = self.root().flushed_at();
629        let ts: i64 = ts
630            .try_into()
631            .map_err(|_| IcechunkFormatErrorKind::InvalidTimestamp)
632            .capture()?;
633        DateTime::from_timestamp_micros(ts)
634            .ok_or(IcechunkFormatErrorKind::InvalidTimestamp)
635            .capture()
636    }
637
638    pub fn message(&self) -> String {
639        self.root().message().to_string()
640    }
641
642    pub fn manifest_files(
643        &self,
644    ) -> impl Iterator<Item = IcechunkResult<ManifestFileInfo>> + '_ {
645        let root = self.root();
646        if let Some(mf2) = root.manifest_files_v2() {
647            Either::Left(mf2.iter().map(|mf| (&mf).try_into()))
648        } else {
649            Either::Right(root.manifest_files().iter().map(|mf| Ok(mf.into())))
650        }
651    }
652
653    /// Cretase a new `Snapshot` with all the same data as `new_child` but `self` as parent
654    #[deprecated(
655        since = "2.0.0",
656        note = "Shouldn't be necessary after 2.0, only to support Icechunk 1 repos"
657    )]
658    pub fn adopt(&self, new_child: &Snapshot) -> IcechunkResult<Self> {
659        // Rust flatbuffers implementation doesn't allow mutation of scalars, so we need to
660        // create a whole new buffer and write to it in full
661
662        Snapshot::from_iter(
663            Some(new_child.id()),
664            Some(self.id()),
665            SpecVersionBin::V1, // 2.0 doesn't use adopt
666            &new_child.message(),
667            Some(new_child.metadata()?.clone()),
668            new_child.manifest_files().try_collect()?,
669            Some(new_child.flushed_at()?),
670            new_child.iter(),
671        )
672    }
673
674    fn _get_node(&self, path: &Path) -> IcechunkResult<NodeSnapshot> {
675        let res = self
676            .root()
677            .nodes()
678            .lookup_by_key(path.to_string().as_str(), |node, path| node.path().cmp(path))
679            .ok_or_else(|| IcechunkFormatErrorKind::NodeNotFound { path: path.clone() })
680            .capture()?;
681        res.try_into()
682    }
683
684    pub fn get_node(&self, path: &Path) -> IcechunkResult<Arc<NodeSnapshot>> {
685        use GuardResult::*;
686        match self.node_cache.get_value_or_guard(path, None) {
687            Value(node) => Ok(node),
688            Timeout => Ok(Arc::new(self._get_node(path)?)),
689            Guard(guard) => {
690                let node = self._get_node(path)?;
691                let node = Arc::new(node);
692                let _ = guard.insert(Arc::clone(&node));
693                Ok(node)
694            }
695        }
696    }
697
698    pub fn get_node_index(&self, path: &Path) -> IcechunkResult<usize> {
699        let path_str = path.to_string();
700        let res =
701            lookup_index_by_key(self.root().nodes(), path_str.as_str(), |node, path| {
702                node.path().cmp(path)
703            })
704            .ok_or_else(|| IcechunkFormatErrorKind::NodeNotFound { path: path.clone() })
705            .capture()?;
706        Ok(res)
707    }
708
709    pub fn iter(&self) -> impl Iterator<Item = IcechunkResult<NodeSnapshot>> + '_ {
710        self.root().nodes().iter().map(|node| node.try_into().inject())
711    }
712
713    pub fn iter_arc(
714        self: Arc<Self>,
715        parent_group: &Path,
716    ) -> impl Iterator<Item = IcechunkResult<NodeSnapshot>> + use<> {
717        NodeIterator::new(self, parent_group)
718    }
719
720    pub fn len(&self) -> usize {
721        self.root().nodes().len()
722    }
723
724    #[must_use]
725    pub fn is_empty(&self) -> bool {
726        self.len() == 0
727    }
728
729    pub fn manifest_info(
730        &self,
731        id: &ManifestId,
732    ) -> IcechunkResult<Option<ManifestFileInfo>> {
733        let root = self.root();
734        if let Some(mf2) = root.manifest_files_v2() {
735            mf2.iter()
736                .find(|mf| mf.id().is_some_and(|mid| mid.0 == id.0))
737                .map(|mf| (&mf).try_into())
738                .transpose()
739        } else {
740            Ok(root
741                .manifest_files()
742                .iter()
743                .find(|mi| mi.id().0 == id.0)
744                .map(|man| man.into()))
745        }
746    }
747}
748
749struct NodeIterator {
750    snapshot: Arc<Snapshot>,
751    next_index: usize,
752    prefix: String,
753}
754
755impl NodeIterator {
756    fn new(snapshot: Arc<Snapshot>, parent_group: &Path) -> Self {
757        let next_index = snapshot.get_node_index(parent_group).unwrap_or_default();
758        let prefix = parent_group.to_string();
759        let prefix = if prefix == "/" { String::new() } else { prefix };
760        NodeIterator { snapshot, next_index, prefix }
761    }
762}
763
764impl Iterator for NodeIterator {
765    type Item = IcechunkResult<NodeSnapshot>;
766
767    fn next(&mut self) -> Option<Self::Item> {
768        let nodes = self.snapshot.root().nodes();
769        loop {
770            // we need to loop over all nodes starting from the index
771            // until we are sure we'll see no more children
772            if self.next_index >= nodes.len() {
773                return None;
774            }
775
776            let node: IcechunkResult<NodeSnapshot> =
777                nodes.get(self.next_index).try_into();
778
779            match node {
780                Ok(res) => {
781                    let node_path = res.path.to_string();
782                    if let Some(after_prefix) =
783                        node_path.strip_prefix(self.prefix.as_str())
784                        && (after_prefix.is_empty() || after_prefix.starts_with('/'))
785                    {
786                        self.next_index += 1;
787                        return Some(Ok(res));
788                    } else if node_path.as_str() > self.prefix.as_str()
789                        && !node_path.starts_with(self.prefix.as_str())
790                    {
791                        // We've passed all possible children of the prefix
792                        return None;
793                    } else {
794                        // Not a match but there may be matches later (foo-bar" comes before "foo/bar")
795                        self.next_index += 1;
796                    }
797                }
798                Err(err) => return Some(Err(err)),
799            }
800        }
801    }
802}
803
804fn mk_node<'bldr>(
805    builder: &mut FlatBufferBuilder<'bldr>,
806    node: &NodeSnapshot,
807    spec_version: SpecVersionBin,
808) -> IcechunkResult<WIPOffset<generated::NodeSnapshot<'bldr>>> {
809    let id = generated::ObjectId8::new(&node.id.0);
810    let path = builder.create_string(node.path.to_string().as_str());
811    let (node_data_type, node_data) =
812        mk_node_data(builder, &node.node_data, spec_version)?;
813    let user_data = Some(builder.create_vector(&node.user_data));
814    Ok(generated::NodeSnapshot::create(
815        builder,
816        &generated::NodeSnapshotArgs {
817            id: Some(&id),
818            path: Some(path),
819            node_data_type,
820            node_data,
821            user_data,
822            ..Default::default()
823        },
824    ))
825}
826
827type ShapeV1<'a> = WIPOffset<Vector<'a, generated::DimensionShape>>;
828type ShapeV2<'a> =
829    WIPOffset<Vector<'a, ForwardsUOffset<generated::DimensionShapeV2<'a>>>>;
830
831fn mk_array_shapes<'a>(
832    builder: &mut FlatBufferBuilder<'a>,
833    spec_version: SpecVersionBin,
834    shape: &ArrayShape,
835) -> (Option<ShapeV1<'a>>, Option<ShapeV2<'a>>) {
836    use SpecVersionBin::*;
837    match spec_version {
838        V1 => {
839            let shape = shape
840                .0
841                .iter()
842                .map(|ds| {
843                    let chunk_length = if ds.num_chunks == 0 {
844                        debug_assert_eq!(ds.dim_length, 0);
845                        0
846                    } else {
847                        // FIXME: this can be *very* wrong, but I'm not sure it really matters
848                        //        we still preserve num_chunks, and return the proper Zarr JSON metadata
849                        //        which is stored in NodeData::user_data which stores the zarr.json exactly
850                        //        as provided by Zarr
851                        ds.dim_length.div_ceil(ds.num_chunks as u64)
852                    };
853
854                    generated::DimensionShape::new(ds.dim_length, chunk_length)
855                })
856                .collect::<Vec<_>>();
857            (Some(builder.create_vector(shape.as_slice())), None)
858        }
859        V2 => {
860            let shape = shape
861                .0
862                .iter()
863                .map(|ds| {
864                    generated::DimensionShapeV2::create(
865                        builder,
866                        &generated::DimensionShapeV2Args {
867                            array_length: ds.dim_length,
868                            num_chunks: ds.num_chunks,
869                        },
870                    )
871                })
872                .collect::<Vec<_>>();
873            let empty_shape_for_v1_compat =
874                builder.create_vector(&[] as &[generated::DimensionShape]);
875            (
876                Some(empty_shape_for_v1_compat),
877                Some(builder.create_vector(shape.as_slice())),
878            )
879        }
880    }
881}
882
883fn mk_node_data(
884    builder: &mut FlatBufferBuilder<'_>,
885    node_data: &NodeData,
886    spec_version: SpecVersionBin,
887) -> IcechunkResult<(generated::NodeData, Option<WIPOffset<UnionWIPOffset>>)> {
888    match node_data {
889        NodeData::Array { manifests, dimension_names, shape } => {
890            let manifests = manifests
891                .iter()
892                .map(|manref| {
893                    let object_id = generated::ObjectId12::new(&manref.object_id.0);
894                    let extents = manref
895                        .extents
896                        .iter()
897                        .map(|range| {
898                            generated::ChunkIndexRange::new(range.start, range.end)
899                        })
900                        .collect::<Vec<_>>();
901                    let extents = builder.create_vector(&extents);
902                    generated::ManifestRef::create(
903                        builder,
904                        &generated::ManifestRefArgs {
905                            object_id: Some(&object_id),
906                            extents: Some(extents),
907                        },
908                    )
909                })
910                .collect::<Vec<_>>();
911            let manifests = builder.create_vector(manifests.as_slice());
912            let dimensions = dimension_names.as_ref().map(|dn| {
913                let names = dn
914                    .iter()
915                    .map(|n| match n {
916                        DimensionName::Name(s) => {
917                            let n = builder.create_shared_string(s.as_str());
918                            generated::DimensionName::create(
919                                builder,
920                                &generated::DimensionNameArgs { name: Some(n) },
921                            )
922                        }
923                        DimensionName::NotSpecified => generated::DimensionName::create(
924                            builder,
925                            &generated::DimensionNameArgs { name: None },
926                        ),
927                    })
928                    .collect::<Vec<_>>();
929                builder.create_vector(names.as_slice())
930            });
931            let (shape_v1, shape_v2) = mk_array_shapes(builder, spec_version, shape);
932            let node_data = generated::ArrayNodeData::create(
933                builder,
934                &generated::ArrayNodeDataArgs {
935                    manifests: Some(manifests),
936                    shape: shape_v1,
937                    shape_v2,
938                    dimension_names: dimensions,
939                },
940            );
941            Ok((generated::NodeData::Array, Some(node_data.as_union_value())))
942        }
943        NodeData::Group => Ok((
944            generated::NodeData::Group,
945            Some(
946                generated::GroupNodeData::create(
947                    builder,
948                    &generated::GroupNodeDataArgs {},
949                )
950                .as_union_value(),
951            ),
952        )),
953    }
954}
955
956#[cfg(test)]
957#[expect(unused_qualifications)] // proptest macros generate fully qualified paths
958mod tests {
959    use crate::{IcechunkFormatError, ObjectId};
960
961    use super::*;
962    use crate::{
963        roundtrip_serialization_tests,
964        strategies::{ShapeDim, manifest_file_info, node_snapshot, shapes_and_dims},
965    };
966    use pretty_assertions::assert_eq;
967    use proptest::prelude::*;
968    use std::iter::{self};
969
970    roundtrip_serialization_tests!(
971        serialize_and_deserialize_node_snapshot - node_snapshot,
972        serialize_and_deserialize_manifest_file_info - manifest_file_info
973    );
974
975    #[icechunk_macros::test]
976    fn test_get_node() -> Result<(), Box<dyn std::error::Error>> {
977        let shape1 = ArrayShape::new(vec![(10u64, 3), (20, 2), (30, 1)]).unwrap();
978        let dim_names1 = Some(vec!["x".into(), "y".into(), "t".into()]);
979
980        let shape2 = shape1.clone();
981        let dim_names2 = Some(vec![
982            DimensionName::NotSpecified,
983            DimensionName::NotSpecified,
984            "t".into(),
985        ]);
986
987        let shape3 = shape1.clone();
988        let dim_names3 = None;
989
990        let man_ref1 = ManifestRef {
991            object_id: ObjectId::random(),
992            extents: ManifestExtents::new(&[0, 0, 0], &[100, 100, 100]),
993        };
994        let man_ref2 = ManifestRef {
995            object_id: ObjectId::random(),
996            extents: ManifestExtents::new(&[0, 0, 0], &[100, 100, 100]),
997        };
998
999        let node_ids = iter::repeat_with(NodeId::random).take(7).collect::<Vec<_>>();
1000        // nodes must be sorted by path
1001        let nodes = vec![
1002            NodeSnapshot {
1003                path: Path::root(),
1004                id: node_ids[0].clone(),
1005                user_data: Bytes::new(),
1006                node_data: NodeData::Group,
1007            },
1008            NodeSnapshot {
1009                path: "/a".try_into().unwrap(),
1010                id: node_ids[1].clone(),
1011                user_data: Bytes::new(),
1012                node_data: NodeData::Group,
1013            },
1014            NodeSnapshot {
1015                path: "/array2".try_into().unwrap(),
1016                id: node_ids[5].clone(),
1017                user_data: Bytes::new(),
1018                node_data: NodeData::Array {
1019                    shape: shape2.clone(),
1020                    dimension_names: dim_names2.clone(),
1021                    manifests: vec![],
1022                },
1023            },
1024            NodeSnapshot {
1025                path: "/b".try_into().unwrap(),
1026                id: node_ids[2].clone(),
1027                user_data: Bytes::new(),
1028                node_data: NodeData::Group,
1029            },
1030            NodeSnapshot {
1031                path: "/b/array1".try_into().unwrap(),
1032                id: node_ids[4].clone(),
1033                user_data: Bytes::copy_from_slice(b"hello"),
1034                node_data: NodeData::Array {
1035                    shape: shape1.clone(),
1036                    dimension_names: dim_names1.clone(),
1037                    manifests: vec![man_ref1.clone(), man_ref2.clone()],
1038                },
1039            },
1040            NodeSnapshot {
1041                path: "/b/array3".try_into().unwrap(),
1042                id: node_ids[6].clone(),
1043                user_data: Bytes::new(),
1044                node_data: NodeData::Array {
1045                    shape: shape3.clone(),
1046                    dimension_names: dim_names3.clone(),
1047                    manifests: vec![],
1048                },
1049            },
1050            NodeSnapshot {
1051                path: "/b/c".try_into().unwrap(),
1052                id: node_ids[3].clone(),
1053                user_data: Bytes::copy_from_slice(b"bye"),
1054                node_data: NodeData::Group,
1055            },
1056        ];
1057        let manifests = vec![
1058            ManifestFileInfo {
1059                id: man_ref1.object_id.clone(),
1060                size_bytes: 1_000_000,
1061                num_chunk_refs: 100_000,
1062            },
1063            ManifestFileInfo {
1064                id: man_ref2.object_id.clone(),
1065                size_bytes: 1_000_000,
1066                num_chunk_refs: 100_000,
1067            },
1068        ];
1069        let st = Snapshot::from_iter(
1070            None,
1071            None,
1072            SpecVersionBin::current(),
1073            "",
1074            Default::default(),
1075            manifests,
1076            None,
1077            nodes.into_iter().map(Ok::<NodeSnapshot, IcechunkFormatError>),
1078        )
1079        .unwrap();
1080
1081        assert!(matches!(
1082            st.get_node(&"/nonexistent".try_into().unwrap()),
1083            Err(IcechunkFormatError {
1084                kind: IcechunkFormatErrorKind::NodeNotFound {
1085                    path
1086                },
1087                ..
1088            }) if path == "/nonexistent".try_into().unwrap()
1089        ));
1090
1091        let node = st.get_node(&"/b/c".try_into().unwrap()).unwrap();
1092        assert_eq!(
1093            node,
1094            Arc::new(NodeSnapshot {
1095                path: "/b/c".try_into().unwrap(),
1096                id: node_ids[3].clone(),
1097                user_data: Bytes::copy_from_slice(b"bye"),
1098                node_data: NodeData::Group,
1099            }),
1100        );
1101        let node = st.get_node(&Path::root()).unwrap();
1102        assert_eq!(
1103            node,
1104            Arc::new(NodeSnapshot {
1105                path: Path::root(),
1106                id: node_ids[0].clone(),
1107                user_data: Bytes::new(),
1108                node_data: NodeData::Group,
1109            }),
1110        );
1111        let node = st.get_node(&"/b/array1".try_into().unwrap()).unwrap();
1112        assert_eq!(
1113            node,
1114            Arc::new(NodeSnapshot {
1115                path: "/b/array1".try_into().unwrap(),
1116                id: node_ids[4].clone(),
1117                user_data: Bytes::copy_from_slice(b"hello"),
1118                node_data: NodeData::Array {
1119                    shape: shape1.clone(),
1120                    dimension_names: dim_names1.clone(),
1121                    manifests: vec![man_ref1, man_ref2]
1122                },
1123            }),
1124        );
1125        let node = st.get_node(&"/array2".try_into().unwrap()).unwrap();
1126        assert_eq!(
1127            node,
1128            Arc::new(NodeSnapshot {
1129                path: "/array2".try_into().unwrap(),
1130                id: node_ids[5].clone(),
1131                user_data: Bytes::new(),
1132                node_data: NodeData::Array {
1133                    shape: shape2.clone(),
1134                    dimension_names: dim_names2.clone(),
1135                    manifests: vec![]
1136                },
1137            }),
1138        );
1139        let node = st.get_node(&"/b/array3".try_into().unwrap()).unwrap();
1140        assert_eq!(
1141            node,
1142            Arc::new(NodeSnapshot {
1143                path: "/b/array3".try_into().unwrap(),
1144                id: node_ids[6].clone(),
1145                user_data: Bytes::new(),
1146                node_data: NodeData::Array {
1147                    shape: shape3.clone(),
1148                    dimension_names: dim_names3.clone(),
1149                    manifests: vec![]
1150                },
1151            }),
1152        );
1153        Ok(())
1154    }
1155
1156    #[icechunk_macros::test]
1157    fn test_valid_chunk_coord() {
1158        let shape1 =
1159            ArrayShape::new(vec![(10_000, 10), (10_001, 11), (9_999, 10)]).unwrap();
1160        let shape2 = ArrayShape::new(vec![(0, 1_000), (0, 1_000), (0, 1_000)]).unwrap();
1161        let coord1 = ChunkIndices(vec![9, 10, 9]);
1162        let coord2 = ChunkIndices(vec![10, 11, 10]);
1163        let coord3 = ChunkIndices(vec![0, 0, 0]);
1164
1165        assert!(shape1.valid_chunk_coord(&coord1));
1166        assert!(!shape1.valid_chunk_coord(&coord2));
1167        assert!(shape2.valid_chunk_coord(&coord3));
1168    }
1169
1170    #[icechunk_macros::test]
1171    fn test_valid_chunk_coord_zero_length_dim() {
1172        let shape = ArrayShape::new(vec![(0, 0)]).unwrap();
1173        assert!(shape.valid_chunk_coord(&ChunkIndices(vec![0])));
1174        assert!(!shape.valid_chunk_coord(&ChunkIndices(vec![1])));
1175    }
1176
1177    #[test_strategy::proptest]
1178    fn test_prop_valid_chunk_coord(
1179        #[strategy(shapes_and_dims(None, Some(1)))] shape_dim: ShapeDim,
1180        axis_offset: usize,
1181    ) {
1182        let shape = &shape_dim.shape;
1183        let ndim = shape.len();
1184        let axis = axis_offset % ndim;
1185
1186        // origin is always valid
1187        let origin = ChunkIndices(vec![0; ndim]);
1188        prop_assert!(shape.valid_chunk_coord(&origin));
1189
1190        // max valid coord is in bounds
1191        let max_valid =
1192            ChunkIndices(shape.num_chunks().map(|nc| nc.saturating_sub(1)).collect());
1193        prop_assert!(shape.valid_chunk_coord(&max_valid));
1194
1195        // bumping one axis out of bounds is rejected
1196        let mut oob = max_valid.0.clone();
1197        oob[axis] = shape.iter().nth(axis).unwrap().num_chunks().max(1);
1198        prop_assert!(!shape.valid_chunk_coord(&ChunkIndices(oob)));
1199    }
1200}