From f29be7b930b7f09a99cb2de06e3826294af685d1 Mon Sep 17 00:00:00 2001 From: Akash Verma Date: Fri, 21 Aug 2026 16:34:50 +0530 Subject: [PATCH 1/5] fix(mssql): classify Azure SQL throttling, warn on description-map failures, log Query Store choice per database - SQLSERVER_ERRORS now recognizes Azure SQL DTU/vCore throttling (10928/10929) instead of surfacing a generic, unclassified error. - Description-map load failures (schema/database/stored-procedure comments) now log at WARNING instead of silent DEBUG, so a real permission/connection issue is visible instead of disappearing. - The ingest-all-databases per-database engine loop now logs which query-log source (Query Store vs plan-cache DMVs) was picked for each database, matching the single-database path's existing log line. --- .../source/database/mssql/connection.py | 7 +++++ .../source/database/mssql/metadata.py | 2 +- .../source/database/mssql/query_parser.py | 11 ++++++++ .../source/database/mssql/test_connection.py | 28 +++++++++++++++++++ 4 files changed, 47 insertions(+), 1 deletion(-) diff --git a/ingestion/src/metadata/ingestion/source/database/mssql/connection.py b/ingestion/src/metadata/ingestion/source/database/mssql/connection.py index b2119a8f2f98..609c88a49a4d 100644 --- a/ingestion/src/metadata/ingestion/source/database/mssql/connection.py +++ b/ingestion/src/metadata/ingestion/source/database/mssql/connection.py @@ -116,6 +116,13 @@ 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, keyed by number only since the message text varies by which limit was hit. + when(_sqlserver_errno(10928, 10929)).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) diff --git a/ingestion/src/metadata/ingestion/source/database/mssql/metadata.py b/ingestion/src/metadata/ingestion/source/database/mssql/metadata.py index 522833987953..d4f5eaf052f9 100644 --- a/ingestion/src/metadata/ingestion/source/database/mssql/metadata.py +++ b/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] diff --git a/ingestion/src/metadata/ingestion/source/database/mssql/query_parser.py b/ingestion/src/metadata/ingestion/source/database/mssql/query_parser.py index 1e4eeee33506..45585fd0cc82 100644 --- a/ingestion/src/metadata/ingestion/source/database/mssql/query_parser.py +++ b/ingestion/src/metadata/ingestion/source/database/mssql/query_parser.py @@ -119,6 +119,17 @@ 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: diff --git a/ingestion/tests/unit/source/database/mssql/test_connection.py b/ingestion/tests/unit/source/database/mssql/test_connection.py index 4a5937fb099c..5e2869ee4287 100644 --- a/ingestion/tests/unit/source/database/mssql/test_connection.py +++ b/ingestion/tests/unit/source/database/mssql/test_connection.py @@ -218,6 +218,14 @@ 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."),) +_THROTTLED_WORKER_LIMIT = ( + ( + 10929, + "Resource ID: 1. The %s minimum guarantee is 10, maximum limit is 60, and the current usage " + "for the database is 61.", + ), +) @pytest.mark.parametrize( @@ -362,6 +370,26 @@ 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_network_pack_is_folded_in(): diagnosis = MSSQL_ERRORS.classify(socket.gaierror("Name or service not known")) assert diagnosis is not None From 507fc121d33f74dd4313cded8ea3fbcdee16a8da Mon Sep 17 00:00:00 2001 From: Akash Verma Date: Fri, 21 Aug 2026 16:55:12 +0530 Subject: [PATCH 2/5] fix(mssql): cover pyodbc and elastic pools in the Azure SQL throttling rule - pyodbc never exposes a SQL Server error number (see _mssql_number), so the number-only throttling rule left it undiagnosed. Add a text fallback, matching the pattern every other pyodbc-reachable rule in this file uses. - Add 10936 (the elastic-pool variant of 10928), missed in the original rule. - Verified 10928/10929/10936 against Microsoft's Azure SQL troubleshooting docs and a live sys.messages lookup; fixed the test fixtures to use the real message text instead of guessed/truncated wording. --- .../source/database/mssql/connection.py | 16 +++++- .../source/database/mssql/test_connection.py | 53 +++++++++++++++++-- 2 files changed, 64 insertions(+), 5 deletions(-) diff --git a/ingestion/src/metadata/ingestion/source/database/mssql/connection.py b/ingestion/src/metadata/ingestion/source/database/mssql/connection.py index 609c88a49a4d..2b1a1b493802 100644 --- a/ingestion/src/metadata/ingestion/source/database/mssql/connection.py +++ b/ingestion/src/metadata/ingestion/source/database/mssql/connection.py @@ -117,8 +117,20 @@ def _sqlserver_errno(*codes: int) -> Matcher: 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, keyed by number only since the message text varies by which limit was hit. - when(_sqlserver_errno(10928, 10929)).diagnose( + # 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("has been reached"), + 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.", diff --git a/ingestion/tests/unit/source/database/mssql/test_connection.py b/ingestion/tests/unit/source/database/mssql/test_connection.py index 5e2869ee4287..f418cf234445 100644 --- a/ingestion/tests/unit/source/database/mssql/test_connection.py +++ b/ingestion/tests/unit/source/database/mssql/test_connection.py @@ -218,12 +218,26 @@ 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."),) +_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 %s minimum guarantee is 10, maximum limit is 60, and the current usage " - "for the database is 61.", + "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.", ), ) @@ -390,6 +404,39 @@ def test_pymssql_throttling_tuple_shape_classifies(): 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 From b59c58a63f7c21a8d9f2e779bf428f83b67139bf Mon Sep 17 00:00:00 2001 From: Akash Verma Date: Fri, 21 Aug 2026 17:04:30 +0530 Subject: [PATCH 3/5] fix(mssql): narrow Azure SQL throttling text fallback "has been reached" was a generic enough substring match that any non-throttling SQL Server error mentioning a limit, on a pyodbc connection with no exposed error number, could misdiagnose as Azure SQL throttling. Match the specific "limit for the database is" / "limit for the elastic pool is" phrases from the real 10928/10936 messages instead - same coverage, narrower match. --- .../src/metadata/ingestion/source/database/mssql/connection.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/ingestion/src/metadata/ingestion/source/database/mssql/connection.py b/ingestion/src/metadata/ingestion/source/database/mssql/connection.py index 2b1a1b493802..60aa5513f8a1 100644 --- a/ingestion/src/metadata/ingestion/source/database/mssql/connection.py +++ b/ingestion/src/metadata/ingestion/source/database/mssql/connection.py @@ -127,7 +127,8 @@ def _sqlserver_errno(*codes: int) -> Matcher: when( Matchers.any_of( _sqlserver_errno(10928, 10929, 10936), - Matchers.contains("has been reached"), + 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( From 48d2a513449c52b578bffd62e4f051f7b5d0e89e Mon Sep 17 00:00:00 2001 From: Akash Verma Date: Fri, 21 Aug 2026 18:36:08 +0530 Subject: [PATCH 4/5] fix(mssql): guard three more unhandled query paths against transient/permission failures - _databases_to_scan() (query_parser.py) now falls back to the single configured engine instead of crashing the whole ingest-all-databases usage/lineage run when the database-listing query fails. - get_sqlalchemy_engine_dateformat() (utils.py) now returns None on failure so callers fall back to the already-documented DEFAULT_DATETIME_FORMAT instead of crashing source construction. - get_stored_procedures() (metadata.py) now reports a permission/syntax failure via status.warning() instead of letting it crash the whole generator uncaught - keeping it out of the success-rate calculation so one schema's stored-procedure permission gap doesn't fail the run. All three were reproduced live against a real SQL Server container via the actual metadata ingest/usage CLI before fixing, and re-verified live after. Adds matching unit tests. --- .../source/database/mssql/metadata.py | 26 +++++++--- .../source/database/mssql/query_parser.py | 12 ++++- .../ingestion/source/database/mssql/utils.py | 20 +++++--- .../unit/topology/database/test_mssql.py | 50 +++++++++++++++++++ 4 files changed, 92 insertions(+), 16 deletions(-) diff --git a/ingestion/src/metadata/ingestion/source/database/mssql/metadata.py b/ingestion/src/metadata/ingestion/source/database/mssql/metadata.py index d4f5eaf052f9..44f4f7c017bd 100644 --- a/ingestion/src/metadata/ingestion/source/database/mssql/metadata.py +++ b/ingestion/src/metadata/ingestion/source/database/mssql/metadata.py @@ -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 + try: + with self.engine.connect() as conn: + results = conn.execute( + text( + MSSQL_GET_STORED_PROCEDURES.format( + database_name=self.context.get().database, + 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()) diff --git a/ingestion/src/metadata/ingestion/source/database/mssql/query_parser.py b/ingestion/src/metadata/ingestion/source/database/mssql/query_parser.py index 45585fd0cc82..3a8521530782 100644 --- a/ingestion/src/metadata/ingestion/source/database/mssql/query_parser.py +++ b/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 @@ -137,8 +138,15 @@ def _per_database_engines(self) -> Iterator[Engine]: 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] diff --git a/ingestion/src/metadata/ingestion/source/database/mssql/utils.py b/ingestion/src/metadata/ingestion/source/database/mssql/utils.py index 4afdb3cb6ab9..9cc98b0fa450 100644 --- a/ingestion/src/metadata/ingestion/source/database/mssql/utils.py +++ b/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": diff --git a/ingestion/tests/unit/topology/database/test_mssql.py b/ingestion/tests/unit/topology/database/test_mssql.py index 27eed74df40a..dcd540187505 100644 --- a/ingestion/tests/unit/topology/database/test_mssql.py +++ b/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([]) From f990caca886a037ef0965f7f81a834fdcedd9afe Mon Sep 17 00:00:00 2001 From: Akash Verma Date: Fri, 21 Aug 2026 18:54:45 +0530 Subject: [PATCH 5/5] fix(mssql): suppress pre-existing TopologyContext pyright false-positive The try/except wrap added in 48d2a5134 re-indented the context.get().database / .database_schema accesses in get_stored_procedures(), shifting their column position enough to break the basedpyright baseline's match and surface them as new CI errors. The underlying reportAttributeAccessIssue is a longstanding, codebase- wide false positive (TopologyContext fields are injected dynamically at runtime via create_model()); suppress inline the same way snowflake/metadata.py already does for the identical case. --- .../src/metadata/ingestion/source/database/mssql/metadata.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/ingestion/src/metadata/ingestion/source/database/mssql/metadata.py b/ingestion/src/metadata/ingestion/source/database/mssql/metadata.py index 44f4f7c017bd..25a764aedafc 100644 --- a/ingestion/src/metadata/ingestion/source/database/mssql/metadata.py +++ b/ingestion/src/metadata/ingestion/source/database/mssql/metadata.py @@ -252,13 +252,13 @@ def get_database_names(self) -> Iterable[str]: def get_stored_procedures(self) -> Iterable[MssqlStoredProcedure]: """List Snowflake stored procedures""" if self.source_config.includeStoredProcedures: - 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, + database_name=self.context.get().database, # pyright: ignore[reportAttributeAccessIssue] schema_name=schema_name, ) )