fix(ingestion): release Engine resources on database switch (#27625) (#27627)

* fix(ingestion): release Engine resources on database switch (#27625)

* fix(ingestion): release engine resources on database switch

Close fairies in _connection_map and clear _inspector_map before
engine.dispose() in CommonDbSourceService.set_inspector/close. Dispose
alone does not free Inspector.info_cache or release checked-out
ConnectionFairies, leaving the old engine GC-pinned across DB switches
and triggering _finalize_fairy RecursionError at interpreter shutdown.

Eagerly fetch multi-DB name queries (MultiDBSource._execute_database_query
and SnowflakeSource.get_database_names_raw) so the cursor closes before
the caller invokes set_inspector, which disposes the engine the cursor
was bound to.

Also rebind scoped_session to the new engine so it doesn't keep the
disposed one alive via sessionmaker.bind.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* py format

* fix(ingestion): address PR review feedback from gitar-bot and Copilot

- Set self.engine = None after dispose in _release_engine (gitar-bot):
  prevents close() from leaving a dangling disposed-engine reference
  that would produce a confusing pool error on accidental later access.

- _FakeSource now has close() and is wrapped in a fixture that cleans
  up its checked-out connection (Copilot #1): avoids resource warnings
  and an interfering fairy across test teardown.

- Rewrite test_generator_survives_engine_dispose_mid_iteration as
  test_generator_survives_connection_close_mid_iteration (Copilot #2):
  Engine.dispose() does not close checked-out connections, so the old
  test did not reproduce what _release_engine actually does. The real
  regression is the explicit conn.close() on the fairy in
  _connection_map before dispose. The new test closes the connection
  mid-iteration, which is what fetchall() needs to survive.

- Switch the query in _FakeSource.get_database_names_raw and the
  seeded INSERT assertions to the TEXT name column (Copilot #3):
  _execute_database_query is typed Iterable[str]; testing on integer
  ids obscured the actual contract.

- Update test_disposes_pool to assert surrogate.engine is None after
  release (follows from the new self.engine = None behavior) and
  verify the original pool's checkedout() is 0.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* fix(ingestion): keep connection_obj in sync with engine across DB switches

self.connection_obj is set once in __init__ to the initial engine and
never updated. After set_inspector rebuilds self.engine, connection_obj
still points at the disposed original engine — pinning its dialect and
compiled_cache alive for the source's lifetime.

Rebind connection_obj when creating the new engine in set_inspector,
and clear it in _release_engine so close() leaves nothing dangling.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: ulixius9

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

@@ -62,7 +62,6 @@
 from metadata.ingestion.models.patch_request import PatchedEntity, PatchRequest
 from metadata.ingestion.ometa.ometa_api import OpenMetadata
 from metadata.ingestion.source.connections import get_connection
-from metadata.ingestion.source.connections_utils import kill_active_connections
 from metadata.ingestion.source.database.database_service import DatabaseServiceSource
 from metadata.ingestion.source.database.sql_column_handler import SqlColumnHandlerMixin
 from metadata.ingestion.source.database.sqlalchemy_source import SqlAlchemySource
@@ -152,15 +151,42 @@ def set_inspector(self, database_name: str) -> None:
         :param database_name: new database to set
         """
 
-        kill_active_connections(self.engine)
+        self._release_engine()
         logger.info(f"Ingesting from database: {database_name}")
 
         new_service_connection = deepcopy(self.service_connection)
         new_service_connection.database = database_name
         self.engine = get_connection(new_service_connection)
+        self.session = create_and_bind_thread_safe_session(self.engine)
+        self.connection_obj = self.engine
 
-        self._connection_map = {}  # Lazy init as well
+    def _release_engine(self) -> None:
+        # Close fairies first so _ConnectionRecord drops its pool reference;
+        # dispose alone leaves them orphaned and causes _finalize_fairy
+        # RecursionErrors at GC time. Clearing _inspector_map is what
+        # actually frees Inspector.info_cache — dispose() does not.
+        if getattr(self, "engine", None) is None:
+            return
+        for conn in self._connection_map.values():
+            try:
+                conn.close()
+            except Exception:  # pylint: disable=broad-except
+                logger.debug("Connection already closed", exc_info=True)
+        self._connection_map = {}
         self._inspector_map = {}
+        session = getattr(self, "session", None)
+        if session is not None:
+            try:
+                session.remove()
+            except Exception:  # pylint: disable=broad-except
+                logger.debug("Session cleanup failed", exc_info=True)
+            self.session = None
+        try:
+            self.engine.dispose()
+        except Exception as exc:  # pylint: disable=broad-except
+            logger.warning(f"Failed to dispose engine: {exc}")
+        self.engine = None
+        self.connection_obj = None
 
     def get_database_names(self) -> Iterable[str]:
         """
@@ -781,14 +807,10 @@ def inspector(self) -> Inspector:
         return self._inspector_map[thread_id]
 
     def close(self):
-        if self.connection is not None:
-            self.connection.close()
-        for connection in self._connection_map.values():
-            connection.close()
+        self._release_engine()
         if hasattr(self, "ssl_manager") and self.ssl_manager:
             self.ssl_manager = cast(SSLManager, self.ssl_manager)
             self.ssl_manager.cleanup_temp_files()
-        self.engine.dispose()
 
     def fetch_table_tags(
         self,

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

@@ -31,7 +31,8 @@ def get_database_names_raw(self) -> Iterable[str]:
         """
 
     def _execute_database_query(self, query: str) -> Iterable[str]:
-        results = self.connection.execute(query)  # pylint: disable=no-member
+        results = self.connection.execute(query).fetchall()  # pylint: disable=no-member
+
         for res in results:
             row = list(res)
             yield row[0]

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

@@ -358,7 +358,7 @@ def get_configured_database(self) -> Optional[str]:
         return self.service_connection.database
 
     def get_database_names_raw(self) -> Iterable[str]:
-        results = self.connection.execute(SNOWFLAKE_GET_DATABASES)
+        results = self.connection.execute(SNOWFLAKE_GET_DATABASES).fetchall()
         for res in results:
             row = list(res)
             yield row[1]