diff --git a/src/Analyzer/Passes/GroupingFunctionsResolvePass.cpp b/src/Analyzer/Passes/GroupingFunctionsResolvePass.cpp
index fc122730b37..d7ccc53041b 100644
--- a/src/Analyzer/Passes/GroupingFunctionsResolvePass.cpp
+++ b/src/Analyzer/Passes/GroupingFunctionsResolvePass.cpp
@@ -4,6 +4,7 @@
 #include <Core/Settings.h>
 
 #include <Functions/grouping.h>
+#include <Functions/IFunctionAdaptors.h>
 
 #include <Interpreters/Context.h>
 
diff --git a/src/Analyzer/Resolve/QueryAnalyzer.cpp b/src/Analyzer/Resolve/QueryAnalyzer.cpp
index 03ebd893c47..d118cb281ae 100644
--- a/src/Analyzer/Resolve/QueryAnalyzer.cpp
+++ b/src/Analyzer/Resolve/QueryAnalyzer.cpp
@@ -13,6 +13,7 @@
 #include <DataTypes/getLeastSupertype.h>
 
 #include <Functions/FunctionFactory.h>
+#include <Functions/IFunctionAdaptors.h>
 #include <Functions/UserDefined/UserDefinedExecutableFunctionFactory.h>
 #include <Functions/UserDefined/UserDefinedSQLFunctionFactory.h>
 #include <Functions/grouping.h>
diff --git a/src/Columns/ColumnDecimal.cpp b/src/Columns/ColumnDecimal.cpp
index bb4433f8956..73366150e7d 100644
--- a/src/Columns/ColumnDecimal.cpp
+++ b/src/Columns/ColumnDecimal.cpp
@@ -8,6 +8,10 @@
 #include <Common/assert_cast.h>
 #include <Common/iota.h>
 
+#include <Core/DecimalFunctions.h>
+#include <Core/TypeId.h>
+
+#include <base/TypeName.h>
 #include <base/sort.h>
 
 #include <IO/WriteHelpers.h>
@@ -30,6 +34,19 @@ namespace ErrorCodes
     extern const int NOT_IMPLEMENTED;
 }
 
+template <is_decimal T>
+const char * ColumnDecimal<T>::getFamilyName() const
+{
+    return TypeName<T>.data();
+}
+
+template <is_decimal T>
+TypeIndex ColumnDecimal<T>::getDataType() const
+{
+    return TypeToTypeIndex<T>;
+}
+
+
 template <is_decimal T>
 #if !defined(DEBUG_OR_SANITIZER_BUILD)
 int ColumnDecimal<T>::compareAt(size_t n, size_t m, const IColumn & rhs_, int) const
@@ -46,6 +63,12 @@ int ColumnDecimal<T>::doCompareAt(size_t n, size_t m, const IColumn & rhs_, int)
     return decimalLess<T>(b, a, other.scale, scale) ? 1 : (decimalLess<T>(a, b, scale, other.scale) ? -1 : 0);
 }
 
+template <is_decimal T>
+Float64 ColumnDecimal<T>::getFloat64(size_t n) const
+{
+    return DecimalUtils::convertTo<Float64>(data[n], scale);
+}
+
 template <is_decimal T>
 const char * ColumnDecimal<T>::deserializeAndInsertFromArena(const char * pos)
 {
diff --git a/src/Columns/ColumnDecimal.h b/src/Columns/ColumnDecimal.h
index 6f8360a54dd..690549e4a56 100644
--- a/src/Columns/ColumnDecimal.h
+++ b/src/Columns/ColumnDecimal.h
@@ -1,14 +1,9 @@
 #pragma once
 
-#include <base/sort.h>
-#include <base/TypeName.h>
-#include <Core/Field.h>
-#include <Core/DecimalFunctions.h>
-#include <Core/TypeId.h>
-#include <Common/typeid_cast.h>
 #include <Columns/ColumnFixedSizeHelper.h>
 #include <Columns/IColumn.h>
 #include <Columns/IColumnImpl.h>
+#include <Core/Field.h>
 
 
 namespace DB
@@ -39,8 +34,8 @@ private:
     {}
 
 public:
-    const char * getFamilyName() const override { return TypeName<T>.data(); }
-    TypeIndex getDataType() const override { return TypeToTypeIndex<T>; }
+    const char * getFamilyName() const override;
+    TypeIndex getDataType() const override;
 
     bool isNumeric() const override { return false; }
     bool canBeInsideNullable() const override { return true; }
@@ -98,7 +93,7 @@ public:
         return StringRef(reinterpret_cast<const char *>(&data[n]), sizeof(data[n]));
     }
 
-    Float64 getFloat64(size_t n) const final { return DecimalUtils::convertTo<Float64>(data[n], scale); }
+    Float64 getFloat64(size_t n) const final;
 
     const char * deserializeAndInsertFromArena(const char * pos) override;
     const char * skipSerializedInArena(const char * pos) const override;
diff --git a/src/Columns/ColumnFunction.cpp b/src/Columns/ColumnFunction.cpp
index cc80d04444e..5e41e95fdc5 100644
--- a/src/Columns/ColumnFunction.cpp
+++ b/src/Columns/ColumnFunction.cpp
@@ -347,7 +347,7 @@ ColumnWithTypeAndName ColumnFunction::reduce() const
     if (is_function_compiled)
         ProfileEvents::increment(ProfileEvents::CompiledFunctionExecute);
 
-    res.column = function->execute(columns, res.type, elements_size);
+    res.column = function->execute(columns, res.type, elements_size, /* dry_run = */ false);
     if (res.column->getDataType() != res.type->getColumnType())
         throw Exception(
             ErrorCodes::LOGICAL_ERROR,
diff --git a/src/Columns/ColumnVector.cpp b/src/Columns/ColumnVector.cpp
index 84fc6ebc61d..3c7727f37c4 100644
--- a/src/Columns/ColumnVector.cpp
+++ b/src/Columns/ColumnVector.cpp
@@ -32,6 +32,8 @@
 #    include <emmintrin.h>
 #endif
 
+#include "config.h"
+
 #if USE_MULTITARGET_CODE
 #    include <immintrin.h>
 #endif
@@ -658,7 +660,7 @@ inline void doFilterAligned(const UInt8 *& filt_pos, const UInt8 *& filt_end_ali
                             reinterpret_cast<void *>(&res_data[current_offset]), mask & KMASK);
                     current_offset += std::popcount(mask & KMASK);
                     /// prepare mask for next iter, if ELEMENTS_PER_VEC = 64, no next iter
-                    if (ELEMENTS_PER_VEC < 64)
+                    if constexpr (ELEMENTS_PER_VEC < 64)
                     {
                         mask >>= ELEMENTS_PER_VEC;
                     }
@@ -992,6 +994,151 @@ ColumnPtr ColumnVector<T>::createWithOffsets(const IColumn::Offsets & offsets, c
     return res;
 }
 
+DECLARE_DEFAULT_CODE(
+    template <typename Container, typename Type> void vectorIndexImpl(
+    const Container & data, const PaddedPODArray<Type> & indexes, size_t limit, Container & res_data)
+    {
+        for (size_t i = 0; i < limit; ++i)
+            res_data[i] = data[indexes[i]];
+    }
+);
+
+DECLARE_AVX512VBMI_SPECIFIC_CODE(
+    template <typename Container, typename Type>
+    void vectorIndexImpl(const Container & data, const PaddedPODArray<Type> & indexes, size_t limit, Container & res_data)
+    {
+        static constexpr UInt64 MASK64 = 0xffffffffffffffff;
+        const size_t limit64 = limit & ~63;
+        size_t pos = 0;
+        size_t data_size = data.size();
+
+        auto data_pos = reinterpret_cast<const UInt8 *>(data.data());
+        auto indexes_pos = reinterpret_cast<const UInt8 *>(indexes.data());
+        auto res_pos = reinterpret_cast<UInt8 *>(res_data.data());
+
+        if (limit == 0)
+            return; /// nothing to do, just return
+
+        if (data_size <= 64)
+        {
+            /// one single mask load for table size <= 64
+            __mmask64 last_mask = MASK64 >> (64 - data_size);
+            __m512i table1 = _mm512_maskz_loadu_epi8(last_mask, data_pos);
+
+            /// 64 bytes table lookup using one single permutexvar_epi8
+            while (pos < limit64)
+            {
+                __m512i vidx = _mm512_loadu_epi8(indexes_pos + pos);
+                __m512i out = _mm512_permutexvar_epi8(vidx, table1);
+                _mm512_storeu_epi8(res_pos + pos, out);
+                pos += 64;
+            }
+            /// tail handling
+            if (limit > limit64)
+            {
+                __mmask64 tail_mask = MASK64 >> (limit64 + 64 - limit);
+                __m512i vidx = _mm512_maskz_loadu_epi8(tail_mask, indexes_pos + pos);
+                __m512i out = _mm512_permutexvar_epi8(vidx, table1);
+                _mm512_mask_storeu_epi8(res_pos + pos, tail_mask, out);
+            }
+        }
+        else if (data_size <= 128)
+        {
+            /// table size (64, 128] requires 2 zmm load
+            __mmask64 last_mask = MASK64 >> (128 - data_size);
+            __m512i table1 = _mm512_loadu_epi8(data_pos);
+            __m512i table2 = _mm512_maskz_loadu_epi8(last_mask, data_pos + 64);
+
+            /// 128 bytes table lookup using one single permute2xvar_epi8
+            while (pos < limit64)
+            {
+                __m512i vidx = _mm512_loadu_epi8(indexes_pos + pos);
+                __m512i out = _mm512_permutex2var_epi8(table1, vidx, table2);
+                _mm512_storeu_epi8(res_pos + pos, out);
+                pos += 64;
+            }
+            if (limit > limit64)
+            {
+                __mmask64 tail_mask = MASK64 >> (limit64 + 64 - limit);
+                __m512i vidx = _mm512_maskz_loadu_epi8(tail_mask, indexes_pos + pos);
+                __m512i out = _mm512_permutex2var_epi8(table1, vidx, table2);
+                _mm512_mask_storeu_epi8(res_pos + pos, tail_mask, out);
+            }
+        }
+        else
+        {
+            if (data_size > 256)
+            {
+                /// byte index will not exceed 256 boundary.
+                data_size = 256;
+            }
+
+            __m512i table1 = _mm512_loadu_epi8(data_pos);
+            __m512i table2 = _mm512_loadu_epi8(data_pos + 64);
+            __m512i table3, table4;
+            if (data_size <= 192)
+            {
+                /// only 3 tables need to load if size <= 192
+                __mmask64 last_mask = MASK64 >> (192 - data_size);
+                table3 = _mm512_maskz_loadu_epi8(last_mask, data_pos + 128);
+                table4 = _mm512_setzero_si512();
+            }
+            else
+            {
+                __mmask64 last_mask = MASK64 >> (256 - data_size);
+                table3 = _mm512_loadu_epi8(data_pos + 128);
+                table4 = _mm512_maskz_loadu_epi8(last_mask, data_pos + 192);
+            }
+
+            /// 256 bytes table lookup can use: 2 permute2xvar_epi8 plus 1 blender with MSB
+            while (pos < limit64)
+            {
+                __m512i vidx = _mm512_loadu_epi8(indexes_pos + pos);
+                __m512i tmp1 = _mm512_permutex2var_epi8(table1, vidx, table2);
+                __m512i tmp2 = _mm512_permutex2var_epi8(table3, vidx, table4);
+                __mmask64 msb = _mm512_movepi8_mask(vidx);
+                __m512i out = _mm512_mask_blend_epi8(msb, tmp1, tmp2);
+                _mm512_storeu_epi8(res_pos + pos, out);
+                pos += 64;
+            }
+            if (limit > limit64)
+            {
+                __mmask64 tail_mask = MASK64 >> (limit64 + 64 - limit);
+                __m512i vidx = _mm512_maskz_loadu_epi8(tail_mask, indexes_pos + pos);
+                __m512i tmp1 = _mm512_permutex2var_epi8(table1, vidx, table2);
+                __m512i tmp2 = _mm512_permutex2var_epi8(table3, vidx, table4);
+                __mmask64 msb = _mm512_movepi8_mask(vidx);
+                __m512i out = _mm512_mask_blend_epi8(msb, tmp1, tmp2);
+                _mm512_mask_storeu_epi8(res_pos + pos, tail_mask, out);
+            }
+        }
+    }
+);
+
+template <typename T>
+template <typename Type>
+ColumnPtr ColumnVector<T>::indexImpl(const PaddedPODArray<Type> & indexes, size_t limit) const
+{
+    chassert(limit <= indexes.size());
+
+    auto res = this->create(limit);
+    typename Self::Container & res_data = res->getData();
+#if USE_MULTITARGET_CODE
+    if constexpr (sizeof(T) == 1 && sizeof(Type) == 1)
+    {
+        /// VBMI optimization only applicable for (U)Int8 types
+        if (isArchSupported(TargetArch::AVX512VBMI))
+        {
+            TargetSpecific::AVX512VBMI::vectorIndexImpl<Container, Type>(data, indexes, limit, res_data);
+            return res;
+        }
+    }
+#endif
+    TargetSpecific::Default::vectorIndexImpl<Container, Type>(data, indexes, limit, res_data);
+
+    return res;
+}
+
 /// Explicit template instantiations - to avoid code bloat in headers.
 template class ColumnVector<UInt8>;
 template class ColumnVector<UInt16>;
@@ -1012,4 +1159,17 @@ template class ColumnVector<UUID>;
 template class ColumnVector<IPv4>;
 template class ColumnVector<IPv6>;
 
+INSTANTIATE_INDEX_TEMPLATE_IMPL(ColumnVector)
+/// Used by ColumnVariant.cpp
+template ColumnPtr ColumnVector<UInt8>::indexImpl<UInt16>(const PaddedPODArray<UInt16> & indexes, size_t limit) const;
+template ColumnPtr ColumnVector<UInt8>::indexImpl<UInt32>(const PaddedPODArray<UInt32> & indexes, size_t limit) const;
+template ColumnPtr ColumnVector<UInt8>::indexImpl<UInt64>(const PaddedPODArray<UInt64> & indexes, size_t limit) const;
+template ColumnPtr ColumnVector<UInt64>::indexImpl<UInt8>(const PaddedPODArray<UInt8> & indexes, size_t limit) const;
+template ColumnPtr ColumnVector<UInt64>::indexImpl<UInt16>(const PaddedPODArray<UInt16> & indexes, size_t limit) const;
+template ColumnPtr ColumnVector<UInt64>::indexImpl<UInt32>(const PaddedPODArray<UInt32> & indexes, size_t limit) const;
+
+#if defined(OS_DARWIN)
+template ColumnPtr ColumnVector<UInt8>::indexImpl<size_t>(const PaddedPODArray<size_t> & indexes, size_t limit) const;
+template ColumnPtr ColumnVector<UInt64>::indexImpl<size_t>(const PaddedPODArray<size_t> & indexes, size_t limit) const;
+#endif
 }
diff --git a/src/Columns/ColumnVector.h b/src/Columns/ColumnVector.h
index e8bb6ad6798..1387cca1ece 100644
--- a/src/Columns/ColumnVector.h
+++ b/src/Columns/ColumnVector.h
@@ -13,10 +13,6 @@
 
 #include "config.h"
 
-#if USE_MULTITARGET_CODE
-#    include <immintrin.h>
-#endif
-
 namespace DB
 {
 
@@ -320,151 +316,6 @@ protected:
     Container data;
 };
 
-DECLARE_DEFAULT_CODE(
-template <typename Container, typename Type>
-inline void vectorIndexImpl(const Container & data, const PaddedPODArray<Type> & indexes, size_t limit, Container & res_data)
-{
-    for (size_t i = 0; i < limit; ++i)
-        res_data[i] = data[indexes[i]];
-}
-);
-
-DECLARE_AVX512VBMI_SPECIFIC_CODE(
-template <typename Container, typename Type>
-inline void vectorIndexImpl(const Container & data, const PaddedPODArray<Type> & indexes, size_t limit, Container & res_data)
-{
-    static constexpr UInt64 MASK64 = 0xffffffffffffffff;
-    const size_t limit64 = limit & ~63;
-    size_t pos = 0;
-    size_t data_size = data.size();
-
-    auto data_pos = reinterpret_cast<const UInt8 *>(data.data());
-    auto indexes_pos = reinterpret_cast<const UInt8 *>(indexes.data());
-    auto res_pos = reinterpret_cast<UInt8 *>(res_data.data());
-
-    if (limit == 0)
-        return;   /// nothing to do, just return
-
-    if (data_size <= 64)
-    {
-        /// one single mask load for table size <= 64
-        __mmask64 last_mask = MASK64 >> (64 - data_size);
-        __m512i table1 = _mm512_maskz_loadu_epi8(last_mask, data_pos);
-
-        /// 64 bytes table lookup using one single permutexvar_epi8
-        while (pos < limit64)
-        {
-            __m512i vidx = _mm512_loadu_epi8(indexes_pos + pos);
-            __m512i out = _mm512_permutexvar_epi8(vidx, table1);
-            _mm512_storeu_epi8(res_pos + pos, out);
-            pos += 64;
-        }
-        /// tail handling
-        if (limit > limit64)
-        {
-            __mmask64 tail_mask = MASK64 >> (limit64 + 64 - limit);
-            __m512i vidx = _mm512_maskz_loadu_epi8(tail_mask, indexes_pos + pos);
-            __m512i out = _mm512_permutexvar_epi8(vidx, table1);
-            _mm512_mask_storeu_epi8(res_pos + pos, tail_mask, out);
-        }
-    }
-    else if (data_size <= 128)
-    {
-        /// table size (64, 128] requires 2 zmm load
-        __mmask64 last_mask = MASK64 >> (128 - data_size);
-        __m512i table1 = _mm512_loadu_epi8(data_pos);
-        __m512i table2 = _mm512_maskz_loadu_epi8(last_mask, data_pos + 64);
-
-        /// 128 bytes table lookup using one single permute2xvar_epi8
-        while (pos < limit64)
-        {
-            __m512i vidx = _mm512_loadu_epi8(indexes_pos + pos);
-            __m512i out = _mm512_permutex2var_epi8(table1, vidx, table2);
-            _mm512_storeu_epi8(res_pos + pos, out);
-            pos += 64;
-        }
-        if (limit > limit64)
-        {
-            __mmask64 tail_mask = MASK64 >> (limit64 + 64 - limit);
-            __m512i vidx = _mm512_maskz_loadu_epi8(tail_mask, indexes_pos + pos);
-            __m512i out = _mm512_permutex2var_epi8(table1, vidx, table2);
-            _mm512_mask_storeu_epi8(res_pos + pos, tail_mask, out);
-        }
-    }
-    else
-    {
-        if (data_size > 256)
-        {
-            /// byte index will not exceed 256 boundary.
-            data_size = 256;
-        }
-
-        __m512i table1 = _mm512_loadu_epi8(data_pos);
-        __m512i table2 = _mm512_loadu_epi8(data_pos + 64);
-        __m512i table3, table4;
-        if (data_size <= 192)
-        {
-            /// only 3 tables need to load if size <= 192
-            __mmask64 last_mask = MASK64 >> (192 - data_size);
-            table3 = _mm512_maskz_loadu_epi8(last_mask, data_pos + 128);
-            table4 = _mm512_setzero_si512();
-        }
-        else
-        {
-            __mmask64 last_mask = MASK64 >> (256 - data_size);
-            table3 = _mm512_loadu_epi8(data_pos + 128);
-            table4 = _mm512_maskz_loadu_epi8(last_mask, data_pos + 192);
-        }
-
-        /// 256 bytes table lookup can use: 2 permute2xvar_epi8 plus 1 blender with MSB
-        while (pos < limit64)
-        {
-            __m512i vidx = _mm512_loadu_epi8(indexes_pos + pos);
-            __m512i tmp1 = _mm512_permutex2var_epi8(table1, vidx, table2);
-            __m512i tmp2 = _mm512_permutex2var_epi8(table3, vidx, table4);
-            __mmask64 msb = _mm512_movepi8_mask(vidx);
-            __m512i out = _mm512_mask_blend_epi8(msb, tmp1, tmp2);
-            _mm512_storeu_epi8(res_pos + pos, out);
-            pos += 64;
-        }
-        if (limit > limit64)
-        {
-            __mmask64 tail_mask = MASK64 >> (limit64 + 64 - limit);
-            __m512i vidx = _mm512_maskz_loadu_epi8(tail_mask, indexes_pos + pos);
-            __m512i tmp1 = _mm512_permutex2var_epi8(table1, vidx, table2);
-            __m512i tmp2 = _mm512_permutex2var_epi8(table3, vidx, table4);
-            __mmask64 msb = _mm512_movepi8_mask(vidx);
-            __m512i out = _mm512_mask_blend_epi8(msb, tmp1, tmp2);
-            _mm512_mask_storeu_epi8(res_pos + pos, tail_mask, out);
-        }
-    }
-}
-);
-
-template <typename T>
-template <typename Type>
-ColumnPtr ColumnVector<T>::indexImpl(const PaddedPODArray<Type> & indexes, size_t limit) const
-{
-    assert(limit <= indexes.size());
-
-    auto res = this->create(limit);
-    typename Self::Container & res_data = res->getData();
-#if USE_MULTITARGET_CODE
-    if constexpr (sizeof(T) == 1 && sizeof(Type) == 1)
-    {
-        /// VBMI optimization only applicable for (U)Int8 types
-        if (isArchSupported(TargetArch::AVX512VBMI))
-        {
-            TargetSpecific::AVX512VBMI::vectorIndexImpl<Container, Type>(data, indexes, limit, res_data);
-            return res;
-        }
-    }
-#endif
-    TargetSpecific::Default::vectorIndexImpl<Container, Type>(data, indexes, limit, res_data);
-
-    return res;
-}
-
 template <class TCol>
 concept is_col_vector = std::is_same_v<TCol, ColumnVector<typename TCol::ValueType>>;
 
diff --git a/src/Columns/ColumnsCommon.h b/src/Columns/ColumnsCommon.h
index 99f1d2da47e..f0d6cff2e35 100644
--- a/src/Columns/ColumnsCommon.h
+++ b/src/Columns/ColumnsCommon.h
@@ -142,4 +142,10 @@ ColumnPtr permuteImpl(const Column & column, const IColumn::Permutation & perm,
     template ColumnPtr Column::indexImpl<UInt16>(const PaddedPODArray<UInt16> & indexes, size_t limit) const; \
     template ColumnPtr Column::indexImpl<UInt32>(const PaddedPODArray<UInt32> & indexes, size_t limit) const; \
     template ColumnPtr Column::indexImpl<UInt64>(const PaddedPODArray<UInt64> & indexes, size_t limit) const;
+
+#define INSTANTIATE_INDEX_TEMPLATE_IMPL(ColumnTemplate) \
+    template ColumnPtr ColumnTemplate<UInt8>::indexImpl<UInt8>(const PaddedPODArray<UInt8> & indexes, size_t limit) const; \
+    template ColumnPtr ColumnTemplate<UInt16>::indexImpl<UInt16>(const PaddedPODArray<UInt16> & indexes, size_t limit) const; \
+    template ColumnPtr ColumnTemplate<UInt32>::indexImpl<UInt32>(const PaddedPODArray<UInt32> & indexes, size_t limit) const; \
+    template ColumnPtr ColumnTemplate<UInt64>::indexImpl<UInt64>(const PaddedPODArray<UInt64> & indexes, size_t limit) const;
 }
diff --git a/src/Core/DecimalComparison.h b/src/Core/DecimalComparison.h
index 77402adf164..4b6783265d0 100644
--- a/src/Core/DecimalComparison.h
+++ b/src/Core/DecimalComparison.h
@@ -2,6 +2,7 @@
 
 #include <base/arithmeticOverflow.h>
 #include <Core/Block.h>
+#include <Core/DecimalFunctions.h>
 #include <Core/AccurateComparison.h>
 #include <Core/callOnTypeIndex.h>
 #include <DataTypes/DataTypesNumber.h>
@@ -52,8 +53,8 @@ struct DecCompareInt
     using TypeB = Type;
 };
 
-template <typename A, typename B, template <typename, typename> typename Operation, bool _check_overflow = true,
-    bool _actual = is_decimal<A> || is_decimal<B>>
+template <typename A, typename B, template <typename, typename> typename Operation>
+requires is_decimal<A> || is_decimal<B>
 class DecimalComparison
 {
 public:
@@ -65,20 +66,17 @@ public:
     using ArrayA = typename ColVecA::Container;
     using ArrayB = typename ColVecB::Container;
 
-    static ColumnPtr apply(const ColumnWithTypeAndName & col_left, const ColumnWithTypeAndName & col_right)
+    static ColumnPtr apply(const ColumnWithTypeAndName & col_left, const ColumnWithTypeAndName & col_right, bool check_overflow)
     {
-        if constexpr (_actual)
-        {
-            ColumnPtr c_res;
-            Shift shift = getScales<A, B>(col_left.type, col_right.type);
+        ColumnPtr c_res;
+        Shift shift = getScales<A, B>(col_left.type, col_right.type);
 
-            return applyWithScale(col_left.column, col_right.column, shift);
-        }
-        else
-            return nullptr;
+        if (check_overflow)
+            return applyWithScale<true>(col_left.column, col_right.column, shift);
+        return applyWithScale<false>(col_left.column, col_right.column, shift);
     }
 
-    static bool compare(A a, B b, UInt32 scale_a, UInt32 scale_b)
+    static bool compare(A a, B b, UInt32 scale_a, UInt32 scale_b, bool check_overflow)
     {
         static const UInt32 max_scale = DecimalUtils::max_precision<Decimal256>;
         if (scale_a > max_scale || scale_b > max_scale)
@@ -90,7 +88,9 @@ public:
         if (scale_a > scale_b)
             shift.b = static_cast<CompareInt>(DecimalUtils::scaleMultiplier<A>(scale_a - scale_b));
 
-        return applyWithScale(a, b, shift);
+        if (check_overflow)
+            return applyWithScale<true>(a, b, shift);
+        return applyWithScale<false>(a, b, shift);
     }
 
 private:
@@ -104,14 +104,14 @@ private:
         bool right() const { return b != 1; }
     };
 
-    template <typename T, typename U>
+    template <bool check_overflow, typename T, typename U>
     static auto applyWithScale(T a, U b, const Shift & shift)
     {
         if (shift.left())
-            return apply<true, false>(a, b, shift.a);
+            return apply<check_overflow, true, false>(a, b, shift.a);
         if (shift.right())
-            return apply<false, true>(a, b, shift.b);
-        return apply<false, false>(a, b, 1);
+            return apply<check_overflow, false, true>(a, b, shift.b);
+        return apply<check_overflow, false, false>(a, b, 1);
     }
 
     template <typename T, typename U>
@@ -125,8 +125,8 @@ private:
         if (decimal0 && decimal1)
         {
             auto result_type = DecimalUtils::binaryOpResult<false, false>(*decimal0, *decimal1);
-            shift.a = static_cast<CompareInt>(result_type.scaleFactorFor(decimal0->getTrait(), false).value);
-            shift.b = static_cast<CompareInt>(result_type.scaleFactorFor(decimal1->getTrait(), false).value);
+            shift.a = static_cast<CompareInt>(result_type.scaleFactorFor(DecimalUtils::DataTypeDecimalTrait<T>{decimal0->getPrecision(), decimal0->getScale()}, false).value);
+            shift.b = static_cast<CompareInt>(result_type.scaleFactorFor(DecimalUtils::DataTypeDecimalTrait<U>{decimal1->getPrecision(), decimal1->getScale()}, false).value);
         }
         else if (decimal0)
             shift.b = static_cast<CompareInt>(decimal0->getScaleMultiplier().value);
@@ -158,66 +158,63 @@ private:
         return shift;
     }
 
-    template <bool scale_left, bool scale_right>
+    template <bool check_overflow, bool scale_left, bool scale_right>
     static ColumnPtr apply(const ColumnPtr & c0, const ColumnPtr & c1, CompareInt scale)
     {
         auto c_res = ColumnUInt8::create();
 
-        if constexpr (_actual)
+        bool c0_is_const = isColumnConst(*c0);
+        bool c1_is_const = isColumnConst(*c1);
+
+        if (c0_is_const && c1_is_const)
         {
-            bool c0_is_const = isColumnConst(*c0);
-            bool c1_is_const = isColumnConst(*c1);
+            const ColumnConst & c0_const = checkAndGetColumnConst<ColVecA>(*c0);
+            const ColumnConst & c1_const = checkAndGetColumnConst<ColVecB>(*c1);
 
-            if (c0_is_const && c1_is_const)
+            A a = c0_const.template getValue<A>();
+            B b = c1_const.template getValue<B>();
+            UInt8 res = apply<check_overflow, scale_left, scale_right>(a, b, scale);
+            return DataTypeUInt8().createColumnConst(c0->size(), toField(res));
+        }
+
+        ColumnUInt8::Container & vec_res = c_res->getData();
+        vec_res.resize(c0->size());
+
+        if (c0_is_const)
+        {
+            const ColumnConst & c0_const = checkAndGetColumnConst<ColVecA>(*c0);
+            A a = c0_const.template getValue<A>();
+            if (const ColVecB * c1_vec = checkAndGetColumn<ColVecB>(c1.get()))
+                constantVector<check_overflow, scale_left, scale_right>(a, c1_vec->getData(), vec_res, scale);
+            else
+                throw Exception(ErrorCodes::LOGICAL_ERROR, "Wrong column in Decimal comparison");
+        }
+        else if (c1_is_const)
+        {
+            const ColumnConst & c1_const = checkAndGetColumnConst<ColVecB>(*c1);
+            B b = c1_const.template getValue<B>();
+            if (const ColVecA * c0_vec = checkAndGetColumn<ColVecA>(c0.get()))
+                vectorConstant<check_overflow, scale_left, scale_right>(c0_vec->getData(), b, vec_res, scale);
+            else
+                throw Exception(ErrorCodes::LOGICAL_ERROR, "Wrong column in Decimal comparison");
+        }
+        else
+        {
+            if (const ColVecA * c0_vec = checkAndGetColumn<ColVecA>(c0.get()))
             {
-                const ColumnConst & c0_const = checkAndGetColumnConst<ColVecA>(*c0);
-                const ColumnConst & c1_const = checkAndGetColumnConst<ColVecB>(*c1);
-
-                A a = c0_const.template getValue<A>();
-                B b = c1_const.template getValue<B>();
-                UInt8 res = apply<scale_left, scale_right>(a, b, scale);
-                return DataTypeUInt8().createColumnConst(c0->size(), toField(res));
-            }
-
-            ColumnUInt8::Container & vec_res = c_res->getData();
-            vec_res.resize(c0->size());
-
-            if (c0_is_const)
-            {
-                const ColumnConst & c0_const = checkAndGetColumnConst<ColVecA>(*c0);
-                A a = c0_const.template getValue<A>();
                 if (const ColVecB * c1_vec = checkAndGetColumn<ColVecB>(c1.get()))
-                    constantVector<scale_left, scale_right>(a, c1_vec->getData(), vec_res, scale);
-                else
-                    throw Exception(ErrorCodes::LOGICAL_ERROR, "Wrong column in Decimal comparison");
-            }
-            else if (c1_is_const)
-            {
-                const ColumnConst & c1_const = checkAndGetColumnConst<ColVecB>(*c1);
-                B b = c1_const.template getValue<B>();
-                if (const ColVecA * c0_vec = checkAndGetColumn<ColVecA>(c0.get()))
-                    vectorConstant<scale_left, scale_right>(c0_vec->getData(), b, vec_res, scale);
+                    vectorVector<check_overflow, scale_left, scale_right>(c0_vec->getData(), c1_vec->getData(), vec_res, scale);
                 else
                     throw Exception(ErrorCodes::LOGICAL_ERROR, "Wrong column in Decimal comparison");
             }
             else
-            {
-                if (const ColVecA * c0_vec = checkAndGetColumn<ColVecA>(c0.get()))
-                {
-                    if (const ColVecB * c1_vec = checkAndGetColumn<ColVecB>(c1.get()))
-                        vectorVector<scale_left, scale_right>(c0_vec->getData(), c1_vec->getData(), vec_res, scale);
-                    else
-                        throw Exception(ErrorCodes::LOGICAL_ERROR, "Wrong column in Decimal comparison");
-                }
-                else
-                    throw Exception(ErrorCodes::LOGICAL_ERROR, "Wrong column in Decimal comparison");
-            }
+                throw Exception(ErrorCodes::LOGICAL_ERROR, "Wrong column in Decimal comparison");
         }
 
         return c_res;
     }
 
-    template <bool scale_left, bool scale_right>
+    template <bool check_overflow, bool scale_left, bool scale_right>
     static NO_INLINE UInt8 apply(A a, B b, CompareInt scale [[maybe_unused]])
     {
         CompareInt x;
@@ -232,7 +229,7 @@ private:
         else
             y = static_cast<CompareInt>(b);
 
-        if constexpr (_check_overflow)
+        if constexpr (check_overflow)
         {
             bool overflow = false;
 
@@ -264,9 +261,8 @@ private:
         return Op::apply(x, y);
     }
 
-    template <bool scale_left, bool scale_right>
-    static void NO_INLINE vectorVector(const ArrayA & a, const ArrayB & b, PaddedPODArray<UInt8> & c,
-                                        CompareInt scale)
+    template <bool check_overflow, bool scale_left, bool scale_right>
+    static void NO_INLINE vectorVector(const ArrayA & a, const ArrayB & b, PaddedPODArray<UInt8> & c, CompareInt scale)
     {
         size_t size = a.size();
         const A * a_pos = a.data();
@@ -276,14 +272,14 @@ private:
 
         while (a_pos < a_end)
         {
-            *c_pos = apply<scale_left, scale_right>(*a_pos, *b_pos, scale);
+            *c_pos = apply<check_overflow, scale_left, scale_right>(*a_pos, *b_pos, scale);
             ++a_pos;
             ++b_pos;
             ++c_pos;
         }
     }
 
-    template <bool scale_left, bool scale_right>
+    template <bool check_overflow, bool scale_left, bool scale_right>
     static void NO_INLINE vectorConstant(const ArrayA & a, B b, PaddedPODArray<UInt8> & c, CompareInt scale)
     {
         size_t size = a.size();
@@ -293,13 +289,13 @@ private:
 
         while (a_pos < a_end)
         {
-            *c_pos = apply<scale_left, scale_right>(*a_pos, b, scale);
+            *c_pos = apply<check_overflow, scale_left, scale_right>(*a_pos, b, scale);
             ++a_pos;
             ++c_pos;
         }
     }
 
-    template <bool scale_left, bool scale_right>
+    template <bool check_overflow, bool scale_left, bool scale_right>
     static void NO_INLINE constantVector(A a, const ArrayB & b, PaddedPODArray<UInt8> & c, CompareInt scale)
     {
         size_t size = b.size();
@@ -309,7 +305,7 @@ private:
 
         while (b_pos < b_end)
         {
-            *c_pos = apply<scale_left, scale_right>(a, *b_pos, scale);
+            *c_pos = apply<check_overflow, scale_left, scale_right>(a, *b_pos, scale);
             ++b_pos;
             ++c_pos;
         }
diff --git a/src/Core/Field.cpp b/src/Core/Field.cpp
index e774a95e19f..90f30b52c0c 100644
--- a/src/Core/Field.cpp
+++ b/src/Core/Field.cpp
@@ -529,22 +529,25 @@ Field Field::restoreFromDump(std::string_view dump_)
 template <typename T>
 bool decimalEqual(T x, T y, UInt32 x_scale, UInt32 y_scale)
 {
+    bool check_overflow = true;
     using Comparator = DecimalComparison<T, T, EqualsOp>;
-    return Comparator::compare(x, y, x_scale, y_scale);
+    return Comparator::compare(x, y, x_scale, y_scale, check_overflow);
 }
 
 template <typename T>
 bool decimalLess(T x, T y, UInt32 x_scale, UInt32 y_scale)
 {
+    bool check_overflow = true;
     using Comparator = DecimalComparison<T, T, LessOp>;
-    return Comparator::compare(x, y, x_scale, y_scale);
+    return Comparator::compare(x, y, x_scale, y_scale, check_overflow);
 }
 
 template <typename T>
 bool decimalLessOrEqual(T x, T y, UInt32 x_scale, UInt32 y_scale)
 {
+    bool check_overflow = true;
     using Comparator = DecimalComparison<T, T, LessOrEqualsOp>;
-    return Comparator::compare(x, y, x_scale, y_scale);
+    return Comparator::compare(x, y, x_scale, y_scale, check_overflow);
 }
 
 
diff --git a/src/Core/Field.h b/src/Core/Field.h
index c08d5c9eb42..5a6ee9cdf29 100644
--- a/src/Core/Field.h
+++ b/src/Core/Field.h
@@ -863,6 +863,9 @@ template <> struct Field::EnumToType<Field::Types::AggregateFunctionState> { usi
 template <> struct Field::EnumToType<Field::Types::CustomType> { using Type = CustomType; };
 template <> struct Field::EnumToType<Field::Types::Bool> { using Type = UInt64; };
 
+/// Use it to prevent inclusion of magic_enum in headers, which is very expensive for the compiler
+std::string_view fieldTypeToString(Field::Types::Which type);
+
 constexpr bool isInt64OrUInt64FieldType(Field::Types::Which t)
 {
     return t == Field::Types::Int64
@@ -886,7 +889,7 @@ auto & Field::safeGet()
     if (target != which &&
         !(which == Field::Types::Bool && (target == Field::Types::UInt64 || target == Field::Types::Int64)) &&
         !(isInt64OrUInt64FieldType(which) && isInt64OrUInt64FieldType(target)))
-            throw Exception(ErrorCodes::BAD_GET, "Bad get: has {}, requested {}", getTypeName(), target);
+        throw Exception(ErrorCodes::BAD_GET, "Bad get: has {}, requested {}", getTypeName(), fieldTypeToString(target));
 
     return get<T>();
 }
@@ -1002,8 +1005,6 @@ void readQuoted(DecimalField<T> & x, ReadBuffer & buf);
 void writeFieldText(const Field & x, WriteBuffer & buf);
 
 String toString(const Field & x);
-
-std::string_view fieldTypeToString(Field::Types::Which type);
 }
 
 template <>
diff --git a/src/Core/callOnTypeIndex.h b/src/Core/callOnTypeIndex.h
index 0c8f2201b0d..09fbc7f1f10 100644
--- a/src/Core/callOnTypeIndex.h
+++ b/src/Core/callOnTypeIndex.h
@@ -87,6 +87,77 @@ static bool callOnBasicType(TypeIndex number, F && f)
     return false;
 }
 
+
+template <typename T, bool _int, bool _float, bool _decimal, bool _datetime, typename F>
+static bool callOnBasicTypeSecondArg(TypeIndex number, F && f)
+{
+    if constexpr (_int)
+    {
+        switch (number)
+        {
+            case TypeIndex::UInt8:        return f(TypePair<UInt8, T>());
+            case TypeIndex::UInt16:       return f(TypePair<UInt16, T>());
+            case TypeIndex::UInt32:       return f(TypePair<UInt32, T>());
+            case TypeIndex::UInt64:       return f(TypePair<UInt64, T>());
+            case TypeIndex::UInt128:      return f(TypePair<UInt128, T>());
+            case TypeIndex::UInt256:      return f(TypePair<UInt256, T>());
+
+            case TypeIndex::Int8:         return f(TypePair<Int8, T>());
+            case TypeIndex::Int16:        return f(TypePair<Int16, T>());
+            case TypeIndex::Int32:        return f(TypePair<Int32, T>());
+            case TypeIndex::Int64:        return f(TypePair<Int64, T>());
+            case TypeIndex::Int128:       return f(TypePair<Int128, T>());
+            case TypeIndex::Int256:       return f(TypePair<Int256, T>());
+
+            case TypeIndex::Enum8:        return f(TypePair<Int8, T>());
+            case TypeIndex::Enum16:       return f(TypePair<Int16, T>());
+
+            default:
+                break;
+        }
+    }
+
+    if constexpr (_decimal)
+    {
+        switch (number)
+        {
+            case TypeIndex::Decimal32:    return f(TypePair<Decimal32, T>());
+            case TypeIndex::Decimal64:    return f(TypePair<Decimal64, T>());
+            case TypeIndex::Decimal128:   return f(TypePair<Decimal128, T>());
+            case TypeIndex::Decimal256:   return f(TypePair<Decimal256, T>());
+            default:
+                break;
+        }
+    }
+
+    if constexpr (_float)
+    {
+        switch (number)
+        {
+            case TypeIndex::BFloat16:     return f(TypePair<BFloat16, T>());
+            case TypeIndex::Float32:      return f(TypePair<Float32, T>());
+            case TypeIndex::Float64:      return f(TypePair<Float64, T>());
+            default:
+                break;
+        }
+    }
+
+    if constexpr (_datetime)
+    {
+        switch (number)
+        {
+            case TypeIndex::Date:         return f(TypePair<UInt16, T>());
+            case TypeIndex::Date32:       return f(TypePair<Int32, T>());
+            case TypeIndex::DateTime:     return f(TypePair<UInt32, T>());
+            case TypeIndex::DateTime64:   return f(TypePair<DateTime64, T>());
+            default:
+                break;
+        }
+    }
+
+    return false;
+}
+
 /// Unroll template using TypeIndex
 template <bool _int, bool _float, bool _decimal, bool _datetime, typename F>
 static inline bool callOnBasicTypes(TypeIndex type_num1, TypeIndex type_num2, F && f)
diff --git a/src/DataTypes/DataTypeDecimalBase.cpp b/src/DataTypes/DataTypeDecimalBase.cpp
index 68bfba475d6..423ab2e4765 100644
--- a/src/DataTypes/DataTypeDecimalBase.cpp
+++ b/src/DataTypes/DataTypeDecimalBase.cpp
@@ -1,7 +1,8 @@
+#include <type_traits>
+#include <Core/DecimalFunctions.h>
 #include <Core/Settings.h>
 #include <DataTypes/DataTypeDecimalBase.h>
 #include <Interpreters/Context.h>
-#include <type_traits>
 
 namespace DB
 {
@@ -14,6 +15,12 @@ namespace ErrorCodes
 {
 }
 
+template <is_decimal T>
+constexpr size_t DataTypeDecimalBase<T>::maxPrecision()
+{
+    return DecimalUtils::max_precision<T>;
+}
+
 bool decimalCheckComparisonOverflow(ContextPtr context)
 {
     return context->getSettingsRef()[Setting::decimal_check_overflow];
@@ -41,6 +48,18 @@ T DataTypeDecimalBase<T>::getScaleMultiplier(UInt32 scale_)
     return DecimalUtils::scaleMultiplier<typename T::NativeType>(scale_);
 }
 
+template <is_decimal T>
+T DataTypeDecimalBase<T>::wholePart(T x) const
+{
+    return DecimalUtils::getWholePart(x, scale);
+}
+
+template <is_decimal T>
+T DataTypeDecimalBase<T>::fractionalPart(T x) const
+{
+    return DecimalUtils::getFractionalPart(x, scale);
+}
+
 
 /// Explicit template instantiations.
 template class DataTypeDecimalBase<Decimal32>;
diff --git a/src/DataTypes/DataTypeDecimalBase.h b/src/DataTypes/DataTypeDecimalBase.h
index c1e1d27557f..beba3c42616 100644
--- a/src/DataTypes/DataTypeDecimalBase.h
+++ b/src/DataTypes/DataTypeDecimalBase.h
@@ -3,11 +3,10 @@
 #include <cmath>
 #include <type_traits>
 
-#include <Core/TypeId.h>
-#include <Core/DecimalFunctions.h>
 #include <Columns/ColumnDecimal.h>
-#include <DataTypes/IDataType.h>
+#include <Core/TypeId.h>
 #include <DataTypes/DataTypesNumber.h>
+#include <DataTypes/IDataType.h>
 #include <Interpreters/Context_fwd.h>
 
 
@@ -64,7 +63,7 @@ public:
 
     static constexpr bool is_parametric = true;
 
-    static constexpr size_t maxPrecision() { return DecimalUtils::max_precision<T>; }
+    static constexpr size_t maxPrecision();
 
     DataTypeDecimalBase(UInt32 precision_, UInt32 scale_)
     :   precision(precision_),
@@ -104,15 +103,8 @@ public:
     UInt32 getScale() const { return scale; }
     T getScaleMultiplier() const { return getScaleMultiplier(scale); }
 
-    T wholePart(T x) const
-    {
-        return DecimalUtils::getWholePart(x, scale);
-    }
-
-    T fractionalPart(T x) const
-    {
-        return DecimalUtils::getFractionalPart(x, scale);
-    }
+    T wholePart(T x) const;
+    T fractionalPart(T x) const;
 
     T maxWholeValue() const { return getScaleMultiplier(precision - scale) - T(1); }
 
@@ -147,11 +139,6 @@ public:
 
     static T getScaleMultiplier(UInt32 scale);
 
-    DecimalUtils::DataTypeDecimalTrait<T> getTrait() const
-    {
-        return {precision, scale};
-    }
-
 protected:
     const UInt32 precision;
     const UInt32 scale;
@@ -167,50 +154,35 @@ inline const DataTypeDecimalBase<T> * checkDecimalBase(const IDataType & data_ty
     return nullptr;
 }
 
-template <bool is_multiply, bool is_division, typename T, typename U, template <typename> typename DecimalType>
-inline auto decimalResultType(const DecimalType<T> & tx, const DecimalType<U> & ty)
-{
-    const auto result_trait = DecimalUtils::binaryOpResult<is_multiply, is_division>(tx, ty);
-    return DecimalType<typename decltype(result_trait)::FieldType>(result_trait.precision, result_trait.scale);
-}
+template <> constexpr size_t DataTypeDecimalBase<Decimal32>::maxPrecision() { return 9; };
+template <> constexpr size_t DataTypeDecimalBase<Decimal64>::maxPrecision() { return 18; };
+template <> constexpr size_t DataTypeDecimalBase<DateTime64>::maxPrecision() { return 18; };
+template <> constexpr size_t DataTypeDecimalBase<Decimal128>::maxPrecision() { return 38; };
+template <> constexpr size_t DataTypeDecimalBase<Decimal256>::maxPrecision() { return 76; };
 
-template <bool is_multiply, bool is_division, typename T, typename U, template <typename> typename DecimalType>
-inline DecimalType<T> decimalResultType(const DecimalType<T> & tx, const DataTypeNumber<U> & ty)
-{
-    const auto result_trait = DecimalUtils::binaryOpResult<is_multiply, is_division>(tx, ty);
-    return DecimalType<typename decltype(result_trait)::FieldType>(result_trait.precision, result_trait.scale);
-}
-
-template <bool is_multiply, bool is_division, typename T, typename U, template <typename> typename DecimalType>
-inline DecimalType<U> decimalResultType(const DataTypeNumber<T> & tx, const DecimalType<U> & ty)
-{
-    const auto result_trait = DecimalUtils::binaryOpResult<is_multiply, is_division>(tx, ty);
-    return DecimalType<typename decltype(result_trait)::FieldType>(result_trait.precision, result_trait.scale);
-}
+extern template class DataTypeDecimalBase<Decimal32>;
+extern template class DataTypeDecimalBase<Decimal64>;
+extern template class DataTypeDecimalBase<DateTime64>;
+extern template class DataTypeDecimalBase<Decimal128>;
+extern template class DataTypeDecimalBase<Decimal256>;
 
 template <template <typename> typename DecimalType>
 inline DataTypePtr createDecimal(UInt64 precision_value, UInt64 scale_value)
 {
-    if (precision_value < DecimalUtils::min_precision || precision_value > DecimalUtils::max_precision<Decimal256>)
+    if (precision_value < 1 || precision_value > DataTypeDecimalBase<Decimal256>::maxPrecision())
         throw Exception(ErrorCodes::ARGUMENT_OUT_OF_BOUND, "Wrong precision: it must be between {} and {}, got {}",
-                        DecimalUtils::min_precision, DecimalUtils::max_precision<Decimal256>, precision_value);
+                        1, DataTypeDecimalBase<Decimal256>::maxPrecision(), precision_value);
 
     if (scale_value > precision_value)
         throw Exception(ErrorCodes::ARGUMENT_OUT_OF_BOUND, "Negative scales and scales larger than precision are not supported");
 
-    if (precision_value <= DecimalUtils::max_precision<Decimal32>)
+    if (precision_value <= DataTypeDecimalBase<Decimal32>::maxPrecision())
         return std::make_shared<DecimalType<Decimal32>>(precision_value, scale_value);
-    if (precision_value <= DecimalUtils::max_precision<Decimal64>)
+    if (precision_value <= DataTypeDecimalBase<Decimal64>::maxPrecision())
         return std::make_shared<DecimalType<Decimal64>>(precision_value, scale_value);
-    if (precision_value <= DecimalUtils::max_precision<Decimal128>)
+    if (precision_value <= DataTypeDecimalBase<Decimal128>::maxPrecision())
         return std::make_shared<DecimalType<Decimal128>>(precision_value, scale_value);
     return std::make_shared<DecimalType<Decimal256>>(precision_value, scale_value);
 }
 
-extern template class DataTypeDecimalBase<Decimal32>;
-extern template class DataTypeDecimalBase<Decimal64>;
-extern template class DataTypeDecimalBase<Decimal128>;
-extern template class DataTypeDecimalBase<Decimal256>;
-extern template class DataTypeDecimalBase<DateTime64>;
-
 }
diff --git a/src/DataTypes/convertMySQLDataType.cpp b/src/DataTypes/convertMySQLDataType.cpp
index fa91e86b43d..2037cfee9c6 100644
--- a/src/DataTypes/convertMySQLDataType.cpp
+++ b/src/DataTypes/convertMySQLDataType.cpp
@@ -111,13 +111,13 @@ DataTypePtr convertMySQLDataType(MultiEnum<MySQLDataTypesSupport> type_support,
     }
     else if (type_support.isSet(MySQLDataTypesSupport::DECIMAL) && (type_name == "numeric" || type_name == "decimal"))
     {
-        if (precision <= DecimalUtils::max_precision<Decimal32>)
+        if (precision <=  DataTypeDecimalBase<Decimal32>::maxPrecision())
             res = std::make_shared<DataTypeDecimal<Decimal32>>(precision, scale);
-        else if (precision <= DecimalUtils::max_precision<Decimal64>)
+        else if (precision <= DataTypeDecimalBase<Decimal64>::maxPrecision())
             res = std::make_shared<DataTypeDecimal<Decimal64>>(precision, scale);
-        else if (precision <= DecimalUtils::max_precision<Decimal128>)
+        else if (precision <= DataTypeDecimalBase<Decimal128>::maxPrecision())
             res = std::make_shared<DataTypeDecimal<Decimal128>>(precision, scale);
-        else if (precision <= DecimalUtils::max_precision<Decimal256>)
+        else if (precision <= DataTypeDecimalBase<Decimal256>::maxPrecision())
             res = std::make_shared<DataTypeDecimal<Decimal256>>(precision, scale);
     }
     else if (type_name == "point")
diff --git a/src/Dictionaries/getDictionaryConfigurationFromAST.cpp b/src/Dictionaries/getDictionaryConfigurationFromAST.cpp
index 4ec2e1f5260..241f2b51705 100644
--- a/src/Dictionaries/getDictionaryConfigurationFromAST.cpp
+++ b/src/Dictionaries/getDictionaryConfigurationFromAST.cpp
@@ -493,7 +493,7 @@ void buildConfigurationFromFunctionWithKeyValueArguments(
             /// We assume that function will not take arguments and will return constant value like tcpPort or hostName
             /// Such functions will return column with size equal to input_rows_count.
             size_t input_rows_count = 1;
-            auto result = function->execute({}, function->getResultType(), input_rows_count);
+            auto result = function->execute({}, function->getResultType(), input_rows_count, /* dry_run = */ false);
 
             Field value;
             result->get(0, value);
diff --git a/src/Formats/ProtobufSerializer.cpp b/src/Formats/ProtobufSerializer.cpp
index ca849136495..c44f8a56a3c 100644
--- a/src/Formats/ProtobufSerializer.cpp
+++ b/src/Formats/ProtobufSerializer.cpp
@@ -1308,7 +1308,8 @@ namespace
                         {
                             if (decimal.value == 0)
                                 writeInt(0);
-                            else if (DecimalComparison<DecimalType, int, EqualsOp>::compare(decimal, 1, scale, 0))
+                            else if (DecimalComparison<DecimalType, int, EqualsOp>::compare(
+                                         decimal, 1, scale, 0, /* check overflow */ true))
                                 writeInt(1);
                             else
                             {
diff --git a/src/Functions/CMakeLists.txt b/src/Functions/CMakeLists.txt
index 19ddadff401..3a57ce42412 100644
--- a/src/Functions/CMakeLists.txt
+++ b/src/Functions/CMakeLists.txt
@@ -7,6 +7,7 @@ add_headers_and_sources(clickhouse_functions .)
 # This allows less dependency and linker work (specially important when building many example executables)
 set(DBMS_FUNCTIONS
     IFunction.cpp # IFunctionOverloadResolver::getLambdaArgumentTypes, IExecutableFunction::execute... (Many AST visitors, analyzer passes, some storages...)
+    IFunctionAdaptors.cpp # FunctionToFunctionBaseAdaptor (Used by FunctionFactory.cpp)
     FunctionDynamicAdaptor.cpp # IFunctionOverloadResolver::getLambdaArgumentTypes, IExecutableFunction::execute... (Many AST visitors, analyzer passes, some storages...)
     FunctionFactory.cpp # FunctionFactory::instance() (Many AST visitors, analyzer passes, some storages...)
     FunctionHelpers.cpp # convertConstTupleToConstantElements, checkAndGetColumnConstStringOrFixedString, checkAndGetNestedArrayOffset ...)
diff --git a/src/Functions/FunctionBinaryArithmetic.h b/src/Functions/FunctionBinaryArithmetic.h
index e7f0d8d43b0..91adf73b679 100644
--- a/src/Functions/FunctionBinaryArithmetic.h
+++ b/src/Functions/FunctionBinaryArithmetic.h
@@ -41,6 +41,7 @@
 #include <Functions/FunctionFactory.h>
 #include <Functions/FunctionHelpers.h>
 #include <Functions/IFunction.h>
+#include <Functions/IFunctionAdaptors.h>
 #include <Functions/IsOperation.h>
 #include <Functions/castTypeToEither.h>
 #include <Interpreters/Context.h>
@@ -231,6 +232,27 @@ public:
 namespace impl_
 {
 
+template <bool is_multiply, bool is_division, typename T, typename U, template <typename> typename DecimalType>
+inline auto decimalResultType(const DecimalType<T> & tx, const DecimalType<U> & ty)
+{
+    const auto result_trait = DecimalUtils::binaryOpResult<is_multiply, is_division>(tx, ty);
+    return DecimalType<typename decltype(result_trait)::FieldType>(result_trait.precision, result_trait.scale);
+}
+
+template <bool is_multiply, bool is_division, typename T, typename U, template <typename> typename DecimalType>
+inline DecimalType<T> decimalResultType(const DecimalType<T> & tx, const DataTypeNumber<U> & ty)
+{
+    const auto result_trait = DecimalUtils::binaryOpResult<is_multiply, is_division>(tx, ty);
+    return DecimalType<typename decltype(result_trait)::FieldType>(result_trait.precision, result_trait.scale);
+}
+
+template <bool is_multiply, bool is_division, typename T, typename U, template <typename> typename DecimalType>
+inline DecimalType<U> decimalResultType(const DataTypeNumber<T> & tx, const DecimalType<U> & ty)
+{
+    const auto result_trait = DecimalUtils::binaryOpResult<is_multiply, is_division>(tx, ty);
+    return DecimalType<typename decltype(result_trait)::FieldType>(result_trait.precision, result_trait.scale);
+}
+
 /** Arithmetic operations: +, -, *, /, %,
   * intDiv (integer division)
   * Bitwise operations: |, &, ^, ~.
@@ -1166,7 +1188,7 @@ class FunctionBinaryArithmetic : public IFunction
             new_arguments[1].type = std::make_shared<DataTypeNumber<DataTypeInterval::FieldType>>();
 
         auto function = function_builder->build(new_arguments);
-        return function->execute(new_arguments, result_type, input_rows_count);
+        return function->execute(new_arguments, result_type, input_rows_count, /* dry_run = */ false);
     }
 
     ColumnPtr executeDateTimeTupleOfIntervalsPlusMinus(const ColumnsWithTypeAndName & arguments, const DataTypePtr & result_type,
@@ -1180,7 +1202,7 @@ class FunctionBinaryArithmetic : public IFunction
 
         auto function = function_builder->build(new_arguments);
 
-        return function->execute(new_arguments, result_type, input_rows_count);
+        return function->execute(new_arguments, result_type, input_rows_count, /* dry_run = */ false);
     }
 
     ColumnPtr executeIntervalTupleOfIntervalsPlusMinus(const ColumnsWithTypeAndName & arguments, const DataTypePtr & result_type,
@@ -1188,7 +1210,7 @@ class FunctionBinaryArithmetic : public IFunction
     {
         auto function = function_builder->build(arguments);
 
-        return function->execute(arguments, result_type, input_rows_count);
+        return function->execute(arguments, result_type, input_rows_count, /* dry_run = */ false);
     }
 
     ColumnPtr executeArraysImpl(const ColumnsWithTypeAndName & arguments, const DataTypePtr & result_type, size_t input_rows_count) const
@@ -1323,7 +1345,7 @@ class FunctionBinaryArithmetic : public IFunction
 
         auto function = function_builder->build(new_arguments);
 
-        return function->execute(new_arguments, result_type, input_rows_count);
+        return function->execute(new_arguments, result_type, input_rows_count, /* dry_run = */ false);
     }
 
     template <typename T, typename ResultDataType>
@@ -2225,7 +2247,7 @@ ColumnPtr executeStringInteger(const ColumnsWithTypeAndName & arguments, const A
         /// Special case when the function is plus, minus or multiply, both arguments are tuples.
         if (auto function_builder = getFunctionForTupleArithmetic(arguments[0].type, arguments[1].type, context))
         {
-            return function_builder->build(arguments)->execute(arguments, result_type, input_rows_count);
+            return function_builder->build(arguments)->execute(arguments, result_type, input_rows_count, /* dry_run = */ false);
         }
 
         /// Special case when the function is multiply or divide, one of arguments is Tuple and another is Number.
diff --git a/src/Functions/FunctionFactory.cpp b/src/Functions/FunctionFactory.cpp
index cd5d9d2c877..d98704f2baf 100644
--- a/src/Functions/FunctionFactory.cpp
+++ b/src/Functions/FunctionFactory.cpp
@@ -1,4 +1,5 @@
 #include <Functions/FunctionFactory.h>
+#include <Functions/IFunctionAdaptors.h>
 
 #include <Interpreters/Context.h>
 
diff --git a/src/Functions/FunctionFactory.h b/src/Functions/FunctionFactory.h
index d05e84439be..350ee984c54 100644
--- a/src/Functions/FunctionFactory.h
+++ b/src/Functions/FunctionFactory.h
@@ -5,7 +5,6 @@
 #include <Common/IFactoryWithAliases.h>
 #include <Common/FunctionDocumentation.h>
 #include <Functions/IFunction.h>
-#include <Functions/IFunctionAdaptors.h>
 
 #include <functional>
 #include <memory>
diff --git a/src/Functions/FunctionMathUnary.h b/src/Functions/FunctionMathUnary.h
index 12a15a072b2..b02795bf787 100644
--- a/src/Functions/FunctionMathUnary.h
+++ b/src/Functions/FunctionMathUnary.h
@@ -1,6 +1,7 @@
 #pragma once
 
 #include <Core/callOnTypeIndex.h>
+#include <Core/DecimalFunctions.h>
 #include <DataTypes/DataTypesNumber.h>
 #include <DataTypes/DataTypesDecimal.h>
 #include <Columns/ColumnsNumber.h>
diff --git a/src/Functions/FunctionUnaryArithmetic.h b/src/Functions/FunctionUnaryArithmetic.h
index 259dc1c42ba..cf79bb06430 100644
--- a/src/Functions/FunctionUnaryArithmetic.h
+++ b/src/Functions/FunctionUnaryArithmetic.h
@@ -339,7 +339,7 @@ public:
         /// Special case when the function is negate, argument is tuple.
         if (auto function_builder = getFunctionForTupleArithmetic(arguments[0].type, context))
         {
-            return function_builder->build(arguments)->execute(arguments, result_type, input_rows_count);
+            return function_builder->build(arguments)->execute(arguments, result_type, input_rows_count, /* dry_run = */ false);
         }
 
         ColumnPtr result_column;
diff --git a/src/Functions/FunctionUnixTimestamp64.h b/src/Functions/FunctionUnixTimestamp64.h
index 3e3a89b8a05..a3c26111cb5 100644
--- a/src/Functions/FunctionUnixTimestamp64.h
+++ b/src/Functions/FunctionUnixTimestamp64.h
@@ -5,6 +5,7 @@
 #include <Functions/FunctionHelpers.h>
 #include <DataTypes/DataTypeDateTime64.h>
 #include <DataTypes/DataTypesNumber.h>
+#include <Core/DecimalFunctions.h>
 #include <Columns/ColumnsNumber.h>
 #include <Interpreters/Context.h>
 
diff --git a/src/Functions/FunctionsCodingULID.cpp b/src/Functions/FunctionsCodingULID.cpp
index b67224a5625..ec8ef211d09 100644
--- a/src/Functions/FunctionsCodingULID.cpp
+++ b/src/Functions/FunctionsCodingULID.cpp
@@ -2,6 +2,7 @@
 
 #if USE_ULID
 
+#include <Core/DecimalFunctions.h>
 #include <Columns/ColumnFixedString.h>
 #include <Columns/ColumnString.h>
 #include <Columns/ColumnsDateTime.h>
diff --git a/src/Functions/FunctionsComparison.h b/src/Functions/FunctionsComparison.h
index c9d3ccb4445..d8ff9b1699e 100644
--- a/src/Functions/FunctionsComparison.h
+++ b/src/Functions/FunctionsComparison.h
@@ -41,12 +41,6 @@
 #include <limits>
 #include <type_traits>
 
-#if USE_EMBEDDED_COMPILER
-#    include <DataTypes/Native.h>
-#    include <llvm/IR/IRBuilder.h>
-#endif
-
-
 namespace DB
 {
 
@@ -59,6 +53,68 @@ namespace ErrorCodes
     extern const int BAD_ARGUMENTS;
 }
 
+template <bool _int, bool _float, bool _decimal, bool _datetime, typename F>
+static inline bool callOnAtLeastOneDecimalType(TypeIndex type_num1, TypeIndex type_num2, F && f)
+{
+    switch (type_num1)
+    {
+        case TypeIndex::DateTime64:
+            return callOnBasicType<DateTime64, _int, _float, _decimal, _datetime>(type_num2, std::forward<F>(f));
+        case TypeIndex::Decimal32:
+            return callOnBasicType<Decimal32, _int, _float, _decimal, _datetime>(type_num2, std::forward<F>(f));
+        case TypeIndex::Decimal64:
+            return callOnBasicType<Decimal64, _int, _float, _decimal, _datetime>(type_num2, std::forward<F>(f));
+        case TypeIndex::Decimal128:
+            return callOnBasicType<Decimal128, _int, _float, _decimal, _datetime>(type_num2, std::forward<F>(f));
+        case TypeIndex::Decimal256:
+            return callOnBasicType<Decimal256, _int, _float, _decimal, _datetime>(type_num2, std::forward<F>(f));
+        default:
+            break;
+    }
+
+    switch (type_num2)
+    {
+        case TypeIndex::DateTime64:
+            return callOnBasicTypeSecondArg<DateTime64, _int, _float, _decimal, _datetime>(type_num1, std::forward<F>(f));
+        case TypeIndex::Decimal32:
+            return callOnBasicTypeSecondArg<Decimal32, _int, _float, _decimal, _datetime>(type_num1, std::forward<F>(f));
+        case TypeIndex::Decimal64:
+            return callOnBasicTypeSecondArg<Decimal64, _int, _float, _decimal, _datetime>(type_num1, std::forward<F>(f));
+        case TypeIndex::Decimal128:
+            return callOnBasicTypeSecondArg<Decimal128, _int, _float, _decimal, _datetime>(type_num1, std::forward<F>(f));
+        case TypeIndex::Decimal256:
+            return callOnBasicTypeSecondArg<Decimal256, _int, _float, _decimal, _datetime>(type_num1, std::forward<F>(f));
+        default:
+            break;
+    }
+
+    return false;
+}
+
+template <template <typename, typename> class Operation, typename Name>
+ColumnPtr executeDecimal(const ColumnWithTypeAndName & col_left, const ColumnWithTypeAndName & col_right, bool check_decimal_overflow)
+{
+    TypeIndex left_number = col_left.type->getTypeId();
+    TypeIndex right_number = col_right.type->getTypeId();
+    ColumnPtr res;
+
+    auto call = [&](const auto & types) -> bool
+    {
+        using Types = std::decay_t<decltype(types)>;
+        using LeftDataType = typename Types::LeftType;
+        using RightDataType = typename Types::RightType;
+
+        return (res = DecimalComparison<LeftDataType, RightDataType, Operation>::apply(col_left, col_right, check_decimal_overflow))
+            != nullptr;
+    };
+
+    if (!callOnAtLeastOneDecimalType<true, false, true, true>(left_number, right_number, call))
+        throw Exception(
+            ErrorCodes::LOGICAL_ERROR, "Wrong call for {} with {} and {}", Name::name, col_left.type->getName(), col_right.type->getName());
+
+    return res;
+}
+
 
 /** Comparison functions: ==, !=, <, >, <=, >=.
   * The comparison functions always return 0 or 1 (UInt8).
@@ -574,62 +630,6 @@ struct GenericComparisonImpl
     }
 };
 
-
-#if USE_EMBEDDED_COMPILER
-
-template <template <typename, typename> typename Op> struct CompileOp;
-
-template <> struct CompileOp<EqualsOp>
-{
-    static llvm::Value * compile(llvm::IRBuilder<> & b, llvm::Value * x, llvm::Value * y, bool /*is_signed*/)
-    {
-        return x->getType()->isIntegerTy() ? b.CreateICmpEQ(x, y) : b.CreateFCmpOEQ(x, y); /// qNaNs always compare false
-    }
-};
-
-template <> struct CompileOp<NotEqualsOp>
-{
-    static llvm::Value * compile(llvm::IRBuilder<> & b, llvm::Value * x, llvm::Value * y, bool /*is_signed*/)
-    {
-        return x->getType()->isIntegerTy() ? b.CreateICmpNE(x, y) : b.CreateFCmpUNE(x, y);
-    }
-};
-
-template <> struct CompileOp<LessOp>
-{
-    static llvm::Value * compile(llvm::IRBuilder<> & b, llvm::Value * x, llvm::Value * y, bool is_signed)
-    {
-        return x->getType()->isIntegerTy() ? (is_signed ? b.CreateICmpSLT(x, y) : b.CreateICmpULT(x, y)) : b.CreateFCmpOLT(x, y);
-    }
-};
-
-template <> struct CompileOp<GreaterOp>
-{
-    static llvm::Value * compile(llvm::IRBuilder<> & b, llvm::Value * x, llvm::Value * y, bool is_signed)
-    {
-        return x->getType()->isIntegerTy() ? (is_signed ? b.CreateICmpSGT(x, y) : b.CreateICmpUGT(x, y)) : b.CreateFCmpOGT(x, y);
-    }
-};
-
-template <> struct CompileOp<LessOrEqualsOp>
-{
-    static llvm::Value * compile(llvm::IRBuilder<> & b, llvm::Value * x, llvm::Value * y, bool is_signed)
-    {
-        return x->getType()->isIntegerTy() ? (is_signed ? b.CreateICmpSLE(x, y) : b.CreateICmpULE(x, y)) : b.CreateFCmpOLE(x, y);
-    }
-};
-
-template <> struct CompileOp<GreaterOrEqualsOp>
-{
-    static llvm::Value * compile(llvm::IRBuilder<> & b, llvm::Value * x, llvm::Value * y, bool is_signed)
-    {
-        return x->getType()->isIntegerTy() ? (is_signed ? b.CreateICmpSGE(x, y) : b.CreateICmpUGE(x, y)) : b.CreateFCmpOGE(x, y);
-    }
-};
-
-#endif
-
-
 struct NameEquals          { static constexpr auto name = "equals"; };
 struct NameNotEquals       { static constexpr auto name = "notEquals"; };
 struct NameLess            { static constexpr auto name = "less"; };
@@ -753,30 +753,6 @@ private:
         return nullptr;
     }
 
-    ColumnPtr executeDecimal(const ColumnWithTypeAndName & col_left, const ColumnWithTypeAndName & col_right) const
-    {
-        TypeIndex left_number = col_left.type->getTypeId();
-        TypeIndex right_number = col_right.type->getTypeId();
-        ColumnPtr res;
-
-        auto call = [&](const auto & types) -> bool
-        {
-            using Types = std::decay_t<decltype(types)>;
-            using LeftDataType = typename Types::LeftType;
-            using RightDataType = typename Types::RightType;
-
-            if (check_decimal_overflow)
-                return (res = DecimalComparison<LeftDataType, RightDataType, Op, true>::apply(col_left, col_right)) != nullptr;
-            return (res = DecimalComparison<LeftDataType, RightDataType, Op, false>::apply(col_left, col_right)) != nullptr;
-        };
-
-        if (!callOnBasicTypes<true, false, true, true>(left_number, right_number, call))
-            throw Exception(ErrorCodes::LOGICAL_ERROR, "Wrong call for {} with {} and {}",
-                            getName(), col_left.type->getName(), col_right.type->getName());
-
-        return res;
-    }
-
     ColumnPtr executeString(const IColumn * c0, const IColumn * c1) const
     {
         const ColumnString * c0_string = checkAndGetColumn<ColumnString>(c0);
@@ -1010,7 +986,7 @@ private:
             convolution_columns[i].type = impl->getResultType();
 
             /// Comparison of the elements.
-            convolution_columns[i].column = impl->execute(tmp_columns, impl->getResultType(), input_rows_count);
+            convolution_columns[i].column = impl->execute(tmp_columns, impl->getResultType(), input_rows_count, /* dry_run = */ false);
         }
 
         if (tuple_size == 1)
@@ -1021,7 +997,7 @@ private:
 
         /// Logical convolution.
         auto impl = func_convolution->build(convolution_columns);
-        return impl->execute(convolution_columns, impl->getResultType(), input_rows_count);
+        return impl->execute(convolution_columns, impl->getResultType(), input_rows_count, /* dry_run = */ false);
     }
 
     ColumnPtr executeTupleLessGreaterImpl(
@@ -1053,18 +1029,18 @@ private:
             {
                 auto impl_head = func_compare_head->build(tmp_columns);
                 less_columns[i].type = impl_head->getResultType();
-                less_columns[i].column = impl_head->execute(tmp_columns, less_columns[i].type, input_rows_count);
+                less_columns[i].column = impl_head->execute(tmp_columns, less_columns[i].type, input_rows_count, /* dry_run = */ false);
 
                 auto impl_equals = func_equals->build(tmp_columns);
                 equal_columns[i].type = impl_equals->getResultType();
-                equal_columns[i].column = impl_equals->execute(tmp_columns, equal_columns[i].type, input_rows_count);
+                equal_columns[i].column = impl_equals->execute(tmp_columns, equal_columns[i].type, input_rows_count, /* dry_run = */ false);
 
             }
             else
             {
                 auto impl_tail = func_compare_tail->build(tmp_columns);
                 less_columns[i].type = impl_tail->getResultType();
-                less_columns[i].column = impl_tail->execute(tmp_columns, less_columns[i].type, input_rows_count);
+                less_columns[i].column = impl_tail->execute(tmp_columns, less_columns[i].type, input_rows_count, /* dry_run = */ false);
             }
         }
 
@@ -1083,13 +1059,13 @@ private:
             tmp_columns[1] = equal_columns[i];
             auto func_and_adaptor = func_and->build(tmp_columns);
 
-            tmp_columns[0].column = func_and_adaptor->execute(tmp_columns, func_and_adaptor->getResultType(), input_rows_count);
+            tmp_columns[0].column = func_and_adaptor->execute(tmp_columns, func_and_adaptor->getResultType(), input_rows_count, /* dry_run = */ false);
             tmp_columns[0].type = func_and_adaptor->getResultType();
 
             tmp_columns[1] = less_columns[i];
             auto func_or_adaptor = func_or->build(tmp_columns);
 
-            tmp_columns[0].column = func_or_adaptor->execute(tmp_columns, func_or_adaptor->getResultType(), input_rows_count);
+            tmp_columns[0].column = func_or_adaptor->execute(tmp_columns, func_or_adaptor->getResultType(), input_rows_count, /* dry_run = */ false);
             tmp_columns[tmp_columns.size() - 1].type = func_or_adaptor->getResultType();
         }
 
@@ -1334,7 +1310,8 @@ public:
                 DataTypePtr common_type = getLeastSupertype(DataTypes{left_type, right_type});
                 ColumnPtr c0_converted = castColumn(col_with_type_and_name_left, common_type);
                 ColumnPtr c1_converted = castColumn(col_with_type_and_name_right, common_type);
-                return executeDecimal({c0_converted, common_type, "left"}, {c1_converted, common_type, "right"});
+                return executeDecimal<Op, Name>(
+                    {c0_converted, common_type, "left"}, {c1_converted, common_type, "right"}, check_decimal_overflow);
             }
 
             /// Check does another data type is comparable to Decimal, includes Int and Float.
@@ -1357,7 +1334,7 @@ public:
                     = ColumnsWithTypeAndName{{c0_converted, converted_type, "left"}, {c1_converted, converted_type, "right"}};
                 return executeImpl(new_arguments, result_type, input_rows_count);
             }
-            return executeDecimal(col_with_type_and_name_left, col_with_type_and_name_right);
+            return executeDecimal<Op, Name>(col_with_type_and_name_left, col_with_type_and_name_right, check_decimal_overflow);
         }
         if (date_and_datetime)
         {
@@ -1367,7 +1344,8 @@ public:
             if (!((res = executeNumLeftType<UInt32>(c0_converted.get(), c1_converted.get()))
                   || (res = executeNumLeftType<UInt64>(c0_converted.get(), c1_converted.get()))
                   || (res = executeNumLeftType<Int32>(c0_converted.get(), c1_converted.get()))
-                  || (res = executeDecimal({c0_converted, common_type, "left"}, {c1_converted, common_type, "right"}))))
+                  || (res = executeDecimal<Op, Name>(
+                          {c0_converted, common_type, "left"}, {c1_converted, common_type, "right"}, check_decimal_overflow))))
                 throw Exception(ErrorCodes::LOGICAL_ERROR, "Date related common types can only be UInt32/UInt64/Int32/Decimal");
             return res;
         }
diff --git a/src/Functions/FunctionsConversion.cpp b/src/Functions/FunctionsConversion.cpp
index d581ae510e8..ead4b6065ac 100644
--- a/src/Functions/FunctionsConversion.cpp
+++ b/src/Functions/FunctionsConversion.cpp
@@ -3243,11 +3243,9 @@ private:
     {
         auto function_adaptor = std::make_unique<FunctionToOverloadResolverAdaptor>(function)->build({ColumnWithTypeAndName{nullptr, from_type, ""}});
 
-        return [function_adaptor]
-            (ColumnsWithTypeAndName & arguments, const DataTypePtr & result_type, const ColumnNullable *, size_t input_rows_count)
-        {
-            return function_adaptor->execute(arguments, result_type, input_rows_count);
-        };
+        return [function_adaptor](
+                   ColumnsWithTypeAndName & arguments, const DataTypePtr & result_type, const ColumnNullable *, size_t input_rows_count)
+        { return function_adaptor->execute(arguments, result_type, input_rows_count, /* dry_run = */ false); };
     }
 
     static WrapperType createToNullableColumnWrapper()
diff --git a/src/Functions/FunctionsExternalDictionaries.h b/src/Functions/FunctionsExternalDictionaries.h
index 4c7e1534ba5..b0f54b74091 100644
--- a/src/Functions/FunctionsExternalDictionaries.h
+++ b/src/Functions/FunctionsExternalDictionaries.h
@@ -644,7 +644,7 @@ private:
         };
 
         auto rows = mask_column->size();
-        result_column = if_func->build(if_args)->execute(if_args, result_type, rows);
+        result_column = if_func->build(if_args)->execute(if_args, result_type, rows, /* dry_run = */ false);
     }
 
 
diff --git a/src/Functions/FunctionsLogical.cpp b/src/Functions/FunctionsLogical.cpp
index f83a958bf31..625a0f4c388 100644
--- a/src/Functions/FunctionsLogical.cpp
+++ b/src/Functions/FunctionsLogical.cpp
@@ -1,5 +1,6 @@
 #include <Functions/FunctionFactory.h>
 #include <Functions/FunctionsLogical.h>
+#include <Functions/IFunctionAdaptors.h>
 #include <Functions/logical.h>
 
 #include <Columns/ColumnConst.h>
diff --git a/src/Functions/FunctionsOpDate.cpp b/src/Functions/FunctionsOpDate.cpp
index c4b154736e0..d8a3aabd9d0 100644
--- a/src/Functions/FunctionsOpDate.cpp
+++ b/src/Functions/FunctionsOpDate.cpp
@@ -73,7 +73,7 @@ public:
         auto op_build = op->build(arguments);
 
         auto res_type = op_build->getResultType();
-        return op_build->execute(arguments, res_type, input_rows_count);
+        return op_build->execute(arguments, res_type, input_rows_count, /* dry_run = */ false);
     }
 
 private:
diff --git a/src/Functions/IFunction.cpp b/src/Functions/IFunction.cpp
index 6a552bd1b01..b9ce4b8f04f 100644
--- a/src/Functions/IFunction.cpp
+++ b/src/Functions/IFunction.cpp
@@ -486,6 +486,13 @@ ColumnPtr IExecutableFunction::execute(
     return executeWithoutSparseColumns(arguments, result_type, input_rows_count, dry_run);
 }
 
+ColumnPtr IFunctionBase::execute(const DB::ColumnsWithTypeAndName& arguments, const DB::DataTypePtr& result_type,
+        size_t input_rows_count, bool dry_run) const
+{
+    checkFunctionArgumentSizes(arguments, input_rows_count);
+    return prepare(arguments)->execute(arguments, result_type, input_rows_count, dry_run);
+}
+
 void IFunctionOverloadResolver::checkNumberOfArguments(size_t number_of_arguments) const
 {
     if (isVariadic())
diff --git a/src/Functions/IFunction.h b/src/Functions/IFunction.h
index 871af2b5503..2e44d14d820 100644
--- a/src/Functions/IFunction.h
+++ b/src/Functions/IFunction.h
@@ -7,7 +7,6 @@
 #include <Core/Names.h>
 #include <Core/ValuesWithType.h>
 #include <DataTypes/IDataType.h>
-#include <Functions/FunctionHelpers.h>
 #include <Common/Exception.h>
 
 #include "config.h"
@@ -141,11 +140,7 @@ public:
         const ColumnsWithTypeAndName & arguments,
         const DataTypePtr & result_type,
         size_t input_rows_count,
-        bool dry_run = false) const
-    {
-        checkFunctionArgumentSizes(arguments, input_rows_count);
-        return prepare(arguments)->execute(arguments, result_type, input_rows_count, dry_run);
-    }
+        bool dry_run) const;
 
     /// Get the main function name.
     virtual String getName() const = 0;
diff --git a/src/Functions/IFunctionAdaptors.cpp b/src/Functions/IFunctionAdaptors.cpp
new file mode 100644
index 00000000000..d490ea7bbb0
--- /dev/null
+++ b/src/Functions/IFunctionAdaptors.cpp
@@ -0,0 +1,19 @@
+#include <Functions/FunctionHelpers.h>
+#include <Functions/IFunctionAdaptors.h>
+
+namespace DB
+{
+ColumnPtr FunctionToExecutableFunctionAdaptor::executeImpl(const ColumnsWithTypeAndName& arguments,
+        const DataTypePtr& result_type, size_t input_rows_count) const
+{
+    checkFunctionArgumentSizes(arguments, input_rows_count);
+    return function->executeImpl(arguments, result_type, input_rows_count);
+}
+
+ColumnPtr FunctionToExecutableFunctionAdaptor::executeDryRunImpl(const ColumnsWithTypeAndName & arguments, const DataTypePtr & result_type, size_t input_rows_count) const
+{
+    checkFunctionArgumentSizes(arguments, input_rows_count);
+    return function->executeImplDryRun(arguments, result_type, input_rows_count);
+}
+
+}
diff --git a/src/Functions/IFunctionAdaptors.h b/src/Functions/IFunctionAdaptors.h
index f7a8e798a8e..ad5f3df5df7 100644
--- a/src/Functions/IFunctionAdaptors.h
+++ b/src/Functions/IFunctionAdaptors.h
@@ -16,17 +16,8 @@ public:
 
 protected:
 
-    ColumnPtr executeImpl(const ColumnsWithTypeAndName & arguments, const DataTypePtr & result_type, size_t input_rows_count) const final
-    {
-        checkFunctionArgumentSizes(arguments, input_rows_count);
-        return function->executeImpl(arguments, result_type, input_rows_count);
-    }
-
-    ColumnPtr executeDryRunImpl(const ColumnsWithTypeAndName & arguments, const DataTypePtr & result_type, size_t input_rows_count) const final
-    {
-        checkFunctionArgumentSizes(arguments, input_rows_count);
-        return function->executeImplDryRun(arguments, result_type, input_rows_count);
-    }
+    ColumnPtr executeImpl(const ColumnsWithTypeAndName & arguments, const DataTypePtr & result_type, size_t input_rows_count) const final;
+    ColumnPtr executeDryRunImpl(const ColumnsWithTypeAndName & arguments, const DataTypePtr & result_type, size_t input_rows_count) const final;
 
     bool useDefaultImplementationForNulls() const final { return function->useDefaultImplementationForNulls(); }
     bool useDefaultImplementationForNothing() const final { return function->useDefaultImplementationForNothing(); }
diff --git a/src/Functions/Kusto/KqlArraySort.cpp b/src/Functions/Kusto/KqlArraySort.cpp
index fb3e6259ee4..c6f0048f7df 100644
--- a/src/Functions/Kusto/KqlArraySort.cpp
+++ b/src/Functions/Kusto/KqlArraySort.cpp
@@ -117,11 +117,11 @@ public:
         }
 
         auto zipped
-            = FunctionFactory::instance().get("arrayZip", context)->build(new_args)->execute(new_args, result_type, input_rows_count);
+            = FunctionFactory::instance().get("arrayZip", context)->build(new_args)->execute(new_args, result_type, input_rows_count, /* dry_run = */ false);
 
         ColumnsWithTypeAndName sort_arg({{zipped, std::make_shared<DataTypeArray>(result_type), "zipped"}});
         auto sorted_tuple
-            = FunctionFactory::instance().get(sort_function, context)->build(sort_arg)->execute(sort_arg, result_type, input_rows_count);
+            = FunctionFactory::instance().get(sort_function, context)->build(sort_arg)->execute(sort_arg, result_type, input_rows_count, /* dry_run = */ false);
 
         auto null_type = std::make_shared<DataTypeNullable>(std::make_shared<DataTypeInt8>());
 
@@ -140,7 +140,7 @@ public:
                     {null_type->createColumnConstWithDefaultValue(input_rows_count), null_type, "NULL"},
                 });
 
-                tuple_columns[i] = fun_array->build(null_array_arg)->execute(null_array_arg, arg_type, input_rows_count);
+                tuple_columns[i] = fun_array->build(null_array_arg)->execute(null_array_arg, arg_type, input_rows_count, /* dry_run = */ false);
                 tuple_columns[i] = tuple_columns[i]->convertToFullColumnIfConst();
             }
             else
@@ -151,7 +151,7 @@ public:
                 auto tuple_coulmn = FunctionFactory::instance()
                                         .get("tupleElement", context)
                                         ->build(untuple_args)
-                                        ->execute(untuple_args, result_type, input_rows_count);
+                                        ->execute(untuple_args, result_type, input_rows_count, /* dry_run = */ false);
 
                 auto out_tmp = ColumnArray::create(nested_types[i]->createColumn());
 
@@ -190,7 +190,7 @@ public:
             slice_index.column = FunctionFactory::instance()
                                      .get("indexOf", context)
                                      ->build(indexof_args)
-                                     ->execute(indexof_args, result_type, input_rows_count);
+                                     ->execute(indexof_args, result_type, input_rows_count, /* dry_run = */ false);
 
             auto null_index_in_array = slice_index.column->get64(0);
             if (null_index_in_array > 0)
@@ -218,15 +218,15 @@ public:
                         ColumnsWithTypeAndName slice_args_right(
                             {{ColumnWithTypeAndName(tuple_columns[i], arg_type, "array")}, slice_index});
                         ColumnWithTypeAndName arr_left{
-                            fun_slice->build(slice_args_left)->execute(slice_args_left, arg_type, input_rows_count), arg_type, ""};
+                            fun_slice->build(slice_args_left)->execute(slice_args_left, arg_type, input_rows_count, /* dry_run = */ false), arg_type, ""};
                         ColumnWithTypeAndName arr_right{
-                            fun_slice->build(slice_args_right)->execute(slice_args_right, arg_type, input_rows_count), arg_type, ""};
+                            fun_slice->build(slice_args_right)->execute(slice_args_right, arg_type, input_rows_count, /* dry_run = */ false), arg_type, ""};
 
                         ColumnsWithTypeAndName arr_cancat({arr_right, arr_left});
                         auto out_tmp = FunctionFactory::instance()
                                            .get("arrayConcat", context)
                                            ->build(arr_cancat)
-                                           ->execute(arr_cancat, arg_type, input_rows_count);
+                                           ->execute(arr_cancat, arg_type, input_rows_count, /* dry_run = */ false);
                         adjusted_columns[i] = std::move(out_tmp);
                     }
                 }
diff --git a/src/Functions/Kusto/KqlFunctionBase.h b/src/Functions/Kusto/KqlFunctionBase.h
index efdf8982f4b..9b636ac2cd7 100644
--- a/src/Functions/Kusto/KqlFunctionBase.h
+++ b/src/Functions/Kusto/KqlFunctionBase.h
@@ -23,7 +23,7 @@ public:
             {DataTypeUInt8().createColumnConst(1, toField(UInt8(1))), std::make_shared<DataTypeUInt8>(), ""},
             {DataTypeUInt8().createColumnConst(1, toField(UInt8(2))), std::make_shared<DataTypeUInt8>(), ""}
         });
-        auto if_res = FunctionFactory::instance().get("if", context)->build(if_columns)->execute(if_columns, std::make_shared<DataTypeUInt8>(), input_rows_count);
+        auto if_res = FunctionFactory::instance().get("if", context)->build(if_columns)->execute(if_columns, std::make_shared<DataTypeUInt8>(), input_rows_count, /* dry_run = */ false);
         auto result = if_res->getUInt(0);
         return (result == 1);
     }
diff --git a/src/Functions/UserDefined/UserDefinedExecutableFunctionFactory.cpp b/src/Functions/UserDefined/UserDefinedExecutableFunctionFactory.cpp
index 3f72073ff31..9b236172c43 100644
--- a/src/Functions/UserDefined/UserDefinedExecutableFunctionFactory.cpp
+++ b/src/Functions/UserDefined/UserDefinedExecutableFunctionFactory.cpp
@@ -10,10 +10,9 @@
 #include <Processors/Sources/SourceFromSingleChunk.h>
 #include <Formats/formatBlock.h>
 
-#include <Functions/FunctionFactory.h>
 #include <Functions/FunctionHelpers.h>
+#include <Functions/IFunctionAdaptors.h>
 #include <Functions/UserDefined/ExternalUserDefinedExecutableFunctionsLoader.h>
-#include <AggregateFunctions/AggregateFunctionFactory.h>
 #include <Interpreters/convertFieldToType.h>
 #include <Interpreters/Context.h>
 #include <Interpreters/castColumn.h>
diff --git a/src/Functions/array/FunctionsMapMiscellaneous.cpp b/src/Functions/array/FunctionsMapMiscellaneous.cpp
index c3586a57161..9bd40cf3549 100644
--- a/src/Functions/array/FunctionsMapMiscellaneous.cpp
+++ b/src/Functions/array/FunctionsMapMiscellaneous.cpp
@@ -10,6 +10,7 @@
 #include <DataTypes/DataTypeTuple.h>
 
 #include <Functions/FunctionHelpers.h>
+#include <Functions/IFunctionAdaptors.h>
 #include <Functions/like.h>
 #include <Functions/array/arrayConcat.h>
 #include <Functions/array/arrayFilter.h>
diff --git a/src/Functions/array/arrayIntersect.cpp b/src/Functions/array/arrayIntersect.cpp
index 64bf75bfd50..deccca4fb5e 100644
--- a/src/Functions/array/arrayIntersect.cpp
+++ b/src/Functions/array/arrayIntersect.cpp
@@ -339,7 +339,7 @@ static ColumnPtr callFunctionNotEquals(ColumnWithTypeAndName first, ColumnWithTy
 {
     ColumnsWithTypeAndName args{first, second};
     auto eq_func = FunctionFactory::instance().get("notEquals", context)->build(args);
-    return eq_func->execute(args, eq_func->getResultType(), args.front().column->size());
+    return eq_func->execute(args, eq_func->getResultType(), args.front().column->size(), /* dry_run = */ false);
 }
 
 template <typename Mode>
diff --git a/src/Functions/array/arrayJaccardIndex.cpp b/src/Functions/array/arrayJaccardIndex.cpp
index 56a8c66fc8d..4c023e88a2d 100644
--- a/src/Functions/array/arrayJaccardIndex.cpp
+++ b/src/Functions/array/arrayJaccardIndex.cpp
@@ -113,7 +113,7 @@ public:
 
         ColumnWithTypeAndName intersect_column;
         intersect_column.type = intersect_array->getResultType();
-        intersect_column.column = intersect_array->execute(arguments, intersect_column.type, input_rows_count);
+        intersect_column.column = intersect_array->execute(arguments, intersect_column.type, input_rows_count, /* dry_run = */ false);
 
         const auto * intersect_column_type = checkAndGetDataType<DataTypeArray>(intersect_column.type.get());
         if (!intersect_column_type)
diff --git a/src/Functions/array/has.cpp b/src/Functions/array/has.cpp
index a17dcdcfbf9..ee7d7fb5a0d 100644
--- a/src/Functions/array/has.cpp
+++ b/src/Functions/array/has.cpp
@@ -1,5 +1,6 @@
 #include "arrayIndex.h"
 #include <Functions/FunctionFactory.h>
+#include <Functions/IFunctionAdaptors.h>
 
 namespace DB
 {
diff --git a/src/Functions/caseWithExpression.cpp b/src/Functions/caseWithExpression.cpp
index f0a620489ef..257fd4c4528 100644
--- a/src/Functions/caseWithExpression.cpp
+++ b/src/Functions/caseWithExpression.cpp
@@ -93,12 +93,12 @@ public:
 
         auto fun_array = FunctionFactory::instance().get("array", context);
 
-        src_array_col.column = fun_array->build(src_array_elems)->execute(src_array_elems, src_array_type, input_rows_count);
-        dst_array_col.column = fun_array->build(dst_array_elems)->execute(dst_array_elems, dst_array_type, input_rows_count);
+        src_array_col.column = fun_array->build(src_array_elems)->execute(src_array_elems, src_array_type, input_rows_count, /* dry_run = */ false);
+        dst_array_col.column = fun_array->build(dst_array_elems)->execute(dst_array_elems, dst_array_type, input_rows_count, /* dry_run = */ false);
 
         /// Execute transform.
         ColumnsWithTypeAndName transform_args{args.front(), src_array_col, dst_array_col, args.back()};
-        return FunctionFactory::instance().get("transform", context)->build(transform_args)->execute(transform_args, result_type, input_rows_count);
+        return FunctionFactory::instance().get("transform", context)->build(transform_args)->execute(transform_args, result_type, input_rows_count, /* dry_run = */ false);
     }
 
 private:
diff --git a/src/Functions/changeDate.cpp b/src/Functions/changeDate.cpp
index 19e4c165ee3..0c8c83bab19 100644
--- a/src/Functions/changeDate.cpp
+++ b/src/Functions/changeDate.cpp
@@ -6,6 +6,7 @@
 #include <Columns/IColumn.h>
 #include <Common/DateLUT.h>
 #include <Common/typeid_cast.h>
+#include <Core/DecimalFunctions.h>
 #include <DataTypes/DataTypeDate.h>
 #include <DataTypes/DataTypeDate32.h>
 #include <DataTypes/DataTypeDateTime.h>
diff --git a/src/Functions/coalesce.cpp b/src/Functions/coalesce.cpp
index 19da6a85b38..2a8cad386d4 100644
--- a/src/Functions/coalesce.cpp
+++ b/src/Functions/coalesce.cpp
@@ -133,11 +133,11 @@ public:
             {
                 tmp_args[0] = filtered_args[i];
                 auto & cond = multi_if_args.emplace_back(ColumnWithTypeAndName{nullptr, std::make_shared<DataTypeUInt8>(), ""});
-                cond.column = is_not_null->build(tmp_args)->execute(tmp_args, cond.type, input_rows_count);
+                cond.column = is_not_null->build(tmp_args)->execute(tmp_args, cond.type, input_rows_count, /* dry_run = */ false);
 
                 tmp_args[0] = filtered_args[i];
                 auto & val = multi_if_args.emplace_back(ColumnWithTypeAndName{nullptr, removeNullable(filtered_args[i].type), ""});
-                val.column = assume_not_null->build(tmp_args)->execute(tmp_args, val.type, input_rows_count);
+                val.column = assume_not_null->build(tmp_args)->execute(tmp_args, val.type, input_rows_count, /* dry_run = */ false);
             }
         }
 
@@ -152,7 +152,7 @@ public:
         /// use function "if" instead, because it's implemented more efficient.
         /// TODO: make "multiIf" the same efficient.
         FunctionOverloadResolverPtr if_or_multi_if = multi_if_args.size() == 3 ? if_function : multi_if_function;
-        ColumnPtr res = if_or_multi_if->build(multi_if_args)->execute(multi_if_args, result_type, input_rows_count);
+        ColumnPtr res = if_or_multi_if->build(multi_if_args)->execute(multi_if_args, result_type, input_rows_count, /* dry_run = */ false);
 
         /// if last argument is not nullable, result should be also not nullable
         if (!multi_if_args.back().column->isNullable() && res->isNullable())
diff --git a/src/Functions/concat.cpp b/src/Functions/concat.cpp
index e0283891428..d2ae91101d8 100644
--- a/src/Functions/concat.cpp
+++ b/src/Functions/concat.cpp
@@ -3,6 +3,7 @@
 #include <DataTypes/DataTypeString.h>
 #include <Functions/FunctionFactory.h>
 #include <Functions/FunctionHelpers.h>
+#include <Functions/IFunctionAdaptors.h>
 #include <Functions/GatherUtils/Algorithms.h>
 #include <Functions/GatherUtils/Sinks.h>
 #include <Functions/GatherUtils/Sources.h>
diff --git a/src/Functions/date_trunc.cpp b/src/Functions/date_trunc.cpp
index 020a2da5004..9c46a65313b 100644
--- a/src/Functions/date_trunc.cpp
+++ b/src/Functions/date_trunc.cpp
@@ -151,10 +151,10 @@ public:
         auto to_start_of_interval = FunctionFactory::instance().get("toStartOfInterval", context);
 
         if (arguments.size() == 2)
-            return to_start_of_interval->build(temp_columns)->execute(temp_columns, result_type, input_rows_count);
+            return to_start_of_interval->build(temp_columns)->execute(temp_columns, result_type, input_rows_count, /* dry_run = */ false);
 
         temp_columns[2] = arguments[2];
-        return to_start_of_interval->build(temp_columns)->execute(temp_columns, result_type, input_rows_count);
+        return to_start_of_interval->build(temp_columns)->execute(temp_columns, result_type, input_rows_count, /* dry_run = */ false);
     }
 
     bool hasInformationAboutMonotonicity() const override
diff --git a/src/Functions/empty.cpp b/src/Functions/empty.cpp
index 2b6a7869845..0864cb9e2fb 100644
--- a/src/Functions/empty.cpp
+++ b/src/Functions/empty.cpp
@@ -1,6 +1,7 @@
 #include <DataTypes/DataTypeString.h>
 #include <Functions/FunctionFactory.h>
 #include <Functions/FunctionStringOrArrayToT.h>
+#include <Functions/IFunctionAdaptors.h>
 #include <Functions/EmptyImpl.h>
 #include <Columns/ColumnObject.h>
 
diff --git a/src/Functions/formatRow.cpp b/src/Functions/formatRow.cpp
index a86b4c6905b..521e560e487 100644
--- a/src/Functions/formatRow.cpp
+++ b/src/Functions/formatRow.cpp
@@ -5,6 +5,7 @@
 #include <Functions/FunctionFactory.h>
 #include <Functions/FunctionHelpers.h>
 #include <Functions/IFunction.h>
+#include <Functions/IFunctionAdaptors.h>
 #include <IO/WriteBufferFromVector.h>
 #include <IO/WriteHelpers.h>
 #include <Processors/Formats/IOutputFormat.h>
diff --git a/src/Functions/greater.cpp b/src/Functions/greater.cpp
index 2b87b376ce0..7d1b9401aee 100644
--- a/src/Functions/greater.cpp
+++ b/src/Functions/greater.cpp
@@ -7,6 +7,7 @@ namespace DB
 
 using FunctionGreater = FunctionComparison<GreaterOp, NameGreater>;
 using FunctionEquals = FunctionComparison<EqualsOp, NameEquals>;
+extern template class FunctionComparison<EqualsOp, NameEquals>;
 
 REGISTER_FUNCTION(Greater)
 {
diff --git a/src/Functions/greaterOrEquals.cpp b/src/Functions/greaterOrEquals.cpp
index c77ca585c76..b131b99db35 100644
--- a/src/Functions/greaterOrEquals.cpp
+++ b/src/Functions/greaterOrEquals.cpp
@@ -2,13 +2,14 @@
 #include <Functions/FunctionsComparison.h>
 #include <Functions/FunctionsLogical.h>
 
-
 namespace DB
 {
 
 using FunctionGreaterOrEquals = FunctionComparison<GreaterOrEqualsOp, NameGreaterOrEquals>;
 using FunctionGreater = FunctionComparison<GreaterOp, NameGreater>;
+extern template class FunctionComparison<GreaterOp, NameGreater>;
 using FunctionEquals = FunctionComparison<EqualsOp, NameEquals>;
+extern template class FunctionComparison<EqualsOp, NameEquals>;
 
 REGISTER_FUNCTION(GreaterOrEquals)
 {
diff --git a/src/Functions/if.cpp b/src/Functions/if.cpp
index 5e1e7067e86..ef358780b20 100644
--- a/src/Functions/if.cpp
+++ b/src/Functions/if.cpp
@@ -26,6 +26,7 @@
 #include <Functions/FunctionIfBase.h>
 #include <Functions/GatherUtils/Algorithms.h>
 #include <Functions/IFunction.h>
+#include <Functions/IFunctionAdaptors.h>
 #include <Interpreters/Context.h>
 #include <Interpreters/castColumn.h>
 #include <Common/assert_cast.h>
diff --git a/src/Functions/ifNotFinite.cpp b/src/Functions/ifNotFinite.cpp
index d7af10eec44..369d39b3bd2 100644
--- a/src/Functions/ifNotFinite.cpp
+++ b/src/Functions/ifNotFinite.cpp
@@ -45,7 +45,7 @@ public:
     {
         ColumnsWithTypeAndName is_finite_columns{arguments[0]};
         auto is_finite = FunctionFactory::instance().get("isFinite", context)->build(is_finite_columns);
-        auto res = is_finite->execute(is_finite_columns, is_finite->getResultType(), input_rows_count);
+        auto res = is_finite->execute(is_finite_columns, is_finite->getResultType(), input_rows_count, /* dry_run = */ false);
 
         ColumnsWithTypeAndName if_columns
         {
@@ -55,7 +55,7 @@ public:
         };
 
         auto func_if = FunctionFactory::instance().get("if", context)->build(if_columns);
-        return func_if->execute(if_columns, result_type, input_rows_count);
+        return func_if->execute(if_columns, result_type, input_rows_count, /* dry_run = */ false);
     }
 
 private:
diff --git a/src/Functions/ifNull.cpp b/src/Functions/ifNull.cpp
index 358a52c8394..f1186a9abdc 100644
--- a/src/Functions/ifNull.cpp
+++ b/src/Functions/ifNull.cpp
@@ -66,11 +66,11 @@ public:
 
         auto is_not_null = FunctionFactory::instance().get("isNotNull", context)->build(columns);
         auto is_not_null_type = std::make_shared<DataTypeUInt8>();
-        auto is_not_null_res = is_not_null->execute(columns, is_not_null_type, input_rows_count);
+        auto is_not_null_res = is_not_null->execute(columns, is_not_null_type, input_rows_count, /* dry_run = */ false);
 
         auto assume_not_null = FunctionFactory::instance().get("assumeNotNull", context)->build(columns);
         auto assume_not_null_type = removeNullable(arguments[0].type);
-        auto assume_nut_null_res = assume_not_null->execute(columns, assume_not_null_type, input_rows_count);
+        auto assume_nut_null_res = assume_not_null->execute(columns, assume_not_null_type, input_rows_count, /* dry_run = */ false);
 
         ColumnsWithTypeAndName if_columns
         {
@@ -80,7 +80,7 @@ public:
         };
 
         auto func_if = FunctionFactory::instance().get("if", context)->build(if_columns);
-        return func_if->execute(if_columns, result_type, input_rows_count);
+        return func_if->execute(if_columns, result_type, input_rows_count, /* dry_run = */ false);
     }
 
 private:
diff --git a/src/Functions/isDecimalOverflow.cpp b/src/Functions/isDecimalOverflow.cpp
index 4ea0deb0c4d..c49ca1e2827 100644
--- a/src/Functions/isDecimalOverflow.cpp
+++ b/src/Functions/isDecimalOverflow.cpp
@@ -7,6 +7,7 @@
 #include <Columns/ColumnDecimal.h>
 #include <Columns/ColumnConst.h>
 #include <Common/intExp.h>
+#include <Core/DecimalFunctions.h>
 
 
 namespace DB
diff --git a/src/Functions/less.cpp b/src/Functions/less.cpp
index 0998dc60292..f21bf8293fa 100644
--- a/src/Functions/less.cpp
+++ b/src/Functions/less.cpp
@@ -8,6 +8,7 @@ namespace DB
 
 using FunctionLess = FunctionComparison<LessOp, NameLess>;
 using FunctionEquals = FunctionComparison<EqualsOp, NameEquals>;
+extern template class FunctionComparison<EqualsOp, NameEquals>;
 
 REGISTER_FUNCTION(Less)
 {
diff --git a/src/Functions/lessOrEquals.cpp b/src/Functions/lessOrEquals.cpp
index e88ae34da75..3d1d9207112 100644
--- a/src/Functions/lessOrEquals.cpp
+++ b/src/Functions/lessOrEquals.cpp
@@ -8,7 +8,9 @@ namespace DB
 
 using FunctionLessOrEquals = FunctionComparison<LessOrEqualsOp, NameLessOrEquals>;
 using FunctionLess = FunctionComparison<LessOp, NameLess>;
+extern template class FunctionComparison<LessOp, NameLess>;
 using FunctionEquals = FunctionComparison<EqualsOp, NameEquals>;
+extern template class FunctionComparison<EqualsOp, NameEquals>;
 
 REGISTER_FUNCTION(LessOrEquals)
 {
diff --git a/src/Functions/makeDate.cpp b/src/Functions/makeDate.cpp
index 21d466d7708..bc94168d6ac 100644
--- a/src/Functions/makeDate.cpp
+++ b/src/Functions/makeDate.cpp
@@ -10,6 +10,7 @@
 #include <Columns/ColumnDecimal.h>
 #include <Columns/ColumnsDateTime.h>
 #include <Columns/ColumnsNumber.h>
+#include <Core/DecimalFunctions.h>
 #include <Interpreters/castColumn.h>
 
 #include <Common/DateLUT.h>
diff --git a/src/Functions/map.cpp b/src/Functions/map.cpp
index 893d0c51713..a20b7370893 100644
--- a/src/Functions/map.cpp
+++ b/src/Functions/map.cpp
@@ -133,13 +133,13 @@ public:
         const DataTypePtr & value_array_type = std::make_shared<DataTypeArray>(value_type);
 
         /// key_array = array(args[0], args[2]...)
-        ColumnPtr key_array = function_array->build(key_args)->execute(key_args, key_array_type, input_rows_count);
+        ColumnPtr key_array = function_array->build(key_args)->execute(key_args, key_array_type, input_rows_count, /* dry_run = */ false);
         /// value_array = array(args[1], args[3]...)
-        ColumnPtr value_array = function_array->build(value_args)->execute(value_args, value_array_type, input_rows_count);
+        ColumnPtr value_array = function_array->build(value_args)->execute(value_args, value_array_type, input_rows_count, /* dry_run = */ false);
 
         /// result = mapFromArrays(key_array, value_array)
         ColumnsWithTypeAndName map_args{{key_array, key_array_type, ""}, {value_array, value_array_type, ""}};
-        return function_map_from_arrays->build(map_args)->execute(map_args, result_type, input_rows_count);
+        return function_map_from_arrays->build(map_args)->execute(map_args, result_type, input_rows_count, /* dry_run = */ false);
     }
 
 private:
diff --git a/src/Functions/monthName.cpp b/src/Functions/monthName.cpp
index 5bd9cd934de..3405756db68 100644
--- a/src/Functions/monthName.cpp
+++ b/src/Functions/monthName.cpp
@@ -70,7 +70,7 @@ public:
         };
 
         auto date_name_func = function_resolver->build(temporary_columns);
-        return date_name_func->execute(temporary_columns, result_type, input_rows_count);
+        return date_name_func->execute(temporary_columns, result_type, input_rows_count, /* dry_run = */ false);
     }
 
 private:
diff --git a/src/Functions/multiIf.cpp b/src/Functions/multiIf.cpp
index 088d4d0555b..e43152ffdc2 100644
--- a/src/Functions/multiIf.cpp
+++ b/src/Functions/multiIf.cpp
@@ -1,5 +1,6 @@
 #include <Functions/FunctionFactory.h>
 #include <Functions/FunctionIfBase.h>
+#include <Functions/IFunctionAdaptors.h>
 #include <Columns/ColumnNullable.h>
 #include <Columns/ColumnConst.h>
 #include <Columns/ColumnsNumber.h>
diff --git a/src/Functions/multiMatchAny.cpp b/src/Functions/multiMatchAny.cpp
index 054a60fce2d..69b1feccd17 100644
--- a/src/Functions/multiMatchAny.cpp
+++ b/src/Functions/multiMatchAny.cpp
@@ -1,6 +1,7 @@
-#include "FunctionsMultiStringSearch.h"
-#include "FunctionFactory.h"
-#include "MultiMatchAnyImpl.h"
+#include <Functions/FunctionsMultiStringSearch.h>
+#include <Functions/FunctionFactory.h>
+#include <Functions/MultiMatchAnyImpl.h>
+#include <Functions/IFunctionAdaptors.h>
 
 
 namespace DB
diff --git a/src/Functions/nullIf.cpp b/src/Functions/nullIf.cpp
index 550287885a1..6c0a06029a6 100644
--- a/src/Functions/nullIf.cpp
+++ b/src/Functions/nullIf.cpp
@@ -49,7 +49,7 @@ public:
         /// nullIf(col1, col2) == if(col1 = col2, NULL, col1)
 
         auto equals_func = FunctionFactory::instance().get("equals", context)->build(arguments);
-        auto eq_res = equals_func->execute(arguments, equals_func->getResultType(), input_rows_count);
+        auto eq_res = equals_func->execute(arguments, equals_func->getResultType(), input_rows_count, /* dry_run = */ false);
 
         ColumnsWithTypeAndName if_columns
         {
@@ -59,7 +59,7 @@ public:
         };
 
         auto func_if = FunctionFactory::instance().get("if", context)->build(if_columns);
-        auto if_res = func_if->execute(if_columns, result_type, input_rows_count);
+        auto if_res = func_if->execute(if_columns, result_type, input_rows_count, /* dry_run = */ false);
 
         return makeNullable(if_res);
     }
diff --git a/src/Functions/printf.cpp b/src/Functions/printf.cpp
index fd76955159c..496fe96346a 100644
--- a/src/Functions/printf.cpp
+++ b/src/Functions/printf.cpp
@@ -242,7 +242,7 @@ public:
             }
         }
 
-        auto res = function_concat->build(concat_args)->execute(concat_args, std::make_shared<DataTypeString>(), input_rows_count);
+        auto res = function_concat->build(concat_args)->execute(concat_args, std::make_shared<DataTypeString>(), input_rows_count, /* dry_run = */ false);
         return res;
     }
 
diff --git a/src/Functions/reverse.cpp b/src/Functions/reverse.cpp
index 119bc614098..d692c42d847 100644
--- a/src/Functions/reverse.cpp
+++ b/src/Functions/reverse.cpp
@@ -3,6 +3,7 @@
 #include <Columns/ColumnArray.h>
 #include <Functions/FunctionFactory.h>
 #include <Functions/FunctionHelpers.h>
+#include <Functions/IFunctionAdaptors.h>
 #include <base/map.h>
 #include "reverse.h"
 
diff --git a/src/Functions/splitByRegexp.cpp b/src/Functions/splitByRegexp.cpp
index eab996c98bf..9c6f00721b2 100644
--- a/src/Functions/splitByRegexp.cpp
+++ b/src/Functions/splitByRegexp.cpp
@@ -3,6 +3,7 @@
 #include <Functions/FunctionFactory.h>
 #include <Functions/FunctionHelpers.h>
 #include <Functions/FunctionTokens.h>
+#include <Functions/IFunctionAdaptors.h>
 #include <Functions/Regexps.h>
 #include <Common/StringUtils.h>
 #include <base/map.h>
diff --git a/src/Functions/toBool.cpp b/src/Functions/toBool.cpp
index ac595d313e3..32dd759d50c 100644
--- a/src/Functions/toBool.cpp
+++ b/src/Functions/toBool.cpp
@@ -55,7 +55,7 @@ namespace
             };
 
             auto func_cast = createInternalCast(arguments[0], result_type, CastType::nonAccurate, {});
-            return func_cast->execute(cast_args, result_type, arguments[0].column->size());
+            return func_cast->execute(cast_args, result_type, arguments[0].column->size(), /* dry_run = */ false);
         }
     };
 }
diff --git a/src/Functions/transform.cpp b/src/Functions/transform.cpp
index e5445b36809..b6c28bc0b12 100644
--- a/src/Functions/transform.cpp
+++ b/src/Functions/transform.cpp
@@ -12,6 +12,7 @@
 #include <Functions/FunctionFactory.h>
 #include <Functions/FunctionHelpers.h>
 #include <Functions/IFunction.h>
+#include <Functions/IFunctionAdaptors.h>
 #include <Interpreters/castColumn.h>
 #include <Interpreters/convertFieldToType.h>
 #include <Common/HashTable/HashMap.h>
@@ -213,7 +214,7 @@ namespace
 
             auto impl = std::make_shared<FunctionToOverloadResolverAdaptor>(std::make_shared<FunctionTransform>())->build(args);
 
-            return impl->execute(args, result_type, input_rows_count);
+            return impl->execute(args, result_type, input_rows_count, /* dry_run = */ false);
         }
 
         void executeAnything(const IColumn * in, IColumn & column_result, const ColumnPtr default_non_const, const IColumn & in_casted, size_t input_rows_count) const
diff --git a/src/Functions/tupleHammingDistance.cpp b/src/Functions/tupleHammingDistance.cpp
index ffdf8c93f15..e45e73dfebc 100644
--- a/src/Functions/tupleHammingDistance.cpp
+++ b/src/Functions/tupleHammingDistance.cpp
@@ -119,7 +119,7 @@ public:
 
             ColumnWithTypeAndName column;
             column.type = elem_compare->getResultType();
-            column.column = elem_compare->execute({left, right}, column.type, input_rows_count);
+            column.column = elem_compare->execute({left, right}, column.type, input_rows_count, /* dry_run = */ false);
 
             if (i == 0)
             {
@@ -129,7 +129,7 @@ public:
             {
                 auto plus_elem = plus->build({res, column});
                 auto res_type = plus_elem->getResultType();
-                res.column = plus_elem->execute({res, column}, res_type, input_rows_count);
+                res.column = plus_elem->execute({res, column}, res_type, input_rows_count, /* dry_run = */ false);
                 res.type = res_type;
             }
         }
diff --git a/src/Functions/tupleNames.cpp b/src/Functions/tupleNames.cpp
index e444478c224..38d2fff3adb 100644
--- a/src/Functions/tupleNames.cpp
+++ b/src/Functions/tupleNames.cpp
@@ -4,6 +4,7 @@
 #include <DataTypes/DataTypeString.h>
 #include <DataTypes/DataTypeTuple.h>
 #include <Functions/FunctionFactory.h>
+#include <Functions/FunctionHelpers.h>
 
 namespace DB
 {
diff --git a/src/Functions/vectorFunctions.cpp b/src/Functions/vectorFunctions.cpp
index 5e23493c86d..7b309313abe 100644
--- a/src/Functions/vectorFunctions.cpp
+++ b/src/Functions/vectorFunctions.cpp
@@ -136,7 +136,7 @@ public:
             ColumnWithTypeAndName left{left_elements[i], left_types[i], {}};
             ColumnWithTypeAndName right{right_elements[i], right_types[i], {}};
             auto elem_func = func->build(ColumnsWithTypeAndName{left, right});
-            columns[i] = elem_func->execute({left, right}, elem_func->getResultType(), input_rows_count)
+            columns[i] = elem_func->execute({left, right}, elem_func->getResultType(), input_rows_count, /* dry_run = */ false)
                                   ->convertToFullColumnIfConst();
         }
 
@@ -221,7 +221,7 @@ public:
         {
             ColumnWithTypeAndName cur{cur_elements[i], cur_types[i], {}};
             auto elem_negate = negate->build(ColumnsWithTypeAndName{cur});
-            columns[i] = elem_negate->execute({cur}, elem_negate->getResultType(), input_rows_count)
+            columns[i] = elem_negate->execute({cur}, elem_negate->getResultType(), input_rows_count, /* dry_run = */ false)
                                     ->convertToFullColumnIfConst();
         }
 
@@ -295,7 +295,7 @@ public:
         {
             ColumnWithTypeAndName cur{cur_elements[i], cur_types[i], {}};
             auto elem_func = func->build(ColumnsWithTypeAndName{cur, p_column});
-            columns[i] = elem_func->execute({cur, p_column}, elem_func->getResultType(), input_rows_count)
+            columns[i] = elem_func->execute({cur, p_column}, elem_func->getResultType(), input_rows_count, /* dry_run = */ false)
                                   ->convertToFullColumnIfConst();
         }
 
@@ -413,7 +413,7 @@ public:
 
             ColumnWithTypeAndName column;
             column.type = elem_multiply->getResultType();
-            column.column = elem_multiply->execute({left, right}, column.type, input_rows_count);
+            column.column = elem_multiply->execute({left, right}, column.type, input_rows_count, /* dry_run = */ false);
 
             if (i == 0)
             {
@@ -423,7 +423,7 @@ public:
             {
                 auto plus_elem = plus->build({res, column});
                 auto res_type = plus_elem->getResultType();
-                res.column = plus_elem->execute({res, column}, res_type, input_rows_count);
+                res.column = plus_elem->execute({res, column}, res_type, input_rows_count, /* dry_run = */ false);
                 res.type = res_type;
             }
         }
@@ -510,7 +510,7 @@ public:
             ColumnWithTypeAndName column{cur_elements[i], cur_types[i], {}};
             auto elem_plus = plus->build(ColumnsWithTypeAndName{i == 0 ? arguments[0] : res, column});
             auto res_type = elem_plus->getResultType();
-            res.column = elem_plus->execute({i == 0 ? arguments[0] : res, column}, res_type, input_rows_count);
+            res.column = elem_plus->execute({i == 0 ? arguments[0] : res, column}, res_type, input_rows_count, /* dry_run = */ false);
             res.type = res_type;
         }
 
@@ -665,14 +665,14 @@ public:
             {
                 auto minus = FunctionFactory::instance().get("minus", context);
                 auto elem_minus = minus->build({left, arguments[1]});
-                last_column = elem_minus->execute({left, arguments[1]}, arguments[1].type, input_rows_count)
+                last_column = elem_minus->execute({left, arguments[1]}, arguments[1].type, input_rows_count, /* dry_run = */ false)
                                         ->convertToFullColumnIfConst();
             }
             else
             {
                 auto plus = FunctionFactory::instance().get("plus", context);
                 auto elem_plus = plus->build({left, arguments[1]});
-                last_column = elem_plus->execute({left, arguments[1]}, arguments[1].type, input_rows_count)
+                last_column = elem_plus->execute({left, arguments[1]}, arguments[1].type, input_rows_count, /* dry_run = */ false)
                                         ->convertToFullColumnIfConst();
             }
         }
@@ -682,7 +682,7 @@ public:
             {
                 auto negate = FunctionFactory::instance().get("negate", context);
                 auto elem_negate = negate->build({arguments[1]});
-                last_column = elem_negate->execute({arguments[1]}, arguments[1].type, input_rows_count);
+                last_column = elem_negate->execute({arguments[1]}, arguments[1].type, input_rows_count, /* dry_run = */ false);
             }
             else
             {
@@ -783,7 +783,7 @@ public:
 
             ColumnWithTypeAndName column;
             column.type = elem_abs->getResultType();
-            column.column = elem_abs->execute({cur}, column.type, input_rows_count);
+            column.column = elem_abs->execute({cur}, column.type, input_rows_count, /* dry_run = */ false);
 
             if (i == 0)
             {
@@ -793,7 +793,7 @@ public:
             {
                 auto plus_elem = plus->build({res, column});
                 auto res_type = plus_elem->getResultType();
-                res.column = plus_elem->execute({res, column}, res_type, input_rows_count);
+                res.column = plus_elem->execute({res, column}, res_type, input_rows_count, /* dry_run = */ false);
                 res.type = res_type;
             }
         }
@@ -885,7 +885,7 @@ public:
 
             ColumnWithTypeAndName column;
             column.type = elem_multiply->getResultType();
-            column.column = elem_multiply->execute({cur, cur}, column.type, input_rows_count);
+            column.column = elem_multiply->execute({cur, cur}, column.type, input_rows_count, /* dry_run = */ false);
 
             if (i == 0)
             {
@@ -895,7 +895,7 @@ public:
             {
                 auto plus_elem = plus->build({res, column});
                 auto res_type = plus_elem->getResultType();
-                res.column = plus_elem->execute({res, column}, res_type, input_rows_count);
+                res.column = plus_elem->execute({res, column}, res_type, input_rows_count, /* dry_run = */ false);
                 res.type = res_type;
             }
         }
@@ -949,7 +949,7 @@ public:
 
         auto sqrt = FunctionFactory::instance().get("sqrt", context);
         auto sqrt_elem = sqrt->build({squared_res});
-        return sqrt_elem->execute({squared_res}, sqrt_elem->getResultType(), input_rows_count);
+        return sqrt_elem->execute({squared_res}, sqrt_elem->getResultType(), input_rows_count, /* dry_run = */ false);
     }
 };
 using FunctionL2Norm = FunctionLNorm<L2Label>;
@@ -1036,7 +1036,7 @@ public:
 
             ColumnWithTypeAndName column;
             column.type = elem_abs->getResultType();
-            column.column = elem_abs->execute({cur}, column.type, input_rows_count);
+            column.column = elem_abs->execute({cur}, column.type, input_rows_count, /* dry_run = */ false);
 
             if (i == 0)
             {
@@ -1046,7 +1046,7 @@ public:
             {
                 auto max_elem = max->build({res, column});
                 auto res_type = max_elem->getResultType();
-                res.column = max_elem->execute({res, column}, res_type, input_rows_count);
+                res.column = max_elem->execute({res, column}, res_type, input_rows_count, /* dry_run = */ false);
                 res.type = res_type;
             }
         }
@@ -1163,14 +1163,14 @@ public:
         {
             ColumnWithTypeAndName cur{cur_elements[i], cur_types[i], {}};
             auto elem_abs = abs->build(ColumnsWithTypeAndName{cur});
-            cur.column = elem_abs->execute({cur}, elem_abs->getResultType(), input_rows_count);
+            cur.column = elem_abs->execute({cur}, elem_abs->getResultType(), input_rows_count, /* dry_run = */ false);
             cur.type = elem_abs->getResultType();
 
             auto elem_pow = pow->build(ColumnsWithTypeAndName{cur, p_column});
 
             ColumnWithTypeAndName column;
             column.type = elem_pow->getResultType();
-            column.column = elem_pow->execute({cur, p_column}, column.type, input_rows_count);
+            column.column = elem_pow->execute({cur, p_column}, column.type, input_rows_count, /* dry_run = */ false);
 
             if (i == 0)
             {
@@ -1180,7 +1180,7 @@ public:
             {
                 auto plus_elem = plus->build({res, column});
                 auto res_type = plus_elem->getResultType();
-                res.column = plus_elem->execute({res, column}, res_type, input_rows_count);
+                res.column = plus_elem->execute({res, column}, res_type, input_rows_count, /* dry_run = */ false);
                 res.type = res_type;
             }
         }
@@ -1188,7 +1188,7 @@ public:
         ColumnWithTypeAndName inv_p_column{DataTypeFloat64().createColumnConst(input_rows_count, 1 / p),
                                            std::make_shared<DataTypeFloat64>(), {}};
         auto pow_elem = pow->build({res, inv_p_column});
-        return pow_elem->execute({res, inv_p_column}, pow_elem->getResultType(), input_rows_count);
+        return pow_elem->execute({res, inv_p_column}, pow_elem->getResultType(), input_rows_count, /* dry_run = */ false);
     }
 };
 using FunctionLpNorm = FunctionLNorm<LpLabel>;
@@ -1247,12 +1247,12 @@ public:
         if constexpr (FuncLabel::name[0] == 'p')
         {
             auto func_elem = func->build({minus_res, arguments[2]});
-            return func_elem->execute({minus_res, arguments[2]}, func_elem->getResultType(), input_rows_count);
+            return func_elem->execute({minus_res, arguments[2]}, func_elem->getResultType(), input_rows_count, /* dry_run = */ false);
         }
         else
         {
             auto func_elem = func->build({minus_res});
-            return func_elem->execute({minus_res}, func_elem->getResultType(), input_rows_count);
+            return func_elem->execute({minus_res}, func_elem->getResultType(), input_rows_count, /* dry_run = */ false);
         }
     }
 };
@@ -1394,16 +1394,16 @@ public:
         ColumnWithTypeAndName multiply_result;
         multiply_result.type = multiply_elem->getResultType();
         multiply_result.column = multiply_elem->execute({first_norm, second_norm},
-                                                        multiply_result.type, input_rows_count);
+                                                        multiply_result.type, input_rows_count, /* dry_run = */ false);
 
         auto divide_elem = divide->build({dot_result, multiply_result});
         ColumnWithTypeAndName divide_result;
         divide_result.type = divide_elem->getResultType();
         divide_result.column = divide_elem->execute({dot_result, multiply_result},
-                                                    divide_result.type, input_rows_count);
+                                                    divide_result.type, input_rows_count, /* dry_run = */ false);
 
         auto minus_elem = minus->build({one, divide_result});
-        return minus_elem->execute({one, divide_result}, minus_elem->getResultType(), input_rows_count);
+        return minus_elem->execute({one, divide_result}, minus_elem->getResultType(), input_rows_count, /* dry_run = */ false);
     }
 };
 
diff --git a/src/Interpreters/ArrayJoinAction.cpp b/src/Interpreters/ArrayJoinAction.cpp
index f1952a21760..3983dde9db4 100644
--- a/src/Interpreters/ArrayJoinAction.cpp
+++ b/src/Interpreters/ArrayJoinAction.cpp
@@ -5,6 +5,7 @@
 #include <Columns/ColumnMap.h>
 #include <DataTypes/DataTypesNumber.h>
 #include <Functions/FunctionFactory.h>
+#include <Functions/IFunctionAdaptors.h>
 #include <Functions/array/length.h>
 #include <Functions/array/arrayResize.h>
 #include <Functions/array/emptyArrayToSingle.h>
@@ -166,7 +167,7 @@ ArrayJoinResultIterator::ArrayJoinResultIterator(const ArrayJoinAction * array_j
 
             ColumnWithTypeAndName array_col = convertArrayJoinColumn(src_col);
             ColumnsWithTypeAndName tmp_block{array_col}; //, {{}, uint64, {}}};
-            auto len_col = function_length->build(tmp_block)->execute(tmp_block, uint64, rows);
+            auto len_col = function_length->build(tmp_block)->execute(tmp_block, uint64, rows, /* dry_run = */ false);
             updateMaxLength(*max_length, *len_col);
         }
 
@@ -177,7 +178,7 @@ ArrayJoinResultIterator::ArrayJoinResultIterator(const ArrayJoinAction * array_j
 
             ColumnWithTypeAndName array_col = convertArrayJoinColumn(src_col);
             ColumnsWithTypeAndName tmp_block{array_col, column_of_max_length};
-            array_col.column = function_array_resize->build(tmp_block)->execute(tmp_block, array_col.type, rows);
+            array_col.column = function_array_resize->build(tmp_block)->execute(tmp_block, array_col.type, rows, /* dry_run = */ false);
 
             src_col = std::move(array_col);
             any_array_map_ptr = src_col.column->convertToFullColumnIfConst();
@@ -194,7 +195,7 @@ ArrayJoinResultIterator::ArrayJoinResultIterator(const ArrayJoinAction * array_j
             const auto & src_col = block.getByName(name);
             ColumnWithTypeAndName array_col = convertArrayJoinColumn(src_col);
             ColumnsWithTypeAndName tmp_block{array_col};
-            non_empty_array_columns[name] = function_builder->build(tmp_block)->execute(tmp_block, array_col.type, array_col.column->size());
+            non_empty_array_columns[name] = function_builder->build(tmp_block)->execute(tmp_block, array_col.type, array_col.column->size(), /* dry_run = */ false);
         }
 
         any_array_map_ptr = non_empty_array_columns.begin()->second->convertToFullColumnIfConst();
diff --git a/src/Interpreters/ExpressionJIT.cpp b/src/Interpreters/ExpressionJIT.cpp
index 16275b23053..9baa48515c7 100644
--- a/src/Interpreters/ExpressionJIT.cpp
+++ b/src/Interpreters/ExpressionJIT.cpp
@@ -266,7 +266,7 @@ public:
     {
         const auto & type = function.getArgumentTypes().at(0);
         ColumnsWithTypeAndName args{{type->createColumnConst(1, value), type, "x" }};
-        auto col = function.execute(args, function.getResultType(), 1);
+        auto col = function.execute(args, function.getResultType(), 1, /* dry_run = */ false);
         col->get(0, value);
     }
 
diff --git a/src/Interpreters/RowRefs.cpp b/src/Interpreters/RowRefs.cpp
index 26524218b2b..17687296e81 100644
--- a/src/Interpreters/RowRefs.cpp
+++ b/src/Interpreters/RowRefs.cpp
@@ -1,10 +1,11 @@
 #include <Interpreters/RowRefs.h>
 
-#include <Common/RadixSort.h>
+#include <Columns/ColumnDecimal.h>
 #include <Columns/IColumn.h>
-#include <DataTypes/IDataType.h>
 #include <Core/Joins.h>
+#include <DataTypes/IDataType.h>
 #include <base/types.h>
+#include <Common/RadixSort.h>
 
 
 namespace DB
diff --git a/src/Interpreters/RowRefs.h b/src/Interpreters/RowRefs.h
index f8ac68191d6..4c8abd44025 100644
--- a/src/Interpreters/RowRefs.h
+++ b/src/Interpreters/RowRefs.h
@@ -7,7 +7,6 @@
 #include <optional>
 #include <variant>
 
-#include <Columns/ColumnDecimal.h>
 #include <Columns/ColumnVector.h>
 #include <Columns/IColumn.h>
 #include <Core/Joins.h>
diff --git a/src/Interpreters/castColumn.cpp b/src/Interpreters/castColumn.cpp
index 3890c81934a..caa52bb040f 100644
--- a/src/Interpreters/castColumn.cpp
+++ b/src/Interpreters/castColumn.cpp
@@ -34,8 +34,8 @@ static ColumnPtr castColumn(CastType cast_type, const ColumnWithTypeAndName & ar
     FunctionBasePtr func_cast = cache ? cache->getOrSet(cast_type, from_name, to_name, std::move(get_cast_func)) : get_cast_func();
 
     if (cast_type == CastType::accurateOrNull)
-        return func_cast->execute(arguments, makeNullable(type), arg.column->size());
-    return func_cast->execute(arguments, type, arg.column->size());
+        return func_cast->execute(arguments, makeNullable(type), arg.column->size(), /* dry_run = */ false);
+    return func_cast->execute(arguments, type, arg.column->size(), /* dry_run = */ false);
 }
 
 ColumnPtr castColumn(const ColumnWithTypeAndName & arg, const DataTypePtr & type, InternalCastFunctionCache * cache)
diff --git a/src/Planner/Utils.cpp b/src/Planner/Utils.cpp
index c2deafb4ff1..589ef8bcd96 100644
--- a/src/Planner/Utils.cpp
+++ b/src/Planner/Utils.cpp
@@ -11,11 +11,13 @@
 #include <DataTypes/DataTypeNullable.h>
 
 #include <Columns/getLeastSuperColumn.h>
+#include <Columns/ColumnConst.h>
 #include <Columns/ColumnSet.h>
 
 #include <IO/WriteBufferFromString.h>
 
 #include <Functions/FunctionFactory.h>
+#include <Functions/IFunctionAdaptors.h>
 #include <Functions/indexHint.h>
 
 #include <Storages/StorageDummy.h>
diff --git a/src/Processors/Formats/Impl/CHColumnToArrowColumn.cpp b/src/Processors/Formats/Impl/CHColumnToArrowColumn.cpp
index 30301b242db..ecc22d48ae5 100644
--- a/src/Processors/Formats/Impl/CHColumnToArrowColumn.cpp
+++ b/src/Processors/Formats/Impl/CHColumnToArrowColumn.cpp
@@ -2,6 +2,7 @@
 
 #if USE_ARROW || USE_PARQUET
 
+#include <Core/DecimalFunctions.h>
 #include <Columns/ColumnFixedString.h>
 #include <Columns/ColumnNullable.h>
 #include <Columns/ColumnString.h>
diff --git a/src/Processors/QueryPlan/FilterStep.cpp b/src/Processors/QueryPlan/FilterStep.cpp
index 9d0dd37b268..49190701553 100644
--- a/src/Processors/QueryPlan/FilterStep.cpp
+++ b/src/Processors/QueryPlan/FilterStep.cpp
@@ -7,6 +7,7 @@
 #include <Common/JSONBuilder.h>
 #include <DataTypes/DataTypeFactory.h>
 #include <DataTypes/DataTypeLowCardinality.h>
+#include <DataTypes/DataTypeNullable.h>
 #include <DataTypes/DataTypesNumber.h>
 #include <Functions/IFunction.h>
 #include <stack>
diff --git a/src/Processors/Transforms/WindowTransform.cpp b/src/Processors/Transforms/WindowTransform.cpp
index 597a6861eeb..9b9c8fccfef 100644
--- a/src/Processors/Transforms/WindowTransform.cpp
+++ b/src/Processors/Transforms/WindowTransform.cpp
@@ -2350,7 +2350,7 @@ struct WindowFunctionLagLeadInFrame final : public StatelessWindowFunction
             }
         };
 
-        return func_cast->execute(arguments, argument_types[0], columns[idx[2]]->size());
+        return func_cast->execute(arguments, argument_types[0], columns[idx[2]]->size(), /* dry_run = */ false);
     }
 
     static DataTypePtr createResultType(const DataTypes & argument_types_, const std::string & name_)
diff --git a/src/Storages/MergeTree/KeyCondition.cpp b/src/Storages/MergeTree/KeyCondition.cpp
index b8c03ba2cd6..1bf568eebba 100644
--- a/src/Storages/MergeTree/KeyCondition.cpp
+++ b/src/Storages/MergeTree/KeyCondition.cpp
@@ -17,6 +17,7 @@
 #include <Functions/indexHint.h>
 #include <Functions/CastOverloadResolver.h>
 #include <Functions/IFunction.h>
+#include <Functions/IFunctionAdaptors.h>
 #include <Functions/IFunctionDateOrDateTime.h>
 #include <Functions/geometryConverters.h>
 #include <Common/FieldVisitorToString.h>
@@ -905,7 +906,7 @@ static Field applyFunctionForField(
         { arg_type->createColumnConst(1, arg_value), arg_type, "x" },
     };
 
-    auto col = func->execute(columns, func->getResultType(), 1);
+    auto col = func->execute(columns, func->getResultType(), 1, /* dry_run = */ false);
     return (*col)[0];
 }
 
@@ -939,7 +940,7 @@ static FieldRef applyFunction(const FunctionBasePtr & func, const DataTypePtr &
         /// When cache is missed, we calculate the whole column where the field comes from. This will avoid repeated calculation.
         ColumnsWithTypeAndName args{(*columns)[field.column_idx]};
         field.columns->emplace_back(ColumnWithTypeAndName {nullptr, func->getResultType(), result_name});
-        (*columns)[result_idx].column = func->execute(args, (*columns)[result_idx].type, columns->front().column->size());
+        (*columns)[result_idx].column = func->execute(args, (*columns)[result_idx].type, columns->front().column->size(), /* dry_run = */ false);
     }
 
     return {field.columns, field.row_idx, result_idx};
@@ -1012,7 +1013,7 @@ bool applyFunctionChainToColumn(
             return false;
 
         result_column = castColumnAccurate({result_column, result_type, ""}, argument_type);
-        result_column = func->execute({{result_column, argument_type, ""}}, func->getResultType(), result_column->size());
+        result_column = func->execute({{result_column, argument_type, ""}}, func->getResultType(), result_column->size(), /* dry_run = */ false);
         result_type = func->getResultType();
 
         // Transforming nullable columns to the nested ones, in case no nulls found
diff --git a/src/Storages/MergeTree/MergeTreeIndexSet.cpp b/src/Storages/MergeTree/MergeTreeIndexSet.cpp
index 142a09cc086..2b18f5fe1e5 100644
--- a/src/Storages/MergeTree/MergeTreeIndexSet.cpp
+++ b/src/Storages/MergeTree/MergeTreeIndexSet.cpp
@@ -12,6 +12,7 @@
 #include <Parsers/ASTSelectQuery.h>
 
 #include <Functions/FunctionFactory.h>
+#include <Functions/IFunctionAdaptors.h>
 #include <Functions/indexHint.h>
 #include <Planner/PlannerActionsVisitor.h>
 
diff --git a/src/Storages/MergeTree/MergeTreeSplitPrewhereIntoReadSteps.cpp b/src/Storages/MergeTree/MergeTreeSplitPrewhereIntoReadSteps.cpp
index 1cc4006a285..4870152d9ce 100644
--- a/src/Storages/MergeTree/MergeTreeSplitPrewhereIntoReadSteps.cpp
+++ b/src/Storages/MergeTree/MergeTreeSplitPrewhereIntoReadSteps.cpp
@@ -3,6 +3,7 @@
 #include <Functions/IFunctionAdaptors.h>
 #include <Storages/SelectQueryInfo.h>
 #include <Storages/MergeTree/MergeTreeRangeReader.h>
+#include <DataTypes/DataTypeNullable.h>
 #include <DataTypes/DataTypeString.h>
 #include <DataTypes/DataTypeLowCardinality.h>
 #include <Interpreters/ExpressionActions.h>
diff --git a/src/Storages/removeGroupingFunctionSpecializations.cpp b/src/Storages/removeGroupingFunctionSpecializations.cpp
index 668596756a1..36bcc8816aa 100644
--- a/src/Storages/removeGroupingFunctionSpecializations.cpp
+++ b/src/Storages/removeGroupingFunctionSpecializations.cpp
@@ -4,6 +4,7 @@
 #include <Analyzer/ColumnNode.h>
 #include <Analyzer/FunctionNode.h>
 #include <Common/Exception.h>
+#include <Functions/IFunctionAdaptors.h>
 #include <Functions/grouping.h>
 
 namespace DB