Warn when table has column without no name during table migration (#2984)

## Changes
Warn when table has column without no name during table migration. 

The code goes with a "do first, apologize later" approach to avoid an
additional call to retrieve the column schema.

### Linked issues
Resolves #2891

### Functionality

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

### Tests

- [x] added unit tests

Author: JCZuurmond

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

@@ -1,14 +1,15 @@
 import dataclasses
 import logging
+import re
 from collections import defaultdict
 from functools import partial
 
 from databricks.labs.blueprint.parallel import Threads
 from databricks.labs.lsql.backends import SqlBackend
-
-from databricks.labs.ucx.framework.utils import escape_sql_identifier
 from databricks.sdk import WorkspaceClient
+from databricks.sdk.errors.platform import DatabricksError
 
+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
 from databricks.labs.ucx.hive_metastore.locations import ExternalLocations
@@ -28,7 +29,6 @@
     ViewsMigrationSequencer,
     ViewToMigrate,
 )
-from databricks.sdk.errors.platform import DatabricksError
 
 logger = logging.getLogger(__name__)
 
@@ -95,7 +95,7 @@ def _migrate_tables(
         for table in tables_in_scope:
             tasks.append(
                 partial(
-                    self._migrate_table,
+                    self._safe_migrate_table,
                     table,
                     managed_table_external_storage,
                     hiveserde_in_place_migrate,
@@ -131,15 +131,35 @@ def _migrate_managed_table(self, managed_table_external_storage: str, src_table:
         logger.warning(f"failed-to-migrate: unknown managed_table_external_storage: {managed_table_external_storage}")
         return True
 
-    def _migrate_table(
+    def _safe_migrate_table(
         self,
         src_table: TableToMigrate,
         managed_table_external_storage: str,
         hiveserde_in_place_migrate: bool = False,
-    ):
+    ) -> bool:
         if self._table_already_migrated(src_table.rule.as_uc_table_key):
             logger.info(f"Table {src_table.src.key} already migrated to {src_table.rule.as_uc_table_key}")
             return True
+        try:
+            return self._migrate_table(src_table, managed_table_external_storage, hiveserde_in_place_migrate)
+        except Exception as e:  # pylint: disable=broad-exception-caught
+            # Catching a Spark AnalysisException here, for which we do not have the dependency to catch explicitly
+            pattern = (  # See https://github.com/databrickslabs/ucx/issues/2891
+                r"INVALID_PARAMETER_VALUE: Invalid input: RPC CreateTable Field managedcatalog.ColumnInfo.name: "
+                r'At columns.\d+: name "" is not a valid name`'
+            )
+            if re.match(pattern, str(e)):
+                logger.warning(f"failed-to-migrate: Table with empty column name '{src_table.src.key}'", exc_info=e)
+            else:
+                logger.warning(f"failed-to-migrate: Unknown reason for table '{src_table.src.key}'", exc_info=e)
+            return False
+
+    def _migrate_table(
+        self,
+        src_table: TableToMigrate,
+        managed_table_external_storage: str,
+        hiveserde_in_place_migrate: bool = False,
+    ) -> bool:
         if src_table.src.what == What.DBFS_ROOT_DELTA:
             return self._migrate_dbfs_root_table(src_table.src, src_table.rule)
         if src_table.src.what == What.DBFS_ROOT_NON_DELTA:

tests/unit/hive_metastore/test_table_migrate.py

@@ -10,6 +10,7 @@
 from databricks.sdk.service.catalog import CatalogInfo, SchemaInfo, TableInfo
 
 from databricks.labs.ucx.framework.owners import AdministratorLocator
+from databricks.labs.ucx.framework.utils import escape_sql_identifier
 from databricks.labs.ucx.hive_metastore.grants import MigrateGrants
 from databricks.labs.ucx.hive_metastore.locations import ExternalLocations
 from databricks.labs.ucx.hive_metastore.mapping import (
@@ -1483,3 +1484,60 @@ def test_table_migration_status_source_table_unknown() -> None:
 
     assert owner == "an_admin"
     table_ownership.owner_of.assert_not_called()
+
+
+class MockBackendWithGeneralException(MockBackend):
+    """Mock backend that allows raising a general exception.
+
+    Note: we want to raise a Spark AnalysisException, for which we do not have the dependency to raise explicitly.
+    """
+
+    @staticmethod
+    def _api_error_from_message(error_message: str):  # No return type to avoid mypy complains on different return type
+        return Exception(error_message)
+
+
+def test_migrate_tables_handles_table_with_empty_column(caplog) -> None:
+    table_crawler = create_autospec(TablesCrawler)
+    table = Table("hive_metastore", "schema", "table", "MANAGED", "DELTA")
+
+    error_message = (
+        "INVALID_PARAMETER_VALUE: Invalid input: RPC CreateTable Field managedcatalog.ColumnInfo.name: "
+        'At columns.21: name "" is not a valid name`'
+    )
+    query = f"ALTER TABLE {escape_sql_identifier(table.full_name)} SET TBLPROPERTIES ('upgraded_to' = 'catalog.schema.table');"
+    backend = MockBackendWithGeneralException(fails_on_first={query: error_message})
+
+    ws = create_autospec(WorkspaceClient)
+    ws.get_workspace_id.return_value = 123456789
+
+    table_mapping = create_autospec(TableMapping)
+    rule = Rule("workspace", "catalog", "schema", "schema", "table", "table")
+    table_to_migrate = TableToMigrate(table, rule)
+    table_mapping.get_tables_to_migrate.return_value = [table_to_migrate]
+
+    migration_status_refresher = create_autospec(TableMigrationStatusRefresher)
+    migration_status_refresher.get_seen_tables.return_value = {}
+    migration_status_refresher.index.return_value = []
+
+    migrate_grants = create_autospec(MigrateGrants)
+    external_locations = create_autospec(ExternalLocations)
+    table_migrator = TablesMigrator(
+        table_crawler,
+        ws,
+        backend,
+        table_mapping,
+        migration_status_refresher,
+        migrate_grants,
+        external_locations,
+    )
+
+    with caplog.at_level(logging.WARN, logger="databricks.labs.ucx.hive_metastore"):
+        table_migrator.migrate_tables(table.what)
+    assert "failed-to-migrate: Table with empty column name 'hive_metastore.schema.table'" in caplog.messages
+
+    table_crawler.snapshot.assert_not_called()  # Mocking table mapping instead
+    ws.get_workspace_id.assert_not_called()  # Errors before getting here
+    migration_status_refresher.index.assert_not_called()  # Only called when migrating view
+    migrate_grants.apply.assert_not_called()  # Errors before getting here
+    external_locations.resolve_mount.assert_not_called()  # Only called when migrating external table