some adjustments

Author: arthurpassos

src/Storages/MergeTree/ExportPartTask.cpp

@@ -50,16 +50,13 @@ bool ExportPartTask::executeStep()
 
     MergeTreeSequentialSourceType read_type = MergeTreeSequentialSourceType::Export;
 
-    NamesAndTypesList partition_columns;
+    Block block_with_partition_values;
     if (metadata_snapshot->hasPartitionKey())
     {
-        const auto & partition_key = metadata_snapshot->getPartitionKey();
-        if (!partition_key.column_names.empty())
-            partition_columns = partition_key.expression->getRequiredColumnsWithTypes();
+        /// todo arthur do I need to init minmax_idx?
+        block_with_partition_values = manifest.data_part->minmax_idx->getBlock(storage);
     }
 
-    auto block_with_partition_values = manifest.data_part->partition.getBlockWithPartitionValues(partition_columns);
-
     auto destination_storage = DatabaseCatalog::instance().tryGetTable(manifest.destination_storage_id, local_context);
     if (!destination_storage)
     {
@@ -92,7 +89,7 @@ bool ExportPartTask::executeStep()
             ProfileEvents::increment(ProfileEvents::PartsExportDuplicated);
         }
 
-        ProfileEvents::incrementNoTrace(ProfileEvents::PartsExportFailures);
+        ProfileEvents::increment(ProfileEvents::PartsExportFailures);
 
         std::lock_guard inner_lock(storage.export_manifests_mutex);
         storage.export_manifests.erase(manifest);

src/Storages/MergeTree/MergeTreeData.cpp

@@ -6272,182 +6272,6 @@ void MergeTreeData::exportPartToTable(
     background_moves_assignee.trigger();
 }
 
-void MergeTreeData::exportPartToTableImpl(
-    const MergeTreePartExportManifest & manifest,
-    ContextPtr local_context)
-{
-    auto metadata_snapshot = getInMemoryMetadataPtr();
-    Names columns_to_read = metadata_snapshot->getColumns().getNamesOfPhysical();
-    StorageSnapshotPtr storage_snapshot = getStorageSnapshot(metadata_snapshot, local_context);
-
-    MergeTreeSequentialSourceType read_type = MergeTreeSequentialSourceType::Export;
-
-    Block block_with_partition_values;
-    if (metadata_snapshot->hasPartitionKey())
-    {
-        /// todo arthur do I need to init minmax_idx?
-        block_with_partition_values = manifest.data_part->minmax_idx->getBlock(*this);
-    }
-
-    auto destination_storage = DatabaseCatalog::instance().tryGetTable(manifest.destination_storage_id, getContext());
-    if (!destination_storage)
-    {
-        std::lock_guard inner_lock(export_manifests_mutex);
-
-        const auto destination_storage_id_name = manifest.destination_storage_id.getNameForLogs();
-        export_manifests.erase(manifest);
-        throw Exception(ErrorCodes::UNKNOWN_TABLE, "Failed to reconstruct destination storage: {}", destination_storage_id_name);
-    }
-
-    SinkToStoragePtr sink;
-    std::string destination_file_path;
-
-    try
-    {
-        auto context_copy = Context::createCopy(local_context);
-        context_copy->setSetting("output_format_parallel_formatting", manifest.parallel_formatting);
-
-        sink = destination_storage->import(
-            manifest.data_part->name + "_" + manifest.data_part->checksums.getTotalChecksumHex(),
-            block_with_partition_values,
-            destination_file_path,
-            manifest.overwrite_file_if_exists,
-            context_copy);
-    }
-    catch (const Exception & e)
-    {
-        if (e.code() == ErrorCodes::FILE_ALREADY_EXISTS)
-        {
-            ProfileEvents::increment(ProfileEvents::PartsExportDuplicated);
-        }
-
-        ProfileEvents::increment(ProfileEvents::PartsExportFailures);
-
-        std::lock_guard inner_lock(export_manifests_mutex);
-        export_manifests.erase(manifest);
-
-        if (manifest.completion_callback)
-            manifest.completion_callback(MergeTreePartExportManifest::CompletionCallbackResult::createFailure(e.message()));
-        return;
-    }
-
-    bool apply_deleted_mask = true;
-    bool read_with_direct_io = local_context->getSettingsRef()[Setting::min_bytes_to_use_direct_io] > manifest.data_part->getBytesOnDisk();
-    bool prefetch = false;
-
-    MergeTreeData::IMutationsSnapshot::Params params
-    {
-        .metadata_version = metadata_snapshot->getMetadataVersion(),
-        .min_part_metadata_version = manifest.data_part->getMetadataVersion(),
-    };
-
-    auto mutations_snapshot = getMutationsSnapshot(params);
-
-    auto alter_conversions = MergeTreeData::getAlterConversionsForPart(
-        manifest.data_part,
-        mutations_snapshot,
-        local_context);
-
-    QueryPlan plan_for_part;
-
-    createReadFromPartStep(
-        read_type,
-        plan_for_part,
-        *this,
-        storage_snapshot,
-        RangesInDataPart(manifest.data_part),
-        alter_conversions,
-        nullptr,
-        columns_to_read,
-        nullptr,
-        apply_deleted_mask,
-        std::nullopt,
-        read_with_direct_io,
-        prefetch,
-        local_context,
-        getLogger("ExportPartition"));
-
-    auto exports_list_entry = getContext()->getExportsList().insert(
-        getStorageID(),
-        manifest.destination_storage_id,
-        manifest.data_part->getBytesOnDisk(),
-        manifest.data_part->name,
-        destination_file_path,
-        manifest.data_part->rows_count,
-        manifest.data_part->getBytesOnDisk(),
-        manifest.data_part->getBytesUncompressedOnDisk(),
-        manifest.create_time,
-        local_context);
-
-    ThreadGroupSwitcher switcher((*exports_list_entry)->thread_group, "");
-
-    QueryPlanOptimizationSettings optimization_settings(local_context);
-    auto pipeline_settings = BuildQueryPipelineSettings(local_context);
-    auto builder = plan_for_part.buildQueryPipeline(optimization_settings, pipeline_settings);
-
-    builder->setProgressCallback([&exports_list_entry](const Progress & progress)
-    {
-        (*exports_list_entry)->bytes_read_uncompressed += progress.read_bytes;
-        (*exports_list_entry)->rows_read += progress.read_rows;
-        (*exports_list_entry)->elapsed = (*exports_list_entry)->watch.elapsedSeconds();
-    });
-
-    auto pipeline = QueryPipelineBuilder::getPipeline(std::move(*builder));
-
-    pipeline.complete(sink);
-
-    try
-    {
-        CompletedPipelineExecutor exec(pipeline);
-        exec.execute();
-
-        std::lock_guard inner_lock(export_manifests_mutex);
-        writePartLog(
-            PartLogElement::Type::EXPORT_PART,
-            {},
-            static_cast<UInt64>((*exports_list_entry)->elapsed * 1000000000),
-            manifest.data_part->name,
-            manifest.data_part,
-            {manifest.data_part},
-            nullptr,
-            nullptr,
-            exports_list_entry.get());
-
-        export_manifests.erase(manifest);
-
-        ProfileEvents::increment(ProfileEvents::PartsExports);
-        ProfileEvents::increment(ProfileEvents::PartsExportTotalMilliseconds, static_cast<UInt64>((*exports_list_entry)->elapsed * 1000));
-
-        if (manifest.completion_callback)
-            manifest.completion_callback(MergeTreePartExportManifest::CompletionCallbackResult::createSuccess(destination_file_path));
-    }
-    catch (const Exception & e)
-    {
-        tryLogCurrentException(__PRETTY_FUNCTION__, fmt::format("while exporting the part {}. User should retry.", manifest.data_part->name));
-
-        ProfileEvents::increment(ProfileEvents::PartsExportFailures);
-
-        std::lock_guard inner_lock(export_manifests_mutex);
-        writePartLog(
-            PartLogElement::Type::EXPORT_PART,
-            ExecutionStatus::fromCurrentException("", true),
-            static_cast<UInt64>((*exports_list_entry)->elapsed * 1000000000),
-            manifest.data_part->name,
-            manifest.data_part,
-            {manifest.data_part},
-            nullptr,
-            nullptr,
-            exports_list_entry.get());
-
-        export_manifests.erase(manifest);
-
-        if (manifest.completion_callback)
-            manifest.completion_callback(MergeTreePartExportManifest::CompletionCallbackResult::createFailure(e.message()));
-
-        throw;
-    }
-}
-
 void MergeTreeData::killExportPart(const String & query_id)
 {
     std::lock_guard lock(export_manifests_mutex);

src/Storages/MergeTree/MergeTreeData.h

@@ -998,10 +998,6 @@ class MergeTreeData : public IStorage, public WithMutableContext
         throw Exception(ErrorCodes::NOT_IMPLEMENTED, "EXPORT PARTITION is not implemented");
     }
 
-    void exportPartToTableImpl(
-        const MergeTreePartExportManifest & manifest,
-        ContextPtr local_context);
-
     /// Checks that Partition could be dropped right now
     /// Otherwise - throws an exception with detailed information.
     /// We do not use mutex because it is not very important that the size could change during the operation.

tests/queries/0_stateless/03572_export_merge_tree_part_to_object_storage.sh

@@ -49,6 +49,8 @@ echo "---- Export 2020_1_1_0 and 2021_2_2_0 to wildcard table"
 query "ALTER TABLE $mt_table EXPORT PART '2020_1_1_0' TO TABLE $s3_table_wildcard SETTINGS allow_experimental_export_merge_tree_part = 1"
 query "ALTER TABLE $mt_table EXPORT PART '2021_2_2_0' TO TABLE $s3_table_wildcard SETTINGS allow_experimental_export_merge_tree_part = 1"
 
+sleep 3
+
 echo "---- Both data parts should appear"
 query "SELECT * FROM s3(s3_conn, filename='$s3_table_wildcard/**.parquet') ORDER BY id"
 
@@ -60,13 +62,14 @@ query "SELECT * FROM s3(s3_conn, filename='$s3_table_wildcard/**.parquet') ORDER
 query "CREATE TABLE $mt_table_partition_expression_with_function (id UInt64, year UInt16) ENGINE = MergeTree() PARTITION BY toString(year) ORDER BY tuple()"
 query "CREATE TABLE $s3_table_wildcard_partition_expression_with_function (id UInt64, year UInt16) ENGINE = S3(s3_conn, filename='$s3_table_wildcard_partition_expression_with_function/{_partition_id}/{_file}.parquet', format=Parquet, partition_strategy='wildcard') PARTITION BY toString(year)"
 
-# insert
 query "INSERT INTO $mt_table_partition_expression_with_function VALUES (1, 2020), (2, 2020), (3, 2020), (4, 2021)"
 
 echo "---- Export 2020_1_1_0 and 2021_2_2_0 to wildcard table with partition expression with function"
 query "ALTER TABLE $mt_table_partition_expression_with_function EXPORT PART 'cb217c742dc7d143b61583011996a160_1_1_0' TO TABLE $s3_table_wildcard_partition_expression_with_function SETTINGS allow_experimental_export_merge_tree_part = 1"
 query "ALTER TABLE $mt_table_partition_expression_with_function EXPORT PART '3be6d49ecf9749a383964bc6fab22d10_2_2_0' TO TABLE $s3_table_wildcard_partition_expression_with_function SETTINGS allow_experimental_export_merge_tree_part = 1"
 
+sleep 1
+
 echo "---- Both data parts should appear"
 query "SELECT * FROM s3(s3_conn, filename='$s3_table_wildcard_partition_expression_with_function/**.parquet') ORDER BY id"