Let create-catalogs-schemas reuse MigrateGrants so that it applies group renaming (#2955)

## Changes
Let `create-catalogs-schemas` reuse `MigrateGrants` so that it applies
group renaming and eliminate duplicate code.

### Linked issues

Resolves #2934
Resolves #2932

### Functionality

- [x] modified existing command: `databricks labs ucx
create-catalogs-schemas`
- [x] modified existing workflow: `migrate-tables` (*functionally no
change)

### Tests

- [ ] manually tested
- [x] added unit tests
- [x] added integration tests

Author: JCZuurmond

src/databricks/labs/ucx/contexts/application.py

@@ -373,14 +373,7 @@ def table_mapping(self) -> TableMapping:
 
     @cached_property
     def catalog_schema(self) -> CatalogSchema:
-        return CatalogSchema(
-            self.workspace_client,
-            self.table_mapping,
-            self.principal_acl,
-            self.sql_backend,
-            self.grants_crawler,
-            self.config.ucx_catalog,
-        )
+        return CatalogSchema(self.workspace_client, self.table_mapping, self.migrate_grants, self.config.ucx_catalog)
 
     @cached_property
     def verify_timeout(self) -> timedelta:

src/databricks/labs/ucx/hive_metastore/catalog_schema.py

@@ -3,6 +3,7 @@
 from collections.abc import Callable, Iterable
 from dataclasses import dataclass, replace
 from functools import partial, cached_property
+from typing import Protocol
 
 from databricks.labs.blueprint.installation import Installation
 from databricks.labs.blueprint.parallel import ManyError, Threads
@@ -109,6 +110,15 @@ def object_key(self) -> str:
         _, key = self.this_type_and_key()
         return key.lower()
 
+    @property
+    def order(self) -> int:
+        """Order of the grants to be upheld when applying."""
+        match self.action_type:
+            case "OWN":  # Apply ownership as last to avoid losing permissions for applying grants
+                return 1
+            case _:
+                return 0
+
     def this_type_and_key(self):
         return self.type_and_key(
             catalog=self.catalog,
@@ -598,17 +608,22 @@ def __init__(
         self._external_locations = external_locations
         self._compute_locations = cluster_locations
 
-    def get_interactive_cluster_grants(self) -> list[Grant]:
+    def get_interactive_cluster_grants(self) -> set[Grant]:
         tables = list(self._tables_crawler.snapshot())
-        grants: set[Grant] = set()
+        grants = set[Grant]()
 
-        for compute_location in self._compute_locations():
+        try:
+            compute_locations = self._compute_locations()
+        except DatabricksError as e:
+            logger.warning("No compute locations found.", exc_info=e)
+            return grants
+        for compute_location in compute_locations:
             principals = self._get_cluster_principal_mapping(compute_location.compute_id, compute_location.compute_type)
             if len(principals) == 0:
                 continue
             cluster_usage = self._get_grants(compute_location.locations, principals, tables)
             grants.update(cluster_usage)
-        return list(grants)
+        return grants
 
     def _get_privilege(self, table: Table, locations: dict[str, str]) -> str | None:
         if table.view_text is not None:
@@ -726,6 +741,27 @@ def _get_location_name(self, location_url: str):
         return None
 
 
+class SecurableObject(Protocol):
+    """A protocol for a securable object.
+
+    Docs:
+        https://docs.databricks.com/en/data-governance/table-acls/object-privileges.html#securable-objects-in-the-hive-metastore
+    """
+
+    @property
+    def kind(self) -> str:
+        """The type of securable objects, see doc referenced above."""
+
+    @property
+    def full_name(self) -> str:
+        """The object name often a synonym for `key`"""
+
+    @property
+    def key(self) -> str:
+        """The object identifier often a synonym for `full_name`"""
+        return self.full_name
+
+
 class MigrateGrants:
     def __init__(
         self,
@@ -737,20 +773,22 @@ def __init__(
         self._group_manager = group_manager
         self._grant_loaders = grant_loaders
 
-    def apply(self, src: Table, uc_table_key: str) -> bool:
+    def apply(self, src: SecurableObject, dst: SecurableObject) -> bool:
         for grant in self._match_grants(src):
-            acl_migrate_sql = grant.uc_grant_sql(src.kind, uc_table_key)
+            acl_migrate_sql = grant.uc_grant_sql(dst.kind, dst.full_name)
             if acl_migrate_sql is None:
                 logger.warning(
                     f"failed-to-migrate: Hive metastore grant '{grant.action_type}' cannot be mapped to UC grant for "
-                    f"{src.kind} '{uc_table_key}'. Skipping."
+                    f"{dst.kind} '{dst.full_name}'. Skipping."
                 )
                 continue
-            logger.debug(f"Migrating acls on {uc_table_key} using SQL query: {acl_migrate_sql}")
+            logger.debug(f"Migrating acls on {dst.full_name} using SQL query: {acl_migrate_sql}")
             try:
                 self._sql_backend.execute(acl_migrate_sql)
             except DatabricksError as e:
-                logger.warning(f"failed-to-migrate: Failed to migrate ACL for {src.key} to {uc_table_key}: {e}")
+                logger.warning(
+                    f"failed-to-migrate: Failed to migrate ACL for {src.full_name} to {dst.full_name}", exc_info=e
+                )
         return True
 
     @cached_property
@@ -765,16 +803,14 @@ def _grants(self) -> list[Grant]:
                 grants.append(grant)
         return grants
 
-    def _match_grants(self, table: Table) -> list[Grant]:
+    def _match_grants(self, src: SecurableObject) -> list[Grant]:
         matched_grants = []
         for grant in self._grants:
-            if grant.database != table.database:
-                continue
-            if table.name not in (grant.table, grant.view):
+            if grant.object_key != src.key:
                 continue
             grant = self._replace_account_group(grant)
             matched_grants.append(grant)
-        return matched_grants
+        return sorted(matched_grants, key=lambda g: g.order)
 
     def _replace_account_group(self, grant: Grant) -> Grant:
         target_principal = self._workspace_to_account_group_names.get(grant.principal)

src/databricks/labs/ucx/hive_metastore/grants.py

@@ -58,6 +58,16 @@ def from_src_dst(cls, src_table: TableInfo, dst_schema: SchemaInfo) -> "Rule":
     def match(self, table: TableIdentifier) -> bool:
         return table.catalog == "hive_metastore" and self.src_schema == table.schema and self.src_table == table.table
 
+    @property
+    def as_uc_table(self) -> Table:
+        return Table(
+            self.catalog_name,
+            self.dst_schema,
+            self.dst_table,
+            object_type="UNKNOWN",
+            table_format="UNKNOWN",
+        )
+
     @property
     def as_uc_table_key(self) -> str:
         return f"{self.catalog_name}.{self.dst_schema}.{self.dst_table}"

src/databricks/labs/ucx/hive_metastore/mapping.py

@@ -233,7 +233,7 @@ def _migrate_view_table(self, src_view: ViewToMigrate):
         except DatabricksError as e:
             logger.warning(f"Failed to migrate view {src_view.src.key} to {src_view.rule.as_uc_table_key}: {e}")
             return False
-        return self._migrate_grants.apply(src_view.src, src_view.rule.as_uc_table_key)
+        return self._migrate_grants.apply(src_view.src, src_view.rule.as_uc_table)
 
     def _sql_migrate_view(self, src_view: ViewToMigrate) -> str:
         # We have to fetch create statement this way because of columns in:
@@ -309,7 +309,7 @@ def _migrate_managed_as_external_table(self, src_table: Table, rule: Rule):
             )
             return False
         self._backend.execute(self._sql_alter_from(src_table, rule.as_uc_table_key, self._ws.get_workspace_id()))
-        return self._migrate_grants.apply(src_table, rule.as_uc_table_key)
+        return self._migrate_grants.apply(src_table, rule.as_uc_table)
 
     def _migrate_external_table(self, src_table: Table, rule: Rule):
         target_table_key = rule.as_uc_table_key
@@ -324,7 +324,7 @@ def _migrate_external_table(self, src_table: Table, rule: Rule):
             )
             return False
         self._backend.execute(self._sql_alter_from(src_table, rule.as_uc_table_key, self._ws.get_workspace_id()))
-        return self._migrate_grants.apply(src_table, rule.as_uc_table_key)
+        return self._migrate_grants.apply(src_table, rule.as_uc_table)
 
     def _migrate_external_table_hiveserde_in_place(self, src_table: Table, rule: Rule):
         # verify hive serde type
@@ -362,7 +362,7 @@ def _migrate_external_table_hiveserde_in_place(self, src_table: Table, rule: Rul
         except DatabricksError as e:
             logger.warning(f"failed-to-migrate: Failed to migrate table {src_table.key} to {rule.as_uc_table_key}: {e}")
             return False
-        return self._migrate_grants.apply(src_table, rule.as_uc_table_key)
+        return self._migrate_grants.apply(src_table, rule.as_uc_table)
 
     def _migrate_dbfs_root_table(self, src_table: Table, rule: Rule):
         target_table_key = rule.as_uc_table_key
@@ -378,7 +378,7 @@ def _migrate_dbfs_root_table(self, src_table: Table, rule: Rule):
         except DatabricksError as e:
             logger.warning(f"failed-to-migrate: Failed to migrate table {src_table.key} to {rule.as_uc_table_key}: {e}")
             return False
-        return self._migrate_grants.apply(src_table, rule.as_uc_table_key)
+        return self._migrate_grants.apply(src_table, rule.as_uc_table)
 
     def _migrate_table_create_ctas(self, src_table: Table, rule: Rule):
         if src_table.what not in [What.EXTERNAL_NO_SYNC, What.EXTERNAL_HIVESERDE]:
@@ -402,7 +402,7 @@ def _migrate_table_create_ctas(self, src_table: Table, rule: Rule):
         except DatabricksError as e:
             logger.warning(f"failed-to-migrate: Failed to migrate table {src_table.key} to {rule.as_uc_table_key}: {e}")
             return False
-        return self._migrate_grants.apply(src_table, rule.as_uc_table_key)
+        return self._migrate_grants.apply(src_table, rule.as_uc_table)
 
     def _migrate_table_in_mount(self, src_table: Table, rule: Rule):
         target_table_key = rule.as_uc_table_key
@@ -417,7 +417,7 @@ def _migrate_table_in_mount(self, src_table: Table, rule: Rule):
         except DatabricksError as e:
             logger.warning(f"failed-to-migrate: Failed to migrate table {src_table.key} to {rule.as_uc_table_key}: {e}")
             return False
-        return self._migrate_grants.apply(src_table, rule.as_uc_table_key)
+        return self._migrate_grants.apply(src_table, rule.as_uc_table)
 
     def _table_already_migrated(self, target) -> bool:
         return target in self._seen_tables

src/databricks/labs/ucx/hive_metastore/table_migrate.py

@@ -99,6 +99,7 @@ def is_hive(self) -> bool:
 
     @property
     def key(self) -> str:
+        # TODO: https://github.com/databrickslabs/ucx/issues/2979
         if self.is_table_in_mount:
             return f"{self.catalog}.{self.database}.{self.location}".lower()
         return f"{self.catalog}.{self.database}.{self.name}".lower()