Skip to main content

icechunk_format/
repo_info.rs

1//! Version info, branches, and tags for a repository.
2
3use 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    // Offline is defined in the flatbuffers, but we won't support it
28    // before better specs on how we want to use it
29}
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// TODO: should we not implement serialize and let the session fetch the repo info?
89#[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        // FIXME: add other fields
117        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, // taken from the default
179    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        // TODO: we should check no loops
324        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                // Absent vector when empty, so unexpired snapshots stay
364                // byte-identical to the pre-feature encoding.
365                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        // TODO: provide accessors for last_updated_at, status, metadata, etc.
437        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        // replace the backup path in the last update, that must be None, by the new backup path
472        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        // A backup_path points to the previous repo info file. It is only meaningful
483        // when there are previous updates (you can't back up what doesn't exist).
484        // However, previous updates CAN exist without a backup_path: during migration,
485        // synthetic ops log entries are generated with no prior repo info file to
486        // reference. (This differs from RepoInitializedUpdate, which has neither
487        // previous updates nor a backup path.)
488        assert!(
489            backup_path.is_none() || last_update.is_some(),
490            "A backup path must not be provided without previous updates"
491        );
492
493        // Reject updates whose timestamp is not strictly newer than the top of the ops log
494        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        // If we didn't overflow (all previous updates fit in the new file),
516        // preserve the old file's chain pointer so older history remains reachable.
517        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            // Once we've reached the target file size, look for a valid overflow
525            // point: an entry whose backup_path we can use as the chain pointer.
526            // Entries without a backup_path (e.g. synthetic migration entries)
527            // can't serve as overflow — keep them in the file instead.
528            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        // This method is basically constant, so it's OK to unwrap in it
595        #[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    /// Read the raw config bytes from the `FlatBuffer` (for pass-through in mutations).
629    pub fn config_bytes_raw(&self) -> IcechunkResult<Option<Vec<u8>>> {
630        Ok(self.root()?.config().map(|v| v.bytes().to_vec()))
631    }
632
633    /// Read the repository configuration from the repo info.
634    /// Returns `None` for repos created before config was embedded,
635    /// or for repos using the default configuration.
636    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    /// None means not set, use the default value
684    /// Some(true) means enabled
685    /// Some(false) means disabled
686    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    /// Doesn't check the validity of `flag_id`
725    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>>, // for testing
812        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                // retain preserves order
939                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                // retain preserves order
1111                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    /// Update the embedded configuration and record a `ConfigChangedUpdate` in the op log.
1196    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    /// Set the repository status.
1235    ///
1236    /// When `update_config` is `Some((previous_file, num_updates_per_file))`, a
1237    /// `RepoStatusChangedUpdate` entry is appended to the ops log (normal behavior).
1238    ///
1239    /// When `update_config` is `None`, the status is changed silently: the existing
1240    /// ops log entries are preserved as-is with no new entry added.
1241    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            // It's OK to panic here because ancestry already found the snapshot, and
1559            // it's always the first element of the ancestry
1560            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    // Generates an instance of RepoInfo which may not deserialize to a valid repository
1852    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            // needs to be micro second rounded
1867            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        // an edited snapshot carrying a non-empty list
1974        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        // the list survives serialization verbatim, order preserved
1999        let read_tip = repo.find_snapshot(&tip_id)?;
2000        assert_eq!(read_tip.pruned_ancestor_tx_logs, pruned);
2001
2002        // an unedited snapshot reads back as empty (absent vector)
2003        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            // needs to be micro second rounded
2015            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        // cannot add existing
2043        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        // tags
2096        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        // delete tags
2120        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        // cannot add deleted
2128        assert!(
2129            repo.add_tag(SpecVersionBin::current(), "tag3", &id1, "tag3-again", 100)
2130                .is_err()
2131        );
2132        // cannot delete deleted
2133        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            // needs to be micro second rounded
2151            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        // check updates for a new repo
2161        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        // check updates after num_updates_per_file changes
2174        // fill the first page of updates by adding branches
2175        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        // check all other updates
2189        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        // Now, if we add another change, it won't fit in the first "page" of repo updates
2209        repo = repo.add_tag(
2210            SpecVersionBin::current(),
2211            "tag",
2212            &id1,
2213            "first-branches",
2214            num_updates_per_file,
2215        )?;
2216        // the file only contains the first "page" worth of updates
2217        assert_eq!(repo.latest_updates()?.count(), n);
2218        // next file is the oldest change
2219        assert_eq!(repo.repo_before_updates()?, Some("0"));
2220        let mut updates = repo.latest_updates()?;
2221        // last change is the tag creation
2222        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        // next change is a branch creation backed up to first-branches
2227        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        // all other changes are branch creation (repo creation is in the next page)
2235        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        // Attempting add_snapshot with a timestamp equal to the top of the ops log
2269        // should fail
2270        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        // Attempting add_snapshot with a timestamp older than the top of the ops log
2300        // should also fail
2301        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        // Attempting add_snapshot with a strictly newer timestamp should succeed
2332        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    /// Verify that `set_status` with `update_config = None` (silent mode)
2360    /// changes the repo status without appending an entry to the ops log.
2361    #[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        // Add a branch to have more than one ops log entry
2377        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        // Normal set_status: should add a RepoStatusChangedUpdate entry
2386        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        // Silent set_status: should NOT add an ops log entry
2411        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        // Ops log unchanged: same number of entries, same content
2422        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        // But the status itself IS changed
2431        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}