fix(backend): backtick-quote hyphenated catalog names in SQL queries

src/back/core/databricks/SQLWarehouse.py

@@ -6,6 +6,7 @@
 """
 
 import queue
+import re
 import threading
 import time
 from contextlib import contextmanager
@@ -96,7 +97,91 @@ def _borrow(self):
                 self._pool.put_nowait(conn)
             except queue.Full:
                 self._close_quietly(conn)
+    @staticmethod
+    def _quote_hyphenated_uc_identifiers(query: str) -> str:
+        """
+        Databricks SQL requires identifiers with hyphens to be backticked.
+
+        Example:
+            arc-dbx-uc-cdp.causal_analytics.table_name
+
+        becomes:
+            `arc-dbx-uc-cdp`.causal_analytics.table_name
+
+        This only modifies SQL outside string literals and outside existing
+        backtick identifiers, so URI strings and already-quoted identifiers
+        are not touched.
+        """
+        if not isinstance(query, str) or "-" not in query:
+            return query
+
+        # Matches unquoted identifier parts containing hyphen only when used
+        # like a SQL multipart identifier before a dot.
+        pattern = re.compile(
+            r"(?<![`A-Za-z0-9_])"
+            r"([A-Za-z_][A-Za-z0-9_]*(?:-[A-Za-z0-9_]+)+)"
+            r"(?=\.)"
+        )
 
+        output = []
+        chunk = []
+
+        in_single_quote = False
+        in_double_quote = False
+        in_backtick = False
+        i = 0
+
+        def flush_chunk():
+            if chunk:
+                text = "".join(chunk)
+                output.append(pattern.sub(r"`\1`", text))
+                chunk.clear()
+
+        while i < len(query):
+            ch = query[i]
+
+            if ch == "'" and not in_double_quote and not in_backtick:
+                if not in_single_quote:
+                    flush_chunk()
+                    output.append(ch)
+                    in_single_quote = True
+                else:
+                    output.append(ch)
+                    in_single_quote = False
+                i += 1
+                continue
+
+            if ch == '"' and not in_single_quote and not in_backtick:
+                if not in_double_quote:
+                    flush_chunk()
+                    output.append(ch)
+                    in_double_quote = True
+                else:
+                    output.append(ch)
+                    in_double_quote = False
+                i += 1
+                continue
+
+            if ch == "`" and not in_single_quote and not in_double_quote:
+                if not in_backtick:
+                    flush_chunk()
+                    output.append(ch)
+                    in_backtick = True
+                else:
+                    output.append(ch)
+                    in_backtick = False
+                i += 1
+                continue
+
+            if in_single_quote or in_double_quote or in_backtick:
+                output.append(ch)
+            else:
+                chunk.append(ch)
+
+            i += 1
+
+        flush_chunk()
+        return "".join(output)
     @staticmethod
     def _close_quietly(pc: _PooledConnection) -> None:
         try:
@@ -138,6 +223,7 @@ def execute_query(self, query: str) -> List[Dict[str, Any]]:
         try:
             with self._borrow() as conn:
                 with conn.cursor() as cur:
+                    query = self._quote_hyphenated_uc_identifiers(query)
                     cur.execute(query)
                     columns = [desc[0] for desc in cur.description]
                     return [dict(zip(columns, row)) for row in cur.fetchall()]
@@ -165,6 +251,7 @@ def iter_rows(
         try:
             with self._borrow() as conn:
                 with conn.cursor() as cur:
+                    query = self._quote_hyphenated_uc_identifiers(query)
                     cur.execute(query)
                     columns = [desc[0] for desc in cur.description]
                     while True:
@@ -183,6 +270,7 @@ def execute_statement(self, statement: str) -> bool:
         try:
             with self._borrow() as conn:
                 with conn.cursor() as cur:
+                    statement = self._quote_hyphenated_uc_identifiers(statement)
                     cur.execute(statement)
                 # UC DDL must be committed before control-plane APIs (e.g. synced
                 # database tables) can resolve catalog.schema in the metastore.

src/back/core/databricks/UnityCatalog.py

@@ -59,7 +59,7 @@ def get_schemas(self, catalog: str) -> List[str]:
             params = self._auth.get_sql_connection_params()
             with sql.connect(**params) as conn:
                 with conn.cursor() as cur:
-                    cur.execute(f"SHOW SCHEMAS IN {catalog}")
+                    cur.execute(f"SHOW SCHEMAS IN `{catalog}`")
                     return [row[0] for row in cur.fetchall()]
         except Exception as exc:
             logger.exception("Error fetching schemas: %s", exc)
@@ -75,7 +75,7 @@ def get_tables(self, catalog: str, schema: str) -> List[str]:
             params = self._auth.get_sql_connection_params()
             with sql.connect(**params) as conn:
                 with conn.cursor() as cur:
-                    cur.execute(f"SHOW TABLES IN {catalog}.{schema}")
+                    cur.execute(f"SHOW TABLES IN `{catalog}`.{schema}")
                     return [row[1] for row in cur.fetchall()]
         except Exception as exc:
             logger.exception("Error fetching tables: %s", exc)
@@ -138,7 +138,7 @@ def get_table_columns(
             params = self._auth.get_sql_connection_params()
             with sql.connect(**params) as conn:
                 with conn.cursor() as cur:
-                    cur.execute(f"DESCRIBE {catalog}.{schema}.{table}")
+                    cur.execute(f"DESCRIBE `{catalog}`.{schema}.{table}")
                     columns = []
                     for row in cur.fetchall():
                         columns.append(
@@ -160,7 +160,7 @@ def get_table_comment(self, catalog: str, schema: str, table: str) -> str:
             with sql.connect(**params) as conn:
                 with conn.cursor() as cur:
                     query = (
-                        f"SELECT comment FROM {catalog}.information_schema.tables "
+                        f"SELECT comment FROM `{catalog}`.information_schema.tables "
                         f"WHERE table_catalog = '{catalog}' "
                         f"AND table_schema = '{schema}' "
                         f"AND table_name = '{table}'"
@@ -179,7 +179,7 @@ def get_volumes(self, catalog: str, schema: str) -> List[str]:
             params = self._auth.get_sql_connection_params()
             with sql.connect(**params) as conn:
                 with conn.cursor() as cur:
-                    cur.execute(f"SHOW VOLUMES IN {catalog}.{schema}")
+                    cur.execute(f"SHOW VOLUMES IN `{catalog}`.{schema}")
                     return [row[1] for row in cur.fetchall()]
         except Exception as exc:
             logger.exception("Error fetching volumes: %s", exc)