dbms: added setting 'select_sequential_consistency' [#METR-16779].

dbms/include/DB/Core/ErrorCodes.h

@@ -291,6 +291,7 @@ namespace ErrorCodes
 		UNSATISFIED_QUORUM_FOR_PREVIOUS_WRITE = 286,
 		UNKNOWN_FORMAT_VERSION = 287,
 		DISTRIBUTED_IN_JOIN_SUBQUERY_DENIED = 288,
+		REPLICA_IS_NOT_IN_QUORUM = 289,
 
 		KEEPER_EXCEPTION = 999,
 		POCO_EXCEPTION = 1000,

dbms/include/DB/Interpreters/Settings.h

@@ -163,6 +163,9 @@ struct Settings
 	\
 	/** Для запросов INSERT в реплицируемую таблицу, ждать записи на указанное число реплик и лианеризовать добавление данных. 0 - отключено. */ \
 	M(SettingUInt64, insert_quorum, 0) \
+	/** Для запросов SELECT из реплицируемой таблицы, кидать исключение, если на реплике нет куска, записанного с кворумом; \
+	  * не читать куски, которые ещё не были записаны с кворумом. */ \
+	M(SettingUInt64, select_sequential_consistency, 0) \
 
 	/// Всевозможные ограничения на выполнение запроса.
 	Limits limits;

dbms/include/DB/Storages/MergeTree/MergeTreeDataSelectExecutor.h

@@ -18,6 +18,7 @@ public:
 
 	/** При чтении, выбирается набор кусков, покрывающий нужный диапазон индекса.
 	  * Если inout_part_index != nullptr, из этого счетчика берутся значения для виртуального столбца _part_index.
+	  * max_block_number_to_read - если не ноль - не читать все куски, у которых правая граница больше этого порога.
 	  */
 	BlockInputStreams read(
 		const Names & column_names,
@@ -25,9 +26,10 @@ public:
 		const Context & context,
 		const Settings & settings,
 		QueryProcessingStage::Enum & processed_stage,
-		size_t max_block_size = DEFAULT_BLOCK_SIZE,
-		unsigned threads = 1,
-		size_t * inout_part_index = nullptr);
+		size_t max_block_size,
+		unsigned threads,
+		size_t * inout_part_index,
+		Int64 max_block_number_to_read);
 
 private:
 	MergeTreeData & data;

dbms/src/Storages/MergeTree/MergeTreeDataSelectExecutor.cpp

@@ -44,7 +44,8 @@ BlockInputStreams MergeTreeDataSelectExecutor::read(
 	QueryProcessingStage::Enum & processed_stage,
 	const size_t max_block_size,
 	const unsigned threads,
-	size_t * part_index)
+	size_t * part_index,
+	Int64 max_block_number_to_read)
 {
 	size_t part_index_var = 0;
 	if (!part_index)
@@ -82,7 +83,8 @@ BlockInputStreams MergeTreeDataSelectExecutor::read(
 	if (settings.force_index_by_date && date_condition.alwaysUnknown())
 		throw Exception("Index by date is not used and setting 'force_index_by_date' is set.", ErrorCodes::INDEX_NOT_USED);
 
-	/// Выберем куски, в которых могут быть данные, удовлетворяющие date_condition, и которые подходят под условие на _part.
+	/// Выберем куски, в которых могут быть данные, удовлетворяющие date_condition, и которые подходят под условие на _part,
+	///  а также max_block_number_to_read.
 	{
 		auto prev_parts = parts;
 		parts.clear();
@@ -98,6 +100,9 @@ BlockInputStreams MergeTreeDataSelectExecutor::read(
 			if (!date_condition.mayBeTrueInRange(&left, &right))
 				continue;
 
+			if (part->right > max_block_number_to_read)
+				continue;
+
 			parts.push_back(part);
 		}
 	}

dbms/src/Storages/StorageMergeTree.cpp

@@ -125,7 +125,7 @@ BlockInputStreams StorageMergeTree::read(
 		if (select.where_expression && !select.prewhere_expression)
 			MergeTreeWhereOptimizer{select, data, column_names, log};
 
-	return reader.read(column_names, query, context, settings, processed_stage, max_block_size, threads);
+	return reader.read(column_names, query, context, settings, processed_stage, max_block_size, threads, nullptr, 0);
 }
 
 BlockOutputStreamPtr StorageMergeTree::write(ASTPtr query, const Settings & settings)

dbms/src/Storages/StorageReplicatedMergeTree.cpp

@@ -2435,7 +2435,7 @@ BlockInputStreams StorageReplicatedMergeTree::read(
 	{
 		res = unreplicated_reader->read(real_column_names, query,
 										context, settings, processed_stage,
-										max_block_size, threads, &part_index);
+										max_block_size, threads, &part_index, 0);
 
 		for (auto & virtual_column : virt_column_names)
 		{
@@ -2449,7 +2449,44 @@ BlockInputStreams StorageReplicatedMergeTree::read(
 
 	if (values.count(1))
 	{
-		auto res2 = reader.read(real_column_names, query, context, settings, processed_stage, max_block_size, threads, &part_index);
+		/** Настройка select_sequential_consistency имеет два смысла:
+		  * 1. Кидать исключение, если на реплике есть не все куски, которые были записаны на кворум остальных реплик.
+		  * 2. Не читать куски, которые ещё не были записаны на кворум реплик.
+		  * Для этого приходится синхронно сходить в ZooKeeper.
+		  */
+		Int64 max_block_number_to_read = 0;
+		if (settings.select_sequential_consistency)
+		{
+			auto zookeeper = getZooKeeper();
+			String last_part;
+			zookeeper->tryGet(zookeeper_path + "/quorum/last_part", last_part);
+
+			if (!last_part.empty() && !data.getPartIfExists(last_part))	/// TODO Отключение реплики при распределённых запросах.
+				throw Exception("Replica doesn't have part " + last_part + " which was successfully written to quorum of other replicas."
+					" Send query to another replica or disable 'select_sequential_consistency' setting.", ErrorCodes::REPLICA_IS_NOT_IN_QUORUM);
+
+			if (last_part.empty())	/// Если ещё ни один кусок не был записан с кворумом.
+			{
+				String quorum_str;
+				if (zookeeper->tryGet(zookeeper_path + "/quorum/status", quorum_str))
+				{
+					ReplicatedMergeTreeQuorumEntry quorum_entry;
+					quorum_entry.fromString(quorum_str);
+					ActiveDataPartSet::Part part_info;
+					ActiveDataPartSet::parsePartName(quorum_entry.part_name, part_info);
+					max_block_number_to_read = part_info.left - 1;
+				}
+			}
+			else
+			{
+				ActiveDataPartSet::Part part_info;
+				ActiveDataPartSet::parsePartName(last_part, part_info);
+				max_block_number_to_read = part_info.right;
+			}
+		}
+
+		auto res2 = reader.read(
+			real_column_names, query, context, settings, processed_stage, max_block_size, threads, &part_index, max_block_number_to_read);
 
 		for (auto & virtual_column : virt_column_names)
 		{