Fixed legacy table ACL ownership migration and other integration testing issues (#722)

Close #717

Author: nfx

pyproject.toml

@@ -132,7 +132,9 @@ ignore = [
     # Ignore complexity
     "C901", "PLR0911", "PLR0912", "PLR0913", "PLR0915",
     # Ignore flaky Import block is un-sorted or un-formatted
-    "I001"
+    "I001",
+    # Ignore Exception must not use a string literal, assign to variable first
+    "EM101",
 ]
 extend-exclude = [
     "notebooks/*.py"

src/databricks/labs/ucx/cli.py

@@ -96,8 +96,8 @@ def validate_external_locations():
     installation = installation_manager.for_user(ws.current_user.me())
     sql_backend = StatementExecutionBackend(ws, installation.config.warehouse_id)
     location_crawler = ExternalLocations(ws, sql_backend, installation.config.inventory_database)
-    path = location_crawler.save_as_terraform_definitions_on_workspace()
-    if len(path) > 0 and prompts.confirm(f"external_locations.tf file written to {path}. Do you want to open it?"):
+    path = location_crawler.save_as_terraform_definitions_on_workspace(installation.path)
+    if path and prompts.confirm(f"external_locations.tf file written to {path}. Do you want to open it?"):
         webbrowser.open(f"{ws.config.host}/#workspace{path}")
 
 

src/databricks/labs/ucx/framework/parallel.py

@@ -26,7 +26,7 @@ def __init__(self, errs):
 class Threads(Generic[Result]):
     def __init__(self, name, tasks: Sequence[Task[Result]], num_threads: int):
         self._name = name
-        self._tasks = tasks
+        self._tasks = list(tasks)
         self._task_fail_error_pct = 50
         self._num_threads = num_threads
         self._started = dt.datetime.now()
@@ -48,6 +48,12 @@ def gather(
                 num_threads = MIN_THREADS
         return cls(name, tasks, num_threads=num_threads)._run()
 
+    @classmethod
+    def strict(cls, name: str, tasks: Sequence[Task[Result]]):
+        _, errs = cls.gather(name, tasks)
+        if errs:
+            raise ManyError(errs)
+
     def _run(self) -> tuple[Iterable[Result], list[Exception]]:
         given_cnt = len(self._tasks)
         if given_cnt == 0:

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

@@ -73,13 +73,17 @@ def this_type_and_key(self):
             anonymous_function=self.anonymous_function,
         )
 
-    def hive_grant_sql(self) -> str:
+    def hive_grant_sql(self) -> list[str]:
         object_type, object_key = self.this_type_and_key()
         # See https://docs.databricks.com/en/sql/language-manual/security-grant.html
-        if self.action_type.upper() == "OWN":
-            return f"ALTER {object_type} {object_key} OWNER TO `{self.principal}`"
-        else:
-            return f"GRANT {self.action_type} ON {object_type} {object_key} TO `{self.principal}`"
+        statements = []
+        actions = self.action_type.split(", ")
+        if "OWN" in actions:
+            actions.remove("OWN")
+            statements.append(f"ALTER {object_type} {object_key} OWNER TO `{self.principal}`")
+        if actions:
+            statements.append(f"GRANT {', '.join(actions)} ON {object_type} {object_key} TO `{self.principal}`")
+        return statements
 
     def hive_revoke_sql(self) -> str:
         object_type, object_key = self.this_type_and_key()

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

@@ -33,7 +33,6 @@ class ExternalLocations(CrawlerBase[ExternalLocation]):
     def __init__(self, ws: WorkspaceClient, sbe: SqlBackend, schema):
         super().__init__(sbe, "hive_metastore", schema, "external_locations", ExternalLocation)
         self._ws = ws
-        self._folder = f"/Users/{ws.current_user.me().user_name}/.ucx"
 
     def _external_locations(self, tables: list[Row], mounts) -> Iterable[ExternalLocation]:
         min_slash = 2
@@ -173,9 +172,7 @@ def _match_table_external_locations(self) -> tuple[list[list], list[ExternalLoca
             missing_locations.append(loc)
         return matching_locations, missing_locations
 
-    def save_as_terraform_definitions_on_workspace(self, folder: str | None = None) -> str:
-        if folder:
-            self._folder = folder
+    def save_as_terraform_definitions_on_workspace(self, folder: str) -> str | None:
         matching_locations, missing_locations = self._match_table_external_locations()
         if len(matching_locations) > 0:
             logger.info("following external locations are already configured.")
@@ -190,13 +187,12 @@ def save_as_terraform_definitions_on_workspace(self, folder: str | None = None)
             for script in self._get_ext_location_definitions(missing_locations):
                 buffer.write(script)
             buffer.seek(0)
-            return self._overwrite_mapping(buffer)
-        else:
-            logger.info("no additional external location to be created.")
-            return ""
+            return self._overwrite_mapping(folder, buffer)
+        logger.info("no additional external location to be created.")
+        return None
 
-    def _overwrite_mapping(self, buffer) -> str:
-        path = f"{self._folder}/external_locations.tf"
+    def _overwrite_mapping(self, folder, buffer) -> str:
+        path = f"{folder}/external_locations.tf"
         self._ws.workspace.upload(path, buffer, overwrite=True, format=ImportFormat.AUTO)
         return path