Merge branch 'antalya-25.8' into bugfix/antalya-25.8/system_drop_filesystem_cache_on_cluster

Author: ianton-ru

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/Interpreters/ClusterFunctionReadTask.cpp

@@ -19,9 +19,11 @@ namespace ErrorCodes
 namespace Setting
 {
     extern const SettingsBool cluster_function_process_archive_on_multiple_nodes;
+    extern const SettingsBool allow_experimental_iceberg_read_optimization;
 }
 
 ClusterFunctionReadTaskResponse::ClusterFunctionReadTaskResponse(ObjectInfoPtr object, const ContextPtr & context)
+    : iceberg_read_optimization_enabled(context->getSettingsRef()[Setting::allow_experimental_iceberg_read_optimization])
 {
     if (!object)
         throw Exception(ErrorCodes::LOGICAL_ERROR, "`object` cannot be null");
@@ -67,7 +69,8 @@ void ClusterFunctionReadTaskResponse::serialize(WriteBuffer & out, size_t protoc
 
     if (protocol_version >= DBMS_CLUSTER_PROCESSING_PROTOCOL_VERSION_WITH_DATA_LAKE_COLUMNS_METADATA)
     {
-        if (file_meta_info.has_value())
+        /// This info is not used when optimization is disabled, so there is no need to send it.
+        if (iceberg_read_optimization_enabled && file_meta_info.has_value())
             file_meta_info.value()->serialize(out);
         else
             DataFileMetaInfo().serialize(out);

src/Interpreters/ClusterFunctionReadTask.h

@@ -23,6 +23,8 @@ struct ClusterFunctionReadTaskResponse
     /// File's columns info
     std::optional<DataFileMetaInfoPtr> file_meta_info;
 
+    const bool iceberg_read_optimization_enabled = false;
+
     /// Convert received response into ObjectInfo.
     ObjectInfoPtr getObjectInfo() const;
 

src/Storages/MergeTree/MergeTreeData.cpp

@@ -213,6 +213,7 @@ namespace Setting
     extern const SettingsUInt64 min_bytes_to_use_direct_io;
     extern const SettingsBool export_merge_tree_part_overwrite_file_if_exists;
     extern const SettingsBool output_format_parallel_formatting;
+    extern const SettingsBool output_format_parquet_parallel_encoding;
 }
 
 namespace MergeTreeSetting
@@ -6244,7 +6245,9 @@ void MergeTreeData::exportPartToTable(const PartitionCommand & command, ContextP
             dest_storage->getStorageID(),
             part,
             query_context->getSettingsRef()[Setting::export_merge_tree_part_overwrite_file_if_exists],
-            query_context->getSettingsRef()[Setting::output_format_parallel_formatting]);
+            query_context->getSettingsRef()[Setting::output_format_parallel_formatting],
+            query_context->getSettingsRef()[Setting::output_format_parquet_parallel_encoding],
+            query_context->getSettingsRef()[Setting::max_threads]);
 
         std::lock_guard lock(export_manifests_mutex);
 
@@ -6292,6 +6295,8 @@ void MergeTreeData::exportPartToTableImpl(
     {
         auto context_copy = Context::createCopy(local_context);
         context_copy->setSetting("output_format_parallel_formatting", manifest.parallel_formatting);
+        context_copy->setSetting("output_format_parquet_parallel_encoding", manifest.parquet_parallel_encoding);
+        context_copy->setSetting("max_threads", manifest.max_threads);
 
         sink = destination_storage->import(
             manifest.data_part->name + "_" + manifest.data_part->checksums.getTotalChecksumHex(),

src/Storages/MergeTree/MergeTreeExportManifest.h

@@ -13,17 +13,24 @@ struct MergeTreeExportManifest
         const StorageID & destination_storage_id_,
         const DataPartPtr & data_part_,
         bool overwrite_file_if_exists_,
-        bool parallel_formatting_)
+        bool parallel_formatting_,
+        bool parallel_formatting_parquet_,
+        std::size_t max_threads_)
         : destination_storage_id(destination_storage_id_),
           data_part(data_part_),
           overwrite_file_if_exists(overwrite_file_if_exists_),
           parallel_formatting(parallel_formatting_),
+          parquet_parallel_encoding(parallel_formatting_parquet_),
+          max_threads(max_threads_),
           create_time(time(nullptr)) {}
 
     StorageID destination_storage_id;
     DataPartPtr data_part;
     bool overwrite_file_if_exists;
     bool parallel_formatting;
+    /// parquet has a different setting for parallel formatting
+    bool parquet_parallel_encoding;
+    std::size_t max_threads;
 
     time_t create_time;
     mutable bool in_progress = false;

src/Storages/MergeTree/MergeTreeSequentialSource.cpp

@@ -169,7 +169,8 @@ MergeTreeSequentialSource::MergeTreeSequentialSource(
             addThrottler(read_settings.local_throttler, context->getMergesThrottler());
             break;
         case Export:
-            read_settings.local_throttler = context->getExportsThrottler();
+            addThrottler(read_settings.local_throttler, context->getExportsThrottler());
+            addThrottler(read_settings.remote_throttler, context->getExportsThrottler());
             break;
     }
 

src/Storages/ObjectStorage/ObjectStorageFilePathGenerator.h

@@ -55,7 +55,7 @@ namespace DB
 
             result += raw_path;
 
-            if (raw_path.back() != '/')
+            if (!raw_path.empty() && raw_path.back() != '/')
             {
                 result += "/";
             }

src/Storages/ObjectStorage/StorageObjectStorage.cpp

@@ -1,3 +1,4 @@
+#include <optional>
 #include <thread>
 #include <Core/ColumnWithTypeAndName.h>
 #include <Storages/ObjectStorage/StorageObjectStorage.h>
@@ -507,7 +508,7 @@ SinkToStoragePtr StorageObjectStorage::import(
         destination_file_path,
         object_storage,
         configuration,
-        format_settings,
+        std::nullopt, /// passing nullopt to force rebuild for format_settings based on query context
         std::make_shared<const Block>(getInMemoryMetadataPtr()->getSampleBlock()),
         local_context);
 }