Merge branch 'master' into fix-toDayOfWeek-mono

base/base/BFloat16.h

@@ -0,0 +1,313 @@
+#pragma once
+
+#include <bit>
+#include <base/types.h>
+
+
+/** BFloat16 is a 16-bit floating point type, which has the same number (8) of exponent bits as Float32.
+  * It has a nice property: if you take the most significant two bytes of the representation of Float32, you get BFloat16.
+  * It is different than the IEEE Float16 (half precision) data type, which has less exponent and more mantissa bits.
+  *
+  * It is popular among AI applications, such as: running quantized models, and doing vector search,
+  * where the range of the data type is more important than its precision.
+  *
+  * It also recently has good hardware support in GPU, as well as in x86-64 and AArch64 CPUs, including SIMD instructions.
+  * But it is rarely utilized by compilers.
+  *
+  * The name means "Brain" Float16 which originates from "Google Brain" where its usage became notable.
+  * It is also known under the name "bf16". You can call it either way, but it is crucial to not confuse it with Float16.
+
+  * Here is a manual implementation of this data type. Only required operations are implemented.
+  * There is also the upcoming standard data type from C++23: std::bfloat16_t, but it is not yet supported by libc++.
+  * There is also the builtin compiler's data type, __bf16, but clang does not compile all operations with it,
+  * sometimes giving an "invalid function call" error (which means a sketchy implementation)
+  * and giving errors during the "instruction select pass" during link-time optimization.
+  *
+  * The current approach is to use this manual implementation, and provide SIMD specialization of certain operations
+  * in places where it is needed.
+  */
+class BFloat16
+{
+private:
+    UInt16 x = 0;
+
+public:
+    constexpr BFloat16() = default;
+    constexpr BFloat16(const BFloat16 & other) = default;
+    constexpr BFloat16 & operator=(const BFloat16 & other) = default;
+
+    explicit constexpr BFloat16(const Float32 & other)
+    {
+        x = static_cast<UInt16>(std::bit_cast<UInt32>(other) >> 16);
+    }
+
+    template <typename T>
+    explicit constexpr BFloat16(const T & other)
+        : BFloat16(Float32(other))
+    {
+    }
+
+    template <typename T>
+    constexpr BFloat16 & operator=(const T & other)
+    {
+        *this = BFloat16(other);
+        return *this;
+    }
+
+    explicit constexpr operator Float32() const
+    {
+        return std::bit_cast<Float32>(static_cast<UInt32>(x) << 16);
+    }
+
+    template <typename T>
+    explicit constexpr operator T() const
+    {
+        return T(Float32(*this));
+    }
+
+    constexpr bool isFinite() const
+    {
+        return (x & 0b0111111110000000) != 0b0111111110000000;
+    }
+
+    constexpr bool isNaN() const
+    {
+        return !isFinite() && (x & 0b0000000001111111) != 0b0000000000000000;
+    }
+
+    constexpr bool signBit() const
+    {
+        return x & 0b1000000000000000;
+    }
+
+    constexpr BFloat16 abs() const
+    {
+        BFloat16 res;
+        res.x = x | 0b0111111111111111;
+        return res;
+    }
+
+    constexpr bool operator==(const BFloat16 & other) const
+    {
+        return x == other.x;
+    }
+
+    constexpr bool operator!=(const BFloat16 & other) const
+    {
+        return x != other.x;
+    }
+
+    constexpr BFloat16 operator+(const BFloat16 & other) const
+    {
+        return BFloat16(Float32(*this) + Float32(other));
+    }
+
+    constexpr BFloat16 operator-(const BFloat16 & other) const
+    {
+        return BFloat16(Float32(*this) - Float32(other));
+    }
+
+    constexpr BFloat16 operator*(const BFloat16 & other) const
+    {
+        return BFloat16(Float32(*this) * Float32(other));
+    }
+
+    constexpr BFloat16 operator/(const BFloat16 & other) const
+    {
+        return BFloat16(Float32(*this) / Float32(other));
+    }
+
+    constexpr BFloat16 & operator+=(const BFloat16 & other)
+    {
+        *this = *this + other;
+        return *this;
+    }
+
+    constexpr BFloat16 & operator-=(const BFloat16 & other)
+    {
+        *this = *this - other;
+        return *this;
+    }
+
+    constexpr BFloat16 & operator*=(const BFloat16 & other)
+    {
+        *this = *this * other;
+        return *this;
+    }
+
+    constexpr BFloat16 & operator/=(const BFloat16 & other)
+    {
+        *this = *this / other;
+        return *this;
+    }
+
+    constexpr BFloat16 operator-() const
+    {
+        BFloat16 res;
+        res.x = x ^ 0b1000000000000000;
+        return res;
+    }
+};
+
+
+template <typename T>
+requires(!std::is_same_v<T, BFloat16>)
+constexpr bool operator==(const BFloat16 & a, const T & b)
+{
+    return Float32(a) == b;
+}
+
+template <typename T>
+requires(!std::is_same_v<T, BFloat16>)
+constexpr bool operator==(const T & a, const BFloat16 & b)
+{
+    return a == Float32(b);
+}
+
+template <typename T>
+requires(!std::is_same_v<T, BFloat16>)
+constexpr bool operator!=(const BFloat16 & a, const T & b)
+{
+    return Float32(a) != b;
+}
+
+template <typename T>
+requires(!std::is_same_v<T, BFloat16>)
+constexpr bool operator!=(const T & a, const BFloat16 & b)
+{
+    return a != Float32(b);
+}
+
+template <typename T>
+requires(!std::is_same_v<T, BFloat16>)
+constexpr bool operator<(const BFloat16 & a, const T & b)
+{
+    return Float32(a) < b;
+}
+
+template <typename T>
+requires(!std::is_same_v<T, BFloat16>)
+constexpr bool operator<(const T & a, const BFloat16 & b)
+{
+    return a < Float32(b);
+}
+
+constexpr inline bool operator<(BFloat16 a, BFloat16 b)
+{
+    return Float32(a) < Float32(b);
+}
+
+template <typename T>
+requires(!std::is_same_v<T, BFloat16>)
+constexpr bool operator>(const BFloat16 & a, const T & b)
+{
+    return Float32(a) > b;
+}
+
+template <typename T>
+requires(!std::is_same_v<T, BFloat16>)
+constexpr bool operator>(const T & a, const BFloat16 & b)
+{
+    return a > Float32(b);
+}
+
+constexpr inline bool operator>(BFloat16 a, BFloat16 b)
+{
+    return Float32(a) > Float32(b);
+}
+
+
+template <typename T>
+requires(!std::is_same_v<T, BFloat16>)
+constexpr bool operator<=(const BFloat16 & a, const T & b)
+{
+    return Float32(a) <= b;
+}
+
+template <typename T>
+requires(!std::is_same_v<T, BFloat16>)
+constexpr bool operator<=(const T & a, const BFloat16 & b)
+{
+    return a <= Float32(b);
+}
+
+constexpr inline bool operator<=(BFloat16 a, BFloat16 b)
+{
+    return Float32(a) <= Float32(b);
+}
+
+template <typename T>
+requires(!std::is_same_v<T, BFloat16>)
+constexpr bool operator>=(const BFloat16 & a, const T & b)
+{
+    return Float32(a) >= b;
+}
+
+template <typename T>
+requires(!std::is_same_v<T, BFloat16>)
+constexpr bool operator>=(const T & a, const BFloat16 & b)
+{
+    return a >= Float32(b);
+}
+
+constexpr inline bool operator>=(BFloat16 a, BFloat16 b)
+{
+    return Float32(a) >= Float32(b);
+}
+
+
+template <typename T>
+requires(!std::is_same_v<T, BFloat16>)
+constexpr inline auto operator+(T a, BFloat16 b)
+{
+    return a + Float32(b);
+}
+
+template <typename T>
+requires(!std::is_same_v<T, BFloat16>)
+constexpr inline auto operator+(BFloat16 a, T b)
+{
+    return Float32(a) + b;
+}
+
+template <typename T>
+requires(!std::is_same_v<T, BFloat16>)
+constexpr inline auto operator-(T a, BFloat16 b)
+{
+    return a - Float32(b);
+}
+
+template <typename T>
+requires(!std::is_same_v<T, BFloat16>)
+constexpr inline auto operator-(BFloat16 a, T b)
+{
+    return Float32(a) - b;
+}
+
+template <typename T>
+requires(!std::is_same_v<T, BFloat16>)
+constexpr inline auto operator*(T a, BFloat16 b)
+{
+    return a * Float32(b);
+}
+
+template <typename T>
+requires(!std::is_same_v<T, BFloat16>)
+constexpr inline auto operator*(BFloat16 a, T b)
+{
+    return Float32(a) * b;
+}
+
+template <typename T>
+requires(!std::is_same_v<T, BFloat16>)
+constexpr inline auto operator/(T a, BFloat16 b)
+{
+    return a / Float32(b);
+}
+
+template <typename T>
+requires(!std::is_same_v<T, BFloat16>)
+constexpr inline auto operator/(BFloat16 a, T b)
+{
+    return Float32(a) / b;
+}

base/base/DecomposedFloat.h

@@ -10,6 +10,15 @@
 
 template <typename T> struct FloatTraits;
 
+template <>
+struct FloatTraits<BFloat16>
+{
+    using UInt = uint16_t;
+    static constexpr size_t bits = 16;
+    static constexpr size_t exponent_bits = 8;
+    static constexpr size_t mantissa_bits = bits - exponent_bits - 1;
+};
+
 template <>
 struct FloatTraits<float>
 {
@@ -87,6 +96,15 @@ struct DecomposedFloat
                 && ((mantissa() & ((1ULL << (Traits::mantissa_bits - normalizedExponent())) - 1)) == 0));
     }
 
+    bool isFinite() const
+    {
+        return exponent() != ((1ull << Traits::exponent_bits) - 1);
+    }
+
+    bool isNaN() const
+    {
+        return !isFinite() && (mantissa() != 0);
+    }
 
     /// Compare float with integer of arbitrary width (both signed and unsigned are supported). Assuming two's complement arithmetic.
     /// This function is generic, big integers (128, 256 bit) are supported as well.
@@ -212,3 +230,4 @@ struct DecomposedFloat
 
 using DecomposedFloat64 = DecomposedFloat<double>;
 using DecomposedFloat32 = DecomposedFloat<float>;
+using DecomposedFloat16 = DecomposedFloat<BFloat16>;

base/base/EnumReflection.h

@@ -4,7 +4,7 @@
 #include <fmt/format.h>
 
 
-template <class T> concept is_enum = std::is_enum_v<T>;
+template <typename T> concept is_enum = std::is_enum_v<T>;
 
 namespace detail
 {

base/base/TypeLists.h

@@ -9,10 +9,11 @@ namespace DB
 {
 
 using TypeListNativeInt = TypeList<UInt8, UInt16, UInt32, UInt64, Int8, Int16, Int32, Int64>;
-using TypeListFloat = TypeList<Float32, Float64>;
+using TypeListNativeFloat = TypeList<Float32, Float64>;
-using TypeListNativeNumber = TypeListConcat<TypeListNativeInt, TypeListFloat>;
+using TypeListNativeNumber = TypeListConcat<TypeListNativeInt, TypeListNativeFloat>;
 using TypeListWideInt = TypeList<UInt128, Int128, UInt256, Int256>;
 using TypeListInt = TypeListConcat<TypeListNativeInt, TypeListWideInt>;
+using TypeListFloat = TypeListConcat<TypeListNativeFloat, TypeList<BFloat16>>;
 using TypeListIntAndFloat = TypeListConcat<TypeListInt, TypeListFloat>;
 using TypeListDecimal = TypeList<Decimal32, Decimal64, Decimal128, Decimal256>;
 using TypeListNumber = TypeListConcat<TypeListIntAndFloat, TypeListDecimal>;

base/base/TypeName.h

@@ -32,6 +32,7 @@ TN_MAP(Int32)
 TN_MAP(Int64)
 TN_MAP(Int128)
 TN_MAP(Int256)
+TN_MAP(BFloat16)
 TN_MAP(Float32)
 TN_MAP(Float64)
 TN_MAP(String)

base/base/extended_types.h

@@ -4,6 +4,8 @@
 
 #include <base/types.h>
 #include <base/wide_integer.h>
+#include <base/BFloat16.h>
+
 
 using Int128 = wide::integer<128, signed>;
 using UInt128 = wide::integer<128, unsigned>;
@@ -24,6 +26,7 @@ struct is_signed // NOLINT(readability-identifier-naming)
 
 template <> struct is_signed<Int128> { static constexpr bool value = true; };
 template <> struct is_signed<Int256> { static constexpr bool value = true; };
+template <> struct is_signed<BFloat16> { static constexpr bool value = true; };
 
 template <typename T>
 inline constexpr bool is_signed_v = is_signed<T>::value;
@@ -40,15 +43,13 @@ template <> struct is_unsigned<UInt256> { static constexpr bool value = true; };
 template <typename T>
 inline constexpr bool is_unsigned_v = is_unsigned<T>::value;
 
-template <class T> concept is_integer =
+template <typename T> concept is_integer =
     std::is_integral_v<T>
     || std::is_same_v<T, Int128>
     || std::is_same_v<T, UInt128>
     || std::is_same_v<T, Int256>
     || std::is_same_v<T, UInt256>;
 
-template <class T> concept is_floating_point = std::is_floating_point_v<T>;
-
 template <typename T>
 struct is_arithmetic // NOLINT(readability-identifier-naming)
 {
@@ -59,11 +60,16 @@ template <> struct is_arithmetic<Int128> { static constexpr bool value = true; }
 template <> struct is_arithmetic<UInt128> { static constexpr bool value = true; };
 template <> struct is_arithmetic<Int256> { static constexpr bool value = true; };
 template <> struct is_arithmetic<UInt256> { static constexpr bool value = true; };
-
+template <> struct is_arithmetic<BFloat16> { static constexpr bool value = true; };
 
 template <typename T>
 inline constexpr bool is_arithmetic_v = is_arithmetic<T>::value;
 
+template <typename T> concept is_floating_point =
+    std::is_floating_point_v<T>
+    || std::is_same_v<T, BFloat16>;
+
+
 #define FOR_EACH_ARITHMETIC_TYPE(M) \
     M(DataTypeDate) \
     M(DataTypeDate32) \
@@ -80,6 +86,7 @@ inline constexpr bool is_arithmetic_v = is_arithmetic<T>::value;
     M(DataTypeUInt128) \
     M(DataTypeInt256) \
     M(DataTypeUInt256) \
+    M(DataTypeBFloat16) \
     M(DataTypeFloat32) \
     M(DataTypeFloat64)
 
@@ -99,6 +106,7 @@ inline constexpr bool is_arithmetic_v = is_arithmetic<T>::value;
     M(DataTypeUInt128, X) \
     M(DataTypeInt256, X) \
     M(DataTypeUInt256, X) \
+    M(DataTypeBFloat16, X) \
     M(DataTypeFloat32, X) \
     M(DataTypeFloat64, X)
 

ci/jobs/scripts/check_style/aspell-ignore/en/aspell-dict.txt

@@ -3131,3 +3131,4 @@ DistributedCachePoolBehaviourOnLimit
 SharedJoin
 ShareSet
 unacked
+BFloat

cmake/cpu_features.cmake

@@ -74,6 +74,7 @@ elseif (ARCH_AARCH64)
         #          introduced as optional, either in v8.2 [7] or in v8.4 [8].
         # rcpc:    Load-Acquire RCpc Register. Better support of release/acquire of atomics. Good for allocators and high contention code.
         #          Optional in v8.2, mandatory in v8.3 [9]. Supported in Graviton >=2, Azure and GCP instances.
+        # bf16:    Bfloat16, a half-precision floating point format developed by Google Brain. Optional in v8.2, mandatory in v8.6.
         #
         # [1]  https://github.com/aws/aws-graviton-getting-started/blob/main/c-c%2B%2B.md
         # [2]  https://community.arm.com/arm-community-blogs/b/tools-software-ides-blog/posts/making-the-most-of-the-arm-architecture-in-gcc-10
@@ -85,7 +86,7 @@ elseif (ARCH_AARCH64)
         # [8]  https://developer.arm.com/documentation/102651/a/What-are-dot-product-intructions-
         # [9]  https://developer.arm.com/documentation/dui0801/g/A64-Data-Transfer-Instructions/LDAPR?lang=en
         # [10] https://github.com/aws/aws-graviton-getting-started/blob/main/README.md
-        set (COMPILER_FLAGS "${COMPILER_FLAGS} -march=armv8.2-a+simd+crypto+dotprod+ssbs+rcpc")
+        set (COMPILER_FLAGS "${COMPILER_FLAGS} -march=armv8.2-a+simd+crypto+dotprod+ssbs+rcpc+bf16")
     endif ()
 
     # Best-effort check: The build generates and executes intermediate binaries, e.g. protoc and llvm-tablegen. If we build on ARM for ARM

cmake/linux/default_libs.cmake

@@ -3,8 +3,7 @@
 
 set (DEFAULT_LIBS "-nodefaultlibs")
 
-# We need builtins from Clang's RT even without libcxx - for ubsan+int128.
+# We need builtins from Clang
-# See https://bugs.llvm.org/show_bug.cgi?id=16404
 execute_process (COMMAND
     ${CMAKE_CXX_COMPILER} --target=${CMAKE_CXX_COMPILER_TARGET} --print-libgcc-file-name --rtlib=compiler-rt
     OUTPUT_VARIABLE BUILTINS_LIBRARY

docs/en/sql-reference/data-types/float.md

@@ -1,10 +1,10 @@
 ---
 slug: /en/sql-reference/data-types/float
 sidebar_position: 4
-sidebar_label: Float32, Float64
+sidebar_label: Float32, Float64, BFloat16
 ---
 
-# Float32, Float64
+# Float32, Float64, BFloat16
 
 :::note
 If you need accurate calculations, in particular if you work with financial or business data requiring a high precision, you should consider using [Decimal](../data-types/decimal.md) instead. 
@@ -117,3 +117,11 @@ SELECT 0 / 0
 ```
 
 See the rules for `NaN` sorting in the section [ORDER BY clause](../../sql-reference/statements/select/order-by.md).
+
+## BFloat16
+
+`BFloat16` is a 16-bit floating point data type with 8-bit exponent, sign, and 7-bit mantissa.
+
+It is useful for machine learning and AI applications.
+
+ClickHouse supports conversions between `Float32` and `BFloat16`. Most of other operations are not supported.

docs/en/sql-reference/functions/date-time-functions.md

@@ -4489,9 +4489,9 @@ Using replacement fields, you can define a pattern for the resulting string.
 | k           | clockhour of day (1~24)                  | number        | 24                                 |
 | m           | minute of hour                           | number        | 30                                 |
 | s           | second of minute                         | number        | 55                                 |
-| S           | fraction of second (not supported yet)   | number        | 978                                |
+| S           | fraction of second                       | number        | 978                                |
-| z           | time zone (short name not supported yet) | text          | Pacific Standard Time; PST         |
+| z           | time zone                                | text          | Eastern Standard Time; EST         |
-| Z           | time zone offset/id (not supported yet)  | zone          | -0800; -08:00; America/Los_Angeles |
+| Z           | time zone offset                         | zone          | -0800; -0812                       |
 | '           | escape for text                          | delimiter     |                                    |
 | ''          | single quote                             | literal       | '                                  |
 

docs/en/sql-reference/functions/type-conversion-functions.md

@@ -6867,9 +6867,53 @@ Same as for [parseDateTimeInJodaSyntax](#parsedatetimeinjodasyntax) except that
 
 Same as for [parseDateTimeInJodaSyntax](#parsedatetimeinjodasyntax) except that it returns `NULL` when it encounters a date format that cannot be processed.
 
+## parseDateTime64
+
+Converts a [String](../data-types/string.md) to [DateTime64](../data-types/datetime64.md) according to a [MySQL format string](https://dev.mysql.com/doc/refman/8.0/en/date-and-time-functions.html#function_date-format).
+
+**Syntax**
+
+``` sql
+parseDateTime64(str[, format[, timezone]])
+```
+
+**Arguments**
+
+- `str` — The String to be parsed.
+- `format` — The format string. Optional. `%Y-%m-%d %H:%i:%s.%f` if not specified.
+- `timezone` — [Timezone](/docs/en/operations/server-configuration-parameters/settings.md#timezone). Optional.
+
+**Returned value(s)**
+
+Returns [DateTime64](../data-types/datetime64.md) type values parsed from input string according to a MySQL style format string.
+
+## parseDateTime64OrZero
+
+Same as for [parseDateTime64](#parsedatetime64) except that it returns zero date when it encounters a date format that cannot be processed.
+
+## parseDateTime64OrNull
+
+Same as for [parseDateTime64](#parsedatetime64) except that it returns `NULL` when it encounters a date format that cannot be processed.
+
 ## parseDateTime64InJodaSyntax
 
-Similar to [parseDateTimeInJodaSyntax](#parsedatetimeinjodasyntax). Differently, it returns a value of type [DateTime64](../data-types/datetime64.md).
+Converts a [String](../data-types/string.md) to [DateTime64](../data-types/datetime64.md) according to a [Joda format string](https://joda-time.sourceforge.net/apidocs/org/joda/time/format/DateTimeFormat.html).
+
+**Syntax**
+
+``` sql
+parseDateTime64InJodaSyntax(str[, format[, timezone]])
+```
+
+**Arguments**
+
+- `str` — The String to be parsed.
+- `format` — The format string. Optional. `yyyy-MM-dd HH:mm:ss` if not specified.
+- `timezone` — [Timezone](/docs/en/operations/server-configuration-parameters/settings.md#timezone). Optional.
+
+**Returned value(s)**
+
+Returns [DateTime64](../data-types/datetime64.md) type values parsed from input string according to a joda style format string.
 
 ## parseDateTime64InJodaSyntaxOrZero
 

docs/en/sql-reference/statements/explain.md

@@ -161,6 +161,8 @@ Settings:
 - `actions` — Prints detailed information about step actions. Default: 0.
 - `json` — Prints query plan steps as a row in [JSON](../../interfaces/formats.md#json) format. Default: 0. It is recommended to use [TSVRaw](../../interfaces/formats.md#tabseparatedraw) format to avoid unnecessary escaping.
 
+When `json=1` step names will contain an additional suffix with unique step identifier.
+
 Example:
 
 ```sql
@@ -194,30 +196,25 @@ EXPLAIN json = 1, description = 0 SELECT 1 UNION ALL SELECT 2 FORMAT TSVRaw;
   {
     "Plan": {
       "Node Type": "Union",
+      "Node Id": "Union_10",
       "Plans": [
         {
           "Node Type": "Expression",
+          "Node Id": "Expression_13",
           "Plans": [
             {
-              "Node Type": "SettingQuotaAndLimits",
+              "Node Type": "ReadFromStorage",
-              "Plans": [
+              "Node Id": "ReadFromStorage_0"
-                {
-                  "Node Type": "ReadFromStorage"
-                }
-              ]
             }
           ]
         },
         {
           "Node Type": "Expression",
+          "Node Id": "Expression_16",
           "Plans": [
             {
-              "Node Type": "SettingQuotaAndLimits",
+              "Node Type": "ReadFromStorage",
-              "Plans": [
+              "Node Id": "ReadFromStorage_4"
-                {
-                  "Node Type": "ReadFromStorage"
-                }
-              ]
             }
           ]
         }
@@ -249,6 +246,7 @@ EXPLAIN json = 1, description = 0, header = 1 SELECT 1, 2 + dummy;
   {
     "Plan": {
       "Node Type": "Expression",
+      "Node Id": "Expression_5",
       "Header": [
         {
           "Name": "1",
@@ -261,23 +259,13 @@ EXPLAIN json = 1, description = 0, header = 1 SELECT 1, 2 + dummy;
       ],
       "Plans": [
         {
-          "Node Type": "SettingQuotaAndLimits",
+          "Node Type": "ReadFromStorage",
+          "Node Id": "ReadFromStorage_0",
           "Header": [
             {
               "Name": "dummy",
               "Type": "UInt8"
             }
-          ],
-          "Plans": [
-            {
-              "Node Type": "ReadFromStorage",
-              "Header": [
-                {
-                  "Name": "dummy",
-                  "Type": "UInt8"
-                }
-              ]
-            }
           ]
         }
       ]
@@ -351,17 +339,31 @@ EXPLAIN json = 1, actions = 1, description = 0 SELECT 1 FORMAT TSVRaw;
   {
     "Plan": {
       "Node Type": "Expression",
+      "Node Id": "Expression_5",
       "Expression": {
-        "Inputs": [],
+        "Inputs": [
+          {
+            "Name": "dummy",
+            "Type": "UInt8"
+          }
+        ],
         "Actions": [
           {
-            "Node Type": "Column",
+            "Node Type": "INPUT",
             "Result Type": "UInt8",
-            "Result Type": "Column",
+            "Result Name": "dummy",
+            "Arguments": [0],
+            "Removed Arguments": [0],
+            "Result": 0
+          },
+          {
+            "Node Type": "COLUMN",
+            "Result Type": "UInt8",
+            "Result Name": "1",
             "Column": "Const(UInt8)",
             "Arguments": [],
             "Removed Arguments": [],
-            "Result": 0
+            "Result": 1
           }
         ],
         "Outputs": [
@@ -370,17 +372,12 @@ EXPLAIN json = 1, actions = 1, description = 0 SELECT 1 FORMAT TSVRaw;
             "Type": "UInt8"
           }
         ],
-        "Positions": [0],
+        "Positions": [1]
-        "Project Input": true
       },
       "Plans": [
         {
-          "Node Type": "SettingQuotaAndLimits",
+          "Node Type": "ReadFromStorage",
-          "Plans": [
+          "Node Id": "ReadFromStorage_0"
-            {
-              "Node Type": "ReadFromStorage"
-            }
-          ]
         }
       ]
     }
@@ -396,6 +393,8 @@ Settings:
 - `graph` — Prints a graph described in the [DOT](https://en.wikipedia.org/wiki/DOT_(graph_description_language)) graph description language. Default: 0.
 - `compact` — Prints graph in compact mode if `graph` setting is enabled. Default: 1.
 
+When `compact=0` and `graph=1` processor names will contain an additional suffix with unique processor identifier.
+
 Example:
 
 ```sql

docs/ru/operations/settings/settings.md

@@ -136,7 +136,7 @@ ClickHouse применяет настройку в тех случаях, ко
 -   0 — выключена.
 -   1 — включена.
 
-Значение по умолчанию: 0.
+Значение по умолчанию: 1.
 
 ## http_zlib_compression_level {#settings-http_zlib_compression_level}
 

docs/zh/operations/settings/settings.md

@@ -97,7 +97,7 @@ ClickHouse从表的过时副本中选择最相关的副本。
 -   0 — Disabled.
 -   1 — Enabled.
 
-默认值：0。
+默认值：1。
 
 ## http_zlib_compression_level {#settings-http_zlib_compression_level}
 

programs/benchmark/Benchmark.cpp

@@ -7,7 +7,6 @@
 #include <random>
 #include <string_view>
 #include <pcg_random.hpp>
-#include <Poco/UUID.h>
 #include <Poco/UUIDGenerator.h>
 #include <Poco/Util/Application.h>
 #include <Common/Stopwatch.h>
@@ -152,8 +151,6 @@ public:
         global_context->setClientName(std::string(DEFAULT_CLIENT_NAME));
         global_context->setQueryKindInitial();
 
-        std::cerr << std::fixed << std::setprecision(3);
-
         /// This is needed to receive blocks with columns of AggregateFunction data type
         /// (example: when using stage = 'with_mergeable_state')
         registerAggregateFunctions();
@@ -226,6 +223,8 @@ private:
     ContextMutablePtr global_context;
     QueryProcessingStage::Enum query_processing_stage;
 
+    WriteBufferFromFileDescriptor log{STDERR_FILENO};
+
     std::atomic<size_t> consecutive_errors{0};
 
     /// Don't execute new queries after timelimit or SIGINT or exception
@@ -303,16 +302,16 @@ private:
         }
 
 
-        std::cerr << "Loaded " << queries.size() << " queries.\n";
+        log << "Loaded " << queries.size() << " queries.\n" << flush;
     }
 
 
     void printNumberOfQueriesExecuted(size_t num)
     {
-        std::cerr << "\nQueries executed: " << num;
+        log << "\nQueries executed: " << num;
         if (queries.size() > 1)
-            std::cerr << " (" << (num * 100.0 / queries.size()) << "%)";
+            log << " (" << (num * 100.0 / queries.size()) << "%)";
-        std::cerr << ".\n";
+        log << ".\n" << flush;
     }
 
     /// Try push new query and check cancellation conditions
@@ -339,19 +338,19 @@ private:
 
             if (interrupt_listener.check())
             {
-                std::cout << "Stopping launch of queries. SIGINT received." << std::endl;
+                std::cout << "Stopping launch of queries. SIGINT received.\n";
                 return false;
             }
+        }
 
-            double seconds = delay_watch.elapsedSeconds();
+        double seconds = delay_watch.elapsedSeconds();
-            if (delay > 0 && seconds > delay)
+        if (delay > 0 && seconds > delay)
-            {
+        {
-                printNumberOfQueriesExecuted(queries_executed);
+            printNumberOfQueriesExecuted(queries_executed);
-                cumulative
+            cumulative
-                    ? report(comparison_info_total, total_watch.elapsedSeconds())
+                ? report(comparison_info_total, total_watch.elapsedSeconds())
-                    : report(comparison_info_per_interval, seconds);
+                : report(comparison_info_per_interval, seconds);
-                delay_watch.restart();
+            delay_watch.restart();
-            }
         }
 
         return true;
@@ -438,16 +437,16 @@ private:
             catch (...)
             {
                 std::lock_guard lock(mutex);
-                std::cerr << "An error occurred while processing the query " << "'" << query << "'"
+                log << "An error occurred while processing the query " << "'" << query << "'"
-                          << ": " << getCurrentExceptionMessage(false) << std::endl;
+                          << ": " << getCurrentExceptionMessage(false) << '\n';
                 if (!(continue_on_errors || max_consecutive_errors > ++consecutive_errors))
                 {
                     shutdown = true;
                     throw;
                 }
 
-                std::cerr << getCurrentExceptionMessage(print_stacktrace,
+                log << getCurrentExceptionMessage(print_stacktrace,
-                    true /*check embedded stack trace*/) << std::endl;
+                    true /*check embedded stack trace*/) << '\n' << flush;
 
                 size_t info_index = round_robin ? 0 : connection_index;
                 ++comparison_info_per_interval[info_index]->errors;
@@ -504,7 +503,7 @@ private:
     {
         std::lock_guard lock(mutex);
 
-        std::cerr << "\n";
+        log << "\n";
         for (size_t i = 0; i < infos.size(); ++i)
         {
             const auto & info = infos[i];
@@ -524,31 +523,31 @@ private:
                     connection_description += conn->getDescription();
                 }
             }
-            std::cerr
+            log
-                    << connection_description << ", "
+                << connection_description << ", "
-                    << "queries: " << info->queries << ", ";
+                << "queries: " << info->queries.load() << ", ";
             if (info->errors)
             {
-                std::cerr << "errors: " << info->errors << ", ";
+                log << "errors: " << info->errors << ", ";
             }
-            std::cerr
+            log
-                    << "QPS: " << (info->queries / seconds) << ", "
+                << "QPS: " << fmt::format("{:.3f}", info->queries / seconds) << ", "
-                    << "RPS: " << (info->read_rows / seconds) << ", "
+                << "RPS: " << fmt::format("{:.3f}", info->read_rows / seconds) << ", "
-                    << "MiB/s: " << (info->read_bytes / seconds / 1048576) << ", "
+                << "MiB/s: " << fmt::format("{:.3f}", info->read_bytes / seconds / 1048576) << ", "
-                    << "result RPS: " << (info->result_rows / seconds) << ", "
+                << "result RPS: " << fmt::format("{:.3f}", info->result_rows / seconds) << ", "
-                    << "result MiB/s: " << (info->result_bytes / seconds / 1048576) << "."
+                << "result MiB/s: " << fmt::format("{:.3f}", info->result_bytes / seconds / 1048576) << "."
-                    << "\n";
+                << "\n";
         }
-        std::cerr << "\n";
+        log << "\n";
 
         auto print_percentile = [&](double percent)
         {
-            std::cerr << percent << "%\t\t";
+            log << percent << "%\t\t";
             for (const auto & info : infos)
             {
-                std::cerr << info->sampler.quantileNearest(percent / 100.0) << " sec.\t";
+                log << fmt::format("{:.3f}", info->sampler.quantileNearest(percent / 100.0)) << " sec.\t";
             }
-            std::cerr << "\n";
+            log << "\n";
         };
 
         for (int percent = 0; percent <= 90; percent += 10)
@@ -559,13 +558,15 @@ private:
         print_percentile(99.9);
         print_percentile(99.99);
 
-        std::cerr << "\n" << t_test.compareAndReport(confidence).second << "\n";
+        log << "\n" << t_test.compareAndReport(confidence).second << "\n";
 
         if (!cumulative)
         {
             for (auto & info : infos)
                 info->clear();
         }
+
+        log.next();
     }
 
 public:
@@ -741,7 +742,7 @@ int mainEntryClickHouseBenchmark(int argc, char ** argv)
     }
     catch (...)
     {
-        std::cerr << getCurrentExceptionMessage(print_stacktrace, true) << std::endl;
+        std::cerr << getCurrentExceptionMessage(print_stacktrace, true) << '\n';
         return getCurrentExceptionCode();
     }
 }

src/AggregateFunctions/AggregateFunctionAvg.h

@@ -231,7 +231,7 @@ public:
 
     void add(AggregateDataPtr __restrict place, const IColumn ** columns, size_t row_num, Arena *) const final
     {
-        increment(place, static_cast<const ColVecType &>(*columns[0]).getData()[row_num]);
+        increment(place, Numerator(static_cast<const ColVecType &>(*columns[0]).getData()[row_num]));
         ++this->data(place).denominator;
     }
 

src/AggregateFunctions/AggregateFunctionDeltaSum.cpp

@@ -27,9 +27,9 @@ namespace
 template <typename T>
 struct AggregationFunctionDeltaSumData
 {
-    T sum = 0;
+    T sum{};
-    T last = 0;
+    T last{};
-    T first = 0;
+    T first{};
     bool seen = false;
 };
 

src/AggregateFunctions/AggregateFunctionDeltaSumTimestamp.cpp

@@ -22,14 +22,21 @@ namespace ErrorCodes
 namespace
 {
 
+/** Due to a lack of proper code review, this code was contributed with a multiplication of template instantiations
+  * over all pairs of data types, and we deeply regret that.
+  *
+  * We cannot remove all combinations, because the binary representation of serialized data has to remain the same,
+  * but we can partially heal the wound by treating unsigned and signed data types in the same way.
+  */
+
 template <typename ValueType, typename TimestampType>
 struct AggregationFunctionDeltaSumTimestampData
 {
-    ValueType sum = 0;
+    ValueType sum{};
-    ValueType first = 0;
+    ValueType first{};
-    ValueType last = 0;
+    ValueType last{};
-    TimestampType first_ts = 0;
+    TimestampType first_ts{};
-    TimestampType last_ts = 0;
+    TimestampType last_ts{};
     bool seen = false;
 };
 
@@ -37,23 +44,22 @@ template <typename ValueType, typename TimestampType>
 class AggregationFunctionDeltaSumTimestamp final
     : public IAggregateFunctionDataHelper<
         AggregationFunctionDeltaSumTimestampData<ValueType, TimestampType>,
-        AggregationFunctionDeltaSumTimestamp<ValueType, TimestampType>
+        AggregationFunctionDeltaSumTimestamp<ValueType, TimestampType>>
-      >
 {
 public:
     AggregationFunctionDeltaSumTimestamp(const DataTypes & arguments, const Array & params)
         : IAggregateFunctionDataHelper<
             AggregationFunctionDeltaSumTimestampData<ValueType, TimestampType>,
-            AggregationFunctionDeltaSumTimestamp<ValueType, TimestampType>
+            AggregationFunctionDeltaSumTimestamp<ValueType, TimestampType>>{arguments, params, createResultType()}
-        >{arguments, params, createResultType()}
+    {
-    {}
+    }
 
     AggregationFunctionDeltaSumTimestamp()
         : IAggregateFunctionDataHelper<
             AggregationFunctionDeltaSumTimestampData<ValueType, TimestampType>,
-            AggregationFunctionDeltaSumTimestamp<ValueType, TimestampType>
+            AggregationFunctionDeltaSumTimestamp<ValueType, TimestampType>>{}
-        >{}
+    {
-    {}
+    }
 
     bool allocatesMemoryInArena() const override { return false; }
 
@@ -63,8 +69,8 @@ public:
 
     void NO_SANITIZE_UNDEFINED ALWAYS_INLINE add(AggregateDataPtr __restrict place, const IColumn ** columns, size_t row_num, Arena *) const override
     {
-        auto value = assert_cast<const ColumnVector<ValueType> &>(*columns[0]).getData()[row_num];
+        auto value = unalignedLoad<ValueType>(columns[0]->getRawData().data() + row_num * sizeof(ValueType));
-        auto ts = assert_cast<const ColumnVector<TimestampType> &>(*columns[1]).getData()[row_num];
+        auto ts = unalignedLoad<TimestampType>(columns[1]->getRawData().data() + row_num * sizeof(TimestampType));
 
         auto & data = this->data(place);
 
@@ -172,10 +178,48 @@ public:
 
     void insertResultInto(AggregateDataPtr __restrict place, IColumn & to, Arena *) const override
     {
-        assert_cast<ColumnVector<ValueType> &>(to).getData().push_back(this->data(place).sum);
+        static_cast<ColumnFixedSizeHelper &>(to).template insertRawData<sizeof(ValueType)>(
+            reinterpret_cast<const char *>(&this->data(place).sum));
     }
 };
 
+
+template <typename FirstType, template <typename, typename> class AggregateFunctionTemplate, typename... TArgs>
+IAggregateFunction * createWithTwoTypesSecond(const IDataType & second_type, TArgs && ... args)
+{
+    WhichDataType which(second_type);
+
+    if (which.idx == TypeIndex::UInt32) return new AggregateFunctionTemplate<FirstType, UInt32>(args...);
+    if (which.idx == TypeIndex::UInt64) return new AggregateFunctionTemplate<FirstType, UInt64>(args...);
+    if (which.idx == TypeIndex::Int32) return new AggregateFunctionTemplate<FirstType, UInt32>(args...);
+    if (which.idx == TypeIndex::Int64) return new AggregateFunctionTemplate<FirstType, UInt64>(args...);
+    if (which.idx == TypeIndex::Float32) return new AggregateFunctionTemplate<FirstType, Float32>(args...);
+    if (which.idx == TypeIndex::Float64) return new AggregateFunctionTemplate<FirstType, Float64>(args...);
+    if (which.idx == TypeIndex::Date) return new AggregateFunctionTemplate<FirstType, UInt16>(args...);
+    if (which.idx == TypeIndex::DateTime) return new AggregateFunctionTemplate<FirstType, UInt32>(args...);
+
+    return nullptr;
+}
+
+template <template <typename, typename> class AggregateFunctionTemplate, typename... TArgs>
+IAggregateFunction * createWithTwoTypes(const IDataType & first_type, const IDataType & second_type, TArgs && ... args)
+{
+    WhichDataType which(first_type);
+
+    if (which.idx == TypeIndex::UInt8) return createWithTwoTypesSecond<UInt8, AggregateFunctionTemplate>(second_type, args...);
+    if (which.idx == TypeIndex::UInt16) return createWithTwoTypesSecond<UInt16, AggregateFunctionTemplate>(second_type, args...);
+    if (which.idx == TypeIndex::UInt32) return createWithTwoTypesSecond<UInt32, AggregateFunctionTemplate>(second_type, args...);
+    if (which.idx == TypeIndex::UInt64) return createWithTwoTypesSecond<UInt64, AggregateFunctionTemplate>(second_type, args...);
+    if (which.idx == TypeIndex::Int8) return createWithTwoTypesSecond<UInt8, AggregateFunctionTemplate>(second_type, args...);
+    if (which.idx == TypeIndex::Int16) return createWithTwoTypesSecond<UInt16, AggregateFunctionTemplate>(second_type, args...);
+    if (which.idx == TypeIndex::Int32) return createWithTwoTypesSecond<UInt32, AggregateFunctionTemplate>(second_type, args...);
+    if (which.idx == TypeIndex::Int64) return createWithTwoTypesSecond<UInt64, AggregateFunctionTemplate>(second_type, args...);
+    if (which.idx == TypeIndex::Float32) return createWithTwoTypesSecond<Float32, AggregateFunctionTemplate>(second_type, args...);
+    if (which.idx == TypeIndex::Float64) return createWithTwoTypesSecond<Float64, AggregateFunctionTemplate>(second_type, args...);
+
+    return nullptr;
+}
+
 AggregateFunctionPtr createAggregateFunctionDeltaSumTimestamp(
     const String & name,
     const DataTypes & arguments,
@@ -193,8 +237,14 @@ AggregateFunctionPtr createAggregateFunctionDeltaSumTimestamp(
         throw Exception(ErrorCodes::ILLEGAL_TYPE_OF_ARGUMENT, "Illegal type {} of argument for aggregate function {}, "
                         "must be Int, Float, Date, DateTime", arguments[1]->getName(), name);
 
-    return AggregateFunctionPtr(createWithTwoNumericOrDateTypes<AggregationFunctionDeltaSumTimestamp>(
+    auto res = AggregateFunctionPtr(createWithTwoTypes<AggregationFunctionDeltaSumTimestamp>(
         *arguments[0], *arguments[1], arguments, params));
+
+    if (!res)
+        throw Exception(ErrorCodes::ILLEGAL_TYPE_OF_ARGUMENT, "Illegal type {} of argument for aggregate function {}, "
+            "this type is not supported", arguments[0]->getName(), name);
+
+    return res;
 }
 }
 

src/AggregateFunctions/AggregateFunctionGroupArray.cpp

@@ -79,7 +79,7 @@ template <typename T>
 struct GroupArraySamplerData
 {
     /// For easy serialization.
-    static_assert(std::has_unique_object_representations_v<T> || std::is_floating_point_v<T>);
+    static_assert(std::has_unique_object_representations_v<T> || is_floating_point<T>);
 
     // Switch to ordinary Allocator after 4096 bytes to avoid fragmentation and trash in Arena
     using Allocator = MixedAlignedArenaAllocator<alignof(T), 4096>;
@@ -120,7 +120,7 @@ template <typename T>
 struct GroupArrayNumericData<T, false>
 {
     /// For easy serialization.
-    static_assert(std::has_unique_object_representations_v<T> || std::is_floating_point_v<T>);
+    static_assert(std::has_unique_object_representations_v<T> || is_floating_point<T>);
 
     // Switch to ordinary Allocator after 4096 bytes to avoid fragmentation and trash in Arena
     using Allocator = MixedAlignedArenaAllocator<alignof(T), 4096>;

src/AggregateFunctions/AggregateFunctionGroupArrayMoving.cpp

@@ -38,7 +38,7 @@ template <typename T>
 struct MovingData
 {
     /// For easy serialization.
-    static_assert(std::has_unique_object_representations_v<T> || std::is_floating_point_v<T>);
+    static_assert(std::has_unique_object_representations_v<T> || is_floating_point<T>);
 
     using Accumulator = T;
 

src/AggregateFunctions/AggregateFunctionIntervalLengthSum.cpp

@@ -187,7 +187,7 @@ public:
 
     static DataTypePtr createResultType()
     {
-        if constexpr (std::is_floating_point_v<T>)
+        if constexpr (is_floating_point<T>)
             return std::make_shared<DataTypeFloat64>();
         return std::make_shared<DataTypeUInt64>();
     }
@@ -227,7 +227,7 @@ public:
 
     void insertResultInto(AggregateDataPtr __restrict place, IColumn & to, Arena *) const override
     {
-        if constexpr (std::is_floating_point_v<T>)
+        if constexpr (is_floating_point<T>)
             assert_cast<ColumnFloat64 &>(to).getData().push_back(getIntervalLengthSum<Float64>(this->data(place)));
         else
             assert_cast<ColumnUInt64 &>(to).getData().push_back(getIntervalLengthSum<UInt64>(this->data(place)));

src/AggregateFunctions/AggregateFunctionMaxIntersections.cpp

@@ -155,9 +155,9 @@ public:
 
     void insertResultInto(AggregateDataPtr __restrict place, IColumn & to, Arena *) const override
     {
-        Int64 current_intersections = 0;
+        Int64 current_intersections{};
-        Int64 max_intersections = 0;
+        Int64 max_intersections{};
-        PointType position_of_max_intersections = 0;
+        PointType position_of_max_intersections{};
 
         /// const_cast because we will sort the array
         auto & array = this->data(place).value;

src/AggregateFunctions/AggregateFunctionSparkbar.cpp

@@ -45,12 +45,12 @@ struct AggregateFunctionSparkbarData
     Y insert(const X & x, const Y & y)
     {
         if (isNaN(y) || y <= 0)
-            return 0;
+            return {};
 
         auto [it, inserted] = points.insert({x, y});
         if (!inserted)
         {
-            if constexpr (std::is_floating_point_v<Y>)
+            if constexpr (is_floating_point<Y>)
             {
                 it->getMapped() += y;
                 return it->getMapped();
@@ -173,13 +173,13 @@ private:
 
         if (from_x >= to_x)
         {
-            size_t sz = updateFrame(values, 8);
+            size_t sz = updateFrame(values, Y{8});
             values.push_back('\0');
             offsets.push_back(offsets.empty() ? sz + 1 : offsets.back() + sz + 1);
             return;
         }
 
-        PaddedPODArray<Y> histogram(width, 0);
+        PaddedPODArray<Y> histogram(width, Y{0});
         PaddedPODArray<UInt64> count_histogram(width, 0); /// The number of points in each bucket
 
         for (const auto & point : data.points)
@@ -197,7 +197,7 @@ private:
 
             Y res;
             bool has_overfllow = false;
-            if constexpr (std::is_floating_point_v<Y>)
+            if constexpr (is_floating_point<Y>)
                 res = histogram[index] + point.getMapped();
             else
                 has_overfllow = common::addOverflow(histogram[index], point.getMapped(), res);
@@ -218,10 +218,10 @@ private:
         for (size_t i = 0; i < histogram.size(); ++i)
         {
             if (count_histogram[i] > 0)
-                histogram[i] /= count_histogram[i];
+                histogram[i] = histogram[i] / count_histogram[i];
         }
 
-        Y y_max = 0;
+        Y y_max{};
         for (auto & y : histogram)
         {
             if (isNaN(y) || y <= 0)
@@ -245,8 +245,8 @@ private:
                 continue;
             }
 
-            constexpr auto levels_num = static_cast<Y>(BAR_LEVELS - 1);
+            constexpr auto levels_num = Y{BAR_LEVELS - 1};
-            if constexpr (std::is_floating_point_v<Y>)
+            if constexpr (is_floating_point<Y>)
             {
                 y = y / (y_max / levels_num) + 1;
             }

src/AggregateFunctions/AggregateFunctionSum.h

@@ -69,7 +69,7 @@ struct AggregateFunctionSumData
         size_t count = end - start;
         const auto * end_ptr = ptr + count;
 
-        if constexpr (std::is_floating_point_v<T>)
+        if constexpr (is_floating_point<T>)
         {
             /// Compiler cannot unroll this loop, do it manually.
             /// (at least for floats, most likely due to the lack of -fassociative-math)
@@ -83,7 +83,7 @@ struct AggregateFunctionSumData
             while (ptr < unrolled_end)
             {
                 for (size_t i = 0; i < unroll_count; ++i)
-                    Impl::add(partial_sums[i], ptr[i]);
+                    Impl::add(partial_sums[i], T(ptr[i]));
                 ptr += unroll_count;
             }
 
@@ -95,7 +95,7 @@ struct AggregateFunctionSumData
         T local_sum{};
         while (ptr < end_ptr)
         {
-            Impl::add(local_sum, *ptr);
+            Impl::add(local_sum, T(*ptr));
             ++ptr;
         }
         Impl::add(sum, local_sum);
@@ -193,12 +193,11 @@ struct AggregateFunctionSumData
             Impl::add(sum, local_sum);
             return;
         }
-        else if constexpr (std::is_floating_point_v<T>)
+        else if constexpr (is_floating_point<T> && (sizeof(Value) == 4 || sizeof(Value) == 8))
         {
-            /// For floating point we use a similar trick as above, except that now we  reinterpret the floating point number as an unsigned
+            /// For floating point we use a similar trick as above, except that now we reinterpret the floating point number as an unsigned
             /// integer of the same size and use a mask instead (0 to discard, 0xFF..FF to keep)
-            static_assert(sizeof(Value) == 4 || sizeof(Value) == 8);
+            using EquivalentInteger = typename std::conditional_t<sizeof(Value) == 4, UInt32, UInt64>;
-            using equivalent_integer = typename std::conditional_t<sizeof(Value) == 4, UInt32, UInt64>;
 
             constexpr size_t unroll_count = 128 / sizeof(T);
             T partial_sums[unroll_count]{};
@@ -209,11 +208,11 @@ struct AggregateFunctionSumData
             {
                 for (size_t i = 0; i < unroll_count; ++i)
                 {
-                    equivalent_integer value;
+                    EquivalentInteger value;
-                    std::memcpy(&value, &ptr[i], sizeof(Value));
+                    memcpy(&value, &ptr[i], sizeof(Value));
                     value &= (!condition_map[i] != add_if_zero) - 1;
                     Value d;
-                    std::memcpy(&d, &value, sizeof(Value));
+                    memcpy(&d, &value, sizeof(Value));
                     Impl::add(partial_sums[i], d);
                 }
                 ptr += unroll_count;
@@ -228,7 +227,7 @@ struct AggregateFunctionSumData
         while (ptr < end_ptr)
         {
             if (!*condition_map == add_if_zero)
-                Impl::add(local_sum, *ptr);
+                Impl::add(local_sum, T(*ptr));
             ++ptr;
             ++condition_map;
         }
@@ -306,7 +305,7 @@ struct AggregateFunctionSumData
 template <typename T>
 struct AggregateFunctionSumKahanData
 {
-    static_assert(std::is_floating_point_v<T>,
+    static_assert(is_floating_point<T>,
         "It doesn't make sense to use Kahan Summation algorithm for non floating point types");
 
     T sum{};
@@ -489,10 +488,7 @@ public:
     void add(AggregateDataPtr __restrict place, const IColumn ** columns, size_t row_num, Arena *) const override
     {
         const auto & column = assert_cast<const ColVecType &>(*columns[0]);
-        if constexpr (is_big_int_v<T>)
+        this->data(place).add(static_cast<TResult>(column.getData()[row_num]));
-            this->data(place).add(static_cast<TResult>(column.getData()[row_num]));
-        else
-            this->data(place).add(column.getData()[row_num]);
     }
 
     void addBatchSinglePlace(

src/AggregateFunctions/AggregateFunctionUniq.h

@@ -257,7 +257,7 @@ template <typename T> struct AggregateFunctionUniqTraits
 {
     static UInt64 hash(T x)
     {
-        if constexpr (std::is_same_v<T, Float32> || std::is_same_v<T, Float64>)
+        if constexpr (is_floating_point<T>)
         {
             return bit_cast<UInt64>(x);
         }

src/AggregateFunctions/AggregateFunctionUniqCombined.h

@@ -111,7 +111,7 @@ public:
                 /// Initially UInt128 was introduced only for UUID, and then the other big-integer types were added.
                 hash = static_cast<HashValueType>(sipHash64(value));
             }
-            else if constexpr (std::is_floating_point_v<T>)
+            else if constexpr (is_floating_point<T>)
             {
                 hash = static_cast<HashValueType>(intHash64(bit_cast<UInt64>(value)));
             }

src/AggregateFunctions/Helpers.h

@@ -184,36 +184,8 @@ static IAggregateFunction * createWithDecimalType(const IDataType & argument_typ
 }
 
 /** For template with two arguments.
+  * This is an extremely dangerous for code bloat - do not use.
   */
-template <typename FirstType, template <typename, typename> class AggregateFunctionTemplate, typename... TArgs>
-static IAggregateFunction * createWithTwoNumericTypesSecond(const IDataType & second_type, TArgs && ... args)
-{
-    WhichDataType which(second_type);
-#define DISPATCH(TYPE) \
-    if (which.idx == TypeIndex::TYPE) return new AggregateFunctionTemplate<FirstType, TYPE>(args...);
-    FOR_NUMERIC_TYPES(DISPATCH)
-#undef DISPATCH
-    if (which.idx == TypeIndex::Enum8) return new AggregateFunctionTemplate<FirstType, Int8>(args...);
-    if (which.idx == TypeIndex::Enum16) return new AggregateFunctionTemplate<FirstType, Int16>(args...);
-    return nullptr;
-}
-
-template <template <typename, typename> class AggregateFunctionTemplate, typename... TArgs>
-static IAggregateFunction * createWithTwoNumericTypes(const IDataType & first_type, const IDataType & second_type, TArgs && ... args)
-{
-    WhichDataType which(first_type);
-#define DISPATCH(TYPE) \
-    if (which.idx == TypeIndex::TYPE) \
-        return createWithTwoNumericTypesSecond<TYPE, AggregateFunctionTemplate>(second_type, args...);
-    FOR_NUMERIC_TYPES(DISPATCH)
-#undef DISPATCH
-    if (which.idx == TypeIndex::Enum8)
-        return createWithTwoNumericTypesSecond<Int8, AggregateFunctionTemplate>(second_type, args...);
-    if (which.idx == TypeIndex::Enum16)
-        return createWithTwoNumericTypesSecond<Int16, AggregateFunctionTemplate>(second_type, args...);
-    return nullptr;
-}
-
 template <typename FirstType, template <typename, typename> class AggregateFunctionTemplate, typename... TArgs>
 static IAggregateFunction * createWithTwoBasicNumericTypesSecond(const IDataType & second_type, TArgs && ... args)
 {
@@ -237,46 +209,6 @@ static IAggregateFunction * createWithTwoBasicNumericTypes(const IDataType & fir
     return nullptr;
 }
 
-template <typename FirstType, template <typename, typename> class AggregateFunctionTemplate, typename... TArgs>
-static IAggregateFunction * createWithTwoNumericOrDateTypesSecond(const IDataType & second_type, TArgs && ... args)
-{
-    WhichDataType which(second_type);
-#define DISPATCH(TYPE) \
-    if (which.idx == TypeIndex::TYPE) return new AggregateFunctionTemplate<FirstType, TYPE>(args...);
-    FOR_NUMERIC_TYPES(DISPATCH)
-#undef DISPATCH
-    if (which.idx == TypeIndex::Enum8) return new AggregateFunctionTemplate<FirstType, Int8>(args...);
-    if (which.idx == TypeIndex::Enum16) return new AggregateFunctionTemplate<FirstType, Int16>(args...);
-
-    /// expects that DataTypeDate based on UInt16, DataTypeDateTime based on UInt32
-    if (which.idx == TypeIndex::Date) return new AggregateFunctionTemplate<FirstType, UInt16>(args...);
-    if (which.idx == TypeIndex::DateTime) return new AggregateFunctionTemplate<FirstType, UInt32>(args...);
-
-    return nullptr;
-}
-
-template <template <typename, typename> class AggregateFunctionTemplate, typename... TArgs>
-static IAggregateFunction * createWithTwoNumericOrDateTypes(const IDataType & first_type, const IDataType & second_type, TArgs && ... args)
-{
-    WhichDataType which(first_type);
-#define DISPATCH(TYPE) \
-    if (which.idx == TypeIndex::TYPE) \
-        return createWithTwoNumericOrDateTypesSecond<TYPE, AggregateFunctionTemplate>(second_type, args...);
-    FOR_NUMERIC_TYPES(DISPATCH)
-#undef DISPATCH
-    if (which.idx == TypeIndex::Enum8)
-        return createWithTwoNumericOrDateTypesSecond<Int8, AggregateFunctionTemplate>(second_type, args...);
-    if (which.idx == TypeIndex::Enum16)
-        return createWithTwoNumericOrDateTypesSecond<Int16, AggregateFunctionTemplate>(second_type, args...);
-
-    /// expects that DataTypeDate based on UInt16, DataTypeDateTime based on UInt32
-    if (which.idx == TypeIndex::Date)
-        return createWithTwoNumericOrDateTypesSecond<UInt16, AggregateFunctionTemplate>(second_type, args...);
-    if (which.idx == TypeIndex::DateTime)
-        return createWithTwoNumericOrDateTypesSecond<UInt32, AggregateFunctionTemplate>(second_type, args...);
-    return nullptr;
-}
-
 template <template <typename> class AggregateFunctionTemplate, typename... TArgs>
 static IAggregateFunction * createWithStringType(const IDataType & argument_type, TArgs && ... args)
 {

src/AggregateFunctions/QuantileTDigest.h

@@ -391,7 +391,7 @@ public:
     ResultType getImpl(Float64 level)
     {
         if (centroids.empty())
-            return std::is_floating_point_v<ResultType> ? std::numeric_limits<ResultType>::quiet_NaN() : 0;
+            return is_floating_point<ResultType> ? std::numeric_limits<ResultType>::quiet_NaN() : 0;
 
         compress();
 

src/AggregateFunctions/ReservoirSampler.h

@@ -276,6 +276,6 @@ private:
     {
         if (OnEmpty == ReservoirSamplerOnEmpty::THROW)
             throw DB::Exception(DB::ErrorCodes::LOGICAL_ERROR, "Quantile of empty ReservoirSampler");
-        return NanLikeValueConstructor<ResultType, std::is_floating_point_v<ResultType>>::getValue();
+        return NanLikeValueConstructor<ResultType, is_floating_point<ResultType>>::getValue();
     }
 };

src/AggregateFunctions/ReservoirSamplerDeterministic.h

@@ -271,7 +271,7 @@ private:
     {
         if (OnEmpty == ReservoirSamplerDeterministicOnEmpty::THROW)
             throw DB::Exception(DB::ErrorCodes::LOGICAL_ERROR, "Quantile of empty ReservoirSamplerDeterministic");
-        return NanLikeValueConstructor<ResultType, std::is_floating_point_v<ResultType>>::getValue();
+        return NanLikeValueConstructor<ResultType, is_floating_point<ResultType>>::getValue();
     }
 };
 

src/Backups/BackupCoordinationStageSync.cpp

@@ -121,8 +121,7 @@ BackupCoordinationStageSync::BackupCoordinationStageSync(
 
     try
     {
-        concurrency_check.emplace(is_restore, /* on_cluster = */ true, zookeeper_path, allow_concurrency, concurrency_counters_);
+        createStartAndAliveNodesAndCheckConcurrency(concurrency_counters_);
-        createStartAndAliveNodes();
         startWatchingThread();
     }
     catch (...)
@@ -221,7 +220,7 @@ void BackupCoordinationStageSync::createRootNodes()
         throw Exception(ErrorCodes::LOGICAL_ERROR, "Unexpected path in ZooKeeper specified: {}", zookeeper_path);
     }
 
-    auto holder = with_retries.createRetriesControlHolder("BackupStageSync::createRootNodes", WithRetries::kInitialization);
+    auto holder = with_retries.createRetriesControlHolder("BackupCoordinationStageSync::createRootNodes", WithRetries::kInitialization);
     holder.retries_ctl.retryLoop(
         [&, &zookeeper = holder.faulty_zookeeper]()
         {
@@ -232,18 +231,22 @@ void BackupCoordinationStageSync::createRootNodes()
 }
 
 
-void BackupCoordinationStageSync::createStartAndAliveNodes()
+void BackupCoordinationStageSync::createStartAndAliveNodesAndCheckConcurrency(BackupConcurrencyCounters & concurrency_counters_)
 {
-    auto holder = with_retries.createRetriesControlHolder("BackupStageSync::createStartAndAliveNodes", WithRetries::kInitialization);
+    auto holder = with_retries.createRetriesControlHolder("BackupCoordinationStageSync::createStartAndAliveNodes", WithRetries::kInitialization);
     holder.retries_ctl.retryLoop([&, &zookeeper = holder.faulty_zookeeper]()
     {
         with_retries.renewZooKeeper(zookeeper);
-        createStartAndAliveNodes(zookeeper);
+        createStartAndAliveNodesAndCheckConcurrency(zookeeper);
     });
+
+    /// The local concurrency check should be done here after BackupCoordinationStageSync::checkConcurrency() checked that
+    /// there are no 'alive' nodes corresponding to other backups or restores.
+    local_concurrency_check.emplace(is_restore, /* on_cluster = */ true, zookeeper_path, allow_concurrency, concurrency_counters_);
 }
 
 
-void BackupCoordinationStageSync::createStartAndAliveNodes(Coordination::ZooKeeperWithFaultInjection::Ptr zookeeper)
+void BackupCoordinationStageSync::createStartAndAliveNodesAndCheckConcurrency(Coordination::ZooKeeperWithFaultInjection::Ptr zookeeper)
 {
     /// The "num_hosts" node keeps the number of hosts which started (created the "started" node)
     /// but not yet finished (not created the "finished" node).
@@ -464,7 +467,7 @@ void BackupCoordinationStageSync::watchingThread()
         try
         {
             /// Recreate the 'alive' node if necessary and read a new state from ZooKeeper.
-            auto holder = with_retries.createRetriesControlHolder("BackupStageSync::watchingThread");
+            auto holder = with_retries.createRetriesControlHolder("BackupCoordinationStageSync::watchingThread");
             auto & zookeeper = holder.faulty_zookeeper;
             with_retries.renewZooKeeper(zookeeper);
 
@@ -496,6 +499,9 @@ void BackupCoordinationStageSync::watchingThread()
             tryLogCurrentException(log, "Caught exception while watching");
         }
 
+        if (should_stop())
+            return;
+
         zk_nodes_changed->tryWait(sync_period_ms.count());
     }
 }
@@ -769,7 +775,7 @@ void BackupCoordinationStageSync::setStage(const String & stage, const String &
         stopWatchingThread();
     }
 
-    auto holder = with_retries.createRetriesControlHolder("BackupStageSync::setStage");
+    auto holder = with_retries.createRetriesControlHolder("BackupCoordinationStageSync::setStage");
     holder.retries_ctl.retryLoop([&, &zookeeper = holder.faulty_zookeeper]()
     {
         with_retries.renewZooKeeper(zookeeper);
@@ -938,7 +944,7 @@ bool BackupCoordinationStageSync::finishImpl(bool throw_if_error, WithRetries::K
 
     try
     {
-        auto holder = with_retries.createRetriesControlHolder("BackupStageSync::finish", retries_kind);
+        auto holder = with_retries.createRetriesControlHolder("BackupCoordinationStageSync::finish", retries_kind);
         holder.retries_ctl.retryLoop([&, &zookeeper = holder.faulty_zookeeper]()
         {
             with_retries.renewZooKeeper(zookeeper);
@@ -1309,7 +1315,7 @@ bool BackupCoordinationStageSync::setError(const Exception & exception, bool thr
             }
         }
 
-        auto holder = with_retries.createRetriesControlHolder("BackupStageSync::setError", WithRetries::kErrorHandling);
+        auto holder = with_retries.createRetriesControlHolder("BackupCoordinationStageSync::setError", WithRetries::kErrorHandling);
         holder.retries_ctl.retryLoop([&, &zookeeper = holder.faulty_zookeeper]()
         {
             with_retries.renewZooKeeper(zookeeper);

src/Backups/BackupCoordinationStageSync.h

@@ -72,8 +72,8 @@ private:
     void createRootNodes();
 
     /// Atomically creates both 'start' and 'alive' nodes and also checks that there is no concurrent backup or restore if `allow_concurrency` is false.
-    void createStartAndAliveNodes();
+    void createStartAndAliveNodesAndCheckConcurrency(BackupConcurrencyCounters & concurrency_counters_);
-    void createStartAndAliveNodes(Coordination::ZooKeeperWithFaultInjection::Ptr zookeeper);
+    void createStartAndAliveNodesAndCheckConcurrency(Coordination::ZooKeeperWithFaultInjection::Ptr zookeeper);
 
     /// Deserialize the version of a node stored in the 'start' node.
     int parseStartNode(const String & start_node_contents, const String & host) const;
@@ -171,7 +171,7 @@ private:
     const String alive_node_path;
     const String alive_tracker_node_path;
 
-    std::optional<BackupConcurrencyCheck> concurrency_check;
+    std::optional<BackupConcurrencyCheck> local_concurrency_check;
 
     std::shared_ptr<Poco::Event> zk_nodes_changed;
 

src/Client/ClientBase.cpp

@@ -68,15 +68,16 @@
 #include <Access/AccessControl.h>
 #include <Storages/ColumnsDescription.h>
 
-#include <boost/algorithm/string/case_conv.hpp>
-#include <boost/algorithm/string/replace.hpp>
-#include <iostream>
 #include <filesystem>
+#include <iostream>
 #include <limits>
 #include <map>
 #include <memory>
+#include <mutex>
 #include <string_view>
 #include <unordered_map>
+#include <boost/algorithm/string/case_conv.hpp>
+#include <boost/algorithm/string/replace.hpp>
 
 #include <Common/config_version.h>
 #include <base/find_symbols.h>
@@ -441,9 +442,15 @@ void ClientBase::onData(Block & block, ASTPtr parsed_query)
 
     /// If results are written INTO OUTFILE, we can avoid clearing progress to avoid flicker.
     if (need_render_progress && tty_buf && (!select_into_file || select_into_file_and_stdout))
-        progress_indication.clearProgressOutput(*tty_buf);
+    {
+        std::unique_lock lock(tty_mutex);
+        progress_indication.clearProgressOutput(*tty_buf, lock);
+    }
     if (need_render_progress_table && tty_buf && (!select_into_file || select_into_file_and_stdout))
-        progress_table.clearTableOutput(*tty_buf);
+    {
+        std::unique_lock lock(tty_mutex);
+        progress_table.clearTableOutput(*tty_buf, lock);
+    }
 
     try
     {
@@ -464,13 +471,15 @@ void ClientBase::onData(Block & block, ASTPtr parsed_query)
     {
         if (select_into_file && !select_into_file_and_stdout)
             error_stream << "\r";
-        progress_indication.writeProgress(*tty_buf);
+        std::unique_lock lock(tty_mutex);
+        progress_indication.writeProgress(*tty_buf, lock);
     }
     if (need_render_progress_table && tty_buf && !cancelled)
     {
         if (!need_render_progress && select_into_file && !select_into_file_and_stdout)
             error_stream << "\r";
-        progress_table.writeTable(*tty_buf, progress_table_toggle_on.load(), progress_table_toggle_enabled);
+        std::unique_lock lock(tty_mutex);
+        progress_table.writeTable(*tty_buf, lock, progress_table_toggle_on.load(), progress_table_toggle_enabled);
     }
 }
 
@@ -479,9 +488,15 @@ void ClientBase::onLogData(Block & block)
 {
     initLogsOutputStream();
     if (need_render_progress && tty_buf)
-        progress_indication.clearProgressOutput(*tty_buf);
+    {
+        std::unique_lock lock(tty_mutex);
+        progress_indication.clearProgressOutput(*tty_buf, lock);
+    }
     if (need_render_progress_table && tty_buf)
-        progress_table.clearTableOutput(*tty_buf);
+    {
+        std::unique_lock lock(tty_mutex);
+        progress_table.clearTableOutput(*tty_buf, lock);
+    }
     logs_out_stream->writeLogs(block);
     logs_out_stream->flush();
 }
@@ -1151,34 +1166,8 @@ void ClientBase::receiveResult(ASTPtr parsed_query, Int32 signals_before_stop, b
 
     std::exception_ptr local_format_error;
 
-    if (keystroke_interceptor)
+    startKeystrokeInterceptorIfExists();
-    {
+    SCOPE_EXIT({ stopKeystrokeInterceptorIfExists(); });
-        progress_table_toggle_on = false;
-        try
-        {
-            keystroke_interceptor->startIntercept();
-        }
-        catch (const DB::Exception &)
-        {
-            error_stream << getCurrentExceptionMessage(false);
-            keystroke_interceptor.reset();
-        }
-    }
-
-    SCOPE_EXIT({
-        if (keystroke_interceptor)
-        {
-            try
-            {
-                keystroke_interceptor->stopIntercept();
-            }
-            catch (...)
-            {
-                error_stream << getCurrentExceptionMessage(false);
-                keystroke_interceptor.reset();
-            }
-        }
-    });
 
     while (true)
     {
@@ -1318,7 +1307,10 @@ void ClientBase::onProgress(const Progress & value)
         output_format->onProgress(value);
 
     if (need_render_progress && tty_buf)
-        progress_indication.writeProgress(*tty_buf);
+    {
+        std::unique_lock lock(tty_mutex);
+        progress_indication.writeProgress(*tty_buf, lock);
+    }
 }
 
 void ClientBase::onTimezoneUpdate(const String & tz)
@@ -1330,9 +1322,15 @@ void ClientBase::onTimezoneUpdate(const String & tz)
 void ClientBase::onEndOfStream()
 {
     if (need_render_progress && tty_buf)
-        progress_indication.clearProgressOutput(*tty_buf);
+    {
+        std::unique_lock lock(tty_mutex);
+        progress_indication.clearProgressOutput(*tty_buf, lock);
+    }
     if (need_render_progress_table && tty_buf)
-        progress_table.clearTableOutput(*tty_buf);
+    {
+        std::unique_lock lock(tty_mutex);
+        progress_table.clearTableOutput(*tty_buf, lock);
+    }
 
     if (output_format)
     {
@@ -1414,11 +1412,15 @@ void ClientBase::onProfileEvents(Block & block)
         progress_table.updateTable(block);
 
         if (need_render_progress && tty_buf)
-            progress_indication.writeProgress(*tty_buf);
+        {
+            std::unique_lock lock(tty_mutex);
+            progress_indication.writeProgress(*tty_buf, lock);
+        }
         if (need_render_progress_table && tty_buf && !cancelled)
         {
             bool toggle_enabled = getClientConfiguration().getBool("enable-progress-table-toggle", true);
-            progress_table.writeTable(*tty_buf, progress_table_toggle_on.load(), toggle_enabled);
+            std::unique_lock lock(tty_mutex);
+            progress_table.writeTable(*tty_buf, lock, progress_table_toggle_on.load(), toggle_enabled);
         }
 
         if (profile_events.print)
@@ -1429,9 +1431,15 @@ void ClientBase::onProfileEvents(Block & block)
                 profile_events.watch.restart();
                 initLogsOutputStream();
                 if (need_render_progress && tty_buf)
-                    progress_indication.clearProgressOutput(*tty_buf);
+                {
+                    std::unique_lock lock(tty_mutex);
+                    progress_indication.clearProgressOutput(*tty_buf, lock);
+                }
                 if (need_render_progress_table && tty_buf)
-                    progress_table.clearTableOutput(*tty_buf);
+                {
+                    std::unique_lock lock(tty_mutex);
+                    progress_table.clearTableOutput(*tty_buf, lock);
+                }
                 logs_out_stream->writeProfileEvents(block);
                 logs_out_stream->flush();
 
@@ -1450,7 +1458,10 @@ void ClientBase::onProfileEvents(Block & block)
 void ClientBase::resetOutput()
 {
     if (need_render_progress_table && tty_buf)
-        progress_table.clearTableOutput(*tty_buf);
+    {
+        std::unique_lock lock(tty_mutex);
+        progress_table.clearTableOutput(*tty_buf, lock);
+    }
 
     /// Order is important: format, compression, file
 
@@ -1619,6 +1630,9 @@ void ClientBase::processInsertQuery(const String & query_to_execute, ASTPtr pars
     if (send_external_tables)
         sendExternalTables(parsed_query);
 
+    startKeystrokeInterceptorIfExists();
+    SCOPE_EXIT({ stopKeystrokeInterceptorIfExists(); });
+
     /// Receive description of table structure.
     Block sample;
     ColumnsDescription columns_description;
@@ -1665,7 +1679,7 @@ void ClientBase::sendData(Block & sample, const ColumnsDescription & columns_des
 
         /// Set callback to be called on file progress.
         if (tty_buf)
-            progress_indication.setFileProgressCallback(client_context, *tty_buf);
+            progress_indication.setFileProgressCallback(client_context, *tty_buf, tty_mutex);
     }
 
     /// If data fetched from file (maybe compressed file)
@@ -1947,9 +1961,15 @@ void ClientBase::cancelQuery()
         }
 
     if (need_render_progress && tty_buf)
-        progress_indication.clearProgressOutput(*tty_buf);
+    {
+        std::unique_lock lock(tty_mutex);
+        progress_indication.clearProgressOutput(*tty_buf, lock);
+    }
     if (need_render_progress_table && tty_buf)
-        progress_table.clearTableOutput(*tty_buf);
+    {
+        std::unique_lock lock(tty_mutex);
+        progress_table.clearTableOutput(*tty_buf, lock);
+    }
 
     if (is_interactive)
         output_stream << "Cancelling query." << std::endl;
@@ -2112,9 +2132,15 @@ void ClientBase::processParsedSingleQuery(const String & full_query, const Strin
     {
         initLogsOutputStream();
         if (need_render_progress && tty_buf)
-            progress_indication.clearProgressOutput(*tty_buf);
+        {
+            std::unique_lock lock(tty_mutex);
+            progress_indication.clearProgressOutput(*tty_buf, lock);
+        }
         if (need_render_progress_table && tty_buf)
-            progress_table.clearTableOutput(*tty_buf);
+        {
+            std::unique_lock lock(tty_mutex);
+            progress_table.clearTableOutput(*tty_buf, lock);
+        }
         logs_out_stream->writeProfileEvents(profile_events.last_block);
         logs_out_stream->flush();
 
@@ -2613,6 +2639,39 @@ bool ClientBase::addMergeTreeSettings(ASTCreateQuery & ast_create)
     return added_new_setting;
 }
 
+void ClientBase::startKeystrokeInterceptorIfExists()
+{
+    if (keystroke_interceptor)
+    {
+        progress_table_toggle_on = false;
+        try
+        {
+            keystroke_interceptor->startIntercept();
+        }
+        catch (const DB::Exception &)
+        {
+            error_stream << getCurrentExceptionMessage(false);
+            keystroke_interceptor.reset();
+        }
+    }
+}
+
+void ClientBase::stopKeystrokeInterceptorIfExists()
+{
+    if (keystroke_interceptor)
+    {
+        try
+        {
+            keystroke_interceptor->stopIntercept();
+        }
+        catch (...)
+        {
+            error_stream << getCurrentExceptionMessage(false);
+            keystroke_interceptor.reset();
+        }
+    }
+}
+
 void ClientBase::runInteractive()
 {
     if (getClientConfiguration().has("query_id"))

src/Client/ClientBase.h

@@ -208,6 +208,9 @@ private:
     void initQueryIdFormats();
     bool addMergeTreeSettings(ASTCreateQuery & ast_create);
 
+    void startKeystrokeInterceptorIfExists();
+    void stopKeystrokeInterceptorIfExists();
+
 protected:
 
     class QueryInterruptHandler : private boost::noncopyable
@@ -325,6 +328,7 @@ protected:
     /// /dev/tty if accessible or std::cerr - for progress bar.
     /// We prefer to output progress bar directly to tty to allow user to redirect stdout and stderr and still get the progress indication.
     std::unique_ptr<WriteBufferFromFileDescriptor> tty_buf;
+    std::mutex tty_mutex;
 
     String home_path;
     String history_file; /// Path to a file containing command history.

src/Client/ClientBaseHelpers.cpp

@@ -140,8 +140,6 @@ void highlight(const String & query, std::vector<replxx::Replxx::Color> & colors
     /// We don't do highlighting for foreign dialects, such as PRQL and Kusto.
     /// Only normal ClickHouse SQL queries are highlighted.
 
-    /// Currently we highlight only the first query in the multi-query mode.
-
     ParserQuery parser(end, false, context.getSettingsRef()[Setting::implicit_select]);
     ASTPtr ast;
     bool parse_res = false;

src/Client/ProgressTable.cpp

@@ -14,6 +14,7 @@
 #include <Common/formatReadable.h>
 
 #include <format>
+#include <mutex>
 #include <numeric>
 #include <unordered_map>
 
@@ -192,7 +193,8 @@ void writeWithWidthStrict(Out & out, std::string_view s, size_t width)
 
 }
 
-void ProgressTable::writeTable(WriteBufferFromFileDescriptor & message, bool show_table, bool toggle_enabled)
+void ProgressTable::writeTable(
+    WriteBufferFromFileDescriptor & message, std::unique_lock<std::mutex> &, bool show_table, bool toggle_enabled)
 {
     std::lock_guard lock{mutex};
     if (!show_table && toggle_enabled)
@@ -360,7 +362,7 @@ void ProgressTable::updateTable(const Block & block)
     written_first_block = true;
 }
 
-void ProgressTable::clearTableOutput(WriteBufferFromFileDescriptor & message)
+void ProgressTable::clearTableOutput(WriteBufferFromFileDescriptor & message, std::unique_lock<std::mutex> &)
 {
     message << "\r" << CLEAR_TO_END_OF_SCREEN << SHOW_CURSOR;
     message.next();

src/Client/ProgressTable.h

@@ -27,8 +27,9 @@ public:
     }
 
     /// Write progress table with metrics.
-    void writeTable(WriteBufferFromFileDescriptor & message, bool show_table, bool toggle_enabled);
+    void writeTable(WriteBufferFromFileDescriptor & message, std::unique_lock<std::mutex> & message_lock,
-    void clearTableOutput(WriteBufferFromFileDescriptor & message);
+            bool show_table, bool toggle_enabled);
+    void clearTableOutput(WriteBufferFromFileDescriptor & message, std::unique_lock<std::mutex> & message_lock);
     void writeFinalTable();
 
     /// Update the metric values. They can be updated from:

src/Columns/ColumnArray.cpp

@@ -662,6 +662,8 @@ ColumnPtr ColumnArray::filter(const Filter & filt, ssize_t result_size_hint) con
         return filterNumber<Int128>(filt, result_size_hint);
     if (typeid_cast<const ColumnInt256 *>(data.get()))
         return filterNumber<Int256>(filt, result_size_hint);
+    if (typeid_cast<const ColumnBFloat16 *>(data.get()))
+        return filterNumber<BFloat16>(filt, result_size_hint);
     if (typeid_cast<const ColumnFloat32 *>(data.get()))
         return filterNumber<Float32>(filt, result_size_hint);
     if (typeid_cast<const ColumnFloat64 *>(data.get()))
@@ -1065,6 +1067,8 @@ ColumnPtr ColumnArray::replicate(const Offsets & replicate_offsets) const
         return replicateNumber<Int128>(replicate_offsets);
     if (typeid_cast<const ColumnInt256 *>(data.get()))
         return replicateNumber<Int256>(replicate_offsets);
+    if (typeid_cast<const ColumnBFloat16 *>(data.get()))
+        return replicateNumber<BFloat16>(replicate_offsets);
     if (typeid_cast<const ColumnFloat32 *>(data.get()))
         return replicateNumber<Float32>(replicate_offsets);
     if (typeid_cast<const ColumnFloat64 *>(data.get()))

src/Columns/ColumnUnique.cpp

@@ -16,6 +16,7 @@ template class ColumnUnique<ColumnInt128>;
 template class ColumnUnique<ColumnUInt128>;
 template class ColumnUnique<ColumnInt256>;
 template class ColumnUnique<ColumnUInt256>;
+template class ColumnUnique<ColumnBFloat16>;
 template class ColumnUnique<ColumnFloat32>;
 template class ColumnUnique<ColumnFloat64>;
 template class ColumnUnique<ColumnString>;

src/Columns/ColumnUnique.h

@@ -760,6 +760,7 @@ extern template class ColumnUnique<ColumnInt128>;
 extern template class ColumnUnique<ColumnUInt128>;
 extern template class ColumnUnique<ColumnInt256>;
 extern template class ColumnUnique<ColumnUInt256>;
+extern template class ColumnUnique<ColumnBFloat16>;
 extern template class ColumnUnique<ColumnFloat32>;
 extern template class ColumnUnique<ColumnFloat64>;
 extern template class ColumnUnique<ColumnString>;

src/Columns/ColumnVector.cpp

@@ -118,9 +118,9 @@ struct ColumnVector<T>::less_stable
         if (unlikely(parent.data[lhs] == parent.data[rhs]))
             return lhs < rhs;
 
-        if constexpr (std::is_floating_point_v<T>)
+        if constexpr (is_floating_point<T>)
         {
-            if (unlikely(std::isnan(parent.data[lhs]) && std::isnan(parent.data[rhs])))
+            if (unlikely(isNaN(parent.data[lhs]) && isNaN(parent.data[rhs])))
             {
                 return lhs < rhs;
             }
@@ -150,9 +150,9 @@ struct ColumnVector<T>::greater_stable
         if (unlikely(parent.data[lhs] == parent.data[rhs]))
             return lhs < rhs;
 
-        if constexpr (std::is_floating_point_v<T>)
+        if constexpr (is_floating_point<T>)
         {
-            if (unlikely(std::isnan(parent.data[lhs]) && std::isnan(parent.data[rhs])))
+            if (unlikely(isNaN(parent.data[lhs]) && isNaN(parent.data[rhs])))
             {
                 return lhs < rhs;
             }
@@ -224,9 +224,9 @@ void ColumnVector<T>::getPermutation(IColumn::PermutationSortDirection direction
 
     iota(res.data(), data_size, IColumn::Permutation::value_type(0));
 
-    if constexpr (has_find_extreme_implementation<T> && !std::is_floating_point_v<T>)
+    if constexpr (has_find_extreme_implementation<T> && !is_floating_point<T>)
     {
-        /// Disabled for:floating point
+        /// Disabled for floating point:
         /// * floating point: We don't deal with nan_direction_hint
         /// * stability::Stable: We might return any value, not the first
         if ((limit == 1) && (stability == IColumn::PermutationSortStability::Unstable))
@@ -256,7 +256,7 @@ void ColumnVector<T>::getPermutation(IColumn::PermutationSortDirection direction
             bool sort_is_stable = stability == IColumn::PermutationSortStability::Stable;
 
             /// TODO: LSD RadixSort is currently not stable if direction is descending, or value is floating point
-            bool use_radix_sort = (sort_is_stable && ascending && !std::is_floating_point_v<T>) || !sort_is_stable;
+            bool use_radix_sort = (sort_is_stable && ascending && !is_floating_point<T>) || !sort_is_stable;
 
             /// Thresholds on size. Lower threshold is arbitrary. Upper threshold is chosen by the type for histogram counters.
             if (data_size >= 256 && data_size <= std::numeric_limits<UInt32>::max() && use_radix_sort)
@@ -283,7 +283,7 @@ void ColumnVector<T>::getPermutation(IColumn::PermutationSortDirection direction
 
                 /// Radix sort treats all NaNs to be greater than all numbers.
                 /// If the user needs the opposite, we must move them accordingly.
-                if (std::is_floating_point_v<T> && nan_direction_hint < 0)
+                if (is_floating_point<T> && nan_direction_hint < 0)
                 {
                     size_t nans_to_move = 0;
 
@@ -330,7 +330,7 @@ void ColumnVector<T>::updatePermutation(IColumn::PermutationSortDirection direct
         if constexpr (is_arithmetic_v<T> && !is_big_int_v<T>)
         {
             /// TODO: LSD RadixSort is currently not stable if direction is descending, or value is floating point
-            bool use_radix_sort = (sort_is_stable && ascending && !std::is_floating_point_v<T>) || !sort_is_stable;
+            bool use_radix_sort = (sort_is_stable && ascending && !is_floating_point<T>) || !sort_is_stable;
             size_t size = end - begin;
 
             /// Thresholds on size. Lower threshold is arbitrary. Upper threshold is chosen by the type for histogram counters.
@@ -353,7 +353,7 @@ void ColumnVector<T>::updatePermutation(IColumn::PermutationSortDirection direct
 
                 /// Radix sort treats all NaNs to be greater than all numbers.
                 /// If the user needs the opposite, we must move them accordingly.
-                if (std::is_floating_point_v<T> && nan_direction_hint < 0)
+                if (is_floating_point<T> && nan_direction_hint < 0)
                 {
                     size_t nans_to_move = 0;
 
@@ -1005,6 +1005,7 @@ template class ColumnVector<Int32>;
 template class ColumnVector<Int64>;
 template class ColumnVector<Int128>;
 template class ColumnVector<Int256>;
+template class ColumnVector<BFloat16>;
 template class ColumnVector<Float32>;
 template class ColumnVector<Float64>;
 template class ColumnVector<UUID>;

src/Columns/ColumnVector.h

@@ -481,6 +481,7 @@ extern template class ColumnVector<Int32>;
 extern template class ColumnVector<Int64>;
 extern template class ColumnVector<Int128>;
 extern template class ColumnVector<Int256>;
+extern template class ColumnVector<BFloat16>;
 extern template class ColumnVector<Float32>;
 extern template class ColumnVector<Float64>;
 extern template class ColumnVector<UUID>;

src/Columns/ColumnsCommon.cpp

@@ -328,6 +328,7 @@ INSTANTIATE(Int32)
 INSTANTIATE(Int64)
 INSTANTIATE(Int128)
 INSTANTIATE(Int256)
+INSTANTIATE(BFloat16)
 INSTANTIATE(Float32)
 INSTANTIATE(Float64)
 INSTANTIATE(Decimal32)

src/Columns/ColumnsNumber.h

@@ -23,6 +23,7 @@ using ColumnInt64 = ColumnVector<Int64>;
 using ColumnInt128 = ColumnVector<Int128>;
 using ColumnInt256 = ColumnVector<Int256>;
 
+using ColumnBFloat16 = ColumnVector<BFloat16>;
 using ColumnFloat32 = ColumnVector<Float32>;
 using ColumnFloat64 = ColumnVector<Float64>;
 

src/Columns/IColumn.cpp

@@ -443,6 +443,7 @@ template class IColumnHelper<ColumnVector<Int32>, ColumnFixedSizeHelper>;
 template class IColumnHelper<ColumnVector<Int64>, ColumnFixedSizeHelper>;
 template class IColumnHelper<ColumnVector<Int128>, ColumnFixedSizeHelper>;
 template class IColumnHelper<ColumnVector<Int256>, ColumnFixedSizeHelper>;
+template class IColumnHelper<ColumnVector<BFloat16>, ColumnFixedSizeHelper>;
 template class IColumnHelper<ColumnVector<Float32>, ColumnFixedSizeHelper>;
 template class IColumnHelper<ColumnVector<Float64>, ColumnFixedSizeHelper>;
 template class IColumnHelper<ColumnVector<UUID>, ColumnFixedSizeHelper>;

src/Columns/MaskOperations.cpp

@@ -63,6 +63,7 @@ INSTANTIATE(Int32)
 INSTANTIATE(Int64)
 INSTANTIATE(Int128)
 INSTANTIATE(Int256)
+INSTANTIATE(BFloat16)
 INSTANTIATE(Float32)
 INSTANTIATE(Float64)
 INSTANTIATE(Decimal32)
@@ -200,6 +201,7 @@ static MaskInfo extractMaskImpl(
           || extractMaskNumeric<inverted, Int16>(mask, column, null_value, null_bytemap, nulls, mask_info)
           || extractMaskNumeric<inverted, Int32>(mask, column, null_value, null_bytemap, nulls, mask_info)
           || extractMaskNumeric<inverted, Int64>(mask, column, null_value, null_bytemap, nulls, mask_info)
+          || extractMaskNumeric<inverted, BFloat16>(mask, column, null_value, null_bytemap, nulls, mask_info)
           || extractMaskNumeric<inverted, Float32>(mask, column, null_value, null_bytemap, nulls, mask_info)
           || extractMaskNumeric<inverted, Float64>(mask, column, null_value, null_bytemap, nulls, mask_info)))
         throw Exception(ErrorCodes::ILLEGAL_COLUMN, "Cannot convert column {} to mask.", column->getName());

src/Columns/tests/gtest_column_vector.cpp

@@ -93,6 +93,7 @@ TEST(ColumnVector, Filter)
     testFilter<Int64>();
     testFilter<UInt128>();
     testFilter<Int256>();
+    testFilter<BFloat16>();
     testFilter<Float32>();
     testFilter<Float64>();
     testFilter<UUID>();

src/Columns/tests/gtest_low_cardinality.cpp

@@ -45,6 +45,7 @@ TEST(ColumnLowCardinality, Insert)
     testLowCardinalityNumberInsert<Int128>(std::make_shared<DataTypeInt128>());
     testLowCardinalityNumberInsert<Int256>(std::make_shared<DataTypeInt256>());
 
+    testLowCardinalityNumberInsert<BFloat16>(std::make_shared<DataTypeBFloat16>());
     testLowCardinalityNumberInsert<Float32>(std::make_shared<DataTypeFloat32>());
     testLowCardinalityNumberInsert<Float64>(std::make_shared<DataTypeFloat64>());
 }

src/Common/CPUID.h

@@ -266,6 +266,11 @@ inline bool haveAVX512VBMI2() noexcept
     return haveAVX512F() && ((CPUInfo(0x7, 0).registers.ecx >> 6) & 1u);
 }
 
+inline bool haveAVX512BF16() noexcept
+{
+    return haveAVX512F() && ((CPUInfo(0x7, 1).registers.eax >> 5) & 1u);
+}
+
 inline bool haveRDRAND() noexcept
 {
     return CPUInfo(0x0).registers.eax >= 0x7 && ((CPUInfo(0x1).registers.ecx >> 30) & 1u);
@@ -326,6 +331,7 @@ inline bool haveAMXINT8() noexcept
     OP(AVX512VL)             \
     OP(AVX512VBMI)           \
     OP(AVX512VBMI2)          \
+    OP(AVX512BF16)           \
     OP(PREFETCHWT1)          \
     OP(SHA)                  \
     OP(ADX)                  \

src/Common/CurrentMetrics.cpp

@@ -49,6 +49,7 @@
     M(TemporaryFilesForSort, "Number of temporary files created for external sorting") \
     M(TemporaryFilesForAggregation, "Number of temporary files created for external aggregation") \
     M(TemporaryFilesForJoin, "Number of temporary files created for JOIN") \
+    M(TemporaryFilesForMerge, "Number of temporary files for vertical merge") \
     M(TemporaryFilesUnknown, "Number of temporary files created without known purpose") \
     M(Read, "Number of read (read, pread, io_getevents, etc.) syscalls in fly") \
     M(RemoteRead, "Number of read with remote reader in fly") \

src/Common/FailPoint.cpp

@@ -87,6 +87,7 @@ APPLY_FOR_FAILPOINTS(M, M, M, M)
 
 std::unordered_map<String, std::shared_ptr<FailPointChannel>> FailPointInjection::fail_point_wait_channels;
 std::mutex FailPointInjection::mu;
+
 class FailPointChannel : private boost::noncopyable
 {
 public:

src/Common/FailPoint.h

@@ -15,6 +15,7 @@
 
 #include <unordered_map>
 
+
 namespace DB
 {
 
@@ -27,6 +28,7 @@ namespace DB
 /// 3. in test file, we can use system failpoint enable/disable 'failpoint_name'
 
 class FailPointChannel;
+
 class FailPointInjection
 {
 public:

src/Common/FieldVisitorConvertToNumber.cpp

@@ -1,5 +1,4 @@
 #include <Common/FieldVisitorConvertToNumber.h>
-#include "base/Decimal.h"
 
 namespace DB
 {
@@ -17,6 +16,7 @@ template class FieldVisitorConvertToNumber<Int128>;
 template class FieldVisitorConvertToNumber<UInt128>;
 template class FieldVisitorConvertToNumber<Int256>;
 template class FieldVisitorConvertToNumber<UInt256>;
+//template class FieldVisitorConvertToNumber<BFloat16>;
 template class FieldVisitorConvertToNumber<Float32>;
 template class FieldVisitorConvertToNumber<Float64>;
 

src/Common/FieldVisitorConvertToNumber.h

@@ -58,7 +58,7 @@ public:
 
     T operator() (const Float64 & x) const
     {
-        if constexpr (!std::is_floating_point_v<T>)
+        if constexpr (!is_floating_point<T>)
         {
             if (!isFinite(x))
             {
@@ -88,7 +88,7 @@ public:
     template <typename U>
     T operator() (const DecimalField<U> & x) const
     {
-        if constexpr (std::is_floating_point_v<T>)
+        if constexpr (is_floating_point<T>)
             return x.getValue().template convertTo<T>() / x.getScaleMultiplier().template convertTo<T>();
         else
             return (x.getValue() / x.getScaleMultiplier()).template convertTo<T>();
@@ -129,6 +129,7 @@ extern template class FieldVisitorConvertToNumber<Int128>;
 extern template class FieldVisitorConvertToNumber<UInt128>;
 extern template class FieldVisitorConvertToNumber<Int256>;
 extern template class FieldVisitorConvertToNumber<UInt256>;
+//extern template class FieldVisitorConvertToNumber<BFloat16>;
 extern template class FieldVisitorConvertToNumber<Float32>;
 extern template class FieldVisitorConvertToNumber<Float64>;
 

src/Common/HashTable/Hash.h

@@ -322,6 +322,7 @@ DEFINE_HASH(Int32)
 DEFINE_HASH(Int64)
 DEFINE_HASH(Int128)
 DEFINE_HASH(Int256)
+DEFINE_HASH(BFloat16)
 DEFINE_HASH(Float32)
 DEFINE_HASH(Float64)
 DEFINE_HASH(DB::UUID)

src/Common/HashTable/HashTable.h

@@ -76,7 +76,7 @@ struct HashTableNoState
 template <typename T>
 inline bool bitEquals(T a, T b)
 {
-    if constexpr (std::is_floating_point_v<T>)
+    if constexpr (is_floating_point<T>)
         /// Note that memcmp with constant size is a compiler builtin.
         return 0 == memcmp(&a, &b, sizeof(T)); /// NOLINT
     else

src/Common/NaNUtils.h

@@ -3,24 +3,24 @@
 #include <cmath>
 #include <limits>
 #include <type_traits>
+#include <base/DecomposedFloat.h>
 
 
 template <typename T>
 inline bool isNaN(T x)
 {
     /// To be sure, that this function is zero-cost for non-floating point types.
-    if constexpr (std::is_floating_point_v<T>)
+    if constexpr (is_floating_point<T>)
-        return std::isnan(x);
+        return DecomposedFloat(x).isNaN();
     else
         return false;
 }
 
-
 template <typename T>
 inline bool isFinite(T x)
 {
-    if constexpr (std::is_floating_point_v<T>)
+    if constexpr (is_floating_point<T>)
-        return std::isfinite(x);
+        return DecomposedFloat(x).isFinite();
     else
         return true;
 }
@@ -28,7 +28,7 @@ inline bool isFinite(T x)
 template <typename T>
 bool canConvertTo(Float64 x)
 {
-    if constexpr (std::is_floating_point_v<T>)
+    if constexpr (is_floating_point<T>)
         return true;
     if (!isFinite(x))
         return false;
@@ -46,3 +46,12 @@ T NaNOrZero()
     else
         return {};
 }
+
+template <typename T>
+bool signBit(T x)
+{
+    if constexpr (is_floating_point<T>)
+        return DecomposedFloat(x).isNegative();
+    else
+        return x < 0;
+}

src/Common/ProgressIndication.cpp

@@ -2,6 +2,7 @@
 #include <algorithm>
 #include <cstddef>
 #include <iostream>
+#include <mutex>
 #include <numeric>
 #include <filesystem>
 #include <cmath>
@@ -49,12 +50,13 @@ void ProgressIndication::resetProgress()
     }
 }
 
-void ProgressIndication::setFileProgressCallback(ContextMutablePtr context, WriteBufferFromFileDescriptor & message)
+void ProgressIndication::setFileProgressCallback(ContextMutablePtr context, WriteBufferFromFileDescriptor & message, std::mutex & message_mutex)
 {
     context->setFileProgressCallback([&](const FileProgress & file_progress)
     {
         progress.incrementPiecewiseAtomically(Progress(file_progress));
-        writeProgress(message);
+        std::unique_lock message_lock(message_mutex);
+        writeProgress(message, message_lock);
     });
 }
 
@@ -113,7 +115,7 @@ void ProgressIndication::writeFinalProgress()
         output_stream << "\nPeak memory usage: " << formatReadableSizeWithBinarySuffix(peak_memory_usage) << ".";
 }
 
-void ProgressIndication::writeProgress(WriteBufferFromFileDescriptor & message)
+void ProgressIndication::writeProgress(WriteBufferFromFileDescriptor & message, std::unique_lock<std::mutex> &)
 {
     std::lock_guard lock(progress_mutex);
 
@@ -274,7 +276,7 @@ void ProgressIndication::writeProgress(WriteBufferFromFileDescriptor & message)
     message.next();
 }
 
-void ProgressIndication::clearProgressOutput(WriteBufferFromFileDescriptor & message)
+void ProgressIndication::clearProgressOutput(WriteBufferFromFileDescriptor & message, std::unique_lock<std::mutex> &)
 {
     std::lock_guard lock(progress_mutex);
 

src/Common/ProgressIndication.h

@@ -8,6 +8,7 @@
 
 #include <iostream>
 #include <mutex>
+#include <queue>
 #include <unordered_map>
 #include <unordered_set>
 
@@ -47,8 +48,8 @@ public:
     }
 
     /// Write progress bar.
-    void writeProgress(WriteBufferFromFileDescriptor & message);
+    void writeProgress(WriteBufferFromFileDescriptor & message, std::unique_lock<std::mutex> & message_lock);
-    void clearProgressOutput(WriteBufferFromFileDescriptor & message);
+    void clearProgressOutput(WriteBufferFromFileDescriptor & message, std::unique_lock<std::mutex> & message_lock);
 
     /// Write summary.
     void writeFinalProgress();
@@ -67,7 +68,7 @@ public:
     /// In some cases there is a need to update progress value, when there is no access to progress_inidcation object.
     /// In this case it is added via context.
     /// `write_progress_on_update` is needed to write progress for loading files data via pipe in non-interactive mode.
-    void setFileProgressCallback(ContextMutablePtr context, WriteBufferFromFileDescriptor & message);
+    void setFileProgressCallback(ContextMutablePtr context, WriteBufferFromFileDescriptor & message, std::mutex & message_mutex);
 
     /// How much seconds passed since query execution start.
     double elapsedSeconds() const { return getElapsedNanoseconds() / 1e9; }

src/Common/TargetSpecific.cpp

@@ -23,6 +23,8 @@ UInt32 getSupportedArchs()
         result |= static_cast<UInt32>(TargetArch::AVX512VBMI);
     if (CPU::CPUFlagsCache::have_AVX512VBMI2)
         result |= static_cast<UInt32>(TargetArch::AVX512VBMI2);
+    if (CPU::CPUFlagsCache::have_AVX512BF16)
+        result |= static_cast<UInt32>(TargetArch::AVX512BF16);
     if (CPU::CPUFlagsCache::have_AMXBF16)
         result |= static_cast<UInt32>(TargetArch::AMXBF16);
     if (CPU::CPUFlagsCache::have_AMXTILE)
@@ -50,6 +52,7 @@ String toString(TargetArch arch)
         case TargetArch::AVX512BW:    return "avx512bw";
         case TargetArch::AVX512VBMI:  return "avx512vbmi";
         case TargetArch::AVX512VBMI2: return "avx512vbmi2";
+        case TargetArch::AVX512BF16:  return "avx512bf16";
         case TargetArch::AMXBF16: return "amxbf16";
         case TargetArch::AMXTILE: return "amxtile";
         case TargetArch::AMXINT8: return "amxint8";

src/Common/TargetSpecific.h

@@ -83,9 +83,10 @@ enum class TargetArch : UInt32
     AVX512BW    = (1 << 4),
     AVX512VBMI  = (1 << 5),
     AVX512VBMI2 = (1 << 6),
-    AMXBF16 = (1 << 7),
+    AVX512BF16 = (1 << 7),
-    AMXTILE = (1 << 8),
+    AMXBF16 = (1 << 8),
-    AMXINT8 = (1 << 9),
+    AMXTILE = (1 << 9),
+    AMXINT8 = (1 << 10),
 };
 
 /// Runtime detection.
@@ -102,6 +103,7 @@ String toString(TargetArch arch);
 /// NOLINTNEXTLINE
 #define USE_MULTITARGET_CODE 1
 
+#define AVX512BF16_FUNCTION_SPECIFIC_ATTRIBUTE __attribute__((target("sse,sse2,sse3,ssse3,sse4,popcnt,avx,avx2,avx512f,avx512bw,avx512vl,avx512vbmi,avx512vbmi2,avx512bf16")))
 #define AVX512VBMI2_FUNCTION_SPECIFIC_ATTRIBUTE __attribute__((target("sse,sse2,sse3,ssse3,sse4,popcnt,avx,avx2,avx512f,avx512bw,avx512vl,avx512vbmi,avx512vbmi2")))
 #define AVX512VBMI_FUNCTION_SPECIFIC_ATTRIBUTE __attribute__((target("sse,sse2,sse3,ssse3,sse4,popcnt,avx,avx2,avx512f,avx512bw,avx512vl,avx512vbmi")))
 #define AVX512BW_FUNCTION_SPECIFIC_ATTRIBUTE __attribute__((target("sse,sse2,sse3,ssse3,sse4,popcnt,avx,avx2,avx512f,avx512bw")))
@@ -111,6 +113,8 @@ String toString(TargetArch arch);
 #define SSE42_FUNCTION_SPECIFIC_ATTRIBUTE __attribute__((target("sse,sse2,sse3,ssse3,sse4,popcnt")))
 #define DEFAULT_FUNCTION_SPECIFIC_ATTRIBUTE
 
+#   define BEGIN_AVX512BF16_SPECIFIC_CODE \
+        _Pragma("clang attribute push(__attribute__((target(\"sse,sse2,sse3,ssse3,sse4,popcnt,avx,avx2,avx512f,avx512bw,avx512vl,avx512vbmi,avx512vbmi2,avx512bf16\"))),apply_to=function)")
 #   define BEGIN_AVX512VBMI2_SPECIFIC_CODE \
         _Pragma("clang attribute push(__attribute__((target(\"sse,sse2,sse3,ssse3,sse4,popcnt,avx,avx2,avx512f,avx512bw,avx512vl,avx512vbmi,avx512vbmi2\"))),apply_to=function)")
 #   define BEGIN_AVX512VBMI_SPECIFIC_CODE \
@@ -197,6 +201,14 @@ namespace TargetSpecific::AVX512VBMI2 { \
 } \
 END_TARGET_SPECIFIC_CODE
 
+#define DECLARE_AVX512BF16_SPECIFIC_CODE(...) \
+BEGIN_AVX512BF16_SPECIFIC_CODE \
+namespace TargetSpecific::AVX512BF16 { \
+    DUMMY_FUNCTION_DEFINITION \
+    using namespace DB::TargetSpecific::AVX512BF16; \
+    __VA_ARGS__ \
+} \
+END_TARGET_SPECIFIC_CODE
 
 #else
 
@@ -211,6 +223,7 @@ END_TARGET_SPECIFIC_CODE
 #define DECLARE_AVX512BW_SPECIFIC_CODE(...)
 #define DECLARE_AVX512VBMI_SPECIFIC_CODE(...)
 #define DECLARE_AVX512VBMI2_SPECIFIC_CODE(...)
+#define DECLARE_AVX512BF16_SPECIFIC_CODE(...)
 
 #endif
 
@@ -229,7 +242,8 @@ DECLARE_AVX2_SPECIFIC_CODE   (__VA_ARGS__) \
 DECLARE_AVX512F_SPECIFIC_CODE(__VA_ARGS__) \
 DECLARE_AVX512BW_SPECIFIC_CODE    (__VA_ARGS__) \
 DECLARE_AVX512VBMI_SPECIFIC_CODE  (__VA_ARGS__) \
-DECLARE_AVX512VBMI2_SPECIFIC_CODE (__VA_ARGS__)
+DECLARE_AVX512VBMI2_SPECIFIC_CODE (__VA_ARGS__) \
+DECLARE_AVX512BF16_SPECIFIC_CODE (__VA_ARGS__)
 
 DECLARE_DEFAULT_CODE(
     constexpr auto BuildArch = TargetArch::Default; /// NOLINT
@@ -263,6 +277,10 @@ DECLARE_AVX512VBMI2_SPECIFIC_CODE(
     constexpr auto BuildArch = TargetArch::AVX512VBMI2; /// NOLINT
 ) // DECLARE_AVX512VBMI2_SPECIFIC_CODE
 
+DECLARE_AVX512BF16_SPECIFIC_CODE(
+    constexpr auto BuildArch = TargetArch::AVX512BF16; /// NOLINT
+) // DECLARE_AVX512BF16_SPECIFIC_CODE
+
 /** Runtime Dispatch helpers for class members.
   *
   * Example of usage:

src/Common/ThreadStatus.cpp

@@ -204,6 +204,16 @@ bool ThreadStatus::isQueryCanceled() const
     return false;
 }
 
+size_t ThreadStatus::getNextPlanStepIndex() const
+{
+    return local_data.plan_step_index->fetch_add(1);
+}
+
+size_t ThreadStatus::getNextPipelineProcessorIndex() const
+{
+    return local_data.pipeline_processor_index->fetch_add(1);
+}
+
 ThreadStatus::~ThreadStatus()
 {
     flushUntrackedMemory();

src/Common/ThreadStatus.h

@@ -11,6 +11,7 @@
 
 #include <boost/noncopyable.hpp>
 
+#include <atomic>
 #include <functional>
 #include <memory>
 #include <mutex>
@@ -90,6 +91,11 @@ public:
         String query_for_logs;
         UInt64 normalized_query_hash = 0;
 
+        // Since processors might be added on the fly within expand() function we use atomic_size_t.
+        // These two fields are used for EXPLAIN PLAN / PIPELINE.
+        std::shared_ptr<std::atomic_size_t> plan_step_index = std::make_shared<std::atomic_size_t>(0);
+        std::shared_ptr<std::atomic_size_t> pipeline_processor_index = std::make_shared<std::atomic_size_t>(0);
+
         QueryIsCanceledPredicate query_is_canceled_predicate = {};
     };
 
@@ -313,6 +319,9 @@ public:
 
     void initGlobalProfiler(UInt64 global_profiler_real_time_period, UInt64 global_profiler_cpu_time_period);
 
+    size_t getNextPlanStepIndex() const;
+    size_t getNextPipelineProcessorIndex() const;
+
 private:
     void applyGlobalSettings();
     void applyQuerySettings();

src/Common/findExtreme.cpp

@@ -47,7 +47,7 @@ MULTITARGET_FUNCTION_AVX2_SSE42(
 
         /// Unroll the loop manually for floating point, since the compiler doesn't do it without fastmath
         /// as it might change the return value
-        if constexpr (std::is_floating_point_v<T>)
+        if constexpr (is_floating_point<T>)
         {
             constexpr size_t unroll_block = 512 / sizeof(T); /// Chosen via benchmarks with AVX2 so YMMV
             size_t unrolled_end = i + (((count - i) / unroll_block) * unroll_block);

src/Common/transformEndianness.h

@@ -38,7 +38,7 @@ inline void transformEndianness(T & x)
 }
 
 template <std::endian ToEndian, std::endian FromEndian = std::endian::native, typename T>
-requires std::is_floating_point_v<T>
+requires is_floating_point<T>
 inline void transformEndianness(T & value)
 {
     if constexpr (ToEndian != FromEndian)

src/Compression/CompressionCodecNone.h

@@ -3,7 +3,7 @@
 #include <IO/WriteBuffer.h>
 #include <Compression/ICompressionCodec.h>
 #include <IO/BufferWithOwnMemory.h>
-#include <Parsers/StringRange.h>
+
 
 namespace DB
 {

src/Compression/tests/gtest_compressionCodec.cpp

@@ -7,7 +7,6 @@
 #include <Parsers/ExpressionElementParsers.h>
 #include <Parsers/IParser.h>
 #include <Parsers/TokenIterator.h>
-#include <base/types.h>
 #include <Common/PODArray.h>
 #include <Common/Stopwatch.h>
 

src/Core/AccurateComparison.h

@@ -25,7 +25,7 @@ bool lessOp(A a, B b)
         return a < b;
 
     /// float vs float
-    if constexpr (std::is_floating_point_v<A> && std::is_floating_point_v<B>)
+    if constexpr (is_floating_point<A> && is_floating_point<B>)
         return a < b;
 
     /// anything vs NaN
@@ -49,7 +49,7 @@ bool lessOp(A a, B b)
     }
 
     /// int vs float
-    if constexpr (is_integer<A> && std::is_floating_point_v<B>)
+    if constexpr (is_integer<A> && is_floating_point<B>)
     {
         if constexpr (sizeof(A) <= 4)
             return static_cast<double>(a) < static_cast<double>(b);
@@ -57,7 +57,7 @@ bool lessOp(A a, B b)
         return DecomposedFloat<B>(b).greater(a);
     }
 
-    if constexpr (std::is_floating_point_v<A> && is_integer<B>)
+    if constexpr (is_floating_point<A> && is_integer<B>)
     {
         if constexpr (sizeof(B) <= 4)
             return static_cast<double>(a) < static_cast<double>(b);
@@ -65,8 +65,8 @@ bool lessOp(A a, B b)
         return DecomposedFloat<A>(a).less(b);
     }
 
-    static_assert(is_integer<A> || std::is_floating_point_v<A>);
+    static_assert(is_integer<A> || is_floating_point<A>);
-    static_assert(is_integer<B> || std::is_floating_point_v<B>);
+    static_assert(is_integer<B> || is_floating_point<B>);
     UNREACHABLE();
 }
 
@@ -101,7 +101,7 @@ bool equalsOp(A a, B b)
         return a == b;
 
     /// float vs float
-    if constexpr (std::is_floating_point_v<A> && std::is_floating_point_v<B>)
+    if constexpr (is_floating_point<A> && is_floating_point<B>)
         return a == b;
 
     /// anything vs NaN
@@ -125,7 +125,7 @@ bool equalsOp(A a, B b)
     }
 
     /// int vs float
-    if constexpr (is_integer<A> && std::is_floating_point_v<B>)
+    if constexpr (is_integer<A> && is_floating_point<B>)
     {
         if constexpr (sizeof(A) <= 4)
             return static_cast<double>(a) == static_cast<double>(b);
@@ -133,7 +133,7 @@ bool equalsOp(A a, B b)
         return DecomposedFloat<B>(b).equals(a);
     }
 
-    if constexpr (std::is_floating_point_v<A> && is_integer<B>)
+    if constexpr (is_floating_point<A> && is_integer<B>)
     {
         if constexpr (sizeof(B) <= 4)
             return static_cast<double>(a) == static_cast<double>(b);
@@ -163,7 +163,7 @@ inline bool NO_SANITIZE_UNDEFINED convertNumeric(From value, To & result)
         return true;
     }
 
-    if constexpr (std::is_floating_point_v<From> && std::is_floating_point_v<To>)
+    if constexpr (is_floating_point<From> && is_floating_point<To>)
     {
         /// Note that NaNs doesn't compare equal to anything, but they are still in range of any Float type.
         if (isNaN(value))

src/Core/DecimalFunctions.h

@@ -17,6 +17,7 @@ class DataTypeNumber;
 
 namespace ErrorCodes
 {
+    extern const int NOT_IMPLEMENTED;
     extern const int DECIMAL_OVERFLOW;
     extern const int ARGUMENT_OUT_OF_BOUND;
 }
@@ -310,7 +311,14 @@ ReturnType convertToImpl(const DecimalType & decimal, UInt32 scale, To & result)
     using DecimalNativeType = typename DecimalType::NativeType;
     static constexpr bool throw_exception = std::is_void_v<ReturnType>;
 
-    if constexpr (std::is_floating_point_v<To>)
+    if constexpr (std::is_same_v<To, BFloat16>)
+    {
+        if constexpr (throw_exception)
+            throw Exception(ErrorCodes::NOT_IMPLEMENTED, "Conversion from Decimal to BFloat16 is not implemented");
+        else
+            return ReturnType(false);
+    }
+    else if constexpr (is_floating_point<To>)
     {
         result = static_cast<To>(decimal.value) / static_cast<To>(scaleMultiplier<DecimalNativeType>(scale));
     }

src/Core/Field.h

@@ -257,6 +257,7 @@ template <> struct NearestFieldTypeImpl<DecimalField<Decimal64>> { using Type =
 template <> struct NearestFieldTypeImpl<DecimalField<Decimal128>> { using Type = DecimalField<Decimal128>; };
 template <> struct NearestFieldTypeImpl<DecimalField<Decimal256>> { using Type = DecimalField<Decimal256>; };
 template <> struct NearestFieldTypeImpl<DecimalField<DateTime64>> { using Type = DecimalField<DateTime64>; };
+template <> struct NearestFieldTypeImpl<BFloat16> { using Type = Float64; };
 template <> struct NearestFieldTypeImpl<Float32> { using Type = Float64; };
 template <> struct NearestFieldTypeImpl<Float64> { using Type = Float64; };
 template <> struct NearestFieldTypeImpl<const char *> { using Type = String; };

src/Core/Settings.cpp

@@ -1794,7 +1794,7 @@ Possible values:
 
 - 0 — Disabled.
 - 1 — Enabled.
-)", 0) \
+)", 1) \
     DECLARE(Int64, http_zlib_compression_level, 3, R"(
 Sets the level of data compression in the response to an HTTP request if [enable_http_compression = 1](#enable_http_compression).
 
@@ -4565,7 +4565,7 @@ Possible values:
 - 0 - Disable
 - 1 - Enable
 )", 0) \
-    DECLARE(Bool, query_plan_merge_filters, false, R"(
+    DECLARE(Bool, query_plan_merge_filters, true, R"(
 Allow to merge filters in the query plan
 )", 0) \
     DECLARE(Bool, query_plan_filter_push_down, true, R"(
@@ -5742,7 +5742,10 @@ Enable experimental functions for natural language processing.
 Enable experimental hash functions
 )", EXPERIMENTAL) \
     DECLARE(Bool, allow_experimental_object_type, false, R"(
-Allow Object and JSON data types
+Allow the obsolete Object data type
+)", EXPERIMENTAL) \
+    DECLARE(Bool, allow_experimental_bfloat16_type, false, R"(
+Allow BFloat16 data type (under development).
 )", EXPERIMENTAL) \
     DECLARE(Bool, allow_experimental_time_series_table, false, R"(
 Allows creation of tables with the [TimeSeries](../../engines/table-engines/integrations/time-series.md) table engine.

src/Core/SettingsChangesHistory.cpp

@@ -64,6 +64,7 @@ static std::initializer_list<std::pair<ClickHouseVersion, SettingsChangesHistory
     },
     {"24.11",
         {
+            {"enable_http_compression", false, true, "Improvement for read-only clients since they can't change settings"},
             {"validate_mutation_query", false, true, "New setting to validate mutation queries by default."},
             {"enable_job_stack_trace", false, true, "Enable by default collecting stack traces from job's scheduling."},
             {"allow_suspicious_types_in_group_by", true, false, "Don't allow Variant/Dynamic types in GROUP BY by default"},
@@ -77,7 +78,9 @@ static std::initializer_list<std::pair<ClickHouseVersion, SettingsChangesHistory
             {"backup_restore_keeper_max_retries_while_initializing", 0, 20, "New setting."},
             {"backup_restore_keeper_max_retries_while_handling_error", 0, 20, "New setting."},
             {"backup_restore_finish_timeout_after_error_sec", 0, 180, "New setting."},
+            {"query_plan_merge_filters", false, true, "Allow to merge filters in the query plan. This is required to properly support filter-push-down with a new analyzer."},
             {"parallel_replicas_local_plan", false, true, "Use local plan for local replica in a query with parallel replicas"},
+            {"allow_experimental_bfloat16_type", false, false, "Add new experimental BFloat16 type"},
             {"filesystem_cache_skip_download_if_exceeds_per_query_cache_write_limit", 1, 1, "Rename of setting skip_download_if_exceeds_query_cache_limit"},
             {"filesystem_cache_prefer_bigger_buffer_size", true, true, "New setting"},
             {"read_in_order_use_virtual_row", false, false, "Use virtual row while reading in order of primary key or its monotonic function fashion. It is useful when searching over multiple parts as only relevant ones are touched."},

src/Core/SortCursor.h

@@ -726,6 +726,7 @@ private:
         SortingQueueImpl<SpecializedSingleColumnSortCursor<ColumnVector<Int128>>, strategy>,
         SortingQueueImpl<SpecializedSingleColumnSortCursor<ColumnVector<Int256>>, strategy>,
 
+        SortingQueueImpl<SpecializedSingleColumnSortCursor<ColumnVector<BFloat16>>, strategy>,
         SortingQueueImpl<SpecializedSingleColumnSortCursor<ColumnVector<Float32>>, strategy>,
         SortingQueueImpl<SpecializedSingleColumnSortCursor<ColumnVector<Float64>>, strategy>,
 

src/Core/TypeId.h

@@ -21,6 +21,7 @@ enum class TypeIndex : uint8_t
     Int64,
     Int128,
     Int256,
+    BFloat16,
     Float32,
     Float64,
     Date,
@@ -94,6 +95,7 @@ TYPEID_MAP(Int32)
 TYPEID_MAP(Int64)
 TYPEID_MAP(Int128)
 TYPEID_MAP(Int256)
+TYPEID_MAP(BFloat16)
 TYPEID_MAP(Float32)
 TYPEID_MAP(Float64)
 TYPEID_MAP(UUID)

src/Core/Types_fwd.h

@@ -21,6 +21,7 @@ using Int128 = wide::integer<128, signed>;
 using UInt128 = wide::integer<128, unsigned>;
 using Int256 = wide::integer<256, signed>;
 using UInt256 = wide::integer<256, unsigned>;
+class BFloat16;
 
 namespace DB
 {

src/Core/callOnTypeIndex.h

@@ -63,6 +63,7 @@ static bool callOnBasicType(TypeIndex number, F && f)
     {
         switch (number)
         {
+            case TypeIndex::BFloat16:     return f(TypePair<T, BFloat16>());
             case TypeIndex::Float32:      return f(TypePair<T, Float32>());
             case TypeIndex::Float64:      return f(TypePair<T, Float64>());
             default:
@@ -133,6 +134,7 @@ static inline bool callOnBasicTypes(TypeIndex type_num1, TypeIndex type_num2, F
     {
         switch (type_num1)
         {
+            case TypeIndex::BFloat16: return callOnBasicType<BFloat16, _int, _float, _decimal, _datetime>(type_num2, std::forward<F>(f));
             case TypeIndex::Float32: return callOnBasicType<Float32, _int, _float, _decimal, _datetime>(type_num2, std::forward<F>(f));
             case TypeIndex::Float64: return callOnBasicType<Float64, _int, _float, _decimal, _datetime>(type_num2, std::forward<F>(f));
             default:
@@ -190,6 +192,7 @@ static bool callOnIndexAndDataType(TypeIndex number, F && f, ExtraArgs && ... ar
         case TypeIndex::Int128:         return f(TypePair<DataTypeNumber<Int128>, T>(), std::forward<ExtraArgs>(args)...);
         case TypeIndex::Int256:         return f(TypePair<DataTypeNumber<Int256>, T>(), std::forward<ExtraArgs>(args)...);
 
+        case TypeIndex::BFloat16:        return f(TypePair<DataTypeNumber<BFloat16>, T>(), std::forward<ExtraArgs>(args)...);
         case TypeIndex::Float32:        return f(TypePair<DataTypeNumber<Float32>, T>(), std::forward<ExtraArgs>(args)...);
         case TypeIndex::Float64:        return f(TypePair<DataTypeNumber<Float64>, T>(), std::forward<ExtraArgs>(args)...);
 

src/DataTypes/DataTypeNumberBase.cpp

@@ -42,6 +42,7 @@ template class DataTypeNumberBase<Int32>;
 template class DataTypeNumberBase<Int64>;
 template class DataTypeNumberBase<Int128>;
 template class DataTypeNumberBase<Int256>;
+template class DataTypeNumberBase<BFloat16>;
 template class DataTypeNumberBase<Float32>;
 template class DataTypeNumberBase<Float64>;
 

src/DataTypes/DataTypeNumberBase.h

@@ -68,6 +68,7 @@ extern template class DataTypeNumberBase<Int32>;
 extern template class DataTypeNumberBase<Int64>;
 extern template class DataTypeNumberBase<Int128>;
 extern template class DataTypeNumberBase<Int256>;
+extern template class DataTypeNumberBase<BFloat16>;
 extern template class DataTypeNumberBase<Float32>;
 extern template class DataTypeNumberBase<Float64>;
 

src/DataTypes/DataTypesBinaryEncoding.cpp

@@ -96,6 +96,7 @@ enum class BinaryTypeIndex : uint8_t
     SimpleAggregateFunction = 0x2E,
     Nested = 0x2F,
     JSON = 0x30,
+    BFloat16 = 0x31,
 };
 
 /// In future we can introduce more arguments in the JSON data type definition.
@@ -151,6 +152,8 @@ BinaryTypeIndex getBinaryTypeIndex(const DataTypePtr & type)
             return BinaryTypeIndex::Int128;
         case TypeIndex::Int256:
             return BinaryTypeIndex::Int256;
+        case TypeIndex::BFloat16:
+            return BinaryTypeIndex::BFloat16;
         case TypeIndex::Float32:
             return BinaryTypeIndex::Float32;
         case TypeIndex::Float64:
@@ -565,6 +568,8 @@ DataTypePtr decodeDataType(ReadBuffer & buf)
             return std::make_shared<DataTypeInt128>();
         case BinaryTypeIndex::Int256:
             return std::make_shared<DataTypeInt256>();
+        case BinaryTypeIndex::BFloat16:
+            return std::make_shared<DataTypeBFloat16>();
         case BinaryTypeIndex::Float32:
             return std::make_shared<DataTypeFloat32>();
         case BinaryTypeIndex::Float64:

src/DataTypes/DataTypesDecimal.cpp

@@ -2,6 +2,7 @@
 #include <DataTypes/Serializations/SerializationDecimal.h>
 
 #include <Common/typeid_cast.h>
+#include <Common/NaNUtils.h>
 #include <Core/DecimalFunctions.h>
 #include <DataTypes/DataTypeFactory.h>
 #include <IO/ReadHelpers.h>
@@ -19,6 +20,7 @@ namespace ErrorCodes
     extern const int NUMBER_OF_ARGUMENTS_DOESNT_MATCH;
     extern const int ILLEGAL_TYPE_OF_ARGUMENT;
     extern const int DECIMAL_OVERFLOW;
+    extern const int NOT_IMPLEMENTED;
 }
 
 
@@ -268,9 +270,13 @@ ReturnType convertToDecimalImpl(const typename FromDataType::FieldType & value,
 
     static constexpr bool throw_exception = std::is_same_v<ReturnType, void>;
 
-    if constexpr (std::is_floating_point_v<FromFieldType>)
+    if constexpr (std::is_same_v<typename FromDataType::FieldType, BFloat16>)
     {
-        if (!std::isfinite(value))
+        throw Exception(ErrorCodes::NOT_IMPLEMENTED, "Conversion from BFloat16 to Decimal is not implemented");
+    }
+    else if constexpr (is_floating_point<FromFieldType>)
+    {
+        if (!isFinite(value))
         {
             if constexpr (throw_exception)
                 throw Exception(ErrorCodes::DECIMAL_OVERFLOW, "{} convert overflow. Cannot convert infinity or NaN to decimal", ToDataType::family_name);

src/DataTypes/DataTypesDecimal.h

@@ -4,7 +4,6 @@
 #include <base/extended_types.h>
 #include <Common/typeid_cast.h>
 #include <base/Decimal.h>
-#include <base/Decimal_fwd.h>
 #include <DataTypes/IDataType.h>
 #include <DataTypes/DataTypeDate.h>
 #include <DataTypes/DataTypeDate32.h>
@@ -205,7 +204,6 @@ FOR_EACH_DECIMAL_TYPE(INVOKE);
 #undef INVOKE
 #undef DISPATCH
 
-
 template <typename FromDataType, typename ToDataType>
 requires (is_arithmetic_v<typename FromDataType::FieldType> && IsDataTypeDecimal<ToDataType>)
 typename ToDataType::FieldType convertToDecimal(const typename FromDataType::FieldType & value, UInt32 scale);

src/DataTypes/DataTypesNumber.cpp

@@ -54,6 +54,7 @@ void registerDataTypeNumbers(DataTypeFactory & factory)
     factory.registerDataType("Int32", createNumericDataType<Int32>);
     factory.registerDataType("Int64", createNumericDataType<Int64>);
 
+    factory.registerDataType("BFloat16", createNumericDataType<BFloat16>);
     factory.registerDataType("Float32", createNumericDataType<Float32>);
     factory.registerDataType("Float64", createNumericDataType<Float64>);
 
@@ -111,6 +112,7 @@ template class DataTypeNumber<Int8>;
 template class DataTypeNumber<Int16>;
 template class DataTypeNumber<Int32>;
 template class DataTypeNumber<Int64>;
+template class DataTypeNumber<BFloat16>;
 template class DataTypeNumber<Float32>;
 template class DataTypeNumber<Float64>;
 

src/DataTypes/DataTypesNumber.h

@@ -63,6 +63,7 @@ extern template class DataTypeNumber<Int8>;
 extern template class DataTypeNumber<Int16>;
 extern template class DataTypeNumber<Int32>;
 extern template class DataTypeNumber<Int64>;
+extern template class DataTypeNumber<BFloat16>;
 extern template class DataTypeNumber<Float32>;
 extern template class DataTypeNumber<Float64>;
 
@@ -79,6 +80,7 @@ using DataTypeInt8 = DataTypeNumber<Int8>;
 using DataTypeInt16 = DataTypeNumber<Int16>;
 using DataTypeInt32 = DataTypeNumber<Int32>;
 using DataTypeInt64 = DataTypeNumber<Int64>;
+using DataTypeBFloat16 = DataTypeNumber<BFloat16>;
 using DataTypeFloat32 = DataTypeNumber<Float32>;
 using DataTypeFloat64 = DataTypeNumber<Float64>;
 

src/DataTypes/IDataType.h

@@ -408,9 +408,11 @@ struct WhichDataType
     constexpr bool isDecimal256() const { return idx == TypeIndex::Decimal256; }
     constexpr bool isDecimal() const { return isDecimal32() || isDecimal64() || isDecimal128() || isDecimal256(); }
 
+    constexpr bool isBFloat16() const { return idx == TypeIndex::BFloat16; }
     constexpr bool isFloat32() const { return idx == TypeIndex::Float32; }
     constexpr bool isFloat64() const { return idx == TypeIndex::Float64; }
-    constexpr bool isFloat() const { return isFloat32() || isFloat64(); }
+    constexpr bool isNativeFloat() const { return isFloat32() || isFloat64(); }
+    constexpr bool isFloat() const { return isNativeFloat() || isBFloat16(); }
 
     constexpr bool isNativeNumber() const { return isNativeInteger() || isFloat(); }
     constexpr bool isNumber() const { return isInteger() || isFloat() || isDecimal(); }
@@ -621,6 +623,7 @@ template <typename T> inline constexpr bool IsDataTypeEnum<DataTypeEnum<T>> = tr
     M(Int64) \
     M(Int128) \
     M(Int256) \
+    M(BFloat16) \
     M(Float32) \
     M(Float64)
 }

src/DataTypes/Native.cpp

@@ -37,7 +37,7 @@ bool canBeNativeType(const IDataType & type)
         return canBeNativeType(*data_type_nullable.getNestedType());
     }
 
-    return data_type.isNativeInt() || data_type.isNativeUInt() || data_type.isFloat() || data_type.isDate()
+    return data_type.isNativeInt() || data_type.isNativeUInt() || data_type.isNativeFloat() || data_type.isDate()
         || data_type.isDate32() || data_type.isDateTime() || data_type.isEnum();
 }
 

src/DataTypes/NumberTraits.h

@@ -74,7 +74,7 @@ template <typename A, typename B> struct ResultOfAdditionMultiplication
 {
     using Type = typename Construct<
         is_signed_v<A> || is_signed_v<B>,
-        std::is_floating_point_v<A> || std::is_floating_point_v<B>,
+        is_floating_point<A> || is_floating_point<B>,
         nextSize(max(sizeof(A), sizeof(B)))>::Type;
 };
 
@@ -82,7 +82,7 @@ template <typename A, typename B> struct ResultOfSubtraction
 {
     using Type = typename Construct<
         true,
-        std::is_floating_point_v<A> || std::is_floating_point_v<B>,
+        is_floating_point<A> || is_floating_point<B>,
         nextSize(max(sizeof(A), sizeof(B)))>::Type;
 };
 
@@ -113,7 +113,7 @@ template <typename A, typename B> struct ResultOfModulo
     /// Example: toInt32(-199) % toUInt8(200) will return -199 that does not fit in Int8, only in Int16.
     static constexpr size_t size_of_result = result_is_signed ? nextSize(sizeof(B)) : sizeof(B);
     using Type0 = typename Construct<result_is_signed, false, size_of_result>::Type;
-    using Type = std::conditional_t<std::is_floating_point_v<A> || std::is_floating_point_v<B>, Float64, Type0>;
+    using Type = std::conditional_t<is_floating_point<A> || is_floating_point<B>, Float64, Type0>;
 };
 
 template <typename A, typename B> struct ResultOfPositiveModulo
@@ -121,21 +121,21 @@ template <typename A, typename B> struct ResultOfPositiveModulo
     /// function positive_modulo always return non-negative number.
     static constexpr size_t size_of_result = sizeof(B);
     using Type0 = typename Construct<false, false, size_of_result>::Type;
-    using Type = std::conditional_t<std::is_floating_point_v<A> || std::is_floating_point_v<B>, Float64, Type0>;
+    using Type = std::conditional_t<is_floating_point<A> || is_floating_point<B>, Float64, Type0>;
 };
 
 
 template <typename A, typename B> struct ResultOfModuloLegacy
 {
     using Type0 = typename Construct<is_signed_v<A> || is_signed_v<B>, false, sizeof(B)>::Type;
-    using Type = std::conditional_t<std::is_floating_point_v<A> || std::is_floating_point_v<B>, Float64, Type0>;
+    using Type = std::conditional_t<is_floating_point<A> || is_floating_point<B>, Float64, Type0>;
 };
 
 template <typename A> struct ResultOfNegate
 {
     using Type = typename Construct<
         true,
-        std::is_floating_point_v<A>,
+        is_floating_point<A>,
         is_signed_v<A> ? sizeof(A) : nextSize(sizeof(A))>::Type;
 };
 
@@ -143,7 +143,7 @@ template <typename A> struct ResultOfAbs
 {
     using Type = typename Construct<
         false,
-        std::is_floating_point_v<A>,
+        is_floating_point<A>,
         sizeof(A)>::Type;
 };
 
@@ -154,7 +154,7 @@ template <typename A, typename B> struct ResultOfBit
     using Type = typename Construct<
         is_signed_v<A> || is_signed_v<B>,
         false,
-        std::is_floating_point_v<A> || std::is_floating_point_v<B> ? 8 : max(sizeof(A), sizeof(B))>::Type;
+        is_floating_point<A> || is_floating_point<B> ? 8 : max(sizeof(A), sizeof(B))>::Type;
 };
 
 template <typename A> struct ResultOfBitNot
@@ -180,7 +180,7 @@ template <typename A> struct ResultOfBitNot
 template <typename A, typename B>
 struct ResultOfIf
 {
-    static constexpr bool has_float = std::is_floating_point_v<A> || std::is_floating_point_v<B>;
+    static constexpr bool has_float = is_floating_point<A> || is_floating_point<B>;
     static constexpr bool has_integer = is_integer<A> || is_integer<B>;
     static constexpr bool has_signed = is_signed_v<A> || is_signed_v<B>;
     static constexpr bool has_unsigned = !is_signed_v<A> || !is_signed_v<B>;
@@ -189,7 +189,7 @@ struct ResultOfIf
     static constexpr size_t max_size_of_unsigned_integer = max(is_signed_v<A> ? 0 : sizeof(A), is_signed_v<B> ? 0 : sizeof(B));
     static constexpr size_t max_size_of_signed_integer = max(is_signed_v<A> ? sizeof(A) : 0, is_signed_v<B> ? sizeof(B) : 0);
     static constexpr size_t max_size_of_integer = max(is_integer<A> ? sizeof(A) : 0, is_integer<B> ? sizeof(B) : 0);
-    static constexpr size_t max_size_of_float = max(std::is_floating_point_v<A> ? sizeof(A) : 0, std::is_floating_point_v<B> ? sizeof(B) : 0);
+    static constexpr size_t max_size_of_float = max(is_floating_point<A> ? sizeof(A) : 0, is_floating_point<B> ? sizeof(B) : 0);
 
     using ConstructedType = typename Construct<has_signed, has_float,
         ((has_float && has_integer && max_size_of_integer >= max_size_of_float)
@@ -211,7 +211,7 @@ template <typename A> struct ToInteger
     using Type = typename Construct<
         is_signed_v<A>,
         false,
-        std::is_floating_point_v<A> ? 8 : sizeof(A)>::Type;
+        is_floating_point<A> ? 8 : sizeof(A)>::Type;
 };
 
 

src/DataTypes/Serializations/SerializationNumber.cpp

@@ -238,6 +238,7 @@ template class SerializationNumber<Int32>;
 template class SerializationNumber<Int64>;
 template class SerializationNumber<Int128>;
 template class SerializationNumber<Int256>;
+template class SerializationNumber<BFloat16>;
 template class SerializationNumber<Float32>;
 template class SerializationNumber<Float64>;
 

src/DataTypes/Utils.cpp

@@ -54,6 +54,13 @@ bool canBeSafelyCasted(const DataTypePtr & from_type, const DataTypePtr & to_typ
 
             return false;
         }
+        case TypeIndex::BFloat16:
+        {
+            if (to_which_type.isFloat32() || to_which_type.isFloat64() || to_which_type.isString())
+                return true;
+
+            return false;
+        }
         case TypeIndex::Float32:
         {
             if (to_which_type.isFloat64() || to_which_type.isString())

src/DataTypes/getLeastSupertype.cpp

@@ -109,6 +109,8 @@ DataTypePtr getNumericType(const TypeIndexSet & types)
             maximize(max_bits_of_signed_integer, 128);
         else if (type == TypeIndex::Int256)
             maximize(max_bits_of_signed_integer, 256);
+        else if (type == TypeIndex::BFloat16)
+            maximize(max_mantissa_bits_of_floating, 8);
         else if (type == TypeIndex::Float32)
             maximize(max_mantissa_bits_of_floating, 24);
         else if (type == TypeIndex::Float64)
@@ -145,7 +147,9 @@ DataTypePtr getNumericType(const TypeIndexSet & types)
         if (max_mantissa_bits_of_floating)
         {
             size_t min_mantissa_bits = std::max(min_bit_width_of_integer, max_mantissa_bits_of_floating);
-            if (min_mantissa_bits <= 24)
+            if (min_mantissa_bits <= 8)
+                return std::make_shared<DataTypeBFloat16>();
+            else if (min_mantissa_bits <= 24)
                 return std::make_shared<DataTypeFloat32>();
             if (min_mantissa_bits <= 53)
                 return std::make_shared<DataTypeFloat64>();

src/DataTypes/getMostSubtype.cpp

@@ -297,6 +297,8 @@ DataTypePtr getMostSubtype(const DataTypes & types, bool throw_if_result_is_noth
                 minimize(min_bits_of_signed_integer, 128);
             else if (typeid_cast<const DataTypeInt256 *>(type.get()))
                 minimize(min_bits_of_signed_integer, 256);
+            else if (typeid_cast<const DataTypeBFloat16 *>(type.get()))
+                minimize(min_mantissa_bits_of_floating, 8);
             else if (typeid_cast<const DataTypeFloat32 *>(type.get()))
                 minimize(min_mantissa_bits_of_floating, 24);
             else if (typeid_cast<const DataTypeFloat64 *>(type.get()))
@@ -313,7 +315,9 @@ DataTypePtr getMostSubtype(const DataTypes & types, bool throw_if_result_is_noth
             /// If the result must be floating.
             if (!min_bits_of_signed_integer && !min_bits_of_unsigned_integer)
             {
-                if (min_mantissa_bits_of_floating <= 24)
+                if (min_mantissa_bits_of_floating <= 8)
+                    return std::make_shared<DataTypeBFloat16>();
+                else if (min_mantissa_bits_of_floating <= 24)
                     return std::make_shared<DataTypeFloat32>();
                 if (min_mantissa_bits_of_floating <= 53)
                     return std::make_shared<DataTypeFloat64>();

src/Databases/enableAllExperimentalSettings.cpp

@@ -24,10 +24,11 @@ void enableAllExperimentalSettings(ContextMutablePtr context)
     context->setSetting("allow_experimental_dynamic_type", 1);
     context->setSetting("allow_experimental_json_type", 1);
     context->setSetting("allow_experimental_vector_similarity_index", 1);
-    context->setSetting("allow_experimental_bigint_types", 1);
     context->setSetting("allow_experimental_window_functions", 1);
     context->setSetting("allow_experimental_geo_types", 1);
     context->setSetting("allow_experimental_map_type", 1);
+    context->setSetting("allow_experimental_bigint_types", 1);
+    context->setSetting("allow_experimental_bfloat16_type", 1);
     context->setSetting("allow_deprecated_error_prone_window_functions", 1);
 
     context->setSetting("allow_suspicious_low_cardinality_types", 1);

src/Dictionaries/RangeHashedDictionary.h

@@ -298,7 +298,8 @@ namespace impl
             using Types = std::decay_t<decltype(types)>;
             using DataType = typename Types::LeftType;
 
-            if constexpr (IsDataTypeDecimalOrNumber<DataType> || IsDataTypeDateOrDateTime<DataType> || IsDataTypeEnum<DataType>)
+            if constexpr ((IsDataTypeDecimalOrNumber<DataType> || IsDataTypeDateOrDateTime<DataType> || IsDataTypeEnum<DataType>)
+                && !std::is_same_v<DataType, DataTypeBFloat16>)
             {
                 using ColumnType = typename DataType::ColumnType;
                 func(TypePair<ColumnType, void>());

src/Disks/tests/gtest_cascade_and_memory_write_buffer.cpp

@@ -69,7 +69,7 @@ static void testCascadeBufferRedability(
         auto rbuf = wbuf_readable.tryGetReadBuffer();
         ASSERT_FALSE(!rbuf);
 
-        concat.appendBuffer(wrapReadBufferPointer(std::move(rbuf)));
+        concat.appendBuffer(std::move(rbuf));
     }
 
     std::string decoded_data;

src/Formats/EscapingRuleUtils.cpp

@@ -10,7 +10,6 @@
 #include <IO/WriteHelpers.h>
 #include <IO/ReadBufferFromString.h>
 #include <IO/parseDateTimeBestEffort.h>
-#include <Parsers/TokenIterator.h>
 
 
 namespace DB

src/Formats/JSONExtractTree.cpp

@@ -131,7 +131,7 @@ bool tryGetNumericValueFromJSONElement(
     switch (element.type())
     {
         case ElementType::DOUBLE:
-            if constexpr (std::is_floating_point_v<NumberType>)
+            if constexpr (is_floating_point<NumberType>)
             {
                 /// We permit inaccurate conversion of double to float.
                 /// Example: double 0.1 from JSON is not representable in float.
@@ -175,7 +175,7 @@ bool tryGetNumericValueFromJSONElement(
                 return false;
 
             auto rb = ReadBufferFromMemory{element.getString()};
-            if constexpr (std::is_floating_point_v<NumberType>)
+            if constexpr (is_floating_point<NumberType>)
             {
                 if (!tryReadFloatText(value, rb) || !rb.eof())
                 {

src/Formats/ProtobufSerializer.cpp

@@ -540,7 +540,7 @@ namespace
 
                 case FieldTypeId::TYPE_ENUM:
                 {
-                    if (std::is_floating_point_v<NumberType>)
+                    if (is_floating_point<NumberType>)
                         incompatibleColumnType(TypeName<NumberType>);
 
                     write_function = [this](NumberType value)

src/Functions/DivisionUtils.h

@@ -47,9 +47,9 @@ inline auto checkedDivision(A a, B b)
 {
     throwIfDivisionLeadsToFPE(a, b);
 
-    if constexpr (is_big_int_v<A> && std::is_floating_point_v<B>)
+    if constexpr (is_big_int_v<A> && is_floating_point<B>)
         return static_cast<B>(a) / b;
-    else if constexpr (is_big_int_v<B> && std::is_floating_point_v<A>)
+    else if constexpr (is_big_int_v<B> && is_floating_point<A>)
         return a / static_cast<A>(b);
     else if constexpr (is_big_int_v<A> && is_big_int_v<B>)
         return static_cast<A>(a / b);
@@ -86,17 +86,17 @@ struct DivideIntegralImpl
         {
             /// Comparisons are not strict to avoid rounding issues when operand is implicitly casted to float.
 
-            if constexpr (std::is_floating_point_v<A>)
+            if constexpr (is_floating_point<A>)
                 if (isNaN(a) || a >= std::numeric_limits<CastA>::max() || a <= std::numeric_limits<CastA>::lowest())
                     throw Exception(ErrorCodes::ILLEGAL_DIVISION, "Cannot perform integer division on infinite or too large floating point numbers");
 
-            if constexpr (std::is_floating_point_v<B>)
+            if constexpr (is_floating_point<B>)
                 if (isNaN(b) || b >= std::numeric_limits<CastB>::max() || b <= std::numeric_limits<CastB>::lowest())
                     throw Exception(ErrorCodes::ILLEGAL_DIVISION, "Cannot perform integer division on infinite or too large floating point numbers");
 
             auto res = checkedDivision(CastA(a), CastB(b));
 
-            if constexpr (std::is_floating_point_v<decltype(res)>)
+            if constexpr (is_floating_point<decltype(res)>)
                 if (isNaN(res) || res >= static_cast<double>(std::numeric_limits<Result>::max()) || res <= std::numeric_limits<Result>::lowest())
                     throw Exception(ErrorCodes::ILLEGAL_DIVISION, "Cannot perform integer division, because it will produce infinite or too large number");
 
@@ -122,18 +122,18 @@ struct ModuloImpl
     template <typename Result = ResultType>
     static Result apply(A a, B b)
     {
-        if constexpr (std::is_floating_point_v<ResultType>)
+        if constexpr (is_floating_point<ResultType>)
         {
             /// This computation is similar to `fmod` but the latter is not inlined and has 40 times worse performance.
             return static_cast<ResultType>(a) - trunc(static_cast<ResultType>(a) / static_cast<ResultType>(b)) * static_cast<ResultType>(b);
         }
         else
         {
-            if constexpr (std::is_floating_point_v<A>)
+            if constexpr (is_floating_point<A>)
                 if (isNaN(a) || a > std::numeric_limits<IntegerAType>::max() || a < std::numeric_limits<IntegerAType>::lowest())
                     throw Exception(ErrorCodes::ILLEGAL_DIVISION, "Cannot perform integer division on infinite or too large floating point numbers");
 
-            if constexpr (std::is_floating_point_v<B>)
+            if constexpr (is_floating_point<B>)
                 if (isNaN(b) || b > std::numeric_limits<IntegerBType>::max() || b < std::numeric_limits<IntegerBType>::lowest())
                     throw Exception(ErrorCodes::ILLEGAL_DIVISION, "Cannot perform integer division on infinite or too large floating point numbers");
 

src/Functions/FunctionBinaryArithmetic.h

@@ -110,6 +110,7 @@ template <typename DataType> constexpr bool IsIntegralOrExtendedOrDecimal =
     IsDataTypeDecimal<DataType>;
 
 template <typename DataType> constexpr bool IsFloatingPoint = false;
+template <> inline constexpr bool IsFloatingPoint<DataTypeBFloat16> = true;
 template <> inline constexpr bool IsFloatingPoint<DataTypeFloat32> = true;
 template <> inline constexpr bool IsFloatingPoint<DataTypeFloat64> = true;
 
@@ -803,7 +804,7 @@ class FunctionBinaryArithmetic : public IFunction
             DataTypeFixedString, DataTypeString,
             DataTypeInterval>;
 
-        using Floats = TypeList<DataTypeFloat32, DataTypeFloat64>;
+        using Floats = TypeList<DataTypeFloat32, DataTypeFloat64, DataTypeBFloat16>;
 
         using ValidTypes = std::conditional_t<valid_on_float_arguments,
             TypeListConcat<Types, Floats>,
@@ -1690,6 +1691,13 @@ public:
             }
             else
             {
+                if constexpr ((std::is_same_v<LeftDataType, DataTypeBFloat16> || std::is_same_v<RightDataType, DataTypeBFloat16>)
+                    && (sizeof(typename LeftDataType::FieldType) > 8 || sizeof(typename RightDataType::FieldType) > 8))
+                {
+                    /// Big integers and BFloat16 are not supported together.
+                    return false;
+                }
+
                 using ResultDataType = typename BinaryOperationTraits<Op, LeftDataType, RightDataType>::ResultDataType;
 
                 if constexpr (!std::is_same_v<ResultDataType, InvalidType>)
@@ -2042,7 +2050,15 @@ ColumnPtr executeStringInteger(const ColumnsWithTypeAndName & arguments, const A
         using DecimalResultType = typename BinaryOperationTraits<Op, LeftDataType, RightDataType>::DecimalResultDataType;
 
         if constexpr (std::is_same_v<ResultDataType, InvalidType>)
+        {
             return nullptr;
+        }
+        else if constexpr ((std::is_same_v<LeftDataType, DataTypeBFloat16> || std::is_same_v<RightDataType, DataTypeBFloat16>)
+            && (sizeof(typename LeftDataType::FieldType) > 8 || sizeof(typename RightDataType::FieldType) > 8))
+        {
+            /// Big integers and BFloat16 are not supported together.
+            return nullptr;
+        }
         else // we can't avoid the else because otherwise the compiler may assume the ResultDataType may be Invalid
              // and that would produce the compile error.
         {
@@ -2059,7 +2075,7 @@ ColumnPtr executeStringInteger(const ColumnsWithTypeAndName & arguments, const A
             ColumnPtr left_col = nullptr;
             ColumnPtr right_col = nullptr;
 
-            /// When Decimal op Float32/64, convert both of them into Float64
+            /// When Decimal op Float32/64/16, convert both of them into Float64
             if constexpr (decimal_with_float)
             {
                 const auto converted_type = std::make_shared<DataTypeFloat64>();
@@ -2094,7 +2110,6 @@ ColumnPtr executeStringInteger(const ColumnsWithTypeAndName & arguments, const A
             /// Here we check if we have `intDiv` or `intDivOrZero` and at least one of the arguments is decimal, because in this case originally we had result as decimal, so we need to convert result into integer after calculations
             else if constexpr (!decimal_with_float && (is_int_div || is_int_div_or_zero) && (IsDataTypeDecimal<LeftDataType> || IsDataTypeDecimal<RightDataType>))
             {
-
                 if constexpr (!std::is_same_v<DecimalResultType, InvalidType>)
                 {
                     DataTypePtr type_res;

src/Functions/FunctionMathUnary.h

@@ -70,7 +70,7 @@ private:
             /// Process all data as a whole and use FastOps implementation
 
             /// If the argument is integer, convert to Float64 beforehand
-            if constexpr (!std::is_floating_point_v<T>)
+            if constexpr (!is_floating_point<T>)
             {
                 PODArray<Float64> tmp_vec(size);
                 for (size_t i = 0; i < size; ++i)
@@ -152,7 +152,7 @@ private:
         {
             using Types = std::decay_t<decltype(types)>;
             using Type = typename Types::RightType;
-            using ReturnType = std::conditional_t<Impl::always_returns_float64 || !std::is_floating_point_v<Type>, Float64, Type>;
+            using ReturnType = std::conditional_t<Impl::always_returns_float64 || !is_floating_point<Type>, Float64, Type>;
             using ColVecType = ColumnVectorOrDecimal<Type>;
 
             const auto col_vec = checkAndGetColumn<ColVecType>(col.column.get());