add special ColumnDecimal instead of ColumnVector<DecimalNN> CLICKHOUSE-3765

dbms/src/Columns/ColumnDecimal.cpp

@@ -0,0 +1,237 @@
+//#include <cstring>
+#include <cmath>
+
+#include <Common/Exception.h>
+#include <Common/Arena.h>
+#include <Common/SipHash.h>
+
+#include <IO/WriteHelpers.h>
+
+#include <Columns/ColumnsCommon.h>
+#include <Columns/ColumnDecimal.h>
+#include <DataStreams/ColumnGathererStream.h>
+
+template <typename T> bool decimalLess(T x, T y, UInt32 x_scale, UInt32 y_scale);
+
+namespace DB
+{
+
+namespace ErrorCodes
+{
+    extern const int PARAMETER_OUT_OF_BOUND;
+    extern const int SIZES_OF_COLUMNS_DOESNT_MATCH;
+}
+
+template <typename T>
+int ColumnDecimal<T>::compareAt(size_t n, size_t m, const IColumn & rhs_, int ) const
+{
+    auto other = static_cast<const Self &>(rhs_);
+    const T & a = data[n];
+    const T & b = static_cast<const Self &>(rhs_).data[m];
+
+    return decimalLess<T>(b, a, other.scale, scale) ? 1 : (decimalLess<T>(a, b, scale, other.scale) ? -1 : 0);
+}
+
+template <typename T>
+StringRef ColumnDecimal<T>::serializeValueIntoArena(size_t n, Arena & arena, char const *& begin) const
+{
+    auto pos = arena.allocContinue(sizeof(T), begin);
+    memcpy(pos, &data[n], sizeof(T));
+    return StringRef(pos, sizeof(T));
+}
+
+template <typename T>
+const char * ColumnDecimal<T>::deserializeAndInsertFromArena(const char * pos)
+{
+    data.push_back(*reinterpret_cast<const T *>(pos));
+    return pos + sizeof(T);
+}
+
+template <typename T>
+void ColumnDecimal<T>::updateHashWithValue(size_t n, SipHash & hash) const
+{
+    hash.update(data[n]);
+}
+
+template <typename T>
+void ColumnDecimal<T>::getPermutation(bool reverse, size_t limit, int , IColumn::Permutation & res) const
+{
+    size_t s = data.size();
+    res.resize(s);
+    for (size_t i = 0; i < s; ++i)
+        res[i] = i;
+
+    if (limit >= s)
+        limit = 0;
+
+    if (limit)
+    {
+        if (reverse)
+            std::partial_sort(res.begin(), res.begin() + limit, res.end(), [](T a, T b) { return a > b; });
+        else
+            std::partial_sort(res.begin(), res.begin() + limit, res.end(), [](T a, T b) { return a < b; });
+    }
+    else
+    {
+        if (reverse)
+            std::sort(res.begin(), res.end(), [](T a, T b) { return a > b; });
+        else
+            std::sort(res.begin(), res.end(), [](T a, T b) { return a < b; });
+    }
+}
+
+template <typename T>
+MutableColumnPtr ColumnDecimal<T>::cloneResized(size_t size) const
+{
+    auto res = this->create(0, scale);
+
+    if (size > 0)
+    {
+        auto & new_col = static_cast<Self &>(*res);
+        new_col.data.resize(size);
+
+        size_t count = std::min(this->size(), size);
+        memcpy(&new_col.data[0], &data[0], count * sizeof(data[0]));
+
+        if (size > count)
+            memset(static_cast<void *>(&new_col.data[count]), static_cast<int>(value_type()), (size - count) * sizeof(value_type));
+    }
+
+    return std::move(res);
+}
+
+template <typename T>
+void ColumnDecimal<T>::insertRangeFrom(const IColumn & src, size_t start, size_t length)
+{
+    const ColumnDecimal & src_vec = static_cast<const ColumnDecimal &>(src);
+
+    if (start + length > src_vec.data.size())
+        throw Exception("Parameters start = "
+            + toString(start) + ", length = "
+            + toString(length) + " are out of bound in ColumnVector<T>::insertRangeFrom method"
+            " (data.size() = " + toString(src_vec.data.size()) + ").",
+            ErrorCodes::PARAMETER_OUT_OF_BOUND);
+
+    size_t old_size = data.size();
+    data.resize(old_size + length);
+    memcpy(&data[old_size], &src_vec.data[start], length * sizeof(data[0]));
+}
+
+template <typename T>
+ColumnPtr ColumnDecimal<T>::filter(const IColumn::Filter & filt, ssize_t result_size_hint) const
+{
+    size_t size = data.size();
+    if (size != filt.size())
+        throw Exception("Size of filter doesn't match size of column.", ErrorCodes::SIZES_OF_COLUMNS_DOESNT_MATCH);
+
+    auto res = this->create(0, scale);
+    Container & res_data = res->getData();
+
+    if (result_size_hint)
+        res_data.reserve(result_size_hint > 0 ? result_size_hint : size);
+
+    const UInt8 * filt_pos = &filt[0];
+    const UInt8 * filt_end = filt_pos + size;
+    const T * data_pos = &data[0];
+
+    while (filt_pos < filt_end)
+    {
+        if (*filt_pos)
+            res_data.push_back(*data_pos);
+
+        ++filt_pos;
+        ++data_pos;
+    }
+
+    return std::move(res);
+}
+
+template <typename T>
+ColumnPtr ColumnDecimal<T>::permute(const IColumn::Permutation & perm, size_t limit) const
+{
+    size_t size = data.size();
+
+    if (limit == 0)
+        limit = size;
+    else
+        limit = std::min(size, limit);
+
+    if (perm.size() < limit)
+        throw Exception("Size of permutation is less than required.", ErrorCodes::SIZES_OF_COLUMNS_DOESNT_MATCH);
+
+    auto res = this->create(limit, scale);
+    typename Self::Container & res_data = res->getData();
+    for (size_t i = 0; i < limit; ++i)
+        res_data[i] = data[perm[i]];
+
+    return std::move(res);
+}
+
+template <typename T>
+ColumnPtr ColumnDecimal<T>::index(const IColumn & indexes, size_t limit) const
+{
+    return selectIndexImpl(*this, indexes, limit);
+}
+
+template <typename T>
+ColumnPtr ColumnDecimal<T>::replicate(const IColumn::Offsets & offsets) const
+{
+    size_t size = data.size();
+    if (size != offsets.size())
+        throw Exception("Size of offsets doesn't match size of column.", ErrorCodes::SIZES_OF_COLUMNS_DOESNT_MATCH);
+
+    auto res = this->create(0, scale);
+    if (0 == size)
+        return res;
+
+    typename Self::Container & res_data = res->getData();
+    res_data.reserve(offsets.back());
+
+    IColumn::Offset prev_offset = 0;
+    for (size_t i = 0; i < size; ++i)
+    {
+        size_t size_to_replicate = offsets[i] - prev_offset;
+        prev_offset = offsets[i];
+
+        for (size_t j = 0; j < size_to_replicate; ++j)
+            res_data.push_back(data[i]);
+    }
+
+    return std::move(res);
+}
+
+template <typename T>
+void ColumnDecimal<T>::gather(ColumnGathererStream & gatherer)
+{
+    gatherer.gather(*this);
+}
+
+template <typename T>
+void ColumnDecimal<T>::getExtremes(Field & min, Field & max) const
+{
+    if (data.size() == 0)
+    {
+        min = typename NearestFieldType<T>::Type(0, scale);
+        max = typename NearestFieldType<T>::Type(0, scale);
+        return;
+    }
+
+    T cur_min = data[0];
+    T cur_max = data[0];
+
+    for (const T & x : data)
+    {
+        if (x < cur_min)
+            cur_min = x;
+        else if (x > cur_max)
+            cur_max = x;
+    }
+
+    min = typename NearestFieldType<T>::Type(cur_min, scale);
+    max = typename NearestFieldType<T>::Type(cur_max, scale);
+}
+
+template class ColumnDecimal<Decimal32>;
+template class ColumnDecimal<Decimal64>;
+template class ColumnDecimal<Decimal128>;
+}

dbms/src/Columns/ColumnDecimal.h

@@ -0,0 +1,162 @@
+#pragma once
+
+#include <cmath>
+
+#include <Columns/IColumn.h>
+
+
+namespace DB
+{
+
+namespace ErrorCodes
+{
+    extern const int LOGICAL_ERROR;
+}
+
+/// PaddedPODArray extended by Decimal scale
+template <typename T>
+class DecimalPaddedPODArray : public PaddedPODArray<T>
+{
+public:
+    using Base = PaddedPODArray<T>;
+    using Base::operator[];
+
+    DecimalPaddedPODArray(size_t size, UInt32 scale_)
+    :   Base(size),
+        scale(scale_)
+    {}
+
+    DecimalPaddedPODArray(const DecimalPaddedPODArray & other)
+    :   Base(other.begin(), other.end()),
+        scale(other.scale)
+    {}
+
+    DecimalPaddedPODArray(DecimalPaddedPODArray && other)
+    {
+        this->swap(other);
+        std::swap(scale, other.scale);
+    }
+
+    DecimalPaddedPODArray & operator=(DecimalPaddedPODArray && other)
+    {
+        this->swap(other);
+        std::swap(scale, other.scale);
+        return *this;
+    }
+
+    UInt32 getScale() const { return scale; }
+
+private:
+    UInt32 scale;
+};
+
+
+/// A ColumnVector for Decimals
+template <typename T>
+class ColumnDecimal final : public COWPtrHelper<IColumn, ColumnDecimal<T>>
+{
+private:
+    using Self = ColumnDecimal<T>;
+    friend class COWPtrHelper<IColumn, Self>;
+
+public:
+    using value_type = T;
+    using Container = DecimalPaddedPODArray<value_type>;
+
+private:
+    ColumnDecimal(const size_t n, UInt32 scale_)
+    :   data(n, scale_),
+        scale(scale_)
+    {}
+
+    ColumnDecimal(const ColumnDecimal & src)
+    :   data(src.data),
+        scale(src.scale)
+    {}
+
+public:
+    const char * getFamilyName() const override { return TypeName<T>::get(); }
+
+    bool isNumeric() const override { return false; }
+    bool canBeInsideNullable() const override { return false; }
+    bool isFixedAndContiguous() const override { return true; }
+    size_t sizeOfValueIfFixed() const override { return sizeof(T); }
+
+    size_t size() const override { return data.size(); }
+    size_t byteSize() const override { return data.size() * sizeof(data[0]); }
+    size_t allocatedBytes() const override { return data.allocated_bytes(); }
+    void reserve(size_t n) override { data.reserve(n); }
+
+    void insertFrom(const IColumn & src, size_t n) override { data.push_back(static_cast<const Self &>(src).getData()[n]); }
+    void insertData(const char * pos, size_t /*length*/) override { data.push_back(*reinterpret_cast<const T *>(pos)); }
+    void insertDefault() override { data.push_back(T()); }
+    void insert(const Field & x) override { data.push_back(DB::get<typename NearestFieldType<T>::Type>(x)); }
+    void insertRangeFrom(const IColumn & src, size_t start, size_t length) override;
+
+    void popBack(size_t n) override { data.resize_assume_reserved(data.size() - n); }
+
+    StringRef serializeValueIntoArena(size_t n, Arena & arena, char const *& begin) const override;
+    const char * deserializeAndInsertFromArena(const char * pos) override;
+    void updateHashWithValue(size_t n, SipHash & hash) const override;
+    int compareAt(size_t n, size_t m, const IColumn & rhs_, int nan_direction_hint) const override;
+    void getPermutation(bool reverse, size_t limit, int nan_direction_hint, IColumn::Permutation & res) const override;
+
+    MutableColumnPtr cloneResized(size_t size) const override;
+
+    Field operator[](size_t n) const override { return DecimalField(data[n], scale); }
+
+    StringRef getRawData() const override { return StringRef(reinterpret_cast<const char*>(data.data()), data.size()); }
+    StringRef getDataAt(size_t n) const override { return StringRef(reinterpret_cast<const char *>(&data[n]), sizeof(data[n])); }
+    void get(size_t n, Field & res) const override { res = (*this)[n]; }
+    bool getBool(size_t n) const override { return bool(data[n]); }
+    Int64 getInt(size_t n) const override { return Int64(data[n] * scale); }
+
+    ColumnPtr filter(const IColumn::Filter & filt, ssize_t result_size_hint) const override;
+    ColumnPtr permute(const IColumn::Permutation & perm, size_t limit) const override;
+    ColumnPtr index(const IColumn & indexes, size_t limit) const override;
+
+    template <typename Type>
+    ColumnPtr indexImpl(const PaddedPODArray<Type> & indexes, size_t limit) const;
+
+    ColumnPtr replicate(const IColumn::Offsets & offsets) const override;
+    void getExtremes(Field & min, Field & max) const override;
+
+    MutableColumns scatter(IColumn::ColumnIndex num_columns, const IColumn::Selector & selector) const override
+    {
+        return this->template scatterImpl<Self>(num_columns, selector);
+    }
+
+    void gather(ColumnGathererStream & gatherer_stream) override;
+
+
+    void insert(const T value) { data.push_back(value); }
+    Container & getData() { return data; }
+    const Container & getData() const { return data; }
+    const T & getElement(size_t n) const { return data[n]; }
+    T & getElement(size_t n) { return data[n]; }
+
+protected:
+    Container data;
+    UInt32 scale;
+};
+
+template <typename T>
+template <typename Type>
+ColumnPtr ColumnDecimal<T>::indexImpl(const PaddedPODArray<Type> & indexes, size_t limit) const
+{
+    size_t size = indexes.size();
+
+    if (limit == 0)
+        limit = size;
+    else
+        limit = std::min(size, limit);
+
+    auto res = this->create(limit, scale);
+    typename Self::Container & res_data = res->getData();
+    for (size_t i = 0; i < limit; ++i)
+        res_data[i] = data[indexes[i]];
+
+    return std::move(res);
+}
+
+}

dbms/src/Columns/ColumnVector.cpp

@@ -278,17 +278,8 @@ void ColumnVector<T>::getExtremes(Field & min, Field & max) const
 
     if (size == 0)
     {
-        if constexpr (IsDecimalNumber<T>)
-        {
-            min = typename NearestFieldType<T>::Type(0, data.getScale());
-            max = typename NearestFieldType<T>::Type(0, data.getScale());
-        }
-        else
-        {
-            min = typename NearestFieldType<T>::Type(0);
-            max = typename NearestFieldType<T>::Type(0);
-        }
-
+        min = typename NearestFieldType<T>::Type(0);
+        max = typename NearestFieldType<T>::Type(0);
         return;
     }
 
@@ -322,16 +313,8 @@ void ColumnVector<T>::getExtremes(Field & min, Field & max) const
             cur_max = x;
     }
 
-    if constexpr (IsDecimalNumber<T>)
-    {
-        min = typename NearestFieldType<T>::Type(cur_min, data.getScale());
-        max = typename NearestFieldType<T>::Type(cur_max, data.getScale());
-    }
-    else
-    {
-        min = typename NearestFieldType<T>::Type(cur_min);
-        max = typename NearestFieldType<T>::Type(cur_max);
-    }
+    min = typename NearestFieldType<T>::Type(cur_min);
+    max = typename NearestFieldType<T>::Type(cur_max);
 }
 
 /// Explicit template instantiations - to avoid code bloat in headers.
@@ -347,8 +330,4 @@ template class ColumnVector<Int64>;
 template class ColumnVector<Int128>;
 template class ColumnVector<Float32>;
 template class ColumnVector<Float64>;
-
-template class ColumnVector<Decimal32>;
-template class ColumnVector<Decimal64>;
-template class ColumnVector<Decimal128>;
 }

dbms/src/Columns/ColumnVector.h

@@ -8,12 +8,6 @@
 namespace DB
 {
 
-namespace ErrorCodes
-{
-    extern const int LOGICAL_ERROR;
-}
-
-
 /** Stuff for comparing numbers.
   * Integer values are compared as usual.
   * Floating-point numbers are compared this way that NaNs always end up at the end
@@ -123,40 +117,6 @@ template <> inline UInt64 unionCastToUInt64(Float32 x)
 }
 
 
-/// PaddedPODArray extended by Decimal scale
-template <typename T>
-class DecimalPaddedPODArray : public PaddedPODArray<T>
-{
-public:
-    using Base = PaddedPODArray<T>;
-    using Base::operator[];
-    using Base::Base;
-
-    DecimalPaddedPODArray(std::initializer_list<T> il)
-        : DecimalPaddedPODArray(std::begin(il), std::end(il))
-    {}
-
-    DecimalPaddedPODArray(DecimalPaddedPODArray && other)
-    {
-        this->swap(other);
-        std::swap(scale, other.scale);
-    }
-
-    DecimalPaddedPODArray & operator=(DecimalPaddedPODArray && other)
-    {
-        this->swap(other);
-        std::swap(scale, other.scale);
-        return *this;
-    }
-
-    void setScale(UInt32 s) { scale = s; }
-    UInt32 getScale() const { return scale; }
-
-private:
-    UInt32 scale = DecimalField<Decimal32>::wrongScale();
-};
-
-
 /** A template for columns that use a simple array to store.
  */
 template <typename T>
@@ -171,7 +131,7 @@ private:
 
 public:
     using value_type = T;
-    using Container = std::conditional_t<IsDecimalNumber<T>, DecimalPaddedPODArray<value_type>, PaddedPODArray<value_type>>;
+    using Container = PaddedPODArray<value_type>;
 
 private:
     ColumnVector() {}
@@ -255,15 +215,7 @@ public:
 
     Field operator[](size_t n) const override
     {
-        if constexpr (IsDecimalNumber<T>)
-        {
-            UInt32 scale = data.getScale();
-            if (scale == DecimalField<Decimal32>::wrongScale())
-                throw Exception("Extracting Decimal field with unknown scale. Scale is lost.", ErrorCodes::LOGICAL_ERROR);
-            return DecimalField(data[n], scale);
-        }
-        else
-            return typename NearestFieldType<T>::Type(data[n]);
+        return typename NearestFieldType<T>::Type(data[n]);
     }
 
     void get(size_t n, Field & res) const override

dbms/src/Core/Field.h

@@ -36,8 +36,6 @@ template <typename T>
 class DecimalField
 {
 public:
-    static constexpr UInt32 wrongScale() { return std::numeric_limits<UInt32>::max(); }
-
     DecimalField(T value, UInt32 scale_)
     :   dec(value),
         scale(scale_)

dbms/src/DataTypes/DataTypesDecimal.cpp

@@ -143,9 +143,7 @@ Field DataTypeDecimal<T>::getDefault() const
 template <typename T>
 MutableColumnPtr DataTypeDecimal<T>::createColumn() const
 {
-    auto column = ColumnType::create();
-    column->getData().setScale(scale);
-    return column;
+    return ColumnType::create(0, scale);
 }
 
 

dbms/src/DataTypes/DataTypesDecimal.h

@@ -1,7 +1,7 @@
 #pragma once
 #include <common/likely.h>
 #include <Common/typeid_cast.h>
-#include <Columns/ColumnVector.h>
+#include <Columns/ColumnDecimal.h>
 #include <DataTypes/IDataType.h>
 #include <DataTypes/DataTypesNumber.h>
 
@@ -93,7 +93,7 @@ class DataTypeDecimal final : public DataTypeSimpleSerialization
 {
 public:
     using FieldType = T;
-    using ColumnType = ColumnVector<T>;
+    using ColumnType = ColumnDecimal<T>;
 
     static constexpr bool is_parametric = true;
 

dbms/src/Functions/FunctionsArithmetic.h

@@ -7,6 +7,7 @@
 #include <DataTypes/DataTypeInterval.h>
 #include <DataTypes/Native.h>
 #include <Columns/ColumnVector.h>
+#include <Columns/ColumnDecimal.h>
 #include <Columns/ColumnConst.h>
 #include <Functions/IFunction.h>
 #include <Functions/FunctionHelpers.h>
@@ -735,9 +736,9 @@ struct DecimalBinaryOperation
     using ResultType = ResultType_;
     using NativeResultType = typename NativeType<ResultType>::Type;
     using Op = Operation<NativeResultType, NativeResultType>;
-    using ArrayA = typename ColumnVector<A>::Container;
-    using ArrayB = typename ColumnVector<B>::Container;
-    using ArrayC = typename ColumnVector<ResultType>::Container;
+    using ArrayA = std::conditional_t<IsDecimalNumber<A>, typename ColumnDecimal<A>::Container, typename ColumnVector<A>::Container>;
+    using ArrayB = std::conditional_t<IsDecimalNumber<B>, typename ColumnDecimal<B>::Container, typename ColumnVector<B>::Container>;
+    using ArrayC = typename ColumnDecimal<ResultType>::Container;
     using XOverflow = DecimalBinaryOperation<A, B, Operation, ResultType_, !_check_overflow>;
 
     static constexpr bool is_plus_minus =   std::is_same_v<Operation<Int32, Int32>, PlusImpl<Int32, Int32>> ||
@@ -1199,9 +1200,9 @@ public:
                 using T0 = typename LeftDataType::FieldType;
                 using T1 = typename RightDataType::FieldType;
                 using ResultType = typename ResultDataType::FieldType;
-                using ColVecT0 = ColumnVector<T0>;
-                using ColVecT1 = ColumnVector<T1>;
-                using ColVecResult = ColumnVector<ResultType>;
+                using ColVecT0 = std::conditional_t<IsDecimalNumber<T0>, ColumnDecimal<T0>, ColumnVector<T0>>;
+                using ColVecT1 = std::conditional_t<IsDecimalNumber<T1>, ColumnDecimal<T1>, ColumnVector<T1>>;
+                using ColVecResult = std::conditional_t<IsDecimalNumber<ResultType>, ColumnDecimal<ResultType>, ColumnVector<ResultType>>;
 
                 /// Decimal operations need scale. Operations are on result type.
                 using OpImpl = std::conditional_t<IsDecimal<ResultDataType>,
@@ -1248,9 +1249,18 @@ public:
                     }
                 }
 
-                auto col_res = ColVecResult::create();
+                typename ColVecResult::MutablePtr col_res = nullptr;
+                if constexpr (result_is_decimal)
+                {
+                    ResultDataType type = decimalResultType(left, right, is_multiply, is_division);
+                    col_res = ColVecResult::create(0, type.getScale());
+                }
+                else
+                    col_res = ColVecResult::create();
+
                 auto & vec_res = col_res->getData();
                 vec_res.resize(block.rows());
+
                 if (auto col_left = checkAndGetColumnConst<ColVecT0>(col_left_raw))
                 {
                     if (auto col_right = checkAndGetColumn<ColVecT1>(col_right_raw))
@@ -1258,7 +1268,6 @@ public:
                         if constexpr (result_is_decimal)
                         {
                             ResultDataType type = decimalResultType(left, right, is_multiply, is_division);
-                            vec_res.setScale(type.getScale());
 
                             typename ResultDataType::FieldType scale_a = type.scaleFactorFor(left, is_multiply);
                             typename ResultDataType::FieldType scale_b = type.scaleFactorFor(right, is_multiply || is_division);
@@ -1281,7 +1290,6 @@ public:
                     if constexpr (result_is_decimal)
                     {
                         ResultDataType type = decimalResultType(left, right, is_multiply, is_division);
-                        vec_res.setScale(type.getScale());
 
                         typename ResultDataType::FieldType scale_a = type.scaleFactorFor(left, is_multiply);
                         typename ResultDataType::FieldType scale_b = type.scaleFactorFor(right, is_multiply || is_division);
@@ -1447,9 +1455,9 @@ public:
             {
                 if constexpr (allow_decimal)
                 {
-                    if (auto col = checkAndGetColumn<ColumnVector<T0>>(block.getByPosition(arguments[0]).column.get()))
+                    if (auto col = checkAndGetColumn<ColumnDecimal<T0>>(block.getByPosition(arguments[0]).column.get()))
                     {
-                        auto col_res = ColumnVector<typename Op<T0>::ResultType>::create();
+                        auto col_res = ColumnDecimal<typename Op<T0>::ResultType>::create(0, type.getScale());
                         auto & vec_res = col_res->getData();
                         vec_res.resize(col->getData().size());
                         UnaryOperationImpl<T0, Op<T0>>::vector(col->getData(), vec_res);

dbms/src/Functions/FunctionsComparison.h

@@ -2,6 +2,7 @@
 
 #include <Columns/ColumnsNumber.h>
 #include <Columns/ColumnConst.h>
+#include <Columns/ColumnDecimal.h>
 #include <Columns/ColumnString.h>
 #include <Columns/ColumnFixedString.h>
 #include <Columns/ColumnTuple.h>
@@ -228,8 +229,8 @@ class DecimalComparison
 public:
     using CompareInt = typename DecCompareInt<A, B>::Type;
     using Op = Operation<CompareInt, CompareInt>;
-    using ColVecA = ColumnVector<A>;
-    using ColVecB = ColumnVector<B>;
+    using ColVecA = std::conditional_t<IsDecimalNumber<A>, ColumnDecimal<A>, ColumnVector<A>>;
+    using ColVecB = std::conditional_t<IsDecimalNumber<B>, ColumnDecimal<B>, ColumnVector<B>>;
     using ArrayA = typename ColVecA::Container;
     using ArrayB = typename ColVecB::Container;
 

dbms/src/Functions/FunctionsConversion.h

@@ -102,18 +102,23 @@ struct ConvertImpl
     {
         const ColumnWithTypeAndName & named_from = block.getByPosition(arguments[0]);
 
-        if (const ColumnVector<FromFieldType> * col_from = checkAndGetColumn<ColumnVector<FromFieldType>>(named_from.column.get()))
+        using ColVecFrom = std::conditional_t<IsDecimalNumber<FromFieldType>, ColumnDecimal<FromFieldType>, ColumnVector<FromFieldType>>;
+        using ColVecTo = std::conditional_t<IsDecimalNumber<ToFieldType>, ColumnDecimal<ToFieldType>, ColumnVector<ToFieldType>>;
+
+        if (const ColVecFrom * col_from = checkAndGetColumn<ColVecFrom>(named_from.column.get()))
         {
-            auto col_to = ColumnVector<ToFieldType>::create();
+            typename ColVecTo::MutablePtr col_to = nullptr;
             if constexpr (IsDecimal<ToDataType>)
             {
                 const ColumnWithTypeAndName & scale_column = block.getByPosition(arguments[1]);
                 UInt32 scale = extractToDecimalScale(scale_column);
-                col_to->getData().setScale(scale);
+                col_to = ColVecTo::create(0, scale);
             }
+            else
+                col_to = ColVecTo::create();
 
-            const typename ColumnVector<FromFieldType>::Container & vec_from = col_from->getData();
-            typename ColumnVector<ToFieldType>::Container & vec_to = col_to->getData();
+            const auto & vec_from = col_from->getData();
+            auto & vec_to = col_to->getData();
             size_t size = vec_from.size();
             vec_to.resize(size);
 
@@ -451,6 +456,8 @@ struct ConvertThroughParsing
 
     static void execute(Block & block, const ColumnNumbers & arguments, size_t result, size_t input_rows_count)
     {
+        using ColVecTo = std::conditional_t<IsDecimalNumber<ToFieldType>, ColumnDecimal<ToFieldType>, ColumnVector<ToFieldType>>;
+
         const DateLUTImpl * local_time_zone [[maybe_unused]] = nullptr;
         const DateLUTImpl * utc_time_zone [[maybe_unused]] = nullptr;
 
@@ -478,16 +485,19 @@ struct ConvertThroughParsing
                 ErrorCodes::ILLEGAL_COLUMN);
 
         size_t size = input_rows_count;
-        auto col_to = ColumnVector<ToFieldType>::create(size);
-        typename ColumnVector<ToFieldType>::Container & vec_to = col_to->getData();
+        typename ColVecTo::MutablePtr col_to = nullptr;
 
         if constexpr (IsDecimal<ToDataType>)
         {
             const ColumnWithTypeAndName & scale_column = block.getByPosition(arguments[1]);
             UInt32 scale = extractToDecimalScale(scale_column);
-            vec_to.setScale(scale);
+            col_to = ColVecTo::create(size, scale);
             ToDataType check_bounds_in_ctor(ToDataType::maxPrecision(), scale);
         }
+        else
+            col_to = ColVecTo::create(size);
+
+        typename ColVecTo::Container & vec_to = col_to->getData();
 
         ColumnUInt8::MutablePtr col_null_map_to;
         ColumnUInt8::Container * vec_null_map_to [[maybe_unused]] = nullptr;

dbms/tests/queries/0_stateless/00700_decimal_in_keys.reference

@@ -0,0 +1,29 @@
+1
+0
+0
+0
+1
+1
+5
+9.00000000	29.00000000	29.00000000
+8.00000000	28.00000000	28.00000000
+7.00000000	27.00000000	27.00000000
+6.00000000	26.00000000	26.00000000
+9.00000000	19.00000000	19.00000000
+8.00000000	18.00000000	18.00000000
+7.00000000	17.00000000	17.00000000
+6.00000000	16.00000000	16.00000000
+9.00000000	9.00000000	9.00000000
+8.00000000	8.00000000	8.00000000
+7.00000000	7.00000000	7.00000000
+6.00000000	6.00000000	6.00000000
+1.00000000	1.00000000	1.00000000
+3.00000000	3.00000000	3.00000000
+1.00000000	11.00000000	11.00000000
+3.00000000	13.00000000	13.00000000
+1.00000000	21.00000000	21.00000000
+3.00000000	23.00000000	23.00000000
+1.00000000	31.00000000	31.00000000
+3.00000000	33.00000000	33.00000000
+1.00000000	41.00000000	41.00000000
+3.00000000	43.00000000	43.00000000

dbms/tests/queries/0_stateless/00700_decimal_in_keys.sql

@@ -12,10 +12,20 @@ ENGINE = MergeTree
 PARTITION BY toInt32(d1)
 ORDER BY (d2, d3);
 
+INSERT INTO test.decimal (d1, d2, d3) VALUES (4.2, 4.2, 4.2);
+
+SELECT count() FROM test.decimal WHERE d1 =  toDecimal32('4.2', 1);
+SELECT count() FROM test.decimal WHERE d1 != toDecimal32('4.2', 2);
+SELECT count() FROM test.decimal WHERE d1 <  toDecimal32('4.2', 3);
+SELECT count() FROM test.decimal WHERE d1 >  toDecimal32('4.2', 4);
+SELECT count() FROM test.decimal WHERE d1 <= toDecimal32('4.2', 5);
+SELECT count() FROM test.decimal WHERE d1 >= toDecimal32('4.2', 6);
+
 INSERT INTO test.decimal (d1, d2, d3)
     SELECT toDecimal32(number % 10, 8), toDecimal64(number, 8), toDecimal128(number, 8) FROM system.numbers LIMIT 50;
 
 SELECT count() FROM test.decimal WHERE d1 = 1;
---SELECT * FROM test.decimal WHERE d1 <= toDecimal32(2, 8) ORDER BY d2 DESC;
+SELECT * FROM test.decimal WHERE d1 > 5 AND d2 < 30 ORDER BY d2 DESC;
+SELECT * FROM test.decimal WHERE d1 IN(1, 3) ORDER BY d2;
 
 DROP TABLE test.decimal;