From fe50e937456e6b5e9afa593c7d4818b8c2567963 Mon Sep 17 00:00:00 2001 From: Joao Amaral <7281460+joaopamaral@users.noreply.github.com> Date: Mon, 18 May 2026 15:58:38 -0300 Subject: [PATCH 1/7] fix(superset): bridge DataModel <-> Dashboard lineage through the Chart node MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit OpenMetadata's lineage graph traverses through explicit AddLineageRequest edges. For Superset the current behaviour was: - `Dashboard.dataModels` was never set, so the structural link from the dashboard to its datamodels was missing in the dashboard's Data Models panel. - `DashboardServiceSource.yield_datamodel_dashboard_lineage()` emitted a direct `DataModel -> Dashboard` lineage edge, so the lineage graph showed datamodels connected to the dashboard but skipped the chart entirely. Net result in production: the dashboard's lineage view showed `Table -> DataModel -> Dashboard` and the chart was invisible in the graph, even though it was a member of the dashboard. This change rewires the relationship so the chart is the bridge node: 1. `SupersetSourceMixin.yield_dashboard_lineage_details` now also emits `DataModel -> Chart` and `Chart -> Dashboard` edges for each chart on the dashboard. Combined with the existing `Table -> DataModel` edge the full chain in the lineage view becomes `Table -> DataModel -> Chart -> Dashboard`. 2. `SupersetSourceMixin.yield_datamodel_dashboard_lineage` is overridden to yield nothing, suppressing the base class' direct `DataModel -> Dashboard` edge so the datamodel doesn't render alongside the chart as two parallel paths into the dashboard. 3. `db_source.yield_dashboard` and `api_source.yield_dashboard` now send `Dashboard.dataModels=[]` (an empty list, not absent) on every PUT so any datamodel entries persisted from prior runs are deleted by the server. The relationship is fully represented by the lineage chain. Two helpers — `_get_chart_entity` and `_get_dashboard_entity` — resolve the Chart and Dashboard entities by FQN via `metadata.get_by_name`. If either isn't yet visible on the server (first run, before the entity POST has been committed), the bridge edges are skipped without crashing and the next run picks them up. Tests ----- Adds `tests/unit/topology/dashboard/test_superset_chart_lineage.py` with five Mockito-style unit tests: - `test_db_source_yields_empty_data_models` / `test_api_source_yields_empty_data_models`: Both sources produce `CreateDashboardRequest` with `dataModels=[]`. - `test_override_yields_no_edges`: the `yield_datamodel_dashboard_lineage` override produces zero edges. - `test_chart_bridge_edges_emitted`: when the chart and dashboard entities resolve, `yield_dashboard_lineage_details` emits both `DataModel -> Chart` and `Chart -> Dashboard` edges. - `test_no_chart_entity_skips_bridge_edges`: when the chart isn't yet resolvable on the server, both bridge edges are skipped gracefully. Three of the five tests fail on the unpatched code, exercising each of the three behaviour changes above. Manually verified in production (Superset metadata-DB connection mode) against a real Meltano dashboard: after this fix the dashboard's lineage graph renders `Table -> DataModel -> Chart -> Dashboard` with the chart visible as the bridge node, and the API confirms `Dashboard.dataModels` is cleared while all `DataModel -> Chart` and `Chart -> Dashboard` edges are present. Co-Authored-By: Claude Opus 4.7 (1M context) --- .../source/dashboard/superset/api_source.py | 6 + .../source/dashboard/superset/db_source.py | 7 + .../source/dashboard/superset/mixin.py | 92 +++++- .../dashboard/test_superset_chart_lineage.py | 263 ++++++++++++++++++ 4 files changed, 367 insertions(+), 1 deletion(-) create mode 100644 ingestion/tests/unit/topology/dashboard/test_superset_chart_lineage.py diff --git a/ingestion/src/metadata/ingestion/source/dashboard/superset/api_source.py b/ingestion/src/metadata/ingestion/source/dashboard/superset/api_source.py index 76f7fb97713c..a4de8346ba4a 100644 --- a/ingestion/src/metadata/ingestion/source/dashboard/superset/api_source.py +++ b/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), ) diff --git a/ingestion/src/metadata/ingestion/source/dashboard/superset/db_source.py b/ingestion/src/metadata/ingestion/source/dashboard/superset/db_source.py index a0b6225c16b1..8e4e3c04b747 100644 --- a/ingestion/src/metadata/ingestion/source/dashboard/superset/db_source.py +++ b/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), ) diff --git a/ingestion/src/metadata/ingestion/source/dashboard/superset/mixin.py b/ingestion/src/metadata/ingestion/source/dashboard/superset/mixin.py index 58551bb87445..f587876ea2f0 100644 --- a/ingestion/src/metadata/ingestion/source/dashboard/superset/mixin.py +++ b/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 ( @@ -319,13 +321,32 @@ 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. """ for chart_json in filter( None, @@ -350,6 +371,27 @@ 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 + dashboard_entity = self._get_dashboard_entity(dashboard_details) + 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 +404,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]: diff --git a/ingestion/tests/unit/topology/dashboard/test_superset_chart_lineage.py b/ingestion/tests/unit/topology/dashboard/test_superset_chart_lineage.py new file mode 100644 index 000000000000..7d79dae7ed4d --- /dev/null +++ b/ingestion/tests/unit/topology/dashboard/test_superset_chart_lineage.py @@ -0,0 +1,263 @@ +# 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. +""" +Regression tests for the Superset Dashboard <-> DataModel <-> Chart linking +behaviour. Verifies: + +1. yield_dashboard sends dataModels=[] on every CreateDashboardRequest, so + stale Dashboard.dataModels entries from previous runs are cleared. + +2. The Superset override of yield_datamodel_dashboard_lineage produces no + edges, suppressing the base class' direct DataModel -> Dashboard edge. + +3. yield_dashboard_lineage_details emits both DataModel -> Chart and + Chart -> Dashboard edges so the lineage graph renders the chart as the + bridge between datamodels and the dashboard. +""" + +import uuid +from unittest import TestCase +from unittest.mock import MagicMock + +from metadata.generated.schema.api.data.createDashboard import CreateDashboardRequest +from metadata.generated.schema.api.lineage.addLineage import AddLineageRequest +from metadata.generated.schema.entity.data.chart import Chart, ChartType +from metadata.generated.schema.entity.data.dashboard import Dashboard, DashboardType +from metadata.generated.schema.entity.data.dashboardDataModel import ( + DashboardDataModel, + DataModelType, +) +from metadata.generated.schema.type.basic import ( + EntityName, + FullyQualifiedEntityName, + SourceUrl, +) +from metadata.generated.schema.type.entityReference import EntityReference +from metadata.ingestion.api.models import Either +from metadata.ingestion.source.dashboard.superset.api_source import SupersetAPISource +from metadata.ingestion.source.dashboard.superset.db_source import SupersetDBSource +from metadata.ingestion.source.dashboard.superset.mixin import SupersetSourceMixin +from metadata.ingestion.source.dashboard.superset.models import ( + DashboardResult, + FetchChart, + FetchDashboard, +) + + +def _make_dashboard_entity() -> Dashboard: + return Dashboard( + id=uuid.uuid4(), + name=EntityName("4"), + fullyQualifiedName=FullyQualifiedEntityName("superset_test.4"), + dashboardType=DashboardType.Dashboard, + service=EntityReference(id=uuid.uuid4(), type="dashboardService"), + ) + + +def _make_chart_entity(chart_id: str) -> Chart: + return Chart( + id=uuid.uuid4(), + name=EntityName(chart_id), + fullyQualifiedName=FullyQualifiedEntityName(f"superset_test.{chart_id}"), + chartType=ChartType.Table, + service=EntityReference(id=uuid.uuid4(), type="dashboardService"), + ) + + +def _make_datamodel_entity(datamodel_id: str) -> DashboardDataModel: + return DashboardDataModel( + id=uuid.uuid4(), + name=EntityName(datamodel_id), + fullyQualifiedName=FullyQualifiedEntityName( + f"superset_test.model.{datamodel_id}" + ), + dataModelType=DataModelType.SupersetDataModel, + columns=[], + service=EntityReference(id=uuid.uuid4(), type="dashboardService"), + ) + + +def _build_context(charts=None, datamodels=None) -> MagicMock: + ctx = MagicMock() + ctx.dashboard_service = "superset_test" + ctx.charts = charts or [] + ctx.dataModels = datamodels or [] + return ctx + + +class TestSupersetDashboardDataModelsCleared(TestCase): + """yield_dashboard must send dataModels=[] (not omit) so server clears + any stale Dashboard.dataModels entries persisted from earlier runs.""" + + def test_db_source_yields_empty_data_models(self): + source = SupersetDBSource.__new__(SupersetDBSource) + source.metadata = MagicMock() + source.service_connection = MagicMock(hostPort="https://superset.example.com") + source.context = MagicMock() + source.context.get.return_value = _build_context( + charts=["10", "11"], datamodels=["45", "46"] + ) + source.get_owner_ref = MagicMock(return_value=None) + + dashboard_details = FetchDashboard( + id=4, + dashboard_title="Meltano", + position_json=None, + published=True, + email=None, + json_metadata=None, + ) + + results = list(SupersetDBSource.yield_dashboard(source, dashboard_details)) + self.assertEqual(len(results), 1) + request: CreateDashboardRequest = results[0].right + # The whole point of the regression fix: dataModels MUST be an empty + # list, not None or absent — that's what tells the server to clear + # the field. + self.assertEqual(request.dataModels, []) + self.assertEqual(len(request.charts), 2) + + def test_api_source_yields_empty_data_models(self): + source = SupersetAPISource.__new__(SupersetAPISource) + source.metadata = MagicMock() + source.service_connection = MagicMock(hostPort="https://superset.example.com") + source.context = MagicMock() + source.context.get.return_value = _build_context( + charts=["10"], datamodels=["45"] + ) + source.get_owner_ref = MagicMock(return_value=None) + + dashboard_details = DashboardResult( + id=4, + dashboard_title="Meltano", + url="/dashboard/4/", + published=True, + position_json=None, + email=None, + json_metadata=None, + ) + + results = list(SupersetAPISource.yield_dashboard(source, dashboard_details)) + self.assertEqual(len(results), 1) + request: CreateDashboardRequest = results[0].right + self.assertEqual(request.dataModels, []) + self.assertEqual(len(request.charts), 1) + + +class TestSupersetSuppressesDirectDataModelDashboardEdge(TestCase): + """The Superset mixin must override yield_datamodel_dashboard_lineage to + emit zero edges — otherwise the base class produces a direct + DataModel -> Dashboard lineage edge that bypasses the chart node.""" + + def test_override_yields_no_edges(self): + source = MagicMock() + # Bind the unbound method to the mock so the override runs + result = list( + SupersetSourceMixin.yield_datamodel_dashboard_lineage(source) + ) + self.assertEqual(result, []) + + +class TestSupersetEmitsChartBridgeEdges(TestCase): + """yield_dashboard_lineage_details must emit DataModel -> Chart and + Chart -> Dashboard edges so the chart bridges the chain in the + lineage graph.""" + + def _make_source(self, chart_entity, datamodel_entity, dashboard_entity): + source = MagicMock() + # Real bound methods we want to exercise + source.yield_dashboard_lineage_details = ( + lambda *a, **kw: SupersetSourceMixin.yield_dashboard_lineage_details( + source, *a, **kw + ) + ) + source._get_chart_entity = lambda chart_json: chart_entity + source._get_dashboard_entity = lambda dashboard_details: dashboard_entity + source._get_dashboard_data_model_entity = lambda chart_json: datamodel_entity + source._get_input_tables = lambda chart_json: [] + source._enrich_raw_input_tables = lambda inputs, to_entity, prefix: [] + source._get_charts_of_dashboard = lambda dashboard_details: ["10"] + source._get_add_lineage_request = ( + SupersetSourceMixin.__mro__[1]._get_add_lineage_request + ) + source.all_charts = { + "10": FetchChart( + id=10, + slice_name="chart-10", + datasource_id=45, + viz_type="table", + table_name="t", + table_id=1, + table_schema=None, + schema_name=None, + sql=None, + params=None, + description=None, + url=None, + ) + } + return source + + def test_chart_bridge_edges_emitted(self): + chart = _make_chart_entity("10") + datamodel = _make_datamodel_entity("45") + dashboard = _make_dashboard_entity() + source = self._make_source(chart, datamodel, dashboard) + + dashboard_details = FetchDashboard( + id=4, + dashboard_title="Meltano", + position_json=None, + published=True, + email=None, + json_metadata=None, + ) + results = [ + r for r in source.yield_dashboard_lineage_details(dashboard_details) + if r is not None and r.right is not None + ] + + edges = [r.right for r in results if isinstance(r.right, AddLineageRequest)] + self.assertEqual(len(edges), 2, f"expected 2 edges, got {edges}") + + from_to = {(e.edge.fromEntity.id.root, e.edge.toEntity.id.root) for e in edges} + self.assertIn( + (datamodel.id.root, chart.id.root), + from_to, + "DataModel -> Chart edge missing", + ) + self.assertIn( + (chart.id.root, dashboard.id.root), + from_to, + "Chart -> Dashboard edge missing", + ) + + def test_no_chart_entity_skips_bridge_edges(self): + # When the Chart entity isn't yet visible on the server, both bridge + # edges should be skipped (no crash, just no emission). + datamodel = _make_datamodel_entity("45") + source = self._make_source( + chart_entity=None, datamodel_entity=datamodel, dashboard_entity=None + ) + dashboard_details = FetchDashboard( + id=4, + dashboard_title="Meltano", + position_json=None, + published=True, + email=None, + json_metadata=None, + ) + results = [ + r for r in source.yield_dashboard_lineage_details(dashboard_details) + if r is not None and r.right is not None + ] + edges = [r.right for r in results if isinstance(r.right, AddLineageRequest)] + self.assertEqual(edges, [], "no edges should be emitted without chart entity") From fd7a46924a69a61f51f85ab7be1804dee0443a66 Mon Sep 17 00:00:00 2001 From: Joao Amaral <7281460+joaopamaral@users.noreply.github.com> Date: Mon, 18 May 2026 16:06:10 -0300 Subject: [PATCH 2/7] review(superset): hoist dashboard entity lookup, drop fragile MRO indexing in test Two review comments from gitar-bot on #28240: - mixin.yield_dashboard_lineage_details was calling self._get_dashboard_entity(dashboard_details) inside the per-chart loop, doing N identical metadata.get_by_name lookups for a dashboard with N charts. Hoist the call out of the loop and reuse the result. - test_superset_chart_lineage.py was wiring SupersetSourceMixin.__mro__[1]._get_add_lineage_request onto a mock, which silently breaks if a new intermediate base class is inserted. Reference DashboardServiceSource._get_add_lineage_request directly. Co-Authored-By: Claude Opus 4.7 (1M context) --- .../metadata/ingestion/source/dashboard/superset/mixin.py | 4 +++- .../unit/topology/dashboard/test_superset_chart_lineage.py | 7 ++++--- 2 files changed, 7 insertions(+), 4 deletions(-) diff --git a/ingestion/src/metadata/ingestion/source/dashboard/superset/mixin.py b/ingestion/src/metadata/ingestion/source/dashboard/superset/mixin.py index f587876ea2f0..cacc4ffe5fcd 100644 --- a/ingestion/src/metadata/ingestion/source/dashboard/superset/mixin.py +++ b/ingestion/src/metadata/ingestion/source/dashboard/superset/mixin.py @@ -348,6 +348,9 @@ def yield_dashboard_lineage_details( 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, [ @@ -384,7 +387,6 @@ def yield_dashboard_lineage_details( ) if dm_to_chart is not None: yield dm_to_chart - dashboard_entity = self._get_dashboard_entity(dashboard_details) if dashboard_entity is not None: chart_to_dash = self._get_add_lineage_request( to_entity=dashboard_entity, diff --git a/ingestion/tests/unit/topology/dashboard/test_superset_chart_lineage.py b/ingestion/tests/unit/topology/dashboard/test_superset_chart_lineage.py index 7d79dae7ed4d..3d84d89bcfef 100644 --- a/ingestion/tests/unit/topology/dashboard/test_superset_chart_lineage.py +++ b/ingestion/tests/unit/topology/dashboard/test_superset_chart_lineage.py @@ -42,6 +42,9 @@ ) from metadata.generated.schema.type.entityReference import EntityReference from metadata.ingestion.api.models import Either +from metadata.ingestion.source.dashboard.dashboard_service import ( + DashboardServiceSource, +) from metadata.ingestion.source.dashboard.superset.api_source import SupersetAPISource from metadata.ingestion.source.dashboard.superset.db_source import SupersetDBSource from metadata.ingestion.source.dashboard.superset.mixin import SupersetSourceMixin @@ -185,9 +188,7 @@ def _make_source(self, chart_entity, datamodel_entity, dashboard_entity): source._get_input_tables = lambda chart_json: [] source._enrich_raw_input_tables = lambda inputs, to_entity, prefix: [] source._get_charts_of_dashboard = lambda dashboard_details: ["10"] - source._get_add_lineage_request = ( - SupersetSourceMixin.__mro__[1]._get_add_lineage_request - ) + source._get_add_lineage_request = DashboardServiceSource._get_add_lineage_request source.all_charts = { "10": FetchChart( id=10, From 25cebbb647b563cef1bcbc5d31191cbb12885a23 Mon Sep 17 00:00:00 2001 From: Joao Amaral <7281460+joaopamaral@users.noreply.github.com> Date: Tue, 2 Jun 2026 16:00:10 -0300 Subject: [PATCH 3/7] fix(ingestion): guard None from add_lineage in write_lineage add_lineage PUTs the edge first, then re-fetches the source entity lineage. When that follow-up GET 404s (e.g. a dangling reference to a deleted chart/datamodel), add_lineage returns None even though the edge was written. write_lineage assumed a dict and crashed on created_lineage.get('error') with 'NoneType' object has no attribute 'get'. Treat None as a soft failure: log a warning and return an empty Either so the bulk sink continues. Co-Authored-By: Claude Opus 4.8 --- ingestion/pyproject.toml | 2 +- .../metadata/ingestion/sink/metadata_rest.py | 8 ++ .../unit/test_sink_lineage_none_response.py | 75 +++++++++++++++++++ 3 files changed, 84 insertions(+), 1 deletion(-) create mode 100644 ingestion/tests/unit/test_sink_lineage_none_response.py diff --git a/ingestion/pyproject.toml b/ingestion/pyproject.toml index 3fa1d32f47e1..2d5ddbb984c8 100644 --- a/ingestion/pyproject.toml +++ b/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.1" dynamic = ["readme", "dependencies", "optional-dependencies"] authors = [ { name = "OpenMetadata Committers" } diff --git a/ingestion/src/metadata/ingestion/sink/metadata_rest.py b/ingestion/src/metadata/ingestion/sink/metadata_rest.py index 2cbf68e8550c..e9dd9dbb86e1 100644 --- a/ingestion/src/metadata/ingestion/sink/metadata_rest.py +++ b/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( diff --git a/ingestion/tests/unit/test_sink_lineage_none_response.py b/ingestion/tests/unit/test_sink_lineage_none_response.py new file mode 100644 index 000000000000..c54e687cfb46 --- /dev/null +++ b/ingestion/tests/unit/test_sink_lineage_none_response.py @@ -0,0 +1,75 @@ +# Copyright 2025 Collate +# Licensed under the Collate Community License, Version 1.0 (the "License"); + +""" +Unit tests for sink-level handling of a None response from add_lineage in +write_lineage. + +add_lineage PUTs the edge first and only then fetches the source entity +lineage back. When that follow-up GET 404s (e.g. the source has a dangling +downstream reference to a deleted entity), add_lineage returns None even +though the edge was written. The sink must not crash on that None. +""" + +from unittest.mock import Mock, patch +from uuid import uuid4 + +import pytest + +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, +) + + +class TestSinkLineageNoneResponse: + """write_lineage must tolerate a None response from add_lineage""" + + @pytest.fixture(autouse=True) + def setup(self): + self.mock_metadata = Mock() + self.config = MetadataRestSinkConfig(bulk_sink_batch_size=10) + self.sink = MetadataRestSink(self.config, self.mock_metadata) + + def _lineage_request(self) -> AddLineageRequest: + return AddLineageRequest( + edge=EntitiesEdge( + fromEntity=EntityReference(id=uuid4(), type="dashboardDataModel"), + toEntity=EntityReference(id=uuid4(), type="chart"), + ) + ) + + @patch("metadata.ingestion.sink.metadata_rest.logger") + def test_none_response_does_not_crash(self, mock_logger): + """A None add_lineage response is logged and yields a non-error result""" + self.mock_metadata.add_lineage.return_value = None + + result = self.sink.write_lineage(self._lineage_request()) + + assert result.left is None + assert result.right is None + mock_logger.warning.assert_called_once() + + def test_error_response_is_propagated(self): + """An error dict from add_lineage is still surfaced as a failure""" + self.mock_metadata.add_lineage.return_value = {"error": "boom"} + + result = self.sink.write_lineage(self._lineage_request()) + + assert result.right is None + assert result.left is not None + assert result.left.error == "boom" + + def test_success_response_returns_fqn(self): + """A successful response returns the source entity FQN""" + self.mock_metadata.add_lineage.return_value = { + "entity": {"fullyQualifiedName": "service.model"} + } + + result = self.sink.write_lineage(self._lineage_request()) + + assert result.left is None + assert result.right == "service.model" From b85f8be935a6861901174a804ebcbf58dfd1e14f Mon Sep 17 00:00:00 2001 From: Joao Amaral <7281460+joaopamaral@users.noreply.github.com> Date: Tue, 2 Jun 2026 16:10:33 -0300 Subject: [PATCH 4/7] feat(ingestion): mark orphaned charts as deleted (backport #27309) Backport of #27309 (Fix #27308) onto 1.12.9. Adds first-class chart cleanup to dashboard ingestion so charts removed from the source (and no longer referenced by any dashboard in the scan) are soft-deleted, mirror- ing markDeletedDataModels/markDeletedDashboards. - new markDeletedCharts config (default true) on dashboard pipeline schema + generated python model - chart_source_state, register_record_chart(), mark_charts_as_deleted() on the base DashboardServiceSource; wired into post_process - connectors register charts via register_record_chart in yield_dashboard_chart This removes the source of the dangling DataModel->Chart lineage edges that triggered the write_lineage None crash: when a datamodel is deleted but its chart survived, the orphaned chart kept a dead upstream edge. Conflicts resolved for 1.12.9 (powerbi/sigma kept local connector shape); the powerbi/sigma/tableau/qliksense test additions from #27309 were left on the 1.13 line as they depend on newer test scaffolding. Co-Authored-By: Claude Opus 4.8 --- .../source/dashboard/dashboard_service.py | 34 +- .../dashboard/domodashboard/metadata.py | 24 +- .../source/dashboard/grafana/metadata.py | 32 +- .../source/dashboard/lightdash/metadata.py | 22 +- .../source/dashboard/looker/metadata.py | 18 +- .../source/dashboard/metabase/metadata.py | 18 +- .../dashboard/microstrategy/metadata.py | 16 +- .../source/dashboard/mode/metadata.py | 16 +- .../source/dashboard/powerbi/metadata.py | 638 +++++++++++++----- .../source/dashboard/qlikcloud/metadata.py | 26 +- .../source/dashboard/qliksense/metadata.py | 30 +- .../source/dashboard/quicksight/metadata.py | 20 +- .../source/dashboard/redash/metadata.py | 42 +- .../source/dashboard/sigma/metadata.py | 30 +- .../source/dashboard/ssrs/metadata.py | 26 +- .../source/dashboard/superset/api_source.py | 5 +- .../source/dashboard/superset/db_source.py | 5 +- .../source/dashboard/tableau/metadata.py | 5 +- .../topology/dashboard/test_domodashboard.py | 9 + .../unit/topology/dashboard/test_grafana.py | 8 + .../unit/topology/dashboard/test_looker.py | 9 + .../topology/dashboard/test_microstrategy.py | 56 ++ .../unit/topology/dashboard/test_qlikcloud.py | 11 + .../topology/dashboard/test_quicksight.py | 13 + .../unit/topology/dashboard/test_ssrs.py | 7 + .../dashboardServiceMetadataPipeline.json | 6 + .../createIngestionPipeline.ts | 5 + .../ingestionPipelines/ingestionPipeline.ts | 5 + .../dashboardServiceMetadataPipeline.ts | 5 + .../generated/metadataIngestion/workflow.ts | 5 + 30 files changed, 823 insertions(+), 323 deletions(-) diff --git a/ingestion/src/metadata/ingestion/source/dashboard/dashboard_service.py b/ingestion/src/metadata/ingestion/source/dashboard/dashboard_service.py index ead0bad5ca42..f1522f806a16 100644 --- a/ingestion/src/metadata/ingestion/source/dashboard/dashboard_service.py +++ b/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], diff --git a/ingestion/src/metadata/ingestion/source/dashboard/domodashboard/metadata.py b/ingestion/src/metadata/ingestion/source/dashboard/domodashboard/metadata.py index 0e422908388b..676add03f258 100644 --- a/ingestion/src/metadata/ingestion/source/dashboard/domodashboard/metadata.py +++ b/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( diff --git a/ingestion/src/metadata/ingestion/source/dashboard/grafana/metadata.py b/ingestion/src/metadata/ingestion/source/dashboard/grafana/metadata.py index 7c6adfd0543c..d85fa4ca035a 100644 --- a/ingestion/src/metadata/ingestion/source/dashboard/grafana/metadata.py +++ b/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( diff --git a/ingestion/src/metadata/ingestion/source/dashboard/lightdash/metadata.py b/ingestion/src/metadata/ingestion/source/dashboard/lightdash/metadata.py index bf5f0a2c9073..0a40488af262 100644 --- a/ingestion/src/metadata/ingestion/source/dashboard/lightdash/metadata.py +++ b/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()) diff --git a/ingestion/src/metadata/ingestion/source/dashboard/looker/metadata.py b/ingestion/src/metadata/ingestion/source/dashboard/looker/metadata.py index 1e6b87ae2f5a..b720c7264d31 100644 --- a/ingestion/src/metadata/ingestion/source/dashboard/looker/metadata.py +++ b/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( diff --git a/ingestion/src/metadata/ingestion/source/dashboard/metabase/metadata.py b/ingestion/src/metadata/ingestion/source/dashboard/metabase/metadata.py index 0ce042dc16c0..fe5f377a6c7f 100644 --- a/ingestion/src/metadata/ingestion/source/dashboard/metabase/metadata.py +++ b/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( diff --git a/ingestion/src/metadata/ingestion/source/dashboard/microstrategy/metadata.py b/ingestion/src/metadata/ingestion/source/dashboard/microstrategy/metadata.py index 9b9086337c26..c54e553c2add 100644 --- a/ingestion/src/metadata/ingestion/source/dashboard/microstrategy/metadata.py +++ b/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( diff --git a/ingestion/src/metadata/ingestion/source/dashboard/mode/metadata.py b/ingestion/src/metadata/ingestion/source/dashboard/mode/metadata.py index b21cbe32f2d9..77ebe7f744c0 100644 --- a/ingestion/src/metadata/ingestion/source/dashboard/mode/metadata.py +++ b/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( diff --git a/ingestion/src/metadata/ingestion/source/dashboard/powerbi/metadata.py b/ingestion/src/metadata/ingestion/source/dashboard/powerbi/metadata.py index 8642bc6f0f42..5e77e5137b29 100644 --- a/ingestion/src/metadata/ingestion/source/dashboard/powerbi/metadata.py +++ b/ingestion/src/metadata/ingestion/source/dashboard/powerbi/metadata.py @@ -140,7 +140,9 @@ def __init__( metadata: OpenMetadata, ): super().__init__(config, metadata) - self.pagination_entity_per_page = min(100, self.service_connection.pagination_entity_per_page) + self.pagination_entity_per_page = min( + 100, self.service_connection.pagination_entity_per_page + ) self.datamodel_file_mappings = [] self.state = WorkspaceState() @@ -154,7 +156,9 @@ def get_org_workspace_data(self) -> Iterable[Optional[Group]]: # noqa: UP045 fetch all the workspace data for non-admin users """ filter_pattern = self.source_config.projectFilterPattern - paginated_filter_patterns = self._paginate_project_filter_pattern(filter_pattern) + paginated_filter_patterns = self._paginate_project_filter_pattern( + filter_pattern + ) if len(paginated_filter_patterns) > 1: logger.info( f"Paginating workspace fetch with {len(paginated_filter_patterns)}" @@ -166,26 +170,41 @@ def get_org_workspace_data(self) -> Iterable[Optional[Group]]: # noqa: UP045 for workspace in workspaces: # add the dashboards to the workspace workspace.dashboards.extend( - self.client.api_client.fetch_all_org_dashboards(group_id=workspace.id) or [] + self.client.api_client.fetch_all_org_dashboards( + group_id=workspace.id + ) + or [] ) for dashboard in workspace.dashboards: # add the tiles to the dashboards dashboard.tiles.extend( - self.client.api_client.fetch_all_org_tiles(group_id=workspace.id, dashboard_id=dashboard.id) + self.client.api_client.fetch_all_org_tiles( + group_id=workspace.id, dashboard_id=dashboard.id + ) or [] ) # add the reports to the workspaces - workspace.reports.extend(self.client.api_client.fetch_all_org_reports(group_id=workspace.id) or []) + workspace.reports.extend( + self.client.api_client.fetch_all_org_reports( + group_id=workspace.id + ) + or [] + ) # add the datasets to the workspaces workspace.datasets.extend( - self.client.api_client.fetch_all_org_datasets(group_id=workspace.id) or [] + self.client.api_client.fetch_all_org_datasets( + group_id=workspace.id + ) + or [] ) for dataset in workspace.datasets: # add the tables to the datasets dataset.tables.extend( - self.client.api_client.fetch_dataset_tables(group_id=workspace.id, dataset_id=dataset.id) + self.client.api_client.fetch_dataset_tables( + group_id=workspace.id, dataset_id=dataset.id + ) or [] ) yield workspace @@ -224,7 +243,9 @@ def get_admin_workspace_data(self) -> Iterable[Optional[Group]]: # noqa: UP045 fetch all the workspace data """ filter_pattern = self.source_config.projectFilterPattern - paginated_filter_patterns = self._paginate_project_filter_pattern(filter_pattern) + paginated_filter_patterns = self._paginate_project_filter_pattern( + filter_pattern + ) if len(paginated_filter_patterns) > 1: logger.info( f"Paginating workspace fetch with {len(paginated_filter_patterns)}" @@ -238,12 +259,18 @@ def get_admin_workspace_data(self) -> Iterable[Optional[Group]]: # noqa: UP045 # Start the scan of the available workspaces for dashboard metadata workspace_paginated_list = [ workspace_id_list[i : i + self.pagination_entity_per_page] - for i in range(0, len(workspace_id_list), self.pagination_entity_per_page) + for i in range( + 0, len(workspace_id_list), self.pagination_entity_per_page + ) ] count = 1 for workspace_ids_chunk in workspace_paginated_list: - logger.info(f"Scanning {count}/{len(workspace_paginated_list)} set of workspaces") - workspace_scan = self.client.api_client.initiate_workspace_scan(workspace_ids_chunk) + logger.info( + f"Scanning {count}/{len(workspace_paginated_list)} set of workspaces" + ) + workspace_scan = self.client.api_client.initiate_workspace_scan( + workspace_ids_chunk + ) if not workspace_scan: logger.error( f"Error initiating workspace scan for ids:{str(workspace_ids_chunk)}\n moving to next set of workspaces" # noqa: RUF010 @@ -252,7 +279,11 @@ def get_admin_workspace_data(self) -> Iterable[Optional[Group]]: # noqa: UP045 continue # Keep polling the scan status endpoint to check if scan is succeeded - workspace_scan_status = self.client.api_client.wait_for_scan_complete(scan_id=workspace_scan.id) + workspace_scan_status = ( + self.client.api_client.wait_for_scan_complete( + scan_id=workspace_scan.id + ) + ) if not workspace_scan_status: logger.error( f"Max poll hit to scan status for scan_id: {workspace_scan.id}, moving to next set of workspaces" @@ -261,9 +292,13 @@ def get_admin_workspace_data(self) -> Iterable[Optional[Group]]: # noqa: UP045 continue # Get scan result for successfull scan - response = self.client.api_client.fetch_workspace_scan_result(scan_id=workspace_scan.id) + response = self.client.api_client.fetch_workspace_scan_result( + scan_id=workspace_scan.id + ) if not response: - logger.error(f"Error getting workspace scan result for scan_id: {workspace_scan.id}") + logger.error( + f"Error getting workspace scan result for scan_id: {workspace_scan.id}" + ) count += 1 continue for active_workspace in response.workspaces: @@ -274,11 +309,15 @@ def get_admin_workspace_data(self) -> Iterable[Optional[Group]]: # noqa: UP045 logger.error("Unable to fetch any PowerBI workspaces") @classmethod - def create(cls, config_dict, metadata: OpenMetadata, pipeline_name: Optional[str] = None): # noqa: UP045 + def create( + cls, config_dict, metadata: OpenMetadata, pipeline_name: Optional[str] = None + ): # noqa: UP045 config = WorkflowSource.model_validate(config_dict) connection: PowerBIConnection = config.serviceConnection.root.config if not isinstance(connection, PowerBIConnection): - raise InvalidSourceException(f"Expected PowerBIConnection, but got {connection}") + raise InvalidSourceException( + f"Expected PowerBIConnection, but got {connection}" + ) return cls(config, metadata) def _prepare_workspace_data(self) -> Iterable[Group]: @@ -289,7 +328,9 @@ def _prepare_workspace_data(self) -> Iterable[Group]: - Workspaces that failed to fetch (None) are filtered out at the producer. """ producer = ( - self.get_admin_workspace_data() if self.service_connection.useAdminApis else self.get_org_workspace_data() + self.get_admin_workspace_data() + if self.service_connection.useAdminApis + else self.get_org_workspace_data() ) for workspace in producer: if workspace is None: @@ -303,7 +344,9 @@ def get_dashboard(self) -> Any: for workspace in self._prepare_workspace_data(): try: self.state.enter(workspace) - self.context.get().workspace = workspace # pyright: ignore[reportAttributeAccessIssue] + self.context.get().workspace = ( + workspace # pyright: ignore[reportAttributeAccessIssue] + ) for dashboard in self.get_dashboards_list() or []: dashboard_details = self.get_dashboard_details(dashboard) dashboard_name = self.get_dashboard_name(dashboard_details) @@ -325,8 +368,12 @@ def get_dashboard(self) -> Any: self.state.add_filtered_dashboard(dashboard_details) yield workspace except Exception as exc: # pylint: disable=broad-except - ws_name = getattr(workspace, "name", None) or getattr(workspace, "id", "") - logger.warning("Failed to process PowerBI workspace '%s': %s", ws_name, exc) + ws_name = getattr(workspace, "name", None) or getattr( + workspace, "id", "" + ) + logger.warning( + "Failed to process PowerBI workspace '%s': %s", ws_name, exc + ) self.status.failed( StackTraceError( name=f"Workspace {ws_name}", @@ -339,13 +386,20 @@ def get_dashboard(self) -> Any: def get_dashboards_list( self, - ) -> Optional[List[Union[PowerBIDashboard, PowerBIReport]]]: # noqa: UP006, UP007, UP045 + ) -> Optional[ + List[Union[PowerBIDashboard, PowerBIReport]] + ]: # noqa: UP006, UP007, UP045 """ Get List of all dashboards """ - return self.context.get().workspace.reports + self.context.get().workspace.dashboards # pyright: ignore[reportAttributeAccessIssue] + return ( + self.context.get().workspace.reports + + self.context.get().workspace.dashboards + ) # pyright: ignore[reportAttributeAccessIssue] - def get_dashboard_name(self, dashboard: Union[PowerBIDashboard, PowerBIReport]) -> str | None: # noqa: UP007 # pyright: ignore[reportIncompatibleMethodOverride] + def get_dashboard_name( + self, dashboard: Union[PowerBIDashboard, PowerBIReport] + ) -> str | None: # noqa: UP007 # pyright: ignore[reportIncompatibleMethodOverride] """ Get Dashboard Name """ @@ -371,17 +425,24 @@ def _get_dashboard_url(self, workspace_id: str, dashboard_id: str) -> str: f"{workspace_id}/dashboards/{dashboard_id}?experience=power-bi" ) - def _get_report_url(self, workspace_id: str, dashboard_details: PowerBIReport) -> str: + def _get_report_url( + self, workspace_id: str, dashboard_details: PowerBIReport + ) -> str: """ Method to build the dashboard url """ page_id = "" dashboard_id = dashboard_details.id reports_prefix = DEFAULT_REPORTS_PREFIX - if isinstance(dashboard_details.format, str) and dashboard_details.format == RDL_REPORT_FORMAT: + if ( + isinstance(dashboard_details.format, str) + and dashboard_details.format == RDL_REPORT_FORMAT + ): reports_prefix = RDL_REPORTS_PREFIX try: - pages: Optional[List[ReportPage]] = self.client.api_client.fetch_report_pages(workspace_id, dashboard_id) # noqa: UP006, UP045 + pages: Optional[List[ReportPage]] = ( + self.client.api_client.fetch_report_pages(workspace_id, dashboard_id) + ) # noqa: UP006, UP045 if ( pages and pages[0].name ): # if there are pages and page has name then only add page id in url: # if there are pages and page has name then only add page id in url @@ -416,14 +477,20 @@ def _get_dataflow_url(self, workspace_id: str, dataflow_id: str) -> str: f"{workspace_id}/dataflows/{dataflow_id}?experience=power-bi" ) - def _get_chart_url(self, report_id: Optional[str], workspace_id: str, dashboard_id: str) -> str: # noqa: UP045 + def _get_chart_url( + self, report_id: Optional[str], workspace_id: str, dashboard_id: str + ) -> str: # noqa: UP045 """ Method to build the chart url """ - chart_url_postfix = f"reports/{report_id}" if report_id else f"dashboards/{dashboard_id}" + chart_url_postfix = ( + f"reports/{report_id}" if report_id else f"dashboards/{dashboard_id}" + ) return f"{clean_uri(self.service_connection.hostPort)}/groups/{workspace_id}/{chart_url_postfix}" - def yield_dashboard(self, dashboard_details: Group) -> Iterable[Either[CreateDashboardRequest]]: + def yield_dashboard( + self, dashboard_details: Group + ) -> Iterable[Either[CreateDashboardRequest]]: """ Method to Get Dashboard Entity, Dashboard Charts & Lineage """ @@ -431,7 +498,9 @@ def yield_dashboard(self, dashboard_details: Group) -> Iterable[Either[CreateDas for dashboard in self.state.filtered_dashboards: dashboard_details = self.get_dashboard_details(dashboard) if isinstance(dashboard_details, PowerBIDashboard): - dashboard_chart_ids = self.state.pop_dashboard_chart_ids(dashboard_details.id) + dashboard_chart_ids = self.state.pop_dashboard_chart_ids( + dashboard_details.id + ) dashboard_request = CreateDashboardRequest( name=EntityName(dashboard_details.id), sourceUrl=SourceUrl( @@ -454,11 +523,17 @@ def yield_dashboard(self, dashboard_details: Group) -> Iterable[Either[CreateDas ) for chart in dashboard_chart_ids ], - service=FullyQualifiedEntityName(self.context.get().dashboard_service), # pyright: ignore[reportAttributeAccessIssue] + service=FullyQualifiedEntityName( + self.context.get().dashboard_service + ), # pyright: ignore[reportAttributeAccessIssue] owners=self.get_owner_ref(dashboard_details=dashboard_details), ) else: - description = Markdown(dashboard_details.description) if dashboard_details.description else None + description = ( + Markdown(dashboard_details.description) + if dashboard_details.description + else None + ) dashboard_request = CreateDashboardRequest( name=EntityName(dashboard_details.id), dashboardType=DashboardType.Report, @@ -485,7 +560,9 @@ def yield_dashboard(self, dashboard_details: Group) -> Iterable[Either[CreateDas ) ) - def yield_dashboard_chart(self, dashboard_details: Group) -> Iterable[Either[CreateChartRequest]]: + def yield_dashboard_chart( + self, dashboard_details: Group + ) -> Iterable[Either[CreateChartRequest]]: """Get chart method Args: dashboard_details: @@ -500,8 +577,12 @@ def yield_dashboard_chart(self, dashboard_details: Group) -> Iterable[Either[Cre try: chart_title = chart.title chart_display_name = chart_title if chart_title else chart.id - if filter_by_chart(self.source_config.chartFilterPattern, chart_display_name): - self.status.filter(chart_display_name, "Chart Pattern not Allowed") + if filter_by_chart( + self.source_config.chartFilterPattern, chart_display_name + ): + self.status.filter( + chart_display_name, "Chart Pattern not Allowed" + ) continue chart_request = CreateChartRequest( name=EntityName(chart.id), @@ -514,10 +595,13 @@ def yield_dashboard_chart(self, dashboard_details: Group) -> Iterable[Either[Cre dashboard_id=dashboard_details.id, ) ), - service=FullyQualifiedEntityName(self.context.get().dashboard_service), # pyright: ignore[reportAttributeAccessIssue] + service=FullyQualifiedEntityName( + self.context.get().dashboard_service + ), # pyright: ignore[reportAttributeAccessIssue] ) yield Either(right=chart_request) self.state.add_dashboard_chart(dashboard_details.id, chart.id) + self.register_record_chart(chart_request=chart_request) except Exception as exc: yield Either( left=StackTraceError( @@ -543,8 +627,14 @@ def _get_child_measures(self, table: PowerBiTable) -> List[Column]: # noqa: UP0 measure_type = DataType.MEASURE_VISIBLE if measure.isHidden: measure_type = DataType.MEASURE_HIDDEN - expression_text = f"Expression : {measure.expression}" if measure.expression else "" - description_text = f"Description : {measure.description}" if measure.description else "" + expression_text = ( + f"Expression : {measure.expression}" if measure.expression else "" + ) + description_text = ( + f"Description : {measure.description}" + if measure.description + else "" + ) description_field_text = f"{expression_text}\n\n{description_text}" parsed_measure = PowerBiMeasureModel( dataType=measure_type, @@ -573,8 +663,12 @@ def _get_child_columns(self, table: PowerBiTable) -> List[Column]: # noqa: UP00 continue try: parsed_column = { - "dataTypeDisplay": (column.dataType if column.dataType else DataType.UNKNOWN.value), - "dataType": ColumnTypeParser.get_column_type(column.dataType if column.dataType else None), + "dataTypeDisplay": ( + column.dataType if column.dataType else DataType.UNKNOWN.value + ), + "dataType": ColumnTypeParser.get_column_type( + column.dataType if column.dataType else None + ), "name": truncate_column_name(column.name), "displayName": column.name, "description": column.description, @@ -587,7 +681,9 @@ def _get_child_columns(self, table: PowerBiTable) -> List[Column]: # noqa: UP00 logger.warning(f"Error processing datamodel nested column: {exc}") return columns - def _get_column_info(self, dataset: Dataset) -> Optional[List[Column]]: # noqa: UP006, UP045 + def _get_column_info( + self, dataset: Dataset + ) -> Optional[List[Column]]: # noqa: UP006, UP045 """Build columns from dataset""" datasource_columns = [] for table in dataset.tables or []: @@ -602,7 +698,9 @@ def _get_column_info(self, dataset: Dataset) -> Optional[List[Column]]: # noqa: if self.service_connection.displayTableNameFromSource: table_display_name = self.parse_table_name_from_source(table=table) if table_display_name: - logger.debug(f"Parsed Table display name: {table_display_name} for table: {table.name}") + logger.debug( + f"Parsed Table display name: {table_display_name} for table: {table.name}" + ) if not table_display_name: table_display_name = table.name parsed_table = { @@ -625,7 +723,9 @@ def _get_column_info(self, dataset: Dataset) -> Optional[List[Column]]: # noqa: logger.warning(f"Error to yield datamodel column: {exc}") return datasource_columns - def _get_dataflow_column_info(self, dataflow_export: DataflowExportResponse) -> Optional[List[Column]]: # noqa: UP006, UP045 + def _get_dataflow_column_info( + self, dataflow_export: DataflowExportResponse + ) -> Optional[List[Column]]: # noqa: UP006, UP045 """Build columns from dataflow export response entities""" datasource_columns = [] for entity in dataflow_export.entities or []: @@ -651,7 +751,11 @@ def _get_dataflow_column_info(self, dataflow_export: DataflowExportResponse) -> continue try: parsed_column = { - "dataTypeDisplay": (attribute.dataType if attribute.dataType else DataType.UNKNOWN.value), + "dataTypeDisplay": ( + attribute.dataType + if attribute.dataType + else DataType.UNKNOWN.value + ), "dataType": ColumnTypeParser.get_column_type( attribute.dataType if attribute.dataType else None ), @@ -659,12 +763,17 @@ def _get_dataflow_column_info(self, dataflow_export: DataflowExportResponse) -> "displayName": attribute.name, "description": attribute.description, } - if attribute.dataType and attribute.dataType == DataType.ARRAY.value: + if ( + attribute.dataType + and attribute.dataType == DataType.ARRAY.value + ): parsed_column["arrayDataType"] = DataType.UNKNOWN child_columns.append(Column(**parsed_column)) except Exception as exc: logger.debug(traceback.format_exc()) - logger.warning(f"Error processing dataflow entity attribute: {exc}") + logger.warning( + f"Error processing dataflow entity attribute: {exc}" + ) if child_columns: parsed_table["children"] = child_columns datasource_columns.append(Column(**parsed_table)) @@ -673,11 +782,16 @@ def _get_dataflow_column_info(self, dataflow_export: DataflowExportResponse) -> logger.warning(f"Error to yield dataflow entity column: {exc}") return datasource_columns - def _get_datamodels_list(self) -> List[Union[Dataset, Dataflow]]: # noqa: UP006, UP007 + def _get_datamodels_list( + self, + ) -> List[Union[Dataset, Dataflow]]: # noqa: UP006, UP007 """ Get All the Powerbi Datasets """ - return self.context.get().workspace.datasets + self.context.get().workspace.dataflows # pyright: ignore[reportAttributeAccessIssue] + return ( + self.context.get().workspace.datasets + + self.context.get().workspace.dataflows + ) # pyright: ignore[reportAttributeAccessIssue] def _filtered_datamodels(self) -> list: """Filtered datamodels for the current workspace, memoised on first call.""" @@ -692,14 +806,18 @@ def _filtered_datamodels(self) -> list: dataset.id, ) continue - if filter_by_datamodel(self.source_config.dataModelFilterPattern, dataset.name): + if filter_by_datamodel( + self.source_config.dataModelFilterPattern, dataset.name + ): self.status.filter(dataset.name, "Data model filtered out.") continue filtered.append(dataset) self.state.set_filtered_datamodels(filtered) return filtered - def yield_datamodel(self, dashboard_details: Group) -> Iterable[Either[CreateDashboardDataModelRequest]]: + def yield_datamodel( + self, dashboard_details: Group + ) -> Iterable[Either[CreateDashboardDataModelRequest]]: """ Get All the Powerbi Datasets """ @@ -735,26 +853,42 @@ def yield_datamodel(self, dashboard_details: Group) -> Iterable[Either[CreateDas # dataflow export api for detailed metadata # api: https://api.powerbi.com/v1.0/myorg/admin/dataflows/DATAFLOW_ID/export # doc: https://learn.microsoft.com/en-us/rest/api/power-bi/admin/dataflows-export-dataflow-as-admin - dataflow_export = self.client.api_client.fetch_dataflow_export(dataflow_id=dataset.id) + dataflow_export = self.client.api_client.fetch_dataflow_export( + dataflow_id=dataset.id + ) if dataflow_export: self.state.cache_dataflow_export(dataset.id, dataflow_export) - datamodel_columns = self._get_dataflow_column_info(dataflow_export) + datamodel_columns = self._get_dataflow_column_info( + dataflow_export + ) else: - logger.warning(f"Unknown dataset type: {type(dataset)}, name: {dataset.name}") + logger.warning( + f"Unknown dataset type: {type(dataset)}, name: {dataset.name}" + ) continue - data_model_request = CreateDashboardDataModelRequest( # pyright: ignore[reportCallIssue] - name=EntityName(dataset.id), - displayName=dataset.name, - description=(Markdown(dataset.description) if dataset.description else None), - service=FullyQualifiedEntityName(self.context.get().dashboard_service), # pyright: ignore[reportAttributeAccessIssue] - dataModelType=data_model_type, - serviceType=DashboardServiceType.PowerBI.value, - columns=datamodel_columns, - project=self.get_project_name(dashboard_details=dataset), - owners=self.get_owner_ref(dashboard_details=dataset), - sourceUrl=SourceUrl(source_url), + data_model_request = ( + CreateDashboardDataModelRequest( # pyright: ignore[reportCallIssue] + name=EntityName(dataset.id), + displayName=dataset.name, + description=( + Markdown(dataset.description) + if dataset.description + else None + ), + service=FullyQualifiedEntityName( + self.context.get().dashboard_service + ), # pyright: ignore[reportAttributeAccessIssue] + dataModelType=data_model_type, + serviceType=DashboardServiceType.PowerBI.value, + columns=datamodel_columns, + project=self.get_project_name(dashboard_details=dataset), + owners=self.get_owner_ref(dashboard_details=dataset), + sourceUrl=SourceUrl(source_url), + ) ) - yield Either(right=data_model_request) # pyright: ignore[reportCallIssue] + yield Either( + right=data_model_request + ) # pyright: ignore[reportCallIssue] self.register_record_datamodel(datamodel_request=data_model_request) except Exception as exc: dataset_name = dataset.name or dataset.id or "" @@ -789,7 +923,9 @@ def create_report_dashboard_lineage( dashboard_details.id, ) return - dashboard_entity = self.metadata.get_by_name(entity=Dashboard, fqn=dashboard_fqn) + dashboard_entity = self.metadata.get_by_name( + entity=Dashboard, fqn=dashboard_fqn + ) if not dashboard_entity: logger.debug( "Dashboard entity not found for tile-pinned report lineage: dashboard=%s", @@ -818,14 +954,20 @@ def create_report_dashboard_lineage( error_name="Report and Dashboard Lineage", ) - def _get_dataset_ids_from_report_datasources(self, report_id: str) -> List[str]: # noqa: UP006 + def _get_dataset_ids_from_report_datasources( + self, report_id: str + ) -> List[str]: # noqa: UP006 """ Fetch report datasources and extract dataset IDs from connectionDetails.database. The database field follows the pattern: sobe_wowvirtualserver-{DATASET_ID} """ dataset_ids = [] - workspace_id = self.context.get().workspace.id # pyright: ignore[reportAttributeAccessIssue] - datasources = self.client.api_client.fetch_report_datasources(group_id=workspace_id, report_id=report_id) + workspace_id = ( + self.context.get().workspace.id + ) # pyright: ignore[reportAttributeAccessIssue] + datasources = self.client.api_client.fetch_report_datasources( + group_id=workspace_id, report_id=report_id + ) if not datasources: return dataset_ids for datasource in datasources: @@ -837,7 +979,9 @@ def _get_dataset_ids_from_report_datasources(self, report_id: str) -> List[str]: if match: dataset_ids.append(match.group(1)) if dataset_ids: - logger.debug(f"Extracted dataset IDs from report datasources API call for report_id={report_id}") + logger.debug( + f"Extracted dataset IDs from report datasources API call for report_id={report_id}" + ) return dataset_ids def create_datamodel_report_lineage( @@ -849,7 +993,9 @@ def create_datamodel_report_lineage( create the lineage between datamodel and report """ try: - logger.debug(f"Processing to create datamodel and report lineage for report: {dashboard_details.id}") + logger.debug( + f"Processing to create datamodel and report lineage for report: {dashboard_details.id}" + ) report_fqn = fqn.build( self.metadata, entity_type=Dashboard, @@ -867,13 +1013,17 @@ def create_datamodel_report_lineage( return dataset_ids = [] if dashboard_details.datasetId: - logger.debug(f"Report linked datasetId is present in api response for report: {dashboard_details.id}") + logger.debug( + f"Report linked datasetId is present in api response for report: {dashboard_details.id}" + ) dataset_ids = [dashboard_details.datasetId] else: logger.debug( f"Processing to get report datasources from API to extract datasetIds for report: {dashboard_details.id} as datasetId is not present in api response" ) - dataset_ids = self._get_dataset_ids_from_report_datasources(report_id=dashboard_details.id) + dataset_ids = self._get_dataset_ids_from_report_datasources( + report_id=dashboard_details.id + ) if dataset_ids: for dataset_id in dataset_ids: @@ -908,13 +1058,17 @@ def create_datamodel_report_lineage( yield Either( left=StackTraceError( name="Datamodel and Report Lineage", - error=(f"Error to yield datamodel and report lineage details: {exc}"), + error=( + f"Error to yield datamodel and report lineage details: {exc}" + ), stackTrace=traceback.format_exc(), ) ) @staticmethod - def _get_data_model_column_fqn(data_model_entity: DashboardDataModel, column: str) -> Optional[str]: # noqa: UP045 + def _get_data_model_column_fqn( + data_model_entity: DashboardDataModel, column: str + ) -> Optional[str]: # noqa: UP045 """ Get fqn of column if exist in data model entity or its child columns """ @@ -930,7 +1084,9 @@ def _get_data_model_column_fqn(data_model_entity: DashboardDataModel, column: st logger.debug(f"Error to get data_model_column_fqn {exc}") logger.debug(traceback.format_exc()) - def parse_table_name_from_source(self, table: PowerBiTable) -> Optional[str]: # noqa: UP045 + def parse_table_name_from_source( + self, table: PowerBiTable + ) -> Optional[str]: # noqa: UP045 """ Parse the snowflake table name """ @@ -944,8 +1100,12 @@ def parse_table_name_from_source(self, table: PowerBiTable) -> Optional[str]: # if "Snowflake.Databases" in source_expression: # snowflake expression - table_match = re.search(r'\[Name=(?:"([^"]+)"|([^,]+)),Kind="Table"\]', source_expression) - view_match = re.search(r'\[Name=(?:"([^"]+)"|([^,]+)),Kind="View"\]', source_expression) + table_match = re.search( + r'\[Name=(?:"([^"]+)"|([^,]+)),Kind="Table"\]', source_expression + ) + view_match = re.search( + r'\[Name=(?:"([^"]+)"|([^,]+)),Kind="View"\]', source_expression + ) table = table_match.group(1) if table_match else None view = view_match.group(1) if view_match else None return table if table else view @@ -961,7 +1121,9 @@ def parse_table_name_from_source(self, table: PowerBiTable) -> Optional[str]: # logger.debug(traceback.format_exc()) return None - def _parse_expression_regex_exp(self, match: re.Match, datamodel_entity: DashboardDataModel) -> Optional[str]: # noqa: UP045 + def _parse_expression_regex_exp( + self, match: re.Match, datamodel_entity: DashboardDataModel + ) -> Optional[str]: # noqa: UP045 """parse snowflake regex expression""" try: if not match: @@ -989,9 +1151,13 @@ def _parse_expression_regex_exp(self, match: re.Match, datamodel_entity: Dashboa logger.debug(traceback.format_exc()) return None - def _parse_redshift_source(self, source_expression: str) -> Optional[List[dict]]: # noqa: UP006, UP045 + def _parse_redshift_source( + self, source_expression: str + ) -> Optional[List[dict]]: # noqa: UP006, UP045 try: - db_match = re.search(r'AmazonRedshift\.Database\("[^"]+","([^"]+)"\)', source_expression) + db_match = re.search( + r'AmazonRedshift\.Database\("[^"]+","([^"]+)"\)', source_expression + ) if not db_match: # not valid redshift source return None @@ -1011,7 +1177,9 @@ def _parse_redshift_source(self, source_expression: str) -> Optional[List[dict]] logger.debug(traceback.format_exc()) return None - def _parse_bigquery_query_source(self, source_expression: str) -> Optional[List[dict]]: # noqa: UP006, UP045 + def _parse_bigquery_query_source( + self, source_expression: str + ) -> Optional[List[dict]]: # noqa: UP006, UP045 """ Parse BigQuery Value.NativeQuery source expressions containing inline SQL. @@ -1021,8 +1189,12 @@ def _parse_bigquery_query_source(self, source_expression: str) -> Optional[List[ """ try: # Strip M language block comments (/* ... */) and line comments (//) - cleaned_expression = re.sub(r"/\*.*?\*/", "", source_expression, flags=re.DOTALL) - cleaned_expression = re.sub(SQL_LINE_COMMENT_PATTERN, "", cleaned_expression) + cleaned_expression = re.sub( + r"/\*.*?\*/", "", source_expression, flags=re.DOTALL + ) + cleaned_expression = re.sub( + SQL_LINE_COMMENT_PATTERN, "", cleaned_expression + ) # Extract the project from BillingProject parameter billing_match = re.search(r'BillingProject="([^"]+)"', cleaned_expression) @@ -1082,7 +1254,9 @@ def _parse_bigquery_query_source(self, source_expression: str) -> Optional[List[ table_name = source_table.raw_name if table_name: - logger.debug(f"BigQuery NativeQuery table found: {database}.{schema}.{table_name}") + logger.debug( + f"BigQuery NativeQuery table found: {database}.{schema}.{table_name}" + ) lineage_tables_list.append( { "database": database, @@ -1123,17 +1297,25 @@ def _parse_bigquery_source( ) if source_ref_match: - ref_name = source_ref_match.group(1).strip().strip('"').strip("#").strip('"') - logger.debug(f"Table source references expression: {ref_name}, resolving...") + ref_name = ( + source_ref_match.group(1).strip().strip('"').strip("#").strip('"') + ) + logger.debug( + f"Table source references expression: {ref_name}, resolving..." + ) # Fetch the dataset to get its expressions dataset = self.state.find_dataset(datamodel_entity.name.root) if dataset and dataset.expressions: for dexpression in dataset.expressions: if dexpression.name == ref_name and dexpression.expression: - logger.debug(f"Found referenced expression '{ref_name}', checking for BigQuery") + logger.debug( + f"Found referenced expression '{ref_name}', checking for BigQuery" + ) # Recursively parse the referenced expression - return self._parse_bigquery_source(dexpression.expression, datamodel_entity, table) + return self._parse_bigquery_source( + dexpression.expression, datamodel_entity, table + ) # Check if this is a direct BigQuery connection if "GoogleBigQuery.Database" not in source_expression: @@ -1157,7 +1339,9 @@ def _parse_bigquery_source( # Pattern: [Name="project"][Data][Name="dataset",Kind="Schema"][Data][Name="table",Kind="Table"] # Extract all Name= patterns - name_matches = re.findall(r'\[Name="([^"]+)"(?:,Kind="([^"]+)")?\]', source_expression) + name_matches = re.findall( + r'\[Name="([^"]+)"(?:,Kind="([^"]+)")?\]', source_expression + ) if not name_matches: logger.debug( @@ -1180,7 +1364,9 @@ def _parse_bigquery_source( # First Name without Kind is likely the project project = name - logger.debug(f"Extracted BigQuery info: project={project}, dataset={dataset}, table={table_name}") + logger.debug( + f"Extracted BigQuery info: project={project}, dataset={dataset}, table={table_name}" + ) if not table_name: logger.debug( "Table name not found in Parsing BigQuery source expression for " @@ -1196,13 +1382,17 @@ def _parse_bigquery_source( logger.debug(traceback.format_exc()) return None - def _parse_snowflake_query_source(self, source_expression: str) -> Optional[List[dict]]: # noqa: UP006, UP045 + def _parse_snowflake_query_source( + self, source_expression: str + ) -> Optional[List[dict]]: # noqa: UP006, UP045 """ Parse snowflake query source source expressions like `Value.NativeQuery(Snowflake.Databases())` """ try: - logger.debug(f"parsing source expression through query parser: {source_expression[:100]}") + logger.debug( + f"parsing source expression through query parser: {source_expression[:100]}" + ) # Look for SQL query after [Data], # The pattern needs to handle the concatenated strings with & operators @@ -1243,7 +1433,9 @@ def _parse_snowflake_query_source(self, source_expression: str) -> Optional[List # 4. Clean up excessive whitespace parser_query = re.sub(r"\s+", " ", parser_query).strip() - logger.debug(f"Attempting LineageParser with cleaned query: {parser_query[:200]}") + logger.debug( + f"Attempting LineageParser with cleaned query: {parser_query[:200]}" + ) try: parser = LineageParser( @@ -1259,10 +1451,14 @@ def _parse_snowflake_query_source(self, source_expression: str) -> Optional[List return None if parser.source_tables: - logger.debug(f"[{query_hash}] LineageParser found {len(parser.source_tables)} source table(s)") + logger.debug( + f"[{query_hash}] LineageParser found {len(parser.source_tables)} source table(s)" + ) for table in parser.source_tables: schema_name = table.schema if hasattr(table, "schema") else "N/A" - logger.debug(f"[{query_hash}] source table: {table.raw_name}, schema: {schema_name}") + logger.debug( + f"[{query_hash}] source table: {table.raw_name}, schema: {schema_name}" + ) lineage_tables_list = [] for source_table in parser.source_tables: # source_table = parser.source_tables[0] @@ -1273,7 +1469,9 @@ def _parse_snowflake_query_source(self, source_expression: str) -> Optional[List if hasattr(source_table, "schema") and source_table.schema: # Log what we have in the schema object - logger.debug(f"Schema object type: {type(source_table.schema)}, value: {source_table.schema}") + logger.debug( + f"Schema object type: {type(source_table.schema)}, value: {source_table.schema}" + ) # Get schema as string first schema_str = ( @@ -1288,7 +1486,11 @@ def _parse_snowflake_query_source(self, source_expression: str) -> Optional[List if len(parts) == 2: # Format: database.schema # Check for placeholder (case insensitive) - database = parts[0] if parts[0].upper() != "PLACEHOLDER_DB" else None + database = ( + parts[0] + if parts[0].upper() != "PLACEHOLDER_DB" + else None + ) schema = parts[1] else: # Just use as is @@ -1296,7 +1498,10 @@ def _parse_snowflake_query_source(self, source_expression: str) -> Optional[List else: schema = schema_str # Check if schema has a parent (database) - if hasattr(source_table.schema, "parent") and source_table.schema.parent: + if ( + hasattr(source_table.schema, "parent") + and source_table.schema.parent + ): database = ( source_table.schema.parent.raw_name if hasattr(source_table.schema.parent, "raw_name") @@ -1330,10 +1535,18 @@ def _parse_catalog_table_definition( self, source_expression: str, datamodel_entity: DashboardDataModel ) -> Optional[List[dict]]: # noqa: UP006, UP045 """parse catalog table definition""" - db_match = re.search(r'\[Name=(?:"([^"]+)"|([^,]+)),Kind="Database"\]', source_expression) - schema_match = re.search(r'\[Name=(?:"([^"]+)"|([^,]+)),Kind="Schema"\]', source_expression) - table_match = re.search(r'\[Name=(?:"([^"]+)"|([^,]+)),Kind="Table"\]', source_expression) - view_match = re.search(r'\[Name=(?:"([^"]+)"|([^,]+)),Kind="View"\]', source_expression) + db_match = re.search( + r'\[Name=(?:"([^"]+)"|([^,]+)),Kind="Database"\]', source_expression + ) + schema_match = re.search( + r'\[Name=(?:"([^"]+)"|([^,]+)),Kind="Schema"\]', source_expression + ) + table_match = re.search( + r'\[Name=(?:"([^"]+)"|([^,]+)),Kind="Table"\]', source_expression + ) + view_match = re.search( + r'\[Name=(?:"([^"]+)"|([^,]+)),Kind="View"\]', source_expression + ) try: database = self._parse_expression_regex_exp(db_match, datamodel_entity) schema = self._parse_expression_regex_exp(schema_match, datamodel_entity) @@ -1367,7 +1580,9 @@ def _parse_databricks_source( parser_type=self.get_query_parser_type(), ) else: # noqa: RET505 - return self._parse_catalog_table_definition(source_expression, datamodel_entity) + return self._parse_catalog_table_definition( + source_expression, datamodel_entity + ) except Exception as exc: logger.debug(f"Error to parse databricks table source: {exc}") logger.debug(traceback.format_exc()) @@ -1383,7 +1598,9 @@ def _parse_snowflake_source( if SNOWFLAKE_QUERY_EXPRESSION_KW in source_expression: # snowflake query source identified return self._parse_snowflake_query_source(source_expression) - return self._parse_catalog_table_definition(source_expression, datamodel_entity) + return self._parse_catalog_table_definition( + source_expression, datamodel_entity + ) except Exception as exc: logger.debug(f"Error to parse snowflake table source: {exc}") logger.debug(traceback.format_exc()) @@ -1401,7 +1618,9 @@ def _parse_table_info_from_source_exp( return None # parse snowflake source - table_info_list = self._parse_snowflake_source(source_expression, datamodel_entity) + table_info_list = self._parse_snowflake_source( + source_expression, datamodel_entity + ) if isinstance(table_info_list, List): # noqa: UP006 return table_info_list @@ -1411,12 +1630,16 @@ def _parse_table_info_from_source_exp( return table_info_list # parse bigquery source - table_info_list = self._parse_bigquery_source(source_expression, datamodel_entity, table) + table_info_list = self._parse_bigquery_source( + source_expression, datamodel_entity, table + ) if isinstance(table_info_list, List): # noqa: UP006 return table_info_list # parse databricks source - table_info_list = self._parse_databricks_source(source_expression, datamodel_entity) + table_info_list = self._parse_databricks_source( + source_expression, datamodel_entity + ) if isinstance(table_info_list, List): # noqa: UP006 return table_info_list @@ -1449,7 +1672,9 @@ def _get_table_and_datamodel_lineage( ) = self.parse_db_service_prefix(db_service_prefix) try: - table_info_list = self._parse_table_info_from_source_exp(table, datamodel_entity) + table_info_list = self._parse_table_info_from_source_exp( + table, datamodel_entity + ) if not table_info_list: # if tables are not found from source expression # try establishing lineage using powerbi's table name. @@ -1469,15 +1694,31 @@ def _get_table_and_datamodel_lineage( table_name = table_info.get("table") or table.name schema_name = table_info.get("schema") database_name = table_info.get("database") - if prefix_table_name and table_name and prefix_table_name.lower() != table_name.lower(): - logger.debug(f"Table {table_name} does not match prefix {prefix_table_name}") + if ( + prefix_table_name + and table_name + and prefix_table_name.lower() != table_name.lower() + ): + logger.debug( + f"Table {table_name} does not match prefix {prefix_table_name}" + ) return - if prefix_schema_name and schema_name and prefix_schema_name.lower() != schema_name.lower(): - logger.debug(f"Schema {table_info.get('schema')} does not match prefix {prefix_schema_name}") + if ( + prefix_schema_name + and schema_name + and prefix_schema_name.lower() != schema_name.lower() + ): + logger.debug( + f"Schema {table_info.get('schema')} does not match prefix {prefix_schema_name}" + ) return - if prefix_database_name and database_name and prefix_database_name.lower() != database_name.lower(): + if ( + prefix_database_name + and database_name + and prefix_database_name.lower() != database_name.lower() + ): logger.debug( f"Database {table_info.get('database')} does not match prefix {prefix_database_name}" ) @@ -1491,7 +1732,9 @@ def _get_table_and_datamodel_lineage( database_name=(prefix_database_name or database_name), ) except ValueError: - logger.debug(f"Skipping table '{table_name}' with invalid FQN characters") + logger.debug( + f"Skipping table '{table_name}' with invalid FQN characters" + ) continue table_entity = self.metadata.search_in_any_service( entity_type=Table, @@ -1503,8 +1746,14 @@ def _get_table_and_datamodel_lineage( table_entity.name.root, # pyright: ignore[reportAttributeAccessIssue] datamodel_entity.name.root, ) - columns_list = [column.name for column in (table.columns or []) if column.name] - column_lineage = self._get_column_lineage(table_entity, datamodel_entity, columns_list) + columns_list = [ + column.name + for column in (table.columns or []) + if column.name + ] + column_lineage = self._get_column_lineage( + table_entity, datamodel_entity, columns_list + ) yield self._get_add_lineage_request( to_entity=datamodel_entity, from_entity=table_entity, @@ -1530,18 +1779,22 @@ def create_table_datamodel_lineage_from_files( """ Method to create lineage between table and datamodels using pbit files """ - (prefix_service_name, *_) = self.parse_db_service_prefix(db_service_prefix) + prefix_service_name, *_ = self.parse_db_service_prefix(db_service_prefix) try: # check if the datamodel_file_mappings is populated or not # if not, then populate the datamodel_file_mappings and process the lineage if not self.datamodel_file_mappings: - self.datamodel_file_mappings = self.client.file_client.get_data_model_schema_mappings() + self.datamodel_file_mappings = ( + self.client.file_client.get_data_model_schema_mappings() + ) # search which file contains the datamodel and for the given datamodel_entity datamodel_file_list = [] for datamodel_schema in self.datamodel_file_mappings or []: - for connections in datamodel_schema.connectionFile.RemoteArtifacts or []: + for connections in ( + datamodel_schema.connectionFile.RemoteArtifacts or [] + ): if connections.DatasetId == model_str(datamodel_entity.name): datamodel_file_list.append(datamodel_schema) # noqa: PERF401 @@ -1570,13 +1823,17 @@ def _emit_om_target_lineage( target_ids: Iterable[Optional[str]], # noqa: UP045 target: LineageTargetSpec, error_name: str, - column_lineage_builder: Optional[Callable[..., Optional[List[ColumnLineage]]]] = None, # noqa: UP006, UP045 + column_lineage_builder: Optional[ + Callable[..., Optional[List[ColumnLineage]]] + ] = None, # noqa: UP006, UP045 ) -> Iterable[Either[AddLineageRequest]]: """Resolve target entities in OM and yield lineage from each into `to_entity`. Silent skip on falsy or missing target; failures surface as `Either.left`. """ - service_name = self.context.get().dashboard_service # pyright: ignore[reportAttributeAccessIssue] + service_name = ( + self.context.get().dashboard_service + ) # pyright: ignore[reportAttributeAccessIssue] for target_id in target_ids: if not target_id: logger.debug( @@ -1613,7 +1870,11 @@ def _emit_om_target_lineage( to_entity.name.root, ) continue - column_lineage = column_lineage_builder(to_entity, target_entity) if column_lineage_builder else None + column_lineage = ( + column_lineage_builder(to_entity, target_entity) + if column_lineage_builder + else None + ) lineage_request = self._get_add_lineage_request( from_entity=target_entity, to_entity=to_entity, @@ -1632,7 +1893,9 @@ def _emit_om_target_lineage( yield Either( left=StackTraceError( name=error_name, - error=(f"Error to yield {error_name} between [{to_entity.name.root}, {target_id!s}]: {exc}"), + error=( + f"Error to yield {error_name} between [{to_entity.name.root}, {target_id!s}]: {exc}" + ), stackTrace=traceback.format_exc(), ), right=None, @@ -1700,7 +1963,11 @@ def _create_dataset_upstream_dataset_column_lineage( column=column.name.root, ) if source_column and target_column: - column_lineage.append(ColumnLineage(fromColumns=[source_column], toColumn=target_column)) + column_lineage.append( + ColumnLineage( + fromColumns=[source_column], toColumn=target_column + ) + ) return column_lineage # noqa: TRY300 except Exception as exc: logger.debug(traceback.format_exc()) @@ -1725,7 +1992,9 @@ def create_dataset_upstream_dataset_lineage( column_lineage_builder=self._create_dataset_upstream_dataset_column_lineage, ) - def _parse_dataflow_m_document(self, dataflow_export: DataflowExportResponse) -> List[dict]: # noqa: UP006 + def _parse_dataflow_m_document( + self, dataflow_export: DataflowExportResponse + ) -> List[dict]: # noqa: UP006 """ Parse Power Query M expressions from the dataflow export document to extract table references for each entity/query in the dataflow. @@ -1756,7 +2025,9 @@ def _parse_dataflow_m_document(self, dataflow_export: DataflowExportResponse) -> # Only process entities that have loadEnabled=true in queriesMetadata query_meta = queries_metadata.get(entity_name, {}) - if isinstance(query_meta, dict) and not query_meta.get("loadEnabled", False): + if isinstance(query_meta, dict) and not query_meta.get( + "loadEnabled", False + ): continue table_info_list = self._parse_sql_source(block) @@ -1774,7 +2045,9 @@ def _parse_dataflow_m_document(self, dataflow_export: DataflowExportResponse) -> ) return results - def _parse_sql_source(self, m_expression: str) -> Optional[List[dict]]: # noqa: UP006, UP045 + def _parse_sql_source( + self, m_expression: str + ) -> Optional[List[dict]]: # noqa: UP006, UP045 """ Parse a Power Query M expression block from a dataflow document to extract database table references. Handles: @@ -1944,7 +2217,9 @@ def create_dataflow_table_lineage( try: parsed_entities = self._parse_dataflow_m_document(dataflow_export) if not parsed_entities: - logger.debug(f"No table references found in dataflow [{datamodel.name}] M document") + logger.debug( + f"No table references found in dataflow [{datamodel.name}] M document" + ) return # Build a map of entity_name -> entity attributes for column lineage. @@ -1960,7 +2235,9 @@ def create_dataflow_table_lineage( datamodel.name, ) continue - entity_attributes_map[entity.name] = [attr.name for attr in entity.attributes or [] if attr.name] + entity_attributes_map[entity.name] = [ + attr.name for attr in entity.attributes or [] if attr.name + ] for parsed_entity in parsed_entities: entity_name = parsed_entity["entity_name"] @@ -1974,13 +2251,25 @@ def create_dataflow_table_lineage( if not table_name: continue - if prefix_table_name and table_name and prefix_table_name.lower() != table_name.lower(): + if ( + prefix_table_name + and table_name + and prefix_table_name.lower() != table_name.lower() + ): continue - if prefix_schema_name and schema_name and prefix_schema_name.lower() != schema_name.lower(): + if ( + prefix_schema_name + and schema_name + and prefix_schema_name.lower() != schema_name.lower() + ): continue - if prefix_database_name and database_name and prefix_database_name.lower() != database_name.lower(): + if ( + prefix_database_name + and database_name + and prefix_database_name.lower() != database_name.lower() + ): continue try: fqn_search_string = build_es_fqn_search_string( @@ -1990,7 +2279,9 @@ def create_dataflow_table_lineage( database_name=prefix_database_name or database_name, ) except ValueError: - logger.debug(f"Skipping table '{table_name}' with invalid FQN characters") + logger.debug( + f"Skipping table '{table_name}' with invalid FQN characters" + ) continue table_entity = self.metadata.search_in_any_service( entity_type=Table, @@ -2001,7 +2292,9 @@ def create_dataflow_table_lineage( table_entity=table_entity, datamodel_entity=datamodel_entity, entity_name=entity_name, - entity_attributes=entity_attributes_map.get(entity_name, []), + entity_attributes=entity_attributes_map.get( + entity_name, [] + ), ) yield self._get_add_lineage_request( to_entity=datamodel_entity, @@ -2013,7 +2306,9 @@ def create_dataflow_table_lineage( yield Either( left=StackTraceError( name="Dataflow Table Lineage", - error=(f"Error to yield dataflow table lineage for dataflow [{datamodel.name}]: {exc}"), + error=( + f"Error to yield dataflow table lineage for dataflow [{datamodel.name}]: {exc}" + ), stackTrace=traceback.format_exc(), ) ) @@ -2033,14 +2328,18 @@ def _get_dataflow_column_lineage( try: column_lineage = [] for attr_name in entity_attributes: - from_column = get_column_fqn(table_entity=table_entity, column=attr_name) + from_column = get_column_fqn( + table_entity=table_entity, column=attr_name + ) to_column = self._get_downstream_data_model_column_fqn( data_model_entity=datamodel_entity, table_name=entity_name, column=attr_name, ) if from_column and to_column: - column_lineage.append(ColumnLineage(fromColumns=[from_column], toColumn=to_column)) + column_lineage.append( + ColumnLineage(fromColumns=[from_column], toColumn=to_column) + ) return column_lineage # noqa: TRY300 except Exception as exc: logger.debug(f"Error getting dataflow column lineage: {exc}") @@ -2069,7 +2368,7 @@ def yield_dashboard_lineage_details( We will build the logic to build the logic as below tables - datamodel - report - dashboard """ - (prefix_service_name, *_) = self.parse_db_service_prefix(db_service_prefix) + prefix_service_name, *_ = self.parse_db_service_prefix(db_service_prefix) for dashboard in self.state.filtered_dashboards: dashboard_details = self.get_dashboard_details(dashboard) @@ -2080,7 +2379,9 @@ def yield_dashboard_lineage_details( dashboard_details=dashboard_details, ) if isinstance(dashboard_details, PowerBIDashboard): - yield from self.create_report_dashboard_lineage(dashboard_details=dashboard_details) + yield from self.create_report_dashboard_lineage( + dashboard_details=dashboard_details + ) except Exception as exc: # pylint: disable=broad-except yield Either( left=StackTraceError( @@ -2122,9 +2423,13 @@ def yield_dashboard_lineage_details( datamodel_entity=datamodel_entity, ) # 2. dataset-upstreamDataflow lineage - yield from self.create_dataset_upstream_dataflow_lineage(datamodel, datamodel_entity) + yield from self.create_dataset_upstream_dataflow_lineage( + datamodel, datamodel_entity + ) # 3. dataset-upstreamDataset lineage - yield from self.create_dataset_upstream_dataset_lineage(datamodel, datamodel_entity) + yield from self.create_dataset_upstream_dataset_lineage( + datamodel, datamodel_entity + ) # create the lineage between table and datamodel using the pbit files if self.client.file_client: yield from self.create_table_datamodel_lineage_from_files( @@ -2142,9 +2447,13 @@ def yield_dashboard_lineage_details( db_service_prefix=db_service_prefix, ) # 6. dataflow-upstreamDataflow lineage - yield from self.create_dataflow_upstream_dataflow_lineage(datamodel, datamodel_entity) + yield from self.create_dataflow_upstream_dataflow_lineage( + datamodel, datamodel_entity + ) else: - logger.warning(f"Unknown datamodel type: {type(datamodel)}, name: {datamodel.name}") + logger.warning( + f"Unknown datamodel type: {type(datamodel)}, name: {datamodel.name}" + ) except Exception as exc: # pylint: disable=broad-except yield Either( left=StackTraceError( @@ -2161,8 +2470,12 @@ def yield_dashboard_lineage( """Flush the sink before lineage resolution so that target lookups in super().yield_dashboard_lineage see this workspace's just-flushed entities. """ - ws_id = self.context.get().workspace.id # pyright: ignore[reportAttributeAccessIssue] - yield Either(right=Barrier(reason=f"powerbi_ws:{ws_id}")) # pyright: ignore[reportCallIssue] + ws_id = ( + self.context.get().workspace.id + ) # pyright: ignore[reportAttributeAccessIssue] + yield Either( + right=Barrier(reason=f"powerbi_ws:{ws_id}") + ) # pyright: ignore[reportCallIssue] yield from super().yield_dashboard_lineage(dashboard_details) def yield_datamodel_dashboard_lineage( @@ -2182,10 +2495,14 @@ def get_project_name(self, dashboard_details: Any) -> Optional[str]: # noqa: UP Get the project / workspace / folder / collection name of the dashboard """ try: - return str(self.context.get().workspace.name) # pyright: ignore[reportAttributeAccessIssue] + return str( + self.context.get().workspace.name + ) # pyright: ignore[reportAttributeAccessIssue] except Exception as exc: logger.debug(traceback.format_exc()) - logger.warning(f"Error fetching project name for {dashboard_details.id}: {exc}") + logger.warning( + f"Error fetching project name for {dashboard_details.id}: {exc}" + ) return None def get_owner_ref( # pylint: disable=unused-argument, useless-return # noqa: C901 @@ -2196,7 +2513,9 @@ def get_owner_ref( # pylint: disable=unused-argument, useless-return # noqa: C """ try: if not self.source_config.includeOwners: - logger.debug(f"Skipping owner processing for {dashboard_details.id} as includeOwners is False") + logger.debug( + f"Skipping owner processing for {dashboard_details.id} as includeOwners is False" + ) return None owner_ref_list = [] # to assign multiple owners to entity if they exist for owner in dashboard_details.users or []: @@ -2216,7 +2535,10 @@ def get_owner_ref( # pylint: disable=unused-argument, useless-return # noqa: C f"User is not a member of {dashboard_details.id}: ({owner.displayName}, {owner.email})" ) continue - if access_right and any(keyword in access_right.lower() for keyword in OWNER_ACCESS_RIGHTS_KEYWORDS): + if access_right and any( + keyword in access_right.lower() + for keyword in OWNER_ACCESS_RIGHTS_KEYWORDS + ): if owner.email: try: owner_email = EmailStr._validate(owner.email) @@ -2225,7 +2547,9 @@ def get_owner_ref( # pylint: disable=unused-argument, useless-return # noqa: C owner_email = None if owner_email: try: - owner_ref = self.metadata.get_reference_by_email(owner_email.lower()) + owner_ref = self.metadata.get_reference_by_email( + owner_email.lower() + ) except Exception as err: logger.debug( f"Could not process owner data with email" @@ -2233,7 +2557,9 @@ def get_owner_ref( # pylint: disable=unused-argument, useless-return # noqa: C ) elif owner.displayName: try: - owner_ref = self.metadata.get_reference_by_name(name=owner.displayName) + owner_ref = self.metadata.get_reference_by_name( + name=owner.displayName + ) except Exception as err: logger.debug( f"Could not process owner data with name" @@ -2254,13 +2580,17 @@ def get_owner_ref( # pylint: disable=unused-argument, useless-return # noqa: C current_active_user = dashboard_details.modifiedBy if current_active_user: try: - owner_ref = self.metadata.get_reference_by_email(current_active_user.lower()) + owner_ref = self.metadata.get_reference_by_email( + current_active_user.lower() + ) if owner_ref and owner_ref.root[0] not in owner_ref_list: owner_ref_list.append(owner_ref.root[0]) except Exception as err: logger.debug(f"Could not fetch current active user due to {err}") if len(owner_ref_list) > 0: - logger.debug(f"Successfully fetched owners data for {dashboard_details.id}") + logger.debug( + f"Successfully fetched owners data for {dashboard_details.id}" + ) return EntityReferenceList(root=owner_ref_list) return None # noqa: TRY300 except Exception as err: diff --git a/ingestion/src/metadata/ingestion/source/dashboard/qlikcloud/metadata.py b/ingestion/src/metadata/ingestion/source/dashboard/qlikcloud/metadata.py index 113d26a1d9a6..1072a843d31e 100644 --- a/ingestion/src/metadata/ingestion/source/dashboard/qlikcloud/metadata.py +++ b/ingestion/src/metadata/ingestion/source/dashboard/qlikcloud/metadata.py @@ -312,20 +312,20 @@ def yield_dashboard_chart( ): self.status.filter(chart.qMeta.title, "Chart Pattern not allowed") continue - yield Either( - right=CreateChartRequest( - name=EntityName(chart.qInfo.qId), - displayName=chart.qMeta.title, - description=( - Markdown(chart.qMeta.description) - if chart.qMeta.description - else None - ), - chartType=ChartType.Other, - sourceUrl=SourceUrl(chart_url), - service=self.context.get().dashboard_service, - ) + chart_request = CreateChartRequest( + name=EntityName(chart.qInfo.qId), + displayName=chart.qMeta.title, + description=( + Markdown(chart.qMeta.description) + if chart.qMeta.description + else None + ), + 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: # pylint: disable=broad-except yield Either( left=StackTraceError( diff --git a/ingestion/src/metadata/ingestion/source/dashboard/qliksense/metadata.py b/ingestion/src/metadata/ingestion/source/dashboard/qliksense/metadata.py index cd7ae20c9b8b..bcb57d6c5cc4 100644 --- a/ingestion/src/metadata/ingestion/source/dashboard/qliksense/metadata.py +++ b/ingestion/src/metadata/ingestion/source/dashboard/qliksense/metadata.py @@ -189,22 +189,22 @@ def yield_dashboard_chart( ): self.status.filter(chart.qMeta.title, "Chart Pattern not allowed") continue - yield Either( - right=CreateChartRequest( - name=EntityName(chart.qInfo.qId), - displayName=chart.qMeta.title, - description=( - Markdown(chart.qMeta.description) - if chart.qMeta.description - else None - ), - chartType=ChartType.Other, - sourceUrl=SourceUrl(chart_url), - service=FullyQualifiedEntityName( - self.context.get().dashboard_service - ), - ) + chart_request = CreateChartRequest( + name=EntityName(chart.qInfo.qId), + displayName=chart.qMeta.title, + description=( + Markdown(chart.qMeta.description) + if chart.qMeta.description + else None + ), + chartType=ChartType.Other, + sourceUrl=SourceUrl(chart_url), + service=FullyQualifiedEntityName( + self.context.get().dashboard_service + ), ) + yield Either(right=chart_request) + self.register_record_chart(chart_request=chart_request) except Exception as exc: # pylint: disable=broad-except yield Either( left=StackTraceError( diff --git a/ingestion/src/metadata/ingestion/source/dashboard/quicksight/metadata.py b/ingestion/src/metadata/ingestion/source/dashboard/quicksight/metadata.py index a24e8266f3a9..9696d6c499e9 100644 --- a/ingestion/src/metadata/ingestion/source/dashboard/quicksight/metadata.py +++ b/ingestion/src/metadata/ingestion/source/dashboard/quicksight/metadata.py @@ -214,17 +214,17 @@ def yield_dashboard_chart( f"https://{self.aws_region}.quicksight.aws.amazon.com/sn/dashboards" f"/{dashboard_details.DashboardId}" ) - yield Either( - right=CreateChartRequest( - name=EntityName(chart.ChartId), - displayName=chart.Name, - chartType=ChartType.Other.value, - sourceUrl=SourceUrl(self.dashboard_url), - service=FullyQualifiedEntityName( - self.context.get().dashboard_service - ), - ) + chart_request = CreateChartRequest( + name=EntityName(chart.ChartId), + displayName=chart.Name, + chartType=ChartType.Other.value, + sourceUrl=SourceUrl(self.dashboard_url), + service=FullyQualifiedEntityName( + 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( diff --git a/ingestion/src/metadata/ingestion/source/dashboard/redash/metadata.py b/ingestion/src/metadata/ingestion/source/dashboard/redash/metadata.py index cc51bd2d0f98..4afb78f361aa 100644 --- a/ingestion/src/metadata/ingestion/source/dashboard/redash/metadata.py +++ b/ingestion/src/metadata/ingestion/source/dashboard/redash/metadata.py @@ -330,28 +330,28 @@ def yield_dashboard_chart( self.status.filter(chart_display_name, "Chart Pattern not allowed") continue - yield Either( - right=CreateChartRequest( - name=EntityName(str(widgets["id"])), - displayName=( - chart_display_name - if visualization and visualization["query"] - else "" - ), - chartType=get_standard_chart_type( - visualization["type"] if visualization else "" - ), - service=FullyQualifiedEntityName( - self.context.get().dashboard_service - ), - sourceUrl=SourceUrl(self.get_dashboard_url(dashboard_details)), - description=( - Markdown(visualization["description"]) - if visualization - else None - ), - ) + chart_request = CreateChartRequest( + name=EntityName(str(widgets["id"])), + displayName=( + chart_display_name + if visualization and visualization["query"] + else "" + ), + chartType=get_standard_chart_type( + visualization["type"] if visualization else "" + ), + service=FullyQualifiedEntityName( + self.context.get().dashboard_service + ), + sourceUrl=SourceUrl(self.get_dashboard_url(dashboard_details)), + description=( + Markdown(visualization["description"]) + if visualization + else None + ), ) + yield Either(right=chart_request) + self.register_record_chart(chart_request=chart_request) except Exception as exc: yield Either( left=StackTraceError( diff --git a/ingestion/src/metadata/ingestion/source/dashboard/sigma/metadata.py b/ingestion/src/metadata/ingestion/source/dashboard/sigma/metadata.py index 9375b644a6b4..1e20c674f45d 100644 --- a/ingestion/src/metadata/ingestion/source/dashboard/sigma/metadata.py +++ b/ingestion/src/metadata/ingestion/source/dashboard/sigma/metadata.py @@ -161,22 +161,22 @@ 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(str(chart.elementId)), - displayName=chart.name, - chartType=get_standard_chart_type(chart.vizualizationType), - service=FullyQualifiedEntityName( - self.context.get().dashboard_service - ), - sourceUrl=SourceUrl(dashboard_details.url), - description=( - Markdown(dashboard_details.description) - if dashboard_details.description - else None - ), - ) + chart_request = CreateChartRequest( + name=EntityName(str(chart.elementId)), + displayName=chart.name, + chartType=get_standard_chart_type(chart.vizualizationType), + service=FullyQualifiedEntityName( + self.context.get().dashboard_service + ), + sourceUrl=SourceUrl(dashboard_details.url), + description=( + Markdown(dashboard_details.description) + if dashboard_details.description + else None + ), ) + yield Either(right=chart_request) + self.register_record_chart(chart_request=chart_request) except Exception as exc: yield Either( left=StackTraceError( diff --git a/ingestion/src/metadata/ingestion/source/dashboard/ssrs/metadata.py b/ingestion/src/metadata/ingestion/source/dashboard/ssrs/metadata.py index 2e6160244c47..6be221396097 100644 --- a/ingestion/src/metadata/ingestion/source/dashboard/ssrs/metadata.py +++ b/ingestion/src/metadata/ingestion/source/dashboard/ssrs/metadata.py @@ -156,20 +156,20 @@ def yield_dashboard_chart( f"{clean_uri(self.service_connection.hostPort)}" f"/report{dashboard_details.path}" ) - yield Either( - right=CreateChartRequest( - name=EntityName(f"{dashboard_details.id}_chart"), - displayName=chart_name, - description=( - Markdown(dashboard_details.description) - if dashboard_details.description - else None - ), - chartType=ChartType.Other.value, - sourceUrl=SourceUrl(chart_url), - service=self.context.get().dashboard_service, - ) + chart_request = CreateChartRequest( + name=EntityName(f"{dashboard_details.id}_chart"), + displayName=chart_name, + description=( + Markdown(dashboard_details.description) + if dashboard_details.description + else None + ), + chartType=ChartType.Other.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 Exception as exc: yield Either( left=StackTraceError( diff --git a/ingestion/src/metadata/ingestion/source/dashboard/superset/api_source.py b/ingestion/src/metadata/ingestion/source/dashboard/superset/api_source.py index a4de8346ba4a..29bf19bb7a3f 100644 --- a/ingestion/src/metadata/ingestion/source/dashboard/superset/api_source.py +++ b/ingestion/src/metadata/ingestion/source/dashboard/superset/api_source.py @@ -158,7 +158,7 @@ def yield_dashboard_chart( f"chart details for id: {chart_id} not found, skipped" ) continue - chart = CreateChartRequest( + chart_request = CreateChartRequest( name=EntityName(str(chart_json.id)), displayName=chart_json.slice_name, description=( @@ -172,7 +172,8 @@ def yield_dashboard_chart( ), service=self.context.get().dashboard_service, ) - yield Either(right=chart) + yield Either(right=chart_request) + self.register_record_chart(chart_request=chart_request) except Exception as exc: # pylint: disable=broad-except yield Either( left=StackTraceError( diff --git a/ingestion/src/metadata/ingestion/source/dashboard/superset/db_source.py b/ingestion/src/metadata/ingestion/source/dashboard/superset/db_source.py index 8e4e3c04b747..1077b378ea4d 100644 --- a/ingestion/src/metadata/ingestion/source/dashboard/superset/db_source.py +++ b/ingestion/src/metadata/ingestion/source/dashboard/superset/db_source.py @@ -192,7 +192,7 @@ def yield_dashboard_chart( ) continue - chart = CreateChartRequest( + chart_request = CreateChartRequest( name=EntityName(str(chart_json.id)), displayName=chart_json.slice_name, description=( @@ -206,7 +206,8 @@ def yield_dashboard_chart( ), service=self.context.get().dashboard_service, ) - yield Either(right=chart) + yield Either(right=chart_request) + self.register_record_chart(chart_request=chart_request) except Exception as exc: yield Either( left=StackTraceError( diff --git a/ingestion/src/metadata/ingestion/source/dashboard/tableau/metadata.py b/ingestion/src/metadata/ingestion/source/dashboard/tableau/metadata.py index 8d9735cd1783..38cfe08167b9 100644 --- a/ingestion/src/metadata/ingestion/source/dashboard/tableau/metadata.py +++ b/ingestion/src/metadata/ingestion/source/dashboard/tableau/metadata.py @@ -808,7 +808,7 @@ def yield_dashboard_chart( f"/{workbook_chart_name.chart_url_name}" ) - chart = CreateChartRequest( + chart_request = CreateChartRequest( name=EntityName(chart.id), displayName=chart.name, chartType=get_standard_chart_type(chart.sheetType), @@ -823,7 +823,8 @@ def yield_dashboard_chart( self.context.get().dashboard_service ), ) - yield Either(right=chart) + yield Either(right=chart_request) + self.register_record_chart(chart_request=chart_request) except Exception as exc: yield Either( left=StackTraceError( diff --git a/ingestion/tests/unit/topology/dashboard/test_domodashboard.py b/ingestion/tests/unit/topology/dashboard/test_domodashboard.py index 2da0f2f9d5c2..7dc90c55f4aa 100644 --- a/ingestion/tests/unit/topology/dashboard/test_domodashboard.py +++ b/ingestion/tests/unit/topology/dashboard/test_domodashboard.py @@ -205,3 +205,12 @@ def test_chart(self): ) is None ) + + def test_chart_source_state_populated(self): + """Verify register_record_chart populates chart_source_state after yield_dashboard_chart.""" + self.domodashboard.chart_source_state = set() + with patch.object(REST, "_request", return_value=mock_data[0]): + list(self.domodashboard.yield_dashboard_chart(MOCK_DASHBOARD)) + assert len(self.domodashboard.chart_source_state) > 0 + for fqn in self.domodashboard.chart_source_state: + assert "domodashboard_source_test" in fqn diff --git a/ingestion/tests/unit/topology/dashboard/test_grafana.py b/ingestion/tests/unit/topology/dashboard/test_grafana.py index 3b2d5d6e309a..e51227c8f162 100644 --- a/ingestion/tests/unit/topology/dashboard/test_grafana.py +++ b/ingestion/tests/unit/topology/dashboard/test_grafana.py @@ -704,3 +704,11 @@ def test_complete_json_parsing(self): parsed_response = GrafanaDashboardResponse(**complete_json) self.assertEqual(parsed_response, expected_output) + + def test_chart_source_state_populated(self): + """Verify register_record_chart populates chart_source_state after yield_dashboard_chart.""" + self.grafana.chart_source_state = set() + list(self.grafana.yield_dashboard_chart(MOCK_DASHBOARD_RESPONSE)) + assert len(self.grafana.chart_source_state) == 3 + for fqn in self.grafana.chart_source_state: + assert "mock_grafana" in fqn diff --git a/ingestion/tests/unit/topology/dashboard/test_looker.py b/ingestion/tests/unit/topology/dashboard/test_looker.py index 22963702f262..86d8eeb3499f 100644 --- a/ingestion/tests/unit/topology/dashboard/test_looker.py +++ b/ingestion/tests/unit/topology/dashboard/test_looker.py @@ -962,3 +962,12 @@ def test_process_view_missing_view_yields_no_error(self): results = list(self.looker._process_view(ViewName("missing_view"), explore)) assert len(results) == 0 + + def test_chart_source_state_populated(self): + """Verify register_record_chart populates chart_source_state after yield_dashboard_chart.""" + self.looker.chart_source_state = set() + list(self.looker.yield_dashboard_chart(MOCK_LOOKER_DASHBOARD)) + assert len(self.looker.chart_source_state) == 1 + assert any( + "looker_source_test" in fqn for fqn in self.looker.chart_source_state + ) diff --git a/ingestion/tests/unit/topology/dashboard/test_microstrategy.py b/ingestion/tests/unit/topology/dashboard/test_microstrategy.py index f2e2161c9ed0..f90896feb48f 100644 --- a/ingestion/tests/unit/topology/dashboard/test_microstrategy.py +++ b/ingestion/tests/unit/topology/dashboard/test_microstrategy.py @@ -17,17 +17,27 @@ from unittest import TestCase from unittest.mock import patch +from metadata.generated.schema.entity.services.dashboardService import ( + DashboardConnection, + DashboardService, + DashboardServiceType, +) from metadata.generated.schema.metadataIngestion.workflow import ( OpenMetadataWorkflowConfig, ) +from metadata.generated.schema.type.basic import FullyQualifiedEntityName from metadata.ingestion.ometa.ometa_api import OpenMetadata from metadata.ingestion.source.dashboard.microstrategy.metadata import ( MicrostrategySource, ) from metadata.ingestion.source.dashboard.microstrategy.models import ( + MstrChapter, MstrDashboard, + MstrDashboardDetails, MstrOwner, + MstrPage, MstrProject, + MstrVisualization, ) mock_micro_config = { @@ -181,3 +191,49 @@ def test_include_owners_flag_with_owner_data(self): self.assertIsNotNone(dashboard.owner) self.assertEqual(dashboard.owner.name, "Administrator") self.assertEqual(dashboard.owner.id, "54F3D26011D2896560009A8E67019608") + + def test_chart_source_state_populated(self): + """Verify register_record_chart populates chart_source_state after yield_dashboard_chart.""" + MOCK_DASHBOARD_SERVICE = DashboardService( + id="c3eb265f-5445-4ad3-ba5e-797d3a3071bb", + name="mock_microstrategy", + fullyQualifiedName=FullyQualifiedEntityName("mock_microstrategy"), + connection=DashboardConnection(), + serviceType=DashboardServiceType.MicroStrategy, + ) + self.microstrategy.context.get().__dict__[ + "dashboard_service" + ] = MOCK_DASHBOARD_SERVICE.fullyQualifiedName.root + mock_details = MstrDashboardDetails( + id="dash1", + name="Test Dashboard", + projectId="proj1", + projectName="Test Project", + currentChapter="ch1", + chapters=[ + MstrChapter( + key="ch1", + name="Chapter 1", + pages=[ + MstrPage( + key="pg1", + name="Page 1", + visualizations=[ + MstrVisualization( + key="v1", name="Chart A", visualizationType="grid" + ), + MstrVisualization( + key="v2", name="Chart B", visualizationType="bar" + ), + ], + ) + ], + ) + ], + datasets=[], + ) + self.microstrategy.chart_source_state = set() + list(self.microstrategy.yield_dashboard_chart(mock_details)) + assert len(self.microstrategy.chart_source_state) == 2 + for fqn in self.microstrategy.chart_source_state: + assert "mock_microstrategy" in fqn diff --git a/ingestion/tests/unit/topology/dashboard/test_qlikcloud.py b/ingestion/tests/unit/topology/dashboard/test_qlikcloud.py index 64f3d26dea3c..8ea445aefc79 100644 --- a/ingestion/tests/unit/topology/dashboard/test_qlikcloud.py +++ b/ingestion/tests/unit/topology/dashboard/test_qlikcloud.py @@ -635,3 +635,14 @@ def test_get_data_files_api_failure(self): data_files = self.qlikcloud.client.get_data_files() assert data_files == [], "Expected empty list when API fails" + + def test_chart_source_state_populated(self): + """Verify register_record_chart populates chart_source_state after yield_dashboard_chart.""" + self.qlikcloud.chart_source_state = set() + with patch.object( + QlikCloudClient, "get_dashboard_charts", return_value=MOCK_CHARTS + ): + list(self.qlikcloud.yield_dashboard_chart(MOCK_DASHBOARD_DETAILS)) + assert len(self.qlikcloud.chart_source_state) == 2 + for fqn in self.qlikcloud.chart_source_state: + assert "qlikcloud_source_test" in fqn diff --git a/ingestion/tests/unit/topology/dashboard/test_quicksight.py b/ingestion/tests/unit/topology/dashboard/test_quicksight.py index 2fae7d73471c..4f1c0741bee2 100644 --- a/ingestion/tests/unit/topology/dashboard/test_quicksight.py +++ b/ingestion/tests/unit/topology/dashboard/test_quicksight.py @@ -388,3 +388,16 @@ def describe_data_set_side_effect(**kwargs): col_names_b = {col.name.root for col in dm_b.columns} assert col_names_b == {"email", "created_at"} + + def test_chart_source_state_populated(self): + """Verify register_record_chart populates chart_source_state after yield_dashboard_chart.""" + dashboard_details = DashboardDetail( + **{**MOCK_DASHBOARD_DETAILS, "Version": mock_data["Version"]} + ) + self.quicksight.chart_source_state = set() + list(self.quicksight.yield_dashboard_chart(dashboard_details)) + assert len(self.quicksight.chart_source_state) == len( + mock_data["Version"]["Sheets"] + ) + for fqn in self.quicksight.chart_source_state: + assert "quicksight_source_test" in fqn diff --git a/ingestion/tests/unit/topology/dashboard/test_ssrs.py b/ingestion/tests/unit/topology/dashboard/test_ssrs.py index 091e7eed8da9..44eb426e69e0 100644 --- a/ingestion/tests/unit/topology/dashboard/test_ssrs.py +++ b/ingestion/tests/unit/topology/dashboard/test_ssrs.py @@ -241,6 +241,13 @@ def test_yield_dashboard_lineage_is_noop(self, ssrs_source): result = ssrs_source.yield_dashboard_lineage_details(MOCK_REPORTS[0]) assert result is None + def test_chart_source_state_populated(self, ssrs_source): + """Verify register_record_chart populates chart_source_state after yield_dashboard_chart.""" + ssrs_source.chart_source_state = set() + list(ssrs_source.yield_dashboard_chart(MOCK_REPORTS[0])) + assert len(ssrs_source.chart_source_state) == 1 + assert any("mock_ssrs" in fqn for fqn in ssrs_source.chart_source_state) + class TestSsrsModels: def test_ssrs_report_parsing(self): diff --git a/openmetadata-spec/src/main/resources/json/schema/metadataIngestion/dashboardServiceMetadataPipeline.json b/openmetadata-spec/src/main/resources/json/schema/metadataIngestion/dashboardServiceMetadataPipeline.json index 1f37e8010323..4c697ec682d0 100644 --- a/openmetadata-spec/src/main/resources/json/schema/metadataIngestion/dashboardServiceMetadataPipeline.json +++ b/openmetadata-spec/src/main/resources/json/schema/metadataIngestion/dashboardServiceMetadataPipeline.json @@ -78,6 +78,12 @@ "default": true, "title": "Mark Deleted Data Models" }, + "markDeletedCharts": { + "description": "Optional configuration to soft delete charts in OpenMetadata if the source charts are deleted.", + "type": "boolean", + "default": true, + "title": "Mark Deleted Charts" + }, "includeTags": { "description": "Optional configuration to toggle the tags ingestion.", "type": "boolean", diff --git a/openmetadata-ui/src/main/resources/ui/src/generated/api/services/ingestionPipelines/createIngestionPipeline.ts b/openmetadata-ui/src/main/resources/ui/src/generated/api/services/ingestionPipelines/createIngestionPipeline.ts index 094565b2d2f6..4496c61f7453 100644 --- a/openmetadata-ui/src/main/resources/ui/src/generated/api/services/ingestionPipelines/createIngestionPipeline.ts +++ b/openmetadata-ui/src/main/resources/ui/src/generated/api/services/ingestionPipelines/createIngestionPipeline.ts @@ -511,6 +511,11 @@ export interface Pipeline { * Details required to generate Lineage */ lineageInformation?: LineageInformation; + /** + * Optional configuration to soft delete charts in OpenMetadata if the source charts are + * deleted. + */ + markDeletedCharts?: boolean; /** * Optional configuration to soft delete dashboards in OpenMetadata if the source dashboards * are deleted. Also, if the dashboard is deleted, all the associated entities like lineage, diff --git a/openmetadata-ui/src/main/resources/ui/src/generated/entity/services/ingestionPipelines/ingestionPipeline.ts b/openmetadata-ui/src/main/resources/ui/src/generated/entity/services/ingestionPipelines/ingestionPipeline.ts index 6d3bfc7c46da..136f6e0eb488 100644 --- a/openmetadata-ui/src/main/resources/ui/src/generated/entity/services/ingestionPipelines/ingestionPipeline.ts +++ b/openmetadata-ui/src/main/resources/ui/src/generated/entity/services/ingestionPipelines/ingestionPipeline.ts @@ -1150,6 +1150,11 @@ export interface Pipeline { * Details required to generate Lineage */ lineageInformation?: LineageInformation; + /** + * Optional configuration to soft delete charts in OpenMetadata if the source charts are + * deleted. + */ + markDeletedCharts?: boolean; /** * Optional configuration to soft delete dashboards in OpenMetadata if the source dashboards * are deleted. Also, if the dashboard is deleted, all the associated entities like lineage, diff --git a/openmetadata-ui/src/main/resources/ui/src/generated/metadataIngestion/dashboardServiceMetadataPipeline.ts b/openmetadata-ui/src/main/resources/ui/src/generated/metadataIngestion/dashboardServiceMetadataPipeline.ts index ced18ced628e..01a98d76ce82 100644 --- a/openmetadata-ui/src/main/resources/ui/src/generated/metadataIngestion/dashboardServiceMetadataPipeline.ts +++ b/openmetadata-ui/src/main/resources/ui/src/generated/metadataIngestion/dashboardServiceMetadataPipeline.ts @@ -49,6 +49,11 @@ export interface DashboardServiceMetadataPipeline { * Details required to generate Lineage */ lineageInformation?: LineageInformation; + /** + * Optional configuration to soft delete charts in OpenMetadata if the source charts are + * deleted. + */ + markDeletedCharts?: boolean; /** * Optional configuration to soft delete dashboards in OpenMetadata if the source dashboards * are deleted. Also, if the dashboard is deleted, all the associated entities like lineage, diff --git a/openmetadata-ui/src/main/resources/ui/src/generated/metadataIngestion/workflow.ts b/openmetadata-ui/src/main/resources/ui/src/generated/metadataIngestion/workflow.ts index 5ecd9b29c0f3..bce7af9a63d7 100644 --- a/openmetadata-ui/src/main/resources/ui/src/generated/metadataIngestion/workflow.ts +++ b/openmetadata-ui/src/main/resources/ui/src/generated/metadataIngestion/workflow.ts @@ -5388,6 +5388,11 @@ export interface Pipeline { * Details required to generate Lineage */ lineageInformation?: LineageInformation; + /** + * Optional configuration to soft delete charts in OpenMetadata if the source charts are + * deleted. + */ + markDeletedCharts?: boolean; /** * Optional configuration to soft delete dashboards in OpenMetadata if the source dashboards * are deleted. Also, if the dashboard is deleted, all the associated entities like lineage, From a00b5857e5d94bbaa1243dfcfbc07ed8126f4ad0 Mon Sep 17 00:00:00 2001 From: Joao Amaral <7281460+joaopamaral@users.noreply.github.com> Date: Tue, 2 Jun 2026 16:17:07 -0300 Subject: [PATCH 5/7] build(ingestion): bump build tag to superset.chart.bridge.2 Co-Authored-By: Claude Opus 4.8 --- ingestion/pyproject.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/ingestion/pyproject.toml b/ingestion/pyproject.toml index 2d5ddbb984c8..0c47f28e6a8d 100644 --- a/ingestion/pyproject.toml +++ b/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+superset.chart.bridge.1" +version = "1.12.9.0+superset.chart.bridge.2" dynamic = ["readme", "dependencies", "optional-dependencies"] authors = [ { name = "OpenMetadata Committers" } From 7d950046a389415715257cac20c2473f1d251e93 Mon Sep 17 00:00:00 2001 From: Joao Amaral <7281460+joaopamaral@users.noreply.github.com> Date: Thu, 11 Jun 2026 14:58:43 -0300 Subject: [PATCH 6/7] fix(lineage): resolve cross-catalog view->table lineage via db-wildcard fallback A view ingested in one catalog can reference a source table that lives under a different database name in another service -- e.g. a view in trino_views_hiyu (catalog hiyu) using a table ingested as trino_hiyu.iceberg... get_table_entities_from_query searched with the view's database (hiyu) for every service (incl. crossDatabaseServiceNames), so the table under database 'iceberg' never matched and no lineage edge was created. Add a final database=None fallback (after the exact and schema fallbacks) so the FQN search wildcards the database (.*..
) and resolves the table across the configured cross-database services. Scoped to service_names, so it only reaches the current service plus the explicitly-allowed crossDatabaseServiceNames. Co-Authored-By: Claude Opus 4.8 --- .../metadata/ingestion/lineage/sql_lineage.py | 30 ++++++++ .../lineage/test_cross_catalog_lineage.py | 76 +++++++++++++++++++ 2 files changed, 106 insertions(+) create mode 100644 ingestion/tests/unit/lineage/test_cross_catalog_lineage.py diff --git a/ingestion/src/metadata/ingestion/lineage/sql_lineage.py b/ingestion/src/metadata/ingestion/lineage/sql_lineage.py index 011d8f7cb4c8..9e51f952d1d5 100644 --- a/ingestion/src/metadata/ingestion/lineage/sql_lineage.py +++ b/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..
`. Retry with database=None so the FQN + # search wildcards the database (`.*..
`) 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 diff --git a/ingestion/tests/unit/lineage/test_cross_catalog_lineage.py b/ingestion/tests/unit/lineage/test_cross_catalog_lineage.py new file mode 100644 index 000000000000..97ae269142a0 --- /dev/null +++ b/ingestion/tests/unit/lineage/test_cross_catalog_lineage.py @@ -0,0 +1,76 @@ +# Copyright 2025 Collate +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# http://www.apache.org/licenses/LICENSE-2.0 +# 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. +""" +Unit tests for the cross-catalog (database-wildcard) fallback in +get_table_entities_from_query. + +A view in one catalog (e.g. `hiyu`) can reference a table that was ingested +under a different database name in another service (e.g. +`trino_hiyu.iceberg..
`). The exact and schema fallbacks search +with the view's database, miss it, and the run produced no lineage. The +database=None fallback wildcards the database so the table resolves across the +configured cross-database services. +""" + +from unittest.mock import MagicMock, patch + +from metadata.ingestion.lineage.sql_lineage import get_table_entities_from_query + + +class TestCrossCatalogFallback: + """get_table_entities_from_query must fall back to a wildcard database.""" + + def _run(self, side_effect): + metadata = MagicMock() + with patch( + "metadata.ingestion.lineage.sql_lineage.search_table_entities", + side_effect=side_effect, + ) as mock_search: + result = get_table_entities_from_query( + metadata=metadata, + service_names=["trino_views_hiyu", "trino_hiyu"], + database_name="hiyu", + database_schema="e2e_test", + table_name="test_current_date", + schema_fallback=True, + ) + return result, mock_search + + def test_database_wildcard_fallback_resolves_cross_catalog(self): + """When exact + schema searches (db=hiyu) miss, retry with db=None.""" + sentinel_table = MagicMock() + + def side_effect(*_, database=None, **__): + # Only the database-wildcard search resolves the table. + return [sentinel_table] if database is None else None + + result, mock_search = self._run(side_effect) + + assert result == [sentinel_table] + # At least one search must have used database=None (the wildcard fallback). + assert any( + call.kwargs.get("database") is None for call in mock_search.call_args_list + ) + + def test_returns_none_when_table_truly_absent(self): + """If no search ever resolves, the function still returns None.""" + result, _ = self._run(lambda *a, **k: None) + + assert result is None + + def test_exact_match_skips_fallbacks(self): + """A first-try hit (db=hiyu) returns without any wildcard search.""" + sentinel_table = MagicMock() + result, mock_search = self._run(lambda *a, **k: [sentinel_table]) + + assert result == [sentinel_table] + # Only the first (exact) search runs; no fallback needed. + assert mock_search.call_count == 1 From 21dee522c6d95bd833d020194b4d1f5cabd4dde2 Mon Sep 17 00:00:00 2001 From: Joao Amaral <7281460+joaopamaral@users.noreply.github.com> Date: Thu, 11 Jun 2026 14:58:43 -0300 Subject: [PATCH 7/7] build(ingestion): bump build tag to superset.chart.bridge.3 Co-Authored-By: Claude Opus 4.8 --- ingestion/pyproject.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/ingestion/pyproject.toml b/ingestion/pyproject.toml index 0c47f28e6a8d..0a2876f66e5b 100644 --- a/ingestion/pyproject.toml +++ b/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+superset.chart.bridge.2" +version = "1.12.9.0+superset.chart.bridge.3" dynamic = ["readme", "dependencies", "optional-dependencies"] authors = [ { name = "OpenMetadata Committers" }