Repository: databrickslabs/ucx

Issue:
Migrate Tables Fail if default catalog is set for the workspace
### Is there an existing issue for this?

- [x] I have searched the existing issues

### Current Behavior

If a workspace default catalog is defined the table migration fails when it tries to describe schema. We don't specify "hive_metastore" catalog in our queries.

### Expected Behavior

Add hive_metastore qualifier to all queries.

### Steps To Reproduce

Set a default catalog for the workspace
Run migrate tables workflow

### Cloud

AWS

### Operating System

macOS

### Version

latest via Databricks CLI

### Relevant log output

```shell

```

Discussion:

Solution patch:
diff --git a/src/databricks/labs/ucx/assessment/jobs.py b/src/databricks/labs/ucx/assessment/jobs.py
index a0dcd4c9..489da76e 100644
--- a/src/databricks/labs/ucx/assessment/jobs.py
+++ b/src/databricks/labs/ucx/assessment/jobs.py
@@ -42,6 +42,8 @@ class JobInfo:
     job_name: str | None = None
     creator: str | None = None
     """User-name of the creator of the pipeline, if known."""
+    last_run: int | None = None
+    """Timestamp of the last run of the pipeline, if known."""
 
     __id_attributes__: ClassVar[tuple[str, ...]] = ("job_id",)
 
@@ -54,6 +56,7 @@ class JobInfo:
             failures="[]",
             job_name=job_name,
             creator=job.creator_user_name or None,
+            last_run=None,
         )
 
 
diff --git a/src/databricks/labs/ucx/hive_metastore/mapping.py b/src/databricks/labs/ucx/hive_metastore/mapping.py
index 78237ea5..c87e7406 100644
--- a/src/databricks/labs/ucx/hive_metastore/mapping.py
+++ b/src/databricks/labs/ucx/hive_metastore/mapping.py
@@ -241,7 +241,9 @@ class TableMapping:
             return database
         describe = {}
         try:
-            for value in self._sql_backend.fetch(f"DESCRIBE SCHEMA EXTENDED {escape_sql_identifier(database)}"):
+            for value in self._sql_backend.fetch(
+                f"DESCRIBE SCHEMA EXTENDED hive_metastore.{escape_sql_identifier(database)}"
+            ):
                 describe[value["database_description_item"]] = value["database_description_value"]
         except NotFound:
             logger.warning(
@@ -290,10 +292,12 @@ class TableMapping:
         return table_to_migrate
 
     def _get_table_properties(self, table: Table):
-        table_identifier = f"{escape_sql_identifier(table.database)}.{escape_sql_identifier(table.name)}"
         if table.is_table_in_mount:
             table_identifier = f"delta.`{table.location}`"
-
+        else:
+            table_identifier = (
+                f"hive_metastore.{escape_sql_identifier(table.database)}.{escape_sql_identifier(table.name)}"
+            )
         try:
             return self._sql_backend.fetch(f"SHOW TBLPROPERTIES {table_identifier}")
         except DatabricksError as err:
diff --git a/src/databricks/labs/ucx/progress/workflows.py b/src/databricks/labs/ucx/progress/workflows.py
index 8206bcec..8d27f962 100644
--- a/src/databricks/labs/ucx/progress/workflows.py
+++ b/src/databricks/labs/ucx/progress/workflows.py
@@ -181,7 +181,7 @@ class MigrationProgress(Workflow):
         """Scans all jobs for migration issues in notebooks.
         Also stores direct filesystem accesses for display in the migration dashboard."""
         # TODO: Ensure these are captured in the history log.
-        ctx.workflow_linter.refresh_report(ctx.sql_backend, ctx.inventory_database)
+        ctx.workflow_linter.refresh_report(ctx.sql_backend, ctx.inventory_database, last_run_days=30)
 
     @job_task(
         depends_on=[
diff --git a/src/databricks/labs/ucx/source_code/linters/jobs.py b/src/databricks/labs/ucx/source_code/linters/jobs.py
index 721fdc8a..f8558ad6 100644
--- a/src/databricks/labs/ucx/source_code/linters/jobs.py
+++ b/src/databricks/labs/ucx/source_code/linters/jobs.py
@@ -1,9 +1,9 @@
 import dataclasses
-import functools
 import logging
 
 from collections.abc import Iterable
 from datetime import datetime, timezone
+from functools import partial
 from pathlib import Path
 
 from databricks.labs.blueprint.parallel import Threads
@@ -56,10 +56,22 @@ class WorkflowLinter:
         self._directfs_crawler = directfs_crawler
         self._used_tables_crawler = used_tables_crawler
 
-    def refresh_report(self, sql_backend: SqlBackend, inventory_database: str) -> None:
+    def refresh_report(
+        self, sql_backend: SqlBackend, inventory_database: str, /, last_run_days: int | None = None
+    ) -> None:
         tasks = []
+
+        def lint_job_limited(job_id: int) -> tuple[list[JobProblem], list[DirectFsAccess], list[UsedTable]]:
+            return self.lint_job(job_id, last_run_days=last_run_days)
+
         for job in self._jobs_crawler.snapshot():
-            tasks.append(functools.partial(self.lint_job, job.job_id))
+            try:
+                job_id = int(job.job_id)
+            except ValueError:
+                logger.warning(f"Invalid job id: {job.job_id}")
+                continue
+            tasks.append(partial(lint_job_limited, job_id))
+        #     TODO: Limit Scope
         logger.info(f"Running {len(tasks)} linting tasks in parallel...")
         job_results, errors = Threads.gather('linting workflows', tasks)
         job_problems: list[JobProblem] = []
@@ -82,12 +94,26 @@ class WorkflowLinter:
             error_messages = "\n".join([str(error) for error in errors])
             logger.warning(f"Errors occurred during linting:\n{error_messages}")
 
-    def lint_job(self, job_id: int) -> tuple[list[JobProblem], list[DirectFsAccess], list[UsedTable]]:
+    def lint_job(
+        self, job_id: int, /, last_run_days: int | None = None
+    ) -> tuple[list[JobProblem], list[DirectFsAccess], list[UsedTable]]:
         try:
             job = self._ws.jobs.get(job_id)
         except NotFound:
             logger.warning(f'Could not find job: {job_id}')
-            return ([], [], [])
+            return [], [], []
+
+        if last_run_days:
+            current_day_ms = int(datetime.now().timestamp() * 1000)
+            last_run_day_ms = current_day_ms - (last_run_days * 24 * 60 * 60 * 1000)
+            runs = self._ws.jobs.list_runs(
+                job_id=job_id,
+                limit=1,
+                start_time_from=last_run_day_ms,
+            )
+            if not runs:
+                logger.warning(f'Could not find job runs in the last {last_run_days} days: {job_id}')
+                return [], [], []
 
         problems, dfsas, tables = self._lint_job(job)
         if len(problems) > 0:


Test patch:
diff --git a/tests/unit/hive_metastore/test_mapping.py b/tests/unit/hive_metastore/test_mapping.py
index c9751c65..01944256 100644
--- a/tests/unit/hive_metastore/test_mapping.py
+++ b/tests/unit/hive_metastore/test_mapping.py
@@ -330,18 +330,18 @@ def test_skip_tables_marked_for_skipping_or_upgraded():
             ["test_schema2"],
             ["test_schema3"],
         ],
-        "SHOW TBLPROPERTIES `test_schema1`.`test_table1`": [
+        "SHOW TBLPROPERTIES hive_metastore.`test_schema1`.`test_table1`": [
             {"key": "upgraded_to", "value": "fake_dest"},
         ],
-        "SHOW TBLPROPERTIES `test_schema1`.`test_view1`": [
+        "SHOW TBLPROPERTIES hive_metastore.`test_schema1`.`test_view1`": [
             {"key": "databricks.labs.ucx.skip", "value": "true"},
         ],
-        "SHOW TBLPROPERTIES `test_schema1`.`test_table2`": [
+        "SHOW TBLPROPERTIES hive_metastore.`test_schema1`.`test_table2`": [
             {"key": "upgraded_to", "value": "fake_dest"},
         ],
-        "DESCRIBE SCHEMA EXTENDED `test_schema1`": [],
-        "DESCRIBE SCHEMA EXTENDED `test_schema2`": [],
-        "DESCRIBE SCHEMA EXTENDED `test_schema3`": [
+        "DESCRIBE SCHEMA EXTENDED hive_metastore.`test_schema1`": [],
+        "DESCRIBE SCHEMA EXTENDED hive_metastore.`test_schema2`": [],
+        "DESCRIBE SCHEMA EXTENDED hive_metastore.`test_schema3`": [
             {
                 "database_description_item": "Properties",
                 "database_description_value": "((databricks.labs.ucx.skip,true))",
@@ -418,7 +418,7 @@ def test_skip_tables_marked_for_skipping_or_upgraded():
 def test_table_with_no_target_reverted():
     errors = {}
     rows = {
-        "SHOW TBLPROPERTIES `schema1`.`table1`": [
+        "SHOW TBLPROPERTIES hive_metastore.`schema1`.`table1`": [
             {"key": "upgraded_to", "value": "non.existing.table"},
         ],
     }
@@ -490,7 +490,7 @@ def test_skipping_rules_existing_targets():
     ]
     table_mapping.get_tables_to_migrate(tables_crawler)
 
-    assert ["DESCRIBE SCHEMA EXTENDED `schema1`"] == backend.queries
+    assert ["DESCRIBE SCHEMA EXTENDED hive_metastore.`schema1`"] == backend.queries
 
 
 def test_mismatch_from_table_raises_exception():
@@ -530,7 +530,7 @@ def test_mismatch_from_table_raises_exception():
             assert len(e.errs) == 1
             raise e.errs[0]
 
-    assert ["DESCRIBE SCHEMA EXTENDED `schema1`"] == backend.queries
+    assert ["DESCRIBE SCHEMA EXTENDED hive_metastore.`schema1`"] == backend.queries
 
 
 def test_table_not_in_crawled_tables():
@@ -548,7 +548,7 @@ def test_table_not_in_crawled_tables():
     tables_crawler.snapshot.return_value = []
     table_mapping.get_tables_to_migrate(tables_crawler)
 
-    assert ["DESCRIBE SCHEMA EXTENDED `schema1`"] == backend.queries
+    assert ["DESCRIBE SCHEMA EXTENDED hive_metastore.`schema1`"] == backend.queries
 
 
 def test_skipping_rules_database_skipped():
@@ -560,7 +560,7 @@ def test_skipping_rules_database_skipped():
     )
     errors = {}
     rows = {
-        "DESCRIBE SCHEMA EXTENDED `schema2`": [
+        "DESCRIBE SCHEMA EXTENDED hive_metastore.`schema2`": [
             {
                 "database_description_item": "Properties",
                 "database_description_value": "((databricks.labs.ucx.skip,true))",
@@ -594,8 +594,8 @@ def test_skipping_rules_database_skipped():
     ]
     table_mapping.get_tables_to_migrate(tables_crawler)
 
-    assert "SHOW TBLPROPERTIES `schema1`.`table1`" in backend.queries
-    assert "SHOW TBLPROPERTIES `schema2`.`table2`" not in backend.queries
+    assert "SHOW TBLPROPERTIES hive_metastore.`schema1`.`table1`" in backend.queries
+    assert "SHOW TBLPROPERTIES hive_metastore.`schema2`.`table2`" not in backend.queries
 
 
 def test_skip_missing_table_in_snapshot():
@@ -754,7 +754,7 @@ def test_tables_in_mounts():
     tables_crawler.snapshot.return_value = []
     table_mapping.get_tables_to_migrate(tables_crawler)
 
-    assert ["DESCRIBE SCHEMA EXTENDED `schema1`"] == backend.queries
+    assert ["DESCRIBE SCHEMA EXTENDED hive_metastore.`schema1`"] == backend.queries
 
 
 def test_mapping_table_in_mount():
@@ -913,7 +913,7 @@ def test_mapping_broken_table(caplog):
 def test_table_with_no_target_reverted_failed(caplog):
     errors = {"ALTER TABLE": "ALTER_TABLE_FAILED"}
     rows = {
-        "SHOW TBLPROPERTIES `schema1`.`table1`": [
+        "SHOW TBLPROPERTIES hive_metastore.`schema1`.`table1`": [
             {"key": "upgraded_to", "value": "non.existing.table"},
         ],
     }
diff --git a/tests/unit/recon/test_migration_recon.py b/tests/unit/recon/test_migration_recon.py
index 1d92745e..2f03e6fc 100644
--- a/tests/unit/recon/test_migration_recon.py
+++ b/tests/unit/recon/test_migration_recon.py
@@ -42,8 +42,12 @@ def test_migrate_recon_should_produce_proper_queries(
             (src_3.schema, src_3.table, tgt_3.catalog, tgt_3.schema, tgt_3.table, "2021-01-01T00:00:00Z"),
             ("schema_none", "table_none", None, "schema_a", "table_a", "2021-01-01T00:00:00Z"),
         ],
-        f"SHOW TBLPROPERTIES {src.schema}.{src.table} \\('upgraded_to'\\)": UPGRADED_TO[("value", "fake_dest"),],
-        f"SHOW TBLPROPERTIES {src_2.schema}.{src_2.table} \\('upgraded_to'\\)": UPGRADED_TO[("value", "fake_dest"),],
+        f"SHOW TBLPROPERTIES hive_metastore.{src.schema}.{src.table} \\('upgraded_to'\\)": UPGRADED_TO[
+            ("value", "fake_dest"),
+        ],
+        f"SHOW TBLPROPERTIES hive_metastore.{src_2.schema}.{src_2.table} \\('upgraded_to'\\)": UPGRADED_TO[
+            ("value", "fake_dest"),
+        ],
         "DESCRIBE TABLE": metadata_row_factory[
             ("col1", "int"),
             ("col2", "string"),
diff --git a/tests/unit/source_code/test_jobs.py b/tests/unit/source_code/test_jobs.py
index e14d63fd..ecff87e4 100644
--- a/tests/unit/source_code/test_jobs.py
+++ b/tests/unit/source_code/test_jobs.py
@@ -10,7 +10,7 @@ from databricks.labs.lsql.backends import MockBackend
 from databricks.sdk import WorkspaceClient
 from databricks.sdk.errors import NotFound
 from databricks.sdk.service import compute, jobs
-from databricks.sdk.service.jobs import Job, SparkPythonTask
+from databricks.sdk.service.jobs import Job, SparkPythonTask, BaseRun, RunResultState
 from databricks.sdk.service.pipelines import (
     GetPipelineResponse,
     FileLibrary,
@@ -589,3 +589,67 @@ def test_workflow_linter_refresh_report(dependency_resolver, mock_path_lookup, m
     sql_backend.has_rows_written_for('test.workflow_problems')
     sql_backend.has_rows_written_for('hive_metastore.test.used_tables_in_paths')
     sql_backend.has_rows_written_for('hive_metastore.test.directfs_in_paths')
+
+
+def test_workflow_linter_refresh_report_time_bound(dependency_resolver, mock_path_lookup, migration_index) -> None:
+    ws = create_autospec(WorkspaceClient)
+    ws.workspace.get_status.return_value = ObjectInfo(object_id=123, language=Language.PYTHON)
+    some_things = mock_path_lookup.resolve(Path("functional/zoo.py"))
+    assert some_things is not None
+    ws.workspace.download.return_value = some_things.read_bytes()
+    notebook_task = jobs.NotebookTask(
+        notebook_path=some_things.absolute().as_posix(),
+        base_parameters={"a": "b", "c": "dbfs:/mnt/foo"},
+    )
+    task = jobs.Task(
+        task_key="test",
+        job_cluster_key="main",
+        notebook_task=notebook_task,
+    )
+    settings = jobs.JobSettings(
+        tasks=[task],
+        name='some',
+        job_clusters=[
+            jobs.JobCluster(
+                job_cluster_key="main",
+                new_cluster=compute.ClusterSpec(
+                    spark_version="15.2.x-photon-scala2.12",
+                    node_type_id="Standard_F4s",
+                    num_workers=2,
+                    data_security_mode=compute.DataSecurityMode.LEGACY_TABLE_ACL,
+                    spark_conf={"spark.databricks.cluster.profile": "singleNode"},
+                ),
+            ),
+        ],
+    )
+    ws.jobs.list.return_value = [Job(job_id=1, settings=settings), Job(job_id=2, settings=settings)]
+    ws.jobs.get.side_effect = [Job(job_id=1, settings=settings), Job(job_id=2, settings=settings)]
+    ws.jobs.list_runs.side_effect = [
+        [
+            BaseRun(
+                state=jobs.RunState(result_state=RunResultState.SUCCESS),
+                run_id=1,
+                job_id=2,
+                run_page_url="http://example.com",
+            )
+        ],
+        [],
+    ]
+    sql_backend = MockBackend()
+    jobs_crawler = JobsCrawler(ws, sql_backend, 'test')
+    directfs_crawler = DirectFsAccessCrawler.for_paths(sql_backend, "test")
+    used_tables_crawler = UsedTablesCrawler.for_paths(sql_backend, "test")
+    linter = WorkflowLinter(
+        ws,
+        jobs_crawler,
+        dependency_resolver,
+        mock_path_lookup,
+        migration_index,
+        directfs_crawler,
+        used_tables_crawler,
+    )
+    linter.refresh_report(sql_backend, 'test', last_run_days=30)
+
+    sql_backend.has_rows_written_for('test.workflow_problems')
+    sql_backend.has_rows_written_for('hive_metastore.test.used_tables_in_paths')
+    sql_backend.has_rows_written_for('hive_metastore.test.directfs_in_paths')