Merge branch 'antalya-25.8' into max_bytes_rows_per_file

zvonand · web-flow · commit 8a3179e5c4a6 · 2025-12-18T18:44:28.000+01:00
diff --git a/.github/workflows/master.yml b/.github/workflows/master.yml
@@ -4178,7 +4178,7 @@ jobs:
     secrets: inherit
     with:
       runner_type: altinity-on-demand, altinity-regression-tester
-      commit: 00a50b5b8f12c9c603b9a3fa17dd2c5ea2012cac
+      commit: 8d2c6d2072771d450a47faca38966da7493821e1
       arch: release
       build_sha: ${{ github.event_name == 'pull_request' && github.event.pull_request.head.sha || github.sha }}
       timeout_minutes: 300
@@ -4190,7 +4190,7 @@ jobs:
     secrets: inherit
     with:
       runner_type: altinity-on-demand, altinity-regression-tester-aarch64
-      commit: 00a50b5b8f12c9c603b9a3fa17dd2c5ea2012cac
+      commit: 8d2c6d2072771d450a47faca38966da7493821e1
       arch: aarch64
       build_sha: ${{ github.event_name == 'pull_request' && github.event.pull_request.head.sha || github.sha }}
       timeout_minutes: 300
diff --git a/.github/workflows/pull_request.yml b/.github/workflows/pull_request.yml
@@ -4134,7 +4134,7 @@ jobs:
     secrets: inherit
     with:
       runner_type: altinity-on-demand, altinity-regression-tester
-      commit: 00a50b5b8f12c9c603b9a3fa17dd2c5ea2012cac
+      commit: 8d2c6d2072771d450a47faca38966da7493821e1
       arch: release
       build_sha: ${{ github.event_name == 'pull_request' && github.event.pull_request.head.sha || github.sha }}
       timeout_minutes: 300
@@ -4146,7 +4146,7 @@ jobs:
     secrets: inherit
     with:
       runner_type: altinity-on-demand, altinity-regression-tester-aarch64
-      commit: 00a50b5b8f12c9c603b9a3fa17dd2c5ea2012cac
+      commit: 8d2c6d2072771d450a47faca38966da7493821e1
       arch: aarch64
       build_sha: ${{ github.event_name == 'pull_request' && github.event.pull_request.head.sha || github.sha }}
       timeout_minutes: 300
diff --git a/ci/praktika/yaml_additional_templates.py b/ci/praktika/yaml_additional_templates.py
@@ -35,7 +35,7 @@ class AltinityWorkflowTemplates:
           echo "Workflow Run Report: [View Report]($REPORT_LINK)" >> $GITHUB_STEP_SUMMARY
 """
     # Additional jobs
-    REGRESSION_HASH = "00a50b5b8f12c9c603b9a3fa17dd2c5ea2012cac"
+    REGRESSION_HASH = "8d2c6d2072771d450a47faca38966da7493821e1"
     ALTINITY_JOBS = {
         "GrypeScan": r"""
   GrypeScanServer:
diff --git a/src/Core/Settings.cpp b/src/Core/Settings.cpp
@@ -7111,9 +7111,6 @@ Use Shuffle aggregation strategy instead of PartialAggregation + Merge in distri
 )", EXPERIMENTAL) \
     DECLARE(Bool, allow_experimental_iceberg_read_optimization, true, R"(
 Allow Iceberg read optimization based on Iceberg metadata.
-)", EXPERIMENTAL) \
-    DECLARE(Bool, allow_retries_in_cluster_requests, false, R"(
-Allow retries in cluster request, when one node goes offline
 )", EXPERIMENTAL) \
     DECLARE(Bool, object_storage_remote_initiator, false, R"(
 Execute request to object storage as remote on one of object_storage_cluster nodes.
@@ -7236,6 +7233,7 @@ Sets the evaluation time to be used with promql dialect. 'auto' means the curren
     MAKE_OBSOLETE(M, Bool, allow_experimental_shared_set_join, true) \
     MAKE_OBSOLETE(M, UInt64, min_external_sort_block_bytes, 100_MiB) \
     MAKE_OBSOLETE(M, UInt64, distributed_cache_read_alignment, 0) \
+    MAKE_OBSOLETE(M, Bool, allow_retries_in_cluster_requests, false) \
     /** The section above is for obsolete settings. Do not add anything there. */
 #endif /// __CLION_IDE__
 
diff --git a/src/Databases/DataLake/RestCatalog.cpp b/src/Databases/DataLake/RestCatalog.cpp
@@ -28,6 +28,11 @@
 #include <Poco/URI.h>
 #include <Poco/JSON/Array.h>
 #include <Poco/JSON/Parser.h>
+#include <Poco/Net/HTTPClientSession.h>
+#include <Poco/Net/HTTPResponse.h>
+#include <Poco/Net/HTTPSClientSession.h>
+#include <Poco/Net/SSLManager.h>
+#include <Poco/StreamCopier.h>
 
 
 namespace DB::ErrorCodes
@@ -203,12 +208,11 @@ std::string RestCatalog::retrieveAccessToken() const
     /// 1. support oauth2-server-uri
     /// https://github.com/apache/iceberg/blob/918f81f3c3f498f46afcea17c1ac9cdc6913cb5c/open-api/rest-catalog-open-api.yaml#L183C82-L183C99
 
-    DB::HTTPHeaderEntries headers;
-    headers.emplace_back("Content-Type", "application/x-www-form-urlencoded");
-    headers.emplace_back("Accepts", "application/json; charset=UTF-8");
-
     Poco::URI url;
     DB::ReadWriteBufferFromHTTP::OutStreamCallback out_stream_callback;
+    size_t body_size = 0;
+    String body;
+
     if (oauth_server_uri.empty() && !oauth_server_use_request_body)
     {
         url = Poco::URI(base_url / oauth_tokens_endpoint);
@@ -223,11 +227,20 @@ std::string RestCatalog::retrieveAccessToken() const
     }
     else
     {
+        String encoded_auth_scope;
+        String encoded_client_id;
+        String encoded_client_secret;
+        Poco::URI::encode(auth_scope, auth_scope, encoded_auth_scope);
+        Poco::URI::encode(client_id, client_id, encoded_client_id);
+        Poco::URI::encode(client_secret, client_secret, encoded_client_secret);
+
+        body = fmt::format(
+            "grant_type=client_credentials&scope={}&client_id={}&client_secret={}",
+            encoded_auth_scope, encoded_client_id, encoded_client_secret);
+        body_size = body.size();
         out_stream_callback = [&](std::ostream & os)
         {
-            os << fmt::format(
-                "grant_type=client_credentials&scope={}&client_id={}&client_secret={}",
-                auth_scope, client_id, client_secret);
+            os << body;
         };
 
         if (oauth_server_uri.empty())
@@ -237,19 +250,23 @@ std::string RestCatalog::retrieveAccessToken() const
     }
 
     const auto & context = getContext();
-    auto wb = DB::BuilderRWBufferFromHTTP(url)
-        .withConnectionGroup(DB::HTTPConnectionGroupType::HTTP)
-        .withMethod(Poco::Net::HTTPRequest::HTTP_POST)
-        .withSettings(context->getReadSettings())
-        .withTimeouts(DB::ConnectionTimeouts::getHTTPTimeouts(context->getSettingsRef(), context->getServerSettings()))
-        .withHostFilter(&context->getRemoteHostFilter())
-        .withOutCallback(std::move(out_stream_callback))
-        .withSkipNotFound(false)
-        .withHeaders(headers)
-        .create(credentials);
+    auto timeouts = DB::ConnectionTimeouts::getHTTPTimeouts(context->getSettingsRef(), context->getServerSettings());
+    auto session = makeHTTPSession(DB::HTTPConnectionGroupType::HTTP, url, timeouts, {});
+
+    Poco::Net::HTTPRequest request(Poco::Net::HTTPRequest::HTTP_POST, url.getPathAndQuery(),
+                                Poco::Net::HTTPMessage::HTTP_1_1);
+    request.setContentType("application/x-www-form-urlencoded");
+    request.setContentLength(body_size);
+    request.set("Accept", "application/json");
+
+    std::ostream & os = session->sendRequest(request);
+    out_stream_callback(os);
+
+    Poco::Net::HTTPResponse response;
+    std::istream & rs = session->receiveResponse(response);
 
     std::string json_str;
-    readJSONObjectPossiblyInvalid(json_str, *wb);
+    Poco::StreamCopier::copyToString(rs, json_str);
 
     Poco::JSON::Parser parser;
     Poco::Dynamic::Var res_json = parser.parse(json_str);
diff --git a/src/QueryPipeline/RemoteQueryExecutor.cpp b/src/QueryPipeline/RemoteQueryExecutor.cpp
@@ -52,7 +52,6 @@ namespace Setting
     extern const SettingsBool use_hedged_requests;
     extern const SettingsBool push_external_roles_in_interserver_queries;
     extern const SettingsMilliseconds parallel_replicas_connect_timeout_ms;
-    extern const SettingsBool allow_retries_in_cluster_requests;
 }
 
 namespace ErrorCodes
@@ -83,7 +82,6 @@ RemoteQueryExecutor::RemoteQueryExecutor(
     , extension(extension_)
     , priority_func(priority_func_)
     , read_packet_type_separately(context->canUseParallelReplicasOnInitiator() && !context->getSettingsRef()[Setting::use_hedged_requests])
-    , allow_retries_in_cluster_requests(context->getSettingsRef()[Setting::allow_retries_in_cluster_requests])
 {
     if (stage == QueryProcessingStage::QueryPlan && !query_plan)
         throw Exception(ErrorCodes::LOGICAL_ERROR, "Query plan is not passed for QueryPlan processing stage");
@@ -469,8 +467,7 @@ int RemoteQueryExecutor::sendQueryAsync()
         read_context = std::make_unique<ReadContext>(
             *this,
             /*suspend_when_query_sent*/ true,
-            read_packet_type_separately,
-            allow_retries_in_cluster_requests);
+            read_packet_type_separately);
 
     /// If query already sent, do nothing. Note that we cannot use sent_query flag here,
     /// because we can still be in process of sending scalars or external tables.
@@ -543,8 +540,7 @@ RemoteQueryExecutor::ReadResult RemoteQueryExecutor::readAsync()
         read_context = std::make_unique<ReadContext>(
             *this,
             /*suspend_when_query_sent*/ false,
-            read_packet_type_separately,
-            allow_retries_in_cluster_requests);
+            read_packet_type_separately);
         recreate_read_context = false;
     }
 
@@ -735,16 +731,13 @@ RemoteQueryExecutor::ReadResult RemoteQueryExecutor::processPacket(Packet packet
             break;
 
         case Protocol::Server::ConnectionLost:
-            if (allow_retries_in_cluster_requests)
+            if (extension && extension->task_iterator && extension->task_iterator->supportRerunTask() && extension->replica_info)
             {
-                if (extension && extension->task_iterator && extension->task_iterator->supportRerunTask() && extension->replica_info)
+                if (!replica_has_processed_data.contains(extension->replica_info->number_of_current_replica))
                 {
-                    if (!replica_has_processed_data.contains(extension->replica_info->number_of_current_replica))
-                    {
-                        finished = true;
-                        extension->task_iterator->rescheduleTasksFromReplica(extension->replica_info->number_of_current_replica);
-                        return ReadResult(Block{});
-                    }
+                    finished = true;
+                    extension->task_iterator->rescheduleTasksFromReplica(extension->replica_info->number_of_current_replica);
+                    return ReadResult(Block{});
                 }
             }
             packet.exception->rethrow();
diff --git a/src/QueryPipeline/RemoteQueryExecutor.h b/src/QueryPipeline/RemoteQueryExecutor.h
@@ -339,8 +339,6 @@ class RemoteQueryExecutor
 
     const bool read_packet_type_separately = false;
 
-    const bool allow_retries_in_cluster_requests = false;
-
     std::unordered_set<size_t> replica_has_processed_data;
 
     /// Send all scalars to remote servers
diff --git a/src/QueryPipeline/RemoteQueryExecutorReadContext.cpp b/src/QueryPipeline/RemoteQueryExecutorReadContext.cpp
@@ -22,13 +22,11 @@ namespace ErrorCodes
 RemoteQueryExecutorReadContext::RemoteQueryExecutorReadContext(
     RemoteQueryExecutor & executor_,
     bool suspend_when_query_sent_,
-    bool read_packet_type_separately_,
-    bool allow_retries_in_cluster_requests_)
+    bool read_packet_type_separately_)
     : AsyncTaskExecutor(std::make_unique<Task>(*this))
     , executor(executor_)
     , suspend_when_query_sent(suspend_when_query_sent_)
     , read_packet_type_separately(read_packet_type_separately_)
-    , allow_retries_in_cluster_requests(allow_retries_in_cluster_requests_)
 {
     if (-1 == pipe2(pipe_fd, O_NONBLOCK))
         throw ErrnoException(ErrorCodes::CANNOT_OPEN_FILE, "Cannot create pipe");
@@ -63,46 +61,38 @@ void RemoteQueryExecutorReadContext::Task::run(AsyncCallback async_callback, Sus
     {
         while (true)
         {
-            try
-            {
-                read_context.has_read_packet_part = PacketPart::None;
-
-                if (read_context.read_packet_type_separately)
-                {
-                    read_context.packet.type = read_context.executor.getConnections().receivePacketTypeUnlocked(async_callback);
-                    read_context.has_read_packet_part = PacketPart::Type;
-                    suspend_callback();
-                }
-                read_context.packet = read_context.executor.getConnections().receivePacketUnlocked(async_callback);
-                read_context.has_read_packet_part = PacketPart::Body;
-                if (read_context.packet.type == Protocol::Server::Data)
-                    read_context.has_data_packets = true;
-            }
-            catch (const Exception & e)
+            read_context.has_read_packet_part = PacketPart::None;
+
+            if (read_context.read_packet_type_separately)
             {
-                /// If cluster node unxepectedly shutted down (kill/segfault/power off/etc.) socket just closes.
-                /// If initiator did not process any data packets before, this fact can be ignored.
-                /// Unprocessed tasks will be executed on other nodes.
-                if (e.code() == ErrorCodes::ATTEMPT_TO_READ_AFTER_EOF
-                    && !read_context.has_data_packets.load() && read_context.executor.skipUnavailableShards())
-                {
-                    read_context.has_read_packet_part = PacketPart::None;
-                }
-                else
-                    throw;
+                read_context.packet.type = read_context.executor.getConnections().receivePacketTypeUnlocked(async_callback);
+                read_context.has_read_packet_part = PacketPart::Type;
+                suspend_callback();
             }
+            read_context.packet = read_context.executor.getConnections().receivePacketUnlocked(async_callback);
+            read_context.has_read_packet_part = PacketPart::Body;
+            if (read_context.packet.type == Protocol::Server::Data)
+                read_context.has_data_packets = true;
 
             suspend_callback();
         }
     }
-    catch (const Exception &)
+    catch (const Exception & e)
     {
-        if (!read_context.allow_retries_in_cluster_requests)
+        /// If cluster node unxepectedly shutted down (kill/segfault/power off/etc.) socket just closes.
+        /// If initiator did not process any data packets before, this fact can be ignored.
+        /// Unprocessed tasks will be executed on other nodes.
+        if (e.code() == ErrorCodes::ATTEMPT_TO_READ_AFTER_EOF
+            && !read_context.has_data_packets.load()
+            && read_context.executor.skipUnavailableShards())
+        {
+            read_context.packet.type = Protocol::Server::ConnectionLost;
+            read_context.packet.exception = std::make_unique<Exception>(getCurrentExceptionMessageAndPattern(true), getCurrentExceptionCode());
+            read_context.has_read_packet_part = PacketPart::Body;
+            suspend_callback();
+        }
+        else
             throw;
-        read_context.packet.type = Protocol::Server::ConnectionLost;
-        read_context.packet.exception = std::make_unique<Exception>(getCurrentExceptionMessageAndPattern(true), getCurrentExceptionCode());
-        read_context.has_read_packet_part = PacketPart::Body;
-        suspend_callback();
     }
 }
 
diff --git a/src/QueryPipeline/RemoteQueryExecutorReadContext.h b/src/QueryPipeline/RemoteQueryExecutorReadContext.h
@@ -28,8 +28,7 @@ class RemoteQueryExecutorReadContext : public AsyncTaskExecutor
     explicit RemoteQueryExecutorReadContext(
         RemoteQueryExecutor & executor_,
         bool suspend_when_query_sent_,
-        bool read_packet_type_separately_,
-        bool allow_retries_in_cluster_requests_);
+        bool read_packet_type_separately_);
 
     ~RemoteQueryExecutorReadContext() override;
 
@@ -112,7 +111,6 @@ class RemoteQueryExecutorReadContext : public AsyncTaskExecutor
     bool suspend_when_query_sent = false;
     bool is_query_sent = false;
     const bool read_packet_type_separately = false;
-    const bool allow_retries_in_cluster_requests = false;
 };
 
 }
diff --git a/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp b/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp
@@ -396,7 +396,15 @@ void StorageObjectStorageCluster::updateQueryToSendIfNeeded(
     }
 
     ASTPtr object_storage_type_arg;
-    configuration->extractDynamicStorageType(args, context, &object_storage_type_arg);
+    if (cluster_name_in_settings)
+        configuration->extractDynamicStorageType(args, context, &object_storage_type_arg);
+    else
+    {
+        auto args_copy = args;
+        // Remove cluster name from args to avoid confusing cluster name and named collection name
+        args_copy.erase(args_copy.begin());
+        configuration->extractDynamicStorageType(args_copy, context, &object_storage_type_arg);
+    }
     ASTPtr settings_temporary_storage = nullptr;
     for (auto * it = args.begin(); it != args.end(); ++it)
     {
diff --git a/tests/integration/helpers/iceberg_utils.py b/tests/integration/helpers/iceberg_utils.py
@@ -196,6 +196,7 @@ def get_creation_expression(
     explicit_metadata_path="",
     storage_type_as_arg=False,
     storage_type_in_named_collection=False,
+    cluster_name_as_literal=True,
     additional_settings = [],
     **kwargs,
 ):
@@ -224,6 +225,8 @@ def get_creation_expression(
     else:
         settings_expression = ""
 
+    cluster_name = "'cluster_simple'" if cluster_name_as_literal else "cluster_simple"
+
     storage_arg = storage_type
     engine_part = ""
     if (storage_type_in_named_collection):
@@ -252,7 +255,7 @@ def get_creation_expression(
 
         if run_on_cluster:
             assert table_function
-            return f"iceberg{engine_part}Cluster('cluster_simple', {storage_arg}, filename = 'iceberg_data/default/{table_name}/', format={format}, url = 'http://minio1:9001/{bucket}/')"
+            return f"iceberg{engine_part}Cluster({cluster_name}, {storage_arg}, filename = 'iceberg_data/default/{table_name}/', format={format}, url = 'http://minio1:9001/{bucket}/')"
         else:
             if table_function:
                 return f"iceberg{engine_part}({storage_arg}, filename = 'iceberg_data/default/{table_name}/', format={format}, url = 'http://minio1:9001/{bucket}/')"
@@ -271,7 +274,7 @@ def get_creation_expression(
         if run_on_cluster:
             assert table_function
             return f"""
-                iceberg{engine_part}Cluster('cluster_simple', {storage_arg}, container = '{cluster.azure_container_name}', storage_account_url = '{cluster.env_variables["AZURITE_STORAGE_ACCOUNT_URL"]}', blob_path = '/iceberg_data/default/{table_name}/', format={format})
+                iceberg{engine_part}Cluster({cluster_name}, {storage_arg}, container = '{cluster.azure_container_name}', storage_account_url = '{cluster.env_variables["AZURITE_STORAGE_ACCOUNT_URL"]}', blob_path = '/iceberg_data/default/{table_name}/', format={format})
             """
         else:
             if table_function:
@@ -293,7 +296,7 @@ def get_creation_expression(
         if run_on_cluster:
             assert table_function
             return f"""
-                iceberg{engine_part}Cluster('cluster_simple', {storage_arg}, path = '/iceberg_data/default/{table_name}/', format={format})
+                iceberg{engine_part}Cluster({cluster_name}, {storage_arg}, path = '/iceberg_data/default/{table_name}/', format={format})
             """
         else:
             if table_function:
diff --git a/tests/integration/test_storage_iceberg/test.py b/tests/integration/test_storage_iceberg/test.py
