diff --git a/rustfs/src/admin/handlers/table_catalog/tests.rs b/rustfs/src/admin/handlers/table_catalog/tests.rs index 7ce53bf22..9ce007d95 100644 --- a/rustfs/src/admin/handlers/table_catalog/tests.rs +++ b/rustfs/src/admin/handlers/table_catalog/tests.rs @@ -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::(&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(); diff --git a/rustfs/src/table_catalog/store/object.rs b/rustfs/src/table_catalog/store/object.rs index 6fdd40385..f62b2b93f 100644 --- a/rustfs/src/table_catalog/store/object.rs +++ b/rustfs/src/table_catalog/store/object.rs @@ -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 { 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> { let mut matched: Option = 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(¤t, prefix_already_checked) + self.reserve_table_warehouse_index(¤t, 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> { - 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, + locked_table_entry: Option<&str>, ) -> TableCatalogStoreResult> { 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::(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::(self.catalog_bucket(), &object).await? + } else { + self.read_entry::(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(_)) => { diff --git a/rustfs/src/table_catalog/tests.rs b/rustfs/src/table_catalog/tests.rs index c32ca56d0..2f61241c3 100644 --- a/rustfs/src/table_catalog/tests.rs +++ b/rustfs/src/table_catalog/tests.rs @@ -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, ¤t, 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::(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, ¤t).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, ¤t).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();