Move parts between storage volumes according to configured TTL expressions

dbms/src/Common/quoteString.cpp

@@ -14,6 +14,15 @@ String quoteString(const StringRef & x)
 }
 
 
+String doubleQuoteString(const StringRef & x)
+{
+    String res(x.size, '\0');
+    WriteBufferFromString wb(res);
+    writeDoubleQuotedString(x, wb);
+    return res;
+}
+
+
 String backQuote(const StringRef & x)
 {
     String res(x.size, '\0');

dbms/src/Common/quoteString.h

@@ -9,6 +9,9 @@ namespace DB
 /// Quote the string.
 String quoteString(const StringRef & x);
 
+/// Double quote the string.
+String doubleQuoteString(const StringRef & x);
+
 /// Quote the identifier with backquotes.
 String backQuote(const StringRef & x);
 

dbms/src/DataStreams/TTLBlockInputStream.cpp

@@ -85,6 +85,8 @@ Block TTLBlockInputStream::readImpl()
 
     removeValuesWithExpiredColumnTTL(block);
 
+    updateMovesTTL(block);
+
     return block;
 }
 
@@ -145,7 +147,8 @@ void TTLBlockInputStream::removeValuesWithExpiredColumnTTL(Block & block)
         defaults_expression->execute(block_with_defaults);
     }
 
-    for (const auto & [name, ttl_entry] : storage.ttl_entries_by_name)
+    std::vector<String> columns_to_remove;
+    for (const auto & [name, ttl_entry] : storage.column_ttl_entries_by_name)
     {
         const auto & old_ttl_info = old_ttl_infos.columns_ttl[name];
         auto & new_ttl_info = new_ttl_infos.columns_ttl[name];
@@ -159,7 +162,10 @@ void TTLBlockInputStream::removeValuesWithExpiredColumnTTL(Block & block)
             continue;
 
         if (!block.has(ttl_entry.result_column))
+        {
+            columns_to_remove.push_back(ttl_entry.result_column);
             ttl_entry.expression->execute(block);
+        }
 
         ColumnPtr default_column = nullptr;
         if (block_with_defaults.has(name))
@@ -192,9 +198,34 @@ void TTLBlockInputStream::removeValuesWithExpiredColumnTTL(Block & block)
         column_with_type.column = std::move(result_column);
     }
 
-    for (const auto & elem : storage.ttl_entries_by_name)
-        if (block.has(elem.second.result_column))
-            block.erase(elem.second.result_column);
+    for (const String & column : columns_to_remove)
+        block.erase(column);
+}
+
+void TTLBlockInputStream::updateMovesTTL(Block & block)
+{
+    std::vector<String> columns_to_remove;
+    for (const auto & ttl_entry : storage.move_ttl_entries)
+    {
+        auto & new_ttl_info = new_ttl_infos.moves_ttl[ttl_entry.result_column];
+
+        if (!block.has(ttl_entry.result_column))
+        {
+            columns_to_remove.push_back(ttl_entry.result_column);
+            ttl_entry.expression->execute(block);
+        }
+
+        const IColumn * ttl_column = block.getByName(ttl_entry.result_column).column.get();
+
+        for (size_t i = 0; i < block.rows(); ++i)
+        {
+            UInt32 cur_ttl = getTimestampByIndex(ttl_column, i);
+            new_ttl_info.update(cur_ttl);
+        }
+    }
+
+    for (const String & column : columns_to_remove)
+        block.erase(column);
 }
 
 UInt32 TTLBlockInputStream::getTimestampByIndex(const IColumn * column, size_t ind)

dbms/src/DataStreams/TTLBlockInputStream.h

@@ -58,6 +58,9 @@ class TTLBlockInputStream : public IBlockInputStream
     /// Removes rows with expired table ttl and computes new ttl_infos for part
     void removeRowsWithExpiredTableTTL(Block & block);
 
+    /// Updates TTL for moves
+    void updateMovesTTL(Block & block);
+
     UInt32 getTimestampByIndex(const IColumn * column, size_t ind);
     bool isTTLExpired(time_t ttl);
 };

dbms/src/Interpreters/Context.cpp

@@ -1487,15 +1487,15 @@ BackgroundProcessingPool & Context::getBackgroundPool()
 {
     auto lock = getLock();
     if (!shared->background_pool)
-        shared->background_pool.emplace(settings.background_pool_size);
+        shared->background_pool.emplace(settings.background_pool_size, getConfigRef());
     return *shared->background_pool;
 }
 
 BackgroundProcessingPool & Context::getBackgroundMovePool()
 {
     auto lock = getLock();
     if (!shared->background_move_pool)
-        shared->background_move_pool.emplace(settings.background_move_pool_size, "BackgroundMovePool", "BgMoveProcPool");
+        shared->background_move_pool.emplace(settings.background_move_pool_size, getConfigRef(), "BackgroundMovePool", "BgMoveProcPool");
     return *shared->background_move_pool;
 }
 

dbms/src/Parsers/ASTAlterQuery.cpp

@@ -176,12 +176,14 @@ void ASTAlterCommand::formatImpl(
         settings.ostr << " TO ";
         switch (move_destination_type)
         {
-            case MoveDestinationType::DISK:
+            case PartDestinationType::DISK:
                 settings.ostr << "DISK ";
                 break;
-            case MoveDestinationType::VOLUME:
+            case PartDestinationType::VOLUME:
                 settings.ostr << "VOLUME ";
                 break;
+            default:
+                break;
         }
         settings.ostr << quoteString(move_destination_name);
     }

dbms/src/Parsers/ASTAlterQuery.h

@@ -3,6 +3,7 @@
 #include <Parsers/IAST.h>
 #include <Parsers/ASTQueryWithTableAndOutput.h>
 #include <Parsers/ASTQueryWithOnCluster.h>
+#include <Parsers/ASTTTLElement.h>
 
 
 namespace DB
@@ -128,15 +129,9 @@ class ASTAlterCommand : public IAST
 
     bool if_exists = false;     /// option for DROP_COLUMN, MODIFY_COLUMN, COMMENT_COLUMN
 
-    enum MoveDestinationType
-    {
-        DISK,
-        VOLUME,
-    };
-
-    MoveDestinationType move_destination_type;
+    PartDestinationType move_destination_type; /// option for MOVE PART/PARTITION
 
-    String move_destination_name;
+    String move_destination_name;             /// option for MOVE PART/PARTITION
 
     /** For FETCH PARTITION - the path in ZK to the shard, from which to download the partition.
      */

dbms/src/Parsers/ASTTTLElement.cpp

@@ -0,0 +1,27 @@
+
+#include <Columns/Collator.h>
+#include <Common/quoteString.h>
+#include <Parsers/ASTTTLElement.h>
+
+
+namespace DB
+{
+
+void ASTTTLElement::formatImpl(const FormatSettings & settings, FormatState & state, FormatStateStacked frame) const
+{
+    children.front()->formatImpl(settings, state, frame);
+    if (destination_type == PartDestinationType::DISK)
+    {
+        settings.ostr << " TO DISK " << quoteString(destination_name);
+    }
+    else if (destination_type == PartDestinationType::VOLUME)
+    {
+        settings.ostr << " TO VOLUME " << quoteString(destination_name);
+    }
+    else if (destination_type == PartDestinationType::DELETE)
+    {
+        /// It would be better to output "DELETE" here but that will break compatibility with earlier versions.
+    }
+}
+
+}

dbms/src/Parsers/ASTTTLElement.h

@@ -0,0 +1,36 @@
+#pragma once
+
+#include <Parsers/IAST.h>
+#include <Storages/MergeTree/PartDestinationType.h>
+
+
+namespace DB
+{
+/** Element of TTL expression.
+  */
+class ASTTTLElement : public IAST
+{
+public:
+    PartDestinationType destination_type;
+    String destination_name;
+
+    ASTTTLElement(PartDestinationType destination_type_, const String & destination_name_)
+        : destination_type(destination_type_)
+        , destination_name(destination_name_)
+    {
+    }
+
+    String getID(char) const override { return "TTLElement"; }
+
+    ASTPtr clone() const override
+    {
+        auto clone = std::make_shared<ASTTTLElement>(*this);
+        clone->cloneChildren();
+        return clone;
+    }
+
+protected:
+    void formatImpl(const FormatSettings & settings, FormatState & state, FormatStateStacked frame) const override;
+};
+
+}

dbms/src/Parsers/ExpressionElementParsers.cpp

@@ -16,6 +16,7 @@
 #include <Parsers/ASTAsterisk.h>
 #include <Parsers/ASTQualifiedAsterisk.h>
 #include <Parsers/ASTQueryParameter.h>
+#include <Parsers/ASTTTLElement.h>
 #include <Parsers/ASTOrderByElement.h>
 #include <Parsers/ASTSubquery.h>
 #include <Parsers/ASTFunctionWithKeyValueArguments.h>
@@ -1477,6 +1478,42 @@ bool ParserFunctionWithKeyValueArguments::parseImpl(Pos & pos, ASTPtr & node, Ex
     return true;
 }
 
+bool ParserTTLElement::parseImpl(Pos & pos, ASTPtr & node, Expected & expected)
+{
+    ParserKeyword s_to_disk("TO DISK");
+    ParserKeyword s_to_volume("TO VOLUME");
+    ParserKeyword s_delete("DELETE");
+    ParserStringLiteral parser_string_literal;
+    ParserExpression parser_exp;
+
+    ASTPtr expr_elem;
+    if (!parser_exp.parse(pos, expr_elem, expected))
+        return false;
+
+    PartDestinationType destination_type = PartDestinationType::DELETE;
+    String destination_name;
+    if (s_to_disk.ignore(pos))
+        destination_type = PartDestinationType::DISK;
+    else if (s_to_volume.ignore(pos))
+        destination_type = PartDestinationType::VOLUME;
+    else
+        s_delete.ignore(pos);
+
+    if (destination_type == PartDestinationType::DISK || destination_type == PartDestinationType::VOLUME)
+    {
+        ASTPtr ast_space_name;
+        if (!parser_string_literal.parse(pos, ast_space_name, expected))
+            return false;
+
+        destination_name = ast_space_name->as<ASTLiteral &>().value.get<const String &>();
+    }
+
+    node = std::make_shared<ASTTTLElement>(destination_type, destination_name);
+    node->children.push_back(expr_elem);
+
+    return true;
+}
+
 bool ParserIdentifierWithOptionalParameters::parseImpl(Pos & pos, ASTPtr & node, Expected & expected)
 {
     ParserIdentifier non_parametric;

dbms/src/Parsers/ExpressionElementParsers.h

@@ -320,4 +320,14 @@ class ParserIdentifierWithOptionalParameters : public IParserBase
     bool parseImpl(Pos & pos, ASTPtr & node, Expected & expected);
 };
 
+/** Element of TTL expression - same as expression element, but in addition,
+ *   TO DISK 'xxx' | TO VOLUME 'xxx' | DELETE could be specified
+  */
+class ParserTTLElement : public IParserBase
+{
+protected:
+    const char * getName() const { return "element of TTL expression"; }
+    bool parseImpl(Pos & pos, ASTPtr & node, Expected & expected);
+};
+
 }

dbms/src/Parsers/ExpressionListParsers.cpp

@@ -557,6 +557,13 @@ bool ParserOrderByExpressionList::parseImpl(Pos & pos, ASTPtr & node, Expected &
 }
 
 
+bool ParserTTLExpressionList::parseImpl(Pos & pos, ASTPtr & node, Expected & expected)
+{
+    return ParserList(std::make_unique<ParserTTLElement>(), std::make_unique<ParserToken>(TokenType::Comma), false)
+        .parse(pos, node, expected);
+}
+
+
 bool ParserNullityChecking::parseImpl(Pos & pos, ASTPtr & node, Expected & expected)
 {
     ASTPtr node_comp;

dbms/src/Parsers/ExpressionListParsers.h

@@ -386,6 +386,7 @@ class ParserKeyValuePair : public IParserBase
     bool parseImpl(Pos & pos, ASTPtr & node, Expected & expected) override;
 };
 
+
 /// Parser for list of key-value pairs.
 class ParserKeyValuePairsList : public IParserBase
 {
@@ -394,4 +395,12 @@ class ParserKeyValuePairsList : public IParserBase
     bool parseImpl(Pos & pos, ASTPtr & node, Expected & expected) override;
 };
 
+
+class ParserTTLExpressionList : public IParserBase
+{
+protected:
+    const char * getName() const { return "ttl expression"; }
+    bool parseImpl(Pos & pos, ASTPtr & node, Expected & expected);
+};
+
 }

dbms/src/Parsers/ParserAlterQuery.cpp

@@ -87,6 +87,7 @@ bool ParserAlterCommand::parseImpl(Pos & pos, ASTPtr & node, Expected & expected
         /* allow_empty = */ false);
     ParserSetQuery parser_settings(true);
     ParserNameList values_p;
+    ParserTTLExpressionList parser_ttl_list;
 
     if (is_live_view)
     {
@@ -236,9 +237,9 @@ bool ParserAlterCommand::parseImpl(Pos & pos, ASTPtr & node, Expected & expected
             command->part = true;
 
             if (s_to_disk.ignore(pos))
-                command->move_destination_type = ASTAlterCommand::MoveDestinationType::DISK;
+                command->move_destination_type = PartDestinationType::DISK;
             else if (s_to_volume.ignore(pos))
-                command->move_destination_type = ASTAlterCommand::MoveDestinationType::VOLUME;
+                command->move_destination_type = PartDestinationType::VOLUME;
             else
                 return false;
 
@@ -256,9 +257,9 @@ bool ParserAlterCommand::parseImpl(Pos & pos, ASTPtr & node, Expected & expected
             command->type = ASTAlterCommand::MOVE_PARTITION;
 
             if (s_to_disk.ignore(pos))
-                command->move_destination_type = ASTAlterCommand::MoveDestinationType::DISK;
+                command->move_destination_type = PartDestinationType::DISK;
             else if (s_to_volume.ignore(pos))
-                command->move_destination_type = ASTAlterCommand::MoveDestinationType::VOLUME;
+                command->move_destination_type = PartDestinationType::VOLUME;
             else
                 return false;
 
@@ -431,7 +432,7 @@ bool ParserAlterCommand::parseImpl(Pos & pos, ASTPtr & node, Expected & expected
         }
         else if (s_modify_ttl.ignore(pos, expected))
         {
-            if (!parser_exp_elem.parse(pos, command->ttl, expected))
+            if (!parser_ttl_list.parse(pos, command->ttl, expected))
                 return false;
             command->type = ASTAlterCommand::MODIFY_TTL;
         }

dbms/src/Parsers/ParserCreateQuery.cpp

@@ -250,6 +250,7 @@ bool ParserStorage::parseImpl(Pos & pos, ASTPtr & node, Expected & expected)
     ParserIdentifierWithOptionalParameters ident_with_optional_params_p;
     ParserExpression expression_p;
     ParserSetQuery settings_p(/* parse_only_internals_ = */ true);
+    ParserTTLExpressionList parser_ttl_list;
 
     ASTPtr engine;
     ASTPtr partition_by;
@@ -303,7 +304,7 @@ bool ParserStorage::parseImpl(Pos & pos, ASTPtr & node, Expected & expected)
 
         if (!ttl_table && s_ttl.ignore(pos, expected))
         {
-            if (expression_p.parse(pos, ttl_table, expected))
+            if (parser_ttl_list.parse(pos, ttl_table, expected))
                 continue;
             else
                 return false;