Improved error handling for migrate-tables workflows (#1674)

## Changes
- improved error handling for `migrate-tables` workflows

### 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 #1673

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

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

Author: nkvuong

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

@@ -8,7 +8,7 @@
 from databricks.labs.blueprint.parallel import Threads
 from databricks.labs.lsql.backends import SqlBackend
 from databricks.sdk import WorkspaceClient
-from databricks.sdk.errors import BadRequest, NotFound, ResourceConflict
+from databricks.sdk.errors import BadRequest, NotFound, ResourceConflict, DatabricksError
 from databricks.sdk.service.catalog import TableInfo, SchemaInfo
 
 from databricks.labs.ucx.account.workspaces import WorkspaceInfo
@@ -182,9 +182,13 @@ def _get_table_in_scope_task(self, table_to_migrate: TableToMigrate) -> TableToM
         if self.exists_in_uc(table, rule.as_uc_table_key):
             logger.info(f"The intended target for {table.key}, {rule.as_uc_table_key}, already exists.")
             return None
-        result = self._sql_backend.fetch(
-            f"SHOW TBLPROPERTIES {escape_sql_identifier(table.database)}.{escape_sql_identifier(table.name)}"
-        )
+        try:
+            result = self._sql_backend.fetch(
+                f"SHOW TBLPROPERTIES {escape_sql_identifier(table.database)}.{escape_sql_identifier(table.name)}"
+            )
+        except DatabricksError as err:
+            logger.warning(f"Failed to get properties for Table {table.key}: {err}")
+            return None
         for value in result:
             if value["key"] == self.UCX_SKIP_PROPERTY:
                 logger.info(f"{table.key} is marked to be skipped")
@@ -199,7 +203,10 @@ def _get_table_in_scope_task(self, table_to_migrate: TableToMigrate) -> TableToM
                     )
                     return None
                 logger.info(f"The upgrade_to target for {table.key} is missing. Unsetting the upgrade_to property")
-                self._sql_backend.execute(table.sql_unset_upgraded_to())
+                try:
+                    self._sql_backend.execute(table.sql_unset_upgraded_to())
+                except DatabricksError as err:
+                    logger.warning(f"Failed to unset upgraded_to property for Table {table.key}: {err}")
 
         return table_to_migrate
 

tests/unit/hive_metastore/test_mapping.py

@@ -638,3 +638,77 @@ def test_is_target_exists():
     )
     assert not table_mapping.exists_in_uc(src_table, "cat1.schema1.dest1")
     assert table_mapping.exists_in_uc(src_table, "cat1.schema2.dest2")
+
+
+def test_mapping_broken_table(caplog):
+    client = create_autospec(WorkspaceClient)
+    tables_crawler = create_autospec(TablesCrawler)
+    # When check table properties, raise token error
+    backend = MockBackend(fails_on_first={"SHOW TBLPROPERTIES": "tokentoken"})
+    installation = MockInstallation(
+        {
+            'mapping.csv': [
+                {
+                    'catalog_name': 'catalog',
+                    'dst_schema': 'schema1',
+                    'dst_table': 'table1',
+                    'src_schema': 'schema1',
+                    'src_table': 'table1',
+                    'workspace_name': 'workspace',
+                },
+            ]
+        }
+    )
+    client.tables.get.side_effect = NotFound()
+    table_mapping = TableMapping(installation, client, backend)
+    tables_crawler.snapshot.return_value = [
+        Table(
+            object_type="EXTERNAL",
+            table_format="DELTA",
+            catalog="hive_metastore",
+            database="schema1",
+            name="table1",
+        ),
+    ]
+    table_mapping.get_tables_to_migrate(tables_crawler)
+    assert "Failed to get properties for Table hive_metastore.schema1.table1" in caplog.text
+
+
+def test_table_with_no_target_reverted_failed(caplog):
+    errors = {"ALTER TABLE": "ALTER_TABLE_FAILED"}
+    rows = {
+        "SHOW TBLPROPERTIES schema1.table1": [
+            {"key": "upgraded_to", "value": "non.existing.table"},
+        ],
+    }
+    backend = MockBackend(fails_on_first=errors, rows=rows)
+    client = create_autospec(WorkspaceClient)
+    client.tables.get.side_effect = NotFound()
+
+    installation = MockInstallation(
+        {
+            'mapping.csv': [
+                {
+                    'workspace_name': "fake_ws",
+                    "catalog_name": 'cat1',
+                    'src_schema': 'schema1',
+                    'dst_schema': 'schema1',
+                    'src_table': 'table1',
+                    'dst_table': 'table1',
+                }
+            ]
+        }
+    )
+    table_mapping = TableMapping(installation, client, backend)
+    tables_crawler = create_autospec(TablesCrawler)
+    tables_crawler.snapshot.return_value = [
+        Table(
+            object_type="EXTERNAL",
+            table_format="DELTA",
+            catalog="hive_metastore",
+            database="schema1",
+            name="table1",
+        ),
+    ]
+    table_mapping.get_tables_to_migrate(tables_crawler)
+    assert "Failed to unset upgraded_to property" in caplog.text