diff --git a/src/iceberg/catalog/rest/rest_catalog.cc b/src/iceberg/catalog/rest/rest_catalog.cc index 9635aebef..ded53ce4e 100644 --- a/src/iceberg/catalog/rest/rest_catalog.cc +++ b/src/iceberg/catalog/rest/rest_catalog.cc @@ -479,6 +479,7 @@ RestCatalog::RestCatalog(RestCatalogProperties config, std::shared_ptr f std::string_view RestCatalog::name() const { return name_; } Result> RestCatalog::AsCatalog() { + std::lock_guard lock(default_catalog_mutex_); if (auto catalog = default_catalog_.lock()) { return catalog; } diff --git a/src/iceberg/catalog/rest/rest_catalog.h b/src/iceberg/catalog/rest/rest_catalog.h index 65b0b5eab..a840c797f 100644 --- a/src/iceberg/catalog/rest/rest_catalog.h +++ b/src/iceberg/catalog/rest/rest_catalog.h @@ -20,6 +20,7 @@ #pragma once #include +#include #include #include @@ -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 { @@ -200,7 +205,8 @@ class ICEBERG_REST_EXPORT RestCatalog final std::shared_ptr catalog_session_; SnapshotMode snapshot_mode_; SessionContext default_context_; - std::weak_ptr default_catalog_; + std::mutex default_catalog_mutex_; + std::weak_ptr default_catalog_; // Guarded by default_catalog_mutex_. std::shared_ptr reporter_; Executor* metrics_executor_ = nullptr; }; diff --git a/src/iceberg/test/rest_catalog_integration_test.cc b/src/iceberg/test/rest_catalog_integration_test.cc index 69632d522..57e699d74 100644 --- a/src/iceberg/test/rest_catalog_integration_test.cc +++ b/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 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) {