feat(ingest/powerbi): PowerBI source updates

[mminichino]

• Fixes dataset column lineage
• Adds MySQL support
• Adds incremental lineage
• Adds ability to filter workspaces by name

Checklist

• The PR conforms to DataHub's Contributing Guideline (particularly Commit Message Format)
• Links to related issues (if applicable)
• Tests for the changes have been added/updated (if applicable)
• Docs related to the changes have been added/updated (if applicable). If a new feature has been added a Usage Guide has been added for the same.
• For any breaking change/potential downtime/deprecation/big changes an entry has been made in Updating DataHub

[codecov]

❌ 2 Tests Failed:

Tests completed	Failed	Passed	Skipped
3774	2	3772	68

View the full list of 2 ❄️ flaky tests

tests.entity_versioning.test_versioning::test_link_unlink_three_versions_unlink_middle_and_latest

Flake rate in main: 27.94% (Passed 49 times, Failed 19 times)

Stack Traces | 9.04s run time

graph_client = DataHubGraph: configured to talk to http://localhost:8080 with token: eyJh**********y_oA

    @pytest.fixture(scope="function", autouse=True)
    def ingest_cleanup_data(graph_client: DataHubGraph):
        try:
            for urn in ENTITY_URN_OBJS:
                graph_client.emit_mcp(
                    MetadataChangeProposalWrapper(
                        entityUrn=urn.urn(),
                        aspect=DatasetKeyClass(
                            platform=urn.platform, name=urn.name, origin=urn.env
                        ),
                    )
                )
            for i in [2, 1, 0, 0, 1, 2]:
>               graph_client.unlink_asset_from_version_set(ENTITY_URNS[i])

tests/entity_versioning/test_versioning.py:31: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
...../ingestion/graph/entity_versioning.py:164: in unlink_asset_from_version_set
    response = self.execute_graphql(self.UNLINK_VERSION_MUTATION, variables)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = DataHubGraph: configured to talk to http://localhost:8080 with token: eyJh**********y_oA
query = '\n        mutation($input: UnlinkVersionInput!) {\n            unlinkAssetVersion(input: $input) {\n                urn\n            }\n        }\n    '
variables = {'input': {'unlinkedEntity': 'urn:li:dataset:(urn:li:dataPlatform:snowflake,versioning_0,PROD)', 'versionSet': 'urn:li:versionSet:(12345678910,dataset)'}}
operation_name = None, format_exception = True

    def execute_graphql(
        self,
        query: str,
        variables: Optional[Dict] = None,
        operation_name: Optional[str] = None,
        format_exception: bool = True,
    ) -> Dict:
        url = f"{self.config.server}/api/graphql"
    
        body: Dict = {
            "query": query,
        }
        if variables:
            body["variables"] = variables
        if operation_name:
            body["operationName"] = operation_name
    
        logger.debug(
            f"Executing {operation_name or ''} graphql query: {query} with variables: {json.dumps(variables)}"
        )
        result = self._post_generic(url, body)
        if result.get("errors"):
            if format_exception:
>               raise GraphError(f"Error executing graphql query: {result['errors']}")
E               datahub.configuration.common.GraphError: Error executing graphql query: [{'message': 'An unknown error occurred.', 'locations': [{'line': 3, 'column': 13}], 'path': ['unlinkAssetVersion'], 'extensions': {'code': 500, 'type': 'SERVER_ERROR', 'classification': 'DataFetchingException'}}]

...../ingestion/graph/client.py:1188: GraphError

tests.entity_versioning.test_versioning::test_link_unlink_three_versions_unlink_and_relink

Flake rate in main: 27.94% (Passed 49 times, Failed 19 times)

Stack Traces | 29.8s run time

graph_client = DataHubGraph: configured to talk to http://localhost:8080 with token: eyJh**********y_oA

    @pytest.fixture(scope="function", autouse=True)
    def ingest_cleanup_data(graph_client: DataHubGraph):
        try:
            for urn in ENTITY_URN_OBJS:
                graph_client.emit_mcp(
                    MetadataChangeProposalWrapper(
                        entityUrn=urn.urn(),
                        aspect=DatasetKeyClass(
                            platform=urn.platform, name=urn.name, origin=urn.env
                        ),
                    )
                )
            for i in [2, 1, 0, 0, 1, 2]:
                graph_client.unlink_asset_from_version_set(ENTITY_URNS[i])
            yield
        finally:
            for i in [2, 1, 0, 0, 1, 2]:
>               graph_client.unlink_asset_from_version_set(ENTITY_URNS[i])

tests/entity_versioning/test_versioning.py:35: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
...../ingestion/graph/entity_versioning.py:164: in unlink_asset_from_version_set
    response = self.execute_graphql(self.UNLINK_VERSION_MUTATION, variables)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = DataHubGraph: configured to talk to http://localhost:8080 with token: eyJh**********y_oA
query = '\n        mutation($input: UnlinkVersionInput!) {\n            unlinkAssetVersion(input: $input) {\n                urn\n            }\n        }\n    '
variables = {'input': {'unlinkedEntity': 'urn:li:dataset:(urn:li:dataPlatform:snowflake,versioning_1,PROD)', 'versionSet': 'urn:li:versionSet:(12345678910,dataset)'}}
operation_name = None, format_exception = True

    def execute_graphql(
        self,
        query: str,
        variables: Optional[Dict] = None,
        operation_name: Optional[str] = None,
        format_exception: bool = True,
    ) -> Dict:
        url = f"{self.config.server}/api/graphql"
    
        body: Dict = {
            "query": query,
        }
        if variables:
            body["variables"] = variables
        if operation_name:
            body["operationName"] = operation_name
    
        logger.debug(
            f"Executing {operation_name or ''} graphql query: {query} with variables: {json.dumps(variables)}"
        )
        result = self._post_generic(url, body)
        if result.get("errors"):
            if format_exception:
>               raise GraphError(f"Error executing graphql query: {result['errors']}")
E               datahub.configuration.common.GraphError: Error executing graphql query: [{'message': 'An unknown error occurred.', 'locations': [{'line': 3, 'column': 13}], 'path': ['unlinkAssetVersion'], 'extensions': {'code': 500, 'type': 'SERVER_ERROR', 'classification': 'DataFetchingException'}}]

...../ingestion/graph/client.py:1188: GraphError

To view more test analytics, go to the Test Analytics Dashboard
_{📋 Got 3 mins? Take this short survey to help us improve Test Analytics.}

[sgomezvillamor]

To ensure a workspace is not filtered out, both the workspace ID and name must be allowed. Is that correct?
Similarly, if either of them is denied, the workspace will be filtered out, right?

If so, we may add such a precise explanation in the description.

IMO this confusion comes from the workspace_id_pattern, which should have never been a pattern. Given workspace id is a uuid-like, having regex patterns there is not practical. Instead, it should be workspace_ids: List[str], with the list of workspaces ids to be included, and then, only if not given, then the workspace_name_pattern would be considered.

We can skip workspace_id_pattern deprecation and just refine docs to avoid confusion.

[mminichino]

No, the filter is an OR condition. You can just specify a name filter, or an ID filter or both. Do if you just have a name filer, all IDs will be allowed. I'll update the docs.

[mminichino]

Docs updated.

[sgomezvillamor]

In this scenario and assuming name2 is not the name of the workspace id id1

workspace_id_pattern:
  allow:
  - id1
workspace_name_pattern:
  allow:
  - name2

No workspace will be allowed, right? that's why I see this as an AND. Of course, if we have allow all (the default) in any of the two, then everything depends on the other

And the same scenario:

workspace_id_pattern:
  deny:
  - id1
workspace_name_pattern:
  deny:
  - name2

Everything will be allowed but workspace id id1 and workspace name name2. So an OR in this case.

[mminichino]

If you allow one by ID and one by name both will be allowed. For the second scenario, if you deny one by ID and another by name, both will be denied.

[sgomezvillamor]

This is a sort of breaking change, why?

[mminichino]

This is actually not a breaking change. The API call is hard coded to return all the possible attributes, including dataset schema info so this should have been always set to true.

[sgomezvillamor]

Not sure about the API call, but emitting or not the aspect depends on that config flag

datahub/metadata-ingestion/src/datahub/ingestion/source/powerbi/powerbi.py

Lines 445 to 446 in 730f1b8

	if self.__config.extract_dataset_schema:
	dataset_mcps.extend(self.extract_dataset_schema(table, ds_urn))

[mminichino]

Correct, though the problem is this should have never been an option based on the way the source works. It is confusing at best. You can set extract_column_level_lineage but if you don't extract the schema data from the API scan result, you won't get column level lineage. However, the dataset schema data, if present, will always be returned.

            params={
                "datasetExpressions": True,
                "datasetSchema": True,
                "datasourceDetails": True,
                "getArtifactUsers": True,
                "lineage": True,
            },

[sgomezvillamor]

Overall looks good
I just miss some testing

Have you considered mocking some test in metadata-ingestion/tests/integration/powerbi/test_powerbi.py?
or has this been end to end tested locally somehow?

[mminichino]

My updates have been tested locally.

[sgomezvillamor]

CI failure

FAILED tests/integration/powerbi/test_profiling.py::test_profiling - Failed: Metadata files differ (use `pytest --update-golden-files` to update):
Urn changed, urn:li:dataset:(urn:li:dataPlatform:powerbi,library-dataset.articles,PROD):
<schemaMetadata> added
= 1 failed, 286 passed, 21 skipped, 1676 deselected, 196 warnings in 1166.63s (0:19:26) =

> Task :metadata-ingestion:testIntegrationBatch0 FAILED

I would say this is related to the breaking change in the extract_dataset_schema config

[mminichino]

The mocked API scan result does not match what the API is actually returning. But the test was wired for a specific input and output. I should just need to either add the setting or update the golden. I'll look at it.

[mminichino]

Yes, that golden does not include field data. However, in the documentation we say that Schema Metadata is enabled by default. So, in my view, that golden was incorrect from the start. That exact issue has been raised by several customers.

[hsheth2]

@mminichino sounds like we need to update the mock to include the fields, and then update/regenerate the golden file accordingly

[mminichino]

Yes, I am in the process of doing that

[mminichino]

Golden files updated. Integration test passes.