Merge pull request ClickHouse#87537 from ClickHouse/backport/25.8/87347

Backport ClickHouse#87347 to 25.8: Fix applying patches to columns with default expression

Author: clickhouse-gh[bot]

src/Storages/MergeTree/MergeTreeReadersChain.cpp

@@ -152,32 +152,33 @@ void MergeTreeReadersChain::executeActionsBeforePrewhere(
     }
 
     auto patch_max_version = getMaxPatchVersionForStep(range_reader);
+    const auto & result_header = range_reader.getReadSampleBlock();
+    auto columns_for_patches = getColumnsForPatches(result_header, read_columns);
 
-    /// Apply patches without min version for new columns because
-    /// if column is read on this step, it is not used by previous steps.
-    applyPatches(
-        range_reader.getReadSampleBlock(),
-        read_columns,
-        result.patch_versions_block,
-        /*min_version=*/ {},
-        patch_max_version,
-        /*after_conversions=*/ false,
-        result.columns_for_patches);
+    auto apply_patches = [&](ColumnForPatch::Order order)
+    {
+        /// Apply patches without min version for new columns because
+        /// if column is read on this step, it is not used by previous steps.
+        applyPatches(
+            result_header,
+            read_columns,
+            result.patch_versions_block,
+            /*min_version=*/ {},
+            patch_max_version,
+            columns_for_patches,
+            {order},
+            result.columns_for_patches);
+    };
+
+    apply_patches(ColumnForPatch::Order::BeforeConversions);
 
     /// If columns not empty, then apply on-fly alter conversions if any required
     if (!prewhere_info || prewhere_info->perform_alter_conversions)
     {
         merge_tree_reader->performRequiredConversions(read_columns);
     }
 
-    applyPatches(
-        range_reader.getReadSampleBlock(),
-        read_columns,
-        result.patch_versions_block,
-        /*min_version=*/ {},
-        patch_max_version,
-        /*after_conversions=*/ true,
-        result.columns_for_patches);
+    apply_patches(ColumnForPatch::Order::AfterConversions);
 
     /// If some columns absent in part, then evaluate default values
     if (should_evaluate_missing_defaults)
@@ -192,6 +193,8 @@ void MergeTreeReadersChain::executeActionsBeforePrewhere(
         addDummyColumnWithRowCount(additional_columns, result.num_rows);
         merge_tree_reader->evaluateMissingDefaults(additional_columns, read_columns);
     }
+
+    apply_patches(ColumnForPatch::Order::AfterEvaluatingDefaults);
 }
 
 void MergeTreeReadersChain::executePrewhereActions(MergeTreeRangeReader & reader, ReadResult & result, const Block & previous_header, bool is_last_reader)
@@ -258,8 +261,57 @@ void MergeTreeReadersChain::readPatches(const Block & result_header, std::vector
     }
 }
 
+ColumnsForPatches MergeTreeReadersChain::getColumnsForPatches(const Block & header, const Columns & columns) const
+{
+    ColumnsForPatches res;
+    auto block = header.cloneWithColumns(columns);
+    using enum ColumnForPatch::Order;
+
+    for (const auto & patch_reader : patch_readers)
+    {
+        const auto & patch = patch_reader->getPatchPart();
+        const auto & patch_columns = patch.part->getColumnsDescription();
+        const auto & alter_conversions = patch.part->getAlterConversions();
+        auto & columns_for_patch = res.emplace_back();
+
+        for (const auto & column : block)
+        {
+            if (isPatchPartSystemColumn(column.name))
+                continue;
+
+            String column_name_in_patch = column.name;
+            if (alter_conversions && alter_conversions->isColumnRenamed(column.name))
+                column_name_in_patch = alter_conversions->getColumnOldName(column.name);
+
+            if (!patch_columns.hasColumnOrSubcolumn(GetColumnsOptions::All, column_name_in_patch))
+                continue;
+
+            /// Add requested column name, not the column name in patch, for correct applying patches.
+            /// This column name is translated to the column name in patch in MergeTree reader.
+            /// If columns is not created at this stage, it will be filled with defaults at the latest stage of reading.
+            if (!column.column)
+            {
+                columns_for_patch.emplace_back(column.name, AfterEvaluatingDefaults);
+            }
+            else if (patch.perform_alter_conversions)
+            {
+                columns_for_patch.emplace_back(column.name, AfterConversions);
+            }
+            else
+            {
+                columns_for_patch.emplace_back(column.name, BeforeConversions);
+            }
+        }
+    }
+
+    return res;
+}
+
 void MergeTreeReadersChain::applyPatchesAfterReader(ReadResult & result, size_t reader_index)
 {
+    if (patch_readers.empty() || result.num_rows == 0 || result.columns.empty())
+        return;
+
     auto & current_reader = range_readers.at(reader_index);
 
     std::optional<UInt64> min_version = getMaxPatchVersionForStep(current_reader);
@@ -277,53 +329,42 @@ void MergeTreeReadersChain::applyPatchesAfterReader(ReadResult & result, size_t
             column.column = ColumnUInt64::create(result.patch_versions_block.rows(), *min_version);
     }
 
+    const auto & result_header = current_reader.getSampleBlock();
+    auto columns_for_patches = getColumnsForPatches(result_header, result.columns);
+
+    using enum ColumnForPatch::Order;
+    std::set<ColumnForPatch::Order> suitable_orders = {AfterConversions, AfterEvaluatingDefaults};
+
     applyPatches(
-        current_reader.getSampleBlock(),
+        result_header,
         result.columns,
         result.patch_versions_block,
         min_version,
         max_version,
-        /*after_conversions=*/ true,
+        columns_for_patches,
+        suitable_orders,
         result.columns_for_patches);
 }
 
-static UInt128 getColumnsHash(const Names & column_names)
-{
-    SipHash hash;
-    for (const auto & column_name : column_names)
-        hash.update(column_name);
-    return hash.get128();
-}
-
-Names getUpdatedColumns(const Block & block, const IMergeTreeDataPartInfoForReader & patch)
+struct NamesHash
 {
-    Names res;
-    const auto & patch_columns = patch.getColumnsDescription();
-    const auto & alter_conversions = patch.getAlterConversions();
-
-    for (const auto & column : block)
+    size_t operator()(const Names & column_names) const
     {
-        if (isPatchPartSystemColumn(column.name))
-            continue;
-
-        String column_name = column.name;
-        if (alter_conversions && alter_conversions->isColumnRenamed(column_name))
-            column_name = alter_conversions->getColumnOldName(column_name);
-
-        if (patch_columns.hasColumnOrSubcolumn(GetColumnsOptions::All, column_name))
-            res.push_back(std::move(column_name));
+        SipHash hash;
+        for (const auto & column_name : column_names)
+            hash.update(column_name);
+        return hash.get64();
     }
-
-    return res;
-}
+};
 
 void MergeTreeReadersChain::applyPatches(
     const Block & result_header,
     Columns & result_columns,
     Block & versions_block,
     std::optional<UInt64> min_version,
     std::optional<UInt64> max_version,
-    bool after_conversions,
+    const ColumnsForPatches & columns_for_patches,
+    const std::set<ColumnForPatch::Order> & suitable_orders,
     const Block & additional_columns) const
 {
     if (patch_readers.empty() || result_columns.empty())
@@ -333,7 +374,7 @@ void MergeTreeReadersChain::applyPatches(
     addPatchVirtuals(result_block, additional_columns);
 
     /// Combine patches with the same structure.
-    std::unordered_map<UInt128, PatchesToApply> patches_to_apply;
+    std::unordered_map<Names, PatchesToApply, NamesHash> patches_to_apply;
     UInt64 source_data_version = patch_readers.front()->getPatchPart().source_data_version;
 
     for (size_t i = 0; i < patch_readers.size(); ++i)
@@ -354,15 +395,17 @@ void MergeTreeReadersChain::applyPatches(
         if (max_version.has_value() && patchHasHigherDataVersion(*patch.part, *max_version))
             continue;
 
-        if (after_conversions != patch.perform_alter_conversions)
-            continue;
+        Names updated_columns;
+        for (const auto & columns_for_patch : columns_for_patches[i])
+        {
+            if (suitable_orders.contains(columns_for_patch.order))
+                updated_columns.push_back(columns_for_patch.column_name);
+        }
 
-        auto updated_columns = getUpdatedColumns(result_block, *patch.part);
         if (updated_columns.empty())
             continue;
 
         std::sort(updated_columns.begin(), updated_columns.end());
-        auto columns_hash = getColumnsHash(updated_columns);
 
         for (const auto & patch_result : patch_results)
         {
@@ -372,16 +415,16 @@ void MergeTreeReadersChain::applyPatches(
             for (auto & patch_to_apply : patches)
             {
                 if (!patch_to_apply->empty())
-                    patches_to_apply[columns_hash].push_back(std::move(patch_to_apply));
+                    patches_to_apply[updated_columns].push_back(std::move(patch_to_apply));
             }
         }
     }
 
     if (min_version.has_value())
         source_data_version = std::max(source_data_version, *min_version);
 
-    for (const auto & [_, patches] : patches_to_apply)
-        applyPatchesToBlock(result_block, versions_block, patches, source_data_version);
+    for (const auto & [updated_columns, patches] : patches_to_apply)
+        applyPatchesToBlock(result_block, versions_block, patches, updated_columns, source_data_version);
 
     result_columns = result_block.getColumns();
     result_columns.resize(result_header.columns());

src/Storages/MergeTree/MergeTreeReadersChain.h

@@ -7,6 +7,27 @@ namespace DB
 
 using RangeReaders = std::vector<MergeTreeRangeReader>;
 
+struct ColumnForPatch
+{
+    enum class Order
+    {
+        /// Apply patch before converting the column to actual type.
+        BeforeConversions,
+        /// Apply patch after converting the column to actual type.
+        AfterConversions,
+        /// Apply patch after evaluating missing defaults for the column.
+        AfterEvaluatingDefaults,
+    };
+
+    ColumnForPatch(const String & column_name_, Order order_) : column_name(column_name_), order(order_) {}
+
+    String column_name;
+    Order order;
+};
+
+using ColumnsForPatch = std::vector<ColumnForPatch>;
+using ColumnsForPatches = std::vector<ColumnsForPatch>;
+
 class MergeTreeReadersChain
 {
 public:
@@ -44,14 +65,16 @@ class MergeTreeReadersChain
     void addPatchVirtuals(Block & to, const Block & from) const;
     void addPatchVirtuals(ReadResult & result, const Block & header) const;
     void applyPatchesAfterReader(ReadResult & result, size_t reader_index);
+    ColumnsForPatches getColumnsForPatches(const Block & header, const Columns & columns) const;
 
     void applyPatches(
         const Block & result_header,
         Columns & result_columns,
         Block & versions_block,
         std::optional<UInt64> min_version,
         std::optional<UInt64> max_version,
-        bool after_conversions,
+        const ColumnsForPatches & columns_for_patches,
+        const std::set<ColumnForPatch::Order> & suitable_orders,
         const Block & additional_columns) const;
 
     RangeReaders range_readers;

src/Storages/MergeTree/PatchParts/applyPatches.cpp

@@ -249,7 +249,7 @@ IColumn::Patch CombinedPatchBuilder::createPatchForColumn(const String & column_
     };
 }
 
-Block getUpdatedHeader(const PatchesToApply & patches, const Block & result_block)
+Block getUpdatedHeader(const PatchesToApply & patches, const NameSet & updated_columns)
 {
     std::vector<Block> headers;
 
@@ -266,9 +266,8 @@ Block getUpdatedHeader(const PatchesToApply & patches, const Block & result_bloc
 
         for (const auto & column : patch->patch_blocks[0])
         {
-            /// System columns may differ in patches because we allow to apply combined patches
-            /// with different modes. Ignore columns that are not present in result block.
-            if (isPatchPartSystemColumn(column.name) || !result_block.has(column.name))
+            /// Ignore columns that are not updated.
+            if (!updated_columns.contains(column.name))
                 header.erase(column.name);
         }
 
@@ -522,13 +521,15 @@ void applyPatchesToBlock(
     Block & result_block,
     Block & versions_block,
     const PatchesToApply & patches,
+    const Names & updated_columns,
     UInt64 source_data_version)
 {
     if (patches.empty())
         return;
 
     ProfileEventTimeIncrement<Microseconds> watch(ProfileEvents::ApplyPatchesMicroseconds);
-    auto updated_header = getUpdatedHeader(patches, result_block);
+    NameSet updated_columns_set(updated_columns.begin(), updated_columns.end());
+    auto updated_header = getUpdatedHeader(patches, updated_columns_set);
 
     if (canApplyPatchesRaw(patches))
         applyPatchesToBlockRaw(result_block, versions_block, patches, updated_header, source_data_version);

src/Storages/MergeTree/PatchParts/applyPatches.h

@@ -66,6 +66,7 @@ void applyPatchesToBlock(
     Block & result_block,
     Block & versions_block,
     const PatchesToApply & patches,
+    const Names & updated_columns,
     UInt64 source_data_version);
 
 }

tests/queries/0_stateless/03100_lwu_44_missing_default.reference

@@ -0,0 +1,40 @@
+0	999
+1	1000
+2	1000
+3	1000
+4	1000
+5	1000
+6	1000
+7	1000
+8	1000
+9	1000
+0	999
+1	1000
+2	1000
+3	1000
+4	1000
+5	1000
+6	1000
+7	1000
+8	1000
+9	1000
+110	999
+111	1000
+112	1000
+113	1000
+114	1000
+115	1000
+116	1000
+117	1000
+118	1000
+119	1000
+110	999
+111	1000
+112	1000
+113	1000
+114	1000
+115	1000
+116	1000
+117	1000
+118	1000
+119	1000