fix(superset): API datamodel fields + SQL lineage, cache datasource/database fetches

ingestion/src/metadata/ingestion/source/dashboard/superset/api_source.py

@@ -39,6 +39,7 @@
     ChartResult,
     DashboardResult,
     FetchChart,
+    SupersetDatasource,
 )
 from metadata.utils import fqn
 from metadata.utils.filters import filter_by_datamodel
@@ -125,11 +126,71 @@ def yield_dashboard(self, dashboard_details: DashboardResult) -> Iterable[Either
                 )
             )
 
-    def _get_datasource_fqn_for_lineage(self, chart_json: ChartResult, db_service_prefix: Optional[str]):  # noqa: UP045
+    def _get_datasource_fqn_for_lineage(self, chart_json: FetchChart | ChartResult, db_service_prefix: str | None):
+        # A SQL-parsed source table carries a table_name; a raw chart carries only a datasource_id.
+        if getattr(chart_json, "table_name", None):
+            return self._get_source_table_fqn(chart_json, db_service_prefix)  # pyright: ignore[reportArgumentType]
         return (
             self._get_datasource_fqn(chart_json.datasource_id, db_service_prefix) if chart_json.datasource_id else None
         )
 
+    def _get_input_tables(self, chart: FetchChart | ChartResult):
+        # Virtual (SQL) datasets: parse the dataset SQL to reach real source tables.
+        datasource_id = getattr(chart, "datasource_id", None)
+        datasource = self.client.fetch_datasource(datasource_id) if datasource_id else None
+        result = datasource.result if datasource else None
+        if result and result.sql:
+            enriched = FetchChart(
+                sql=result.sql,
+                schema=result.table_schema,
+                datasource_id=datasource_id,
+            )
+            input_tables = self._parse_lineage_from_dataset_sql(enriched)
+        else:
+            input_tables = super()._get_input_tables(chart)  # pyright: ignore[reportArgumentType]
+        return input_tables
+
+    def _resolve_lineage_database_name(
+        self, datasource_json: SupersetDatasource, db_service_name: str | None
+    ) -> str | None:
+        database = datasource_json.result.database if datasource_json.result else None
+        database_id = database.id if database else None
+        default_database_name = None
+        if database_id is not None:
+            database_json = self.client.fetch_database(database_id)
+            default_database_name = (
+                database_json.result.parameters.database if database_json.result.parameters else None
+            )
+        db_service_entity = (
+            self.metadata.get_by_name(entity=DatabaseService, fqn=db_service_name) if db_service_name else None
+        )
+        if db_service_entity is None:
+            return default_database_name
+        return get_database_name_for_lineage(db_service_entity, default_database_name)
+
+    def _get_source_table_fqn(self, chart_json: FetchChart, db_service_prefix: str | None) -> str | None:
+        try:
+            (
+                db_service_name,
+                prefix_database_name,
+                prefix_schema_name,
+                prefix_table_name,
+            ) = self.parse_db_service_prefix(db_service_prefix)
+            database_name = None
+            if db_service_name and chart_json.datasource_id:
+                datasource_json = self.client.fetch_datasource(chart_json.datasource_id)
+                database_name = self._resolve_lineage_database_name(datasource_json, db_service_name)
+            return build_es_fqn_search_string(
+                database_name=prefix_database_name or database_name or "*",
+                schema_name=prefix_schema_name or chart_json.table_schema,
+                service_name=db_service_name or "*",
+                table_name=prefix_table_name or chart_json.table_name,
+            )
+        except Exception as err:
+            logger.debug(traceback.format_exc())
+            logger.warning(f"Failed to build source table fqn for [{getattr(chart_json, 'table_name', None)}]: {err}")
+        return None
+
     def yield_dashboard_chart(self, dashboard_details: DashboardResult) -> Iterable[Either[CreateChartRequest]]:
         """Method to fetch charts linked to dashboard"""
         for chart_id in self._get_charts_of_dashboard(dashboard_details):
@@ -169,12 +230,7 @@ def _get_datasource_fqn(self, datasource_id: str, db_service_prefix: Optional[st
             if datasource_json:
                 database_name = None
                 if db_service_prefix:
-                    database_json = self.client.fetch_database(datasource_json.result.database.id)
-                    default_database_name = (
-                        database_json.result.parameters.database if database_json.result.parameters else None
-                    )
-                    db_service_entity = self.metadata.get_by_name(entity=DatabaseService, fqn=db_service_name)
-                    database_name = get_database_name_for_lineage(db_service_entity, default_database_name)
+                    database_name = self._resolve_lineage_database_name(datasource_json, db_service_name)
 
                     if prefix_database_name and database_name and prefix_database_name.lower() != database_name.lower():
                         logger.debug(f"Database {database_name} does not match prefix {prefix_database_name}")

ingestion/src/metadata/ingestion/source/dashboard/superset/client.py

@@ -15,6 +15,8 @@
 import json
 import traceback
 
+from cachetools import LRUCache
+
 from metadata.generated.schema.entity.services.connections.dashboard.supersetConnection import (
     SupersetConnection,
 )
@@ -102,6 +104,8 @@ def __init__(self, config: SupersetConnection):
             verify=get_verify_ssl(config.connection.sslConfig),
         )
         self.client = TrackedREST(client_config, source_name="superset")
+        self._datasource_cache = LRUCache(maxsize=512)
+        self._database_cache = LRUCache(maxsize=512)
 
     def get_dashboard_count(self) -> int:
         resp_dashboards = self.client.get("/dashboard/?q=(page:0,page_size:1)")
@@ -223,12 +227,16 @@ def fetch_datasource(self, datasource_id: str) -> SupersetDatasource:
         Returns:
             requests.Response
         """
-
+        # Cache only real responses; a transient failure must stay retryable, not
+        # poison the id with an empty result for the rest of the run.
+        if datasource_id in self._datasource_cache:
+            return self._datasource_cache[datasource_id]
         try:
             datasource_response = self.client.get(f"/dataset/{datasource_id}")
             if datasource_response:
                 datasource_list = SupersetDatasource(**datasource_response)
-                return datasource_list  # noqa: RET504
+                self._datasource_cache[datasource_id] = datasource_list
+                return datasource_list
         except Exception:
             logger.debug(traceback.format_exc())
             logger.warning("Failed to fetch the datasource list")
@@ -244,12 +252,15 @@ def fetch_database(self, database_id: str) -> ListDatabaseResult:
         Returns:
             requests.Response
         """
-
+        # Cache only real responses; keep transient failures retryable.
+        if database_id in self._database_cache:
+            return self._database_cache[database_id]
         try:
             database_response = self.client.get(f"/database/{database_id}")
             if database_response:
                 database_list = ListDatabaseResult(**database_response)
-                return database_list  # noqa: RET504
+                self._database_cache[database_id] = database_list
+                return database_list
         except Exception:
             logger.debug(traceback.format_exc())
             logger.warning("Failed to fetch the database list")

ingestion/src/metadata/ingestion/source/dashboard/superset/mixin.py

@@ -232,6 +232,7 @@ def _parse_lineage_from_dataset_sql(self, chart_json: FetchChart) -> List[Tuple[
                         table_name=table_name,
                         schema=table_schema,
                         sqlalchemy_uri=chart_json.sqlalchemy_uri,
+                        datasource_id=chart_json.datasource_id,
                     ),
                     column_mapping,
                 )

ingestion/tests/integration/superset/test_superset.py

@@ -64,6 +64,7 @@
 from metadata.ingestion.source.dashboard.superset.db_source import SupersetDBSource
 from metadata.ingestion.source.dashboard.superset.metadata import SupersetSource
 from metadata.ingestion.source.dashboard.superset.models import (
+    ChartResult,
     DatabaseResult,
     DataSourceResult,
     FetchChart,
@@ -664,6 +665,73 @@ def test_datamodel_fields_api(self):
         assert str(data_model.sourceUrl.root).endswith("/tablemodelview/edit/99")
         assert data_model.columns[0].name.root == "Population"
 
+    def test_api_get_input_tables_parses_dataset_sql(self):
+        """
+        API _get_input_tables parses the virtual dataset SQL to reach the real source tables
+        """
+        with patch.object(
+            self.superset_api.client,
+            "fetch_datasource",
+            return_value=MOCK_DATASOURCE_RESPONSE_WITH_SQL,
+        ):
+            result = self.superset_api._get_input_tables(ChartResult(datasource_id=99))
+        source_tables = [fetch_chart.table_name for fetch_chart, _ in result]
+        self.assertIn("sample_table", source_tables)
+
+    def test_api_get_source_table_fqn_uses_parsed_table(self):
+        """
+        SQL-parsed source table fqn uses the parsed table name, not the datasource's own table
+        """
+        with (
+            patch.object(OpenMetadata, "get_by_name", return_value=MOCK_DB_POSTGRES_SERVICE),
+            patch.object(self.superset_api.client, "fetch_datasource", return_value=MOCK_DATASOURCE_RESPONSE),
+            patch.object(self.superset_api.client, "fetch_database", return_value=MOCK_DATABASE_RESPONSE),
+        ):
+            fqn = self.superset_api._get_source_table_fqn(  # pylint: disable=protected-access
+                FetchChart(table_name="orders", schema="main", datasource_id=1),
+                MOCK_DB_POSTGRES_SERVICE.name.root,
+            )
+        self.assertEqual(fqn, "test_postgres.*.main.orders")
+
+    def test_api_fetch_datasource_is_cached(self):
+        """
+        Repeated fetch_datasource for the same id resolves from cache, hitting the network once
+        """
+        with patch.object(self.superset_api.client.client, "get", return_value={"id": 1}) as mock_get:
+            self.superset_api.client.fetch_datasource(7)
+            self.superset_api.client.fetch_datasource(7)
+            self.superset_api.client.fetch_datasource(8)
+        self.assertEqual(mock_get.call_count, 2)
+
+    def test_api_fetch_datasource_failure_is_retryable(self):
+        """
+        A failed/empty fetch is not cached, so a later call for the same id retries instead of
+        being served the poisoned empty result
+        """
+        with patch.object(self.superset_api.client.client, "get", side_effect=[None, {"id": 5}]) as mock_get:
+            first = self.superset_api.client.fetch_datasource(5)
+            second = self.superset_api.client.fetch_datasource(5)
+        self.assertIsNone(first.id)
+        self.assertEqual(second.id, 5)
+        self.assertEqual(mock_get.call_count, 2)
+
+    def test_api_source_table_fqn_missing_db_service_does_not_crash(self):
+        """
+        When the db service prefix is not registered in OM, fqn resolution degrades gracefully
+        instead of raising on a None DatabaseService
+        """
+        with (
+            patch.object(OpenMetadata, "get_by_name", return_value=None),
+            patch.object(self.superset_api.client, "fetch_datasource", return_value=MOCK_DATASOURCE_RESPONSE),
+            patch.object(self.superset_api.client, "fetch_database", return_value=MOCK_DATABASE_RESPONSE),
+        ):
+            fqn = self.superset_api._get_source_table_fqn(  # pylint: disable=protected-access
+                FetchChart(table_name="orders", schema="main", datasource_id=1),
+                "missing_service",
+            )
+        self.assertIn("orders", fqn)
+        self.assertIn("missing_service", fqn)
+
     def test_is_table_to_table_lineage(self):
         table = Table(name="table_name", schema=Schema(name="schema_name"))