Enables cli cmd databricks labs ucx create-catalog-schemas to apply catalog/schema acl from legacy hive_metastore (#2676)

<!-- REMOVE IRRELEVANT COMMENTS BEFORE CREATING A PULL REQUEST -->
## Changes
<!-- Summary of your changes that are easy to understand. Add
screenshots when necessary -->

### Linked issues
<!-- DOC: Link issue with a keyword: close, closes, closed, fix, fixes,
fixed, resolve, resolves, resolved. See
https://docs.github.com/en/issues/tracking-your-work-with-issues/linking-a-pull-request-to-an-issue#linking-a-pull-request-to-an-issue-using-a-keyword
-->

Resolves #2514 

### Functionality

- [ ] modified existing command: `databricks labs ucx
create-catalog-schemas`

### Tests
<!-- How is this tested? Please see the checklist below and also
describe any other relevant tests -->

- [ ] added unit tests
- [ ] added integration tests

Author: HariGS-DB

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

@@ -348,7 +348,13 @@ def table_mapping(self):
 
     @cached_property
     def catalog_schema(self):
-        return CatalogSchema(self.workspace_client, self.table_mapping, self.principal_acl, self.sql_backend)
+        return CatalogSchema(
+            self.workspace_client,
+            self.table_mapping,
+            self.principal_acl,
+            self.sql_backend,
+            self.grants_crawler,
+        )
 
     @cached_property
     def verify_timeout(self):

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

@@ -6,7 +6,7 @@
 
 from databricks.labs.blueprint.tui import Prompts
 from databricks.labs.lsql.backends import SqlBackend
-from databricks.labs.ucx.hive_metastore.grants import PrincipalACL, Grant
+from databricks.labs.ucx.hive_metastore.grants import PrincipalACL, Grant, GrantsCrawler
 from databricks.sdk import WorkspaceClient
 from databricks.sdk.errors import NotFound
 from databricks.sdk.service.catalog import SchemaInfo
@@ -19,13 +19,19 @@
 
 class CatalogSchema:
     def __init__(
-        self, ws: WorkspaceClient, table_mapping: TableMapping, principal_grants: PrincipalACL, sql_backend: SqlBackend
+        self,
+        ws: WorkspaceClient,
+        table_mapping: TableMapping,
+        principal_grants: PrincipalACL,
+        sql_backend: SqlBackend,
+        grants_crawler: GrantsCrawler,
     ):
         self._ws = ws
         self._table_mapping = table_mapping
         self._external_locations = self._ws.external_locations.list()
         self._principal_grants = principal_grants
         self._backend = sql_backend
+        self._hive_grants_crawler = grants_crawler
 
     def create_all_catalogs_schemas(self, prompts: Prompts) -> None:
         candidate_catalogs, candidate_schemas = self._get_missing_catalogs_schemas()
@@ -46,10 +52,22 @@ def create_all_catalogs_schemas(self, prompts: Prompts) -> None:
                             f"Schema {candidate_schema} in catalog {candidate_catalog} " f"already exists. Skipping."
                         )
                         continue
+        self._apply_from_legacy_table_acls()
         self._update_principal_acl()
 
+    def _apply_from_legacy_table_acls(self):
+        grants = self._get_catalog_schema_hive_grants()
+        for grant in grants:
+            acl_migrate_sql = grant.uc_grant_sql()
+            if acl_migrate_sql is None:
+                logger.warning(f"Cannot identify UC grant for {grant.this_type_and_key()}. Skipping.")
+                continue
+            logger.debug(f"Migrating acls on {grant.this_type_and_key()} using SQL query: {acl_migrate_sql}")
+            self._backend.execute(acl_migrate_sql)
+
     def _update_principal_acl(self):
-        grants = self._get_catalog_schema_grants()
+
+        grants = self._get_catalog_schema_principal_acl_grants()
         for grant in grants:
             acl_migrate_sql = grant.uc_grant_sql()
             if acl_migrate_sql is None:
@@ -58,7 +76,21 @@ def _update_principal_acl(self):
             logger.debug(f"Migrating acls on {grant.this_type_and_key()} using SQL query: {acl_migrate_sql}")
             self._backend.execute(acl_migrate_sql)
 
-    def _get_catalog_schema_grants(self) -> list[Grant]:
+    def _get_catalog_schema_hive_grants(self) -> list[Grant]:
+        src_dst_schema_mapping = self._get_database_source_target_mapping()
+        hive_grants = self._hive_grants_crawler.snapshot()
+        new_grants: list[Grant] = []
+        for grant in hive_grants:
+            if grant.this_type_and_key()[0] == "DATABASE" and grant.database:
+                for schema in src_dst_schema_mapping[grant.database]:
+                    new_grants.append(replace(grant, catalog=schema.catalog_name, database=schema.name))
+        catalog_grants: set[Grant] = set()
+        for grant in new_grants:
+            catalog_grants.add(replace(grant, database=None))
+        new_grants.extend(catalog_grants)
+        return new_grants
+
+    def _get_catalog_schema_principal_acl_grants(self) -> list[Grant]:
         src_trg_schema_mapping = self._get_database_source_target_mapping()
         grants = self._principal_grants.get_interactive_cluster_grants()
         # filter on grants to only get database level grants

tests/unit/hive_metastore/test_catalog_schema.py

@@ -9,7 +9,7 @@
 from databricks.sdk.service.catalog import CatalogInfo, ExternalLocationInfo, SchemaInfo
 
 from databricks.labs.ucx.hive_metastore.catalog_schema import CatalogSchema
-from databricks.labs.ucx.hive_metastore.grants import PrincipalACL, Grant
+from databricks.labs.ucx.hive_metastore.grants import PrincipalACL, Grant, GrantsCrawler
 from databricks.labs.ucx.hive_metastore.mapping import TableMapping
 
 
@@ -83,15 +83,35 @@ def prepare_test(ws, backend: MockBackend | None = None) -> CatalogSchema:
     )
     table_mapping = TableMapping(installation, ws, backend)
     principal_acl = create_autospec(PrincipalACL)
+    hive_acl = create_autospec(GrantsCrawler)
     grants = [
         Grant('user1', 'SELECT', 'catalog1', 'schema3', 'table'),
         Grant('user1', 'MODIFY', 'catalog2', 'schema2', 'table'),
         Grant('user1', 'SELECT', 'catalog2', 'schema3', 'table2'),
         Grant('user1', 'USAGE', 'hive_metastore', 'schema3'),
         Grant('user1', 'USAGE', 'hive_metastore', 'schema2'),
     ]
+    hive_grants = [
+        Grant(principal="princ1", catalog="hive_metastore", action_type="USE"),
+        Grant(principal="princ2", catalog="hive_metastore", database="schema3", action_type="USAGE"),
+        Grant(
+            principal="princ33",
+            catalog="hive_metastore",
+            database="database_one",
+            view="table_one",
+            action_type="SELECT",
+        ),
+        Grant(
+            principal="princ5",
+            catalog="hive_metastore",
+            database="schema2",
+            action_type="USAGE",
+        ),
+    ]
     principal_acl.get_interactive_cluster_grants.return_value = grants
-    return CatalogSchema(ws, table_mapping, principal_acl, backend)
+    hive_acl.snapshot.return_value = hive_grants
+
+    return CatalogSchema(ws, table_mapping, principal_acl, backend, hive_acl)
 
 
 @pytest.mark.parametrize("location", ["s3://foo/bar", "s3://foo/bar/test", "s3://foo/bar/test/baz"])
@@ -171,6 +191,11 @@ def test_catalog_schema_acl():
         'GRANT USE SCHEMA ON DATABASE `catalog2`.`schema3` TO `user1`',
         'GRANT USE CATALOG ON CATALOG `catalog1` TO `user1`',
         'GRANT USE CATALOG ON CATALOG `catalog2` TO `user1`',
+        'GRANT USE CATALOG ON CATALOG `catalog1` TO `princ2`',
+        'GRANT USE SCHEMA ON DATABASE `catalog1`.`schema3` TO `princ2`',
+        'GRANT USE SCHEMA ON DATABASE `catalog2`.`schema2` TO `princ5`',
+        'GRANT USE SCHEMA ON DATABASE `catalog2`.`schema3` TO `princ5`',
+        'GRANT USE CATALOG ON CATALOG `catalog2` TO `princ5`',
     ]
     assert len(backend.queries) == len(queries)
     for query in queries: