Backport ClickHouse#88828 to 25.8: Revert "Revert "Fix coalescing merge tree for tuple types""

Author: robot-clickhouse

src/Processors/Merges/Algorithms/SummingSortedAlgorithm.cpp

@@ -1,12 +1,15 @@
+#include <memory>
 #include <Processors/Merges/Algorithms/SummingSortedAlgorithm.h>
 
 #include <AggregateFunctions/AggregateFunctionFactory.h>
 #include <Columns/ColumnAggregateFunction.h>
 #include <Columns/ColumnTuple.h>
+#include <Common/Exception.h>
 #include <Common/AlignedBuffer.h>
 #include <Common/Arena.h>
 #include <Common/FieldVisitorSum.h>
 #include <Common/StringUtils.h>
+#include <DataTypes/DataTypeTuple.h>
 #include <DataTypes/DataTypeArray.h>
 #include <DataTypes/DataTypeCustomSimpleAggregateFunction.h>
 #include <DataTypes/NestedUtils.h>
@@ -51,7 +54,7 @@ struct SummingSortedAlgorithm::AggregateDescription
     bool is_agg_func_type = false;
     bool is_simple_agg_func_type = false;
     bool remove_default_values;
-    bool aggregate_all_columns;
+    bool aggregate_all_columns = false;
 
     String sum_function_map_name;
 
@@ -257,6 +260,45 @@ static SummingSortedAlgorithm::ColumnsDefinition defineColumns(
         const ColumnWithTypeAndName & column = header.safeGetByPosition(i);
 
         const auto * simple = dynamic_cast<const DataTypeCustomSimpleAggregateFunction *>(column.type->getCustomName());
+        bool is_non_empty_tuple = typeid_cast<const DataTypeTuple *>(column.type.get()) && !typeid_cast<const DataTypeTuple *>(column.type.get())->getElements().empty();
+        if (aggregate_all_columns && (is_non_empty_tuple || typeid_cast<const DataTypeArray *>(column.type.get())) && !simple)
+        {
+            const auto map_name = Nested::extractTableName(column.name);
+            /// if nested table name ends with `Map` it is a possible candidate for special handling
+            if (map_name == column.name || !endsWith(map_name, "Map"))
+            {
+                bool is_agg_func = WhichDataType(column.type).isAggregateFunction();
+
+                SummingSortedAlgorithm::AggregateDescription desc;
+                desc.remove_default_values = remove_default_values;
+                desc.aggregate_all_columns = aggregate_all_columns;
+                desc.sum_function_map_name = sum_function_map_name;
+                desc.is_agg_func_type = is_agg_func;
+                desc.column_numbers = {i};
+
+                desc.real_type = column.type;
+                desc.nested_type = recursiveRemoveLowCardinality(desc.real_type);
+                if (desc.real_type.get() == desc.nested_type.get())
+                    desc.nested_type = nullptr;
+
+                if (simple)
+                {
+                    // simple aggregate function
+                    desc.init(simple->getFunction(), true);
+                    if (desc.function->allocatesMemoryInArena())
+                        def.allocates_memory_in_arena = true;
+                }
+                else if (!is_agg_func)
+                {
+                    desc.init(sum_function_name.c_str(), {column.type});
+                }
+
+                def.columns_to_aggregate.emplace_back(std::move(desc));
+            }
+
+            continue;
+        }
+
         if (column.name == BlockNumberColumn::name || column.name == BlockOffsetColumn::name)
         {
             def.column_numbers_not_to_aggregate.push_back(i);
@@ -281,7 +323,15 @@ static SummingSortedAlgorithm::ColumnsDefinition defineColumns(
             bool is_agg_func = WhichDataType(column.type).isAggregateFunction();
 
             /// There are special const columns for example after prewhere sections.
-            if (!aggregate_all_columns && ((!column.type->isSummable() && !is_agg_func && !simple) || isColumnConst(*column.column)))
+            if (!aggregate_all_columns)
+            {
+                if ((!column.type->isSummable() && !is_agg_func && !simple) || isColumnConst(*column.column))
+                {
+                    def.column_numbers_not_to_aggregate.push_back(i);
+                    continue;
+                }
+            }
+            else if (column.type->getTypeId() == TypeIndex::Tuple)
             {
                 def.column_numbers_not_to_aggregate.push_back(i);
                 continue;
@@ -452,10 +502,17 @@ static void postprocessChunk(
 
         if (!desc.is_agg_func_type && !desc.is_simple_agg_func_type && isTuple(desc.function->getResultType()))
         {
-            /// Unpack tuple into block.
-            size_t tuple_size = desc.column_numbers.size();
-            for (size_t i = 0; i < tuple_size; ++i)
-                res_columns[desc.column_numbers[i]] = assert_cast<const ColumnTuple &>(*column).getColumnPtr(i);
+            if (desc.aggregate_all_columns)
+            {
+                res_columns[desc.column_numbers[0]] = column;
+            }
+            else
+            {
+                /// Unpack tuple into block.
+                size_t tuple_size = desc.column_numbers.size();
+                for (size_t i = 0; i < tuple_size; ++i)
+                   res_columns[desc.column_numbers[i]] = assert_cast<const ColumnTuple &>(*column).getColumnPtr(i);
+            }
         }
         else if (desc.nested_type)
         {
@@ -535,12 +592,24 @@ void SummingSortedAlgorithm::SummingMergedData::initialize(const DB::Block & hea
         // Wrap aggregated columns in a tuple to match function signature
         if (!desc.is_agg_func_type && !desc.is_simple_agg_func_type && isTuple(desc.function->getResultType()))
         {
-            size_t tuple_size = desc.column_numbers.size();
-            MutableColumns tuple_columns(tuple_size);
-            for (size_t i = 0; i < tuple_size; ++i)
-                tuple_columns[i] = std::move(columns[desc.column_numbers[i]]);
+            if (desc.aggregate_all_columns)
+            {
+                auto column = desc.real_type->createColumn();
+                size_t tuple_size = static_cast<const ColumnTuple &>(*column).tupleSize();
+                MutableColumns tuple_columns(tuple_size);
+                for (size_t i = 0; i < tuple_size; ++i)
+                    tuple_columns[i] = static_cast<const ColumnTuple &>(*column).getColumnPtr(i)->cloneEmpty();
+                new_columns.emplace_back(ColumnTuple::create(std::move(tuple_columns)));
+            }
+            else
+            {
+                size_t tuple_size = desc.column_numbers.size();
+                MutableColumns tuple_columns(tuple_size);
+                for (size_t i = 0; i < tuple_size; ++i)
+                    tuple_columns[i] = std::move(columns[desc.column_numbers[i]]);
 
-            new_columns.emplace_back(ColumnTuple::create(std::move(tuple_columns)));
+                new_columns.emplace_back(ColumnTuple::create(std::move(tuple_columns)));
+            }
         }
         else
         {
@@ -734,7 +803,6 @@ Chunk SummingSortedAlgorithm::SummingMergedData::pull()
 {
     auto chunk = MergedData::pull();
     postprocessChunk(chunk, def.column_names.size(), def);
-
     initAggregateDescription();
 
     return chunk;

tests/queries/0_stateless/03652_coalescing_merge_tree_fix_empty_tuple.reference

@@ -0,0 +1,6 @@
+(1,2)
+(3,4)
+[1,2]
+[3,4]
+()
+()

tests/queries/0_stateless/03652_coalescing_merge_tree_fix_empty_tuple.sql

@@ -0,0 +1,40 @@
+DROP TABLE IF EXISTS t0;
+
+CREATE TABLE t0 (c0 Tuple(a Int32, b Nullable(Int32)), c1 Int32) ENGINE = SummingMergeTree() ORDER BY c1;
+INSERT INTO t0 VALUES ((1,2), 0);
+INSERT INTO t0 VALUES ((3,4), 0);
+SELECT c0 FROM t0 FINAL;
+
+DROP TABLE IF EXISTS t0;
+
+CREATE TABLE t0 (c0 Tuple(a Int32, b Nullable(Int32)), c1 Int32) ENGINE = CoalescingMergeTree() ORDER BY c1;
+INSERT INTO t0 VALUES ((1,2), 0); -- return this one because tuple has not a nullable type so we can not aggregate by tuple columns
+INSERT INTO t0 VALUES ((3,4), 0);
+SELECT c0 FROM t0 FINAL;
+
+DROP TABLE IF EXISTS t0;
+
+CREATE TABLE t0 (c0 Array(Nullable(Int32)), c1 Int32) ENGINE = SummingMergeTree() ORDER BY c1;
+INSERT INTO t0 VALUES ([1,2], 0);
+INSERT INTO t0 VALUES ([3,4], 0);
+SELECT c0 FROM t0 FINAL;
+
+
+DROP TABLE IF EXISTS t0;
+
+CREATE TABLE t0 (c0 Array(Nullable(Int32)), c1 Int32) ENGINE = CoalescingMergeTree() ORDER BY c1;
+INSERT INTO t0 VALUES ([1,2], 0); -- return this one because array has not a nullable type so we can not aggregate by tuple columns
+INSERT INTO t0 VALUES ([3,4], 0);
+SELECT c0 FROM t0 FINAL;
+
+DROP TABLE IF EXISTS t0;
+
+CREATE TABLE t0 (c0 Tuple) ENGINE = CoalescingMergeTree() ORDER BY tuple();
+INSERT INTO t0 (c0) VALUES (());
+SELECT c0 FROM t0 FINAL;
+
+DROP TABLE IF EXISTS t0;
+
+CREATE TABLE t0 (c0 Tuple) ENGINE = SummingMergeTree() ORDER BY tuple();
+INSERT INTO t0 (c0) VALUES (());
+SELECT c0 FROM t0 FINAL;