Merge branch 'master' into format-settings-parsing

2024-11-25 09:02:00 +00:00 · 2024-06-23 03:14:51 +02:00 · 2024-06-23 03:14:51 +02:00 · c21cce7a04
commit c21cce7a04
parent 9dcaade3bb af361a73ff
87 changed files with 1067 additions and 462 deletions
--- a/README.md
+++ b/README.md
@ -34,20 +34,18 @@ curl https://clickhouse.com/ | sh

 Every month we get together with the community (users, contributors, customers, those interested in learning more about ClickHouse) to discuss what is coming in the latest release. If you are interested in sharing what you've built on ClickHouse, let us know.

-* [v24.5 Community Call](https://clickhouse.com/company/events/v24-5-community-release-call) - May 30
+* [v24.6 Community Call](https://clickhouse.com/company/events/v24-6-community-release-call) - Jul 2

 ## Upcoming Events

 Keep an eye out for upcoming meetups and events around the world. Somewhere else you want us to be? Please feel free to reach out to tyler `<at>` clickhouse `<dot>` com. You can also peruse [ClickHouse Events](https://clickhouse.com/company/news-events) for a list of all upcoming trainings, meetups, speaking engagements, etc.

-* [ClickHouse Happy Hour @ Tom's Watch Bar - Los Angeles](https://www.meetup.com/clickhouse-los-angeles-user-group/events/300740584/) - May 22
-* [ClickHouse & Confluent Meetup in Dubai](https://www.meetup.com/clickhouse-dubai-meetup-group/events/299629189/) - May 28
-* [ClickHouse Meetup in Stockholm](https://www.meetup.com/clickhouse-stockholm-user-group/events/299752651/) - Jun 3
-* [ClickHouse Meetup @ Cloudflare - San Francisco](https://www.meetup.com/clickhouse-silicon-valley-meetup-group/events/300523061/) - Jun 4
-* [ClickHouse (クリックハウス) Meetup Tokyo](https://www.meetup.com/clickhouse-tokyo-user-group/events/300798053/) - Jun 5
+* [AWS Summit in DC](https://clickhouse.com/company/events/2024-06-aws-summit-dc) - Jun 26
 * [ClickHouse Meetup in Amsterdam](https://www.meetup.com/clickhouse-netherlands-user-group/events/300781068/) - Jun 27
 * [ClickHouse Meetup in Paris](https://www.meetup.com/clickhouse-france-user-group/events/300783448/) - Jul 9
+* [ClickHouse Cloud - Live Update Call](https://clickhouse.com/company/events/202407-cloud-update-live) - Jul 9
 * [ClickHouse Meetup @ Ramp - New York City](https://www.meetup.com/clickhouse-new-york-user-group/events/300595845/) - Jul 9
+* [AWS Summit in New York](https://clickhouse.com/company/events/2024-07-awssummit-nyc) - Jul 10
 * [ClickHouse Meetup @ Klaviyo - Boston](https://www.meetup.com/clickhouse-boston-user-group/events/300907870) - Jul 11

 ## Recent Recordings
--- a/base/poco/Net/include/Poco/Net/HTTPBasicStreamBuf.h
+++ b/base/poco/Net/include/Poco/Net/HTTPBasicStreamBuf.h
@ -26,7 +26,7 @@ namespace Poco
 {
 namespace Net
 {
-    constexpr size_t HTTP_DEFAULT_BUFFER_SIZE = 1024 * 1024;
+    constexpr size_t HTTP_DEFAULT_BUFFER_SIZE = 8 * 1024;

    typedef Poco::BasicBufferedStreamBuf<char, std::char_traits<char>> HTTPBasicStreamBuf;

--- a/docker/test/stateless/run.sh
+++ b/docker/test/stateless/run.sh
@ -285,7 +285,7 @@ stop_logs_replication

 # Try to get logs while server is running
 failed_to_save_logs=0
-for table in query_log zookeeper_log trace_log transactions_info_log metric_log blob_storage_log
+for table in query_log zookeeper_log trace_log transactions_info_log metric_log blob_storage_log error_log
 do
    err=$(clickhouse-client -q "select * from system.$table into outfile '/test_output/$table.tsv.gz' format TSVWithNamesAndTypes")
    echo "$err"
@ -339,7 +339,7 @@ if [ $failed_to_save_logs -ne 0 ]; then
    #   directly
    # - even though ci auto-compress some files (but not *.tsv) it does this only
    #   for files >64MB, we want this files to be compressed explicitly
-    for table in query_log zookeeper_log trace_log transactions_info_log metric_log blob_storage_log
+    for table in query_log zookeeper_log trace_log transactions_info_log metric_log blob_storage_log error_log
    do
        clickhouse-local "$data_path_config" --only-system-tables --stacktrace -q "select * from system.$table format TSVWithNamesAndTypes" | zstd --threads=0 > /test_output/$table.tsv.zst ||:
        if [[ -n "$USE_DATABASE_REPLICATED" ]] && [[ "$USE_DATABASE_REPLICATED" -eq 1 ]]; then
--- a/docs/en/operations/server-configuration-parameters/settings.md
+++ b/docs/en/operations/server-configuration-parameters/settings.md
@ -954,6 +954,38 @@ Or it can be set in hex:

 Everything mentioned above can be applied for `aes_256_gcm_siv` (but the key must be 32 bytes long).

+## error_log {#error_log}
+
+It is disabled by default.
+
+**Enabling**
+
+To manually turn on error history collection [`system.error_log`](../../operations/system-tables/error_log.md), create `/etc/clickhouse-server/config.d/error_log.xml` with the following content:
+
+``` xml
+<clickhouse>
+    <error_log>
+        <database>system</database>
+        <table>error_log</table>
+        <flush_interval_milliseconds>7500</flush_interval_milliseconds>
+        <collect_interval_milliseconds>1000</collect_interval_milliseconds>
+        <max_size_rows>1048576</max_size_rows>
+        <reserved_size_rows>8192</reserved_size_rows>
+        <buffer_size_rows_flush_threshold>524288</buffer_size_rows_flush_threshold>
+        <flush_on_crash>false</flush_on_crash>
+    </error_log>
+</clickhouse>
+```
+
+**Disabling**
+
+To disable `error_log` setting, you should create the following file `/etc/clickhouse-server/config.d/disable_error_log.xml` with the following content:
+
+``` xml
+<clickhouse>
+<error_log remove="1" />
+</clickhouse>
+```

 ## custom_settings_prefixes {#custom_settings_prefixes}

@ -1917,7 +1949,7 @@ For more information, see the MergeTreeSettings.h header file.

 ## metric_log {#metric_log}

-It is enabled by default. If it`s not, you can do this manually.
+It is disabled by default.

 **Enabling**

--- a/docs/en/operations/system-tables/error_log.md
+++ b/docs/en/operations/system-tables/error_log.md
@ -0,0 +1,39 @@
+---
+slug: /en/operations/system-tables/error_log
+---
+# error_log
+
+Contains history of error values from table `system.errors`, periodically flushed to disk.
+
+Columns:
+- `hostname` ([LowCardinality(String)](../../sql-reference/data-types/string.md)) — Hostname of the server executing the query.
+- `event_date` ([Date](../../sql-reference/data-types/date.md)) — Event date.
+- `event_time` ([DateTime](../../sql-reference/data-types/datetime.md)) — Event time.
+- `code` ([Int32](../../sql-reference/data-types/int-uint.md)) — Code number of the error.
+- `error` ([LowCardinality(String)](../../sql-reference/data-types/string.md)) - Name of the error.
+- `value` ([UInt64](../../sql-reference/data-types/int-uint.md)) — The number of times this error happened.
+- `remote` ([UInt8](../../sql-reference/data-types/int-uint.md)) — Remote exception (i.e. received during one of the distributed queries).
+
+**Example**
+
+``` sql
+SELECT * FROM system.error_log LIMIT 1 FORMAT Vertical;
+```
+
+``` text
+Row 1:
+──────
+hostname:   clickhouse.eu-central1.internal
+event_date: 2024-06-18
+event_time: 2024-06-18 07:32:39
+code:       999
+error:      KEEPER_EXCEPTION
+value:      2
+remote:     0
+```
+
+**See also**
+
+- [error_log setting](../../operations/server-configuration-parameters/settings.md#error_log) — Enabling and disabling the setting.
+- [system.errors](../../operations/system-tables/errors.md) — Contains error codes with the number of times they have been triggered.
+- [Monitoring](../../operations/monitoring.md) — Base concepts of ClickHouse monitoring.
--- a/docs/en/sql-reference/aggregate-functions/reference/stddevpop.md
+++ b/docs/en/sql-reference/aggregate-functions/reference/stddevpop.md
@ -25,7 +25,7 @@ stddevPop(x)

 **Returned value**

-Square root of standard deviation of `x`. [Float64](../../data-types/float.md).
+- Square root of standard deviation of `x`. [Float64](../../data-types/float.md).


 **Example**
--- a/docs/en/sql-reference/aggregate-functions/reference/varpop.md
+++ b/docs/en/sql-reference/aggregate-functions/reference/varpop.md
@ -4,30 +4,25 @@ slug: "/en/sql-reference/aggregate-functions/reference/varpop"
 sidebar_position: 32
 ---

-This page covers the `varPop` and `varPopStable` functions available in ClickHouse.
-
 ## varPop

-Calculates the population covariance between two data columns. The population covariance measures the degree to which two variables vary together. Calculates the amount `Σ((x - x̅)^2) / n`, where `n` is the sample size and `x̅`is the average value of `x`.
+Calculates the population variance.

 **Syntax**

 ```sql
-covarPop(x, y)
+varPop(x)
 ```

+Alias: `VAR_POP`.
+
 **Parameters**

- `x`: The first data column. [Numeric](../../../native-protocol/columns.md)
- `y`: The second data column. [Numeric](../../../native-protocol/columns.md)
+- `x`: Population of values to find the population variance of. [(U)Int*](../../data-types/int-uint.md), [Float*](../../data-types/float.md), [Decimal*](../../data-types/decimal.md).

 **Returned value**

-Returns an integer of type `Float64`.
-
-**Implementation details**
-
-This function uses a numerically unstable algorithm. If you need numerical stability in calculations, use the slower but more stable [`varPopStable`](#varpopstable) function.
+- Returns the population variance of `x`. [`Float64`](../../data-types/float.md).

 **Example**

@ -37,69 +32,21 @@ Query:
 DROP TABLE IF EXISTS test_data;
 CREATE TABLE test_data
 (
-    x Int32,
-    y Int32
+    x UInt8,
 )
 ENGINE = Memory;

-INSERT INTO test_data VALUES (1, 2), (2, 3), (3, 5), (4, 6), (5, 8);
+INSERT INTO test_data VALUES (3), (3), (3), (4), (4), (5), (5), (7), (11), (15);

 SELECT
-    covarPop(x, y) AS covar_pop
+    varPop(x) AS var_pop
 FROM test_data;
 ```

 Result:

 ```response
-3
-```
-
-## varPopStable
-
-Calculates population covariance between two data columns using a stable, numerically accurate method to calculate the variance. This function is designed to provide reliable results even with large datasets or values that might cause numerical instability in other implementations.
-
-**Syntax**
-
-```sql
-covarPopStable(x, y)
-```
-
-**Parameters**
-
- `x`: The first data column. [String literal](../../syntax#syntax-string-literal)
- `y`: The second data column. [Expression](../../syntax#syntax-expressions)
-
-**Returned value**
-
-Returns an integer of type `Float64`.
-
-**Implementation details**
-
-Unlike [`varPop`](#varpop), this function uses a stable, numerically accurate algorithm to calculate the population variance to avoid issues like catastrophic cancellation or loss of precision. This function also handles `NaN` and `Inf` values correctly, excluding them from calculations.
-
-**Example**
-
-Query:
-
-```sql
-DROP TABLE IF EXISTS test_data;
-CREATE TABLE test_data
-(
-    x Int32,
-    y Int32
-)
-ENGINE = Memory;
-
-INSERT INTO test_data VALUES (1, 2), (2, 9), (9, 5), (4, 6), (5, 8);
-
-SELECT
-    covarPopStable(x, y) AS covar_pop_stable
-FROM test_data;
-```
-
-Result:
-
-```response
-0.5999999999999999
+┌─var_pop─┐
+│    14.4 │
+└─────────┘
 ```
--- a/docs/en/sql-reference/aggregate-functions/reference/varpopstable.md
+++ b/docs/en/sql-reference/aggregate-functions/reference/varpopstable.md
@ -0,0 +1,52 @@
+---
+title: "varPopStable"
+slug: "/en/sql-reference/aggregate-functions/reference/varpopstable"
+sidebar_position: 32
+---
+
+## varPopStable
+
+Returns the population variance. Unlike [`varPop`](../reference/varpop.md), this function uses a [numerically stable](https://en.wikipedia.org/wiki/Numerical_stability) algorithm. It works slower but provides a lower computational error.
+
+**Syntax**
+
+```sql
+varPopStable(x)
+```
+
+Alias: `VAR_POP_STABLE`.
+
+**Parameters**
+
+- `x`: Population of values to find the population variance of. [(U)Int*](../../data-types/int-uint.md), [Float*](../../data-types/float.md), [Decimal*](../../data-types/decimal.md).
+
+**Returned value**
+
+- Returns the population variance of `x`. [Float64](../../data-types/float.md).
+
+**Example**
+
+Query:
+
+```sql
+DROP TABLE IF EXISTS test_data;
+CREATE TABLE test_data
+(
+    x UInt8,
+)
+ENGINE = Memory;
+
+INSERT INTO test_data VALUES (3),(3),(3),(4),(4),(5),(5),(7),(11),(15);
+
+SELECT
+    varPopStable(x) AS var_pop_stable
+FROM test_data;
+```
+
+Result:
+
+```response
+┌─var_pop_stable─┐
+│           14.4 │
+└────────────────┘
+```
--- a/docs/en/sql-reference/aggregate-functions/reference/varsamp.md
+++ b/docs/en/sql-reference/aggregate-functions/reference/varsamp.md
@ -4,8 +4,6 @@ slug: /en/sql-reference/aggregate-functions/reference/varsamp
 sidebar_position: 33
 ---

-This page contains information on the `varSamp` and `varSampStable` ClickHouse functions.
-
 ## varSamp

 Calculate the sample variance of a data set.
@ -13,24 +11,27 @@ Calculate the sample variance of a data set.
 **Syntax**

 ```sql
-varSamp(expr)
+varSamp(x)
 ```

+Alias: `VAR_SAMP`.
+
 **Parameters**

- `expr`: An expression representing the data set for which you want to calculate the sample variance. [Expression](../../syntax#syntax-expressions)
+- `x`: The population for which you want to calculate the sample variance. [(U)Int*](../../data-types/int-uint.md), [Float*](../../data-types/float.md), [Decimal*](../../data-types/decimal.md).

 **Returned value**

-Returns a Float64 value representing the sample variance of the input data set.
+
+- Returns the sample variance of the input data set `x`. [Float64](../../data-types/float.md).

 **Implementation details**

-The `varSamp()` function calculates the sample variance using the following formula:
+The `varSamp` function calculates the sample variance using the following formula:

-```plaintext
-∑(x - mean(x))^2 / (n - 1)
-```
+$$
+\sum\frac{(x - \text{mean}(x))^2}{(n - 1)}
+$$

 Where:

@ -38,91 +39,29 @@ Where:
 - `mean(x)` is the arithmetic mean of the data set.
 - `n` is the number of data points in the data set.

-The function assumes that the input data set represents a sample from a larger population. If you want to calculate the variance of the entire population (when you have the complete data set), you should use the [`varPop()` function](./varpop#varpop) instead.
-
-This function uses a numerically unstable algorithm. If you need numerical stability in calculations, use the slower but more stable [`varSampStable`](#varsampstable) function.
+The function assumes that the input data set represents a sample from a larger population. If you want to calculate the variance of the entire population (when you have the complete data set), you should use [`varPop`](../reference/varpop.md) instead.

 **Example**

 Query:

 ```sql
-CREATE TABLE example_table
+DROP TABLE IF EXISTS test_data;
+CREATE TABLE test_data
 (
-    id UInt64,
-    value Float64
+    x Float64
 )
-ENGINE = MergeTree
-ORDER BY id;
+ENGINE = Memory;

-INSERT INTO example_table VALUES (1, 10.5), (2, 12.3), (3, 9.8), (4, 11.2), (5, 10.7);
+INSERT INTO test_data VALUES (10.5), (12.3), (9.8), (11.2), (10.7);

-SELECT varSamp(value) FROM example_table;
+SELECT round(varSamp(x),3) AS var_samp FROM test_data;
 ```

 Response:

 ```response
-0.8650000000000091
+┌─var_samp─┐
+│    0.865 │
+└──────────┘
 ```
-
-## varSampStable
-
-Calculate the sample variance of a data set using a numerically stable algorithm.
-
-**Syntax**
-
-```sql
-varSampStable(expr)
-```
-
-**Parameters**
-
- `expr`: An expression representing the data set for which you want to calculate the sample variance. [Expression](../../syntax#syntax-expressions)
-
-**Returned value**
-
-The `varSampStable` function returns a Float64 value representing the sample variance of the input data set.
-
-**Implementation details**
-
-The `varSampStable` function calculates the sample variance using the same formula as the [`varSamp`](#varsamp) function:
-
-```plaintext
-∑(x - mean(x))^2 / (n - 1)
-```
-
-Where:
- `x` is each individual data point in the data set.
- `mean(x)` is the arithmetic mean of the data set.
- `n` is the number of data points in the data set.
-
-The difference between `varSampStable` and `varSamp` is that `varSampStable` is designed to provide a more deterministic and stable result when dealing with floating-point arithmetic. It uses an algorithm that minimizes the accumulation of rounding errors, which can be particularly important when dealing with large data sets or data with a wide range of values.
-
-Like `varSamp`, the `varSampStable` function assumes that the input data set represents a sample from a larger population. If you want to calculate the variance of the entire population (when you have the complete data set), you should use the [`varPopStable`](./varpop#varpopstable) function instead.
-
-**Example**
-
-Query:
-
-```sql
-CREATE TABLE example_table
-(
-    id UInt64,
-    value Float64
-)
-ENGINE = MergeTree
-ORDER BY id;
-
-INSERT INTO example_table VALUES (1, 10.5), (2, 12.3), (3, 9.8), (4, 11.2), (5, 10.7);
-
-SELECT varSampStable(value) FROM example_table;
-```
-
-Response:
-
-```response
-0.865
-```
-
-This query calculates the sample variance of the `value` column in the `example_table` using the `varSampStable()` function. The result shows that the sample variance of the values `[10.5, 12.3, 9.8, 11.2, 10.7]` is approximately 0.865, which may differ slightly from the result of `varSamp` due to the more precise handling of floating-point arithmetic.
--- a/docs/en/sql-reference/aggregate-functions/reference/varsampstable.md
+++ b/docs/en/sql-reference/aggregate-functions/reference/varsampstable.md
@ -0,0 +1,63 @@
+---
+title: "varSampStable"
+slug: /en/sql-reference/aggregate-functions/reference/varsampstable
+sidebar_position: 33
+---
+
+## varSampStable
+
+Calculate the sample variance of a data set. Unlike [`varSamp`](../reference/varsamp.md), this function uses a numerically stable algorithm. It works slower but provides a lower computational error.
+
+**Syntax**
+
+```sql
+varSampStable(x)
+```
+
+Alias: `VAR_SAMP_STABLE`
+
+**Parameters**
+
+- `x`: The population for which you want to calculate the sample variance. [(U)Int*](../../data-types/int-uint.md), [Float*](../../data-types/float.md), [Decimal*](../../data-types/decimal.md).
+
+**Returned value**
+
+- Returns the sample variance of the input data set. [Float64](../../data-types/float.md).
+
+**Implementation details**
+
+The `varSampStable` function calculates the sample variance using the same formula as the [`varSamp`](../reference/varsamp.md):
+
+$$
+\sum\frac{(x - \text{mean}(x))^2}{(n - 1)}
+$$
+
+Where:
+- `x` is each individual data point in the data set.
+- `mean(x)` is the arithmetic mean of the data set.
+- `n` is the number of data points in the data set.
+
+**Example**
+
+Query:
+
+```sql
+DROP TABLE IF EXISTS test_data;
+CREATE TABLE test_data
+(
+    x Float64
+)
+ENGINE = Memory;
+
+INSERT INTO test_data VALUES (10.5), (12.3), (9.8), (11.2), (10.7);
+
+SELECT round(varSampStable(x),3) AS var_samp_stable FROM test_data;
+```
+
+Response:
+
+```response
+┌─var_samp_stable─┐
+│           0.865 │
+└─────────────────┘
+```
--- a/programs/local/LocalServer.cpp
+++ b/programs/local/LocalServer.cpp
@ -732,11 +732,8 @@ void LocalServer::processConfig()
        attachInformationSchema(global_context, *createMemoryDatabaseIfNotExists(global_context, DatabaseCatalog::INFORMATION_SCHEMA_UPPERCASE));
    }

-    server_display_name = config().getString("display_name", getFQDNOrHostName());
-    prompt_by_server_display_name = config().getRawString("prompt_by_server_display_name.default", "{display_name} :) ");
-    std::map<String, String> prompt_substitutions{{"display_name", server_display_name}};
-    for (const auto & [key, value] : prompt_substitutions)
-        boost::replace_all(prompt_by_server_display_name, "{" + key + "}", value);
+    server_display_name = config().getString("display_name", "");
+    prompt_by_server_display_name = config().getRawString("prompt_by_server_display_name.default", ":) ");

    global_context->setQueryKindInitial();
    global_context->setQueryKind(query_kind);
--- a/programs/server/config.xml
+++ b/programs/server/config.xml
@ -1155,6 +1155,18 @@
        <flush_on_crash>false</flush_on_crash>
    </metric_log>

+    <!-- Error log contains rows with current values of errors collected with "collect_interval_milliseconds" interval. -->
+    <error_log>
+        <database>system</database>
+        <table>error_log</table>
+        <flush_interval_milliseconds>7500</flush_interval_milliseconds>
+        <max_size_rows>1048576</max_size_rows>
+        <reserved_size_rows>8192</reserved_size_rows>
+        <buffer_size_rows_flush_threshold>524288</buffer_size_rows_flush_threshold>
+        <collect_interval_milliseconds>1000</collect_interval_milliseconds>
+        <flush_on_crash>false</flush_on_crash>
+    </error_log>
+
    <!--
        Asynchronous metric log contains values of metrics from
        system.asynchronous_metrics.
--- a/programs/server/config.yaml.example
+++ b/programs/server/config.yaml.example
@ -726,6 +726,13 @@ metric_log:
    flush_interval_milliseconds: 7500
    collect_interval_milliseconds: 1000

+# Error log contains rows with current values of errors collected with "collect_interval_milliseconds" interval.
+error_log:
+    database: system
+    table: error_log
+    flush_interval_milliseconds: 7500
+    collect_interval_milliseconds: 1000
+
 # Asynchronous metric log contains values of metrics from
 # system.asynchronous_metrics.
 asynchronous_metric_log:
--- a/src/Backups/BackupImpl.cpp
+++ b/src/Backups/BackupImpl.cpp
@ -24,8 +24,6 @@
 #include <Poco/Util/XMLConfiguration.h>
 #include <Poco/DOM/DOMParser.h>

-#include <ranges>
-

 namespace ProfileEvents
 {
@ -93,6 +91,7 @@ BackupImpl::BackupImpl(
    const std::optional<BackupInfo> & base_backup_info_,
    std::shared_ptr<IBackupReader> reader_,
    const ContextPtr & context_,
+    bool is_internal_backup_,
    bool use_same_s3_credentials_for_base_backup_)
    : backup_info(backup_info_)
    , backup_name_for_logging(backup_info.toStringForLogging())
@ -101,7 +100,7 @@ BackupImpl::BackupImpl(
    , open_mode(OpenMode::READ)
    , reader(std::move(reader_))
    , context(context_)
-    , is_internal_backup(false)
+    , is_internal_backup(is_internal_backup_)
    , version(INITIAL_BACKUP_VERSION)
    , base_backup_info(base_backup_info_)
    , use_same_s3_credentials_for_base_backup(use_same_s3_credentials_for_base_backup_)
@ -256,6 +255,7 @@ std::shared_ptr<const IBackup> BackupImpl::getBaseBackupUnlocked() const
        params.backup_info = *base_backup_info;
        params.open_mode = OpenMode::READ;
        params.context = context;
+        params.is_internal_backup = is_internal_backup;
        /// use_same_s3_credentials_for_base_backup should be inherited for base backups
        params.use_same_s3_credentials_for_base_backup = use_same_s3_credentials_for_base_backup;

--- a/src/Backups/BackupImpl.h
+++ b/src/Backups/BackupImpl.h
@ -40,6 +40,7 @@ public:
        const std::optional<BackupInfo> & base_backup_info_,
        std::shared_ptr<IBackupReader> reader_,
        const ContextPtr & context_,
+        bool is_internal_backup_,
        bool use_same_s3_credentials_for_base_backup_);

    BackupImpl(
--- a/src/Backups/registerBackupEngineAzureBlobStorage.cpp
+++ b/src/Backups/registerBackupEngineAzureBlobStorage.cpp
@ -153,6 +153,7 @@ void registerBackupEngineAzureBlobStorage(BackupFactory & factory)
                params.base_backup_info,
                reader,
                params.context,
+                params.is_internal_backup,
                /* use_same_s3_credentials_for_base_backup*/ false);
        }
        else
--- a/src/Backups/registerBackupEngineS3.cpp
+++ b/src/Backups/registerBackupEngineS3.cpp
@ -119,6 +119,7 @@ void registerBackupEngineS3(BackupFactory & factory)
                params.base_backup_info,
                reader,
                params.context,
+                params.is_internal_backup,
                params.use_same_s3_credentials_for_base_backup);
        }
        else
--- a/src/Backups/registerBackupEnginesFileAndDisk.cpp
+++ b/src/Backups/registerBackupEnginesFileAndDisk.cpp
@ -177,6 +177,7 @@ void registerBackupEnginesFileAndDisk(BackupFactory & factory)
                params.base_backup_info,
                reader,
                params.context,
+                params.is_internal_backup,
                params.use_same_s3_credentials_for_base_backup);
        }
        else
--- a/src/Common/CurrentMetrics.cpp
+++ b/src/Common/CurrentMetrics.cpp
@ -1,6 +1,7 @@
 #include <Common/CurrentMetrics.h>


+// clang-format off
 /// Available metrics. Add something here as you wish.
 /// If the metric is generic (i.e. not server specific)
 /// it should be also added to src/Coordination/KeeperConstant.cpp
--- a/src/Common/Dwarf.cpp
+++ b/src/Common/Dwarf.cpp
@ -202,7 +202,10 @@ uint64_t readU64(std::string_view & sp)
 {
    SAFE_CHECK(sp.size() >= N, "underflow");
    uint64_t x = 0;
+    if constexpr (std::endian::native == std::endian::little)
        memcpy(&x, sp.data(), N);
+    else
+        memcpy(reinterpret_cast<char*>(&x) + sizeof(uint64_t) - N, sp.data(), N);
    sp.remove_prefix(N);
    return x;
 }
--- a/src/Common/ErrorCodes.h
+++ b/src/Common/ErrorCodes.h
@ -1,8 +1,6 @@
 #pragma once

 #include <cstddef>
-#include <cstdint>
-#include <utility>
 #include <mutex>
 #include <string_view>
 #include <vector>
@ -35,7 +33,7 @@ namespace ErrorCodes

    struct Error
    {
-        /// Number of times Exception with this ErrorCode had been throw.
+        /// Number of times Exception with this ErrorCode has been thrown.
        Value count = 0;
        /// Time of the last error.
        UInt64 error_time_ms = 0;
--- a/src/Common/ProfileEvents.cpp
+++ b/src/Common/ProfileEvents.cpp
@ -3,6 +3,7 @@
 #include <Common/TraceSender.h>


+// clang-format off
 /// Available events. Add something here as you wish.
 /// If the event is generic (i.e. not server specific)
 /// it should be also added to src/Coordination/KeeperConstant.cpp
@ -14,6 +15,7 @@
    M(QueriesWithSubqueries, "Count queries with all subqueries") \
    M(SelectQueriesWithSubqueries, "Count SELECT queries with all subqueries") \
    M(InsertQueriesWithSubqueries, "Count INSERT queries with all subqueries") \
+    M(SelectQueriesWithPrimaryKeyUsage, "Count SELECT queries which use the primary key to evaluate the WHERE condition") \
    M(AsyncInsertQuery, "Same as InsertQuery, but only for asynchronous INSERT queries.") \
    M(AsyncInsertBytes, "Data size in bytes of asynchronous INSERT queries.") \
    M(AsyncInsertRows, "Number of rows inserted by asynchronous INSERT queries.") \
--- a/src/Common/SystemLogBase.cpp
+++ b/src/Common/SystemLogBase.cpp
@ -1,5 +1,6 @@
 #include <Interpreters/AsynchronousMetricLog.h>
 #include <Interpreters/CrashLog.h>
+#include <Interpreters/ErrorLog.h>
 #include <Interpreters/MetricLog.h>
 #include <Interpreters/OpenTelemetrySpanLog.h>
 #include <Interpreters/PartLog.h>
--- a/src/Common/SystemLogBase.h
+++ b/src/Common/SystemLogBase.h
@ -1,9 +1,7 @@
 #pragma once

-#include <atomic>
 #include <condition_variable>
 #include <memory>
-#include <thread>
 #include <vector>
 #include <base/types.h>

@ -32,7 +30,8 @@
    M(FilesystemReadPrefetchesLogElement) \
    M(AsynchronousInsertLogElement) \
    M(BackupLogElement) \
-    M(BlobStorageLogElement)
+    M(BlobStorageLogElement) \
+    M(ErrorLogElement)

 namespace Poco
 {
--- a/src/Core/ServerSettings.h
+++ b/src/Core/ServerSettings.h
@ -14,6 +14,7 @@ class AbstractConfiguration;
 namespace DB
 {

+// clang-format off
 #define SERVER_SETTINGS(M, ALIAS) \
    M(Bool, show_addresses_in_stack_traces, true, "If it is set true will show addresses in stack traces", 0) \
    M(Bool, shutdown_wait_unfinished_queries, false, "If set true ClickHouse will wait for running queries finish before shutdown.", 0) \
--- a/src/Core/Settings.h
+++ b/src/Core/Settings.h
@ -31,6 +31,7 @@ class IColumn;
  * for tracking settings changes in different versions and for special `compatibility` setting to work correctly.
  */

+// clang-format off
 #define COMMON_SETTINGS(M, ALIAS) \
    M(Dialect, dialect, Dialect::clickhouse, "Which dialect will be used to parse query", 0)\
    M(UInt64, min_compress_block_size, 65536, "The actual size of the block to compress, if the uncompressed data less than max_compress_block_size is no less than this value and no less than the volume of data for one mark.", 0) \
--- a/src/Core/SettingsChangesHistory.h
+++ b/src/Core/SettingsChangesHistory.h
@ -75,6 +75,7 @@ namespace SettingsChangesHistory
    using SettingsChanges = std::vector<SettingChange>;
 }

+// clang-format off
 /// History of settings changes that controls some backward incompatible changes
 /// across all ClickHouse versions. It maps ClickHouse version to settings changes that were done
 /// in this version. This history contains both changes to existing settings and newly added settings.
--- a/src/Interpreters/AsynchronousMetricLog.h
+++ b/src/Interpreters/AsynchronousMetricLog.h
@ -8,8 +8,6 @@
 #include <Core/NamesAndAliases.h>
 #include <Storages/ColumnsDescription.h>

-#include <vector>
-#include <atomic>
 #include <ctime>


--- a/src/Interpreters/ErrorLog.cpp
+++ b/src/Interpreters/ErrorLog.cpp
@ -0,0 +1,123 @@
+#include <DataTypes/DataTypeDate.h>
+#include <DataTypes/DataTypeDateTime.h>
+#include <DataTypes/DataTypeDateTime64.h>
+#include <DataTypes/DataTypeLowCardinality.h>
+#include <DataTypes/DataTypeString.h>
+#include <DataTypes/DataTypesNumber.h>
+#include <Interpreters/ErrorLog.h>
+#include <base/getFQDNOrHostName.h>
+#include <Common/DateLUTImpl.h>
+#include <Common/ThreadPool.h>
+#include <Common/ErrorCodes.h>
+#include <Parsers/ExpressionElementParsers.h>
+#include <Parsers/parseQuery.h>
+
+#include <vector>
+
+namespace DB
+{
+
+ColumnsDescription ErrorLogElement::getColumnsDescription()
+{
+    ParserCodec codec_parser;
+    return ColumnsDescription {
+        {
+                "hostname",
+                std::make_shared<DataTypeLowCardinality>(std::make_shared<DataTypeString>()),
+                parseQuery(codec_parser, "(ZSTD(1))", 0, DBMS_DEFAULT_MAX_PARSER_DEPTH, DBMS_DEFAULT_MAX_PARSER_BACKTRACKS),
+                "Hostname of the server executing the query."
+            },
+        {
+                "event_date",
+                std::make_shared<DataTypeDate>(),
+                parseQuery(codec_parser, "(Delta(2), ZSTD(1))", 0, DBMS_DEFAULT_MAX_PARSER_DEPTH, DBMS_DEFAULT_MAX_PARSER_BACKTRACKS),
+                "Event date."
+            },
+        {
+                "event_time",
+                std::make_shared<DataTypeDateTime>(),
+                parseQuery(codec_parser, "(Delta(4), ZSTD(1))", 0, DBMS_DEFAULT_MAX_PARSER_DEPTH, DBMS_DEFAULT_MAX_PARSER_BACKTRACKS),
+                "Event time."
+            },
+        {
+                "code",
+                std::make_shared<DataTypeInt32>(),
+                parseQuery(codec_parser, "(ZSTD(1))", 0, DBMS_DEFAULT_MAX_PARSER_DEPTH, DBMS_DEFAULT_MAX_PARSER_BACKTRACKS),
+                "Error code."
+            },
+        {
+                "error",
+                std::make_shared<DataTypeLowCardinality>(std::make_shared<DataTypeString>()),
+                parseQuery(codec_parser, "(ZSTD(1))", 0, DBMS_DEFAULT_MAX_PARSER_DEPTH, DBMS_DEFAULT_MAX_PARSER_BACKTRACKS),
+                "Error name."
+            },
+        {
+                "value",
+                std::make_shared<DataTypeUInt64>(),
+                parseQuery(codec_parser, "(ZSTD(3))", 0, DBMS_DEFAULT_MAX_PARSER_DEPTH, DBMS_DEFAULT_MAX_PARSER_BACKTRACKS),
+                "Number of errors happened in time interval."
+            },
+        {
+                "remote",
+                std::make_shared<DataTypeUInt8>(),
+                parseQuery(codec_parser, "(ZSTD(1))", 0, DBMS_DEFAULT_MAX_PARSER_DEPTH, DBMS_DEFAULT_MAX_PARSER_BACKTRACKS),
+                "Remote exception (i.e. received during one of the distributed queries)."
+            }
+    };
+}
+
+void ErrorLogElement::appendToBlock(MutableColumns & columns) const
+{
+    size_t column_idx = 0;
+
+    columns[column_idx++]->insert(getFQDNOrHostName());
+    columns[column_idx++]->insert(DateLUT::instance().toDayNum(event_time).toUnderType());
+    columns[column_idx++]->insert(event_time);
+    columns[column_idx++]->insert(code);
+    columns[column_idx++]->insert(ErrorCodes::getName(code));
+    columns[column_idx++]->insert(value);
+    columns[column_idx++]->insert(remote);
+}
+
+struct ValuePair
+{
+    UInt64 local = 0;
+    UInt64 remote = 0;
+};
+
+void ErrorLog::stepFunction(TimePoint current_time)
+{
+    /// Static lazy initialization to avoid polluting the header with implementation details
+    static std::vector<ValuePair> previous_values(ErrorCodes::end());
+
+    auto event_time = std::chrono::system_clock::to_time_t(current_time);
+
+    for (ErrorCodes::ErrorCode code = 0, end = ErrorCodes::end(); code < end; ++code)
+    {
+        const auto & error = ErrorCodes::values[code].get();
+        if (error.local.count != previous_values.at(code).local)
+        {
+            ErrorLogElement local_elem {
+                .event_time=event_time,
+                .code=code,
+                .value=error.local.count - previous_values.at(code).local,
+                .remote=false
+            };
+            this->add(std::move(local_elem));
+            previous_values[code].local = error.local.count;
+        }
+        if (error.remote.count != previous_values.at(code).remote)
+        {
+            ErrorLogElement remote_elem {
+                .event_time=event_time,
+                .code=code,
+                .value=error.remote.count - previous_values.at(code).remote,
+                .remote=true
+            };
+            this->add(std::move(remote_elem));
+            previous_values[code].remote = error.remote.count;
+        }
+    }
+}
+
+}
--- a/src/Interpreters/ErrorLog.h
+++ b/src/Interpreters/ErrorLog.h
@ -0,0 +1,39 @@
+#pragma once
+
+#include <Interpreters/SystemLog.h>
+#include <Interpreters/PeriodicLog.h>
+#include <Common/ErrorCodes.h>
+#include <Core/NamesAndTypes.h>
+#include <Core/NamesAndAliases.h>
+#include <Storages/ColumnsDescription.h>
+
+
+namespace DB
+{
+
+/** ErrorLog is a log of error values measured at regular time interval.
+  */
+
+struct ErrorLogElement
+{
+    time_t event_time{};
+    ErrorCodes::ErrorCode code{};
+    ErrorCodes::Value value{};
+    bool remote{};
+
+    static std::string name() { return "ErrorLog"; }
+    static ColumnsDescription getColumnsDescription();
+    static NamesAndAliases getNamesAndAliases() { return {}; }
+    void appendToBlock(MutableColumns & columns) const;
+};
+
+
+class ErrorLog : public PeriodicLog<ErrorLogElement>
+{
+    using PeriodicLog<ErrorLogElement>::PeriodicLog;
+
+protected:
+    void stepFunction(TimePoint current_time) override;
+};
+
+}
--- a/src/Interpreters/MetricLog.cpp
+++ b/src/Interpreters/MetricLog.cpp
@ -56,44 +56,11 @@ void MetricLogElement::appendToBlock(MutableColumns & columns) const
        columns[column_idx++]->insert(current_metrics[i].toUnderType());
 }

-
-void MetricLog::startCollectMetric(size_t collect_interval_milliseconds_)
+void MetricLog::stepFunction(const std::chrono::system_clock::time_point current_time)
 {
-    collect_interval_milliseconds = collect_interval_milliseconds_;
-    is_shutdown_metric_thread = false;
-    metric_flush_thread = std::make_unique<ThreadFromGlobalPool>([this] { metricThreadFunction(); });
-}
-
-
-void MetricLog::stopCollectMetric()
-{
-    bool old_val = false;
-    if (!is_shutdown_metric_thread.compare_exchange_strong(old_val, true))
-        return;
-    if (metric_flush_thread)
-        metric_flush_thread->join();
-}
-
-
-void MetricLog::shutdown()
-{
-    stopCollectMetric();
-    stopFlushThread();
-}
-
-
-void MetricLog::metricThreadFunction()
-{
-    auto desired_timepoint = std::chrono::system_clock::now();
-
+    /// Static lazy initialization to avoid polluting the header with implementation details
    /// For differentiation of ProfileEvents counters.
-    std::vector<ProfileEvents::Count> prev_profile_events(ProfileEvents::end());
-
-    while (!is_shutdown_metric_thread)
-    {
-        try
-        {
-            const auto current_time = std::chrono::system_clock::now();
+    static std::vector<ProfileEvents::Count> prev_profile_events(ProfileEvents::end());

    MetricLogElement elem;
    elem.event_time = std::chrono::system_clock::to_time_t(current_time);
@ -115,19 +82,6 @@ void MetricLog::metricThreadFunction()
    }

    this->add(std::move(elem));
-
-            /// We will record current time into table but align it to regular time intervals to avoid time drift.
-            /// We may drop some time points if the server is overloaded and recording took too much time.
-            while (desired_timepoint <= current_time)
-                desired_timepoint += std::chrono::milliseconds(collect_interval_milliseconds);
-
-            std::this_thread::sleep_until(desired_timepoint);
-        }
-        catch (...)
-        {
-            tryLogCurrentException(__PRETTY_FUNCTION__);
-        }
-    }
 }

 }
--- a/src/Interpreters/MetricLog.h
+++ b/src/Interpreters/MetricLog.h
@ -1,6 +1,7 @@
 #pragma once

 #include <Interpreters/SystemLog.h>
+#include <Interpreters/PeriodicLog.h>
 #include <Common/ProfileEvents.h>
 #include <Common/CurrentMetrics.h>
 #include <Common/ThreadPool_fwd.h>
@ -9,7 +10,6 @@
 #include <Storages/ColumnsDescription.h>

 #include <vector>
-#include <atomic>
 #include <ctime>


@ -33,26 +33,12 @@ struct MetricLogElement
    void appendToBlock(MutableColumns & columns) const;
 };

-
-class MetricLog : public SystemLog<MetricLogElement>
+class MetricLog : public PeriodicLog<MetricLogElement>
 {
-    using SystemLog<MetricLogElement>::SystemLog;
+    using PeriodicLog<MetricLogElement>::PeriodicLog;

-public:
-    void shutdown() override;
-
-    /// Launches a background thread to collect metrics with interval
-    void startCollectMetric(size_t collect_interval_milliseconds_);
-
-    /// Stop background thread. Call before shutdown.
-    void stopCollectMetric();
-
-private:
-    void metricThreadFunction();
-
-    std::unique_ptr<ThreadFromGlobalPool> metric_flush_thread;
-    size_t collect_interval_milliseconds;
-    std::atomic<bool> is_shutdown_metric_thread{false};
+protected:
+    void stepFunction(TimePoint current_time) override;
 };

 }
--- a/src/Interpreters/PeriodicLog.cpp
+++ b/src/Interpreters/PeriodicLog.cpp
@ -0,0 +1,62 @@
+#include <Interpreters/PeriodicLog.h>
+#include <Interpreters/ErrorLog.h>
+#include <Interpreters/MetricLog.h>
+
+namespace DB
+{
+
+template <typename LogElement>
+void PeriodicLog<LogElement>::startCollect(size_t collect_interval_milliseconds_)
+{
+    collect_interval_milliseconds = collect_interval_milliseconds_;
+    is_shutdown_metric_thread = false;
+    flush_thread = std::make_unique<ThreadFromGlobalPool>([this] { threadFunction(); });
+}
+
+template <typename LogElement>
+void PeriodicLog<LogElement>::stopCollect()
+{
+    bool old_val = false;
+    if (!is_shutdown_metric_thread.compare_exchange_strong(old_val, true))
+        return;
+    if (flush_thread)
+        flush_thread->join();
+}
+
+template <typename LogElement>
+void PeriodicLog<LogElement>::shutdown()
+{
+    stopCollect();
+    this->stopFlushThread();
+}
+
+template <typename LogElement>
+void PeriodicLog<LogElement>::threadFunction()
+{
+    auto desired_timepoint = std::chrono::system_clock::now();
+    while (!is_shutdown_metric_thread)
+    {
+        try
+        {
+            const auto current_time = std::chrono::system_clock::now();
+
+            stepFunction(current_time);
+
+            /// We will record current time into table but align it to regular time intervals to avoid time drift.
+            /// We may drop some time points if the server is overloaded and recording took too much time.
+            while (desired_timepoint <= current_time)
+                desired_timepoint += std::chrono::milliseconds(collect_interval_milliseconds);
+
+            std::this_thread::sleep_until(desired_timepoint);
+        }
+        catch (...)
+        {
+            tryLogCurrentException(__PRETTY_FUNCTION__);
+        }
+    }
+}
+
+#define INSTANTIATE_SYSTEM_LOG(ELEMENT) template class PeriodicLog<ELEMENT>;
+SYSTEM_PERIODIC_LOG_ELEMENTS(INSTANTIATE_SYSTEM_LOG)
+
+}
--- a/src/Interpreters/PeriodicLog.h
+++ b/src/Interpreters/PeriodicLog.h
@ -0,0 +1,43 @@
+#pragma once
+
+#include <Interpreters/SystemLog.h>
+#include <Common/ThreadPool.h>
+
+#include <atomic>
+#include <chrono>
+
+#define SYSTEM_PERIODIC_LOG_ELEMENTS(M) \
+    M(ErrorLogElement) \
+    M(MetricLogElement)
+
+namespace DB
+{
+
+template <typename LogElement>
+class PeriodicLog : public SystemLog<LogElement>
+{
+    using SystemLog<LogElement>::SystemLog;
+
+public:
+    using TimePoint = std::chrono::system_clock::time_point;
+
+    /// Launches a background thread to collect metrics with interval
+    void startCollect(size_t collect_interval_milliseconds_);
+
+    /// Stop background thread
+    void stopCollect();
+
+    void shutdown() final;
+
+protected:
+    virtual void stepFunction(TimePoint current_time) = 0;
+
+private:
+    void threadFunction();
+
+    std::unique_ptr<ThreadFromGlobalPool> flush_thread;
+    size_t collect_interval_milliseconds;
+    std::atomic<bool> is_shutdown_metric_thread{false};
+};
+
+}
--- a/src/Interpreters/StorageID.h
+++ b/src/Interpreters/StorageID.h
@ -1,7 +1,6 @@
 #pragma once
 #include <base/types.h>
 #include <Core/UUID.h>
-#include <tuple>
 #include <Parsers/IAST_fwd.h>
 #include <Core/QualifiedTableName.h>
 #include <Common/Exception.h>
--- a/src/Interpreters/SystemLog.cpp
+++ b/src/Interpreters/SystemLog.cpp
@ -12,6 +12,7 @@
 #include <Interpreters/Context.h>
 #include <Interpreters/CrashLog.h>
 #include <Interpreters/DatabaseCatalog.h>
+#include <Interpreters/ErrorLog.h>
 #include <Interpreters/FilesystemCacheLog.h>
 #include <Interpreters/FilesystemReadPrefetchesLog.h>
 #include <Interpreters/InterpreterCreateQuery.h>
@ -116,6 +117,7 @@ namespace
 {

 constexpr size_t DEFAULT_METRIC_LOG_COLLECT_INTERVAL_MILLISECONDS = 1000;
+constexpr size_t DEFAULT_ERROR_LOG_COLLECT_INTERVAL_MILLISECONDS = 1000;

 /// Creates a system log with MergeTree engine using parameters from config
 template <typename TSystemLog>
@ -286,6 +288,7 @@ SystemLogs::SystemLogs(ContextPtr global_context, const Poco::Util::AbstractConf
    crash_log = createSystemLog<CrashLog>(global_context, "system", "crash_log", config, "crash_log", "Contains information about stack traces for fatal errors. The table does not exist in the database by default, it is created only when fatal errors occur.");
    text_log = createSystemLog<TextLog>(global_context, "system", "text_log", config, "text_log", "Contains logging entries which are normally written to a log file or to stdout.");
    metric_log = createSystemLog<MetricLog>(global_context, "system", "metric_log", config, "metric_log", "Contains history of metrics values from tables system.metrics and system.events, periodically flushed to disk.");
+    error_log = createSystemLog<ErrorLog>(global_context, "system", "error_log", config, "error_log", "Contains history of error values from table system.errors, periodically flushed to disk.");
    filesystem_cache_log = createSystemLog<FilesystemCacheLog>(global_context, "system", "filesystem_cache_log", config, "filesystem_cache_log", "Contains a history of all events occurred with filesystem cache for objects on a remote filesystem.");
    filesystem_read_prefetches_log = createSystemLog<FilesystemReadPrefetchesLog>(
        global_context, "system", "filesystem_read_prefetches_log", config, "filesystem_read_prefetches_log", "Contains a history of all prefetches done during reading from MergeTables backed by a remote filesystem.");
@ -320,6 +323,8 @@ SystemLogs::SystemLogs(ContextPtr global_context, const Poco::Util::AbstractConf
        logs.emplace_back(text_log.get());
    if (metric_log)
        logs.emplace_back(metric_log.get());
+    if (error_log)
+        logs.emplace_back(error_log.get());
    if (asynchronous_metric_log)
        logs.emplace_back(asynchronous_metric_log.get());
    if (opentelemetry_span_log)
@ -366,7 +371,14 @@ SystemLogs::SystemLogs(ContextPtr global_context, const Poco::Util::AbstractConf
    {
        size_t collect_interval_milliseconds = config.getUInt64("metric_log.collect_interval_milliseconds",
                                                                DEFAULT_METRIC_LOG_COLLECT_INTERVAL_MILLISECONDS);
-        metric_log->startCollectMetric(collect_interval_milliseconds);
+        metric_log->startCollect(collect_interval_milliseconds);
+    }
+
+    if (error_log)
+    {
+        size_t collect_interval_milliseconds = config.getUInt64("error_log.collect_interval_milliseconds",
+                                                                DEFAULT_ERROR_LOG_COLLECT_INTERVAL_MILLISECONDS);
+        error_log->startCollect(collect_interval_milliseconds);
    }

    if (crash_log)
--- a/src/Interpreters/SystemLog.h
+++ b/src/Interpreters/SystemLog.h
@ -40,6 +40,7 @@ class PartLog;
 class TextLog;
 class TraceLog;
 class CrashLog;
+class ErrorLog;
 class MetricLog;
 class AsynchronousMetricLog;
 class OpenTelemetrySpanLog;
@ -72,6 +73,7 @@ struct SystemLogs
    std::shared_ptr<CrashLog> crash_log;                /// Used to log server crashes.
    std::shared_ptr<TextLog> text_log;                  /// Used to log all text messages.
    std::shared_ptr<MetricLog> metric_log;              /// Used to log all metrics.
+    std::shared_ptr<ErrorLog> error_log;                /// Used to log errors.
    std::shared_ptr<FilesystemCacheLog> filesystem_cache_log;
    std::shared_ptr<FilesystemReadPrefetchesLog> filesystem_read_prefetches_log;
    std::shared_ptr<S3QueueLog> s3_queue_log;
--- a/src/Parsers/ASTAlterQuery.cpp
+++ b/src/Parsers/ASTAlterQuery.cpp
@ -60,6 +60,8 @@ ASTPtr ASTAlterCommand::clone() const
        res->settings_resets = res->children.emplace_back(settings_resets->clone()).get();
    if (select)
        res->select = res->children.emplace_back(select->clone()).get();
+    if (sql_security)
+        res->sql_security = res->children.emplace_back(sql_security->clone()).get();
    if (rename_to)
        res->rename_to = res->children.emplace_back(rename_to->clone()).get();

@ -522,6 +524,7 @@ void ASTAlterCommand::forEachPointerToChild(std::function<void(void**)> f)
    f(reinterpret_cast<void **>(&settings_changes));
    f(reinterpret_cast<void **>(&settings_resets));
    f(reinterpret_cast<void **>(&select));
+    f(reinterpret_cast<void **>(&sql_security));
    f(reinterpret_cast<void **>(&rename_to));
 }

--- a/src/Parsers/ASTSQLSecurity.cpp
+++ b/src/Parsers/ASTSQLSecurity.cpp
@ -7,7 +7,7 @@ namespace DB

 void ASTSQLSecurity::formatImpl(const FormatSettings & settings, FormatState & state, FormatStateStacked frame) const
 {
-    if (!type.has_value())
+    if (!type)
        return;

    if (definer || is_definer_current_user)
--- a/src/Processors/QueryPlan/Optimizations/Optimizations.h
+++ b/src/Processors/QueryPlan/Optimizations/Optimizations.h
@ -107,7 +107,7 @@ struct Frame
 using Stack = std::vector<Frame>;

 /// Second pass optimizations
-void optimizePrimaryKeyCondition(const Stack & stack);
+void optimizePrimaryKeyConditionAndLimit(const Stack & stack);
 void optimizePrewhere(Stack & stack, QueryPlan::Nodes & nodes);
 void optimizeReadInOrder(QueryPlan::Node & node, QueryPlan::Nodes & nodes);
 void optimizeAggregationInOrder(QueryPlan::Node & node, QueryPlan::Nodes &);
--- a/src/Processors/QueryPlan/Optimizations/optimizePrimaryKeyConditionAndLimit.cpp
+++ b/src/Processors/QueryPlan/Optimizations/optimizePrimaryKeyConditionAndLimit.cpp
@ -1,13 +1,13 @@
 #include <Processors/QueryPlan/Optimizations/Optimizations.h>
 #include <Processors/QueryPlan/ExpressionStep.h>
 #include <Processors/QueryPlan/FilterStep.h>
-#include <Processors/QueryPlan/ReadFromMergeTree.h>
+#include <Processors/QueryPlan/LimitStep.h>
 #include <Processors/QueryPlan/SourceStepWithFilter.h>

 namespace DB::QueryPlanOptimizations
 {

-void optimizePrimaryKeyCondition(const Stack & stack)
+void optimizePrimaryKeyConditionAndLimit(const Stack & stack)
 {
    const auto & frame = stack.back();

@ -26,16 +26,26 @@ void optimizePrimaryKeyCondition(const Stack & stack)
    for (auto iter = stack.rbegin() + 1; iter != stack.rend(); ++iter)
    {
        if (auto * filter_step = typeid_cast<FilterStep *>(iter->node->step.get()))
+        {
            source_step_with_filter->addFilter(filter_step->getExpression(), filter_step->getFilterColumnName());
-
+        }
+        else if (auto * limit_step = typeid_cast<LimitStep *>(iter->node->step.get()))
+        {
+            source_step_with_filter->setLimit(limit_step->getLimitForSorting());
+            break;
+        }
+        else if (typeid_cast<ExpressionStep *>(iter->node->step.get()))
+        {
            /// Note: actually, plan optimizations merge Filter and Expression steps.
            /// Ideally, chain should look like (Expression -> ...) -> (Filter -> ...) -> ReadFromStorage,
            /// So this is likely not needed.
-        else if (typeid_cast<ExpressionStep *>(iter->node->step.get()))
            continue;
+        }
        else
+        {
            break;
        }
+    }

    source_step_with_filter->applyFilters();
 }
--- a/src/Processors/QueryPlan/Optimizations/optimizeTree.cpp
+++ b/src/Processors/QueryPlan/Optimizations/optimizeTree.cpp
@ -115,7 +115,7 @@ void optimizeTreeSecondPass(const QueryPlanOptimizationSettings & optimization_s

    while (!stack.empty())
    {
-        optimizePrimaryKeyCondition(stack);
+        optimizePrimaryKeyConditionAndLimit(stack);

        /// NOTE: optimizePrewhere can modify the stack.
        /// Prewhere optimization relies on PK optimization (getConditionSelectivityEstimatorByPredicate)
--- a/src/Processors/QueryPlan/ReadFromMergeTree.cpp
+++ b/src/Processors/QueryPlan/ReadFromMergeTree.cpp
@ -120,6 +120,7 @@ namespace ProfileEvents
    extern const Event SelectedParts;
    extern const Event SelectedRanges;
    extern const Event SelectedMarks;
+    extern const Event SelectQueriesWithPrimaryKeyUsage;
 }

 namespace DB
@ -500,11 +501,11 @@ Pipe ReadFromMergeTree::readInOrder(
    Names required_columns,
    PoolSettings pool_settings,
    ReadType read_type,
-    UInt64 limit)
+    UInt64 read_limit)
 {
    /// For reading in order it makes sense to read only
    /// one range per task to reduce number of read rows.
-    bool has_limit_below_one_block = read_type != ReadType::Default && limit && limit < block_size.max_block_size_rows;
+    bool has_limit_below_one_block = read_type != ReadType::Default && read_limit && read_limit < block_size.max_block_size_rows;
    MergeTreeReadPoolPtr pool;

    if (is_parallel_reading_from_replicas)
@ -1569,12 +1570,18 @@ ReadFromMergeTree::AnalysisResultPtr ReadFromMergeTree::selectRangesToRead(
    if (indexes->part_values && indexes->part_values->empty())
        return std::make_shared<AnalysisResult>(std::move(result));

-    if (settings.force_primary_key && indexes->key_condition.alwaysUnknownOrTrue())
+    if (indexes->key_condition.alwaysUnknownOrTrue())
+    {
+        if (settings.force_primary_key)
        {
            throw Exception(ErrorCodes::INDEX_NOT_USED,
                "Primary key ({}) is not used and setting 'force_primary_key' is set",
                fmt::join(primary_key_column_names, ", "));
        }
+    } else
+    {
+        ProfileEvents::increment(ProfileEvents::SelectQueriesWithPrimaryKeyUsage);
+    }

    LOG_DEBUG(log, "Key condition: {}", indexes->key_condition.toString());

@ -1669,7 +1676,7 @@ ReadFromMergeTree::AnalysisResultPtr ReadFromMergeTree::selectRangesToRead(
    return std::make_shared<AnalysisResult>(std::move(result));
 }

-bool ReadFromMergeTree::requestReadingInOrder(size_t prefix_size, int direction, size_t limit)
+bool ReadFromMergeTree::requestReadingInOrder(size_t prefix_size, int direction, size_t read_limit)
 {
    /// if dirction is not set, use current one
    if (!direction)
@ -1680,7 +1687,7 @@ bool ReadFromMergeTree::requestReadingInOrder(size_t prefix_size, int direction,
    if (direction != 1 && query_info.isFinal())
        return false;

-    query_info.input_order_info = std::make_shared<InputOrderInfo>(SortDescription{}, prefix_size, direction, limit);
+    query_info.input_order_info = std::make_shared<InputOrderInfo>(SortDescription{}, prefix_size, direction, read_limit);
    reader_settings.read_in_order = true;

    /// In case or read-in-order, don't create too many reading streams.
--- a/src/Processors/QueryPlan/ReadFromSystemNumbersStep.cpp
+++ b/src/Processors/QueryPlan/ReadFromSystemNumbersStep.cpp
@ -393,7 +393,7 @@ ReadFromSystemNumbersStep::ReadFromSystemNumbersStep(
    , num_streams{num_streams_}
    , limit_length_and_offset(InterpreterSelectQuery::getLimitLengthAndOffset(query_info.query->as<ASTSelectQuery &>(), context))
    , should_pushdown_limit(shouldPushdownLimit(query_info, limit_length_and_offset.first))
-    , limit(query_info.limit)
+    , query_info_limit(query_info.limit)
    , storage_limits(query_info.storage_limits)
 {
    storage_snapshot->check(column_names);
@ -563,7 +563,7 @@ Pipe ReadFromSystemNumbersStep::makePipe()
        {
            auto rows_appr = (*numbers_storage.limit - 1) / numbers_storage.step + 1;
            if (limit > 0 && limit < rows_appr)
-                rows_appr = limit;
+                rows_appr = query_info_limit;
            source->addTotalRowsApprox(rows_appr);
        }

--- a/src/Processors/QueryPlan/ReadFromSystemNumbersStep.h
+++ b/src/Processors/QueryPlan/ReadFromSystemNumbersStep.h
@ -41,7 +41,7 @@ private:
    size_t num_streams;
    std::pair<UInt64, UInt64> limit_length_and_offset;
    bool should_pushdown_limit;
-    UInt64 limit;
+    UInt64 query_info_limit;
    std::shared_ptr<const StorageLimitsList> storage_limits;
 };

--- a/src/Processors/QueryPlan/SourceStepWithFilter.h
+++ b/src/Processors/QueryPlan/SourceStepWithFilter.h
@ -8,8 +8,9 @@
 namespace DB
 {

-/** Source step that can use filters for more efficient pipeline initialization.
+/** Source step that can use filters and limit for more efficient pipeline initialization.
  * Filters must be added before pipeline initialization.
+  * Limit must be set before pipeline initialization.
  */
 class SourceStepWithFilter : public ISourceStep
 {
@ -49,6 +50,11 @@ public:
        filter_dags.push_back(std::move(filter_dag));
    }

+    void setLimit(size_t limit_value)
+    {
+        limit = limit_value;
+    }
+
    /// Apply filters that can optimize reading from storage.
    void applyFilters()
    {
@ -72,6 +78,7 @@ protected:
    PrewhereInfoPtr prewhere_info;
    StorageSnapshotPtr storage_snapshot;
    ContextPtr context;
+    std::optional<size_t> limit;

    ActionsDAGPtr filter_actions_dag;

--- a/src/Storages/StoragePostgreSQL.cpp
+++ b/src/Storages/StoragePostgreSQL.cpp
@ -35,9 +35,12 @@
 #include <Parsers/getInsertQuery.h>
 #include <Parsers/ASTFunction.h>

+#include <Processors/QueryPlan/QueryPlan.h>
+#include <Processors/QueryPlan/SourceStepWithFilter.h>
 #include <Processors/Sinks/SinkToStorage.h>

 #include <QueryPipeline/Pipe.h>
+#include <QueryPipeline/QueryPipelineBuilder.h>

 #include <Storages/StorageFactory.h>
 #include <Storages/transformQueryForExternalDatabase.h>
@ -106,28 +109,79 @@ ColumnsDescription StoragePostgreSQL::getTableStructureFromData(
    return ColumnsDescription{columns_info->columns};
 }

-Pipe StoragePostgreSQL::read(
-    const Names & column_names_,
-    const StorageSnapshotPtr & storage_snapshot,
-    SelectQueryInfo & query_info_,
-    ContextPtr context_,
-    QueryProcessingStage::Enum /*processed_stage*/,
-    size_t max_block_size_,
-    size_t /*num_streams*/)
+namespace
 {
-    storage_snapshot->check(column_names_);
+
+class ReadFromPostgreSQL : public SourceStepWithFilter
+{
+public:
+    ReadFromPostgreSQL(
+        const Names & column_names_,
+        const SelectQueryInfo & query_info_,
+        const StorageSnapshotPtr & storage_snapshot_,
+        const ContextPtr & context_,
+        Block sample_block,
+        size_t max_block_size_,
+        String remote_table_schema_,
+        String remote_table_name_,
+        postgres::ConnectionHolderPtr connection_)
+        : SourceStepWithFilter(DataStream{.header = std::move(sample_block)}, column_names_, query_info_, storage_snapshot_, context_)
+        , logger(getLogger("ReadFromPostgreSQL"))
+        , max_block_size(max_block_size_)
+        , remote_table_schema(remote_table_schema_)
+        , remote_table_name(remote_table_name_)
+        , connection(std::move(connection_))
+    {
+    }
+
+    std::string getName() const override { return "ReadFromPostgreSQL"; }
+
+    void initializePipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings &) override
+    {
+        std::optional<size_t> transform_query_limit;
+        if (limit && !filter_actions_dag)
+            transform_query_limit = limit;

        /// Connection is already made to the needed database, so it should not be present in the query;
        /// remote_table_schema is empty if it is not specified, will access only table_name.
        String query = transformQueryForExternalDatabase(
-        query_info_,
-        column_names_,
+            query_info,
+            required_source_columns,
            storage_snapshot->metadata->getColumns().getOrdinary(),
-        IdentifierQuotingStyle::DoubleQuotes, LiteralEscapingStyle::PostgreSQL, remote_table_schema, remote_table_name, context_);
-    LOG_TRACE(log, "Query: {}", query);
+            IdentifierQuotingStyle::DoubleQuotes,
+            LiteralEscapingStyle::PostgreSQL,
+            remote_table_schema,
+            remote_table_name,
+            context,
+            transform_query_limit);
+        LOG_TRACE(logger, "Query: {}", query);
+
+        pipeline.init(Pipe(std::make_shared<PostgreSQLSource<>>(std::move(connection), query, getOutputStream().header, max_block_size)));
+    }
+
+    LoggerPtr logger;
+    size_t max_block_size;
+    String remote_table_schema;
+    String remote_table_name;
+    postgres::ConnectionHolderPtr connection;
+};
+
+}
+
+void StoragePostgreSQL::read(
+    QueryPlan & query_plan,
+    const Names & column_names,
+    const StorageSnapshotPtr & storage_snapshot,
+    SelectQueryInfo & query_info,
+    ContextPtr local_context,
+    QueryProcessingStage::Enum /*processed_stage*/,
+    size_t max_block_size,
+    size_t /*num_streams*/)
+{
+    storage_snapshot->check(column_names);

    Block sample_block;
-    for (const String & column_name : column_names_)
+    for (const String & column_name : column_names)
    {
        auto column_data = storage_snapshot->metadata->getColumns().getPhysical(column_name);
        WhichDataType which(column_data.type);
@ -136,7 +190,17 @@ Pipe StoragePostgreSQL::read(
        sample_block.insert({ column_data.type, column_data.name });
    }

-    return Pipe(std::make_shared<PostgreSQLSource<>>(pool->get(), query, sample_block, max_block_size_));
+    auto reading = std::make_unique<ReadFromPostgreSQL>(
+        column_names,
+        query_info,
+        storage_snapshot,
+        local_context,
+        sample_block,
+        max_block_size,
+        remote_table_schema,
+        remote_table_name,
+        pool->get());
+    query_plan.addStep(std::move(reading));
 }


--- a/src/Storages/StoragePostgreSQL.h
+++ b/src/Storages/StoragePostgreSQL.h
@ -37,11 +37,12 @@ public:

    String getName() const override { return "PostgreSQL"; }

-    Pipe read(
+    void read(
+        QueryPlan & query_plan,
        const Names & column_names,
        const StorageSnapshotPtr & storage_snapshot,
        SelectQueryInfo & query_info,
-        ContextPtr context,
+        ContextPtr local_context,
        QueryProcessingStage::Enum processed_stage,
        size_t max_block_size,
        size_t num_streams) override;
--- a/src/Storages/transformQueryForExternalDatabase.cpp
+++ b/src/Storages/transformQueryForExternalDatabase.cpp
@ -288,7 +288,8 @@ String transformQueryForExternalDatabaseImpl(
    LiteralEscapingStyle literal_escaping_style,
    const String & database,
    const String & table,
-    ContextPtr context)
+    ContextPtr context,
+    std::optional<size_t> limit)
 {
    bool strict = context->getSettingsRef().external_table_strict_query;

@ -374,6 +375,9 @@ String transformQueryForExternalDatabaseImpl(
        select->setExpression(ASTSelectQuery::Expression::WHERE, std::move(original_where));
    }

+    if (limit)
+        select->setExpression(ASTSelectQuery::Expression::LIMIT_LENGTH, std::make_shared<ASTLiteral>(*limit));
+
    ASTPtr select_ptr = select;
    dropAliases(select_ptr);

@ -399,7 +403,8 @@ String transformQueryForExternalDatabase(
    LiteralEscapingStyle literal_escaping_style,
    const String & database,
    const String & table,
-    ContextPtr context)
+    ContextPtr context,
+    std::optional<size_t> limit)
 {
    if (!query_info.syntax_analyzer_result)
    {
@ -424,7 +429,8 @@ String transformQueryForExternalDatabase(
            literal_escaping_style,
            database,
            table,
-            context);
+            context,
+            limit);
    }

    auto clone_query = query_info.query->clone();
@ -436,7 +442,8 @@ String transformQueryForExternalDatabase(
        literal_escaping_style,
        database,
        table,
-        context);
+        context,
+        limit);
 }

 }
--- a/src/Storages/transformQueryForExternalDatabase.h
+++ b/src/Storages/transformQueryForExternalDatabase.h
@ -21,6 +21,8 @@ class IAST;
  * and WHERE contains subset of (AND-ed) conditions from original query,
  * that contain only compatible expressions.
  *
+  * If limit is passed additionally apply LIMIT in result query.
+  *
  * Compatible expressions are comparisons of identifiers, constants, and logical operations on them.
  *
  * Throws INCORRECT_QUERY if external_table_strict_query (from context settings)
@ -34,6 +36,7 @@ String transformQueryForExternalDatabase(
    LiteralEscapingStyle literal_escaping_style,
    const String & database,
    const String & table,
-    ContextPtr context);
+    ContextPtr context,
+    std::optional<size_t> limit = {});

 }
--- a/tests/ci/ci.py
+++ b/tests/ci/ci.py
@ -1065,18 +1065,19 @@ def main() -> int:
            )

            # rerun helper check
-            # FIXME: remove rerun_helper check and rely on ci cache only
            if check_name not in (
                CI.JobNames.BUILD_CHECK,
            ):  # we might want to rerun build report job
                rerun_helper = RerunHelper(commit, check_name_with_group)
                if rerun_helper.is_already_finished_by_status():
+                    print("WARNING: Rerunning job with GH status ")
                    status = rerun_helper.get_finished_status()
                    assert status
-                    previous_status = status.state
                    print("::group::Commit Status")
                    print(status)
                    print("::endgroup::")
+                    # FIXME: try rerun, even if status is present. To enable manual restart via GH interface
+                    # previous_status = status.state

            # ci cache check
            if not previous_status and not ci_settings.no_ci_cache:
--- a/tests/integration/test_MemoryTracking/configs/no_system_log.xml
+++ b/tests/integration/test_MemoryTracking/configs/no_system_log.xml
@ -5,6 +5,7 @@
    <query_log remove="remove" />
    <query_views_log remove="remove" />
    <metric_log remove="remove"/>
+    <error_log remove="remove"/>
    <text_log remove="remove"/>
    <trace_log remove="remove"/>
    <asynchronous_metric_log remove="remove" />
--- a/tests/integration/test_backup_restore_new/test.py
+++ b/tests/integration/test_backup_restore_new/test.py
@ -1,5 +1,4 @@
 import pytest
-import asyncio
 import glob
 import re
 import random
@ -1486,6 +1485,7 @@ def test_backup_all(exclude_system_log_tables):
            "processors_profile_log",
            "asynchronous_insert_log",
            "backup_log",
+            "error_log",
        ]
        exclude_from_backup += ["system." + table_name for table_name in log_tables]

--- a/tests/integration/test_backup_restore_s3/test.py
+++ b/tests/integration/test_backup_restore_s3/test.py
@ -627,67 +627,126 @@ def test_user_specific_auth(start_cluster):
    create_user("superuser2")
    create_user("regularuser")

-    node.query("CREATE TABLE specific_auth (col UInt64) ENGINE=Memory")
+    node.query("CREATE TABLE specific_auth (col UInt64) ENGINE=MergeTree ORDER BY col")
+    node.query("INSERT INTO specific_auth VALUES (1)")

-    assert "Access" in node.query_and_get_error(
-        "BACKUP TABLE specific_auth TO S3('http://minio1:9001/root/data/backups/limited/backup1.zip')"
+    def backup_restore(backup, user, should_fail, on_cluster=False, base_backup=None):
+        on_cluster_clause = "ON CLUSTER 'cluster'" if on_cluster else ""
+        base_backup = (
+            f" SETTINGS base_backup = {base_backup}" if base_backup is not None else ""
        )
-    assert "Access" in node.query_and_get_error(
-        "BACKUP TABLE specific_auth TO S3('http://minio1:9001/root/data/backups/limited/backup1.zip')",
+        backup_query = (
+            f"BACKUP TABLE specific_auth {on_cluster_clause} TO {backup} {base_backup}"
+        )
+        restore_query = f"RESTORE TABLE specific_auth {on_cluster_clause} FROM {backup}"
+
+        if should_fail:
+            assert "Access" in node.query_and_get_error(backup_query, user=user)
+        else:
+            node.query(backup_query, user=user)
+            node.query("DROP TABLE specific_auth SYNC")
+            node.query(restore_query, user=user)
+
+    backup_restore(
+        "S3('http://minio1:9001/root/data/backups/limited/backup1/')",
+        user=None,
+        should_fail=True,
+    )
+
+    backup_restore(
+        "S3('http://minio1:9001/root/data/backups/limited/backup1/')",
        user="regularuser",
+        should_fail=True,
    )

-    node.query(
-        "BACKUP TABLE specific_auth TO S3('http://minio1:9001/root/data/backups/limited/backup1.zip')",
-        user="superuser1",
-    )
-    node.query(
-        "RESTORE TABLE specific_auth FROM S3('http://minio1:9001/root/data/backups/limited/backup1.zip')",
+    backup_restore(
+        "S3('http://minio1:9001/root/data/backups/limited/backup1/')",
        user="superuser1",
+        should_fail=False,
    )

-    node.query(
-        "BACKUP TABLE specific_auth TO S3('http://minio1:9001/root/data/backups/limited/backup2.zip')",
-        user="superuser2",
-    )
-    node.query(
-        "RESTORE TABLE specific_auth FROM S3('http://minio1:9001/root/data/backups/limited/backup2.zip')",
+    backup_restore(
+        "S3('http://minio1:9001/root/data/backups/limited/backup2/')",
        user="superuser2",
+        should_fail=False,
    )

    assert "Access" in node.query_and_get_error(
-        "RESTORE TABLE specific_auth FROM S3('http://minio1:9001/root/data/backups/limited/backup1.zip')",
+        "RESTORE TABLE specific_auth FROM S3('http://minio1:9001/root/data/backups/limited/backup1/')",
        user="regularuser",
    )

-    assert "HTTP response code: 403" in node.query_and_get_error(
-        "SELECT * FROM s3('http://minio1:9001/root/data/backups/limited/backup1.zip', 'RawBLOB')",
+    node.query("INSERT INTO specific_auth VALUES (2)")
+
+    backup_restore(
+        "S3('http://minio1:9001/root/data/backups/limited/backup1_inc/')",
        user="regularuser",
+        should_fail=True,
+        base_backup="S3('http://minio1:9001/root/data/backups/limited/backup1/')",
    )

-    node.query(
-        "SELECT * FROM s3('http://minio1:9001/root/data/backups/limited/backup1.zip', 'RawBLOB')",
+    backup_restore(
+        "S3('http://minio1:9001/root/data/backups/limited/backup1_inc/')",
        user="superuser1",
+        should_fail=False,
+        base_backup="S3('http://minio1:9001/root/data/backups/limited/backup1/')",
+    )
+
+    assert "Access" in node.query_and_get_error(
+        "RESTORE TABLE specific_auth FROM S3('http://minio1:9001/root/data/backups/limited/backup1_inc/')",
+        user="regularuser",
    )

    assert "Access Denied" in node.query_and_get_error(
-        "BACKUP TABLE specific_auth ON CLUSTER 'cluster' TO S3('http://minio1:9001/root/data/backups/limited/backup3/')",
+        "SELECT * FROM s3('http://minio1:9001/root/data/backups/limited/backup1/*', 'RawBLOB')",
        user="regularuser",
    )

    node.query(
-        "BACKUP TABLE specific_auth ON CLUSTER 'cluster' TO S3('http://minio1:9001/root/data/backups/limited/backup3/')",
+        "SELECT * FROM s3('http://minio1:9001/root/data/backups/limited/backup1/*', 'RawBLOB')",
        user="superuser1",
    )

+    backup_restore(
+        "S3('http://minio1:9001/root/data/backups/limited/backup3/')",
+        user="regularuser",
+        should_fail=True,
+        on_cluster=True,
+    )
+
+    backup_restore(
+        "S3('http://minio1:9001/root/data/backups/limited/backup3/')",
+        user="superuser1",
+        should_fail=False,
+        on_cluster=True,
+    )
+
    assert "Access Denied" in node.query_and_get_error(
        "RESTORE TABLE specific_auth ON CLUSTER 'cluster' FROM S3('http://minio1:9001/root/data/backups/limited/backup3/')",
        user="regularuser",
    )

-    node.query(
-        "RESTORE TABLE specific_auth ON CLUSTER 'cluster' FROM S3('http://minio1:9001/root/data/backups/limited/backup3/')",
+    node.query("INSERT INTO specific_auth VALUES (3)")
+
+    backup_restore(
+        "S3('http://minio1:9001/root/data/backups/limited/backup3_inc/')",
+        user="regularuser",
+        should_fail=True,
+        on_cluster=True,
+        base_backup="S3('http://minio1:9001/root/data/backups/limited/backup3/')",
+    )
+
+    backup_restore(
+        "S3('http://minio1:9001/root/data/backups/limited/backup3_inc/')",
        user="superuser1",
+        should_fail=False,
+        on_cluster=True,
+        base_backup="S3('http://minio1:9001/root/data/backups/limited/backup3/')",
+    )
+
+    assert "Access Denied" in node.query_and_get_error(
+        "RESTORE TABLE specific_auth ON CLUSTER 'cluster' FROM S3('http://minio1:9001/root/data/backups/limited/backup3_inc/')",
+        user="regularuser",
    )

    assert "Access Denied" in node.query_and_get_error(
--- a/tests/integration/test_checking_s3_blobs_paranoid/test.py
+++ b/tests/integration/test_checking_s3_blobs_paranoid/test.py
@ -300,7 +300,7 @@ def test_when_s3_broken_pipe_at_upload_is_retried(cluster, broken_s3):
        LIMIT 1000000
        SETTINGS
            s3_max_single_part_upload_size=100,
-            s3_min_upload_part_size=100000,
+            s3_min_upload_part_size=1000000,
            s3_check_objects_after_upload=0
        """,
        query_id=insert_query_id,
@ -311,7 +311,7 @@ def test_when_s3_broken_pipe_at_upload_is_retried(cluster, broken_s3):
    )

    assert create_multipart == 1
-    assert upload_parts == 69
+    assert upload_parts == 7
    assert s3_errors == 3

    broken_s3.setup_at_part_upload(
--- a/tests/integration/test_config_xml_full/configs/config.d/error_log.xml
+++ b/tests/integration/test_config_xml_full/configs/config.d/error_log.xml
@ -0,0 +1,8 @@
+<clickhouse>
+    <error_log>
+        <database>system</database>
+        <table>error_log</table>
+        <flush_interval_milliseconds>7500</flush_interval_milliseconds>
+        <collect_interval_milliseconds>1000</collect_interval_milliseconds>
+    </error_log>
+</clickhouse>
--- a/tests/integration/test_config_xml_full/configs/config.xml
+++ b/tests/integration/test_config_xml_full/configs/config.xml
@ -756,6 +756,14 @@
        <collect_interval_milliseconds>1000</collect_interval_milliseconds>
    </metric_log>

+    <!-- Error log contains rows with current values of errors collected with "collect_interval_milliseconds" interval. -->
+    <error_log>
+        <database>system</database>
+        <table>error_log</table>
+        <flush_interval_milliseconds>7500</flush_interval_milliseconds>
+        <collect_interval_milliseconds>1000</collect_interval_milliseconds>
+    </error_log>
+
    <!--
        Asynchronous metric log contains values of metrics from
        system.asynchronous_metrics.
--- a/tests/integration/test_config_xml_full/test.py
+++ b/tests/integration/test_config_xml_full/test.py
@ -1,10 +1,3 @@
-import time
-import threading
-from os import path as p, unlink
-from tempfile import NamedTemporaryFile
-
-import helpers
-import pytest
 from helpers.cluster import ClickHouseCluster


@ -16,6 +9,7 @@ def test_xml_full_conf():

    all_confd = [
        "configs/config.d/access_control.xml",
+        "configs/config.d/error_log.xml",
        "configs/config.d/keeper_port.xml",
        "configs/config.d/logging_no_rotate.xml",
        "configs/config.d/log_to_console.xml",
--- a/tests/integration/test_config_xml_main/configs/config.d/error_log.yaml
+++ b/tests/integration/test_config_xml_main/configs/config.d/error_log.yaml
@ -0,0 +1,6 @@
+error_log:
+  database: system
+  table: error_log 
+  flush_interval_milliseconds: 7500
+  collect_interval_milliseconds: 1000
+
--- a/tests/integration/test_config_xml_main/configs/config.xml
+++ b/tests/integration/test_config_xml_main/configs/config.xml
@ -129,6 +129,13 @@
        <collect_interval_milliseconds>1000</collect_interval_milliseconds>
    </metric_log>

+    <error_log>
+        <database>system</database>
+        <table>error_log</table>
+        <flush_interval_milliseconds>7500</flush_interval_milliseconds>
+        <collect_interval_milliseconds>1000</collect_interval_milliseconds>
+    </error_log>
+
    <asynchronous_metric_log>
        <database>system</database>
        <table>asynchronous_metric_log</table>
--- a/tests/integration/test_config_xml_main/test.py
+++ b/tests/integration/test_config_xml_main/test.py
@ -1,10 +1,3 @@
-import time
-import threading
-from os import path as p, unlink
-from tempfile import NamedTemporaryFile
-
-import helpers
-import pytest
 from helpers.cluster import ClickHouseCluster


@ -16,6 +9,7 @@ def test_xml_main_conf():

    all_confd = [
        "configs/config.d/access_control.yaml",
+        "configs/config.d/error_log.yaml",
        "configs/config.d/keeper_port.yaml",
        "configs/config.d/logging_no_rotate.yaml",
        "configs/config.d/log_to_console.yaml",
--- a/tests/integration/test_config_xml_yaml_mix/configs/config.d/error_log.xml
+++ b/tests/integration/test_config_xml_yaml_mix/configs/config.d/error_log.xml
@ -0,0 +1,8 @@
+<clickhouse>
+    <error_log>
+        <database>system</database>
+        <table>error_log</table>
+        <flush_interval_milliseconds>7500</flush_interval_milliseconds>
+        <collect_interval_milliseconds>1000</collect_interval_milliseconds>
+    </error_log>
+</clickhouse>
--- a/tests/integration/test_config_xml_yaml_mix/configs/config.xml
+++ b/tests/integration/test_config_xml_yaml_mix/configs/config.xml
@ -129,6 +129,13 @@
        <collect_interval_milliseconds>1000</collect_interval_milliseconds>
    </metric_log>

+    <error_log>
+        <database>system</database>
+        <table>error_log</table>
+        <flush_interval_milliseconds>7500</flush_interval_milliseconds>
+        <collect_interval_milliseconds>1000</collect_interval_milliseconds>
+    </error_log>
+
    <asynchronous_metric_log>
        <database>system</database>
        <table>asynchronous_metric_log</table>
--- a/tests/integration/test_config_xml_yaml_mix/test.py
+++ b/tests/integration/test_config_xml_yaml_mix/test.py
@ -1,10 +1,3 @@
-import time
-import threading
-from os import path as p, unlink
-from tempfile import NamedTemporaryFile
-
-import helpers
-import pytest
 from helpers.cluster import ClickHouseCluster


@ -17,6 +10,7 @@ def test_extra_yaml_mix():
    all_confd = [
        "configs/config.d/0_common_instance_config.yaml",
        "configs/config.d/access_control.yaml",
+        "configs/config.d/error_log.xml",
        "configs/config.d/keeper_port.xml",
        "configs/config.d/logging_no_rotate.xml",
        "configs/config.d/log_to_console.yaml",
--- a/tests/integration/test_config_yaml_full/configs/config.d/error_log.yaml
+++ b/tests/integration/test_config_yaml_full/configs/config.d/error_log.yaml
@ -0,0 +1,6 @@
+error_log:
+  database: system
+  table: error_log
+  flush_interval_milliseconds: 7500
+  collect_interval_milliseconds: 1000
+
--- a/tests/integration/test_config_yaml_full/configs/config.yaml
+++ b/tests/integration/test_config_yaml_full/configs/config.yaml
@ -90,6 +90,11 @@ metric_log:
  table: metric_log
  flush_interval_milliseconds: 7500
  collect_interval_milliseconds: 1000
+error_log:
+  database: system
+  table: error_log
+  flush_interval_milliseconds: 7500
+  collect_interval_milliseconds: 1000
 asynchronous_metric_log:
  database: system
  table: asynchronous_metric_log
--- a/tests/integration/test_config_yaml_full/test.py
+++ b/tests/integration/test_config_yaml_full/test.py
@ -1,10 +1,3 @@
-import time
-import threading
-from os import path as p, unlink
-from tempfile import NamedTemporaryFile
-
-import helpers
-import pytest
 from helpers.cluster import ClickHouseCluster


@ -17,6 +10,7 @@ def test_yaml_full_conf():
    all_confd = [
        "configs/config.d/0_common_instance_config.yaml",
        "configs/config.d/access_control.yaml",
+        "configs/config.d/error_log.yaml",
        "configs/config.d/keeper_port.yaml",
        "configs/config.d/logging_no_rotate.yaml",
        "configs/config.d/log_to_console.yaml",
--- a/tests/integration/test_config_yaml_main/configs/config.d/error_log.xml
+++ b/tests/integration/test_config_yaml_main/configs/config.d/error_log.xml
@ -0,0 +1,8 @@
+<clickhouse>
+    <error_log>
+        <database>system</database>
+        <table>error_log</table>
+        <flush_interval_milliseconds>7500</flush_interval_milliseconds>
+        <collect_interval_milliseconds>1000</collect_interval_milliseconds>
+    </error_log>
+</clickhouse>
--- a/tests/integration/test_config_yaml_main/configs/config.yaml
+++ b/tests/integration/test_config_yaml_main/configs/config.yaml
@ -90,6 +90,11 @@ metric_log:
  table: metric_log
  flush_interval_milliseconds: 7500
  collect_interval_milliseconds: 1000
+error_log:
+  database: system
+  table: error_log
+  flush_interval_milliseconds: 7500
+  collect_interval_milliseconds: 1000
 asynchronous_metric_log:
  database: system
  table: asynchronous_metric_log
--- a/tests/integration/test_config_yaml_main/test.py
+++ b/tests/integration/test_config_yaml_main/test.py
@ -1,10 +1,3 @@
-import time
-import threading
-from os import path as p, unlink
-from tempfile import NamedTemporaryFile
-
-import helpers
-import pytest
 from helpers.cluster import ClickHouseCluster


@ -17,6 +10,7 @@ def test_yaml_main_conf():
    all_confd = [
        "configs/config.d/0_common_instance_config.yaml",
        "configs/config.d/access_control.xml",
+        "configs/config.d/error_log.xml",
        "configs/config.d/keeper_port.xml",
        "configs/config.d/logging_no_rotate.xml",
        "configs/config.d/log_to_console.xml",
--- a/tests/integration/test_input_format_parallel_parsing_memory_tracking/configs/conf.xml
+++ b/tests/integration/test_input_format_parallel_parsing_memory_tracking/configs/conf.xml
@ -6,6 +6,7 @@
    <query_log remove="remove" />
    <query_views_log remove="remove" />
    <metric_log remove="remove"/>
+    <error_log remove="remove"/>
    <text_log remove="remove"/>
    <trace_log remove="remove"/>
    <asynchronous_metric_log remove="remove" />
--- a/tests/integration/test_keeper_snapshots/test.py
+++ b/tests/integration/test_keeper_snapshots/test.py
@ -189,7 +189,7 @@ def test_invalid_snapshot(started_cluster):
                f"/var/lib/clickhouse/coordination/snapshots/{last_snapshot}",
            ]
        )
-        node.start_clickhouse(expected_to_fail=True)
+        node.start_clickhouse(start_wait_sec=120, expected_to_fail=True)
        assert node.contains_in_log(
            "Aborting because of failure to load from latest snapshot with index"
        )
--- a/tests/integration/test_system_flush_logs/test.py
+++ b/tests/integration/test_system_flush_logs/test.py
@ -2,7 +2,6 @@
 # pylint: disable=unused-argument
 # pylint: disable=redefined-outer-name

-import time
 import pytest
 from helpers.cluster import ClickHouseCluster
 from helpers.test_tools import assert_eq_with_retry
@ -22,6 +21,7 @@ system_logs = [
    ("system.part_log", 1),
    ("system.trace_log", 1),
    ("system.metric_log", 1),
+    ("system.error_log", 1),
 ]


--- a/tests/integration/test_system_logs_recreate/test.py
+++ b/tests/integration/test_system_logs_recreate/test.py
@ -30,6 +30,7 @@ def test_system_logs_recreate():
        "part_log",
        "trace_log",
        "metric_log",
+        "error_log",
    ]

    node.query("SYSTEM FLUSH LOGS")
--- a/tests/queries/0_stateless/01926_order_by_desc_limit.sql
+++ b/tests/queries/0_stateless/01926_order_by_desc_limit.sql
@ -12,10 +12,10 @@ INSERT INTO order_by_desc SELECT number, repeat('a', 1024) FROM numbers(1024 * 3
 OPTIMIZE TABLE order_by_desc FINAL;

 SELECT s FROM order_by_desc ORDER BY u DESC LIMIT 10 FORMAT Null
-SETTINGS max_memory_usage = '600M';
+SETTINGS max_memory_usage = '400M';

 SELECT s FROM order_by_desc ORDER BY u LIMIT 10 FORMAT Null
-SETTINGS max_memory_usage = '600M';
+SETTINGS max_memory_usage = '400M';

 SYSTEM FLUSH LOGS;

--- a/tests/queries/0_stateless/02152_http_external_tables_memory_tracking.reference
+++ b/tests/queries/0_stateless/02152_http_external_tables_memory_tracking.reference
@ -1,16 +0,0 @@
-Checking input_format_parallel_parsing=false&
-1
-Checking input_format_parallel_parsing=false&cancel_http_readonly_queries_on_client_close=1&readonly=1
-1
-Checking input_format_parallel_parsing=false&send_progress_in_http_headers=true
-1
-Checking input_format_parallel_parsing=false&cancel_http_readonly_queries_on_client_close=1&readonly=1&send_progress_in_http_headers=true
-1
-Checking input_format_parallel_parsing=true&
-1
-Checking input_format_parallel_parsing=true&cancel_http_readonly_queries_on_client_close=1&readonly=1
-1
-Checking input_format_parallel_parsing=true&send_progress_in_http_headers=true
-1
-Checking input_format_parallel_parsing=true&cancel_http_readonly_queries_on_client_close=1&readonly=1&send_progress_in_http_headers=true
-1
--- a/tests/queries/0_stateless/02152_http_external_tables_memory_tracking.sh
+++ b/tests/queries/0_stateless/02152_http_external_tables_memory_tracking.sh
@ -1,57 +0,0 @@
-#!/usr/bin/env bash
-# Tags: no-tsan, no-cpu-aarch64, no-parallel, no-debug
-# TSan does not supports tracing.
-# trace_log doesn't work on aarch64
-
-# Regression for proper release of Context,
-# via tracking memory of external tables.
-
-CURDIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
-# shellcheck source=../shell_config.sh
-. "$CURDIR"/../shell_config.sh
-
-tmp_file=$(mktemp "$CURDIR/clickhouse.XXXXXX.csv")
-trap 'rm $tmp_file' EXIT
-
-$CLICKHOUSE_CLIENT -q "SELECT toString(number) FROM numbers(1e6) FORMAT TSV" > "$tmp_file"
-
-function run_and_check()
-{
-    local query_id
-    query_id="$(${CLICKHOUSE_CURL} -sS "${CLICKHOUSE_URL}" --data-binary @- <<<'SELECT generateUUIDv4()')"
-
-    echo "Checking $*"
-
-    # Run query with external table (implicit StorageMemory user)
-    $CLICKHOUSE_CURL -sS -F "s=@$tmp_file;" "$CLICKHOUSE_URL&s_structure=key+Int&query=SELECT+count()+FROM+s&memory_profiler_sample_probability=1&max_untracked_memory=0&query_id=$query_id&$*" -o /dev/null
-
-    ${CLICKHOUSE_CURL} -sS "${CLICKHOUSE_URL}" --data-binary @- <<<'SYSTEM FLUSH LOGS'
-
-    # Check that temporary table had been destroyed.
-    ${CLICKHOUSE_CURL} -sS "${CLICKHOUSE_URL}&allow_introspection_functions=1" --data-binary @- <<<"
-    WITH arrayStringConcat(arrayMap(x -> demangle(addressToSymbol(x)), trace), '\n') AS sym
-    SELECT 1 FROM system.trace_log
-    PREWHERE
-        query_id = '$query_id' AND
-        trace_type = 'MemorySample' AND
-        /* only deallocations */
-        size < 0 AND
-        event_date >= yesterday()
-    WHERE
-        sym LIKE '%DB::StorageMemory::drop%\n%TemporaryTableHolder::~TemporaryTableHolder%'
-    LIMIT 1
-    "
-}
-
-for input_format_parallel_parsing in false true; do
-    query_args_variants=(
-        ""
-        "cancel_http_readonly_queries_on_client_close=1&readonly=1"
-        "send_progress_in_http_headers=true"
-        # nested progress callback
-        "cancel_http_readonly_queries_on_client_close=1&readonly=1&send_progress_in_http_headers=true"
-    )
-    for query_args in "${query_args_variants[@]}"; do
-        run_and_check "input_format_parallel_parsing=$input_format_parallel_parsing&$query_args"
-    done
-done
--- a/tests/queries/0_stateless/02340_parts_refcnt_mergetree.sh
+++ b/tests/queries/0_stateless/02340_parts_refcnt_mergetree.sh
@ -16,7 +16,7 @@ function check_refcnt_for_table()
        -- queue may hold the parts lock for awhile as well
        system stop pulling replication log $table;
    "
-    $CLICKHOUSE_CLIENT --insert_keeper_fault_injection_probability=0 -q "insert into $table select number, number%4 from numbers(200)"
+    $CLICKHOUSE_CLIENT --insert_keeper_fault_injection_probability=0 -q "insert into $table select number, number%4 from numbers(2000)"

    local query_id
    query_id="$table-$(random_str 10)"
@ -35,7 +35,7 @@ function check_refcnt_for_table()
    )

    # Notes:
-    # - query may sleep 0.1*(200/4)=5 seconds maximum, it is enough to check system.parts
+    # - query may sleep 0.1*(2000/4)=50 seconds maximum, it is enough to check system.parts
    # - "part = 1" condition should prune all parts except first
    # - max_block_size=1 with index_granularity=1 will allow to cancel the query earlier
    $CLICKHOUSE_CLIENT "${args[@]}" -q "select sleepEachRow(0.1) from $table where part = 1" &
--- a/tests/queries/0_stateless/02884_create_view_with_sql_security_option.reference
+++ b/tests/queries/0_stateless/02884_create_view_with_sql_security_option.reference
@ -35,5 +35,8 @@ OK
 2	2
 6	6
 9	9
+===== TestInsertChain =====
 1000
 1000
+===== TestOnCluster =====
+1
--- a/tests/queries/0_stateless/02884_create_view_with_sql_security_option.sh
+++ b/tests/queries/0_stateless/02884_create_view_with_sql_security_option.sh
@ -260,6 +260,8 @@ EOF

 ${CLICKHOUSE_CLIENT} --user $user2 --query "SELECT * FROM $db.test_mv_row_2"

+echo "===== TestInsertChain ====="
+
 ${CLICKHOUSE_CLIENT} --multiquery <<EOF
 CREATE TABLE $db.session_events(
    clientId UUID,
@ -299,4 +301,16 @@ ${CLICKHOUSE_CLIENT} --user $user3 --query "INSERT INTO $db.session_events SELEC
 ${CLICKHOUSE_CLIENT} --user $user3 --query "SELECT count(*) FROM session_events"
 ${CLICKHOUSE_CLIENT} --query "SELECT count(*) FROM materialized_events"

+echo "===== TestOnCluster ====="
+${CLICKHOUSE_CLIENT} --multiquery <<EOF
+
+CREATE TABLE $db.test_cluster ON CLUSTER test_shard_localhost (a String) Engine = MergeTree() ORDER BY a FORMAT Null;
+CREATE TABLE $db.test_cluster_2 ON CLUSTER test_shard_localhost (a String) Engine = MergeTree() ORDER BY a FORMAT Null;
+CREATE MATERIALIZED VIEW $db.cluster_mv ON CLUSTER test_shard_localhost TO $db.test_cluster_2 AS SELECT * FROM $db.test_cluster FORMAT Null;
+ALTER TABLE $db.cluster_mv ON CLUSTER test_shard_localhost MODIFY DEFINER = $user3 FORMAT Null;
+EOF
+
+${CLICKHOUSE_CLIENT} --query "SHOW CREATE TABLE $db.cluster_mv" | grep -c "DEFINER = $user3"
+
+
 ${CLICKHOUSE_CLIENT} --query "DROP USER IF EXISTS $user1, $user2, $user3";
--- a/tests/queries/0_stateless/02922_deduplication_with_zero_copy.sh
+++ b/tests/queries/0_stateless/02922_deduplication_with_zero_copy.sh
@ -31,11 +31,11 @@ create table r2 (n int)
 function get_shared_locks()
 {
  table_shared_id="$1"
-  for part in $($CLICKHOUSE_KEEPER_CLIENT -q "ls /clickhouse/zero_copy/zero_copy_s3/${table_shared_id}")
+  for part in $($CLICKHOUSE_KEEPER_CLIENT -q "ls '/clickhouse/zero_copy/zero_copy_s3/${table_shared_id}'")
  do
-    for blob in $($CLICKHOUSE_KEEPER_CLIENT -q "ls /clickhouse/zero_copy/zero_copy_s3/${table_shared_id}/${part}")
+    for blob in $($CLICKHOUSE_KEEPER_CLIENT -q "ls '/clickhouse/zero_copy/zero_copy_s3/${table_shared_id}/${part}'")
    do
-      for lock in $($CLICKHOUSE_KEEPER_CLIENT -q "ls /clickhouse/zero_copy/zero_copy_s3/${table_shared_id}/${part}/${blob}")
+      for lock in $($CLICKHOUSE_KEEPER_CLIENT -q "ls '/clickhouse/zero_copy/zero_copy_s3/${table_shared_id}/${part}/${blob}'")
      do
        echo "/clickhouse/zero_copy/zero_copy_s3/${table_shared_id}/${part}/${blob}/${lock}"
      done
@ -48,7 +48,7 @@ function filter_temporary_locks()
 {
  while read -r lock
  do
-    owner="$($CLICKHOUSE_KEEPER_CLIENT -q "get_stat ${lock}" | grep 'ephemeralOwner' | sed 's/.*= //')"
+    owner="$($CLICKHOUSE_KEEPER_CLIENT -q "get_stat '${lock}'" | grep 'ephemeralOwner' | sed 's/.*= //')"
    if [[ "${owner}" -eq "0" ]]
    then
      echo "${lock}"
@ -111,7 +111,7 @@ export -f insert_duplicates
 export -f get_shared_locks
 export -f loop

-table_shared_id="$($CLICKHOUSE_KEEPER_CLIENT -q "get /test/02922/${CLICKHOUSE_DATABASE}/table/table_shared_id")"
+table_shared_id="$($CLICKHOUSE_KEEPER_CLIENT -q "get '/test/02922/${CLICKHOUSE_DATABASE}/table/table_shared_id'")"

 exit_code=0
 timeout 40 bash -c "loop '${table_shared_id}'" || exit_code="${?}"
@ -128,11 +128,11 @@ function list_keeper_nodes() {
  table_shared_id=$1

  echo "zero_copy:"
-  $CLICKHOUSE_KEEPER_CLIENT -q "ls /clickhouse/zero_copy/zero_copy_s3" | grep -o "${table_shared_id}" | \
+  $CLICKHOUSE_KEEPER_CLIENT -q "ls '/clickhouse/zero_copy/zero_copy_s3'" | grep -o "${table_shared_id}" | \
    sed "s/${table_shared_id}/<table_shared_id>/g" || :

  echo "tables:"
-  $CLICKHOUSE_KEEPER_CLIENT -q "ls /test/02922/${CLICKHOUSE_DATABASE}" | grep -o "table" || :
+  $CLICKHOUSE_KEEPER_CLIENT -q "ls '/test/02922/${CLICKHOUSE_DATABASE}'" | grep -o "table" || :
 }

 list_keeper_nodes "${table_shared_id}"
--- a/tests/queries/0_stateless/03164_selects_with_pk_usage_profile_event.reference
+++ b/tests/queries/0_stateless/03164_selects_with_pk_usage_profile_event.reference
@ -0,0 +1,8 @@
+selects_with_pk_usage
+0
+selects_with_pk_usage
+0
+selects_with_pk_usage
+1
+selects_with_pk_usage
+1
--- a/tests/queries/0_stateless/03164_selects_with_pk_usage_profile_event.sh
+++ b/tests/queries/0_stateless/03164_selects_with_pk_usage_profile_event.sh
@ -0,0 +1,101 @@
+#!/usr/bin/env bash
+
+# Tests profile event "SelectedMarksByPrimaryKeyUsage"
+
+CUR_DIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
+# shellcheck source=../shell_config.sh
+. "$CUR_DIR"/../shell_config.sh
+
+table_id="$(random_str 10)"
+
+$CLICKHOUSE_CLIENT -q "
+    DROP TABLE IF EXISTS table_$table_id;"
+
+$CLICKHOUSE_CLIENT -q "
+    CREATE TABLE table_$table_id (
+        pk Int64,
+        col1 Int64,
+        col2 Int64,
+        INDEX idx(col2) TYPE minmax
+    ) ENGINE = MergeTree ORDER BY pk PARTITION BY (pk % 2);";
+
+$CLICKHOUSE_CLIENT -q "
+    ALTER TABLE table_$table_id ADD PROJECTION proj (SELECT * ORDER BY col1);"
+
+# Populate two partitions with 50k rows each. Each partition has >1 granules.
+# We want SelectQueriesWithPrimaryKeyUsage to increase by +1 in each query, not by +1 per partition or by +1 per granule.
+$CLICKHOUSE_CLIENT -q "
+    INSERT INTO table_$table_id SELECT number, number, number FROM numbers(100000);"
+
+# Run SELECTs
+
+# -- No filter
+query_id="$(random_str 10)"
+$CLICKHOUSE_CLIENT --query_id "$query_id" -q "
+    SELECT count(*) FROM table_$table_id FORMAT Null;"
+$CLICKHOUSE_CLIENT -mn -q "
+    SYSTEM FLUSH LOGS;
+    SELECT
+        ProfileEvents['SelectQueriesWithPrimaryKeyUsage'] AS selects_with_pk_usage
+    FROM
+        system.query_log
+    WHERE
+        current_database = currentDatabase()
+        AND type = 'QueryFinish'
+        AND query_id = '$query_id'
+    FORMAT TSVWithNames;
+"
+
+# -- Filter on non-PK column. However, it has a minmax-index defined. We expect the profile event to not increase.
+query_id="$(random_str 10)"
+$CLICKHOUSE_CLIENT --query_id "$query_id" -q "
+    SELECT count(*) FROM table_$table_id WHERE col2 >= 50000 FORMAT Null;"
+$CLICKHOUSE_CLIENT -mn -q "
+    SYSTEM FLUSH LOGS;
+    SELECT
+        ProfileEvents['SelectQueriesWithPrimaryKeyUsage'] AS selects_with_pk_usage
+    FROM
+        system.query_log
+    WHERE
+        current_database = currentDatabase()
+        AND type = 'QueryFinish'
+        AND query_id = '$query_id'
+    FORMAT TSVWithNames;
+"
+
+# Filter on PK
+query_id="$(random_str 10)"
+$CLICKHOUSE_CLIENT --query_id "$query_id" -q "
+    SELECT count(*) FROM table_$table_id WHERE pk >= 50000 FORMAT Null;"
+$CLICKHOUSE_CLIENT -mn -q "
+    SYSTEM FLUSH LOGS;
+    SELECT
+        ProfileEvents['SelectQueriesWithPrimaryKeyUsage'] AS selects_with_pk_usage
+    FROM
+        system.query_log
+    WHERE
+        current_database = currentDatabase()
+        AND type = 'QueryFinish'
+        AND query_id = '$query_id'
+    FORMAT TSVWithNames;
+"
+
+# Filter on PK in projection
+query_id="$(random_str 10)"
+$CLICKHOUSE_CLIENT --query_id "$query_id" -q "
+    SELECT count(*) FROM table_$table_id WHERE col1 >= 50000 FORMAT Null;"
+$CLICKHOUSE_CLIENT -mn -q "
+    SYSTEM FLUSH LOGS;
+    SELECT
+        ProfileEvents['SelectQueriesWithPrimaryKeyUsage'] AS selects_with_pk_usage
+    FROM
+        system.query_log
+    WHERE
+        current_database = currentDatabase()
+        AND type = 'QueryFinish'
+        AND query_id = '$query_id'
+    FORMAT TSVWithNames;
+"
+
+$CLICKHOUSE_CLIENT -q "
+    DROP TABLE table_$table_id;"
--- a/tests/queries/0_stateless/03172_error_log_table_not_empty.reference
+++ b/tests/queries/0_stateless/03172_error_log_table_not_empty.reference
@ -0,0 +1,6 @@
+1
+1
+1
+1
+1
+1
--- a/tests/queries/0_stateless/03172_error_log_table_not_empty.sql
+++ b/tests/queries/0_stateless/03172_error_log_table_not_empty.sql
@ -0,0 +1,25 @@
+-- Throw three random errors: 111, 222 and 333
+SELECT throwIf(true, 'error_log', toInt16(111)) SETTINGS allow_custom_error_code_in_throwif=1; -- { serverError 111 }
+SELECT throwIf(true, 'error_log', toInt16(222)) SETTINGS allow_custom_error_code_in_throwif=1; -- { serverError 222 }
+SELECT throwIf(true, 'error_log', toInt16(333)) SETTINGS allow_custom_error_code_in_throwif=1; -- { serverError 333 }
+
+-- Wait for more than collect_interval_milliseconds to ensure system.error_log is flushed
+SELECT sleep(2) FORMAT NULL;
+SYSTEM FLUSH LOGS;
+
+-- Check that the three random errors are propagated
+SELECT sum(value) > 0 FROM system.error_log WHERE code = 111 AND event_time > now() - INTERVAL 1 MINUTE;
+SELECT sum(value) > 0 FROM system.error_log WHERE code = 222 AND event_time > now() - INTERVAL 1 MINUTE;
+SELECT sum(value) > 0 FROM system.error_log WHERE code = 333 AND event_time > now() - INTERVAL 1 MINUTE;
+
+-- Ensure that if we throw them again, they're still propagated
+SELECT throwIf(true, 'error_log', toInt16(111)) SETTINGS allow_custom_error_code_in_throwif=1; -- { serverError 111 }
+SELECT throwIf(true, 'error_log', toInt16(222)) SETTINGS allow_custom_error_code_in_throwif=1; -- { serverError 222 }
+SELECT throwIf(true, 'error_log', toInt16(333)) SETTINGS allow_custom_error_code_in_throwif=1; -- { serverError 333 }
+
+SELECT sleep(2) FORMAT NULL;
+SYSTEM FLUSH LOGS;
+
+SELECT sum(value) > 1 FROM system.error_log WHERE code = 111 AND event_time > now() - INTERVAL 1 MINUTE;
+SELECT sum(value) > 1 FROM system.error_log WHERE code = 222 AND event_time > now() - INTERVAL 1 MINUTE;
+SELECT sum(value) > 1 FROM system.error_log WHERE code = 333 AND event_time > now() - INTERVAL 1 MINUTE;
--- a/utils/check-style/aspell-ignore/en/aspell-dict.txt
+++ b/utils/check-style/aspell-ignore/en/aspell-dict.txt
@ -2852,7 +2852,9 @@ variantElement
 variantType
 varint
 varpop
+varpopstable
 varsamp
+varsampstable
 vectorized
 vectorscan
 vendoring
--- a/utils/check-style/check-style
+++ b/utils/check-style/check-style
@ -93,6 +93,7 @@ EXTERN_TYPES_EXCLUDES=(
    ErrorCodes::values
    ErrorCodes::values[i]
    ErrorCodes::getErrorCodeByName
+    ErrorCodes::Value
 )
 for extern_type in ${!EXTERN_TYPES[@]}; do
    type_of_extern=${EXTERN_TYPES[$extern_type]}