pub async fn load<S>(
        object_store: Arc<ObjectStore>,
        server_id: ServerId,
        db_name: String,
        state_data: S::EmptyInput,
    ) -> Result<Option<(Self, S)>>
    where
        S: CatalogState + Send + Sync,
    {
        // parse all paths into revisions
        let mut transactions: HashMap<u64, Uuid> = HashMap::new();
        let mut max_revision = None;
        let mut last_checkpoint = None;
        for (_path, file_type, revision_counter, uuid) in
            list_files(&object_store, server_id, &db_name).await?
        {
            // keep track of the max
            max_revision = Some(
                max_revision
                    .map(|m: u64| m.max(revision_counter))
                    .unwrap_or(revision_counter),
            );

            // keep track of latest checkpoint
            if matches!(file_type, FileType::Checkpoint) {
                last_checkpoint = Some(
                    last_checkpoint
                        .map(|m: u64| m.max(revision_counter))
                        .unwrap_or(revision_counter),
                );
            }

            // insert but check for duplicates
            match transactions.entry(revision_counter) {
                Occupied(o) => {
                    // sort for determinism
                    let (uuid1, uuid2) = if *o.get() < uuid {
                        (*o.get(), uuid)
                    } else {
                        (uuid, *o.get())
                    };

                    if uuid1 != uuid2 {
                        Fork {
                            revision_counter,
                            uuid1,
                            uuid2,
                        }
                        .fail()?;
                    }
                }
                Vacant(v) => {
                    v.insert(uuid);
                }
            }
        }

        // Check if there is any catalog stored at all
        if transactions.is_empty() {
            return Ok(None);
        }

        // setup empty state
        let mut state = S::new_empty(&db_name, state_data);
        let mut last_tkey = None;

        // detect replay start
        let start_revision = last_checkpoint.unwrap_or(0);

        // detect end of replay process
        let max_revision = max_revision.expect("transactions list is not empty here");

        // read and replay delta revisions
        for rev in start_revision..=max_revision {
            let uuid = transactions.get(&rev).context(MissingTransaction {
                revision_counter: rev,
            })?;
            let tkey = TransactionKey {
                revision_counter: rev,
                uuid: *uuid,
            };
            let file_type = if Some(rev) == last_checkpoint {
                FileType::Checkpoint
            } else {
                FileType::Transaction
            };
            OpenTransaction::load_and_apply(
                &object_store,
                server_id,
                &db_name,
                &tkey,
                &mut state,
                &last_tkey,
                file_type,
            )
            .await?;
            last_tkey = Some(tkey);
        }

        Ok(Some((
            Self {
                previous_tkey: RwLock::new(last_tkey),
                transaction_semaphore: Semaphore::new(1),
                object_store,
                server_id,
                db_name,
            },
            state,
        )))
    }
/// Load existing catalog from store, if it exists.
///
/// Loading starts at the latest checkpoint or -- if none exists -- at transaction `0`. Transactions before that
/// point are neither verified nor are they required to exist.
Load existing catalog from store, if it exists.
Loading starts at the latest checkpoint or -- if none exists -- at transaction `0`. Transactions before that
point are neither verified nor are they required to exist.
Load existing catalog from store if it exists . Loading starts at the latest checkpoint or -- if none exists -- at transaction ` 0 ` . Transactions before that point are neither verified nor are they required to exist .
pub async fn load<S>(
        object_store: Arc<ObjectStore>,
        server_id: ServerId,
        db_name: String,
        state_data: S::EmptyInput,
    ) -> Result<Option<(Self, S)>>
    where
        S: CatalogState + Send + Sync,
    {
        // parse all paths into revisions
        let mut transactions: HashMap<u64, Uuid> = HashMap::new();
        let mut max_revision = None;
        let mut last_checkpoint = None;
        for (_path, file_type, revision_counter, uuid) in
            list_files(&object_store, server_id, &db_name).await?
        {
            // keep track of the max
            max_revision = Some(
                max_revision
                    .map(|m: u64| m.max(revision_counter))
                    .unwrap_or(revision_counter),
            );
            // keep track of latest checkpoint
            if matches!(file_type, FileType::Checkpoint) {
                last_checkpoint = Some(
                    last_checkpoint
                        .map(|m: u64| m.max(revision_counter))
                        .unwrap_or(revision_counter),
                );
            }
            // insert but check for duplicates
            match transactions.entry(revision_counter) {
                Occupied(o) => {
                    // sort for determinism
                    let (uuid1, uuid2) = if *o.get() < uuid {
                        (*o.get(), uuid)
                    } else {
                        (uuid, *o.get())
                    };
                    if uuid1 != uuid2 {
                        Fork {
                            revision_counter,
                            uuid1,
                            uuid2,
                        }
                        .fail()?;
                    }
                }
                Vacant(v) => {
                    v.insert(uuid);
                }
            }
        }
        // Check if there is any catalog stored at all
        if transactions.is_empty() {
            return Ok(None);
        }
        // setup empty state
        let mut state = S::new_empty(&db_name, state_data);
        let mut last_tkey = None;
        // detect replay start
        let start_revision = last_checkpoint.unwrap_or(0);
        // detect end of replay process
        let max_revision = max_revision.expect("transactions list is not empty here");
        // read and replay delta revisions
        for rev in start_revision..=max_revision {
            let uuid = transactions.get(&rev).context(MissingTransaction {
                revision_counter: rev,
            })?;
            let tkey = TransactionKey {
                revision_counter: rev,
                uuid: *uuid,
            };
            let file_type = if Some(rev) == last_checkpoint {
                FileType::Checkpoint
            } else {
                FileType::Transaction
            };
            OpenTransaction::load_and_apply(
                &object_store,
                server_id,
                &db_name,
                &tkey,
                &mut state,
                &last_tkey,
                file_type,
            )
            .await?;
            last_tkey = Some(tkey);
        }
        Ok(Some((
            Self {
                previous_tkey: RwLock::new(last_tkey),
                transaction_semaphore: Semaphore::new(1),
                object_store,
                server_id,
                db_name,
            },
            state,
        )))
    }
pub async fn load < S > ( object_store : Arc < ObjectStore > , server_id : ServerId , db_name : String , state_data : S :: EmptyInput , ) -> Result < Option < ( Self , S ) > > where S : CatalogState + Send + Sync , { // parse all paths into revisions let mut transactions : HashMap < u64 , Uuid > = HashMap :: new ( ) ; let mut max_revision = None ; let mut last_checkpoint = None ; for ( _path , file_type , revision_counter , uuid ) in list_files ( & object_store , server_id , & db_name ) . await ? { // keep track of the max max_revision = Some ( max_revision . map ( | m : u64 | m . max ( revision_counter ) ) . unwrap_or ( revision_counter ) , ) ; // keep track of latest checkpoint if matches ! ( file_type FileType Checkpoint ) { last_checkpoint = Some ( last_checkpoint . map ( | m : u64 | m . max ( revision_counter ) ) . unwrap_or ( revision_counter ) , ) ; } // insert but check for duplicates match transactions . entry ( revision_counter ) { Occupied ( o ) => { // sort for determinism let ( uuid1 , uuid2 ) = if * o . get ( ) < uuid { ( * o . get ( ) , uuid ) } else { ( uuid , * o . get ( ) ) } ; if uuid1 != uuid2 { Fork { revision_counter , uuid1 , uuid2 , } . fail ( ) ? ; } } Vacant ( v ) => { v . insert ( uuid ) ; } } } // Check if there is any catalog stored at all if transactions . is_empty ( ) { return Ok ( None ) ; } // setup empty state let mut state = S :: new_empty ( & db_name , state_data ) ; let mut last_tkey = None ; // detect replay start let start_revision = last_checkpoint . unwrap_or ( 0 ) ; // detect end of replay process let max_revision = max_revision . expect ( " " ) ; // read and replay delta revisions for rev in start_revision ..= max_revision { let uuid = transactions . get ( & rev ) . context ( MissingTransaction { revision_counter : rev , } ) ? ; let tkey = TransactionKey { revision_counter : rev , uuid : * uuid , } ; let file_type = if Some ( rev ) == last_checkpoint { FileType :: Checkpoint } else { FileType :: Transaction } ; OpenTransaction :: load_and_apply ( & object_store , server_id , & db_name , & tkey , & mut state , & last_tkey , file_type , ) . await ? ; last_tkey = Some ( tkey ) ; } Ok ( Some ( ( Self { previous_tkey : RwLock :: new ( last_tkey ) , transaction_semaphore : Semaphore :: new ( 1 ) , object_store , server_id , db_name , } , state , ) ) ) }
Load existing catalog from store, if it exists.
Load existing catalog from store if it exists .
object_store
Arc<ObjectStore>
server_id
ServerId
db_name
String
state_data
S::EmptyInput
object_store
Arc<ObjectStore>
server_id
ServerId
db_name
String
state_data
S::EmptyInput