Add ACL migration to migrate-tables workflow (#1135)

## Changes
- Add ACL migration strategy as a user configuration
- Pass this strategy to `migrate-tables` workflow

### Functionality 

- [x] modified existing workflow: `migrate-tables`

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

- [x] manually tested
- [ ] verified on staging environment (screenshot attached)

Author: nkvuong

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

@@ -80,25 +80,22 @@ def for_cli(cls, ws: WorkspaceClient, product='ucx'):
     def index(self):
         return self._migration_status_refresher.index()
 
-    def migrate_tables(self, *, what: What | None = None, acl_strategy: AclMigrationWhat | None = None):
+    def migrate_tables(self, *, what: What | None = None, acl_strategy: list[AclMigrationWhat] | None = None):
         self._init_seen_tables()
         tables_to_migrate = self._tm.get_tables_to_migrate(self._tc)
+        tasks = []
+        grants = []
         if acl_strategy is not None:
             grants_to_migrate = self._gc.snapshot()
             migrated_groups = self._group.snapshot()
-        tasks = []
+        else:
+            acl_strategy = []
         for table in tables_to_migrate:
             if what is not None and table.src.what != what:
                 continue
-            match acl_strategy:
-                case None:
-                    tasks.append(partial(self._migrate_table, table.src, table.rule))
-                case AclMigrationWhat.LEGACY_TACL:
-                    grants = self._match_grants(table.src, grants_to_migrate, migrated_groups)
-                    tasks.append(partial(self._migrate_table, table.src, table.rule, grants))
-                case AclMigrationWhat.PRINCIPAL:
-                    # TODO: Implement principal-based ACL migration
-                    pass
+            if AclMigrationWhat.LEGACY_TACL in acl_strategy:
+                grants.extend(self._match_grants(table.src, grants_to_migrate, migrated_groups))
+            tasks.append(partial(self._migrate_table, table.src, table.rule, grants))
         Threads.strict("migrate tables", tasks)
 
     def _migrate_table(self, src_table: Table, rule: Rule, grants: list[Grant] | None = None):

src/databricks/labs/ucx/runtime.py

@@ -25,7 +25,7 @@
     TablesMigrate,
 )
 from databricks.labs.ucx.hive_metastore.table_size import TableSizeCrawler
-from databricks.labs.ucx.hive_metastore.tables import What
+from databricks.labs.ucx.hive_metastore.tables import AclMigrationWhat, What
 from databricks.labs.ucx.hive_metastore.udfs import UdfsCrawler
 from databricks.labs.ucx.hive_metastore.verification import VerifyHasMetastore
 from databricks.labs.ucx.workspace_access.generic import WorkspaceListing
@@ -439,7 +439,7 @@ def migrate_external_tables_sync(
     group_manager = GroupManager(sql_backend, ws, cfg.inventory_database)
     TablesMigrate(
         table_crawler, grant_crawler, ws, sql_backend, table_mapping, group_manager, migration_status_refresher
-    ).migrate_tables(what=What.EXTERNAL_SYNC)
+    ).migrate_tables(what=What.EXTERNAL_SYNC, acl_strategy=[AclMigrationWhat.LEGACY_TACL])
 
 
 @task("migrate-tables", job_cluster="table_migration")
@@ -459,7 +459,7 @@ def migrate_dbfs_root_delta_tables(
     group_manager = GroupManager(sql_backend, ws, cfg.inventory_database)
     TablesMigrate(
         table_crawler, grant_crawler, ws, sql_backend, table_mapping, group_manager, migration_status_refresher
-    ).migrate_tables(what=What.DBFS_ROOT_DELTA)
+    ).migrate_tables(what=What.DBFS_ROOT_DELTA, acl_strategy=[AclMigrationWhat.LEGACY_TACL])
 
 
 @task("migrate-groups-experimental", depends_on=[crawl_groups])

tests/integration/hive_metastore/test_migrate.py

@@ -486,7 +486,7 @@ def test_migrate_managed_tables_with_acl(
         table_crawler, grant_crawler, ws, sql_backend, table_mapping, group_manager, migration_status_refresher
     )
 
-    table_migrate.migrate_tables(acl_strategy=AclMigrationWhat.LEGACY_TACL)
+    table_migrate.migrate_tables(acl_strategy=[AclMigrationWhat.LEGACY_TACL])
 
     target_tables = list(sql_backend.fetch(f"SHOW TABLES IN {dst_schema.full_name}"))
     assert len(target_tables) == 1

tests/unit/hive_metastore/test_table_migrate.py

@@ -628,7 +628,7 @@ def test_migrate_acls_should_produce_proper_queries(ws, caplog):
     table_migrate = TablesMigrate(
         table_crawler, grant_crawler, ws, backend, table_mapping, group_manager, migration_status_refresher
     )
-    table_migrate.migrate_tables(acl_strategy=AclMigrationWhat.LEGACY_TACL)
+    table_migrate.migrate_tables(acl_strategy=[AclMigrationWhat.LEGACY_TACL])
 
     assert "GRANT SELECT ON TABLE ucx_default.db1_dst.managed_dbfs TO `account group`" in backend.queries
     assert "GRANT MODIFY ON TABLE ucx_default.db1_dst.managed_mnt TO `account group`" in backend.queries