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(|a, b| a.id.0.cmp(&b.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 let args = generated::SnapshotInfoArgs {
363 id: Some(&id),
364 parent_offset,
365 flushed_at: snap.flushed_at.timestamp_micros() as u64,
366 message: Some(builder.create_string(snap.message.as_str())),
367 metadata: Some(metadata),
368 };
369 Ok::<_, IcechunkFormatError>(generated::SnapshotInfo::create(
370 &mut builder,
371 &args,
372 ))
373 })
374 .try_collect()?;
375 let snapshots = builder.create_vector(&snapshots);
376
377 let limited_reason = status
378 .limited_availability_reason
379 .as_deref()
380 .map(|s| builder.create_string(s));
381 let status = generated::RepoStatus::create(
382 &mut builder,
383 &generated::RepoStatusArgs {
384 availability: status.availability.into(),
385 set_at: status.set_at.timestamp_micros() as u64,
386 limited_availability_reason: limited_reason,
387 },
388 );
389
390 let metadata_items: Vec<_> = metadata
391 .iter()
392 .map(|(k, v)| {
393 let name = builder.create_shared_string(k.as_str());
394 let serialized = flexbuffers::to_vec(v).map_err(Box::new).capture()?;
395 let value = builder.create_vector(serialized.as_slice());
396 let item = generated::MetadataItem::create(
397 &mut builder,
398 &generated::MetadataItemArgs { name: Some(name), value: Some(value) },
399 );
400 Ok::<_, IcechunkFormatError>(item)
401 })
402 .try_collect()?;
403
404 let metadata = builder.create_vector(metadata_items.as_slice());
405
406 let enabled_feature_flags =
407 sorted_enabled_feature_flags.map(|it| builder.create_vector_from_iter(it));
408 let disabled_feature_flags =
409 sorted_disabled_feature_flags.map(|it| builder.create_vector_from_iter(it));
410
411 let (latest_updates, repo_before_updates) = build_updates(&mut builder)?;
412
413 let config = config_bytes.map(|bytes| builder.create_vector(bytes));
414
415 let repo_args = generated::RepoArgs {
417 tags: Some(tags),
418 branches: Some(branches),
419 deleted_tags: Some(deleted_tags),
420 snapshots: Some(snapshots),
421 spec_version: spec_version as u8,
422 status: Some(status),
423 metadata: Some(metadata),
424 latest_updates: Some(latest_updates),
425 repo_before_updates,
426 config,
427 enabled_feature_flags,
428 disabled_feature_flags,
429 ..Default::default()
430 };
431 let repo = generated::Repo::create(&mut builder, &repo_args);
432 builder.finish(repo, Some("Ichk"));
433 let (mut buffer, offset) = builder.collapse();
434 buffer.drain(0..offset);
435 buffer.shrink_to_fit();
436 Ok(Self { buffer })
437 }
438
439 fn mk_latest_updates<
440 'bldr,
441 'a,
442 I: IntoIterator<Item = IcechunkResult<(UpdateType, DateTime<Utc>, Option<&'a str>)>>,
443 >(
444 builder: &mut flatbuffers::FlatBufferBuilder<'bldr>,
445 update: UpdateInfo<I>,
446 backup_path: Option<&'a str>,
447 num_updates_per_file: u16,
448 previous_info: Option<&'a str>,
449 ) -> LatestUpdatesResult<'bldr> {
450 let mut previous_updates = update.previous_updates.into_iter();
452 let last_update = previous_updates.next().map(|maybe_data| {
453 maybe_data.map(|(ut, dt, path)| {
454 assert!(
455 path.is_none(),
456 "Invalid latest update iterator, last element has backup path"
457 );
458 (ut, dt, backup_path)
459 })
460 });
461 assert!(
468 backup_path.is_none() || last_update.is_some(),
469 "A backup path must not be provided without previous updates"
470 );
471
472 if let Some(Ok((_, latest_time, _))) = &last_update
474 && update.update_time <= *latest_time
475 {
476 return Err(IcechunkFormatErrorKind::InvalidUpdateTimestamp {
477 latest_time: *latest_time,
478 new_time: update.update_time,
479 })
480 .capture();
481 }
482
483 let new_updates: Box<dyn Iterator<Item = _>> =
484 if let Some(last_update) = last_update {
485 Box::new(
486 [Ok((update.update_type, update.update_time, None)), last_update]
487 .into_iter(),
488 )
489 } else {
490 Box::new([Ok((update.update_type, update.update_time, None))].into_iter())
491 };
492
493 let all_updates = new_updates.into_iter().chain(previous_updates);
494 let mut repo_before_updates = previous_info;
497
498 let num_updates = num_updates_per_file as usize;
499 let mut updates = Vec::new();
500 for maybe_data in all_updates {
501 let (u_type, u_time, file) = maybe_data?;
502
503 if updates.len() >= num_updates
508 && let Some(bp) = file
509 {
510 repo_before_updates = Some(bp);
511 break;
512 }
513
514 let (update_type_type, update_type) = update_type_to_fb(builder, &u_type)?;
515 let file = file.map(|file| builder.create_string(file));
516 updates.push(generated::Update::create(
517 builder,
518 &generated::UpdateArgs {
519 update_type_type,
520 update_type: Some(update_type),
521 updated_at: u_time.timestamp_micros() as u64,
522 backup_path: file,
523 },
524 ));
525 }
526
527 debug_assert!(!updates.is_empty(), "Must have at least one update in repo file");
528
529 let updates = builder.create_vector(&updates);
530 let repo_before_updates = repo_before_updates.map(|s| builder.create_string(s));
531 Ok((updates, repo_before_updates))
532 }
533
534 fn mk_latest_updates_passthrough<'bldr, 'a>(
535 builder: &mut flatbuffers::FlatBufferBuilder<'bldr>,
536 updates: impl IntoIterator<Item = UpdateTuple<'a>>,
537 repo_before_updates: Option<&'a str>,
538 ) -> LatestUpdatesResult<'bldr> {
539 let mut fb_updates = Vec::new();
540 for maybe_data in updates {
541 let (u_type, u_time, file) = maybe_data?;
542 let (update_type_type, update_type) = update_type_to_fb(builder, &u_type)?;
543 let file = file.map(|file| builder.create_string(file));
544 fb_updates.push(generated::Update::create(
545 builder,
546 &generated::UpdateArgs {
547 update_type_type,
548 update_type: Some(update_type),
549 updated_at: u_time.timestamp_micros() as u64,
550 backup_path: file,
551 },
552 ));
553 }
554 debug_assert!(
555 !fb_updates.is_empty(),
556 "Must have at least one update in repo file"
557 );
558 let updates = builder.create_vector(&fb_updates);
559 let repo_before_updates = repo_before_updates.map(|s| builder.create_string(s));
560 Ok((updates, repo_before_updates))
561 }
562
563 pub fn initial<C: Serialize>(
564 spec_version: SpecVersionBin,
565 snapshot: SnapshotInfo,
566 num_updates_per_file: u16,
567 config: Option<&C>,
568 update_time: Option<DateTime<Utc>>,
569 ) -> Self {
570 #[expect(clippy::expect_used)]
571 let config_bytes =
572 config.map(|c| flexbuffers::to_vec(c).expect("Cannot serialize config"));
573 #[expect(clippy::expect_used)]
575 Self::from_parts(
576 spec_version,
577 [],
578 [("main", 0)],
579 [],
580 [snapshot],
581 &Default::default(),
582 |builder| {
583 Self::mk_latest_updates(
584 builder,
585 UpdateInfo {
586 update_type: UpdateType::RepoInitializedUpdate,
587 update_time: update_time.unwrap_or_else(Utc::now),
588 previous_updates: [],
589 },
590 None,
591 num_updates_per_file,
592 None,
593 )
594 },
595 config_bytes.as_deref(),
596 None::<std::iter::Empty<u16>>,
597 None::<std::iter::Empty<u16>>,
598 &RepoStatus {
599 availability: RepoAvailability::Online,
600 set_at: update_time.unwrap_or_else(Utc::now),
601 limited_availability_reason: None,
602 },
603 )
604 .expect("Cannot generate initial snapshot")
605 }
606
607 pub fn config_bytes_raw(&self) -> IcechunkResult<Option<Vec<u8>>> {
609 Ok(self.root()?.config().map(|v| v.bytes().to_vec()))
610 }
611
612 pub fn config<C: serde::de::DeserializeOwned>(&self) -> IcechunkResult<Option<C>> {
616 match self.root()?.config() {
617 None => Ok(None),
618 Some(config_fb) => {
619 let config: C = flexbuffers::from_slice(config_fb.bytes())
620 .map_err(Box::new)
621 .capture()?;
622 Ok(Some(config))
623 }
624 }
625 }
626
627 pub fn metadata(&self) -> IcechunkResult<SnapshotProperties> {
628 self.root()?
629 .metadata()
630 .unwrap_or_default()
631 .iter()
632 .map(|item| {
633 let key = item.name().to_string();
634 let value = flexbuffers::from_slice(item.value().bytes())
635 .map_err(Box::new)
636 .capture()?;
637 Ok((key, value))
638 })
639 .try_collect()
640 }
641
642 pub fn status(&self) -> IcechunkResult<RepoStatus> {
643 let root = self.root()?;
644 let fb_status = root.status();
645 fb_status.try_into()
646 }
647
648 pub fn enabled_feature_flags(
649 &self,
650 ) -> IcechunkResult<Option<impl DoubleEndedIterator<Item = u16> + ExactSizeIterator>>
651 {
652 Ok(self.root()?.enabled_feature_flags().map(|v| v.iter()))
653 }
654
655 pub fn disabled_feature_flags(
656 &self,
657 ) -> IcechunkResult<Option<impl DoubleEndedIterator<Item = u16> + ExactSizeIterator>>
658 {
659 Ok(self.root()?.disabled_feature_flags().map(|v| v.iter()))
660 }
661
662 pub fn feature_flag_enabled(&self, id: u16) -> IcechunkResult<Option<bool>> {
666 let root = self.root()?;
667 if root
668 .enabled_feature_flags()
669 .and_then(|v| v.lookup_by_key(id, |this, key| this.cmp(key)))
670 .is_some()
671 {
672 return Ok(Some(true));
673 }
674 if root
675 .disabled_feature_flags()
676 .and_then(|v| v.lookup_by_key(id, |this, key| this.cmp(key)))
677 .is_some()
678 {
679 return Ok(Some(false));
680 }
681 Ok(None)
682 }
683
684 fn all_tags(&self) -> IcechunkResult<impl Iterator<Item = (&str, u32)>> {
685 Ok(self.root()?.tags().iter().map(|r| (r.name(), r.snapshot_index())))
686 }
687
688 fn all_branches(&self) -> IcechunkResult<impl Iterator<Item = (&str, u32)>> {
689 Ok(self.root()?.branches().iter().map(|r| (r.name(), r.snapshot_index())))
690 }
691
692 pub fn deleted_tags(&self) -> IcechunkResult<impl Iterator<Item = &str>> {
693 Ok(self.root()?.deleted_tags().iter())
694 }
695
696 pub fn all_snapshots(
697 &self,
698 ) -> IcechunkResult<impl Iterator<Item = IcechunkResult<SnapshotInfo>>> {
699 let root = self.root()?;
700 Ok(root.snapshots().iter().map(move |snap| mk_snapshot_info(&root, &snap)))
701 }
702
703 pub fn update_feature_flag(
705 &self,
706 spec_version: SpecVersionBin,
707 flag_id: u16,
708 enabled: Option<bool>,
709 previous_file: &str,
710 num_updates_per_file: u16,
711 ) -> IcechunkResult<Self> {
712 let snaps: Vec<_> = self.all_snapshots()?.try_collect()?;
713 let (eff, dff): (Option<Vec<_>>, Option<Vec<_>>) = match enabled {
714 Some(false) => {
715 let e = self
716 .enabled_feature_flags()?
717 .map(|it| it.filter(|x| *x != flag_id).collect());
718
719 let mut d: BTreeSet<_> = self
720 .disabled_feature_flags()?
721 .map(|it| it.collect())
722 .unwrap_or_default();
723 d.insert(flag_id);
724 let d = d.into_iter().collect();
725 (e, Some(d))
726 }
727 Some(true) => {
728 let d = self
729 .disabled_feature_flags()?
730 .map(|it| it.filter(|x| *x != flag_id).collect());
731
732 let mut e: BTreeSet<_> = self
733 .enabled_feature_flags()?
734 .map(|it| it.collect())
735 .unwrap_or_default();
736 e.insert(flag_id);
737 let e = e.into_iter().collect();
738 (Some(e), d)
739 }
740 None => {
741 let e = self
742 .enabled_feature_flags()?
743 .map(|it| it.filter(|x| *x != flag_id).collect());
744 let d = self
745 .disabled_feature_flags()?
746 .map(|it| it.filter(|x| *x != flag_id).collect());
747 (e, d)
748 }
749 };
750
751 let latest_updates = self.latest_updates()?;
752 let rbu = self.repo_before_updates()?;
753 Self::from_parts(
754 spec_version,
755 self.all_tags()?,
756 self.all_branches()?,
757 self.deleted_tags()?,
758 snaps,
759 &self.metadata()?,
760 |builder| {
761 Self::mk_latest_updates(
762 builder,
763 UpdateInfo {
764 update_type: UpdateType::FeatureFlagChanged {
765 id: flag_id,
766 new_value: enabled,
767 },
768 update_time: Utc::now(),
769 previous_updates: latest_updates,
770 },
771 Some(previous_file),
772 num_updates_per_file,
773 rbu,
774 )
775 },
776 self.config_bytes_raw()?.as_deref(),
777 eff.map(|it| it.into_iter()),
778 dff.map(|it| it.into_iter()),
779 &self.status()?,
780 )
781 }
782
783 #[expect(clippy::too_many_arguments)]
784 pub fn add_snapshot(
785 &self,
786 spec_version: SpecVersionBin,
787 snap: SnapshotInfo,
788 branch: Option<&str>,
789 update_type: UpdateType,
790 update_time: Option<DateTime<Utc>>, previous_file: &str,
792 num_updates_per_file: u16,
793 ) -> IcechunkResult<Self> {
794 let mut snapshots: Vec<_> = self.all_snapshots()?.try_collect()?;
795 let new_index = match snapshots.binary_search_by_key(&&snap.id, |snap| &snap.id) {
796 Ok(_) => Err(IcechunkFormatErrorKind::DuplicateSnapshotId {
797 snapshot_id: snap.id.clone(),
798 })
799 .capture(),
800 Err(idx) => Ok(idx),
801 }?;
802
803 snapshots.insert(new_index, snap);
804
805 let tags = self.all_tags()?.map(|(name, idx)| {
806 if idx as usize >= new_index { (name, idx + 1) } else { (name, idx) }
807 });
808 let branches = self.all_branches()?.map(|(name, idx)| {
809 if Some(name) == branch {
810 (name, new_index as u32)
811 } else if idx as usize >= new_index {
812 (name, idx + 1)
813 } else {
814 (name, idx)
815 }
816 });
817
818 let latest_updates = self.latest_updates()?;
819 let rbu = self.repo_before_updates()?;
820 let res = Self::from_parts(
821 spec_version,
822 tags,
823 branches,
824 self.deleted_tags()?,
825 snapshots,
826 &self.metadata()?,
827 |builder| {
828 Self::mk_latest_updates(
829 builder,
830 UpdateInfo {
831 update_type,
832 update_time: update_time.unwrap_or_else(Utc::now),
833 previous_updates: latest_updates,
834 },
835 Some(previous_file),
836 num_updates_per_file,
837 rbu,
838 )
839 },
840 self.config_bytes_raw()?.as_deref(),
841 self.enabled_feature_flags()?,
842 self.disabled_feature_flags()?,
843 &self.status()?,
844 )?;
845 Ok(res)
846 }
847
848 pub fn add_branch(
849 &self,
850 spec_version: SpecVersionBin,
851 name: &str,
852 snap: &SnapshotId,
853 previous_file: &str,
854 num_updates_per_file: u16,
855 ) -> IcechunkResult<Self> {
856 if let Ok(snapshot_id) = self.resolve_branch(name) {
857 return Err(IcechunkFormatErrorKind::BranchAlreadyExists {
858 branch: name.to_string(),
859 snapshot_id,
860 })
861 .capture();
862 }
863
864 match self.resolve_snapshot_index(snap)? {
865 Some(snap_idx) => {
866 let mut branches: Vec<_> = self.all_branches()?.collect();
867 branches.push((name, snap_idx as u32));
868 branches.sort_by(|(name1, _), (name2, _)| name1.cmp(name2));
869 let snaps: Vec<_> = self.all_snapshots()?.try_collect()?;
870 let latest_updates = self.latest_updates()?;
871 let rbu = self.repo_before_updates()?;
872 Ok(Self::from_parts(
873 spec_version,
874 self.all_tags()?,
875 branches,
876 self.deleted_tags()?,
877 snaps,
878 &self.metadata()?,
879 |builder| {
880 Self::mk_latest_updates(
881 builder,
882 UpdateInfo {
883 update_type: UpdateType::BranchCreatedUpdate {
884 name: name.to_string(),
885 },
886 update_time: Utc::now(),
887 previous_updates: latest_updates,
888 },
889 Some(previous_file),
890 num_updates_per_file,
891 rbu,
892 )
893 },
894 self.config_bytes_raw()?.as_deref(),
895 self.enabled_feature_flags()?,
896 self.disabled_feature_flags()?,
897 &self.status()?,
898 )?)
899 }
900 None => Err(IcechunkFormatErrorKind::SnapshotIdNotFound {
901 snapshot_id: snap.clone(),
902 })
903 .capture(),
904 }
905 }
906
907 pub fn delete_branch(
908 &self,
909 spec_version: SpecVersionBin,
910 name: &str,
911 previous_file: &str,
912 num_updates_per_file: u16,
913 ) -> IcechunkResult<Self> {
914 match self.resolve_branch(name) {
915 Ok(previous_snap_id) => {
916 let mut branches: Vec<_> = self.all_branches()?.collect();
917 branches.retain(|(n, _)| n != &name);
919 let snaps: Vec<_> = self.all_snapshots()?.try_collect()?;
920 let latest_updates = self.latest_updates()?;
921 let rbu = self.repo_before_updates()?;
922 Self::from_parts(
923 spec_version,
924 self.all_tags()?,
925 branches,
926 self.deleted_tags()?,
927 snaps,
928 &self.metadata()?,
929 |builder| {
930 Self::mk_latest_updates(
931 builder,
932 UpdateInfo {
933 update_type: UpdateType::BranchDeletedUpdate {
934 name: name.to_string(),
935 previous_snap_id,
936 },
937 update_time: Utc::now(),
938 previous_updates: latest_updates,
939 },
940 Some(previous_file),
941 num_updates_per_file,
942 rbu,
943 )
944 },
945 self.config_bytes_raw()?.as_deref(),
946 self.enabled_feature_flags()?,
947 self.disabled_feature_flags()?,
948 &self.status()?,
949 )
950 }
951 Err(IcechunkFormatError {
952 kind: IcechunkFormatErrorKind::BranchNotFound { .. },
953 ..
954 }) => {
955 Err(IcechunkFormatErrorKind::BranchNotFound { branch: name.to_string() })
956 .capture()
957 }
958 Err(err) => Err(err),
959 }
960 }
961
962 pub fn update_branch(
963 &self,
964 spec_version: SpecVersionBin,
965 name: &str,
966 new_snap: &SnapshotId,
967 previous_file: &str,
968 num_updates_per_file: u16,
969 ) -> IcechunkResult<Self> {
970 let previous_snap_id = self.resolve_branch(name)?;
971 match self.resolve_snapshot_index(new_snap)? {
972 Some(snap_idx) => {
973 let branches = self.all_branches()?.map(|(br, idx)| {
974 if br == name { (br, snap_idx as u32) } else { (br, idx) }
975 });
976 let snaps: Vec<_> = self.all_snapshots()?.try_collect()?;
977 let latest_updates = self.latest_updates()?;
978 let rbu = self.repo_before_updates()?;
979 Ok(Self::from_parts(
980 spec_version,
981 self.all_tags()?,
982 branches,
983 self.deleted_tags()?,
984 snaps,
985 &self.metadata()?,
986 |builder| {
987 Self::mk_latest_updates(
988 builder,
989 UpdateInfo {
990 update_type: UpdateType::BranchResetUpdate {
991 name: name.to_string(),
992 previous_snap_id,
993 },
994 update_time: Utc::now(),
995 previous_updates: latest_updates,
996 },
997 Some(previous_file),
998 num_updates_per_file,
999 rbu,
1000 )
1001 },
1002 self.config_bytes_raw()?.as_deref(),
1003 self.enabled_feature_flags()?,
1004 self.disabled_feature_flags()?,
1005 &self.status()?,
1006 )?)
1007 }
1008 None => Err(IcechunkFormatErrorKind::SnapshotIdNotFound {
1009 snapshot_id: new_snap.clone(),
1010 })
1011 .capture(),
1012 }
1013 }
1014
1015 pub fn add_tag(
1016 &self,
1017 spec_version: SpecVersionBin,
1018 name: &str,
1019 snap: &SnapshotId,
1020 previous_file: &str,
1021 num_updates_per_file: u16,
1022 ) -> IcechunkResult<Self> {
1023 if self.resolve_tag(name).is_ok() {
1024 return Err(IcechunkFormatErrorKind::TagAlreadyExists {
1025 tag: name.to_string(),
1026 })
1027 .capture();
1028 }
1029 if self.tag_was_deleted(name)? {
1030 return Err(IcechunkFormatErrorKind::TagPreviouslyDeleted {
1031 tag: name.to_string(),
1032 })
1033 .capture();
1034 }
1035
1036 match self.resolve_snapshot_index(snap)? {
1037 Some(snap_idx) => {
1038 let mut tags: Vec<_> = self.all_tags()?.collect();
1039 tags.push((name, snap_idx as u32));
1040 tags.sort_by(|(name1, _), (name2, _)| name1.cmp(name2));
1041 let snaps: Vec<_> = self.all_snapshots()?.try_collect()?;
1042 let latest_updates = self.latest_updates()?;
1043 let rbu = self.repo_before_updates()?;
1044 Ok(Self::from_parts(
1045 spec_version,
1046 tags,
1047 self.all_branches()?,
1048 self.deleted_tags()?,
1049 snaps,
1050 &self.metadata()?,
1051 |builder| {
1052 Self::mk_latest_updates(
1053 builder,
1054 UpdateInfo {
1055 update_type: UpdateType::TagCreatedUpdate {
1056 name: name.to_string(),
1057 },
1058 update_time: Utc::now(),
1059 previous_updates: latest_updates,
1060 },
1061 Some(previous_file),
1062 num_updates_per_file,
1063 rbu,
1064 )
1065 },
1066 self.config_bytes_raw()?.as_deref(),
1067 self.enabled_feature_flags()?,
1068 self.disabled_feature_flags()?,
1069 &self.status()?,
1070 )?)
1071 }
1072 None => Err(IcechunkFormatErrorKind::SnapshotIdNotFound {
1073 snapshot_id: snap.clone(),
1074 })
1075 .capture(),
1076 }
1077 }
1078
1079 pub fn delete_tag(
1080 &self,
1081 spec_version: SpecVersionBin,
1082 name: &str,
1083 previous_file: &str,
1084 num_updates_per_file: u16,
1085 ) -> IcechunkResult<Self> {
1086 match self.resolve_tag(name) {
1087 Ok(previous_snap_id) => {
1088 let mut tags: Vec<_> = self.all_tags()?.collect();
1089 tags.retain(|(n, _)| n != &name);
1091
1092 let mut deleted_tags: BTreeSet<_> = self.deleted_tags()?.collect();
1093 debug_assert!(!deleted_tags.contains(name));
1094 deleted_tags.insert(name);
1095
1096 let snaps: Vec<_> = self.all_snapshots()?.try_collect()?;
1097 let latest_updates = self.latest_updates()?;
1098 let rbu = self.repo_before_updates()?;
1099 Self::from_parts(
1100 spec_version,
1101 tags,
1102 self.all_branches()?,
1103 deleted_tags,
1104 snaps,
1105 &self.metadata()?,
1106 |builder| {
1107 Self::mk_latest_updates(
1108 builder,
1109 UpdateInfo {
1110 update_type: UpdateType::TagDeletedUpdate {
1111 name: name.to_string(),
1112 previous_snap_id,
1113 },
1114 update_time: Utc::now(),
1115 previous_updates: latest_updates,
1116 },
1117 Some(previous_file),
1118 num_updates_per_file,
1119 rbu,
1120 )
1121 },
1122 self.config_bytes_raw()?.as_deref(),
1123 self.enabled_feature_flags()?,
1124 self.disabled_feature_flags()?,
1125 &self.status()?,
1126 )
1127 }
1128 Err(IcechunkFormatError {
1129 kind: IcechunkFormatErrorKind::TagNotFound { .. },
1130 ..
1131 }) => Err(IcechunkFormatErrorKind::TagNotFound { tag: name.to_string() })
1132 .capture(),
1133 Err(err) => Err(err),
1134 }
1135 }
1136
1137 pub fn set_metadata(
1138 &self,
1139 spec_version: SpecVersionBin,
1140 metadata: &SnapshotProperties,
1141 previous_file: &str,
1142 num_updates_per_file: u16,
1143 ) -> IcechunkResult<Self> {
1144 let snaps: Vec<_> = self.all_snapshots()?.try_collect()?;
1145 let latest_updates = self.latest_updates()?;
1146 let rbu = self.repo_before_updates()?;
1147 Self::from_parts(
1148 spec_version,
1149 self.all_tags()?,
1150 self.all_branches()?,
1151 self.deleted_tags()?,
1152 snaps,
1153 metadata,
1154 |builder| {
1155 Self::mk_latest_updates(
1156 builder,
1157 UpdateInfo {
1158 update_type: UpdateType::MetadataChangedUpdate,
1159 update_time: Utc::now(),
1160 previous_updates: latest_updates,
1161 },
1162 Some(previous_file),
1163 num_updates_per_file,
1164 rbu,
1165 )
1166 },
1167 self.config_bytes_raw()?.as_deref(),
1168 self.enabled_feature_flags()?,
1169 self.disabled_feature_flags()?,
1170 &self.status()?,
1171 )
1172 }
1173
1174 pub fn set_config<C: Serialize>(
1176 &self,
1177 spec_version: SpecVersionBin,
1178 config: &C,
1179 previous_file: &str,
1180 num_updates_per_file: u16,
1181 ) -> IcechunkResult<Self> {
1182 let config_bytes = flexbuffers::to_vec(config).map_err(Box::new).capture()?;
1183 let snaps: Vec<_> = self.all_snapshots()?.try_collect()?;
1184 let latest_updates = self.latest_updates()?;
1185 let rbu = self.repo_before_updates()?;
1186 Self::from_parts(
1187 spec_version,
1188 self.all_tags()?,
1189 self.all_branches()?,
1190 self.deleted_tags()?,
1191 snaps,
1192 &self.metadata()?,
1193 |builder| {
1194 Self::mk_latest_updates(
1195 builder,
1196 UpdateInfo {
1197 update_type: UpdateType::ConfigChangedUpdate,
1198 update_time: Utc::now(),
1199 previous_updates: latest_updates,
1200 },
1201 Some(previous_file),
1202 num_updates_per_file,
1203 rbu,
1204 )
1205 },
1206 Some(config_bytes.as_slice()),
1207 self.enabled_feature_flags()?,
1208 self.disabled_feature_flags()?,
1209 &self.status()?,
1210 )
1211 }
1212
1213 pub fn set_status(
1221 &self,
1222 spec_version: SpecVersionBin,
1223 status: &RepoStatus,
1224 update_config: Option<(&str, u16)>,
1225 ) -> IcechunkResult<Self> {
1226 let snaps: Vec<_> = self.all_snapshots()?.try_collect()?;
1227 let latest_updates = self.latest_updates()?;
1228 let rbu = self.repo_before_updates()?;
1229 if let Some((previous_file, num_updates_per_file)) = update_config {
1230 let status_update = status.clone();
1231 Self::from_parts(
1232 spec_version,
1233 self.all_tags()?,
1234 self.all_branches()?,
1235 self.deleted_tags()?,
1236 snaps,
1237 &self.metadata()?,
1238 |builder| {
1239 Self::mk_latest_updates(
1240 builder,
1241 UpdateInfo {
1242 update_type: UpdateType::RepoStatusChangedUpdate {
1243 status: status_update,
1244 },
1245 update_time: Utc::now(),
1246 previous_updates: latest_updates,
1247 },
1248 Some(previous_file),
1249 num_updates_per_file,
1250 rbu,
1251 )
1252 },
1253 self.config_bytes_raw()?.as_deref(),
1254 self.enabled_feature_flags()?,
1255 self.disabled_feature_flags()?,
1256 status,
1257 )
1258 } else {
1259 Self::from_parts(
1260 spec_version,
1261 self.all_tags()?,
1262 self.all_branches()?,
1263 self.deleted_tags()?,
1264 snaps,
1265 &self.metadata()?,
1266 |builder| {
1267 Self::mk_latest_updates_passthrough(builder, latest_updates, rbu)
1268 },
1269 self.config_bytes_raw()?.as_deref(),
1270 self.enabled_feature_flags()?,
1271 self.disabled_feature_flags()?,
1272 status,
1273 )
1274 }
1275 }
1276
1277 pub fn from_buffer(buffer: Vec<u8>) -> IcechunkResult<RepoInfo> {
1278 let _ = flatbuffers::root_with_opts::<generated::Repo<'_>>(
1279 &ROOT_OPTIONS,
1280 buffer.as_slice(),
1281 )
1282 .capture()?;
1283 Ok(RepoInfo { buffer })
1284 }
1285
1286 pub fn bytes(&self) -> &[u8] {
1287 self.buffer.as_slice()
1288 }
1289
1290 fn root(&self) -> IcechunkResult<generated::Repo<'_>> {
1291 flatbuffers::root::<generated::Repo<'_>>(&self.buffer).capture()
1292 }
1293
1294 pub fn tag_names(&self) -> IcechunkResult<impl Iterator<Item = &str>> {
1295 Ok(self.root()?.tags().iter().map(|r| r.name()))
1296 }
1297
1298 pub fn branch_names(&self) -> IcechunkResult<impl Iterator<Item = &str>> {
1299 Ok(self.root()?.branches().iter().map(|r| r.name()))
1300 }
1301
1302 pub fn tags(&self) -> IcechunkResult<impl Iterator<Item = (&str, SnapshotId)>> {
1303 let root = self.root()?;
1304 Ok(self.all_tags()?.map(move |(name, idx)| {
1305 (name, SnapshotId::new(root.snapshots().get(idx as usize).id().0))
1306 }))
1307 }
1308
1309 pub fn branches(&self) -> IcechunkResult<impl Iterator<Item = (&str, SnapshotId)>> {
1310 let root = self.root()?;
1311 Ok(self.all_branches()?.map(move |(name, idx)| {
1312 (name, SnapshotId::new(root.snapshots().get(idx as usize).id().0))
1313 }))
1314 }
1315
1316 pub fn resolve_tag(&self, name: &str) -> IcechunkResult<SnapshotId> {
1317 let root = self.root()?;
1318 let res = root
1319 .tags()
1320 .lookup_by_key(name, |r, key| r.name().cmp(key))
1321 .map(|r| {
1322 let index = r.snapshot_index();
1323 SnapshotId::new(root.snapshots().get(index as usize).id().0)
1324 })
1325 .ok_or_else(|| IcechunkFormatErrorKind::TagNotFound { tag: name.to_string() })
1326 .capture()?;
1327
1328 Ok(res)
1329 }
1330
1331 pub fn tag_was_deleted(&self, name: &str) -> IcechunkResult<bool> {
1332 let root = self.root()?;
1333 let res = root.deleted_tags().lookup_by_key(name, |name, key| name.cmp(key));
1334 Ok(res.is_some())
1335 }
1336
1337 pub fn resolve_branch(&self, name: &str) -> IcechunkResult<SnapshotId> {
1338 let root = self.root()?;
1339 let res = root
1340 .branches()
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::BranchNotFound {
1347 branch: name.to_string(),
1348 })
1349 .capture()?;
1350
1351 Ok(res)
1352 }
1353
1354 pub fn spec_version(&self) -> IcechunkResult<SpecVersionBin> {
1355 let raw = self.root()?.spec_version();
1356 raw.try_into()
1357 .map_err(|_| IcechunkFormatErrorKind::InvalidSpecVersion {
1358 found: raw,
1359 max_supported: SpecVersionBin::current() as u8,
1360 })
1361 .capture()
1362 }
1363
1364 pub fn latest_updates(
1365 &self,
1366 ) -> IcechunkResult<impl Iterator<Item = UpdateTuple<'_>>> {
1367 let res = self.root()?.latest_updates().iter().map(|up| self.update_to_tuple(up));
1368 Ok(res)
1369 }
1370
1371 fn update_to_tuple<'a>(
1372 &'a self,
1373 update: generated::Update<'a>,
1374 ) -> IcechunkResult<(UpdateType, DateTime<Utc>, Option<&'a str>)> {
1375 let ty = self.mk_update_type(&update)?;
1376 let ts = timestamp_to_timestamp(update.updated_at())?;
1377 let bp = update.backup_path();
1378 Ok((ty, ts, bp))
1379 }
1380
1381 fn mk_update_type(
1382 &self,
1383 update: &generated::Update<'_>,
1384 ) -> IcechunkResult<UpdateType> {
1385 #[expect(clippy::unwrap_used)]
1386 match update.update_type_type() {
1387 generated::UpdateType::RepoInitializedUpdate => {
1388 Ok(UpdateType::RepoInitializedUpdate)
1389 }
1390 generated::UpdateType::RepoMigratedUpdate => {
1391 let up = update.update_type_as_repo_migrated_update().unwrap();
1392 let from_raw = up.from_version();
1393 let to_raw = up.to_version();
1394 Ok(UpdateType::RepoMigratedUpdate {
1395 from_version: from_raw
1396 .try_into()
1397 .map_err(|_| IcechunkFormatErrorKind::InvalidSpecVersion {
1398 found: from_raw,
1399 max_supported: SpecVersionBin::current() as u8,
1400 })
1401 .capture()?,
1402 to_version: to_raw
1403 .try_into()
1404 .map_err(|_| IcechunkFormatErrorKind::InvalidSpecVersion {
1405 found: to_raw,
1406 max_supported: SpecVersionBin::current() as u8,
1407 })
1408 .capture()?,
1409 })
1410 }
1411 generated::UpdateType::RepoStatusChangedUpdate => {
1412 let up = update.update_type_as_repo_status_changed_update().unwrap();
1413 let fb_status = up.status().unwrap();
1414 let status = fb_status.try_into()?;
1415
1416 Ok(UpdateType::RepoStatusChangedUpdate { status })
1417 }
1418 generated::UpdateType::ConfigChangedUpdate => {
1419 Ok(UpdateType::ConfigChangedUpdate)
1420 }
1421 generated::UpdateType::MetadataChangedUpdate => {
1422 Ok(UpdateType::MetadataChangedUpdate)
1423 }
1424 generated::UpdateType::TagCreatedUpdate => {
1425 let up = update.update_type_as_tag_created_update().unwrap();
1426 Ok(UpdateType::TagCreatedUpdate { name: up.name().to_string() })
1427 }
1428 generated::UpdateType::TagDeletedUpdate => {
1429 let up = update.update_type_as_tag_deleted_update().unwrap();
1430 let previous_snap_id = SnapshotId::new(up.previous_snap_id().0);
1431 Ok(UpdateType::TagDeletedUpdate {
1432 name: up.name().to_string(),
1433 previous_snap_id,
1434 })
1435 }
1436 generated::UpdateType::BranchCreatedUpdate => {
1437 let up = update.update_type_as_branch_created_update().unwrap();
1438 Ok(UpdateType::BranchCreatedUpdate { name: up.name().to_string() })
1439 }
1440 generated::UpdateType::BranchDeletedUpdate => {
1441 let up = update.update_type_as_branch_deleted_update().unwrap();
1442 let previous_snap_id = SnapshotId::new(up.previous_snap_id().0);
1443 Ok(UpdateType::BranchDeletedUpdate {
1444 name: up.name().to_string(),
1445 previous_snap_id,
1446 })
1447 }
1448 generated::UpdateType::BranchResetUpdate => {
1449 let up = update.update_type_as_branch_reset_update().unwrap();
1450 let previous_snap_id = SnapshotId::new(up.previous_snap_id().0);
1451 Ok(UpdateType::BranchResetUpdate {
1452 name: up.name().to_string(),
1453 previous_snap_id,
1454 })
1455 }
1456 generated::UpdateType::NewCommitUpdate => {
1457 let up = update.update_type_as_new_commit_update().unwrap();
1458 let new_snap_id = SnapshotId::new(up.new_snap_id().0);
1459 Ok(UpdateType::NewCommitUpdate {
1460 branch: up.branch().to_string(),
1461 new_snap_id,
1462 })
1463 }
1464 generated::UpdateType::CommitAmendedUpdate => {
1465 let up = update.update_type_as_commit_amended_update().unwrap();
1466 let previous_snap_id = SnapshotId::new(up.previous_snap_id().0);
1467 let new_snap_id = SnapshotId::new(up.new_snap_id().0);
1468 Ok(UpdateType::CommitAmendedUpdate {
1469 branch: up.branch().to_string(),
1470 previous_snap_id,
1471 new_snap_id,
1472 })
1473 }
1474 generated::UpdateType::NewDetachedSnapshotUpdate => {
1475 let up = update.update_type_as_new_detached_snapshot_update().unwrap();
1476 let new_snap_id = SnapshotId::new(up.new_snap_id().0);
1477 Ok(UpdateType::NewDetachedSnapshotUpdate { new_snap_id })
1478 }
1479 generated::UpdateType::GCRanUpdate => Ok(UpdateType::GCRanUpdate),
1480 generated::UpdateType::ExpirationRanUpdate => {
1481 Ok(UpdateType::ExpirationRanUpdate)
1482 }
1483 generated::UpdateType::FeatureFlagChangedUpdate => {
1484 let up = update.update_type_as_feature_flag_changed_update().unwrap();
1485 Ok(UpdateType::FeatureFlagChanged {
1486 id: up.id(),
1487 new_value: if up.is_set() { Some(up.new_value()) } else { None },
1488 })
1489 }
1490 _ => Err(IcechunkFormatErrorKind::InvalidFlatBuffer(
1491 flatbuffers::InvalidFlatbuffer::InconsistentUnion {
1492 field: Cow::Borrowed("latest_update_type"),
1493 field_type: Cow::Borrowed("UpdateType"),
1494 error_trace: Default::default(),
1495 },
1496 ))
1497 .capture(),
1498 }
1499 }
1500
1501 pub fn ancestry<'a>(
1502 &'a self,
1503 snapshot: &SnapshotId,
1504 ) -> IcechunkResult<impl Iterator<Item = IcechunkResult<SnapshotInfo>> + Send + use<'a>>
1505 {
1506 let root = self.root()?;
1507 if let Some(start) = self.resolve_snapshot_index(snapshot)? {
1508 let mut index = Some(start as i32);
1509 let iter = std::iter::from_fn(move || {
1510 if let Some(ix) = index {
1511 if ix >= 0 {
1512 let snap = root.snapshots().get(ix as usize);
1513 index = Some(snap.parent_offset());
1514 Some(mk_snapshot_info(&root, &snap))
1515 } else {
1516 index = None;
1517 None
1518 }
1519 } else {
1520 None
1521 }
1522 });
1523 Ok(iter)
1524 } else {
1525 Err(IcechunkFormatErrorKind::SnapshotIdNotFound {
1526 snapshot_id: snapshot.clone(),
1527 })
1528 .capture()
1529 }
1530 }
1531
1532 pub fn find_snapshot(&self, id: &SnapshotId) -> IcechunkResult<SnapshotInfo> {
1533 let mut anc = self.ancestry(id)?;
1534 #[expect(clippy::panic)]
1535 match anc.next() {
1536 Some(snap) => snap,
1537 None => panic!("Ancestry head snapshot not found"),
1540 }
1541 }
1542
1543 pub fn repo_before_updates(&self) -> IcechunkResult<Option<&str>> {
1544 Ok(self.root()?.repo_before_updates())
1545 }
1546
1547 fn resolve_snapshot_index(&self, id: &SnapshotId) -> IcechunkResult<Option<usize>> {
1548 Ok(lookup_index_by_key(self.root()?.snapshots(), &id.0, |snap, key| {
1549 snap.id().0.cmp(key)
1550 }))
1551 }
1552}
1553
1554fn resolve_ref_iter<'a>(
1555 sorted_snapshots: &[SnapshotInfo],
1556 it: impl IntoIterator<Item = (&'a str, SnapshotId)>,
1557) -> IcechunkResult<Vec<(&'a str, u32)>> {
1558 let mut res: Vec<_> = it
1559 .into_iter()
1560 .map(|(name, id)| {
1561 let idx = sorted_snapshots
1562 .binary_search_by_key(&&id.0, |snap| &snap.id.0)
1563 .map_err(|_| IcechunkFormatErrorKind::SnapshotIdNotFound {
1564 snapshot_id: id.clone(),
1565 })
1566 .capture()? as u32;
1567 Ok::<_, IcechunkFormatError>((name, idx))
1568 })
1569 .try_collect()?;
1570 res.sort_by(|(name1, _), (name2, _)| name1.cmp(name2));
1571 Ok(res)
1572}
1573
1574fn timestamp_to_timestamp(ts: u64) -> IcechunkResult<DateTime<Utc>> {
1575 let ts: i64 =
1576 ts.try_into().map_err(|_| IcechunkFormatErrorKind::InvalidTimestamp).capture()?;
1577 DateTime::from_timestamp_micros(ts)
1578 .ok_or(IcechunkFormatErrorKind::InvalidTimestamp)
1579 .capture()
1580}
1581
1582fn mk_snapshot_info(
1583 repo: &generated::Repo<'_>,
1584 snap: &generated::SnapshotInfo<'_>,
1585) -> IcechunkResult<SnapshotInfo> {
1586 let flushed_at = timestamp_to_timestamp(snap.flushed_at())?;
1587 let parent_id = if snap.parent_offset() >= 0 {
1588 let parent = repo.snapshots().get(snap.parent_offset() as usize).id();
1589 Some(parent)
1590 } else {
1591 None
1592 };
1593 let metadata = snap
1594 .metadata()
1595 .map(|items| {
1596 let items = items
1597 .iter()
1598 .map(|item| {
1599 let name = item.name().to_string();
1600 let value = flexbuffers::from_slice(item.value().bytes())
1601 .map_err(Box::new)
1602 .capture()?;
1603 Ok::<_, IcechunkFormatError>((name, value))
1604 })
1605 .try_collect()?;
1606 Ok::<_, IcechunkFormatError>(items)
1607 })
1608 .transpose()?
1609 .unwrap_or_default();
1610
1611 Ok(SnapshotInfo {
1612 id: SnapshotId::new(snap.id().0),
1613 flushed_at,
1614 message: snap.message().to_string(),
1615 metadata,
1616 parent_id: parent_id.map(|buf| SnapshotId::new(buf.0)),
1617 })
1618}
1619
1620fn update_type_to_fb<'bldr>(
1621 builder: &mut flatbuffers::FlatBufferBuilder<'bldr>,
1622 update: &UpdateType,
1623) -> IcechunkResult<(generated::UpdateType, WIPOffset<UnionWIPOffset>)> {
1624 match update {
1625 UpdateType::RepoInitializedUpdate => Ok((
1626 generated::UpdateType::RepoInitializedUpdate,
1627 generated::RepoInitializedUpdate::create(
1628 builder,
1629 &generated::RepoInitializedUpdateArgs {},
1630 )
1631 .as_union_value(),
1632 )),
1633 UpdateType::RepoMigratedUpdate { from_version, to_version } => Ok((
1634 generated::UpdateType::RepoMigratedUpdate,
1635 generated::RepoMigratedUpdate::create(
1636 builder,
1637 &generated::RepoMigratedUpdateArgs {
1638 from_version: *from_version as u8,
1639 to_version: *to_version as u8,
1640 },
1641 )
1642 .as_union_value(),
1643 )),
1644 UpdateType::RepoStatusChangedUpdate { status } => {
1645 let limited_availability_reason = status
1646 .limited_availability_reason
1647 .as_ref()
1648 .map(|r| builder.create_string(r));
1649 let status = generated::RepoStatus::create(
1650 builder,
1651 &generated::RepoStatusArgs {
1652 availability: status.availability.into(),
1653 set_at: status.set_at.timestamp_micros() as u64,
1654 limited_availability_reason,
1655 },
1656 );
1657
1658 Ok((
1659 generated::UpdateType::RepoStatusChangedUpdate,
1660 generated::RepoStatusChangedUpdate::create(
1661 builder,
1662 &generated::RepoStatusChangedUpdateArgs { status: Some(status) },
1663 )
1664 .as_union_value(),
1665 ))
1666 }
1667 UpdateType::ConfigChangedUpdate => Ok((
1668 generated::UpdateType::ConfigChangedUpdate,
1669 generated::ConfigChangedUpdate::create(
1670 builder,
1671 &generated::ConfigChangedUpdateArgs {},
1672 )
1673 .as_union_value(),
1674 )),
1675 UpdateType::MetadataChangedUpdate => Ok((
1676 generated::UpdateType::MetadataChangedUpdate,
1677 generated::MetadataChangedUpdate::create(
1678 builder,
1679 &generated::MetadataChangedUpdateArgs {},
1680 )
1681 .as_union_value(),
1682 )),
1683 UpdateType::TagCreatedUpdate { name } => {
1684 let name = Some(builder.create_string(name));
1685 Ok((
1686 generated::UpdateType::TagCreatedUpdate,
1687 generated::TagCreatedUpdate::create(
1688 builder,
1689 &generated::TagCreatedUpdateArgs { name },
1690 )
1691 .as_union_value(),
1692 ))
1693 }
1694 UpdateType::TagDeletedUpdate { name, previous_snap_id } => {
1695 let name = Some(builder.create_string(name));
1696 let object_id12 = generated::ObjectId12::new(&previous_snap_id.0);
1697 let previous_snap_id = Some(&object_id12);
1698 Ok((
1699 generated::UpdateType::TagDeletedUpdate,
1700 generated::TagDeletedUpdate::create(
1701 builder,
1702 &generated::TagDeletedUpdateArgs { name, previous_snap_id },
1703 )
1704 .as_union_value(),
1705 ))
1706 }
1707 UpdateType::BranchCreatedUpdate { name } => {
1708 let name = Some(builder.create_string(name));
1709 Ok((
1710 generated::UpdateType::BranchCreatedUpdate,
1711 generated::BranchCreatedUpdate::create(
1712 builder,
1713 &generated::BranchCreatedUpdateArgs { name },
1714 )
1715 .as_union_value(),
1716 ))
1717 }
1718 UpdateType::BranchDeletedUpdate { name, previous_snap_id } => {
1719 let name = Some(builder.create_string(name));
1720 let object_id12 = generated::ObjectId12::new(&previous_snap_id.0);
1721 let previous_snap_id = Some(&object_id12);
1722 Ok((
1723 generated::UpdateType::BranchDeletedUpdate,
1724 generated::BranchDeletedUpdate::create(
1725 builder,
1726 &generated::BranchDeletedUpdateArgs { name, previous_snap_id },
1727 )
1728 .as_union_value(),
1729 ))
1730 }
1731 UpdateType::BranchResetUpdate { name, previous_snap_id } => {
1732 let name = Some(builder.create_string(name));
1733 let object_id12 = generated::ObjectId12::new(&previous_snap_id.0);
1734 let previous_snap_id = Some(&object_id12);
1735 Ok((
1736 generated::UpdateType::BranchResetUpdate,
1737 generated::BranchResetUpdate::create(
1738 builder,
1739 &generated::BranchResetUpdateArgs { name, previous_snap_id },
1740 )
1741 .as_union_value(),
1742 ))
1743 }
1744 UpdateType::NewCommitUpdate { branch, new_snap_id } => {
1745 let branch = Some(builder.create_string(branch));
1746 let object_id12 = generated::ObjectId12::new(&new_snap_id.0);
1747 let new_snap_id = Some(&object_id12);
1748 Ok((
1749 generated::UpdateType::NewCommitUpdate,
1750 generated::NewCommitUpdate::create(
1751 builder,
1752 &generated::NewCommitUpdateArgs { branch, new_snap_id },
1753 )
1754 .as_union_value(),
1755 ))
1756 }
1757 UpdateType::CommitAmendedUpdate { branch, previous_snap_id, new_snap_id } => {
1758 let branch = Some(builder.create_string(branch));
1759 let object_id12 = generated::ObjectId12::new(&previous_snap_id.0);
1760 let previous_snap_id = Some(&object_id12);
1761 let object_id12 = generated::ObjectId12::new(&new_snap_id.0);
1762 let new_snap_id = Some(&object_id12);
1763 Ok((
1764 generated::UpdateType::CommitAmendedUpdate,
1765 generated::CommitAmendedUpdate::create(
1766 builder,
1767 &generated::CommitAmendedUpdateArgs {
1768 branch,
1769 previous_snap_id,
1770 new_snap_id,
1771 },
1772 )
1773 .as_union_value(),
1774 ))
1775 }
1776 UpdateType::NewDetachedSnapshotUpdate { new_snap_id } => {
1777 let object_id12 = generated::ObjectId12::new(&new_snap_id.0);
1778 let new_snap_id = Some(&object_id12);
1779 Ok((
1780 generated::UpdateType::NewDetachedSnapshotUpdate,
1781 generated::NewDetachedSnapshotUpdate::create(
1782 builder,
1783 &generated::NewDetachedSnapshotUpdateArgs { new_snap_id },
1784 )
1785 .as_union_value(),
1786 ))
1787 }
1788 UpdateType::GCRanUpdate => Ok((
1789 generated::UpdateType::GCRanUpdate,
1790 generated::GCRanUpdate::create(builder, &generated::GCRanUpdateArgs {})
1791 .as_union_value(),
1792 )),
1793 UpdateType::ExpirationRanUpdate => Ok((
1794 generated::UpdateType::ExpirationRanUpdate,
1795 generated::ExpirationRanUpdate::create(
1796 builder,
1797 &generated::ExpirationRanUpdateArgs {},
1798 )
1799 .as_union_value(),
1800 )),
1801 UpdateType::FeatureFlagChanged { id, new_value } => Ok((
1802 generated::UpdateType::FeatureFlagChangedUpdate,
1803 generated::FeatureFlagChangedUpdate::create(
1804 builder,
1805 &generated::FeatureFlagChangedUpdateArgs {
1806 id: *id,
1807 new_value: new_value.unwrap_or_default(),
1808 is_set: new_value.is_some(),
1809 },
1810 )
1811 .as_union_value(),
1812 )),
1813 }
1814}
1815
1816#[cfg(test)]
1817mod tests {
1818
1819 use super::*;
1820 use crate::roundtrip_serialization_tests;
1821 use proptest::prelude::*;
1822 use std::collections::HashSet;
1823
1824 fn potentially_invalid_repo_info() -> impl Strategy<Value = RepoInfo> {
1826 any::<Vec<u8>>().prop_map(|buffer| RepoInfo { buffer })
1827 }
1828
1829 roundtrip_serialization_tests!(
1830 serialize_and_deserialize_repo_info - potentially_invalid_repo_info
1831 );
1832
1833 #[test]
1834 fn test_add_snapshot() -> Result<(), Box<dyn std::error::Error>> {
1835 let id1 = SnapshotId::random();
1836 let snap1 = SnapshotInfo {
1837 id: id1.clone(),
1838 parent_id: None,
1839 flushed_at: DateTime::from_timestamp_micros(1_000_000).unwrap(),
1841 message: "snap 1".to_string(),
1842 metadata: Default::default(),
1843 };
1844 let repo = RepoInfo::initial(
1845 SpecVersionBin::current(),
1846 snap1.clone(),
1847 100,
1848 None::<&()>,
1849 None,
1850 );
1851 assert_eq!(repo.all_snapshots()?.next().unwrap().unwrap(), snap1);
1852
1853 let id2 = SnapshotId::random();
1854 let snap2 = SnapshotInfo {
1855 id: id2.clone(),
1856 parent_id: Some(id1.clone()),
1857 flushed_at: DateTime::from_timestamp_micros(2_000_000).unwrap(),
1858 message: "snap 2".to_string(),
1859 ..snap1.clone()
1860 };
1861 let repo = repo.add_snapshot(
1862 SpecVersionBin::current(),
1863 snap2.clone(),
1864 Some("main"),
1865 UpdateType::NewCommitUpdate {
1866 branch: "main".to_string(),
1867 new_snap_id: snap2.id.clone(),
1868 },
1869 None,
1870 "foo/bar",
1871 100,
1872 )?;
1873 assert_eq!(&repo.resolve_branch("main")?, &snap2.id);
1874 assert_eq!(repo.repo_before_updates()?, None);
1875
1876 let all: HashSet<_> = repo.all_snapshots()?.try_collect()?;
1877 assert_eq!(all, HashSet::from_iter([snap1.clone(), snap2.clone()]));
1878
1879 let anc: Vec<_> = repo.ancestry(&id1)?.try_collect()?;
1880 assert_eq!(anc, std::slice::from_ref(&snap1));
1881
1882 let anc: Vec<_> = repo.ancestry(&id2)?.try_collect()?;
1883 assert_eq!(anc, [snap2.clone(), snap1.clone()]);
1884
1885 assert!(repo.ancestry(&SnapshotId::random()).is_err());
1886
1887 let id3 = SnapshotId::random();
1888 let snap3 = SnapshotInfo {
1889 id: id3.clone(),
1890 parent_id: Some(id2.clone()),
1891 flushed_at: DateTime::from_timestamp_micros(3_000_000).unwrap(),
1892 message: "snap 3".to_string(),
1893 ..snap2.clone()
1894 };
1895 let repo = repo.add_snapshot(
1896 SpecVersionBin::current(),
1897 snap3.clone(),
1898 Some("main"),
1899 UpdateType::NewCommitUpdate {
1900 branch: "main".to_string(),
1901 new_snap_id: snap3.id.clone(),
1902 },
1903 None,
1904 "foo",
1905 100,
1906 )?;
1907 assert_eq!(&repo.resolve_branch("main")?, &snap3.id);
1908 let all: HashSet<_> = repo.all_snapshots()?.try_collect()?;
1909 assert_eq!(
1910 all,
1911 HashSet::from_iter([snap1.clone(), snap2.clone(), snap3.clone()])
1912 );
1913
1914 let all: HashSet<_> = repo.all_snapshots()?.try_collect()?;
1915 assert_eq!(
1916 all,
1917 HashSet::from_iter([snap1.clone(), snap2.clone(), snap3.clone()])
1918 );
1919
1920 let anc: Vec<_> = repo.ancestry(&id3)?.try_collect()?;
1921 assert_eq!(anc, [snap3.clone(), snap2.clone(), snap1.clone()]);
1922 Ok(())
1923 }
1924
1925 #[test]
1926 fn test_tags_and_branches() -> Result<(), Box<dyn std::error::Error>> {
1927 let id1 = SnapshotId::random();
1928 let snap1 = SnapshotInfo {
1929 id: id1.clone(),
1930 parent_id: None,
1931 flushed_at: DateTime::from_timestamp_micros(1_000_000).unwrap(),
1933 message: "snap 1".to_string(),
1934 metadata: Default::default(),
1935 };
1936 let repo = RepoInfo::initial(
1937 SpecVersionBin::current(),
1938 snap1.clone(),
1939 100,
1940 None::<&()>,
1941 None,
1942 );
1943 let repo = repo.add_branch(SpecVersionBin::current(), "foo", &id1, "foo", 100)?;
1944 let repo = repo.add_branch(SpecVersionBin::current(), "bar", &id1, "bar", 100)?;
1945 assert!(matches!(
1946 repo.add_branch(
1947 SpecVersionBin::current(),
1948 "bad-snap",
1949 &SnapshotId::random(),
1950 "bad",
1951 100
1952 ),
1953 Err(IcechunkFormatError {
1954 kind: IcechunkFormatErrorKind::SnapshotIdNotFound { .. },
1955 ..
1956 })
1957 ));
1958 assert!(matches!(
1960 repo.add_branch(SpecVersionBin::current(), "bar", &id1, "/foo/bar", 100),
1961 Err(IcechunkFormatError {
1962 kind: IcechunkFormatErrorKind::BranchAlreadyExists { .. },
1963 ..
1964 })
1965 ));
1966
1967 assert_eq!(
1968 repo.all_branches()?.collect::<HashSet<_>>(),
1969 [("main", 0), ("foo", 0), ("bar", 0)].into()
1970 );
1971
1972 let id2 = SnapshotId::random();
1973 let snap2 = SnapshotInfo {
1974 id: id2.clone(),
1975 parent_id: Some(id1.clone()),
1976 flushed_at: Utc::now(),
1977 message: "snap 2".to_string(),
1978 ..snap1.clone()
1979 };
1980 let repo = repo.add_snapshot(
1981 SpecVersionBin::current(),
1982 snap2.clone(),
1983 Some("main"),
1984 UpdateType::NewCommitUpdate {
1985 branch: "main".to_string(),
1986 new_snap_id: snap2.id.clone(),
1987 },
1988 None,
1989 "foo",
1990 100,
1991 )?;
1992 let repo =
1993 repo.add_branch(SpecVersionBin::current(), "baz", &id2, "/foo/bar", 100)?;
1994 assert_eq!(repo.resolve_branch("main")?, id2.clone());
1995 assert_eq!(repo.resolve_branch("foo")?, id1.clone());
1996 assert_eq!(repo.resolve_branch("bar")?, id1.clone());
1997 assert_eq!(repo.resolve_branch("baz")?, id2.clone());
1998
1999 let repo = repo.delete_branch(SpecVersionBin::current(), "bar", "bar", 100)?;
2000 assert!(repo.resolve_branch("bar").is_err());
2001 assert_eq!(
2002 repo.all_branches()?.map(|(n, _)| n).collect::<HashSet<_>>(),
2003 ["main", "foo", "baz"].into()
2004 );
2005
2006 assert!(
2007 repo.delete_branch(SpecVersionBin::current(), "bad-branch", "bad", 100)
2008 .is_err()
2009 );
2010
2011 let repo = repo.add_tag(SpecVersionBin::current(), "tag1", &id1, "tag1", 100)?;
2013 let repo = repo.add_tag(SpecVersionBin::current(), "tag2", &id2, "tag2", 100)?;
2014 assert!(
2015 repo.add_tag(
2016 SpecVersionBin::current(),
2017 "bad-snap",
2018 &SnapshotId::random(),
2019 "bad",
2020 100
2021 )
2022 .is_err()
2023 );
2024 assert!(
2025 repo.add_tag(SpecVersionBin::current(), "tag1", &id1, "tag1-again", 100)
2026 .is_err()
2027 );
2028 assert_eq!(repo.resolve_tag("tag1")?, id1.clone());
2029 assert_eq!(repo.resolve_tag("tag2")?, id2.clone());
2030 assert_eq!(
2031 repo.all_tags()?.map(|(n, _)| n).collect::<HashSet<_>>(),
2032 ["tag1", "tag2"].into()
2033 );
2034
2035 let repo = repo.add_tag(SpecVersionBin::current(), "tag3", &id1, "tag3", 100)?;
2037 let repo =
2038 repo.delete_tag(SpecVersionBin::current(), "tag3", "delete-tag3", 100)?;
2039 assert_eq!(
2040 repo.all_tags()?.map(|(n, _)| n).collect::<HashSet<_>>(),
2041 ["tag1", "tag2"].into()
2042 );
2043 assert!(
2045 repo.add_tag(SpecVersionBin::current(), "tag3", &id1, "tag3-again", 100)
2046 .is_err()
2047 );
2048 assert!(
2050 repo.delete_tag(SpecVersionBin::current(), "tag3", "delete-tag3-again", 100)
2051 .is_err()
2052 );
2053 assert_eq!(
2054 repo.all_tags()?.map(|(n, _)| n).collect::<HashSet<_>>(),
2055 ["tag1", "tag2"].into()
2056 );
2057 Ok(())
2058 }
2059
2060 #[test]
2061 fn test_repo_info_updates() -> Result<(), Box<dyn std::error::Error>> {
2062 let id1 = SnapshotId::random();
2063 let snap1 = SnapshotInfo {
2064 id: id1.clone(),
2065 parent_id: None,
2066 flushed_at: DateTime::from_timestamp_micros(1_000_000).unwrap(),
2068 message: "snap 1".to_string(),
2069 metadata: Default::default(),
2070 };
2071
2072 let num_updates_per_file: u16 = 10;
2073 let n = num_updates_per_file as usize;
2074
2075 let mut repo = RepoInfo::initial(
2077 SpecVersionBin::current(),
2078 snap1,
2079 num_updates_per_file,
2080 None::<&()>,
2081 None,
2082 );
2083 assert_eq!(repo.latest_updates()?.count(), 1);
2084 let (last_update, _, file) = repo.latest_updates()?.next().unwrap()?;
2085 assert!(file.is_none());
2086 assert_eq!(last_update, UpdateType::RepoInitializedUpdate);
2087 assert!(repo.repo_before_updates()?.is_none());
2088 for i in 1..=(n - 1) {
2091 repo = repo.add_branch(
2092 SpecVersionBin::current(),
2093 i.to_string().as_str(),
2094 &id1,
2095 (i - 1).to_string().as_str(),
2096 num_updates_per_file,
2097 )?;
2098 }
2099
2100 assert_eq!(repo.latest_updates()?.count(), n);
2101 let updates = repo.latest_updates()?;
2102
2103 for (idx, update) in updates.enumerate() {
2105 let (update, _, file) = update?;
2106 if idx == n - 1 {
2107 assert_eq!(update, UpdateType::RepoInitializedUpdate);
2108 assert_eq!(file, Some("0"));
2109 } else {
2110 assert_eq!(
2111 update,
2112 UpdateType::BranchCreatedUpdate { name: (n - 1 - idx).to_string() }
2113 );
2114 if idx == 0 {
2115 assert!(file.is_none());
2116 } else {
2117 assert_eq!(file, Some((n - 1 - idx).to_string().as_str()));
2118 }
2119 }
2120 }
2121 assert!(repo.repo_before_updates()?.is_none());
2122
2123 repo = repo.add_tag(
2125 SpecVersionBin::current(),
2126 "tag",
2127 &id1,
2128 "first-branches",
2129 num_updates_per_file,
2130 )?;
2131 assert_eq!(repo.latest_updates()?.count(), n);
2133 assert_eq!(repo.repo_before_updates()?, Some("0"));
2135 let mut updates = repo.latest_updates()?;
2136 let (last_update, _, file) = updates.next().unwrap()?;
2138 assert_eq!(last_update, UpdateType::TagCreatedUpdate { name: "tag".to_string() });
2139 assert!(file.is_none());
2140
2141 let (last_update, _, file) = updates.next().unwrap()?;
2143 assert_eq!(
2144 last_update,
2145 UpdateType::BranchCreatedUpdate { name: (n - 1).to_string() }
2146 );
2147 assert_eq!(file, Some("first-branches"));
2148
2149 for (idx, update) in updates.enumerate() {
2151 let (update, _, file) = update?;
2152 assert_eq!(file, Some((n - 2 - idx).to_string().as_str()));
2153 assert_eq!(
2154 update,
2155 UpdateType::BranchCreatedUpdate { name: (n - 2 - idx).to_string() }
2156 );
2157 }
2158
2159 Ok(())
2160 }
2161
2162 #[test]
2163 fn test_update_timestamp_ordering_rejected() -> Result<(), Box<dyn std::error::Error>>
2164 {
2165 let flushed_at = DateTime::from_timestamp_micros(1_000_000).unwrap();
2166 let id1 = SnapshotId::random();
2167 let snap1 = SnapshotInfo {
2168 id: id1.clone(),
2169 parent_id: None,
2170 flushed_at,
2171 message: "snap 1".to_string(),
2172 metadata: Default::default(),
2173 };
2174 let repo = RepoInfo::initial(
2175 SpecVersionBin::current(),
2176 snap1,
2177 100,
2178 None::<&()>,
2179 Some(flushed_at),
2180 );
2181
2182 let id2 = SnapshotId::random();
2185 let snap2 = SnapshotInfo {
2186 id: id2.clone(),
2187 parent_id: Some(id1.clone()),
2188 flushed_at: DateTime::from_timestamp_micros(1_000_000).unwrap(),
2189 message: "snap 2".to_string(),
2190 metadata: Default::default(),
2191 };
2192 let result = repo.add_snapshot(
2193 SpecVersionBin::current(),
2194 snap2,
2195 Some("main"),
2196 UpdateType::NewCommitUpdate {
2197 branch: "main".to_string(),
2198 new_snap_id: id2.clone(),
2199 },
2200 Some(flushed_at),
2201 "backup",
2202 100,
2203 );
2204 assert!(matches!(
2205 result,
2206 Err(IcechunkFormatError {
2207 kind: IcechunkFormatErrorKind::InvalidUpdateTimestamp { .. },
2208 ..
2209 })
2210 ));
2211
2212 let flushed_at = DateTime::from_timestamp_micros(500_000).unwrap();
2215 let id3 = SnapshotId::random();
2216 let snap3 = SnapshotInfo {
2217 id: id3.clone(),
2218 parent_id: Some(id1.clone()),
2219 flushed_at,
2220 message: "snap 3".to_string(),
2221 metadata: Default::default(),
2222 };
2223 let result = repo.add_snapshot(
2224 SpecVersionBin::current(),
2225 snap3,
2226 Some("main"),
2227 UpdateType::NewCommitUpdate {
2228 branch: "main".to_string(),
2229 new_snap_id: id3.clone(),
2230 },
2231 Some(flushed_at),
2232 "backup",
2233 100,
2234 );
2235 assert!(matches!(
2236 result,
2237 Err(IcechunkFormatError {
2238 kind: IcechunkFormatErrorKind::InvalidUpdateTimestamp { .. },
2239 ..
2240 })
2241 ));
2242
2243 let flushed_at = DateTime::from_timestamp_micros(2_000_000).unwrap();
2245 let id4 = SnapshotId::random();
2246 let snap4 = SnapshotInfo {
2247 id: id4.clone(),
2248 parent_id: Some(id1.clone()),
2249 flushed_at,
2250 message: "snap 4".to_string(),
2251 metadata: Default::default(),
2252 };
2253 let result = repo.add_snapshot(
2254 SpecVersionBin::current(),
2255 snap4,
2256 Some("main"),
2257 UpdateType::NewCommitUpdate {
2258 branch: "main".to_string(),
2259 new_snap_id: id4.clone(),
2260 },
2261 Some(flushed_at),
2262 "backup",
2263 100,
2264 );
2265 assert!(result.is_ok());
2266
2267 Ok(())
2268 }
2269
2270 #[test]
2273 fn test_set_status_silent() -> Result<(), Box<dyn std::error::Error>> {
2274 let id1 = SnapshotId::random();
2275 let snap1 = SnapshotInfo {
2276 id: id1.clone(),
2277 parent_id: None,
2278 flushed_at: DateTime::from_timestamp_micros(1_000_000).unwrap(),
2279 message: "snap 1".to_string(),
2280 metadata: Default::default(),
2281 };
2282
2283 let repo =
2284 RepoInfo::initial(SpecVersionBin::current(), snap1, 100, None::<&()>, None);
2285
2286 let repo =
2288 repo.add_branch(SpecVersionBin::current(), "dev", &id1, "backup0", 100)?;
2289
2290 let updates_before: Vec<_> =
2291 repo.latest_updates()?.collect::<Result<Vec<_>, _>>()?;
2292 let rbu_before = repo.repo_before_updates()?;
2293 let num_updates_before = updates_before.len();
2294
2295 let new_status = RepoStatus {
2297 availability: RepoAvailability::ReadOnly,
2298 set_at: Utc::now(),
2299 limited_availability_reason: Some("maintenance".to_string()),
2300 };
2301 let repo_normal = repo.set_status(
2302 SpecVersionBin::current(),
2303 &new_status,
2304 Some(("backup1", 100)),
2305 )?;
2306 let updates_after_normal: Vec<_> =
2307 repo_normal.latest_updates()?.collect::<Result<Vec<_>, _>>()?;
2308 assert_eq!(updates_after_normal.len(), num_updates_before + 1);
2309 assert!(matches!(
2310 &updates_after_normal[0].0,
2311 UpdateType::RepoStatusChangedUpdate { .. }
2312 ));
2313 let normal_status = repo_normal.status()?;
2314 assert_eq!(normal_status.availability, RepoAvailability::ReadOnly);
2315 assert_eq!(
2316 normal_status.limited_availability_reason.as_deref(),
2317 Some("maintenance")
2318 );
2319
2320 let silent_status = RepoStatus {
2322 availability: RepoAvailability::ReadOnly,
2323 set_at: Utc::now(),
2324 limited_availability_reason: None,
2325 };
2326 let repo_silent =
2327 repo.set_status(SpecVersionBin::current(), &silent_status, None)?;
2328
2329 let updates_after_silent: Vec<_> =
2330 repo_silent.latest_updates()?.collect::<Result<Vec<_>, _>>()?;
2331 assert_eq!(updates_after_silent.len(), num_updates_before);
2333 for (before, after) in updates_before.iter().zip(updates_after_silent.iter()) {
2334 assert_eq!(before.0, after.0);
2335 assert_eq!(before.1, after.1);
2336 assert_eq!(before.2, after.2);
2337 }
2338 assert_eq!(repo_silent.repo_before_updates()?, rbu_before);
2339
2340 let silent_result_status = repo_silent.status()?;
2342 assert_eq!(silent_result_status.availability, RepoAvailability::ReadOnly);
2343 assert!(silent_result_status.limited_availability_reason.is_none());
2344
2345 Ok(())
2346 }
2347}