Merge branch 'antalya-25.8' into export_replicated_mt_partition_v2

Author: arthurpassos

src/Storages/IStorage.h

@@ -468,6 +468,7 @@ It is currently only implemented in StorageObjectStorage.
         Block & /* block_with_partition_values */,
         std::string & /* destination_file_path */,
         bool /* overwrite_if_exists */,
+        const std::optional<FormatSettings> & /* format_settings */,
         ContextPtr /* context */)
     {
       throw Exception(ErrorCodes::NOT_IMPLEMENTED, "Import is not implemented for storage {}", getName());

src/Storages/MergeTree/ExportPartTask.cpp

@@ -86,17 +86,13 @@ bool ExportPartTask::executeStep()
 
     try
     {
-        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(),
             block_with_partition_values,
             (*exports_list_entry)->destination_file_path,
             manifest.file_already_exists_policy == MergeTreePartExportManifest::FileAlreadyExistsPolicy::overwrite,
-            context_copy);
+            manifest.format_settings,
+            local_context);
     }
     catch (const Exception & e)
     {

src/Storages/MergeTree/MergeTreeData.cpp

@@ -48,6 +48,7 @@
 #include <Disks/SingleDiskVolume.h>
 #include <Disks/TemporaryFileOnDisk.h>
 #include <Disks/createVolume.h>
+#include <Formats/FormatFactory.h>
 #include <IO/Operators.h>
 #include <IO/S3Common.h>
 #include <IO/SharedThreadPools.h>
@@ -6254,14 +6255,13 @@ void MergeTreeData::exportPartToTable(
                         part_name, getStorageID().getFullTableName());
 
     {
+        const auto format_settings = getFormatSettings(query_context);
         MergeTreePartExportManifest manifest(
             dest_storage->getStorageID(),
             part,
             transaction_id,
             query_context->getSettingsRef()[Setting::export_merge_tree_part_file_already_exists_policy].value,
-            query_context->getSettingsRef()[Setting::output_format_parallel_formatting],
-            query_context->getSettingsRef()[Setting::output_format_parquet_parallel_encoding],
-            query_context->getSettingsRef()[Setting::max_threads],
+            format_settings,
             completion_callback);
 
         std::lock_guard lock(export_manifests_mutex);

src/Storages/MergeTree/MergeTreePartExportManifest.h

@@ -45,17 +45,13 @@ struct MergeTreePartExportManifest
         const DataPartPtr & data_part_,
         const String & transaction_id_,
         FileAlreadyExistsPolicy file_already_exists_policy_,
-        bool parallel_formatting_,
-        bool parquet_parallel_encoding_,
-        std::size_t max_threads_,
+        const FormatSettings & format_settings_,
         std::function<void(CompletionCallbackResult)> completion_callback_ = {})
         : destination_storage_id(destination_storage_id_),
           data_part(data_part_),
           transaction_id(transaction_id_),
           file_already_exists_policy(file_already_exists_policy_),
-          parallel_formatting(parallel_formatting_),
-          parquet_parallel_encoding(parquet_parallel_encoding_),
-          max_threads(max_threads_),
+          format_settings(format_settings_),
           completion_callback(completion_callback_),
           create_time(time(nullptr)) {}
 
@@ -64,10 +60,7 @@ struct MergeTreePartExportManifest
     /// Used for killing the export.
     String transaction_id;
     FileAlreadyExistsPolicy file_already_exists_policy;
-    bool parallel_formatting;
-    /// parquet has a different setting for parallel formatting
-    bool parquet_parallel_encoding;
-    std::size_t max_threads;
+    FormatSettings format_settings;
 
     std::function<void(CompletionCallbackResult)> completion_callback;
 

src/Storages/ObjectStorage/StorageObjectStorage.cpp

@@ -484,6 +484,7 @@ SinkToStoragePtr StorageObjectStorage::import(
     Block & block_with_partition_values,
     std::string & destination_file_path,
     bool overwrite_if_exists,
+    const std::optional<FormatSettings> & format_settings_,
     ContextPtr local_context)
 {
     std::string partition_key;
@@ -509,7 +510,7 @@ SinkToStoragePtr StorageObjectStorage::import(
         destination_file_path,
         object_storage,
         configuration,
-        std::nullopt, /// passing nullopt to force rebuild for format_settings based on query context
+        format_settings_ ? format_settings_ : format_settings,
         std::make_shared<const Block>(getInMemoryMetadataPtr()->getSampleBlock()),
         local_context);
 }

src/Storages/ObjectStorage/StorageObjectStorage.h

@@ -86,6 +86,7 @@ class StorageObjectStorage : public IStorage
         Block & /* block_with_partition_values */,
         std::string & /* destination_file_path */,
         bool /* overwrite_if_exists */,
+        const std::optional<FormatSettings> & /* format_settings_ */,
         ContextPtr /* context */) override;
 
     void commitExportPartitionTransaction(

src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp

@@ -1,3 +1,4 @@
+#include <optional>
 #include <Storages/ObjectStorage/StorageObjectStorageCluster.h>
 
 #include <Common/Exception.h>
@@ -575,12 +576,13 @@ SinkToStoragePtr StorageObjectStorageCluster::import(
     Block & block_with_partition_values,
     std::string & destination_file_path,
     bool overwrite_if_exists,
+    const std::optional<FormatSettings> & format_settings_,
     ContextPtr context)
 {
     if (pure_storage)
-        return pure_storage->import(file_name, block_with_partition_values, destination_file_path, overwrite_if_exists, context);
+        return pure_storage->import(file_name, block_with_partition_values, destination_file_path, overwrite_if_exists, format_settings_, context);
 
-    return IStorageCluster::import(file_name, block_with_partition_values, destination_file_path, overwrite_if_exists, context);
+    return IStorageCluster::import(file_name, block_with_partition_values, destination_file_path, overwrite_if_exists, format_settings_, context);
 }
 
 void StorageObjectStorageCluster::readFallBackToPure(

src/Storages/ObjectStorage/StorageObjectStorageCluster.h

@@ -130,6 +130,7 @@ class StorageObjectStorageCluster : public IStorageCluster
         Block & /* block_with_partition_values */,
         std::string & /* destination_file_path */,
         bool /* overwrite_if_exists */,
+        const std::optional<FormatSettings> & /* format_settings_ */,
         ContextPtr /* context */) override;
     bool prefersLargeBlocks() const override;