Added escaping for column name when backup is created

ch_backup/clickhouse/control.py

@@ -524,6 +524,47 @@
 """
 )
 
+UNESCAPED_KEYWORDS = [
+    "TOP",
+]
+
+
+def _fix_create_statement(create_statement: str) -> str:
+    """
+    Fix DDL that was stored by ClickHouse without properly escaping reserved
+    keywords used as identifiers (column names, aliases).
+
+    This is a known ClickHouse bug where keywords like TOP are not backtick-escaped
+    during DDL serialization to .sql files, making stored DDL unparsable on ATTACH.
+
+    Example of broken DDL:
+        SELECT TOP AS c FROM default.test_3
+        -- TOP is treated as TOP N syntax
+
+    Fixed DDL:
+        SELECT `TOP` AS c FROM default.test_3
+    """
+
+    fixed = create_statement
+    for keyword in UNESCAPED_KEYWORDS:
+        fixed = re.sub(
+            rf"(?<![`\w]){keyword}(?![`\w])\s+AS\s",
+            f"`{keyword}` AS ",
+            fixed,
+            flags=re.IGNORECASE,
+        )
+
+    if fixed != create_statement:
+        logging.warning(
+            "DDL was modified to fix unescaped reserved keywords. "
+            "This is likely a ClickHouse bug. "
+            "Original: {!r}. Fixed: {!r}",
+            create_statement,
+            fixed,
+        )
+
+    return fixed
+
 
 # pylint: disable=too-many-public-methods
 class ClickhouseCTL:
@@ -959,7 +1000,8 @@ def create_table(self, table: Table) -> None:
         """
         Restore table.
         """
-        self._ch_client.query(table.create_statement)
+        fixed_statement = _fix_create_statement(table.create_statement)
+        self._ch_client.query(fixed_statement)
 
     def restore_replica(self, table: Table) -> None:
         """

tests/unit/test_control.py

@@ -1,6 +1,10 @@
 import pytest
 
-from ch_backup.clickhouse.control import _format_string_array, _parse_version
+from ch_backup.clickhouse.control import (
+    _fix_create_statement,
+    _format_string_array,
+    _parse_version,
+)
 from tests.unit.utils import parametrize
 
 
@@ -49,3 +53,25 @@ def test_format_string_array(value, result):
 )
 def test_parse_version(version: str, expected: list[int]) -> None:
     assert _parse_version(version) == expected
+
+
+def test_fix_create_statement_top_keyword():
+    broken = (
+        "ATTACH VIEW test_4 UUID '1c465905-164d-400b-8a39-ff382c92fb81' "
+        "(c String) AS WITH A AS (SELECT TOP AS c FROM default.test_3) "
+        "SELECT c FROM A"
+    )
+    fixed = _fix_create_statement(broken)
+
+    assert "`TOP`" in fixed
+    assert "SELECT `TOP` AS c" in fixed
+
+
+def test_fix_create_statement_no_change_when_valid():
+    valid = (
+        "ATTACH VIEW test_4 UUID '1c465905-164d-400b-8a39-ff382c92fb81' "
+        "(c String) AS SELECT c FROM default.test_3"
+    )
+    fixed = _fix_create_statement(valid)
+
+    assert fixed == valid