fix(starrocks): reuse shared connection for table comments (#27030)

Author: SaaiAravindhRaja

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

@@ -282,24 +282,20 @@ def query_view_names_and_types(
         return tables
 
     def get_table_description(
-        self, schema_name: str, table_name: str, inspector: Inspector
+        self, schema_name: str, table_name: str, _inspector: Inspector
     ) -> Optional[str]:
-        description = None
         try:
-            with self.engine.connect() as conn:
-                rows = conn.execute(
-                    sql.text(STARROCKS_TABLE_COMMENTS),
-                    {"table_name": table_name, "schema": schema_name},
-                )
-                for row in rows:
-                    description = row[0]
-                    break
+            row = self.connection.execute(
+                sql.text(STARROCKS_TABLE_COMMENTS),
+                {"table_name": table_name, "schema": schema_name},
+            ).fetchone()
+            return row[0] if row else None
         except Exception as exc:  # pylint: disable=broad-except
             logger.debug(traceback.format_exc())
             logger.warning(
                 f"Table description error for table [{schema_name}.{table_name}]: {exc}"
             )
-        return description or None
+        return None
 
     def _get_columns(self, table_name, schema=None):
         """Get column information and primary key columns of the specified table"""

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

@@ -14,7 +14,7 @@
 """
 
 from unittest import TestCase
-from unittest.mock import patch
+from unittest.mock import Mock, patch
 
 import pytest
 from sqlalchemy import types as sqltypes
@@ -114,6 +114,32 @@ def test_close_connection(self, engine, connection):
         connection.return_value = True
         self.starrocks_source.close()
 
+    def test_get_table_description_returns_comment(self):
+        mock_result = Mock()
+        mock_result.fetchone.return_value = ("table comment",)
+        mock_connection = Mock()
+        mock_connection.execute.return_value = mock_result
+        self.starrocks_source.connection = mock_connection
+
+        result = self.starrocks_source.get_table_description(
+            schema_name="public", table_name="my_table", _inspector=Mock()
+        )
+
+        self.assertEqual(result, "table comment")
+
+    def test_get_table_description_returns_none_when_missing(self):
+        mock_result = Mock()
+        mock_result.fetchone.return_value = None
+        mock_connection = Mock()
+        mock_connection.execute.return_value = mock_result
+        self.starrocks_source.connection = mock_connection
+
+        result = self.starrocks_source.get_table_description(
+            schema_name="public", table_name="my_table", _inspector=Mock()
+        )
+
+        self.assertIsNone(result)
+
 
 class StarRocksSSLUnitTest(TestCase):
     @patch(