fix(superset): bridge DataModel <-> Dashboard lineage through the Chart node (1.12.9)

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.9.0"
+version = "1.12.9.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,36 @@ 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 table was ingested under -- e.g. a view in catalog
+    # `hiyu` (service trino_views_hiyu) referencing a table ingested as
+    # `trino_hiyu.iceberg.<schema>.<table>`. Retry with database=None so the FQN
+    # search wildcards the database (`<service>.*.<schema>.<table>`) and resolves
+    # the table across the configured cross-database services. Scoped to
+    # `service_names`, which already only contains the current service plus the
+    # explicitly-allowed crossDatabaseServiceNames, so this can't 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

@@ -463,6 +463,14 @@ 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 is None:
+            logger.warning(
+                "Lineage edge was written but the source entity lineage could not "
+                "be fetched back (commonly a dangling downstream reference to a "
+                f"deleted entity): {add_lineage.edge.fromEntity.type} "
+                f"{add_lineage.edge.fromEntity.id.root}"
+            )
+            return Either(right=None)
         if created_lineage.get("error"):
             return Either(
                 left=StackTraceError(

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

@@ -124,7 +124,11 @@ class DashboardServiceTopology(ServiceTopology):
             ),
         ],
         children=["bulk_data_model", "dashboard"],
-        post_process=["mark_dashboards_as_deleted", "mark_datamodels_as_deleted"],
+        post_process=[
+            "mark_dashboards_as_deleted",
+            "mark_datamodels_as_deleted",
+            "mark_charts_as_deleted",
+        ],
     )
     # Dashboard Services have very different approaches when
     # when dealing with data models. Tableau has the models
@@ -219,6 +223,7 @@ class DashboardServiceSource(TopologyRunnerMixin, Source, ABC):
     context = TopologyContextManager(topology)
     dashboard_source_state: Set = set()
     datamodel_source_state: Set = set()
+    chart_source_state: Set = set()
 
     @retry_with_docker_host()
     def __init__(
@@ -486,6 +491,20 @@ def mark_datamodels_as_deleted(self) -> Iterable[Either[DeleteEntity]]:
                 params={"service": self.context.get().dashboard_service},
             )
 
+    def mark_charts_as_deleted(self) -> Iterable[Either[DeleteEntity]]:
+        """
+        Method to mark the charts as deleted
+        """
+        if self.source_config.markDeletedCharts:
+            logger.info("Mark Deleted Charts set to True")
+            yield from delete_entity_from_source(
+                metadata=self.metadata,
+                entity_type=Chart,
+                entity_source_state=self.chart_source_state,
+                mark_deleted_entity=self.source_config.markDeletedCharts,
+                params={"service": self.context.get().dashboard_service},
+            )
+
     def get_owner_ref(  # pylint: disable=unused-argument, useless-return
         self, dashboard_details
     ) -> Optional[EntityReferenceList]:
@@ -525,6 +544,19 @@ def register_record_datamodel(
 
         self.datamodel_source_state.add(datamodel_fqn)
 
+    def register_record_chart(self, chart_request: CreateChartRequest) -> None:
+        """
+        Mark the chart record as scanned and update the chart_source_state
+        """
+        chart_fqn = fqn.build(
+            self.metadata,
+            entity_type=Chart,
+            service_name=chart_request.service.root,
+            chart_name=chart_request.name.root,
+        )
+
+        self.chart_source_state.add(chart_fqn)
+
     @staticmethod
     def _get_add_lineage_request(
         to_entity: Union[Dashboard, DashboardDataModel, Chart],

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

@@ -237,20 +237,18 @@ def yield_dashboard_chart(
                     self.status.filter(chart.name, "Chart Pattern not allowed")
                     continue
                 if chart.name:
-                    yield Either(
-                        right=CreateChartRequest(
-                            name=EntityName(str(chart_id)),
-                            description=(
-                                Markdown(chart.description)
-                                if chart.description
-                                else None
-                            ),
-                            displayName=chart.name,
-                            sourceUrl=SourceUrl(chart_url),
-                            service=self.context.get().dashboard_service,
-                            chartType=get_standard_chart_type(chart.metadata.chartType),
-                        )
+                    chart_request = CreateChartRequest(
+                        name=EntityName(str(chart_id)),
+                        description=(
+                            Markdown(chart.description) if chart.description else None
+                        ),
+                        displayName=chart.name,
+                        sourceUrl=SourceUrl(chart_url),
+                        service=self.context.get().dashboard_service,
+                        chartType=get_standard_chart_type(chart.metadata.chartType),
                     )
+                    yield Either(right=chart_request)
+                    self.register_record_chart(chart_request=chart_request)
             except Exception as exc:
                 name = chart.name if chart else ""
                 yield Either(

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

@@ -224,23 +224,23 @@ def yield_dashboard_chart(
                 # Map Grafana panel types to standard chart types
                 chart_type = self._map_panel_type_to_chart_type(panel.type)
 
-                yield Either(
-                    right=CreateChartRequest(
-                        name=EntityName(chart_name),
-                        displayName=chart_display_name,
-                        description=(
-                            Markdown(panel.description) if panel.description else None
-                        ),
-                        chartType=chart_type,
-                        service=FullyQualifiedEntityName(
-                            self.context.get().dashboard_service
-                        ),
-                        sourceUrl=SourceUrl(
-                            f"{clean_uri(self.service_connection.hostPort)}"
-                            f"{dashboard_details.meta.url}?viewPanel={panel.id}"
-                        ),
-                    )
+                chart_request = CreateChartRequest(
+                    name=EntityName(chart_name),
+                    displayName=chart_display_name,
+                    description=(
+                        Markdown(panel.description) if panel.description else None
+                    ),
+                    chartType=chart_type,
+                    service=FullyQualifiedEntityName(
+                        self.context.get().dashboard_service
+                    ),
+                    sourceUrl=SourceUrl(
+                        f"{clean_uri(self.service_connection.hostPort)}"
+                        f"{dashboard_details.meta.url}?viewPanel={panel.id}"
+                    ),
                 )
+                yield Either(right=chart_request)
+                self.register_record_chart(chart_request=chart_request)
             except Exception as exc:
                 yield Either(
                     left=StackTraceError(

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

@@ -176,18 +176,18 @@ def yield_dashboard_chart(
                 if filter_by_chart(self.source_config.chartFilterPattern, chart.name):
                     self.status.filter(chart.name, "Chart Pattern not allowed")
                     continue
-                yield Either(
-                    right=CreateChartRequest(
-                        name=EntityName(chart.uuid),
-                        displayName=chart.name,
-                        description=(
-                            Markdown(chart.description) if chart.description else None
-                        ),
-                        sourceUrl=SourceUrl(chart_url),
-                        service=self.context.get().dashboard_service,
-                        chartType=chart_type,
-                    )
+                chart_request = CreateChartRequest(
+                    name=EntityName(chart.uuid),
+                    displayName=chart.name,
+                    description=(
+                        Markdown(chart.description) if chart.description else None
+                    ),
+                    sourceUrl=SourceUrl(chart_url),
+                    service=self.context.get().dashboard_service,
+                    chartType=chart_type,
                 )
+                yield Either(right=chart_request)
+                self.register_record_chart(chart_request=chart_request)
                 self.status.scanned(chart.name)
             except Exception as exc:  # pylint: disable=broad-except
                 logger.debug(traceback.format_exc())

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

@@ -1762,16 +1762,16 @@ def yield_dashboard_chart(
                     source_url = chart.result_maker.query.share_url
                 else:
                     source_url = f"{clean_uri(self.service_connection.hostPort)}/merge?mid={chart.merge_result_id}"
-                yield Either(
-                    right=CreateChartRequest(
-                        name=EntityName(chart.id),
-                        displayName=chart.title or chart.id,
-                        description=Markdown(description) if description else None,
-                        chartType=get_standard_chart_type(chart.type).value,
-                        sourceUrl=SourceUrl(source_url),
-                        service=self.context.get().dashboard_service,
-                    )
+                chart_request = CreateChartRequest(
+                    name=EntityName(chart.id),
+                    displayName=chart.title or chart.id,
+                    description=Markdown(description) if description else None,
+                    chartType=get_standard_chart_type(chart.type).value,
+                    sourceUrl=SourceUrl(source_url),
+                    service=self.context.get().dashboard_service,
                 )
+                yield Either(right=chart_request)
+                self.register_record_chart(chart_request=chart_request)
 
             except Exception as exc:
                 yield Either(

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

@@ -271,16 +271,16 @@ def yield_dashboard_chart(
                 ):
                     self.status.filter(chart_details.name, "Chart Pattern not allowed")
                     continue
-                yield Either(
-                    right=CreateChartRequest(
-                        name=EntityName(chart_details.id),
-                        displayName=chart_details.name,
-                        description=chart_details.description,
-                        chartType=get_standard_chart_type(chart_details.display).value,
-                        sourceUrl=SourceUrl(chart_url),
-                        service=self.context.get().dashboard_service,
-                    )
+                chart_request = CreateChartRequest(
+                    name=EntityName(chart_details.id),
+                    displayName=chart_details.name,
+                    description=chart_details.description,
+                    chartType=get_standard_chart_type(chart_details.display).value,
+                    sourceUrl=SourceUrl(chart_url),
+                    service=self.context.get().dashboard_service,
                 )
+                yield Either(right=chart_request)
+                self.register_record_chart(chart_request=chart_request)
             except KeyError as exc:
                 yield Either(
                     left=StackTraceError(

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

@@ -323,16 +323,14 @@ def _yield_chart_from_visualization(
                     self.status.filter(chart.name, "Chart Pattern not allowed")
                     continue
 
-                yield Either(
-                    right=CreateChartRequest(
-                        name=f"{page.key}{chart.key}",
-                        displayName=chart.name,
-                        chartType=get_standard_chart_type(
-                            chart.visualizationType
-                        ).value,
-                        service=self.context.get().dashboard_service,
-                    )
+                chart_request = CreateChartRequest(
+                    name=f"{page.key}{chart.key}",
+                    displayName=chart.name,
+                    chartType=get_standard_chart_type(chart.visualizationType).value,
+                    service=self.context.get().dashboard_service,
                 )
+                yield Either(right=chart_request)
+                self.register_record_chart(chart_request=chart_request)
             except Exception as exc:
                 yield Either(
                     left=StackTraceError(

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

@@ -269,15 +269,15 @@ def yield_dashboard_chart(
                     chart_url = (
                         f"{clean_uri(self.service_connection.hostPort)}{chart_path}"
                     )
-                    yield Either(
-                        right=CreateChartRequest(
-                            name=EntityName(chart.get(client.TOKEN)),
-                            displayName=chart_name,
-                            chartType=ChartType.Other,
-                            sourceUrl=SourceUrl(chart_url),
-                            service=self.context.get().dashboard_service,
-                        )
+                    chart_request = CreateChartRequest(
+                        name=EntityName(chart.get(client.TOKEN)),
+                        displayName=chart_name,
+                        chartType=ChartType.Other,
+                        sourceUrl=SourceUrl(chart_url),
+                        service=self.context.get().dashboard_service,
                     )
+                    yield Either(right=chart_request)
+                    self.register_record_chart(chart_request=chart_request)
                 except Exception as exc:
                     name = chart_name if chart_name else ""
                     yield Either(