Merge branch 'master' into CurtizJ-patch-7

2024-11-27 10:02:01 +00:00 · 2023-04-11 15:06:07 +02:00 · 2023-04-11 15:06:07 +02:00 · 5a77a2aa4c
commit 5a77a2aa4c
parent e88938d6c8 7210b50551
370 changed files with 2794 additions and 1261 deletions
--- a/README.md
+++ b/README.md
@ -1,4 +1,4 @@
-[![ClickHouse — open source distributed column-oriented DBMS](https://github.com/ClickHouse/clickhouse-presentations/raw/master/images/logo-400x240.png)](https://clickhouse.com)
+[<img alt="ClickHouse — open source distributed column-oriented DBMS" width="400px" src="https://clickhouse.com/images/ch_gh_logo_rounded.png" />](https://clickhouse.com?utm_source=github)

 ClickHouse® is an open-source column-oriented database management system that allows generating analytical data reports in real-time.

--- a/base/base/wide_integer_impl.h
+++ b/base/base/wide_integer_impl.h
@ -155,13 +155,13 @@ struct common_type<wide::integer<Bits, Signed>, Arithmetic>
        std::is_floating_point_v<Arithmetic>,
        Arithmetic,
        std::conditional_t<
-            sizeof(Arithmetic) < Bits * sizeof(long),
+            sizeof(Arithmetic) * 8 < Bits,
            wide::integer<Bits, Signed>,
            std::conditional_t<
-                Bits * sizeof(long) < sizeof(Arithmetic),
+                Bits < sizeof(Arithmetic) * 8,
                Arithmetic,
                std::conditional_t<
-                    Bits * sizeof(long) == sizeof(Arithmetic) && (std::is_same_v<Signed, signed> || std::is_signed_v<Arithmetic>),
+                    Bits == sizeof(Arithmetic) * 8 && (std::is_same_v<Signed, signed> || std::is_signed_v<Arithmetic>),
                    Arithmetic,
                    wide::integer<Bits, Signed>>>>>;
 };
--- a/contrib/cctz
+++ b/contrib/cctz
@ -1 +1 @@
-Subproject commit 7c78edd52b4d65acc103c2f195818ffcabe6fe0d
+Subproject commit 5e05432420f9692418e2e12aff09859e420b14a2
--- a/contrib/krb5
+++ b/contrib/krb5
@ -1 +1 @@
-Subproject commit 9453aec0d50e5aff9b189051611b321b40935d02
+Subproject commit b56ce6ba690e1f320df1a64afa34980c3e462617
--- a/contrib/krb5-cmake/CMakeLists.txt
+++ b/contrib/krb5-cmake/CMakeLists.txt
@ -15,10 +15,6 @@ if(NOT AWK_PROGRAM)
    message(FATAL_ERROR "You need the awk program to build ClickHouse with krb5 enabled.")
 endif()

-if (NOT (ENABLE_OPENSSL OR ENABLE_OPENSSL_DYNAMIC))
-    add_compile_definitions(USE_BORINGSSL=1)
-endif ()
-
 set(KRB5_SOURCE_DIR "${ClickHouse_SOURCE_DIR}/contrib/krb5/src")
 set(KRB5_ET_BIN_DIR "${CMAKE_CURRENT_BINARY_DIR}/include_private")

@ -162,6 +158,11 @@ set(ALL_SRCS

    "${KRB5_SOURCE_DIR}/lib/crypto/builtin/kdf.c"
    "${KRB5_SOURCE_DIR}/lib/crypto/builtin/cmac.c"
+    "${KRB5_SOURCE_DIR}/lib/crypto/builtin/des/des_keys.c"
+    "${KRB5_SOURCE_DIR}/lib/crypto/builtin/des/f_parity.c"
+    "${KRB5_SOURCE_DIR}/lib/crypto/builtin/enc_provider/rc4.c"
+    "${KRB5_SOURCE_DIR}/lib/crypto/builtin/hash_provider/hash_md4.c"
+    "${KRB5_SOURCE_DIR}/lib/crypto/builtin/md4/md4.c"
    "${KRB5_SOURCE_DIR}/lib/crypto/krb/prng.c"
    "${KRB5_SOURCE_DIR}/lib/crypto/krb/enc_dk_cmac.c"
    # "${KRB5_SOURCE_DIR}/lib/crypto/krb/crc32.c"
@ -226,7 +227,6 @@ set(ALL_SRCS
    # "${KRB5_SOURCE_DIR}/lib/crypto/openssl/enc_provider/des.c"
    "${KRB5_SOURCE_DIR}/lib/crypto/openssl/enc_provider/rc4.c"
    "${KRB5_SOURCE_DIR}/lib/crypto/openssl/enc_provider/des3.c"
-    #"${KRB5_SOURCE_DIR}/lib/crypto/openssl/enc_provider/camellia.c"
    "${KRB5_SOURCE_DIR}/lib/crypto/openssl/cmac.c"
    "${KRB5_SOURCE_DIR}/lib/crypto/openssl/sha256.c"
    "${KRB5_SOURCE_DIR}/lib/crypto/openssl/hmac.c"
@ -474,6 +474,14 @@ set(ALL_SRCS
    "${KRB5_SOURCE_DIR}/lib/krb5/krb5_libinit.c"
 )

+if (NOT (ENABLE_OPENSSL OR ENABLE_OPENSSL_DYNAMIC))
+    add_compile_definitions(USE_BORINGSSL=1)
+else()
+    set(ALL_SRCS ${ALL_SRCS}
+        "${KRB5_SOURCE_DIR}/lib/crypto/openssl/enc_provider/camellia.c"
+    )
+endif()
+
 add_custom_command(
    OUTPUT "${CMAKE_CURRENT_BINARY_DIR}/compile_et"
    COMMAND /bin/sh
@ -673,6 +681,7 @@ target_include_directories(_krb5 PRIVATE
    "${KRB5_SOURCE_DIR}/lib/gssapi/krb5"
    "${KRB5_SOURCE_DIR}/lib/gssapi/spnego"
    "${KRB5_SOURCE_DIR}/util/et"
+    "${KRB5_SOURCE_DIR}/lib/crypto/builtin/md4"
    "${KRB5_SOURCE_DIR}/lib/crypto/openssl"
    "${KRB5_SOURCE_DIR}/lib/crypto/krb"
    "${KRB5_SOURCE_DIR}/util/profile"
--- a/docs/en/interfaces/formats.md
+++ b/docs/en/interfaces/formats.md
@ -78,7 +78,7 @@ The supported formats are:
 | [Null](#null)                                                                             | ✗    | ✔      |
 | [XML](#xml)                                                                               | ✗    | ✔      |
 | [CapnProto](#capnproto)                                                                   | ✔    | ✔      |
-| [LineAsString](#lineasstring)                                                             | ✔    | ✗      |
+| [LineAsString](#lineasstring)                                                             | ✔    | ✔      |
 | [Regexp](#data-format-regexp)                                                             | ✔    | ✗      |
 | [RawBLOB](#rawblob)                                                                       | ✔    | ✔      |
 | [MsgPack](#msgpack)                                                                       | ✔    | ✔      |
@ -1877,6 +1877,13 @@ Column names must:

 Output Avro file compression and sync interval can be configured with [output_format_avro_codec](/docs/en/operations/settings/settings-formats.md/#output_format_avro_codec) and [output_format_avro_sync_interval](/docs/en/operations/settings/settings-formats.md/#output_format_avro_sync_interval) respectively.

+### Example Data {#example-data-avro}
+
+Using the ClickHouse [DESCRIBE](/docs/en/sql-reference/statements/describe-table) function, you can quickly view the inferred format of an Avro file like the following example. This example includes the URL of a publicly accessible Avro file in the ClickHouse S3 public bucket:
+
+``` DESCRIBE url('https://clickhouse-public-datasets.s3.eu-central-1.amazonaws.com/hits.avro','Avro');
+```
+
 ## AvroConfluent {#data-format-avro-confluent}

 AvroConfluent supports decoding single-object Avro messages commonly used with [Kafka](https://kafka.apache.org/) and [Confluent Schema Registry](https://docs.confluent.io/current/schema-registry/index.html).
--- a/docs/en/operations/system-tables/replicas.md
+++ b/docs/en/operations/system-tables/replicas.md
@ -50,6 +50,7 @@ last_queue_update:           2021-10-12 14:50:08
 absolute_delay:              99
 total_replicas:              5
 active_replicas:             5
+lost_part_count:             0
 last_queue_update_exception:
 zookeeper_exception:
 replica_is_active:           {'r1':1,'r2':1}
@ -90,6 +91,7 @@ The next 4 columns have a non-zero value only where there is an active session w
 -   `absolute_delay` (`UInt64`) - How big lag in seconds the current replica has.
 -   `total_replicas` (`UInt8`) - The total number of known replicas of this table.
 -   `active_replicas` (`UInt8`) - The number of replicas of this table that have a session in ClickHouse Keeper (i.e., the number of functioning replicas).
+-   `lost_part_count` (`UInt64`) - The number of data parts lost in the table by all replicas in total since table creation. Value is persisted in ClickHouse Keeper and can only increase.
 -   `last_queue_update_exception` (`String`) - When the queue contains broken entries. Especially important when ClickHouse breaks backward compatibility between versions and log entries written by newer versions aren't parseable by old versions.
 -   `zookeeper_exception` (`String`) - The last exception message, got if the error happened when fetching the info from ClickHouse Keeper.
 -   `replica_is_active` ([Map(String, UInt8)](../../sql-reference/data-types/map.md)) — Map between replica name and is replica active.
--- a/docs/en/operations/system-tables/storage_policies.md
+++ b/docs/en/operations/system-tables/storage_policies.md
@ -11,8 +11,16 @@ Columns:
 -   `volume_name` ([String](../../sql-reference/data-types/string.md)) — Volume name defined in the storage policy.
 -   `volume_priority` ([UInt64](../../sql-reference/data-types/int-uint.md)) — Volume order number in the configuration, the data fills the volumes according this priority, i.e. data during inserts and merges is written to volumes with a lower priority (taking into account other rules: TTL, `max_data_part_size`, `move_factor`).
 -   `disks` ([Array(String)](../../sql-reference/data-types/array.md)) — Disk names, defined in the storage policy.
+-   `volume_type` ([Enum8](../../sql-reference/data-types/enum.md))  — Type of volume. Can have one of the following values:
+    -   `JBOD` 
+    -   `SINGLE_DISK`
+    -   `UNKNOWN`
 -   `max_data_part_size` ([UInt64](../../sql-reference/data-types/int-uint.md)) — Maximum size of a data part that can be stored on volume disks (0 — no limit).
 -   `move_factor` ([Float64](../../sql-reference/data-types/float.md)) — Ratio of free disk space. When the ratio exceeds the value of configuration parameter, ClickHouse start to move data to the next volume in order.
 -   `prefer_not_to_merge` ([UInt8](../../sql-reference/data-types/int-uint.md)) — Value of the `prefer_not_to_merge` setting. When this setting is enabled, merging data on this volume is not allowed. This allows controlling how ClickHouse works with slow disks.
+-   `perform_ttl_move_on_insert` ([UInt8](../../sql-reference/data-types/int-uint.md)) — Value of the `perform_ttl_move_on_insert` setting. — Disables TTL move on data part INSERT. By default if we insert a data part that already expired by the TTL move rule it immediately goes to a volume/disk declared in move rule. This can significantly slowdown insert in case if destination volume/disk is slow (e.g. S3).
+-   `load_balancing` ([Enum8](../../sql-reference/data-types/enum.md))  — Policy for disk balancing. Can have one of the following values:
+    -   `ROUND_ROBIN`
+    -   `LEAST_USED`

 If the storage policy contains more then one volume, then information for each volume is stored in the individual row of the table.
--- a/docs/en/sql-reference/aggregate-functions/reference/deltasumtimestamp.md
+++ b/docs/en/sql-reference/aggregate-functions/reference/deltasumtimestamp.md
@ -6,7 +6,7 @@ title: deltaSumTimestamp

 Adds the difference between consecutive rows. If the difference is negative, it is ignored.

-This function is primarily for [materialized views](../../../sql-reference/statements/create/view.md#materialized) that are ordered by some time bucket-aligned timestamp, for example, a `toStartOfMinute` bucket. Because the rows in such a materialized view will all have the same timestamp, it is impossible for them to be merged in the "right" order. This function keeps track of the `timestamp` of the values it's seen, so it's possible to order the states correctly during merging.
+This function is primarily for [materialized views](../../../sql-reference/statements/create/view.md#materialized) that store data ordered by some time bucket-aligned timestamp, for example, a `toStartOfMinute` bucket. Because the rows in such a materialized view will all have the same timestamp, it is impossible for them to be merged in the correct order, without storing the original, unrounded timestamp value. The `deltaSumTimestamp` function keeps track of the original `timestamp` of the values it's seen, so the values (states) of the function are correctly computed during merging of parts.

 To calculate the delta sum across an ordered collection you can simply use the [deltaSum](../../../sql-reference/aggregate-functions/reference/deltasum.md#agg_functions-deltasum) function.

--- a/docs/en/sql-reference/functions/date-time-functions.md
+++ b/docs/en/sql-reference/functions/date-time-functions.md
@ -1264,7 +1264,7 @@ Using replacement fields, you can define a pattern for the resulting string. “
 | %d       | day of the month, zero-padded (01-31)                   | 02         |
 | %D       | Short MM/DD/YY date, equivalent to %m/%d/%y             | 01/02/18   |
 | %e       | day of the month, space-padded (1-31)                   | &nbsp; 2   |
-| %f       | fractional second from the fractional part of DateTime64 | 1234560   |
+| %f       | fractional second, see 'Note 1' below                   | 1234560    |
 | %F       | short YYYY-MM-DD date, equivalent to %Y-%m-%d           | 2018-01-02 |
 | %g       | two-digit year format, aligned to ISO 8601, abbreviated from four-digit notation                                | 18       |
 | %G       | four-digit year format for ISO week number, calculated from the week-based year [defined by the ISO 8601](https://en.wikipedia.org/wiki/ISO_8601#Week_dates) standard, normally useful only with %V  | 2018         |
@ -1276,7 +1276,7 @@ Using replacement fields, you can define a pattern for the resulting string. “
 | %k       | hour in 24h format (00-23)                              | 22         |
 | %l       | hour in 12h format (01-12)                              | 09         |
 | %m       | month as an integer number (01-12)                      | 01         |
-| %M       | full month name (January-December), see (*) below       | January    |
+| %M       | full month name (January-December), see 'Note 2' below  | January    |
 | %n       | new-line character (‘’)                                 |            |
 | %p       | AM or PM designation                                    | PM         |
 | %Q       | Quarter (1-4)                                           | 1          |
@ -1295,7 +1295,9 @@ Using replacement fields, you can define a pattern for the resulting string. “
 | %z       | Time offset from UTC as +HHMM or -HHMM                  | -0500      |
 | %%       | a % sign                                                | %          |

-(*) In ClickHouse versions earlier than v23.4, `%M` prints the minute (00-59) instead of the full month name (January-December). The previous behavior can be restored using setting `formatdatetime_parsedatetime_m_is_month_name = 0`.
+Note 1: In ClickHouse versions earlier than v23.4, `%f` prints a single zero (0) if the formatted value is a Date, Date32 or DateTime (which have no fractional seconds) or a DateTime64 with a precision of 0. The previous behavior can be restored using setting `formatdatetime_f_prints_single_zero = 1`.
+
+Note 2: In ClickHouse versions earlier than v23.4, `%M` prints the minute (00-59) instead of the full month name (January-December). The previous behavior can be restored using setting `formatdatetime_parsedatetime_m_is_month_name = 0`.

 **Example**

--- a/docs/en/sql-reference/functions/hash-functions.md
+++ b/docs/en/sql-reference/functions/hash-functions.md
@ -441,11 +441,11 @@ SELECT farmHash64(array('e','x','a'), 'mple', 10, toDateTime('2019-06-15 23:00:0

 ## javaHash

-Calculates JavaHash from a [string](http://hg.openjdk.java.net/jdk8u/jdk8u/jdk/file/478a4add975b/src/share/classes/java/lang/String.java#l1452), 
-[Byte](https://hg.openjdk.java.net/jdk8u/jdk8u/jdk/file/478a4add975b/src/share/classes/java/lang/Byte.java#l405), 
-[Short](https://hg.openjdk.java.net/jdk8u/jdk8u/jdk/file/478a4add975b/src/share/classes/java/lang/Short.java#l410), 
-[Integer](https://hg.openjdk.java.net/jdk8u/jdk8u/jdk/file/478a4add975b/src/share/classes/java/lang/Integer.java#l959), 
-[Long](https://hg.openjdk.java.net/jdk8u/jdk8u/jdk/file/478a4add975b/src/share/classes/java/lang/Long.java#l1060). 
+Calculates JavaHash from a [string](http://hg.openjdk.java.net/jdk8u/jdk8u/jdk/file/478a4add975b/src/share/classes/java/lang/String.java#l1452),
+[Byte](https://hg.openjdk.java.net/jdk8u/jdk8u/jdk/file/478a4add975b/src/share/classes/java/lang/Byte.java#l405),
+[Short](https://hg.openjdk.java.net/jdk8u/jdk8u/jdk/file/478a4add975b/src/share/classes/java/lang/Short.java#l410),
+[Integer](https://hg.openjdk.java.net/jdk8u/jdk8u/jdk/file/478a4add975b/src/share/classes/java/lang/Integer.java#l959),
+[Long](https://hg.openjdk.java.net/jdk8u/jdk8u/jdk/file/478a4add975b/src/share/classes/java/lang/Long.java#l1060).
 This hash function is neither fast nor having a good quality. The only reason to use it is when this algorithm is already used in another system and you have to calculate exactly the same result.

 Note that Java only support calculating signed integers hash, so if you want to calculate unsigned integers hash you must cast it to proper signed ClickHouse types.
@ -660,6 +660,45 @@ Result:
 └──────────────────────┴─────────────────────┘
 ```

+
+## kafkaMurmurHash
+
+Calculates a 32-bit [MurmurHash2](https://github.com/aappleby/smhasher) hash value using the same hash seed as [Kafka](https://github.com/apache/kafka/blob/461c5cfe056db0951d9b74f5adc45973670404d7/clients/src/main/java/org/apache/kafka/common/utils/Utils.java#L482) and without the highest bit to be compatible with [Default Partitioner](https://github.com/apache/kafka/blob/139f7709bd3f5926901a21e55043388728ccca78/clients/src/main/java/org/apache/kafka/clients/producer/internals/BuiltInPartitioner.java#L328).
+
+**Syntax**
+
+```sql
+MurmurHash(par1, ...)
+```
+
+**Arguments**
+
+-   `par1, ...` — A variable number of parameters that can be any of the [supported data types](/docs/en/sql-reference/data-types/index.md/#data_types).
+
+**Returned value**
+
+-   Calculated hash value.
+
+Type: [UInt32](/docs/en/sql-reference/data-types/int-uint.md).
+
+**Example**
+
+Query:
+
+```sql
+SELECT
+    kafkaMurmurHash('foobar') AS res1,
+    kafkaMurmurHash(array('e','x','a'), 'mple', 10, toDateTime('2019-06-15 23:00:00')) AS res2
+```
+
+Result:
+
+```response
+┌───────res1─┬─────res2─┐
+│ 1357151166 │ 85479775 │
+└────────────┴──────────┘
+```
+
 ## murmurHash3_32, murmurHash3_64

 Produces a [MurmurHash3](https://github.com/aappleby/smhasher) hash value.
--- a/docs/en/sql-reference/functions/type-conversion-functions.md
+++ b/docs/en/sql-reference/functions/type-conversion-functions.md
@ -1245,7 +1245,6 @@ Returns DateTime values parsed from input string according to a MySQL style form
 **Supported format specifiers**

 All format specifiers listed in [formatDateTime](/docs/en/sql-reference/functions/date-time-functions.md#date_time_functions-formatDateTime) except:
- %f: fractional second
 - %Q: Quarter (1-4) 

 **Example**
--- a/docs/en/sql-reference/statements/drop.md
+++ b/docs/en/sql-reference/statements/drop.md
@ -22,6 +22,10 @@ DROP DATABASE [IF EXISTS] db [ON CLUSTER cluster] [SYNC]

 Deletes the table.

+:::tip
+Also see [UNDROP TABLE](/docs/en/sql-reference/statements/undrop.md)
+:::
+
 Syntax:

 ``` sql
--- a/docs/en/sql-reference/statements/undrop.md
+++ b/docs/en/sql-reference/statements/undrop.md
@ -0,0 +1,99 @@
+---
+slug: /en/sql-reference/statements/undrop
+sidebar_label: UNDROP
+---
+
+# UNDROP TABLE
+
+Cancels the dropping of the table.
+
+Beginning with ClickHouse version 23.3 it is possible to UNDROP a table in an Atomic database
+within `database_atomic_delay_before_drop_table_sec` (8 minutes by default) of issuing the DROP TABLE statement.  Dropped tables are listed in 
+a system table called `system.dropped_tables`.
+
+If you have a materialized view without a `TO` clause associated with the dropped table, then you will also have to UNDROP the inner table of that view.
+
+:::note
+UNDROP TABLE is experimental.  To use it add this setting: 
+```sql
+set allow_experimental_undrop_table_query = 1;
+```
+:::
+
+:::tip
+Also see [DROP TABLE](/docs/en/sql-reference/statements/drop.md)
+:::
+
+Syntax:
+
+``` sql
+UNDROP TABLE [db.]name [UUID '<uuid>'] [ON CLUSTER cluster]
+```
+
+**Example**
+
+``` sql
+set allow_experimental_undrop_table_query = 1;
+```
+
+```sql
+CREATE TABLE undropMe
+(
+    `id` UInt8
+)
+ENGINE = MergeTree
+ORDER BY id
+```
+
+```sql
+DROP TABLE undropMe
+```
+```sql
+SELECT *
+FROM system.dropped_tables
+FORMAT Vertical
+```
+```response
+Row 1:
+──────
+index:                 0
+database:              default
+table:                 undropMe
+uuid:                  aa696a1a-1d70-4e60-a841-4c80827706cc
+engine:                MergeTree
+metadata_dropped_path: /var/lib/clickhouse/metadata_dropped/default.undropMe.aa696a1a-1d70-4e60-a841-4c80827706cc.sql
+table_dropped_time:    2023-04-05 14:12:12
+
+1 row in set. Elapsed: 0.001 sec. 
+```
+```sql
+UNDROP TABLE undropMe
+```
+```response
+Ok.
+```
+```sql
+SELECT *
+FROM system.dropped_tables
+FORMAT Vertical
+```
+```response
+Ok.
+
+0 rows in set. Elapsed: 0.001 sec. 
+```
+```sql
+DESCRIBE TABLE undropMe
+FORMAT Vertical
+```
+```response
+Row 1:
+──────
+name:               id
+type:               UInt8
+default_type:       
+default_expression: 
+comment:            
+codec_expression:   
+ttl_expression:     
+```
--- a/docs/en/sql-reference/syntax.md
+++ b/docs/en/sql-reference/syntax.md
@ -14,7 +14,7 @@ The `INSERT` query uses both parsers:
 INSERT INTO t VALUES (1, 'Hello, world'), (2, 'abc'), (3, 'def')
 ```

-The `INSERT INTO t VALUES` fragment is parsed by the full parser, and the data `(1, 'Hello, world'), (2, 'abc'), (3, 'def')` is parsed by the fast stream parser. You can also turn on the full parser for the data by using the [input_format_values_interpret_expressions](../operations/settings/settings-formats.md#settings-input_format_values_interpret_expressions) setting. When `input_format_values_interpret_expressions = 1`, ClickHouse first tries to parse values with the fast stream parser. If it fails, ClickHouse tries to use the full parser for the data, treating it like an SQL [expression](#syntax-expressions).
+The `INSERT INTO t VALUES` fragment is parsed by the full parser, and the data `(1, 'Hello, world'), (2, 'abc'), (3, 'def')` is parsed by the fast stream parser. You can also turn on the full parser for the data by using the [input_format_values_interpret_expressions](../operations/settings/settings-formats.md#input_format_values_interpret_expressions) setting. When `input_format_values_interpret_expressions = 1`, ClickHouse first tries to parse values with the fast stream parser. If it fails, ClickHouse tries to use the full parser for the data, treating it like an SQL [expression](#expressions).

 Data can have any format. When a query is received, the server calculates no more than [max_query_size](../operations/settings/settings.md#settings-max_query_size) bytes of the request in RAM (by default, 1 MB), and the rest is stream parsed.
 It allows for avoiding issues with large `INSERT` queries.
@ -45,7 +45,7 @@ You can check whether a data type name is case-sensitive in the [system.data_typ

 In contrast to standard SQL, all other keywords (including functions names) are **case-sensitive**.

-Keywords are not reserved; they are treated as such only in the corresponding context. If you use [identifiers](#syntax-identifiers) with the same name as the keywords, enclose them into double-quotes or backticks. For example, the query `SELECT "FROM" FROM table_name` is valid if the table `table_name` has column with the name `"FROM"`.
+Keywords are not reserved; they are treated as such only in the corresponding context. If you use [identifiers](#identifiers) with the same name as the keywords, enclose them into double-quotes or backticks. For example, the query `SELECT "FROM" FROM table_name` is valid if the table `table_name` has column with the name `"FROM"`.

 ## Identifiers

@ -54,7 +54,7 @@ Identifiers are:
 -   Cluster, database, table, partition, and column names.
 -   Functions.
 -   Data types.
-   [Expression aliases](#syntax-expression_aliases).
+-   [Expression aliases](#expression_aliases).

 Identifiers can be quoted or non-quoted. The latter is preferred.

@ -108,7 +108,7 @@ Depending on the data format (input or output), `NULL` may have a different repr

 There are many nuances to processing `NULL`. For example, if at least one of the arguments of a comparison operation is `NULL`, the result of this operation is also `NULL`. The same is true for multiplication, addition, and other operations. For more information, read the documentation for each operation.

-In queries, you can check `NULL` using the [IS NULL](../sql-reference/operators/index.md#operator-is-null) and [IS NOT NULL](../sql-reference/operators/index.md) operators and the related functions `isNull` and `isNotNull`.
+In queries, you can check `NULL` using the [IS NULL](../sql-reference/operators/index.md#is-null) and [IS NOT NULL](../sql-reference/operators/index.md#is-not-null) operators and the related functions `isNull` and `isNotNull`.

 ### Heredoc

@ -149,7 +149,7 @@ For example, the following SQL defines parameters named `a`, `b`, `c` and `d` -
 SET param_a = 13;
 SET param_b = 'str';
 SET param_c = '2022-08-04 18:30:53';
-SET param_d = {'10': [11, 12], '13': [14, 15]}';
+SET param_d = {'10': [11, 12], '13': [14, 15]};

 SELECT
   {a: UInt32},
@ -166,7 +166,7 @@ Result:

 If you are using `clickhouse-client`, the parameters are specified as `--param_name=value`. For example, the following parameter has the name `message` and it is retrieved as a `String`:

-```sql
+```bash
 clickhouse-client --param_message='hello' --query="SELECT {message: String}"
 ```

@ -190,7 +190,7 @@ Query parameters are not general text substitutions which can be used in arbitra
 ## Functions

 Function calls are written like an identifier with a list of arguments (possibly empty) in round brackets. In contrast to standard SQL, the brackets are required, even for an empty argument list. Example: `now()`.
-There are regular and aggregate functions (see the section “Aggregate functions”). Some aggregate functions can contain two lists of arguments in brackets. Example: `quantile (0.9) (x)`. These aggregate functions are called “parametric” functions, and the arguments in the first list are called “parameters”. The syntax of aggregate functions without parameters is the same as for regular functions.
+There are regular and aggregate functions (see the section [Aggregate functions](/docs/en/sql-reference/aggregate-functions/index.md)). Some aggregate functions can contain two lists of arguments in brackets. Example: `quantile (0.9) (x)`. These aggregate functions are called “parametric” functions, and the arguments in the first list are called “parameters”. The syntax of aggregate functions without parameters is the same as for regular functions.

 ## Operators

@ -199,7 +199,7 @@ For example, the expression `1 + 2 * 3 + 4` is transformed to `plus(plus(1, mult

 ## Data Types and Database Table Engines

-Data types and table engines in the `CREATE` query are written the same way as identifiers or functions. In other words, they may or may not contain an argument list in brackets. For more information, see the sections “Data types,” “Table engines,” and “CREATE”.
+Data types and table engines in the `CREATE` query are written the same way as identifiers or functions. In other words, they may or may not contain an argument list in brackets. For more information, see the sections [Data types](/docs/en/sql-reference/data-types/index.md), [Table engines](/docs/en/engines/table-engines/index.md), and [CREATE](/docs/en/sql-reference/statements/create/index.md).

 ## Expression Aliases

@ -211,17 +211,17 @@ expr AS alias

 -   `AS` — The keyword for defining aliases. You can define the alias for a table name or a column name in a `SELECT` clause without using the `AS` keyword.

-        For example, `SELECT table_name_alias.column_name FROM table_name table_name_alias`.
+    For example, `SELECT table_name_alias.column_name FROM table_name table_name_alias`.

-        In the [CAST](./functions/type-conversion-functions.md#type_conversion_function-cast) function, the `AS` keyword has another meaning. See the description of the function.
+    In the [CAST](./functions/type-conversion-functions.md#castx-t) function, the `AS` keyword has another meaning. See the description of the function.

 -   `expr` — Any expression supported by ClickHouse.

-        For example, `SELECT column_name * 2 AS double FROM some_table`.
+    For example, `SELECT column_name * 2 AS double FROM some_table`.

-   `alias` — Name for `expr`. Aliases should comply with the [identifiers](#syntax-identifiers) syntax.
+-   `alias` — Name for `expr`. Aliases should comply with the [identifiers](#identifiers) syntax.

-        For example, `SELECT "table t".column_name FROM table_name AS "table t"`.
+    For example, `SELECT "table t".column_name FROM table_name AS "table t"`.

 ### Notes on Usage

@ -254,11 +254,11 @@ Received exception from server (version 18.14.17):
 Code: 184. DB::Exception: Received from localhost:9000, 127.0.0.1. DB::Exception: Aggregate function sum(b) is found inside another aggregate function in query.
 ```

-In this example, we declared table `t` with column `b`. Then, when selecting data, we defined the `sum(b) AS b` alias. As aliases are global, ClickHouse substituted the literal `b` in the expression `argMax(a, b)` with the expression `sum(b)`. This substitution caused the exception. You can change this default behavior by setting [prefer_column_name_to_alias](../operations/settings/settings.md#prefer_column_name_to_alias) to `1`.
+In this example, we declared table `t` with column `b`. Then, when selecting data, we defined the `sum(b) AS b` alias. As aliases are global, ClickHouse substituted the literal `b` in the expression `argMax(a, b)` with the expression `sum(b)`. This substitution caused the exception. You can change this default behavior by setting [prefer_column_name_to_alias](../operations/settings/settings.md#prefer-column-name-to-alias) to `1`.

 ## Asterisk

-In a `SELECT` query, an asterisk can replace the expression. For more information, see the section “SELECT”.
+In a `SELECT` query, an asterisk can replace the expression. For more information, see the section [SELECT](/docs/en/sql-reference/statements/select/index.md#asterisk).

 ## Expressions

--- a/docs/en/sql-reference/table-functions/executable.md
+++ b/docs/en/sql-reference/table-functions/executable.md
@ -20,7 +20,7 @@ A key advantage between ordinary UDF functions and the `executable` table functi
 The `executable` table function requires three parameters and accepts an optional list of input queries:

 ```sql
-executable(script_name, format, structure, [input_query...])
+executable(script_name, format, structure, [input_query...] [,SETTINGS ...])
 ```

 - `script_name`: the file name of the script. saved in the `user_scripts` folder (the default folder of the `user_scripts_path` setting)
@ -83,6 +83,15 @@ The response looks like:
 └────┴────────────┘
 ```

+## Settings
+
+-   `send_chunk_header` - controls whether to send row count before sending a chunk of data to process. Default value is `false`.
+-   `pool_size` — Size of pool. If 0 is specified as `pool_size` then there is no pool size restrictions. Default value is `16`.
+-   `max_command_execution_time` — Maximum executable script command execution time for processing block of data. Specified in seconds. Default value is 10.
+-   `command_termination_timeout` — executable script should contain main read-write loop. After table function is destroyed, pipe is closed, and executable file will have `command_termination_timeout` seconds to shutdown, before ClickHouse will send SIGTERM signal to child process. Specified in seconds. Default value is 10.
+-   `command_read_timeout` - timeout for reading data from command stdout in milliseconds. Default value 10000.
+-   `command_write_timeout` - timeout for writing data to command stdin in milliseconds. Default value 10000.
+
 ## Passing Query Results to a Script

 Be sure to check out the example in the `Executable` table engine on [how to pass query results to a script](../../engines/table-engines/special/executable.md#passing-query-results-to-a-script). Here is how you execute the same script in that example using the `executable` table function:
@ -94,4 +103,4 @@ SELECT * FROM executable(
    'id UInt64, sentiment Float32',
    (SELECT id, comment FROM hackernews WHERE id > 0 AND comment != '' LIMIT 20)
 );
-```
+```
--- a/docs/ru/sql-reference/aggregate-functions/reference/deltasumtimestamp.md
+++ b/docs/ru/sql-reference/aggregate-functions/reference/deltasumtimestamp.md
@ -7,7 +7,7 @@ sidebar_position: 141

 Суммирует разницу между последовательными строками. Если разница отрицательна — она будет проигнорирована.

-Эта функция предназначена в первую очередь для [материализованных представлений](../../../sql-reference/statements/create/view.md#materialized), упорядоченных по некоторому временному бакету согласно timestamp, например, по бакету `toStartOfMinute`. Поскольку строки в таком материализованном представлении будут иметь одинаковый timestamp, невозможно объединить их в "правом" порядке. Функция отслеживает `timestamp` наблюдаемых значений, поэтому возможно правильно упорядочить состояния во время слияния.
+Эта функция предназначена в первую очередь для [материализованных представлений](../../../sql-reference/statements/create/view.md#materialized), хранящих данные, упорядоченные по некоторому округленному временному интервалу, согласно timestamp, например, по бакету `toStartOfMinute`. Поскольку строки в таком материализованном представлении будут иметь одинаковый timestamp, их невозможно объединить в правильном порядке без хранения исходного, неокругленного значения timestamp. Функция  `deltaSumTimestamp` отслеживает исходные `timestamp` наблюдаемых значений, поэтому значения (состояния) функции правильно вычисляются во время слияния кусков.

 Чтобы вычислить разницу между упорядоченными последовательными строками, вы можете использовать функцию [deltaSum](../../../sql-reference/aggregate-functions/reference/deltasum.md#agg_functions-deltasum) вместо функции `deltaSumTimestamp`.

--- a/programs/client/Client.cpp
+++ b/programs/client/Client.cpp
@ -277,11 +277,11 @@ void Client::initialize(Poco::Util::Application & self)
      */

    const char * env_user = getenv("CLICKHOUSE_USER"); // NOLINT(concurrency-mt-unsafe)
-    if (env_user)
+    if (env_user && !config().has("user"))
        config().setString("user", env_user);

    const char * env_password = getenv("CLICKHOUSE_PASSWORD"); // NOLINT(concurrency-mt-unsafe)
-    if (env_password)
+    if (env_password && !config().has("password"))
        config().setString("password", env_password);

    parseConnectionsCredentials();
--- a/programs/install/Install.cpp
+++ b/programs/install/Install.cpp
@ -375,15 +375,22 @@ int mainEntryClickHouseInstall(int argc, char ** argv)

                try
                {
-                    ReadBufferFromFile in(binary_self_path.string());
-                    WriteBufferFromFile out(main_bin_tmp_path.string());
-                    copyData(in, out);
-                    out.sync();
+                    String source = binary_self_path.string();
+                    String destination = main_bin_tmp_path.string();

-                    if (0 != fchmod(out.getFD(), S_IRUSR | S_IRGRP | S_IROTH | S_IXUSR | S_IXGRP | S_IXOTH))
+                    /// Try to make a hard link first, as an optimization.
+                    /// It is possible if the source and the destination are on the same filesystems.
+                    if (0 != link(source.c_str(), destination.c_str()))
+                    {
+                        ReadBufferFromFile in(binary_self_path.string());
+                        WriteBufferFromFile out(main_bin_tmp_path.string());
+                        copyData(in, out);
+                        out.sync();
+                        out.finalize();
+                    }
+
+                    if (0 != chmod(destination.c_str(), S_IRUSR | S_IRGRP | S_IROTH | S_IXUSR | S_IXGRP | S_IXOTH))
                        throwFromErrno(fmt::format("Cannot chmod {}", main_bin_tmp_path.string()), ErrorCodes::SYSTEM_ERROR);
-
-                    out.finalize();
                }
                catch (const Exception & e)
                {
--- a/programs/keeper/Keeper.cpp
+++ b/programs/keeper/Keeper.cpp
@ -17,7 +17,6 @@
 #include <Poco/Net/TCPServerParams.h>
 #include <Poco/Net/TCPServer.h>
 #include <Poco/Util/HelpFormatter.h>
-#include <Poco/Version.h>
 #include <Poco/Environment.h>
 #include <sys/stat.h>
 #include <pwd.h>
--- a/programs/server/Server.cpp
+++ b/programs/server/Server.cpp
@ -981,7 +981,7 @@ try

    StatusFile status{path / "status", StatusFile::write_full_info};

-    DB::ServerUUID::load(path / "uuid", log);
+    ServerUUID::load(path / "uuid", log);

    /// Try to increase limit on number of open files.
    {
--- a/src/Access/DiskAccessStorage.cpp
+++ b/src/Access/DiskAccessStorage.cpp
@ -10,6 +10,7 @@
 #include <Interpreters/Access/InterpreterCreateUserQuery.h>
 #include <Interpreters/Access/InterpreterShowGrantsQuery.h>
 #include <Common/logger_useful.h>
+#include <Common/ThreadPool.h>
 #include <Poco/JSON/JSON.h>
 #include <Poco/JSON/Object.h>
 #include <Poco/JSON/Stringifier.h>
@ -19,6 +20,7 @@
 #include <base/range.h>
 #include <filesystem>
 #include <fstream>
+#include <memory>


 namespace DB
@ -317,15 +319,15 @@ void DiskAccessStorage::scheduleWriteLists(AccessEntityType type)
        return; /// If the lists' writing thread is still waiting we can update `types_of_lists_to_write` easily,
                /// without restarting that thread.

-    if (lists_writing_thread.joinable())
-        lists_writing_thread.join();
+    if (lists_writing_thread && lists_writing_thread->joinable())
+        lists_writing_thread->join();

    /// Create the 'need_rebuild_lists.mark' file.
    /// This file will be used later to find out if writing lists is successful or not.
    std::ofstream out{getNeedRebuildListsMarkFilePath(directory_path)};
    out.close();

-    lists_writing_thread = ThreadFromGlobalPool{&DiskAccessStorage::listsWritingThreadFunc, this};
+    lists_writing_thread = std::make_unique<ThreadFromGlobalPool>(&DiskAccessStorage::listsWritingThreadFunc, this);
    lists_writing_thread_is_waiting = true;
 }

@ -349,10 +351,10 @@ void DiskAccessStorage::listsWritingThreadFunc()

 void DiskAccessStorage::stopListsWritingThread()
 {
-    if (lists_writing_thread.joinable())
+    if (lists_writing_thread && lists_writing_thread->joinable())
    {
        lists_writing_thread_should_exit.notify_one();
-        lists_writing_thread.join();
+        lists_writing_thread->join();
    }
 }

--- a/src/Access/DiskAccessStorage.h
+++ b/src/Access/DiskAccessStorage.h
@ -1,7 +1,7 @@
 #pragma once

 #include <Access/MemoryAccessStorage.h>
-#include <Common/ThreadPool.h>
+#include <Common/ThreadPool_fwd.h>
 #include <boost/container/flat_set.hpp>


@ -81,7 +81,7 @@ private:
    bool failed_to_write_lists TSA_GUARDED_BY(mutex) = false;

    /// List files are written in a separate thread.
-    ThreadFromGlobalPool lists_writing_thread;
+    std::unique_ptr<ThreadFromGlobalPool> lists_writing_thread;

    /// Signals `lists_writing_thread` to exit.
    std::condition_variable lists_writing_thread_should_exit;
--- a/src/Access/ReplicatedAccessStorage.cpp
+++ b/src/Access/ReplicatedAccessStorage.cpp
@ -1,3 +1,4 @@
+#include <memory>
 #include <Access/AccessEntityIO.h>
 #include <Access/MemoryAccessStorage.h>
 #include <Access/ReplicatedAccessStorage.h>
@ -15,6 +16,7 @@
 #include <Common/ZooKeeper/ZooKeeper.h>
 #include <Common/escapeForFileName.h>
 #include <Common/setThreadName.h>
+#include <Common/ThreadPool.h>
 #include <base/range.h>
 #include <base/sleep.h>
 #include <boost/range/algorithm_ext/erase.hpp>
@ -72,7 +74,7 @@ void ReplicatedAccessStorage::startWatchingThread()
 {
    bool prev_watching_flag = watching.exchange(true);
    if (!prev_watching_flag)
-        watching_thread = ThreadFromGlobalPool(&ReplicatedAccessStorage::runWatchingThread, this);
+        watching_thread = std::make_unique<ThreadFromGlobalPool>(&ReplicatedAccessStorage::runWatchingThread, this);
 }

 void ReplicatedAccessStorage::stopWatchingThread()
@ -81,8 +83,8 @@ void ReplicatedAccessStorage::stopWatchingThread()
    if (prev_watching_flag)
    {
        watched_queue->finish();
-        if (watching_thread.joinable())
-            watching_thread.join();
+        if (watching_thread && watching_thread->joinable())
+            watching_thread->join();
    }
 }

--- a/src/Access/ReplicatedAccessStorage.h
+++ b/src/Access/ReplicatedAccessStorage.h
@ -2,7 +2,7 @@

 #include <atomic>

-#include <Common/ThreadPool.h>
+#include <Common/ThreadPool_fwd.h>
 #include <Common/ZooKeeper/Common.h>
 #include <Common/ZooKeeper/ZooKeeper.h>
 #include <Common/ConcurrentBoundedQueue.h>
@ -21,7 +21,7 @@ public:
    static constexpr char STORAGE_TYPE[] = "replicated";

    ReplicatedAccessStorage(const String & storage_name, const String & zookeeper_path, zkutil::GetZooKeeper get_zookeeper, AccessChangesNotifier & changes_notifier_, bool allow_backup);
-    virtual ~ReplicatedAccessStorage() override;
+    ~ReplicatedAccessStorage() override;

    const char * getStorageType() const override { return STORAGE_TYPE; }

@ -43,7 +43,7 @@ private:
    std::mutex cached_zookeeper_mutex;

    std::atomic<bool> watching = false;
-    ThreadFromGlobalPool watching_thread;
+    std::unique_ptr<ThreadFromGlobalPool> watching_thread;
    std::shared_ptr<ConcurrentBoundedQueue<UUID>> watched_queue;

    std::optional<UUID> insertImpl(const AccessEntityPtr & entity, bool replace_if_exists, bool throw_if_exists) override;
--- a/src/AggregateFunctions/AggregateFunctionMaxIntersections.h
+++ b/src/AggregateFunctions/AggregateFunctionMaxIntersections.h
@ -1,6 +1,5 @@
 #pragma once

-#include <Common/logger_useful.h>
 #include <base/sort.h>

 #include <DataTypes/DataTypesNumber.h>
--- a/src/AggregateFunctions/AggregateFunctionSparkbar.h
+++ b/src/AggregateFunctions/AggregateFunctionSparkbar.h
@ -11,7 +11,6 @@
 #include <IO/WriteHelpers.h>
 #include <Columns/ColumnString.h>
 #include <Common/PODArray.h>
-#include <Common/logger_useful.h>
 #include <IO/ReadBufferFromString.h>
 #include <Common/HashTable/HashMap.h>
 #include <Columns/IColumn.h>
--- a/src/AggregateFunctions/AggregateFunctionSumMap.h
+++ b/src/AggregateFunctions/AggregateFunctionSumMap.h
@ -18,7 +18,6 @@
 #include <AggregateFunctions/IAggregateFunction.h>
 #include <AggregateFunctions/FactoryHelpers.h>
 #include <map>
-#include <Common/logger_useful.h>
 #include <Common/ClickHouseRevision.h>


--- a/src/AggregateFunctions/IAggregateFunction.h
+++ b/src/AggregateFunctions/IAggregateFunction.h
@ -9,7 +9,7 @@
 #include <Interpreters/Context_fwd.h>
 #include <base/types.h>
 #include <Common/Exception.h>
-#include <Common/ThreadPool.h>
+#include <Common/ThreadPool_fwd.h>
 #include <Core/IResolvedFunction.h>

 #include "config.h"
--- a/src/Analyzer/Passes/QueryAnalysisPass.cpp
+++ b/src/Analyzer/Passes/QueryAnalysisPass.cpp
@ -32,6 +32,7 @@
 #include <AggregateFunctions/AggregateFunctionFactory.h>

 #include <TableFunctions/TableFunctionFactory.h>
+#include <Formats/FormatFactory.h>

 #include <Databases/IDatabase.h>

@ -75,6 +76,7 @@
 #include <Analyzer/InDepthQueryTreeVisitor.h>
 #include <Analyzer/QueryTreeBuilder.h>
 #include <Analyzer/IQueryTreeNode.h>
+#include <Analyzer/Identifier.h>

 namespace ProfileEvents
 {
@ -112,6 +114,8 @@ namespace ErrorCodes
    extern const int ALIAS_REQUIRED;
    extern const int NUMBER_OF_ARGUMENTS_DOESNT_MATCH;
    extern const int UNKNOWN_TABLE;
+    extern const int ILLEGAL_COLUMN;
+    extern const int NUMBER_OF_COLUMNS_DOESNT_MATCH;
 }

 /** Query analyzer implementation overview. Please check documentation in QueryAnalysisPass.h before.
@ -6079,6 +6083,18 @@ void QueryAnalyzer::initializeTableExpressionData(const QueryTreeNodePtr & table
    scope.table_expression_node_to_data.emplace(table_expression_node, std::move(table_expression_data));
 }

+bool findIdentifier(const FunctionNode & function)
+{
+    for (const auto & argument : function.getArguments())
+    {
+        if (argument->as<IdentifierNode>())
+            return true;
+        if (const auto * f = argument->as<FunctionNode>(); f && findIdentifier(*f))
+            return true;
+    }
+    return false;
+}
+
 /// Resolve table function node in scope
 void QueryAnalyzer::resolveTableFunction(QueryTreeNodePtr & table_function_node,
    IdentifierResolveScope & scope,
@ -6090,12 +6106,11 @@ void QueryAnalyzer::resolveTableFunction(QueryTreeNodePtr & table_function_node,
    if (!nested_table_function)
        expressions_visitor.visit(table_function_node_typed.getArgumentsNode());

-    const auto & table_function_factory = TableFunctionFactory::instance();
    const auto & table_function_name = table_function_node_typed.getTableFunctionName();

    auto & scope_context = scope.context;

-    TableFunctionPtr table_function_ptr = table_function_factory.tryGet(table_function_name, scope_context);
+    TableFunctionPtr table_function_ptr = TableFunctionFactory::instance().tryGet(table_function_name, scope_context);
    if (!table_function_ptr)
    {
        auto hints = TableFunctionFactory::instance().getHints(table_function_name);
@ -6110,17 +6125,131 @@ void QueryAnalyzer::resolveTableFunction(QueryTreeNodePtr & table_function_node,
                table_function_name);
    }

+    uint64_t use_structure_from_insertion_table_in_table_functions = scope_context->getSettingsRef().use_structure_from_insertion_table_in_table_functions;
    if (!nested_table_function &&
-        scope_context->getSettingsRef().use_structure_from_insertion_table_in_table_functions &&
+        use_structure_from_insertion_table_in_table_functions &&
        scope_context->hasInsertionTable() &&
        table_function_ptr->needStructureHint())
    {
        const auto & insertion_table = scope_context->getInsertionTable();
        if (!insertion_table.empty())
        {
-            auto insertion_table_storage = DatabaseCatalog::instance().getTable(insertion_table, scope_context);
-            const auto & structure_hint = insertion_table_storage->getInMemoryMetadataPtr()->columns;
-            table_function_ptr->setStructureHint(structure_hint);
+            const auto & insert_structure = DatabaseCatalog::instance().getTable(insertion_table, scope_context)->getInMemoryMetadataPtr()->getColumns();
+            DB::ColumnsDescription structure_hint;
+
+            bool use_columns_from_insert_query = true;
+
+            /// Insert table matches columns against SELECT expression by position, so we want to map
+            /// insert table columns to table function columns through names from SELECT expression.
+
+            auto insert_column = insert_structure.begin();
+            auto insert_structure_end = insert_structure.end();  /// end iterator of the range covered by possible asterisk
+            auto virtual_column_names = table_function_ptr->getVirtualsToCheckBeforeUsingStructureHint();
+            bool asterisk = false;
+            const auto & expression_list = scope.scope_node->as<QueryNode &>().getProjection();
+            auto expression = expression_list.begin();
+
+            /// We want to go through SELECT expression list and correspond each expression to column in insert table
+            /// which type will be used as a hint for the file structure inference.
+            for (; expression != expression_list.end() && insert_column != insert_structure_end; ++expression)
+            {
+                if (auto * identifier_node = (*expression)->as<IdentifierNode>())
+                {
+
+                    if (!virtual_column_names.contains(identifier_node->getIdentifier().getFullName()))
+                    {
+                        if (asterisk)
+                        {
+                            if (use_structure_from_insertion_table_in_table_functions == 1)
+                                throw Exception(ErrorCodes::ILLEGAL_COLUMN, "Asterisk cannot be mixed with column list in INSERT SELECT query.");
+
+                            use_columns_from_insert_query = false;
+                            break;
+                        }
+
+                        structure_hint.add({ identifier_node->getIdentifier().getFullName(), insert_column->type });
+                    }
+
+                    /// Once we hit asterisk we want to find end of the range covered by asterisk
+                    /// contributing every further SELECT expression to the tail of insert structure
+                    if (asterisk)
+                        --insert_structure_end;
+                    else
+                        ++insert_column;
+                }
+                else if (auto * matcher_node = (*expression)->as<MatcherNode>(); matcher_node && matcher_node->getMatcherType() == MatcherNodeType::ASTERISK)
+                {
+                    if (asterisk)
+                    {
+                        if (use_structure_from_insertion_table_in_table_functions == 1)
+                            throw Exception(ErrorCodes::ILLEGAL_COLUMN, "Only one asterisk can be used in INSERT SELECT query.");
+
+                        use_columns_from_insert_query = false;
+                        break;
+                    }
+                    if (!structure_hint.empty())
+                    {
+                        if (use_structure_from_insertion_table_in_table_functions == 1)
+                            throw Exception(ErrorCodes::ILLEGAL_COLUMN, "Asterisk cannot be mixed with column list in INSERT SELECT query.");
+
+                        use_columns_from_insert_query = false;
+                        break;
+                    }
+
+                    asterisk = true;
+                }
+                else if (auto * function = (*expression)->as<FunctionNode>())
+                {
+                    if (use_structure_from_insertion_table_in_table_functions == 2 && findIdentifier(*function))
+                    {
+                        use_columns_from_insert_query = false;
+                        break;
+                    }
+
+                    /// Once we hit asterisk we want to find end of the range covered by asterisk
+                    /// contributing every further SELECT expression to the tail of insert structure
+                    if (asterisk)
+                        --insert_structure_end;
+                    else
+                        ++insert_column;
+                }
+                else
+                {
+                    /// Once we hit asterisk we want to find end of the range covered by asterisk
+                    /// contributing every further SELECT expression to the tail of insert structure
+                    if (asterisk)
+                        --insert_structure_end;
+                    else
+                        ++insert_column;
+                }
+            }
+
+            if (use_structure_from_insertion_table_in_table_functions == 2 && !asterisk)
+            {
+                /// For input function we should check if input format supports reading subset of columns.
+                if (table_function_ptr->getName() == "input")
+                    use_columns_from_insert_query = FormatFactory::instance().checkIfFormatSupportsSubsetOfColumns(scope.context->getInsertFormat());
+                else
+                    use_columns_from_insert_query = table_function_ptr->supportsReadingSubsetOfColumns();
+            }
+
+            if (use_columns_from_insert_query)
+            {
+                if (expression == expression_list.end())
+                {
+                    /// Append tail of insert structure to the hint
+                    if (asterisk)
+                    {
+                        for (; insert_column != insert_structure_end; ++insert_column)
+                            structure_hint.add({ insert_column->name, insert_column->type });
+                    }
+
+                    if (!structure_hint.empty())
+                        table_function_ptr->setStructureHint(structure_hint);
+
+                } else if (use_structure_from_insertion_table_in_table_functions == 1)
+                    throw Exception(ErrorCodes::NUMBER_OF_COLUMNS_DOESNT_MATCH, "Number of columns in insert table less than required by SELECT expression.");
+            }
        }
    }

--- a/src/Backups/BackupCoordinationFileInfos.cpp
+++ b/src/Backups/BackupCoordinationFileInfos.cpp
@ -1,5 +1,6 @@
 #include <Backups/BackupCoordinationFileInfos.h>
 #include <Common/quoteString.h>
+#include <Common/Exception.h>


 namespace DB
--- a/src/Backups/BackupEntriesCollector.cpp
+++ b/src/Backups/BackupEntriesCollector.cpp
@ -15,6 +15,7 @@
 #include <base/sleep.h>
 #include <Common/escapeForFileName.h>
 #include <boost/range/algorithm/copy.hpp>
+#include <base/scope_guard.h>
 #include <filesystem>

 namespace fs = std::filesystem;
--- a/src/Backups/BackupFileInfo.cpp
+++ b/src/Backups/BackupFileInfo.cpp
@ -6,6 +6,7 @@
 #include <Common/logger_useful.h>
 #include <Common/scope_guard_safe.h>
 #include <Common/setThreadName.h>
+#include <Common/ThreadPool.h>
 #include <IO/HashingReadBuffer.h>


--- a/src/Backups/BackupFileInfo.h
+++ b/src/Backups/BackupFileInfo.h
@ -1,8 +1,9 @@
 #pragma once

 #include <Core/Types.h>
-#include <Common/ThreadPool.h>
+#include <Common/ThreadPool_fwd.h>

+namespace Poco { class Logger; }

 namespace DB
 {
--- a/src/Backups/BackupUtils.h
+++ b/src/Backups/BackupUtils.h
@ -1,7 +1,6 @@
 #pragma once

 #include <Parsers/ASTBackupQuery.h>
-#include <Common/ThreadPool.h>


 namespace DB
--- a/src/Backups/BackupsWorker.cpp
+++ b/src/Backups/BackupsWorker.cpp
@ -23,6 +23,7 @@
 #include <Common/CurrentMetrics.h>
 #include <Common/setThreadName.h>
 #include <Common/scope_guard_safe.h>
+#include <Common/ThreadPool.h>


 namespace CurrentMetrics
@ -182,8 +183,8 @@ namespace


 BackupsWorker::BackupsWorker(size_t num_backup_threads, size_t num_restore_threads, bool allow_concurrent_backups_, bool allow_concurrent_restores_)
-    : backups_thread_pool(CurrentMetrics::BackupsThreads, CurrentMetrics::BackupsThreadsActive, num_backup_threads, /* max_free_threads = */ 0, num_backup_threads)
-    , restores_thread_pool(CurrentMetrics::RestoreThreads, CurrentMetrics::RestoreThreadsActive, num_restore_threads, /* max_free_threads = */ 0, num_restore_threads)
+    : backups_thread_pool(std::make_unique<ThreadPool>(CurrentMetrics::BackupsThreads, CurrentMetrics::BackupsThreadsActive, num_backup_threads, /* max_free_threads = */ 0, num_backup_threads))
+    , restores_thread_pool(std::make_unique<ThreadPool>(CurrentMetrics::RestoreThreads, CurrentMetrics::RestoreThreadsActive, num_restore_threads, /* max_free_threads = */ 0, num_restore_threads))
    , log(&Poco::Logger::get("BackupsWorker"))
    , allow_concurrent_backups(allow_concurrent_backups_)
    , allow_concurrent_restores(allow_concurrent_restores_)
@ -248,7 +249,7 @@ OperationID BackupsWorker::startMakingBackup(const ASTPtr & query, const Context

        if (backup_settings.async)
        {
-            backups_thread_pool.scheduleOrThrowOnError(
+            backups_thread_pool->scheduleOrThrowOnError(
                [this, backup_query, backup_id, backup_name_for_logging, backup_info, backup_settings, backup_coordination, context_in_use, mutable_context]
                {
                    doBackup(
@ -435,7 +436,7 @@ void BackupsWorker::buildFileInfosForBackupEntries(const BackupPtr & backup, con
    LOG_TRACE(log, "{}", Stage::BUILDING_FILE_INFOS);
    backup_coordination->setStage(Stage::BUILDING_FILE_INFOS, "");
    backup_coordination->waitForStage(Stage::BUILDING_FILE_INFOS);
-    backup_coordination->addFileInfos(::DB::buildFileInfosForBackupEntries(backup_entries, backup->getBaseBackup(), backups_thread_pool));
+    backup_coordination->addFileInfos(::DB::buildFileInfosForBackupEntries(backup_entries, backup->getBaseBackup(), *backups_thread_pool));
 }


@ -522,7 +523,7 @@ void BackupsWorker::writeBackupEntries(BackupMutablePtr backup, BackupEntries &&
            }
        };

-        if (always_single_threaded || !backups_thread_pool.trySchedule([job] { job(true); }))
+        if (always_single_threaded || !backups_thread_pool->trySchedule([job] { job(true); }))
            job(false);
    }

@ -581,7 +582,7 @@ OperationID BackupsWorker::startRestoring(const ASTPtr & query, ContextMutablePt

        if (restore_settings.async)
        {
-            restores_thread_pool.scheduleOrThrowOnError(
+            restores_thread_pool->scheduleOrThrowOnError(
                [this, restore_query, restore_id, backup_name_for_logging, backup_info, restore_settings, restore_coordination, context_in_use]
                {
                    doRestore(
@ -716,7 +717,7 @@ void BackupsWorker::doRestore(
            }

            /// Execute the data restoring tasks.
-            restoreTablesData(restore_id, backup, std::move(data_restore_tasks), restores_thread_pool);
+            restoreTablesData(restore_id, backup, std::move(data_restore_tasks), *restores_thread_pool);

            /// We have restored everything, we need to tell other hosts (they could be waiting for it).
            restore_coordination->setStage(Stage::COMPLETED, "");
@ -941,8 +942,8 @@ void BackupsWorker::shutdown()
    if (has_active_backups_and_restores)
        LOG_INFO(log, "Waiting for {} backups and {} restores to be finished", num_active_backups, num_active_restores);

-    backups_thread_pool.wait();
-    restores_thread_pool.wait();
+    backups_thread_pool->wait();
+    restores_thread_pool->wait();

    if (has_active_backups_and_restores)
        LOG_INFO(log, "All backup and restore tasks have finished");
--- a/src/Backups/BackupsWorker.h
+++ b/src/Backups/BackupsWorker.h
@ -1,7 +1,8 @@
 #pragma once

 #include <Backups/BackupStatus.h>
-#include <Common/ThreadPool.h>
+#include <Common/ThreadPool_fwd.h>
+#include <Interpreters/Context_fwd.h>
 #include <Core/UUID.h>
 #include <Parsers/IAST_fwd.h>
 #include <unordered_map>
@ -132,8 +133,8 @@ private:
    void setNumFilesAndSize(const OperationID & id, size_t num_files, UInt64 total_size, size_t num_entries,
                            UInt64 uncompressed_size, UInt64 compressed_size, size_t num_read_files, UInt64 num_read_bytes);

-    ThreadPool backups_thread_pool;
-    ThreadPool restores_thread_pool;
+    std::unique_ptr<ThreadPool> backups_thread_pool;
+    std::unique_ptr<ThreadPool> restores_thread_pool;

    std::unordered_map<OperationID, Info> infos;
    std::condition_variable status_changed;
--- a/src/Bridge/CMakeLists.txt
+++ b/src/Bridge/CMakeLists.txt
@ -2,4 +2,4 @@ add_library (bridge
    IBridge.cpp
 )

-target_link_libraries (bridge PRIVATE daemon dbms Poco::Data Poco::Data::ODBC)
+target_link_libraries (bridge PRIVATE daemon dbms)
--- a/src/Bridge/IBridge.cpp
+++ b/src/Bridge/IBridge.cpp
@ -14,17 +14,13 @@
 #include <Server/HTTP/HTTPServer.h>
 #include <base/errnoToString.h>
 #include <base/range.h>
+#include <base/scope_guard.h>

 #include <sys/time.h>
 #include <sys/resource.h>

 #include "config.h"

-#if USE_ODBC
-#    include <Poco/Data/ODBC/Connector.h>
-#endif
-
-
 namespace DB
 {

--- a/src/BridgeHelper/IBridgeHelper.h
+++ b/src/BridgeHelper/IBridgeHelper.h
@ -5,7 +5,6 @@
 #include <Poco/Util/AbstractConfiguration.h>
 #include <Poco/Net/HTTPRequest.h>
 #include <Common/ShellCommand.h>
-#include <Common/logger_useful.h>


 namespace DB
--- a/src/BridgeHelper/XDBCBridgeHelper.h
+++ b/src/BridgeHelper/XDBCBridgeHelper.h
@ -11,7 +11,6 @@
 #include <Poco/Util/AbstractConfiguration.h>
 #include <Common/BridgeProtocolVersion.h>
 #include <Common/ShellCommand.h>
-#include <Common/logger_useful.h>
 #include <IO/ConnectionTimeouts.h>
 #include <base/range.h>
 #include <BridgeHelper/IBridgeHelper.h>
--- a/src/Client/ClientBase.cpp
+++ b/src/Client/ClientBase.cpp
@ -7,6 +7,7 @@

 #include <base/argsToConfig.h>
 #include <base/safeExit.h>
+#include <base/scope_guard.h>
 #include <Core/Block.h>
 #include <Core/Protocol.h>
 #include <Common/DateLUT.h>
@ -2219,9 +2220,6 @@ void ClientBase::runInteractive()
    LineReader lr(history_file, config().has("multiline"), query_extenders, query_delimiters);
 #endif

-    /// Enable bracketed-paste-mode so that we are able to paste multiline queries as a whole.
-    lr.enableBracketedPaste();
-
    static const std::initializer_list<std::pair<String, String>> backslash_aliases =
        {
            { "\\l", "SHOW DATABASES" },
@ -2239,7 +2237,18 @@ void ClientBase::runInteractive()

    do
    {
-        auto input = lr.readLine(prompt(), ":-] ");
+        String input;
+        {
+            /// Enable bracketed-paste-mode so that we are able to paste multiline queries as a whole.
+            /// But keep it disabled outside of query input, because it breaks password input
+            /// (e.g. if we need to reconnect and show a password prompt).
+            /// (Alternatively, we could make the password input ignore the control sequences.)
+            lr.enableBracketedPaste();
+            SCOPE_EXIT({ lr.disableBracketedPaste(); });
+
+            input = lr.readLine(prompt(), ":-] ");
+        }
+
        if (input.empty())
            break;

--- a/src/Client/Connection.cpp
+++ b/src/Client/Connection.cpp
@ -22,7 +22,8 @@
 #include <Common/StringUtils/StringUtils.h>
 #include <Common/OpenSSLHelpers.h>
 #include <Common/randomSeed.h>
-#include "Core/Block.h"
+#include <Common/logger_useful.h>
+#include <Core/Block.h>
 #include <Interpreters/ClientInfo.h>
 #include <Interpreters/OpenTelemetrySpanLog.h>
 #include <Compression/CompressionFactory.h>
--- a/src/Client/Connection.h
+++ b/src/Client/Connection.h
@ -1,6 +1,5 @@
 #pragma once

-#include <Common/logger_useful.h>

 #include <Poco/Net/StreamSocket.h>

--- a/src/Client/LineReader.h
+++ b/src/Client/LineReader.h
@ -46,7 +46,10 @@ public:
    /// clickhouse-client so that without -m flag, one can still paste multiline queries, and
    /// possibly get better pasting performance. See https://cirw.in/blog/bracketed-paste for
    /// more details.
+    /// These methods (if implemented) emit the control characters immediately, without waiting
+    /// for the next readLine() call.
    virtual void enableBracketedPaste() {}
+    virtual void disableBracketedPaste() {}

 protected:
    enum InputStatus
--- a/src/Client/LocalConnection.cpp
+++ b/src/Client/LocalConnection.cpp
@ -6,6 +6,7 @@
 #include <Processors/Executors/PushingAsyncPipelineExecutor.h>
 #include <Storages/IStorage.h>
 #include <Common/ConcurrentBoundedQueue.h>
+#include <Common/CurrentThread.h>
 #include <Core/Protocol.h>


--- a/src/Client/LocalConnection.h
+++ b/src/Client/LocalConnection.h
@ -7,6 +7,7 @@
 #include <Interpreters/Session.h>
 #include <Interpreters/ProfileEventsExt.h>
 #include <Storages/ColumnsDescription.h>
+#include <Common/CurrentThread.h>


 namespace DB
--- a/src/Client/ReplxxLineReader.cpp
+++ b/src/Client/ReplxxLineReader.cpp
@ -519,4 +519,10 @@ void ReplxxLineReader::enableBracketedPaste()
    rx.enable_bracketed_paste();
 }

+void ReplxxLineReader::disableBracketedPaste()
+{
+    bracketed_paste_enabled = false;
+    rx.disable_bracketed_paste();
+}
+
 }
--- a/src/Client/ReplxxLineReader.h
+++ b/src/Client/ReplxxLineReader.h
@ -19,6 +19,7 @@ public:
    ~ReplxxLineReader() override;

    void enableBracketedPaste() override;
+    void disableBracketedPaste() override;

    /// If highlight is on, we will set a flag to denote whether the last token is a delimiter.
    /// This is useful to determine the behavior of <ENTER> key when multiline is enabled.
--- a/src/Common/AsynchronousMetrics.cpp
+++ b/src/Common/AsynchronousMetrics.cpp
@ -3,6 +3,7 @@
 #include <Common/setThreadName.h>
 #include <Common/CurrentMetrics.h>
 #include <Common/filesystemHelpers.h>
+#include <Common/logger_useful.h>
 #include <IO/UncompressedCache.h>
 #include <IO/MMappedFileCache.h>
 #include <IO/ReadHelpers.h>
--- a/src/Common/CacheBase.h
+++ b/src/Common/CacheBase.h
@ -12,7 +12,6 @@
 #include <mutex>
 #include <unordered_map>

-#include <Common/logger_useful.h>
 #include <base/defines.h>


--- a/src/Common/Config/ConfigProcessor.cpp
+++ b/src/Common/Config/ConfigProcessor.cpp
@ -21,6 +21,7 @@
 #include <Common/Exception.h>
 #include <Common/getResource.h>
 #include <Common/XMLUtils.h>
+#include <Common/logger_useful.h>
 #include <base/errnoToString.h>
 #include <base/sort.h>
 #include <IO/WriteBufferFromString.h>
--- a/src/Common/Config/ConfigProcessor.h
+++ b/src/Common/Config/ConfigProcessor.h
@ -16,9 +16,10 @@
 #include <Poco/DirectoryIterator.h>
 #include <Poco/ConsoleChannel.h>
 #include <Poco/Util/AbstractConfiguration.h>
-#include <Common/logger_useful.h>


+namespace Poco { class Logger; }
+
 namespace zkutil
 {
    class ZooKeeperNodeCache;
--- a/src/Common/Config/YAMLParser.h
+++ b/src/Common/Config/YAMLParser.h
@ -2,11 +2,11 @@

 #include "config.h"

-#include <string>
-
+#include <Common/ErrorCodes.h>
+#include <Common/Exception.h>
+#include <base/types.h>
 #include <Poco/DOM/Document.h>
-#include "Poco/DOM/AutoPtr.h"
-#include <Common/logger_useful.h>
+#include <Poco/DOM/AutoPtr.h>

 #if USE_YAML_CPP

--- a/src/Common/Config/configReadClient.cpp
+++ b/src/Common/Config/configReadClient.cpp
@ -4,6 +4,7 @@
 #include "ConfigProcessor.h"
 #include <filesystem>
 #include <iostream>
+#include <base/types.h>

 namespace fs = std::filesystem;

--- a/src/Common/CurrentThread.cpp
+++ b/src/Common/CurrentThread.cpp
@ -90,7 +90,7 @@ void CurrentThread::attachInternalTextLogsQueue(const std::shared_ptr<InternalTe
 }


-ThreadGroupStatusPtr CurrentThread::getGroup()
+ThreadGroupPtr CurrentThread::getGroup()
 {
    if (unlikely(!current_thread))
        return nullptr;
--- a/src/Common/CurrentThread.h
+++ b/src/Common/CurrentThread.h
@ -39,7 +39,7 @@ public:
    static ThreadStatus & get();

    /// Group to which belongs current thread
-    static ThreadGroupStatusPtr getGroup();
+    static ThreadGroupPtr getGroup();

    /// A logs queue used by TCPHandler to pass logs to a client
    static void attachInternalTextLogsQueue(const std::shared_ptr<InternalTextLogsQueue> & logs_queue,
@ -69,9 +69,9 @@ public:

    /// You must call one of these methods when create a query child thread:
    /// Add current thread to a group associated with the thread group
-    static void attachToGroup(const ThreadGroupStatusPtr & thread_group);
+    static void attachToGroup(const ThreadGroupPtr & thread_group);
    /// Is useful for a ThreadPool tasks
-    static void attachToGroupIfDetached(const ThreadGroupStatusPtr & thread_group);
+    static void attachToGroupIfDetached(const ThreadGroupPtr & thread_group);

    /// Non-master threads call this method in destructor automatically
    static void detachFromGroupIfNotDetached();
--- a/src/Common/DNSResolver.cpp
+++ b/src/Common/DNSResolver.cpp
@ -3,6 +3,7 @@
 #include <Common/Exception.h>
 #include <Common/ProfileEvents.h>
 #include <Common/thread_local_rng.h>
+#include <Common/logger_useful.h>
 #include <Core/Names.h>
 #include <base/types.h>
 #include <Poco/Net/IPAddress.h>
--- a/src/Common/DNSResolver.h
+++ b/src/Common/DNSResolver.h
@ -5,9 +5,10 @@
 #include <base/types.h>
 #include <Core/Names.h>
 #include <boost/noncopyable.hpp>
-#include <Common/logger_useful.h>


+namespace Poco { class Logger; }
+
 namespace DB
 {

--- a/src/Common/ErrorHandlers.h
+++ b/src/Common/ErrorHandlers.h
@ -1,7 +1,6 @@
 #pragma once

 #include <Poco/ErrorHandler.h>
-#include <Common/logger_useful.h>
 #include <Common/Exception.h>


--- a/src/Common/Exception.h
+++ b/src/Common/Exception.h
@ -4,7 +4,6 @@
 #include <vector>
 #include <memory>

-#include <Poco/Version.h>
 #include <Poco/Exception.h>

 #include <base/defines.h>
--- a/src/Common/FileChecker.cpp
+++ b/src/Common/FileChecker.cpp
@ -1,5 +1,7 @@
 #include <Common/FileChecker.h>
 #include <Common/escapeForFileName.h>
+#include <Common/logger_useful.h>
+#include <Common/ErrorCodes.h>
 #include <Disks/IDisk.h>
 #include <IO/WriteBufferFromFile.h>
 #include <IO/ReadBufferFromFile.h>
@ -25,7 +27,9 @@ FileChecker::FileChecker(const String & file_info_path_) : FileChecker(nullptr,
 {
 }

-FileChecker::FileChecker(DiskPtr disk_, const String & file_info_path_) : disk(std::move(disk_))
+FileChecker::FileChecker(DiskPtr disk_, const String & file_info_path_)
+    : disk(std::move(disk_))
+    , log(&Poco::Logger::get("FileChecker"))
 {
    setPath(file_info_path_);
    try
--- a/src/Common/FileChecker.h
+++ b/src/Common/FileChecker.h
@ -1,8 +1,10 @@
 #pragma once

-#include <Common/logger_useful.h>
 #include <Storages/CheckResults.h>
+#include <map>
+#include <base/types.h>

+namespace Poco { class Logger; }

 namespace DB
 {
@ -46,7 +48,7 @@ private:
    size_t getRealFileSize(const String & path_) const;

    const DiskPtr disk;
-    const Poco::Logger * log = &Poco::Logger::get("FileChecker");
+    const Poco::Logger * log;

    String files_info_path;
    std::map<String, size_t> map;
--- a/src/Common/LRUCachePolicy.h
+++ b/src/Common/LRUCachePolicy.h
@ -5,8 +5,6 @@
 #include <list>
 #include <unordered_map>

-#include <Common/logger_useful.h>
-
 namespace DB
 {
 /// Cache policy LRU evicts entries which are not used for a long time.
@ -174,7 +172,7 @@ private:
            auto it = cells.find(key);
            if (it == cells.end())
            {
-                LOG_ERROR(&Poco::Logger::get("LRUCache"), "LRUCache became inconsistent. There must be a bug in it.");
+                // Queue became inconsistent
                abort();
            }

@ -192,7 +190,7 @@ private:

        if (current_size_in_bytes > (1ull << 63))
        {
-            LOG_ERROR(&Poco::Logger::get("LRUCache"), "LRUCache became inconsistent. There must be a bug in it.");
+            // Queue became inconsistent
            abort();
        }
    }
--- a/src/Common/Macros.cpp
+++ b/src/Common/Macros.cpp
@ -1,8 +1,9 @@
 #include <Poco/Util/AbstractConfiguration.h>
 #include <Common/Macros.h>
 #include <Common/Exception.h>
-#include <IO/WriteHelpers.h>
 #include <Common/logger_useful.h>
+#include <Core/ServerUUID.h>
+#include <IO/WriteHelpers.h>


 namespace DB
@ -11,6 +12,8 @@ namespace DB
 namespace ErrorCodes
 {
    extern const int SYNTAX_ERROR;
+    extern const int BAD_ARGUMENTS;
+    extern const int NO_ELEMENTS_IN_CONFIG;
 }

 Macros::Macros(const Poco::Util::AbstractConfiguration & config, const String & root_key, Poco::Logger * log)
@ -95,7 +98,7 @@ String Macros::expand(const String & s,
        else if (macro_name == "uuid" && !info.expand_special_macros_only)
        {
            if (info.table_id.uuid == UUIDHelpers::Nil)
-                throw Exception(ErrorCodes::SYNTAX_ERROR, "Macro 'uuid' and empty arguments of ReplicatedMergeTree "
+                throw Exception(ErrorCodes::BAD_ARGUMENTS, "Macro 'uuid' and empty arguments of ReplicatedMergeTree "
                                "are supported only for ON CLUSTER queries with Atomic database engine");
            /// For ON CLUSTER queries we don't want to require all macros definitions in initiator's config.
            /// However, initiator must check that for cross-replication cluster zookeeper_path does not contain {uuid} macro.
@ -105,6 +108,15 @@ String Macros::expand(const String & s,
            res += toString(info.table_id.uuid);
            info.expanded_uuid = true;
        }
+        else if (macro_name == "server_uuid")
+        {
+            auto uuid = ServerUUID::get();
+            if (UUIDHelpers::Nil == uuid)
+                throw Exception(ErrorCodes::BAD_ARGUMENTS,
+                    "Macro {server_uuid} expanded to zero, which means the UUID is not initialized (most likely it's not a server application)");
+            res += toString(uuid);
+            info.expanded_other = true;
+        }
        else if (info.shard && macro_name == "shard")
        {
            res += *info.shard;
@ -125,7 +137,7 @@ String Macros::expand(const String & s,
            info.has_unknown = true;
        }
        else
-            throw Exception(ErrorCodes::SYNTAX_ERROR, "No macro '{}' in config while processing substitutions in "
+            throw Exception(ErrorCodes::NO_ELEMENTS_IN_CONFIG, "No macro '{}' in config while processing substitutions in "
                            "'{}' at '{}' or macro is not supported here", macro_name, s, toString(begin));

        pos = end + 1;
@ -142,7 +154,7 @@ String Macros::getValue(const String & key) const
 {
    if (auto it = macros.find(key); it != macros.end())
        return it->second;
-    throw Exception(ErrorCodes::SYNTAX_ERROR, "No macro {} in config", key);
+    throw Exception(ErrorCodes::NO_ELEMENTS_IN_CONFIG, "No macro {} in config", key);
 }


--- a/src/Common/MemoryTracker.cpp
+++ b/src/Common/MemoryTracker.cpp
@ -82,26 +82,6 @@ inline std::string_view toDescription(OvercommitResult result)
    }
 }

-inline void debugLogBigAllocationWithoutCheck(Int64 size [[maybe_unused]])
-{
-    /// Big allocations through allocNoThrow (without checking memory limits) may easily lead to OOM (and it's hard to debug).
-    /// Let's find them.
-#ifdef ABORT_ON_LOGICAL_ERROR
-    if (size < 0)
-        return;
-
-    constexpr Int64 threshold = 16 * 1024 * 1024;   /// The choice is arbitrary (maybe we should decrease it)
-    if (size < threshold)
-        return;
-
-    MemoryTrackerBlockerInThread blocker;
-    LOG_TEST(&Poco::Logger::get("MemoryTracker"), "Too big allocation ({} bytes) without checking memory limits, "
-                                                   "it may lead to OOM. Stack trace: {}", size, StackTrace().toString());
-#else
-    return;     /// Avoid trash logging in release builds
-#endif
-}
-
 }

 namespace ProfileEvents
@ -175,6 +155,26 @@ void MemoryTracker::injectFault() const
        description ? description : "");
 }

+void MemoryTracker::debugLogBigAllocationWithoutCheck(Int64 size [[maybe_unused]])
+{
+    /// Big allocations through allocNoThrow (without checking memory limits) may easily lead to OOM (and it's hard to debug).
+    /// Let's find them.
+#ifdef ABORT_ON_LOGICAL_ERROR
+    if (size < 0)
+        return;
+
+    constexpr Int64 threshold = 16 * 1024 * 1024;   /// The choice is arbitrary (maybe we should decrease it)
+    if (size < threshold)
+        return;
+
+    MemoryTrackerBlockerInThread blocker(VariableContext::Global);
+    LOG_TEST(&Poco::Logger::get("MemoryTracker"), "Too big allocation ({} bytes) without checking memory limits, "
+                                                   "it may lead to OOM. Stack trace: {}", size, StackTrace().toString());
+#else
+    return;     /// Avoid trash logging in release builds
+#endif
+}
+
 void MemoryTracker::allocImpl(Int64 size, bool throw_if_memory_exceeded, MemoryTracker * query_tracker)
 {
    if (size < 0)
--- a/src/Common/MemoryTracker.h
+++ b/src/Common/MemoryTracker.h
@ -215,6 +215,8 @@ public:

    /// Prints info about peak memory consumption into log.
    void logPeakMemoryUsage();
+
+    void debugLogBigAllocationWithoutCheck(Int64 size [[maybe_unused]]);
 };

 extern MemoryTracker total_memory_tracker;
--- a/src/Common/OvercommitTracker.h
+++ b/src/Common/OvercommitTracker.h
@ -1,7 +1,7 @@
 #pragma once

-#include <Common/logger_useful.h>
 #include <base/types.h>
+#include <Core/Types.h>
 #include <boost/core/noncopyable.hpp>
 #include <Poco/Logger.h>
 #include <cassert>
--- a/src/Common/PoolBase.h
+++ b/src/Common/PoolBase.h
@ -144,12 +144,17 @@ public:
                return Entry(*items.back());
            }

-            LOG_INFO(log, "No free connections in pool. Waiting.");
-
            if (timeout < 0)
+            {
+                LOG_INFO(log, "No free connections in pool. Waiting undefinitelly.");
                available.wait(lock);
+            }
            else
-                available.wait_for(lock, std::chrono::microseconds(timeout));
+            {
+                auto timeout_ms = std::chrono::microseconds(timeout);
+                LOG_INFO(log, "No free connections in pool. Waiting {} ms.", timeout_ms.count());
+                available.wait_for(lock, timeout_ms);
+            }
        }
    }

--- a/src/Common/SLRUCachePolicy.h
+++ b/src/Common/SLRUCachePolicy.h
@ -5,8 +5,6 @@
 #include <list>
 #include <unordered_map>

-#include <Common/logger_useful.h>
-
 namespace DB
 {

@ -236,7 +234,7 @@ private:
            auto it = cells.find(key);
            if (it == cells.end())
            {
-                LOG_ERROR(&Poco::Logger::get("SLRUCache"), "SLRUCache became inconsistent. There must be a bug in it.");
+                // Queue became inconsistent
                abort();
            }

@ -264,7 +262,7 @@ private:

        if (current_size_in_bytes > (1ull << 63))
        {
-            LOG_ERROR(&Poco::Logger::get("SLRUCache"), "SLRUCache became inconsistent. There must be a bug in it.");
+            // Queue became inconsistent
            abort();
        }
    }
--- a/src/Common/SpaceSaving.h
+++ b/src/Common/SpaceSaving.h
@ -51,6 +51,9 @@ struct SpaceSavingArena<StringRef>
 {
    StringRef emplace(StringRef key)
    {
+        if (!key.data)
+            return key;
+
        return copyStringInArena(arena, key);
    }

--- a/src/Common/SystemLogBase.cpp
+++ b/src/Common/SystemLogBase.cpp
@ -18,6 +18,7 @@

 #include <Common/MemoryTrackerBlockerInThread.h>
 #include <Common/SystemLogBase.h>
+#include <Common/ThreadPool.h>

 #include <Common/logger_useful.h>
 #include <base/scope_guard.h>
@ -35,20 +36,18 @@ namespace
    constexpr size_t DBMS_SYSTEM_LOG_QUEUE_SIZE = 1048576;
 }

+ISystemLog::~ISystemLog() = default;
+
 void ISystemLog::stopFlushThread()
 {
    {
        std::lock_guard lock(mutex);

-        if (!saving_thread.joinable())
-        {
+        if (!saving_thread || !saving_thread->joinable())
            return;
-        }

        if (is_shutdown)
-        {
            return;
-        }

        is_shutdown = true;

@ -56,13 +55,13 @@ void ISystemLog::stopFlushThread()
        flush_event.notify_all();
    }

-    saving_thread.join();
+    saving_thread->join();
 }

 void ISystemLog::startup()
 {
    std::lock_guard lock(mutex);
-    saving_thread = ThreadFromGlobalPool([this] { savingThreadFunction(); });
+    saving_thread = std::make_unique<ThreadFromGlobalPool>([this] { savingThreadFunction(); });
 }

 static thread_local bool recursive_add_call = false;
--- a/src/Common/SystemLogBase.h
+++ b/src/Common/SystemLogBase.h
@ -10,7 +10,7 @@
 #include <Interpreters/Context_fwd.h>
 #include <Parsers/IAST_fwd.h>
 #include <Storages/IStorage_fwd.h>
-#include <Common/ThreadPool.h>
+#include <Common/ThreadPool_fwd.h>

 #define SYSTEM_LOG_ELEMENTS(M) \
    M(AsynchronousMetricLogElement) \
@ -60,12 +60,12 @@ public:
    /// Stop the background flush thread before destructor. No more data will be written.
    virtual void shutdown() = 0;

-    virtual ~ISystemLog() = default;
+    virtual ~ISystemLog();

    virtual void savingThreadFunction() = 0;

 protected:
-    ThreadFromGlobalPool saving_thread;
+    std::unique_ptr<ThreadFromGlobalPool> saving_thread;

    /// Data shared between callers of add()/flush()/shutdown(), and the saving thread
    std::mutex mutex;
--- a/src/Common/ThreadPool.h
+++ b/src/Common/ThreadPool.h
@ -17,6 +17,7 @@
 #include <Common/ThreadStatus.h>
 #include <Common/OpenTelemetryTraceContext.h>
 #include <Common/CurrentMetrics.h>
+#include <Common/ThreadPool_fwd.h>
 #include <base/scope_guard.h>

 /** Very simple thread pool similar to boost::threadpool.
--- a/src/Common/ThreadPool_fwd.h
+++ b/src/Common/ThreadPool_fwd.h
@ -0,0 +1,13 @@
+#pragma once
+
+template <typename Thread>
+class ThreadPoolImpl;
+
+template <bool propagate_opentelemetry_context>
+class ThreadFromGlobalPoolImpl;
+
+using ThreadFromGlobalPoolNoTracingContextPropagation = ThreadFromGlobalPoolImpl<false>;
+
+using ThreadFromGlobalPool = ThreadFromGlobalPoolImpl<true>;
+
+using ThreadPool = ThreadPoolImpl<ThreadFromGlobalPoolNoTracingContextPropagation>;
--- a/src/Common/ThreadProfileEvents.cpp
+++ b/src/Common/ThreadProfileEvents.cpp
@ -23,6 +23,7 @@
 #include <boost/algorithm/string/split.hpp>

 #include <base/errnoToString.h>
+#include <Common/logger_useful.h>


 namespace ProfileEvents
--- a/src/Common/ThreadProfileEvents.h
+++ b/src/Common/ThreadProfileEvents.h
@ -2,11 +2,13 @@

 #include <base/types.h>
 #include <base/getThreadId.h>
+#include <base/defines.h>
+#include <Common/Exception.h>
 #include <Common/ProfileEvents.h>
 #include <sys/time.h>
 #include <sys/resource.h>
 #include <pthread.h>
-#include <Common/logger_useful.h>
+#include <boost/noncopyable.hpp>


 #if defined(OS_LINUX)
--- a/src/Common/ThreadStatus.cpp
+++ b/src/Common/ThreadStatus.cpp
@ -2,6 +2,8 @@
 #include <Common/ThreadProfileEvents.h>
 #include <Common/QueryProfiler.h>
 #include <Common/ThreadStatus.h>
+#include <Common/CurrentThread.h>
+#include <Common/logger_useful.h>
 #include <base/errnoToString.h>
 #include <Interpreters/Context.h>

@ -61,7 +63,7 @@ static thread_local ThreadStack alt_stack;
 static thread_local bool has_alt_stack = false;
 #endif

-ThreadGroupStatus::ThreadGroupStatus()
+ThreadGroup::ThreadGroup()
    : master_thread_id(CurrentThread::get().thread_id)
 {}

@ -119,7 +121,7 @@ ThreadStatus::ThreadStatus()
 #endif
 }

-ThreadGroupStatusPtr ThreadStatus::getThreadGroup() const
+ThreadGroupPtr ThreadStatus::getThreadGroup() const
 {
    return thread_group;
 }
@ -139,7 +141,7 @@ ContextPtr ThreadStatus::getGlobalContext() const
    return global_context.lock();
 }

-void ThreadGroupStatus::attachInternalTextLogsQueue(const InternalTextLogsQueuePtr & logs_queue, LogsLevel logs_level)
+void ThreadGroup::attachInternalTextLogsQueue(const InternalTextLogsQueuePtr & logs_queue, LogsLevel logs_level)
 {
    std::lock_guard lock(mutex);
    shared_data.logs_queue_ptr = logs_queue;
--- a/src/Common/ThreadStatus.h
+++ b/src/Common/ThreadStatus.h
@ -41,7 +41,6 @@ class TaskStatsInfoGetter;
 class InternalTextLogsQueue;
 struct ViewRuntimeData;
 class QueryViewsLog;
-class ThreadGroupSwitcher;
 using InternalTextLogsQueuePtr = std::shared_ptr<InternalTextLogsQueue>;
 using InternalTextLogsQueueWeakPtr = std::weak_ptr<InternalTextLogsQueue>;

@ -58,15 +57,15 @@ using ThreadStatusPtr = ThreadStatus *;
  * Create via CurrentThread::initializeQuery (for queries) or directly (for various background tasks).
  * Use via CurrentThread::getGroup.
  */
-class ThreadGroupStatus;
-using ThreadGroupStatusPtr = std::shared_ptr<ThreadGroupStatus>;
+class ThreadGroup;
+using ThreadGroupPtr = std::shared_ptr<ThreadGroup>;

-class ThreadGroupStatus
+class ThreadGroup
 {
 public:
-    ThreadGroupStatus();
+    ThreadGroup();
    using FatalErrorCallback = std::function<void()>;
-    ThreadGroupStatus(ContextPtr query_context_, FatalErrorCallback fatal_error_callback_ = {});
+    ThreadGroup(ContextPtr query_context_, FatalErrorCallback fatal_error_callback_ = {});

    /// The first thread created this thread group
    const UInt64 master_thread_id;
@ -104,9 +103,9 @@ public:
    void attachInternalProfileEventsQueue(const InternalProfileEventsQueuePtr & profile_queue);

    /// When new query starts, new thread group is created for it, current thread becomes master thread of the query
-    static ThreadGroupStatusPtr createForQuery(ContextPtr query_context_, FatalErrorCallback fatal_error_callback_ = {});
+    static ThreadGroupPtr createForQuery(ContextPtr query_context_, FatalErrorCallback fatal_error_callback_ = {});

-    static ThreadGroupStatusPtr createForBackgroundProcess(ContextPtr storage_context);
+    static ThreadGroupPtr createForBackgroundProcess(ContextPtr storage_context);

    std::vector<UInt64> getInvolvedThreadIds() const;
    void linkThread(UInt64 thread_it);
@ -120,6 +119,21 @@ private:
    std::unordered_set<UInt64> thread_ids;
 };

+/**
+ * Since merge is executed with multiple threads, this class
+ * switches the parent MemoryTracker as part of the thread group to account all the memory used.
+ */
+class ThreadGroupSwitcher : private boost::noncopyable
+{
+public:
+    explicit ThreadGroupSwitcher(ThreadGroupPtr thread_group);
+    ~ThreadGroupSwitcher();
+
+private:
+    ThreadGroupPtr prev_thread_group;
+};
+
+
 /**
 * We use **constinit** here to tell the compiler the current_thread variable is initialized.
 * If we didn't help the compiler, then it would most likely add a check before every use of the variable to initialize it if needed.
@ -163,7 +177,7 @@ public:

 private:
    /// Group of threads, to which this thread attached
-    ThreadGroupStatusPtr thread_group;
+    ThreadGroupPtr thread_group;

    /// Is set once
    ContextWeakPtr global_context;
@ -174,7 +188,7 @@ private:
    using FatalErrorCallback = std::function<void()>;
    FatalErrorCallback fatal_error_callback;

-    ThreadGroupStatus::SharedData local_data;
+    ThreadGroup::SharedData local_data;

    bool performance_counters_finalized = false;

@ -215,7 +229,7 @@ public:
    ThreadStatus();
    ~ThreadStatus();

-    ThreadGroupStatusPtr getThreadGroup() const;
+    ThreadGroupPtr getThreadGroup() const;

    const String & getQueryId() const;

@ -239,7 +253,7 @@ public:
    void setInternalThread();

    /// Attaches slave thread to existing thread group
-    void attachToGroup(const ThreadGroupStatusPtr & thread_group_, bool check_detached = true);
+    void attachToGroup(const ThreadGroupPtr & thread_group_, bool check_detached = true);

    /// Detaches thread from the thread group and the query, dumps performance counters if they have not been dumped
    void detachFromGroup();
@ -287,7 +301,7 @@ private:

    void logToQueryThreadLog(QueryThreadLog & thread_log, const String & current_database);

-    void attachToGroupImpl(const ThreadGroupStatusPtr & thread_group_);
+    void attachToGroupImpl(const ThreadGroupPtr & thread_group_);
 };

 /**
--- a/src/Common/ZooKeeper/Common.h
+++ b/src/Common/ZooKeeper/Common.h
@ -3,12 +3,10 @@
 #include <functional>

 #include <Common/ZooKeeper/ZooKeeper.h>
-#include <Common/ZooKeeper/ZooKeeperWithFaultInjection.h>

 namespace zkutil
 {

 using GetZooKeeper = std::function<ZooKeeperPtr()>;
-using GetZooKeeperWithFaultInjection = std::function<Coordination::ZooKeeperWithFaultInjection::Ptr()>;

 }
--- a/src/Common/ZooKeeper/IKeeper.h
+++ b/src/Common/ZooKeeper/IKeeper.h
@ -273,7 +273,7 @@ struct SetRequest : virtual Request
    void addRootPath(const String & root_path) override;
    String getPath() const override { return path; }

-    size_t bytesSize() const override { return data.size() + data.size() + sizeof(version); }
+    size_t bytesSize() const override { return path.size() + data.size() + sizeof(version); }
 };

 struct SetResponse : virtual Response
--- a/src/Common/ZooKeeper/ZooKeeper.cpp
+++ b/src/Common/ZooKeeper/ZooKeeper.cpp
@ -15,6 +15,7 @@
 #include "Common/ZooKeeper/IKeeper.h"
 #include <Common/StringUtils/StringUtils.h>
 #include <Common/Exception.h>
+#include <Common/logger_useful.h>

 #include <Poco/Net/NetException.h>
 #include <Poco/Net/DNS.h>
--- a/src/Common/ZooKeeper/ZooKeeper.h
+++ b/src/Common/ZooKeeper/ZooKeeper.h
@ -7,7 +7,6 @@
 #include <memory>
 #include <mutex>
 #include <string>
-#include <Common/logger_useful.h>
 #include <Common/ProfileEvents.h>
 #include <Common/CurrentMetrics.h>
 #include <Common/Stopwatch.h>
--- a/src/Common/ZooKeeper/ZooKeeperImpl.cpp
+++ b/src/Common/ZooKeeper/ZooKeeperImpl.cpp
@ -669,8 +669,8 @@ void ZooKeeper::receiveThread()
                    earliest_operation = operations.begin()->second;
                    auto earliest_operation_deadline = earliest_operation->time + std::chrono::microseconds(args.operation_timeout_ms * 1000);
                    if (now > earliest_operation_deadline)
-                        throw Exception(Error::ZOPERATIONTIMEOUT, "Operation timeout (deadline already expired) for path: {}",
-                                        earliest_operation->request->getPath());
+                        throw Exception(Error::ZOPERATIONTIMEOUT, "Operation timeout (deadline of {} ms already expired) for path: {}",
+                                        args.operation_timeout_ms, earliest_operation->request->getPath());
                    max_wait_us = std::chrono::duration_cast<std::chrono::microseconds>(earliest_operation_deadline - now).count();
                }
            }
@ -687,12 +687,12 @@ void ZooKeeper::receiveThread()
            {
                if (earliest_operation)
                {
-                    throw Exception(Error::ZOPERATIONTIMEOUT, "Operation timeout (no response) for request {} for path: {}",
-                        toString(earliest_operation->request->getOpNum()), earliest_operation->request->getPath());
+                    throw Exception(Error::ZOPERATIONTIMEOUT, "Operation timeout (no response in {} ms) for request {} for path: {}",
+                        args.operation_timeout_ms, toString(earliest_operation->request->getOpNum()), earliest_operation->request->getPath());
                }
                waited_us += max_wait_us;
                if (waited_us >= args.session_timeout_ms * 1000)
-                    throw Exception(Error::ZOPERATIONTIMEOUT, "Nothing is received in session timeout");
+                    throw Exception(Error::ZOPERATIONTIMEOUT, "Nothing is received in session timeout of {} ms", args.session_timeout_ms);

            }

@ -1080,7 +1080,7 @@ void ZooKeeper::pushRequest(RequestInfo && info)
            if (requests_queue.isFinished())
                throw Exception(Error::ZSESSIONEXPIRED, "Session expired");

-            throw Exception(Error::ZOPERATIONTIMEOUT, "Cannot push request to queue within operation timeout");
+            throw Exception(Error::ZOPERATIONTIMEOUT, "Cannot push request to queue within operation timeout of {} ms", args.operation_timeout_ms);
        }
    }
    catch (...)
@ -1332,7 +1332,7 @@ void ZooKeeper::close()
    request_info.request = std::make_shared<ZooKeeperCloseRequest>(std::move(request));

    if (!requests_queue.tryPush(std::move(request_info), args.operation_timeout_ms))
-        throw Exception(Error::ZOPERATIONTIMEOUT, "Cannot push close request to queue within operation timeout");
+        throw Exception(Error::ZOPERATIONTIMEOUT, "Cannot push close request to queue within operation timeout of {} ms", args.operation_timeout_ms);

    ProfileEvents::increment(ProfileEvents::ZooKeeperClose);
 }
--- a/src/Common/ZooKeeper/ZooKeeperLock.cpp
+++ b/src/Common/ZooKeeper/ZooKeeperLock.cpp
@ -1,4 +1,6 @@
 #include <Common/ZooKeeper/ZooKeeperLock.h>
+#include <Common/logger_useful.h>
+#include <Common/ErrorCodes.h>
 #include <filesystem>

 namespace DB
--- a/src/Common/ZooKeeper/ZooKeeperLock.h
+++ b/src/Common/ZooKeeper/ZooKeeperLock.h
@ -3,7 +3,8 @@
 #include <Common/ZooKeeper/KeeperException.h>
 #include <memory>
 #include <string>
-#include <Common/logger_useful.h>
+
+namespace Poco { class Logger; }

 namespace zkutil
 {
--- a/src/Common/ZooKeeper/ZooKeeperWithFaultInjection.h
+++ b/src/Common/ZooKeeper/ZooKeeperWithFaultInjection.h
@ -4,6 +4,7 @@
 #include <Common/ZooKeeper/Types.h>
 #include <Common/ZooKeeper/ZooKeeper.h>
 #include <Common/ZooKeeper/ZooKeeperCommon.h>
+#include <Common/logger_useful.h>
 #include <Common/randomSeed.h>

 namespace DB
--- a/src/Common/mysqlxx/PoolFactory.cpp
+++ b/src/Common/mysqlxx/PoolFactory.cpp
@ -23,12 +23,6 @@ PoolWithFailover PoolFactory::get(const std::string & config_name, unsigned defa
    return get(Poco::Util::Application::instance().config(), config_name, default_connections, max_connections, max_tries);
 }

-/// Duplicate of code from StringUtils.h. Copied here for less dependencies.
-static bool startsWith(const std::string & s, const char * prefix)
-{
-    return s.size() >= strlen(prefix) && 0 == memcmp(s.data(), prefix, strlen(prefix));
-}
-
 static std::string getPoolEntryName(const Poco::Util::AbstractConfiguration & config,
        const std::string & config_name)
 {
@ -55,7 +49,7 @@ static std::string getPoolEntryName(const Poco::Util::AbstractConfiguration & co
        for (const auto & replica_config_key : replica_keys)
        {
            /// There could be another elements in the same level in configuration file, like "user", "port"...
-            if (startsWith(replica_config_key, "replica"))
+            if (replica_config_key.starts_with("replica"))
            {
                std::string replica_name = config_name + "." + replica_config_key;
                std::string tmp_host = config.getString(replica_name + ".host", host);
--- a/src/Common/scope_guard_safe.h
+++ b/src/Common/scope_guard_safe.h
@ -1,7 +1,6 @@
 #pragma once

 #include <base/scope_guard.h>
-#include <Common/logger_useful.h>
 #include <Common/LockMemoryExceptionInThread.h>

 /// Same as SCOPE_EXIT() but block the MEMORY_LIMIT_EXCEEDED errors.
--- a/src/Coordination/Changelog.h
+++ b/src/Coordination/Changelog.h
@ -10,6 +10,7 @@
 #include <libnuraft/nuraft.hxx>
 #include <libnuraft/raft_server.hxx>
 #include <Common/ConcurrentBoundedQueue.h>
+#include <Common/ThreadPool.h>

 namespace DB
 {
--- a/src/Coordination/KeeperDispatcher.cpp
+++ b/src/Coordination/KeeperDispatcher.cpp
@ -10,6 +10,7 @@
 #include <Common/checkStackSize.h>
 #include <Common/CurrentMetrics.h>
 #include <Common/ProfileEvents.h>
+#include <Common/logger_useful.h>

 #include <future>
 #include <chrono>
--- a/src/Coordination/KeeperDispatcher.h
+++ b/src/Coordination/KeeperDispatcher.h
@ -9,7 +9,6 @@
 #include <Common/ConcurrentBoundedQueue.h>
 #include <Poco/Util/AbstractConfiguration.h>
 #include <Common/Exception.h>
-#include <Common/logger_useful.h>
 #include <functional>
 #include <Coordination/KeeperServer.h>
 #include <Coordination/CoordinationSettings.h>
--- a/src/Coordination/KeeperLogStore.cpp
+++ b/src/Coordination/KeeperLogStore.cpp
@ -1,5 +1,6 @@
 #include <Coordination/KeeperLogStore.h>
 #include <IO/CompressionMethod.h>
+#include <Common/logger_useful.h>

 namespace DB
 {
--- a/src/Coordination/KeeperLogStore.h
+++ b/src/Coordination/KeeperLogStore.h
@ -4,7 +4,6 @@
 #include <mutex>
 #include <Core/Types.h>
 #include <Coordination/Changelog.h>
-#include <Common/logger_useful.h>
 #include <base/defines.h>

 namespace DB
--- a/src/Coordination/KeeperSnapshotManagerS3.h
+++ b/src/Coordination/KeeperSnapshotManagerS3.h
@ -9,7 +9,6 @@
 #if USE_AWS_S3
 #include <Common/ConcurrentBoundedQueue.h>
 #include <Common/ThreadPool.h>
-#include <Common/logger_useful.h>

 #include <string>
 #endif
--- a/src/Coordination/KeeperStateMachine.cpp
+++ b/src/Coordination/KeeperStateMachine.cpp
@ -8,9 +8,10 @@
 #include <Coordination/WriteBufferFromNuraftBuffer.h>
 #include <IO/ReadHelpers.h>
 #include <sys/mman.h>
-#include "Common/ZooKeeper/ZooKeeperCommon.h"
+#include <Common/ZooKeeper/ZooKeeperCommon.h>
 #include <Common/ZooKeeper/ZooKeeperIO.h>
 #include <Common/ProfileEvents.h>
+#include <Common/logger_useful.h>
 #include "Coordination/KeeperStorage.h"


--- a/src/Coordination/KeeperStateMachine.h
+++ b/src/Coordination/KeeperStateMachine.h
@ -8,7 +8,6 @@

 #include <libnuraft/nuraft.hxx>
 #include <Common/ConcurrentBoundedQueue.h>
-#include <Common/logger_useful.h>


 namespace DB
--- a/src/Coordination/KeeperStateManager.cpp
+++ b/src/Coordination/KeeperStateManager.cpp
@ -7,6 +7,7 @@
 #include <Common/isLocalAddress.h>
 #include <IO/ReadHelpers.h>
 #include <Common/getMultipleKeysFromConfig.h>
+#include <Common/logger_useful.h>

 namespace DB
 {
--- a/src/Coordination/KeeperStorage.cpp
+++ b/src/Coordination/KeeperStorage.cpp
@ -12,6 +12,7 @@
 #include <Common/StringUtils/StringUtils.h>
 #include <Common/ZooKeeper/IKeeper.h>
 #include <base/hex.h>
+#include <base/scope_guard.h>
 #include <Common/logger_useful.h>
 #include <Common/setThreadName.h>
 #include <Common/LockMemoryExceptionInThread.h>
@ -61,16 +62,10 @@ String getSHA1(const String & userdata)
    return String{digest_id.begin(), digest_id.end()};
 }

-String generateDigest(const String & userdata)
-{
-    std::vector<String> user_password;
-    boost::split(user_password, userdata, [](char character) { return character == ':'; });
-    return user_password[0] + ":" + base64Encode(getSHA1(userdata));
-}
-
 bool fixupACL(
    const std::vector<Coordination::ACL> & request_acls,
-    const std::vector<KeeperStorage::AuthID> & current_ids,
+    int64_t session_id,
+    const KeeperStorage::UncommittedState & uncommitted_state,
    std::vector<Coordination::ACL> & result_acls)
 {
    if (request_acls.empty())
@ -81,14 +76,18 @@ bool fixupACL(
    {
        if (request_acl.scheme == "auth")
        {
-            for (const auto & current_id : current_ids)
-            {
-                valid_found = true;
-                Coordination::ACL new_acl = request_acl;
-                new_acl.scheme = current_id.scheme;
-                new_acl.id = current_id.id;
-                result_acls.push_back(new_acl);
-            }
+            uncommitted_state.forEachAuthInSession(
+                session_id,
+                [&](const KeeperStorage::AuthID & auth_id)
+                {
+                    valid_found = true;
+                    Coordination::ACL new_acl = request_acl;
+
+                    new_acl.scheme = auth_id.scheme;
+                    new_acl.id = auth_id.id;
+
+                    result_acls.push_back(new_acl);
+                });
        }
        else if (request_acl.scheme == "world" && request_acl.id == "anyone")
        {
@ -564,6 +563,32 @@ Coordination::ACLs KeeperStorage::UncommittedState::getACLs(StringRef path) cons
    return storage.acl_map.convertNumber(node_it->value.acl_id);
 }

+void KeeperStorage::UncommittedState::forEachAuthInSession(int64_t session_id, std::function<void(const AuthID &)> func) const
+{
+    const auto call_for_each_auth = [&func](const auto & auth_ids)
+    {
+        for (const auto & auth : auth_ids)
+        {
+            using TAuth = std::remove_reference_t<decltype(auth)>;
+
+            const AuthID * auth_ptr = nullptr;
+            if constexpr (std::is_pointer_v<TAuth>)
+                auth_ptr = auth;
+            else
+                auth_ptr = &auth;
+
+            func(*auth_ptr);
+        }
+    };
+
+    // for committed
+    if (storage.session_and_auth.contains(session_id))
+        call_for_each_auth(storage.session_and_auth.at(session_id));
+    // for uncommitted
+    if (session_and_auth.contains(session_id))
+        call_for_each_auth(session_and_auth.at(session_id));
+}
+
 namespace
 {

@ -927,7 +952,7 @@ struct KeeperStorageCreateRequestProcessor final : public KeeperStorageRequestPr
            return {KeeperStorage::Delta{zxid, Coordination::Error::ZBADARGUMENTS}};

        Coordination::ACLs node_acls;
-        if (!fixupACL(request.acls, storage.session_and_auth[session_id], node_acls))
+        if (!fixupACL(request.acls, session_id, storage.uncommitted_state, node_acls))
            return {KeeperStorage::Delta{zxid, Coordination::Error::ZINVALIDACL}};

        if (request.is_ephemeral)
@ -1533,10 +1558,8 @@ struct KeeperStorageSetACLRequestProcessor final : public KeeperStorageRequestPr
            return {KeeperStorage::Delta{zxid, Coordination::Error::ZBADVERSION}};


-        auto & session_auth_ids = storage.session_and_auth[session_id];
        Coordination::ACLs node_acls;
-
-        if (!fixupACL(request.acls, session_auth_ids, node_acls))
+        if (!fixupACL(request.acls, session_id, uncommitted_state, node_acls))
            return {KeeperStorage::Delta{zxid, Coordination::Error::ZINVALIDACL}};

        std::vector<KeeperStorage::Delta> new_deltas
@ -1840,7 +1863,7 @@ struct KeeperStorageAuthRequestProcessor final : public KeeperStorageRequestProc
            return {KeeperStorage::Delta{zxid, Coordination::Error::ZAUTHFAILED}};

        std::vector<KeeperStorage::Delta> new_deltas;
-        auto auth_digest = generateDigest(auth_request.data);
+        auto auth_digest = KeeperStorage::generateDigest(auth_request.data);
        if (auth_digest == storage.superdigest)
        {
            KeeperStorage::AuthID auth{"super", ""};
@ -2420,5 +2443,12 @@ void KeeperStorage::recalculateStats()
    container.recalculateDataSize();
 }

+String KeeperStorage::generateDigest(const String & userdata)
+{
+    std::vector<String> user_password;
+    boost::split(user_password, userdata, [](char character) { return character == ':'; });
+    return user_password[0] + ":" + base64Encode(getSHA1(userdata));
+}
+

 }
--- a/src/Coordination/KeeperStorage.h
+++ b/src/Coordination/KeeperStorage.h
@ -105,6 +105,8 @@ public:
        return first.value == second.value;
    }

+    static String generateDigest(const String & userdata);
+
    struct RequestForSession
    {
        int64_t session_id;
@ -263,6 +265,8 @@ public:
            return check_auth(auth_it->second);
        }

+        void forEachAuthInSession(int64_t session_id, std::function<void(const AuthID &)> func) const;
+
        std::shared_ptr<Node> tryGetNodeFromStorage(StringRef path) const;

        std::unordered_map<int64_t, std::list<const AuthID *>> session_and_auth;
--- a/Show More
+++ b/Show More