fix(ingest): multi-database ingestion config supported by odbc connections

Per the docs if the database field is omitted from the config then all databases are considered for ingestion.
This was not the case as the metadata ingested into a datahub instance contained only the initial connection database contents

Author: Santos, Tyler (Boston)

metadata-ingestion/src/datahub/ingestion/source/sql/mssql/source.py

@@ -160,7 +160,10 @@ def get_sql_alchemy_url(
             uri_opts=uri_opts,
         )
         if self.use_odbc:
-            uri = f"{uri}?{urllib.parse.urlencode(self.uri_args)}"
+            uri_args = ({k:v for k,v in self.uri_args.items() if k.lower() != 'database'}
+                       if current_db else
+                       self.uri_args)
+            uri = f"{uri}?{urllib.parse.urlencode(uri_args)}"
         return uri
 
     @property
@@ -939,6 +942,24 @@ def construct_flow_workunits(
                 aspect=data_flow.as_container_aspect,
             ).as_workunit()
 
+    def _database_names_from_engine(self, engine):
+        """
+        Helper method to get database names from the engine.
+        This is used to fetch the list of databases in the SQL Server instance.
+        """
+        with engine.begin() as conn:
+            databases = conn.execute(
+                "SELECT name FROM master.sys.databases WHERE name NOT IN \
+                  ('master', 'model', 'msdb', 'tempdb', 'Resource', \
+                       'distribution' , 'reportserver', 'reportservertempdb'); "
+            ).fetchall()
+        return [db["name"] for db in databases]
+    
+    def _inspector_for_database(self, db_name: str) -> Inspector:
+        url = self.config.get_sql_alchemy_url(current_db=db_name)
+        engine = create_engine(url, **self.config.options)
+        return inspect(engine)
+
     def get_inspectors(self) -> Iterable[Inspector]:
         # This method can be overridden in the case that you want to dynamically
         # run on multiple databases.
@@ -950,19 +971,10 @@ def get_inspectors(self) -> Iterable[Inspector]:
             inspector = inspect(engine)
             yield inspector
         else:
-            with engine.begin() as conn:
-                databases = conn.execute(
-                    "SELECT name FROM master.sys.databases WHERE name NOT IN \
-                  ('master', 'model', 'msdb', 'tempdb', 'Resource', \
-                       'distribution' , 'reportserver', 'reportservertempdb'); "
-                ).fetchall()
-
-            for db in databases:
-                if self.config.database_pattern.allowed(db["name"]):
-                    url = self.config.get_sql_alchemy_url(current_db=db["name"])
-                    engine = create_engine(url, **self.config.options)
-                    inspector = inspect(engine)
-                    self.current_database = db["name"]
+            for db_name in self._database_names_from_engine(engine):
+                if self.config.database_pattern.allowed(db_name):
+                    inspector = self._inspector_for_database(db_name)
+                    self.current_database = db_name
                     yield inspector
 
     def get_identifier(

metadata-ingestion/tests/unit/test_mssql.py

@@ -299,3 +299,36 @@ def test_stored_procedure_vs_direct_query_compatibility(mssql_source):
     # Verify database_name is properly set
     assert sp_step["database_name"] == "test_db"
     assert direct_step["database_name"] == "test_db"
+
+
+def test_get_sql_alchemy_url_ignores_config_db_when_override_is_provided():
+    """Test that the database name is ignored when an override is provided"""
+    override_db = "override_db"
+    config_defined_db = "config_defined_connection_db"
+
+    config = SQLServerConfig(
+        host_port="localhost:1433",
+        username="test",
+        password="test",
+        use_odbc=True,
+        uri_args={'database': config_defined_db, 'driver': 'some_driver_value'},
+    )
+
+    # Mock to avoid DB connections
+    with patch("datahub.ingestion.source.sql.sql_common.SQLAlchemySource.__init__"), \
+         patch("datahub.ingestion.source.sql.mssql.source.SQLServerSource._database_names_from_engine") as mock_db_names, \
+         patch("datahub.ingestion.source.sql.mssql.source.SQLServerSource._inspector_for_database"):
+        mock_db_names.return_value = [override_db]
+        source = SQLServerSource(config, MagicMock())
+
+        # Call the method with the override
+        default_connection = source.config.get_sql_alchemy_url()
+
+        # assert that when not overridden, the default connection uses the config defined database
+        assert config_defined_db in default_connection
+        assert override_db not in default_connection
+
+        # assert that when overridden, the override database is used
+        override_db_connection = source.config.get_sql_alchemy_url(current_db=override_db)
+        assert override_db in override_db_connection
+        assert config_defined_db not in override_db_connection