diff --git a/base/poco/Net/include/Poco/Net/HTTPRequest.h b/base/poco/Net/include/Poco/Net/HTTPRequest.h index 269167feb834..1ec89520677c 100644 --- a/base/poco/Net/include/Poco/Net/HTTPRequest.h +++ b/base/poco/Net/include/Poco/Net/HTTPRequest.h @@ -157,6 +157,16 @@ namespace Net static const std::string UPGRADE; static const std::string EXPECT; + void setSuppressKeepAliveHeader(bool suppress); + /// Suppresses the hop-by-hop `Keep-Alive: timeout=..., max=...` request header that + /// HTTPClientSession::sendRequest would otherwise add. The header is deprecated by RFC 7230 + /// and some peers reject it outright: Amazon S3 Tables answers `S3TablesUnsupportedHeader` + /// to any PUT carrying it. Keep-alive itself and connection reuse are unaffected -- they are + /// driven by the `Connection` header and by ClickHouse's own connection pool. + + bool getSuppressKeepAliveHeader() const; + /// Returns true if the `Keep-Alive` request header is suppressed for this request. + protected: void getCredentials(const std::string & header, std::string & scheme, std::string & authInfo) const; /// Returns the authentication scheme and additional authentication @@ -179,6 +189,7 @@ namespace Net std::string _method; std::string _uri; + bool _suppress_keep_alive_header = false; HTTPRequest(const HTTPRequest &); HTTPRequest & operator=(const HTTPRequest &); @@ -200,6 +211,18 @@ namespace Net } + inline void HTTPRequest::setSuppressKeepAliveHeader(bool suppress) + { + _suppress_keep_alive_header = suppress; + } + + + inline bool HTTPRequest::getSuppressKeepAliveHeader() const + { + return _suppress_keep_alive_header; + } + + } } // namespace Poco::Net diff --git a/base/poco/Net/src/HTTPClientSession.cpp b/base/poco/Net/src/HTTPClientSession.cpp index 209eeba75e0e..1754f59d8e5c 100644 --- a/base/poco/Net/src/HTTPClientSession.cpp +++ b/base/poco/Net/src/HTTPClientSession.cpp @@ -283,7 +283,8 @@ std::ostream& HTTPClientSession::sendRequest(HTTPRequest& request, uint64_t * co reconnect(connect_time); if (!request.has(HTTPMessage::CONNECTION)) request.setKeepAlive(keepAlive); - if (keepAlive && !request.has(HTTPMessage::CONNECTION_KEEP_ALIVE) && _keepAliveTimeout.totalSeconds() > 0) + if (keepAlive && !request.getSuppressKeepAliveHeader() && !request.has(HTTPMessage::CONNECTION_KEEP_ALIVE) + && _keepAliveTimeout.totalSeconds() > 0) request.setKeepAliveTimeout(_keepAliveTimeout.totalSeconds(), _keepAliveMaxRequests); if (!request.has(HTTPRequest::HOST) && !_host.empty()) request.setHost(_host, _port); diff --git a/src/Databases/DataLake/GlueCatalog.cpp b/src/Databases/DataLake/GlueCatalog.cpp index a8dda240e7ce..0494f7651db7 100644 --- a/src/Databases/DataLake/GlueCatalog.cpp +++ b/src/Databases/DataLake/GlueCatalog.cpp @@ -753,7 +753,7 @@ bool GlueCatalog::updateSchema( return updateMetadata(namespace_name, table_name, new_metadata_path, nullptr); } -void GlueCatalog::dropTable(const String & namespace_name, const String & table_name) const +void GlueCatalog::dropTable(const String & namespace_name, const String & table_name, bool /*delete_data*/) const { if (!isNamespaceAllowed(namespace_name)) throw DB::Exception(DB::ErrorCodes::CATALOG_NAMESPACE_DISABLED, diff --git a/src/Databases/DataLake/GlueCatalog.h b/src/Databases/DataLake/GlueCatalog.h index 8d10ba0c8667..9ed2bea3828f 100644 --- a/src/Databases/DataLake/GlueCatalog.h +++ b/src/Databases/DataLake/GlueCatalog.h @@ -79,7 +79,7 @@ class GlueCatalog final : public ICatalog, private DB::WithContext Int32 new_last_column_id, Poco::JSON::Object::Ptr metadata = nullptr) const override; - void dropTable(const String & namespace_name, const String & table_name) const override; + void dropTable(const String & namespace_name, const String & table_name, bool delete_data) const override; /// Returns a callback that re-vends fresh AWS credentials from the configured /// credentials provider chain. Invoked by `ReadBufferFromS3` when an S3 call diff --git a/src/Databases/DataLake/ICatalog.cpp b/src/Databases/DataLake/ICatalog.cpp index 62cf44930225..226f4457402f 100644 --- a/src/Databases/DataLake/ICatalog.cpp +++ b/src/Databases/DataLake/ICatalog.cpp @@ -357,7 +357,7 @@ bool ICatalog::updateSchema( throw DB::Exception(DB::ErrorCodes::NOT_IMPLEMENTED, "updateSchema is not implemented"); } -void ICatalog::dropTable(const String & /*namespace_name*/, const String & /*table_name*/) const +void ICatalog::dropTable(const String & /*namespace_name*/, const String & /*table_name*/, bool /*delete_data*/) const { throw DB::Exception(DB::ErrorCodes::NOT_IMPLEMENTED, "dropTable is not implemented"); } diff --git a/src/Databases/DataLake/ICatalog.h b/src/Databases/DataLake/ICatalog.h index f77bfcff0405..893c2ce8fa3e 100644 --- a/src/Databases/DataLake/ICatalog.h +++ b/src/Databases/DataLake/ICatalog.h @@ -212,7 +212,7 @@ class ICatalog Poco::JSON::Object::Ptr metadata = nullptr) const; /// Drop table from catalog. - virtual void dropTable(const String & namespace_name, const String & table_name) const; + virtual void dropTable(const String & namespace_name, const String & table_name, bool delete_data) const; /// Does the catalog support transactions or anything like that? /// For example, the Iceberg REST catalog supports atomic operations "compare if snapshot X is equal to" and "add new snapshot Y". diff --git a/src/Databases/DataLake/RestCatalog.cpp b/src/Databases/DataLake/RestCatalog.cpp index 58bf459992f0..be751924d052 100644 --- a/src/Databases/DataLake/RestCatalog.cpp +++ b/src/Databases/DataLake/RestCatalog.cpp @@ -1785,7 +1785,7 @@ bool RestCatalog::updateSchema( return true; } -void RestCatalog::dropTable(const String & namespace_name, const String & table_name) const +void RestCatalog::dropTable(const String & namespace_name, const String & table_name, bool /*delete_data*/) const { if (!allowed_namespaces.isNamespaceAllowed(namespace_name, /*nested*/ false)) throw DB::Exception(DB::ErrorCodes::CATALOG_NAMESPACE_DISABLED, diff --git a/src/Databases/DataLake/RestCatalog.h b/src/Databases/DataLake/RestCatalog.h index 4eb33d1045ab..f85449dc34b9 100644 --- a/src/Databases/DataLake/RestCatalog.h +++ b/src/Databases/DataLake/RestCatalog.h @@ -102,7 +102,7 @@ class RestCatalog : public ICatalog, public DB::WithContext bool isTransactional() const override { return true; } - void dropTable(const String & namespace_name, const String & table_name) const override; + void dropTable(const String & namespace_name, const String & table_name, bool delete_data) const override; ICatalog::CredentialsRefreshCallback getCredentialsConfigurationCallback(const DB::StorageID & storage_id) override; diff --git a/src/Databases/DataLake/S3TablesCatalog.cpp b/src/Databases/DataLake/S3TablesCatalog.cpp index 07cd7e723da3..46d3c287b341 100644 --- a/src/Databases/DataLake/S3TablesCatalog.cpp +++ b/src/Databases/DataLake/S3TablesCatalog.cpp @@ -33,6 +33,7 @@ namespace DB::ErrorCodes { extern const int BAD_ARGUMENTS; extern const int DATALAKE_DATABASE_ERROR; + extern const int SUPPORT_IS_DISABLED; } namespace DB::Setting @@ -157,13 +158,17 @@ bool S3TablesCatalog::tryGetTableMetadata( if (!RestCatalog::tryGetTableMetadata(namespace_name, table_name, context_, result)) return false; - if (!result.requiresCredentials()) - return true; + /// For S3 Tables the catalog and the underlying data live in AWS S3 under the same + /// AWS principal, so endpoint/metadata-location normalization and fallback IAM + /// credential injection must always run - even when the engine was created with + /// `vended_credentials=0` (which leaves `requiresCredentials()` == false). Otherwise + /// reads and inserts can lose the catalog endpoint/credentials and fail with auth + /// or routing errors. bool need_credentials = true; - if (const auto storage_credentials = result.getStorageCredentials()) + if (result.hasStorageCredentials()) { - auto creds = std::dynamic_pointer_cast(storage_credentials); + auto creds = std::dynamic_pointer_cast(result.getStorageCredentials()); if (creds && !creds->isEmpty()) need_credentials = false; } @@ -178,6 +183,7 @@ bool S3TablesCatalog::tryGetTableMetadata( "S3 Tables: catalog IAM credentials are empty for {}.{}, " "check AWS credentials configuration", namespace_name, table_name); + result.withStorageCredentials(); result.setStorageCredentials(std::make_shared( aws_creds.GetAWSAccessKeyId(), aws_creds.GetAWSSecretKey(), aws_creds.GetSessionToken())); } @@ -191,6 +197,20 @@ bool S3TablesCatalog::tryGetTableMetadata( result.setEndpoint(endpoint); } + if (auto props = result.getDataLakeSpecificProperties(); + props && !props->iceberg_metadata_file_location.empty()) + { + const String & loc = props->iceberg_metadata_file_location; + auto scheme_end = loc.find("://"); + if (scheme_end != String::npos) + { + auto path_start = loc.find('/', scheme_end + 3); + if (path_start != String::npos) + props->iceberg_metadata_file_location = loc.substr(path_start + 1); + } + result.setDataLakeSpecificProperties(std::move(props)); + } + return true; } @@ -214,8 +234,16 @@ ICatalog::CredentialsRefreshCallback S3TablesCatalog::getCredentialsConfiguratio }; } -void S3TablesCatalog::dropTable(const String & namespace_name, const String & table_name) const +void S3TablesCatalog::dropTable(const String & namespace_name, const String & table_name, bool delete_data) const { + /// https://docs.aws.amazon.com/AmazonS3/latest/userguide/s3-tables-delete.html + if (!delete_data) + throw DB::Exception( + DB::ErrorCodes::SUPPORT_IS_DISABLED, + "S3 Tables cannot drop table {}.{} without deleting its data, and `iceberg_delete_data_on_drop` is disabled. " + "Enable `iceberg_delete_data_on_drop` to drop the table together with its data", + namespace_name, table_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 aff432c1b679..dc0cc71cd108 100644 --- a/src/Databases/DataLake/S3TablesCatalog.h +++ b/src/Databases/DataLake/S3TablesCatalog.h @@ -41,7 +41,7 @@ class S3TablesCatalog final : public RestCatalog DB::ContextPtr context_, TableMetadata & result) const override; - void dropTable(const String & namespace_name, const String & table_name) const override; + void dropTable(const String & namespace_name, const String & table_name, bool delete_data) const override; ICatalog::CredentialsRefreshCallback getCredentialsConfigurationCallback(const DB::StorageID & storage_id) override; diff --git a/src/IO/S3/PocoHTTPClient.cpp b/src/IO/S3/PocoHTTPClient.cpp index 2b76300bbbc9..5e97d4cc0781 100644 --- a/src/IO/S3/PocoHTTPClient.cpp +++ b/src/IO/S3/PocoHTTPClient.cpp @@ -566,6 +566,7 @@ void PocoHTTPClient::makeRequestInternalImpl( Poco::Net::HTTPRequest poco_request(Poco::Net::HTTPRequest::HTTP_1_1); + poco_request.setSuppressKeepAliveHeader(true); /** According to RFC-2616, Request-URI is allowed to be encoded. * However, there is no clear agreement on which exact symbols must be encoded. * Effectively, `Poco::URI` chooses smaller subset of characters to encode, diff --git a/src/Storages/ObjectStorage/StorageObjectStorage.cpp b/src/Storages/ObjectStorage/StorageObjectStorage.cpp index 01e4a1d4ff6e..3dfeff8a50fb 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorage.cpp +++ b/src/Storages/ObjectStorage/StorageObjectStorage.cpp @@ -54,6 +54,7 @@ namespace Setting extern const SettingsInt64 delta_lake_snapshot_start_version; extern const SettingsInt64 delta_lake_snapshot_end_version; extern const SettingsUInt64 max_streams_for_files_processing_in_cluster_functions; + extern const SettingsBool iceberg_delete_data_on_drop; } namespace ErrorCodes @@ -833,13 +834,14 @@ void StorageObjectStorage::truncate( void StorageObjectStorage::drop() { + /// We cannot use query context here, because drop is executed in the background. + auto drop_context = Context::getGlobalContextInstance(); if (catalog) { const auto [namespace_name, table_name] = DataLake::parseTableName(storage_id.getTableName()); - catalog->dropTable(namespace_name, table_name); + catalog->dropTable(namespace_name, table_name, drop_context->getSettingsRef()[Setting::iceberg_delete_data_on_drop]); } - /// We cannot use query context here, because drop is executed in the background. - configuration->drop(Context::getGlobalContextInstance()); + configuration->drop(drop_context); } std::unique_ptr StorageObjectStorage::createReadBufferIterator(