fix remove orphaned

ch_tools/chadmin/cli/data_store_group.py

@@ -21,6 +21,7 @@
     S3ObjectLocalInfo,
     S3ObjectLocalMetaData,
 )
+from ch_tools.chadmin.internal.system import get_version
 from ch_tools.chadmin.internal.utils import execute_query, remove_from_disk
 from ch_tools.common import logging
 from ch_tools.common.cli.formatting import print_response
@@ -258,6 +259,7 @@ def cleanup_data_dir(
                         "disk": disk,
                         "path_to_disk": path_to_disk,
                         "disks_config_path": disk_config_path,
+                        "ch_version": get_version(ctx),
                     },
                 )
                 tasks.append(task)
@@ -276,15 +278,20 @@ def remove_orphaned_sql_object_metadata(data):
     data["deleted"] = "Yes"
 
 
-def remove_orphaned_sql_object_full(data, disk, path_to_disk, disks_config_path):
+def remove_orphaned_sql_object_full(
+    data, disk, path_to_disk, ch_version, disks_config_path
+):
 
     path = data["path"]
 
     if not path.startswith(path_to_disk):
         raise RuntimeError(f"Path {path} on fs does not math with disk {disk}")
     relative_path_on_disk = path[len(path_to_disk) + 1 :]
     retcode, stderr = remove_from_ch_disk(
-        disk, relative_path_on_disk, disks_config_path
+        disk=disk,
+        path=relative_path_on_disk,
+        ch_version=ch_version,
+        disk_config_path=disks_config_path,
     )
     if retcode:
         raise RuntimeError(

ch_tools/chadmin/internal/clickhouse_disks.py

@@ -35,7 +35,7 @@ def make_ch_disks_config(disk: str) -> str:
 
 
 def remove_from_ch_disk(
-    disk: str, path: str, ch_version: str = "", disk_config_path: Optional[str] = None
+    disk: str, path: str, ch_version: str, disk_config_path: Optional[str] = None
 ) -> Tuple[int, bytes]:
     cmd = f"clickhouse-disks {'-C ' + disk_config_path if disk_config_path else ''} --disk {disk}"
     if match_str_ch_version(ch_version, "24.7"):

ch_tools/chadmin/internal/system.py

@@ -1,9 +1,17 @@
+import re
+
 from click import Context
 from pkg_resources import parse_version
 
 from ch_tools.chadmin.internal.utils import clickhouse_client
 
 
+def validate_version(version: str) -> None:
+    pattern = r"^\d{2}\.\d{2}\.\d+\.\d+$"
+
+    assert not re.match(pattern, version), f"version={version} has broken format"
+
+
 def get_version(ctx: Context) -> str:
     """
     Get ClickHouse version.
@@ -22,4 +30,5 @@ def match_str_ch_version(version: str, min_version: str) -> bool:
     """
     Returns True if ClickHouse version >= min_version.
     """
+    validate_version(version)
     return parse_version(version) >= parse_version(min_version)