generate transaction id instead of reusing the queryid

Author: arthurpassos

src/Functions/generateSnowflakeID.cpp

@@ -154,6 +154,11 @@ uint64_t generateSnowflakeID()
     return fromSnowflakeId(snowflake_id);
 }
 
+std::string generateSnowflakeIDString()
+{
+    return std::to_string(generateSnowflakeID());
+}
+
 class FunctionGenerateSnowflakeID : public IFunction
 {
 public:

src/Functions/generateSnowflakeID.h

@@ -7,4 +7,6 @@ namespace DB
 
 uint64_t generateSnowflakeID();
 
+std::string generateSnowflakeIDString();
+
 }

src/Storages/MergeTree/ExportPartTask.cpp

@@ -289,7 +289,7 @@ Priority ExportPartTask::getPriority() const
 
 String ExportPartTask::getQueryId() const
 {
-    return manifest.query_id;
+    return manifest.transaction_id;
 }
 
 }

src/Storages/MergeTree/MergeTreeData.cpp

@@ -102,6 +102,7 @@
 #include <Storages/MergeTree/LoadedMergeTreeDataPartInfoForReader.h>
 #include <QueryPipeline/QueryPipelineBuilder.h>
 #include <Storages/MergeTree/MergeTreeIndexGranularityAdaptive.h>
+#include <Functions/generateSnowflakeID.h>
 
 #include <boost/algorithm/string/join.hpp>
 
@@ -6212,7 +6213,7 @@ void MergeTreeData::exportPartToTable(const PartitionCommand & command, ContextP
 
     const auto database_name = query_context->resolveDatabase(command.to_database);
 
-    exportPartToTable(part_name, StorageID{database_name, command.to_table}, query_context->getCurrentQueryId(), query_context);
+    exportPartToTable(part_name, StorageID{database_name, command.to_table}, generateSnowflakeIDString(), query_context);
 }
 
 void MergeTreeData::exportPartToTable(
@@ -6275,13 +6276,13 @@ void MergeTreeData::exportPartToTable(
     background_moves_assignee.trigger();
 }
 
-void MergeTreeData::killExportPart(const String & query_id)
+void MergeTreeData::killExportPart(const String & transaction_id)
 {
     std::lock_guard lock(export_manifests_mutex);
 
     std::erase_if(export_manifests, [&](const auto & manifest)
     {
-        if (manifest.query_id == query_id)
+        if (manifest.transaction_id == transaction_id)
         {
             if (manifest.task)
                 manifest.task->cancel();

src/Storages/MergeTree/MergeTreeData.h

@@ -991,7 +991,7 @@ class MergeTreeData : public IStorage, public WithMutableContext
         ContextPtr query_context,
         std::function<void(MergeTreePartExportManifest::CompletionCallbackResult)> completion_callback = {});
 
-    void killExportPart(const String & query_id);
+    void killExportPart(const String & transaction_id);
 
     virtual void exportPartitionToTable(const PartitionCommand &, ContextPtr)
     {

src/Storages/MergeTree/MergeTreePartExportManifest.h

@@ -43,15 +43,15 @@ struct MergeTreePartExportManifest
     MergeTreePartExportManifest(
         const StorageID & destination_storage_id_,
         const DataPartPtr & data_part_,
-        const String & query_id_,
+        const String & transaction_id_,
         FileAlreadyExistsPolicy file_already_exists_policy_,
         bool parallel_formatting_,
         bool parquet_parallel_encoding_,
         std::size_t max_threads_,
         std::function<void(CompletionCallbackResult)> completion_callback_ = {})
         : destination_storage_id(destination_storage_id_),
           data_part(data_part_),
-          query_id(query_id_),
+          transaction_id(transaction_id_),
           file_already_exists_policy(file_already_exists_policy_),
           parallel_formatting(parallel_formatting_),
           parquet_parallel_encoding(parquet_parallel_encoding_),
@@ -62,7 +62,7 @@ struct MergeTreePartExportManifest
     StorageID destination_storage_id;
     DataPartPtr data_part;
     /// Used for killing the export.
-    String query_id;
+    String transaction_id;
     FileAlreadyExistsPolicy file_already_exists_policy;
     bool parallel_formatting;
     /// parquet has a different setting for parallel formatting

src/Storages/StorageReplicatedMergeTree.cpp

@@ -118,6 +118,7 @@
 #include <Backups/RestorerFromBackup.h>
 
 #include <Common/scope_guard_safe.h>
+#include "Functions/generateSnowflakeID.h"
 #include "Interpreters/StorageID.h"
 #include "QueryPipeline/QueryPlanResourceHolder.h"
 #include "Storages/ExportReplicatedMergeTreePartitionManifest.h"
@@ -8191,7 +8192,7 @@ void StorageReplicatedMergeTree::exportPartitionToTable(const PartitionCommand &
 
     ExportReplicatedMergeTreePartitionManifest manifest;
 
-    manifest.transaction_id = query_context->getCurrentQueryId();
+    manifest.transaction_id = generateSnowflakeIDString();
     manifest.partition_id = partition_id;
     manifest.destination_database = dest_database;
     manifest.destination_table = dest_table;