Add max_execution_time for freeze queries (#284)

* Add max_execution_time for freeze queries

* add kill

Author: MikhailBurdukov

ch_backup/clickhouse/control.py

@@ -207,6 +207,12 @@
 """
 )
 
+KILL_RUNNING_FREEZE_SQL = strip_query(
+    """
+    KILL QUERY WHERE query_kind='Alter' AND user='{user}' AND query ilike '%ALTER TABLE%FREEZE%'
+"""
+)
+
 DROP_TABLE_IF_EXISTS_SQL = strip_query(
     """
     DROP TABLE IF EXISTS `{db_name}`.`{table_name}` NO DELAY
@@ -586,6 +592,14 @@ def attach_table(self, table: Union[TableMetadata, Table]) -> None:
 
         self._ch_client.query(query_sql)
 
+    def kill_old_freeze_queries(self):
+        """
+        Collect all freeze queries from prev backup and kill them all.
+        """
+        self._ch_client.query(
+            KILL_RUNNING_FREEZE_SQL.format(user=self._ch_ctl_config["user"])
+        )
+
     def freeze_table(
         self,
         backup_name: str,
@@ -598,6 +612,12 @@ def freeze_table(
         # Table has no partitions or created with deprecated syntax.
         # FREEZE PARTITION ID with deprecated syntax throws segmentation fault in CH.
         freeze_by_partitions = threads > 0 and "PARTITION BY" in table.create_statement
+
+        query_settings = None
+        # Since https://github.com/ClickHouse/ClickHouse/pull/75016
+        if self.ch_version_ge("25.2"):
+            query_settings = {"max_execution_time": self._freeze_timeout}
+
         if freeze_by_partitions:
             with ThreadExecPool(max(1, threads)) as pool:
                 if freeze_by_partitions:
@@ -613,6 +633,7 @@ def freeze_table(
                             f'Freeze partition "{partition}"',
                             self._ch_client.query,
                             query_sql,
+                            settings=query_settings,
                             timeout=self._freeze_timeout,
                             should_retry=False,
                             new_session=True,
@@ -629,6 +650,7 @@ def freeze_table(
             )
             self._ch_client.query(
                 query_sql,
+                settings=query_settings,
                 timeout=self._freeze_timeout,
                 should_retry=False,
                 new_session=True,

ch_backup/config.py

@@ -73,6 +73,7 @@ def _as_seconds(t: str) -> int:
         "validate_part_after_upload": False,
         "restore_fail_on_attach_error": False,
         "update_metadata_interval": _as_seconds("30 min"),
+        "kill_old_freeze_queries": True,
     },
     "restore": {
         "use_inplace_cloud_restore": False,

ch_backup/logic/table.py

@@ -77,6 +77,13 @@ def backup(
             logging.debug('Cloud Storage "shadow" backup will be compressed')
             context.backup_meta.cloud_storage.compress()
 
+        # Since https://github.com/ClickHouse/ClickHouse/pull/75016
+        if (
+            context.ch_ctl.ch_version_ge("25.2")
+            and context.config["kill_old_freeze_queries"]
+        ):
+            context.ch_ctl.kill_old_freeze_queries()
+
         for db in databases:
             self._backup(
                 context,