diff --git a/docs/en/operations/server-configuration-parameters/settings.md b/docs/en/operations/server-configuration-parameters/settings.md
index a6ae517e4011..e9f0f0dae002 100644
--- 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,
diff --git a/docs/en/operations/system-tables/index.md b/docs/en/operations/system-tables/index.md
index 1b720098fc79..a46f306f6777 100644
--- 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>
 ```
diff --git a/docs/ru/operations/server-configuration-parameters/settings.md b/docs/ru/operations/server-configuration-parameters/settings.md
index 421df3fe3eb7..81a696bcfc10 100644
--- 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}
 
 Правила, основанные на регулярных выражениях, которые будут применены для всех запросов, а также для всех сообщений перед сохранением их в лог на сервере,
diff --git a/docs/ru/operations/system-tables/index.md b/docs/ru/operations/system-tables/index.md
index 7ff368b19101..24f79cae212c 100644
--- 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>
 ```
diff --git a/programs/server/config.xml b/programs/server/config.xml
index 2a7dc1e576a7..153cb728bb43 100644
--- 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>
diff --git a/src/Common/SystemLogBase.cpp b/src/Common/SystemLogBase.cpp
index 3d68fe63227c..919020f12c91 100644
--- 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,11 +217,26 @@ 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>
@@ -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>
diff --git a/src/Common/SystemLogBase.h b/src/Common/SystemLogBase.h
index f6e4a579edfe..28ccc91b014c 100644
--- a/src/Common/SystemLogBase.h
+++ b/src/Common/SystemLogBase.h
@@ -62,6 +62,9 @@ class ISystemLog
 
     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 @@ class ISystemLog
     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 @@ class SystemLogQueue
     bool is_shutdown = false;
 
     std::condition_variable flush_event;
-    const size_t flush_interval_milliseconds;
+
+    const SystemLogQueueSettings settings;
 };
 
 
@@ -131,8 +147,7 @@ class SystemLogBase : public ISystemLog
     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 @@ class SystemLogBase : public ISystemLog
     /** 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;
diff --git a/src/Common/ZooKeeper/ZooKeeperImpl.cpp b/src/Common/ZooKeeper/ZooKeeperImpl.cpp
index 5e16a437be31..38f91880ddff 100644
--- 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
diff --git a/src/Daemon/BaseDaemon.cpp b/src/Daemon/BaseDaemon.cpp
index f61ca054b2ae..3e9c13f8c5dd 100644
--- a/src/Daemon/BaseDaemon.cpp
+++ b/src/Daemon/BaseDaemon.cpp
@@ -466,6 +466,10 @@ class SignalListener : public Poco::Runnable
         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)
         {
diff --git a/src/Disks/IO/AsynchronousBoundedReadBuffer.cpp b/src/Disks/IO/AsynchronousBoundedReadBuffer.cpp
index 86ee541dcbd5..57f6af8178b1 100644
--- 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));
 }
 
 
diff --git a/src/Disks/IO/CachedOnDiskReadBufferFromFile.cpp b/src/Disks/IO/CachedOnDiskReadBufferFromFile.cpp
index 6674eefeab12..6f08c6ba912a 100644
--- 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)
diff --git a/src/Disks/IO/CachedOnDiskWriteBufferFromFile.cpp b/src/Disks/IO/CachedOnDiskWriteBufferFromFile.cpp
index 33d4ed7b3d7f..128b8df20460 100644
--- 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()
diff --git a/src/Disks/IO/ReadBufferFromRemoteFSGather.cpp b/src/Disks/IO/ReadBufferFromRemoteFSGather.cpp
index 800cc0883e67..ada1b2cbecd1 100644
--- 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)
diff --git a/src/Interpreters/AsynchronousInsertQueue.cpp b/src/Interpreters/AsynchronousInsertQueue.cpp
index 01ebea86b669..4f15192f46cb 100644
--- 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
             {
diff --git a/src/Interpreters/Context.cpp b/src/Interpreters/Context.cpp
index f83e524ffb9d..1bedbe030572 100644
--- 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();
diff --git a/src/Interpreters/Context.h b/src/Interpreters/Context.h
index 75752774d4ce..7f6ad37208ce 100644
--- a/src/Interpreters/Context.h
+++ b/src/Interpreters/Context.h
@@ -998,6 +998,9 @@ class Context: public std::enable_shared_from_this<Context>
     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.
diff --git a/src/Interpreters/CrashLog.cpp b/src/Interpreters/CrashLog.cpp
index 379c9122cc81..ec693eb7931e 100644
--- 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));
     }
 }
diff --git a/src/Interpreters/CrashLog.h b/src/Interpreters/CrashLog.h
index 78794574c826..65714295be42 100644
--- a/src/Interpreters/CrashLog.h
+++ b/src/Interpreters/CrashLog.h
@@ -45,6 +45,11 @@ class CrashLog : public SystemLog<CrashLogElement>
     {
         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; }
 };
 
 }
diff --git a/src/Interpreters/MetricLog.cpp b/src/Interpreters/MetricLog.cpp
index 24f77f7d0ba3..24e9e4487ae8 100644
--- 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.
diff --git a/src/Interpreters/PartLog.cpp b/src/Interpreters/PartLog.cpp
index 881fcae4de66..a97f1f405bcd 100644
--- 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 (...)
diff --git a/src/Interpreters/ProcessorsProfileLog.cpp b/src/Interpreters/ProcessorsProfileLog.cpp
index e78a07bb752e..14159ad3438b 100644
--- 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_)
-{
-}
 
 }
diff --git a/src/Interpreters/ProcessorsProfileLog.h b/src/Interpreters/ProcessorsProfileLog.h
index 81d58edd913c..63791c0374c2 100644
--- 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;
 };
 
 }
diff --git a/src/Interpreters/Session.cpp b/src/Interpreters/Session.cpp
index 97b056cfc32b..cadf619700cb 100644
--- a/src/Interpreters/Session.cpp
+++ b/src/Interpreters/Session.cpp
@@ -240,7 +240,7 @@ class NamedSessionsStorage
 
             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);
 
diff --git a/src/Interpreters/SessionLog.cpp b/src/Interpreters/SessionLog.cpp
index c930013e52b1..0a8a7fc18c55 100644
--- 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));
 }
 
 }
diff --git a/src/Interpreters/SystemLog.cpp b/src/Interpreters/SystemLog.cpp
index 0b89b1dec26f..be0468aa8764 100644
--- 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>
diff --git a/src/Interpreters/SystemLog.h b/src/Interpreters/SystemLog.h
index 5d8bb30150dd..437b1b2a6bb4 100644
--- 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 @@ class SystemLog : public SystemLogBase<LogElement>, private boost::noncopyable,
       *   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.
diff --git a/src/Interpreters/TextLog.cpp b/src/Interpreters/TextLog.cpp
index 108135c78b39..3951a41f0c5d 100644
--- 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);
 }
 
 }
diff --git a/src/Interpreters/TextLog.h b/src/Interpreters/TextLog.h
index 60ca11632aaf..4bfed5327f33 100644
--- 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_);
-
-    static std::shared_ptr<Queue> getLogQueue(size_t flush_interval_milliseconds)
+    explicit TextLog(ContextPtr context_, const SystemLogSettings & settings);
+
+    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; }
 };
 
 }
diff --git a/src/Interpreters/ThreadStatusExt.cpp b/src/Interpreters/ThreadStatusExt.cpp
index 398bea26b87c..7a6bc45c1188 100644
--- 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)
diff --git a/src/Interpreters/TraceCollector.cpp b/src/Interpreters/TraceCollector.cpp
index cb00e37df69b..19cc5c4e6bdb 100644
--- 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));
             }
         }
     }
diff --git a/src/Interpreters/TransactionLog.cpp b/src/Interpreters/TransactionLog.cpp
index 2ef4f4d62183..631e7f5c7465 100644
--- 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 (...)
 {
diff --git a/src/Interpreters/TransactionsInfoLog.cpp b/src/Interpreters/TransactionsInfoLog.cpp
index b62cd4672d84..90f5022a444b 100644
--- 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 (...)
 {
diff --git a/src/Loggers/Loggers.cpp b/src/Loggers/Loggers.cpp
index 271ab39cd88b..90b3457b7d8d 100644
--- 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
 }
diff --git a/src/Loggers/OwnSplitChannel.cpp b/src/Loggers/OwnSplitChannel.cpp
index b5ac42d6041d..cdf8402745f7 100644
--- 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
 }
diff --git a/src/Storages/MergeTree/MergeTreeData.cpp b/src/Storages/MergeTree/MergeTreeData.cpp
index 6179c70ca575..013a9d6923c4 100644
--- 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 (...)
 {
diff --git a/tests/integration/parallel_skip.json b/tests/integration/parallel_skip.json
index 1075fbaa0f8f..6e1604f4eb57 100644
--- 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"
 ]
diff --git a/tests/integration/test_crash_log/configs/crash_log.xml b/tests/integration/test_crash_log/configs/crash_log.xml
new file mode 100644
index 000000000000..f4fbfaba08e0
--- /dev/null
+++ 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>
diff --git a/tests/integration/test_crash_log/test.py b/tests/integration/test_crash_log/test.py
index 9f6eca794b1f..1b7e7f382429 100644
--- a/tests/integration/test_crash_log/test.py
+++ b/tests/integration/test_crash_log/test.py
@@ -12,7 +12,9 @@
 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"
diff --git a/tests/integration/test_system_flush_logs/test.py b/tests/integration/test_system_flush_logs/test.py
index d9ab76d2d61e..bf225ac30f87 100644
--- 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"]
+    )
diff --git a/tests/integration/test_system_logs/test_system_logs.py b/tests/integration/test_system_logs/test_system_logs.py
index aac5ee538196..72249cd64ee6 100644
--- 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()