clickhouse: optimized FINAL [#CONV-7363].

dbms/include/DB/DataStreams/CollapsingFinalBlockInputStream.h

@@ -226,7 +226,7 @@ private:
 	
 	Queue queue;
 	
-	Cursor current;			/// Текущий первичный ключ.
+	Cursor previous;
 	
 	Cursor first_negative;	/// Первая отрицательная строка для текущего первичного ключа.
 	Cursor last_positive;	/// Последняя положительная строка для текущего первичного ключа.

dbms/src/DataStreams/CollapsingFinalBlockInputStream.cpp

@@ -7,7 +7,7 @@ namespace DB
 CollapsingFinalBlockInputStream::~CollapsingFinalBlockInputStream()
 {
 	/// Нужно обезвредить все MergingBlockPtr, чтобы они не пытались класть блоки в output_blocks.
-	current.block.cancel();
+	previous.block.cancel();
 	first_negative.block.cancel();
 	last_positive.block.cancel();
 	
@@ -63,7 +63,7 @@ void CollapsingFinalBlockInputStream::commitCurrent()
 		
 		first_negative = Cursor();
 		last_positive = Cursor();
-		current = Cursor();
+		previous = Cursor();
 	}
 	
 	count_negative = 0;
@@ -85,44 +85,59 @@ Block CollapsingFinalBlockInputStream::readImpl()
 	{
 		while (!queue.empty() && output_blocks.empty())
 		{
-			Cursor next = queue.top();
+			Cursor current = queue.top();
 			queue.pop();
 			
-			if (!next.equal(current))
+			bool has_next = !queue.empty();
+			Cursor next = has_next ? queue.top() : Cursor();
+			
+			/// Будем продвигаться в текущем блоке, не используя очередь, пока возможно.
+			while (true)
+			{
+				if (!current.equal(previous))
 				{
 					commitCurrent();
-				current = next;
+					previous = current;
 				}
 				
-			Int8 sign = next.getSign();
+				Int8 sign = current.getSign();
 				if (sign == 1)
 				{
-				last_positive = next;
+					last_positive = current;
 					++count_positive;
 				}
 				else if (sign == -1)
 				{
 					if (!count_negative)
-					first_negative = next;
+						first_negative = current;
 					++count_negative;
 				}
 				else
 					reportBadSign(sign);
 				
-			if (next.isLast())
+				if (current.isLast())
 				{
-				fetchNextBlock(next.block->stream_index);
+					fetchNextBlock(current.block->stream_index);
 					
 					/// Все потоки кончились. Обработаем последний ключ.
-				if (queue.empty())
+					if (!has_next)
 					{
 						commitCurrent();
 					}
+					
+					break;
 				}
 				else
 				{
-				next.next();
-				queue.push(next);
+					current.next();
+					
+					if (has_next && !(next < current))
+					{
+						queue.push(current);
+						
+						break;
+					}
+				}
 			}
 		}