#pragma once #include #include #include #include #include #include #include #include #include #include namespace DB { /** Функции сравнения: ==, !=, <, >, <=, >=. * Функции сравнения возвращают всегда 0 или 1 (UInt8). * * Сравнивать можно следующие типы: * - числа; * - строки и фиксированные строки; * - даты; * - даты-с-временем; * внутри каждой группы, но не из разных групп. */ /** Игнорируем warning о сравнении signed и unsigned. * (Результат может быть некорректным.) */ #pragma GCC diagnostic push #pragma GCC diagnostic ignored "-Wsign-compare" template struct EqualsOp { static UInt8 apply(A a, B b) { return a == b; } }; template struct NotEqualsOp { static UInt8 apply(A a, B b) { return a != b; } }; template struct LessOp { static UInt8 apply(A a, B b) { return a < b; } }; template struct GreaterOp { static UInt8 apply(A a, B b) { return a > b; } }; template struct LessOrEqualsOp { static UInt8 apply(A a, B b) { return a <= b; } }; template struct GreaterOrEqualsOp { static UInt8 apply(A a, B b) { return a >= b; } }; #pragma GCC diagnostic pop template struct NumComparisonImpl { static void vector_vector(const PODArray & a, const PODArray & b, PODArray & c) { /** GCC 4.8.2 векторизует цикл только если его записать в такой форме. * В данном случае, если сделать цикл по индексу массива (код будет выглядеть проще), * цикл не будет векторизовываться. */ size_t size = a.size(); const A * a_pos = &a[0]; const B * b_pos = &b[0]; UInt8 * c_pos = &c[0]; const A * a_end = a_pos + size; while (a_pos < a_end) { *c_pos = Op::apply(*a_pos, *b_pos); ++a_pos; ++b_pos; ++c_pos; } } static void vector_constant(const PODArray & a, B b, PODArray & c) { size_t size = a.size(); const A * a_pos = &a[0]; UInt8 * c_pos = &c[0]; const A * a_end = a_pos + size; while (a_pos < a_end) { *c_pos = Op::apply(*a_pos, b); ++a_pos; ++c_pos; } } static void constant_vector(A a, const PODArray & b, PODArray & c) { size_t size = b.size(); const B * b_pos = &b[0]; UInt8 * c_pos = &c[0]; const B * b_end = b_pos + size; while (b_pos < b_end) { *c_pos = Op::apply(a, *b_pos); ++b_pos; ++c_pos; } } static void constant_constant(A a, B b, UInt8 & c) { c = Op::apply(a, b); } }; template struct StringComparisonImpl { static void string_vector_string_vector( const ColumnString::Chars_t & a_data, const ColumnString::Offsets_t & a_offsets, const ColumnString::Chars_t & b_data, const ColumnString::Offsets_t & b_offsets, PODArray & c) { size_t size = a_offsets.size(); for (size_t i = 0; i < size; ++i) { if (i == 0) { /// Завершающий ноль в меньшей по длине строке входит в сравнение. c[i] = Op::apply(memcmp(&a_data[0], &b_data[0], std::min(a_offsets[0], b_offsets[0])), 0); } else { c[i] = Op::apply(memcmp(&a_data[a_offsets[i - 1]], &b_data[b_offsets[i - 1]], std::min(a_offsets[i] - a_offsets[i - 1], b_offsets[i] - b_offsets[i - 1])), 0); } } } static void string_vector_fixed_string_vector( const ColumnString::Chars_t & a_data, const ColumnString::Offsets_t & a_offsets, const ColumnString::Chars_t & b_data, ColumnString::Offset_t b_n, PODArray & c) { size_t size = a_offsets.size(); for (size_t i = 0; i < size; ++i) { if (i == 0) { int res = memcmp(&a_data[0], &b_data[0], std::min(a_offsets[0] - 1, b_n)); c[i] = Op::apply(res, 0) || (res == 0 && Op::apply(a_offsets[0], b_n + 1)); } else { int res = memcmp(&a_data[a_offsets[i - 1]], &b_data[i * b_n], std::min(a_offsets[i] - a_offsets[i - 1] - 1, b_n)); c[i] = Op::apply(res, 0) || (res == 0 && Op::apply(a_offsets[i] - a_offsets[i - 1], b_n + 1)); } } } static void string_vector_constant( const ColumnString::Chars_t & a_data, const ColumnString::Offsets_t & a_offsets, const std::string & b, PODArray & c) { size_t size = a_offsets.size(); ColumnString::Offset_t b_n = b.size(); const UInt8 * b_data = reinterpret_cast(b.data()); for (size_t i = 0; i < size; ++i) { if (i == 0) { c[i] = Op::apply(memcmp(&a_data[0], b_data, std::min(a_offsets[0], b_n + 1)), 0); } else { c[i] = Op::apply(memcmp(&a_data[a_offsets[i - 1]], b_data, std::min(a_offsets[i] - a_offsets[i - 1], b_n + 1)), 0); } } } static void fixed_string_vector_string_vector( const ColumnString::Chars_t & a_data, ColumnString::Offset_t a_n, const ColumnString::Chars_t & b_data, const ColumnString::Offsets_t & b_offsets, PODArray & c) { size_t size = b_offsets.size(); for (size_t i = 0; i < size; ++i) { if (i == 0) { int res = memcmp(&a_data[0], &b_data[0], std::min(b_offsets[0] - 1, a_n)); c[i] = Op::apply(res, 0) || (res == 0 && Op::apply(a_n + 1, b_offsets[0])); } else { int res = memcmp(&a_data[i * a_n], &b_data[b_offsets[i - 1]], std::min(b_offsets[i] - b_offsets[i - 1] - 1, a_n)); c[i] = Op::apply(res, 0) || (res == 0 && Op::apply(a_n + 1, b_offsets[i] - b_offsets[i - 1])); } } } static void fixed_string_vector_fixed_string_vector( const ColumnString::Chars_t & a_data, ColumnString::Offset_t a_n, const ColumnString::Chars_t & b_data, ColumnString::Offset_t b_n, PODArray & c) { size_t size = a_data.size(); for (size_t i = 0, j = 0; i < size; i += a_n, ++j) { int res = memcmp(&a_data[i], &b_data[i], std::min(a_n, b_n)); c[j] = Op::apply(res, 0) || (res == 0 && Op::apply(a_n, b_n)); } } static void fixed_string_vector_constant( const ColumnString::Chars_t & a_data, ColumnString::Offset_t a_n, const std::string & b, PODArray & c) { size_t size = a_data.size(); const UInt8 * b_data = reinterpret_cast(b.data()); ColumnString::Offset_t b_n = b.size(); for (size_t i = 0, j = 0; i < size; i += a_n, ++j) { int res = memcmp(&a_data[i], b_data, std::min(a_n, b_n)); c[j] = Op::apply(res, 0) || (res == 0 && Op::apply(a_n, b_n)); } } static void constant_string_vector( const std::string & a, const ColumnString::Chars_t & b_data, const ColumnString::Offsets_t & b_offsets, PODArray & c) { size_t size = b_offsets.size(); ColumnString::Offset_t a_n = a.size(); const UInt8 * a_data = reinterpret_cast(a.data()); for (size_t i = 0; i < size; ++i) { if (i == 0) { c[i] = Op::apply(memcmp(a_data, &b_data[0], std::min(b_offsets[0], a_n + 1)), 0); } else { c[i] = Op::apply(memcmp(a_data, &b_data[b_offsets[i - 1]], std::min(b_offsets[i] - b_offsets[i - 1], a_n + 1)), 0); } } } static void constant_fixed_string_vector( const std::string & a, const ColumnString::Chars_t & b_data, ColumnString::Offset_t b_n, PODArray & c) { size_t size = b_data.size(); const UInt8 * a_data = reinterpret_cast(a.data()); ColumnString::Offset_t a_n = a.size(); for (size_t i = 0, j = 0; i < size; i += b_n, ++j) { int res = memcmp(a_data, &b_data[i], std::min(a_n, b_n)); c[j] = Op::apply(res, 0) || (res == 0 && Op::apply(b_n, a_n)); } } static void constant_constant( const std::string & a, const std::string & b, UInt8 & c) { c = Op::apply(memcmp(a.data(), b.data(), std::min(a.size(), b.size()) + 1), 0); } }; /// Сравнения на равенство/неравенство реализованы несколько более эффективно. template struct StringEqualsImpl { static void string_vector_string_vector( const ColumnString::Chars_t & a_data, const ColumnString::Offsets_t & a_offsets, const ColumnString::Chars_t & b_data, const ColumnString::Offsets_t & b_offsets, PODArray & c) { size_t size = a_offsets.size(); for (size_t i = 0; i < size; ++i) c[i] = positive == ((i == 0) ? (a_offsets[0] == b_offsets[0] && !memcmp(&a_data[0], &b_data[0], a_offsets[0] - 1)) : (a_offsets[i] - a_offsets[i - 1] == b_offsets[i] - b_offsets[i - 1] && !memcmp(&a_data[a_offsets[i - 1]], &b_data[b_offsets[i - 1]], a_offsets[i] - a_offsets[i - 1] - 1))); } static void string_vector_fixed_string_vector( const ColumnString::Chars_t & a_data, const ColumnString::Offsets_t & a_offsets, const ColumnString::Chars_t & b_data, ColumnString::Offset_t b_n, PODArray & c) { size_t size = a_offsets.size(); for (size_t i = 0; i < size; ++i) c[i] = positive == ((i == 0) ? (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 ColumnString::Chars_t & a_data, const ColumnString::Offsets_t & a_offsets, const std::string & b, PODArray & c) { size_t size = a_offsets.size(); ColumnString::Offset_t b_n = b.size(); const UInt8 * b_data = reinterpret_cast(b.data()); for (size_t i = 0; i < size; ++i) c[i] = positive == ((i == 0) ? (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_fixed_string_vector( const ColumnString::Chars_t & a_data, ColumnString::Offset_t a_n, const ColumnString::Chars_t & b_data, ColumnString::Offset_t b_n, PODArray & c) { size_t size = a_data.size(); for (size_t i = 0, j = 0; i < size; i += a_n, ++j) c[j] = positive == (a_n == b_n && !memcmp(&a_data[i], &b_data[i], a_n)); } static void fixed_string_vector_constant( const ColumnString::Chars_t & a_data, ColumnString::Offset_t a_n, const std::string & b, PODArray & c) { size_t size = a_data.size(); const UInt8 * b_data = reinterpret_cast(b.data()); ColumnString::Offset_t b_n = b.size(); for (size_t i = 0, j = 0; i < size; i += a_n, ++j) c[j] = positive == (a_n == b_n && !memcmp(&a_data[i], b_data, a_n)); } static void constant_constant( const std::string & a, const std::string & b, UInt8 & c) { c = positive == (a == b); } static void fixed_string_vector_string_vector( const ColumnString::Chars_t & a_data, ColumnString::Offset_t a_n, const ColumnString::Chars_t & b_data, const ColumnString::Offsets_t & b_offsets, PODArray & c) { string_vector_fixed_string_vector(b_data, b_offsets, a_data, a_n, c); } static void constant_string_vector( const std::string & a, const ColumnString::Chars_t & b_data, const ColumnString::Offsets_t & b_offsets, PODArray & c) { string_vector_constant(b_data, b_offsets, a, c); } static void constant_fixed_string_vector( const std::string & a, const ColumnString::Chars_t & b_data, ColumnString::Offset_t b_n, PODArray & c) { fixed_string_vector_constant(b_data, b_n, a, c); } }; template struct StringComparisonImpl> : StringEqualsImpl {}; template struct StringComparisonImpl> : StringEqualsImpl {}; template < template class Op, typename Name> class FunctionComparison : public IFunction { private: template bool executeNumRightType(Block & block, const ColumnNumbers & arguments, size_t result, const ColumnVector * col_left) { if (ColumnVector * col_right = typeid_cast *>(&*block.getByPosition(arguments[1]).column)) { ColumnUInt8 * col_res = new ColumnUInt8; block.getByPosition(result).column = col_res; ColumnUInt8::Container_t & vec_res = col_res->getData(); vec_res.resize(col_left->getData().size()); NumComparisonImpl>::vector_vector(col_left->getData(), col_right->getData(), vec_res); return true; } else if (ColumnConst * col_right = typeid_cast *>(&*block.getByPosition(arguments[1]).column)) { ColumnUInt8 * col_res = new ColumnUInt8; block.getByPosition(result).column = col_res; ColumnUInt8::Container_t & vec_res = col_res->getData(); vec_res.resize(col_left->getData().size()); NumComparisonImpl>::vector_constant(col_left->getData(), col_right->getData(), vec_res); return true; } return false; } template bool executeNumConstRightType(Block & block, const ColumnNumbers & arguments, size_t result, const ColumnConst * col_left) { if (ColumnVector * col_right = typeid_cast *>(&*block.getByPosition(arguments[1]).column)) { ColumnUInt8 * col_res = new ColumnUInt8; block.getByPosition(result).column = col_res; ColumnUInt8::Container_t & vec_res = col_res->getData(); vec_res.resize(col_left->size()); NumComparisonImpl>::constant_vector(col_left->getData(), col_right->getData(), vec_res); return true; } else if (ColumnConst * col_right = typeid_cast *>(&*block.getByPosition(arguments[1]).column)) { UInt8 res = 0; NumComparisonImpl>::constant_constant(col_left->getData(), col_right->getData(), res); ColumnConstUInt8 * col_res = new ColumnConstUInt8(col_left->size(), res); block.getByPosition(result).column = col_res; return true; } return false; } template bool executeNumLeftType(Block & block, const ColumnNumbers & arguments, size_t result) { if (ColumnVector * col_left = typeid_cast *>(&*block.getByPosition(arguments[0]).column)) { if ( executeNumRightType(block, arguments, result, col_left) || executeNumRightType(block, arguments, result, col_left) || executeNumRightType(block, arguments, result, col_left) || executeNumRightType(block, arguments, result, col_left) || executeNumRightType(block, arguments, result, col_left) || executeNumRightType(block, arguments, result, col_left) || executeNumRightType(block, arguments, result, col_left) || executeNumRightType(block, arguments, result, col_left) || executeNumRightType(block, arguments, result, col_left) || executeNumRightType(block, arguments, result, col_left)) return true; else throw Exception("Illegal column " + block.getByPosition(arguments[1]).column->getName() + " of second argument of function " + getName(), ErrorCodes::ILLEGAL_COLUMN); } else if (ColumnConst * col_left = typeid_cast *>(&*block.getByPosition(arguments[0]).column)) { if ( executeNumConstRightType(block, arguments, result, col_left) || executeNumConstRightType(block, arguments, result, col_left) || executeNumConstRightType(block, arguments, result, col_left) || executeNumConstRightType(block, arguments, result, col_left) || executeNumConstRightType(block, arguments, result, col_left) || executeNumConstRightType(block, arguments, result, col_left) || executeNumConstRightType(block, arguments, result, col_left) || executeNumConstRightType(block, arguments, result, col_left) || executeNumConstRightType(block, arguments, result, col_left) || executeNumConstRightType(block, arguments, result, col_left)) return true; else throw Exception("Illegal column " + block.getByPosition(arguments[1]).column->getName() + " of second argument of function " + getName(), ErrorCodes::ILLEGAL_COLUMN); } return false; } void executeString(Block & block, const ColumnNumbers & arguments, size_t result) { IColumn * c0 = &*block.getByPosition(arguments[0]).column; IColumn * c1 = &*block.getByPosition(arguments[1]).column; ColumnString * c0_string = typeid_cast(c0); ColumnString * c1_string = typeid_cast(c1); ColumnFixedString * c0_fixed_string = typeid_cast(c0); ColumnFixedString * c1_fixed_string = typeid_cast(c1); ColumnConstString * c0_const = typeid_cast(c0); ColumnConstString * c1_const = typeid_cast(c1); using StringImpl = StringComparisonImpl>; if (c0_const && c1_const) { ColumnConstUInt8 * c_res = new ColumnConstUInt8(c0_const->size(), 0); block.getByPosition(result).column = c_res; StringImpl::constant_constant(c0_const->getData(), c1_const->getData(), c_res->getData()); } else { ColumnUInt8 * c_res = new ColumnUInt8; block.getByPosition(result).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( c0_string->getChars(), c0_string->getOffsets(), c1_string->getChars(), c1_string->getOffsets(), c_res->getData()); else if (c0_string && c1_fixed_string) StringImpl::string_vector_fixed_string_vector( c0_string->getChars(), c0_string->getOffsets(), c1_fixed_string->getChars(), c1_fixed_string->getN(), c_res->getData()); else if (c0_string && c1_const) StringImpl::string_vector_constant( c0_string->getChars(), c0_string->getOffsets(), c1_const->getData(), c_res->getData()); else if (c0_fixed_string && c1_string) StringImpl::fixed_string_vector_string_vector( c0_fixed_string->getChars(), c0_fixed_string->getN(), c1_string->getChars(), c1_string->getOffsets(), c_res->getData()); else if (c0_fixed_string && c1_fixed_string) StringImpl::fixed_string_vector_fixed_string_vector( c0_fixed_string->getChars(), c0_fixed_string->getN(), c1_fixed_string->getChars(), c1_fixed_string->getN(), c_res->getData()); else if (c0_fixed_string && c1_const) StringImpl::fixed_string_vector_constant( c0_fixed_string->getChars(), c0_fixed_string->getN(), c1_const->getData(), c_res->getData()); else if (c0_const && c1_string) StringImpl::constant_string_vector( c0_const->getData(), c1_string->getChars(), c1_string->getOffsets(), c_res->getData()); else if (c0_const && c1_fixed_string) StringImpl::constant_fixed_string_vector( c0_const->getData(), c1_fixed_string->getChars(), c1_fixed_string->getN(), c_res->getData()); else throw Exception("Illegal columns " + block.getByPosition(arguments[0]).column->getName() + " and " + block.getByPosition(arguments[1]).column->getName() + " of arguments of function " + getName(), ErrorCodes::ILLEGAL_COLUMN); } } public: /// Получить имя функции. String getName() const { return Name::get(); } /// Получить типы результата по типам аргументов. Если функция неприменима для данных аргументов - кинуть исключение. DataTypePtr getReturnType(const DataTypes & arguments) const { if (arguments.size() != 2) throw Exception("Number of arguments for function " + getName() + " doesn't match: passed " + toString(arguments.size()) + ", should be 2.", ErrorCodes::NUMBER_OF_ARGUMENTS_DOESNT_MATCH); if (!( ( arguments[0]->isNumeric() && arguments[0]->behavesAsNumber() && arguments[1]->isNumeric() && arguments[1]->behavesAsNumber()) || ( (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); return new DataTypeUInt8; } /// Выполнить функцию над блоком. void execute(Block & block, const ColumnNumbers & arguments, size_t result) { if (block.getByPosition(arguments[0]).column->isNumeric()) { if (!( executeNumLeftType(block, arguments, result) || executeNumLeftType(block, arguments, result) || executeNumLeftType(block, arguments, result) || executeNumLeftType(block, arguments, result) || executeNumLeftType(block, arguments, result) || executeNumLeftType(block, arguments, result) || executeNumLeftType(block, arguments, result) || executeNumLeftType(block, arguments, result) || executeNumLeftType(block, arguments, result) || executeNumLeftType(block, arguments, result))) throw Exception("Illegal column " + block.getByPosition(arguments[0]).column->getName() + " of first argument of function " + getName(), ErrorCodes::ILLEGAL_COLUMN); } else executeString(block, arguments, result); } }; struct NameEquals { static const char * get() { return "equals"; } }; struct NameNotEquals { static const char * get() { return "notEquals"; } }; struct NameLess { static const char * get() { return "less"; } }; struct NameGreater { static const char * get() { return "greater"; } }; struct NameLessOrEquals { static const char * get() { return "lessOrEquals"; } }; struct NameGreaterOrEquals { static const char * get() { return "greaterOrEquals"; } }; typedef FunctionComparison FunctionEquals; typedef FunctionComparison FunctionNotEquals; typedef FunctionComparison FunctionLess; typedef FunctionComparison FunctionGreater; typedef FunctionComparison FunctionLessOrEquals; typedef FunctionComparison FunctionGreaterOrEquals; }