Merge pull request ClickHouse#78022 from GrigoryPervakov/summing-keys

Validate SummungMergeTree columns to sum are not in sort key, fix partition key validation. Fix auto resolve of summing cols.

Author: GrigoryPervakov

docs/en/engines/table-engines/mergetree-family/summingmergetree.md

@@ -35,9 +35,9 @@ For a description of request parameters, see [request description](../../../sql-
 #### columns {#columns}
 
 `columns` - a tuple with the names of columns where values will be summarized. Optional parameter.
-    The columns must be of a numeric type and must not be in the primary key.
+    The columns must be of a numeric type and must not be in the partition or sorting key.
 
- If `columns` is not specified, ClickHouse summarizes the values in all columns with a numeric data type that are not in the primary key.
+ If `columns` is not specified, ClickHouse summarizes the values in all columns with a numeric data type that are not in the sorting key.
 
 ### Query clauses {#query-clauses}
 

src/Core/SettingsChangesHistory.cpp

@@ -701,7 +701,7 @@ const VersionToSettingsChangesMap & getMergeTreeSettingsChangesHistory()
             {"max_postpone_time_for_failed_replicated_fetches_ms", 1ULL * 60 * 1000, 1ULL * 60 * 1000, "Added new setting to enable postponing fetch tasks in the replication queue."},
             {"max_postpone_time_for_failed_replicated_merges_ms", 1ULL * 60 * 1000, 1ULL * 60 * 1000, "Added new setting to enable postponing merge tasks in the replication queue."},
             {"max_postpone_time_for_failed_replicated_tasks_ms", 5ULL * 60 * 1000, 5ULL * 60 * 1000, "Added new setting to enable postponing tasks in the replication queue."},
-
+            {"allow_summing_columns_in_partition_or_order_key", true, false, "New setting to allow summing of partition or sorting key columns"},
         });
         addSettingsChanges(merge_tree_settings_changes_history, "25.3",
         {

src/Processors/Merges/Algorithms/SummingSortedAlgorithm.cpp

@@ -99,7 +99,7 @@ struct SummingSortedAlgorithm::AggregateDescription
 };
 
 
-static bool isInPrimaryKey(const SortDescription & description, const std::string & name)
+static bool isInSortingKey(const SortDescription & description, const std::string & name)
 {
     for (const auto & desc : description)
         if (desc.column_name == name)
@@ -108,10 +108,10 @@ static bool isInPrimaryKey(const SortDescription & description, const std::strin
     return false;
 }
 
-static bool isInPartitionKey(const std::string & column_name, const Names & partition_key_columns)
+static bool isInNames(const std::string & column_name, const Names & names)
 {
-    auto is_in_partition_key = std::find(partition_key_columns.begin(), partition_key_columns.end(), column_name);
-    return is_in_partition_key != partition_key_columns.end();
+    auto is_in_partition_key = std::find(names.begin(), names.end(), column_name);
+    return is_in_partition_key != names.end();
 }
 
 
@@ -205,7 +205,7 @@ static SummingSortedAlgorithm::ColumnsDefinition defineColumns(
     const Block & header,
     const SortDescription & description,
     const Names & column_names_to_sum,
-    const Names & partition_key_columns)
+    const Names & partition_and_sorting_required_columns)
 {
     size_t num_columns = header.columns();
     SummingSortedAlgorithm::ColumnsDefinition def;
@@ -254,16 +254,14 @@ static SummingSortedAlgorithm::ColumnsDefinition defineColumns(
                 continue;
             }
 
-            /// Are they inside the primary key or partition key?
-            if (isInPrimaryKey(description, column.name) || isInPartitionKey(column.name, partition_key_columns))
+            /// Are they inside the sorting key or partition key? Check both to ignore columns with order expression.
+            if (isInSortingKey(description, column.name) || isInNames(column.name, partition_and_sorting_required_columns))
             {
                 def.column_numbers_not_to_aggregate.push_back(i);
                 continue;
             }
 
-            if (column_names_to_sum.empty()
-                || column_names_to_sum.end() !=
-                   std::find(column_names_to_sum.begin(), column_names_to_sum.end(), column.name))
+            if (column_names_to_sum.empty() || isInNames(column.name, column_names_to_sum))
             {
                 // Create aggregator to sum this column
                 SummingSortedAlgorithm::AggregateDescription desc;
@@ -311,7 +309,8 @@ static SummingSortedAlgorithm::ColumnsDefinition defineColumns(
         /// no elements of map could be in primary key
         auto column_num_it = map.second.begin();
         for (; column_num_it != map.second.end(); ++column_num_it)
-            if (isInPrimaryKey(description, header.safeGetByPosition(*column_num_it).name))
+            if (isInSortingKey(description, header.safeGetByPosition(*column_num_it).name)
+                || isInNames(header.safeGetByPosition(*column_num_it).name, partition_and_sorting_required_columns))
                 break;
         if (column_num_it != map.second.end())
         {
@@ -690,11 +689,12 @@ SummingSortedAlgorithm::SummingSortedAlgorithm(
     size_t num_inputs,
     SortDescription description_,
     const Names & column_names_to_sum,
-    const Names & partition_key_columns,
+    const Names & partition_and_sorting_required_columns,
     size_t max_block_size_rows,
     size_t max_block_size_bytes)
     : IMergingAlgorithmWithDelayedChunk(header_, num_inputs, std::move(description_))
-    , columns_definition(defineColumns(header_, description, column_names_to_sum, partition_key_columns))
+    , columns_definition(
+          defineColumns(header_, description, column_names_to_sum, partition_and_sorting_required_columns))
     , merged_data(max_block_size_rows, max_block_size_bytes, columns_definition)
 {
 }

src/Processors/Merges/Algorithms/SummingSortedAlgorithm.h

@@ -21,7 +21,7 @@ class SummingSortedAlgorithm final : public IMergingAlgorithmWithDelayedChunk
         /// List of columns to be summed. If empty, all numeric columns that are not in the description are taken.
         const Names & column_names_to_sum,
         /// List of partition key columns. They have to be excluded.
-        const Names & partition_key_columns,
+        const Names & partition_and_sorting_required_columns,
         size_t max_block_size_rows,
         size_t max_block_size_bytes);
 

src/Processors/Merges/SummingSortedTransform.h

@@ -20,7 +20,7 @@ class SummingSortedTransform final : public IMergingTransform<SummingSortedAlgor
         const Block & header, size_t num_inputs,
         SortDescription description_,
         /// List of columns to be summed. If empty, all numeric columns that are not in the description are taken.
-        const Names & column_names_to_sum,
+        const Names & partition_and_sorting_required_columns,
         const Names & partition_key_columns,
         size_t max_block_size_rows,
         size_t max_block_size_bytes
@@ -30,7 +30,7 @@ class SummingSortedTransform final : public IMergingTransform<SummingSortedAlgor
             header,
             num_inputs,
             std::move(description_),
-            column_names_to_sum,
+            partition_and_sorting_required_columns,
             partition_key_columns,
             max_block_size_rows,
             max_block_size_bytes)

src/Processors/QueryPlan/ReadFromMergeTree.cpp

@@ -1283,7 +1283,7 @@ static void addMergingFinal(
     Pipe & pipe,
     const SortDescription & sort_description,
     MergeTreeData::MergingParams merging_params,
-    Names partition_key_columns,
+    const StorageMetadataPtr & metadata_snapshot,
     size_t max_block_size_rows,
     bool enable_vertical_final)
 {
@@ -1304,9 +1304,12 @@ static void addMergingFinal(
                 return std::make_shared<CollapsingSortedTransform>(header, num_outputs,
                             sort_description, merging_params.sign_column, true, max_block_size_rows, /*max_block_size_bytes=*/0);
 
-            case MergeTreeData::MergingParams::Summing:
+            case MergeTreeData::MergingParams::Summing: {
+                auto required_columns = metadata_snapshot->getPartitionKey().expression->getRequiredColumns();
+                required_columns.append_range(metadata_snapshot->getSortingKey().expression->getRequiredColumns());
                 return std::make_shared<SummingSortedTransform>(header, num_outputs,
-                            sort_description, merging_params.columns_to_sum, partition_key_columns, max_block_size_rows, /*max_block_size_bytes=*/0);
+                            sort_description, merging_params.columns_to_sum, required_columns, max_block_size_rows, /*max_block_size_bytes=*/0);
+            }
 
             case MergeTreeData::MergingParams::Aggregating:
                 return std::make_shared<AggregatingSortedTransform>(header, num_outputs,
@@ -1518,8 +1521,6 @@ Pipe ReadFromMergeTree::spreadMarkRangesAmongStreamsFinal(
         size_t sort_columns_size = sort_columns.size();
         sort_description.reserve(sort_columns_size);
 
-        Names partition_key_columns = storage_snapshot->metadata->getPartitionKey().column_names;
-
         for (size_t i = 0; i < sort_columns_size; ++i)
         {
             if (!reverse_flags.empty() && reverse_flags[i])
@@ -1533,7 +1534,7 @@ Pipe ReadFromMergeTree::spreadMarkRangesAmongStreamsFinal(
                 pipe,
                 sort_description,
                 data.merging_params,
-                partition_key_columns,
+                storage_snapshot->metadata,
                 block_size.max_block_size_rows,
                 enable_vertical_final);
 

src/Storages/MergeTree/MergeTask.cpp

@@ -1524,7 +1524,7 @@ class MergePartsStep : public ITransformingStep
     MergePartsStep(
         const Header & input_header_,
         const SortDescription & sort_description_,
-        const Names partition_key_columns_,
+        const Names partition_and_sorting_required_columns_,
         const MergeTreeData::MergingParams & merging_params_,
         const String & rows_sources_temporary_file_name_,
         UInt64 merge_block_size_rows_,
@@ -1534,7 +1534,7 @@ class MergePartsStep : public ITransformingStep
         time_t time_of_merge_)
         : ITransformingStep(input_header_, input_header_, getTraits())
         , sort_description(sort_description_)
-        , partition_key_columns(partition_key_columns_)
+        , partition_and_sorting_required_columns(partition_and_sorting_required_columns_)
         , merging_params(merging_params_)
         , rows_sources_temporary_file_name(rows_sources_temporary_file_name_)
         , merge_block_size_rows(merge_block_size_rows_)
@@ -1588,7 +1588,7 @@ class MergePartsStep : public ITransformingStep
 
             case MergeTreeData::MergingParams::Summing:
                 merged_transform = std::make_shared<SummingSortedTransform>(
-                    header, input_streams_count, sort_description, merging_params.columns_to_sum, partition_key_columns, merge_block_size_rows, merge_block_size_bytes);
+                    header, input_streams_count, sort_description, merging_params.columns_to_sum, partition_and_sorting_required_columns, merge_block_size_rows, merge_block_size_bytes);
                 break;
 
             case MergeTreeData::MergingParams::Aggregating:
@@ -1651,7 +1651,7 @@ class MergePartsStep : public ITransformingStep
     }
 
     const SortDescription sort_description;
-    const Names partition_key_columns;
+    const Names partition_and_sorting_required_columns;
     const MergeTreeData::MergingParams merging_params{};
     const String rows_sources_temporary_file_name;
     const UInt64 merge_block_size_rows;
@@ -1807,7 +1807,8 @@ void MergeTask::ExecuteAndFinalizeHorizontalPart::createMergedStream() const
         size_t sort_columns_size = sort_columns.size();
         sort_description.reserve(sort_columns_size);
 
-        Names partition_key_columns = global_ctx->metadata_snapshot->getPartitionKey().column_names;
+        auto partition_and_sorting_required_columns = global_ctx->metadata_snapshot->getPartitionKey().expression->getRequiredColumns();
+        partition_and_sorting_required_columns.append_range(global_ctx->metadata_snapshot->getSortingKey().expression->getRequiredColumns());
 
         for (size_t i = 0; i < sort_columns_size; ++i)
         {
@@ -1829,7 +1830,7 @@ void MergeTask::ExecuteAndFinalizeHorizontalPart::createMergedStream() const
         auto merge_step = std::make_unique<MergePartsStep>(
             merge_parts_query_plan.getCurrentHeader(),
             sort_description,
-            partition_key_columns,
+            partition_and_sorting_required_columns,
             global_ctx->merging_params,
             (is_vertical_merge ? RowsSourcesTemporaryFile::FILE_ID : ""), /// rows_sources' temporary file is used only for vertical merge
             (*merge_tree_settings)[MergeTreeSetting::merge_max_block_size],

src/Storages/MergeTree/MergeTreeData.cpp

@@ -196,6 +196,7 @@ namespace MergeTreeSetting
     extern const MergeTreeSettingsBool allow_nullable_key;
     extern const MergeTreeSettingsBool allow_remote_fs_zero_copy_replication;
     extern const MergeTreeSettingsBool allow_suspicious_indices;
+    extern const MergeTreeSettingsBool allow_summing_columns_in_partition_or_order_key;
     extern const MergeTreeSettingsBool assign_part_uuids;
     extern const MergeTreeSettingsBool async_insert;
     extern const MergeTreeSettingsBool check_sample_column_is_correct;
@@ -556,7 +557,7 @@ MergeTreeData::MergeTreeData(
     setProperties(metadata_, metadata_, !sanity_checks);
 
     /// NOTE: using the same columns list as is read when performing actual merges.
-    merging_params.check(metadata_);
+    merging_params.check(*settings, metadata_);
 
     if (metadata_.sampling_key.definition_ast != nullptr)
     {
@@ -1123,7 +1124,7 @@ void MergeTreeData::checkStoragePolicy(const StoragePolicyPtr & new_storage_poli
 }
 
 
-void MergeTreeData::MergingParams::check(const StorageInMemoryMetadata & metadata) const
+void MergeTreeData::MergingParams::check(const MergeTreeSettings & settings, const StorageInMemoryMetadata & metadata) const
 {
     const auto columns = metadata.getColumns().getAllPhysical();
 
@@ -1238,9 +1239,9 @@ void MergeTreeData::MergingParams::check(const StorageInMemoryMetadata & metadat
 
     if (mode == MergingParams::Summing)
     {
-        auto columns_to_sum_copy = columns_to_sum;
-        std::sort(columns_to_sum_copy.begin(), columns_to_sum_copy.end());
-        if (const auto it = std::adjacent_find(columns_to_sum_copy.begin(), columns_to_sum_copy.end()); it != columns_to_sum_copy.end())
+        auto columns_to_sum_sorted = columns_to_sum;
+        std::sort(columns_to_sum_sorted.begin(), columns_to_sum_sorted.end());
+        if (const auto it = std::adjacent_find(columns_to_sum_sorted.begin(), columns_to_sum_sorted.end()); it != columns_to_sum_sorted.end())
             throw Exception(ErrorCodes::BAD_ARGUMENTS, "Column {} is listed multiple times in the list of columns to sum", *it);
 
         /// If columns_to_sum are set, then check that such columns exist.
@@ -1257,20 +1258,39 @@ void MergeTreeData::MergingParams::check(const StorageInMemoryMetadata & metadat
                     column_to_sum);
         }
 
+        auto allow_summing_columns_in_partition_or_order_key = settings[MergeTreeSetting::allow_summing_columns_in_partition_or_order_key];
+
         /// Check that summing columns are not in partition key.
-        if (metadata.isPartitionKeyDefined())
+        if (!allow_summing_columns_in_partition_or_order_key && metadata.isPartitionKeyDefined())
         {
-            auto partition_key_columns = metadata.getPartitionKey().column_names;
+            auto partition_key_columns = metadata.getPartitionKey().expression->getRequiredColumns();
+            std::sort(partition_key_columns.begin(), partition_key_columns.end());
 
             Names names_intersection;
-            std::set_intersection(columns_to_sum.begin(), columns_to_sum.end(),
+            std::set_intersection(columns_to_sum_sorted.begin(), columns_to_sum_sorted.end(),
                                   partition_key_columns.begin(), partition_key_columns.end(),
                                   std::back_inserter(names_intersection));
 
             if (!names_intersection.empty())
                 throw Exception(ErrorCodes::BAD_ARGUMENTS, "Columns: {} listed both in columns to sum and in partition key. "
                 "That is not allowed.", boost::algorithm::join(names_intersection, ", "));
         }
+
+        /// Check that summing columns are not in sorting key.
+        if (!allow_summing_columns_in_partition_or_order_key && metadata.isSortingKeyDefined())
+        {
+            auto sorting_key_columns = metadata.getSortingKey().expression->getRequiredColumns();
+            std::sort(sorting_key_columns.begin(), sorting_key_columns.end());
+
+            Names names_intersection;
+            std::set_intersection(columns_to_sum_sorted.begin(), columns_to_sum_sorted.end(),
+                                  sorting_key_columns.begin(), sorting_key_columns.end(),
+                                  std::back_inserter(names_intersection));
+
+            if (!names_intersection.empty())
+                throw Exception(ErrorCodes::BAD_ARGUMENTS, "Columns: {} listed both in columns to sum and in sorting key. "
+                "That is not allowed.", boost::algorithm::join(names_intersection, ", "));
+        }
     }
 
     if (mode == MergingParams::Replacing)

src/Storages/MergeTree/MergeTreeData.h

@@ -368,7 +368,7 @@ class MergeTreeData : public IStorage, public WithMutableContext
         Graphite::Params graphite_params;
 
         /// Check that needed columns are present and have correct types.
-        void check(const StorageInMemoryMetadata & metadata) const;
+        void check(const MergeTreeSettings & settings, const StorageInMemoryMetadata & metadata) const;
 
         String getModeName() const;
     };