1use itertools::Itertools as _;
4use serde::{Deserialize, Serialize};
5use std::{
6 borrow::Cow,
7 collections::{BTreeSet, HashMap},
8};
9use tracing::trace;
10
11use crate::{
12 IcechunkFormatError, IcechunkFormatErrorKind, IcechunkResult, SnapshotId,
13 flatbuffers::generated,
14 format_constants::SpecVersionBin,
15 lookup_index_by_key,
16 snapshot::{SnapshotInfo, SnapshotProperties},
17};
18
19use chrono::{DateTime, Utc};
20use flatbuffers::{UnionWIPOffset, VerifierOptions, WIPOffset};
21use icechunk_types::ICResultExt as _;
22
23#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
24pub enum RepoAvailability {
25 Online,
26 ReadOnly,
27 }
30
31impl From<generated::RepoAvailability> for RepoAvailability {
32 fn from(value: generated::RepoAvailability) -> Self {
33 match value {
34 generated::RepoAvailability::Online => RepoAvailability::Online,
35 generated::RepoAvailability::ReadOnly => RepoAvailability::ReadOnly,
36 _ => RepoAvailability::Online,
37 }
38 }
39}
40
41impl From<RepoAvailability> for generated::RepoAvailability {
42 fn from(value: RepoAvailability) -> Self {
43 match value {
44 RepoAvailability::Online => generated::RepoAvailability::Online,
45 RepoAvailability::ReadOnly => generated::RepoAvailability::ReadOnly,
46 }
47 }
48}
49
50#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
51pub struct RepoStatus {
52 pub availability: RepoAvailability,
53 pub set_at: DateTime<Utc>,
54 pub limited_availability_reason: Option<String>,
55}
56
57impl TryFrom<generated::RepoStatus<'_>> for RepoStatus {
58 type Error = IcechunkFormatError;
59
60 fn try_from(fb_status: generated::RepoStatus<'_>) -> Result<Self, Self::Error> {
61 let ts: i64 = fb_status
62 .set_at()
63 .try_into()
64 .map_err(|_| IcechunkFormatErrorKind::InvalidTimestamp)
65 .capture()?;
66 let set_at = DateTime::from_timestamp_micros(ts)
67 .ok_or(IcechunkFormatErrorKind::InvalidTimestamp)
68 .capture()?;
69 Ok(RepoStatus {
70 availability: fb_status.availability().into(),
71 set_at,
72 limited_availability_reason: fb_status
73 .limited_availability_reason()
74 .map(|s| s.to_string()),
75 })
76 }
77}
78
79impl RepoStatus {
80 pub fn error_msg(&self) -> String {
81 format!(
82 "Repo status is {0:?}, set at {1}, reason: {2:?}",
83 self.availability, self.set_at, self.limited_availability_reason
84 )
85 }
86}
87
88#[derive(PartialEq, Serialize, Deserialize)]
90pub struct RepoInfo {
91 buffer: Vec<u8>,
92}
93
94impl std::fmt::Debug for RepoInfo {
95 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
96 let tags = self.tags().map(Vec::from_iter).unwrap_or_default();
97 let tags =
98 tags.into_iter().map(|(name, snap)| format!("{name} -> {snap}")).join(", ");
99 let branches = self.branches().map(Vec::from_iter).unwrap_or_default();
100 let branches = branches
101 .into_iter()
102 .map(|(name, snap)| format!("{name} -> {snap}"))
103 .join(", ");
104 let snaps = self.all_snapshots().map(Vec::from_iter).unwrap_or_default();
105 let snaps = snaps
106 .into_iter()
107 .map(|ms| match ms {
108 Ok(snap) => format!(
109 "{} -> {}",
110 snap.id,
111 snap.parent_id.map(|s| s.to_string()).unwrap_or_default()
112 ),
113 Err(_) => "#err".to_string(),
114 })
115 .join(", ");
116 f.debug_struct("RepoInfo")
118 .field("tags", &tags)
119 .field("branches", &branches)
120 .field("snapshots", &snaps)
121 .finish_non_exhaustive()
122 }
123}
124
125#[derive(Debug, Clone, PartialEq, Eq)]
126pub enum UpdateType {
127 RepoInitializedUpdate,
128 RepoMigratedUpdate {
129 from_version: SpecVersionBin,
130 to_version: SpecVersionBin,
131 },
132 RepoStatusChangedUpdate {
133 status: RepoStatus,
134 },
135 ConfigChangedUpdate,
136 MetadataChangedUpdate,
137 TagCreatedUpdate {
138 name: String,
139 },
140 TagDeletedUpdate {
141 name: String,
142 previous_snap_id: SnapshotId,
143 },
144 BranchCreatedUpdate {
145 name: String,
146 },
147 BranchDeletedUpdate {
148 name: String,
149 previous_snap_id: SnapshotId,
150 },
151 BranchResetUpdate {
152 name: String,
153 previous_snap_id: SnapshotId,
154 },
155 NewCommitUpdate {
156 branch: String,
157 new_snap_id: SnapshotId,
158 },
159 CommitAmendedUpdate {
160 branch: String,
161 previous_snap_id: SnapshotId,
162 new_snap_id: SnapshotId,
163 },
164 NewDetachedSnapshotUpdate {
165 new_snap_id: SnapshotId,
166 },
167 GCRanUpdate,
168 ExpirationRanUpdate,
169 FeatureFlagChanged {
170 id: u16,
171 new_value: Option<bool>,
172 },
173}
174
175static ROOT_OPTIONS: VerifierOptions = VerifierOptions {
176 max_depth: 10,
177 max_tables: 5_000_000,
178 max_apparent_size: 1 << 31, ignore_missing_null_terminator: true,
180};
181
182#[derive(Debug, Clone)]
183pub struct UpdateInfo<I> {
184 pub update_type: UpdateType,
185 pub update_time: DateTime<Utc>,
186 pub previous_updates: I,
187}
188
189type UpdateTuple<'a> = IcechunkResult<(UpdateType, DateTime<Utc>, Option<&'a str>)>;
190
191type LatestUpdatesResult<'bldr> = IcechunkResult<(
192 WIPOffset<
193 flatbuffers::Vector<
194 'bldr,
195 flatbuffers::ForwardsUOffset<generated::Update<'bldr>>,
196 >,
197 >,
198 Option<WIPOffset<&'bldr str>>,
199)>;
200
201impl RepoInfo {
202 #[expect(clippy::too_many_arguments)]
203 pub fn new<
204 'a,
205 I: IntoIterator<Item = IcechunkResult<(UpdateType, DateTime<Utc>, Option<&'a str>)>>,
206 EFFIt: DoubleEndedIterator<Item = u16> + ExactSizeIterator,
207 DFFIt: DoubleEndedIterator<Item = u16> + ExactSizeIterator,
208 >(
209 spec_version: SpecVersionBin,
210 tags: impl IntoIterator<Item = (&'a str, SnapshotId)>,
211 branches: impl IntoIterator<Item = (&'a str, SnapshotId)>,
212 deleted_tags: impl IntoIterator<Item = &'a str>,
213 snapshots: impl IntoIterator<Item = SnapshotInfo>,
214 metadata: &SnapshotProperties,
215 update: UpdateInfo<I>,
216 backup_path: Option<&'a str>,
217 num_updates_per_file: u16,
218 previous_info: Option<&'a str>,
219 config_bytes: Option<&[u8]>,
220 sorted_enabled_feature_flags: Option<EFFIt>,
221 sorted_disabled_feature_flags: Option<DFFIt>,
222 status: &RepoStatus,
223 ) -> IcechunkResult<Self> {
224 let mut snapshots: Vec<_> = snapshots.into_iter().collect();
225 snapshots.sort_by_key(|a| a.id.0);
226 let tags = resolve_ref_iter(&snapshots, tags)?;
227 let branches = resolve_ref_iter(&snapshots, branches)?;
228 let mut deleted_tags: Vec<_> = deleted_tags.into_iter().collect();
229 deleted_tags.sort();
230 Self::from_parts(
231 spec_version,
232 tags,
233 branches,
234 deleted_tags,
235 snapshots,
236 metadata,
237 |builder| {
238 Self::mk_latest_updates(
239 builder,
240 update,
241 backup_path,
242 num_updates_per_file,
243 previous_info,
244 )
245 },
246 config_bytes,
247 sorted_enabled_feature_flags,
248 sorted_disabled_feature_flags,
249 status,
250 )
251 }
252
253 #[expect(clippy::too_many_arguments)]
254 fn from_parts<
255 'a,
256 F: for<'bldr> FnOnce(
257 &mut flatbuffers::FlatBufferBuilder<'bldr>,
258 ) -> LatestUpdatesResult<'bldr>,
259 EFFIt: DoubleEndedIterator<Item = u16> + ExactSizeIterator,
260 DFFIt: DoubleEndedIterator<Item = u16> + ExactSizeIterator,
261 >(
262 spec_version: SpecVersionBin,
263 sorted_tags: impl IntoIterator<Item = (&'a str, u32)>,
264 sorted_branches: impl IntoIterator<Item = (&'a str, u32)>,
265 sorted_deleted_tags: impl IntoIterator<Item = &'a str>,
266 sorted_snapshots: impl IntoIterator<Item = SnapshotInfo>,
267 metadata: &SnapshotProperties,
268 build_updates: F,
269 config_bytes: Option<&[u8]>,
270 sorted_enabled_feature_flags: Option<EFFIt>,
271 sorted_disabled_feature_flags: Option<DFFIt>,
272 status: &RepoStatus,
273 ) -> IcechunkResult<Self> {
274 trace!("Creating new repo info from parts");
275 let mut builder = flatbuffers::FlatBufferBuilder::with_capacity(4_096);
276 let tags = sorted_tags
277 .into_iter()
278 .map(|(name, offset)| {
279 let args = generated::RefArgs {
280 name: Some(builder.create_string(name)),
281 snapshot_index: offset,
282 };
283 generated::Ref::create(&mut builder, &args)
284 })
285 .collect::<Vec<_>>();
286 let tags = builder.create_vector(&tags);
287
288 let mut main_found = false;
289 let branches = sorted_branches
290 .into_iter()
291 .map(|(name, offset)| {
292 if name == icechunk_types::DEFAULT_BRANCH {
293 main_found = true;
294 }
295 let args = generated::RefArgs {
296 name: Some(builder.create_string(name)),
297 snapshot_index: offset,
298 };
299 generated::Ref::create(&mut builder, &args)
300 })
301 .collect::<Vec<_>>();
302 if !main_found {
303 return Err(IcechunkFormatErrorKind::BranchNotFound {
304 branch: icechunk_types::DEFAULT_BRANCH.to_string(),
305 })
306 .capture();
307 }
308
309 let branches = builder.create_vector(&branches);
310
311 let deleted_tags = sorted_deleted_tags
312 .into_iter()
313 .map(|name| builder.create_string(name))
314 .collect::<Vec<_>>();
315 let deleted_tags = builder.create_vector(&deleted_tags);
316
317 let snapshots: Vec<_> = sorted_snapshots.into_iter().collect();
318 debug_assert!(snapshots.is_sorted_by(|a, b| a.id.0 <= b.id.0));
319
320 let snapshot_index: HashMap<_, _> =
321 snapshots.iter().enumerate().map(|(ix, sn)| (sn.id.clone(), ix)).collect();
322
323 let snapshots: Vec<_> = snapshots
325 .iter()
326 .map(|snap| {
327 let id = &snap.id.0;
328 let id = generated::ObjectId12::new(id);
329 let parent_offset = match snap.parent_id.as_ref() {
330 Some(parent_id) => {
331 let index = snapshot_index
332 .get(parent_id)
333 .ok_or_else(|| IcechunkFormatErrorKind::SnapshotIdNotFound {
334 snapshot_id: snap.id.clone(),
335 })
336 .capture()?;
337 Ok(*index as i32)
338 }
339 None => Ok::<_, IcechunkFormatError>(-1),
340 }?;
341
342 let metadata_items: Vec<_> = snap
343 .metadata
344 .iter()
345 .map(|(k, v)| {
346 let name = builder.create_shared_string(k.as_str());
347 let serialized =
348 flexbuffers::to_vec(v).map_err(Box::new).capture()?;
349 let value = builder.create_vector(serialized.as_slice());
350 let item = generated::MetadataItem::create(
351 &mut builder,
352 &generated::MetadataItemArgs {
353 name: Some(name),
354 value: Some(value),
355 },
356 );
357 Ok::<_, IcechunkFormatError>(item)
358 })
359 .try_collect()?;
360
361 let metadata = builder.create_vector(metadata_items.as_slice());
362
363 let pruned_ancestor_tx_logs = if snap.pruned_ancestor_tx_logs.is_empty() {
366 None
367 } else {
368 let ids: Vec<_> = snap
369 .pruned_ancestor_tx_logs
370 .iter()
371 .map(|id| generated::ObjectId12::new(&id.0))
372 .collect();
373 Some(builder.create_vector(ids.as_slice()))
374 };
375 debug_assert_eq!(
376 pruned_ancestor_tx_logs.is_none(),
377 snap.pruned_ancestor_tx_logs.is_empty(),
378 "empty pruned_ancestor_tx_logs must serialize as a null vector, \
379 not a present empty one"
380 );
381
382 let args = generated::SnapshotInfoArgs {
383 id: Some(&id),
384 parent_offset,
385 flushed_at: snap.flushed_at.timestamp_micros() as u64,
386 message: Some(builder.create_string(snap.message.as_str())),
387 metadata: Some(metadata),
388 pruned_ancestor_tx_logs,
389 };
390 Ok::<_, IcechunkFormatError>(generated::SnapshotInfo::create(
391 &mut builder,
392 &args,
393 ))
394 })
395 .try_collect()?;
396 let snapshots = builder.create_vector(&snapshots);
397
398 let limited_reason = status
399 .limited_availability_reason
400 .as_deref()
401 .map(|s| builder.create_string(s));
402 let status = generated::RepoStatus::create(
403 &mut builder,
404 &generated::RepoStatusArgs {
405 availability: status.availability.into(),
406 set_at: status.set_at.timestamp_micros() as u64,
407 limited_availability_reason: limited_reason,
408 },
409 );
410
411 let metadata_items: Vec<_> = metadata
412 .iter()
413 .map(|(k, v)| {
414 let name = builder.create_shared_string(k.as_str());
415 let serialized = flexbuffers::to_vec(v).map_err(Box::new).capture()?;
416 let value = builder.create_vector(serialized.as_slice());
417 let item = generated::MetadataItem::create(
418 &mut builder,
419 &generated::MetadataItemArgs { name: Some(name), value: Some(value) },
420 );
421 Ok::<_, IcechunkFormatError>(item)
422 })
423 .try_collect()?;
424
425 let metadata = builder.create_vector(metadata_items.as_slice());
426
427 let enabled_feature_flags =
428 sorted_enabled_feature_flags.map(|it| builder.create_vector_from_iter(it));
429 let disabled_feature_flags =
430 sorted_disabled_feature_flags.map(|it| builder.create_vector_from_iter(it));
431
432 let (latest_updates, repo_before_updates) = build_updates(&mut builder)?;
433
434 let config = config_bytes.map(|bytes| builder.create_vector(bytes));
435
436 let repo_args = generated::RepoArgs {
438 tags: Some(tags),
439 branches: Some(branches),
440 deleted_tags: Some(deleted_tags),
441 snapshots: Some(snapshots),
442 spec_version: spec_version as u8,
443 status: Some(status),
444 metadata: Some(metadata),
445 latest_updates: Some(latest_updates),
446 repo_before_updates,
447 config,
448 enabled_feature_flags,
449 disabled_feature_flags,
450 ..Default::default()
451 };
452 let repo = generated::Repo::create(&mut builder, &repo_args);
453 builder.finish(repo, Some("Ichk"));
454 let (mut buffer, offset) = builder.collapse();
455 buffer.drain(0..offset);
456 buffer.shrink_to_fit();
457 Ok(Self { buffer })
458 }
459
460 fn mk_latest_updates<
461 'bldr,
462 'a,
463 I: IntoIterator<Item = IcechunkResult<(UpdateType, DateTime<Utc>, Option<&'a str>)>>,
464 >(
465 builder: &mut flatbuffers::FlatBufferBuilder<'bldr>,
466 update: UpdateInfo<I>,
467 backup_path: Option<&'a str>,
468 num_updates_per_file: u16,
469 previous_info: Option<&'a str>,
470 ) -> LatestUpdatesResult<'bldr> {
471 let mut previous_updates = update.previous_updates.into_iter();
473 let last_update = previous_updates.next().map(|maybe_data| {
474 maybe_data.map(|(ut, dt, path)| {
475 assert!(
476 path.is_none(),
477 "Invalid latest update iterator, last element has backup path"
478 );
479 (ut, dt, backup_path)
480 })
481 });
482 assert!(
489 backup_path.is_none() || last_update.is_some(),
490 "A backup path must not be provided without previous updates"
491 );
492
493 if let Some(Ok((_, latest_time, _))) = &last_update
495 && update.update_time <= *latest_time
496 {
497 return Err(IcechunkFormatErrorKind::InvalidUpdateTimestamp {
498 latest_time: *latest_time,
499 new_time: update.update_time,
500 })
501 .capture();
502 }
503
504 let new_updates: Box<dyn Iterator<Item = _>> =
505 if let Some(last_update) = last_update {
506 Box::new(
507 [Ok((update.update_type, update.update_time, None)), last_update]
508 .into_iter(),
509 )
510 } else {
511 Box::new([Ok((update.update_type, update.update_time, None))].into_iter())
512 };
513
514 let all_updates = new_updates.into_iter().chain(previous_updates);
515 let mut repo_before_updates = previous_info;
518
519 let num_updates = num_updates_per_file as usize;
520 let mut updates = Vec::new();
521 for maybe_data in all_updates {
522 let (u_type, u_time, file) = maybe_data?;
523
524 if updates.len() >= num_updates
529 && let Some(bp) = file
530 {
531 repo_before_updates = Some(bp);
532 break;
533 }
534
535 let (update_type_type, update_type) = update_type_to_fb(builder, &u_type)?;
536 let file = file.map(|file| builder.create_string(file));
537 updates.push(generated::Update::create(
538 builder,
539 &generated::UpdateArgs {
540 update_type_type,
541 update_type: Some(update_type),
542 updated_at: u_time.timestamp_micros() as u64,
543 backup_path: file,
544 },
545 ));
546 }
547
548 debug_assert!(!updates.is_empty(), "Must have at least one update in repo file");
549
550 let updates = builder.create_vector(&updates);
551 let repo_before_updates = repo_before_updates.map(|s| builder.create_string(s));
552 Ok((updates, repo_before_updates))
553 }
554
555 fn mk_latest_updates_passthrough<'bldr, 'a>(
556 builder: &mut flatbuffers::FlatBufferBuilder<'bldr>,
557 updates: impl IntoIterator<Item = UpdateTuple<'a>>,
558 repo_before_updates: Option<&'a str>,
559 ) -> LatestUpdatesResult<'bldr> {
560 let mut fb_updates = Vec::new();
561 for maybe_data in updates {
562 let (u_type, u_time, file) = maybe_data?;
563 let (update_type_type, update_type) = update_type_to_fb(builder, &u_type)?;
564 let file = file.map(|file| builder.create_string(file));
565 fb_updates.push(generated::Update::create(
566 builder,
567 &generated::UpdateArgs {
568 update_type_type,
569 update_type: Some(update_type),
570 updated_at: u_time.timestamp_micros() as u64,
571 backup_path: file,
572 },
573 ));
574 }
575 debug_assert!(
576 !fb_updates.is_empty(),
577 "Must have at least one update in repo file"
578 );
579 let updates = builder.create_vector(&fb_updates);
580 let repo_before_updates = repo_before_updates.map(|s| builder.create_string(s));
581 Ok((updates, repo_before_updates))
582 }
583
584 pub fn initial<C: Serialize>(
585 spec_version: SpecVersionBin,
586 snapshot: SnapshotInfo,
587 num_updates_per_file: u16,
588 config: Option<&C>,
589 update_time: Option<DateTime<Utc>>,
590 ) -> Self {
591 #[expect(clippy::expect_used)]
592 let config_bytes =
593 config.map(|c| flexbuffers::to_vec(c).expect("Cannot serialize config"));
594 #[expect(clippy::expect_used)]
596 Self::from_parts(
597 spec_version,
598 [],
599 [("main", 0)],
600 [],
601 [snapshot],
602 &Default::default(),
603 |builder| {
604 Self::mk_latest_updates(
605 builder,
606 UpdateInfo {
607 update_type: UpdateType::RepoInitializedUpdate,
608 update_time: update_time.unwrap_or_else(Utc::now),
609 previous_updates: [],
610 },
611 None,
612 num_updates_per_file,
613 None,
614 )
615 },
616 config_bytes.as_deref(),
617 None::<std::iter::Empty<u16>>,
618 None::<std::iter::Empty<u16>>,
619 &RepoStatus {
620 availability: RepoAvailability::Online,
621 set_at: update_time.unwrap_or_else(Utc::now),
622 limited_availability_reason: None,
623 },
624 )
625 .expect("Cannot generate initial snapshot")
626 }
627
628 pub fn config_bytes_raw(&self) -> IcechunkResult<Option<Vec<u8>>> {
630 Ok(self.root()?.config().map(|v| v.bytes().to_vec()))
631 }
632
633 pub fn config<C: serde::de::DeserializeOwned>(&self) -> IcechunkResult<Option<C>> {
637 match self.root()?.config() {
638 None => Ok(None),
639 Some(config_fb) => {
640 let config: C = flexbuffers::from_slice(config_fb.bytes())
641 .map_err(Box::new)
642 .capture()?;
643 Ok(Some(config))
644 }
645 }
646 }
647
648 pub fn metadata(&self) -> IcechunkResult<SnapshotProperties> {
649 self.root()?
650 .metadata()
651 .unwrap_or_default()
652 .iter()
653 .map(|item| {
654 let key = item.name().to_string();
655 let value = flexbuffers::from_slice(item.value().bytes())
656 .map_err(Box::new)
657 .capture()?;
658 Ok((key, value))
659 })
660 .try_collect()
661 }
662
663 pub fn status(&self) -> IcechunkResult<RepoStatus> {
664 let root = self.root()?;
665 let fb_status = root.status();
666 fb_status.try_into()
667 }
668
669 pub fn enabled_feature_flags(
670 &self,
671 ) -> IcechunkResult<Option<impl DoubleEndedIterator<Item = u16> + ExactSizeIterator>>
672 {
673 Ok(self.root()?.enabled_feature_flags().map(|v| v.iter()))
674 }
675
676 pub fn disabled_feature_flags(
677 &self,
678 ) -> IcechunkResult<Option<impl DoubleEndedIterator<Item = u16> + ExactSizeIterator>>
679 {
680 Ok(self.root()?.disabled_feature_flags().map(|v| v.iter()))
681 }
682
683 pub fn feature_flag_enabled(&self, id: u16) -> IcechunkResult<Option<bool>> {
687 let root = self.root()?;
688 if root
689 .enabled_feature_flags()
690 .and_then(|v| v.lookup_by_key(id, |this, key| this.cmp(key)))
691 .is_some()
692 {
693 return Ok(Some(true));
694 }
695 if root
696 .disabled_feature_flags()
697 .and_then(|v| v.lookup_by_key(id, |this, key| this.cmp(key)))
698 .is_some()
699 {
700 return Ok(Some(false));
701 }
702 Ok(None)
703 }
704
705 fn all_tags(&self) -> IcechunkResult<impl Iterator<Item = (&str, u32)>> {
706 Ok(self.root()?.tags().iter().map(|r| (r.name(), r.snapshot_index())))
707 }
708
709 fn all_branches(&self) -> IcechunkResult<impl Iterator<Item = (&str, u32)>> {
710 Ok(self.root()?.branches().iter().map(|r| (r.name(), r.snapshot_index())))
711 }
712
713 pub fn deleted_tags(&self) -> IcechunkResult<impl Iterator<Item = &str>> {
714 Ok(self.root()?.deleted_tags().iter())
715 }
716
717 pub fn all_snapshots(
718 &self,
719 ) -> IcechunkResult<impl Iterator<Item = IcechunkResult<SnapshotInfo>>> {
720 let root = self.root()?;
721 Ok(root.snapshots().iter().map(move |snap| mk_snapshot_info(&root, &snap)))
722 }
723
724 pub fn update_feature_flag(
726 &self,
727 spec_version: SpecVersionBin,
728 flag_id: u16,
729 enabled: Option<bool>,
730 previous_file: &str,
731 num_updates_per_file: u16,
732 ) -> IcechunkResult<Self> {
733 let snaps: Vec<_> = self.all_snapshots()?.try_collect()?;
734 let (eff, dff): (Option<Vec<_>>, Option<Vec<_>>) = match enabled {
735 Some(false) => {
736 let e = self
737 .enabled_feature_flags()?
738 .map(|it| it.filter(|x| *x != flag_id).collect());
739
740 let mut d: BTreeSet<_> = self
741 .disabled_feature_flags()?
742 .map(|it| it.collect())
743 .unwrap_or_default();
744 d.insert(flag_id);
745 let d = d.into_iter().collect();
746 (e, Some(d))
747 }
748 Some(true) => {
749 let d = self
750 .disabled_feature_flags()?
751 .map(|it| it.filter(|x| *x != flag_id).collect());
752
753 let mut e: BTreeSet<_> = self
754 .enabled_feature_flags()?
755 .map(|it| it.collect())
756 .unwrap_or_default();
757 e.insert(flag_id);
758 let e = e.into_iter().collect();
759 (Some(e), d)
760 }
761 None => {
762 let e = self
763 .enabled_feature_flags()?
764 .map(|it| it.filter(|x| *x != flag_id).collect());
765 let d = self
766 .disabled_feature_flags()?
767 .map(|it| it.filter(|x| *x != flag_id).collect());
768 (e, d)
769 }
770 };
771
772 let latest_updates = self.latest_updates()?;
773 let rbu = self.repo_before_updates()?;
774 Self::from_parts(
775 spec_version,
776 self.all_tags()?,
777 self.all_branches()?,
778 self.deleted_tags()?,
779 snaps,
780 &self.metadata()?,
781 |builder| {
782 Self::mk_latest_updates(
783 builder,
784 UpdateInfo {
785 update_type: UpdateType::FeatureFlagChanged {
786 id: flag_id,
787 new_value: enabled,
788 },
789 update_time: Utc::now(),
790 previous_updates: latest_updates,
791 },
792 Some(previous_file),
793 num_updates_per_file,
794 rbu,
795 )
796 },
797 self.config_bytes_raw()?.as_deref(),
798 eff.map(|it| it.into_iter()),
799 dff.map(|it| it.into_iter()),
800 &self.status()?,
801 )
802 }
803
804 #[expect(clippy::too_many_arguments)]
805 pub fn add_snapshot(
806 &self,
807 spec_version: SpecVersionBin,
808 snap: SnapshotInfo,
809 branch: Option<&str>,
810 update_type: UpdateType,
811 update_time: Option<DateTime<Utc>>, previous_file: &str,
813 num_updates_per_file: u16,
814 ) -> IcechunkResult<Self> {
815 let mut snapshots: Vec<_> = self.all_snapshots()?.try_collect()?;
816 let new_index = match snapshots.binary_search_by_key(&&snap.id, |snap| &snap.id) {
817 Ok(_) => Err(IcechunkFormatErrorKind::DuplicateSnapshotId {
818 snapshot_id: snap.id.clone(),
819 })
820 .capture(),
821 Err(idx) => Ok(idx),
822 }?;
823
824 snapshots.insert(new_index, snap);
825
826 let tags = self.all_tags()?.map(|(name, idx)| {
827 if idx as usize >= new_index { (name, idx + 1) } else { (name, idx) }
828 });
829 let branches = self.all_branches()?.map(|(name, idx)| {
830 if Some(name) == branch {
831 (name, new_index as u32)
832 } else if idx as usize >= new_index {
833 (name, idx + 1)
834 } else {
835 (name, idx)
836 }
837 });
838
839 let latest_updates = self.latest_updates()?;
840 let rbu = self.repo_before_updates()?;
841 let res = Self::from_parts(
842 spec_version,
843 tags,
844 branches,
845 self.deleted_tags()?,
846 snapshots,
847 &self.metadata()?,
848 |builder| {
849 Self::mk_latest_updates(
850 builder,
851 UpdateInfo {
852 update_type,
853 update_time: update_time.unwrap_or_else(Utc::now),
854 previous_updates: latest_updates,
855 },
856 Some(previous_file),
857 num_updates_per_file,
858 rbu,
859 )
860 },
861 self.config_bytes_raw()?.as_deref(),
862 self.enabled_feature_flags()?,
863 self.disabled_feature_flags()?,
864 &self.status()?,
865 )?;
866 Ok(res)
867 }
868
869 pub fn add_branch(
870 &self,
871 spec_version: SpecVersionBin,
872 name: &str,
873 snap: &SnapshotId,
874 previous_file: &str,
875 num_updates_per_file: u16,
876 ) -> IcechunkResult<Self> {
877 if let Ok(snapshot_id) = self.resolve_branch(name) {
878 return Err(IcechunkFormatErrorKind::BranchAlreadyExists {
879 branch: name.to_string(),
880 snapshot_id,
881 })
882 .capture();
883 }
884
885 match self.resolve_snapshot_index(snap)? {
886 Some(snap_idx) => {
887 let mut branches: Vec<_> = self.all_branches()?.collect();
888 branches.push((name, snap_idx as u32));
889 branches.sort_by_key(|(name1, _)| *name1);
890 let snaps: Vec<_> = self.all_snapshots()?.try_collect()?;
891 let latest_updates = self.latest_updates()?;
892 let rbu = self.repo_before_updates()?;
893 Ok(Self::from_parts(
894 spec_version,
895 self.all_tags()?,
896 branches,
897 self.deleted_tags()?,
898 snaps,
899 &self.metadata()?,
900 |builder| {
901 Self::mk_latest_updates(
902 builder,
903 UpdateInfo {
904 update_type: UpdateType::BranchCreatedUpdate {
905 name: name.to_string(),
906 },
907 update_time: Utc::now(),
908 previous_updates: latest_updates,
909 },
910 Some(previous_file),
911 num_updates_per_file,
912 rbu,
913 )
914 },
915 self.config_bytes_raw()?.as_deref(),
916 self.enabled_feature_flags()?,
917 self.disabled_feature_flags()?,
918 &self.status()?,
919 )?)
920 }
921 None => Err(IcechunkFormatErrorKind::SnapshotIdNotFound {
922 snapshot_id: snap.clone(),
923 })
924 .capture(),
925 }
926 }
927
928 pub fn delete_branch(
929 &self,
930 spec_version: SpecVersionBin,
931 name: &str,
932 previous_file: &str,
933 num_updates_per_file: u16,
934 ) -> IcechunkResult<Self> {
935 match self.resolve_branch(name) {
936 Ok(previous_snap_id) => {
937 let mut branches: Vec<_> = self.all_branches()?.collect();
938 branches.retain(|(n, _)| n != &name);
940 let snaps: Vec<_> = self.all_snapshots()?.try_collect()?;
941 let latest_updates = self.latest_updates()?;
942 let rbu = self.repo_before_updates()?;
943 Self::from_parts(
944 spec_version,
945 self.all_tags()?,
946 branches,
947 self.deleted_tags()?,
948 snaps,
949 &self.metadata()?,
950 |builder| {
951 Self::mk_latest_updates(
952 builder,
953 UpdateInfo {
954 update_type: UpdateType::BranchDeletedUpdate {
955 name: name.to_string(),
956 previous_snap_id,
957 },
958 update_time: Utc::now(),
959 previous_updates: latest_updates,
960 },
961 Some(previous_file),
962 num_updates_per_file,
963 rbu,
964 )
965 },
966 self.config_bytes_raw()?.as_deref(),
967 self.enabled_feature_flags()?,
968 self.disabled_feature_flags()?,
969 &self.status()?,
970 )
971 }
972 Err(IcechunkFormatError {
973 kind: IcechunkFormatErrorKind::BranchNotFound { .. },
974 ..
975 }) => {
976 Err(IcechunkFormatErrorKind::BranchNotFound { branch: name.to_string() })
977 .capture()
978 }
979 Err(err) => Err(err),
980 }
981 }
982
983 pub fn update_branch(
984 &self,
985 spec_version: SpecVersionBin,
986 name: &str,
987 new_snap: &SnapshotId,
988 previous_file: &str,
989 num_updates_per_file: u16,
990 ) -> IcechunkResult<Self> {
991 let previous_snap_id = self.resolve_branch(name)?;
992 match self.resolve_snapshot_index(new_snap)? {
993 Some(snap_idx) => {
994 let branches = self.all_branches()?.map(|(br, idx)| {
995 if br == name { (br, snap_idx as u32) } else { (br, idx) }
996 });
997 let snaps: Vec<_> = self.all_snapshots()?.try_collect()?;
998 let latest_updates = self.latest_updates()?;
999 let rbu = self.repo_before_updates()?;
1000 Ok(Self::from_parts(
1001 spec_version,
1002 self.all_tags()?,
1003 branches,
1004 self.deleted_tags()?,
1005 snaps,
1006 &self.metadata()?,
1007 |builder| {
1008 Self::mk_latest_updates(
1009 builder,
1010 UpdateInfo {
1011 update_type: UpdateType::BranchResetUpdate {
1012 name: name.to_string(),
1013 previous_snap_id,
1014 },
1015 update_time: Utc::now(),
1016 previous_updates: latest_updates,
1017 },
1018 Some(previous_file),
1019 num_updates_per_file,
1020 rbu,
1021 )
1022 },
1023 self.config_bytes_raw()?.as_deref(),
1024 self.enabled_feature_flags()?,
1025 self.disabled_feature_flags()?,
1026 &self.status()?,
1027 )?)
1028 }
1029 None => Err(IcechunkFormatErrorKind::SnapshotIdNotFound {
1030 snapshot_id: new_snap.clone(),
1031 })
1032 .capture(),
1033 }
1034 }
1035
1036 pub fn add_tag(
1037 &self,
1038 spec_version: SpecVersionBin,
1039 name: &str,
1040 snap: &SnapshotId,
1041 previous_file: &str,
1042 num_updates_per_file: u16,
1043 ) -> IcechunkResult<Self> {
1044 if self.resolve_tag(name).is_ok() {
1045 return Err(IcechunkFormatErrorKind::TagAlreadyExists {
1046 tag: name.to_string(),
1047 })
1048 .capture();
1049 }
1050 if self.tag_was_deleted(name)? {
1051 return Err(IcechunkFormatErrorKind::TagPreviouslyDeleted {
1052 tag: name.to_string(),
1053 })
1054 .capture();
1055 }
1056
1057 match self.resolve_snapshot_index(snap)? {
1058 Some(snap_idx) => {
1059 let mut tags: Vec<_> = self.all_tags()?.collect();
1060 tags.push((name, snap_idx as u32));
1061 tags.sort_by_key(|(name1, _)| *name1);
1062 let snaps: Vec<_> = self.all_snapshots()?.try_collect()?;
1063 let latest_updates = self.latest_updates()?;
1064 let rbu = self.repo_before_updates()?;
1065 Ok(Self::from_parts(
1066 spec_version,
1067 tags,
1068 self.all_branches()?,
1069 self.deleted_tags()?,
1070 snaps,
1071 &self.metadata()?,
1072 |builder| {
1073 Self::mk_latest_updates(
1074 builder,
1075 UpdateInfo {
1076 update_type: UpdateType::TagCreatedUpdate {
1077 name: name.to_string(),
1078 },
1079 update_time: Utc::now(),
1080 previous_updates: latest_updates,
1081 },
1082 Some(previous_file),
1083 num_updates_per_file,
1084 rbu,
1085 )
1086 },
1087 self.config_bytes_raw()?.as_deref(),
1088 self.enabled_feature_flags()?,
1089 self.disabled_feature_flags()?,
1090 &self.status()?,
1091 )?)
1092 }
1093 None => Err(IcechunkFormatErrorKind::SnapshotIdNotFound {
1094 snapshot_id: snap.clone(),
1095 })
1096 .capture(),
1097 }
1098 }
1099
1100 pub fn delete_tag(
1101 &self,
1102 spec_version: SpecVersionBin,
1103 name: &str,
1104 previous_file: &str,
1105 num_updates_per_file: u16,
1106 ) -> IcechunkResult<Self> {
1107 match self.resolve_tag(name) {
1108 Ok(previous_snap_id) => {
1109 let mut tags: Vec<_> = self.all_tags()?.collect();
1110 tags.retain(|(n, _)| n != &name);
1112
1113 let mut deleted_tags: BTreeSet<_> = self.deleted_tags()?.collect();
1114 debug_assert!(!deleted_tags.contains(name));
1115 deleted_tags.insert(name);
1116
1117 let snaps: Vec<_> = self.all_snapshots()?.try_collect()?;
1118 let latest_updates = self.latest_updates()?;
1119 let rbu = self.repo_before_updates()?;
1120 Self::from_parts(
1121 spec_version,
1122 tags,
1123 self.all_branches()?,
1124 deleted_tags,
1125 snaps,
1126 &self.metadata()?,
1127 |builder| {
1128 Self::mk_latest_updates(
1129 builder,
1130 UpdateInfo {
1131 update_type: UpdateType::TagDeletedUpdate {
1132 name: name.to_string(),
1133 previous_snap_id,
1134 },
1135 update_time: Utc::now(),
1136 previous_updates: latest_updates,
1137 },
1138 Some(previous_file),
1139 num_updates_per_file,
1140 rbu,
1141 )
1142 },
1143 self.config_bytes_raw()?.as_deref(),
1144 self.enabled_feature_flags()?,
1145 self.disabled_feature_flags()?,
1146 &self.status()?,
1147 )
1148 }
1149 Err(IcechunkFormatError {
1150 kind: IcechunkFormatErrorKind::TagNotFound { .. },
1151 ..
1152 }) => Err(IcechunkFormatErrorKind::TagNotFound { tag: name.to_string() })
1153 .capture(),
1154 Err(err) => Err(err),
1155 }
1156 }
1157
1158 pub fn set_metadata(
1159 &self,
1160 spec_version: SpecVersionBin,
1161 metadata: &SnapshotProperties,
1162 previous_file: &str,
1163 num_updates_per_file: u16,
1164 ) -> IcechunkResult<Self> {
1165 let snaps: Vec<_> = self.all_snapshots()?.try_collect()?;
1166 let latest_updates = self.latest_updates()?;
1167 let rbu = self.repo_before_updates()?;
1168 Self::from_parts(
1169 spec_version,
1170 self.all_tags()?,
1171 self.all_branches()?,
1172 self.deleted_tags()?,
1173 snaps,
1174 metadata,
1175 |builder| {
1176 Self::mk_latest_updates(
1177 builder,
1178 UpdateInfo {
1179 update_type: UpdateType::MetadataChangedUpdate,
1180 update_time: Utc::now(),
1181 previous_updates: latest_updates,
1182 },
1183 Some(previous_file),
1184 num_updates_per_file,
1185 rbu,
1186 )
1187 },
1188 self.config_bytes_raw()?.as_deref(),
1189 self.enabled_feature_flags()?,
1190 self.disabled_feature_flags()?,
1191 &self.status()?,
1192 )
1193 }
1194
1195 pub fn set_config<C: Serialize>(
1197 &self,
1198 spec_version: SpecVersionBin,
1199 config: &C,
1200 previous_file: &str,
1201 num_updates_per_file: u16,
1202 ) -> IcechunkResult<Self> {
1203 let config_bytes = flexbuffers::to_vec(config).map_err(Box::new).capture()?;
1204 let snaps: Vec<_> = self.all_snapshots()?.try_collect()?;
1205 let latest_updates = self.latest_updates()?;
1206 let rbu = self.repo_before_updates()?;
1207 Self::from_parts(
1208 spec_version,
1209 self.all_tags()?,
1210 self.all_branches()?,
1211 self.deleted_tags()?,
1212 snaps,
1213 &self.metadata()?,
1214 |builder| {
1215 Self::mk_latest_updates(
1216 builder,
1217 UpdateInfo {
1218 update_type: UpdateType::ConfigChangedUpdate,
1219 update_time: Utc::now(),
1220 previous_updates: latest_updates,
1221 },
1222 Some(previous_file),
1223 num_updates_per_file,
1224 rbu,
1225 )
1226 },
1227 Some(config_bytes.as_slice()),
1228 self.enabled_feature_flags()?,
1229 self.disabled_feature_flags()?,
1230 &self.status()?,
1231 )
1232 }
1233
1234 pub fn set_status(
1242 &self,
1243 spec_version: SpecVersionBin,
1244 status: &RepoStatus,
1245 update_config: Option<(&str, u16)>,
1246 ) -> IcechunkResult<Self> {
1247 let snaps: Vec<_> = self.all_snapshots()?.try_collect()?;
1248 let latest_updates = self.latest_updates()?;
1249 let rbu = self.repo_before_updates()?;
1250 if let Some((previous_file, num_updates_per_file)) = update_config {
1251 let status_update = status.clone();
1252 Self::from_parts(
1253 spec_version,
1254 self.all_tags()?,
1255 self.all_branches()?,
1256 self.deleted_tags()?,
1257 snaps,
1258 &self.metadata()?,
1259 |builder| {
1260 Self::mk_latest_updates(
1261 builder,
1262 UpdateInfo {
1263 update_type: UpdateType::RepoStatusChangedUpdate {
1264 status: status_update,
1265 },
1266 update_time: Utc::now(),
1267 previous_updates: latest_updates,
1268 },
1269 Some(previous_file),
1270 num_updates_per_file,
1271 rbu,
1272 )
1273 },
1274 self.config_bytes_raw()?.as_deref(),
1275 self.enabled_feature_flags()?,
1276 self.disabled_feature_flags()?,
1277 status,
1278 )
1279 } else {
1280 Self::from_parts(
1281 spec_version,
1282 self.all_tags()?,
1283 self.all_branches()?,
1284 self.deleted_tags()?,
1285 snaps,
1286 &self.metadata()?,
1287 |builder| {
1288 Self::mk_latest_updates_passthrough(builder, latest_updates, rbu)
1289 },
1290 self.config_bytes_raw()?.as_deref(),
1291 self.enabled_feature_flags()?,
1292 self.disabled_feature_flags()?,
1293 status,
1294 )
1295 }
1296 }
1297
1298 pub fn from_buffer(buffer: Vec<u8>) -> IcechunkResult<RepoInfo> {
1299 let _ = flatbuffers::root_with_opts::<generated::Repo<'_>>(
1300 &ROOT_OPTIONS,
1301 buffer.as_slice(),
1302 )
1303 .capture()?;
1304 Ok(RepoInfo { buffer })
1305 }
1306
1307 pub fn bytes(&self) -> &[u8] {
1308 self.buffer.as_slice()
1309 }
1310
1311 fn root(&self) -> IcechunkResult<generated::Repo<'_>> {
1312 flatbuffers::root::<generated::Repo<'_>>(&self.buffer).capture()
1313 }
1314
1315 pub fn tag_names(&self) -> IcechunkResult<impl Iterator<Item = &str>> {
1316 Ok(self.root()?.tags().iter().map(|r| r.name()))
1317 }
1318
1319 pub fn branch_names(&self) -> IcechunkResult<impl Iterator<Item = &str>> {
1320 Ok(self.root()?.branches().iter().map(|r| r.name()))
1321 }
1322
1323 pub fn tags(&self) -> IcechunkResult<impl Iterator<Item = (&str, SnapshotId)>> {
1324 let root = self.root()?;
1325 Ok(self.all_tags()?.map(move |(name, idx)| {
1326 (name, SnapshotId::new(root.snapshots().get(idx as usize).id().0))
1327 }))
1328 }
1329
1330 pub fn branches(&self) -> IcechunkResult<impl Iterator<Item = (&str, SnapshotId)>> {
1331 let root = self.root()?;
1332 Ok(self.all_branches()?.map(move |(name, idx)| {
1333 (name, SnapshotId::new(root.snapshots().get(idx as usize).id().0))
1334 }))
1335 }
1336
1337 pub fn resolve_tag(&self, name: &str) -> IcechunkResult<SnapshotId> {
1338 let root = self.root()?;
1339 let res = root
1340 .tags()
1341 .lookup_by_key(name, |r, key| r.name().cmp(key))
1342 .map(|r| {
1343 let index = r.snapshot_index();
1344 SnapshotId::new(root.snapshots().get(index as usize).id().0)
1345 })
1346 .ok_or_else(|| IcechunkFormatErrorKind::TagNotFound { tag: name.to_string() })
1347 .capture()?;
1348
1349 Ok(res)
1350 }
1351
1352 pub fn tag_was_deleted(&self, name: &str) -> IcechunkResult<bool> {
1353 let root = self.root()?;
1354 let res = root.deleted_tags().lookup_by_key(name, |name, key| name.cmp(key));
1355 Ok(res.is_some())
1356 }
1357
1358 pub fn resolve_branch(&self, name: &str) -> IcechunkResult<SnapshotId> {
1359 let root = self.root()?;
1360 let res = root
1361 .branches()
1362 .lookup_by_key(name, |r, key| r.name().cmp(key))
1363 .map(|r| {
1364 let index = r.snapshot_index();
1365 SnapshotId::new(root.snapshots().get(index as usize).id().0)
1366 })
1367 .ok_or_else(|| IcechunkFormatErrorKind::BranchNotFound {
1368 branch: name.to_string(),
1369 })
1370 .capture()?;
1371
1372 Ok(res)
1373 }
1374
1375 pub fn spec_version(&self) -> IcechunkResult<SpecVersionBin> {
1376 let raw = self.root()?.spec_version();
1377 raw.try_into()
1378 .map_err(|_| IcechunkFormatErrorKind::InvalidSpecVersion {
1379 found: raw,
1380 max_supported: SpecVersionBin::current() as u8,
1381 })
1382 .capture()
1383 }
1384
1385 pub fn latest_updates(
1386 &self,
1387 ) -> IcechunkResult<impl Iterator<Item = UpdateTuple<'_>>> {
1388 let res = self.root()?.latest_updates().iter().map(|up| self.update_to_tuple(up));
1389 Ok(res)
1390 }
1391
1392 fn update_to_tuple<'a>(
1393 &'a self,
1394 update: generated::Update<'a>,
1395 ) -> IcechunkResult<(UpdateType, DateTime<Utc>, Option<&'a str>)> {
1396 let ty = self.mk_update_type(&update)?;
1397 let ts = timestamp_to_timestamp(update.updated_at())?;
1398 let bp = update.backup_path();
1399 Ok((ty, ts, bp))
1400 }
1401
1402 fn mk_update_type(
1403 &self,
1404 update: &generated::Update<'_>,
1405 ) -> IcechunkResult<UpdateType> {
1406 #[expect(clippy::unwrap_used)]
1407 match update.update_type_type() {
1408 generated::UpdateType::RepoInitializedUpdate => {
1409 Ok(UpdateType::RepoInitializedUpdate)
1410 }
1411 generated::UpdateType::RepoMigratedUpdate => {
1412 let up = update.update_type_as_repo_migrated_update().unwrap();
1413 let from_raw = up.from_version();
1414 let to_raw = up.to_version();
1415 Ok(UpdateType::RepoMigratedUpdate {
1416 from_version: from_raw
1417 .try_into()
1418 .map_err(|_| IcechunkFormatErrorKind::InvalidSpecVersion {
1419 found: from_raw,
1420 max_supported: SpecVersionBin::current() as u8,
1421 })
1422 .capture()?,
1423 to_version: to_raw
1424 .try_into()
1425 .map_err(|_| IcechunkFormatErrorKind::InvalidSpecVersion {
1426 found: to_raw,
1427 max_supported: SpecVersionBin::current() as u8,
1428 })
1429 .capture()?,
1430 })
1431 }
1432 generated::UpdateType::RepoStatusChangedUpdate => {
1433 let up = update.update_type_as_repo_status_changed_update().unwrap();
1434 let fb_status = up.status().unwrap();
1435 let status = fb_status.try_into()?;
1436
1437 Ok(UpdateType::RepoStatusChangedUpdate { status })
1438 }
1439 generated::UpdateType::ConfigChangedUpdate => {
1440 Ok(UpdateType::ConfigChangedUpdate)
1441 }
1442 generated::UpdateType::MetadataChangedUpdate => {
1443 Ok(UpdateType::MetadataChangedUpdate)
1444 }
1445 generated::UpdateType::TagCreatedUpdate => {
1446 let up = update.update_type_as_tag_created_update().unwrap();
1447 Ok(UpdateType::TagCreatedUpdate { name: up.name().to_string() })
1448 }
1449 generated::UpdateType::TagDeletedUpdate => {
1450 let up = update.update_type_as_tag_deleted_update().unwrap();
1451 let previous_snap_id = SnapshotId::new(up.previous_snap_id().0);
1452 Ok(UpdateType::TagDeletedUpdate {
1453 name: up.name().to_string(),
1454 previous_snap_id,
1455 })
1456 }
1457 generated::UpdateType::BranchCreatedUpdate => {
1458 let up = update.update_type_as_branch_created_update().unwrap();
1459 Ok(UpdateType::BranchCreatedUpdate { name: up.name().to_string() })
1460 }
1461 generated::UpdateType::BranchDeletedUpdate => {
1462 let up = update.update_type_as_branch_deleted_update().unwrap();
1463 let previous_snap_id = SnapshotId::new(up.previous_snap_id().0);
1464 Ok(UpdateType::BranchDeletedUpdate {
1465 name: up.name().to_string(),
1466 previous_snap_id,
1467 })
1468 }
1469 generated::UpdateType::BranchResetUpdate => {
1470 let up = update.update_type_as_branch_reset_update().unwrap();
1471 let previous_snap_id = SnapshotId::new(up.previous_snap_id().0);
1472 Ok(UpdateType::BranchResetUpdate {
1473 name: up.name().to_string(),
1474 previous_snap_id,
1475 })
1476 }
1477 generated::UpdateType::NewCommitUpdate => {
1478 let up = update.update_type_as_new_commit_update().unwrap();
1479 let new_snap_id = SnapshotId::new(up.new_snap_id().0);
1480 Ok(UpdateType::NewCommitUpdate {
1481 branch: up.branch().to_string(),
1482 new_snap_id,
1483 })
1484 }
1485 generated::UpdateType::CommitAmendedUpdate => {
1486 let up = update.update_type_as_commit_amended_update().unwrap();
1487 let previous_snap_id = SnapshotId::new(up.previous_snap_id().0);
1488 let new_snap_id = SnapshotId::new(up.new_snap_id().0);
1489 Ok(UpdateType::CommitAmendedUpdate {
1490 branch: up.branch().to_string(),
1491 previous_snap_id,
1492 new_snap_id,
1493 })
1494 }
1495 generated::UpdateType::NewDetachedSnapshotUpdate => {
1496 let up = update.update_type_as_new_detached_snapshot_update().unwrap();
1497 let new_snap_id = SnapshotId::new(up.new_snap_id().0);
1498 Ok(UpdateType::NewDetachedSnapshotUpdate { new_snap_id })
1499 }
1500 generated::UpdateType::GCRanUpdate => Ok(UpdateType::GCRanUpdate),
1501 generated::UpdateType::ExpirationRanUpdate => {
1502 Ok(UpdateType::ExpirationRanUpdate)
1503 }
1504 generated::UpdateType::FeatureFlagChangedUpdate => {
1505 let up = update.update_type_as_feature_flag_changed_update().unwrap();
1506 Ok(UpdateType::FeatureFlagChanged {
1507 id: up.id(),
1508 new_value: if up.is_set() { Some(up.new_value()) } else { None },
1509 })
1510 }
1511 _ => Err(IcechunkFormatErrorKind::InvalidFlatBuffer(
1512 flatbuffers::InvalidFlatbuffer::InconsistentUnion {
1513 field: Cow::Borrowed("latest_update_type"),
1514 field_type: Cow::Borrowed("UpdateType"),
1515 error_trace: Default::default(),
1516 },
1517 ))
1518 .capture(),
1519 }
1520 }
1521
1522 pub fn ancestry<'a>(
1523 &'a self,
1524 snapshot: &SnapshotId,
1525 ) -> IcechunkResult<impl Iterator<Item = IcechunkResult<SnapshotInfo>> + Send + use<'a>>
1526 {
1527 let root = self.root()?;
1528 if let Some(start) = self.resolve_snapshot_index(snapshot)? {
1529 let mut index = Some(start as i32);
1530 let iter = std::iter::from_fn(move || {
1531 if let Some(ix) = index {
1532 if ix >= 0 {
1533 let snap = root.snapshots().get(ix as usize);
1534 index = Some(snap.parent_offset());
1535 Some(mk_snapshot_info(&root, &snap))
1536 } else {
1537 index = None;
1538 None
1539 }
1540 } else {
1541 None
1542 }
1543 });
1544 Ok(iter)
1545 } else {
1546 Err(IcechunkFormatErrorKind::SnapshotIdNotFound {
1547 snapshot_id: snapshot.clone(),
1548 })
1549 .capture()
1550 }
1551 }
1552
1553 pub fn find_snapshot(&self, id: &SnapshotId) -> IcechunkResult<SnapshotInfo> {
1554 let mut anc = self.ancestry(id)?;
1555 #[expect(clippy::panic)]
1556 match anc.next() {
1557 Some(snap) => snap,
1558 None => panic!("Ancestry head snapshot not found"),
1561 }
1562 }
1563
1564 pub fn repo_before_updates(&self) -> IcechunkResult<Option<&str>> {
1565 Ok(self.root()?.repo_before_updates())
1566 }
1567
1568 fn resolve_snapshot_index(&self, id: &SnapshotId) -> IcechunkResult<Option<usize>> {
1569 Ok(lookup_index_by_key(self.root()?.snapshots(), &id.0, |snap, key| {
1570 snap.id().0.cmp(key)
1571 }))
1572 }
1573}
1574
1575fn resolve_ref_iter<'a>(
1576 sorted_snapshots: &[SnapshotInfo],
1577 it: impl IntoIterator<Item = (&'a str, SnapshotId)>,
1578) -> IcechunkResult<Vec<(&'a str, u32)>> {
1579 let mut res: Vec<_> = it
1580 .into_iter()
1581 .map(|(name, id)| {
1582 let idx = sorted_snapshots
1583 .binary_search_by_key(&&id.0, |snap| &snap.id.0)
1584 .map_err(|_| IcechunkFormatErrorKind::SnapshotIdNotFound {
1585 snapshot_id: id.clone(),
1586 })
1587 .capture()? as u32;
1588 Ok::<_, IcechunkFormatError>((name, idx))
1589 })
1590 .try_collect()?;
1591 res.sort_by_key(|(name1, _)| *name1);
1592 Ok(res)
1593}
1594
1595fn timestamp_to_timestamp(ts: u64) -> IcechunkResult<DateTime<Utc>> {
1596 let ts: i64 =
1597 ts.try_into().map_err(|_| IcechunkFormatErrorKind::InvalidTimestamp).capture()?;
1598 DateTime::from_timestamp_micros(ts)
1599 .ok_or(IcechunkFormatErrorKind::InvalidTimestamp)
1600 .capture()
1601}
1602
1603fn mk_snapshot_info(
1604 repo: &generated::Repo<'_>,
1605 snap: &generated::SnapshotInfo<'_>,
1606) -> IcechunkResult<SnapshotInfo> {
1607 let flushed_at = timestamp_to_timestamp(snap.flushed_at())?;
1608 let parent_id = if snap.parent_offset() >= 0 {
1609 let parent = repo.snapshots().get(snap.parent_offset() as usize).id();
1610 Some(parent)
1611 } else {
1612 None
1613 };
1614 let metadata = snap
1615 .metadata()
1616 .map(|items| {
1617 let items = items
1618 .iter()
1619 .map(|item| {
1620 let name = item.name().to_string();
1621 let value = flexbuffers::from_slice(item.value().bytes())
1622 .map_err(Box::new)
1623 .capture()?;
1624 Ok::<_, IcechunkFormatError>((name, value))
1625 })
1626 .try_collect()?;
1627 Ok::<_, IcechunkFormatError>(items)
1628 })
1629 .transpose()?
1630 .unwrap_or_default();
1631
1632 let pruned_ancestor_tx_logs = snap
1633 .pruned_ancestor_tx_logs()
1634 .map(|ids| ids.iter().map(|oid| SnapshotId::new(oid.0)).collect())
1635 .unwrap_or_default();
1636
1637 Ok(SnapshotInfo {
1638 id: SnapshotId::new(snap.id().0),
1639 flushed_at,
1640 message: snap.message().to_string(),
1641 metadata,
1642 parent_id: parent_id.map(|buf| SnapshotId::new(buf.0)),
1643 pruned_ancestor_tx_logs,
1644 })
1645}
1646
1647fn update_type_to_fb<'bldr>(
1648 builder: &mut flatbuffers::FlatBufferBuilder<'bldr>,
1649 update: &UpdateType,
1650) -> IcechunkResult<(generated::UpdateType, WIPOffset<UnionWIPOffset>)> {
1651 match update {
1652 UpdateType::RepoInitializedUpdate => Ok((
1653 generated::UpdateType::RepoInitializedUpdate,
1654 generated::RepoInitializedUpdate::create(
1655 builder,
1656 &generated::RepoInitializedUpdateArgs {},
1657 )
1658 .as_union_value(),
1659 )),
1660 UpdateType::RepoMigratedUpdate { from_version, to_version } => Ok((
1661 generated::UpdateType::RepoMigratedUpdate,
1662 generated::RepoMigratedUpdate::create(
1663 builder,
1664 &generated::RepoMigratedUpdateArgs {
1665 from_version: *from_version as u8,
1666 to_version: *to_version as u8,
1667 },
1668 )
1669 .as_union_value(),
1670 )),
1671 UpdateType::RepoStatusChangedUpdate { status } => {
1672 let limited_availability_reason = status
1673 .limited_availability_reason
1674 .as_ref()
1675 .map(|r| builder.create_string(r));
1676 let status = generated::RepoStatus::create(
1677 builder,
1678 &generated::RepoStatusArgs {
1679 availability: status.availability.into(),
1680 set_at: status.set_at.timestamp_micros() as u64,
1681 limited_availability_reason,
1682 },
1683 );
1684
1685 Ok((
1686 generated::UpdateType::RepoStatusChangedUpdate,
1687 generated::RepoStatusChangedUpdate::create(
1688 builder,
1689 &generated::RepoStatusChangedUpdateArgs { status: Some(status) },
1690 )
1691 .as_union_value(),
1692 ))
1693 }
1694 UpdateType::ConfigChangedUpdate => Ok((
1695 generated::UpdateType::ConfigChangedUpdate,
1696 generated::ConfigChangedUpdate::create(
1697 builder,
1698 &generated::ConfigChangedUpdateArgs {},
1699 )
1700 .as_union_value(),
1701 )),
1702 UpdateType::MetadataChangedUpdate => Ok((
1703 generated::UpdateType::MetadataChangedUpdate,
1704 generated::MetadataChangedUpdate::create(
1705 builder,
1706 &generated::MetadataChangedUpdateArgs {},
1707 )
1708 .as_union_value(),
1709 )),
1710 UpdateType::TagCreatedUpdate { name } => {
1711 let name = Some(builder.create_string(name));
1712 Ok((
1713 generated::UpdateType::TagCreatedUpdate,
1714 generated::TagCreatedUpdate::create(
1715 builder,
1716 &generated::TagCreatedUpdateArgs { name },
1717 )
1718 .as_union_value(),
1719 ))
1720 }
1721 UpdateType::TagDeletedUpdate { name, previous_snap_id } => {
1722 let name = Some(builder.create_string(name));
1723 let object_id12 = generated::ObjectId12::new(&previous_snap_id.0);
1724 let previous_snap_id = Some(&object_id12);
1725 Ok((
1726 generated::UpdateType::TagDeletedUpdate,
1727 generated::TagDeletedUpdate::create(
1728 builder,
1729 &generated::TagDeletedUpdateArgs { name, previous_snap_id },
1730 )
1731 .as_union_value(),
1732 ))
1733 }
1734 UpdateType::BranchCreatedUpdate { name } => {
1735 let name = Some(builder.create_string(name));
1736 Ok((
1737 generated::UpdateType::BranchCreatedUpdate,
1738 generated::BranchCreatedUpdate::create(
1739 builder,
1740 &generated::BranchCreatedUpdateArgs { name },
1741 )
1742 .as_union_value(),
1743 ))
1744 }
1745 UpdateType::BranchDeletedUpdate { name, previous_snap_id } => {
1746 let name = Some(builder.create_string(name));
1747 let object_id12 = generated::ObjectId12::new(&previous_snap_id.0);
1748 let previous_snap_id = Some(&object_id12);
1749 Ok((
1750 generated::UpdateType::BranchDeletedUpdate,
1751 generated::BranchDeletedUpdate::create(
1752 builder,
1753 &generated::BranchDeletedUpdateArgs { name, previous_snap_id },
1754 )
1755 .as_union_value(),
1756 ))
1757 }
1758 UpdateType::BranchResetUpdate { name, previous_snap_id } => {
1759 let name = Some(builder.create_string(name));
1760 let object_id12 = generated::ObjectId12::new(&previous_snap_id.0);
1761 let previous_snap_id = Some(&object_id12);
1762 Ok((
1763 generated::UpdateType::BranchResetUpdate,
1764 generated::BranchResetUpdate::create(
1765 builder,
1766 &generated::BranchResetUpdateArgs { name, previous_snap_id },
1767 )
1768 .as_union_value(),
1769 ))
1770 }
1771 UpdateType::NewCommitUpdate { branch, new_snap_id } => {
1772 let branch = Some(builder.create_string(branch));
1773 let object_id12 = generated::ObjectId12::new(&new_snap_id.0);
1774 let new_snap_id = Some(&object_id12);
1775 Ok((
1776 generated::UpdateType::NewCommitUpdate,
1777 generated::NewCommitUpdate::create(
1778 builder,
1779 &generated::NewCommitUpdateArgs { branch, new_snap_id },
1780 )
1781 .as_union_value(),
1782 ))
1783 }
1784 UpdateType::CommitAmendedUpdate { branch, previous_snap_id, new_snap_id } => {
1785 let branch = Some(builder.create_string(branch));
1786 let object_id12 = generated::ObjectId12::new(&previous_snap_id.0);
1787 let previous_snap_id = Some(&object_id12);
1788 let object_id12 = generated::ObjectId12::new(&new_snap_id.0);
1789 let new_snap_id = Some(&object_id12);
1790 Ok((
1791 generated::UpdateType::CommitAmendedUpdate,
1792 generated::CommitAmendedUpdate::create(
1793 builder,
1794 &generated::CommitAmendedUpdateArgs {
1795 branch,
1796 previous_snap_id,
1797 new_snap_id,
1798 },
1799 )
1800 .as_union_value(),
1801 ))
1802 }
1803 UpdateType::NewDetachedSnapshotUpdate { new_snap_id } => {
1804 let object_id12 = generated::ObjectId12::new(&new_snap_id.0);
1805 let new_snap_id = Some(&object_id12);
1806 Ok((
1807 generated::UpdateType::NewDetachedSnapshotUpdate,
1808 generated::NewDetachedSnapshotUpdate::create(
1809 builder,
1810 &generated::NewDetachedSnapshotUpdateArgs { new_snap_id },
1811 )
1812 .as_union_value(),
1813 ))
1814 }
1815 UpdateType::GCRanUpdate => Ok((
1816 generated::UpdateType::GCRanUpdate,
1817 generated::GCRanUpdate::create(builder, &generated::GCRanUpdateArgs {})
1818 .as_union_value(),
1819 )),
1820 UpdateType::ExpirationRanUpdate => Ok((
1821 generated::UpdateType::ExpirationRanUpdate,
1822 generated::ExpirationRanUpdate::create(
1823 builder,
1824 &generated::ExpirationRanUpdateArgs {},
1825 )
1826 .as_union_value(),
1827 )),
1828 UpdateType::FeatureFlagChanged { id, new_value } => Ok((
1829 generated::UpdateType::FeatureFlagChangedUpdate,
1830 generated::FeatureFlagChangedUpdate::create(
1831 builder,
1832 &generated::FeatureFlagChangedUpdateArgs {
1833 id: *id,
1834 new_value: new_value.unwrap_or_default(),
1835 is_set: new_value.is_some(),
1836 },
1837 )
1838 .as_union_value(),
1839 )),
1840 }
1841}
1842
1843#[cfg(test)]
1844mod tests {
1845
1846 use super::*;
1847 use crate::roundtrip_serialization_tests;
1848 use proptest::prelude::*;
1849 use std::collections::HashSet;
1850
1851 fn potentially_invalid_repo_info() -> impl Strategy<Value = RepoInfo> {
1853 any::<Vec<u8>>().prop_map(|buffer| RepoInfo { buffer })
1854 }
1855
1856 roundtrip_serialization_tests!(
1857 serialize_and_deserialize_repo_info - potentially_invalid_repo_info
1858 );
1859
1860 #[test]
1861 fn test_add_snapshot() -> Result<(), Box<dyn std::error::Error>> {
1862 let id1 = SnapshotId::random();
1863 let snap1 = SnapshotInfo {
1864 id: id1.clone(),
1865 parent_id: None,
1866 flushed_at: DateTime::from_timestamp_micros(1_000_000).unwrap(),
1868 message: "snap 1".to_string(),
1869 metadata: Default::default(),
1870 pruned_ancestor_tx_logs: vec![],
1871 };
1872 let repo = RepoInfo::initial(
1873 SpecVersionBin::current(),
1874 snap1.clone(),
1875 100,
1876 None::<&()>,
1877 None,
1878 );
1879 assert_eq!(repo.all_snapshots()?.next().unwrap().unwrap(), snap1);
1880
1881 let id2 = SnapshotId::random();
1882 let snap2 = SnapshotInfo {
1883 id: id2.clone(),
1884 parent_id: Some(id1.clone()),
1885 flushed_at: DateTime::from_timestamp_micros(2_000_000).unwrap(),
1886 message: "snap 2".to_string(),
1887 ..snap1.clone()
1888 };
1889 let repo = repo.add_snapshot(
1890 SpecVersionBin::current(),
1891 snap2.clone(),
1892 Some("main"),
1893 UpdateType::NewCommitUpdate {
1894 branch: "main".to_string(),
1895 new_snap_id: snap2.id.clone(),
1896 },
1897 None,
1898 "foo/bar",
1899 100,
1900 )?;
1901 assert_eq!(&repo.resolve_branch("main")?, &snap2.id);
1902 assert_eq!(repo.repo_before_updates()?, None);
1903
1904 let all: HashSet<_> = repo.all_snapshots()?.try_collect()?;
1905 assert_eq!(all, HashSet::from_iter([snap1.clone(), snap2.clone()]));
1906
1907 let anc: Vec<_> = repo.ancestry(&id1)?.try_collect()?;
1908 assert_eq!(anc, std::slice::from_ref(&snap1));
1909
1910 let anc: Vec<_> = repo.ancestry(&id2)?.try_collect()?;
1911 assert_eq!(anc, [snap2.clone(), snap1.clone()]);
1912
1913 assert!(repo.ancestry(&SnapshotId::random()).is_err());
1914
1915 let id3 = SnapshotId::random();
1916 let snap3 = SnapshotInfo {
1917 id: id3.clone(),
1918 parent_id: Some(id2.clone()),
1919 flushed_at: DateTime::from_timestamp_micros(3_000_000).unwrap(),
1920 message: "snap 3".to_string(),
1921 ..snap2.clone()
1922 };
1923 let repo = repo.add_snapshot(
1924 SpecVersionBin::current(),
1925 snap3.clone(),
1926 Some("main"),
1927 UpdateType::NewCommitUpdate {
1928 branch: "main".to_string(),
1929 new_snap_id: snap3.id.clone(),
1930 },
1931 None,
1932 "foo",
1933 100,
1934 )?;
1935 assert_eq!(&repo.resolve_branch("main")?, &snap3.id);
1936 let all: HashSet<_> = repo.all_snapshots()?.try_collect()?;
1937 assert_eq!(
1938 all,
1939 HashSet::from_iter([snap1.clone(), snap2.clone(), snap3.clone()])
1940 );
1941
1942 let all: HashSet<_> = repo.all_snapshots()?.try_collect()?;
1943 assert_eq!(
1944 all,
1945 HashSet::from_iter([snap1.clone(), snap2.clone(), snap3.clone()])
1946 );
1947
1948 let anc: Vec<_> = repo.ancestry(&id3)?.try_collect()?;
1949 assert_eq!(anc, [snap3.clone(), snap2.clone(), snap1.clone()]);
1950 Ok(())
1951 }
1952
1953 #[test]
1954 fn test_pruned_ancestor_tx_logs_roundtrip() -> Result<(), Box<dyn std::error::Error>>
1955 {
1956 let root_id = SnapshotId::random();
1957 let root = SnapshotInfo {
1958 id: root_id.clone(),
1959 parent_id: None,
1960 flushed_at: DateTime::from_timestamp_micros(1_000_000).unwrap(),
1961 message: "root".to_string(),
1962 metadata: Default::default(),
1963 pruned_ancestor_tx_logs: vec![],
1964 };
1965 let repo = RepoInfo::initial(
1966 SpecVersionBin::current(),
1967 root.clone(),
1968 100,
1969 None::<&()>,
1970 None,
1971 );
1972
1973 let pruned =
1975 vec![SnapshotId::random(), SnapshotId::random(), SnapshotId::random()];
1976 let tip_id = SnapshotId::random();
1977 let tip = SnapshotInfo {
1978 id: tip_id.clone(),
1979 parent_id: Some(root_id.clone()),
1980 flushed_at: DateTime::from_timestamp_micros(2_000_000).unwrap(),
1981 message: "tip".to_string(),
1982 metadata: Default::default(),
1983 pruned_ancestor_tx_logs: pruned.clone(),
1984 };
1985 let repo = repo.add_snapshot(
1986 SpecVersionBin::current(),
1987 tip.clone(),
1988 Some("main"),
1989 UpdateType::NewCommitUpdate {
1990 branch: "main".to_string(),
1991 new_snap_id: tip_id.clone(),
1992 },
1993 None,
1994 "foo",
1995 100,
1996 )?;
1997
1998 let read_tip = repo.find_snapshot(&tip_id)?;
2000 assert_eq!(read_tip.pruned_ancestor_tx_logs, pruned);
2001
2002 let read_root = repo.find_snapshot(&root_id)?;
2004 assert!(read_root.pruned_ancestor_tx_logs.is_empty());
2005 Ok(())
2006 }
2007
2008 #[test]
2009 fn test_tags_and_branches() -> Result<(), Box<dyn std::error::Error>> {
2010 let id1 = SnapshotId::random();
2011 let snap1 = SnapshotInfo {
2012 id: id1.clone(),
2013 parent_id: None,
2014 flushed_at: DateTime::from_timestamp_micros(1_000_000).unwrap(),
2016 message: "snap 1".to_string(),
2017 metadata: Default::default(),
2018 pruned_ancestor_tx_logs: vec![],
2019 };
2020 let repo = RepoInfo::initial(
2021 SpecVersionBin::current(),
2022 snap1.clone(),
2023 100,
2024 None::<&()>,
2025 None,
2026 );
2027 let repo = repo.add_branch(SpecVersionBin::current(), "foo", &id1, "foo", 100)?;
2028 let repo = repo.add_branch(SpecVersionBin::current(), "bar", &id1, "bar", 100)?;
2029 assert!(matches!(
2030 repo.add_branch(
2031 SpecVersionBin::current(),
2032 "bad-snap",
2033 &SnapshotId::random(),
2034 "bad",
2035 100
2036 ),
2037 Err(IcechunkFormatError {
2038 kind: IcechunkFormatErrorKind::SnapshotIdNotFound { .. },
2039 ..
2040 })
2041 ));
2042 assert!(matches!(
2044 repo.add_branch(SpecVersionBin::current(), "bar", &id1, "/foo/bar", 100),
2045 Err(IcechunkFormatError {
2046 kind: IcechunkFormatErrorKind::BranchAlreadyExists { .. },
2047 ..
2048 })
2049 ));
2050
2051 assert_eq!(
2052 repo.all_branches()?.collect::<HashSet<_>>(),
2053 [("main", 0), ("foo", 0), ("bar", 0)].into()
2054 );
2055
2056 let id2 = SnapshotId::random();
2057 let snap2 = SnapshotInfo {
2058 id: id2.clone(),
2059 parent_id: Some(id1.clone()),
2060 flushed_at: Utc::now(),
2061 message: "snap 2".to_string(),
2062 ..snap1.clone()
2063 };
2064 let repo = repo.add_snapshot(
2065 SpecVersionBin::current(),
2066 snap2.clone(),
2067 Some("main"),
2068 UpdateType::NewCommitUpdate {
2069 branch: "main".to_string(),
2070 new_snap_id: snap2.id.clone(),
2071 },
2072 None,
2073 "foo",
2074 100,
2075 )?;
2076 let repo =
2077 repo.add_branch(SpecVersionBin::current(), "baz", &id2, "/foo/bar", 100)?;
2078 assert_eq!(repo.resolve_branch("main")?, id2.clone());
2079 assert_eq!(repo.resolve_branch("foo")?, id1.clone());
2080 assert_eq!(repo.resolve_branch("bar")?, id1.clone());
2081 assert_eq!(repo.resolve_branch("baz")?, id2.clone());
2082
2083 let repo = repo.delete_branch(SpecVersionBin::current(), "bar", "bar", 100)?;
2084 assert!(repo.resolve_branch("bar").is_err());
2085 assert_eq!(
2086 repo.all_branches()?.map(|(n, _)| n).collect::<HashSet<_>>(),
2087 ["main", "foo", "baz"].into()
2088 );
2089
2090 assert!(
2091 repo.delete_branch(SpecVersionBin::current(), "bad-branch", "bad", 100)
2092 .is_err()
2093 );
2094
2095 let repo = repo.add_tag(SpecVersionBin::current(), "tag1", &id1, "tag1", 100)?;
2097 let repo = repo.add_tag(SpecVersionBin::current(), "tag2", &id2, "tag2", 100)?;
2098 assert!(
2099 repo.add_tag(
2100 SpecVersionBin::current(),
2101 "bad-snap",
2102 &SnapshotId::random(),
2103 "bad",
2104 100
2105 )
2106 .is_err()
2107 );
2108 assert!(
2109 repo.add_tag(SpecVersionBin::current(), "tag1", &id1, "tag1-again", 100)
2110 .is_err()
2111 );
2112 assert_eq!(repo.resolve_tag("tag1")?, id1.clone());
2113 assert_eq!(repo.resolve_tag("tag2")?, id2.clone());
2114 assert_eq!(
2115 repo.all_tags()?.map(|(n, _)| n).collect::<HashSet<_>>(),
2116 ["tag1", "tag2"].into()
2117 );
2118
2119 let repo = repo.add_tag(SpecVersionBin::current(), "tag3", &id1, "tag3", 100)?;
2121 let repo =
2122 repo.delete_tag(SpecVersionBin::current(), "tag3", "delete-tag3", 100)?;
2123 assert_eq!(
2124 repo.all_tags()?.map(|(n, _)| n).collect::<HashSet<_>>(),
2125 ["tag1", "tag2"].into()
2126 );
2127 assert!(
2129 repo.add_tag(SpecVersionBin::current(), "tag3", &id1, "tag3-again", 100)
2130 .is_err()
2131 );
2132 assert!(
2134 repo.delete_tag(SpecVersionBin::current(), "tag3", "delete-tag3-again", 100)
2135 .is_err()
2136 );
2137 assert_eq!(
2138 repo.all_tags()?.map(|(n, _)| n).collect::<HashSet<_>>(),
2139 ["tag1", "tag2"].into()
2140 );
2141 Ok(())
2142 }
2143
2144 #[test]
2145 fn test_repo_info_updates() -> Result<(), Box<dyn std::error::Error>> {
2146 let id1 = SnapshotId::random();
2147 let snap1 = SnapshotInfo {
2148 id: id1.clone(),
2149 parent_id: None,
2150 flushed_at: DateTime::from_timestamp_micros(1_000_000).unwrap(),
2152 message: "snap 1".to_string(),
2153 metadata: Default::default(),
2154 pruned_ancestor_tx_logs: vec![],
2155 };
2156
2157 let num_updates_per_file: u16 = 10;
2158 let n = num_updates_per_file as usize;
2159
2160 let mut repo = RepoInfo::initial(
2162 SpecVersionBin::current(),
2163 snap1,
2164 num_updates_per_file,
2165 None::<&()>,
2166 None,
2167 );
2168 assert_eq!(repo.latest_updates()?.count(), 1);
2169 let (last_update, _, file) = repo.latest_updates()?.next().unwrap()?;
2170 assert!(file.is_none());
2171 assert_eq!(last_update, UpdateType::RepoInitializedUpdate);
2172 assert!(repo.repo_before_updates()?.is_none());
2173 for i in 1..=(n - 1) {
2176 repo = repo.add_branch(
2177 SpecVersionBin::current(),
2178 i.to_string().as_str(),
2179 &id1,
2180 (i - 1).to_string().as_str(),
2181 num_updates_per_file,
2182 )?;
2183 }
2184
2185 assert_eq!(repo.latest_updates()?.count(), n);
2186 let updates = repo.latest_updates()?;
2187
2188 for (idx, update) in updates.enumerate() {
2190 let (update, _, file) = update?;
2191 if idx == n - 1 {
2192 assert_eq!(update, UpdateType::RepoInitializedUpdate);
2193 assert_eq!(file, Some("0"));
2194 } else {
2195 assert_eq!(
2196 update,
2197 UpdateType::BranchCreatedUpdate { name: (n - 1 - idx).to_string() }
2198 );
2199 if idx == 0 {
2200 assert!(file.is_none());
2201 } else {
2202 assert_eq!(file, Some((n - 1 - idx).to_string().as_str()));
2203 }
2204 }
2205 }
2206 assert!(repo.repo_before_updates()?.is_none());
2207
2208 repo = repo.add_tag(
2210 SpecVersionBin::current(),
2211 "tag",
2212 &id1,
2213 "first-branches",
2214 num_updates_per_file,
2215 )?;
2216 assert_eq!(repo.latest_updates()?.count(), n);
2218 assert_eq!(repo.repo_before_updates()?, Some("0"));
2220 let mut updates = repo.latest_updates()?;
2221 let (last_update, _, file) = updates.next().unwrap()?;
2223 assert_eq!(last_update, UpdateType::TagCreatedUpdate { name: "tag".to_string() });
2224 assert!(file.is_none());
2225
2226 let (last_update, _, file) = updates.next().unwrap()?;
2228 assert_eq!(
2229 last_update,
2230 UpdateType::BranchCreatedUpdate { name: (n - 1).to_string() }
2231 );
2232 assert_eq!(file, Some("first-branches"));
2233
2234 for (idx, update) in updates.enumerate() {
2236 let (update, _, file) = update?;
2237 assert_eq!(file, Some((n - 2 - idx).to_string().as_str()));
2238 assert_eq!(
2239 update,
2240 UpdateType::BranchCreatedUpdate { name: (n - 2 - idx).to_string() }
2241 );
2242 }
2243
2244 Ok(())
2245 }
2246
2247 #[test]
2248 fn test_update_timestamp_ordering_rejected() -> Result<(), Box<dyn std::error::Error>>
2249 {
2250 let flushed_at = DateTime::from_timestamp_micros(1_000_000).unwrap();
2251 let id1 = SnapshotId::random();
2252 let snap1 = SnapshotInfo {
2253 id: id1.clone(),
2254 parent_id: None,
2255 flushed_at,
2256 message: "snap 1".to_string(),
2257 metadata: Default::default(),
2258 pruned_ancestor_tx_logs: vec![],
2259 };
2260 let repo = RepoInfo::initial(
2261 SpecVersionBin::current(),
2262 snap1,
2263 100,
2264 None::<&()>,
2265 Some(flushed_at),
2266 );
2267
2268 let id2 = SnapshotId::random();
2271 let snap2 = SnapshotInfo {
2272 id: id2.clone(),
2273 parent_id: Some(id1.clone()),
2274 flushed_at: DateTime::from_timestamp_micros(1_000_000).unwrap(),
2275 message: "snap 2".to_string(),
2276 metadata: Default::default(),
2277 pruned_ancestor_tx_logs: vec![],
2278 };
2279 let result = repo.add_snapshot(
2280 SpecVersionBin::current(),
2281 snap2,
2282 Some("main"),
2283 UpdateType::NewCommitUpdate {
2284 branch: "main".to_string(),
2285 new_snap_id: id2.clone(),
2286 },
2287 Some(flushed_at),
2288 "backup",
2289 100,
2290 );
2291 assert!(matches!(
2292 result,
2293 Err(IcechunkFormatError {
2294 kind: IcechunkFormatErrorKind::InvalidUpdateTimestamp { .. },
2295 ..
2296 })
2297 ));
2298
2299 let flushed_at = DateTime::from_timestamp_micros(500_000).unwrap();
2302 let id3 = SnapshotId::random();
2303 let snap3 = SnapshotInfo {
2304 id: id3.clone(),
2305 parent_id: Some(id1.clone()),
2306 flushed_at,
2307 message: "snap 3".to_string(),
2308 metadata: Default::default(),
2309 pruned_ancestor_tx_logs: vec![],
2310 };
2311 let result = repo.add_snapshot(
2312 SpecVersionBin::current(),
2313 snap3,
2314 Some("main"),
2315 UpdateType::NewCommitUpdate {
2316 branch: "main".to_string(),
2317 new_snap_id: id3.clone(),
2318 },
2319 Some(flushed_at),
2320 "backup",
2321 100,
2322 );
2323 assert!(matches!(
2324 result,
2325 Err(IcechunkFormatError {
2326 kind: IcechunkFormatErrorKind::InvalidUpdateTimestamp { .. },
2327 ..
2328 })
2329 ));
2330
2331 let flushed_at = DateTime::from_timestamp_micros(2_000_000).unwrap();
2333 let id4 = SnapshotId::random();
2334 let snap4 = SnapshotInfo {
2335 id: id4.clone(),
2336 parent_id: Some(id1.clone()),
2337 flushed_at,
2338 message: "snap 4".to_string(),
2339 metadata: Default::default(),
2340 pruned_ancestor_tx_logs: vec![],
2341 };
2342 let result = repo.add_snapshot(
2343 SpecVersionBin::current(),
2344 snap4,
2345 Some("main"),
2346 UpdateType::NewCommitUpdate {
2347 branch: "main".to_string(),
2348 new_snap_id: id4.clone(),
2349 },
2350 Some(flushed_at),
2351 "backup",
2352 100,
2353 );
2354 assert!(result.is_ok());
2355
2356 Ok(())
2357 }
2358
2359 #[test]
2362 fn test_set_status_silent() -> Result<(), Box<dyn std::error::Error>> {
2363 let id1 = SnapshotId::random();
2364 let snap1 = SnapshotInfo {
2365 id: id1.clone(),
2366 parent_id: None,
2367 flushed_at: DateTime::from_timestamp_micros(1_000_000).unwrap(),
2368 message: "snap 1".to_string(),
2369 metadata: Default::default(),
2370 pruned_ancestor_tx_logs: vec![],
2371 };
2372
2373 let repo =
2374 RepoInfo::initial(SpecVersionBin::current(), snap1, 100, None::<&()>, None);
2375
2376 let repo =
2378 repo.add_branch(SpecVersionBin::current(), "dev", &id1, "backup0", 100)?;
2379
2380 let updates_before: Vec<_> =
2381 repo.latest_updates()?.collect::<Result<Vec<_>, _>>()?;
2382 let rbu_before = repo.repo_before_updates()?;
2383 let num_updates_before = updates_before.len();
2384
2385 let new_status = RepoStatus {
2387 availability: RepoAvailability::ReadOnly,
2388 set_at: Utc::now(),
2389 limited_availability_reason: Some("maintenance".to_string()),
2390 };
2391 let repo_normal = repo.set_status(
2392 SpecVersionBin::current(),
2393 &new_status,
2394 Some(("backup1", 100)),
2395 )?;
2396 let updates_after_normal: Vec<_> =
2397 repo_normal.latest_updates()?.collect::<Result<Vec<_>, _>>()?;
2398 assert_eq!(updates_after_normal.len(), num_updates_before + 1);
2399 assert!(matches!(
2400 &updates_after_normal[0].0,
2401 UpdateType::RepoStatusChangedUpdate { .. }
2402 ));
2403 let normal_status = repo_normal.status()?;
2404 assert_eq!(normal_status.availability, RepoAvailability::ReadOnly);
2405 assert_eq!(
2406 normal_status.limited_availability_reason.as_deref(),
2407 Some("maintenance")
2408 );
2409
2410 let silent_status = RepoStatus {
2412 availability: RepoAvailability::ReadOnly,
2413 set_at: Utc::now(),
2414 limited_availability_reason: None,
2415 };
2416 let repo_silent =
2417 repo.set_status(SpecVersionBin::current(), &silent_status, None)?;
2418
2419 let updates_after_silent: Vec<_> =
2420 repo_silent.latest_updates()?.collect::<Result<Vec<_>, _>>()?;
2421 assert_eq!(updates_after_silent.len(), num_updates_before);
2423 for (before, after) in updates_before.iter().zip(updates_after_silent.iter()) {
2424 assert_eq!(before.0, after.0);
2425 assert_eq!(before.1, after.1);
2426 assert_eq!(before.2, after.2);
2427 }
2428 assert_eq!(repo_silent.repo_before_updates()?, rbu_before);
2429
2430 let silent_result_status = repo_silent.status()?;
2432 assert_eq!(silent_result_status.availability, RepoAvailability::ReadOnly);
2433 assert!(silent_result_status.limited_availability_reason.is_none());
2434
2435 Ok(())
2436 }
2437}