Merge pull request ClickHouse#78515 from CurtizJ/alter-conversions-refactoring

Minor refactoring near `AlterConversions`

Author: CurtizJ

src/Processors/QueryPlan/ReadFromMergeTree.cpp

@@ -2279,12 +2279,7 @@ void ReadFromMergeTree::initializePipeline(QueryPipelineBuilder & pipeline, cons
     {
         for (const auto & ranges_in_data_part : result.parts_with_ranges)
         {
-            auto alter_conversions = MergeTreeData::getAlterConversionsForPart(
-                ranges_in_data_part.data_part,
-                mutations_snapshot,
-                storage_snapshot->metadata,
-                getContext());
-
+            auto alter_conversions = MergeTreeData::getAlterConversionsForPart(ranges_in_data_part.data_part, mutations_snapshot, getContext());
             auto part_info = std::make_shared<LoadedMergeTreeDataPartInfoForReader>(ranges_in_data_part.data_part, std::move(alter_conversions));
             lazily_read_info->data_part_infos->emplace(ranges_in_data_part.part_index_in_query, std::move(part_info));
         }

src/Storages/MergeTree/AlterConversions.cpp

@@ -64,18 +64,38 @@ static MutationCommand createCommandWithUpdatedColumns(
     return res;
 }
 
+AlterConversions::AlterConversions(
+    const MutationCommands & mutation_commands_,
+    const ContextPtr & context)
+{
+    for (const auto & command : mutation_commands_)
+        addMutationCommand(command, context);
+
+    /// Do not throw if there are no mutations or patches.
+    if (number_of_alter_mutations > 1)
+    {
+        if (!mutation_commands.empty())
+            throw Exception(ErrorCodes::NOT_IMPLEMENTED,
+                "Applying mutations on-fly is not supported with more than one ALTER MODIFY");
+    }
+}
+
 bool AlterConversions::isSupportedDataMutation(MutationCommand::Type type)
 {
-    using enum MutationCommand::Type;
-    return type == READ_COLUMN || type == UPDATE || type == DELETE;
+    return type == MutationCommand::UPDATE || type == MutationCommand::DELETE;
+}
+
+bool AlterConversions::isSupportedAlterMutation(MutationCommand::Type type)
+{
+    return type == MutationCommand::READ_COLUMN;
 }
 
 bool AlterConversions::isSupportedMetadataMutation(MutationCommand::Type type)
 {
     return type == MutationCommand::Type::RENAME_COLUMN;
 }
 
-void AlterConversions::addMutationCommand(const MutationCommand & command)
+void AlterConversions::addMutationCommand(const MutationCommand & command, const ContextPtr & context)
 {
     using enum MutationCommand::Type;
 
@@ -85,12 +105,12 @@ void AlterConversions::addMutationCommand(const MutationCommand & command)
     }
     else if (command.type == READ_COLUMN)
     {
-        ++number_of_alter_conversions;
+        ++number_of_alter_mutations;
         position_of_alter_conversion = mutation_commands.size();
     }
     else if (command.type == UPDATE || command.type == DELETE)
     {
-        const auto result = findFirstNonDeterministicFunction(command, getContext());
+        const auto result = findFirstNonDeterministicFunction(command, context);
         if (result.subquery)
             throw Exception(ErrorCodes::BAD_ARGUMENTS,
                 "ALTER UPDATE/ALTER DELETE statement with subquery may be nondeterministic and cannot be applied on fly");
@@ -105,11 +125,6 @@ void AlterConversions::addMutationCommand(const MutationCommand & command)
 
         mutation_commands.push_back(command);
     }
-
-    /// Do not throw if there are no mutations.
-    if (number_of_alter_conversions > 1 && !mutation_commands.empty())
-        throw Exception(ErrorCodes::NOT_IMPLEMENTED,
-            "Applying mutations on-fly is not supported with more than one ALTER MODIFY");
 }
 
 bool AlterConversions::columnHasNewName(const std::string & old_name) const
@@ -157,10 +172,12 @@ std::string AlterConversions::getColumnOldName(const std::string & new_name) con
 
 PrewhereExprSteps AlterConversions::getMutationSteps(
     const IMergeTreeDataPartInfoForReader & part_info,
-    const NamesAndTypesList & read_columns) const
+    const NamesAndTypesList & read_columns,
+    const StorageMetadataPtr & metadata_snapshot,
+    const ContextPtr & context) const
 {
-    auto actions_chain = getMutationActions(part_info, read_columns, true);
-    auto settings = ExpressionActionsSettings(getContext());
+    auto actions_chain = getMutationActions(part_info, read_columns, metadata_snapshot, context);
+    auto settings = ExpressionActionsSettings(context);
 
     PrewhereExprSteps steps;
     for (size_t i = 0; i < actions_chain.size(); ++i)
@@ -191,7 +208,8 @@ PrewhereExprSteps AlterConversions::getMutationSteps(
 std::vector<MutationActions> AlterConversions::getMutationActions(
     const IMergeTreeDataPartInfoForReader & part_info,
     const NamesAndTypesList & read_columns,
-    bool can_execute) const
+    const StorageMetadataPtr & metadata_snapshot,
+    const ContextPtr & context) const
 {
     if (mutation_commands.empty())
         return {};
@@ -211,19 +229,16 @@ std::vector<MutationActions> AlterConversions::getMutationActions(
             storage_read_columns.emplace_back(name_in_storage);
     }
 
-    addColumnsRequiredForMaterialized(storage_read_columns, storage_read_columns_set);
+    addColumnsRequiredForMaterialized(storage_read_columns, storage_read_columns_set, metadata_snapshot, context);
     auto filtered_commands = filterMutationCommands(storage_read_columns, std::move(storage_read_columns_set));
 
     if (filtered_commands.empty())
         return {};
 
-    if (can_execute)
-    {
-        ProfileEvents::increment(ProfileEvents::ReadTasksWithAppliedMutationsOnFly);
-        ProfileEvents::increment(ProfileEvents::MutationsAppliedOnFlyInAllReadTasks, filtered_commands.size());
-    }
+    ProfileEvents::increment(ProfileEvents::ReadTasksWithAppliedMutationsOnFly);
+    ProfileEvents::increment(ProfileEvents::MutationsAppliedOnFlyInAllReadTasks, filtered_commands.size());
 
-    MutationsInterpreter::Settings settings(can_execute);
+    MutationsInterpreter::Settings settings(true);
     settings.return_all_columns = true;
     settings.recalculate_dependencies_of_updated_columns = false;
 
@@ -237,13 +252,17 @@ std::vector<MutationActions> AlterConversions::getMutationActions(
         metadata_snapshot,
         std::move(filtered_commands),
         std::move(storage_read_columns),
-        getContext(),
+        context,
         settings);
 
     return interpreter.getMutationActions();
 }
 
-void AlterConversions::addColumnsRequiredForMaterialized(Names & read_columns, NameSet & read_columns_set) const
+void AlterConversions::addColumnsRequiredForMaterialized(
+    Names & read_columns,
+    NameSet & read_columns_set,
+    const StorageMetadataPtr & metadata_snapshot,
+    const ContextPtr & context) const
 {
     NameSet required_source_columns;
     const auto & columns_desc = metadata_snapshot->getColumns();
@@ -255,7 +274,7 @@ void AlterConversions::addColumnsRequiredForMaterialized(Names & read_columns, N
         if (default_desc && default_desc->kind == ColumnDefaultKind::Materialized)
         {
             auto query = default_desc->expression->clone();
-            auto syntax_result = TreeRewriter(getContext()).analyze(query, source_columns);
+            auto syntax_result = TreeRewriter(context).analyze(query, source_columns);
 
             for (const auto & dependency : syntax_result->requiredSourceColumns())
             {
@@ -324,4 +343,16 @@ MutationCommands AlterConversions::filterMutationCommands(Names & read_columns,
     return filtered_commands;
 }
 
+void MutationCounters::assertNotNegative() const
+{
+    if (num_data < 0)
+        throw Exception(ErrorCodes::LOGICAL_ERROR, "On-fly data mutations counter is negative ({})", num_data);
+
+    if (num_alter < 0)
+        throw Exception(ErrorCodes::LOGICAL_ERROR, "On-fly alter mutations counter is negative ({})", num_alter);
+
+    if (num_metadata < 0)
+        throw Exception(ErrorCodes::LOGICAL_ERROR, "On-fly metadata mutations counter is negative ({})", num_metadata);
+}
+
 }

src/Storages/MergeTree/AlterConversions.h

@@ -8,31 +8,27 @@ namespace DB
 {
 
 struct StorageInMemoryMetadata;
-using StorageMetadataPtr = std::shared_ptr<const StorageInMemoryMetadata>;
-
 class IMergeTreeDataPartInfoForReader;
+using StorageMetadataPtr = std::shared_ptr<const StorageInMemoryMetadata>;
 
 /// Alter conversions which should be applied on-fly for part.
 /// Built from of the most recent mutation commands for part.
 /// Now ALTER RENAME COLUMN, ALTER UPDATE and ALTER DELETE are applied.
-class AlterConversions : private WithContext, boost::noncopyable
+class AlterConversions : private boost::noncopyable
 {
 public:
     AlterConversions() = default;
 
-    AlterConversions(StorageMetadataPtr metadata_snapshot_, ContextPtr context_)
-        : WithContext(context_)
-        , metadata_snapshot(std::move(metadata_snapshot_))
-    {
-    }
+    AlterConversions(
+        const MutationCommands & mutation_commands_,
+        const ContextPtr & context);
 
     struct RenamePair
     {
         std::string rename_to;
         std::string rename_from;
     };
 
-    void addMutationCommand(const MutationCommand & command);
     const std::vector<RenamePair> & getRenameMap() const { return rename_map; }
 
     /// Column was renamed (lookup by value in rename_map)
@@ -45,6 +41,7 @@ class AlterConversions : private WithContext, boost::noncopyable
     std::string getColumnOldName(const std::string & new_name) const;
 
     static bool isSupportedDataMutation(MutationCommand::Type type);
+    static bool isSupportedAlterMutation(MutationCommand::Type type);
     static bool isSupportedMetadataMutation(MutationCommand::Type type);
 
     const NameSet & getAllUpdatedColumns() const { return all_updated_columns; }
@@ -54,25 +51,32 @@ class AlterConversions : private WithContext, boost::noncopyable
     /// mutations that affect columns from @read_columns.
     PrewhereExprSteps getMutationSteps(
         const IMergeTreeDataPartInfoForReader & part_info,
-        const NamesAndTypesList & read_columns) const;
+        const NamesAndTypesList & read_columns,
+        const StorageMetadataPtr & metadata_snapshot,
+        const ContextPtr & context) const;
 
 private:
+    void addMutationCommand(const MutationCommand & command, const ContextPtr & context);
+
     /// Returns a chain of actions that can be
     /// applied to block to execute mutation commands
     /// that affect columns from @read_columns.
     std::vector<MutationActions> getMutationActions(
         const IMergeTreeDataPartInfoForReader & part_info,
         const NamesAndTypesList & read_columns,
-        bool can_execute) const;
+        const StorageMetadataPtr & metadata_snapshot,
+        const ContextPtr & context) const;
 
     /// Adds source columns of expressions of MATERIALIZED columns from @read_columns if any.
-    void addColumnsRequiredForMaterialized(Names & read_columns, NameSet & read_columns_set) const;
+    void addColumnsRequiredForMaterialized(
+        Names & read_columns,
+        NameSet & read_columns_set,
+        const StorageMetadataPtr & metadata_snapshot,
+        const ContextPtr & context) const;
 
     /// Returns only mutations commands that affect columns from set.
     MutationCommands filterMutationCommands(Names & read_columns, NameSet read_columns_set) const;
 
-    StorageMetadataPtr metadata_snapshot;
-
     /// Rename map new_name -> old_name.
     std::vector<RenamePair> rename_map;
 
@@ -85,12 +89,21 @@ class AlterConversions : private WithContext, boost::noncopyable
     /// The number of ALTER MODIFY COLUMN commands.
     /// Applying on-fly mutations is not supported
     /// in presence of more than one of such commands.
-    size_t number_of_alter_conversions = 0;
+    size_t number_of_alter_mutations = 0;
 
     /// Names of columns which are updated by mutation commands.
     NameSet all_updated_columns;
 };
 
 using AlterConversionsPtr = std::shared_ptr<const AlterConversions>;
 
+struct MutationCounters
+{
+    Int64 num_data = 0;
+    Int64 num_alter = 0;
+    Int64 num_metadata = 0;
+
+    void assertNotNegative() const;
+};
+
 }

src/Storages/MergeTree/MergeTask.cpp

@@ -487,7 +487,7 @@ bool MergeTask::ExecuteAndFinalizeHorizontalPart::prepare() const
             infos.add(part_infos);
         }
 
-        global_ctx->alter_conversions.push_back(MergeTreeData::getAlterConversionsForPart(part, mutations_snapshot, global_ctx->metadata_snapshot, global_ctx->context));
+        global_ctx->alter_conversions.push_back(MergeTreeData::getAlterConversionsForPart(part, mutations_snapshot, global_ctx->context));
     }
 
     const auto & local_part_min_ttl = global_ctx->new_data_part->ttl_infos.part_min_ttl;