RestCatalog.cpp

#include <Poco/JSON/Object.h>
#include <Poco/JSON/Stringifier.h>
#include <Poco/Net/HTTPRequest.h>
#include <Common/Exception.h>
#include <Common/logger_useful.h>
#include <Common/setThreadName.h>
#include <Common/CurrentThread.h>
#include <Storages/ObjectStorage/DataLakes/Iceberg/IcebergWrites.h>
#include <mutex>
#include <chrono>
#include <Core/SettingsEnums.h>
#include "config.h"

#if USE_AVRO
#include <Databases/DataLake/RestCatalog.h>
#include <Databases/DataLake/StorageCredentials.h>

#include <base/find_symbols.h>
#include <Core/Settings.h>
#include <Common/escapeForFileName.h>
#include <Common/threadPoolCallbackRunner.h>
#include <Common/Base64.h>
#include <Common/checkStackSize.h>

#include <IO/ConnectionTimeouts.h>
#include <IO/GCPOAuth.h>
#include <IO/HTTPCommon.h>
#include <IO/ReadBuffer.h>
#include <IO/ReadBufferFromFile.h>
#include <IO/WriteBufferFromString.h>
#include <IO/Operators.h>
#include <Interpreters/Context.h>
#include <filesystem>

#include <Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.h>
#include <Server/HTTP/HTMLForm.h>
#include <Formats/FormatFactory.h>

#include <Poco/URI.h>
#include <Poco/JSON/Array.h>
#include <Poco/JSON/Parser.h>
#include <Poco/Net/HTTPClientSession.h>
#include <Poco/Net/HTTPResponse.h>
#include <Poco/Net/HTTPSClientSession.h>
#include <Poco/Net/SSLManager.h>
#include <Poco/StreamCopier.h>
#include <Poco/DateTime.h>
#include <Poco/DateTimeFormat.h>
#include <Poco/DateTimeParser.h>
#include <Poco/StringTokenizer.h>
#include <Poco/Timestamp.h>


namespace DB::ErrorCodes
{
    extern const int DATALAKE_DATABASE_ERROR;
    extern const int LOGICAL_ERROR;
    extern const int BAD_ARGUMENTS;
    extern const int CATALOG_NAMESPACE_DISABLED;
}

namespace ProfileEvents
{
    extern const Event DataLakeRestCatalogLoadConfig;
    extern const Event DataLakeRestCatalogLoadConfigMicroseconds;
    extern const Event DataLakeRestCatalogGetNamespaces;
    extern const Event DataLakeRestCatalogGetNamespacesMicroseconds;
    extern const Event DataLakeRestCatalogGetTables;
    extern const Event DataLakeRestCatalogGetTablesMicroseconds;
    extern const Event DataLakeRestCatalogGetTableMetadata;
    extern const Event DataLakeRestCatalogGetTableMetadataMicroseconds;
    extern const Event DataLakeRestCatalogGetCredentials;
    extern const Event DataLakeRestCatalogGetCredentialsMicroseconds;
    extern const Event DataLakeRestCatalogCredentialsVended;
    extern const Event DataLakeRestCatalogCredentialsCacheHits;
    extern const Event DataLakeRestCatalogAuthTokenCacheHits;
    extern const Event DataLakeRestCatalogAuthTokenRefreshed;
    extern const Event DataLakeRestCatalogAuthTokenRefreshedMicroseconds;
    extern const Event DataLakeRestCatalogAuthTokenRefreshedOnUnauthorized;
    extern const Event DataLakeRestCatalogCreateNamespace;
    extern const Event DataLakeRestCatalogCreateNamespaceMicroseconds;
    extern const Event DataLakeRestCatalogCreateTable;
    extern const Event DataLakeRestCatalogCreateTableMicroseconds;
    extern const Event DataLakeRestCatalogUpdateTable;
    extern const Event DataLakeRestCatalogUpdateTableMicroseconds;
    extern const Event DataLakeRestCatalogDropTable;
    extern const Event DataLakeRestCatalogDropTableMicroseconds;
}

namespace DataLake
{

static constexpr auto CONFIG_ENDPOINT = "config";
static constexpr auto NAMESPACES_ENDPOINT = "namespaces";

namespace
{

std::pair<std::string, std::string> parseCatalogCredential(const std::string & catalog_credential)
{
    /// Parse a string of format "<client_id>:<client_secret>"
    /// into separare strings client_id and client_secret.

    std::string client_id;
    std::string client_secret;
    if (!catalog_credential.empty())
    {
        auto pos = catalog_credential.find(':');
        if (pos == std::string::npos)
        {
            throw DB::Exception(
                DB::ErrorCodes::BAD_ARGUMENTS, "Unexpected format of catalog credential: "
                "expected client_id and client_secret separated by `:`");
        }
        client_id = catalog_credential.substr(0, pos);
        client_secret = catalog_credential.substr(pos + 1);
    }
    return std::pair(client_id, client_secret);
}

DB::HTTPHeaderEntry parseAuthHeader(const std::string & auth_header)
{
    /// Parse a string of format "Authorization: <auth_scheme> <auth_token>"
    /// into a key-value header "Authorization", "<auth_scheme> <auth_token>"

    auto pos = auth_header.find(':');
    if (pos == std::string::npos)
        throw DB::Exception(DB::ErrorCodes::BAD_ARGUMENTS, "Unexpected format of auth header");

    return DB::HTTPHeaderEntry(auth_header.substr(0, pos), auth_header.substr(pos + 1));
}

std::string correctAPIURI(const std::string & uri)
{
    if (uri.ends_with("v1"))
        return uri;
    return std::filesystem::path(uri) / "v1";
}

String encodeNamespaceForURI(const String & namespace_name)
{
    String encoded;
    for (const auto & ch : namespace_name)
    {
        if (ch == '.')
            encoded += "%1F";
        else
            encoded.push_back(ch);
    }
    return encoded;
}

}

std::string RestCatalog::Config::toString() const
{
    DB::WriteBufferFromOwnString wb;

    if (!prefix.empty())
        wb << "prefix: " << prefix.string() << ", ";

    if (!default_base_location.empty())
        wb << "default_base_location: " << default_base_location << ", ";

    return wb.str();
}

RestCatalog::RestCatalog(
    const std::string & warehouse_,
    const std::string & base_url_,
    const std::string & catalog_credential_,
    const std::string & auth_scope_,
    const std::string & auth_header_,
    const std::string & oauth_server_uri_,
    bool oauth_server_use_request_body_,
    const std::string & namespaces_,
    DB::ContextPtr context_)
    : ICatalog(warehouse_)
    , DB::WithContext(context_)
    , base_url(correctAPIURI(base_url_))
    , log(getLogger("RestCatalog(" + warehouse_ + ")"))
    , auth_scope(auth_scope_)
    , oauth_server_uri(oauth_server_uri_)
    , oauth_server_use_request_body(oauth_server_use_request_body_)
    , allowed_namespaces(namespaces_)
{
    if (!catalog_credential_.empty())
    {
        std::tie(client_id, client_secret) = parseCatalogCredential(catalog_credential_);
        update_token_if_expired = true;
    }
    else if (!auth_header_.empty())
    {
        auth_header = parseAuthHeader(auth_header_);
    }
    config = loadConfig();
}

RestCatalog::RestCatalog(
    const std::string & warehouse_,
    const std::string & base_url_,
    const std::string & auth_scope_,
    const std::string & oauth_server_uri_,
    bool oauth_server_use_request_body_,
    const std::string & namespaces_,
    DB::ContextPtr context_)
    : ICatalog(warehouse_)
    , DB::WithContext(context_)
    , base_url(correctAPIURI(base_url_))
    , log(getLogger("RestCatalog(" + warehouse_ + ")"))
    , auth_scope(auth_scope_)
    , oauth_server_uri(oauth_server_uri_)
    , oauth_server_use_request_body(oauth_server_use_request_body_)
    , allowed_namespaces(namespaces_)
{
}


RestCatalog::Config RestCatalog::loadConfig()
{
    Poco::URI::QueryParameters params = {{"warehouse", warehouse}};

    std::string json_str;

    {
        ProfileEvents::increment(ProfileEvents::DataLakeRestCatalogLoadConfig);
        auto timer = DB::CurrentThread::getProfileEvents().timer(ProfileEvents::DataLakeRestCatalogLoadConfigMicroseconds);
        auto buf = createReadBuffer(CONFIG_ENDPOINT, params);
        readJSONObjectPossiblyInvalid(json_str, *buf);
    }

    LOG_DEBUG(log, "Received catalog configuration settings: {}", json_str);

    Poco::JSON::Parser parser;
    Poco::Dynamic::Var json = parser.parse(json_str);
    const Poco::JSON::Object::Ptr & object = json.extract<Poco::JSON::Object::Ptr>();

    Config result;

    auto defaults_object = object->get("defaults").extract<Poco::JSON::Object::Ptr>();
    parseCatalogConfigurationSettings(defaults_object, result);

    auto overrides_object = object->get("overrides").extract<Poco::JSON::Object::Ptr>();
    parseCatalogConfigurationSettings(overrides_object, result);

    LOG_DEBUG(log, "Parsed catalog configuration settings: {}", result.toString());
    return result;
}

void RestCatalog::parseCatalogConfigurationSettings(const Poco::JSON::Object::Ptr & object, Config & result)
{
    if (!object)
        return;

    if (object->has("prefix"))
        result.prefix = object->get("prefix").extract<String>();

    if (object->has("default-base-location"))
        result.default_base_location = object->get("default-base-location").extract<String>();
}

DB::HTTPHeaderEntries RestCatalog::getAuthHeaders(
    bool update_token,
    const String & /*method*/,
    const Poco::URI & /*url*/,
    const DB::HTTPHeaderEntries & /*extra_headers*/,
    const String & /*body*/,
    bool * used_cached_oauth_token) const
{
    if (used_cached_oauth_token)
        *used_cached_oauth_token = false;

    /// Option 1: user specified auth header manually.
    /// Header has format: 'Authorization: <scheme> <token>'.
    if (auth_header.has_value())
    {
        return DB::HTTPHeaderEntries{auth_header.value()};
    }

    /// Option 2: user provided grant_type, client_id and client_secret.
    /// We would make OAuthClientCredentialsRequest
    /// https://github.com/apache/iceberg/blob/3badfe0c1fcf0c0adfc7aa4a10f0b50365c48cf9/open-api/rest-catalog-open-api.yaml#L3498C5-L3498C34
    if (!client_id.empty())
    {
        if (!access_token.has_value() || update_token || access_token->isExpired())
        {
            access_token = retrieveAccessToken();
        }
        else if (used_cached_oauth_token)
        {
            *used_cached_oauth_token = true;
        }

        DB::HTTPHeaderEntries headers;
        headers.emplace_back("Authorization", "Bearer " + access_token.value().token);
        return headers;
    }
    return {};
}

OneLakeCatalog::OneLakeCatalog(
    const std::string & warehouse_,
    const std::string & base_url_,
    const std::string & onelake_tenant_id,
    const std::string & onelake_client_id,
    const std::string & onelake_client_secret,
    const std::string & auth_scope_,
    const std::string & oauth_server_uri_,
    bool oauth_server_use_request_body_,
    const std::string & namespaces_,
    DB::ContextPtr context_)
    : RestCatalog(warehouse_, base_url_, auth_scope_, oauth_server_uri_, oauth_server_use_request_body_, namespaces_, context_)
    , tenant_id(onelake_tenant_id)
{
    client_id = onelake_client_id;
    client_secret = onelake_client_secret;
    update_token_if_expired = true;
    // Get token before loading config so getAuthHeaders() can work
    if (!client_id.empty() && !client_secret.empty())
    {
        access_token = retrieveAccessToken();
    }
    config = loadConfig();
}

AccessToken RestCatalog::retrieveAccessToken() const
{
    ProfileEvents::increment(ProfileEvents::DataLakeRestCatalogAuthTokenRefreshed);
    auto timer = DB::CurrentThread::getProfileEvents().timer(ProfileEvents::DataLakeRestCatalogAuthTokenRefreshedMicroseconds);

    static constexpr auto oauth_tokens_endpoint = "oauth/tokens";

    /// TODO:
    /// 1. support oauth2-server-uri
    /// https://github.com/apache/iceberg/blob/918f81f3c3f498f46afcea17c1ac9cdc6913cb5c/open-api/rest-catalog-open-api.yaml#L183C82-L183C99

    Poco::URI url;
    DB::ReadWriteBufferFromHTTP::OutStreamCallback out_stream_callback;
    size_t body_size = 0;
    String body;

    if (oauth_server_uri.empty() && !oauth_server_use_request_body)
    {
        url = Poco::URI(base_url / oauth_tokens_endpoint);

        Poco::URI::QueryParameters params = {
            {"grant_type", "client_credentials"},
            {"scope", auth_scope},
            {"client_id", client_id},
            {"client_secret", client_secret},
        };
        url.setQueryParameters(params);
    }
    else
    {
        String encoded_auth_scope;
        String encoded_client_id;
        String encoded_client_secret;
        Poco::URI::encode(auth_scope, auth_scope, encoded_auth_scope);
        Poco::URI::encode(client_id, client_id, encoded_client_id);
        Poco::URI::encode(client_secret, client_secret, encoded_client_secret);

        body = fmt::format(
            "grant_type=client_credentials&scope={}&client_id={}&client_secret={}",
            encoded_auth_scope, encoded_client_id, encoded_client_secret);
        body_size = body.size();
        out_stream_callback = [&](std::ostream & os)
        {
            os << body;
        };

        if (oauth_server_uri.empty())
            url = Poco::URI(base_url / oauth_tokens_endpoint);
        else
            url = Poco::URI(oauth_server_uri);
    }

    const auto & context = getContext();
    auto timeouts = DB::ConnectionTimeouts::getHTTPTimeouts(context->getSettingsRef(), context->getServerSettings());
    auto session = makeHTTPSession(DB::HTTPConnectionGroupType::HTTP, url, timeouts, {});

    Poco::Net::HTTPRequest request(Poco::Net::HTTPRequest::HTTP_POST, url.getPathAndQuery(),
                                Poco::Net::HTTPMessage::HTTP_1_1);
    request.setContentType("application/x-www-form-urlencoded");
    request.setContentLength(body_size);
    request.set("Accept", "application/json");

    std::ostream & os = session->sendRequest(request);
    out_stream_callback(os);

    Poco::Net::HTTPResponse response;
    std::istream & rs = session->receiveResponse(response);

    std::string json_str;
    Poco::StreamCopier::copyToString(rs, json_str);

    Poco::JSON::Parser parser;
    Poco::Dynamic::Var res_json = parser.parse(json_str);
    const Poco::JSON::Object::Ptr & object = res_json.extract<Poco::JSON::Object::Ptr>();

    AccessToken token;
    token.token = object->get("access_token").extract<String>();

    if (object->has("expires_in"))
    {
        Int64 expires_in = object->getValue<Int64>("expires_in");
        /// Use 90% of the token lifetime as the validity window so that short-lived tokens
        /// (e.g. expires_in=300) still get a sensible buffer instead of going non-positive.
        token.expires_at = std::chrono::system_clock::now() + std::chrono::seconds(expires_in * 9 / 10);
    }

    return token;
}

BigLakeCatalog::BigLakeCatalog(
    const std::string & warehouse_,
    const std::string & base_url_,
    const std::string & google_project_id_,
    const std::string & google_service_account_,
    const std::string & google_metadata_service_,
    const std::string & google_adc_client_id_,
    const std::string & google_adc_client_secret_,
    const std::string & google_adc_refresh_token_,
    const std::string & google_adc_quota_project_id_,
    const std::string & namespaces_,
    DB::ContextPtr context_)
    : RestCatalog(warehouse_, base_url_, "", "", false, namespaces_, context_)
    , google_project_id(google_project_id_)
    , google_service_account(google_service_account_)
    , google_metadata_service(google_metadata_service_)
    , google_adc_client_id(google_adc_client_id_)
    , google_adc_client_secret(google_adc_client_secret_)
    , google_adc_refresh_token(google_adc_refresh_token_)
    , google_adc_quota_project_id(google_adc_quota_project_id_)
{
    update_token_if_expired = true;
    // Get token before loading config so getAuthHeaders() can work
    if (!google_project_id.empty() || !google_adc_client_id.empty())
    {
        access_token = retrieveGoogleCloudAccessToken();
    }
    config = loadConfig();
}

DB::HTTPHeaderEntries BigLakeCatalog::getAuthHeaders(
    bool update_token,
    const String & method,
    const Poco::URI & url,
    const DB::HTTPHeaderEntries & extra_headers,
    const String & body,
    bool * used_cached_oauth_token) const
{
    /// Google Cloud OAuth2 for BigLake.
    /// Uses GCP metadata service or Application Default Credentials to get access token.
    /// Only use Google OAuth if explicitly configured (google_project_id or google_adc_client_id).
    /// https://developers.google.com/identity/protocols/oauth2
    if (!google_project_id.empty() || !google_adc_client_id.empty())
    {
        if (used_cached_oauth_token)
            *used_cached_oauth_token = false;

        if (!access_token.has_value() || update_token || access_token->isExpired())
        {
            access_token = retrieveGoogleCloudAccessToken();
        }
        else if (used_cached_oauth_token)
        {
            *used_cached_oauth_token = true;
        }

        DB::HTTPHeaderEntries headers;
        headers.emplace_back("Authorization", "Bearer " + access_token->token);

        std::string project_id = google_project_id;
        if (project_id.empty() && !google_adc_quota_project_id.empty())
        {
            project_id = google_adc_quota_project_id;
        }

        if (!project_id.empty())
        {
            headers.emplace_back("x-goog-user-project", project_id);
        }

        return headers;
    }

    return RestCatalog::getAuthHeaders(update_token, method, url, extra_headers, body, used_cached_oauth_token);
}

AccessToken BigLakeCatalog::retrieveGoogleCloudAccessTokenFromRefreshToken() const
{
    if (google_adc_client_id.empty() || google_adc_client_secret.empty() || google_adc_refresh_token.empty())
        throw DB::Exception(
            DB::ErrorCodes::BAD_ARGUMENTS,
            "Invalid ADC credentials: client_id, client_secret, and refresh_token are required");

    const auto & context = getContext();
    auto timeouts = DB::ConnectionTimeouts::getHTTPTimeouts(context->getSettingsRef(), context->getServerSettings());
    auto result = fetchGCPOAuthToken(google_adc_client_id, google_adc_client_secret, google_adc_refresh_token, timeouts);

    AccessToken token;
    token.token = std::move(result.access_token);
    token.expires_at = std::chrono::system_clock::now() + std::chrono::seconds(result.expires_in * 9 / 10);
    return token;
}

AccessToken BigLakeCatalog::retrieveGoogleCloudAccessToken() const
{
    ProfileEvents::increment(ProfileEvents::DataLakeRestCatalogAuthTokenRefreshed);
    auto timer = DB::CurrentThread::getProfileEvents().timer(ProfileEvents::DataLakeRestCatalogAuthTokenRefreshedMicroseconds);

    if (!google_adc_client_id.empty() && !google_adc_client_secret.empty() && !google_adc_refresh_token.empty())
    {
        try
        {
            return retrieveGoogleCloudAccessTokenFromRefreshToken();
        }
        catch (const DB::Exception & e)
        {
            LOG_DEBUG(log, "Failed to use ADC credentials, falling back to metadata service: {}", e.what());
        }
    }

    /// Fallback to GCP metadata service (works inside GCP infrastructure)
    /// https://cloud.google.com/compute/docs/metadata/overview
    static constexpr auto DEFAULT_REQUEST_TOKEN_PATH = "/computeMetadata/v1/instance/service-accounts";

    Poco::URI url;
    url.setScheme("http");
    url.setHost(google_metadata_service);
    url.setPath(fmt::format("{}/{}/token", DEFAULT_REQUEST_TOKEN_PATH, google_service_account));

    Poco::Net::HTTPRequest request(Poco::Net::HTTPRequest::HTTP_GET, url.toString(), Poco::Net::HTTPRequest::HTTP_1_1);
    request.add("metadata-flavor", "Google");

    LOG_DEBUG(log, "Requesting Google Cloud access token from metadata service: {}", url.toString());

    const auto & context = getContext();
    auto timeouts = DB::ConnectionTimeouts::getHTTPTimeouts(context->getSettingsRef(), context->getServerSettings());
    auto session = makeHTTPSession(DB::HTTPConnectionGroupType::HTTP, url, timeouts, {});

    if (!session)
        throw DB::Exception(DB::ErrorCodes::BAD_ARGUMENTS, "Can not create HTTP session");
    session->sendRequest(request);

    Poco::Net::HTTPResponse response;
    auto & in = session->receiveResponse(response);

    if (response.getStatus() != Poco::Net::HTTPResponse::HTTP_OK)
    {
        throw DB::Exception(
            DB::ErrorCodes::BAD_ARGUMENTS,
            "Failed to request Google Cloud bearer token from metadata service: {} (status: {})",
            response.getReason(),
            static_cast<int>(response.getStatus()));
    }

    String token_json_raw;
    Poco::StreamCopier::copyToString(in, token_json_raw);

    LOG_DEBUG(log, "Received Google Cloud token response from metadata service");

    Poco::JSON::Parser parser;
    auto object = parser.parse(token_json_raw).extract<Poco::JSON::Object::Ptr>();

    if (!object->has("access_token") || !object->has("expires_in") || !object->has("token_type"))
    {
        throw DB::Exception(
            DB::ErrorCodes::BAD_ARGUMENTS,
            "Unexpected structure of Google Cloud token response. Response should have fields: 'access_token', 'expires_in', 'token_type'");
    }

    auto token_type = object->getValue<String>("token_type");
    if (token_type != "Bearer")
    {
        throw DB::Exception(
            DB::ErrorCodes::BAD_ARGUMENTS,
            "Unexpected token type in Google Cloud response. Expected Bearer token, got {}",
            token_type);
    }

    AccessToken token;
    token.token = object->getValue<String>("access_token");

    if (object->has("expires_in"))
    {
        Int64 expires_in = object->getValue<Int64>("expires_in");
        token.expires_at = std::chrono::system_clock::now() + std::chrono::seconds(expires_in * 9 / 10);
    }

    return token;
}

std::optional<StorageType> RestCatalog::getStorageType() const
{
    if (config.default_base_location.empty())
        return std::nullopt;
    return parseStorageTypeFromLocation(config.default_base_location);
}

DB::ReadWriteBufferFromHTTPPtr RestCatalog::createReadBuffer(
    const std::string & endpoint,
    const Poco::URI::QueryParameters & params,
    const DB::HTTPHeaderEntries & headers) const
{
    const auto & context = getContext();

    /// enable_url_encoding=false to allow use tables with encoded sequences in names like 'foo%2Fbar'
    Poco::URI url(base_url / endpoint, /* enable_url_encoding */ false);
    if (!params.empty())
        url.setQueryParameters(params);

    auto create_buffer = [&](bool update_token, bool & used_cached_oauth_token)
    {
        auto result_headers = getAuthHeaders(update_token, Poco::Net::HTTPRequest::HTTP_GET, url, headers, {}, &used_cached_oauth_token);
        std::move(headers.begin(), headers.end(), std::back_inserter(result_headers));

        return DB::BuilderRWBufferFromHTTP(url)
            .withConnectionGroup(DB::HTTPConnectionGroupType::HTTP)
            .withSettings(getContext()->getReadSettings())
            .withTimeouts(DB::ConnectionTimeouts::getHTTPTimeouts(context->getSettingsRef(), context->getServerSettings()))
            .withHostFilter(&getContext()->getRemoteHostFilter())
            .withHeaders(result_headers)
            .withDelayInit(false)
            .withSkipNotFound(false)
            .create(credentials);
    };

    LOG_DEBUG(log, "Requesting: {}", url.toString());

    try
    {
        bool used_cached_oauth_token = false;
        auto buf = create_buffer(false, used_cached_oauth_token);
        if (used_cached_oauth_token)
            ProfileEvents::increment(ProfileEvents::DataLakeRestCatalogAuthTokenCacheHits);
        return buf;
    }
    catch (const DB::HTTPException & e)
    {
        const auto status = e.getHTTPStatus();
        if (update_token_if_expired &&
            (status == Poco::Net::HTTPResponse::HTTPStatus::HTTP_UNAUTHORIZED
             || status == Poco::Net::HTTPResponse::HTTPStatus::HTTP_FORBIDDEN))
        {
            ProfileEvents::increment(ProfileEvents::DataLakeRestCatalogAuthTokenRefreshedOnUnauthorized);
            bool used_cached_oauth_token_on_retry = false;
            return create_buffer(true, used_cached_oauth_token_on_retry);
        }
        throw;
    }
}

bool RestCatalog::empty() const
{
    /// TODO: add a test with empty namespaces and zero namespaces.
    bool found_table = false;
    auto stop_condition = [&](const std::string & namespace_name) -> bool
    {
        if (!allowed_namespaces.isNamespaceAllowed(namespace_name, /*nested*/ false))
            return false;
        const auto tables = getTables(namespace_name, /* limit */1);
        found_table = !tables.empty();
        return found_table;
    };

    Namespaces namespaces;
    getNamespacesRecursive("", namespaces, stop_condition, /* execute_func */{});

    return found_table;
}

DB::Names RestCatalog::getTables() const
{
    auto & pool = getContext()->getIcebergCatalogThreadpool();
    DB::Names tables;
    std::mutex mutex;

    {
        /// Ensure tables and mutex (capture by reference) outlive runner
        DB::ThreadPoolCallbackRunnerLocal<void> runner(pool, DB::ThreadName::DATALAKE_REST_CATALOG);

        auto execute_for_each_namespace = [&](const std::string & current_namespace)
        {
            if (!allowed_namespaces.isNamespaceAllowed(current_namespace, /*nested*/ false))
                return;
            runner.enqueueAndKeepTrack(
            [=, &tables, &mutex, this]
            {
                auto tables_in_namespace = getTables(current_namespace);
                std::lock_guard lock(mutex);
                std::move(tables_in_namespace.begin(), tables_in_namespace.end(), std::back_inserter(tables));
            });
        };

        Namespaces namespaces;
        getNamespacesRecursive(
            /* base_namespace */"", /// Empty base namespace means starting from root.
            namespaces,
            /* stop_condition */{},
            /* execute_func */execute_for_each_namespace);

        runner.waitForAllToFinishAndRethrowFirstError();
    }

    return tables;
}

void RestCatalog::getNamespacesRecursive(
    const std::string & base_namespace,
    Namespaces & result,
    StopCondition stop_condition,
    ExecuteFunc func) const
{
    checkStackSize();

    auto namespaces = getNamespaces(base_namespace);
    result.reserve(result.size() + namespaces.size());
    result.insert(result.end(), namespaces.begin(), namespaces.end());

    for (const auto & current_namespace : namespaces)
    {
        chassert(current_namespace.starts_with(base_namespace));

        /// Protection from subnamepsaces with empty names
        if (current_namespace == base_namespace)
        {
            LOG_WARNING(log, "Namespace {} has a subnamespace with empty name. This is an error in catalog implementation.", base_namespace);
            continue;
        }

        if (stop_condition && stop_condition(current_namespace))
            break;

        if (func)
        {
            if (allowed_namespaces.isNamespaceAllowed(current_namespace, /*nested*/ false))
                func(current_namespace);
            else
            {
                LOG_DEBUG(log, "Tables in namespace {} are filtered", current_namespace);
            }
        }

        if (allowed_namespaces.isNamespaceAllowed(current_namespace, /*nested*/ true))
            getNamespacesRecursive(current_namespace, result, stop_condition, func);
        else
        {
            LOG_DEBUG(log, "Nested namespaces in namespace {} are filtered", current_namespace);
        }
    }
}

Poco::URI::QueryParameters RestCatalog::createParentNamespaceParams(const std::string & base_namespace) const
{
    std::vector<std::string_view> parts;
    splitInto<'.'>(parts, base_namespace);
    std::string parent_param;
    for (const auto & part : parts)
    {
        /// 0x1F is a unit separator
        /// https://github.com/apache/iceberg/blob/70d87f1750627b14b3b25a0216a97db86a786992/open-api/rest-catalog-open-api.yaml#L264
        if (!parent_param.empty())
            parent_param += static_cast<char>(0x1F);
        parent_param += part;
    }
    return {{"parent", parent_param}};
}

RestCatalog::Namespaces RestCatalog::getNamespaces(const std::string & base_namespace) const
{
    Poco::URI::QueryParameters params;
    if (!base_namespace.empty())
        params = createParentNamespaceParams(base_namespace);

    try
    {
        ProfileEvents::increment(ProfileEvents::DataLakeRestCatalogGetNamespaces);
        auto timer = DB::CurrentThread::getProfileEvents().timer(ProfileEvents::DataLakeRestCatalogGetNamespacesMicroseconds);
        auto buf = createReadBuffer(config.prefix / NAMESPACES_ENDPOINT, params);
        auto namespaces = parseNamespaces(*buf, base_namespace);
        LOG_DEBUG(log, "Loaded {} namespaces in base namespace {}", namespaces.size(), base_namespace);
        return namespaces;
    }
    catch (const DB::HTTPException & e)
    {
        std::string message = fmt::format(
            "Received error while fetching list of namespaces from iceberg catalog `{}`. ",
            warehouse);

        if (e.code() == Poco::Net::HTTPResponse::HTTPStatus::HTTP_NOT_FOUND)
            message += "Namespace provided in the `parent` query parameter is not found. ";

        message += fmt::format(
            "Code: {}, status: {}, message: {}",
            e.code(), e.getHTTPStatus(), e.displayText());

        throw DB::Exception(DB::ErrorCodes::DATALAKE_DATABASE_ERROR, "{}", message);
    }
}

RestCatalog::Namespaces RestCatalog::parseNamespaces(DB::ReadBuffer & buf, const std::string & base_namespace) const
{
    if (buf.eof())
        return {};

    String json_str;
    readJSONObjectPossiblyInvalid(json_str, buf);

    LOG_DEBUG(log, "Received response: {}", json_str);

    try
    {
        Poco::JSON::Parser parser;
        Poco::Dynamic::Var json = parser.parse(json_str);
        if (json.type() == typeid(Poco::JSON::Object::Ptr))
        {
            const Poco::JSON::Object::Ptr & obj = json.extract<Poco::JSON::Object::Ptr>();
            if (obj->size() == 0)
                return {};
        }
        const Poco::JSON::Object::Ptr & object = json.extract<Poco::JSON::Object::Ptr>();

        auto namespaces_object = object->get("namespaces").extract<Poco::JSON::Array::Ptr>();
        if (!namespaces_object)
            throw DB::Exception(DB::ErrorCodes::LOGICAL_ERROR, "Cannot parse result");

        Namespaces namespaces;
        for (size_t i = 0; i < namespaces_object->size(); ++i)
        {
            auto current_namespace_array = namespaces_object->get(static_cast<int>(i)).extract<Poco::JSON::Array::Ptr>();
            if (current_namespace_array->size() == 0)
                throw DB::Exception(DB::ErrorCodes::LOGICAL_ERROR, "Expected namespace array to be non-empty");

            const int idx = static_cast<int>(current_namespace_array->size()) - 1;
            const auto current_namespace = current_namespace_array->get(idx).extract<String>();
            /// BigLake does not support multi-level namespaces. When asked for sub-namespaces of
            /// a non-empty parent (via ?parent=X), BigLake ignores the filter and returns other
            /// top-level namespaces instead. Skip all sub-namespace results to avoid constructing
            /// fake multi-level paths like "ns1.ns2" that BigLake will reject with HTTP 400.
            if (getCatalogType() == DB::DatabaseDataLakeCatalogType::ICEBERG_BIGLAKE && !base_namespace.empty())
            {
                continue;
            }
            const auto full_namespace = base_namespace.empty()
                ? current_namespace
                : base_namespace + "." + current_namespace;

            namespaces.push_back(full_namespace);
        }

        return namespaces;
    }
    catch (DB::Exception & e)
    {
        e.addMessage("while parsing JSON: " + json_str);
        throw;
    }
}

DB::Names RestCatalog::getTables(const std::string & base_namespace, size_t limit) const
{
    if (!allowed_namespaces.isNamespaceAllowed(base_namespace, /*nested*/ false))
        throw DB::Exception(DB::ErrorCodes::CATALOG_NAMESPACE_DISABLED,
            "Namespace {} is filtered by `namespaces` database parameter", base_namespace);

    auto encoded_namespace = encodeNamespaceForURI(base_namespace);
    const std::string endpoint = std::filesystem::path(NAMESPACES_ENDPOINT) / encoded_namespace / "tables";

    ProfileEvents::increment(ProfileEvents::DataLakeRestCatalogGetTables);
    auto timer = DB::CurrentThread::getProfileEvents().timer(ProfileEvents::DataLakeRestCatalogGetTablesMicroseconds);
    auto buf = createReadBuffer(config.prefix / endpoint);
    return parseTables(*buf, base_namespace, limit);
}

DB::Names RestCatalog::parseTables(DB::ReadBuffer & buf, const std::string & base_namespace, size_t limit) const
{
    if (buf.eof())
        return {};

    String json_str;
    readJSONObjectPossiblyInvalid(json_str, buf);

    LOG_DEBUG(log, "Received tables response for namespace: {}", base_namespace);

    try
    {
        Poco::JSON::Parser parser;
        Poco::Dynamic::Var json = parser.parse(json_str);
        const Poco::JSON::Object::Ptr & object = json.extract<Poco::JSON::Object::Ptr>();

        auto identifiers_object = object->get("identifiers").extract<Poco::JSON::Array::Ptr>();
        if (!identifiers_object)
            throw DB::Exception(DB::ErrorCodes::LOGICAL_ERROR, "Cannot parse result");

        DB::Names tables;
        for (size_t i = 0; i < identifiers_object->size(); ++i)
        {
            const auto current_table_json = identifiers_object->get(static_cast<int>(i)).extract<Poco::JSON::Object::Ptr>();
            /// If table has encoded sequence (like 'foo%2Fbar')
            /// catalog returns decoded character instead of sequence ('foo/bar')
            /// Here name encoded back to 'foo%2Fbar' format
            const auto table_name_raw = current_table_json->get("name").extract<String>();
            std::string table_name;
            Poco::URI::encode(table_name_raw, "/", table_name);

            tables.push_back(base_namespace + "." + table_name);
            if (limit && tables.size() >= limit)
                break;
        }

        return tables;
    }
    catch (DB::Exception & e)
    {
        e.addMessage("while parsing JSON: " + json_str);
        throw;
    }
}

bool RestCatalog::existsTable(const std::string & namespace_name, const std::string & table_name) const
{
    TableMetadata table_metadata;
    return tryGetTableMetadata(namespace_name, table_name, getContext(), table_metadata);
}

bool RestCatalog::tryGetTableMetadata(
    const std::string & namespace_name,
    const std::string & table_name,
    DB::ContextPtr context_,
    TableMetadata & result) const
{
    try
    {
        return getTableMetadataImpl(namespace_name, table_name, context_, result);
    }
    catch (const DB::Exception & ex)
    {
        if (ex.code() == DB::ErrorCodes::CATALOG_NAMESPACE_DISABLED)
            throw;
        LOG_DEBUG(log, "tryGetTableMetadata response: {}", ex.what());
        return false;
    }
}

void RestCatalog::getTableMetadata(
    const std::string & namespace_name,
    const std::string & table_name,
    DB::ContextPtr context_,
    TableMetadata & result) const
{
    if (!getTableMetadataImpl(namespace_name, table_name, context_, result))
        throw DB::Exception(DB::ErrorCodes::DATALAKE_DATABASE_ERROR, "No response from iceberg catalog");
}

bool RestCatalog::getTableMetadataImpl(
    const std::string & namespace_name,
    const std::string & table_name,
    DB::ContextPtr context_,
    TableMetadata & result) const
{
    LOG_DEBUG(log, "Checking table {} in namespace {}", table_name, namespace_name);

    if (!allowed_namespaces.isNamespaceAllowed(namespace_name, /*nested*/ false))
        throw DB::Exception(DB::ErrorCodes::CATALOG_NAMESPACE_DISABLED,
            "Namespace {} is filtered by `namespaces` database parameter", namespace_name);

    DB::HTTPHeaderEntries headers;

    const bool want_credentials = result.requiresCredentials();

    /// Reuse previously vended credentials is possible
    std::optional<VendedStorageCredentials> cached_credentials;
    if (want_credentials)
    {
        cached_credentials = tryGetCachedCredentials(namespace_name, table_name);

        /// Header `X-Iceberg-Access-Delegation` tells catalog to include storage credentials in LoadTableResponse.
        /// Value can be one of the two:
        /// 1. `vended-credentials`
        /// 2. `remote-signing`
        /// Currently we support only the first.
        /// https://github.com/apache/iceberg/blob/3badfe0c1fcf0c0adfc7aa4a10f0b50365c48cf9/open-api/rest-catalog-open-api.yaml#L1832
        if (cached_credentials)
            ProfileEvents::increment(ProfileEvents::DataLakeRestCatalogCredentialsCacheHits);
        else
        {
            ProfileEvents::increment(ProfileEvents::DataLakeRestCatalogCredentialsVended);
            headers.emplace_back("X-Iceberg-Access-Delegation", "vended-credentials");
        }
    }

    const std::string endpoint = std::filesystem::path(NAMESPACES_ENDPOINT) / encodeNamespaceForURI(namespace_name) / "tables" / table_name;
    String json_str;

    {
        ProfileEvents::increment(ProfileEvents::DataLakeRestCatalogGetTableMetadata);
        auto timer = DB::CurrentThread::getProfileEvents().timer(ProfileEvents::DataLakeRestCatalogGetTableMetadataMicroseconds);
        auto buf = createReadBuffer(config.prefix / endpoint, /* params */{}, headers);

        if (buf->eof())
        {
            LOG_DEBUG(log, "Table doesn't exist (endpoint: {})", endpoint);
            return false;
        }

        readJSONObjectPossiblyInvalid(json_str, *buf);
    }

#ifdef DEBUG_OR_SANITIZER_BUILD
    /// This log message might contain credentials,
    /// so log it only for debugging.
    LOG_DEBUG(log, "Received metadata for table {}: {}", table_name, json_str);
#endif

    Poco::JSON::Parser parser;
    Poco::Dynamic::Var json = parser.parse(json_str);
    const Poco::JSON::Object::Ptr & object = json.extract<Poco::JSON::Object::Ptr>();

    auto metadata_object = object->get("metadata").extract<Poco::JSON::Object::Ptr>();
    if (!metadata_object)
        throw DB::Exception(DB::ErrorCodes::LOGICAL_ERROR, "Cannot parse result");

    std::string location;
    if (result.requiresLocation())
    {
        if (metadata_object->has("location"))
        {
            location = metadata_object->get("location").extract<String>();
            result.setLocation(location);
            LOG_DEBUG(log, "Location for table {}: {}", table_name, location);
        }
        else
        {
            result.setTableIsNotReadable(fmt::format("Cannot read table {}, because no 'location' in response", table_name));
        }
    }

    if (result.requiresSchema())
    {
        // int format_version = metadata_object->getValue<int>("format-version");
        auto schema_processor = DB::Iceberg::IcebergSchemaProcessor(context_);
        auto id = DB::IcebergMetadata::parseTableSchema(metadata_object, schema_processor, context_, log);
        auto schema = schema_processor.getClickhouseTableSchemaById(id);
        result.setSchema(*schema);
    }

    if (want_credentials && result.isDefaultReadableTable())
    {
        if (cached_credentials)
        {
            result.setStorageCredentials(cached_credentials->credentials);
            if (!cached_credentials->endpoint.empty())
                result.setEndpoint(cached_credentials->endpoint);
        }
        else if (object->has("config"))
        {
            auto config_object = object->get("config").extract<Poco::JSON::Object::Ptr>();
            if (!config_object)
                throw DB::Exception(DB::ErrorCodes::LOGICAL_ERROR, "Cannot parse config result");
            auto parsed = getCredentialsAndEndpoint(config_object, location);
            if (parsed.credentials)
            {
                result.setStorageCredentials(parsed.credentials);
                cacheCredentials(namespace_name, table_name, parsed);
            }
            if (!parsed.endpoint.empty())
                result.setEndpoint(parsed.endpoint);
        }
    }

    if (result.requiresDataLakeSpecificProperties())
    {
        if (object->has("metadata-location") && !object->get("metadata-location").isEmpty())
        {
            auto metadata_location = object->get("metadata-location").extract<String>();
            result.setDataLakeSpecificProperties(DataLakeSpecificProperties{ .iceberg_metadata_file_location = metadata_location });
        }
    }

    if (metadata_object->has("table-uuid"))
        result.setTableUUID(metadata_object->get("table-uuid").extract<String>());

    return true;
}

void RestCatalog::sendRequest(const String & endpoint, Poco::JSON::Object::Ptr request_body, const String & method, bool ignore_result) const
{
    std::ostringstream oss;  // STYLE_CHECK_ALLOW_STD_STRING_STREAM
    if (request_body)
        request_body->stringify(oss);
    const std::string body_str = DB::removeEscapedSlashes(oss.str());

    const auto & context = getContext();

    DB::ReadWriteBufferFromHTTP::OutStreamCallback out_stream_callback;
    if (!body_str.empty())
    {
        out_stream_callback = [body_str](std::ostream & os)
        {
            os << body_str;
        };
    }

    /// enable_url_encoding=false to allow use tables with encoded sequences in names like 'foo%2Fbar'
    Poco::URI url(endpoint, /* enable_url_encoding */ false);

    DB::HTTPHeaderEntries extra_headers;
    extra_headers.emplace_back("Content-Type", "application/json");

    auto create_buffer = [&](bool update_token, bool & used_cached_oauth_token)
    {
        DB::HTTPHeaderEntries headers = getAuthHeaders(update_token, method, url, extra_headers, body_str, &used_cached_oauth_token);
        headers.emplace_back("Content-Type", "application/json");
        return DB::BuilderRWBufferFromHTTP(url)
            .withConnectionGroup(DB::HTTPConnectionGroupType::HTTP)
            .withMethod(method)
            .withSettings(context->getReadSettings())
            .withTimeouts(DB::ConnectionTimeouts::getHTTPTimeouts(context->getSettingsRef(), context->getServerSettings()))
            .withHostFilter(&context->getRemoteHostFilter())
            .withHeaders(headers)
            .withOutCallback(out_stream_callback)
            .withSkipNotFound(false)
            .create(credentials);
    };

    try
    {
        bool used_cached_oauth_token = false;
        auto wb = create_buffer(false, used_cached_oauth_token);

        String response_str;
        if (!ignore_result)
            readJSONObjectPossiblyInvalid(response_str, *wb);
        else
            wb->ignoreAll();

        if (used_cached_oauth_token)
            ProfileEvents::increment(ProfileEvents::DataLakeRestCatalogAuthTokenCacheHits);
    }
    catch (const DB::HTTPException & e)
    {
        const auto status = e.getHTTPStatus();
        if (update_token_if_expired &&
            (status == Poco::Net::HTTPResponse::HTTPStatus::HTTP_UNAUTHORIZED
             || status == Poco::Net::HTTPResponse::HTTPStatus::HTTP_FORBIDDEN))
        {
            ProfileEvents::increment(ProfileEvents::DataLakeRestCatalogAuthTokenRefreshedOnUnauthorized);
            bool used_cached_oauth_token_on_retry = false;
            auto wb = create_buffer(true, used_cached_oauth_token_on_retry);

            String response_str;
            if (!ignore_result)
                readJSONObjectPossiblyInvalid(response_str, *wb);
            else
                wb->ignoreAll();
        }
        else
        {
            throw;
        }
    }
}

void RestCatalog::createNamespaceIfNotExists(const String & namespace_name, const String & location) const
{
    const std::string endpoint = base_url / config.prefix / "namespaces";

    Poco::JSON::Object::Ptr request_body = new Poco::JSON::Object;
    {
        Poco::JSON::Array::Ptr namespaces = new Poco::JSON::Array;
        namespaces->add(namespace_name);
        request_body->set("namespace", namespaces);
    }
    {
        Poco::JSON::Object::Ptr properties = new Poco::JSON::Object;
        properties->set("location", location);
        request_body->set("properties", properties);
    }

    try
    {
        ProfileEvents::increment(ProfileEvents::DataLakeRestCatalogCreateNamespace);
        auto timer = DB::CurrentThread::getProfileEvents().timer(ProfileEvents::DataLakeRestCatalogCreateNamespaceMicroseconds);
        sendRequest(endpoint, request_body);
    }
    catch (...)
    {
        DB::tryLogCurrentException(log);
    }
}

void RestCatalog::createTable(const String & namespace_name, const String & table_name, const String & /*new_metadata_path*/, Poco::JSON::Object::Ptr metadata_content) const
{
    if (!allowed_namespaces.isNamespaceAllowed(namespace_name, /*nested*/ false))
        throw DB::Exception(DB::ErrorCodes::CATALOG_NAMESPACE_DISABLED,
            "Failed to create table {}, namespace {} is filtered by `namespaces` database parameter", table_name, namespace_name);

    createNamespaceIfNotExists(namespace_name, metadata_content->getValue<String>("location"));

    const std::string endpoint = base_url / config.prefix / "namespaces" / namespace_name / "tables";

    Poco::JSON::Object::Ptr request_body = new Poco::JSON::Object;
    request_body->set("name", table_name);
    request_body->set("location", metadata_content->getValue<String>("location"));
    {
        Poco::JSON::Object::Ptr initial_schema = metadata_content->getArray("schemas")->getObject(0);
        Poco::JSON::Array::Ptr identifier_fields = new Poco::JSON::Array;
        initial_schema->set("identifier-field-ids", identifier_fields);
        request_body->set("schema", initial_schema);
    }
    request_body->set("partition-spec", metadata_content->getArray("partition-specs")->get(0));

    {
        Poco::JSON::Object::Ptr write_order = new Poco::JSON::Object;
        write_order->set("order-id", 0);
        Poco::JSON::Array::Ptr fields = new Poco::JSON::Array;
        write_order->set("fields", fields);
        request_body->set("write-order", write_order);
    }
    request_body->set("stage-create", false);
    Poco::JSON::Object::Ptr properties = new Poco::JSON::Object;
    request_body->set("properties", properties);

    try
    {
        ProfileEvents::increment(ProfileEvents::DataLakeRestCatalogCreateTable);
        auto timer = DB::CurrentThread::getProfileEvents().timer(ProfileEvents::DataLakeRestCatalogCreateTableMicroseconds);
        sendRequest(endpoint, request_body);
    }
    catch (const DB::HTTPException & ex)
    {
        throw DB::Exception(DB::ErrorCodes::DATALAKE_DATABASE_ERROR, "Failed to create table {}", ex.displayText());
    }
}


bool RestCatalog::updateMetadata(const String & namespace_name, const String & table_name, const String & /*new_metadata_path*/, Poco::JSON::Object::Ptr new_snapshot) const
{
    const std::string endpoint = base_url / config.prefix / "namespaces" / namespace_name / "tables" / table_name;

    Poco::JSON::Object::Ptr request_body = new Poco::JSON::Object;
    {
        Poco::JSON::Object::Ptr identifier = new Poco::JSON::Object;
        identifier->set("name", table_name);
        Poco::JSON::Array::Ptr namespaces = new Poco::JSON::Array;
        namespaces->add(namespace_name);
        identifier->set("namespace", namespaces);

        request_body->set("identifier", identifier);
    }

    if (new_snapshot->has("parent-snapshot-id"))
    {
        auto parent_snapshot_id = new_snapshot->getValue<Int64>("parent-snapshot-id");
        if (parent_snapshot_id != -1)
        {
            Poco::JSON::Object::Ptr requirement = new Poco::JSON::Object;
            requirement->set("type", "assert-ref-snapshot-id");
            requirement->set("ref", "main");
            requirement->set("snapshot-id", parent_snapshot_id);

            Poco::JSON::Array::Ptr requirements = new Poco::JSON::Array;
            requirements->add(requirement);

            request_body->set("requirements", requirements);
        }
    }

    {
        Poco::JSON::Array::Ptr updates = new Poco::JSON::Array;

        {
            Poco::JSON::Object::Ptr add_snapshot = new Poco::JSON::Object;
            add_snapshot->set("action", "add-snapshot");
            add_snapshot->set("snapshot", new_snapshot);
            updates->add(add_snapshot);
        }

        {
            Poco::JSON::Object::Ptr set_snapshot = new Poco::JSON::Object;
            set_snapshot->set("action", "set-snapshot-ref");
            set_snapshot->set("ref-name", "main");
            set_snapshot->set("type", "branch");
            set_snapshot->set("snapshot-id", new_snapshot->getValue<Int64>("snapshot-id"));

            updates->add(set_snapshot);
        }
        request_body->set("updates", updates);
    }

    try
    {
        ProfileEvents::increment(ProfileEvents::DataLakeRestCatalogUpdateTable);
        auto timer = DB::CurrentThread::getProfileEvents().timer(ProfileEvents::DataLakeRestCatalogUpdateTableMicroseconds);
        sendRequest(endpoint, request_body);
    }
    catch (const DB::HTTPException &)
    {
        return false;
    }
    return true;
}

void RestCatalog::dropTable(const String & namespace_name, const String & table_name) const
{
    if (!allowed_namespaces.isNamespaceAllowed(namespace_name, /*nested*/ false))
        throw DB::Exception(DB::ErrorCodes::CATALOG_NAMESPACE_DISABLED,
            "Failed to drop table {}, namespace {} is filtered by `namespaces` database parameter",
            table_name, namespace_name);

    const std::string endpoint
        = (base_url / config.prefix / "namespaces" / namespace_name / "tables" / table_name).string()
        + "?purgeRequested=False";

    Poco::JSON::Object::Ptr request_body = nullptr;
    try
    {
        ProfileEvents::increment(ProfileEvents::DataLakeRestCatalogDropTable);
        auto timer = DB::CurrentThread::getProfileEvents().timer(ProfileEvents::DataLakeRestCatalogDropTableMicroseconds);
        sendRequest(endpoint, request_body, Poco::Net::HTTPRequest::HTTP_DELETE, true);
    }
    catch (const DB::HTTPException & ex)
    {
        throw DB::Exception(DB::ErrorCodes::DATALAKE_DATABASE_ERROR, "Failed to drop table {}", ex.displayText());
    }
}

namespace
{
std::optional<std::chrono::system_clock::time_point> parseSasTokenExpiry(const std::string & sas_token)
{
    std::string token = sas_token;
    if (!token.empty() && token.front() == '?')
        token.erase(0, 1);

    Poco::StringTokenizer params(token, "&", Poco::StringTokenizer::TOK_IGNORE_EMPTY | Poco::StringTokenizer::TOK_TRIM);
    for (const auto & param : params)
    {
        if (!param.starts_with("se="))
            continue;

        try
        {
            std::string decoded;
            Poco::URI::decode(param.substr(3), decoded);

            int time_zone_differential = 0;
            Poco::DateTime date_time;
            if (Poco::DateTimeParser::tryParse(Poco::DateTimeFormat::ISO8601_FORMAT, decoded, date_time, time_zone_differential))
            {
                date_time.makeUTC(time_zone_differential);
                return std::chrono::system_clock::from_time_t(date_time.timestamp().epochTime());
            }
        }
        catch (...)
        {
            return std::nullopt;
        }
        return std::nullopt;
    }
    return std::nullopt;
}
}

VendedStorageCredentials RestCatalog::getCredentialsAndEndpoint(Poco::JSON::Object::Ptr object, const String & location) const
{
    auto storage_type = parseStorageTypeFromLocation(location);
    switch (storage_type)
    {
        case StorageType::S3:
        {
            static constexpr auto access_key_id_str = "s3.access-key-id";
            static constexpr auto secret_access_key_str = "s3.secret-access-key";
            static constexpr auto session_token_str = "s3.session-token";
            static constexpr auto storage_endpoint_str = "s3.endpoint";
            static constexpr auto session_token_expires_at_ms_str = "s3.session-token-expires-at-ms";

            std::string access_key_id;
            std::string secret_access_key;
            std::string session_token;
            std::string storage_endpoint;
            std::optional<std::chrono::system_clock::time_point> expires_at;
            if (object->has(access_key_id_str))
                access_key_id = object->get(access_key_id_str).extract<String>();
            if (object->has(secret_access_key_str))
                secret_access_key = object->get(secret_access_key_str).extract<String>();
            if (object->has(session_token_str))
                session_token = object->get(session_token_str).extract<String>();
            if (object->has(storage_endpoint_str))
                storage_endpoint = object->get(storage_endpoint_str).extract<String>();
            if (object->has(session_token_expires_at_ms_str))
            {
                try
                {
                    static constexpr Int64 max_representable_sec
                        = std::chrono::duration_cast<std::chrono::seconds>(std::chrono::system_clock::duration::max()).count();
                    const Int64 expires_at_ms = object->get(session_token_expires_at_ms_str).convert<Int64>();
                    if (expires_at_ms <= 0)
                        expires_at = std::chrono::system_clock::time_point{}; /// Already invalid: do not cache.
                    else if (expires_at_ms / 1000 < max_representable_sec)
                        expires_at = std::chrono::system_clock::from_time_t(static_cast<std::time_t>(expires_at_ms / 1000));
                }
                catch (...)
                {
                    LOG_DEBUG(log, "Failed to parse '{}' from vended credentials config", session_token_expires_at_ms_str);
                }
            }

            LOG_DEBUG(log, "get tokens for location {}", location);
            return {std::make_shared<S3Credentials>(access_key_id, secret_access_key, session_token), storage_endpoint, expires_at};
        }
        case StorageType::Azure:
        {
            /// Azure ADLS Gen2 vended credentials use SAS tokens.
            /// The config keys follow the pattern: adls.sas-token.<account_name>
            /// or adls.sas-token.<account_name>.dfs.core.windows.net
            /// We look for any key starting with "adls.sas-token." and use the first one found.
            String sas_token;
            std::vector<std::string> names;
            object->getNames(names);
            for (const auto & name : names)
            {
                if (name.starts_with("adls.sas-token."))
                {
                    sas_token = object->get(name).extract<String>();
                    LOG_DEBUG(log, "Found Azure SAS token with key: {}", name);
                    break;
                }
            }

            if (!sas_token.empty())
                return {std::make_shared<AzureCredentials>(sas_token), "", parseSasTokenExpiry(sas_token)};
            break;
        }
        default:
            break;
    }
    return {nullptr, "", std::nullopt};
}

std::optional<VendedStorageCredentials> RestCatalog::tryGetCachedCredentials(
    const std::string & namespace_name, const std::string & table_name) const
{
    if (vended_credentials_cache_ttl.load(std::memory_order_relaxed) <= std::chrono::seconds::zero())
        return std::nullopt;

    std::lock_guard lock(credentials_cache_mutex);
    auto it = credentials_cache.find({namespace_name, table_name});
    if (it == credentials_cache.end())
        return std::nullopt;
    if (std::chrono::system_clock::now() >= it->second.expires_at.value())
    {
        credentials_cache.erase(it); /// Drop the stale entry.
        return std::nullopt;
    }

    return it->second;
}

void RestCatalog::cacheCredentials(
    const std::string & namespace_name, const std::string & table_name, const VendedStorageCredentials & parsed) const
{
    const auto ttl = vended_credentials_cache_ttl.load(std::memory_order_relaxed);
    if (ttl <= std::chrono::seconds::zero())
        return;

    if (!parsed.credentials || parsed.credentials->isEmpty())
        return;

    const auto now = std::chrono::system_clock::now();

    /// Cap at the configured TTL so an entry never outlives the documented maximum lifetime.
    auto refresh_after = now + ttl;
    if (parsed.expires_at && parsed.expires_at.value() < refresh_after)
        refresh_after = parsed.expires_at.value();
    if (refresh_after <= now)
        return;

    std::lock_guard lock(credentials_cache_mutex);

    if (credentials_cache.size() >= credentials_cache_cleanup_threshold)
        std::erase_if(credentials_cache, [&now](const auto & entry) { return now >= entry.second.expires_at.value(); });
    credentials_cache[{namespace_name, table_name}]
        = VendedStorageCredentials{parsed.credentials, parsed.endpoint, refresh_after};
}

ICatalog::CredentialsRefreshCallback RestCatalog::getCredentialsConfigurationCallback(const DB::StorageID & storage_id)
{
    return [this, storage_id] () -> std::shared_ptr<IStorageCredentials>
    {
        LOG_DEBUG(log, "Update credentials in the catalog");

        DB::HTTPHeaderEntries headers;
        headers.emplace_back("X-Iceberg-Access-Delegation", "vended-credentials");

        const auto & table = storage_id.getTableName();
        auto [namespace_name, table_name] = DataLake::parseTableName(table);
        const std::string endpoint = std::filesystem::path(NAMESPACES_ENDPOINT) / encodeNamespaceForURI(namespace_name) / "tables" / table_name;
        String json_str;

        {
            ProfileEvents::increment(ProfileEvents::DataLakeRestCatalogGetCredentials);
            auto timer = DB::CurrentThread::getProfileEvents().timer(ProfileEvents::DataLakeRestCatalogGetCredentialsMicroseconds);
            auto buf = createReadBuffer(config.prefix / endpoint, /* params */{}, headers);

            if (buf->eof())
            {
                LOG_DEBUG(log, "Table doesn't exist (endpoint: {})", endpoint);
                return nullptr;
            }

            readJSONObjectPossiblyInvalid(json_str, *buf);
        }

        Poco::JSON::Parser parser;
        Poco::Dynamic::Var json = parser.parse(json_str);
        const Poco::JSON::Object::Ptr & object = json.extract<Poco::JSON::Object::Ptr>();

        if (!object->has("config"))
        {
            LOG_DEBUG(log, "No 'config' in response for table {} – catalog does not support credential vending", table_name);
            return nullptr;
        }

        auto config_object = object->get("config").extract<Poco::JSON::Object::Ptr>();
        if (!config_object)
        {
            LOG_DEBUG(log, "Empty 'config' in response for table {}", table_name);
            return nullptr;
        }

        std::string location;
        if (object->has("metadata-location"))
        {
            location = object->get("metadata-location").extract<String>();
            LOG_DEBUG(log, "Location for table {}: {}", table_name, location);
        }
        else
        {
            throw DB::Exception(DB::ErrorCodes::BAD_ARGUMENTS, "Cannot read table {}, because no 'metadata-location' in response", table_name);
        }

        auto parsed = getCredentialsAndEndpoint(config_object, location);
        /// Refresh the per-table cache so subsequent queries reuse these freshly vended credentials.
        cacheCredentials(namespace_name, table_name, parsed);
        return parsed.credentials;
    };
}

/// "alpha,alpha.a1,bravo,bravo.*,charlie,delta.d1,echo.*"
/// allows tables from
/// - "alpha" namespace
/// - "alpha.a1" namespace
/// - "bravo" namespace
/// - any nested namespaces of "bravo"
/// - "charlie" namespace, but not from nested of "charlie"
/// - "delta.d1" namespace, but not from "delta"
/// - any nested namespaces of "echo", but not "echo" itself
/// "bravo.*.b2" makes no sense for now, asterisk allows all nested
RestCatalog::AllowedNamespaces::AllowedNamespaces(const std::string & namespaces_)
{
    std::vector<std::string> list_of_namespaces;
    boost::split(list_of_namespaces, namespaces_, boost::is_any_of(", "), boost::token_compress_on);
    for (const auto & ns : list_of_namespaces)
    {
        std::vector<std::string> list_of_nested_namespaces;
        boost::split(list_of_nested_namespaces, ns, boost::is_any_of("."));

        size_t len = list_of_nested_namespaces.size();
        if (!len)
            continue;

        AllowedNamespaces * current = &(nested_namespaces[list_of_nested_namespaces[0]]);
        for (size_t i = 1; i <= len; ++i)
        {
            if (i == len)
                current->allow_tables = true;
            else
            {
                current = &(current->nested_namespaces[list_of_nested_namespaces[i]]);
                if (list_of_nested_namespaces[i] == "*")
                {
                    current->allow_tables = true;
                    break;
                }
            }
        }
    }
}

bool RestCatalog::AllowedNamespaces::isNamespaceAllowed(const std::string & namespace_, bool nested) const
{
    // Trivial case, check here to avoid split namespace on nested
    if (nested_namespaces.contains("*"))
        return true;

    std::vector<std::string> list_of_nested_namespaces;
    boost::split(list_of_nested_namespaces, namespace_, boost::is_any_of("."));

    const AllowedNamespaces * current = this;
    for (const auto & nns : list_of_nested_namespaces)
    {
        if (current->nested_namespaces.contains("*"))
            return true;
        auto it = current->nested_namespaces.find(nns);
        if (it == current->nested_namespaces.end())
            return false;
        current = &(it->second);
    }

    return nested ? !current->nested_namespaces.empty() : current->allow_tables;
}

}

#endif