Parallel FREEZE TABLE

ch_backup/ch_backup.py

@@ -48,7 +48,9 @@ def __init__(self, config: Config) -> None:
         self._config = config
         self._access_backup_manager = AccessBackup()
         self._database_backup_manager = DatabaseBackup()
-        self._table_backup_manager = TableBackup()
+        self._table_backup_manager = TableBackup(
+            self._config["multiprocessing"]["freeze_threads"]
+        )
         self._udf_backup_manager = UDFBackup()
         self._nc_backup_manager = NamedCollectionsBackup()
 

ch_backup/clickhouse/control.py

@@ -24,6 +24,7 @@
 from ch_backup.exceptions import ClickhouseBackupError
 from ch_backup.util import (
     chown_dir_contents,
+    chown_file,
     escape,
     list_dir_files,
     retry,
@@ -474,15 +475,27 @@ def system_unfreeze(self, backup_name: str) -> None:
             query_sql = SYSTEM_UNFREEZE_SQL.format(backup_name=backup_name)
             self._ch_client.query(query_sql, timeout=self._unfreeze_timeout)
 
-    def remove_freezed_data(self) -> None:
+    def remove_freezed_data(
+        self, backup_name: Optional[str] = None, table: Optional[Table] = None
+    ) -> None:
         """
         Remove all freezed partitions from all local disks.
         """
-        for disk in self._disks.values():
-            if disk.type == "local":
-                shadow_path = os.path.join(disk.path, "shadow")
-                logging.debug("Removing shadow data: {}", shadow_path)
-                self._remove_shadow_data(shadow_path)
+        if backup_name and table:
+            for table_data_path, disk in table.paths_with_disks:
+                if disk.type == "local":
+                    table_relative_path = os.path.relpath(table_data_path, disk.path)
+                    shadow_path = os.path.join(
+                        disk.path, "shadow", backup_name, table_relative_path
+                    )
+                    logging.debug("Removing shadow data: {}", shadow_path)
+                    self._remove_shadow_data(shadow_path)
+        else:
+            for disk in self._disks.values():
+                if disk.type == "local":
+                    shadow_path = os.path.join(disk.path, "shadow")
+                    logging.debug("Removing shadow data: {}", shadow_path)
+                    self._remove_shadow_data(shadow_path)
 
     def remove_freezed_part(self, part: FrozenPart) -> None:
         """
@@ -776,6 +789,26 @@ def chown_dir(self, dir_path: str) -> None:
             need_recursion,
         )
 
+    def create_shadow_increment(self) -> None:
+        """
+        Create shadow/increment.txt to fix race condition with parallel freeze.
+        Must be used before freezing more than one table at once.
+        """
+        default_shadow_path = Path(self._root_data_path) / "shadow"
+        increment_path = default_shadow_path / "increment.txt"
+        if os.path.exists(increment_path):
+            return
+        if not os.path.exists(default_shadow_path):
+            os.mkdir(default_shadow_path)
+            self.chown_dir(str(default_shadow_path))
+        with open(increment_path, "w", encoding="utf-8") as file:
+            file.write("0")
+            chown_file(
+                self._ch_ctl_config["user"],
+                self._ch_ctl_config["group"],
+                str(increment_path),
+            )
+
     @retry(OSError)
     def _remove_shadow_data(self, path: str) -> None:
         if path.find("/shadow") == -1:

ch_backup/config.py

@@ -156,6 +156,8 @@ def _as_seconds(t: str) -> int:
         "workers": 4,
         # The number of processes for parts restoring from S3 disks.
         "cloud_storage_restore_workers": 4,
+        # The number of threads for parallel freeze of tables
+        "freeze_threads": 4,
     },
     "pipeline": {
         # Is asynchronous pipelines used (based on Pypeln library)