perf(s3-tables): avoid duplicate registration prefix scans (#7310)

Co-authored-by: houseme <housemecn@gmail.com>
Co-authored-by: Zhengchao An <anzhengchao@gmail.com>

rustfs/src/table_catalog/store/object.rs

@@ -1336,7 +1336,11 @@ where
         Ok(())
     }
 
-    async fn reserve_table_warehouse_index(&self, entry: &TableEntry) -> TableCatalogStoreResult<WarehouseIndexReservation> {
+    async fn reserve_table_warehouse_index(
+        &self,
+        entry: &TableEntry,
+        prefix_already_checked: bool,
+    ) -> TableCatalogStoreResult<WarehouseIndexReservation> {
         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(&current).await.map(|_| ())
+        self.reserve_table_warehouse_index(&current, 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 {

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);