feat: add `PreservedCatalog.exists`
@@ -250,6 +250,17 @@ impl<S> PreservedCatalog<S>
where
S: CatalogState,
{
+ /// Checks if a preserved catalog exists.
+ pub async fn exists(
+ object_store: &ObjectStore,
+ server_id: ServerId,
+ db_name: &str,
+ ) -> Result<bool> {
+ Ok(!list_transaction_files(object_store, server_id, db_name)
+ .await?
+ .is_empty())
+ }
+
/// Create new catalog w/o any data.
///
/// An empty transaction will be used to mark the catalog start so that concurrent open but still-empty catalogs can
@@ -288,18 +299,11 @@ where
state_data: S::EmptyInput,
) -> Result<Option<Self>> {
// parse all paths into revisions
- let list_path = transactions_path(&object_store, server_id, &db_name);
- let paths = object_store
- .list(Some(&list_path))
- .await
- .context(Read {})?
- .try_concat()
- .await
- .context(Read {})?;
let mut transactions: HashMap<u64, Uuid> = HashMap::new();
let mut max_revision = None;
- for path in paths {
- if let Some((revision_counter, uuid)) = parse_transaction_path(path) {
+ for (_path, revision_counter, uuid) in
+ list_transaction_files(&object_store, server_id, &db_name).await?
+ {
// keep track of the max
max_revision = Some(
max_revision
@@ -329,7 +333,6 @@ where
}
}
}
- }
// Check if there is any catalog stored at all
if transactions.is_empty() {
@@ -478,6 +481,34 @@ fn parse_transaction_path(path: Path) -> Option<(u64, Uuid)> {
}
}
+/// Load a list of all transaction file from object store. Also parse revision counter and transaction UUID.
+///
+/// The files are in no particular order!
+async fn list_transaction_files(
+ object_store: &ObjectStore,
+ server_id: ServerId,
+ db_name: &str,
+) -> Result<Vec<(Path, u64, Uuid)>> {
+ let list_path = transactions_path(&object_store, server_id, &db_name);
+ let paths = object_store
+ .list(Some(&list_path))
+ .await
+ .context(Read {})?
+ .map_ok(|paths| {
+ paths
+ .into_iter()
+ .filter_map(|path| {
+ parse_transaction_path(path.clone())
+ .map(|(revision_counter, uuid)| (path, revision_counter, uuid))
+ })
+ .collect()
+ })
+ .try_concat()
+ .await
+ .context(Read {})?;
+ Ok(paths)
+}
+
/// Serialize and store protobuf-encoded transaction.
async fn store_transaction_proto(
object_store: &ObjectStore,
@@ -974,6 +1005,11 @@ pub mod tests {
let server_id = make_server_id();
let db_name = "db1";
+ assert!(
+ !PreservedCatalog::<TestCatalogState>::exists(&object_store, server_id, db_name,)
+ .await
+ .unwrap()
+ );
assert!(PreservedCatalog::<TestCatalogState>::load(
Arc::clone(&object_store),
server_id,
@@ -993,6 +1029,11 @@ pub mod tests {
.await
.unwrap();
+ assert!(
+ PreservedCatalog::<TestCatalogState>::exists(&object_store, server_id, db_name,)
+ .await
+ .unwrap()
+ );
assert!(PreservedCatalog::<TestCatalogState>::load(
Arc::clone(&object_store),
server_id,
