[WIP] Export merge tree partition to object storage

src/Access/Common/AccessType.h

@@ -74,6 +74,7 @@ enum class AccessType : uint8_t
                                                                        enabled implicitly by the grant ALTER_TABLE */\
     M(ALTER_SETTINGS, "ALTER SETTING, ALTER MODIFY SETTING, MODIFY SETTING, RESET SETTING", TABLE, ALTER_TABLE) /* allows to execute ALTER MODIFY SETTING */\
     M(ALTER_MOVE_PARTITION, "ALTER MOVE PART, MOVE PARTITION, MOVE PART", TABLE, ALTER_TABLE) \
+    M(ALTER_EXPORT_PARTITION, "ALTER EXPORT PART, EXPORT PARTITION, EXPORT PART", TABLE, ALTER_TABLE) \
     M(ALTER_FETCH_PARTITION, "ALTER FETCH PART, FETCH PARTITION", TABLE, ALTER_TABLE) \
     M(ALTER_FREEZE_PARTITION, "FREEZE PARTITION, UNFREEZE", TABLE, ALTER_TABLE) \
     M(ALTER_UNLOCK_SNAPSHOT, "UNLOCK SNAPSHOT", TABLE, ALTER_TABLE) \

src/CMakeLists.txt

@@ -130,6 +130,7 @@ add_headers_and_sources(dbms Storages/ObjectStorage/Azure)
 add_headers_and_sources(dbms Storages/ObjectStorage/S3)
 add_headers_and_sources(dbms Storages/ObjectStorage/HDFS)
 add_headers_and_sources(dbms Storages/ObjectStorage/Local)
+add_headers_and_sources(dbms Storages/ObjectStorage/MergeTree)
 add_headers_and_sources(dbms Storages/ObjectStorage/DataLakes)
 add_headers_and_sources(dbms Storages/ObjectStorage/DataLakes/Iceberg)
 add_headers_and_sources(dbms Storages/ObjectStorage/DataLakes/DeltaLake)

src/Common/CurrentMetrics.cpp

@@ -10,6 +10,7 @@
     M(Merge, "Number of executing background merges") \
     M(MergeParts, "Number of source parts participating in current background merges") \
     M(Move, "Number of currently executing moves") \
+    M(Export, "Number of currently executing exports") \
     M(PartMutation, "Number of mutations (ALTER DELETE/UPDATE)") \
     M(ReplicatedFetch, "Number of data parts being fetched from replica") \
     M(ReplicatedSend, "Number of data parts being sent to replicas") \

src/Core/Settings.cpp

@@ -6874,6 +6874,12 @@ Possible values:
     DECLARE_WITH_ALIAS(Bool, allow_experimental_time_series_aggregate_functions, false, R"(
 Experimental timeSeries* aggregate functions for Prometheus-like timeseries resampling, rate, delta calculation.
 )", EXPERIMENTAL, allow_experimental_ts_to_grid_aggregate_function) \
+    DECLARE_WITH_ALIAS(Bool, allow_experimental_export_merge_tree_partition, false, R"(
+Experimental export merge tree partition.
+)", EXPERIMENTAL, allow_experimental_export_merge_tree_partition) \
+    DECLARE_WITH_ALIAS(Bool, export_merge_tree_partition_background_execution, true, R"(
+    Process exports asynchronously in background threads
+)", EXPERIMENTAL, export_merge_tree_partition_background_execution) \
     \
     /* ####################################################### */ \
     /* ############ END OF EXPERIMENTAL FEATURES ############# */ \

src/Databases/DatabaseS3.cpp

@@ -19,6 +19,7 @@
 #include <Storages/IStorage.h>
 #include <Storages/NamedCollectionsHelpers.h>
 #include <TableFunctions/TableFunctionFactory.h>
+#include <Parsers/ASTInsertQuery.h>
 
 #include <boost/algorithm/string.hpp>
 #include <filesystem>

src/Disks/ObjectStorages/IObjectStorage.h

@@ -96,6 +96,7 @@ struct RelativePathWithMetadata
     virtual ~RelativePathWithMetadata() = default;
 
     virtual std::string getFileName() const { return std::filesystem::path(relative_path).filename(); }
+    virtual std::string getFileNameWithoutExtension() const { return std::filesystem::path(relative_path).stem(); }
     virtual std::string getPath() const { return relative_path; }
     virtual bool isArchive() const { return false; }
     virtual std::string getPathToArchive() const { throw Exception(ErrorCodes::LOGICAL_ERROR, "Not an archive"); }

src/Interpreters/Context.cpp

@@ -34,6 +34,7 @@
 #include <Storages/MarkCache.h>
 #include <Storages/MergeTree/MergeList.h>
 #include <Storages/MergeTree/MovesList.h>
+#include <Storages/ExportsList.h>
 #include <Storages/MergeTree/ReplicatedFetchList.h>
 #include <Storages/MergeTree/MergeTreeData.h>
 #include <Storages/MergeTree/MergeTreeSettings.h>
@@ -463,6 +464,7 @@ struct ContextSharedPart : boost::noncopyable
     GlobalOvercommitTracker global_overcommit_tracker;
     MergeList merge_list;                                       /// The list of executable merge (for (Replicated)?MergeTree)
     MovesList moves_list;                                       /// The list of executing moves (for (Replicated)?MergeTree)
+    ExportsList exports_list;                                   /// The list of executing exports MergeTree -> Object storage
     ReplicatedFetchList replicated_fetch_list;
     RefreshSet refresh_set;                                 /// The list of active refreshes (for MaterializedView)
     ConfigurationPtr users_config TSA_GUARDED_BY(mutex);                              /// Config with the users, profiles and quotas sections.
@@ -1148,6 +1150,8 @@ MergeList & Context::getMergeList() { return shared->merge_list; }
 const MergeList & Context::getMergeList() const { return shared->merge_list; }
 MovesList & Context::getMovesList() { return shared->moves_list; }
 const MovesList & Context::getMovesList() const { return shared->moves_list; }
+ExportsList & Context::getExportsList() { return shared->exports_list; }
+const ExportsList & Context::getExportsList() const { return shared->exports_list; }
 ReplicatedFetchList & Context::getReplicatedFetchList() { return shared->replicated_fetch_list; }
 const ReplicatedFetchList & Context::getReplicatedFetchList() const { return shared->replicated_fetch_list; }
 RefreshSet & Context::getRefreshSet() { return shared->refresh_set; }

src/Interpreters/Context.h

@@ -88,6 +88,7 @@ class AsynchronousMetrics;
 class BackgroundSchedulePool;
 class MergeList;
 class MovesList;
+class ExportsList;
 class ReplicatedFetchList;
 class RefreshSet;
 class Cluster;
@@ -1141,6 +1142,9 @@ class Context: public ContextData, public std::enable_shared_from_this<Context>
     MovesList & getMovesList();
     const MovesList & getMovesList() const;
 
+    ExportsList & getExportsList();
+    const ExportsList & getExportsList() const;
+
     ReplicatedFetchList & getReplicatedFetchList();
     const ReplicatedFetchList & getReplicatedFetchList() const;
 

src/Interpreters/InterpreterAlterQuery.cpp

@@ -502,6 +502,12 @@ AccessRightsElements InterpreterAlterQuery::getRequiredAccessForCommand(const AS
             required_access.emplace_back(AccessType::ALTER_DELETE | AccessType::INSERT, database, table);
             break;
         }
+        case ASTAlterCommand::EXPORT_PARTITION:
+        {
+            required_access.emplace_back(AccessType::ALTER_EXPORT_PARTITION, database, table);
+            required_access.emplace_back(AccessType::INSERT, command.to_database, command.to_table);
+            break;
+        }
         case ASTAlterCommand::FETCH_PARTITION:
         {
             required_access.emplace_back(AccessType::ALTER_FETCH_PARTITION, database, table);

src/Interpreters/PartLog.cpp

@@ -69,6 +69,7 @@ ColumnsDescription PartLogElement::getColumnsDescription()
             {"MovePart",      static_cast<Int8>(MOVE_PART)},
             {"MergePartsStart", static_cast<Int8>(MERGE_PARTS_START)},
             {"MutatePartStart", static_cast<Int8>(MUTATE_PART_START)},
+            {"ExportPart",      static_cast<Int8>(EXPORT_PART)},
         }
     );
 
@@ -109,7 +110,8 @@ ColumnsDescription PartLogElement::getColumnsDescription()
             "RemovePart — Removing or detaching a data part using [DETACH PARTITION](/sql-reference/statements/alter/partition#detach-partitionpart)."
             "MutatePartStart — Mutating of a data part has started, "
             "MutatePart — Mutating of a data part has finished, "
-            "MovePart — Moving the data part from the one disk to another one."},
+            "MovePart — Moving the data part from the one disk to another one."
+            "ExportPart — Exporting the data part from a merge tree table to one (e.g, object storage)."},
         {"merge_reason", std::move(merge_reason_datatype),
             "The reason for the event with type MERGE_PARTS. Can have one of the following values: "
             "NotAMerge — The current event has the type other than MERGE_PARTS, "

src/Interpreters/PartLog.h

@@ -30,6 +30,7 @@ struct PartLogElement
         MOVE_PART = 6,
         MERGE_PARTS_START = 7,
         MUTATE_PART_START = 8,
+        EXPORT_PART = 9,
     };
 
     /// Copy of MergeAlgorithm since values are written to disk.

src/Parsers/ASTAlterQuery.cpp

@@ -355,6 +355,29 @@ void ASTAlterCommand::formatImpl(WriteBuffer & ostr, const FormatSettings & sett
             ostr << quoteString(move_destination_name);
         }
     }
+    else if (type == ASTAlterCommand::EXPORT_PARTITION)
+    {
+        ostr << (settings.hilite ? hilite_keyword : "") << "EXPORT " << (part ? "PART " : "PARTITION ")
+             << (settings.hilite ? hilite_none : "");
+        partition->format(ostr, settings, state, frame);
+        ostr << " TO ";
+        switch (move_destination_type)
+        {
+            case DataDestinationType::TABLE:
+                ostr << "TABLE ";
+                if (!to_database.empty())
+                {
+                    ostr << (settings.hilite ? hilite_identifier : "") << backQuoteIfNeed(to_database)
+                         << (settings.hilite ? hilite_none : "") << ".";
+                }
+                ostr << (settings.hilite ? hilite_identifier : "") << backQuoteIfNeed(to_table)
+                     << (settings.hilite ? hilite_none : "");
+                return;
+            default:
+                break;
+        }
+
+    }
     else if (type == ASTAlterCommand::REPLACE_PARTITION)
     {
         ostr << (settings.hilite ? hilite_keyword : "") << (replace ? "REPLACE" : "ATTACH") << " PARTITION "

src/Parsers/ASTAlterQuery.h

@@ -71,6 +71,7 @@ class ASTAlterCommand : public IAST
         FREEZE_ALL,
         UNFREEZE_PARTITION,
         UNFREEZE_ALL,
+        EXPORT_PARTITION,
 
         DELETE,
         UPDATE,

src/Parsers/CommonParsers.h

@@ -331,6 +331,7 @@ namespace DB
     MR_MACROS(MONTHS, "MONTHS") \
     MR_MACROS(MOVE_PART, "MOVE PART") \
     MR_MACROS(MOVE_PARTITION, "MOVE PARTITION") \
+    MR_MACROS(EXPORT_PARTITION, "EXPORT PARTITION") \
     MR_MACROS(MOVE, "MOVE") \
     MR_MACROS(MS, "MS") \
     MR_MACROS(MUTATION, "MUTATION") \

src/Parsers/ParserAlterQuery.cpp

@@ -82,6 +82,7 @@ bool ParserAlterCommand::parseImpl(Pos & pos, ASTPtr & node, Expected & expected
     ParserKeyword s_forget_partition(Keyword::FORGET_PARTITION);
     ParserKeyword s_move_partition(Keyword::MOVE_PARTITION);
     ParserKeyword s_move_part(Keyword::MOVE_PART);
+    ParserKeyword s_export_partition(Keyword::EXPORT_PARTITION);
     ParserKeyword s_drop_detached_partition(Keyword::DROP_DETACHED_PARTITION);
     ParserKeyword s_drop_detached_part(Keyword::DROP_DETACHED_PART);
     ParserKeyword s_fetch_partition(Keyword::FETCH_PARTITION);
@@ -564,6 +565,22 @@ bool ParserAlterCommand::parseImpl(Pos & pos, ASTPtr & node, Expected & expected
                     command->move_destination_name = ast_space_name->as<ASTLiteral &>().value.safeGet<String>();
                 }
             }
+            else if (s_export_partition.ignore(pos, expected))
+            {
+                if (!parser_partition.parse(pos, command_partition, expected))
+                    return false;
+
+                command->type = ASTAlterCommand::EXPORT_PARTITION;
+
+                if (!s_to_table.ignore(pos, expected))
+                {
+                    return false;
+                }
+
+                if (!parseDatabaseAndTableName(pos, expected, command->to_database, command->to_table))
+                    return false;
+                command->move_destination_type = DataDestinationType::TABLE;
+            }
             else if (s_add_constraint.ignore(pos, expected))
             {
                 if (s_if_not_exists.ignore(pos, expected))

src/Storages/ExportsList.cpp

@@ -0,0 +1,34 @@
+#include <Storages/ExportsList.h>
+#include "base/getThreadId.h"
+
+namespace DB
+{
+
+ExportsListElement::ExportsListElement(
+    const StorageID & source_table_id_,
+    const StorageID & destination_table_id_,
+    const std::string & part_name_,
+    const std::string & destination_path_)
+: source_table_id(source_table_id_),
+    destination_table_id(destination_table_id_),
+    part_name(part_name_),
+    destination_path(destination_path_),
+    thread_id(getThreadId())
+{
+}
+
+ExportInfo ExportsListElement::getInfo() const
+{
+    ExportInfo res;
+    res.source_database = source_table_id.database_name;
+    res.source_table = source_table_id.table_name;
+    res.destination_database = destination_table_id.database_name;
+    res.destination_table = destination_table_id.table_name;
+    res.part_name = part_name;
+    res.destination_path = destination_path;
+    res.elapsed = watch.elapsedSeconds();
+    res.thread_id = thread_id;
+    return res;
+}
+
+}

src/Storages/ExportsList.h

@@ -0,0 +1,62 @@
+#pragma once
+
+#include <Storages/MergeTree/BackgroundProcessList.h>
+#include <Interpreters/StorageID.h>
+#include <Common/Stopwatch.h>
+#include <Common/CurrentMetrics.h>
+#include <boost/noncopyable.hpp>
+
+namespace CurrentMetrics
+{
+extern const Metric Export;
+}
+
+namespace DB
+{
+
+struct ExportInfo
+{
+    std::string source_database;
+    std::string destination_database;
+    std::string source_table;
+    std::string destination_table;
+    std::string part_name;
+    std::string destination_path;
+
+    Float64 elapsed;
+    UInt64 thread_id;
+};
+
+struct ExportsListElement : private boost::noncopyable
+{
+    const StorageID source_table_id;
+    const StorageID destination_table_id;
+    const std::string part_name;
+    const std::string destination_path;
+
+    Stopwatch watch;
+    const UInt64 thread_id;
+
+    ExportsListElement(
+        const StorageID & source_table_id_,
+        const StorageID & destination_table_id_,
+        const std::string & part_name_,
+        const std::string & destination_path_);
+
+    ExportInfo getInfo() const;
+};
+
+
+/// List of currently processing moves
+class ExportsList final : public BackgroundProcessList<ExportsListElement, ExportInfo>
+{
+private:
+    using Parent = BackgroundProcessList<ExportsListElement, ExportInfo>;
+
+public:
+    ExportsList()
+        : Parent(CurrentMetrics::Export)
+    {}
+};
+
+}

src/Storages/IPartitionStrategy.cpp

@@ -332,32 +332,27 @@ ColumnPtr HiveStylePartitionStrategy::computePartitionKey(const Chunk & chunk)
     return block_with_partition_by_expr.getByName(actions_with_column_name.column_name).column;
 }
 
-ColumnRawPtrs HiveStylePartitionStrategy::getFormatChunkColumns(const Chunk & chunk)
+Chunk HiveStylePartitionStrategy::getFormatChunk(const Chunk & chunk)
 {
-    ColumnRawPtrs result;
+    Chunk result;
+
     if (partition_columns_in_data_file)
     {
         for (const auto & column : chunk.getColumns())
         {
-            result.emplace_back(column.get());
+            result.addColumn(column);
         }
 
         return result;
     }
 
-    if (chunk.getNumColumns() != sample_block.columns())
-    {
-        throw Exception(
-            ErrorCodes::LOGICAL_ERROR,
-            "Incorrect number of columns in chunk. Expected {}, found {}",
-            sample_block.columns(), chunk.getNumColumns());
-    }
+    chassert(chunk.getColumns().size() == sample_block.columns());
 
     for (size_t i = 0; i < sample_block.columns(); i++)
     {
         if (!partition_columns_name_set.contains(sample_block.getByPosition(i).name))
         {
-            result.emplace_back(chunk.getColumns()[i].get());
+            result.addColumn(chunk.getColumns()[i]);
         }
     }