dbms: development.

dbms/include/DB/Columns/ColumnArray.h

@@ -57,23 +57,26 @@ public:
 
 	void cut(size_t start, size_t length)
 	{
-		if (start + length > offsets.size())
+		if (length == 0 || start + length > offsets.size())
 			throw Exception("Parameter out of bound in IColumnArray::cut() method.",
 				ErrorCodes::PARAMETER_OUT_OF_BOUND);
 
+		size_t nested_offset = start == 0 ? 0 : offsets[start - 1];
+		size_t nested_length = offsets[start + length - 1] - nested_offset;
+
+		data->cut(nested_offset, nested_length);
+
 		if (start == 0)
 			offsets.resize(length);
 		else
 		{
 			Offsets_t tmp(length);
-			memcpy(&tmp[0], &offsets[start], length * sizeof(offsets[0]));
+
+			for (size_t i = 0; i < length; ++i)
+				tmp[i] = offsets[start + i] - nested_offset;
+			
 			tmp.swap(offsets);
 		}
-
-		size_t nested_offset = start == 0 ? 0 : offsets[start - 1];
-		size_t nested_length = offsets[start + length] - nested_offset;
-		
-		data->cut(nested_offset, nested_length);
 	}
 
 	void insert(const Field & x)

dbms/include/DB/Columns/ColumnConst.h

@@ -1,5 +1,4 @@
-#ifndef DBMS_CORE_COLUMN_CONST_H
-#define DBMS_CORE_COLUMN_CONST_H
+#pragma once
 
 #include <Poco/SharedPtr.h>
 
@@ -24,6 +23,7 @@ public:
 	
 	ColumnConst(size_t s_, const T & data_) : s(s_), data(data_) {}
 
+	bool isNumeric() const { return IsNumber<T>::value; }
 	ColumnPtr cloneEmpty() const { return new ColumnConst(0, data); }
 	size_t size() const { return s; }
 	Field operator[](size_t n) const { return typename NearestFieldType<T>::Type(data); }
@@ -48,6 +48,6 @@ private:
 };
 
 
-}
+typedef ColumnConst<String> ColumnConstString;
 
-#endif
+}

dbms/include/DB/Columns/ColumnFixedArray.h

@@ -82,6 +82,11 @@ public:
 		return *data;
 	}
 
+	size_t getN() const
+	{
+		return n;
+	}
+
 protected:
 	ColumnPtr data;
 	const size_t n;

dbms/include/DB/Columns/ColumnNullable.h

@@ -48,7 +48,7 @@ public:
 
 	void cut(size_t start, size_t length)
 	{
-		if (start + length > nulls.size())
+		if (length == 0 || start + length > nulls.size())
 			throw Exception("Parameter out of bound in IColumnNullable::cut() method.",
 				ErrorCodes::PARAMETER_OUT_OF_BOUND);
 

dbms/include/DB/Columns/ColumnVector.h

@@ -24,6 +24,8 @@ public:
 	ColumnVector() {}
 	ColumnVector(size_t n) : data(n) {}
 
+	bool isNumeric() const { return IsNumber<T>::value; }
+
 	ColumnPtr cloneEmpty() const
 	{
 		return new ColumnVector<T>;
@@ -41,8 +43,11 @@ public:
 	
 	void cut(size_t start, size_t length)
 	{
-		if (start + length > data.size())
-			throw Exception("Parameter out of bound in IColumnVector<T>::cut() method.",
+		if (length == 0 || start + length > data.size())
+			throw Exception("Parameters start = "
+				+ Poco::NumberFormatter::format(start) + ", length = "
+				+ Poco::NumberFormatter::format(length) + " are out of bound in IColumnVector<T>::cut() method"
+				" (data.size() = " + Poco::NumberFormatter::format(data.size()) + ").",
 				ErrorCodes::PARAMETER_OUT_OF_BOUND);
 
 		if (start == 0)

dbms/include/DB/Columns/IColumn.h

@@ -15,6 +15,10 @@ using Poco::SharedPtr;
 class IColumn
 {
 public:
+	/** Столбец представляет собой вектор чисел или числовую константу. 
+	  */
+	virtual bool isNumeric() const { return false; }
+	
 	/** Создать пустой столбец такого же типа */
 	virtual SharedPtr<IColumn> cloneEmpty() const = 0;
 

dbms/include/DB/Core/Field.h

@@ -139,20 +139,31 @@ public:
 
 template <typename T> struct NearestFieldType;
 
-template <> struct NearestFieldType<UInt8> { typedef UInt64 Type; };
-template <> struct NearestFieldType<UInt16> { typedef UInt64 Type; };
-template <> struct NearestFieldType<UInt32> { typedef UInt64 Type; };
-template <> struct NearestFieldType<UInt64> { typedef UInt64 Type; };
+template <> struct NearestFieldType<UInt8> 		{ typedef UInt64 	Type; };
+template <> struct NearestFieldType<UInt16> 	{ typedef UInt64 	Type; };
+template <> struct NearestFieldType<UInt32> 	{ typedef UInt64 	Type; };
+template <> struct NearestFieldType<UInt64> 	{ typedef UInt64 	Type; };
+template <> struct NearestFieldType<Int8> 		{ typedef Int64 	Type; };
+template <> struct NearestFieldType<Int16> 		{ typedef Int64 	Type; };
+template <> struct NearestFieldType<Int32> 		{ typedef Int64 	Type; };
+template <> struct NearestFieldType<Int64> 		{ typedef Int64 	Type; };
+template <> struct NearestFieldType<Float32> 	{ typedef Float64 	Type; };
+template <> struct NearestFieldType<Float64> 	{ typedef Float64 	Type; };
+template <> struct NearestFieldType<String> 	{ typedef String 	Type; };
 
-template <> struct NearestFieldType<Int8> { typedef Int64 Type; };
-template <> struct NearestFieldType<Int16> { typedef Int64 Type; };
-template <> struct NearestFieldType<Int32> { typedef Int64 Type; };
-template <> struct NearestFieldType<Int64> { typedef Int64 Type; };
 
-template <> struct NearestFieldType<Float32> { typedef Float64 Type; };
-template <> struct NearestFieldType<Float64> { typedef Float64 Type; };
+template <typename T> struct IsNumber 	{ static const bool value = false; };
 
-template <> struct NearestFieldType<String> { typedef String Type; };
+template <> struct IsNumber<UInt8> 		{ static const bool value = true; };
+template <> struct IsNumber<UInt16> 	{ static const bool value = true; };
+template <> struct IsNumber<UInt32> 	{ static const bool value = true; };
+template <> struct IsNumber<UInt64> 	{ static const bool value = true; };
+template <> struct IsNumber<Int8> 		{ static const bool value = true; };
+template <> struct IsNumber<Int16> 		{ static const bool value = true; };
+template <> struct IsNumber<Int32> 		{ static const bool value = true; };
+template <> struct IsNumber<Int64> 		{ static const bool value = true; };
+template <> struct IsNumber<Float32> 	{ static const bool value = true; };
+template <> struct IsNumber<Float64> 	{ static const bool value = true; };
 
 }
 

dbms/include/DB/Functions/FunctionsArithmetic.h

@@ -13,7 +13,6 @@ namespace DB
 
 /** Арифметические функции: +, -, *, /, %,
   * div (целочисленное деление),
-  * TODO: <<, >>, <<<, >>>, &, |, ^, ~
   */
 
 template<typename A, typename B>
@@ -185,26 +184,30 @@ struct ModuloImpl
 	{
 		size_t size = a.size();
 		for (size_t i = 0; i < size; ++i)
-			c[i] = a[i] % b[i];
+			c[i] = typename NumberTraits::ToInteger<A>::Type(a[i])
+				% typename NumberTraits::ToInteger<A>::Type(b[i]);
 	}
 
 	static void vector_constant(const std::vector<A> & a, B b, std::vector<ResultType> & c)
 	{
 		size_t size = a.size();
 		for (size_t i = 0; i < size; ++i)
-			c[i] = a[i] % b;
+			c[i] = typename NumberTraits::ToInteger<A>::Type(a[i])
+				% typename NumberTraits::ToInteger<A>::Type(b);
 	}
 
 	static void constant_vector(A a, const std::vector<B> & b, std::vector<ResultType> & c)
 	{
 		size_t size = b.size();
 		for (size_t i = 0; i < size; ++i)
-			c[i] = a % b[i];
+			c[i] = typename NumberTraits::ToInteger<A>::Type(a)
+				% typename NumberTraits::ToInteger<A>::Type(b[i]);
 	}
 
 	static void constant_constant(A a, B b, ResultType & c)
 	{
-		c = a % b;
+		c = typename NumberTraits::ToInteger<A>::Type(a)
+			% typename NumberTraits::ToInteger<A>::Type(b);
 	}
 };
 

dbms/include/DB/Functions/FunctionsComparison.h

@@ -2,12 +2,18 @@
 
 #include <Poco/NumberFormatter.h>
 
+#include <DB/Columns/ColumnsNumber.h>
+#include <DB/Columns/ColumnConst.h>
+#include <DB/Columns/ColumnString.h>
+#include <DB/Columns/ColumnFixedString.h>
+
 #include <DB/DataTypes/DataTypesNumberFixed.h>
 #include <DB/DataTypes/DataTypesNumberVariable.h>
 #include <DB/DataTypes/DataTypeDateTime.h>
 #include <DB/DataTypes/DataTypeDate.h>
 #include <DB/DataTypes/DataTypeString.h>
 #include <DB/DataTypes/DataTypeFixedString.h>
+
 #include <DB/Functions/IFunction.h>
 
 
@@ -25,42 +31,50 @@ namespace DB
   *   внутри каждой группы, но не из разных групп.
   */
 
+/** Игнорируем warning о сравнении signed и unsigned.
+  * (Результат может быть некорректным.)
+  */ 
+#pragma GCC diagnostic push
+#pragma GCC diagnostic ignored "-Wsign-compare"
+
 template<typename A, typename B>
 struct EqualsNumImpl
 {
-	typedef UInt8 ResultType;
-
-	static void vector_vector(const std::vector<A> & a, const std::vector<B> & b, std::vector<ResultType> & c)
+	static void vector_vector(const std::vector<A> & a, const std::vector<B> & b, std::vector<UInt8> & c)
 	{
 		size_t size = a.size();
 		for (size_t i = 0; i < size; ++i)
 			c[i] = a[i] == b[i];
 	}
 
-	static void vector_constant(const std::vector<A> & a, B b, std::vector<ResultType> & c)
+	static void vector_constant(const std::vector<A> & a, B b, std::vector<UInt8> & c)
 	{
 		size_t size = a.size();
 		for (size_t i = 0; i < size; ++i)
 			c[i] = a[i] == b;
 	}
 
-	static void constant_constant(A a, B b, ResultType & c)
+	static void constant_vector(A a, const std::vector<B> & b, std::vector<UInt8> & c)
+	{
+		size_t size = b.size();
+		for (size_t i = 0; i < size; ++i)
+			c[i] = a == b[i];
+	}
+
+	static void constant_constant(A a, B b, UInt8 & c)
 	{
 		c = a == b;
 	}
 };
 
-template<typename A, typename B>
 struct EqualsStringImpl
 {
-	typedef UInt8 ResultType;
-
 	static void string_vector_string_vector(
 		const std::vector<UInt8> & a_data, const std::vector<size_t> & a_offsets,
 		const std::vector<UInt8> & b_data, const std::vector<size_t> & b_offsets,
-		std::vector<ResultType> & c)
+		std::vector<UInt8> & c)
 	{
-		size_t size = a_data.size();
+		size_t size = a_offsets.size();
 		for (size_t i = 0; i < size; ++i)
 			c[i] = (i == 0)
 				? (a_offsets[0] == b_offsets[0] && !memcmp(&a_data[0], &b_data[0], a_offsets[0]))
@@ -71,130 +85,135 @@ struct EqualsStringImpl
 	static void string_vector_fixed_string_vector(
 		const std::vector<UInt8> & a_data, const std::vector<size_t> & a_offsets,
 		const std::vector<UInt8> & b_data, size_t b_n,
-		std::vector<ResultType> & c)
+		std::vector<UInt8> & c)
 	{
-		size_t size = a_data.size();
+		size_t size = a_offsets.size();
 		for (size_t i = 0; i < size; ++i)
 			c[i] = (i == 0)
-				? (a_offsets[0] == b_n && !memcmp(&a_data[0], &b_data[0], b_n))
-				: (a_offsets[i] - a_offsets[i - 1] == b_n
+				? (a_offsets[0] == b_n + 1 && !memcmp(&a_data[0], &b_data[0], b_n))
+				: (a_offsets[i] - a_offsets[i - 1] == b_n + 1
 					&& !memcmp(&a_data[a_offsets[i - 1]], &b_data[b_n * i], b_n));
 	}
 
 	static void string_vector_constant(
 		const std::vector<UInt8> & a_data, const std::vector<size_t> & a_offsets,
 		const std::string & b,
-		std::vector<ResultType> & c)
+		std::vector<UInt8> & c)
 	{
-		size_t size = a_data.size();
+		size_t size = a_offsets.size();
 		size_t b_n = b.size();
-		const UInt8 * b_data = reinterpret_cast<const UInt8 *>(b.data())
+		const UInt8 * b_data = reinterpret_cast<const UInt8 *>(b.data());
 		for (size_t i = 0; i < size; ++i)
 			c[i] = (i == 0)
-				? (a_offsets[0] == b_n && !memcmp(&a_data[0], b_data, b_n))
-				: (a_offsets[i] - a_offsets[i - 1] == b_n
+				? (a_offsets[0] == b_n + 1 && !memcmp(&a_data[0], b_data, b_n))
+				: (a_offsets[i] - a_offsets[i - 1] == b_n + 1
 					&& !memcmp(&a_data[a_offsets[i - 1]], b_data, b_n));
 	}
 
+	static void fixed_string_vector_string_vector(
+		const std::vector<UInt8> & a_data, size_t a_n,
+		const std::vector<UInt8> & b_data, const std::vector<size_t> & b_offsets,
+		std::vector<UInt8> & c)
+	{
+		size_t size = b_offsets.size();
+		for (size_t i = 0; i < size; ++i)
+			c[i] = (i == 0)
+				? (b_offsets[0] == a_n + 1 && !memcmp(&b_data[0], &a_data[0], a_n))
+				: (b_offsets[i] - b_offsets[i - 1] == a_n + 1
+					&& !memcmp(&b_data[b_offsets[i - 1]], &a_data[a_n * i], a_n));
+	}
+
 	static void fixed_string_vector_fixed_string_vector(
 		const std::vector<UInt8> & a_data, size_t a_n,
 		const std::vector<UInt8> & b_data, size_t b_n,
-		std::vector<ResultType> & c)
+		std::vector<UInt8> & c)
 	{
 		size_t size = a_data.size();
-		for (size_t i = 0; i < size; i += n)
-			c[i] = a_n == b_n && !memcmp(&a_data[i], &b_data[i], n);
+		for (size_t i = 0; i < size; i += a_n)
+			c[i] = a_n == b_n && !memcmp(&a_data[i], &b_data[i], a_n);
 	}
 
 	static void fixed_string_vector_constant(
 		const std::vector<UInt8> & a_data, size_t a_n,
 		const std::string & b,
-		std::vector<ResultType> & c)
+		std::vector<UInt8> & c)
 	{
 		size_t size = a_data.size();
-		for (size_t i = 0; i < size; i += n)
-			c[i] = !memcmp(&a_data[i], &b_data[i], n);
+		const UInt8 * b_data = reinterpret_cast<const UInt8 *>(b.data());
+		size_t b_n = b.size();
+		for (size_t i = 0; i < size; i += a_n)
+			c[i] = a_n == b_n && !memcmp(&a_data[i], b_data, a_n);
+	}
+
+	static void constant_string_vector(
+		const std::string & a,
+		const std::vector<UInt8> & b_data, const std::vector<size_t> & b_offsets,
+		std::vector<UInt8> & c)
+	{
+		size_t size = b_offsets.size();
+		size_t a_n = a.size();
+		const UInt8 * a_data = reinterpret_cast<const UInt8 *>(a.data());
+		for (size_t i = 0; i < size; ++i)
+			c[i] = (i == 0)
+				? (b_offsets[0] == a_n + 1 && !memcmp(&b_data[0], a_data, a_n))
+				: (b_offsets[i] - b_offsets[i - 1] == a_n + 1
+					&& !memcmp(&b_data[b_offsets[i - 1]], a_data, a_n));
+	}
+
+	static void constant_fixed_string_vector(
+		const std::string & a,
+		const std::vector<UInt8> & b_data, size_t b_n,
+		std::vector<UInt8> & c)
+	{
+		size_t size = b_data.size();
+		const UInt8 * a_data = reinterpret_cast<const UInt8 *>(a.data());
+		size_t a_n = a.size();
+		for (size_t i = 0; i < size; i += b_n)
+			c[i] = a_n == b_n && !memcmp(&b_data[i], a_data, b_n);
 	}
 
 	static void constant_constant(
 		const std::string & a,
 		const std::string & b,
-		std::string & c)
+		UInt8 & c)
 	{
 		c = a == b;
 	}
 };
 
+#pragma GCC diagnostic pop
 
 
-template <template <typename, typename> class Impl, typename Name>
+template <
+	template <typename, typename> class NumImpl,
+	typename StringImpl,
+	typename Name>
 class FunctionComparison : public IFunction
 {
 private:
-	template <typename T0, typename T1>
-	bool checkRightType(const DataTypes & arguments, DataTypes & types_res) const
-	{
-		if (dynamic_cast<const T1 *>(&*arguments[1]))
-		{
-			types_res.push_back(new typename DataTypeFromFieldType<
-				typename Impl<typename T0::FieldType, typename T1::FieldType>::ResultType>::Type);
-			return true;
-		}
-		return false;
-	}
-
-	template <typename T0>
-	bool checkLeftType(const DataTypes & arguments, DataTypes & types_res) const
-	{
-		if (dynamic_cast<const T0 *>(&*arguments[0]))
-		{
-			if (	checkRightType<T0, DataTypeUInt8>(arguments, types_res)
-				||	checkRightType<T0, DataTypeUInt16>(arguments, types_res)
-				||	checkRightType<T0, DataTypeUInt32>(arguments, types_res)
-				||	checkRightType<T0, DataTypeUInt64>(arguments, types_res)
-				||	checkRightType<T0, DataTypeInt8>(arguments, types_res)
-				||	checkRightType<T0, DataTypeInt16>(arguments, types_res)
-				||	checkRightType<T0, DataTypeInt32>(arguments, types_res)
-				||	checkRightType<T0, DataTypeInt64>(arguments, types_res)
-				||	checkRightType<T0, DataTypeFloat32>(arguments, types_res)
-				||	checkRightType<T0, DataTypeFloat64>(arguments, types_res)
-				||	checkRightType<T0, DataTypeVarUInt>(arguments, types_res)
-				||	checkRightType<T0, DataTypeVarInt>(arguments, types_res))
-				return true;
-			else
-				throw Exception("Illegal type " + arguments[1]->getName() + " of second argument of function " + getName(),
-					ErrorCodes::ILLEGAL_TYPE_OF_ARGUMENT);
-		}
-		return false;
-	}
-
 
 	template <typename T0, typename T1>
-	bool executeRightType(Block & block, const ColumnNumbers & arguments, const ColumnNumbers & result, const ColumnVector<T0> * col_left)
+	bool executeNumRightType(Block & block, const ColumnNumbers & arguments, const ColumnNumbers & result, const ColumnVector<T0> * col_left)
 	{
 		if (ColumnVector<T1> * col_right = dynamic_cast<ColumnVector<T1> *>(&*block.getByPosition(arguments[1]).column))
 		{
-			typedef typename Impl<T0, T1>::ResultType ResultType;
-
-			ColumnVector<ResultType> * col_res = new ColumnVector<ResultType>;
+			ColumnUInt8 * col_res = new ColumnUInt8;
 			block.getByPosition(result[0]).column = col_res;
 
-			typename ColumnVector<ResultType>::Container_t & vec_res = col_res->getData();
+			ColumnUInt8::Container_t & vec_res = col_res->getData();
 			vec_res.resize(col_left->getData().size());
-			Impl<T0, T1>::vector_vector(col_left->getData(), col_right->getData(), vec_res);
+			NumImpl<T0, T1>::vector_vector(col_left->getData(), col_right->getData(), vec_res);
 
 			return true;
 		}
 		else if (ColumnConst<T1> * col_right = dynamic_cast<ColumnConst<T1> *>(&*block.getByPosition(arguments[1]).column))
 		{
-			typedef typename Impl<T0, T1>::ResultType ResultType;
-
-			ColumnVector<ResultType> * col_res = new ColumnVector<ResultType>;
+			ColumnUInt8 * col_res = new ColumnUInt8;
 			block.getByPosition(result[0]).column = col_res;
 
-			typename ColumnVector<ResultType>::Container_t & vec_res = col_res->getData();
+			ColumnUInt8::Container_t & vec_res = col_res->getData();
 			vec_res.resize(col_left->getData().size());
-			Impl<T0, T1>::vector_constant(col_left->getData(), col_right->getData(), vec_res);
+			NumImpl<T0, T1>::vector_constant(col_left->getData(), col_right->getData(), vec_res);
 
 			return true;
 		}
@@ -203,29 +222,25 @@ private:
 	}
 
 	template <typename T0, typename T1>
-	bool executeConstRightType(Block & block, const ColumnNumbers & arguments, const ColumnNumbers & result, const ColumnConst<T0> * col_left)
+	bool executeNumConstRightType(Block & block, const ColumnNumbers & arguments, const ColumnNumbers & result, const ColumnConst<T0> * col_left)
 	{
 		if (ColumnVector<T1> * col_right = dynamic_cast<ColumnVector<T1> *>(&*block.getByPosition(arguments[1]).column))
 		{
-			typedef typename Impl<T0, T1>::ResultType ResultType;
-
-			ColumnVector<ResultType> * col_res = new ColumnVector<ResultType>;
+			ColumnUInt8 * col_res = new ColumnUInt8;
 			block.getByPosition(result[0]).column = col_res;
 
-			typename ColumnVector<ResultType>::Container_t & vec_res = col_res->getData();
+			ColumnUInt8::Container_t & vec_res = col_res->getData();
 			vec_res.resize(col_left->size());
-			Impl<T0, T1>::constant_vector(col_left->getData(), col_right->getData(), vec_res);
+			NumImpl<T0, T1>::constant_vector(col_left->getData(), col_right->getData(), vec_res);
 
 			return true;
 		}
 		else if (ColumnConst<T1> * col_right = dynamic_cast<ColumnConst<T1> *>(&*block.getByPosition(arguments[1]).column))
 		{
-			typedef typename Impl<T0, T1>::ResultType ResultType;
-
-			ResultType res = 0;
-			Impl<T0, T1>::constant_constant(col_left->getData(), col_right->getData(), res);
+			UInt8 res = 0;
+			NumImpl<T0, T1>::constant_constant(col_left->getData(), col_right->getData(), res);
 			
-			ColumnConst<ResultType> * col_res = new ColumnConst<ResultType>(col_left->size(), res);
+			ColumnConstUInt8 * col_res = new ColumnConstUInt8(col_left->size(), res);
 			block.getByPosition(result[0]).column = col_res;
 
 			return true;
@@ -235,20 +250,20 @@ private:
 	}
 
 	template <typename T0>
-	bool executeLeftType(Block & block, const ColumnNumbers & arguments, const ColumnNumbers & result)
+	bool executeNumLeftType(Block & block, const ColumnNumbers & arguments, const ColumnNumbers & result)
 	{
 		if (ColumnVector<T0> * col_left = dynamic_cast<ColumnVector<T0> *>(&*block.getByPosition(arguments[0]).column))
 		{
-			if (	executeRightType<T0, UInt8>(block, arguments, result, col_left)
-				||	executeRightType<T0, UInt16>(block, arguments, result, col_left)
-				||	executeRightType<T0, UInt32>(block, arguments, result, col_left)
-				||	executeRightType<T0, UInt64>(block, arguments, result, col_left)
-				||	executeRightType<T0, Int8>(block, arguments, result, col_left)
-				||	executeRightType<T0, Int16>(block, arguments, result, col_left)
-				||	executeRightType<T0, Int32>(block, arguments, result, col_left)
-				||	executeRightType<T0, Int64>(block, arguments, result, col_left)
-				||	executeRightType<T0, Float32>(block, arguments, result, col_left)
-				||	executeRightType<T0, Float64>(block, arguments, result, col_left))
+			if (	executeNumRightType<T0, UInt8>(block, arguments, result, col_left)
+				||	executeNumRightType<T0, UInt16>(block, arguments, result, col_left)
+				||	executeNumRightType<T0, UInt32>(block, arguments, result, col_left)
+				||	executeNumRightType<T0, UInt64>(block, arguments, result, col_left)
+				||	executeNumRightType<T0, Int8>(block, arguments, result, col_left)
+				||	executeNumRightType<T0, Int16>(block, arguments, result, col_left)
+				||	executeNumRightType<T0, Int32>(block, arguments, result, col_left)
+				||	executeNumRightType<T0, Int64>(block, arguments, result, col_left)
+				||	executeNumRightType<T0, Float32>(block, arguments, result, col_left)
+				||	executeNumRightType<T0, Float64>(block, arguments, result, col_left))
 				return true;
 			else
 				throw Exception("Illegal column of second argument of function " + getName(),
@@ -256,16 +271,16 @@ private:
 		}
 		else if (ColumnConst<T0> * col_left = dynamic_cast<ColumnConst<T0> *>(&*block.getByPosition(arguments[0]).column))
 		{
-			if (	executeConstRightType<T0, UInt8>(block, arguments, result, col_left)
-				||	executeConstRightType<T0, UInt16>(block, arguments, result, col_left)
-				||	executeConstRightType<T0, UInt32>(block, arguments, result, col_left)
-				||	executeConstRightType<T0, UInt64>(block, arguments, result, col_left)
-				||	executeConstRightType<T0, Int8>(block, arguments, result, col_left)
-				||	executeConstRightType<T0, Int16>(block, arguments, result, col_left)
-				||	executeConstRightType<T0, Int32>(block, arguments, result, col_left)
-				||	executeConstRightType<T0, Int64>(block, arguments, result, col_left)
-				||	executeConstRightType<T0, Float32>(block, arguments, result, col_left)
-				||	executeConstRightType<T0, Float64>(block, arguments, result, col_left))
+			if (	executeNumConstRightType<T0, UInt8>(block, arguments, result, col_left)
+				||	executeNumConstRightType<T0, UInt16>(block, arguments, result, col_left)
+				||	executeNumConstRightType<T0, UInt32>(block, arguments, result, col_left)
+				||	executeNumConstRightType<T0, UInt64>(block, arguments, result, col_left)
+				||	executeNumConstRightType<T0, Int8>(block, arguments, result, col_left)
+				||	executeNumConstRightType<T0, Int16>(block, arguments, result, col_left)
+				||	executeNumConstRightType<T0, Int32>(block, arguments, result, col_left)
+				||	executeNumConstRightType<T0, Int64>(block, arguments, result, col_left)
+				||	executeNumConstRightType<T0, Float32>(block, arguments, result, col_left)
+				||	executeNumConstRightType<T0, Float64>(block, arguments, result, col_left))
 				return true;
 			else
 				throw Exception("Illegal column of second argument of function " + getName(),
@@ -274,9 +289,80 @@ private:
 		
 		return false;
 	}
+
+	void executeString(Block & block, const ColumnNumbers & arguments, const ColumnNumbers & result)
+	{
+		IColumn * c0 = &*block.getByPosition(arguments[0]).column;
+		IColumn * c1 = &*block.getByPosition(arguments[1]).column;
+
+		ColumnString * c0_string = dynamic_cast<ColumnString *>(c0);
+		ColumnString * c1_string = dynamic_cast<ColumnString *>(c1);
+		ColumnFixedString * c0_fixed_string = dynamic_cast<ColumnFixedString *>(c0);
+		ColumnFixedString * c1_fixed_string = dynamic_cast<ColumnFixedString *>(c1);
+		ColumnConstString * c0_const = dynamic_cast<ColumnConstString *>(c0);
+		ColumnConstString * c1_const = dynamic_cast<ColumnConstString *>(c1);
+
+		if (c0_const && c1_const)
+		{
+			ColumnConstUInt8 * c_res = new ColumnConstUInt8(c0_const->size(), 0);
+			block.getByPosition(result[0]).column = c_res;
+			StringImpl::constant_constant(c0_const->getData(), c1_const->getData(), c_res->getData());
+		}
+		else
+		{
+			ColumnUInt8 * c_res = new ColumnUInt8;
+			block.getByPosition(result[0]).column = c_res;
+			ColumnUInt8::Container_t & vec_res = c_res->getData();
+			vec_res.resize(c0->size());
+
+			if (c0_string && c1_string)
+				StringImpl::string_vector_string_vector(
+					dynamic_cast<ColumnUInt8 &>(c0_string->getData()).getData(), c0_string->getOffsets(),
+					dynamic_cast<ColumnUInt8 &>(c1_string->getData()).getData(), c1_string->getOffsets(),
+					c_res->getData());
+			else if (c0_string && c1_fixed_string)
+				StringImpl::string_vector_fixed_string_vector(
+					dynamic_cast<ColumnUInt8 &>(c0_string->getData()).getData(), c0_string->getOffsets(),
+					dynamic_cast<ColumnUInt8 &>(c1_fixed_string->getData()).getData(), c1_fixed_string->getN(),
+					c_res->getData());
+			else if (c0_string && c1_const)
+				StringImpl::string_vector_constant(
+					dynamic_cast<ColumnUInt8 &>(c0_string->getData()).getData(), c0_string->getOffsets(),
+					c1_const->getData(),
+					c_res->getData());
+			else if (c0_fixed_string && c1_string)
+				StringImpl::fixed_string_vector_string_vector(
+					dynamic_cast<ColumnUInt8 &>(c0_fixed_string->getData()).getData(), c0_fixed_string->getN(),
+					dynamic_cast<ColumnUInt8 &>(c1_string->getData()).getData(), c1_string->getOffsets(),
+					c_res->getData());
+			else if (c0_fixed_string && c1_fixed_string)
+				StringImpl::fixed_string_vector_fixed_string_vector(
+					dynamic_cast<ColumnUInt8 &>(c0_fixed_string->getData()).getData(), c0_fixed_string->getN(),
+					dynamic_cast<ColumnUInt8 &>(c1_fixed_string->getData()).getData(), c1_fixed_string->getN(),
+					c_res->getData());
+			else if (c0_fixed_string && c1_const)
+				StringImpl::fixed_string_vector_constant(
+					dynamic_cast<ColumnUInt8 &>(c0_fixed_string->getData()).getData(), c0_fixed_string->getN(),
+					c1_const->getData(),
+					c_res->getData());
+			else if (c0_const && c1_string)
+				StringImpl::constant_string_vector(
+					c0_const->getData(),
+					dynamic_cast<ColumnUInt8 &>(c1_string->getData()).getData(), c1_string->getOffsets(),
+					c_res->getData());
+			else if (c0_const && c1_fixed_string)
+				StringImpl::constant_fixed_string_vector(
+					c0_const->getData(),
+					dynamic_cast<ColumnUInt8 &>(c1_fixed_string->getData()).getData(), c1_fixed_string->getN(),
+					c_res->getData());
+			else
+				throw Exception("Illegal columns of arguments of function " + getName(),
+					ErrorCodes::ILLEGAL_COLUMN);
+		}
+	}
 	
 public:
-	/// Получить все имена функции.
+	/// Получить имя функции.
 	String getName() const
 	{
 		return Name::get();
@@ -290,23 +376,16 @@ public:
 				+ Poco::NumberFormatter::format(arguments.size()) + ", should be 2.",
 				ErrorCodes::NUMBER_OF_ARGUMENTS_DOESNT_MATCH);
 
+		if (!(	(arguments[0]->isNumeric() && arguments[1]->isNumeric())
+			||	(	(arguments[0]->getName() == "String" || arguments[0]->getName().substr(0, 11) == "FixedString")
+				&& 	(arguments[1]->getName() == "String" || arguments[1]->getName().substr(0, 11) == "FixedString"))
+			||	(arguments[0]->getName() == "Date" && arguments[1]->getName() == "Date")
+			||	(arguments[0]->getName() == "DateTime" && arguments[1]->getName() == "DateTime")))
+			throw Exception("Illegal types of arguments (" + arguments[0]->getName() + ", " + arguments[1]->getName() + ")"
+				" of function " + getName(), ErrorCodes::ILLEGAL_TYPE_OF_ARGUMENT);
+			
 		DataTypes types_res;
-
-		if (!(	checkLeftType<DataTypeUInt8>(arguments, types_res)
-			||	checkLeftType<DataTypeUInt16>(arguments, types_res)
-			||	checkLeftType<DataTypeUInt32>(arguments, types_res)
-			||	checkLeftType<DataTypeUInt64>(arguments, types_res)
-			||	checkLeftType<DataTypeInt8>(arguments, types_res)
-			||	checkLeftType<DataTypeInt16>(arguments, types_res)
-			||	checkLeftType<DataTypeInt32>(arguments, types_res)
-			||	checkLeftType<DataTypeInt64>(arguments, types_res)
-			||	checkLeftType<DataTypeFloat32>(arguments, types_res)
-			||	checkLeftType<DataTypeFloat64>(arguments, types_res)
-			||	checkLeftType<DataTypeVarUInt>(arguments, types_res)
-			||	checkLeftType<DataTypeVarInt>(arguments, types_res)))
-			throw Exception("Illegal type " + arguments[0]->getName() + " of first argument of function " + getName(),
-				ErrorCodes::ILLEGAL_TYPE_OF_ARGUMENT);
-
+		types_res.push_back(new DataTypeUInt8);
 		return types_res;
 	}
 
@@ -317,35 +396,29 @@ public:
 			throw Exception("Wrong number of result columns in function " + getName() + ", should be 1.",
 				ErrorCodes::ILLEGAL_NUMBER_OF_RESULT_COLUMNS);
 
-		if (!(	executeLeftType<UInt8>(block, arguments, result)
-			||	executeLeftType<UInt16>(block, arguments, result)
-			||	executeLeftType<UInt32>(block, arguments, result)
-			||	executeLeftType<UInt64>(block, arguments, result)
-			||	executeLeftType<Int8>(block, arguments, result)
-			||	executeLeftType<Int16>(block, arguments, result)
-			||	executeLeftType<Int32>(block, arguments, result)
-			||	executeLeftType<Int64>(block, arguments, result)
-			||	executeLeftType<Float32>(block, arguments, result)
-			||	executeLeftType<Float64>(block, arguments, result)))
-		   throw Exception("Illegal column of first argument of function " + getName(),
-				ErrorCodes::ILLEGAL_COLUMN);
+		if (block.getByPosition(arguments[0]).column->isNumeric())
+		{
+			if (!(	executeNumLeftType<UInt8>(block, arguments, result)
+				||	executeNumLeftType<UInt16>(block, arguments, result)
+				||	executeNumLeftType<UInt32>(block, arguments, result)
+				||	executeNumLeftType<UInt64>(block, arguments, result)
+				||	executeNumLeftType<Int8>(block, arguments, result)
+				||	executeNumLeftType<Int16>(block, arguments, result)
+				||	executeNumLeftType<Int32>(block, arguments, result)
+				||	executeNumLeftType<Int64>(block, arguments, result)
+				||	executeNumLeftType<Float32>(block, arguments, result)
+				||	executeNumLeftType<Float64>(block, arguments, result)))
+				throw Exception("Illegal column of first argument of function " + getName(),
+					ErrorCodes::ILLEGAL_COLUMN);
+		}
+		else
+			executeString(block, arguments, result);
 	}
 };
 
 
-struct NamePlus 			{ static const char * get() { return "plus"; } };
-struct NameMinus 			{ static const char * get() { return "minus"; } };
-struct NameMultiply 		{ static const char * get() { return "multiply"; } };
-struct NameDivideFloating	{ static const char * get() { return "divide"; } };
-struct NameDivideIntegral 	{ static const char * get() { return "div"; } };
-struct NameModulo 			{ static const char * get() { return "modulo"; } };
-
-typedef FunctionBinaryArithmetic<PlusImpl, 				NamePlus> 			FunctionPlus;
-typedef FunctionBinaryArithmetic<MinusImpl, 			NameMinus> 			FunctionMinus;
-typedef FunctionBinaryArithmetic<MultiplyImpl, 			NameMultiply> 		FunctionMultiply;
-typedef FunctionBinaryArithmetic<DivideFloatingImpl, 	NameDivideFloating> FunctionDivideFloating;
-typedef FunctionBinaryArithmetic<DivideIntegralImpl, 	NameDivideIntegral> FunctionDivideIntegral;
-typedef FunctionBinaryArithmetic<ModuloImpl, 			NameModulo> 		FunctionModulo;
+struct NameEquals 			{ static const char * get() { return "equals"; } };
 
+typedef FunctionComparison<EqualsNumImpl, 		EqualsStringImpl, 		NameEquals>			FunctionEquals;
 
 }

dbms/include/DB/Functions/NumberTraits.h

@@ -134,9 +134,21 @@ namespace NumberTraits
 	{
 		typedef typename Construct<
 			typename boost::mpl::or_<typename Traits<A>::Sign, typename Traits<B>::Sign>::type,
-			typename boost::mpl::or_<typename Traits<A>::Floatness, typename Traits<B>::Floatness>::type,
+			Integer,
 			typename Traits<B>::Bits>::Type Type;
 	};
+
+	/** Перед применением оператора %, операнды приводятся к целым числам. */
+	template <typename A> struct ToInteger
+	{
+		typedef typename Construct<
+			typename Traits<A>::Sign,
+			Integer,
+			typename boost::mpl::if_<
+				typename Traits<A>::Floatness,
+				Bits64,
+				typename Traits<A>::Bits>::type>::Type Type;
+	};
 }
 
 	

dbms/src/Core/Block.cpp

@@ -20,6 +20,7 @@ Block & Block::operator= (const Block & other)
 {
 	data = other.data;
 	rebuildIndexByPosition();
+	index_by_name.clear();
 	
 	for (IndexByName_t::const_iterator it = other.index_by_name.begin(); it != other.index_by_name.end(); ++it)
 	{
@@ -76,19 +77,32 @@ void Block::erase(size_t position)
 
 ColumnWithNameAndType & Block::getByPosition(size_t position)
 {
+	if (position >= index_by_position.size())
+		throw Exception("Position out of bound in Block::getByPosition(), max position = "
+			+ Poco::NumberFormatter::format(index_by_position.size()), ErrorCodes::POSITION_OUT_OF_BOUND);
+		
 	return *index_by_position[position];
 }
 
 
 const ColumnWithNameAndType & Block::getByPosition(size_t position) const
 {
+	if (position >= index_by_position.size())
+		throw Exception("Position out of bound in Block::getByPosition(), max position = "
+			+ Poco::NumberFormatter::format(index_by_position.size()), ErrorCodes::POSITION_OUT_OF_BOUND);
+		
 	return *index_by_position[position];
 }
 
 
 ColumnWithNameAndType & Block::getByName(const std::string & name)
 {
-	return *index_by_name[name];
+	IndexByName_t::const_iterator it = index_by_name.find(name);
+	if (index_by_name.end() == it)
+		throw Exception("Not found column " + name + " in block. There are only columns: " + dumpNames()
+			, ErrorCodes::NOT_FOUND_COLUMN_IN_BLOCK);
+
+	return *it->second;
 }
 
 

dbms/src/Interpreters/Expression.cpp

@@ -178,7 +178,7 @@ void Expression::executeImpl(ASTPtr ast, Block & block)
 		column.column = node->type->createConstColumn(block.rows(), node->value);
 		column.type = node->type;
 		column.name = node->getTreeID();
-		
+
 		block.insert(column);
 	}
 

dbms/src/Interpreters/tests/expression.cpp

@@ -8,6 +8,7 @@
 #include <DB/DataTypes/DataTypesNumberFixed.h>
 
 #include <DB/Functions/FunctionsArithmetic.h>
+#include <DB/Functions/FunctionsComparison.h>
 
 #include <DB/Parsers/ASTSelectQuery.h>
 #include <DB/Parsers/ParserSelectQuery.h>
@@ -78,7 +79,7 @@ int main(int argc, char ** argv)
 	{
 		DB::ParserSelectQuery parser;
 		DB::ASTPtr ast;
-		std::string input = "SELECT 2 + x * 2, x * 2";
+		std::string input = "SELECT x, s1, s2, 2 + x * 2, x * 2, x % 3 == 1, s1 == 'abc', s1 == s2";
 		std::string expected;
 
 		const char * begin = input.data();
@@ -100,8 +101,12 @@ int main(int argc, char ** argv)
 
 		DB::Context context;
 		context.columns["x"] = new DB::DataTypeInt16;
+		context.columns["s1"] = new DB::DataTypeString;
+		context.columns["s2"] = new DB::DataTypeString;
 		(*context.functions)["plus"] = new DB::FunctionPlus;
 		(*context.functions)["multiply"] = new DB::FunctionMultiply;
+		(*context.functions)["modulo"] = new DB::FunctionModulo;
+		(*context.functions)["equals"] = new DB::FunctionEquals;
 
 		DB::Expression expression(ast, context);
 
@@ -110,19 +115,40 @@ int main(int argc, char ** argv)
 		size_t n = argc == 2 ? atoi(argv[1]) : 10;
 
 		DB::Block block;
+		
 		DB::ColumnWithNameAndType column_x;
 		column_x.name = "x";
 		column_x.type = new DB::DataTypeInt16;
 		DB::ColumnInt16 * x = new DB::ColumnInt16;
 		column_x.column = x;
-		std::vector<Int16> & vec = x->getData();
+		std::vector<Int16> & vec_x = x->getData();
 
-		vec.resize(n);
+		vec_x.resize(n);
 		for (size_t i = 0; i < n; ++i)
-			vec[i] = i;
+			vec_x[i] = i;
 
 		block.insert(column_x);
 
+		DB::ColumnWithNameAndType column_s1;
+		column_s1.name = "s1";
+		column_s1.type = new DB::DataTypeString;
+		column_s1.column = new DB::ColumnString;
+
+		for (size_t i = 0; i < n; ++i)
+			column_s1.column->insert(i % 2 ? "abc" : "def");
+
+		block.insert(column_s1);
+
+		DB::ColumnWithNameAndType column_s2;
+		column_s2.name = "s2";
+		column_s2.type = new DB::DataTypeString;
+		column_s2.column = new DB::ColumnString;
+
+		for (size_t i = 0; i < n; ++i)
+			column_s2.column->insert(i % 3 ? "abc" : "def");
+
+		block.insert(column_s2);
+
 		{
 			Poco::Stopwatch stopwatch;
 			stopwatch.start();