How SummingMergeTree and AggregatingMergeTree Merge Rows

TL;DR: These engines do not aggregate at query time and they do not have a separate “aggregate” step. Their merge algorithm walks the sorted streams, and whenever the sorting key changes it finishes the current group: the summed columns are added up, the aggregate states are merged with the aggregate function’s own merge, and the group collapses to one row.

A MergeTree table is a stream of immutable parts, each sorted by the sorting key. The natural way to build a “pre-aggregated” table on top of it is to run a GROUP BY in a periodic job and write the result into another table. That works, but it doubles the storage, it decouples the result from the source, and the result is only as fresh as the last job run.

SummingMergeTree and AggregatingMergeTree take a different route: the merge itself does the aggregation. Because the parts are sorted by the same key, all the rows that belong together are adjacent in the merge’s input, so the merge can collapse them on the fly — no hash table, no second table, no job. The result is a part that is already aggregated, and SELECT becomes a plain scan.

The interesting part is what “collapse” means for two very different kinds of columns: a number that can simply be added, and an aggregate function state — an opaque blob that only the aggregate function itself knows how to combine.

The engine arguments decide what is summed

SummingMergeTree takes an optional list of columns to sum. It is the last engine argument, and the parser recognises it by the fact that it is not a literal (a literal would be the index_granularity, or a replica name):

cpp
// src/Storages/MergeTree/registerStorageMergeTree.cpp:651
    else if (merging_params.mode == MergeTreeData::MergingParams::Summing || merging_params.mode == MergeTreeData::MergingParams::Coalescing)
    {
        /// If the last element is not index_granularity or replica_name (a literal), then this is a list of summable columns.
        if (arg_cnt && !engine_args[arg_cnt - 1]->as<ASTLiteral>())
        {
            merging_params.columns_to_sum = extractColumnNames(engine_args[arg_cnt - 1]);
            --arg_cnt;
        }
    }

So ENGINE = SummingMergeTree(d, (k1, k2), 8192, (a, b)) sums a and b, and ENGINE = SummingMergeTree() sums every summable column that is not part of the sorting key.

The list is a filter, not a declaration of what to do. The actual decision — which columns are numeric enough to be summed, which are part of the key, which are AggregateFunction columns — is made by the merge algorithm, once, when the merge starts, by walking the header:

cpp
// src/Processors/Merges/Algorithms/SummingSortedAlgorithm.cpp:289
    /** Fill in the column numbers, which must be summed.
        * This can only be numeric columns that are not part of the sort key.
        * If a non-empty column_names_to_sum is specified, then we only take these columns.
        * Some columns from column_names_to_sum may not be found. This is ignored.
        */
    for (size_t i = 0; i < num_columns; ++i)
    {

“Summable” is a property of the type, and it is deliberately narrow:

cpp
// src/DataTypes/IDataType.h:299
    /** Values of data type can be summed (possibly with overflow, within the same data type).
      * Example: numbers, even nullable. Not Date/DateTime. Not Enum.
      * Enums can be passed to aggregate function 'sum', but the result is Int64, not Enum, so they are not summable.
      */
    virtual bool isSummable() const { return false; }

Only DataTypeNumber and Decimal say yes; Date, DateTime, DateTime64, Enum and String all say no. That is not an accident — summing a DateTime would produce a number that is not a date, and summing an Enum would produce an Int64 that is not an Enum, so the merge could not write the result back into the column without changing the schema.

The per-column decision, in the merge algorithm:

cpp
// src/Processors/Merges/Algorithms/SummingSortedAlgorithm.cpp:410
        {
            bool is_agg_func = WhichDataType(column.type).isAggregateFunction();

            /// There are special const columns for example after prewhere sections
            /// or when skip index expressions produce constants.
            if (isColumnConst(*column.column))
            {
                def.column_numbers_not_to_aggregate.push_back(i);
                continue;
            }

            if (!aggregate_all_columns)
            {
                if (!column.type->isSummable() && !is_agg_func && !simple)
                {
                    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;
            }

            /// Are they inside the sorting key or partition key? Check both to ignore columns with order expression.
            if (isInSortingKey(description, column.name)
                || isColumnOrAncestorInNames(i, header_flatten, flatten_ancestors, partition_and_sorting_required_columns))
            {
                def.column_numbers_not_to_aggregate.push_back(i);
                continue;
            }

            if (column_names_to_sum.empty()
                || isColumnOrAncestorInNames(i, header_flatten, flatten_ancestors, column_names_to_sum))
            {
                // Create aggregator to sum this column

Read top to bottom, the rules are: a constant column is never summed; a column whose type is not summable is never summed (unless it is an AggregateFunction column or a SimpleAggregateFunction); a column that is part of the sorting key or the partition key is never summed, because the merge would destroy the ordering it is relying on; and the engine’s columns list, if given, filters what is left.

Everything that is not summed goes into column_numbers_not_to_aggregate, and for those columns the merge keeps the first value of the group. That is the “any” semantics of a dimension column: SummingMergeTree collapses the rows but keeps one representative value of everything that is not a measure.

Walking the group

The merge algorithm is a k-way merge over the sorted inputs, and the group boundary is simply “the sorting key changed”. The whole algorithm is about starting and finishing groups:

cpp
// src/Processors/Merges/Algorithms/SummingSortedAlgorithm.cpp:1027
        {
            detail::RowRef current_key;
            setRowRef(current_key, current);

            key_differs = last_key.empty() || rowsHaveDifferentSortColumns(last_key, current_key);

            last_key = current_key;
            last_chunk_sort_columns.clear();
        }

        if (key_differs)
        {
            if (merged_data.isGroupStarted())
                /// Write the data for the previous group.
                merged_data.finishGroup();

            if (merged_data.hasEnoughRows())
            {
                /// The block is now full and the last row is calculated completely.
                last_key.reset();
                return Status(merged_data.pull());
            }

            merged_data.startGroup(current->all_columns, current->getRow());
        }
        else
            merged_data.addRow(current->all_columns, current->getRow());

startGroup creates one aggregate state per summed column; addRow feeds every row of the group into those states; finishGroup writes the results out. No rows are buffered beyond the current group, which is why a SummingMergeTree merge is O(1) in memory regardless of how many rows it collapses.

Summing: the state is the sum

For a plain numeric column, the “aggregate state” is the sumWithOverflow aggregate function’s state, which is a running total. The merge creates one such function per column at start-up:

cpp
// src/Processors/Merges/Algorithms/SummingSortedAlgorithm.cpp:466
                else if (!is_agg_func)
                {
                    desc.init(sum_function_name.c_str(), {column.type});
                }

where sum_function_name is "sumWithOverflow" — with overflow, because the result has to fit back into the column’s own type, and a merge must not fail because a sum exceeded UInt64. (An ordinary sum would promote to Int64 or Decimal128 and change the column’s type, which a merge cannot do.)

When the group finishes, the state is finalised into the output column, and — here is the subtle part — the row is rolled back if every summed value came out zero:

cpp
// src/Processors/Merges/Algorithms/SummingSortedAlgorithm.cpp:815
void SummingSortedAlgorithm::SummingMergedData::finishGroup()
{
    is_group_started = false;

    /// We have nothing to aggregate. It means that it could be non-zero, because we have columns_not_to_aggregate.
    if (def.columns_to_aggregate.empty())
        current_row_is_zero = false;

    for (auto & desc : def.columns_to_aggregate)
    {
        // Do not insert if the aggregation state hasn't been created
        if (desc.created)
        {
            if (desc.is_agg_func_type)
            {
                current_row_is_zero = false;
            }
            else
            {
                try
                {
                    desc.function->insertResultInto(desc.state.data(), *desc.merged_column, def.arena.get());
                    /// Update zero status of current row
                    if (!desc.is_simple_agg_func_type && desc.column_numbers.size() == 1 && desc.remove_default_values)
                    {
                        // Flag row as non-empty if at least one column number if non-zero
                        current_row_is_zero = current_row_is_zero
                                            && desc.merged_column->isDefaultAt(desc.merged_column->size() - 1);
                    }
                    else
                    {
                        /// It is sumMapWithOverflow aggregate function.
                        /// Assume that the row isn't empty in this case
                        ///   (just because it is compatible with previous version)
                        current_row_is_zero = false;
                    }
                }
                catch (...)
                {
                    desc.destroyState();
                    throw;
                }
            }
            desc.destroyState();
        }
        else
            desc.merged_column->insertDefault();
    }

    /// If it is "zero" row, then rollback the insertion
    /// (at this moment we need rollback only cols from columns_to_aggregate)
    if (current_row_is_zero)
    {
        for (auto & desc : def.columns_to_aggregate)
            desc.merged_column->popBack(1);

        return;
    }

The popBack(1) is the whole trick: the value was inserted into the output column and then taken back out, so a group whose measures all summed to zero produces no row at all. This is what makes a SummingMergeTree table shrink over time instead of accumulating zero rows — a common pattern for “insert an initial value, then a correction” workloads.

Note the asymmetry: a group is dropped only if every summed column is zero. A row with hits = 0 but bytes = 1024 survives. And a group that has columns which are not summed is never dropped, because those columns are dimensions whose value might be meaningful.

The same rule is not applied by SELECT ... FINAL, and the header says why:

cpp
// src/Processors/Merges/SummingSortedTransform.h:29
        /// Whether to remove the rows in which all summed columns are zero after summing.
        /// Merges do it, but `SELECT ... FINAL` does not: a query may read only a subset of
        /// the summed columns, so it cannot decide whether a row is zero the way a merge does.
        bool remove_zero_rows = true

This is a real semantic difference between “what the merge would produce” and “what FINAL produces”, and it is a deliberate one: FINAL may project away the column that would have told it the row was zero.

Aggregating: merging opaque states

AggregatingMergeTree is the same idea for arbitrary aggregate functions. The column’s type is AggregateFunction(sum, UInt64) and its value on disk is a serialised aggregate state. The merge cannot add those states up itself — only the aggregate function knows how to combine two of them — so the merge asks the function:

cpp
// src/Processors/Merges/Algorithms/AggregatingSortedAlgorithm.cpp:227
void AggregatingSortedAlgorithm::AggregatingMergedData::addRow(SortCursor & cursor)
{
    if (!is_group_started)
        throw Exception(ErrorCodes::LOGICAL_ERROR, "Can't add a row to the group because it was not started.");

    for (auto & desc : def.columns_to_aggregate)
        desc.column->insertMergeFrom(*cursor->all_columns[desc.column_number], cursor->getRow());

    for (auto & desc : def.columns_to_simple_aggregate)
    {
        auto & col = cursor->all_columns[desc.column_number];
        desc.add_function(desc.function.get(), desc.state.data(), &col, cursor->getRow(), def.arena.get());
    }
}

insertMergeFrom is the whole of it, and it ends up in the column:

cpp
// src/Columns/ColumnAggregateFunction.cpp:602
void ColumnAggregateFunction::insertMergeFrom(ConstAggregateDataPtr place)
{
    func->merge(data.back(), place, &createOrGetArena());
}

void ColumnAggregateFunction::insertMergeFrom(const IColumn & from, size_t n)
{
    insertMergeFrom(assert_cast<const ColumnAggregateFunction &>(from).data[n]);
}

func->merge(...) is the aggregate function’s own combine operator. For sum it adds the two totals; for uniq it merges the hash sets; for topK it merges the heaps. The merge algorithm does not know or care which — and that is the point: the state is opaque, and the state knows how to combine itself.

The two flavours of aggregate column are handled differently:

AggregateFunction(...) columns keep their state on disk. The merge starts the group by inserting a default (empty) state and merging every row’s state into it:

cpp
// src/Processors/Merges/Algorithms/AggregatingSortedAlgorithm.cpp:181
void AggregatingSortedAlgorithm::AggregatingMergedData::startGroup(const ColumnRawPtrs & raw_columns, size_t row)
{
    /// We will write the data for the group. We copy the values of ordinary columns.
    for (auto column_number : def.column_numbers_not_to_aggregate)
        columns[column_number]->insertFrom(*raw_columns[column_number], row);

    /// Add the empty aggregation state to the aggregate columns. The state will be updated in the `addRow` function.
    for (auto & column_to_aggregate : def.columns_to_aggregate)
        column_to_aggregate.column->insertDefault();

SimpleAggregateFunction(...) columns keep only the final value of the aggregate, not a state. The merge keeps a real state in memory during the group and writes the result out at the end — which is why the file format of a SimpleAggregateFunction column is just the underlying type.

The `sumMap` special case

SummingMergeTree has one more trick: a Nested map column whose name ends in Map is summed by key, not as a whole. The key and the value columns of the map are discovered by name convention — the first column, or one ending in ID, Key or Type, is the key; the rest are values:

cpp
// src/Processors/Merges/Algorithms/SummingSortedAlgorithm.cpp:510
        column_num_it = map.second.begin();
        for (; column_num_it != map.second.end(); ++column_num_it)
        {
            const ColumnWithTypeAndName & key_col = header_flatten.safeGetByPosition(*column_num_it);
            const String & name = key_col.name;
            const DataTypePtr nested_type_holder = recursiveRemoveLowCardinality(assert_cast<const DataTypeArray &>(*key_col.type).getNestedType());
            const IDataType & nested_type = *nested_type_holder;

            if (column_num_it == map.second.begin()
                || endsWith(name, "ID")
                || endsWith(name, "Key")
                || endsWith(name, "Type"))
            {
                if (!nested_type.isValueRepresentedByInteger() &&
                    !isStringOrFixedString(nested_type) &&
                    !typeid_cast<const DataTypeIPv6 *>(&nested_type) &&
                    !typeid_cast<const DataTypeUUID *>(&nested_type))
                        break;

                map_desc.key_col_nums.push_back(*column_num_it);
            }
            else
            {
                if (!nested_type.isSummable())
                    break;

                map_desc.val_col_nums.push_back(*column_num_it);
            }

With exactly one key column, the whole map is handed to sumMapWithOverflow, a single aggregate function that sums the values of matching keys. With a composite key, the engine falls back to the legacy mergeMap path, which merges the maps row by row in C++.

The reason a map needs its own treatment is that the keys are data, not structure: two rows with the same sorting key may have different map keys, and summing the two Array(Array(...)) columns element-wise would produce nonsense. A sumMap adds the values that share a key.

Choosing the algorithm

The engine mode selects the merge algorithm, and the mode comes from the engine name:

cpp
// src/Storages/MergeTree/MergeTask.cpp:3017
            case MergeTreeData::MergingParams::Summing:
                merged_transform = std::make_shared<SummingSortedTransform>(
                    header, input_streams_count, sort_description, merging_params.columns_to_sum, partition_and_sorting_required_columns, merge_block_size_rows, merge_block_size_bytes, max_dynamic_subcolumns, merging_params.allow_tuple_element_aggregation);
                break;

            case MergeTreeData::MergingParams::Aggregating:
                merged_transform = std::make_shared<AggregatingSortedTransform>(header, input_streams_count, sort_description, merge_block_size_rows, merge_block_size_bytes, max_dynamic_subcolumns, merging_params.allow_tuple_element_aggregation);
                break;

Every other mode — Ordinary, Collapsing, Replacing, Graphite, VersionedCollapsing, Coalescing — has its own algorithm in the same switch. Collapsing and VersionedCollapsing use the sign-column trick (a +1 row and a -1 row cancel), Replacing keeps the row with the highest version, Coalescing keeps the last value of each column.

Trade-offs

What this design buys:

No extra storage and no extra job. The pre-aggregated data lives in the same table, in the same parts, and is refreshed by the merges that were going to happen anyway.
O(1) memory. Only the current group is buffered, and the group is collapsed as soon as the sorting key changes.
Correctness by construction for the state case. The aggregate function’s own merge is used, so uniq, topK, quantiles and friends behave exactly as they would in a GROUP BY.
The engine’s columns list is a filter, not a contract. The merge refuses to sum a column that is not summable or that is part of the key, so a mistake in the DDL degrades to “not summed” rather than to a corrupt part.

What it costs:

The result depends on the merge. Until the parts are merged, the table still holds the un-aggregated rows, and a query sees a mix. SELECT ... FINAL forces the aggregation at read time — and then it must decide, on its own, which rows to drop.
FINAL and a merge do not agree on zero rows. A merge drops a group whose measures all sum to zero; FINAL does not, because it may not be reading the column that would have revealed the zero.
The set of summed columns is decided by types. A DateTime or an Enum measure is silently not summed, which surprises people who expected “all numeric columns”.
sumWithOverflow means overflow is silent. The sum is computed in the column’s own type, so a counter that exceeds UInt64 wraps rather than failing. That is the price of not changing the column’s type.
Dimension columns keep the first value of the group, not the last and not an aggregate. For a dimension that varies inside a group, the value in the merged row is arbitrary.