From 4eb7ba0e60a686fccacb3aff12d0a1fe7ce083eb Mon Sep 17 00:00:00 2001
From: Alexey Milovidov <milovidov@yandex-team.ru>
Date: Fri, 29 Nov 2013 22:10:15 +0000
Subject: [PATCH] dbms: more clear code [#METR-2944].

---
 dbms/include/DB/Interpreters/sortBlock.h      |  8 +++++
 .../MergeTree/MergeTreeBlockOutputStream.h    |  2 +-
 dbms/src/Interpreters/sortBlock.cpp           | 31 ++++++++++++++++++-
 3 files changed, 39 insertions(+), 2 deletions(-)

diff --git a/dbms/include/DB/Interpreters/sortBlock.h b/dbms/include/DB/Interpreters/sortBlock.h
index 6dcdac97eb5..e012b218aa0 100644
--- a/dbms/include/DB/Interpreters/sortBlock.h
+++ b/dbms/include/DB/Interpreters/sortBlock.h
@@ -10,4 +10,12 @@ namespace DB
 /// Отсортировать один блок по описанию desc. Если limit != 0, то производится partial sort первых limit строк.
 void sortBlock(Block & block, const SortDescription & description, size_t limit = 0);
 
+
+/** Используется только в StorageMergeTree для сортировки данных при INSERT-е.
+  * Сортировка стабильная. Это важно для сохранения порядка записей в движке CollapsingMergeTree
+  *  - так как на основе порядка записей определяется, удалять ли или оставлять группы строчек при коллапсировании.
+  * Не поддерживаются collations. Не поддерживается частичная сортировка.
+  */ 
+void stableSortBlock(Block & block, const SortDescription & description);
+
 }
diff --git a/dbms/include/DB/Storages/MergeTree/MergeTreeBlockOutputStream.h b/dbms/include/DB/Storages/MergeTree/MergeTreeBlockOutputStream.h
index 9905a8dd852..c1b5c5c60c0 100644
--- a/dbms/include/DB/Storages/MergeTree/MergeTreeBlockOutputStream.h
+++ b/dbms/include/DB/Storages/MergeTree/MergeTreeBlockOutputStream.h
@@ -122,7 +122,7 @@ private:
 		LOG_TRACE(storage.log, "Sorting by primary key.");
 		
 		/// Сортируем.
-		sortBlock(block, storage.sort_descr);
+		stableSortBlock(block, storage.sort_descr);
 		
 		/// Наконец-то можно писать данные на диск.
 		LOG_TRACE(storage.log, "Writing index.");
diff --git a/dbms/src/Interpreters/sortBlock.cpp b/dbms/src/Interpreters/sortBlock.cpp
index a6b40681465..8e21fefad3e 100644
--- a/dbms/src/Interpreters/sortBlock.cpp
+++ b/dbms/src/Interpreters/sortBlock.cpp
@@ -131,7 +131,7 @@ void sortBlock(Block & block, const SortDescription & description, size_t limit)
 			if (limit)
 				std::partial_sort(perm.begin(), perm.begin() + limit, perm.end(), less);
 			else
-				std::stable_sort(perm.begin(), perm.end(), less);
+				std::sort(perm.begin(), perm.end(), less);
 		}
 
 		size_t columns = block.columns();
@@ -140,4 +140,33 @@ void sortBlock(Block & block, const SortDescription & description, size_t limit)
 	}
 }
 
+
+void stableSortBlock(Block & block, const SortDescription & description)
+{
+	if (!block)
+		return;
+
+	size_t size = block.rows();
+	IColumn::Permutation perm(size);
+	for (size_t i = 0; i < size; ++i)
+		perm[i] = i;
+
+	ColumnsWithSortDescriptions columns_with_sort_desc;
+
+	for (size_t i = 0, size = description.size(); i < size; ++i)
+	{
+		IColumn * column = !description[i].column_name.empty()
+			? block.getByName(description[i].column_name).column
+			: block.getByPosition(description[i].column_number).column;
+
+		columns_with_sort_desc.push_back(std::make_pair(column, description[i]));
+	}
+
+	std::stable_sort(perm.begin(), perm.end(), PartialSortingLess(columns_with_sort_desc));
+
+	size_t columns = block.columns();
+	for (size_t i = 0; i < columns; ++i)
+		block.getByPosition(i).column = block.getByPosition(i).column->permute(perm, 0);
+}
+
 }