fix(rest): make RestCatalog safe to share across threads

src/iceberg/catalog/rest/rest_catalog.cc

@@ -479,6 +479,7 @@ RestCatalog::RestCatalog(RestCatalogProperties config, std::shared_ptr<FileIO> f
 std::string_view RestCatalog::name() const { return name_; }
 
 Result<std::shared_ptr<Catalog>> RestCatalog::AsCatalog() {
+  std::lock_guard lock(default_catalog_mutex_);
   if (auto catalog = default_catalog_.lock()) {
     return catalog;
   }

src/iceberg/catalog/rest/rest_catalog.h

@@ -20,6 +20,7 @@
 #pragma once
 
 #include <memory>
+#include <mutex>
 #include <string>
 #include <unordered_set>
 
@@ -40,6 +41,10 @@
 namespace iceberg::rest {
 
 /// \brief Session-aware REST catalog root.
+///
+/// Thread-safe, as are the catalogs returned by AsCatalog() and WithContext(), provided
+/// the configured FileIO implementation, metrics reporters, metrics executor and auth
+/// manager (including its sessions) are thread-safe too.
 class ICEBERG_REST_EXPORT RestCatalog final
     : public SessionCatalog,
       public std::enable_shared_from_this<RestCatalog> {
@@ -200,7 +205,8 @@ class ICEBERG_REST_EXPORT RestCatalog final
   std::shared_ptr<auth::AuthSession> catalog_session_;
   SnapshotMode snapshot_mode_;
   SessionContext default_context_;
-  std::weak_ptr<Catalog> default_catalog_;
+  std::mutex default_catalog_mutex_;
+  std::weak_ptr<Catalog> default_catalog_;  // Guarded by default_catalog_mutex_.
   std::shared_ptr<MetricsReporter> reporter_;
   Executor* metrics_executor_ = nullptr;
 };

src/iceberg/test/rest_catalog_integration_test.cc

@@ -445,6 +445,52 @@ TEST_F(RestCatalogIntegrationTest, ConcurrentHttpClientRequests) {
   }
 }
 
+// Query engines share one catalog between concurrent queries.
+TEST_F(RestCatalogIntegrationTest, ConcurrentCatalogUse) {
+  constexpr int kThreads = 16;
+  constexpr int kAsCatalogCalls = 10000;
+  constexpr int kRequests = 50;
+
+  auto config = RestCatalogProperties::default_properties();
+  config.Set(RestCatalogProperties::kUri, CatalogUri())
+      .Set(RestCatalogProperties::kName, std::string(kCatalogName))
+      .Set(RestCatalogProperties::kWarehouse, std::string(kWarehouseName));
+  config.mutable_configs()[std::string(RestCatalogProperties::kIOImpl.key())] =
+      std::string(kStdFileIOImpl);
+  ICEBERG_UNWRAP_OR_FAIL(auto root, RestCatalog::Make(config));
+
+  Namespace ns{.levels = {"test_concurrent_catalog_use"}};
+  TableIdentifier table_id{.ns = ns, .name = "events"};
+  {
+    ICEBERG_UNWRAP_OR_FAIL(auto catalog, root->AsCatalog());
+    ASSERT_THAT(catalog->CreateNamespace(ns, {}), IsOk());
+    ASSERT_THAT(CreateDefaultTable(catalog, table_id), IsOk());
+  }
+
+  std::vector<std::thread> threads;
+  threads.reserve(kThreads);
+  for (int i = 0; i < kThreads; ++i) {
+    threads.emplace_back([&] {
+      // Nothing else holds the default catalog, so these calls keep recreating it.
+      for (int j = 0; j < kAsCatalogCalls; ++j) {
+        ASSERT_THAT(root->AsCatalog(), IsOk());
+      }
+      ICEBERG_UNWRAP_OR_FAIL(auto catalog, root->AsCatalog());
+      for (int j = 0; j < kRequests; ++j) {
+        ASSERT_THAT(catalog->LoadTable(table_id), IsOk());
+        ASSERT_THAT(catalog->ListNamespaces(ns), IsOk());
+      }
+    });
+  }
+  for (auto& thread : threads) {
+    thread.join();
+  }
+
+  ICEBERG_UNWRAP_OR_FAIL(auto catalog, root->AsCatalog());
+  ASSERT_THAT(catalog->DropTable(table_id, /*purge=*/false), IsOk());
+  ASSERT_THAT(catalog->DropNamespace(ns), IsOk());
+}
+
 // -- Namespace operations --
 
 TEST_F(RestCatalogIntegrationTest, ListNamespaces) {