Backport superset + cross-catalog lineage fixes onto 1.12.11 + Looker UI base env var

ingestion/pyproject.toml

@@ -6,7 +6,7 @@ build-backend = "setuptools.build_meta"
 # since it helps us organize and isolate version management
 [project]
 name = "openmetadata-ingestion"
-version = "1.12.11.0"
+version = "1.12.11.0+superset.chart.bridge.3"
 dynamic = ["readme", "dependencies", "optional-dependencies"]
 authors = [
     { name = "OpenMetadata Committers" }

ingestion/src/metadata/ingestion/lineage/sql_lineage.py

@@ -608,6 +608,38 @@ def get_table_entities_from_query(
         if table_entities:
             return table_entities
 
+    # Cross-catalog fallback: the database resolved from the query/context may not
+    # match the database the source table was ingested under. This happens when the
+    # same physical data is exposed through different catalogs across services, so a
+    # view in one catalog references a table that another service ingested under a
+    # different database name. Searching with the view's database never matches the
+    # table under a different database name, so no lineage edge is created even with
+    # crossDatabaseServiceNames configured. Retry with database=None so the FQN search
+    # wildcards the database (`<service>.*.<schema>.<table>`) and resolves the table
+    # across the configured cross-database services. This is scoped to `service_names`,
+    # which only holds the current service plus the explicitly allowed
+    # crossDatabaseServiceNames, so it cannot reach unrelated services.
+    table_entities = search_table_entities(
+        metadata=metadata,
+        service_names=service_names,
+        database=None,
+        database_schema=schema_query if schema_query else database_schema,
+        table=table,
+    )
+    if table_entities:
+        return table_entities
+
+    if schema_fallback:
+        table_entities = search_table_entities(
+            metadata=metadata,
+            service_names=service_names,
+            database=None,
+            database_schema=None,
+            table=table,
+        )
+        if table_entities:
+            return table_entities
+
     return None
 
 

ingestion/src/metadata/ingestion/sink/metadata_rest.py

@@ -453,7 +453,9 @@ def _flush_query_buffer(self) -> Either[Entity]:
 
         return self._record_query_flush_result(result)
 
-    def _record_query_flush_result(self, result: Optional[BulkOperationResult]) -> Either[Entity]:  # noqa: UP045
+    def _record_query_flush_result(
+        self, result: Optional[BulkOperationResult]
+    ) -> Either[Entity]:  # noqa: UP045
         """Record a query bulk response. Already-present queries are reported as warnings
         (not failures) so a lineage run is not marked failed over queries that lost no
         metadata. Any other failure is still recorded as a failure."""
@@ -468,7 +470,10 @@ def _record_query_flush_result(self, result: Optional[BulkOperationResult]) -> E
         for failed in result.failedRequest or []:
             query_ref = failed.request or "unknown"
             if is_duplicate_query_conflict(failed.message):
-                self.status.warning("Query", f"Skipped already-present query [{query_ref}]: {failed.message}")
+                self.status.warning(
+                    "Query",
+                    f"Skipped already-present query [{query_ref}]: {failed.message}",
+                )
             else:
                 failure = StackTraceError(
                     name="Query Buffer",
@@ -478,7 +483,9 @@ def _record_query_flush_result(self, result: Optional[BulkOperationResult]) -> E
                 self.status.failed(failure)
                 first_failure = first_failure or failure
 
-        return Either(left=first_failure) if first_failure else Either(right=result)  # pyright: ignore[reportCallIssue]
+        return (
+            Either(left=first_failure) if first_failure else Either(right=result)
+        )  # pyright: ignore[reportCallIssue]
 
     @_run_dispatch.register
     def patch_entity(self, record: PatchRequest) -> Either[Entity]:
@@ -566,14 +573,20 @@ def write_classification_and_tag(
     @_run_dispatch.register
     def write_lineage(self, add_lineage: AddLineageRequest) -> Either[Dict[str, Any]]:
         created_lineage = self.metadata.add_lineage(add_lineage, check_patch=True)
-        if created_lineage.get("error"):
+        if created_lineage and created_lineage.get("error"):
             return Either(
                 left=StackTraceError(
                     name="AddLineageRequestError", error=created_lineage["error"]
                 )
             )
 
-        return Either(right=created_lineage["entity"]["fullyQualifiedName"])
+        # The edge was written, but reading back the origin node's lineage can
+        # fail (e.g. 404 when its graph holds an edge to a deleted entity).
+        # Fall back to the request's from-entity so this is not a false error.
+        entity_fqn = (created_lineage or {}).get("entity", {}).get(
+            "fullyQualifiedName"
+        ) or model_str(add_lineage.edge.fromEntity.id)
+        return Either(right=entity_fqn)
 
     @_run_dispatch.register
     def write_override_lineage(

ingestion/src/metadata/ingestion/source/dashboard/looker/metadata.py

@@ -219,6 +219,22 @@ def __init__(
 
         self._added_lineage: Optional[Dict] = {}
 
+    @property
+    def _ui_base(self) -> str:
+        """
+        Base URL for human-facing "View in Looker" links.
+
+        `hostPort` is bound to the Looker API host (the SDK uses it as
+        LOOKERSDK_BASE_URL), which is not always the URL a user should open in
+        the browser. When the API host differs from the UI host, set the
+        LOOKER_UI_BASE env var to the UI base URL; otherwise we fall back to
+        hostPort so behavior is unchanged.
+        """
+        return clean_uri(
+            os.environ.get("LOOKER_UI_BASE")
+            or str(self.service_connection.hostPort)
+        )
+
     @classmethod
     def create(
         cls,
@@ -1368,7 +1384,7 @@ def yield_dashboard(
             # like LookML assets, but rather just organised in folders.
             project=self.get_project_name(dashboard_details),
             sourceUrl=SourceUrl(
-                f"{clean_uri(self.service_connection.hostPort)}/dashboards/{dashboard_details.id}"
+                f"{self._ui_base}/dashboards/{dashboard_details.id}"
             ),
             service=self.context.get().dashboard_service,
             owners=self.get_owner_ref(dashboard_details=dashboard_details),
@@ -1760,8 +1776,18 @@ def yield_dashboard_chart(
                     source_url = chart.query.share_url
                 elif getattr(chart.result_maker, "query", None) is not None:
                     source_url = chart.result_maker.query.share_url
+                elif chart.merge_result_id is not None:
+                    source_url = f"{self._ui_base}/merge?mid={chart.merge_result_id}"
                 else:
-                    source_url = f"{clean_uri(self.service_connection.hostPort)}/merge?mid={chart.merge_result_id}"
+                    # No query, no result-maker query, and no merge result:
+                    # a data-less Looker tile (text/markdown/decoration). Skip
+                    # instead of emitting a dead {host}/merge?mid=None link.
+                    logger.info(
+                        f"Skipping data-less Looker tile id={chart.id} "
+                        f"title={chart.title!r} type={chart.type}: no query, "
+                        "no result-maker query, no merge result"
+                    )
+                    continue
                 yield Either(
                     right=CreateChartRequest(
                         name=EntityName(chart.id),

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

@@ -117,6 +117,12 @@ def yield_dashboard(
                     )
                     for chart in self.context.get().charts or []
                 ],
+                # Force-clear Dashboard.dataModels by sending an empty list.
+                # See comment in SupersetDBSource.yield_dashboard for why
+                # we represent the DataModel<->Dashboard relationship via
+                # the DataModel -> Chart -> Dashboard lineage chain
+                # instead of the structural Dashboard.dataModels field.
+                dataModels=[],
                 service=FullyQualifiedEntityName(self.context.get().dashboard_service),
                 owners=self.get_owner_ref(dashboard_details=dashboard_details),
             )

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

@@ -144,6 +144,13 @@ def yield_dashboard(
                     )
                     for chart in self.context.get().charts or []
                 ],
+                # Force-clear Dashboard.dataModels by sending an empty list.
+                # The DataModel<->Dashboard relationship is represented via
+                # the DataModel -> Chart -> Dashboard lineage chain emitted
+                # in SupersetSourceMixin.yield_dashboard_lineage_details.
+                # Sending [] (instead of omitting the field) ensures any
+                # datamodel entries left over from prior runs are deleted.
+                dataModels=[],
                 service=FullyQualifiedEntityName(self.context.get().dashboard_service),
                 owners=self.get_owner_ref(dashboard_details=dashboard_details),
             )

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

@@ -19,6 +19,8 @@
 from collate_sqllineage.core.models import Table as LineageTable
 
 from metadata.generated.schema.api.lineage.addLineage import AddLineageRequest
+from metadata.generated.schema.entity.data.chart import Chart
+from metadata.generated.schema.entity.data.dashboard import Dashboard
 from metadata.generated.schema.entity.data.dashboardDataModel import DashboardDataModel
 from metadata.generated.schema.entity.data.table import Column, DataType, Table
 from metadata.generated.schema.entity.services.connections.dashboard.supersetConnection import (
@@ -145,8 +147,11 @@ def _get_charts_of_dashboard(
             raw_position_data = dashboard_details.position_json
 
             # For Superset 5.0.0+, position_json may be missing from list endpoint
-            # In this case, fetch individual dashboard details
-            if not raw_position_data and hasattr(self, "client"):
+            # In this case, fetch individual dashboard details. Only the API
+            # client supports this; in DB mode self.client is a SQLAlchemy Engine.
+            if not raw_position_data and hasattr(
+                getattr(self, "client", None), "fetch_dashboard"
+            ):
                 dashboard_response = self.client.fetch_dashboard(dashboard_details.id)
                 if dashboard_response and dashboard_response.result:
                     raw_position_data = dashboard_response.result.position_json
@@ -319,14 +324,36 @@ def _get_dashboard_data_model_entity(
             fqn=datamodel_fqn,
         )
 
+    def yield_datamodel_dashboard_lineage(
+        self,
+    ) -> Iterable[Either[AddLineageRequest]]:
+        """
+        Skip the base class' direct DataModel -> Dashboard lineage edge.
+        For Superset we bridge the chain through the Chart node so the graph
+        renders DataModel -> Chart -> Dashboard rather than DataModel ->
+        Dashboard alongside the chart list. The DataModel -> Chart edge is
+        emitted in yield_dashboard_lineage_details.
+        """
+        return
+        yield  # pragma: no cover  # noqa: F841  # mark this as a generator
+
     def yield_dashboard_lineage_details(
         self,
         dashboard_details: Union[FetchDashboard, DashboardResult],
         db_service_prefix: Optional[str] = None,
     ) -> Iterable[Either[AddLineageRequest]]:
         """
-        Get lineage between datamodel and table
+        Emit lineage edges Table -> DataModel -> Chart -> Dashboard for every
+        chart on this dashboard. Dashboard.charts (set in yield_dashboard)
+        is a structural ref only — the dashboard lineage graph traverses
+        through explicit lineage edges, so we also emit Chart -> Dashboard
+        here. The base class' direct DataModel -> Dashboard edge is
+        suppressed by the override of yield_datamodel_dashboard_lineage so
+        the chart node bridges the chain in the rendered graph.
         """
+        # Resolve the dashboard entity once per dashboard, not once per chart,
+        # to avoid an N+1 lookup against the metadata server.
+        dashboard_entity = self._get_dashboard_entity(dashboard_details)
         for chart_json in filter(
             None,
             [
@@ -350,6 +377,26 @@ def yield_dashboard_lineage_details(
                             from_entity=from_entity_table,
                             column_lineage=column_lineage,
                         )
+
+                    # DataModel -> Chart -> Dashboard bridge: emit BOTH edges
+                    # so the dashboard's lineage graph renders the chart
+                    # between the datamodel and the dashboard, instead of
+                    # the datamodel hanging off the dashboard directly.
+                    chart_entity = self._get_chart_entity(chart_json)
+                    if chart_entity is not None:
+                        dm_to_chart = self._get_add_lineage_request(
+                            to_entity=chart_entity,
+                            from_entity=to_entity,
+                        )
+                        if dm_to_chart is not None:
+                            yield dm_to_chart
+                        if dashboard_entity is not None:
+                            chart_to_dash = self._get_add_lineage_request(
+                                to_entity=dashboard_entity,
+                                from_entity=chart_entity,
+                            )
+                            if chart_to_dash is not None:
+                                yield chart_to_dash
             except Exception as exc:
                 yield Either(
                     left=StackTraceError(
@@ -362,6 +409,54 @@ def yield_dashboard_lineage_details(
                     )
                 )
 
+    def _get_dashboard_entity(self, dashboard_details) -> Optional[Dashboard]:
+        """
+        Look up the Dashboard entity created earlier so we can emit a
+        Chart -> Dashboard lineage edge.
+        """
+        dashboard_id = getattr(dashboard_details, "id", None)
+        if dashboard_id is None:
+            return None
+        try:
+            dashboard_fqn = fqn.build(
+                self.metadata,
+                entity_type=Dashboard,
+                service_name=self.context.get().dashboard_service,
+                dashboard_name=str(dashboard_id),
+            )
+            return self.metadata.get_by_name(entity=Dashboard, fqn=dashboard_fqn)
+        except Exception as exc:  # pylint: disable=broad-except
+            logger.warning(
+                "Failed to resolve dashboard entity for dashboard_id=%s: %s",
+                dashboard_id,
+                exc,
+            )
+            return None
+
+    def _get_chart_entity(self, chart_json) -> Optional[Chart]:
+        """
+        Look up the Chart entity created earlier in this pipeline so we can
+        emit a DataModel -> Chart lineage edge.
+        """
+        chart_id = getattr(chart_json, "id", None)
+        if chart_id is None:
+            return None
+        try:
+            chart_fqn = fqn.build(
+                self.metadata,
+                entity_type=Chart,
+                service_name=self.context.get().dashboard_service,
+                chart_name=str(chart_id),
+            )
+            return self.metadata.get_by_name(entity=Chart, fqn=chart_fqn)
+        except Exception as exc:  # pylint: disable=broad-except
+            logger.warning(
+                "Failed to resolve chart entity for chart_id=%s: %s",
+                chart_id,
+                exc,
+            )
+            return None
+
     def _get_datamodel(
         self, datamodel: Union[SupersetDatasource, FetchChart]
     ) -> Optional[DashboardDataModel]:

ingestion/tests/unit/test_sink_lineage_none_response.py

@@ -0,0 +1,69 @@
+#  Copyright 2026 Collate
+#  Licensed under the Collate Community License, Version 1.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#  https://github.com/open-metadata/OpenMetadata/blob/main/ingestion/LICENSE
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+"""
+MetadataRestSink.write_lineage must not crash when add_lineage returns None,
+which happens when the edge is written but reading back the origin node's
+lineage fails (e.g. 404 because its graph references a deleted entity).
+"""
+from unittest.mock import MagicMock
+
+from metadata.generated.schema.api.lineage.addLineage import AddLineageRequest
+from metadata.generated.schema.type.entityLineage import EntitiesEdge
+from metadata.generated.schema.type.entityReference import EntityReference
+from metadata.ingestion.sink.metadata_rest import (
+    MetadataRestSink,
+    MetadataRestSinkConfig,
+)
+
+FROM_ID = "ab8f49d7-afba-4164-ba06-a0dc6300c5d2"
+TO_ID = "615ff273-1180-4fc2-8682-b5e4a646a6fe"
+
+LINEAGE_REQUEST = AddLineageRequest(
+    edge=EntitiesEdge(
+        fromEntity=EntityReference(id=FROM_ID, type="dashboardDataModel"),
+        toEntity=EntityReference(id=TO_ID, type="chart"),
+    )
+)
+
+
+def _sink_with_add_lineage_returning(value) -> MetadataRestSink:
+    metadata = MagicMock()
+    metadata.add_lineage.return_value = value
+    return MetadataRestSink(config=MetadataRestSinkConfig(), metadata=metadata)
+
+
+def test_write_lineage_handles_none_response():
+    sink = _sink_with_add_lineage_returning(None)
+
+    result = sink.write_lineage(LINEAGE_REQUEST)
+
+    assert result.left is None
+    assert result.right == FROM_ID
+
+
+def test_write_lineage_error_response_is_reported():
+    sink = _sink_with_add_lineage_returning({"error": "Error 400 trying to PUT"})
+
+    result = sink.write_lineage(LINEAGE_REQUEST)
+
+    assert result.left is not None
+    assert "Error 400" in result.left.error
+
+
+def test_write_lineage_success_returns_entity_fqn():
+    sink = _sink_with_add_lineage_returning(
+        {"entity": {"fullyQualifiedName": "service.model_1"}}
+    )
+
+    result = sink.write_lineage(LINEAGE_REQUEST)
+
+    assert result.left is None
+    assert result.right == "service.model_1"