Fix tests.

src/Processors/Merges/Algorithms/AggregatingSortedAlgorithm.cpp

@@ -305,6 +305,8 @@ IMergingAlgorithm::Status AggregatingSortedAlgorithm::merge()
 
         if (current->isLast() && skipLastRowFor(current->pos))
         {
+            /// If we skip this row, it's not equals with any key we process.
+            last_key.reset();
             /// Get the next block from the corresponding source, if there is one.
             queue.removeTop();
             return Status(current.impl->order);

src/Processors/Merges/Algorithms/IMergingAlgorithmWithDelayedChunk.cpp

@@ -33,15 +33,8 @@ void IMergingAlgorithmWithDelayedChunk::updateCursor(Input & input, size_t sourc
     auto & current_input = current_inputs[source_num];
 
     /// Extend lifetime of last chunk.
-    if (current_input.skip_last_row && current_input.chunk.getNumRows() <= 1)
+    last_chunk.swap(current_input.chunk);
-    {
+    last_chunk_sort_columns = std::move(cursors[source_num].sort_columns);
-        /// But if chunk has only single skipped row, ignore it.
-    }
-    else
-    {
-        last_chunk.swap(current_input.chunk);
-        last_chunk_sort_columns = std::move(cursors[source_num].sort_columns);
-    }
 
     current_input.swap(input);
     cursors[source_num].reset(current_input.chunk.getColumns(), {});

src/Processors/Merges/Algorithms/SummingSortedAlgorithm.cpp

@@ -649,6 +649,8 @@ IMergingAlgorithm::Status SummingSortedAlgorithm::merge()
 
         if (current->isLast() && skipLastRowFor(current->pos))
         {
+            /// If we skip this row, it's not equals with any key we process.
+            last_key.reset();
             /// Get the next block from the corresponding source, if there is one.
             queue.removeTop();
             return Status(current.impl->order);