Merge branch 'antalya-25.8' into feature/antalya-25.8/s3cluster_global_join_fixes

Author: ianton-ru

.github/workflows/master.yml

@@ -4178,7 +4178,7 @@ jobs:
     secrets: inherit
     with:
       runner_type: altinity-on-demand, altinity-regression-tester
-      commit: 3fbe58a0ebe8fa5f97b7f36c45a2a69b1d3b6568
+      commit: fc19ce3a7322a10ab791de755c950a56744a12e7
       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: 3fbe58a0ebe8fa5f97b7f36c45a2a69b1d3b6568
+      commit: fc19ce3a7322a10ab791de755c950a56744a12e7
       arch: aarch64
       build_sha: ${{ github.event_name == 'pull_request' && github.event.pull_request.head.sha || github.sha }}
       timeout_minutes: 300

.github/workflows/pull_request.yml

@@ -4134,7 +4134,7 @@ jobs:
     secrets: inherit
     with:
       runner_type: altinity-on-demand, altinity-regression-tester
-      commit: 3fbe58a0ebe8fa5f97b7f36c45a2a69b1d3b6568
+      commit: fc19ce3a7322a10ab791de755c950a56744a12e7
       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: 3fbe58a0ebe8fa5f97b7f36c45a2a69b1d3b6568
+      commit: fc19ce3a7322a10ab791de755c950a56744a12e7
       arch: aarch64
       build_sha: ${{ github.event_name == 'pull_request' && github.event.pull_request.head.sha || github.sha }}
       timeout_minutes: 300

.gitmodules

@@ -6,7 +6,7 @@
 	url = https://github.com/Thalhammer/jwt-cpp
 [submodule "contrib/zstd"]
 	path = contrib/zstd
-	url = https://github.com/facebook/zstd
+	url = https://github.com/ClickHouse/zstd.git
 [submodule "contrib/lz4"]
 	path = contrib/lz4
 	url = https://github.com/lz4/lz4
@@ -45,7 +45,7 @@
 	url = https://github.com/ClickHouse/arrow
 [submodule "contrib/thrift"]
 	path = contrib/thrift
-	url = https://github.com/apache/thrift
+	url = https://github.com/ClickHouse/thrift.git
 [submodule "contrib/libhdfs3"]
 	path = contrib/libhdfs3
 	url = https://github.com/ClickHouse/libhdfs3

ci/praktika/native_jobs.py

@@ -18,6 +18,7 @@
 from .runtime import RunConfig
 from .settings import Settings
 from .utils import Shell, Utils
+from ci.defs.defs import ArtifactNames
 
 assert Settings.CI_CONFIG_RUNS_ON
 
@@ -403,17 +404,22 @@ def check_affected_jobs():
             # NOTE (strtgbb): We always want these build artifacts for our report and regression tests.
             # If we make FinishCIReport and regression tests into praktika jobs, we can remove this.
             if "CIReport" in workflow.additional_jobs:
-                all_required_artifacts.update(["CH_AMD_RELEASE", "CH_ARM_RELEASE"])
+                all_required_artifacts.update(
+                    [
+                        ArtifactNames.CH_AMD_RELEASE,
+                        ArtifactNames.CH_ARM_RELEASE,
+                    ]
+                )
             if (
                 "Regression" in workflow.additional_jobs
                 and "regression"
                 not in workflow_config.custom_data.get("ci_exclude_tags", [])
             ):
-                all_required_artifacts.update(["CH_AMD_BINARY"])
+                all_required_artifacts.update([ArtifactNames.CH_AMD_BINARY])
                 if "aarch64" not in workflow_config.custom_data.get(
                     "ci_exclude_tags", []
                 ):
-                    all_required_artifacts.update(["CH_ARM_BINARY"])
+                    all_required_artifacts.update([ArtifactNames.CH_ARM_BINARY])
             print(f"Including artifacts for custom jobs [{all_required_artifacts}]")
 
             for job in workflow.jobs:

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 = "3fbe58a0ebe8fa5f97b7f36c45a2a69b1d3b6568"
+    REGRESSION_HASH = "fc19ce3a7322a10ab791de755c950a56744a12e7"
     ALTINITY_JOBS = {
         "GrypeScan": r"""
   GrypeScanServer:

src/Client/MultiplexedConnections.cpp

@@ -232,7 +232,7 @@ void MultiplexedConnections::sendIgnoredPartUUIDs(const std::vector<UUID> & uuid
 void MultiplexedConnections::sendClusterFunctionReadTaskResponse(const ClusterFunctionReadTaskResponse & response)
 {
     std::lock_guard lock(cancel_mutex);
-    if (cancelled)
+    if (cancelled || !current_connection || !current_connection->isConnected())
         return;
     current_connection->sendClusterFunctionReadTaskResponse(response);
 }
@@ -241,7 +241,7 @@ void MultiplexedConnections::sendClusterFunctionReadTaskResponse(const ClusterFu
 void MultiplexedConnections::sendMergeTreeReadTaskResponse(const ParallelReadResponse & response)
 {
     std::lock_guard lock(cancel_mutex);
-    if (cancelled)
+    if (cancelled || !current_connection || !current_connection->isConnected())
         return;
     current_connection->sendMergeTreeReadTaskResponse(response);
 }
@@ -527,9 +527,12 @@ MultiplexedConnections::ReplicaState & MultiplexedConnections::getReplicaForRead
 
 void MultiplexedConnections::invalidateReplica(ReplicaState & state)
 {
+    Connection * old_connection = state.connection;
     state.connection = nullptr;
     state.pool_entry = IConnectionPool::Entry();
     --active_connection_count;
+    if (current_connection == old_connection)
+        current_connection = nullptr;
 }
 
 void MultiplexedConnections::setAsyncCallback(AsyncCallback async_callback)

src/Common/ProfileEvents.cpp

@@ -299,6 +299,11 @@
     M(IcebergTrivialCountOptimizationApplied, "Trivial count optimization applied while reading from Iceberg", ValueType::Number) \
     M(IcebergVersionHintUsed, "Number of times version-hint.text has been used.", ValueType::Number) \
     M(IcebergMinMaxIndexPrunedFiles, "Number of skipped files by using MinMax index in Iceberg", ValueType::Number) \
+    M(IcebergAvroFileParsing, "Number of times avro metadata files have been parsed.", ValueType::Number) \
+    M(IcebergAvroFileParsingMicroseconds, "Time spent for parsing avro metadata files for Iceberg tables.", ValueType::Microseconds) \
+    M(IcebergJsonFileParsing, "Number of times json metadata files have been parsed.", ValueType::Number) \
+    M(IcebergJsonFileParsingMicroseconds, "Time spent for parsing json metadata files for Iceberg tables.", ValueType::Microseconds) \
+    \
     M(JoinBuildTableRowCount, "Total number of rows in the build table for a JOIN operation.", ValueType::Number) \
     M(JoinProbeTableRowCount, "Total number of rows in the probe table for a JOIN operation.", ValueType::Number) \
     M(JoinResultRowCount, "Total number of rows in the result of a JOIN operation.", ValueType::Number) \
@@ -580,7 +585,9 @@ The server successfully detected this situation and will download merged part fr
     M(S3DeleteObjects, "Number of S3 API DeleteObject(s) calls.", ValueType::Number) \
     M(S3CopyObject, "Number of S3 API CopyObject calls.", ValueType::Number) \
     M(S3ListObjects, "Number of S3 API ListObjects calls.", ValueType::Number) \
+    M(S3ListObjectsMicroseconds, "Time of S3 API ListObjects execution.", ValueType::Microseconds) \
     M(S3HeadObject,  "Number of S3 API HeadObject calls.", ValueType::Number) \
+    M(S3HeadObjectMicroseconds,  "Time of S3 API HeadObject execution.", ValueType::Microseconds) \
     M(S3GetObjectAttributes, "Number of S3 API GetObjectAttributes calls.", ValueType::Number) \
     M(S3CreateMultipartUpload, "Number of S3 API CreateMultipartUpload calls.", ValueType::Number) \
     M(S3UploadPartCopy, "Number of S3 API UploadPartCopy calls.", ValueType::Number) \
@@ -634,6 +641,7 @@ The server successfully detected this situation and will download merged part fr
     M(AzureCopyObject, "Number of Azure blob storage API CopyObject calls", ValueType::Number) \
     M(AzureDeleteObjects, "Number of Azure blob storage API DeleteObject(s) calls.", ValueType::Number) \
     M(AzureListObjects, "Number of Azure blob storage API ListObjects calls.", ValueType::Number) \
+    M(AzureListObjectsMicroseconds, "Time of Azure blob storage API ListObjects execution.", ValueType::Microseconds) \
     M(AzureGetProperties, "Number of Azure blob storage API GetProperties calls.", ValueType::Number) \
     M(AzureCreateContainer, "Number of Azure blob storage API CreateContainer calls.", ValueType::Number) \
     \

src/Core/Settings.cpp

@@ -6873,6 +6873,16 @@ Use roaring bitmap for iceberg positional deletes.
 )", 0) \
     DECLARE(Bool, export_merge_tree_part_overwrite_file_if_exists, false, R"(
 Overwrite file if it already exists when exporting a merge tree part
+)", 0) \
+    DECLARE(Timezone, iceberg_timezone_for_timestamptz, "UTC", R"(
+Timezone for Iceberg timestamptz field.
+
+Possible values:
+
+- Any valid timezone, e.g. `Europe/Berlin`, `UTC` or `Zulu`
+- `` (empty value) - use session timezone
+
+Default value is `UTC`.
 )", 0) \
     \
     /* ####################################################### */ \

src/Core/SettingsChangesHistory.cpp

@@ -49,6 +49,7 @@ const VersionToSettingsChangesMap & getSettingsChangesHistory()
             {"allow_experimental_export_merge_tree_part", false, false, "New setting."},
             {"export_merge_tree_part_overwrite_file_if_exists", false, false, "New setting."},
             {"allow_experimental_export_merge_tree_part", false, true, "Turned ON by default for Antalya."},
+            {"iceberg_timezone_for_timestamptz", "UTC", "UTC", "New setting."}
         });
         addSettingsChanges(settings_changes_history, "25.8",
         {

src/Databases/DataLake/Common.cpp

@@ -61,14 +61,14 @@ std::vector<String> splitTypeArguments(const String & type_str)
     return args;
 }
 
-DB::DataTypePtr getType(const String & type_name, bool nullable, const String & prefix)
+DB::DataTypePtr getType(const String & type_name, bool nullable, DB::ContextPtr context, const String & prefix)
 {
     String name = trim(type_name);
 
     if (name.starts_with("array<") && name.ends_with(">"))
     {
         String inner = name.substr(6, name.size() - 7);
-        return std::make_shared<DB::DataTypeArray>(getType(inner, nullable));
+        return std::make_shared<DB::DataTypeArray>(getType(inner, nullable, context));
     }
 
     if (name.starts_with("map<") && name.ends_with(">"))
@@ -79,7 +79,7 @@ DB::DataTypePtr getType(const String & type_name, bool nullable, const String &
         if (args.size() != 2)
             throw DB::Exception(DB::ErrorCodes::DATALAKE_DATABASE_ERROR, "Invalid data type {}", type_name);
 
-        return std::make_shared<DB::DataTypeMap>(getType(args[0], false), getType(args[1], nullable));
+        return std::make_shared<DB::DataTypeMap>(getType(args[0], false, context), getType(args[1], nullable, context));
     }
 
     if (name.starts_with("struct<") && name.ends_with(">"))
@@ -101,13 +101,13 @@ DB::DataTypePtr getType(const String & type_name, bool nullable, const String &
             String full_field_name = prefix.empty() ? field_name : prefix + "." + field_name;
 
             field_names.push_back(full_field_name);
-            field_types.push_back(getType(field_type, nullable, full_field_name));
+            field_types.push_back(getType(field_type, nullable, context, full_field_name));
         }
         return std::make_shared<DB::DataTypeTuple>(field_types, field_names);
     }
 
-    return nullable ? DB::makeNullable(DB::Iceberg::IcebergSchemaProcessor::getSimpleType(name))
-                    : DB::Iceberg::IcebergSchemaProcessor::getSimpleType(name);
+    return nullable ? DB::makeNullable(DB::Iceberg::IcebergSchemaProcessor::getSimpleType(name, context))
+                    : DB::Iceberg::IcebergSchemaProcessor::getSimpleType(name, context);
 }
 
 std::pair<std::string, std::string> parseTableName(const std::string & name)