Merge pull request #52174 from arenadata/ADQM-988

Added possibility to save logs on crash and options to configure logs buffer
2024-11-10 01:25:21 +00:00 · 2023-07-28 16:55:00 +03:00 · 2023-07-28 16:55:00 +03:00 · 39200606ec
commit 39200606ec
parent 0160a95266 112058d889
40 changed files with 794 additions and 178 deletions
--- a/docs/en/operations/server-configuration-parameters/settings.md
+++ b/docs/en/operations/server-configuration-parameters/settings.md
@ -512,7 +512,7 @@ Both the cache for `local_disk`, and temporary data will be stored in `/tiny_loc
                <type>cache</type>
                <disk>local_disk</disk>
                <path>/tiny_local_cache/</path>
-                <max_size>10M</max_size>
+                <max_size_rows>10M</max_size_rows>
                <max_file_segment_size>1M</max_file_segment_size>
                <cache_on_write_operations>1</cache_on_write_operations>
                <do_not_evict_index_and_mark_files>0</do_not_evict_index_and_mark_files>
@ -1592,6 +1592,10 @@ To manually turn on metrics history collection [`system.metric_log`](../../opera
        <table>metric_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>
    </metric_log>
 </clickhouse>
 ```
@ -1695,6 +1699,14 @@ Use the following parameters to configure logging:
 - `order_by` - [Custom sorting key](../../engines/table-engines/mergetree-family/mergetree.md#order_by) for a system table. Can't be used if `engine` defined.
 - `engine` - [MergeTree Engine Definition](../../engines/table-engines/mergetree-family/mergetree.md#table_engine-mergetree-creating-a-table) for a system table. Can't be used if `partition_by` or `order_by` defined.
 - `flush_interval_milliseconds` – Interval for flushing data from the buffer in memory to the table.
+- `max_size_rows` – Maximal size in lines for the logs. When non-flushed logs amount reaches max_size, logs dumped to the disk.
+Default: 1048576.
+- `reserved_size_rows` –  Pre-allocated memory size in lines for the logs.
+Default: 8192.
+- `buffer_size_rows_flush_threshold` – Lines amount threshold, reaching it launches flushing logs to the disk in background.
+Default: `max_size_rows / 2`.
+- `flush_on_crash` - Indication whether logs should be dumped to the disk in case of a crash.
+Default: false.
 - `storage_policy` – Name of storage policy to use for the table (optional)
 - `settings` - [Additional parameters](../../engines/table-engines/mergetree-family/mergetree.md/#settings) that control the behavior of the MergeTree (optional).

@ -1706,6 +1718,10 @@ Use the following parameters to configure logging:
    <table>part_log</table>
    <partition_by>toMonday(event_date)</partition_by>
    <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>
+    <flush_on_crash>false</flush_on_crash>
 </part_log>
 ```

@ -1773,6 +1789,14 @@ Use the following parameters to configure logging:
 - `order_by` - [Custom sorting key](../../engines/table-engines/mergetree-family/mergetree.md#order_by) for a system table. Can't be used if `engine` defined.
 - `engine` - [MergeTree Engine Definition](../../engines/table-engines/mergetree-family/mergetree.md#table_engine-mergetree-creating-a-table) for a system table. Can't be used if `partition_by` or `order_by` defined.
 - `flush_interval_milliseconds` – Interval for flushing data from the buffer in memory to the table.
+- `max_size_rows` – Maximal size in lines for the logs. When non-flushed logs amount reaches max_size, logs dumped to the disk.
+Default: 1048576.
+- `reserved_size_rows` –  Pre-allocated memory size in lines for the logs.
+Default: 8192.
+- `buffer_size_rows_flush_threshold` – Lines amount threshold, reaching it launches flushing logs to the disk in background.
+Default: `max_size_rows / 2`.
+- `flush_on_crash` - Indication whether logs should be dumped to the disk in case of a crash.
+Default: false.
 - `storage_policy` – Name of storage policy to use for the table (optional)
 - `settings` - [Additional parameters](../../engines/table-engines/mergetree-family/mergetree.md/#settings) that control the behavior of the MergeTree (optional).

@ -1786,6 +1810,10 @@ If the table does not exist, ClickHouse will create it. If the structure of the
    <table>query_log</table>
    <engine>Engine = MergeTree PARTITION BY event_date ORDER BY event_time TTL event_date + INTERVAL 30 day</engine>
    <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>
+    <flush_on_crash>false</flush_on_crash>
 </query_log>
 ```

@ -1831,6 +1859,14 @@ Use the following parameters to configure logging:
 - `order_by` - [Custom sorting key](../../engines/table-engines/mergetree-family/mergetree.md#order_by) for a system table. Can't be used if `engine` defined.
 - `engine` - [MergeTree Engine Definition](../../engines/table-engines/mergetree-family/mergetree.md#table_engine-mergetree-creating-a-table) for a system table. Can't be used if `partition_by` or `order_by` defined.
 - `flush_interval_milliseconds` – Interval for flushing data from the buffer in memory to the table.
+- `max_size_rows` – Maximal size in lines for the logs. When non-flushed logs amount reaches max_size_rows, logs dumped to the disk.
+Default: 1048576.
+- `reserved_size_rows` –  Pre-allocated memory size in lines for the logs.
+Default: 8192.
+- `buffer_size_rows_flush_threshold` – Lines amount threshold, reaching it launches flushing logs to the disk in background.
+Default: `max_size_rows / 2`.
+- `flush_on_crash` - Indication whether logs should be dumped to the disk in case of a crash.
+Default: false.
 - `storage_policy` – Name of storage policy to use for the table (optional)
 - `settings` - [Additional parameters](../../engines/table-engines/mergetree-family/mergetree.md/#settings) that control the behavior of the MergeTree (optional).

@ -1844,6 +1880,10 @@ If the table does not exist, ClickHouse will create it. If the structure of the
    <table>query_thread_log</table>
    <partition_by>toMonday(event_date)</partition_by>
    <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>  
+    <flush_on_crash>false</flush_on_crash>
 </query_thread_log>
 ```

@ -1861,6 +1901,14 @@ Use the following parameters to configure logging:
 - `order_by` - [Custom sorting key](../../engines/table-engines/mergetree-family/mergetree.md#order_by) for a system table. Can't be used if `engine` defined.
 - `engine` - [MergeTree Engine Definition](../../engines/table-engines/mergetree-family/mergetree.md#table_engine-mergetree-creating-a-table) for a system table. Can't be used if `partition_by` or `order_by` defined.
 - `flush_interval_milliseconds` – Interval for flushing data from the buffer in memory to the table.
+- `max_size_rows` – Maximal size in lines for the logs. When non-flushed logs amount reaches max_size, logs dumped to the disk.
+Default: 1048576.
+- `reserved_size_rows` –  Pre-allocated memory size in lines for the logs.
+Default: 8192.
+- `buffer_size_rows_flush_threshold` – Lines amount threshold, reaching it launches flushing logs to the disk in background.
+Default: `max_size_rows / 2`.
+- `flush_on_crash` - Indication whether logs should be dumped to the disk in case of a crash.
+Default: false.
 - `storage_policy` – Name of storage policy to use for the table (optional)
 - `settings` - [Additional parameters](../../engines/table-engines/mergetree-family/mergetree.md/#settings) that control the behavior of the MergeTree (optional).

@ -1874,6 +1922,10 @@ If the table does not exist, ClickHouse will create it. If the structure of the
    <table>query_views_log</table>
    <partition_by>toYYYYMM(event_date)</partition_by>
    <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>
+    <flush_on_crash>false</flush_on_crash>
 </query_views_log>
 ```

@ -1890,6 +1942,14 @@ Parameters:
 - `order_by` - [Custom sorting key](../../engines/table-engines/mergetree-family/mergetree.md#order_by) for a system table. Can't be used if `engine` defined.
 - `engine` - [MergeTree Engine Definition](../../engines/table-engines/mergetree-family/mergetree.md#table_engine-mergetree-creating-a-table) for a system table. Can't be used if `partition_by` or `order_by` defined.
 - `flush_interval_milliseconds` — Interval for flushing data from the buffer in memory to the table.
+- `max_size_rows` – Maximal size in lines for the logs. When non-flushed logs amount reaches max_size, logs dumped to the disk.
+Default: 1048576.
+- `reserved_size_rows` –  Pre-allocated memory size in lines for the logs.
+Default: 8192.
+- `buffer_size_rows_flush_threshold` – Lines amount threshold, reaching it launches flushing logs to the disk in background.
+Default: `max_size_rows / 2`.
+- `flush_on_crash` - Indication whether logs should be dumped to the disk in case of a crash.
+Default: false.
 - `storage_policy` – Name of storage policy to use for the table (optional)
 - `settings` - [Additional parameters](../../engines/table-engines/mergetree-family/mergetree.md/#settings) that control the behavior of the MergeTree (optional).

@ -1901,13 +1961,16 @@ Parameters:
        <database>system</database>
        <table>text_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>
+        <flush_on_crash>false</flush_on_crash>
        <!-- <partition_by>event_date</partition_by> -->
        <engine>Engine = MergeTree PARTITION BY event_date ORDER BY event_time TTL event_date + INTERVAL 30 day</engine>
    </text_log>
 </clickhouse>
 ```

-
 ## trace_log {#server_configuration_parameters-trace_log}

 Settings for the [trace_log](../../operations/system-tables/trace_log.md#system_tables-trace_log) system table operation.
@ -1920,6 +1983,12 @@ Parameters:
 - `order_by` - [Custom sorting key](../../engines/table-engines/mergetree-family/mergetree.md#order_by) for a system table. Can't be used if `engine` defined.
 - `engine` - [MergeTree Engine Definition](../../engines/table-engines/mergetree-family/index.md) for a system table. Can't be used if `partition_by` or `order_by` defined.
 - `flush_interval_milliseconds` — Interval for flushing data from the buffer in memory to the table.
+- `max_size_rows` – Maximal size in lines for the logs. When non-flushed logs amount reaches max_size, logs dumped to the disk.
+Default: 1048576.
+- `reserved_size_rows` –  Pre-allocated memory size in lines for the logs.
+Default: 8192.
+- `buffer_size_rows_flush_threshold` – Lines amount threshold, reaching it launches flushing logs to the disk in background.
+Default: `max_size_rows / 2`.
 - `storage_policy` – Name of storage policy to use for the table (optional)
 - `settings` - [Additional parameters](../../engines/table-engines/mergetree-family/mergetree.md/#settings) that control the behavior of the MergeTree (optional).

@ -1931,6 +2000,10 @@ The default server configuration file `config.xml` contains the following settin
    <table>trace_log</table>
    <partition_by>toYYYYMM(event_date)</partition_by>
    <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>
+    <flush_on_crash>false</flush_on_crash>
 </trace_log>
 ```

@ -1945,9 +2018,18 @@ Parameters:
 - `partition_by` — [Custom partitioning key](../../engines/table-engines/mergetree-family/custom-partitioning-key.md) for a system table. Can't be used if `engine` defined.
 - `engine` - [MergeTree Engine Definition](../../engines/table-engines/mergetree-family/mergetree.md#table_engine-mergetree-creating-a-table) for a system table. Can't be used if `partition_by` defined.
 - `flush_interval_milliseconds` — Interval for flushing data from the buffer in memory to the table.
+- `max_size_rows` – Maximal size in lines for the logs. When non-flushed logs amount reaches max_size, logs dumped to the disk.
+Default: 1048576.
+- `reserved_size_rows` –  Pre-allocated memory size in lines for the logs.
+Default: 8192.
+- `buffer_size_rows_flush_threshold` – Lines amount threshold, reaching it launches flushing logs to the disk in background.
+Default: `max_size_rows / 2`.
+- `flush_on_crash` - Indication whether logs should be dumped to the disk in case of a crash.
+Default: false.
 - `storage_policy` – Name of storage policy to use for the table (optional)

 **Example**
+
 ```xml
 <clickhouse>
    <asynchronous_insert_log>
@ -1955,11 +2037,53 @@ Parameters:
        <table>asynchronous_insert_log</table>
        <flush_interval_milliseconds>7500</flush_interval_milliseconds>
        <partition_by>toYYYYMM(event_date)</partition_by>
+        <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>
        <!-- <engine>Engine = MergeTree PARTITION BY event_date ORDER BY event_time TTL event_date + INTERVAL 30 day</engine> -->
    </asynchronous_insert_log>
 </clickhouse>
 ```

+## crash_log {#server_configuration_parameters-crash_log}
+
+Settings for the [crash_log](../../operations/system-tables/crash-log.md) system table operation.
+
+Parameters:
+
+- `database` — Database for storing a table.
+- `table` — Table name.
+- `partition_by` — [Custom partitioning key](../../engines/table-engines/mergetree-family/custom-partitioning-key.md) for a system table. Can't be used if `engine` defined.
+- `order_by` - [Custom sorting key](../../engines/table-engines/mergetree-family/mergetree.md#order_by) for a system table. Can't be used if `engine` defined.
+- `engine` - [MergeTree Engine Definition](../../engines/table-engines/mergetree-family/index.md) for a system table. Can't be used if `partition_by` or `order_by` defined.
+- `flush_interval_milliseconds` — Interval for flushing data from the buffer in memory to the table.
+- `max_size_rows` – Maximal size in lines for the logs. When non-flushed logs amount reaches max_size, logs dumped to the disk.
+Default: 1048576.
+- `reserved_size_rows` –  Pre-allocated memory size in lines for the logs.
+Default: 8192.
+- `buffer_size_rows_flush_threshold` – Lines amount threshold, reaching it launches flushing logs to the disk in background.
+Default: `max_size_rows / 2`.
+- `flush_on_crash` - Indication whether logs should be dumped to the disk in case of a crash.
+Default: false.
+- `storage_policy` – Name of storage policy to use for the table (optional)
+- `settings` - [Additional parameters](../../engines/table-engines/mergetree-family/mergetree.md/#settings) that control the behavior of the MergeTree (optional).
+
+The default server configuration file `config.xml` contains the following settings section:
+
+``` xml
+<crash_log>
+    <database>system</database>
+    <table>crash_log</table>
+    <partition_by>toYYYYMM(event_date)</partition_by>
+    <flush_interval_milliseconds>7500</flush_interval_milliseconds>
+    <max_size_rows>1024</max_size_rows>
+    <reserved_size_rows>1024</reserved_size_rows>
+    <buffer_size_rows_flush_threshold>512</buffer_size_rows_flush_threshold>
+    <flush_on_crash>false</flush_on_crash>
+</crash_log>
+```
+
 ## query_masking_rules {#query-masking-rules}

 Regexp-based rules, which will be applied to queries as well as all log messages before storing them in server logs,
--- a/docs/en/operations/system-tables/index.md
+++ b/docs/en/operations/system-tables/index.md
@ -47,6 +47,10 @@ An example:
        <engine>ENGINE = MergeTree PARTITION BY toYYYYMM(event_date) ORDER BY (event_date, event_time) SETTINGS index_granularity = 1024</engine>
        -->
        <flush_interval_milliseconds>7500</flush_interval_milliseconds>
+        <max_size_rows>1048576</max_size>
+        <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>
    </query_log>
 </clickhouse>
 ```
--- a/docs/ru/operations/server-configuration-parameters/settings.md
+++ b/docs/ru/operations/server-configuration-parameters/settings.md
@ -1058,6 +1058,10 @@ ClickHouse использует потоки из глобального пул
        <table>metric_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>
    </metric_log>
 </clickhouse>
 ```
@ -1155,12 +1159,19 @@ ClickHouse использует потоки из глобального пул

 При настройке логирования используются следующие параметры:

-   `database` — имя базы данных;
-   `table` — имя таблицы;
-   `partition_by` — устанавливает [произвольный ключ партиционирования](../../operations/server-configuration-parameters/settings.md). Нельзя использовать если используется `engine`
-   `engine` - устанавливает [настройки MergeTree Engine](../../engines/table-engines/mergetree-family/mergetree.md#table_engine-mergetree-creating-a-table) для системной таблицы. Нельзя использовать если используется `partition_by`.
-   `flush_interval_milliseconds` — период сброса данных из буфера в памяти в таблицу.
-
+- `database` — имя базы данных;
+- `table` — имя таблицы;
+- `partition_by` — устанавливает [произвольный ключ партиционирования](../../operations/server-configuration-parameters/settings.md). Нельзя использовать если используется `engine`
+- `engine` - устанавливает [настройки MergeTree Engine](../../engines/table-engines/mergetree-family/mergetree.md#table_engine-mergetree-creating-a-table) для системной таблицы. Нельзя использовать если используется `partition_by`.
+- `flush_interval_milliseconds` — период сброса данных из буфера в памяти в таблицу.
+- `max_size_rows` – максимальный размер в строках для буфера с логами. Когда буфер будет заполнен полностью, сбрасывает логи на диск.
+Значение по умолчанию: 1048576.
+- `reserved_size_rows` –  преаллоцированный размер в строках для буфера с логами.
+Значение по умолчанию: 8192.
+- `buffer_size_bytes_flush_threshold` – количество линий в логе при достижении которого логи начнут скидываться на диск в неблокирующем режиме.
+Значение по умолчанию: `max_size / 2`.
+- `flush_on_crash` - должны ли логи быть сброшены на диск в случае неожиданной остановки программы.
+Значение по умолчанию: false.
 **Пример**

 ``` xml
@ -1169,6 +1180,10 @@ ClickHouse использует потоки из глобального пул
    <table>part_log</table>
    <partition_by>toMonday(event_date)</partition_by>
    <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>
+    <flush_on_crash>false</flush_on_crash>
 </part_log>
 ```

@ -1218,11 +1233,19 @@ ClickHouse использует потоки из глобального пул

 При настройке логирования используются следующие параметры:

-   `database` — имя базы данных;
-   `table` — имя таблицы, куда будет записываться лог;
-   `partition_by` — устанавливает [произвольный ключ партиционирования](../../operations/server-configuration-parameters/settings.md). Нельзя использовать если используется `engine`
-   `engine` - устанавливает [настройки MergeTree Engine](../../engines/table-engines/mergetree-family/mergetree.md#table_engine-mergetree-creating-a-table) для системной таблицы. Нельзя использовать если используется `partition_by`.
-   `flush_interval_milliseconds` — период сброса данных из буфера в памяти в таблицу.
+- `database` — имя базы данных;
+- `table` — имя таблицы;
+- `partition_by` — устанавливает [произвольный ключ партиционирования](../../operations/server-configuration-parameters/settings.md). Нельзя использовать если используется `engine`
+- `engine` - устанавливает [настройки MergeTree Engine](../../engines/table-engines/mergetree-family/mergetree.md#table_engine-mergetree-creating-a-table) для системной таблицы. Нельзя использовать если используется `partition_by`.
+- `flush_interval_milliseconds` — период сброса данных из буфера в памяти в таблицу.
+- `max_size_rows` – максимальный размер в строках для буфера с логами. Когда буфер будет заполнен полностью, сбрасывает логи на диск.
+Значение по умолчанию: 1048576.
+- `reserved_size_rows` –  преаллоцированный размер в строках для буфера с логами.
+Значение по умолчанию: 8192.
+- `buffer_size_bytes_flush_threshold` – количество линий в логе при достижении которого логи начнут скидываться на диск в неблокирующем режиме.
+Значение по умолчанию: `max_size / 2`.
+- `flush_on_crash` - должны ли логи быть сброшены на диск в случае неожиданной остановки программы.
+Значение по умолчанию: false.

 Если таблица не существует, то ClickHouse создаст её. Если структура журнала запросов изменилась при обновлении сервера ClickHouse, то таблица со старой структурой переименовывается, а новая таблица создается автоматически.

@ -1234,6 +1257,10 @@ ClickHouse использует потоки из глобального пул
    <table>query_log</table>
    <engine>Engine = MergeTree PARTITION BY event_date ORDER BY event_time TTL event_date + INTERVAL 30 day</engine>
    <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>
+    <flush_on_crash>false</flush_on_crash>
 </query_log>
 ```

@ -1245,11 +1272,19 @@ ClickHouse использует потоки из глобального пул

 При настройке логирования используются следующие параметры:

-   `database` — имя базы данных;
-   `table` — имя таблицы, куда будет записываться лог;
-   `partition_by` — устанавливает [произвольный ключ партиционирования](../../operations/server-configuration-parameters/settings.md). Нельзя использовать если используется `engine`
-   `engine` - устанавливает [настройки MergeTree Engine](../../engines/table-engines/mergetree-family/mergetree.md#table_engine-mergetree-creating-a-table) для системной таблицы. Нельзя использовать если используется `partition_by`.
-   `flush_interval_milliseconds` — период сброса данных из буфера в памяти в таблицу.
+- `database` — имя базы данных;
+- `table` — имя таблицы;
+- `partition_by` — устанавливает [произвольный ключ партиционирования](../../operations/server-configuration-parameters/settings.md). Нельзя использовать если используется `engine`
+- `engine` - устанавливает [настройки MergeTree Engine](../../engines/table-engines/mergetree-family/mergetree.md#table_engine-mergetree-creating-a-table) для системной таблицы. Нельзя использовать если используется `partition_by`.
+- `flush_interval_milliseconds` — период сброса данных из буфера в памяти в таблицу.
+- `max_size_rows` – максимальный размер в строках для буфера с логами. Когда буфер будет заполнен полностью, сбрасывает логи на диск.
+Значение по умолчанию: 1048576.
+- `reserved_size_rows` –  преаллоцированный размер в строках для буфера с логами.
+Значение по умолчанию: 8192.
+- `buffer_size_bytes_flush_threshold` – количество линий в логе при достижении которого логи начнут скидываться на диск в неблокирующем режиме.
+Значение по умолчанию: `max_size / 2`.
+- `flush_on_crash` - должны ли логи быть сброшены на диск в случае неожиданной остановки программы.
+Значение по умолчанию: false.

 Если таблица не существует, то ClickHouse создаст её. Если структура журнала запросов изменилась при обновлении сервера ClickHouse, то таблица со старой структурой переименовывается, а новая таблица создается автоматически.

@ -1261,6 +1296,10 @@ ClickHouse использует потоки из глобального пул
    <table>query_thread_log</table>
    <partition_by>toMonday(event_date)</partition_by>
    <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>
+    <flush_on_crash>false</flush_on_crash>
 </query_thread_log>
 ```

@ -1272,11 +1311,19 @@ ClickHouse использует потоки из глобального пул

 При настройке логирования используются следующие параметры:

-   `database` – имя базы данных.
-   `table` – имя системной таблицы, где будут логироваться запросы.
-   `partition_by` — устанавливает [произвольный ключ партиционирования](../../engines/table-engines/mergetree-family/custom-partitioning-key.md). Нельзя использовать, если задан параметр `engine`.
-   `engine` — устанавливает [настройки MergeTree Engine](../../engines/table-engines/mergetree-family/mergetree.md#table_engine-mergetree-creating-a-table) для системной таблицы. Нельзя использовать, если задан параметр `partition_by`.
-   `flush_interval_milliseconds` — период сброса данных из буфера в памяти в таблицу.
+- `database` — имя базы данных;
+- `table` — имя таблицы;
+- `partition_by` — устанавливает [произвольный ключ партиционирования](../../operations/server-configuration-parameters/settings.md). Нельзя использовать если используется `engine`
+- `engine` - устанавливает [настройки MergeTree Engine](../../engines/table-engines/mergetree-family/mergetree.md#table_engine-mergetree-creating-a-table) для системной таблицы. Нельзя использовать если используется `partition_by`.
+- `flush_interval_milliseconds` — период сброса данных из буфера в памяти в таблицу.
+- `max_size_rows` – максимальный размер в строках для буфера с логами. Когда буфер будет заполнен полностью, сбрасывает логи на диск.
+Значение по умолчанию: 1048576.
+- `reserved_size_rows` –  преаллоцированный размер в строках для буфера с логами.
+Значение по умолчанию: 8192.
+- `buffer_size_bytes_flush_threshold` – количество линий в логе при достижении которого логи начнут скидываться на диск в неблокирующем режиме.
+Значение по умолчанию: `max_size / 2`.
+- `flush_on_crash` - должны ли логи быть сброшены на диск в случае неожиданной остановки программы.
+Значение по умолчанию: false.

 Если таблица не существует, то ClickHouse создаст её. Если структура журнала запросов изменилась при обновлении сервера ClickHouse, то таблица со старой структурой переименовывается, а новая таблица создается автоматически.

@ -1288,6 +1335,10 @@ ClickHouse использует потоки из глобального пул
    <table>query_views_log</table>
    <partition_by>toYYYYMM(event_date)</partition_by>
    <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>
+    <flush_on_crash>false</flush_on_crash>
 </query_views_log>
 ```

@ -1297,12 +1348,20 @@ ClickHouse использует потоки из глобального пул

 Параметры:

-   `level` — Максимальный уровень сообщения (по умолчанию `Trace`) которое будет сохранено в таблице.
-   `database` — имя базы данных для хранения таблицы.
-   `table` — имя таблицы, куда будут записываться текстовые сообщения.
-   `partition_by` — устанавливает [произвольный ключ партиционирования](../../engines/table-engines/mergetree-family/custom-partitioning-key.md). Нельзя использовать если используется `engine`
-   `engine` - устанавливает [настройки MergeTree Engine](../../engines/table-engines/mergetree-family/mergetree.md#table_engine-mergetree-creating-a-table) для системной таблицы. Нельзя использовать если используется `partition_by`.
-   `flush_interval_milliseconds` — период сброса данных из буфера в памяти в таблицу.
+- `level` — Максимальный уровень сообщения (по умолчанию `Trace`) которое будет сохранено в таблице.
+- `database` — имя базы данных;
+- `table` — имя таблицы;
+- `partition_by` — устанавливает [произвольный ключ партиционирования](../../operations/server-configuration-parameters/settings.md). Нельзя использовать если используется `engine`
+- `engine` - устанавливает [настройки MergeTree Engine](../../engines/table-engines/mergetree-family/mergetree.md#table_engine-mergetree-creating-a-table) для системной таблицы. Нельзя использовать если используется `partition_by`.
+- `flush_interval_milliseconds` — период сброса данных из буфера в памяти в таблицу.
+- `max_size_rows` – максимальный размер в строках для буфера с логами. Когда буфер будет заполнен полностью, сбрасывает логи на диск.
+Значение по умолчанию: 1048576.
+- `reserved_size_rows` –  преаллоцированный размер в строках для буфера с логами.
+Значение по умолчанию: 8192.
+- `buffer_size_bytes_flush_threshold` – количество линий в логе при достижении которого логи начнут скидываться на диск в неблокирующем режиме.
+Значение по умолчанию: `max_size / 2`.
+- `flush_on_crash` - должны ли логи быть сброшены на диск в случае неожиданной остановки программы.
+Значение по умолчанию: false.

 **Пример**
 ```xml
@ -1312,6 +1371,10 @@ ClickHouse использует потоки из глобального пул
        <database>system</database>
        <table>text_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>
+        <flush_on_crash>false</flush_on_crash>
        <!-- <partition_by>event_date</partition_by> -->
        <engine>Engine = MergeTree PARTITION BY event_date ORDER BY event_time TTL event_date + INTERVAL 30 day</engine>
    </text_log>
@ -1323,13 +1386,21 @@ ClickHouse использует потоки из глобального пул

 Настройки для [trace_log](../../operations/system-tables/trace_log.md#system_tables-trace_log) system table operation.

-Parameters:
+Параметры:

-   `database` — Database for storing a table.
-   `table` — Table name.
-   `partition_by` — устанавливает [произвольный ключ партиционирования](../../operations/server-configuration-parameters/settings.md). Нельзя использовать если используется `engine`
-   `engine` - устанавливает [настройки MergeTree Engine](../../engines/table-engines/mergetree-family/mergetree.md#table_engine-mergetree-creating-a-table) для системной таблицы. Нельзя использовать если используется `partition_by`.
-   `flush_interval_milliseconds` — Interval for flushing data from the buffer in memory to the table.
+- `database` — имя базы данных;
+- `table` — имя таблицы;
+- `partition_by` — устанавливает [произвольный ключ партиционирования](../../operations/server-configuration-parameters/settings.md). Нельзя использовать если используется `engine`
+- `engine` - устанавливает [настройки MergeTree Engine](../../engines/table-engines/mergetree-family/mergetree.md#table_engine-mergetree-creating-a-table) для системной таблицы. Нельзя использовать если используется `partition_by`.
+- `flush_interval_milliseconds` — период сброса данных из буфера в памяти в таблицу.
+- `max_size_rows` – максимальный размер в строках для буфера с логами. Когда буфер будет заполнен полностью, сбрасывает логи на диск.
+Значение по умолчанию: 1048576.
+- `reserved_size_rows` –  преаллоцированный размер в строках для буфера с логами.
+Значение по умолчанию: 8192.
+- `buffer_size_bytes_flush_threshold` – количество линий в логе при достижении которого логи начнут скидываться на диск в неблокирующем режиме.
+Значение по умолчанию: `max_size / 2`.
+- `flush_on_crash` - должны ли логи быть сброшены на диск в случае неожиданной остановки программы.
+Значение по умолчанию: false.

 По умолчанию файл настроек сервера `config.xml` содержит следующие настройки:

@ -1339,9 +1410,84 @@ Parameters:
    <table>trace_log</table>
    <partition_by>toYYYYMM(event_date)</partition_by>
    <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>
 </trace_log>
 ```

+## asynchronous_insert_log {#server_configuration_parameters-asynchronous_insert_log}
+
+Настройки для asynchronous_insert_log Система для логирования ассинхронных вставок.
+
+Параметры:
+
+- `database` — имя базы данных;
+- `table` — имя таблицы;
+- `partition_by` — устанавливает [произвольный ключ партиционирования](../../operations/server-configuration-parameters/settings.md). Нельзя использовать если используется `engine`
+- `engine` - устанавливает [настройки MergeTree Engine](../../engines/table-engines/mergetree-family/mergetree.md#table_engine-mergetree-creating-a-table) для системной таблицы. Нельзя использовать если используется `partition_by`.
+- `flush_interval_milliseconds` — период сброса данных из буфера в памяти в таблицу.
+- `max_size_rows` – максимальный размер в строках для буфера с логами. Когда буфер будет заполнен полностью, сбрасывает логи на диск.
+Значение по умолчанию: 1048576.
+- `reserved_size_rows` –  преаллоцированный размер в строках для буфера с логами.
+Значение по умолчанию: 8192.
+- `buffer_size_bytes_flush_threshold` – количество линий в логе при достижении которого логи начнут скидываться на диск в неблокирующем режиме.
+Значение по умолчанию: `max_size / 2`.
+- `flush_on_crash` - должны ли логи быть сброшены на диск в случае неожиданной остановки программы.
+Значение по умолчанию: false.
+
+**Пример**
+
+```xml
+<clickhouse>
+    <asynchronous_insert_log>
+        <database>system</database>
+        <table>asynchronous_insert_log</table>
+        <flush_interval_milliseconds>7500</flush_interval_milliseconds>
+        <partition_by>toYYYYMM(event_date)</partition_by>
+        <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>
+        <!-- <engine>Engine = MergeTree PARTITION BY event_date ORDER BY event_time TTL event_date + INTERVAL 30 day</engine> -->
+    </asynchronous_insert_log>
+</clickhouse>
+```
+
+## crash_log {#server_configuration_parameters-crash_log}
+
+Настройки для таблицы [crash_log](../../operations/system-tables/crash-log.md).
+
+Параметры:
+
+- `database` — имя базы данных;
+- `table` — имя таблицы;
+- `partition_by` — устанавливает [произвольный ключ партиционирования](../../operations/server-configuration-parameters/settings.md). Нельзя использовать если используется `engine`
+- `engine` - устанавливает [настройки MergeTree Engine](../../engines/table-engines/mergetree-family/mergetree.md#table_engine-mergetree-creating-a-table) для системной таблицы. Нельзя использовать если используется `partition_by`.
+- `flush_interval_milliseconds` — период сброса данных из буфера в памяти в таблицу.
+- `max_size_rows` – максимальный размер в строках для буфера с логами. Когда буфер будет заполнен полностью, сбрасывает логи на диск.
+Значение по умолчанию: 1024.
+- `reserved_size_rows` –  преаллоцированный размер в строках для буфера с логами.
+Значение по умолчанию: 1024.
+- `buffer_size_bytes_flush_threshold` – количество линий в логе при достижении которого логи начнут скидываться на диск в неблокирующем режиме.
+Значение по умолчанию: `max_size / 2`.
+- `flush_on_crash` - должны ли логи быть сброшены на диск в случае неожиданной остановки программы.
+Значение по умолчанию: true.
+
+**Пример**
+
+``` xml
+<crash_log>
+    <database>system</database>
+    <table>crash_log</table>
+    <partition_by>toYYYYMM(event_date)</partition_by>
+    <flush_interval_milliseconds>7500</flush_interval_milliseconds>
+    <max_size_rows>1024</max_size_rows>
+    <reserved_size_rows>1024</reserved_size_rows>
+    <buffer_size_rows_flush_threshold>512</buffer_size_rows_flush_threshold>
+    <flush_on_crash>true</flush_on_crash>
+</crash_log>
+```
+
 ## query_masking_rules {#query-masking-rules}

 Правила, основанные на регулярных выражениях, которые будут применены для всех запросов, а также для всех сообщений перед сохранением их в лог на сервере,
--- a/docs/ru/operations/system-tables/index.md
+++ b/docs/ru/operations/system-tables/index.md
@ -45,6 +45,10 @@ sidebar_label: "Системные таблицы"
        <engine>ENGINE = MergeTree PARTITION BY toYYYYMM(event_date) ORDER BY (event_date, event_time) SETTINGS index_granularity = 1024</engine>
        -->
        <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>
+        <flush_on_crash>false</flush_on_crash>
    </query_log>
 </clickhouse>
 ```
--- a/programs/server/config.xml
+++ b/programs/server/config.xml
@ -1026,6 +1026,14 @@

        <!-- Interval of flushing data. -->
        <flush_interval_milliseconds>7500</flush_interval_milliseconds>
+        <!-- Maximal size in lines for the logs. When non-flushed logs amount reaches max_size, logs dumped to the disk. -->
+        <max_size_rows>1048576</max_size_rows>
+        <!-- Pre-allocated size in lines for the logs. -->
+        <reserved_size_rows>8192</reserved_size_rows>
+        <!-- Lines amount threshold, reaching it launches flushing logs to the disk in background. -->
+        <buffer_size_rows_flush_threshold>524288</buffer_size_rows_flush_threshold>
+        <!-- Indication whether logs should be dumped to the disk in case of a crash -->
+        <flush_on_crash>false</flush_on_crash>

        <!-- example of using a different storage policy for a system table -->
        <!-- storage_policy>local_ssd</storage_policy -->
@ -1039,6 +1047,11 @@

        <partition_by>toYYYYMM(event_date)</partition_by>
        <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>
+        <!-- Indication whether logs should be dumped to the disk in case of a crash -->
+        <flush_on_crash>false</flush_on_crash>
    </trace_log>

    <!-- Query thread log. Has information about all threads participated in query execution.
@ -1048,6 +1061,10 @@
        <table>query_thread_log</table>
        <partition_by>toYYYYMM(event_date)</partition_by>
        <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>
+        <flush_on_crash>false</flush_on_crash>
    </query_thread_log>

    <!-- Query views log. Has information about all dependent views associated with a query.
@ -1066,6 +1083,10 @@
        <table>part_log</table>
        <partition_by>toYYYYMM(event_date)</partition_by>
        <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>
+        <flush_on_crash>false</flush_on_crash>
    </part_log>

    <!-- Uncomment to write text log into table.
@ -1075,6 +1096,10 @@
        <database>system</database>
        <table>text_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>
+        <flush_on_crash>false</flush_on_crash>
        <level></level>
    </text_log>
    -->
@ -1084,7 +1109,11 @@
        <database>system</database>
        <table>metric_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>
    </metric_log>

    <!--
@ -1095,6 +1124,10 @@
        <database>system</database>
        <table>asynchronous_metric_log</table>
        <flush_interval_milliseconds>7000</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>
+        <flush_on_crash>false</flush_on_crash>
    </asynchronous_metric_log>

    <!--
@ -1119,6 +1152,10 @@
        <database>system</database>
        <table>opentelemetry_span_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>
+        <flush_on_crash>false</flush_on_crash>
    </opentelemetry_span_log>


@ -1130,6 +1167,10 @@

        <partition_by />
        <flush_interval_milliseconds>1000</flush_interval_milliseconds>
+        <max_size_rows>1024</max_size_rows>
+        <reserved_size_rows>1024</reserved_size_rows>
+        <buffer_size_rows_flush_threshold>512</buffer_size_rows_flush_threshold>
+        <flush_on_crash>true</flush_on_crash>
    </crash_log>

    <!-- Session log. Stores user log in (successful or not) and log out events.
@ -1142,6 +1183,10 @@

        <partition_by>toYYYYMM(event_date)</partition_by>
        <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>
+        <flush_on_crash>false</flush_on_crash>
    </session_log> -->

    <!-- Profiling on Processors level. -->
@ -1151,6 +1196,10 @@

        <partition_by>toYYYYMM(event_date)</partition_by>
        <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>
+        <flush_on_crash>false</flush_on_crash>
    </processors_profile_log>

    <!-- Log of asynchronous inserts. It allows to check status
@ -1161,6 +1210,10 @@
        <table>asynchronous_insert_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>
+        <flush_on_crash>false</flush_on_crash>
        <partition_by>event_date</partition_by>
        <ttl>event_date + INTERVAL 3 DAY</ttl>
    </asynchronous_insert_log>
--- a/src/Common/SystemLogBase.cpp
+++ b/src/Common/SystemLogBase.cpp
@ -31,30 +31,25 @@ namespace ErrorCodes
    extern const int TIMEOUT_EXCEEDED;
 }

-namespace
-{
-    constexpr size_t DBMS_SYSTEM_LOG_QUEUE_SIZE = 1048576;
-}
-
 ISystemLog::~ISystemLog() = default;


 template <typename LogElement>
-SystemLogQueue<LogElement>::SystemLogQueue(
-    const String & table_name_,
-    size_t flush_interval_milliseconds_,
-    bool turn_off_logger_)
-    : log(&Poco::Logger::get("SystemLogQueue (" + table_name_ + ")"))
-    , flush_interval_milliseconds(flush_interval_milliseconds_)
+SystemLogQueue<LogElement>::SystemLogQueue(const SystemLogQueueSettings & settings_)
+    : log(&Poco::Logger::get("SystemLogQueue (" + settings_.database + "." +settings_.table + ")"))
+    , settings(settings_)
+
 {
-    if (turn_off_logger_)
+    queue.reserve(settings.reserved_size_rows);
+
+    if (settings.turn_off_logger)
        log->setLevel(0);
 }

 static thread_local bool recursive_push_call = false;

 template <typename LogElement>
-void SystemLogQueue<LogElement>::push(const LogElement & element)
+void SystemLogQueue<LogElement>::push(LogElement&& element)
 {
    /// It is possible that the method will be called recursively.
    /// Better to drop these events to avoid complications.
@ -70,7 +65,7 @@ void SystemLogQueue<LogElement>::push(const LogElement & element)
    MemoryTrackerBlockerInThread temporarily_disable_memory_tracker;

    /// Should not log messages under mutex.
-    bool queue_is_half_full = false;
+    bool buffer_size_rows_flush_threshold_exceeded = false;

    {
        std::unique_lock lock(mutex);
@ -78,9 +73,9 @@ void SystemLogQueue<LogElement>::push(const LogElement & element)
        if (is_shutdown)
            return;

-        if (queue.size() == DBMS_SYSTEM_LOG_QUEUE_SIZE / 2)
+        if (queue.size() == settings.buffer_size_rows_flush_threshold)
        {
-            queue_is_half_full = true;
+            buffer_size_rows_flush_threshold_exceeded = true;

            // The queue more than half full, time to flush.
            // We only check for strict equality, because messages are added one
@ -94,7 +89,7 @@ void SystemLogQueue<LogElement>::push(const LogElement & element)
            flush_event.notify_all();
        }

-        if (queue.size() >= DBMS_SYSTEM_LOG_QUEUE_SIZE)
+        if (queue.size() >= settings.max_size_rows)
        {
            // Ignore all further entries until the queue is flushed.
            // Log a message about that. Don't spam it -- this might be especially
@ -108,27 +103,28 @@ void SystemLogQueue<LogElement>::push(const LogElement & element)
                // TextLog sets its logger level to 0, so this log is a noop and
                // there is no recursive logging.
                lock.unlock();
-                LOG_ERROR(log, "Queue is full for system log '{}' at {}", demangle(typeid(*this).name()), queue_front_index);
+                LOG_ERROR(log, "Queue is full for system log '{}' at {}. max_size_rows {}",
+                          demangle(typeid(*this).name()),
+                          queue_front_index,
+                          settings.max_size_rows);
            }

            return;
        }

-        queue.push_back(element);
+        queue.push_back(std::move(element));
    }

-    if (queue_is_half_full)
-        LOG_INFO(log, "Queue is half full for system log '{}'.", demangle(typeid(*this).name()));
+    if (buffer_size_rows_flush_threshold_exceeded)
+        LOG_INFO(log, "Queue is half full for system log '{}'. buffer_size_rows_flush_threshold {}",
+                 demangle(typeid(*this).name()), settings.buffer_size_rows_flush_threshold);
 }

 template <typename LogElement>
-void SystemLogBase<LogElement>::flush(bool force)
+void SystemLogQueue<LogElement>::handleCrash()
 {
-    uint64_t this_thread_requested_offset = queue->notifyFlush(force);
-    if (this_thread_requested_offset == uint64_t(-1))
-        return;
-
-    queue->waitFlush(this_thread_requested_offset);
+    if (settings.notify_flush_on_crash)
+        notifyFlush(/* force */ true);
 }

 template <typename LogElement>
@ -185,11 +181,13 @@ void SystemLogQueue<LogElement>::confirm(uint64_t to_flush_end)
 }

 template <typename LogElement>
-typename SystemLogQueue<LogElement>::Index SystemLogQueue<LogElement>::pop(std::vector<LogElement>& output, bool& should_prepare_tables_anyway, bool& exit_this_thread)
+typename SystemLogQueue<LogElement>::Index SystemLogQueue<LogElement>::pop(std::vector<LogElement> & output,
+                                                                           bool & should_prepare_tables_anyway,
+                                                                           bool & exit_this_thread)
 {
    std::unique_lock lock(mutex);
    flush_event.wait_for(lock,
-        std::chrono::milliseconds(flush_interval_milliseconds),
+        std::chrono::milliseconds(settings.flush_interval_milliseconds),
        [&] ()
        {
            return requested_flush_up_to > flushed_up_to || is_shutdown || is_force_prepare_tables;
@ -219,13 +217,28 @@ void SystemLogQueue<LogElement>::shutdown()

 template <typename LogElement>
 SystemLogBase<LogElement>::SystemLogBase(
-    const String& table_name_,
-    size_t flush_interval_milliseconds_,
+    const SystemLogQueueSettings & settings_,
    std::shared_ptr<SystemLogQueue<LogElement>> queue_)
-    : queue(queue_ ? queue_ : std::make_shared<SystemLogQueue<LogElement>>(table_name_, flush_interval_milliseconds_))
+    : queue(queue_ ? queue_ : std::make_shared<SystemLogQueue<LogElement>>(settings_))
 {
 }

+template <typename LogElement>
+void SystemLogBase<LogElement>::flush(bool force)
+{
+    uint64_t this_thread_requested_offset = queue->notifyFlush(force);
+    if (this_thread_requested_offset == uint64_t(-1))
+        return;
+
+    queue->waitFlush(this_thread_requested_offset);
+}
+
+template <typename LogElement>
+void SystemLogBase<LogElement>::handleCrash()
+{
+    queue->handleCrash();
+}
+
 template <typename LogElement>
 void SystemLogBase<LogElement>::startup()
 {
@ -234,9 +247,9 @@ void SystemLogBase<LogElement>::startup()
 }

 template <typename LogElement>
-void SystemLogBase<LogElement>::add(const LogElement & element)
+void SystemLogBase<LogElement>::add(LogElement element)
 {
-    queue->push(element);
+    queue->push(std::move(element));
 }

 template <typename LogElement>
--- a/src/Common/SystemLogBase.h
+++ b/src/Common/SystemLogBase.h
@ -62,6 +62,9 @@ public:

    virtual void stopFlushThread() = 0;

+    /// Handles crash, flushes log without blocking if notify_flush_on_crash is set
+    virtual void handleCrash() = 0;
+
    virtual ~ISystemLog();

    virtual void savingThreadFunction() = 0;
@ -73,26 +76,38 @@ protected:
    bool is_shutdown = false;
 };

+struct SystemLogQueueSettings
+{
+    String database;
+    String table;
+    size_t reserved_size_rows;
+    size_t max_size_rows;
+    size_t buffer_size_rows_flush_threshold;
+    size_t flush_interval_milliseconds;
+    bool notify_flush_on_crash;
+    bool turn_off_logger;
+};
+
 template <typename LogElement>
 class SystemLogQueue
 {
    using Index = uint64_t;

 public:
-    SystemLogQueue(
-        const String & table_name_,
-        size_t flush_interval_milliseconds_,
-        bool turn_off_logger_ = false);
+    SystemLogQueue(const SystemLogQueueSettings & settings_);

    void shutdown();

    // producer methods
-    void push(const LogElement & element);
+    void push(LogElement && element);
    Index notifyFlush(bool should_prepare_tables_anyway);
    void waitFlush(Index expected_flushed_up_to);

+    /// Handles crash, flushes log without blocking if notify_flush_on_crash is set
+    void handleCrash();
+
     // consumer methods
-    Index pop(std::vector<LogElement>& output, bool& should_prepare_tables_anyway, bool& exit_this_thread);
+    Index pop(std::vector<LogElement>& output, bool & should_prepare_tables_anyway, bool & exit_this_thread);
    void confirm(Index to_flush_end);

 private:
@ -120,7 +135,8 @@ private:
    bool is_shutdown = false;

    std::condition_variable flush_event;
-    const size_t flush_interval_milliseconds;
+
+    const SystemLogQueueSettings settings;
 };


@ -131,8 +147,7 @@ public:
    using Self = SystemLogBase;

    SystemLogBase(
-        const String& table_name_,
-        size_t flush_interval_milliseconds_,
+        const SystemLogQueueSettings & settings_,
        std::shared_ptr<SystemLogQueue<LogElement>> queue_ = nullptr);

    void startup() override;
@ -140,17 +155,25 @@ public:
    /** Append a record into log.
      * Writing to table will be done asynchronously and in case of failure, record could be lost.
      */
-    void add(const LogElement & element);
+    void add(LogElement element);

    /// Flush data in the buffer to disk. Block the thread until the data is stored on disk.
    void flush(bool force) override;

+    /// Handles crash, flushes log without blocking if notify_flush_on_crash is set
+    void handleCrash() override;
+
    /// Non-blocking flush data in the buffer to disk.
    void notifyFlush(bool force);

    String getName() const override { return LogElement::name(); }

    static const char * getDefaultOrderBy() { return "event_date, event_time"; }
+    static consteval size_t getDefaultMaxSize() { return 1048576; }
+    static consteval size_t getDefaultReservedSize() { return 8192; }
+    static consteval size_t getDefaultFlushIntervalMilliseconds() { return 7500; }
+    static consteval bool shouldNotifyFlushOnCrash() { return false; }
+    static consteval bool shouldTurnOffLogger() { return false; }

 protected:
    std::shared_ptr<SystemLogQueue<LogElement>> queue;
--- a/src/Common/ZooKeeper/ZooKeeperImpl.cpp
+++ b/src/Common/ZooKeeper/ZooKeeperImpl.cpp
@ -1440,7 +1440,7 @@ void ZooKeeper::logOperationIfNeeded(const ZooKeeperRequestPtr & request, const
            elem.thread_id = request->thread_id;
            elem.query_id = request->query_id;
        }
-        maybe_zk_log->add(elem);
+        maybe_zk_log->add(std::move(elem));
    }
 }
 #else
--- a/src/Daemon/BaseDaemon.cpp
+++ b/src/Daemon/BaseDaemon.cpp
@ -466,6 +466,10 @@ private:
        if (collectCrashLog)
            collectCrashLog(sig, thread_num, query_id, stack_trace);

+#ifndef CLICKHOUSE_PROGRAM_STANDALONE_BUILD
+        Context::getGlobalContextInstance()->handleCrash();
+#endif
+
        /// Send crash report to developers (if configured)
        if (sig != SanitizerTrap)
        {
--- a/src/Disks/IO/AsynchronousBoundedReadBuffer.cpp
+++ b/src/Disks/IO/AsynchronousBoundedReadBuffer.cpp
@ -147,7 +147,7 @@ void AsynchronousBoundedReadBuffer::appendToPrefetchLog(
    };

    if (prefetches_log)
-        prefetches_log->add(elem);
+        prefetches_log->add(std::move(elem));
 }


--- a/src/Disks/IO/CachedOnDiskReadBufferFromFile.cpp
+++ b/src/Disks/IO/CachedOnDiskReadBufferFromFile.cpp
@ -108,7 +108,7 @@ void CachedOnDiskReadBufferFromFile::appendFilesystemCacheLog(
            break;
    }

-    cache_log->add(elem);
+    cache_log->add(std::move(elem));
 }

 void CachedOnDiskReadBufferFromFile::initialize(size_t offset, size_t size)
--- a/src/Disks/IO/CachedOnDiskWriteBufferFromFile.cpp
+++ b/src/Disks/IO/CachedOnDiskWriteBufferFromFile.cpp
@ -171,7 +171,7 @@ void FileSegmentRangeWriter::appendFilesystemCacheLog(const FileSegment & file_s
        .profile_counters = nullptr,
    };

-    cache_log->add(elem);
+    cache_log->add(std::move(elem));
 }

 void FileSegmentRangeWriter::completeFileSegment()
--- a/src/Disks/IO/ReadBufferFromRemoteFSGather.cpp
+++ b/src/Disks/IO/ReadBufferFromRemoteFSGather.cpp
@ -112,7 +112,7 @@ void ReadBufferFromRemoteFSGather::appendUncachedReadInfo()
        .file_segment_size = current_object.bytes_size,
        .read_from_cache_attempted = false,
    };
-    cache_log->add(elem);
+    cache_log->add(std::move(elem));
 }

 IAsynchronousReader::Result ReadBufferFromRemoteFSGather::readInto(char * data, size_t size, size_t offset, size_t ignore)
--- a/src/Interpreters/AsynchronousInsertQueue.cpp
+++ b/src/Interpreters/AsynchronousInsertQueue.cpp
@ -438,7 +438,7 @@ try
        elem.flush_query_id = flush_query_id;
        elem.exception = flush_exception;
        elem.status = flush_exception.empty() ? Status::Ok : Status::FlushError;
-        log.add(elem);
+        log.add(std::move(elem));
    }
 }
 catch (...)
@ -608,7 +608,7 @@ try
            if (!elem.exception.empty())
            {
                elem.status = AsynchronousInsertLogElement::ParsingError;
-                insert_log->add(elem);
+                insert_log->add(std::move(elem));
            }
            else
            {
--- a/src/Interpreters/Context.cpp
+++ b/src/Interpreters/Context.cpp
@ -3210,6 +3210,12 @@ void Context::initializeMergeTreeMetadataCache(const String & dir, size_t size)
 }
 #endif

+/// Call after unexpected crash happen.
+void Context::handleCrash() const
+{
+    shared->system_logs->handleCrash();
+}
+
 bool Context::hasTraceCollector() const
 {
    return shared->hasTraceCollector();
--- a/src/Interpreters/Context.h
+++ b/src/Interpreters/Context.h
@ -998,6 +998,9 @@ public:
    void initializeMergeTreeMetadataCache(const String & dir, size_t size);
 #endif

+    /// Call after unexpected crash happen.
+    void handleCrash() const;
+
    bool hasTraceCollector() const;

    /// Nullptr if the query log is not ready for this moment.
--- a/src/Interpreters/CrashLog.cpp
+++ b/src/Interpreters/CrashLog.cpp
@ -83,9 +83,6 @@ void collectCrashLog(Int32 signal, UInt64 thread_id, const String & query_id, co
        stack_trace.toStringEveryLine([&trace_full](std::string_view line) { trace_full.push_back(line); });

        CrashLogElement element{static_cast<time_t>(time / 1000000000), time, signal, thread_id, query_id, trace, trace_full};
-        crash_log_owned->add(element);
-        /// Notify savingThreadFunction to start flushing crash log
-        /// Crash log is storing in parallel with the signal processing thread.
-        crash_log_owned->notifyFlush(true);
+        crash_log_owned->add(std::move(element));
    }
 }
--- a/src/Interpreters/CrashLog.h
+++ b/src/Interpreters/CrashLog.h
@ -45,6 +45,11 @@ public:
    {
        crash_log = crash_log_;
    }
+
+    static consteval size_t getDefaultMaxSize() { return 1024; }
+    static consteval size_t getDefaultReservedSize() { return 1024; }
+    static consteval size_t getDefaultFlushIntervalMilliseconds() { return 1000; }
+    static consteval size_t shouldNotifyFlushOnCrash() { return true; }
 };

 }
--- a/src/Interpreters/MetricLog.cpp
+++ b/src/Interpreters/MetricLog.cpp
@ -113,7 +113,7 @@ void MetricLog::metricThreadFunction()
                elem.current_metrics[i] = CurrentMetrics::values[i];
            }

-            this->add(elem);
+            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.
--- a/src/Interpreters/PartLog.cpp
+++ b/src/Interpreters/PartLog.cpp
@ -242,7 +242,7 @@ bool PartLog::addNewParts(

            elem.profile_counters = part_log_entry.profile_counters;

-            part_log->add(elem);
+            part_log->add(std::move(elem));
        }
    }
    catch (...)
--- a/src/Interpreters/ProcessorsProfileLog.cpp
+++ b/src/Interpreters/ProcessorsProfileLog.cpp
@ -73,12 +73,5 @@ void ProcessorProfileLogElement::appendToBlock(MutableColumns & columns) const
    columns[i++]->insert(output_bytes);
 }

-ProcessorsProfileLog::ProcessorsProfileLog(ContextPtr context_, const String & database_name_,
-        const String & table_name_, const String & storage_def_,
-        size_t flush_interval_milliseconds_)
-  : SystemLog<ProcessorProfileLogElement>(context_, database_name_, table_name_,
-        storage_def_, flush_interval_milliseconds_)
-{
-}

 }
--- a/src/Interpreters/ProcessorsProfileLog.h
+++ b/src/Interpreters/ProcessorsProfileLog.h
@ -45,12 +45,7 @@ struct ProcessorProfileLogElement
 class ProcessorsProfileLog : public SystemLog<ProcessorProfileLogElement>
 {
 public:
-    ProcessorsProfileLog(
-        ContextPtr context_,
-        const String & database_name_,
-        const String & table_name_,
-        const String & storage_def_,
-        size_t flush_interval_milliseconds_);
+    using SystemLog<ProcessorProfileLogElement>::SystemLog;
 };

 }
--- a/src/Interpreters/Session.cpp
+++ b/src/Interpreters/Session.cpp
@ -240,7 +240,7 @@ private:

            if (session != sessions.end() && session->second->close_cycle <= current_cycle)
            {
-                if (!session->second.unique())
+                if (session->second.use_count() != 1)
                {
                    LOG_TEST(log, "Delay closing session with session_id: {}, user_id: {}", key.second, key.first);

--- a/src/Interpreters/SessionLog.cpp
+++ b/src/Interpreters/SessionLog.cpp
@ -227,7 +227,7 @@ void SessionLog::addLoginSuccess(const UUID & auth_id, std::optional<String> ses
    for (const auto & s : settings.allChanged())
        log_entry.settings.emplace_back(s.getName(), s.getValueString());

-    add(log_entry);
+    add(std::move(log_entry));
 }

 void SessionLog::addLoginFailure(
@ -243,7 +243,7 @@ void SessionLog::addLoginFailure(
    log_entry.client_info = info;
    log_entry.user_identified_with = AuthenticationType::NO_PASSWORD;

-    add(log_entry);
+    add(std::move(log_entry));
 }

 void SessionLog::addLogOut(const UUID & auth_id, const UserPtr & login_user, const ClientInfo & client_info)
@ -257,7 +257,7 @@ void SessionLog::addLogOut(const UUID & auth_id, const UserPtr & login_user, con
    log_entry.external_auth_server = login_user ? login_user->auth_data.getLDAPServerName() : "";
    log_entry.client_info = client_info;

-    add(log_entry);
+    add(std::move(log_entry));
 }

 }
--- a/src/Interpreters/SystemLog.cpp
+++ b/src/Interpreters/SystemLog.cpp
@ -101,7 +101,6 @@ namespace
 namespace
 {

-constexpr size_t DEFAULT_SYSTEM_LOG_FLUSH_INTERVAL_MILLISECONDS = 7500;
 constexpr size_t DEFAULT_METRIC_LOG_COLLECT_INTERVAL_MILLISECONDS = 1000;

 /// Creates a system log with MergeTree engine using parameters from config
@ -124,18 +123,23 @@ std::shared_ptr<TSystemLog> createSystemLog(
    LOG_DEBUG(&Poco::Logger::get("SystemLog"),
              "Creating {}.{} from {}", default_database_name, default_table_name, config_prefix);

-    String database = config.getString(config_prefix + ".database", default_database_name);
-    String table = config.getString(config_prefix + ".table", default_table_name);
+    SystemLogSettings log_settings;
+    log_settings.queue_settings.database = config.getString(config_prefix + ".database", default_database_name);
+    log_settings.queue_settings.table = config.getString(config_prefix + ".table", default_table_name);

-    if (database != default_database_name)
+    if (log_settings.queue_settings.database != default_database_name)
    {
        /// System tables must be loaded before other tables, but loading order is undefined for all databases except `system`
-        LOG_ERROR(&Poco::Logger::get("SystemLog"), "Custom database name for a system table specified in config."
-            " Table `{}` will be created in `system` database instead of `{}`", table, database);
-        database = default_database_name;
+        LOG_ERROR(
+            &Poco::Logger::get("SystemLog"),
+            "Custom database name for a system table specified in config."
+            " Table `{}` will be created in `system` database instead of `{}`",
+            log_settings.queue_settings.table,
+            log_settings.queue_settings.database);
+
+        log_settings.queue_settings.database = default_database_name;
    }

-    String engine;
    if (config.has(config_prefix + ".engine"))
    {
        if (config.has(config_prefix + ".partition_by"))
@ -159,26 +163,26 @@ std::shared_ptr<TSystemLog> createSystemLog(
                            "If 'engine' is specified for system table, SETTINGS parameters should "
                            "be specified directly inside 'engine' and 'settings' setting doesn't make sense");

-        engine = config.getString(config_prefix + ".engine");
+        log_settings.engine = config.getString(config_prefix + ".engine");
    }
    else
    {
        /// ENGINE expr is necessary.
-        engine = "ENGINE = MergeTree";
+        log_settings.engine = "ENGINE = MergeTree";

        /// PARTITION expr is not necessary.
        String partition_by = config.getString(config_prefix + ".partition_by", "toYYYYMM(event_date)");
        if (!partition_by.empty())
-            engine += " PARTITION BY (" + partition_by + ")";
+            log_settings.engine += " PARTITION BY (" + partition_by + ")";

        /// TTL expr is not necessary.
        String ttl = config.getString(config_prefix + ".ttl", "");
        if (!ttl.empty())
-            engine += " TTL " + ttl;
+            log_settings.engine += " TTL " + ttl;

        /// ORDER BY expr is necessary.
        String order_by = config.getString(config_prefix + ".order_by", TSystemLog::getDefaultOrderBy());
-        engine += " ORDER BY (" + order_by + ")";
+        log_settings.engine += " ORDER BY (" + order_by + ")";

        /// SETTINGS expr is not necessary.
        ///   https://clickhouse.com/docs/en/engines/table-engines/mergetree-family/mergetree#settings
@ -188,24 +192,52 @@ std::shared_ptr<TSystemLog> createSystemLog(
        String settings = config.getString(config_prefix + ".settings", "");
        if (!storage_policy.empty() || !settings.empty())
        {
-            engine += " SETTINGS";
+            log_settings.engine += " SETTINGS";
            /// If 'storage_policy' is repeated, the 'settings' configuration is preferred.
            if (!storage_policy.empty())
-                engine += " storage_policy = " + quoteString(storage_policy);
+                log_settings.engine += " storage_policy = " + quoteString(storage_policy);
            if (!settings.empty())
-                engine += (storage_policy.empty() ? " " : ", ") + settings;
+                log_settings.engine += (storage_policy.empty() ? " " : ", ") + settings;
        }
    }

    /// Validate engine definition syntax to prevent some configuration errors.
    ParserStorageWithComment storage_parser;
-    parseQuery(storage_parser, engine.data(), engine.data() + engine.size(),
+    parseQuery(storage_parser, log_settings.engine.data(), log_settings.engine.data() + log_settings.engine.size(),
            "Storage to create table for " + config_prefix, 0, DBMS_DEFAULT_MAX_PARSER_DEPTH);

-    size_t flush_interval_milliseconds = config.getUInt64(config_prefix + ".flush_interval_milliseconds",
-                                                          DEFAULT_SYSTEM_LOG_FLUSH_INTERVAL_MILLISECONDS);
+    log_settings.queue_settings.flush_interval_milliseconds = config.getUInt64(config_prefix + ".flush_interval_milliseconds",
+                                                                               TSystemLog::getDefaultFlushIntervalMilliseconds());

-    return std::make_shared<TSystemLog>(context, database, table, engine, flush_interval_milliseconds);
+    log_settings.queue_settings.max_size_rows = config.getUInt64(config_prefix + ".max_size_rows",
+                                                                 TSystemLog::getDefaultMaxSize());
+
+    if (log_settings.queue_settings.max_size_rows < 1)
+        throw Exception(ErrorCodes::BAD_ARGUMENTS, "{0}.max_size_rows {1} should be 1 at least",
+                        config_prefix,
+                        log_settings.queue_settings.max_size_rows);
+
+    log_settings.queue_settings.reserved_size_rows = config.getUInt64(config_prefix + ".reserved_size_rows",
+                                                                      TSystemLog::getDefaultReservedSize());
+
+    if (log_settings.queue_settings.max_size_rows < log_settings.queue_settings.reserved_size_rows)
+    {
+         throw Exception(ErrorCodes::BAD_ARGUMENTS,
+                         "{0}.max_size_rows {1} should be greater or equal to {0}.reserved_size_rows {2}",
+                         config_prefix,
+                         log_settings.queue_settings.max_size_rows,
+                         log_settings.queue_settings.reserved_size_rows);
+    }
+
+    log_settings.queue_settings.buffer_size_rows_flush_threshold = config.getUInt64(config_prefix + ".buffer_size_rows_flush_threshold",
+                                                                                    log_settings.queue_settings.max_size_rows / 2);
+
+    log_settings.queue_settings.notify_flush_on_crash = config.getBool(config_prefix + ".flush_on_crash",
+                                                                       TSystemLog::shouldNotifyFlushOnCrash());
+
+    log_settings.queue_settings.turn_off_logger = TSystemLog::shouldTurnOffLogger();
+
+    return std::make_shared<TSystemLog>(context, log_settings);
 }


@ -325,23 +357,25 @@ void SystemLogs::shutdown()
        log->shutdown();
 }

+void SystemLogs::handleCrash()
+{
+    for (auto & log : logs)
+        log->handleCrash();
+}

 template <typename LogElement>
 SystemLog<LogElement>::SystemLog(
    ContextPtr context_,
-    const String & database_name_,
-    const String & table_name_,
-    const String & storage_def_,
-    size_t flush_interval_milliseconds_,
+    const SystemLogSettings & settings_,
    std::shared_ptr<SystemLogQueue<LogElement>> queue_)
-    : Base(database_name_ + "." + table_name_, flush_interval_milliseconds_, queue_)
+    : Base(settings_.queue_settings, queue_)
    , WithContext(context_)
-    , log(&Poco::Logger::get("SystemLog (" + database_name_ + "." + table_name_ + ")"))
-    , table_id(database_name_, table_name_)
-    , storage_def(storage_def_)
+    , log(&Poco::Logger::get("SystemLog (" + settings_.queue_settings.database + "." + settings_.queue_settings.table + ")"))
+    , table_id(settings_.queue_settings.database, settings_.queue_settings.table)
+    , storage_def(settings_.engine)
    , create_query(serializeAST(*getCreateTableQuery()))
 {
-    assert(database_name_ == DatabaseCatalog::SYSTEM_DATABASE);
+    assert(settings_.queue_settings.database == DatabaseCatalog::SYSTEM_DATABASE);
 }

 template <typename LogElement>
--- a/src/Interpreters/SystemLog.h
+++ b/src/Interpreters/SystemLog.h
@ -58,6 +58,7 @@ struct SystemLogs
    ~SystemLogs();

    void shutdown();
+    void handleCrash();

    std::shared_ptr<QueryLog> query_log;                /// Used to log queries.
    std::shared_ptr<QueryThreadLog> query_thread_log;   /// Used to log query threads.
@ -87,6 +88,12 @@ struct SystemLogs
    std::vector<ISystemLog *> logs;
 };

+struct SystemLogSettings
+{
+    SystemLogQueueSettings queue_settings;
+
+    String engine;
+};

 template <typename LogElement>
 class SystemLog : public SystemLogBase<LogElement>, private boost::noncopyable, WithContext
@ -103,13 +110,9 @@ public:
      *   where N - is a minimal number from 1, for that table with corresponding name doesn't exist yet;
      *   and new table get created - as if previous table was not exist.
      */
-    SystemLog(
-        ContextPtr context_,
-        const String & database_name_,
-        const String & table_name_,
-        const String & storage_def_,
-        size_t flush_interval_milliseconds_,
-        std::shared_ptr<SystemLogQueue<LogElement>> queue_ = nullptr);
+    SystemLog(ContextPtr context_,
+              const SystemLogSettings& settings_,
+              std::shared_ptr<SystemLogQueue<LogElement>> queue_ = nullptr);

    /** Append a record into log.
      * Writing to table will be done asynchronously and in case of failure, record could be lost.
--- a/src/Interpreters/TextLog.cpp
+++ b/src/Interpreters/TextLog.cpp
@ -80,15 +80,10 @@ void TextLogElement::appendToBlock(MutableColumns & columns) const
    columns[i++]->insert(message_format_string);
 }

-TextLog::TextLog(ContextPtr context_, const String & database_name_,
-        const String & table_name_, const String & storage_def_,
-        size_t flush_interval_milliseconds_)
-  : SystemLog<TextLogElement>(context_, database_name_, table_name_,
-        storage_def_, flush_interval_milliseconds_, getLogQueue(flush_interval_milliseconds_))
+TextLog::TextLog(ContextPtr context_,
+                 const SystemLogSettings & settings)
+    : SystemLog<TextLogElement>(context_, settings, getLogQueue(settings.queue_settings))
 {
-    // SystemLog methods may write text logs, so we disable logging for the text
-    // log table to avoid recursion.
-    log->setLevel(0);
 }

 }
--- a/src/Interpreters/TextLog.h
+++ b/src/Interpreters/TextLog.h
@ -42,18 +42,15 @@ class TextLog : public SystemLog<TextLogElement>
 public:
    using Queue = SystemLogQueue<TextLogElement>;

-    TextLog(
-        ContextPtr context_,
-        const String & database_name_,
-        const String & table_name_,
-        const String & storage_def_,
-        size_t flush_interval_milliseconds_);
+    explicit TextLog(ContextPtr context_, const SystemLogSettings & settings);

-    static std::shared_ptr<Queue> getLogQueue(size_t flush_interval_milliseconds)
+    static std::shared_ptr<Queue> getLogQueue(const SystemLogQueueSettings & settings)
    {
-        static std::shared_ptr<Queue> queue = std::make_shared<Queue>("text_log", flush_interval_milliseconds, true);
+        static std::shared_ptr<Queue> queue = std::make_shared<Queue>(settings);
        return queue;
    }
+
+    static consteval bool shouldTurnOffLogger() { return true; }
 };

 }
--- a/src/Interpreters/ThreadStatusExt.cpp
+++ b/src/Interpreters/ThreadStatusExt.cpp
@ -513,7 +513,7 @@ void ThreadStatus::logToQueryThreadLog(QueryThreadLog & thread_log, const String
        }
    }

-    thread_log.add(elem);
+    thread_log.add(std::move(elem));
 }

 static String getCleanQueryAst(const ASTPtr q, ContextPtr context)
@ -573,7 +573,7 @@ void ThreadStatus::logToQueryViewsLog(const ViewRuntimeData & vinfo)
            element.stack_trace = getExceptionStackTraceString(vinfo.exception);
    }

-    views_log->add(element);
+    views_log->add(std::move(element));
 }

 void CurrentThread::attachToGroup(const ThreadGroupPtr & thread_group)
--- a/src/Interpreters/TraceCollector.cpp
+++ b/src/Interpreters/TraceCollector.cpp
@ -128,7 +128,7 @@ void TraceCollector::run()
                UInt64 time = static_cast<UInt64>(ts.tv_sec * 1000000000LL + ts.tv_nsec);
                UInt64 time_in_microseconds = static_cast<UInt64>((ts.tv_sec * 1000000LL) + (ts.tv_nsec / 1000));
                TraceLogElement element{time_t(time / 1000000000), time_in_microseconds, time, trace_type, thread_id, query_id, trace, size, event, increment};
-                trace_log->add(element);
+                trace_log->add(std::move(element));
            }
        }
    }
--- a/src/Interpreters/TransactionLog.cpp
+++ b/src/Interpreters/TransactionLog.cpp
@ -34,7 +34,7 @@ try
    elem.tid = tid;
    elem.csn = csn;
    elem.fillCommonFields(nullptr);
-    system_log->add(elem);
+    system_log->add(std::move(elem));
 }
 catch (...)
 {
--- a/src/Interpreters/TransactionsInfoLog.cpp
+++ b/src/Interpreters/TransactionsInfoLog.cpp
@ -101,7 +101,7 @@ try
    elem.type = type;
    elem.tid = tid;
    elem.fillCommonFields(&context);
-    system_log->add(elem);
+    system_log->add(std::move(elem));
 }
 catch (...)
 {
--- a/src/Loggers/Loggers.cpp
+++ b/src/Loggers/Loggers.cpp
@ -21,6 +21,12 @@ namespace fs = std::filesystem;
 namespace DB
 {
    class SensitiveDataMasker;
+
+namespace ErrorCodes
+{
+    extern const int BAD_ARGUMENTS;
+}
+
 }


@ -44,10 +50,6 @@ static std::string renderFileNameTemplate(time_t now, const std::string & file_p
    return path.replace_filename(ss.str());
 }

-#ifndef WITHOUT_TEXT_LOG
-constexpr size_t DEFAULT_SYSTEM_LOG_FLUSH_INTERVAL_MILLISECONDS = 7500;
-#endif
-
 void Loggers::buildLoggers(Poco::Util::AbstractConfiguration & config, Poco::Logger & logger /*_root*/, const std::string & cmd_name)
 {
    auto current_logger = config.getString("logger", "");
@ -271,9 +273,37 @@ void Loggers::buildLoggers(Poco::Util::AbstractConfiguration & config, Poco::Log
    {
        String text_log_level_str = config.getString("text_log.level", "trace");
        int text_log_level = Poco::Logger::parseLevel(text_log_level_str);
-        size_t flush_interval_milliseconds = config.getUInt64("text_log.flush_interval_milliseconds",
-            DEFAULT_SYSTEM_LOG_FLUSH_INTERVAL_MILLISECONDS);
-        split->addTextLog(DB::TextLog::getLogQueue(flush_interval_milliseconds), text_log_level);
+
+        DB::SystemLogQueueSettings log_settings;
+        log_settings.flush_interval_milliseconds = config.getUInt64("text_log.flush_interval_milliseconds",
+                                                                    DB::TextLog::getDefaultFlushIntervalMilliseconds());
+
+        log_settings.max_size_rows = config.getUInt64("text_log.max_size_rows",
+                                                      DB::TextLog::getDefaultMaxSize());
+
+        if (log_settings.max_size_rows< 1)
+            throw DB::Exception(DB::ErrorCodes::BAD_ARGUMENTS, "text_log.max_size_rows {} should be 1 at least",
+                                log_settings.max_size_rows);
+
+        log_settings.reserved_size_rows = config.getUInt64("text_log.reserved_size_rows", DB::TextLog::getDefaultReservedSize());
+
+        if (log_settings.max_size_rows < log_settings.reserved_size_rows)
+        {
+            throw DB::Exception(DB::ErrorCodes::BAD_ARGUMENTS,
+                                "text_log.max_size {0} should be greater or equal to text_log.reserved_size_rows {1}",
+                                log_settings.max_size_rows,
+                                log_settings.reserved_size_rows);
+        }
+
+        log_settings.buffer_size_rows_flush_threshold = config.getUInt64("text_log.buffer_size_rows_flush_threshold",
+                                                                         log_settings.max_size_rows / 2);
+
+        log_settings.notify_flush_on_crash = config.getBool("text_log.flush_on_crash",
+                                                            DB::TextLog::shouldNotifyFlushOnCrash());
+
+        log_settings.turn_off_logger = DB::TextLog::shouldTurnOffLogger();
+
+        split->addTextLog(DB::TextLog::getLogQueue(log_settings), text_log_level);
    }
 #endif
 }
--- a/src/Loggers/OwnSplitChannel.cpp
+++ b/src/Loggers/OwnSplitChannel.cpp
@ -138,7 +138,7 @@ void OwnSplitChannel::logSplit(const Poco::Message & msg)
        std::shared_ptr<SystemLogQueue<TextLogElement>> text_log_locked{};
        text_log_locked = text_log.lock();
        if (text_log_locked)
-            text_log_locked->push(elem);
+            text_log_locked->push(std::move(elem));
    }
 #endif
 }
--- a/src/Storages/MergeTree/MergeTreeData.cpp
+++ b/src/Storages/MergeTree/MergeTreeData.cpp
@ -7775,7 +7775,7 @@ try
        LOG_WARNING(log, "Profile counters are not set");
    }

-    part_log->add(part_log_elem);
+    part_log->add(std::move(part_log_elem));
 }
 catch (...)
 {
--- a/tests/integration/parallel_skip.json
+++ b/tests/integration/parallel_skip.json
@ -74,5 +74,12 @@
  
  "test_http_failover/test.py::test_url_destination_host_with_multiple_addrs",
  "test_http_failover/test.py::test_url_invalid_hostname",
-  "test_http_failover/test.py::test_url_ip_change"
+  "test_http_failover/test.py::test_url_ip_change",
+
+  "test_system_logs/test_system_logs.py::test_max_size_0",
+  "test_system_logs/test_system_logs.py::test_reserved_size_greater_max_size",
+  "test_system_flush_logs/test.py::test_log_buffer_size_rows_flush_threshold",
+  "test_system_flush_logs/test.py::test_log_max_size",
+  "test_crash_log/test.py::test_pkill_query_log",
+  "test_crash_log/test.py::test_pkill"
 ]
--- a/tests/integration/test_crash_log/configs/crash_log.xml
+++ b/tests/integration/test_crash_log/configs/crash_log.xml
@ -0,0 +1,16 @@
+<clickhouse>
+    <crash_log replace="replace">
+        <flush_interval_milliseconds>1000000</flush_interval_milliseconds>
+        <max_size_rows>1</max_size_rows>
+        <reserved_size_rows>1</reserved_size_rows>
+        <buffer_size_rows_flush_threshold>1</buffer_size_rows_flush_threshold>
+        <flush_on_crash>true</flush_on_crash>
+    </crash_log>
+    <query_log replace="replace">
+        <flush_interval_milliseconds>1000000</flush_interval_milliseconds>
+        <max_size_rows>100</max_size_rows>
+        <reserved_size_rows>100</reserved_size_rows>
+        <buffer_size_rows_flush_threshold>100</buffer_size_rows_flush_threshold>
+        <flush_on_crash>true</flush_on_crash>
+    </query_log>
+</clickhouse>
--- a/tests/integration/test_crash_log/test.py
+++ b/tests/integration/test_crash_log/test.py
@ -12,7 +12,9 @@ SCRIPT_DIR = os.path.dirname(os.path.realpath(__file__))
 def started_node():
    cluster = helpers.cluster.ClickHouseCluster(__file__)
    try:
-        node = cluster.add_instance("node", stay_alive=True)
+        node = cluster.add_instance(
+            "node", main_configs=["configs/crash_log.xml"], stay_alive=True
+        )

        cluster.start()
        yield node
@ -55,3 +57,18 @@ def test_pkill(started_node):
            started_node.query("SELECT COUNT(*) FROM system.crash_log")
            == f"{crashes_count}\n"
        )
+
+
+def test_pkill_query_log(started_node):
+    for signal in ["SEGV", "4"]:
+        # force create query_log if it was not created
+        started_node.query("SYSTEM FLUSH LOGS")
+        started_node.query("TRUNCATE TABLE IF EXISTS system.query_log")
+        started_node.query("SELECT COUNT(*) FROM system.query_log")
+        # logs don't flush
+        assert started_node.query("SELECT COUNT(*) FROM system.query_log") == f"{0}\n"
+
+        send_signal(started_node, signal)
+        wait_for_clickhouse_stop(started_node)
+        started_node.restart_clickhouse()
+        assert started_node.query("SELECT COUNT(*) FROM system.query_log") >= f"3\n"
--- a/tests/integration/test_system_flush_logs/test.py
+++ b/tests/integration/test_system_flush_logs/test.py
@ -2,11 +2,16 @@
 # 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

 cluster = ClickHouseCluster(__file__)
-node = cluster.add_instance("node_default")
+node = cluster.add_instance(
+    "node_default",
+    stay_alive=True,
+)

 system_logs = [
    # disabled by default
@ -64,3 +69,95 @@ def test_system_suspend():
    node.query("SYSTEM SUSPEND FOR 1 SECOND;")
    node.query("INSERT INTO t VALUES (now());")
    assert "1\n" == node.query("SELECT max(x) - min(x) >= 1 FROM t;")
+
+
+def test_log_max_size(start_cluster):
+    node.exec_in_container(
+        [
+            "bash",
+            "-c",
+            f"""echo "
+        <clickhouse>
+            <query_log>
+                <flush_interval_milliseconds replace=\\"replace\\">1000000</flush_interval_milliseconds>
+                <max_size_rows replace=\\"replace\\">10</max_size_rows>
+                <reserved_size_rows replace=\\"replace\\">10</reserved_size_rows>
+            </query_log>
+        </clickhouse>
+        " > /etc/clickhouse-server/config.d/yyy-override-query_log.xml
+        """,
+        ]
+    )
+    node.restart_clickhouse()
+    for i in range(10):
+        node.query(f"select {i}")
+
+    assert node.query("select count() >= 10 from system.query_log") == "1\n"
+    node.exec_in_container(
+        ["rm", f"/etc/clickhouse-server/config.d/yyy-override-query_log.xml"]
+    )
+
+
+def test_log_buffer_size_rows_flush_threshold(start_cluster):
+    node.exec_in_container(
+        [
+            "bash",
+            "-c",
+            f"""echo "
+        <clickhouse>
+            <query_log>
+                <flush_interval_milliseconds replace=\\"replace\\">1000000</flush_interval_milliseconds>
+                <buffer_size_rows_flush_threshold replace=\\"replace\\">10</buffer_size_rows_flush_threshold>
+                <max_size_rows replace=\\"replace\\">10000</max_size_rows>
+            </query_log>
+        </clickhouse>
+        " > /etc/clickhouse-server/config.d/yyy-override-query_log.xml
+        """,
+        ]
+    )
+    node.restart_clickhouse()
+    node.query(f"TRUNCATE TABLE IF EXISTS system.query_log")
+    for i in range(10):
+        node.query(f"select {i}")
+
+    assert_eq_with_retry(
+        node,
+        f"select count() >= 11 from system.query_log",
+        "1",
+        sleep_time=0.2,
+        retry_count=100,
+    )
+
+    node.query(f"TRUNCATE TABLE IF EXISTS system.query_log")
+    node.exec_in_container(
+        [
+            "bash",
+            "-c",
+            f"""echo "
+        <clickhouse>
+            <query_log>
+                <flush_interval_milliseconds replace=\\"replace\\">1000000</flush_interval_milliseconds>
+                <buffer_size_rows_flush_threshold replace=\\"replace\\">10000</buffer_size_rows_flush_threshold>
+                <max_size_rows replace=\\"replace\\">10000</max_size_rows>
+            </query_log>
+        </clickhouse>
+        " > /etc/clickhouse-server/config.d/yyy-override-query_log.xml
+        """,
+        ]
+    )
+    node.restart_clickhouse()
+    for i in range(10):
+        node.query(f"select {i}")
+
+    # Logs aren't flushed
+    assert_eq_with_retry(
+        node,
+        f"select count() < 10 from system.query_log",
+        "1",
+        sleep_time=0.2,
+        retry_count=100,
+    )
+
+    node.exec_in_container(
+        ["rm", f"/etc/clickhouse-server/config.d/yyy-override-query_log.xml"]
+    )
--- a/tests/integration/test_system_logs/test_system_logs.py
+++ b/tests/integration/test_system_logs/test_system_logs.py
@ -88,3 +88,53 @@ def test_system_logs_settings_expr(start_cluster):
    assert expected in node3.query(
        "SELECT engine_full FROM system.tables WHERE database='system' and name='query_log'"
    )
+
+
+def test_max_size_0(start_cluster):
+    node1.exec_in_container(
+        [
+            "bash",
+            "-c",
+            f"""echo "
+        <clickhouse>
+            <query_log>
+                <max_size_rows replace=\\"replace\\">0</max_size_rows> 
+                <reserved_size_rows replace=\\"replace\\">0</reserved_size_rows>
+            </query_log>
+        </clickhouse>
+        " > /etc/clickhouse-server/config.d/yyy-override-query_log.xml
+        """,
+        ]
+    )
+    with pytest.raises(Exception):
+        node1.restart_clickhouse()
+
+    node1.exec_in_container(
+        ["rm", f"/etc/clickhouse-server/config.d/yyy-override-query_log.xml"]
+    )
+    node1.restart_clickhouse()
+
+
+def test_reserved_size_greater_max_size(start_cluster):
+    node1.exec_in_container(
+        [
+            "bash",
+            "-c",
+            f"""echo "
+        <clickhouse>
+            <query_log>
+                <max_size_rows replace=\\"replace\\">10</max_size_rows>
+                <reserved_size_rows replace=\\"replace\\">11</reserved_size_rows> 
+            </query_log>
+        </clickhouse>
+        " > /etc/clickhouse-server/config.d/yyy-override-query_log.xml
+        """,
+        ]
+    )
+    with pytest.raises(Exception):
+        node1.restart_clickhouse()
+
+    node1.exec_in_container(
+        ["rm", f"/etc/clickhouse-server/config.d/yyy-override-query_log.xml"]
+    )
+    node1.restart_clickhouse()