diff --git a/rustfs/src/table_catalog/store/object.rs b/rustfs/src/table_catalog/store/object.rs index e2c7249ee..0762473d0 100644 --- a/rustfs/src/table_catalog/store/object.rs +++ b/rustfs/src/table_catalog/store/object.rs @@ -1336,7 +1336,11 @@ where Ok(()) } - async fn reserve_table_warehouse_index(&self, entry: &TableEntry) -> TableCatalogStoreResult { + async fn reserve_table_warehouse_index( + &self, + entry: &TableEntry, + prefix_already_checked: bool, + ) -> TableCatalogStoreResult { let index = table_warehouse_index_entry(entry)?; let object = self .paths @@ -1350,7 +1354,10 @@ where return Ok(WarehouseIndexReservation::AlreadyReserved); } } - self.ensure_table_warehouse_prefix_available(entry).await?; + // Registration already checked the prefix while holding the bucket publication fence. + if !prefix_already_checked { + self.ensure_table_warehouse_prefix_available(entry).await?; + } loop { match self .write_entry(self.catalog_bucket(), &object, &index, TableCatalogPutPrecondition::IfAbsent) @@ -1531,7 +1538,7 @@ where } 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).await { + if let Err(err) = self.reserve_table_warehouse_index(entry, false).await { tracing::warn!( table_bucket = %entry.table_bucket, namespace = %entry.namespace, @@ -1618,7 +1625,7 @@ where if current.state != TableCatalogEntryState::Active { return Ok(()); } - self.reserve_table_warehouse_index(¤t).await.map(|_| ()) + self.reserve_table_warehouse_index(¤t, false).await.map(|_| ()) } pub(in crate::table_catalog) async fn backfill_table_warehouse_index( @@ -1847,7 +1854,7 @@ where )); } self.ensure_table_warehouse_prefix_available(&entry).await?; - let reservation = self.reserve_table_warehouse_index(&entry).await?; + let reservation = self.reserve_table_warehouse_index(&entry, true).await?; if !publication.holds_table_bucket(&entry.table_bucket) || !publication.holds_table(&entry.table_bucket, &entry.namespace, &entry.table) { @@ -5470,7 +5477,7 @@ where if next.warehouse_location != current.warehouse_location { self.ensure_table_warehouse_prefix_available(&next).await?; } - let reservation = self.reserve_table_warehouse_index(&next).await?; + let reservation = self.reserve_table_warehouse_index(&next, false).await?; let staged_write_result = async { if !has_existing_commit { diff --git a/rustfs/src/table_catalog/tests.rs b/rustfs/src/table_catalog/tests.rs index 6ed547c03..62d2ec7b6 100644 --- a/rustfs/src/table_catalog/tests.rs +++ b/rustfs/src/table_catalog/tests.rs @@ -3899,6 +3899,38 @@ async fn table_data_plane_resource_does_not_match_sibling_prefix() { assert_eq!(backend.list_call_count().await, 1); } +#[tokio::test] +async fn object_catalog_registration_scans_warehouse_prefixes_once() { + let backend = TestCatalogObjectBackend::default(); + let store = ObjectTableCatalogStore::new(backend.clone()); + let bucket = "analytics"; + let namespace = Namespace::parse("sales").expect("namespace should parse"); + store.put_table_bucket(test_bucket_entry(bucket)).await.unwrap(); + store + .create_namespace(test_namespace_entry(bucket, &namespace)) + .await + .unwrap(); + + let mut reads = Vec::new(); + for name in ["orders", "returns", "shipments"] { + let table = IdentifierSegment::parse(name).expect("table should parse"); + let mut entry = test_table_entry( + bucket, + &namespace, + &table, + default_table_metadata_file_path(&namespace, &table, "00001.metadata.json"), + ); + entry.table_id = format!("table-{name}"); + entry.warehouse_location = format!("s3://{bucket}/tables/{name}"); + backend.reset_call_counts().await; + store.create_table(entry).await.expect("table should be created"); + reads.push(backend.read_call_count().await); + assert_eq!(backend.list_call_count().await, 2, "one table scan and one warehouse index scan"); + } + assert_eq!(reads[2] - reads[1], 2, "each existing table and its index should be read once"); + assert_eq!(store.list_tables(bucket, &namespace.public_name()).await.unwrap().len(), 3); +} + #[tokio::test] async fn catalog_backings_reject_overlapping_registered_warehouse_prefixes() { let backend = TestCatalogObjectBackend::default(); @@ -15044,6 +15076,7 @@ async fn object_table_catalog_store_commits_with_token_match_and_writes_log() { .unwrap(); backend.seed_object(bucket, &new_metadata, b"{}".to_vec()).await; + backend.reset_call_counts().await; let result = store .commit_table(TableCommitRequest { table_bucket: bucket.to_string(), @@ -15061,6 +15094,11 @@ async fn object_table_catalog_store_commits_with_token_match_and_writes_log() { .await .unwrap(); + assert_eq!( + backend.list_call_count().await, + 0, + "an unchanged warehouse must not trigger a catalog scan" + ); assert_eq!(result.table.metadata_location, new_metadata); assert_ne!(result.table.version_token, "token-v1"); assert_eq!(result.table.generation, 2);