Use SchemaDeployer to deploy schema

Author: JCZuurmond

src/databricks/labs/ucx/progress/install.py

@@ -1,40 +1,21 @@
-import datetime as dt
 import logging
 
-from databricks.labs.lsql.backends import Dataclass, SqlBackend
-from databricks.sdk.errors import InternalError
-from databricks.sdk.retries import retried
-
-from databricks.labs.ucx.framework.utils import escape_sql_identifier
+from databricks.labs.lsql.backends import SqlBackend
+from databricks.labs.lsql.deployment import SchemaDeployer
 
 
 logger = logging.getLogger(__name__)
 
 
 class HistoryInstallation:
-    """Install resources for UCX's artifacts history.
-
-    `InternalError` are retried on create statements for resilience on sporadic Databricks issues.
-    """
+    """Install resources for UCX's artifacts history."""
 
     _SCHEMA = "history"
 
     def __init__(self, sql_backend: SqlBackend, ucx_catalog: str) -> None:
-        self._backend = sql_backend
-        self._ucx_catalog = ucx_catalog
+        # `mod` is required parameter, but it's not used in this context.
+        self._schema_deployer = SchemaDeployer(sql_backend, self._SCHEMA, mod=None, catalog=ucx_catalog)
 
     def run(self) -> None:
-        self._create_schema()
+        self._schema_deployer.deploy_schema()
         logger.info("Installation completed successfully!")
-
-    @retried(on=[InternalError], timeout=dt.timedelta(minutes=1))
-    def _create_schema(self) -> None:
-        schema = f"{self._ucx_catalog}.{self._SCHEMA}"
-        logger.info(f"Creating {schema} database...")
-        self._backend.execute(f"CREATE SCHEMA IF NOT EXISTS {escape_sql_identifier(schema, maxsplit=1)}")
-
-    @retried(on=[InternalError], timeout=dt.timedelta(minutes=1))
-    def _create_table(self, name: str, klass: Dataclass) -> None:
-        full_name = f"{self._ucx_catalog}.{self._SCHEMA}.{name}"
-        logger.info(f"Create {full_name} table ...")
-        self._backend.create_table(escape_sql_identifier(full_name), klass)

tests/unit/progress/test_install.py

@@ -4,4 +4,4 @@
 def test_history_installation_run_creates_history_schema(mock_backend) -> None:
     installation = HistoryInstallation(mock_backend, "ucx")
     installation.run()
-    assert mock_backend.queries[0] == "CREATE SCHEMA IF NOT EXISTS `ucx`.`history`"
+    assert mock_backend.queries[0] == "CREATE SCHEMA IF NOT EXISTS ucx.history"