SummingMergeTree FINAL and merges: copy rows of parts without duplicate keys in whole batches, 53-68% faster FINAL on tables with new parts

SummingSortedAlgorithm::merge handles one row per loop iteration. When the key changes, finishGroup writes the sums of the previous group, checks whether they are all zero and writes the copied columns back; in merges, which set a byte limit, hasEnoughRows adds up the byte size of every column; startGroup copies the columns that are not summed into Fields (one-row columns for Float types), creates the sum states and adds the row. Since https://github.com/ClickHouse/ClickHouse/pull/58142, rowsHaveDifferentSortColumns treats two rows of a part with a non-zero level as different keys without comparing them: every merged part, and every part that INSERT writes with optimize_on_insert, has one row per key.

In a table that receives new rows for existing keys, most rows come from one large part in long runs between the keys of the small parts, and the batch queue from https://github.com/ClickHouse/ClickHouse/pull/115869 already finds these runs. All rows of a batch come from one source and are less than the current rows of the other sources; only the last row may equal the next row of a later source. So every row strictly inside a batch from a part with a non-zero level is a group of its own: the current code already starts and finishes a group for each of them.

The patch:

1. When every summed column is a number summed by sumWithOverflow (no AggregateFunction, SimpleAggregateFunction or ...Map columns, not CoalescingMergeTree), a group of one row results in that row: the sum of one value x is 0 + x, which is x except that −0.0 becomes +0.0. When the current batch comes from a part with a non-zero level and has at least 8 rows, takeRowsOfOwnGroups finishes the current group and copies the rows inside the batch with insertRangeFrom for every column, then adds 0 to Float sums. In merges, which remove rows whose sums are all zero, it stops before such a row, which then takes the normal path. It keeps the max_block_size and max_block_size_bytes limits with rowsToInsertBeforeFlush. The last row of the batch also takes the normal path, and so do chunks with replicated or dynamic-structure columns.
2. For a single-column key the queue currently does not detect batches because the comparison is cheap (IMergingAlgorithmWithDelayedChunk::initializeQueue). The patch enables the detection when a sample merge of the first rows of the current chunks (at most 1024, up to the first chunk end) puts at least half of them into such batches. The choice is made once per merge, so merges whose first rows interleave keep today's per-row path.
3. The merge loop becomes a template with two instantiations, with the batch path when the detection is on and some source comes from a part with a non-zero level, and without it otherwise, so other merges run the loop as before. The batch path is a separate non-inlined function.

https://github.com/ClickHouse/ClickHouse/issues/70667 shows that the one-row-per-key assumption fails after a MODIFY ORDER BY that removes a key column; it can also fail for a part adopted by ATTACH PART or RESTORE, which keeps a non-zero level today (https://github.com/ClickHouse/ClickHouse/pull/109686). The patch does not extend the assumption: it copies only rows that the current code already treats as groups of one row, and the shortcut, the sample and the batch path now share one condition, sourceHasUniqueKeys, so a fix there applies to all of them.

Correctness: 328 result digests are byte-identical between master and the patch: SELECT … FINAL with max_threads 1/4 and max_block_size 1000/65536 (order-insensitive, ORDER BY key and filtered), and OPTIMIZE FINAL of a copy of each table with default limits, merge_max_block_size = 1000, merge_max_block_size_bytes = 65536 and both (content, physical order, and the merged part's rows, marks, uncompressed and compressed bytes). The 13 tables cover a single UInt64 key with Int64/Float64 sums, new rows that cancel existing ones to zero, NaN, −0.0 and all-zero rows written into the merged part by mutations; a (sid, Date) key with String, Nullable, LowCardinality and Array columns; an explicit list of columns to sum; Float32, Decimal(18, 4), Int128 and UInt8 sums; a ...Map Nested column, a SimpleAggregateFunction(max) and an AggregateFunction(uniq) column (batch path off); unmerged level-0 duplicates; two parts with alternating keys; a DateTime64 key; and a CoalescingMergeTree table. The 223 stateless tests whose names match summing, coalescing, aggregating, simple_agg or final give identical output on both builds, apart from one whose output also varies between runs of master; 39 of them exit non-zero, identically on both builds.

Peak memory per query stayed within −1% … +6% of master.

Master 7a987ba2469 (2026-11-12), the base of the patch; the changed files and the batch queue in src/Core/SortCursor.h have not changed in master since (checked at 7d943bb5224b, 2026-11-13). Release build with clang-21 without ThinLTO, on AWS Graviton3 (Neoverse V1, 4 vCPU). Not measured on x86.

CREATE TABLE sum1 (k UInt64, v1 UInt64, v2 Float64) ENGINE = SummingMergeTree ORDER BY k
    SETTINGS auto_statistics_types = '';
INSERT INTO sum1 SELECT intHash64(number), number % 1000 + 1, number / 7 FROM numbers(8000000);
OPTIMIZE TABLE sum1 FINAL;
ALTER TABLE sum1 MODIFY SETTING max_bytes_to_merge_at_max_space_in_pool = 1, max_bytes_to_merge_at_min_space_in_pool = 1;
-- 10 parts of new rows for existing keys: run once for each i = 1 … 10 (clickhouse-client --param_i=1, …, --param_i=10)
INSERT INTO sum1 SELECT intHash64(cityHash64(number, {i:UInt8}) % 8000000), 1, 0.5 FROM numbers(20000);

SELECT k, v1, v2 FROM sum1 FINAL FORMAT Null;
SELECT count(), sum(v1), sum(v2) FROM sum1 FINAL FORMAT Null;

-- merge: copy the 11 parts into an empty table and merge them
CREATE TABLE w (k UInt64, v1 UInt64, v2 Float64) ENGINE = SummingMergeTree ORDER BY k
    SETTINGS auto_statistics_types = '', max_bytes_to_merge_at_max_space_in_pool = 1, max_bytes_to_merge_at_min_space_in_pool = 1;
ALTER TABLE w ATTACH PARTITION tuple() FROM sum1;
OPTIMIZE TABLE w FINAL;

CREATE TABLE sum2 (sid UInt32, d Date, hits UInt64, bytes UInt64, cost Float64) ENGINE = SummingMergeTree ORDER BY (sid, d)
    SETTINGS auto_statistics_types = '';
INSERT INTO sum2 SELECT number % 200000, toDate('2026-10-01') + intDiv(number, 200000), number % 97 + 1, number % 100003, number / 13 FROM numbers(6000000);
OPTIMIZE TABLE sum2 FINAL;  -- then the same MODIFY SETTING as above, and for each i = 1 … 10:
INSERT INTO sum2 SELECT cityHash64(number, {i:UInt8}) % 200000, toDate('2026-10-30') + (number % 2), 1, 100, 0.25 FROM numbers(20000);

SELECT * FROM sum2 FINAL FORMAT Null;
SELECT d, sum(hits), sum(cost) FROM sum2 FINAL GROUP BY d FORMAT Null;

suml is sum2 with region LowCardinality(String) and page String after d: base rows concat('region-', toString(number % 23)) and concat('/page/', toString(number % 200000)), new rows the same functions of their sid. Controls: sumi is two tables of 4M rows with keys number * 2 and number * 2 + 1 (v1 = number % 1000 + 1, v2 = number / 7), each merged with OPTIMIZE FINAL and attached into one table with ATTACH PARTITION ... FROM; sum2i is the same with (sid, d) = (number % 200000, toDate('2026-10-01') + 2 * intDiv(number, 200000)) and the next day, 3M rows each. agg1 is AggregatingMergeTree ORDER BY k with countState(), sumState(number) and uniqState(toUInt64(number % 1000)) grouped by k = intHash64(number) over 4M rows, merged, plus 10 parts of 20k existing keys. The INSERT control writes 4M rows with columns sid UInt64, ts DateTime64(3), value Float64, ver UInt32 (sid = number % 20000) in one block into an empty SummingMergeTree ORDER BY (sid, ts). suml, sumi, sum2i, agg1, the INSERT table and every merge copy were created with SETTINGS auto_statistics_types = '' too. Point lookups are 200 queries SELECT * FROM sum1 FINAL WHERE k = intHash64(N * 39989 + 17), N = 0 … 199, in one client; their per-round time is the sum of the query durations.

Measurement IDs: arm-sum-sfin: sum1 FINAL; arm-sum-sfint: with max_threads = 1; arm-sum-sagg: count(), sum(v1), sum(v2) over FINAL; arm-sum-sc-c4q1, arm-sum-sc-c4q2, arm-sum-sc-c4q3, arm-sum-sc-c4q4: 4 concurrent sum1 FINAL queries; arm-sum-scold: sum1 FINAL with cold caches; arm-sum-smrg: OPTIMIZE of sum1; arm-sum-s2fin: sum2 FINAL; arm-sum-s2agg: sum2 FINAL GROUP BY d; arm-sum-s2mrg: OPTIMIZE of sum2; arm-sum-slfin: suml FINAL; arm-sum-slmrg: OPTIMIZE of suml; arm-sum-sptq, arm-sum-sptq1: point lookups with 4 threads / 1 thread; arm-sum-sifin, arm-sum-simrg: sumi FINAL / OPTIMIZE; arm-sum-s2ifin, arm-sum-s2imrg: sum2i FINAL / OPTIMIZE; arm-sum-sins: INSERT of 4M rows into a SummingMergeTree; arm-sum-amrg, arm-sum-afin: agg1 OPTIMIZE / FINAL GROUP BY.

Two servers (master and patched) on copies of the same data directory with physically identical parts, both pinned to CPUs 0-3, max_threads = 4 unless stated; 11 rounds with alternating server order, the first round is a warm-up; the metric is query_duration_ms from system.query_log; the 95% interval is a Student-t interval of the mean log ratio of the paired rounds. The cold-cache and concurrent runs come last in each round. Automatic statistics were off, so the OPTIMIZE and INSERT times exclude the column statistics that the master default auto_statistics_types = 'basic, uniq_v2' builds in SummingMergeTree merges and INSERTs. The patch does not change that work: with default statistics, the absolute merge saving should be about the same but the percentage smaller (not measured). FINAL builds no statistics and is unaffected.

A row that has no other row with its key should cost about as much as copying it, as in a FINAL over parts that do not overlap. With a few small parts of new rows next to one large part, FINAL should then cost little more than reading the parts.

• https://github.com/ClickHouse/ClickHouse/pull/58142 (rows of a part with a non-zero level are taken to have different keys)
• https://github.com/ClickHouse/ClickHouse/issues/70667 (that assumption fails after MODIFY ORDER BY removes a key column; the patch keeps one shared condition for it)
• https://github.com/ClickHouse/ClickHouse/pull/115869 (batch queue for the merging algorithms, which finds the runs used here)
• https://github.com/ClickHouse/ClickHouse/pull/115032 (FINAL keeps rows whose sums are all zero; merges remove them)
• https://github.com/ClickHouse/ClickHouse/pull/109686 (open: resets the level of parts adopted by ATTACH PART, and by RESTORE unless the table definition matched the backup)
• https://github.com/ClickHouse/ClickHouse/pull/98971 and https://github.com/ClickHouse/ClickHouse/pull/99041 (draft FINAL BY: they turn off the level shortcut only in rowsHaveDifferentSortColumns. With this patch the switch belongs in sourceHasUniqueKeys; otherwise the batch path would copy rows that share the FINAL BY key without summing them)

diff --git a/src/Processors/Merges/Algorithms/IMergingAlgorithmWithDelayedChunk.cpp b/src/Processors/Merges/Algorithms/IMergingAlgorithmWithDelayedChunk.cpp
index b48d3a16805c..940eb43d1ba5 100644
--- a/src/Processors/Merges/Algorithms/IMergingAlgorithmWithDelayedChunk.cpp
+++ b/src/Processors/Merges/Algorithms/IMergingAlgorithmWithDelayedChunk.cpp
@@ -37,9 +37,53 @@ void IMergingAlgorithmWithDelayedChunk::initializeQueue(Inputs inputs)
     /// A single-column key without a collation compares cheaply even through the generic
     /// `SortCursor`, so it keeps the heap container (see `sortDescriptionCompareIsExpensive`).
     /// The algorithms consume rows one by one, so the batches only pay off by saving
-    /// comparisons: with a cheap comparator the detection is skipped altogether.
+    /// comparisons: with a cheap comparator the detection is skipped, unless the algorithm
+    /// takes long batches of rows without duplicate keys in one go.
     bool compare_is_expensive = sortDescriptionCompareIsExpensive(description);
-    queue = SortingQueueForCursor<SortCursor, SortingQueueStrategy::Batch>(cursors, compare_is_expensive, compare_is_expensive);
+    batch_detection_enabled = compare_is_expensive || (uses_batches_of_unique_keys && sourcesHaveLongBatchesOfUniqueKeys());
+    queue = SortingQueueForCursor<SortCursor, SortingQueueStrategy::Batch>(cursors, compare_is_expensive, batch_detection_enabled);
+}
+
+bool IMergingAlgorithmWithDelayedChunk::sourcesHaveLongBatchesOfUniqueKeys() const
+{
+    /// Merge the beginning of the current chunks with the batch detection on, without taking any rows.
+    /// Interleaved keys (e.g. parts that each hold every key once, or two parts of a similar size) come
+    /// in batches of one or two rows, where the detection only costs. A large part merged with a few
+    /// small ones (e.g. a table with recent updates) comes in long batches, which save the work per row.
+    /// A small sample: it runs at the start of every merge, including those of a few granules for a point query.
+    static constexpr size_t max_rows_to_check = 1024;
+
+    bool has_source_with_unique_keys = false;
+    for (size_t source_num = 0; source_num < cursors.size(); ++source_num)
+    {
+        const auto & cursor = cursors[source_num];
+        if (!cursor.empty() && !cursor.permutation && sourceHasUniqueKeys(source_num))
+            has_source_with_unique_keys = true;
+    }
+    if (!has_source_with_unique_keys)
+        return false;
+
+    SortCursorImpls sample_cursors = cursors;
+    SortingQueueForCursor<SortCursor, SortingQueueStrategy::Batch> sample_queue(
+        sample_cursors, /*enable_array_container=*/ false, /*enable_batch_detection=*/ true);
+
+    size_t rows = 0;
+    size_t rows_in_long_batches = 0;
+    while (sample_queue.isValid() && rows < max_rows_to_check)
+    {
+        auto [cursor, batch_size] = sample_queue.current();
+        /// The next chunk of the source is not available here.
+        if ((*cursor)->isLast(batch_size))
+            break;
+
+        rows += batch_size;
+        if (batch_size >= min_batch_size_to_take && !(*cursor)->permutation && sourceHasUniqueKeys((*cursor)->order))
+            rows_in_long_batches += batch_size - 2;
+
+        sample_queue.next(batch_size);
+    }
+
+    return rows_in_long_batches * 2 >= rows && rows_in_long_batches > 0;
 }
 
 void IMergingAlgorithmWithDelayedChunk::updateCursor(Input & input, size_t source_num)
diff --git a/src/Processors/Merges/Algorithms/IMergingAlgorithmWithDelayedChunk.h b/src/Processors/Merges/Algorithms/IMergingAlgorithmWithDelayedChunk.h
index 5b85e34e3b2c..657aeb9bbc74 100644
--- a/src/Processors/Merges/Algorithms/IMergingAlgorithmWithDelayedChunk.h
+++ b/src/Processors/Merges/Algorithms/IMergingAlgorithmWithDelayedChunk.h
@@ -37,7 +37,7 @@ protected:
     {
         /// By the time this method is called, `inputs_origin_merge_tree_part_level[lhs.source_stream_index]` must have been
         /// initialized in either `initializeQueue` or `updateCursor`
-        if (lhs.source_stream_index == rhs.source_stream_index && inputs_origin_merge_tree_part_level[lhs.source_stream_index] > 0)
+        if (lhs.source_stream_index == rhs.source_stream_index && sourceHasUniqueKeys(lhs.source_stream_index))
             return true;
 
         auto first_non_equal = lhs.firstNonEqualSortColumnsWith(prev_unequal_column, rhs);
@@ -53,7 +53,27 @@ protected:
 
     SharedHeader header;
 
+    /// Set by a derived algorithm before `initializeQueue` when it takes the rows inside a batch from a
+    /// source without duplicate keys (a part with a non-zero level, see `sourceHasUniqueKeys`)
+    /// in one go (see `SummingSortedAlgorithm`). The batch detection is then enabled also for cheap
+    /// comparators if such batches cover enough rows (see `sourcesHaveLongBatchesOfUniqueKeys`).
+    bool uses_batches_of_unique_keys = false;
+
+    /// Whether the queue detects batches longer than one row (decided in `initializeQueue`).
+    bool batch_detection_enabled = false;
+
+    /// The smallest batch from such a source that is taken in one go: shorter ones save too little.
+    static constexpr size_t min_batch_size_to_take = 8;
+
+    /// Whether the current chunk of the source comes from a part without duplicate keys: a part with a non-zero level.
+    /// Both `rowsHaveDifferentSortColumns` and the batches of unique keys rely on this one condition.
+    bool sourceHasUniqueKeys(size_t source_num) const { return inputs_origin_merge_tree_part_level[source_num] > 0; }
+
 private:
+    /// Whether most rows at the beginning of the merge come in batches of at least `min_batch_size_to_take`
+    /// rows from sources without duplicate keys.
+    bool sourcesHaveLongBatchesOfUniqueKeys() const;
+
     /// Inputs currently being merged.
     Inputs current_inputs;
     SortCursorImpls cursors;
diff --git a/src/Processors/Merges/Algorithms/SummingSortedAlgorithm.cpp b/src/Processors/Merges/Algorithms/SummingSortedAlgorithm.cpp
index 897113a284ab..dce27177de15 100644
--- a/src/Processors/Merges/Algorithms/SummingSortedAlgorithm.cpp
+++ b/src/Processors/Merges/Algorithms/SummingSortedAlgorithm.cpp
@@ -7,6 +7,7 @@
 #include <AggregateFunctions/AggregateFunctionFactory.h>
 #include <Columns/ColumnAggregateFunction.h>
 #include <Columns/ColumnTuple.h>
+#include <Columns/ColumnVector.h>
 #include <DataTypes/DataTypeArray.h>
 #include <DataTypes/DataTypeCustomSimpleAggregateFunction.h>
 #include <DataTypes/DataTypeIPv4andIPv6.h>
@@ -22,6 +23,8 @@
 #include <Common/Exception.h>
 #include <Common/FieldVisitorSum.h>
 #include <Common/StringUtils.h>
+#include <Common/assert_cast.h>
+#include <base/unaligned.h>
 
 namespace DB
 {
@@ -585,6 +588,32 @@ static SummingSortedAlgorithm::ColumnsDefinition defineColumns(
             def.columns_need_exact_copy[i] = true;
     }
 
+    /// A group of one row results in that row when every aggregated column is a number summed by
+    /// `sumWithOverflow`: the sum of one value `x` is `0 + x` (see `SummingMergedData::insertRowsOfOwnGroups`).
+    def.can_copy_rows_of_own_groups = true;
+    for (const auto & desc : def.columns_to_aggregate)
+    {
+        bool is_number = false;
+        if (desc.real_type)
+        {
+            WhichDataType which(desc.real_type);
+            is_number = which.isInt() || which.isUInt() || which.isFloat32() || which.isFloat64() || which.isDecimal();
+        }
+
+        if (!is_number || desc.is_agg_func_type || desc.is_simple_agg_func_type || desc.aggregate_all_columns || desc.nested_type
+            || desc.column_numbers.size() != 1 || !desc.function || desc.function->getName() != "sumWithOverflow"
+            || !desc.function->getResultType()->equals(*desc.real_type))
+        {
+            def.can_copy_rows_of_own_groups = false;
+            break;
+        }
+    }
+
+    /// See `finishGroup`: a row is removed only if each aggregated column is a single summed column with
+    /// `remove_default_values` and all of them are zero.
+    def.removes_zero_rows = def.can_copy_rows_of_own_groups && def.maps_to_sum.empty() && !def.columns_to_aggregate.empty()
+        && std::all_of(def.columns_to_aggregate.begin(), def.columns_to_aggregate.end(), [](const auto & desc) { return desc.remove_default_values; });
+
     return def;
 }
 
@@ -937,6 +966,120 @@ void SummingSortedAlgorithm::SummingMergedData::addRowImpl(ColumnRawPtrs & raw_c
     }
 }
 
+/// The sum of one value `x` by `sumWithOverflow` is `0 + x`: the same as `x`, except that -0.0 becomes +0.0
+/// (and a signaling NaN a quiet one).
+template <typename T>
+static void addZeroToValuesFrom(IColumn & column, size_t from)
+{
+    auto & data = assert_cast<ColumnVector<T> &>(column).getData();
+    for (size_t i = from; i < data.size(); ++i)
+        data[i] = T(0) + data[i];
+}
+
+template <typename T>
+static bool hasZeroBitsValue(const char * data, size_t length)
+{
+    bool has_zero = false;
+    for (size_t i = 0; i < length; ++i)
+        has_zero |= unalignedLoad<T>(data + i * sizeof(T)) == 0;
+    return has_zero;
+}
+
+/// Whether any of the `length` values of a column of numbers starting at `from` is the default one:
+/// for `ColumnVector` and `ColumnDecimal` (see their `isDefaultAt`) it is the value with all bits zero.
+static bool hasDefaultValueInRange(const IColumn & column, size_t from, size_t length)
+{
+    const size_t value_size = column.sizeOfValueIfFixed();
+    const char * data = column.getRawData().data() + from * value_size;
+    switch (value_size)
+    {
+        case 1: return hasZeroBitsValue<UInt8>(data, length);
+        case 2: return hasZeroBitsValue<UInt16>(data, length);
+        case 4: return hasZeroBitsValue<UInt32>(data, length);
+        case 8: return hasZeroBitsValue<UInt64>(data, length);
+        default:
+            for (size_t i = from; i < from + length; ++i)
+                if (column.isDefaultAt(i))
+                    return true;
+            return false;
+    }
+}
+
+size_t SummingSortedAlgorithm::SummingMergedData::insertRowsOfOwnGroups(const ColumnRawPtrs & raw_columns, size_t begin, size_t end)
+{
+    chassert(!is_group_started && def.can_copy_rows_of_own_groups);
+
+    size_t length = end - begin;
+    for (auto & desc : def.columns_to_aggregate)
+    {
+        const size_t old_size = desc.merged_column->size();
+        desc.merged_column->insertRangeFrom(*raw_columns[desc.column_numbers[0]], begin, length);
+
+        WhichDataType which(desc.real_type);
+        if (which.isFloat64())
+            addZeroToValuesFrom<Float64>(*desc.merged_column, old_size);
+        else if (which.isFloat32())
+            addZeroToValuesFrom<Float32>(*desc.merged_column, old_size);
+    }
+
+    if (def.removes_zero_rows)
+    {
+        /// A row is removed only if all its sums are zero (see `finishGroup`): usually some column has no zero at all.
+        bool every_column_has_zero = true;
+        for (const auto & desc : def.columns_to_aggregate)
+        {
+            if (!hasDefaultValueInRange(*desc.merged_column, desc.merged_column->size() - length, length))
+            {
+                every_column_has_zero = false;
+                break;
+            }
+        }
+
+        size_t rows_before_zero_row = length;
+        if (every_column_has_zero)
+        {
+            for (size_t i = 0; i < length; ++i)
+            {
+                bool is_zero = true;
+                for (const auto & desc : def.columns_to_aggregate)
+                {
+                    if (!desc.merged_column->isDefaultAt(desc.merged_column->size() - length + i))
+                    {
+                        is_zero = false;
+                        break;
+                    }
+                }
+
+                if (is_zero)
+                {
+                    rows_before_zero_row = i;
+                    break;
+                }
+            }
+        }
+
+        /// Stop before the zero row: it goes through `startGroup` and `finishGroup`, which remove it.
+        if (rows_before_zero_row < length)
+        {
+            for (auto & desc : def.columns_to_aggregate)
+                desc.merged_column->popBack(length - rows_before_zero_row);
+            length = rows_before_zero_row;
+        }
+    }
+
+    size_t next_column = columns.size() - def.column_numbers_not_to_aggregate.size();
+    for (auto column_number : def.column_numbers_not_to_aggregate)
+    {
+        if (length)
+            columns[next_column]->insertRangeFrom(*raw_columns[column_number], begin, length);
+        ++next_column;
+    }
+
+    total_merged_rows += length;
+    merged_rows += length;
+    return begin + length;
+}
+
 void SummingSortedAlgorithm::SummingMergedData::initAggregateDescription()
 {
     size_t num_columns = def.columns_to_aggregate.size();
@@ -975,6 +1118,9 @@ SummingSortedAlgorithm::SummingSortedAlgorithm(
     , merged_data(max_block_size_rows, max_block_size_bytes, max_dynamic_subcolumns_, columns_definition)
 {
     columns_definition.origin_header = header_;
+
+    /// The rows of a source without duplicate keys can be taken by whole batches (see `mergeImpl`).
+    uses_batches_of_unique_keys = columns_definition.can_copy_rows_of_own_groups;
 }
 
 void SummingSortedAlgorithm::initialize(Inputs inputs)
@@ -993,7 +1139,14 @@ void SummingSortedAlgorithm::initialize(Inputs inputs)
             preprocessChunk(input.chunk, columns_definition);
 
     merged_data.initialize(*header, inputs);
+    const size_t num_inputs = inputs.size();
     initializeQueue(std::move(inputs));
+
+    take_batches_of_unique_keys = false;
+    if (uses_batches_of_unique_keys && batch_detection_enabled)
+        for (size_t source_num = 0; source_num < num_inputs; ++source_num)
+            if (sourceHasUniqueKeys(source_num))
+                take_batches_of_unique_keys = true;
 }
 
 void SummingSortedAlgorithm::consume(Input & input, size_t source_num)
@@ -1006,14 +1159,56 @@ void SummingSortedAlgorithm::consume(Input & input, size_t source_num)
     updateCursor(input, source_num);
 }
 
+/// Out of line: it runs once per batch, and inlining it would make the loop over the other rows in `merge` slower.
+NO_INLINE bool SummingSortedAlgorithm::takeRowsOfOwnGroups(SortCursor current, size_t batch_size)
+{
+    const ColumnRawPtrs & columns = current->all_columns;
+    for (const auto * column : columns)
+        if (column->isReplicated() || column->hasDynamicStructure())
+            return false;
+
+    /// The group of the current row ends with it (see `mergeImpl`).
+    merged_data.finishGroup();
+
+    const size_t row = current->getRow();
+    const size_t begin = row + 1;
+    size_t end = row + batch_size - 1;
+
+    /// Keep the limits of the block size: write only the rows that fit before a flush. The row `end` then
+    /// starts its group as usual, and like for each row in `mergeImpl` a full block is emitted before that.
+    if (merged_data.hasEnoughRows())
+        end = begin;
+    else
+    {
+        const size_t max_rows = std::min<size_t>(end - begin, merged_data.maxBlockSize() - merged_data.mergedRows());
+        end = begin + merged_data.rowsToInsertBeforeFlush(columns, begin, max_rows, current->getSize());
+    }
+
+    if (begin < end)
+        end = merged_data.insertRowsOfOwnGroups(columns, begin, end);
+
+    /// `last_key` refers to the current row, so the row `end` from the same source is compared as a different key.
+    queue.next(end - row);
+    return true;
+}
+
 IMergingAlgorithm::Status SummingSortedAlgorithm::merge()
+{
+    if (take_batches_of_unique_keys)
+        return mergeImpl<true>();
+    return mergeImpl<false>();
+}
+
+template <bool take_batches>
+IMergingAlgorithm::Status SummingSortedAlgorithm::mergeImpl()
 {
     /// Take the rows in needed order and put them in `merged_columns` until rows no more than `max_block_size`
     while (queue.isValid())
     {
         bool key_differs = false;
 
-        SortCursor current = *queue.current().first;
+        const auto current_and_batch_size = queue.current();
+        SortCursor current = *current_and_batch_size.first;
 
         if (current->isLast() && skipLastRowFor(current->order))
         {
@@ -1052,6 +1247,19 @@ IMergingAlgorithm::Status SummingSortedAlgorithm::merge()
         else
             merged_data.addRow(current->all_columns, current->getRow());
 
+        /// A part with a non-zero level is taken to have no duplicate keys (see `sourceHasUniqueKeys`). All rows of one
+        /// batch come consecutively from the same cursor and are less than the current rows of the other cursors,
+        /// except that the last row of the batch may have the same key as the current row of a cursor of a later
+        /// source. So in a batch from such a part the group of the current row (which may have started in another
+        /// source) ends with it, and every row between it and the last row is a group of its own: take them in one
+        /// go instead of starting and finishing a group for each of them.
+        if constexpr (take_batches)
+        {
+            if (unlikely(current_and_batch_size.second >= min_batch_size_to_take) && !current->permutation
+                && sourceHasUniqueKeys(current->order) && takeRowsOfOwnGroups(current, current_and_batch_size.second))
+                continue;
+        }
+
         if (!current->isLast())
         {
             queue.next(1);
diff --git a/src/Processors/Merges/Algorithms/SummingSortedAlgorithm.h b/src/Processors/Merges/Algorithms/SummingSortedAlgorithm.h
index 372e75e94f53..a9daf18f9396 100644
--- a/src/Processors/Merges/Algorithms/SummingSortedAlgorithm.h
+++ b/src/Processors/Merges/Algorithms/SummingSortedAlgorithm.h
@@ -79,6 +79,12 @@ public:
         SharedHeader origin_header;
 
         bool allow_tuple_element_aggregation = false;
+
+        /// Whether the result of a group of one row is that row: all aggregated columns are summed numbers
+        /// (`0 + x` is `x`, except for -0.0), so such rows can be copied (see `insertRowsOfOwnGroups`).
+        bool can_copy_rows_of_own_groups = false;
+        /// Whether a row whose sums are all zero is removed (see `finishGroup`).
+        bool removes_zero_rows = false;
     };
 
     /// Specialization for SummingSortedTransform. Inserts only data for non-aggregated columns.
@@ -99,6 +105,11 @@ public:
         bool isGroupStarted() const { return is_group_started; }
         void addRow(ColumnRawPtrs & raw_columns, size_t row); /// Possible only when group was started.
 
+        /// Writes the rows [begin, end), each a group of its own, like `startGroup` and `finishGroup` for each of
+        /// them would (possible only when no group is started and `can_copy_rows_of_own_groups`). Stops before the first
+        /// row with only zero sums if such rows are removed. Returns the row it stopped at.
+        size_t insertRowsOfOwnGroups(const ColumnRawPtrs & raw_columns, size_t begin, size_t end);
+
         Chunk pull();
 
     private:
@@ -121,6 +132,16 @@ private:
     /// Order between members is important because merged_data has reference to columns_definition.
     ColumnsDefinition columns_definition;
     SummingMergedData merged_data;
+
+    /// Whether `merge` takes batches of rows from sources without duplicate keys (decided in `initialize`).
+    /// Without them the merge loop is instantiated without that check, to keep the per-row path as it is.
+    bool take_batches_of_unique_keys = false;
+    template <bool take_batches>
+    Status mergeImpl();
+
+    /// Finishes the group of the current row and writes the following rows of the batch, each a group
+    /// of its own, in one go (see `mergeImpl`). Returns false if it did nothing.
+    bool takeRowsOfOwnGroups(SortCursor current, size_t batch_size);
 };
 
 }

@groeneai could you take a look? If the analysis and patch hold up, feel free to open a PR if it fits.