fix(starrocks): query INFORMATION_SCHEMA for table comments (#26692)

StarRocks DDL places COMMENT on a separate line, causing SQLAlchemy's
MySQL parser to silently drop table descriptions. Override
get_table_description() to fetch TABLE_COMMENT directly from
INFORMATION_SCHEMA.TABLES instead.

Author: yash-agarwa-l

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

@@ -41,6 +41,7 @@
     STARROCKS_GET_TABLE_NAMES,
     STARROCKS_PARTITION_DETAILS,
     STARROCKS_SHOW_FULL_COLUMNS,
+    STARROCKS_TABLE_COMMENTS,
 )
 from metadata.utils.logger import ingestion_logger
 from metadata.utils.ssl_manager import SSLManager, check_ssl_and_init
@@ -269,26 +270,26 @@ def query_view_names_and_types(
 
         return tables
 
-    @staticmethod
     def get_table_description(
-        schema_name: str, table_name: str, inspector: Inspector
+        self, schema_name: str, table_name: str, inspector: Inspector
     ) -> Optional[str]:
-        description = None
+        # Fixes #26692: StarRocks formats SHOW CREATE TABLE with COMMENT on a
+        # separate line, which SQLAlchemy's MySQL dialect parser cannot parse.
+        # Query INFORMATION_SCHEMA.TABLES directly to get the table comment.
         try:
-            table_info: dict = inspector.get_table_comment(table_name, schema_name)
+            result = self.connection.execute(
+                sql.text(STARROCKS_TABLE_COMMENTS),
+                {"schema": schema_name, "table_name": table_name},
+            )
+            row = result.first()
+            if row:
+                return row[0] or 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}"
             )
-        else:
-            description = table_info.get("text")
-
-        if description is None:
-            return None
-        if isinstance(description, (list, tuple)) and len(description) > 0:
-            return description[0]
-        return description
+        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 MagicMock, patch
 
 from metadata.generated.schema.metadataIngestion.workflow import (
     OpenMetadataWorkflowConfig,
@@ -135,3 +135,80 @@ def test_iceberg_relkind_mapping(self):
         from metadata.ingestion.source.database.starrocks.metadata import RELKIND_MAP
 
         assert RELKIND_MAP["ICEBERG"] == TableType.Iceberg
+
+
+class TestStarRocksGetTableDescription(TestCase):
+    """Tests for get_table_description querying INFORMATION_SCHEMA directly (fixes #26692)"""
+
+    @patch(
+        "metadata.ingestion.source.database.common_db_source.CommonDbSourceService.test_connection"
+    )
+    def setUp(self, test_connection):
+        test_connection.return_value = False
+        self.config = OpenMetadataWorkflowConfig.model_validate(mock_starrocks_config)
+        self.source = StarRocksSource.create(
+            mock_starrocks_config["source"],
+            self.config.workflowConfig.openMetadataServerConfig,
+        )
+
+    @patch(
+        "metadata.ingestion.source.database.common_db_source.CommonDbSourceService.connection"
+    )
+    def test_returns_table_comment(self, mock_connection):
+        mock_result = MagicMock()
+        mock_result.first.return_value = ("审计日志表",)
+        mock_connection.execute.return_value = mock_result
+        self.source.connection = mock_connection
+
+        description = self.source.get_table_description(
+            schema_name="test_db",
+            table_name="audit_tbl",
+            inspector=MagicMock(),
+        )
+        assert description == "审计日志表"
+
+    @patch(
+        "metadata.ingestion.source.database.common_db_source.CommonDbSourceService.connection"
+    )
+    def test_returns_none_for_empty_comment(self, mock_connection):
+        mock_result = MagicMock()
+        mock_result.first.return_value = ("",)
+        mock_connection.execute.return_value = mock_result
+        self.source.connection = mock_connection
+
+        description = self.source.get_table_description(
+            schema_name="test_db",
+            table_name="no_comment_tbl",
+            inspector=MagicMock(),
+        )
+        assert description is None
+
+    @patch(
+        "metadata.ingestion.source.database.common_db_source.CommonDbSourceService.connection"
+    )
+    def test_returns_none_when_no_row(self, mock_connection):
+        mock_result = MagicMock()
+        mock_result.first.return_value = None
+        mock_connection.execute.return_value = mock_result
+        self.source.connection = mock_connection
+
+        description = self.source.get_table_description(
+            schema_name="test_db",
+            table_name="missing_tbl",
+            inspector=MagicMock(),
+        )
+        assert description is None
+
+    @patch(
+        "metadata.ingestion.source.database.common_db_source.CommonDbSourceService.connection"
+    )
+    def test_returns_none_on_exception(self, mock_connection):
+        mock_connection.execute.side_effect = Exception("connection error")
+        self.source.connection = mock_connection
+
+        description = self.source.get_table_description(
+            schema_name="test_db",
+            table_name="error_tbl",
+            inspector=MagicMock(),
+        )
+        assert description is None