From 61cf4829e47c2da75ea62075a6a9074662d6fddc Mon Sep 17 00:00:00 2001 From: Andrey Zvonov <32552679+zvonand@users.noreply.github.com> Date: Tue, 2 Jun 2026 13:24:19 +0200 Subject: [PATCH 1/4] Cherry-pick of https://github.com/Altinity/ClickHouse/pull/1808 with unresolved conflict markers (resolution in next commit) --- Original cherry-pick message follows: Merge pull request #1808 from Altinity/antalya_26_3_s3_tables Antalya 26.3: S3 tables iceberg support # Conflicts: # src/Databases/DataLake/DatabaseDataLake.cpp # src/Databases/DataLake/RestCatalog.cpp --- src/Core/Settings.cpp | 3 + src/Core/SettingsChangesHistory.cpp | 1 + src/Core/SettingsEnums.cpp | 3 +- src/Core/SettingsEnums.h | 1 + src/Databases/DataLake/AWSV4Signer.cpp | 110 ++++++++ src/Databases/DataLake/AWSV4Signer.h | 34 +++ src/Databases/DataLake/DatabaseDataLake.cpp | 58 +++- src/Databases/DataLake/ICatalog.h | 1 - src/Databases/DataLake/RestCatalog.cpp | 48 +++- src/Databases/DataLake/RestCatalog.h | 14 +- src/Databases/DataLake/S3TablesCatalog.cpp | 260 ++++++++++++++++++ src/Databases/DataLake/S3TablesCatalog.h | 66 +++++ .../DataLake/S3TablesCredentialRefresh.cpp | 43 +++ .../DataLake/S3TablesCredentialRefresh.h | 23 ++ src/Databases/DataLake/StorageCredentials.h | 2 + .../tests/gtest_azure_abfss_parsing.cpp | 27 ++ .../gtest_s3tables_credential_refresh.cpp | 112 ++++++++ .../enableAllExperimentalSettings.cpp | 1 + src/IO/S3/URI.cpp | 15 + src/IO/S3/URI.h | 4 + src/IO/S3/tests/gtest_s3_uri.cpp | 19 ++ 21 files changed, 834 insertions(+), 11 deletions(-) create mode 100644 src/Databases/DataLake/AWSV4Signer.cpp create mode 100644 src/Databases/DataLake/AWSV4Signer.h create mode 100644 src/Databases/DataLake/S3TablesCatalog.cpp create mode 100644 src/Databases/DataLake/S3TablesCatalog.h create mode 100644 src/Databases/DataLake/S3TablesCredentialRefresh.cpp create mode 100644 src/Databases/DataLake/S3TablesCredentialRefresh.h create mode 100644 src/Databases/DataLake/tests/gtest_s3tables_credential_refresh.cpp diff --git a/src/Core/Settings.cpp b/src/Core/Settings.cpp index 7f5f7a2a00ff..503215f90e13 100644 --- a/src/Core/Settings.cpp +++ b/src/Core/Settings.cpp @@ -8321,6 +8321,9 @@ Multiple algorithms can be specified as a comma-separated list, e.g. `dphyp,gree )", EXPERIMENTAL) \ DECLARE(Bool, allow_experimental_database_paimon_rest_catalog, false, R"( Allow experimental database engine DataLakeCatalog with catalog_type = 'paimon_rest' +)", EXPERIMENTAL) \ + DECLARE(Bool, allow_experimental_database_s3_tables, false, R"( +Allow experimental database engine DataLakeCatalog with catalog_type = 's3tables' (Amazon S3 Tables Iceberg REST with SigV4) )", EXPERIMENTAL) \ DECLARE(UInt64, webassembly_udf_max_fuel, 100'000, R"( Fuel limit per WebAssembly UDF instance execution. Each WebAssembly instruction consumes some amount of fuel. The value is scaled by 1024 before being passed to the runtime, so `webassembly_udf_max_fuel = 1` corresponds to approximately 1024 fuel units. Set to 0 for no finite limit. Applies only to functions whose per-function setting `webassembly_udf_enable_fuel` is true, which is the default. diff --git a/src/Core/SettingsChangesHistory.cpp b/src/Core/SettingsChangesHistory.cpp index c2b62c209863..5d1dcfa03cf3 100644 --- a/src/Core/SettingsChangesHistory.cpp +++ b/src/Core/SettingsChangesHistory.cpp @@ -229,6 +229,7 @@ const VersionToSettingsChangesMap & getSettingsChangesHistory() {"iceberg_expire_default_max_ref_age_ms", 9223372036854775807, 9223372036854775807, "New setting."}, {"max_skip_unavailable_shards_num", 0, 0, "New setting to limit the number of shards that can be silently skipped when skip_unavailable_shards is enabled."}, {"max_skip_unavailable_shards_ratio", 0, 0, "New setting to limit the ratio of shards that can be silently skipped when skip_unavailable_shards is enabled."}, + {"allow_experimental_database_s3_tables", false, false, "New setting to enable experimental database S3 tables (AWS Iceberg REST catalog)."}, }); addSettingsChanges(settings_changes_history, "26.2", { diff --git a/src/Core/SettingsEnums.cpp b/src/Core/SettingsEnums.cpp index 0ff8daf7faa0..d498bd5a9db6 100644 --- a/src/Core/SettingsEnums.cpp +++ b/src/Core/SettingsEnums.cpp @@ -359,7 +359,8 @@ IMPLEMENT_SETTING_ENUM( {"hive", DatabaseDataLakeCatalogType::ICEBERG_HIVE}, {"onelake", DatabaseDataLakeCatalogType::ICEBERG_ONELAKE}, {"biglake", DatabaseDataLakeCatalogType::ICEBERG_BIGLAKE}, - {"paimon_rest", DatabaseDataLakeCatalogType::PAIMON_REST}}) + {"paimon_rest", DatabaseDataLakeCatalogType::PAIMON_REST}, + {"s3tables", DatabaseDataLakeCatalogType::S3_TABLES}}) IMPLEMENT_SETTING_ENUM( FileCachePolicy, diff --git a/src/Core/SettingsEnums.h b/src/Core/SettingsEnums.h index 09d175e72680..e6a9b2b6a53b 100644 --- a/src/Core/SettingsEnums.h +++ b/src/Core/SettingsEnums.h @@ -442,6 +442,7 @@ enum class DatabaseDataLakeCatalogType : uint8_t ICEBERG_ONELAKE, ICEBERG_BIGLAKE, PAIMON_REST, + S3_TABLES, }; DECLARE_SETTING_ENUM(DatabaseDataLakeCatalogType) diff --git a/src/Databases/DataLake/AWSV4Signer.cpp b/src/Databases/DataLake/AWSV4Signer.cpp new file mode 100644 index 000000000000..f5bd8c4bc6ed --- /dev/null +++ b/src/Databases/DataLake/AWSV4Signer.cpp @@ -0,0 +1,110 @@ +#include "config.h" + +#if USE_AVRO && USE_SSL && USE_AWS_S3 + +#include + +#include +#include +#include + +#include +#include +#include +#include + +#include +#include + +namespace DB +{ +namespace ErrorCodes +{ + extern const int BAD_ARGUMENTS; + extern const int S3_ERROR; +} +} + +namespace DataLake +{ +namespace +{ + +Aws::Http::HttpMethod mapPocoMethodToAws(const String & method) +{ + using Aws::Http::HttpMethod; + using Poco::Net::HTTPRequest; + + static const std::pair supported_methods[] = { + {HTTPRequest::HTTP_GET, HttpMethod::HTTP_GET}, + {HTTPRequest::HTTP_POST, HttpMethod::HTTP_POST}, + {HTTPRequest::HTTP_PUT, HttpMethod::HTTP_PUT}, + {HTTPRequest::HTTP_DELETE, HttpMethod::HTTP_DELETE}, + {HTTPRequest::HTTP_HEAD, HttpMethod::HTTP_HEAD}, + {HTTPRequest::HTTP_PATCH, HttpMethod::HTTP_PATCH}, + }; + + for (const auto & [poco_method, aws_method] : supported_methods) + if (method == poco_method) + return aws_method; + + throw DB::Exception(DB::ErrorCodes::BAD_ARGUMENTS, "Unsupported HTTP method for AWS SigV4 signing: {}", method); +} + +} + +void signRequestWithAWSV4( + const String & method, + const Poco::URI & uri, + const DB::HTTPHeaderEntries & extra_headers, + const String & payload, + Aws::Client::AWSAuthV4Signer & signer, + const String & region, + const String & service, + DB::HTTPHeaderEntries & out_headers) +{ + const Aws::Http::URI aws_uri(uri.toString().c_str()); + Aws::Http::Standard::StandardHttpRequest request(aws_uri, mapPocoMethodToAws(method)); + + for (const auto & h : extra_headers) + { + if (Poco::icompare(h.name, "authorization") == 0) + continue; + request.SetHeaderValue(Aws::String(h.name.c_str(), h.name.size()), Aws::String(h.value.c_str(), h.value.size())); + } + + if (!payload.empty()) + { + auto body_stream = Aws::MakeShared("AWSV4Signer"); + body_stream->write(payload.data(), static_cast(payload.size())); + body_stream->seekg(0); + request.AddContentBody(body_stream); + } + + static constexpr bool sign_body = true; + if (!signer.SignRequest(request, region.c_str(), service.c_str(), sign_body)) + throw DB::Exception(DB::ErrorCodes::S3_ERROR, "AWS SigV4 signing failed"); + + bool has_authorization = false; + for (const auto & [key, value] : request.GetHeaders()) + { + if (Poco::icompare(key, "authorization") == 0 && !value.empty()) + has_authorization = true; + } + if (!has_authorization) + throw DB::Exception( + DB::ErrorCodes::BAD_ARGUMENTS, + "AWS credentials are missing or incomplete; cannot sign S3 Tables REST request"); + + out_headers.clear(); + for (const auto & [key, value] : request.GetHeaders()) + { + if (Poco::icompare(key, "host") == 0) + continue; + out_headers.emplace_back(String(key.c_str(), key.size()), String(value.c_str(), value.size())); + } +} + +} + +#endif diff --git a/src/Databases/DataLake/AWSV4Signer.h b/src/Databases/DataLake/AWSV4Signer.h new file mode 100644 index 000000000000..cdc42adaca5f --- /dev/null +++ b/src/Databases/DataLake/AWSV4Signer.h @@ -0,0 +1,34 @@ +#pragma once + +#include "config.h" + +#if USE_AVRO && USE_SSL && USE_AWS_S3 + +#include +#include +#include + +namespace Aws::Client +{ +class AWSAuthV4Signer; +} + +namespace DataLake +{ + +/// Sign a Poco-style HTTP request using the AWS SDK's AWSAuthV4Signer. +/// Builds a temporary Aws::Http::StandardHttpRequest, signs it, then extracts +/// the resulting headers into out_headers (excluding Host; ReadWriteBufferFromHTTP sets it from the URI). +void signRequestWithAWSV4( + const String & method, + const Poco::URI & uri, + const DB::HTTPHeaderEntries & extra_headers, + const String & payload, + Aws::Client::AWSAuthV4Signer & signer, + const String & region, + const String & service, + DB::HTTPHeaderEntries & out_headers); + +} + +#endif diff --git a/src/Databases/DataLake/DatabaseDataLake.cpp b/src/Databases/DataLake/DatabaseDataLake.cpp index 12fbb051ba4a..f8492a2f9766 100644 --- a/src/Databases/DataLake/DatabaseDataLake.cpp +++ b/src/Databases/DataLake/DatabaseDataLake.cpp @@ -27,6 +27,9 @@ #include #include #include +#if USE_AWS_S3 && USE_SSL +#include +#endif #include #include @@ -93,6 +96,7 @@ namespace Setting extern const SettingsBool allow_experimental_database_glue_catalog; extern const SettingsBool allow_experimental_database_hms_catalog; extern const SettingsBool allow_experimental_database_paimon_rest_catalog; + extern const SettingsBool allow_experimental_database_s3_tables; extern const SettingsBool use_hive_partitioning; extern const SettingsBool log_queries; extern const SettingsBool parallel_replicas_for_cluster_engines; @@ -155,8 +159,20 @@ void DatabaseDataLake::validateSettings() { if (settings[DatabaseDataLakeSetting::region].value.empty()) throw Exception( - ErrorCodes::BAD_ARGUMENTS, "`region` setting cannot be empty for Glue Catalog. " + ErrorCodes::BAD_ARGUMENTS, "`region` setting cannot be empty for Glue catalog. " + "Please specify 'SETTINGS region=' in the CREATE DATABASE query"); + } + else if (settings[DatabaseDataLakeSetting::catalog_type].value == DB::DatabaseDataLakeCatalogType::S3_TABLES) + { + if (settings[DatabaseDataLakeSetting::region].value.empty()) + throw Exception( + ErrorCodes::BAD_ARGUMENTS, "`region` setting cannot be empty for S3 Tables catalog. " "Please specify 'SETTINGS region=' in the CREATE DATABASE query"); + + if (settings[DatabaseDataLakeSetting::warehouse].value.empty()) + throw Exception( + ErrorCodes::BAD_ARGUMENTS, "`warehouse` setting cannot be empty for S3 Tables catalog. " + "Please specify 'SETTINGS warehouse=' in the CREATE DATABASE query"); } else if (settings[DatabaseDataLakeSetting::warehouse].value.empty()) { @@ -310,6 +326,23 @@ void DatabaseDataLake::initialize() const } break; } + case DB::DatabaseDataLakeCatalogType::S3_TABLES: + { +#if USE_AWS_S3 && USE_SSL + catalog_impl = std::make_shared( + settings[DatabaseDataLakeSetting::warehouse].value, + url, + settings[DatabaseDataLakeSetting::region].value, + catalog_parameters, + settings[DatabaseDataLakeSetting::namespaces].value, + Context::getGlobalContextInstance()); +#else + throw Exception( + ErrorCodes::SUPPORT_IS_DISABLED, + "Amazon S3 Tables catalog requires ClickHouse built with USE_AWS_S3 and USE_SSL"); +#endif + break; + } } } @@ -350,6 +383,7 @@ std::shared_ptr DatabaseDataLake::getConfigur case DatabaseDataLakeCatalogType::ICEBERG_HIVE: case DatabaseDataLakeCatalogType::ICEBERG_REST: case DatabaseDataLakeCatalogType::ICEBERG_BIGLAKE: + case DatabaseDataLakeCatalogType::S3_TABLES: { switch (type) { @@ -1055,6 +1089,15 @@ void registerDatabaseDataLake(DatabaseFactory & factory) throw Exception(ErrorCodes::BAD_ARGUMENTS, "Engine `{}` must have arguments", database_engine_name); } +<<<<<<< HEAD +======= + if (database_engine_name == "Iceberg" && catalog_type != DatabaseDataLakeCatalogType::ICEBERG_REST + && catalog_type != DatabaseDataLakeCatalogType::S3_TABLES) + { + throw Exception(ErrorCodes::BAD_ARGUMENTS, "Engine `Iceberg` must use `rest` or `s3tables` catalog type only"); + } + +>>>>>>> 7cd07274788 (Merge pull request #1808 from Altinity/antalya_26_3_s3_tables) for (auto & engine_arg : engine_args) engine_arg = evaluateConstantExpressionOrIdentifierAsLiteral(engine_arg, args.context); @@ -1138,6 +1181,19 @@ void registerDatabaseDataLake(DatabaseFactory & factory) engine_func->name = "Paimon"; break; } + case DatabaseDataLakeCatalogType::S3_TABLES: + { + if (!args.create_query.attach + && !args.context->getSettingsRef()[Setting::allow_experimental_database_s3_tables]) + { + throw Exception(ErrorCodes::SUPPORT_IS_DISABLED, + "DatabaseDataLake with S3 Tables catalog is experimental. " + "To allow its usage, enable setting allow_experimental_database_s3_tables"); + } + + engine_func->name = "Iceberg"; + break; + } case DatabaseDataLakeCatalogType::NONE: break; } diff --git a/src/Databases/DataLake/ICatalog.h b/src/Databases/DataLake/ICatalog.h index e14b00ac3732..8211ba7aa39c 100644 --- a/src/Databases/DataLake/ICatalog.h +++ b/src/Databases/DataLake/ICatalog.h @@ -38,7 +38,6 @@ class TableMetadata bool hasLocation() const; bool hasSchema() const; bool hasStorageCredentials() const; - bool hasDataLakeSpecificProperties() const; void setLocation(const std::string & location_); std::string getLocation() const; diff --git a/src/Databases/DataLake/RestCatalog.cpp b/src/Databases/DataLake/RestCatalog.cpp index 9f85e9d80d6d..b253705b4d80 100644 --- a/src/Databases/DataLake/RestCatalog.cpp +++ b/src/Databases/DataLake/RestCatalog.cpp @@ -255,7 +255,12 @@ void RestCatalog::parseCatalogConfigurationSettings(const Poco::JSON::Object::Pt result.default_base_location = object->get("default-base-location").extract(); } -DB::HTTPHeaderEntries RestCatalog::getAuthHeaders(bool update_token) const +DB::HTTPHeaderEntries RestCatalog::getAuthHeaders( + bool update_token, + const String & /*method*/, + const Poco::URI & /*url*/, + const DB::HTTPHeaderEntries & /*extra_headers*/, + const String & /*body*/) const { fiu_do_on(DB::FailPoints::check_database_datalake_negative, { @@ -428,7 +433,12 @@ BigLakeCatalog::BigLakeCatalog( config = loadConfig(); } -DB::HTTPHeaderEntries BigLakeCatalog::getAuthHeaders(bool update_token) const +DB::HTTPHeaderEntries BigLakeCatalog::getAuthHeaders( + bool update_token, + const String & /*method*/, + const Poco::URI & /*url*/, + const DB::HTTPHeaderEntries & /*extra_headers*/, + const String & /*body*/) const { /// Google Cloud OAuth2 for BigLake. /// Uses GCP metadata service or Application Default Credentials to get access token. @@ -602,7 +612,7 @@ DB::ReadWriteBufferFromHTTPPtr RestCatalog::createReadBuffer( auto create_buffer = [&](bool update_token) { - auto result_headers = getAuthHeaders(update_token); + auto result_headers = getAuthHeaders(update_token, Poco::Net::HTTPRequest::HTTP_GET, url, headers, {}); std::move(headers.begin(), headers.end(), std::back_inserter(result_headers)); return DB::BuilderRWBufferFromHTTP(url) @@ -1135,9 +1145,6 @@ void RestCatalog::sendRequest(const String & endpoint, Poco::JSON::Object::Ptr r request_body->stringify(oss); const std::string body_str = DB::removeEscapedSlashes(oss.str()); - DB::HTTPHeaderEntries headers = getAuthHeaders(/* update_token = */ true); - headers.emplace_back("Content-Type", "application/json"); - const auto & context = getContext(); DB::ReadWriteBufferFromHTTP::OutStreamCallback out_stream_callback; @@ -1151,6 +1158,12 @@ void RestCatalog::sendRequest(const String & endpoint, Poco::JSON::Object::Ptr r /// 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"); + + DB::HTTPHeaderEntries headers = getAuthHeaders(/* update_token = */ true, method, url, extra_headers, body_str); + headers.emplace_back("Content-Type", "application/json"); auto wb = DB::BuilderRWBufferFromHTTP(url) .withConnectionGroup(DB::HTTPConnectionGroupType::HTTP) .withMethod(method) @@ -1171,7 +1184,11 @@ void RestCatalog::sendRequest(const String & endpoint, Poco::JSON::Object::Ptr r void RestCatalog::createNamespaceIfNotExists(const String & namespace_name, const String & location) const { +<<<<<<< HEAD const std::string endpoint = (base_url / config.prefix / NAMESPACES_ENDPOINT).generic_string(); +======= + const std::string endpoint = base_url / config.prefix / "namespaces"; +>>>>>>> 7cd07274788 (Merge pull request #1808 from Altinity/antalya_26_3_s3_tables) Poco::JSON::Object::Ptr request_body = new Poco::JSON::Object; { @@ -1199,7 +1216,11 @@ void RestCatalog::createTable(const String & namespace_name, const String & tabl { createNamespaceIfNotExists(namespace_name, metadata_content->getValue("location")); +<<<<<<< HEAD const std::string endpoint = (base_url / config.prefix / NAMESPACES_ENDPOINT / encodeNamespaceForURI(namespace_name) / "tables").generic_string(); +======= + const std::string endpoint = base_url / config.prefix / "namespaces" / namespace_name / "tables"; +>>>>>>> 7cd07274788 (Merge pull request #1808 from Altinity/antalya_26_3_s3_tables) Poco::JSON::Object::Ptr request_body = new Poco::JSON::Object; request_body->set("name", table_name); @@ -1240,7 +1261,11 @@ void RestCatalog::createTable(const String & namespace_name, const String & tabl bool RestCatalog::updateMetadata(const String & namespace_name, const String & table_name, const String & /*new_metadata_path*/, Poco::JSON::Object::Ptr new_snapshot) const { +<<<<<<< HEAD const std::string endpoint = (base_url / config.prefix / NAMESPACES_ENDPOINT / encodeNamespaceForURI(namespace_name) / "tables" / table_name).generic_string(); +======= + const std::string endpoint = base_url / config.prefix / "namespaces" / namespace_name / "tables" / table_name; +>>>>>>> 7cd07274788 (Merge pull request #1808 from Altinity/antalya_26_3_s3_tables) Poco::JSON::Object::Ptr request_body = new Poco::JSON::Object; { @@ -1368,7 +1393,18 @@ bool RestCatalog::updateSchema( void RestCatalog::dropTable(const String & namespace_name, const String & table_name) const { +<<<<<<< HEAD const std::string endpoint = fmt::format("{}/namespaces/{}/tables/{}?purgeRequested=False", base_url, namespace_name, table_name); +======= + 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"; +>>>>>>> 7cd07274788 (Merge pull request #1808 from Altinity/antalya_26_3_s3_tables) Poco::JSON::Object::Ptr request_body = nullptr; try diff --git a/src/Databases/DataLake/RestCatalog.h b/src/Databases/DataLake/RestCatalog.h index 982475ee2c96..90c481a899b1 100644 --- a/src/Databases/DataLake/RestCatalog.h +++ b/src/Databases/DataLake/RestCatalog.h @@ -163,7 +163,12 @@ class RestCatalog : public ICatalog, public DB::WithContext TableMetadata & result) const; Config loadConfig(); - virtual DB::HTTPHeaderEntries getAuthHeaders(bool update_token) const; + virtual DB::HTTPHeaderEntries getAuthHeaders( + bool update_token, + const String & method = {}, + const Poco::URI & url = {}, + const DB::HTTPHeaderEntries & extra_headers = {}, + const String & body = {}) const; static void parseCatalogConfigurationSettings(const Poco::JSON::Object::Ptr & object, Config & result); void sendRequest( @@ -223,7 +228,12 @@ class BigLakeCatalog : public RestCatalog return DB::DatabaseDataLakeCatalogType::ICEBERG_BIGLAKE; } - DB::HTTPHeaderEntries getAuthHeaders(bool update_token) const override; + DB::HTTPHeaderEntries getAuthHeaders( + bool update_token, + const String & method = {}, + const Poco::URI & url = {}, + const DB::HTTPHeaderEntries & extra_headers = {}, + const String & body = {}) const override; const std::string & getGoogleADCClientId() const { return google_adc_client_id; } const std::string & getGoogleADCClientSecret() const { return google_adc_client_secret; } diff --git a/src/Databases/DataLake/S3TablesCatalog.cpp b/src/Databases/DataLake/S3TablesCatalog.cpp new file mode 100644 index 000000000000..bf0be97754e5 --- /dev/null +++ b/src/Databases/DataLake/S3TablesCatalog.cpp @@ -0,0 +1,260 @@ +#include "config.h" + +#if USE_AVRO && USE_SSL && USE_AWS_S3 + +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include + +#include + +namespace DB::ErrorCodes +{ + extern const int BAD_ARGUMENTS; + extern const int DATALAKE_DATABASE_ERROR; + extern const int CATALOG_NAMESPACE_DISABLED; +} + +namespace DB::Setting +{ + extern const SettingsUInt64 s3_max_redirects; + extern const SettingsUInt64 s3_retry_attempts; + extern const SettingsBool s3_slow_all_threads_after_network_error; + extern const SettingsBool enable_s3_requests_logging; +} + +namespace DB::ServerSetting +{ + extern const ServerSettingsUInt64 s3_max_redirects; + extern const ServerSettingsUInt64 s3_retry_attempts; +} + +namespace DataLake +{ + +S3TablesCatalog::S3TablesCatalog( + const String & warehouse_, + const String & base_url_, + const String & region_, + const CatalogSettings & catalog_settings_, + const String & namespaces_, + DB::ContextPtr context_) + : RestCatalog(warehouse_, base_url_, "", "", false, namespaces_, context_) + , region(region_) + , storage_endpoint(catalog_settings_.storage_endpoint) +{ + if (region.empty()) + throw DB::Exception(DB::ErrorCodes::BAD_ARGUMENTS, "S3 Tables catalog requires non-empty `region` setting"); + + DB::S3::CredentialsConfiguration creds_config; + creds_config.use_environment_credentials = true; + creds_config.role_arn = catalog_settings_.aws_role_arn; + creds_config.role_session_name = catalog_settings_.aws_role_session_name; + + const auto & server_settings = getContext()->getGlobalContext()->getServerSettings(); + const DB::Settings & global_settings = getContext()->getGlobalContext()->getSettingsRef(); + + int s3_max_redirects = static_cast(server_settings[DB::ServerSetting::s3_max_redirects]); + if (global_settings.isChanged("s3_max_redirects")) + s3_max_redirects = static_cast(global_settings[DB::Setting::s3_max_redirects]); + + int s3_retry_attempts = static_cast(server_settings[DB::ServerSetting::s3_retry_attempts]); + if (global_settings.isChanged("s3_retry_attempts")) + s3_retry_attempts = static_cast(global_settings[DB::Setting::s3_retry_attempts]); + + bool s3_slow_all_threads_after_network_error = global_settings[DB::Setting::s3_slow_all_threads_after_network_error]; + bool s3_slow_all_threads_after_retryable_error = false; + bool enable_s3_requests_logging = global_settings[DB::Setting::enable_s3_requests_logging]; + + DB::S3::PocoHTTPClientConfiguration poco_config = DB::S3::ClientFactory::instance().createClientConfiguration( + region, + getContext()->getRemoteHostFilter(), + s3_max_redirects, + DB::S3::PocoHTTPClientConfiguration::RetryStrategy{.max_retries = static_cast(s3_retry_attempts)}, + s3_slow_all_threads_after_network_error, + s3_slow_all_threads_after_retryable_error, + enable_s3_requests_logging, + /* for_disk_s3 = */ false, + /* opt_disk_name = */ {}, + /* request_throttler = */ {}); + + Aws::Auth::AWSCredentials credentials(catalog_settings_.aws_access_key_id, catalog_settings_.aws_secret_access_key); + credentials_provider = DB::S3::getCredentialsProvider(poco_config, credentials, creds_config); + + signer = std::make_unique( + credentials_provider, + "s3tables", + Aws::String(region.data(), region.size()), + Aws::Client::AWSAuthV4Signer::PayloadSigningPolicy::Always, + /* urlEscapePath = */ false); + + config = loadConfig(); + + if (config.prefix.empty()) + { + String encoded_warehouse; + Poco::URI::encode(warehouse_, "", encoded_warehouse); + config.prefix = encoded_warehouse; + } +} + +/// S3 Tables only supports a single level of namespaces (no nesting), +/// so we use flat getNamespaces() instead of the base class's getNamespacesRecursive(). +DB::Names S3TablesCatalog::getTables() const +{ + auto namespaces = getNamespaces(""); + + auto & pool = getContext()->getIcebergCatalogThreadpool(); + DB::ThreadPoolCallbackRunnerLocal runner(pool, DB::ThreadName::DATALAKE_REST_CATALOG); + + DB::Names tables; + std::mutex mutex; + for (const auto & ns : namespaces) + { + if (!allowed_namespaces.isNamespaceAllowed(ns, /*nested*/ false)) + continue; + runner.enqueueAndKeepTrack( + [&, ns] + { + auto tables_in_ns = RestCatalog::getTables(ns); + std::lock_guard lock(mutex); + std::move(tables_in_ns.begin(), tables_in_ns.end(), std::back_inserter(tables)); + }); + } + runner.waitForAllToFinishAndRethrowFirstError(); + return tables; +} + +bool S3TablesCatalog::tryGetTableMetadata( + const std::string & namespace_name, + const std::string & table_name, + DB::ContextPtr context_, + TableMetadata & result) const +{ + if (!RestCatalog::tryGetTableMetadata(namespace_name, table_name, context_, result)) + return false; + + if (!result.requiresCredentials()) + return true; + + bool need_credentials = true; + if (const auto storage_credentials = result.getStorageCredentials()) + { + auto creds = std::dynamic_pointer_cast(storage_credentials); + if (creds && !creds->isEmpty()) + need_credentials = false; + } + + if (need_credentials) + { + LOG_DEBUG(log, "S3 Tables: no vended credentials for {}.{}, injecting catalog IAM credentials", namespace_name, table_name); + auto aws_creds = credentials_provider->GetAWSCredentials(); + if (aws_creds.GetAWSAccessKeyId().empty() || aws_creds.GetAWSSecretKey().empty()) + throw DB::Exception( + DB::ErrorCodes::BAD_ARGUMENTS, + "S3 Tables: catalog IAM credentials are empty for {}.{}, " + "check AWS credentials configuration", + namespace_name, table_name); + result.setStorageCredentials(std::make_shared( + aws_creds.GetAWSAccessKeyId(), aws_creds.GetAWSSecretKey(), aws_creds.GetSessionToken())); + } + + if (result.getEndpoint().empty()) + { + String endpoint = storage_endpoint.empty() + ? DB::S3::resolveS3Endpoint(region) + : storage_endpoint; + LOG_DEBUG(log, "S3 Tables: no endpoint for {}.{}, injecting: {}", namespace_name, table_name, endpoint); + result.setEndpoint(endpoint); + } + + return true; +} + +ICatalog::CredentialsRefreshCallback S3TablesCatalog::getCredentialsConfigurationCallback(const DB::StorageID & storage_id) +{ + auto base_cb = RestCatalog::getCredentialsConfigurationCallback(storage_id); + return [this, base_callback = std::move(base_cb)] () -> std::shared_ptr + { + if (base_callback) + { + if (auto creds = (*base_callback)()) + { + auto s3_creds = std::dynamic_pointer_cast(creds); + if (s3_creds && !s3_creds->isEmpty()) + return creds; + } + LOG_DEBUG(log, "S3 Tables: vended credentials unavailable on refresh, falling back to catalog IAM credentials"); + } + + return resolveS3TablesRefreshCredentials(std::nullopt, *credentials_provider); + }; +} + +void S3TablesCatalog::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=True"; + + Poco::JSON::Object::Ptr request_body = nullptr; + try + { + sendRequest(endpoint, request_body, Poco::Net::HTTPRequest::HTTP_DELETE, true); + } + catch (const DB::HTTPException & ex) + { + if (ex.getHTTPStatus() == Poco::Net::HTTPResponse::HTTP_NOT_FOUND) + LOG_DEBUG(log, "S3 Tables: table {}.{} already does not exist (404 on purge-delete)", namespace_name, table_name); + else + throw DB::Exception(DB::ErrorCodes::DATALAKE_DATABASE_ERROR, "Failed to drop table {}", ex.displayText()); + } +} + +DB::HTTPHeaderEntries S3TablesCatalog::getAuthHeaders( + bool /*update_token*/, + const String & method, + const Poco::URI & url, + const DB::HTTPHeaderEntries & extra_headers, + const String & body) const +{ + DB::HTTPHeaderEntries all_signed; + signRequestWithAWSV4(method, url, extra_headers, body, *signer, region, "s3tables", all_signed); + + DB::HTTPHeaderEntries auth_headers; + for (auto & h : all_signed) + { + if (h.name == "authorization" || h.name.starts_with("x-amz-")) + auth_headers.push_back(std::move(h)); + } + return auth_headers; +} + +} + +#endif diff --git a/src/Databases/DataLake/S3TablesCatalog.h b/src/Databases/DataLake/S3TablesCatalog.h new file mode 100644 index 000000000000..45ad049f0199 --- /dev/null +++ b/src/Databases/DataLake/S3TablesCatalog.h @@ -0,0 +1,66 @@ +#pragma once + +#include "config.h" + +#if USE_AVRO && USE_SSL && USE_AWS_S3 + +#include +#include + +#include + +#include + +namespace Aws::Auth +{ +class AWSCredentialsProvider; +} + +namespace DataLake +{ + +/// Iceberg REST catalog for Amazon S3 Tables (SigV4, signing name `s3tables`). +/// https://docs.aws.amazon.com/AmazonS3/latest/userguide/s3-tables-integrating-open-source.html +class S3TablesCatalog final : public RestCatalog +{ +public: + S3TablesCatalog( + const String & warehouse_, + const String & base_url_, + const String & region_, + const DataLake::CatalogSettings & catalog_settings_, + const String & namespaces_, + DB::ContextPtr context_); + + DB::DatabaseDataLakeCatalogType getCatalogType() const override { return DB::DatabaseDataLakeCatalogType::S3_TABLES; } + + DB::Names getTables() const override; + + bool tryGetTableMetadata( + const std::string & namespace_name, + const std::string & table_name, + DB::ContextPtr context_, + TableMetadata & result) const override; + + void dropTable(const String & namespace_name, const String & table_name) const override; + + ICatalog::CredentialsRefreshCallback getCredentialsConfigurationCallback(const DB::StorageID & storage_id) override; + +protected: + DB::HTTPHeaderEntries getAuthHeaders( + bool update_token, + const String & method = {}, + const Poco::URI & url = {}, + const DB::HTTPHeaderEntries & extra_headers = {}, + const String & body = {}) const override; + +private: + const String region; + const String storage_endpoint; + std::shared_ptr credentials_provider; + std::unique_ptr signer; +}; + +} + +#endif diff --git a/src/Databases/DataLake/S3TablesCredentialRefresh.cpp b/src/Databases/DataLake/S3TablesCredentialRefresh.cpp new file mode 100644 index 000000000000..8de09e4d553e --- /dev/null +++ b/src/Databases/DataLake/S3TablesCredentialRefresh.cpp @@ -0,0 +1,43 @@ +#include "config.h" + +#if USE_AVRO && USE_SSL && USE_AWS_S3 + +#include + +namespace DataLake +{ + +namespace +{ + +std::shared_ptr getCatalogIAMCredentials(Aws::Auth::AWSCredentialsProvider & provider) +{ + auto aws_creds = provider.GetAWSCredentials(); + if (aws_creds.GetAWSAccessKeyId().empty() || aws_creds.GetAWSSecretKey().empty()) + return nullptr; + return std::make_shared( + aws_creds.GetAWSAccessKeyId(), aws_creds.GetAWSSecretKey(), aws_creds.GetSessionToken()); +} + +} + +std::shared_ptr resolveS3TablesRefreshCredentials( + const ICatalog::CredentialsRefreshCallback & base_callback, + Aws::Auth::AWSCredentialsProvider & credentials_provider) +{ + if (base_callback) + { + if (auto creds = (*base_callback)()) + { + auto s3_creds = std::dynamic_pointer_cast(creds); + if (s3_creds && !s3_creds->isEmpty()) + return creds; + } + } + + return getCatalogIAMCredentials(credentials_provider); +} + +} + +#endif diff --git a/src/Databases/DataLake/S3TablesCredentialRefresh.h b/src/Databases/DataLake/S3TablesCredentialRefresh.h new file mode 100644 index 000000000000..e5b959f88860 --- /dev/null +++ b/src/Databases/DataLake/S3TablesCredentialRefresh.h @@ -0,0 +1,23 @@ +#pragma once + +#include "config.h" + +#if USE_AVRO && USE_SSL && USE_AWS_S3 + +#include +#include + +#include + +#include + +namespace DataLake +{ + +std::shared_ptr resolveS3TablesRefreshCredentials( + const ICatalog::CredentialsRefreshCallback & base_callback, + Aws::Auth::AWSCredentialsProvider & credentials_provider); + +} + +#endif diff --git a/src/Databases/DataLake/StorageCredentials.h b/src/Databases/DataLake/StorageCredentials.h index 3a2f6f793e89..7e73d66ff673 100644 --- a/src/Databases/DataLake/StorageCredentials.h +++ b/src/Databases/DataLake/StorageCredentials.h @@ -32,6 +32,8 @@ class S3Credentials final : public IStorageCredentials , session_token(session_token_) {} + bool isEmpty() const { return access_key_id.empty() || secret_access_key.empty(); } + void addCredentialsToEngineArgs(DB::ASTs & engine_args) const override { if (engine_args.size() != 1) diff --git a/src/Databases/DataLake/tests/gtest_azure_abfss_parsing.cpp b/src/Databases/DataLake/tests/gtest_azure_abfss_parsing.cpp index f1c93eee6dc6..1d710b01a7b4 100644 --- a/src/Databases/DataLake/tests/gtest_azure_abfss_parsing.cpp +++ b/src/Databases/DataLake/tests/gtest_azure_abfss_parsing.cpp @@ -240,4 +240,31 @@ TEST_F(AzureAbfssParsingTest, TableMetadataGetLocationWithEndpointVirtualHostedD EXPECT_EQ(location, "https://my.dotted.bucket.s3.mycompany.com/path/to/table/"); } +TEST_F(AzureAbfssParsingTest, TableMetadataGetMetadataLocationS3TablesWithAwsEndpoint) +{ + TableMetadata metadata; + metadata.withLocation(); + metadata.setLocation("s3://bucket/table-uuid/"); + metadata.setEndpoint("https://s3.us-east-2.amazonaws.com"); + + EXPECT_EQ(metadata.getLocation(), "https://s3.us-east-2.amazonaws.com/bucket/table-uuid/"); + + const std::string metadata_file = + "s3://bucket/table-uuid/metadata/v1.metadata.json"; + EXPECT_EQ(metadata.getMetadataLocation(metadata_file), "metadata/v1.metadata.json"); +} + +TEST_F(AzureAbfssParsingTest, TableMetadataGetMetadataLocationS3TablesEmptyPathWithAwsEndpoint) +{ + TableMetadata metadata; + metadata.withLocation(); + metadata.setLocation("s3://bucket"); + metadata.setEndpoint("https://s3.us-east-2.amazonaws.com"); + + EXPECT_EQ(metadata.getLocation(), "https://s3.us-east-2.amazonaws.com/bucket/"); + + const std::string metadata_file = "s3://bucket/metadata/v1.metadata.json"; + EXPECT_EQ(metadata.getMetadataLocation(metadata_file), "metadata/v1.metadata.json"); +} + } diff --git a/src/Databases/DataLake/tests/gtest_s3tables_credential_refresh.cpp b/src/Databases/DataLake/tests/gtest_s3tables_credential_refresh.cpp new file mode 100644 index 000000000000..6c5bc705fea5 --- /dev/null +++ b/src/Databases/DataLake/tests/gtest_s3tables_credential_refresh.cpp @@ -0,0 +1,112 @@ +#include "config.h" + +#if USE_AVRO && USE_SSL && USE_AWS_S3 + +#include + +#include +#include + +#include +#include + +#include + +namespace +{ + +class RotatingAWSCredentialsProvider : public Aws::Auth::AWSCredentialsProvider +{ +public: + explicit RotatingAWSCredentialsProvider(std::vector credentials_sets_) + : credentials_sets(std::move(credentials_sets_)) + { + } + + Aws::Auth::AWSCredentials GetAWSCredentials() override + { + std::lock_guard lock(mutex); + const size_t index = call_count++; + if (index >= credentials_sets.size()) + return credentials_sets.back(); + return credentials_sets[index]; + } + +private: + std::vector credentials_sets; + std::mutex mutex; + size_t call_count = 0; +}; + +} + +TEST(S3TablesCredentialRefresh, FallsBackToCatalogIAMWhenVendedCredentialsMissing) +{ + RotatingAWSCredentialsProvider provider({ + Aws::Auth::AWSCredentials("access_key_1", "secret_key_1", "session_token_1"), + Aws::Auth::AWSCredentials("access_key_2", "secret_key_2", "session_token_2"), + }); + + DataLake::ICatalog::CredentialsRefreshCallback base_callback = []() -> std::shared_ptr + { + return nullptr; + }; + + auto first = DataLake::resolveS3TablesRefreshCredentials(base_callback, provider); + ASSERT_NE(first, nullptr); + auto first_s3 = std::dynamic_pointer_cast(first); + ASSERT_NE(first_s3, nullptr); + EXPECT_EQ(first_s3->getAccessKeyId(), "access_key_1"); + EXPECT_EQ(first_s3->getSecretAccessKey(), "secret_key_1"); + EXPECT_EQ(first_s3->getSessionToken(), "session_token_1"); + + auto second = DataLake::resolveS3TablesRefreshCredentials(base_callback, provider); + ASSERT_NE(second, nullptr); + auto second_s3 = std::dynamic_pointer_cast(second); + ASSERT_NE(second_s3, nullptr); + EXPECT_EQ(second_s3->getAccessKeyId(), "access_key_2"); + EXPECT_EQ(second_s3->getSecretAccessKey(), "secret_key_2"); + EXPECT_EQ(second_s3->getSessionToken(), "session_token_2"); +} + +TEST(S3TablesCredentialRefresh, PrefersVendedCredentialsWhenPresent) +{ + RotatingAWSCredentialsProvider provider({ + Aws::Auth::AWSCredentials("catalog_access", "catalog_secret", "catalog_token"), + }); + + DataLake::ICatalog::CredentialsRefreshCallback base_callback = []() -> std::shared_ptr + { + return std::make_shared("vended_access", "vended_secret", "vended_token"); + }; + + auto creds = DataLake::resolveS3TablesRefreshCredentials(base_callback, provider); + ASSERT_NE(creds, nullptr); + auto s3_creds = std::dynamic_pointer_cast(creds); + ASSERT_NE(s3_creds, nullptr); + EXPECT_EQ(s3_creds->getAccessKeyId(), "vended_access"); + EXPECT_EQ(s3_creds->getSecretAccessKey(), "vended_secret"); + EXPECT_EQ(s3_creds->getSessionToken(), "vended_token"); +} + +TEST(S3TablesCredentialRefresh, FallsBackWhenVendedCredentialsEmpty) +{ + RotatingAWSCredentialsProvider provider({ + Aws::Auth::AWSCredentials("catalog_access", "catalog_secret", "catalog_token"), + }); + + DataLake::ICatalog::CredentialsRefreshCallback base_callback = []() -> std::shared_ptr + { + return std::make_shared("", "", ""); + }; + + auto creds = DataLake::resolveS3TablesRefreshCredentials(base_callback, provider); + ASSERT_NE(creds, nullptr); + auto s3_creds = std::dynamic_pointer_cast(creds); + ASSERT_NE(s3_creds, nullptr); + EXPECT_EQ(s3_creds->getAccessKeyId(), "catalog_access"); + EXPECT_EQ(s3_creds->getSecretAccessKey(), "catalog_secret"); + EXPECT_EQ(s3_creds->getSessionToken(), "catalog_token"); +} + +#endif diff --git a/src/Databases/enableAllExperimentalSettings.cpp b/src/Databases/enableAllExperimentalSettings.cpp index ac29261994ee..d20e34d5f9ae 100644 --- a/src/Databases/enableAllExperimentalSettings.cpp +++ b/src/Databases/enableAllExperimentalSettings.cpp @@ -69,6 +69,7 @@ void enableAllExperimentalSettings(ContextMutablePtr context) context->setSetting("allow_dynamic_type_in_join_keys", 1); context->setSetting("allow_experimental_alias_table_engine", 1); context->setSetting("allow_experimental_database_paimon_rest_catalog", 1); + context->setSetting("allow_experimental_database_s3_tables", 1); context->setSetting("allow_experimental_object_storage_queue_hive_partitioning", 1); context->setSetting("allow_experimental_json_lazy_type_hints", 1); context->setSetting("allow_experimental_full_text_index", 1); diff --git a/src/IO/S3/URI.cpp b/src/IO/S3/URI.cpp index dd5429dcf55f..8e88c6257876 100644 --- a/src/IO/S3/URI.cpp +++ b/src/IO/S3/URI.cpp @@ -8,6 +8,8 @@ #include #include +#include + #include #include @@ -262,6 +264,19 @@ void URI::validateKey(const String & key, const Poco::URI & uri) } } +std::string resolveS3Endpoint(const std::string & region) +{ + Aws::S3::Endpoint::S3EndpointProvider provider; + provider.AccessBuiltInParameters().SetStringParameter("Region", Aws::String(region)); + auto outcome = provider.ResolveEndpoint({}); + if (outcome.IsSuccess()) + { + auto uri = outcome.GetResult().GetURI(); + return uri.GetURIString(); + } + return "https://s3." + region + ".amazonaws.com"; +} + } } diff --git a/src/IO/S3/URI.h b/src/IO/S3/URI.h index 64b4def76744..28504e39cb7f 100644 --- a/src/IO/S3/URI.h +++ b/src/IO/S3/URI.h @@ -52,6 +52,10 @@ struct URI bool tryInitVirtualHostedStyle(bool is_using_aws_private_link_interface, bool use_strict_pattern); }; +/// Resolve the S3 endpoint URL for a given AWS region using the SDK's +/// Smithy endpoint rules (handles all partitions: standard, China, GovCloud, etc.). +std::string resolveS3Endpoint(const std::string & region); + } #endif diff --git a/src/IO/S3/tests/gtest_s3_uri.cpp b/src/IO/S3/tests/gtest_s3_uri.cpp index a429c645fca3..5c0d3debc0c0 100644 --- a/src/IO/S3/tests/gtest_s3_uri.cpp +++ b/src/IO/S3/tests/gtest_s3_uri.cpp @@ -38,4 +38,23 @@ TEST(IOTestS3URI, PathStyleWithKey) ASSERT_EQ(uri_with_no_key_and_with_slash.key, "key/key/key/key"); } +TEST(IOTestS3URI, ResolveS3Endpoint) +{ + using namespace DB; + + ASSERT_EQ(S3::resolveS3Endpoint("us-east-1"), + "https://s3.us-east-1.amazonaws.com"); + ASSERT_EQ(S3::resolveS3Endpoint("eu-west-1"), + "https://s3.eu-west-1.amazonaws.com"); + + auto cn_north = S3::resolveS3Endpoint("cn-north-1"); + ASSERT_TRUE(cn_north.ends_with(".amazonaws.com.cn")) + << "China region should resolve to .amazonaws.com.cn suffix, got: " << cn_north; + ASSERT_TRUE(cn_north.find("cn-north-1") != std::string::npos) + << "Got: " << cn_north; + + ASSERT_EQ(S3::resolveS3Endpoint("us-gov-west-1"), + "https://s3.us-gov-west-1.amazonaws.com"); +} + #endif From ed1b88d802c49af0845451d782e15ac9afaf659a Mon Sep 17 00:00:00 2001 From: Andrey Zvonov <32552679+zvonand@users.noreply.github.com> Date: Fri, 7 Aug 2026 08:40:43 +0200 Subject: [PATCH 2/4] Resolve conflicts in cherry-pick of #1808 RestCatalog.cpp: kept antalya-26.6's endpoint construction for createNamespaceIfNotExists / createTable / updateMetadata (it already composes base_url / config.prefix / NAMESPACES_ENDPOINT, which is what the PR was introducing, plus namespace URI encoding). Applied the PR's dropTable endpoint change (adds config.prefix); dropped the allowed_namespaces filter block, which is not part of #1808's diff and has no counterpart on antalya-26.6. DatabaseDataLake.cpp: the "Engine `Iceberg` must have `rest` catalog type only" check that #1808 extended no longer exists on antalya-26.6 (Iceberg engine now also serves onelake/biglake/hive catalogs), so the extended check is not re-introduced. Source-PR: #1808 (https://github.com/Altinity/ClickHouse/pull/1808) --- src/Databases/DataLake/DatabaseDataLake.cpp | 9 --------- src/Databases/DataLake/RestCatalog.cpp | 21 --------------------- 2 files changed, 30 deletions(-) diff --git a/src/Databases/DataLake/DatabaseDataLake.cpp b/src/Databases/DataLake/DatabaseDataLake.cpp index f8492a2f9766..7eda271ec0b3 100644 --- a/src/Databases/DataLake/DatabaseDataLake.cpp +++ b/src/Databases/DataLake/DatabaseDataLake.cpp @@ -1089,15 +1089,6 @@ void registerDatabaseDataLake(DatabaseFactory & factory) throw Exception(ErrorCodes::BAD_ARGUMENTS, "Engine `{}` must have arguments", database_engine_name); } -<<<<<<< HEAD -======= - if (database_engine_name == "Iceberg" && catalog_type != DatabaseDataLakeCatalogType::ICEBERG_REST - && catalog_type != DatabaseDataLakeCatalogType::S3_TABLES) - { - throw Exception(ErrorCodes::BAD_ARGUMENTS, "Engine `Iceberg` must use `rest` or `s3tables` catalog type only"); - } - ->>>>>>> 7cd07274788 (Merge pull request #1808 from Altinity/antalya_26_3_s3_tables) for (auto & engine_arg : engine_args) engine_arg = evaluateConstantExpressionOrIdentifierAsLiteral(engine_arg, args.context); diff --git a/src/Databases/DataLake/RestCatalog.cpp b/src/Databases/DataLake/RestCatalog.cpp index b253705b4d80..35dc10def36f 100644 --- a/src/Databases/DataLake/RestCatalog.cpp +++ b/src/Databases/DataLake/RestCatalog.cpp @@ -1184,11 +1184,7 @@ void RestCatalog::sendRequest(const String & endpoint, Poco::JSON::Object::Ptr r void RestCatalog::createNamespaceIfNotExists(const String & namespace_name, const String & location) const { -<<<<<<< HEAD const std::string endpoint = (base_url / config.prefix / NAMESPACES_ENDPOINT).generic_string(); -======= - const std::string endpoint = base_url / config.prefix / "namespaces"; ->>>>>>> 7cd07274788 (Merge pull request #1808 from Altinity/antalya_26_3_s3_tables) Poco::JSON::Object::Ptr request_body = new Poco::JSON::Object; { @@ -1216,11 +1212,7 @@ void RestCatalog::createTable(const String & namespace_name, const String & tabl { createNamespaceIfNotExists(namespace_name, metadata_content->getValue("location")); -<<<<<<< HEAD const std::string endpoint = (base_url / config.prefix / NAMESPACES_ENDPOINT / encodeNamespaceForURI(namespace_name) / "tables").generic_string(); -======= - const std::string endpoint = base_url / config.prefix / "namespaces" / namespace_name / "tables"; ->>>>>>> 7cd07274788 (Merge pull request #1808 from Altinity/antalya_26_3_s3_tables) Poco::JSON::Object::Ptr request_body = new Poco::JSON::Object; request_body->set("name", table_name); @@ -1261,11 +1253,7 @@ void RestCatalog::createTable(const String & namespace_name, const String & tabl bool RestCatalog::updateMetadata(const String & namespace_name, const String & table_name, const String & /*new_metadata_path*/, Poco::JSON::Object::Ptr new_snapshot) const { -<<<<<<< HEAD const std::string endpoint = (base_url / config.prefix / NAMESPACES_ENDPOINT / encodeNamespaceForURI(namespace_name) / "tables" / table_name).generic_string(); -======= - const std::string endpoint = base_url / config.prefix / "namespaces" / namespace_name / "tables" / table_name; ->>>>>>> 7cd07274788 (Merge pull request #1808 from Altinity/antalya_26_3_s3_tables) Poco::JSON::Object::Ptr request_body = new Poco::JSON::Object; { @@ -1393,18 +1381,9 @@ bool RestCatalog::updateSchema( void RestCatalog::dropTable(const String & namespace_name, const String & table_name) const { -<<<<<<< HEAD - const std::string endpoint = fmt::format("{}/namespaces/{}/tables/{}?purgeRequested=False", base_url, namespace_name, table_name); -======= - 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"; ->>>>>>> 7cd07274788 (Merge pull request #1808 from Altinity/antalya_26_3_s3_tables) Poco::JSON::Object::Ptr request_body = nullptr; try From 49824d64f8bb656ce682208d77c9f49b5b745620 Mon Sep 17 00:00:00 2001 From: Andrey Zvonov <32552679+zvonand@users.noreply.github.com> Date: Thu, 11 Jun 2026 14:03:41 +0200 Subject: [PATCH 3/4] Cherry-pick of https://github.com/Altinity/ClickHouse/pull/1868 with unresolved conflict markers (resolution in next commit) --- Original cherry-pick message follows: Merge pull request #1868 from Altinity/fix/datalake-rest-catalog-profile-events Profile events for Iceberg catalogs # Conflicts: # src/Common/ProfileEvents.cpp # src/Databases/DataLake/RestCatalog.cpp # src/Databases/DataLake/UnityCatalog.cpp --- src/Common/ProfileEvents.cpp | 53 +++++++++ src/Databases/DataLake/GlueCatalog.cpp | 70 +++++++++++- src/Databases/DataLake/RestCatalog.cpp | 94 ++++++++++++--- src/Databases/DataLake/S3TablesCatalog.cpp | 9 ++ src/Databases/DataLake/UnityCatalog.cpp | 127 ++++++++++++++++++++- 5 files changed, 327 insertions(+), 26 deletions(-) diff --git a/src/Common/ProfileEvents.cpp b/src/Common/ProfileEvents.cpp index 7f35ef869e92..dd4e623c4b14 100644 --- a/src/Common/ProfileEvents.cpp +++ b/src/Common/ProfileEvents.cpp @@ -1503,6 +1503,59 @@ The server successfully detected this situation and will download merged part fr M(AIRowsProcessed, "Number of rows that received an AI result.", ValueType::Number) \ M(AIRowsSkipped, "Number of rows that received a default value due to quota or error.", ValueType::Number) \ \ +<<<<<<< HEAD +======= + M(ObjectStorageListObjectsCacheHits, "Number of times object storage list objects operation hit the cache.", ValueType::Number) \ + M(ObjectStorageListObjectsCacheMisses, "Number of times object storage list objects operation miss the cache.", ValueType::Number) \ + M(ObjectStorageListObjectsCacheExactMatchHits, "Number of times object storage list objects operation hit the cache with an exact match.", ValueType::Number) \ + M(ObjectStorageListObjectsCachePrefixMatchHits, "Number of times object storage list objects operation miss the cache using prefix matching.", ValueType::Number) \ + \ + M(DataLakeRestCatalogLoadConfig, "Number of 'load config' requests to Iceberg REST catalog.", ValueType::Number) \ + M(DataLakeRestCatalogLoadConfigMicroseconds, "Total time of 'load config' requests to Iceberg REST catalog.", ValueType::Microseconds) \ + M(DataLakeRestCatalogGetNamespaces, "Number of 'get namespaces' requests to Iceberg REST catalog.", ValueType::Number) \ + M(DataLakeRestCatalogGetNamespacesMicroseconds, "Total time of 'get namespaces' requests to Iceberg REST catalog.", ValueType::Microseconds) \ + M(DataLakeRestCatalogGetTables, "Number of 'get tables' requests to Iceberg REST catalog.", ValueType::Number) \ + M(DataLakeRestCatalogGetTablesMicroseconds, "Total time of 'get tables' requests to Iceberg REST catalog.", ValueType::Microseconds) \ + M(DataLakeRestCatalogGetTableMetadata, "Number of 'get table metadata' requests to Iceberg REST catalog.", ValueType::Number) \ + M(DataLakeRestCatalogGetTableMetadataMicroseconds, "Total time of 'get table metadata' requests to Iceberg REST catalog.", ValueType::Microseconds) \ + M(DataLakeRestCatalogGetCredentials, "Number of 'get credentials' requests to Iceberg REST catalog.", ValueType::Number) \ + M(DataLakeRestCatalogGetCredentialsMicroseconds, "Total time of 'get credentials' requests to Iceberg REST catalog.", ValueType::Microseconds) \ + M(DataLakeRestCatalogCreateNamespace, "Number of 'create namespace' requests to Iceberg REST catalog.", ValueType::Number) \ + M(DataLakeRestCatalogCreateNamespaceMicroseconds, "Total time of 'create namespace' requests to Iceberg REST catalog.", ValueType::Microseconds) \ + M(DataLakeRestCatalogCreateTable, "Number of 'create table' requests to Iceberg REST catalog.", ValueType::Number) \ + M(DataLakeRestCatalogCreateTableMicroseconds, "Total time of 'create table' requests to Iceberg REST catalog.", ValueType::Microseconds) \ + M(DataLakeRestCatalogUpdateTable, "Number of 'update table' requests to Iceberg REST catalog.", ValueType::Number) \ + M(DataLakeRestCatalogUpdateTableMicroseconds, "Total time of 'update table' requests to Iceberg REST catalog.", ValueType::Microseconds) \ + M(DataLakeRestCatalogDropTable, "Number of 'drop table' requests to Iceberg REST catalog.", ValueType::Number) \ + M(DataLakeRestCatalogDropTableMicroseconds, "Total time of 'drop table' requests to Iceberg REST catalog.", ValueType::Microseconds) \ + \ + M(DataLakeGlueCatalogGetDatabases, "Number of 'get databases' requests to Iceberg Glue catalog.", ValueType::Number) \ + M(DataLakeGlueCatalogGetDatabasesMicroseconds, "Total time of 'get databases' requests to Iceberg Glue catalog.", ValueType::Microseconds) \ + M(DataLakeGlueCatalogGetTables, "Number of 'get tables' requests to Iceberg Glue catalog.", ValueType::Number) \ + M(DataLakeGlueCatalogGetTablesMicroseconds, "Total time of 'get tables' requests to Iceberg Glue catalog.", ValueType::Microseconds) \ + M(DataLakeGlueCatalogGetTable, "Number of 'get table' requests to Iceberg Glue catalog.", ValueType::Number) \ + M(DataLakeGlueCatalogGetTableMicroseconds, "Total time of 'get table' requests to Iceberg Glue catalog.", ValueType::Microseconds) \ + M(DataLakeGlueCatalogCreateDatabase, "Number of 'create database' requests to Iceberg Glue catalog.", ValueType::Number) \ + M(DataLakeGlueCatalogCreateDatabaseMicroseconds, "Total time of 'create database' requests to Iceberg Glue catalog.", ValueType::Microseconds) \ + M(DataLakeGlueCatalogCreateTable, "Number of 'create table' requests to Iceberg Glue catalog.", ValueType::Number) \ + M(DataLakeGlueCatalogCreateTableMicroseconds, "Total time of 'create table' requests to Iceberg Glue catalog.", ValueType::Microseconds) \ + M(DataLakeGlueCatalogUpdateTable, "Number of 'update table' requests to Iceberg Glue catalog.", ValueType::Number) \ + M(DataLakeGlueCatalogUpdateTableMicroseconds, "Total time of 'update table' requests to Iceberg Glue catalog.", ValueType::Microseconds) \ + M(DataLakeGlueCatalogDropTable, "Number of 'drop table' requests to Iceberg Glue catalog.", ValueType::Number) \ + M(DataLakeGlueCatalogDropTableMicroseconds, "Total time of 'drop table' requests to Iceberg Glue catalog.", ValueType::Microseconds) \ + \ + M(DataLakeUnityCatalogGetTables, "Number of 'get tables' requests to Iceberg Unity catalog.", ValueType::Number) \ + M(DataLakeUnityCatalogGetTablesMicroseconds, "Total time of 'get tables' requests to Iceberg Unity catalog.", ValueType::Microseconds) \ + M(DataLakeUnityCatalogGetTable, "Number of 'get table' requests to Iceberg Unity catalog.", ValueType::Number) \ + M(DataLakeUnityCatalogGetTableMicroseconds, "Total time of 'get table' requests to Iceberg Unity catalog.", ValueType::Microseconds) \ + M(DataLakeUnityCatalogGetTableMetadata, "Number of 'get table metadata' requests to Iceberg Unity catalog.", ValueType::Number) \ + M(DataLakeUnityCatalogGetTableMetadataMicroseconds, "Total time of 'get table metadata' requests to Iceberg Unity catalog.", ValueType::Microseconds) \ + M(DataLakeUnityCatalogGetSchemas, "Number of 'get schemas' requests to Iceberg Unity catalog.", ValueType::Number) \ + M(DataLakeUnityCatalogGetSchemasMicroseconds, "Total time of 'get schemas' requests to Iceberg Unity catalog.", ValueType::Microseconds) \ + M(DataLakeUnityCatalogGetCredentials, "Number of 'get credentials' requests to Iceberg Unity catalog.", ValueType::Number) \ + M(DataLakeUnityCatalogGetCredentialsMicroseconds, "Total time of 'get credentials' requests to Iceberg Unity catalog.", ValueType::Microseconds) \ + +>>>>>>> 383c8d11e60 (Merge pull request #1868 from Altinity/fix/datalake-rest-catalog-profile-events) #ifdef APPLY_FOR_EXTERNAL_EVENTS #define APPLY_FOR_EVENTS(M) APPLY_FOR_BUILTIN_EVENTS(M) APPLY_FOR_EXTERNAL_EVENTS(M) diff --git a/src/Databases/DataLake/GlueCatalog.cpp b/src/Databases/DataLake/GlueCatalog.cpp index 53ac171c79ff..ca6555aa6f39 100644 --- a/src/Databases/DataLake/GlueCatalog.cpp +++ b/src/Databases/DataLake/GlueCatalog.cpp @@ -20,6 +20,7 @@ #include #include +#include #include #include @@ -83,6 +84,24 @@ namespace DB::ServerSetting extern const ServerSettingsUInt64 s3_retry_attempts; } +namespace ProfileEvents +{ + extern const Event DataLakeGlueCatalogGetDatabases; + extern const Event DataLakeGlueCatalogGetDatabasesMicroseconds; + extern const Event DataLakeGlueCatalogGetTables; + extern const Event DataLakeGlueCatalogGetTablesMicroseconds; + extern const Event DataLakeGlueCatalogGetTable; + extern const Event DataLakeGlueCatalogGetTableMicroseconds; + extern const Event DataLakeGlueCatalogCreateDatabase; + extern const Event DataLakeGlueCatalogCreateDatabaseMicroseconds; + extern const Event DataLakeGlueCatalogCreateTable; + extern const Event DataLakeGlueCatalogCreateTableMicroseconds; + extern const Event DataLakeGlueCatalogUpdateTable; + extern const Event DataLakeGlueCatalogUpdateTableMicroseconds; + extern const Event DataLakeGlueCatalogDropTable; + extern const Event DataLakeGlueCatalogDropTableMicroseconds; +} + namespace CurrentMetrics { extern const Metric MarkCacheBytes; @@ -191,7 +210,14 @@ DataLake::ICatalog::Namespaces GlueCatalog::getDatabases(const std::string & pre do { request.SetNextToken(next_token); - auto outcome = glue_client->GetDatabases(request); + + Aws::Glue::Model::GetDatabasesOutcome outcome; + { + ProfileEvents::increment(ProfileEvents::DataLakeGlueCatalogGetDatabases); + auto timer = DB::CurrentThread::getProfileEvents().timer(ProfileEvents::DataLakeGlueCatalogGetDatabasesMicroseconds); + outcome = glue_client->GetDatabases(request); + } + if (outcome.IsSuccess()) { const auto & databases_result = outcome.GetResult(); @@ -240,7 +266,12 @@ DB::Names GlueCatalog::getTablesForDatabase(const std::string & db_name, size_t do { request.SetNextToken(next_token); - auto outcome = glue_client->GetTables(request); + Aws::Glue::Model::GetTablesOutcome outcome; + { + ProfileEvents::increment(ProfileEvents::DataLakeGlueCatalogGetTables); + auto timer = DB::CurrentThread::getProfileEvents().timer(ProfileEvents::DataLakeGlueCatalogGetTablesMicroseconds); + outcome = glue_client->GetTables(request); + } if (outcome.IsSuccess()) { const auto & tables_result = outcome.GetResult(); @@ -290,6 +321,8 @@ bool GlueCatalog::existsTable(const std::string & database_name, const std::stri request.SetDatabaseName(database_name); request.SetName(table_name); + ProfileEvents::increment(ProfileEvents::DataLakeGlueCatalogGetTable); + auto timer = DB::CurrentThread::getProfileEvents().timer(ProfileEvents::DataLakeGlueCatalogGetTableMicroseconds); auto outcome = glue_client->GetTable(request); return outcome.IsSuccess(); } @@ -303,7 +336,12 @@ bool GlueCatalog::tryGetTableMetadata( request.SetDatabaseName(database_name); request.SetName(table_name); - auto outcome = glue_client->GetTable(request); + Aws::Glue::Model::GetTableOutcome outcome; + { + ProfileEvents::increment(ProfileEvents::DataLakeGlueCatalogGetTable); + auto timer = DB::CurrentThread::getProfileEvents().timer(ProfileEvents::DataLakeGlueCatalogGetTableMicroseconds); + outcome = glue_client->GetTable(request); + } if (outcome.IsSuccess()) { const auto & table_outcome = outcome.GetResult().GetTable(); @@ -599,6 +637,8 @@ void GlueCatalog::createNamespaceIfNotExists(const String & namespace_name) cons db_input.SetName(namespace_name); create_request.SetDatabaseInput(db_input); + ProfileEvents::increment(ProfileEvents::DataLakeGlueCatalogCreateDatabase); + auto timer = DB::CurrentThread::getProfileEvents().timer(ProfileEvents::DataLakeGlueCatalogCreateDatabaseMicroseconds); glue_client->CreateDatabase(create_request); } @@ -631,7 +671,13 @@ void GlueCatalog::createTable(const String & namespace_name, const String & tabl request.SetTableInput(table_input); - auto response = glue_client->CreateTable(request); + Aws::Glue::Model::CreateTableOutcome response; + + { + ProfileEvents::increment(ProfileEvents::DataLakeGlueCatalogCreateTable); + auto timer = DB::CurrentThread::getProfileEvents().timer(ProfileEvents::DataLakeGlueCatalogCreateTableMicroseconds); + response = glue_client->CreateTable(request); + } if (!response.IsSuccess()) throw DB::Exception(DB::ErrorCodes::DATALAKE_DATABASE_ERROR, "Can not create metadata in glue catalog: {}", response.GetError().GetMessage()); @@ -666,7 +712,13 @@ bool GlueCatalog::updateMetadata(const String & namespace_name, const String & t request.SetTableInput(table_input); - auto response = glue_client->UpdateTable(request); + Aws::Glue::Model::UpdateTableOutcome response; + + { + ProfileEvents::increment(ProfileEvents::DataLakeGlueCatalogUpdateTable); + auto timer = DB::CurrentThread::getProfileEvents().timer(ProfileEvents::DataLakeGlueCatalogUpdateTableMicroseconds); + response = glue_client->UpdateTable(request); + } if (!response.IsSuccess()) throw DB::Exception(DB::ErrorCodes::DATALAKE_DATABASE_ERROR, "Can not update metadata in glue catalog {}", response.GetError().GetMessage()); @@ -690,7 +742,13 @@ void GlueCatalog::dropTable(const String & namespace_name, const String & table_ request.SetDatabaseName(namespace_name); request.SetName(table_name); - auto response = glue_client->DeleteTable(request); + Aws::Glue::Model::DeleteTableOutcome response; + + { + ProfileEvents::increment(ProfileEvents::DataLakeGlueCatalogDropTable); + auto timer = DB::CurrentThread::getProfileEvents().timer(ProfileEvents::DataLakeGlueCatalogDropTableMicroseconds); + response = glue_client->DeleteTable(request); + } if (!response.IsSuccess()) throw DB::Exception( diff --git a/src/Databases/DataLake/RestCatalog.cpp b/src/Databases/DataLake/RestCatalog.cpp index 35dc10def36f..cec14e788371 100644 --- a/src/Databases/DataLake/RestCatalog.cpp +++ b/src/Databases/DataLake/RestCatalog.cpp @@ -5,6 +5,7 @@ #include #include #include +#include #include #include #include @@ -68,6 +69,28 @@ namespace DB::FailPoints extern const char check_database_datalake_negative[]; } +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 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 { @@ -220,10 +243,15 @@ RestCatalog::RestCatalog( RestCatalog::Config RestCatalog::loadConfig() { Poco::URI::QueryParameters params = {{"warehouse", warehouse}}; - auto buf = createReadBuffer(CONFIG_ENDPOINT, params); std::string json_str; - readJSONObjectPossiblyInvalid(json_str, *buf); + + { + 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); @@ -765,6 +793,7 @@ RestCatalog::Namespaces RestCatalog::getNamespaces(const std::string & base_name try { +<<<<<<< HEAD while (true) { /// The Iceberg REST OpenAPI spec uses `pageToken` (request) / `next-page-token` (response) @@ -803,6 +832,14 @@ RestCatalog::Namespaces RestCatalog::getNamespaces(const std::string & base_name page_token = std::move(next_page_token); } return all_namespaces; +======= + 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; +>>>>>>> 383c8d11e60 (Merge pull request #1868 from Altinity/fix/datalake-rest-catalog-profile-events) } catch (const DB::HTTPException & e) { @@ -907,6 +944,7 @@ DB::Names RestCatalog::getTables(const std::string & base_namespace, size_t limi auto encoded_namespace = encodeNamespaceForURI(base_namespace); const std::string endpoint = std::filesystem::path(NAMESPACES_ENDPOINT) / encoded_namespace / "tables"; +<<<<<<< HEAD DB::Names tables; String page_token; /// Cycle-detection guard: tracks every non-empty `next-page-token` we have seen on this @@ -956,6 +994,12 @@ DB::Names RestCatalog::getTables(const std::string & base_namespace, size_t limi } return 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); +>>>>>>> 383c8d11e60 (Merge pull request #1868 from Altinity/fix/datalake-rest-catalog-profile-events) } DB::Names RestCatalog::parseTables(DB::ReadBuffer & buf, const std::string & base_namespace, size_t limit, String & next_page_token) const @@ -1061,16 +1105,21 @@ bool RestCatalog::getTableMetadataImpl( } const std::string endpoint = std::filesystem::path(NAMESPACES_ENDPOINT) / encodeNamespaceForURI(namespace_name) / "tables" / table_name; - auto buf = createReadBuffer(config.prefix / endpoint, /* params */{}, headers); + String json_str; - if (buf->eof()) { - LOG_DEBUG(log, "Table doesn't exist (endpoint: {})", endpoint); - return false; - } + ProfileEvents::increment(ProfileEvents::DataLakeRestCatalogGetTableMetadata); + auto timer = DB::CurrentThread::getProfileEvents().timer(ProfileEvents::DataLakeRestCatalogGetTableMetadataMicroseconds); + auto buf = createReadBuffer(config.prefix / endpoint, /* params */{}, headers); - String json_str; - readJSONObjectPossiblyInvalid(json_str, *buf); + 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, @@ -1200,6 +1249,8 @@ void RestCatalog::createNamespaceIfNotExists(const String & namespace_name, cons try { + ProfileEvents::increment(ProfileEvents::DataLakeRestCatalogCreateNamespace); + auto timer = DB::CurrentThread::getProfileEvents().timer(ProfileEvents::DataLakeRestCatalogCreateNamespaceMicroseconds); sendRequest(endpoint, request_body); } catch (...) @@ -1242,6 +1293,8 @@ void RestCatalog::createTable(const String & namespace_name, const String & tabl try { + ProfileEvents::increment(ProfileEvents::DataLakeRestCatalogCreateTable); + auto timer = DB::CurrentThread::getProfileEvents().timer(ProfileEvents::DataLakeRestCatalogCreateTableMicroseconds); sendRequest(endpoint, request_body); } catch (const DB::HTTPException & ex) @@ -1307,6 +1360,8 @@ bool RestCatalog::updateMetadata(const String & namespace_name, const String & t try { + ProfileEvents::increment(ProfileEvents::DataLakeRestCatalogUpdateTable); + auto timer = DB::CurrentThread::getProfileEvents().timer(ProfileEvents::DataLakeRestCatalogUpdateTableMicroseconds); sendRequest(endpoint, request_body); } catch (const DB::HTTPException & ex) @@ -1388,6 +1443,8 @@ void RestCatalog::dropTable(const String & namespace_name, const String & table_ 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) @@ -1475,16 +1532,21 @@ ICatalog::CredentialsRefreshCallback RestCatalog::getCredentialsConfigurationCal 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; - auto buf = createReadBuffer(config.prefix / endpoint, /* params */{}, headers); + String json_str; - if (buf->eof()) { - LOG_DEBUG(log, "Table doesn't exist (endpoint: {})", endpoint); - return nullptr; - } + ProfileEvents::increment(ProfileEvents::DataLakeRestCatalogGetCredentials); + auto timer = DB::CurrentThread::getProfileEvents().timer(ProfileEvents::DataLakeRestCatalogGetCredentialsMicroseconds); + auto buf = createReadBuffer(config.prefix / endpoint, /* params */{}, headers); - String json_str; - readJSONObjectPossiblyInvalid(json_str, *buf); + 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); diff --git a/src/Databases/DataLake/S3TablesCatalog.cpp b/src/Databases/DataLake/S3TablesCatalog.cpp index bf0be97754e5..e00a7de586c7 100644 --- a/src/Databases/DataLake/S3TablesCatalog.cpp +++ b/src/Databases/DataLake/S3TablesCatalog.cpp @@ -9,6 +9,7 @@ #include #include #include +#include #include #include #include @@ -49,6 +50,12 @@ namespace DB::ServerSetting extern const ServerSettingsUInt64 s3_retry_attempts; } +namespace ProfileEvents +{ + extern const Event DataLakeRestCatalogDropTable; + extern const Event DataLakeRestCatalogDropTableMicroseconds; +} + namespace DataLake { @@ -225,6 +232,8 @@ void S3TablesCatalog::dropTable(const String & namespace_name, const String & ta 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) diff --git a/src/Databases/DataLake/UnityCatalog.cpp b/src/Databases/DataLake/UnityCatalog.cpp index 414b7e439ecf..7374e21d086b 100644 --- a/src/Databases/DataLake/UnityCatalog.cpp +++ b/src/Databases/DataLake/UnityCatalog.cpp @@ -8,6 +8,7 @@ #include #include #include +#include #include #include #include @@ -15,6 +16,20 @@ #include #include +namespace ProfileEvents +{ + extern const Event DataLakeUnityCatalogGetTables; + extern const Event DataLakeUnityCatalogGetTablesMicroseconds; + extern const Event DataLakeUnityCatalogGetTable; + extern const Event DataLakeUnityCatalogGetTableMicroseconds; + extern const Event DataLakeUnityCatalogGetTableMetadata; + extern const Event DataLakeUnityCatalogGetTableMetadataMicroseconds; + extern const Event DataLakeUnityCatalogGetCredentials; + extern const Event DataLakeUnityCatalogGetCredentialsMicroseconds; + extern const Event DataLakeUnityCatalogGetSchemas; + extern const Event DataLakeUnityCatalogGetSchemasMicroseconds; +} + namespace DB::ErrorCodes { extern const int DATALAKE_DATABASE_ERROR; @@ -144,6 +159,7 @@ void UnityCatalog::getCredentials(const String & table_id, TableMetadata & metad std::shared_ptr creds; switch (storage_type) { +<<<<<<< HEAD case StorageType::S3: creds = parseS3Credentials(response); break; @@ -152,6 +168,70 @@ void UnityCatalog::getCredentials(const String & table_id, TableMetadata & metad break; default: break; +======= + case StorageType::S3: + { + auto callback = [table_id] (std::ostream & os) + { + Poco::JSON::Object obj; + obj.set("table_id", table_id); + obj.set("operation", "READ"); + obj.stringify(os); + }; + + Poco::Dynamic::Var json; + { + ProfileEvents::increment(ProfileEvents::DataLakeUnityCatalogGetCredentials); + auto timer = DB::CurrentThread::getProfileEvents().timer(ProfileEvents::DataLakeUnityCatalogGetCredentialsMicroseconds); + std::string _; + std::tie(json, _) = postJSONRequest(TEMPORARY_CREDENTIALS_ENDPOINT, callback); + } + const Poco::JSON::Object::Ptr & object = json.extract(); + + if (hasValueAndItsNotNone("aws_temp_credentials", object)) + { + const Poco::JSON::Object::Ptr & creds_object = object->getObject("aws_temp_credentials"); + std::string access_key_id = creds_object->get("access_key_id").extract(); + std::string secret_access_key = creds_object->get("secret_access_key").extract(); + std::string session_token = creds_object->get("session_token").extract(); + + auto creds = std::make_shared(access_key_id, secret_access_key, session_token); + metadata.setStorageCredentials(creds); + } + break; + } + case StorageType::Azure: + { + auto callback = [table_id] (std::ostream & os) + { + Poco::JSON::Object obj; + obj.set("table_id", table_id); + obj.set("operation", "READ"); + obj.stringify(os); + }; + + Poco::Dynamic::Var json; + { + ProfileEvents::increment(ProfileEvents::DataLakeUnityCatalogGetCredentials); + auto timer = DB::CurrentThread::getProfileEvents().timer(ProfileEvents::DataLakeUnityCatalogGetCredentialsMicroseconds); + std::string _; + std::tie(json, _) = postJSONRequest(TEMPORARY_CREDENTIALS_ENDPOINT, callback); + } + const Poco::JSON::Object::Ptr & object = json.extract(); + + if (hasValueAndItsNotNone("azure_user_delegation_sas", object)) + { + const Poco::JSON::Object::Ptr & creds_object = object->getObject("azure_user_delegation_sas"); + std::string sas_token = creds_object->get("sas_token").extract(); + + auto creds = std::make_shared(sas_token); + metadata.setStorageCredentials(creds); + } + break; + } + default: + break; +>>>>>>> 383c8d11e60 (Merge pull request #1868 from Altinity/fix/datalake-rest-catalog-profile-events) } if (creds) metadata.setStorageCredentials(creds); @@ -167,7 +247,11 @@ bool UnityCatalog::tryGetTableMetadata( std::string json_str; try { - std::tie(json, json_str) = getJSONRequest(std::filesystem::path{TABLES_ENDPOINT} / full_table_name); + { + ProfileEvents::increment(ProfileEvents::DataLakeUnityCatalogGetTableMetadata); + auto timer = DB::CurrentThread::getProfileEvents().timer(ProfileEvents::DataLakeUnityCatalogGetTableMetadataMicroseconds); + std::tie(json, json_str) = getJSONRequest(std::filesystem::path{TABLES_ENDPOINT} / full_table_name); + } const Poco::JSON::Object::Ptr & object = json.extract(); if (hasValueAndItsNotNone("name", object) && object->get("name").extract() == table_name) { @@ -288,7 +372,11 @@ bool UnityCatalog::existsTable(const std::string & schema_name, const std::strin Poco::Dynamic::Var json; try { - std::tie(json, json_str) = getJSONRequest(std::filesystem::path{TABLES_ENDPOINT} / (warehouse + "." + schema_name + "." + table_name)); + { + ProfileEvents::increment(ProfileEvents::DataLakeUnityCatalogGetTable); + auto timer = DB::CurrentThread::getProfileEvents().timer(ProfileEvents::DataLakeUnityCatalogGetTableMicroseconds); + std::tie(json, json_str) = getJSONRequest(std::filesystem::path{TABLES_ENDPOINT} / (warehouse + "." + schema_name + "." + table_name)); + } const Poco::JSON::Object::Ptr & object = json.extract(); if (hasValueAndItsNotNone("name", object) && object->get("name").extract() == table_name) return true; @@ -316,7 +404,11 @@ DB::Names UnityCatalog::getTablesForSchema(const std::string & schema, size_t li try { - std::tie(json, json_str) = getJSONRequest(TABLES_ENDPOINT, params); + { + ProfileEvents::increment(ProfileEvents::DataLakeUnityCatalogGetTables); + auto timer = DB::CurrentThread::getProfileEvents().timer(ProfileEvents::DataLakeUnityCatalogGetTablesMicroseconds); + std::tie(json, json_str) = getJSONRequest(TABLES_ENDPOINT, params); + } const Poco::JSON::Object::Ptr & object = json.extract(); if (!hasValueAndItsNotNone("tables", object)) @@ -380,7 +472,11 @@ DataLake::ICatalog::Namespaces UnityCatalog::getSchemas(const std::string & base try { - std::tie(json, json_str) = getJSONRequest(SCHEMAS_ENDPOINT, params); + { + ProfileEvents::increment(ProfileEvents::DataLakeUnityCatalogGetSchemas); + auto timer = DB::CurrentThread::getProfileEvents().timer(ProfileEvents::DataLakeUnityCatalogGetSchemasMicroseconds); + std::tie(json, json_str) = getJSONRequest(SCHEMAS_ENDPOINT, params); + } const Poco::JSON::Object::Ptr & object = json.extract(); auto schemas_object = object->get("schemas").extract(); @@ -458,7 +554,30 @@ ICatalog::CredentialsRefreshCallback UnityCatalog::getCredentialsConfigurationCa return [this, unity_table_id] () -> std::shared_ptr { LOG_DEBUG(log, "Update credentials in the catalog"); +<<<<<<< HEAD return parseS3Credentials(requestReadCredentials(unity_table_id)); +======= + Poco::Dynamic::Var json; + { + ProfileEvents::increment(ProfileEvents::DataLakeUnityCatalogGetCredentials); + auto timer = DB::CurrentThread::getProfileEvents().timer(ProfileEvents::DataLakeUnityCatalogGetCredentialsMicroseconds); + std::string _; + std::tie(json, _) = postJSONRequest(TEMPORARY_CREDENTIALS_ENDPOINT, {}); + } + const Poco::JSON::Object::Ptr & object = json.extract(); + + if (hasValueAndItsNotNone("aws_temp_credentials", object)) + { + const Poco::JSON::Object::Ptr & creds_object = object->getObject("aws_temp_credentials"); + std::string access_key_id = creds_object->get("access_key_id").extract(); + std::string secret_access_key = creds_object->get("secret_access_key").extract(); + std::string session_token = creds_object->get("session_token").extract(); + + auto creds = std::make_shared(access_key_id, secret_access_key, session_token); + return creds; + } + return nullptr; +>>>>>>> 383c8d11e60 (Merge pull request #1868 from Altinity/fix/datalake-rest-catalog-profile-events) }; } From b0f9dd262b1212ad39fd1e4e5109d74859d0065d Mon Sep 17 00:00:00 2001 From: Andrey Zvonov <32552679+zvonand@users.noreply.github.com> Date: Fri, 7 Aug 2026 08:44:45 +0200 Subject: [PATCH 4/4] Resolve conflicts in cherry-pick of #1868 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Kept antalya-26.6's paginated getNamespaces()/getTables() loops in RestCatalog and its extracted UnityCatalog credential helpers, placing the PR's profile-event counters/timers at the corresponding request sites. Dropped the ObjectStorageListObjectsCache* rows that came along as context from the merge commit but are not part of #1868 and do not exist on antalya-26.6. Adapted: RestCatalog::getNamespaces/getTables — counters/timers moved inside antalya-26.6's page loops Adapted: UnityCatalog credentials counter/timer moved into the extracted requestReadCredentials() helper Source-PR: #1868 (https://github.com/Altinity/ClickHouse/pull/1868) --- src/Common/ProfileEvents.cpp | 8 -- src/Core/SettingsChangesHistory.cpp | 2 +- src/Databases/DataLake/DatabaseDataLake.cpp | 1 - src/Databases/DataLake/ICatalog.cpp | 29 +++++- src/Databases/DataLake/RestCatalog.cpp | 20 +---- src/Databases/DataLake/S3TablesCatalog.cpp | 14 +-- src/Databases/DataLake/S3TablesCatalog.h | 2 - src/Databases/DataLake/UnityCatalog.cpp | 97 ++------------------- 8 files changed, 42 insertions(+), 131 deletions(-) diff --git a/src/Common/ProfileEvents.cpp b/src/Common/ProfileEvents.cpp index dd4e623c4b14..6032f1d7fda9 100644 --- a/src/Common/ProfileEvents.cpp +++ b/src/Common/ProfileEvents.cpp @@ -1503,13 +1503,6 @@ The server successfully detected this situation and will download merged part fr M(AIRowsProcessed, "Number of rows that received an AI result.", ValueType::Number) \ M(AIRowsSkipped, "Number of rows that received a default value due to quota or error.", ValueType::Number) \ \ -<<<<<<< HEAD -======= - M(ObjectStorageListObjectsCacheHits, "Number of times object storage list objects operation hit the cache.", ValueType::Number) \ - M(ObjectStorageListObjectsCacheMisses, "Number of times object storage list objects operation miss the cache.", ValueType::Number) \ - M(ObjectStorageListObjectsCacheExactMatchHits, "Number of times object storage list objects operation hit the cache with an exact match.", ValueType::Number) \ - M(ObjectStorageListObjectsCachePrefixMatchHits, "Number of times object storage list objects operation miss the cache using prefix matching.", ValueType::Number) \ - \ M(DataLakeRestCatalogLoadConfig, "Number of 'load config' requests to Iceberg REST catalog.", ValueType::Number) \ M(DataLakeRestCatalogLoadConfigMicroseconds, "Total time of 'load config' requests to Iceberg REST catalog.", ValueType::Microseconds) \ M(DataLakeRestCatalogGetNamespaces, "Number of 'get namespaces' requests to Iceberg REST catalog.", ValueType::Number) \ @@ -1555,7 +1548,6 @@ The server successfully detected this situation and will download merged part fr M(DataLakeUnityCatalogGetCredentials, "Number of 'get credentials' requests to Iceberg Unity catalog.", ValueType::Number) \ M(DataLakeUnityCatalogGetCredentialsMicroseconds, "Total time of 'get credentials' requests to Iceberg Unity catalog.", ValueType::Microseconds) \ ->>>>>>> 383c8d11e60 (Merge pull request #1868 from Altinity/fix/datalake-rest-catalog-profile-events) #ifdef APPLY_FOR_EXTERNAL_EVENTS #define APPLY_FOR_EVENTS(M) APPLY_FOR_BUILTIN_EVENTS(M) APPLY_FOR_EXTERNAL_EVENTS(M) diff --git a/src/Core/SettingsChangesHistory.cpp b/src/Core/SettingsChangesHistory.cpp index 5d1dcfa03cf3..e6ab2a70b36a 100644 --- a/src/Core/SettingsChangesHistory.cpp +++ b/src/Core/SettingsChangesHistory.cpp @@ -86,6 +86,7 @@ const VersionToSettingsChangesMap & getSettingsChangesHistory() {"allow_experimental_query_deduplication", false, false, "The setting is obsolete, the feature has been removed."}, {"query_plan_min_columns_for_join_lazy_indexing", 0, 3, "Control the minimum number of payload columns from the left side required for enabling lazy indexing optimization in JOIN"}, {"query_plan_max_limit_for_join_lazy_indexing", 1000, 1000, "Added new setting to control maximum limit value that allows to use query plan for lazy join indexing optimization. If zero, there is no limit"}, + {"allow_experimental_database_s3_tables", false, false, "New setting to enable experimental database S3 tables (AWS Iceberg REST catalog)."}, }); addSettingsChanges(settings_changes_history, "26.5", @@ -229,7 +230,6 @@ const VersionToSettingsChangesMap & getSettingsChangesHistory() {"iceberg_expire_default_max_ref_age_ms", 9223372036854775807, 9223372036854775807, "New setting."}, {"max_skip_unavailable_shards_num", 0, 0, "New setting to limit the number of shards that can be silently skipped when skip_unavailable_shards is enabled."}, {"max_skip_unavailable_shards_ratio", 0, 0, "New setting to limit the ratio of shards that can be silently skipped when skip_unavailable_shards is enabled."}, - {"allow_experimental_database_s3_tables", false, false, "New setting to enable experimental database S3 tables (AWS Iceberg REST catalog)."}, }); addSettingsChanges(settings_changes_history, "26.2", { diff --git a/src/Databases/DataLake/DatabaseDataLake.cpp b/src/Databases/DataLake/DatabaseDataLake.cpp index 7eda271ec0b3..37b9bd2cf8f0 100644 --- a/src/Databases/DataLake/DatabaseDataLake.cpp +++ b/src/Databases/DataLake/DatabaseDataLake.cpp @@ -334,7 +334,6 @@ void DatabaseDataLake::initialize() const url, settings[DatabaseDataLakeSetting::region].value, catalog_parameters, - settings[DatabaseDataLakeSetting::namespaces].value, Context::getGlobalContextInstance()); #else throw Exception( diff --git a/src/Databases/DataLake/ICatalog.cpp b/src/Databases/DataLake/ICatalog.cpp index 432b4d8b61c5..50200b3d3cc6 100644 --- a/src/Databases/DataLake/ICatalog.cpp +++ b/src/Databases/DataLake/ICatalog.cpp @@ -103,7 +103,18 @@ void TableMetadata::setLocation(const std::string & location_) auto pos_to_path = location_.substr(pos_to_bucket).find('/'); if (pos_to_path == std::string::npos) + { + /// An empty path is allowed for AWS S3 Tables: the table location is just `s3://`. + if (storage_type_str == "s3://") + { + location_without_path = location_; + path.clear(); + bucket = location_.substr(pos_to_bucket); + return; + } + throw DB::Exception(DB::ErrorCodes::NOT_IMPLEMENTED, "Unexpected location format: {}", location_); + } pos_to_path = pos_to_bucket + pos_to_path; @@ -284,8 +295,22 @@ std::string TableMetadata::getMetadataLocation(const std::string & iceberg_metad metadata_location = metadata_location.substr(storage_type_str.size()); if (data_location.starts_with(storage_type_str)) data_location = data_location.substr(storage_type_str.size()); - else if (!endpoint.empty() && data_location.starts_with(endpoint)) - data_location = data_location.substr(endpoint.size()); + else if (!endpoint.empty()) + { + std::string normalized_endpoint = endpoint; + if (normalized_endpoint.ends_with('/')) + normalized_endpoint.pop_back(); + + if (data_location.starts_with(normalized_endpoint)) + { + data_location = data_location.substr(normalized_endpoint.size()); + /// `metadata_location` is relative to the bucket (the `s3://` prefix is stripped above), + /// while `data_location` still has the leading slash left over from the endpoint, + /// e.g. "/bucket/table-uuid/". Drop it so that the prefix comparison below works. + if (azure_account_with_suffix.empty() && data_location.starts_with('/')) + data_location = data_location.substr(1); + } + } if (metadata_location.starts_with(data_location)) { diff --git a/src/Databases/DataLake/RestCatalog.cpp b/src/Databases/DataLake/RestCatalog.cpp index cec14e788371..8e2ff1d3f694 100644 --- a/src/Databases/DataLake/RestCatalog.cpp +++ b/src/Databases/DataLake/RestCatalog.cpp @@ -793,7 +793,6 @@ RestCatalog::Namespaces RestCatalog::getNamespaces(const std::string & base_name try { -<<<<<<< HEAD while (true) { /// The Iceberg REST OpenAPI spec uses `pageToken` (request) / `next-page-token` (response) @@ -804,6 +803,8 @@ RestCatalog::Namespaces RestCatalog::getNamespaces(const std::string & base_name if (!page_token.empty()) params.push_back({"pageToken", page_token}); + ProfileEvents::increment(ProfileEvents::DataLakeRestCatalogGetNamespaces); + auto timer = DB::CurrentThread::getProfileEvents().timer(ProfileEvents::DataLakeRestCatalogGetNamespacesMicroseconds); auto buf = createReadBuffer(config.prefix / NAMESPACES_ENDPOINT, params); String next_page_token; auto page_namespaces = parseNamespaces(*buf, base_namespace, next_page_token); @@ -832,14 +833,6 @@ RestCatalog::Namespaces RestCatalog::getNamespaces(const std::string & base_name page_token = std::move(next_page_token); } return all_namespaces; -======= - 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; ->>>>>>> 383c8d11e60 (Merge pull request #1868 from Altinity/fix/datalake-rest-catalog-profile-events) } catch (const DB::HTTPException & e) { @@ -944,7 +937,6 @@ DB::Names RestCatalog::getTables(const std::string & base_namespace, size_t limi auto encoded_namespace = encodeNamespaceForURI(base_namespace); const std::string endpoint = std::filesystem::path(NAMESPACES_ENDPOINT) / encoded_namespace / "tables"; -<<<<<<< HEAD DB::Names tables; String page_token; /// Cycle-detection guard: tracks every non-empty `next-page-token` we have seen on this @@ -963,6 +955,8 @@ DB::Names RestCatalog::getTables(const std::string & base_namespace, size_t limi if (!page_token.empty()) params.push_back({"pageToken", page_token}); + ProfileEvents::increment(ProfileEvents::DataLakeRestCatalogGetTables); + auto timer = DB::CurrentThread::getProfileEvents().timer(ProfileEvents::DataLakeRestCatalogGetTablesMicroseconds); auto buf = createReadBuffer(config.prefix / endpoint, params); /// Pass through the remaining limit so that single-page short-circuiting still works @@ -994,12 +988,6 @@ DB::Names RestCatalog::getTables(const std::string & base_namespace, size_t limi } return 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); ->>>>>>> 383c8d11e60 (Merge pull request #1868 from Altinity/fix/datalake-rest-catalog-profile-events) } DB::Names RestCatalog::parseTables(DB::ReadBuffer & buf, const std::string & base_namespace, size_t limit, String & next_page_token) const diff --git a/src/Databases/DataLake/S3TablesCatalog.cpp b/src/Databases/DataLake/S3TablesCatalog.cpp index e00a7de586c7..5b9e8f93fae8 100644 --- a/src/Databases/DataLake/S3TablesCatalog.cpp +++ b/src/Databases/DataLake/S3TablesCatalog.cpp @@ -33,7 +33,6 @@ namespace DB::ErrorCodes { extern const int BAD_ARGUMENTS; extern const int DATALAKE_DATABASE_ERROR; - extern const int CATALOG_NAMESPACE_DISABLED; } namespace DB::Setting @@ -64,9 +63,8 @@ S3TablesCatalog::S3TablesCatalog( const String & base_url_, const String & region_, const CatalogSettings & catalog_settings_, - const String & namespaces_, DB::ContextPtr context_) - : RestCatalog(warehouse_, base_url_, "", "", false, namespaces_, context_) + : RestCatalog(warehouse_, base_url_, "", "", false, context_) , region(region_) , storage_endpoint(catalog_settings_.storage_endpoint) { @@ -138,8 +136,6 @@ DB::Names S3TablesCatalog::getTables() const std::mutex mutex; for (const auto & ns : namespaces) { - if (!allowed_namespaces.isNamespaceAllowed(ns, /*nested*/ false)) - continue; runner.enqueueAndKeepTrack( [&, ns] { @@ -155,10 +151,9 @@ DB::Names S3TablesCatalog::getTables() const bool S3TablesCatalog::tryGetTableMetadata( const std::string & namespace_name, const std::string & table_name, - DB::ContextPtr context_, TableMetadata & result) const { - if (!RestCatalog::tryGetTableMetadata(namespace_name, table_name, context_, result)) + if (!RestCatalog::tryGetTableMetadata(namespace_name, table_name, result)) return false; if (!result.requiresCredentials()) @@ -220,11 +215,6 @@ ICatalog::CredentialsRefreshCallback S3TablesCatalog::getCredentialsConfiguratio void S3TablesCatalog::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=True"; diff --git a/src/Databases/DataLake/S3TablesCatalog.h b/src/Databases/DataLake/S3TablesCatalog.h index 45ad049f0199..4ac24c03747e 100644 --- a/src/Databases/DataLake/S3TablesCatalog.h +++ b/src/Databases/DataLake/S3TablesCatalog.h @@ -29,7 +29,6 @@ class S3TablesCatalog final : public RestCatalog const String & base_url_, const String & region_, const DataLake::CatalogSettings & catalog_settings_, - const String & namespaces_, DB::ContextPtr context_); DB::DatabaseDataLakeCatalogType getCatalogType() const override { return DB::DatabaseDataLakeCatalogType::S3_TABLES; } @@ -39,7 +38,6 @@ class S3TablesCatalog final : public RestCatalog bool tryGetTableMetadata( const std::string & namespace_name, const std::string & table_name, - DB::ContextPtr context_, TableMetadata & result) const override; void dropTable(const String & namespace_name, const String & table_name) const override; diff --git a/src/Databases/DataLake/UnityCatalog.cpp b/src/Databases/DataLake/UnityCatalog.cpp index 7374e21d086b..f3d7077b7923 100644 --- a/src/Databases/DataLake/UnityCatalog.cpp +++ b/src/Databases/DataLake/UnityCatalog.cpp @@ -121,7 +121,14 @@ Poco::JSON::Object::Ptr UnityCatalog::requestReadCredentials(const String & tabl request_body.set("operation", "READ"); auto callback = [&request_body] (std::ostream & os) { request_body.stringify(os); }; - auto [json, _] = postJSONRequest(TEMPORARY_CREDENTIALS_ENDPOINT, callback); + + Poco::Dynamic::Var json; + { + ProfileEvents::increment(ProfileEvents::DataLakeUnityCatalogGetCredentials); + auto timer = DB::CurrentThread::getProfileEvents().timer(ProfileEvents::DataLakeUnityCatalogGetCredentialsMicroseconds); + std::string _; + std::tie(json, _) = postJSONRequest(TEMPORARY_CREDENTIALS_ENDPOINT, callback); + } return json.extract(); } @@ -159,7 +166,6 @@ void UnityCatalog::getCredentials(const String & table_id, TableMetadata & metad std::shared_ptr creds; switch (storage_type) { -<<<<<<< HEAD case StorageType::S3: creds = parseS3Credentials(response); break; @@ -168,70 +174,6 @@ void UnityCatalog::getCredentials(const String & table_id, TableMetadata & metad break; default: break; -======= - case StorageType::S3: - { - auto callback = [table_id] (std::ostream & os) - { - Poco::JSON::Object obj; - obj.set("table_id", table_id); - obj.set("operation", "READ"); - obj.stringify(os); - }; - - Poco::Dynamic::Var json; - { - ProfileEvents::increment(ProfileEvents::DataLakeUnityCatalogGetCredentials); - auto timer = DB::CurrentThread::getProfileEvents().timer(ProfileEvents::DataLakeUnityCatalogGetCredentialsMicroseconds); - std::string _; - std::tie(json, _) = postJSONRequest(TEMPORARY_CREDENTIALS_ENDPOINT, callback); - } - const Poco::JSON::Object::Ptr & object = json.extract(); - - if (hasValueAndItsNotNone("aws_temp_credentials", object)) - { - const Poco::JSON::Object::Ptr & creds_object = object->getObject("aws_temp_credentials"); - std::string access_key_id = creds_object->get("access_key_id").extract(); - std::string secret_access_key = creds_object->get("secret_access_key").extract(); - std::string session_token = creds_object->get("session_token").extract(); - - auto creds = std::make_shared(access_key_id, secret_access_key, session_token); - metadata.setStorageCredentials(creds); - } - break; - } - case StorageType::Azure: - { - auto callback = [table_id] (std::ostream & os) - { - Poco::JSON::Object obj; - obj.set("table_id", table_id); - obj.set("operation", "READ"); - obj.stringify(os); - }; - - Poco::Dynamic::Var json; - { - ProfileEvents::increment(ProfileEvents::DataLakeUnityCatalogGetCredentials); - auto timer = DB::CurrentThread::getProfileEvents().timer(ProfileEvents::DataLakeUnityCatalogGetCredentialsMicroseconds); - std::string _; - std::tie(json, _) = postJSONRequest(TEMPORARY_CREDENTIALS_ENDPOINT, callback); - } - const Poco::JSON::Object::Ptr & object = json.extract(); - - if (hasValueAndItsNotNone("azure_user_delegation_sas", object)) - { - const Poco::JSON::Object::Ptr & creds_object = object->getObject("azure_user_delegation_sas"); - std::string sas_token = creds_object->get("sas_token").extract(); - - auto creds = std::make_shared(sas_token); - metadata.setStorageCredentials(creds); - } - break; - } - default: - break; ->>>>>>> 383c8d11e60 (Merge pull request #1868 from Altinity/fix/datalake-rest-catalog-profile-events) } if (creds) metadata.setStorageCredentials(creds); @@ -554,30 +496,7 @@ ICatalog::CredentialsRefreshCallback UnityCatalog::getCredentialsConfigurationCa return [this, unity_table_id] () -> std::shared_ptr { LOG_DEBUG(log, "Update credentials in the catalog"); -<<<<<<< HEAD return parseS3Credentials(requestReadCredentials(unity_table_id)); -======= - Poco::Dynamic::Var json; - { - ProfileEvents::increment(ProfileEvents::DataLakeUnityCatalogGetCredentials); - auto timer = DB::CurrentThread::getProfileEvents().timer(ProfileEvents::DataLakeUnityCatalogGetCredentialsMicroseconds); - std::string _; - std::tie(json, _) = postJSONRequest(TEMPORARY_CREDENTIALS_ENDPOINT, {}); - } - const Poco::JSON::Object::Ptr & object = json.extract(); - - if (hasValueAndItsNotNone("aws_temp_credentials", object)) - { - const Poco::JSON::Object::Ptr & creds_object = object->getObject("aws_temp_credentials"); - std::string access_key_id = creds_object->get("access_key_id").extract(); - std::string secret_access_key = creds_object->get("secret_access_key").extract(); - std::string session_token = creds_object->get("session_token").extract(); - - auto creds = std::make_shared(access_key_id, secret_access_key, session_token); - return creds; - } - return nullptr; ->>>>>>> 383c8d11e60 (Merge pull request #1868 from Altinity/fix/datalake-rest-catalog-profile-events) }; }