fix(mssql): harden error handling and observability across throttling, description maps, and unguarded query paths

ingestion/src/metadata/ingestion/source/database/mssql/connection.py

@@ -116,6 +116,26 @@ def _sqlserver_errno(*codes: int) -> Matcher:
         "Insufficient privileges",
         fix="Grant the login SELECT on the objects the failing step reads (and VIEW SERVER STATE for query history).",
     ),
+    # Azure SQL DTU/vCore resource-governance throttling - not an auth or permission
+    # error. Verified against Microsoft's own error catalog (Azure SQL troubleshooting
+    # docs + a live `sys.messages` lookup): 10928 = per-database worker/request limit
+    # reached, 10936 = the same for an elastic pool, 10929 = server too busy to admit
+    # new requests for this database. pyodbc never exposes a number (see _mssql_number),
+    # so it needs the text fallback like every other pyodbc-reachable rule here. The
+    # reference URL in these messages has changed across SQL Server versions, so match
+    # on the stable message body instead.
+    when(
+        Matchers.any_of(
+            _sqlserver_errno(10928, 10929, 10936),
+            Matchers.contains("limit for the database is"),
+            Matchers.contains("limit for the elastic pool is"),
+            Matchers.contains("is currently too busy to support requests"),
+        )
+    ).diagnose(
+        "Azure SQL resource limit reached (throttled)",
+        fix="The database's DTU/vCore limit was reached. Reduce concurrent load or scale up the "
+        "Azure SQL database, then retry.",
+    ),
 )
 
 MSSQL_ERRORS = SQLSERVER_ERRORS.including(NETWORK_ERRORS)

ingestion/src/metadata/ingestion/source/database/mssql/metadata.py

@@ -210,7 +210,7 @@ def _load_description_maps(self) -> None:
                 load_description_map()
             except Exception as exc:
                 logger.debug(traceback.format_exc())
-                logger.debug(f"Could not load MSSQL {description_type} descriptions, continuing without them: {exc}")
+                logger.warning(f"Could not load MSSQL {description_type} descriptions, continuing without them: {exc}")
 
     def get_database_names(self) -> Iterable[str]:
         if not self.config.serviceConnection.root.config.ingestAllDatabases:  # pyright: ignore[reportAttributeAccessIssue]
@@ -252,15 +252,25 @@ def get_database_names(self) -> Iterable[str]:
     def get_stored_procedures(self) -> Iterable[MssqlStoredProcedure]:
         """List Snowflake stored procedures"""
         if self.source_config.includeStoredProcedures:
-            with self.engine.connect() as conn:
-                results = conn.execute(
-                    text(
-                        MSSQL_GET_STORED_PROCEDURES.format(
-                            database_name=self.context.get().database,
-                            schema_name=self.context.get().database_schema,
+            schema_name = self.context.get().database_schema  # pyright: ignore[reportAttributeAccessIssue]
+            try:
+                with self.engine.connect() as conn:
+                    results = conn.execute(
+                        text(
+                            MSSQL_GET_STORED_PROCEDURES.format(
+                                database_name=self.context.get().database,  # pyright: ignore[reportAttributeAccessIssue]
+                                schema_name=schema_name,
+                            )
                         )
-                    )
-                ).all()
+                    ).all()
+            except Exception as exc:
+                logger.debug(traceback.format_exc())
+                logger.warning(f"Error listing stored procedures for schema {schema_name}: {exc}")
+                self.status.warning(
+                    schema_name,
+                    f"Error listing stored procedures for schema {schema_name}: {exc}",
+                )
+                return
             for row in results:
                 try:
                     stored_procedure = MssqlStoredProcedure.model_validate(row._asdict())

ingestion/src/metadata/ingestion/source/database/mssql/query_parser.py

@@ -12,6 +12,7 @@
 MSSQL usage module
 """
 
+import traceback
 from abc import ABC
 from copy import deepcopy
 from typing import Iterator, Optional  # noqa: UP035
@@ -119,15 +120,33 @@ def _per_database_engines(self) -> Iterator[Engine]:
         for database in databases:
             engine = self._engine_for_database(database)
             self._active_query_store = is_query_store_enabled(engine)
+            if self._active_query_store:
+                logger.info(
+                    "MSSQL query history for database %s: Query Store is enabled.",
+                    database,
+                )
+            else:
+                logger.info(
+                    "MSSQL query history for database %s: Query Store is not enabled or not accessible, "
+                    "using plan-cache DMVs.",
+                    database,
+                )
             try:
                 yield engine
             finally:
                 engine.dispose()
         self._active_query_store = None
 
     def _databases_to_scan(self) -> Iterator[str]:
-        with self.engine.connect() as conn:
-            rows = conn.execute(text(MSSQL_GET_QUERY_STORE_DATABASES)).fetchall()
+        try:
+            with self.engine.connect() as conn:
+                rows = conn.execute(text(MSSQL_GET_QUERY_STORE_DATABASES)).fetchall()
+        except Exception as exc:
+            logger.debug(traceback.format_exc())
+            logger.warning(
+                f"Could not list databases to scan, falling back to the configured connection database: {exc}"
+            )
+            return
         database_filter = getattr(self.source_config, "databaseFilterPattern", None)
         for row in rows:
             database = row[0]

ingestion/src/metadata/ingestion/source/database/mssql/utils.py

@@ -12,11 +12,11 @@
 MSSQL SQLAlchemy Helper Methods
 """
 
-from typing import Optional  # noqa: I001
+import traceback
+from typing import Optional
 
-from sqlalchemy import Column, Integer, MetaData, String, Table, alias, sql, text
+from sqlalchemy import Column, Integer, MetaData, String, Table, alias, sql, text, util
 from sqlalchemy import types as sqltypes
-from sqlalchemy import util
 from sqlalchemy.dialects.mssql import information_schema as ischema
 from sqlalchemy.dialects.mssql.base import (
     MSBinary,
@@ -478,10 +478,18 @@ def get_view_names(self, connection, dbname, owner, schema, **kw):  # pylint: di
 
 def get_sqlalchemy_engine_dateformat(engine: Engine) -> Optional[str]:  # noqa: UP045
     """
-    returns sqlaclhemdy engine date format by running config query
+    returns sqlaclhemdy engine date format by running config query.
+    Returns None (falling back to the caller's default format) if the
+    probe itself fails, e.g. a transient connection issue - the caller
+    already has a documented default for this case.
     """
-    with engine.connect() as conn:
-        result = conn.execute(text(GET_DB_CONFIGS)).all()
+    try:
+        with engine.connect() as conn:
+            result = conn.execute(text(GET_DB_CONFIGS)).all()
+    except Exception as exc:
+        logger.warning(f"Could not determine MSSQL dateformat, falling back to the default: {exc}")
+        logger.debug(traceback.format_exc())
+        return None
     for row in result:
         row_dict = row._asdict()
         if row_dict.get("Set Option") == "dateformat":

ingestion/tests/unit/source/database/mssql/test_connection.py

@@ -218,6 +218,28 @@ def _wrapped(orig: Exception) -> Exception:
     (300, "VIEW SERVER STATE permission was denied on object 'server', database 'master'."),
     (297, "The user does not have permission to perform this action."),
 )
+_THROTTLED_LOGIN_LIMIT = (
+    (
+        10928,
+        "Resource ID: 1. The request limit for the database is 60 and has been reached. See "
+        "'https://docs.microsoft.com/azure/azure-sql/database/resource-limits-logical-server' for assistance.",
+    ),
+)
+_THROTTLED_ELASTIC_POOL_LIMIT = (
+    (
+        10936,
+        "Resource ID: 1. The request limit for the elastic pool is 200 and has been reached. See "
+        "'https://docs.microsoft.com/azure/azure-sql/database/resource-limits-logical-server' for assistance.",
+    ),
+)
+_THROTTLED_WORKER_LIMIT = (
+    (
+        10929,
+        "Resource ID: 1. The worker minimum guarantee is 10, maximum limit is 60, and the current usage "
+        "for the database is 61. However, the server is currently too busy to support requests greater "
+        "than 60 for this database. Otherwise, please try again later.",
+    ),
+)
 
 
 @pytest.mark.parametrize(
@@ -362,6 +384,59 @@ def test_statement_permission_denied_is_not_diagnosed():
     assert MSSQL_ERRORS.classify(error) is None
 
 
+def test_pytds_throttling_limit_reached_classifies():
+    error = _wrapped(_pytds_error(*_THROTTLED_LOGIN_LIMIT))
+    diagnosis = MSSQL_ERRORS.classify(error)
+    assert diagnosis is not None
+    assert diagnosis.title == "Azure SQL resource limit reached (throttled)"
+
+
+def test_pytds_throttling_worker_limit_classifies():
+    error = _wrapped(_pytds_error(*_THROTTLED_WORKER_LIMIT))
+    diagnosis = MSSQL_ERRORS.classify(error)
+    assert diagnosis is not None
+    assert diagnosis.title == "Azure SQL resource limit reached (throttled)"
+
+
+def test_pymssql_throttling_tuple_shape_classifies():
+    diagnosis = MSSQL_ERRORS.classify(_wrapped(_pymssql_error(10928, "The request limit for the database is 60.")))
+    assert diagnosis is not None
+    assert diagnosis.title == "Azure SQL resource limit reached (throttled)"
+
+
+def test_pytds_throttling_elastic_pool_limit_classifies():
+    error = _wrapped(_pytds_error(*_THROTTLED_ELASTIC_POOL_LIMIT))
+    diagnosis = MSSQL_ERRORS.classify(error)
+    assert diagnosis is not None
+    assert diagnosis.title == "Azure SQL resource limit reached (throttled)"
+
+
+def test_pyodbc_throttling_limit_reached_classifies():
+    """pyodbc never exposes a SQL Server error number (see _mssql_number), so this rule
+    must fall through to the text match for pyodbc connections to get diagnosed at all."""
+    diagnosis = MSSQL_ERRORS.classify(
+        _pyodbc_error(
+            "HY000",
+            "Resource ID: 1. The request limit for the database is 60 and has been reached.",
+        )
+    )
+    assert diagnosis is not None
+    assert diagnosis.title == "Azure SQL resource limit reached (throttled)"
+
+
+def test_pyodbc_throttling_worker_limit_classifies():
+    diagnosis = MSSQL_ERRORS.classify(
+        _pyodbc_error(
+            "HY000",
+            "Resource ID: 1. The worker minimum guarantee is 10, maximum limit is 60, and the current "
+            "usage for the database is 61. However, the server is currently too busy to support "
+            "requests greater than 60 for this database.",
+        )
+    )
+    assert diagnosis is not None
+    assert diagnosis.title == "Azure SQL resource limit reached (throttled)"
+
+
 def test_network_pack_is_folded_in():
     diagnosis = MSSQL_ERRORS.classify(socket.gaierror("Name or service not known"))
     assert diagnosis is not None

ingestion/tests/unit/topology/database/test_mssql.py

@@ -311,6 +311,25 @@ def test_get_stored_procedures(self):
         self.assertEqual(len(results), 1)
         self.assertEqual(results[0].name, "sp_include")
 
+    def test_get_stored_procedures_degrades_gracefully_on_query_failure(self):
+        """A permission/syntax failure listing stored procedures for one schema
+        is reported as a warning, not a hard failure that tanks the workflow's
+        success percentage - the rest of ingestion is unaffected."""
+        self.mssql.source_config.includeStoredProcedures = True
+        self.mssql.context.get().__dict__["database"] = MOCK_DATABASE.name.root
+        self.mssql.context.get().__dict__["database_schema"] = MOCK_DATABASE_SCHEMA.name.root
+        self.mssql.status = MagicMock()
+
+        mock_engine = MagicMock()
+        mock_engine.connect.side_effect = Exception("The SELECT permission was denied on sql_modules")
+        self.mssql.engine = mock_engine
+
+        results = list(self.mssql.get_stored_procedures())
+
+        self.assertEqual(results, [])
+        self.mssql.status.warning.assert_called_once()
+        self.mssql.status.failed.assert_not_called()
+
 
 class TestUpdateMssqlIschemaNames:
     """Verify update_mssql_ischema_names mutates the dict in-place and returns None."""
@@ -746,6 +765,27 @@ def test_stored_procedure_statement_falls_back_to_dmv(self):
         assert "sys.dm_exec_procedure_stats" in statement
 
 
+class TestMssqlDateformatProbe:
+    """get_sqlalchemy_engine_dateformat: falls back to the caller's documented
+    default instead of crashing source/usage/lineage construction when the
+    DBCC USEROPTIONS probe itself fails (e.g. a transient connection issue)."""
+
+    def test_returns_dateformat_value(self):
+        engine = MagicMock()
+        conn = engine.connect.return_value.__enter__.return_value
+        row = MagicMock()
+        row._asdict.return_value = {"Set Option": "dateformat", "Value": "ymd"}
+        conn.execute.return_value.all.return_value = [row]
+
+        assert mssql_dialet.get_sqlalchemy_engine_dateformat(engine) == "ymd"
+
+    def test_returns_none_when_probe_errors(self):
+        engine = MagicMock()
+        engine.connect.side_effect = Exception("connection reset")
+
+        assert mssql_dialet.get_sqlalchemy_engine_dateformat(engine) is None
+
+
 class TestMssqlPerDatabaseQueryStore:
     """Per-database Query Store engine iteration for ingest-all-databases runs."""
 
@@ -847,6 +887,16 @@ def test_databases_to_scan_applies_database_filter(self):
 
         assert list(source._databases_to_scan()) == ["SalesDW", "Inventory"]
 
+    def test_databases_to_scan_degrades_gracefully_on_query_failure(self):
+        """A transient failure listing databases (network blip, timeout) must not
+        crash the whole ingest-all-databases run - it falls back to the single,
+        already-connected engine instead."""
+        source = self._source(query_store_enabled=True, ingest_all_databases=True)
+        source.engine.connect.side_effect = Exception("connection reset")
+
+        assert list(source._databases_to_scan()) == []
+        assert list(source.get_engine()) == [source.engine]
+
     def test_falls_back_to_dmv_when_no_user_databases_scanned(self):
         source = self._source(query_store_enabled=True, ingest_all_databases=True)
         source._databases_to_scan = lambda: iter([])