Merge branch 'master' into fix_weird_problem

2024-11-21 15:12:02 +00:00 · 2024-11-16 23:57:18 +01:00 · 2024-11-16 23:57:18 +01:00 · ade2dadd01
commit ade2dadd01
parent a82ab36c08 208fd6efe1
320 changed files with 5345 additions and 2298 deletions
--- a/base/base/BFloat16.h
+++ b/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;
+}
--- a/base/base/DecomposedFloat.h
+++ 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>;
--- a/base/base/EnumReflection.h
+++ b/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
 {
--- a/base/base/TypeLists.h
+++ b/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 TypeListNativeNumber = TypeListConcat<TypeListNativeInt, TypeListFloat>;
+using TypeListNativeFloat = TypeList<Float32, Float64>;
+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>;
--- a/base/base/TypeName.h
+++ b/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)
--- a/base/base/extended_types.h
+++ b/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)

--- a/ci/jobs/scripts/check_style/aspell-ignore/en/aspell-dict.txt
+++ b/ci/jobs/scripts/check_style/aspell-ignore/en/aspell-dict.txt
@ -3131,3 +3131,4 @@ DistributedCachePoolBehaviourOnLimit
 SharedJoin
 ShareSet
 unacked
+BFloat
--- a/cmake/cpu_features.cmake
+++ b/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
--- a/cmake/linux/default_libs.cmake
+++ b/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.
-# See https://bugs.llvm.org/show_bug.cgi?id=16404
+# We need builtins from Clang
 execute_process (COMMAND
    ${CMAKE_CXX_COMPILER} --target=${CMAKE_CXX_COMPILER_TARGET} --print-libgcc-file-name --rtlib=compiler-rt
    OUTPUT_VARIABLE BUILTINS_LIBRARY
--- a/docs/en/operations/server-configuration-parameters/settings.md
+++ b/docs/en/operations/server-configuration-parameters/settings.md
@ -597,6 +597,30 @@ If number of tables is greater than this value, server will throw an exception.
 <max_table_num_to_throw>400</max_table_num_to_throw>
 ```

+## max\_replicated\_table\_num\_to\_throw {#max-replicated-table-num-to-throw}
+If number of replicated tables is greater than this value, server will throw an exception. 0 means no limitation. Only count table in Atomic/Ordinary/Replicated/Lazy database engine.
+
+**Example**
+```xml
+<max_replicated_table_num_to_throw>400</max_replicated_table_num_to_throw>
+```
+
+## max\_dictionary\_num\_to\_throw {#max-dictionary-num-to-throw}
+If number of dictionaries is greater than this value, server will throw an exception. 0 means no limitation. Only count table in Atomic/Ordinary/Replicated/Lazy database engine.
+
+**Example**
+```xml
+<max_dictionary_num_to_throw>400</max_dictionary_num_to_throw>
+```
+
+## max\_view\_num\_to\_throw {#max-view-num-to-throw}
+If number of views is greater than this value, server will throw an exception. 0 means no limitation. Only count table in Atomic/Ordinary/Replicated/Lazy database engine.
+
+**Example**
+```xml
+<max_view_num_to_throw>400</max_view_num_to_throw>
+```
+
 ## max\_database\_num\_to\_throw {#max-table-num-to-throw}
 If number of _database is greater than this value, server will throw an exception. 0 means no limitation.
 Default value: 0
--- a/docs/en/sql-reference/data-types/float.md
+++ b/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.
--- a/docs/en/sql-reference/functions/date-time-functions.md
+++ b/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                                |
-| z           | time zone (short name not supported yet) | text          | Pacific Standard Time; PST         |
-| Z           | time zone offset/id (not supported yet)  | zone          | -0800; -08:00; America/Los_Angeles |
+| S           | fraction of second                       | number        | 978                                |
+| z           | time zone                                | text          | Eastern Standard Time; EST         |
+| Z           | time zone offset                         | zone          | -0800; -0812                       |
 | '           | escape for text                          | delimiter     |                                    |
 | ''          | single quote                             | literal       | '                                  |

--- a/docs/en/sql-reference/functions/type-conversion-functions.md
+++ b/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

--- a/docs/en/sql-reference/statements/explain.md
+++ b/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",
-              "Plans": [
-                {
-                  "Node Type": "ReadFromStorage"
-                }
-              ]
+              "Node Type": "ReadFromStorage",
+              "Node Id": "ReadFromStorage_0"
            }
          ]
        },
        {
          "Node Type": "Expression",
+          "Node Id": "Expression_16",
          "Plans": [
            {
-              "Node Type": "SettingQuotaAndLimits",
-              "Plans": [
-                {
-                  "Node Type": "ReadFromStorage"
-                }
-              ]
+              "Node Type": "ReadFromStorage",
+              "Node Id": "ReadFromStorage_4"
            }
          ]
        }
@ -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],
-        "Project Input": true
+        "Positions": [1]
      },
      "Plans": [
        {
-          "Node Type": "SettingQuotaAndLimits",
-          "Plans": [
-            {
-              "Node Type": "ReadFromStorage"
-            }
-          ]
+          "Node Type": "ReadFromStorage",
+          "Node Id": "ReadFromStorage_0"
        }
      ]
    }
@ -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
--- a/docs/ru/operations/settings/settings.md
+++ b/docs/ru/operations/settings/settings.md
@ -136,7 +136,7 @@ ClickHouse применяет настройку в тех случаях, ко
 -   0 — выключена.
 -   1 — включена.

-Значение по умолчанию: 0.
+Значение по умолчанию: 1.

 ## http_zlib_compression_level {#settings-http_zlib_compression_level}

--- a/docs/zh/operations/settings/settings.md
+++ b/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}

--- a/programs/benchmark/Benchmark.cpp
+++ b/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()) << "%)";
-        std::cerr << ".\n";
+            log << " (" << (num * 100.0 / queries.size()) << "%)";
+        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();
-            if (delay > 0 && seconds > delay)
-            {
-                printNumberOfQueriesExecuted(queries_executed);
-                cumulative
-                    ? report(comparison_info_total, total_watch.elapsedSeconds())
-                    : report(comparison_info_per_interval, seconds);
-                delay_watch.restart();
-            }
+        double seconds = delay_watch.elapsedSeconds();
+        if (delay > 0 && seconds > delay)
+        {
+            printNumberOfQueriesExecuted(queries_executed);
+            cumulative
+                ? report(comparison_info_total, total_watch.elapsedSeconds())
+                : report(comparison_info_per_interval, seconds);
+            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 << "'"
-                          << ": " << getCurrentExceptionMessage(false) << std::endl;
+                log << "An error occurred while processing the query " << "'" << query << "'"
+                          << ": " << getCurrentExceptionMessage(false) << '\n';
                if (!(continue_on_errors || max_consecutive_errors > ++consecutive_errors))
                {
                    shutdown = true;
                    throw;
                }

-                std::cerr << getCurrentExceptionMessage(print_stacktrace,
-                    true /*check embedded stack trace*/) << std::endl;
+                log << getCurrentExceptionMessage(print_stacktrace,
+                    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
-                    << connection_description << ", "
-                    << "queries: " << info->queries << ", ";
+            log
+                << connection_description << ", "
+                << "queries: " << info->queries.load() << ", ";
            if (info->errors)
            {
-                std::cerr << "errors: " << info->errors << ", ";
+                log << "errors: " << info->errors << ", ";
            }
-            std::cerr
-                    << "QPS: " << (info->queries / seconds) << ", "
-                    << "RPS: " << (info->read_rows / seconds) << ", "
-                    << "MiB/s: " << (info->read_bytes / seconds / 1048576) << ", "
-                    << "result RPS: " << (info->result_rows / seconds) << ", "
-                    << "result MiB/s: " << (info->result_bytes / seconds / 1048576) << "."
-                    << "\n";
+            log
+                << "QPS: " << fmt::format("{:.3f}", info->queries / seconds) << ", "
+                << "RPS: " << fmt::format("{:.3f}", info->read_rows / seconds) << ", "
+                << "MiB/s: " << fmt::format("{:.3f}", info->read_bytes / seconds / 1048576) << ", "
+                << "result RPS: " << fmt::format("{:.3f}", info->result_rows / seconds) << ", "
+                << "result MiB/s: " << fmt::format("{:.3f}", info->result_bytes / seconds / 1048576) << "."
+                << "\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();
    }
 }
--- a/src/AggregateFunctions/AggregateFunctionAvg.h
+++ b/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;
    }

--- a/src/AggregateFunctions/AggregateFunctionDeltaSum.cpp
+++ b/src/AggregateFunctions/AggregateFunctionDeltaSum.cpp
@ -27,9 +27,9 @@ namespace
 template <typename T>
 struct AggregationFunctionDeltaSumData
 {
-    T sum = 0;
-    T last = 0;
-    T first = 0;
+    T sum{};
+    T last{};
+    T first{};
    bool seen = false;
 };

--- a/src/AggregateFunctions/AggregateFunctionDeltaSumTimestamp.cpp
+++ b/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 first = 0;
-    ValueType last = 0;
-    TimestampType first_ts = 0;
-    TimestampType last_ts = 0;
+    ValueType sum{};
+    ValueType first{};
+    ValueType last{};
+    TimestampType first_ts{};
+    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>
-        >{arguments, params, createResultType()}
-    {}
+            AggregationFunctionDeltaSumTimestamp<ValueType, TimestampType>>{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 ts = assert_cast<const ColumnVector<TimestampType> &>(*columns[1]).getData()[row_num];
+        auto value = unalignedLoad<ValueType>(columns[0]->getRawData().data() + row_num * sizeof(ValueType));
+        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;
 }
 }

--- a/src/AggregateFunctions/AggregateFunctionGroupArray.cpp
+++ b/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>;
--- a/src/AggregateFunctions/AggregateFunctionGroupArrayMoving.cpp
+++ b/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;

--- a/src/AggregateFunctions/AggregateFunctionIntervalLengthSum.cpp
+++ b/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)));
--- a/src/AggregateFunctions/AggregateFunctionMaxIntersections.cpp
+++ b/src/AggregateFunctions/AggregateFunctionMaxIntersections.cpp
@ -155,9 +155,9 @@ public:

    void insertResultInto(AggregateDataPtr __restrict place, IColumn & to, Arena *) const override
    {
-        Int64 current_intersections = 0;
-        Int64 max_intersections = 0;
-        PointType position_of_max_intersections = 0;
+        Int64 current_intersections{};
+        Int64 max_intersections{};
+        PointType position_of_max_intersections{};

        /// const_cast because we will sort the array
        auto & array = this->data(place).value;
--- a/src/AggregateFunctions/AggregateFunctionSparkbar.cpp
+++ b/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);
-            if constexpr (std::is_floating_point_v<Y>)
+            constexpr auto levels_num = Y{BAR_LEVELS - 1};
+            if constexpr (is_floating_point<Y>)
            {
                y = y / (y_max / levels_num) + 1;
            }
--- a/src/AggregateFunctions/AggregateFunctionSum.h
+++ b/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 equivalent_integer = typename std::conditional_t<sizeof(Value) == 4, UInt32, UInt64>;
+            using EquivalentInteger = 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;
-                    std::memcpy(&value, &ptr[i], sizeof(Value));
+                    EquivalentInteger 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]));
-        else
-            this->data(place).add(column.getData()[row_num]);
+        this->data(place).add(static_cast<TResult>(column.getData()[row_num]));
    }

    void addBatchSinglePlace(
--- a/src/AggregateFunctions/AggregateFunctionUniq.h
+++ b/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);
        }
--- a/src/AggregateFunctions/AggregateFunctionUniqCombined.h
+++ b/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)));
            }
--- a/src/AggregateFunctions/Helpers.h
+++ b/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)
 {
--- a/src/AggregateFunctions/QuantileTDigest.h
+++ b/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();

--- a/src/AggregateFunctions/ReservoirSampler.h
+++ b/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();
    }
 };
--- a/src/AggregateFunctions/ReservoirSamplerDeterministic.h
+++ b/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();
    }
 };

--- a/src/Backups/BackupCoordinationStageSync.cpp
+++ b/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_);
-        createStartAndAliveNodes();
+        createStartAndAliveNodesAndCheckConcurrency(concurrency_counters_);
        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);
--- a/src/Backups/BackupCoordinationStageSync.h
+++ b/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 createStartAndAliveNodes(Coordination::ZooKeeperWithFaultInjection::Ptr zookeeper);
+    void createStartAndAliveNodesAndCheckConcurrency(BackupConcurrencyCounters & concurrency_counters_);
+    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;

--- a/src/Client/ClientBase.cpp
+++ b/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)
-    {
-        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();
-            }
-        }
-    });
+    startKeystrokeInterceptorIfExists();
+    SCOPE_EXIT({ stopKeystrokeInterceptorIfExists(); });

    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"))
--- a/src/Client/ClientBase.h
+++ b/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.
--- a/src/Client/ClientBaseHelpers.cpp
+++ b/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;
--- a/src/Client/ProgressTable.cpp
+++ b/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();
--- a/src/Client/ProgressTable.h
+++ b/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 clearTableOutput(WriteBufferFromFileDescriptor & message);
+    void writeTable(WriteBufferFromFileDescriptor & message, std::unique_lock<std::mutex> & message_lock,
+            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:
--- a/src/Columns/ColumnArray.cpp
+++ b/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()))
--- a/src/Columns/ColumnUnique.cpp
+++ b/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>;
--- a/src/Columns/ColumnUnique.h
+++ b/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>;
--- a/src/Columns/ColumnVector.cpp
+++ b/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>;
--- a/src/Columns/ColumnVector.h
+++ b/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>;
--- a/src/Columns/ColumnsCommon.cpp
+++ b/src/Columns/ColumnsCommon.cpp
@ -328,6 +328,7 @@ INSTANTIATE(Int32)
 INSTANTIATE(Int64)
 INSTANTIATE(Int128)
 INSTANTIATE(Int256)
+INSTANTIATE(BFloat16)
 INSTANTIATE(Float32)
 INSTANTIATE(Float64)
 INSTANTIATE(Decimal32)
--- a/src/Columns/ColumnsNumber.h
+++ b/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>;

--- a/src/Columns/IColumn.cpp
+++ b/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>;
--- a/src/Columns/MaskOperations.cpp
+++ b/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());
--- a/src/Columns/tests/gtest_column_vector.cpp
+++ b/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>();
--- a/src/Columns/tests/gtest_low_cardinality.cpp
+++ b/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>());
 }
--- a/src/Common/CPUID.h
+++ b/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)                  \
--- a/src/Common/CurrentMetrics.cpp
+++ b/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") \
@ -255,6 +256,7 @@
    M(PartsActive, "Active data part, used by current and upcoming SELECTs.") \
    M(AttachedDatabase, "Active databases.") \
    M(AttachedTable, "Active tables.") \
+    M(AttachedReplicatedTable, "Active replicated tables.") \
    M(AttachedView, "Active views.") \
    M(AttachedDictionary, "Active dictionaries.") \
    M(PartsOutdated, "Not active data part, but could be used by only current SELECTs, could be deleted after SELECTs finishes.") \
--- a/src/Common/FailPoint.cpp
+++ b/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:
--- a/src/Common/FailPoint.h
+++ b/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:
--- a/src/Common/FieldVisitorConvertToNumber.cpp
+++ b/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>;

--- a/src/Common/FieldVisitorConvertToNumber.h
+++ b/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>;

--- a/src/Common/HashTable/Hash.h
+++ b/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)
--- a/src/Common/HashTable/HashTable.h
+++ b/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
--- a/src/Common/HostResolvePool.cpp
+++ b/src/Common/HostResolvePool.cpp
@ -9,6 +9,7 @@

 #include <mutex>
 #include <algorithm>
+#include <Poco/Timespan.h>


 namespace ProfileEvents
@ -49,16 +50,18 @@ HostResolver::WeakPtr HostResolver::getWeakFromThis()
 }

 HostResolver::HostResolver(String host_, Poco::Timespan history_)
-    : host(std::move(host_))
-    , history(history_)
-    , resolve_function([](const String & host_to_resolve) { return DNSResolver::instance().resolveHostAllInOriginOrder(host_to_resolve); })
-{
-    update();
-}
+    : HostResolver(
+        [](const String & host_to_resolve) { return DNSResolver::instance().resolveHostAllInOriginOrder(host_to_resolve); },
+        host_,
+        history_)
+{}

 HostResolver::HostResolver(
    ResolveFunction && resolve_function_, String host_, Poco::Timespan history_)
-    : host(std::move(host_)), history(history_), resolve_function(std::move(resolve_function_))
+    : host(std::move(host_))
+    , history(history_)
+    , resolve_interval(history_.totalMicroseconds() / 3)
+    , resolve_function(std::move(resolve_function_))
 {
    update();
 }
@ -203,7 +206,7 @@ bool HostResolver::isUpdateNeeded()
    Poco::Timestamp now;

    std::lock_guard lock(mutex);
-    return last_resolve_time + history < now || records.empty();
+    return last_resolve_time + resolve_interval < now || records.empty();
 }

 void HostResolver::updateImpl(Poco::Timestamp now, std::vector<Poco::Net::IPAddress> & next_gen)
--- a/src/Common/HostResolvePool.h
+++ b/src/Common/HostResolvePool.h
@ -26,7 +26,7 @@
 //    a) it still occurs in resolve set after `history_` time or b) all other addresses are pessimized as well.
 // - resolve schedule
 //    Addresses are resolved through `DB::DNSResolver::instance()`.
-//    Usually it does not happen more often than once in `history_` time.
+//    Usually it does not happen more often than 3 times in `history_` period.
 //    But also new resolve performed each `setFail()` call.

 namespace DB
@ -212,6 +212,7 @@ protected:

    const String host;
    const Poco::Timespan history;
+    const Poco::Timespan resolve_interval;
    const HostResolverMetrics metrics = getMetrics();

    // for tests purpose
@ -245,4 +246,3 @@ private:
 };

 }
-
--- a/src/Common/NaNUtils.h
+++ b/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>)
-        return std::isnan(x);
+    if constexpr (is_floating_point<T>)
+        return DecomposedFloat(x).isNaN();
    else
        return false;
 }

-
 template <typename T>
 inline bool isFinite(T x)
 {
-    if constexpr (std::is_floating_point_v<T>)
-        return std::isfinite(x);
+    if constexpr (is_floating_point<T>)
+        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;
+}
--- a/src/Common/ProgressIndication.cpp
+++ b/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);

--- a/src/Common/ProgressIndication.h
+++ b/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 clearProgressOutput(WriteBufferFromFileDescriptor & message);
+    void writeProgress(WriteBufferFromFileDescriptor & message, std::unique_lock<std::mutex> & message_lock);
+    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; }
--- a/src/Common/TargetSpecific.cpp
+++ b/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";
--- a/src/Common/TargetSpecific.h
+++ b/src/Common/TargetSpecific.h
@ -83,9 +83,10 @@ enum class TargetArch : UInt32
    AVX512BW    = (1 << 4),
    AVX512VBMI  = (1 << 5),
    AVX512VBMI2 = (1 << 6),
-    AMXBF16 = (1 << 7),
-    AMXTILE = (1 << 8),
-    AMXINT8 = (1 << 9),
+    AVX512BF16 = (1 << 7),
+    AMXBF16 = (1 << 8),
+    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:
--- a/src/Common/ThreadStatus.cpp
+++ b/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();
--- a/src/Common/ThreadStatus.h
+++ b/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();
--- a/src/Common/findExtreme.cpp
+++ b/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);
--- a/src/Common/transformEndianness.h
+++ b/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)
--- a/src/Compression/CompressionCodecNone.h
+++ b/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
 {
--- a/src/Compression/tests/gtest_compressionCodec.cpp
+++ b/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>

--- a/src/Core/AccurateComparison.h
+++ b/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<B> || std::is_floating_point_v<B>);
+    static_assert(is_integer<A> || is_floating_point<A>);
+    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))
--- a/src/Core/DecimalFunctions.h
+++ b/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));
    }
--- a/src/Core/Field.h
+++ b/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; };
--- a/src/Core/ServerSettings.cpp
+++ b/src/Core/ServerSettings.cpp
@ -131,6 +131,9 @@ namespace DB
    DECLARE(UInt64, max_database_num_to_warn, 1000lu, "If the number of databases is greater than this value, the server will create a warning that will displayed to user.", 0) \
    DECLARE(UInt64, max_part_num_to_warn, 100000lu, "If the number of parts is greater than this value, the server will create a warning that will displayed to user.", 0) \
    DECLARE(UInt64, max_table_num_to_throw, 0lu, "If number of tables is greater than this value, server will throw an exception. 0 means no limitation. View, remote tables, dictionary, system tables are not counted. Only count table in Atomic/Ordinary/Replicated/Lazy database engine.", 0) \
+    DECLARE(UInt64, max_replicated_table_num_to_throw, 0lu, "If number of replicated tables is greater than this value, server will throw an exception. 0 means no limitation. Only count table in Atomic/Ordinary/Replicated/Lazy database engine.", 0) \
+    DECLARE(UInt64, max_dictionary_num_to_throw, 0lu, "If number of dictionaries is greater than this value, server will throw an exception. 0 means no limitation. Only count table in Atomic/Ordinary/Replicated/Lazy database engine.", 0) \
+    DECLARE(UInt64, max_view_num_to_throw, 0lu, "If number of views is greater than this value, server will throw an exception. 0 means no limitation. Only count table in Atomic/Ordinary/Replicated/Lazy database engine.", 0) \
    DECLARE(UInt64, max_database_num_to_throw, 0lu, "If number of databases is greater than this value, server will throw an exception. 0 means no limitation.", 0) \
    DECLARE(UInt64, max_authentication_methods_per_user, 100, "The maximum number of authentication methods a user can be created with or altered. Changing this setting does not affect existing users. Zero means unlimited", 0) \
    DECLARE(UInt64, concurrent_threads_soft_limit_num, 0, "Sets how many concurrent thread can be allocated before applying CPU pressure. Zero means unlimited.", 0) \
--- a/src/Core/Settings.cpp
+++ b/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).

@ -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.
--- a/src/Core/SettingsChangesHistory.cpp
+++ b/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"},
@ -79,6 +80,7 @@ static std::initializer_list<std::pair<ClickHouseVersion, SettingsChangesHistory
            {"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."},
@ -127,7 +129,7 @@ static std::initializer_list<std::pair<ClickHouseVersion, SettingsChangesHistory
            {"allow_experimental_refreshable_materialized_view", false, true, "Not experimental anymore"},
            {"max_parts_to_move", 0, 1000, "New setting"},
            {"hnsw_candidate_list_size_for_search", 64, 256, "New setting. Previously, the value was optionally specified in CREATE INDEX and 64 by default."},
-            {"allow_reorder_prewhere_conditions", false, true, "New setting"},
+            {"allow_reorder_prewhere_conditions", true, true, "New setting"},
            {"input_format_parquet_bloom_filter_push_down", false, true, "When reading Parquet files, skip whole row groups based on the WHERE/PREWHERE expressions and bloom filter in the Parquet metadata."},
            {"date_time_64_output_format_cut_trailing_zeros_align_to_groups_of_thousands", false, false, "Dynamically trim the trailing zeros of datetime64 values to adjust the output scale to (0, 3, 6), corresponding to 'seconds', 'milliseconds', and 'microseconds'."},
        }
--- a/src/Core/SortCursor.h
+++ b/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>,

--- a/src/Core/TypeId.h
+++ b/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)
--- a/src/Core/Types_fwd.h
+++ b/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
 {
--- a/src/Core/callOnTypeIndex.h
+++ b/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)...);

--- a/src/DataTypes/DataTypeNumberBase.cpp
+++ b/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>;

--- a/src/DataTypes/DataTypeNumberBase.h
+++ b/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>;

--- a/src/DataTypes/DataTypesBinaryEncoding.cpp
+++ b/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:
--- a/src/DataTypes/DataTypesDecimal.cpp
+++ b/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);
--- a/src/DataTypes/DataTypesDecimal.h
+++ b/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);
--- a/src/DataTypes/DataTypesNumber.cpp
+++ b/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>;

--- a/src/DataTypes/DataTypesNumber.h
+++ b/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>;

--- a/src/DataTypes/IDataType.h
+++ b/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)
 }
--- a/src/DataTypes/Native.cpp
+++ b/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();
 }

--- a/src/DataTypes/NumberTraits.h
+++ b/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;
 };


--- a/src/DataTypes/Serializations/SerializationNumber.cpp
+++ b/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>;

--- a/src/DataTypes/Utils.cpp
+++ b/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())
--- a/src/DataTypes/getLeastSupertype.cpp
+++ b/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>();
--- a/src/DataTypes/getMostSubtype.cpp
+++ b/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>();
--- a/src/Databases/DatabasesCommon.cpp
+++ b/src/Databases/DatabasesCommon.cpp
@ -382,7 +382,8 @@ StoragePtr DatabaseWithOwnTablesBase::detachTableUnlocked(const String & table_n
    if (!table_storage->isSystemStorage() && !DatabaseCatalog::isPredefinedDatabase(database_name))
    {
        LOG_TEST(log, "Counting detached table {} to database {}", table_name, database_name);
-        CurrentMetrics::sub(getAttachedCounterForStorage(table_storage));
+        for (auto metric : getAttachedCountersForStorage(table_storage))
+            CurrentMetrics::sub(metric);
    }

    auto table_id = table_storage->getStorageID();
@ -430,7 +431,8 @@ void DatabaseWithOwnTablesBase::attachTableUnlocked(const String & table_name, c
    if (!table->isSystemStorage() && !DatabaseCatalog::isPredefinedDatabase(database_name))
    {
        LOG_TEST(log, "Counting attached table {} to database {}", table_name, database_name);
-        CurrentMetrics::add(getAttachedCounterForStorage(table));
+        for (auto metric : getAttachedCountersForStorage(table))
+            CurrentMetrics::add(metric);
    }
 }

--- a/src/Databases/enableAllExperimentalSettings.cpp
+++ b/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);
--- a/src/Dictionaries/RangeHashedDictionary.h
+++ b/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>());
--- a/src/Disks/tests/gtest_cascade_and_memory_write_buffer.cpp
+++ b/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;
--- a/src/Formats/EscapingRuleUtils.cpp
+++ b/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
--- a/Show More
+++ b/Show More