1use 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 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 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
306pub 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
341const 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 pub pruned_ancestor_tx_logs: Vec<SnapshotId>,
390}
391
392impl SnapshotInfo {
393 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 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, 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, ]);
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 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 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 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 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 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 #[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 Snapshot::from_iter(
663 Some(new_child.id()),
664 Some(self.id()),
665 SpecVersionBin::V1, &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 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 return None;
793 } else {
794 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 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)] mod 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 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 let origin = ChunkIndices(vec![0; ndim]);
1188 prop_assert!(shape.valid_chunk_coord(&origin));
1189
1190 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 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}