Store code migration progress in multiworkspace history log and visualize in dashboard

src/databricks/labs/ucx/assessment/workflows.py

@@ -189,7 +189,7 @@ def assess_dashboards(self, ctx: RuntimeContext):
 
         Also, stores direct filesystem accesses for display in the migration dashboard.
         """
-        ctx.query_linter.refresh_report(ctx.sql_backend, ctx.inventory_database)
+        ctx.query_linter.refresh_report()
 
     @job_task
     def assess_workflows(self, ctx: RuntimeContext):

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

@@ -44,8 +44,9 @@
 )
 from databricks.labs.ucx.hive_metastore.mapping import TableMapping
 from databricks.labs.ucx.hive_metastore.table_migration_status import TableMigrationIndex
-from databricks.labs.ucx.hive_metastore.ownership import TableMigrationOwnership, TableOwnership
+from databricks.labs.ucx.hive_metastore.table_ownership import TableOwnership, UsedTableOwnership
 from databricks.labs.ucx.hive_metastore.table_migrate import (
+    TableMigrationOwnership,
     TableMigrationStatusRefresher,
     TablesMigrator,
 )
@@ -63,7 +64,7 @@
     NotebookLoader,
 )
 from databricks.labs.ucx.source_code.path_lookup import PathLookup
-from databricks.labs.ucx.source_code.queries import QueryLinter
+from databricks.labs.ucx.source_code.queries import QueryLinter, QueryProblemOwnership
 from databricks.labs.ucx.source_code.redash import Redash
 from databricks.labs.ucx.workspace_access import generic, redash
 from databricks.labs.ucx.workspace_access.groups import GroupManager
@@ -262,16 +263,23 @@ def tables_crawler(self) -> TablesCrawler:
         return TablesCrawler(self.sql_backend, self.inventory_database, self.config.include_databases)
 
     @cached_property
-    def table_ownership(self) -> TableOwnership:
-        return TableOwnership(
+    def used_table_ownership(self) -> UsedTableOwnership:
+        return UsedTableOwnership(
             self.administrator_locator,
-            self.grants_crawler,
             self.used_tables_crawler_for_paths,
             self.used_tables_crawler_for_queries,
             self.legacy_query_ownership,
             self.workspace_path_ownership,
         )
 
+    @cached_property
+    def table_ownership(self) -> TableOwnership:
+        return TableOwnership(
+            self.administrator_locator,
+            self.grants_crawler,
+            self.used_table_ownership,
+        )
+
     @cached_property
     def workspace_path_ownership(self) -> WorkspacePathOwnership:
         return WorkspacePathOwnership(self.administrator_locator, self.workspace_client)
@@ -281,7 +289,11 @@ def legacy_query_ownership(self) -> LegacyQueryOwnership:
         return LegacyQueryOwnership(self.administrator_locator, self.workspace_client)
 
     @cached_property
-    def directfs_access_ownership(self) -> DirectFsAccessOwnership:
+    def query_problem_ownership(self) -> QueryProblemOwnership:
+        return QueryProblemOwnership(self.administrator_locator, self.workspace_client)
+
+    @cached_property
+    def direct_filesystem_access_ownership(self) -> DirectFsAccessOwnership:
         return DirectFsAccessOwnership(
             self.administrator_locator,
             self.workspace_path_ownership,
@@ -511,6 +523,8 @@ def workflow_linter(self) -> WorkflowLinter:
     def query_linter(self) -> QueryLinter:
         return QueryLinter(
             self.workspace_client,
+            self.sql_backend,
+            self.inventory_database,
             TableMigrationIndex([]),  # TODO: bring back self.tables_migrator.index()
             self.directfs_access_crawler_for_queries,
             self.used_tables_crawler_for_queries,

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

@@ -24,11 +24,13 @@
 from databricks.labs.ucx.hive_metastore.tables import FasterTableScanCrawler
 from databricks.labs.ucx.hive_metastore.udfs import Udf
 from databricks.labs.ucx.installer.logs import TaskRunWarningRecorder
+from databricks.labs.ucx.progress.directfs_access import DirectFsAccessProgressEncoder
 from databricks.labs.ucx.progress.grants import GrantProgressEncoder
 from databricks.labs.ucx.progress.history import ProgressEncoder
 from databricks.labs.ucx.progress.jobs import JobsProgressEncoder
-from databricks.labs.ucx.progress.tables import TableProgressEncoder
+from databricks.labs.ucx.progress.tables import TableProgressEncoder, UsedTableProgressEncoder
 from databricks.labs.ucx.progress.workflow_runs import WorkflowRunRecorder
+from databricks.labs.ucx.progress.queries import QueryProblemProgressEncoder
 
 # As with GlobalContext, service factories unavoidably have a lot of public methods.
 # pylint: disable=too-many-public-methods
@@ -240,3 +242,33 @@ def udfs_progress(self) -> ProgressEncoder[Udf]:
             self.workspace_id,
             self.config.ucx_catalog,
         )
+
+    @cached_property
+    def query_problem_progress(self) -> QueryProblemProgressEncoder:
+        return QueryProblemProgressEncoder(
+            self.sql_backend,
+            self.query_problem_ownership,
+            self.parent_run_id,
+            self.workspace_id,
+            self.config.ucx_catalog,
+        )
+
+    @cached_property
+    def direct_filesystem_access_progress(self) -> DirectFsAccessProgressEncoder:
+        return DirectFsAccessProgressEncoder(
+            self.sql_backend,
+            self.direct_filesystem_access_ownership,
+            self.parent_run_id,
+            self.workspace_id,
+            self.config.ucx_catalog,
+        )
+
+    @cached_property
+    def used_table_progress(self) -> UsedTableProgressEncoder:
+        return UsedTableProgressEncoder(
+            self.sql_backend,
+            self.used_table_ownership,
+            self.parent_run_id,
+            self.workspace_id,
+            self.config.ucx_catalog,
+        )

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

@@ -12,6 +12,7 @@
 from databricks.sdk.service.iam import User, PermissionLevel
 from databricks.sdk.service.workspace import ObjectType
 
+
 logger = logging.getLogger(__name__)
 
 
@@ -242,16 +243,34 @@ def _infer_from_first_can_manage(object_permissions):
         return None
 
 
-class LegacyQueryOwnership(Ownership[str]):
-    def __init__(self, administrator_locator: AdministratorLocator, workspace_client: WorkspaceClient) -> None:
-        super().__init__(administrator_locator)
-        self._workspace_client = workspace_client
+class LegacyQueryOwnershipMixin:
+    """Retrieve ownership of legacy queries.
 
-    def _maybe_direct_owner(self, record: str) -> str | None:
+    This ownership class is different from most ownership implementations as it fetches the query from the workspace,
+    where most other classes expect an object that contains the ownership thus not requiring workspace access.
+
+    A mixin is introduced to get query ownership for both plain query ids and class:QueryProblem, while maintaining
+    the type hinting and reducing risk by handling the call to the workspace in **one** place (in case another exception
+    needs to be caught later).
+    """
+
+    @staticmethod
+    def _maybe_direct_owner_from_query_id(ws: WorkspaceClient, query_id: str) -> str | None:
         try:
-            legacy_query = self._workspace_client.queries.get(record)
+            legacy_query = ws.queries.get(query_id)
             return legacy_query.owner_user_name
         except NotFound:
             return None
         except InternalError:  # redash is very naughty and throws 500s instead of proper 404s
             return None
+
+
+class LegacyQueryOwnership(Ownership[str], LegacyQueryOwnershipMixin):
+    """Query ownership given a query id"""
+
+    def __init__(self, administrator_locator: AdministratorLocator, workspace_client: WorkspaceClient) -> None:
+        super().__init__(administrator_locator)
+        self._workspace_client = workspace_client
+
+    def _maybe_direct_owner(self, record: str) -> str | None:
+        return self._maybe_direct_owner_from_query_id(self._workspace_client, record)

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

@@ -13,8 +13,7 @@
 
 from databricks.labs.ucx.account.workspaces import WorkspaceInfo
 from databricks.labs.ucx.framework.utils import escape_sql_identifier
-from databricks.labs.ucx.hive_metastore import TablesCrawler
-from databricks.labs.ucx.hive_metastore.tables import Table
+from databricks.labs.ucx.hive_metastore.tables import Table, TablesCrawler
 from databricks.labs.ucx.recon.base import TableIdentifier
 
 logger = logging.getLogger(__name__)

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

@@ -9,6 +9,7 @@
 from databricks.sdk import WorkspaceClient
 from databricks.sdk.errors.platform import DatabricksError
 
+from databricks.labs.ucx.framework.owners import Ownership
 from databricks.labs.ucx.framework.utils import escape_sql_identifier
 from databricks.labs.ucx.hive_metastore import TablesCrawler
 from databricks.labs.ucx.hive_metastore.grants import MigrateGrants
@@ -18,8 +19,11 @@
     TableMapping,
     TableToMigrate,
 )
-
-from databricks.labs.ucx.hive_metastore.table_migration_status import TableMigrationStatusRefresher
+from databricks.labs.ucx.hive_metastore.table_migration_status import (
+    TableMigrationStatusRefresher,
+    TableMigrationStatus,
+)
+from databricks.labs.ucx.hive_metastore.table_ownership import TableOwnership
 from databricks.labs.ucx.hive_metastore.tables import (
     MigrationCount,
     Table,
@@ -596,3 +600,29 @@ def _sql_alter_from(self, table: Table, target_table_key: str, ws_id: int):
     def _is_migrated(self, schema: str, table: str) -> bool:
         index = self._migration_status_refresher.index()
         return index.is_migrated(schema, table)
+
+
+class TableMigrationOwnership(Ownership[TableMigrationStatus]):
+    """Determine ownership of table migration records in the inventory.
+
+    This is the owner of the source table, if (and only if) the source table is present in the inventory.
+    """
+
+    def __init__(self, tables_crawler: TablesCrawler, table_ownership: TableOwnership) -> None:
+        super().__init__(table_ownership._administrator_locator)  # TODO: Fix this
+        self._tables_crawler = tables_crawler
+        self._table_ownership = table_ownership
+        self._indexed_tables: dict[tuple[str, str], Table] | None = None
+
+    def _tables_snapshot_index(self, reindex: bool = False) -> dict[tuple[str, str], Table]:
+        index = self._indexed_tables
+        if index is None or reindex:
+            snapshot = self._tables_crawler.snapshot()
+            index = {(table.database, table.name): table for table in snapshot}
+            self._indexed_tables = index
+        return index
+
+    def _maybe_direct_owner(self, record: TableMigrationStatus) -> str | None:
+        index = self._tables_snapshot_index()
+        source_table = index.get((record.src_schema, record.src_table), None)
+        return self._table_ownership.owner_of(source_table) if source_table is not None else None