fix(table-catalog): reuse held entry locks in warehouse scans (#8418)

This commit is contained in:
Henry Guo
2026-10-08 01:01:56 -07:00
committed by GitHub
parent 76386572f6
commit 3e87004032
3 changed files with 495 additions and 20 deletions
@@ -3277,6 +3277,157 @@ async fn create_view_holds_publication_fences_from_metadata_write_through_regist
);
}
#[tokio::test]
async fn object_create_table_response_recreates_renamed_source_without_relocking_or_crossing_identities() {
for destination in ["analytics", "curated"] {
let backend = TestTableCatalogObjectBackend {
reject_reads_while_write_locked: true,
..TestTableCatalogObjectBackend::content_addressed()
};
let store = crate::table_catalog::ObjectTableCatalogStore::new(backend.clone());
let namespace = crate::table_catalog::Namespace::parse("analytics").unwrap();
create_standard_events_table(&store, &backend, &namespace).await;
let first = store.load_table("warehouse", "analytics", "events").await.unwrap().unwrap();
let initial_metadata = backend
.read_object("warehouse", &first.metadata_location)
.await
.unwrap()
.unwrap();
let commit_id = "11111111-1111-4111-8111-111111111111";
let first_commit = standard_commit_table_response(
&store,
&trusted_table_commit_backend(&backend),
"warehouse",
&namespace,
"events",
standard_property_commit_request(commit_id, &first.table_uuid, "original"),
)
.await
.expect("original table should commit before rename");
let first_commit_object = table_metadata_location_for_catalog("warehouse", &first_commit.metadata_location).unwrap();
let committed_metadata = backend.read_object("warehouse", &first_commit_object).await.unwrap().unwrap();
if destination != "analytics" {
create_namespace_response(
&store,
"warehouse",
CreateNamespaceRequest {
namespace: vec![destination.to_string()],
properties: BTreeMap::new(),
},
true,
)
.await
.unwrap();
}
store
.rename_table("warehouse", "analytics", "events", destination, "events_v2")
.await
.unwrap();
let restarted = crate::table_catalog::ObjectTableCatalogStore::new(backend.clone());
let renamed = restarted
.load_table("warehouse", destination, "events_v2")
.await
.unwrap()
.unwrap();
let original_receipt = restarted
.get_commit_by_id("warehouse", &renamed.table_id, commit_id)
.await
.unwrap()
.unwrap();
let request: CreateTableRequest = serde_json::from_value(serde_json::json!({
"name": "events",
"schema": {"type": "struct", "schema-id": 0, "fields": [
{"id": 1, "name": "id", "required": true, "type": "long"}
]}
}))
.unwrap();
let response = create_table_response(
&restarted,
&TableCommitObjectBackend::trusted(backend.clone()),
"warehouse",
&namespace,
request,
true,
)
.await
.expect("source recreation should publish without re-locking its tombstone");
let recreated = restarted
.load_table("warehouse", "analytics", "events")
.await
.unwrap()
.unwrap();
assert_ne!(recreated.table_id, renamed.table_id);
assert_ne!(recreated.table_uuid, renamed.table_uuid);
assert_ne!(recreated.warehouse_location, renamed.warehouse_location);
assert_ne!(recreated.metadata_location, renamed.metadata_location);
assert_ne!(recreated.metadata_location, first.metadata_location);
assert_eq!(response.metadata["table-uuid"], recreated.table_uuid);
assert_eq!(response.metadata["location"], recreated.warehouse_location);
let published = backend
.read_object("warehouse", &recreated.metadata_location)
.await
.unwrap()
.unwrap();
assert_eq!(serde_json::from_slice::<serde_json::Value>(&published.data).unwrap(), response.metadata);
let second_commit = standard_commit_table_response(
&restarted,
&trusted_table_commit_backend(&backend),
"warehouse",
&namespace,
"events",
standard_property_commit_request(commit_id, &recreated.table_uuid, "replacement"),
)
.await
.expect("recreated table should commit with its own scoped receipt and metadata");
let fresh = crate::table_catalog::ObjectTableCatalogStore::new(backend.clone());
let advanced = fresh.load_table("warehouse", "analytics", "events").await.unwrap().unwrap();
assert_eq!(advanced.generation, recreated.generation + 1);
assert_ne!(advanced.version_token, recreated.version_token);
assert_eq!(
table_metadata_location_for_client("warehouse", &advanced.metadata_location),
second_commit.metadata_location
);
assert_eq!(second_commit.metadata["table-uuid"], recreated.table_uuid);
assert_eq!(second_commit.metadata["properties"]["owner"], "replacement");
assert_ne!(second_commit.metadata_location, first_commit.metadata_location);
assert_eq!(
fresh.load_table("warehouse", destination, "events_v2").await.unwrap(),
Some(renamed.clone())
);
assert_eq!(
fresh
.get_commit_by_id("warehouse", &renamed.table_id, commit_id)
.await
.unwrap(),
Some(original_receipt)
);
let replacement_receipt = fresh
.get_commit_by_id("warehouse", &recreated.table_id, commit_id)
.await
.unwrap()
.unwrap();
assert_eq!(replacement_receipt.table_id, recreated.table_id);
assert_eq!(replacement_receipt.new_metadata_location, advanced.metadata_location);
assert_eq!(
backend.read_object("warehouse", &first.metadata_location).await.unwrap(),
Some(initial_metadata)
);
assert_eq!(
backend.read_object("warehouse", &first_commit_object).await.unwrap(),
Some(committed_metadata)
);
drop_table_in_store(&fresh, "warehouse", &namespace, "events").await.unwrap();
let destination_namespace = crate::table_catalog::Namespace::parse(destination).unwrap();
drop_table_in_store(&fresh, "warehouse", &destination_namespace, "events_v2")
.await
.unwrap();
drop_namespace_in_store(&fresh, "warehouse", "analytics").await.unwrap();
if destination != "analytics" {
drop_namespace_in_store(&fresh, "warehouse", destination).await.unwrap();
}
}
}
#[tokio::test]
async fn create_table_response_recreates_dropped_identifier_without_overwriting_retained_metadata() {
let metadata_backend = TestTableCatalogObjectBackend::content_addressed();
+39 -17
View File
@@ -1389,10 +1389,17 @@ where
Ok(true)
}
async fn ensure_table_warehouse_prefix_available(&self, entry: &TableEntry) -> TableCatalogStoreResult<()> {
async fn ensure_table_warehouse_prefix_available(
&self,
entry: &TableEntry,
locked_table_entry: Option<&str>,
) -> TableCatalogStoreResult<()> {
let candidate = table_warehouse_index_entry(entry)?;
validate_table_entry_version_and_id(entry)?;
for existing in self.list_all_table_entries(&candidate.table_bucket).await? {
for existing in self
.list_all_table_entries_with_limit(&candidate.table_bucket, None, locked_table_entry)
.await?
{
if existing.state != TableCatalogEntryState::Active {
continue;
}
@@ -1447,6 +1454,7 @@ where
&self,
entry: &TableEntry,
prefix_already_checked: bool,
locked_table_entry: &str,
) -> TableCatalogStoreResult<WarehouseIndexReservation> {
let index = table_warehouse_index_entry(entry)?;
let object = self
@@ -1463,7 +1471,8 @@ where
}
// Registration already checked the prefix while holding the bucket publication fence.
if !prefix_already_checked {
self.ensure_table_warehouse_prefix_available(entry).await?;
self.ensure_table_warehouse_prefix_available(entry, Some(locked_table_entry))
.await?;
}
loop {
match self
@@ -1704,8 +1713,13 @@ where
.await
}
async fn restore_table_warehouse_index_after_failed_drop(&self, entry: &TableEntry, reason: &'static str) {
if let Err(err) = self.reserve_table_warehouse_index(entry, false).await {
async fn restore_table_warehouse_index_after_failed_drop(
&self,
entry: &TableEntry,
locked_table_entry: &str,
reason: &'static str,
) {
if let Err(err) = self.reserve_table_warehouse_index(entry, false, locked_table_entry).await {
tracing::warn!(
table_bucket = %entry.table_bucket,
namespace = %entry.namespace,
@@ -1874,7 +1888,7 @@ where
) -> TableCatalogStoreResult<Option<TableDataPlaneResource>> {
let mut matched: Option<TableDataPlaneResource> = None;
for table in self
.list_all_table_entries_with_limit(table_bucket, Some(TABLE_DATA_PLANE_INDEX_MISS_SCAN_MAX_CATALOG_OBJECTS))
.list_all_table_entries_with_limit(table_bucket, Some(TABLE_DATA_PLANE_INDEX_MISS_SCAN_MAX_CATALOG_OBJECTS), None)
.await?
{
if table.state != TableCatalogEntryState::Active {
@@ -1933,7 +1947,7 @@ where
if current.state != TableCatalogEntryState::Active {
return Ok(());
}
self.reserve_table_warehouse_index(&current, prefix_already_checked)
self.reserve_table_warehouse_index(&current, prefix_already_checked, &table_path)
.await
.map(|_| ())
}
@@ -1966,7 +1980,7 @@ where
return Ok(());
}
let tables = self
.list_all_table_entries_with_limit(table_bucket, max_catalog_objects)
.list_all_table_entries_with_limit(table_bucket, max_catalog_objects, None)
.await?
.into_iter()
.filter(|table| table.state == TableCatalogEntryState::Active)
@@ -2047,7 +2061,7 @@ where
}
async fn list_all_table_entries(&self, table_bucket: &str) -> TableCatalogStoreResult<Vec<TableEntry>> {
self.list_all_table_entries_with_limit(table_bucket, None).await
self.list_all_table_entries_with_limit(table_bucket, None, None).await
}
async fn list_table_entry_objects_for_data_plane_scan(
@@ -2095,6 +2109,7 @@ where
&self,
table_bucket: &str,
max_catalog_objects: Option<usize>,
locked_table_entry: Option<&str>,
) -> TableCatalogStoreResult<Vec<TableEntry>> {
let mut entries = Vec::new();
let table_objects = match max_catalog_objects {
@@ -2111,7 +2126,13 @@ where
.collect(),
};
for object in table_objects {
let Some((entry, _)) = self.read_entry::<TableEntry>(self.catalog_bucket(), &object).await? else {
// Only the exact entry protected by the caller's write lock can bypass a read lock.
let entry = if locked_table_entry == Some(object.as_str()) {
self.read_entry_unlocked::<TableEntry>(self.catalog_bucket(), &object).await?
} else {
self.read_entry::<TableEntry>(self.catalog_bucket(), &object).await?
};
let Some((entry, _)) = entry else {
continue;
};
validate_table_entry_object(&self.paths, &object, &entry)?;
@@ -2275,8 +2296,9 @@ where
"table registration requires a table publication fence".to_string(),
));
}
self.ensure_table_warehouse_prefix_available(&entry).await?;
let reservation = self.reserve_table_warehouse_index(&entry, true).await?;
self.ensure_table_warehouse_prefix_available(&entry, Some(&table_path))
.await?;
let reservation = self.reserve_table_warehouse_index(&entry, true, &table_path).await?;
if !publication.holds_table_bucket(&entry.table_bucket)
|| !publication.holds_table(&entry.table_bucket, &entry.namespace, &entry.table)
{
@@ -5314,7 +5336,7 @@ where
}
async fn ensure_table_warehouse_location_available(&self, candidate: &TableEntry) -> TableCatalogStoreResult<()> {
self.ensure_table_warehouse_prefix_available(candidate).await
self.ensure_table_warehouse_prefix_available(candidate, None).await
}
async fn list_tables_page(
@@ -5940,9 +5962,9 @@ where
next.version_token = staged_commit_log.new_version_token.clone();
next.generation = current.generation.saturating_add(1);
if next.warehouse_location != current.warehouse_location {
self.ensure_table_warehouse_prefix_available(&next).await?;
self.ensure_table_warehouse_prefix_available(&next, Some(&table_path)).await?;
}
let reservation = self.reserve_table_warehouse_index(&next, false).await?;
let reservation = self.reserve_table_warehouse_index(&next, false, &table_path).await?;
let staged_write_result = async {
if !has_existing_commit {
@@ -6082,7 +6104,7 @@ where
if !publication.holds_table_bucket(table_bucket)
|| !publication.holds_table(table_bucket, &namespace.public_name(), table.as_str())
{
self.restore_table_warehouse_index_after_failed_drop(&entry, "table publication fence lost")
self.restore_table_warehouse_index_after_failed_drop(&entry, &object, "table publication fence lost")
.await;
return Err(TableCatalogStoreError::Internal(
"table drop publication fence was lost before catalog update".to_string(),
@@ -6092,7 +6114,7 @@ where
match self.read_table_with_etag_unlocked(table_bucket, &namespace, &table).await {
Ok(None) => return Ok(()),
Ok(Some((current, _))) if current == entry => {
self.restore_table_warehouse_index_after_failed_drop(&entry, "table entry delete failed")
self.restore_table_warehouse_index_after_failed_drop(&entry, &object, "table entry delete failed")
.await;
}
Ok(Some(_)) => {
+305 -3
View File
@@ -6216,7 +6216,10 @@ async fn table_data_plane_resource_falls_back_to_scan_without_index_state() {
#[tokio::test]
async fn object_table_catalog_store_backfill_skips_table_deleted_after_listing() {
let backend = TestCatalogObjectBackend::default();
let backend = TestCatalogObjectBackend {
reject_reads_while_write_locked: true,
..Default::default()
};
let store = ObjectTableCatalogStore::new(backend.clone());
let bucket = "analytics";
let namespace = Namespace::parse("sales").expect("namespace should parse");
@@ -6251,6 +6254,15 @@ async fn object_table_catalog_store_backfill_skips_table_deleted_after_listing()
.await
.expect("table entry should be seeded without an index");
store
.backfill_active_table_warehouse_index(bucket, &namespace.public_name(), table.as_str())
.await
.expect("active backfill should reuse its held table entry lock");
backend
.delete_object(RUSTFS_META_BUCKET, &index_path)
.await
.expect("backfilled index should be removed before testing a stale listing");
let listed = store
.list_tables(bucket, &namespace.public_name())
.await
@@ -6816,7 +6828,10 @@ async fn active_table_warehouse_prefix_takes_precedence_over_a_deleted_parent()
#[tokio::test]
async fn object_table_catalog_store_restores_index_when_table_entry_delete_fails() {
let backend = TestCatalogObjectBackend::default();
let backend = TestCatalogObjectBackend {
reject_reads_while_write_locked: true,
..Default::default()
};
let store = ObjectTableCatalogStore::new(backend.clone());
let bucket = "analytics";
let namespace = Namespace::parse("sales").expect("namespace should parse");
@@ -17173,7 +17188,10 @@ async fn catalog_backings_reject_metadata_changed_after_validation() {
#[tokio::test]
async fn object_table_catalog_store_syncs_warehouse_location_from_committed_metadata() {
let backend = TestCatalogObjectBackend::default();
let backend = TestCatalogObjectBackend {
reject_reads_while_write_locked: true,
..Default::default()
};
let store = ObjectTableCatalogStore::new(backend.clone());
let bucket = "analytics";
let namespace = Namespace::parse("sales").unwrap();
@@ -20858,6 +20876,290 @@ async fn object_catalog_table_rename_preserves_identity_index_and_reuses_source_
);
}
#[tokio::test]
async fn object_catalog_source_recreation_after_rename_preserves_the_original_identity_and_receipts() {
for destination_namespace in ["sales", "curated"] {
let backend = TestCatalogObjectBackend {
reject_reads_while_write_locked: true,
..Default::default()
};
let store = ObjectTableCatalogStore::new(backend.clone());
let bucket = "analytics";
let source_namespace = Namespace::parse("sales").unwrap();
let source_table = IdentifierSegment::parse("orders").unwrap();
let destination = Namespace::parse(destination_namespace).unwrap();
let destination_table = IdentifierSegment::parse("orders_v2").unwrap();
let current = default_table_metadata_file_path(&source_namespace, &source_table, "00001.metadata.json");
let next = default_table_metadata_file_path(&source_namespace, &source_table, "00002.metadata.json");
seed_table_for_metadata_maintenance(&store, bucket, &source_namespace, &source_table, current.clone()).await;
if destination_namespace != "sales" {
store
.create_namespace(test_namespace_entry(bucket, &destination))
.await
.unwrap();
}
let original = store.load_table(bucket, "sales", "orders").await.unwrap().unwrap();
let mut metadata = super::test_support::table_metadata_json(&original.table_uuid, &original.warehouse_location);
metadata["format-version"] = serde_json::json!(original.format_version);
let metadata_bytes = serde_json::to_vec(&metadata).unwrap();
backend.seed_object(bucket, &current, metadata_bytes.clone()).await;
backend.seed_object(bucket, &next, metadata_bytes).await;
backend
.seed_object(bucket, "tables/table-id/data/retained.bin", b"original table data".to_vec())
.await;
store
.commit_table(TableCommitRequest {
table_bucket: bucket.to_string(),
namespace: source_namespace.public_name(),
table: source_table.as_str().to_string(),
commit_id: "source-recreate-commit".to_string(),
idempotency_key: Some("source-recreate-request".to_string()),
operation: "update-metadata".to_string(),
expected_version_token: original.version_token,
expected_metadata_location: current.clone(),
new_metadata_location: next.clone(),
requirements: Vec::new(),
writer: Some("source-recreate-test".to_string()),
})
.await
.unwrap();
store
.rename_table(bucket, "sales", "orders", destination_namespace, "orders_v2")
.await
.unwrap();
let restarted = ObjectTableCatalogStore::new(backend.clone());
let renamed = restarted
.load_table(bucket, destination_namespace, "orders_v2")
.await
.unwrap()
.unwrap();
let source_path = restarted.paths.table_entry_path(bucket, &source_namespace, &source_table);
let (tombstone, _) = restarted
.read_entry::<TableEntry>(RUSTFS_META_BUCKET, &source_path)
.await
.unwrap()
.unwrap();
assert_eq!(tombstone.state, TableCatalogEntryState::Deleted);
assert_eq!(tombstone.table_id, renamed.table_id);
let before_commit = restarted
.get_commit_by_id(bucket, &renamed.table_id, "source-recreate-commit")
.await
.unwrap()
.unwrap();
let before_idempotency = restarted
.get_commit_by_idempotency_key(bucket, &renamed.table_id, "source-recreate-request")
.await
.unwrap()
.unwrap();
let renamed_path = restarted.paths.table_entry_path(bucket, &destination, &destination_table);
let index_path = restarted.paths.warehouse_index_entry_path(bucket, "tables/table-id/");
let before_entry = backend.read_object(RUSTFS_META_BUCKET, &renamed_path).await.unwrap().unwrap();
let before_index = backend.read_object(RUSTFS_META_BUCKET, &index_path).await.unwrap().unwrap();
let before_current = backend.read_object(bucket, &current).await.unwrap().unwrap();
let before_next = backend.read_object(bucket, &next).await.unwrap().unwrap();
let before_data = backend
.read_object(bucket, "tables/table-id/data/retained.bin")
.await
.unwrap()
.unwrap();
let replacement_metadata =
default_table_metadata_file_path(&source_namespace, &source_table, "00001-replacement.metadata.json");
let mut replacement = test_table_entry(bucket, &source_namespace, &source_table, replacement_metadata.clone());
replacement.table_id = "replacement-table-id".to_string();
replacement.table_uuid = "replacement-table-uuid".to_string();
replacement.warehouse_location = "s3://analytics/tables/replacement-table-id".to_string();
let replacement_json = super::test_support::table_metadata_json(&replacement.table_uuid, &replacement.warehouse_location);
backend
.seed_object(bucket, &replacement_metadata, serde_json::to_vec(&replacement_json).unwrap())
.await;
restarted
.create_table(replacement.clone())
.await
.expect("source recreation should not re-lock its retained tombstone");
assert_ne!(replacement.table_id, renamed.table_id);
assert_ne!(replacement.table_uuid, renamed.table_uuid);
assert_ne!(replacement.warehouse_location, renamed.warehouse_location);
assert_ne!(replacement.metadata_location, renamed.metadata_location);
for reader in [restarted, ObjectTableCatalogStore::new(backend.clone())] {
assert_eq!(reader.load_table(bucket, "sales", "orders").await.unwrap(), Some(replacement.clone()));
assert_eq!(
reader.load_table(bucket, destination_namespace, "orders_v2").await.unwrap(),
Some(renamed.clone())
);
assert_eq!(
reader
.get_commit_by_id(bucket, &renamed.table_id, "source-recreate-commit")
.await
.unwrap(),
Some(before_commit.clone())
);
assert_eq!(
reader
.get_commit_by_idempotency_key(bucket, &renamed.table_id, "source-recreate-request")
.await
.unwrap(),
Some(before_idempotency.clone())
);
assert!(
reader
.get_commit_by_id(bucket, &replacement.table_id, "source-recreate-commit")
.await
.unwrap()
.is_none()
);
assert!(
reader
.get_commit_by_idempotency_key(bucket, &replacement.table_id, "source-recreate-request")
.await
.unwrap()
.is_none()
);
let original_resource = reader
.resolve_table_data_plane_resource(bucket, "tables/table-id/data/retained.bin")
.await
.unwrap()
.unwrap();
assert_eq!(original_resource.table_id, renamed.table_id);
assert_eq!(original_resource.namespace, destination_namespace);
assert_eq!(original_resource.table, "orders_v2");
let replacement_resource = reader
.resolve_table_data_plane_resource(bucket, "tables/replacement-table-id/data/file.bin")
.await
.unwrap()
.unwrap();
assert_eq!(replacement_resource.table_id, replacement.table_id);
assert_eq!(replacement_resource.namespace, "sales");
assert_eq!(replacement_resource.table, "orders");
}
assert_eq!(backend.read_object(RUSTFS_META_BUCKET, &renamed_path).await.unwrap(), Some(before_entry));
assert_eq!(backend.read_object(RUSTFS_META_BUCKET, &index_path).await.unwrap(), Some(before_index));
assert_eq!(backend.read_object(bucket, &current).await.unwrap(), Some(before_current));
assert_eq!(backend.read_object(bucket, &next).await.unwrap(), Some(before_next));
assert_eq!(
backend
.read_object(bucket, "tables/table-id/data/retained.bin")
.await
.unwrap(),
Some(before_data)
);
}
}
#[tokio::test]
async fn object_catalog_source_recreation_keeps_identifier_and_warehouse_conflicts_fail_closed() {
for existing_identifier in [true, false] {
let backend = TestCatalogObjectBackend {
reject_reads_while_write_locked: true,
..Default::default()
};
let store = ObjectTableCatalogStore::new(backend.clone());
let bucket = "analytics";
let source_namespace = Namespace::parse("sales").unwrap();
let source_table = IdentifierSegment::parse("orders").unwrap();
let destination_namespace = Namespace::parse("curated").unwrap();
let destination_table = IdentifierSegment::parse("orders_v2").unwrap();
let current = default_table_metadata_file_path(&source_namespace, &source_table, "00001.metadata.json");
seed_table_for_metadata_maintenance(&store, bucket, &source_namespace, &source_table, current).await;
store
.create_namespace(test_namespace_entry(bucket, &destination_namespace))
.await
.unwrap();
store
.rename_table(bucket, "sales", "orders", "curated", "orders_v2")
.await
.unwrap();
let store = ObjectTableCatalogStore::new(backend.clone());
let (namespace, table) = if existing_identifier {
(&destination_namespace, &destination_table)
} else {
(&source_namespace, &source_table)
};
let mut candidate = test_table_entry(
bucket,
namespace,
table,
default_table_metadata_file_path(namespace, table, "00001-candidate.metadata.json"),
);
candidate.table_id = "candidate-table-id".to_string();
candidate.table_uuid = "candidate-table-uuid".to_string();
candidate.warehouse_location = if existing_identifier {
"s3://analytics/tables/candidate-table-id".to_string()
} else {
"s3://analytics/tables/table-id/nested".to_string()
};
let paths = backend.list_objects(RUSTFS_META_BUCKET, "").await.unwrap();
let mut before = Vec::new();
for path in &paths {
before.push((path.clone(), backend.read_object(RUSTFS_META_BUCKET, path).await.unwrap().unwrap()));
}
let error = store
.create_table(candidate)
.await
.expect_err("conflicting registration must fail closed");
assert_matches!(error, TableCatalogStoreError::Conflict(message) if if existing_identifier {
message.contains("already exists")
} else {
message.contains("overlaps an active table")
});
assert_eq!(backend.list_objects(RUSTFS_META_BUCKET, "").await.unwrap(), paths);
for (path, object) in before {
assert_eq!(backend.read_object(RUSTFS_META_BUCKET, &path).await.unwrap(), Some(object));
}
}
}
#[tokio::test]
async fn object_catalog_warehouse_scans_keep_read_locks_for_unheld_table_entries() {
let backend = TestCatalogObjectBackend {
reject_reads_while_write_locked: true,
..Default::default()
};
let store = ObjectTableCatalogStore::new(backend.clone());
let bucket = "analytics";
let namespace = Namespace::parse("sales").unwrap();
let owner_table = IdentifierSegment::parse("orders").unwrap();
let candidate_table = IdentifierSegment::parse("returns").unwrap();
let current = default_table_metadata_file_path(&namespace, &owner_table, "00001.metadata.json");
seed_table_for_metadata_maintenance(&store, bucket, &namespace, &owner_table, current).await;
let owner = store.load_table(bucket, "sales", "orders").await.unwrap().unwrap();
let owner_path = store.paths.table_entry_path(bucket, &namespace, &owner_table);
let before_owner = backend.read_object(RUSTFS_META_BUCKET, &owner_path).await.unwrap().unwrap();
let mut candidate = test_table_entry(
bucket,
&namespace,
&candidate_table,
default_table_metadata_file_path(&namespace, &candidate_table, "00001-candidate.metadata.json"),
);
candidate.table_id = "candidate-table-id".to_string();
candidate.table_uuid = "candidate-table-uuid".to_string();
candidate.warehouse_location = "s3://analytics/tables/candidate-table-id".to_string();
let guard = backend.acquire_write_lock(RUSTFS_META_BUCKET, &owner_path).await.unwrap();
assert_matches!(
store.ensure_table_warehouse_location_available(&candidate).await,
Err(TableCatalogStoreError::Internal(message)) if message.contains("read attempted while its write lock is held") && message.contains(&owner_path)
);
assert_matches!(
store.create_table(candidate.clone()).await,
Err(TableCatalogStoreError::Internal(message)) if message.contains("read attempted while its write lock is held") && message.contains(&owner_path)
);
drop(guard);
assert_eq!(store.load_table(bucket, "sales", "orders").await.unwrap(), Some(owner));
assert!(store.load_table(bucket, "sales", "returns").await.unwrap().is_none());
assert_eq!(backend.read_object(RUSTFS_META_BUCKET, &owner_path).await.unwrap(), Some(before_owner));
store.ensure_table_warehouse_location_available(&candidate).await.unwrap();
store
.create_table(candidate.clone())
.await
.expect("unlocked foreign entries should remain normally readable");
assert_eq!(store.load_table(bucket, "sales", "returns").await.unwrap(), Some(candidate));
}
#[tokio::test]
async fn object_catalog_table_rename_rejects_missing_and_conflicting_destinations() {
let backend = TestCatalogObjectBackend::default();