Merge

dbms/include/DB/Storages/PkCondition.h

@@ -0,0 +1,280 @@
+#pragma once
+
+#include <sstream>
+
+#include <DB/Interpreters/Context.h>
+#include <DB/Interpreters/Expression.h>
+#include <DB/Storages/IStorage.h>
+#include <DB/Core/SortDescription.h>
+#include <DB/Parsers/ASTExpressionList.h>
+#include <DB/Parsers/ASTSelectQuery.h>
+#include <DB/Parsers/ASTFunction.h>
+#include <DB/Parsers/ASTLiteral.h>
+
+
+namespace DB
+{
+	
+/** Диапазон с открытыми или закрытыми концами; возможно, неограниченный.
+	*/
+struct Range
+{
+	Field left;				/// левая граница, если есть
+	Field right;			/// правая граница, если есть
+	bool left_bounded;		/// ограничен ли слева
+	bool right_bounded; 	/// ограничен ли справа
+	bool left_included; 	/// включает левую границу, если есть
+	bool right_included;	/// включает правую границу, если есть
+	
+	/// Всё множество.
+	Range() : left(), right(), left_bounded(false), right_bounded(false), left_included(false), right_included(false) {}
+	
+	/// Одна точка.
+	Range(const Field & point) : left(point), right(point), left_bounded(true), right_bounded(true), left_included(true), right_included(true) {}
+	
+	/// Ограниченный с двух сторон диапазон.
+	Range(const Field & left_, bool left_included_, const Field & right_, bool right_included_)
+		: left(left_), right(right_), left_bounded(true), right_bounded(true), left_included(left_included_), right_included(right_included_) {}
+	
+	static Range RightBounded(const Field & right_point, bool right_included)
+	{
+		Range r;
+		r.right = right_point;
+		r.right_bounded = true;
+		r.right_included = right_included;
+		return r;
+	}
+	
+	static Range LeftBounded(const Field & left_point, bool left_included)
+	{
+		Range r;
+		r.left= left_point;
+		r.left_bounded = true;
+		r.left_included = left_included;
+		return r;
+	}
+	
+	/// Установить левую границу.
+	void setLeft(const Field & point, bool included)
+	{
+		left = point;
+		left_bounded = true;
+		left_included = included;
+	}
+	
+	/// Установить правую границу.
+	void setRight(const Field & point, bool included)
+	{
+		right = point;
+		right_bounded = true;
+		right_included = included;
+	}
+	
+	/// x входит в range
+	bool contains(const Field & x)
+	{
+		return !leftThan(x) && !rightThan(x);
+	}
+	
+	/// x находится левее
+	bool rightThan(const Field & x)
+	{
+		return (left_bounded
+		? !(boost::apply_visitor(FieldVisitorGreater(), x, left) || (left_included && x == left))
+		: false);
+	}
+	
+	/// x находится правее
+	bool leftThan(const Field & x)
+	{
+		return (right_bounded
+		? !(boost::apply_visitor(FieldVisitorLess(), x, right) || (right_included && x == right))
+		: false);
+	}
+	
+	bool intersectsRange(const Range & r)
+	{
+		/// r левее меня.
+		if (r.right_bounded &&
+			left_bounded &&
+			(boost::apply_visitor(FieldVisitorLess(), r.right, left) ||
+			((!left_included || !r.right_included) &&
+			 r.right == left)))
+			return false;
+		/// r правее меня.
+		if (r.left_bounded &&
+			right_bounded &&
+			(boost::apply_visitor(FieldVisitorGreater(), r.left, right) ||
+			((!right_included || !r.left_included) &&
+			r.left== right)))
+			return false;
+		return true;
+	}
+	
+	bool containsRange(const Range & r)
+	{
+		/// r начинается левее меня.
+		if (left_bounded &&
+			(!r.left_bounded ||
+			boost::apply_visitor(FieldVisitorLess(), r.left, left) ||
+			(r.left_included &&
+			!left_included &&
+			r.left == left)))
+			return false;
+		/// r заканчивается правее меня.
+		if (right_bounded &&
+			(!r.right_bounded ||
+			boost::apply_visitor(FieldVisitorGreater(), r.right, right) ||
+			(r.right_included &&
+			!right_included &&
+			r.right == right)))
+			return false;
+		return true;
+	}
+	
+	void uniteWithRange(const Range & r)
+	{
+		left_bounded &= r.left_bounded;
+		if (left_bounded && boost::apply_visitor(FieldVisitorLess(), r.left, left) || (r.left_included && r.left == left))
+		{
+			left = r.left;
+			left_included = r.left_included;
+		}
+		right_bounded &= r.right_bounded;
+		if (right_bounded && boost::apply_visitor(FieldVisitorGreater(), r.right, right) || (r.right_included && r.right == right))
+		{
+			right = r.right;
+			right_included = r.right_included;
+		}
+	}
+	
+	void intersectWithRange(const Range & r)
+	{
+		if (!left_bounded)
+		{
+			left_bounded = r.left_bounded;
+			left = r.left;
+			left_included = r.left_included;
+		}
+		else if (r.left_bounded)
+		{
+			if (boost::apply_visitor(FieldVisitorGreater(), r.left, left) || (!r.left_included && r.left == left))
+			{
+				left = r.left;
+				left_included = r.left_included;
+			}
+		}
+		if (!right_bounded)
+		{
+			right_bounded = r.right_bounded;
+			right = r.right;
+			right_included = r.right_included;
+		}
+		else if (r.right_bounded)
+		{
+			if (boost::apply_visitor(FieldVisitorLess(), r.right, right) || (!r.right_included && r.right == right))
+			{
+				right = r.right;
+				right_included = r.right_included;
+			}
+		}
+	}
+	
+	String toString()
+	{
+		std::stringstream str;
+		
+		if (!left_bounded)
+			str << "(-inf, ";
+		else
+			str << (left_included ? '[' : '(') << boost::apply_visitor(FieldVisitorToString(), left) << ", ";
+		
+		if (!right_bounded)
+			str << "+inf)";
+		else
+			str << boost::apply_visitor(FieldVisitorToString(), right) << (right_included ? ']' : ')');
+		
+		return str.str();
+	}
+};
+
+
+class PkCondition
+{
+public:
+	PkCondition(ASTPtr query, const Context & context, const SortDescription & sort_descr);
+	
+	/// left_pk и right_pk должны содержать все поля из sort_descr в соответствующем порядке.
+	bool mayBeTrueInRange(const Row & left_pk, const Row & right_pk);
+	
+	bool alwaysTrue()
+	{
+		return rpn.size() == 1 && rpn[0].function == RPNElement::FUNCTION_UNKNOWN;
+	}
+	
+	String toString();
+private:
+	struct RPNElement
+	{
+		enum Function
+		{
+			/// Атомы логического выражения.
+			FUNCTION_IN_RANGE,
+			FUNCTION_NOT_IN_RANGE,
+			FUNCTION_UNKNOWN, /// Может принимать любое значение.
+			/// Операторы логического выражения.
+			FUNCTION_NOT,
+			FUNCTION_AND,
+			FUNCTION_OR,
+		};
+		
+		RPNElement() {}
+		RPNElement(Function function_) : function(function_) {}
+		RPNElement(Function function_, size_t key_column_) : function(function_), key_column(key_column_) {}
+		
+		String toString()
+		{
+			switch (function)
+			{
+				case FUNCTION_AND:
+					return "and";
+				case FUNCTION_OR:
+					return "or";
+				case FUNCTION_NOT:
+					return "not";
+				case FUNCTION_UNKNOWN:
+					return "unknown";
+				case FUNCTION_IN_RANGE:
+				case FUNCTION_NOT_IN_RANGE:
+				{
+					std::ostringstream ss;
+					ss << "(column " << key_column << (function == FUNCTION_NOT_IN_RANGE ? " not" : "") << " in " << range.toString() << ")";
+					return ss.str();
+				}
+				default:
+					return "ERROR";
+			}
+		}
+		
+		Function function;
+		
+		/// Для FUNCTION_IN_RANGE и FUNCTION_NOT_IN_RANGE.
+		Range range;
+		size_t key_column;
+	};
+	
+	typedef std::vector<RPNElement> RPN;
+	typedef std::map<String, size_t> ColumnIndices;
+	
+	void traverseAST(ASTPtr & node, Block & block_with_constants);
+	bool atomFromAST(ASTPtr & node, Block & block_with_constants, RPNElement & out);
+	bool operatorFromAST(ASTFunction * func, RPNElement & out);
+	
+	RPN rpn;
+	
+	Context context;
+	SortDescription sort_descr;
+	ColumnIndices pk_columns;
+};
+
+}

dbms/include/DB/Storages/StorageMergeTree.h

@@ -263,14 +263,6 @@ private:
 	/// Удалить неактуальные куски.
 	void clearOldParts();
 
-	/** Определить диапазоны индексов для запроса.
-	  * Возвращается:
-	  *  date_range - диапазон дат;
-	  *  primary_prefix - префикс первичного ключа, для которого требуется равенство;
-	  *  primary_range - диапазон значений следующего после префикса столбца первичного ключа.
-	  */
-	void getIndexRanges(ASTPtr & query, Range & date_range, Row & primary_prefix, Range & primary_range);
-
 	/// Определяет, какие куски нужно объединять, и запускает их слияние в отдельном потоке. Если iterations=0, объединяет, пока это возможно.
 	void merge(size_t iterations = 1, bool async = true);
 	/// Если while_can, объединяет в цикле, пока можно; иначе выбирает и объединяет только одну пару кусков.

dbms/src/Storages/PkCondition.cpp

@@ -0,0 +1,286 @@
+#include <DB/Storages/PkCondition.h>
+#include <DB/DataTypes/DataTypesNumberFixed.h>
+
+namespace DB
+{
+
+PkCondition::PkCondition(ASTPtr query, const Context & context_, const SortDescription & sort_descr_)
+	: context(context_), sort_descr(sort_descr_)
+{
+	for (size_t i = 0; i < sort_descr.size(); ++i)
+	{
+		pk_columns[sort_descr[i].column_name] = i;
+	}
+	
+	/** Вычисление выражений, зависящих только от констант.
+	 * Чтобы индекс мог использоваться, если написано, например WHERE Date = toDate(now()).
+	 */
+	Expression expr_for_constant_folding(query, context);
+	Block block_with_constants;
+	
+	/// В блоке должен быть хотя бы один столбец, чтобы у него было известно число строк.
+	ColumnWithNameAndType dummy_column;
+	dummy_column.name = "_dummy";
+	dummy_column.type = new DataTypeUInt8;
+	dummy_column.column = new ColumnConstUInt8(1, 0);
+	block_with_constants.insert(dummy_column);
+	
+	expr_for_constant_folding.execute(block_with_constants, 0, true);
+	
+	/// Преобразуем секцию WHERE в обратную польскую строку.
+	ASTSelectQuery & select = dynamic_cast<ASTSelectQuery &>(*query);
+	if (select.where_expression)
+	{
+		traverseAST(select.where_expression, block_with_constants);
+	}
+	else
+	{
+		rpn.push_back(RPNElement(RPNElement::FUNCTION_UNKNOWN));
+	}
+}
+
+/** Получить значение константного выражения.
+ * Вернуть false, если выражение не константно.
+ */
+static bool getConstant(ASTPtr & expr, Block & block_with_constants, Field & value)
+{
+	String column_name = expr->getColumnName();
+	
+	if (ASTLiteral * lit = dynamic_cast<ASTLiteral *>(&*expr))
+	{
+		/// литерал
+		value = lit->value;
+		return true;
+	}
+	else if (block_with_constants.has(column_name) && block_with_constants.getByName(column_name).column->isConst())
+	{
+		/// выражение, вычислившееся в константу
+		value = (*block_with_constants.getByName(column_name).column)[0];
+		return true;
+	}
+	else
+		return false;
+}
+
+void PkCondition::traverseAST(ASTPtr & node, Block & block_with_constants)
+{
+	RPNElement element;
+	
+	if (ASTFunction * func = dynamic_cast<ASTFunction *>(&*node))
+	{
+		if (operatorFromAST(func, element))
+		{
+			ASTs & args = dynamic_cast<ASTExpressionList &>(*func->arguments).children;
+			for (size_t i = 0; i < args.size(); ++i)
+			{
+				traverseAST(args[i], block_with_constants);
+			}
+			rpn.push_back(element);
+			
+			return;
+		}
+	}
+	
+	if (!atomFromAST(node, block_with_constants, element))
+	{
+		element.function = RPNElement::FUNCTION_UNKNOWN;
+	}
+	
+	rpn.push_back(element);
+}
+
+bool PkCondition::atomFromAST(ASTPtr & node, Block & block_with_constants, RPNElement & out)
+{
+	/// Фнукции < > = != <= >=, у которых один агрумент константа, другой - один из столбцов первичного ключа.
+	if (ASTFunction * func = dynamic_cast<ASTFunction *>(&*node))
+	{
+		ASTs & args = dynamic_cast<ASTExpressionList &>(*func->arguments).children;
+		
+		if (args.size() != 2)
+			return false;
+		
+		/// Если true, слева константа.
+		bool inverted;
+		size_t column;
+		Field value;
+		if (pk_columns.count(args[0]->getColumnName()) && getConstant(args[1], block_with_constants, value))
+		{
+			inverted = false;
+			column = pk_columns[args[0]->getColumnName()];
+		}
+		else if (pk_columns.count(args[1]->getColumnName()) && getConstant(args[0], block_with_constants, value))
+		{
+			inverted = true;
+			column = pk_columns[args[1]->getColumnName()];
+		}
+		else
+			return false;
+		
+		std::string func_name = func->name;
+		
+		/// Заменим <const> <sign> <column> на <column> <-sign> <const>
+		if (inverted)
+		{
+			if (func_name == "less")
+				func_name = "greater";
+			else if (func_name == "greater")
+				func_name = "less";
+			else if (func_name == "greaterOrEquals")
+				func_name = "lessOrEquals";
+			else if (func_name == "lessOrEquals")
+				func_name = "greaterOrEquals";
+		}
+		
+		out.function = RPNElement::FUNCTION_IN_RANGE;
+		out.key_column = column;
+		
+		if (func->name == "notEquals")
+		{
+			out.function = RPNElement::FUNCTION_NOT_IN_RANGE;
+			out.range = Range(value);
+		}
+		else if (func->name == "equals")
+			out.range = Range(value);
+		else if (func->name == "less")
+			out.range = Range::RightBounded(value, false);
+		else if (func->name == "greater")
+			out.range = Range::LeftBounded(value, false);
+		else if (func->name == "lessOrEquals")
+			out.range = Range::RightBounded(value, true);
+		else if (func->name == "greaterOrEquals")
+			out.range = Range::LeftBounded(value, true);
+		else
+			return false;
+		
+		return true;
+	}
+	
+	return false;
+}
+
+bool PkCondition::operatorFromAST(ASTFunction * func, RPNElement & out)
+{
+	/// Фнукции AND, OR, NOT.
+	ASTs & args = dynamic_cast<ASTExpressionList &>(*func->arguments).children;
+	
+	if (func->name == "not")
+	{
+		if (args.size() != 1)
+			return false;
+		
+		out.function = RPNElement::FUNCTION_NOT;
+	}
+	else
+	{
+		if (args.size() != 2)
+			return false;
+	
+		if (func->name == "and")
+			out.function = RPNElement::FUNCTION_AND;
+		else if (func->name == "or")
+			out.function = RPNElement::FUNCTION_OR;
+		else
+			return false;
+	}
+	
+	return true;
+}
+
+String PkCondition::toString()
+{
+	String res;
+	for (size_t i = 0; i < rpn.size(); ++i)
+	{
+		if (i)
+			res += ", ";
+		res += rpn[i].toString();
+	}
+	return res;
+}
+
+/// Множество значений булевой переменной. То есть два булевых значения: может ли быть true, может ли быть false.
+struct BoolMask
+{
+	bool can_be_true;
+	bool can_be_false;
+	
+	BoolMask() {}
+	BoolMask(bool can_be_true_, bool can_be_false_) : can_be_true(can_be_true_), can_be_false(can_be_false_) {}
+	
+	BoolMask operator &(const BoolMask & m)
+	{
+		return BoolMask(can_be_true && m.can_be_true, can_be_false || m.can_be_false);
+	}
+	BoolMask operator |(const BoolMask & m)
+	{
+		return BoolMask(can_be_true || m.can_be_true, can_be_false && m.can_be_false);
+	}
+	BoolMask operator !()
+	{
+		return BoolMask(can_be_false, can_be_true);
+	}
+};
+
+bool PkCondition::mayBeTrueInRange(const Row & left_pk, const Row & right_pk)
+{
+	/// Найдем диапазоны элементов ключа.
+	std::vector<Range> key_ranges(sort_descr.size(), Range());
+	for (size_t i = 0; i < sort_descr.size(); ++i)
+	{
+		if (left_pk[i] == right_pk[i])
+		{
+			key_ranges[i] = Range(left_pk[i]);
+		}
+		else
+		{
+			key_ranges[i] = Range(left_pk[i], true, right_pk[i], true);
+			break;
+		}
+	}
+	
+	std::vector<BoolMask> rpn_stack;
+	for (size_t i = 0; i < rpn.size(); ++i)
+	{
+		RPNElement & element = rpn[i];
+		if (element.function == RPNElement::FUNCTION_UNKNOWN)
+		{
+			rpn_stack.push_back(BoolMask(true, true));
+		}
+		else if (element.function == RPNElement::FUNCTION_NOT_IN_RANGE || element.function == RPNElement::FUNCTION_IN_RANGE)
+		{
+			Range & key_range = key_ranges[element.key_column];
+			bool intersects = element.range.intersectsRange(key_range);
+			bool contains = element.range.containsRange(key_range);
+			rpn_stack.push_back(BoolMask(intersects, !contains));
+			if (element.function == RPNElement::FUNCTION_NOT_IN_RANGE)
+				rpn_stack.back() = !rpn_stack.back();
+		}
+		else if (element.function == RPNElement::FUNCTION_NOT)
+		{
+			rpn_stack.back() = !rpn_stack.back();
+		}
+		else if (element.function == RPNElement::FUNCTION_AND)
+		{
+			BoolMask arg1 = rpn_stack.back();
+			rpn_stack.pop_back();
+			BoolMask arg2 = rpn_stack.back();
+			rpn_stack.back() = arg1 & arg2;
+		}
+		else if (element.function == RPNElement::FUNCTION_OR)
+		{
+			BoolMask arg1 = rpn_stack.back();
+			rpn_stack.pop_back();
+			BoolMask arg2 = rpn_stack.back();
+			rpn_stack.back() = arg1 | arg2;
+		}
+		else
+			throw Exception("Unexpected function type in PkCondition::RPNElement", ErrorCodes::LOGICAL_ERROR);
+	}
+	
+	if (rpn_stack.size() != 1)
+		throw Exception("Unexpected stack size in PkCondition::mayBeTrueInRange", ErrorCodes::LOGICAL_ERROR);
+	
+	return rpn_stack[0].can_be_true;
+}
+
+}

dbms/src/Storages/StorageMergeTree.cpp

@@ -38,6 +38,7 @@
 #include <DB/Interpreters/sortBlock.h>
 
 #include <DB/Storages/StorageMergeTree.h>
+#include <DB/Storages/PkCondition.h>
 
 
 #define MERGE_TREE_MARK_SIZE (2 * sizeof(size_t))
@@ -453,93 +454,6 @@ private:
 typedef Poco::SharedPtr<MergedBlockOutputStream> MergedBlockOutputStreamPtr;
 
 
-/** Диапазон с открытыми или закрытыми концами; возможно, неограниченный.
-  * Определяет, какую часть данных читать, при наличии индекса.
-  */
-struct Range
-{
-	Field left;				/// левая граница, если есть
-	Field right;			/// правая граница, если есть
-	bool left_bounded;		/// ограничен ли слева
-	bool right_bounded; 	/// ограничен ли справа
-	bool left_included; 	/// включает левую границу, если есть
-	bool right_included;	/// включает правую границу, если есть
-
-	/// Всё множество.
-	Range() : left(), right(), left_bounded(false), right_bounded(false), left_included(false), right_included(false) {}
-
-	/// Одна точка.
-	Range(const Field & point) : left(point), right(point), left_bounded(true), right_bounded(true), left_included(true), right_included(true) {}
-
-	/// Установить левую границу.
-	void setLeft(const Field & point, bool included)
-	{
-		left = point;
-		left_bounded = true;
-		left_included = included;
-	}
-
-	/// Установить правую границу.
-	void setRight(const Field & point, bool included)
-	{
-		right = point;
-		right_bounded = true;
-		right_included = included;
-	}
-
-	/// x входит в range
-	bool contains(const Field & x)
-	{
-		return !leftThan(x) && !rightThan(x);
-	}
-
-	/// x находится левее
-	bool rightThan(const Field & x)
-	{
-		return (left_bounded
-			? !(boost::apply_visitor(FieldVisitorGreater(), x, left) || (left_included && x == left))
-			: false);
-	}
-
-	/// x находится правее
-	bool leftThan(const Field & x)
-	{
-		return (right_bounded
-			? !(boost::apply_visitor(FieldVisitorLess(), x, right) || (right_included && x == right))
-			: false);
-	}
-
-	/// Пересекает отрезок
-	bool intersectsSegment(const Field & segment_left, const Field & segment_right)
-	{
-		if (!left_bounded)
-			return contains(segment_left);
-		if (!right_bounded)
-			return contains(segment_right);
-
-		return (boost::apply_visitor(FieldVisitorLess(), segment_left, right) || (right_included && segment_left == right))
-			&& (boost::apply_visitor(FieldVisitorGreater(), segment_right, left) || (left_included && segment_right == left));
-	}
-
-	String toString()
-	{
-		std::stringstream str;
-
-		if (!left_bounded)
-			str << "(-inf, ";
-		else
-			str << (left_included ? '[' : '(') << boost::apply_visitor(FieldVisitorToString(), left) << ", ";
-
-		if (!right_bounded)
-			str << "+inf)";
-		else
-			str << boost::apply_visitor(FieldVisitorToString(), right) << (right_included ? ']' : ')');
-
-		return str.str();
-	}
-};
-
-
 /// Для чтения из одного куска. Для чтения сразу из многих, Storage использует сразу много таких объектов.
 class MergeTreeBlockInputStream : public IProfilingBlockInputStream
 {
@@ -571,18 +485,18 @@ public:
 	}
 	
 	/// Получает диапазон засечек, вне которого не могут находиться ключи из заданного диапазона.
+	/// Сейчас может работать некорректно, это промежуточная версия для тестирования.
 	static void markRangeFromPkRange(const String & path,
 									  size_t marks_count,
 									  StorageMergeTree & storage,
-								      Row & requested_pk_prefix,
-								      Range & requested_pk_range,
+								      PkCondition & key_condition,
 									  size_t & out_first_mark,
 									  size_t & out_last_mark)
 	{
 		size_t last_mark_in_file = (marks_count == 0) ? 0 : (marks_count - 1);
 		
 		/// Если индекс не используется.
-		if (requested_pk_prefix.size() == 0 && !requested_pk_range.left_bounded && !requested_pk_range.right_bounded)
+		if (key_condition.alwaysTrue())
 		{
 			out_first_mark = 0;
 			out_last_mark = last_mark_in_file;
@@ -597,10 +511,8 @@ public:
 			String index_path = path + "primary.idx";
 			ReadBufferFromFile index(index_path, std::min(static_cast<size_t>(DBMS_DEFAULT_BUFFER_SIZE), Poco::File(index_path).getSize()));
 			
-			size_t prefix_size = requested_pk_prefix.size();
-			Row pk(storage.sort_descr.size());
-			Row pk_prefix(prefix_size);
-			
+			ssize_t last_read_mark;
+			Row prev_pk;
 			for (size_t current_mark_number = 0; !index.eof(); ++current_mark_number)
 			{
 				/// Читаем очередное значение PK
@@ -608,31 +520,23 @@ public:
 				for (size_t i = 0, size = pk.size(); i < size; ++i)
 					storage.primary_key_sample.getByPosition(i).type->deserializeBinary(pk[i], index);
 				
-				pk_prefix.assign(pk.begin(), pk.begin() + pk_prefix.size());
+				if (current_mark_number > 0)
+				{
+					if (key_condition.mayBeTrueInRange(prev_pk, pk))
+					{
+						if (min_mark_number == -1)
+							min_mark_number = current_mark_number - 1;
+						max_mark_number = current_mark_number - 1;
+					}
+				}
 				
-				if (pk_prefix < requested_pk_prefix)
-				{
-					min_mark_number = current_mark_number;
-				}
-				else if (pk_prefix == requested_pk_prefix)
-				{
-					if (requested_pk_range.rightThan(pk[prefix_size]))
-					{
-						min_mark_number = current_mark_number;
-					}
-					else if (requested_pk_range.leftThan(pk[prefix_size]))
-					{
-						max_mark_number = current_mark_number == 0 ? 0 : (current_mark_number - 1);
-						break;
-					}
-				}
-				else
-				{
-					max_mark_number = current_mark_number == 0 ? 0 : (current_mark_number - 1);
-					break;
-				}
+				prev_pk = pk;
+				last_read_mark = current_mark_number;
 			}
 			
+			if (max_mark_number == last_read_mark - 1)
+				++max_mark_number;
+			
 			out_first_mark = min_mark_number == -1 ? 0 : min_mark_number;
 			out_last_mark = max_mark_number == -1 ? last_mark_in_file : max_mark_number;
 		}
@@ -866,203 +770,6 @@ BlockOutputStreamPtr StorageMergeTree::write(ASTPtr query)
 }
 
 
-/// Собирает список отношений в конъюнкции в секции WHERE для определения того, можно ли использовать индекс.
-static void getRelationsFromConjunction(ASTPtr & node, ASTs & relations)
-{
-	if (ASTFunction * func = dynamic_cast<ASTFunction *>(&*node))
-	{
-		if (func->name == "equals"
-			|| func->name == "less" || func->name == "greater"
-			|| func->name == "lessOrEquals" || func->name == "greaterOrEquals")
-		{
-			relations.push_back(node);
-		}
-		else if (func->name == "and")
-		{
-			/// Обходим рекурсивно.
-			ASTs & args = dynamic_cast<ASTExpressionList &>(*func->arguments).children;
-
-			getRelationsFromConjunction(args.at(0), relations);
-			getRelationsFromConjunction(args.at(1), relations);
-		}
-	}
-}
-
-
-/** Получить значение константного выражения.
-  * Вернуть false, если выражение не константно.
-  */
-static bool getConstant(ASTPtr & expr, Block & block_with_constants, Field & value)
-{
-	String column_name = expr->getColumnName();
-
-	if (ASTLiteral * lit = dynamic_cast<ASTLiteral *>(&*expr))
-	{
-		/// литерал
-		value = lit->value;
-		return true;
-	}
-	else if (block_with_constants.has(column_name) && block_with_constants.getByName(column_name).column->isConst())
-	{
-		/// выражение, вычислившееся в константу
-		value = (*block_with_constants.getByName(column_name).column)[0];
-		return true;
-	}
-	else
-		return false;
-}
-
-
-/** Получить значение константного аргумента функции вида f(name, const_expr) или f(const_expr, name).
-  * block_with_constants содержит вычисленные значения константных выражений.
-  * Вернуть false, если такого нет.
-  */
-static bool getConstantArgument(ASTs & args, Block & block_with_constants, Field & rhs)
-{
-	if (args.size() != 2)
-		return false;
-	
-	return getConstant(args[0], block_with_constants, rhs)
-		|| getConstant(args[1], block_with_constants, rhs);
-}
-
-
-/// Составить диапазон возможных значений для столбца на основе секции WHERE с вычисленными константными выражениями.
-static Range getRangeForColumn(ASTs & relations, const String & column_name, Block & block_with_constants)
-{
-	Range range;
-	
-	for (ASTs::iterator jt = relations.begin(); jt != relations.end(); ++jt)
-	{
-		ASTFunction * func = dynamic_cast<ASTFunction *>(&**jt);
-		if (!func)
-			continue;
-
-		ASTs & args = dynamic_cast<ASTExpressionList &>(*func->arguments).children;
-
-		if (args.size() != 2)
-			continue;
-
-		/// Шаблон: col rel const или const rel col
-		bool inverted;
-		if (column_name == args[0]->getColumnName())
-			inverted = false;
-		else if (column_name == args[1]->getColumnName())
-			inverted = true;
-		else
-			continue;
-
-		Field rhs;
-		if (!getConstantArgument(args, block_with_constants, rhs))
-			continue;
-		
-		if (func->name == "equals")
-		{
-			range = Range(rhs);
-			break;
-		}
-		else if (func->name == "greater")
-			!inverted ? range.setLeft(rhs, false) : range.setRight(rhs, false);
-		else if (func->name == "greaterOrEquals")
-			!inverted ? range.setLeft(rhs, true) : range.setRight(rhs, true);
-		else if (func->name == "less")
-			!inverted ? range.setRight(rhs, false) : range.setLeft(rhs, false);
-		else if (func->name == "lessOrEquals")
-			!inverted ? range.setRight(rhs, true) : range.setLeft(rhs, true);
-	}
-
-	return range;
-}
-
-
-/** Выделяет значение, которому должен быть равен столбец на основе секции WHERE с вычисленными константными выражениями.
-  * Если такого нет - возвращает false.
-  */
-static bool getEqualityForColumn(ASTs & relations, const String & column_name, Block & block_with_constants, Field & value)
-{
-	for (ASTs::iterator jt = relations.begin(); jt != relations.end(); ++jt)
-	{
-		ASTFunction * func = dynamic_cast<ASTFunction *>(&**jt);
-		if (!func || func->name != "equals")
-			continue;
-
-		ASTs & args = dynamic_cast<ASTExpressionList &>(*func->arguments).children;
-
-		if (args.size() != 2)
-			continue;
-
-		if (args[0]->getColumnName() != column_name && args[1]->getColumnName() != column_name)
-			continue;
-
-		if (getConstantArgument(args, block_with_constants, value))
-			return true;
-		else
-			continue;
-	}
-	
-	return false;
-}
-
-
-void StorageMergeTree::getIndexRanges(ASTPtr & query, Range & date_range, Row & primary_prefix, Range & primary_range)
-{
-	/** Вычисление выражений, зависящих только от констант.
-	  * Чтобы индекс мог использоваться, если написано, например WHERE Date = toDate(now()).
-	  */
-	Expression expr_for_constant_folding(query, context);
-	Block block_with_constants;
-
-	/// В блоке должен быть хотя бы один столбец, чтобы у него было известно число строк.
-	ColumnWithNameAndType dummy_column;
-	dummy_column.name = "_dummy";
-	dummy_column.type = new DataTypeUInt8;
-	dummy_column.column = new ColumnConstUInt8(1, 0);
-	block_with_constants.insert(dummy_column);
-	
-	expr_for_constant_folding.execute(block_with_constants, 0, true);
-
-	/// Выделяем из конъюнкции в секции WHERE все отношения.
-	ASTSelectQuery & select = dynamic_cast<ASTSelectQuery &>(*query);
-	if (select.where_expression)
-	{
-		ASTs relations;
-		getRelationsFromConjunction(select.where_expression, relations);
-
-		/// Ищем отношения, которые могут быть использованы для индекса по дате.
-		date_range = getRangeForColumn(relations, date_column_name, block_with_constants);
-
-		/** Теперь ищем отношения, которые могут быть использованы для первичного ключа.
-		  * Сначала находим максимальное количество отношений равенства константе для первых столбцов PK.
-		  */
-		for (SortDescription::const_iterator it = sort_descr.begin(); it != sort_descr.end(); ++it)
-		{
-			Field rhs;
-			if (getEqualityForColumn(relations, it->column_name, block_with_constants, rhs))
-				primary_prefix.push_back(rhs);
-			else
-				break;
-		}
-
-		/// Если не для всех столбцов PK записано равенство, то ищем отношения для следующего столбца PK.
-		if (primary_prefix.size() < sort_descr.size())
-			primary_range = getRangeForColumn(relations, sort_descr[primary_prefix.size()].column_name, block_with_constants);
-	}
-
-	LOG_DEBUG(log, "Date range: " << date_range.toString());
-
-	std::stringstream primary_prefix_str;
-	for (Row::const_iterator it = primary_prefix.begin(); it != primary_prefix.end(); ++it)
-		primary_prefix_str << (it != primary_prefix.begin() ? ", " : "") << boost::apply_visitor(FieldVisitorToString(), *it);
-
-	LOG_DEBUG(log, "Primary key prefix: (" << primary_prefix_str.str() << ")");
-
-	if (primary_prefix.size() < sort_descr.size())
-	{
-		LOG_DEBUG(log, "Primary key range for column " << sort_descr[primary_prefix.size()].column_name << ": " << primary_range.toString());
-	}
-}
-
-
 BlockInputStreams StorageMergeTree::read(
 	const Names & column_names,
 	ASTPtr query,
@@ -1070,24 +777,21 @@ BlockInputStreams StorageMergeTree::read(
 	size_t max_block_size,
 	unsigned threads)
 {
-	/// Диапазон дат.
-	Range date_range;
-	/// Префикс первичного ключа, для которого требуется равенство. Может быть пустым.
-	Row primary_prefix;
-	/// Диапазон следующего после префикса столбца первичного ключа.
-	Range primary_range;
-
-	getIndexRanges(query, date_range, primary_prefix, primary_range);
+	PkCondition key_condition(query, context, sort_descr);
+	PkCondition date_condition(query, context, SortDescription(1, SortColumnDescription(date_column_name, 1)));
 
+	LOG_DEBUG(log, "key condition: " << key_condition.toString());
+	LOG_DEBUG(log, "date condition: " << date_condition.toString());
+	
 	typedef std::vector<DataPartRange> PartsRanges;
 	PartsRanges parts;
 	
-	/// Выберем куски, в которых могут быть данные для date_range.
+	/// Выберем куски, в которых могут быть данные, удовлетворяющие key_condition.
 	{
 		Poco::ScopedLock<Poco::FastMutex> lock(data_parts_mutex);
 		
 		for (DataParts::iterator it = data_parts.begin(); it != data_parts.end(); ++it)
-			if (date_range.intersectsSegment(static_cast<UInt64>((*it)->left_date), static_cast<UInt64>((*it)->right_date)))
+			if (date_condition.mayBeTrueInRange(Row(static_cast<UInt64>((*it)->left_date)),Row(static_cast<UInt64>((*it)->right_date))))
 				parts.push_back(DataPartRange(*it, 0, 0));
 	}
 	
@@ -1099,8 +803,7 @@ BlockInputStreams StorageMergeTree::read(
 		MergeTreeBlockInputStream::markRangeFromPkRange(full_path + part.data_part->name + '/',
 		                                                part.data_part->size,
 												         *this,
-		                                                primary_prefix,
-		                                                primary_range,
+		                                                key_condition,
 		                                                part.first_mark,
 		                                                part.last_mark);
 		sum_marks += part.last_mark - part.first_mark + 1;