Antalya 26.6 Backport of #113505: Added support for inserting data to iceberg s3 tables(AWS)

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
 

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);

src/Databases/DataLake/GlueCatalog.cpp

@@ -888,7 +888,7 @@ bool GlueCatalog::updateSchema(
     return updateTableInGlue(namespace_name, table_name, new_metadata_path, columns);
 }
 
-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,

src/Databases/DataLake/GlueCatalog.h

@@ -82,7 +82,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

src/Databases/DataLake/ICatalog.cpp

@@ -368,7 +368,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");
 }

src/Databases/DataLake/ICatalog.h

@@ -239,7 +239,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".

src/Databases/DataLake/RestCatalog.cpp

@@ -2121,7 +2121,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,

src/Databases/DataLake/RestCatalog.h

@@ -103,7 +103,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;
 

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
@@ -160,13 +161,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<S3Credentials>(storage_credentials);
+        auto creds = std::dynamic_pointer_cast<S3Credentials>(result.getStorageCredentials());
         if (creds && !creds->isEmpty())
             need_credentials = false;
     }
@@ -181,6 +186,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<S3Credentials>(
             aws_creds.GetAWSAccessKeyId(), aws_creds.GetAWSSecretKey(), aws_creds.GetSessionToken()));
     }
@@ -194,6 +200,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;
 }
 
@@ -217,8 +237,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 auto state_snapshot = state.get();
     const std::string endpoint
         = (base_url / state_snapshot->config.prefix / "namespaces" / namespace_name / "tables" / table_name).string()

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;
 

src/IO/S3/PocoHTTPClient.cpp

@@ -578,6 +578,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,

src/Storages/ObjectStorage/StorageObjectStorage.cpp

@@ -59,6 +59,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
@@ -1015,13 +1016,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<ReadBufferIterator> StorageObjectStorage::createReadBufferIterator(