Merge branch 'datahub-project:master' into master

Author: hsheth2

.github/workflows/pr-labeler.yml

@@ -51,7 +51,8 @@ jobs:
                 "acrylJonny",
                 "chakru-r",
                 "brock-acryl",
-                "mminichino"
+                "mminichino",
+                "jayacryl"
               ]'), 
               github.actor
             ) 

datahub-web-react/src/app/entityV2/dataset/profile/schema/utils/utils.ts

@@ -69,6 +69,11 @@ export function pathMatchesNewPath(fieldPathA?: string | null, fieldPathB?: stri
     return fieldPathA === fieldPathB || fieldPathA === downgradeV2FieldPath(fieldPathB);
 }
 
+// should use pathMatchesExact when rendering editable info so the user edits the correct field
+export function pathMatchesExact(fieldPathA?: string | null, fieldPathB?: string | null) {
+    return fieldPathA === fieldPathB;
+}
+
 // group schema fields by fieldPath and grouping for hierarchy in schema table
 export function groupByFieldPath(
     schemaRows?: Array<SchemaField>,

datahub-web-react/src/app/entityV2/shared/tabs/Dataset/Schema/components/SchemaFieldDrawer/AboutFieldTab.tsx

@@ -10,7 +10,7 @@ import {
     UsageQueryResult,
 } from '../../../../../../../../types.generated';
 import { useMutationUrn } from '../../../../../../../entity/shared/EntityContext';
-import { pathMatchesNewPath } from '../../../../../../dataset/profile/schema/utils/utils';
+import { pathMatchesExact } from '../../../../../../dataset/profile/schema/utils/utils';
 import NotesSection from '../../../../../notes/NotesSection';
 import FieldDescription from './FieldDescription';
 import { FieldDetails } from './FieldDetails';
@@ -55,7 +55,7 @@ export function AboutFieldTab({ properties }: AboutFieldTabProps) {
             : undefined;
     const editableFieldInfo = properties.editableSchemaMetadata?.editableSchemaFieldInfo.find(
         (candidateEditableFieldInfo) =>
-            pathMatchesNewPath(candidateEditableFieldInfo.fieldPath, expandedField?.fieldPath),
+            pathMatchesExact(candidateEditableFieldInfo.fieldPath, expandedField?.fieldPath),
     );
 
     const notes = properties.notes?.sort((a, b) => moment(b.lastModified.time).diff(moment(a.lastModified.time))) || [];

datahub-web-react/src/app/entityV2/shared/tabs/Dataset/Schema/utils/useDescriptionRenderer.tsx

@@ -2,7 +2,7 @@ import React, { useState } from 'react';
 
 import { EditableSchemaMetadata, SchemaField, SubResourceType } from '../../../../../../../types.generated';
 import DescriptionField from '../../../../../dataset/profile/schema/components/SchemaDescriptionField';
-import { pathMatchesNewPath } from '../../../../../dataset/profile/schema/utils/utils';
+import { pathMatchesExact } from '../../../../../dataset/profile/schema/utils/utils';
 import { useUpdateDescriptionMutation } from '../../../../../../../graphql/mutations.generated';
 import { useMutationUrn, useRefetch } from '../../../../../../entity/shared/EntityContext';
 import { useSchemaRefetch } from '../SchemaContext';
@@ -31,7 +31,7 @@ export default function useDescriptionRenderer(
 
     return (description: string | undefined, record: SchemaField, index: number): JSX.Element => {
         const editableFieldInfo = editableSchemaMetadata?.editableSchemaFieldInfo.find((candidateEditableFieldInfo) =>
-            pathMatchesNewPath(candidateEditableFieldInfo.fieldPath, record.fieldPath),
+            pathMatchesExact(candidateEditableFieldInfo.fieldPath, record.fieldPath),
         );
         const { schemaFieldEntity } = record;
         const { displayedDescription, sanitizedDescription, isPropagated, sourceDetail } = extractFieldDescription(

datahub-web-react/src/app/entityV2/shared/tabs/Dataset/Schema/utils/useExtractFieldDescriptionInfo.ts

@@ -1,4 +1,4 @@
-import { pathMatchesNewPath } from '@src/app/entityV2/dataset/profile/schema/utils/utils';
+import { pathMatchesExact } from '@src/app/entityV2/dataset/profile/schema/utils/utils';
 import { EditableSchemaMetadata, SchemaField } from '@src/types.generated';
 import { getFieldDescriptionDetails } from './getFieldDescriptionDetails';
 import { sanitizeRichText } from '../../../Documentation/components/editor/utils';
@@ -8,7 +8,7 @@ export default function useExtractFieldDescriptionInfo(
 ) {
     return (record: SchemaField, description: string | undefined | null = null) => {
         const editableFieldInfoB = editableSchemaMetadata?.editableSchemaFieldInfo.find((candidateEditableFieldInfo) =>
-            pathMatchesNewPath(candidateEditableFieldInfo.fieldPath, record.fieldPath),
+            pathMatchesExact(candidateEditableFieldInfo.fieldPath, record.fieldPath),
         );
         const { displayedDescription, isPropagated, sourceDetail } = getFieldDescriptionDetails({
             schemaFieldEntity: record.schemaFieldEntity,

datahub-web-react/src/app/entityV2/shared/tabs/Dataset/Schema/utils/useExtractFieldGlossaryTermsInfo.ts

@@ -1,12 +1,12 @@
-import { pathMatchesNewPath } from '@src/app/entityV2/dataset/profile/schema/utils/utils';
+import { pathMatchesExact } from '@src/app/entityV2/dataset/profile/schema/utils/utils';
 import { EditableSchemaMetadata, GlossaryTerms, SchemaField } from '@src/types.generated';
 
 export default function useExtractFieldGlossaryTermsInfo(
     editableSchemaMetadata: EditableSchemaMetadata | null | undefined,
 ) {
     return (record: SchemaField, defaultUneditableTerms: GlossaryTerms | null = null) => {
         const editableTerms = editableSchemaMetadata?.editableSchemaFieldInfo.find((candidateEditableFieldInfo) =>
-            pathMatchesNewPath(candidateEditableFieldInfo.fieldPath, record.fieldPath),
+            pathMatchesExact(candidateEditableFieldInfo.fieldPath, record.fieldPath),
         )?.glossaryTerms;
 
         const uneditableTerms = defaultUneditableTerms || record?.glossaryTerms;

datahub-web-react/src/app/entityV2/shared/tabs/Dataset/Schema/utils/useExtractFieldTagsInfo.ts

@@ -1,10 +1,10 @@
-import { pathMatchesNewPath } from '@src/app/entityV2/dataset/profile/schema/utils/utils';
+import { pathMatchesExact } from '@src/app/entityV2/dataset/profile/schema/utils/utils';
 import { EditableSchemaMetadata, GlobalTags, SchemaField } from '@src/types.generated';
 
 export default function useExtractFieldTagsInfo(editableSchemaMetadata: EditableSchemaMetadata | null | undefined) {
     return (record: SchemaField, defaultUneditableTags: GlobalTags | null = null) => {
         const editableTags = editableSchemaMetadata?.editableSchemaFieldInfo.find((candidateEditableFieldInfo) =>
-            pathMatchesNewPath(candidateEditableFieldInfo.fieldPath, record.fieldPath),
+            pathMatchesExact(candidateEditableFieldInfo.fieldPath, record.fieldPath),
         )?.globalTags;
 
         const uneditableTags = defaultUneditableTags || record?.globalTags;

datahub-web-react/src/app/entityV2/shared/tabs/Dataset/Schema/utils/useTagsAndTermsRendererFeatureTable.tsx

@@ -1,7 +1,7 @@
 import React from 'react';
 import { EditableSchemaMetadata, EntityType, GlobalTags, SchemaField } from '../../../../../../../types.generated';
 import TagTermGroup from '../../../../../../sharedV2/tags/TagTermGroup';
-import { pathMatchesNewPath } from '../../../../../dataset/profile/schema/utils/utils';
+import { pathMatchesExact } from '../../../../../dataset/profile/schema/utils/utils';
 import { useEntityData, useRefetch } from '../../../../../../entity/shared/EntityContext';
 
 export default function useTagsAndTermsRendererFeatureTable(
@@ -15,7 +15,7 @@ export default function useTagsAndTermsRendererFeatureTable(
 
     const tagAndTermRender = (tags: GlobalTags, record: SchemaField, rowIndex: number | undefined) => {
         const relevantEditableFieldInfo = editableSchemaMetadata?.editableSchemaFieldInfo.find(
-            (candidateEditableFieldInfo) => pathMatchesNewPath(candidateEditableFieldInfo.fieldPath, record.fieldPath),
+            (candidateEditableFieldInfo) => pathMatchesExact(candidateEditableFieldInfo.fieldPath, record.fieldPath),
         );
 
         return (

metadata-ingestion/docs/sources/dynamodb/dynamodb_pre.md

@@ -1,6 +1,6 @@
 ### Prerequisities
 
-Notice of breaking change: in the latest version of the DynamoDB connector, `aws_region` is now a required configuration. The connector will no longer loop through all AWS regions; instead, it will only use the region passed into the recipe configuration.
+Notice of breaking change: Starting v0.13.3, `aws_region` is now a required configuration for DynamoDB Connector. The connector will no longer loop through all AWS regions; instead, it will only use the region passed into the recipe configuration.
 
 In order to execute this source, you need to attach the `AmazonDynamoDBReadOnlyAccess` policy to a user in your AWS account. Then create an API access key and secret for the user.
 
@@ -24,3 +24,10 @@ For a user to be able to create API access key, it needs the following access ke
   ]
 }
 ```
+
+### Concept Mapping
+
+| Source Concept | DataHub Concept                                           | Notes |
+| -------------- | --------------------------------------------------------- | ----- |
+| `"dynamodb"`   | [Data Platform](../../metamodel/entities/dataPlatform.md) |       |
+| DynamoDB Table | [Dataset](../../metamodel/entities/dataset.md)            |       |

metadata-ingestion/src/datahub/cli/iceberg_cli.py

@@ -14,6 +14,7 @@
 from datahub.configuration.common import GraphError
 from datahub.ingestion.graph.client import DataHubGraph, get_default_graph
 from datahub.metadata.schema_classes import SystemMetadataClass
+from datahub.telemetry import telemetry
 
 logger = logging.getLogger(__name__)
 
@@ -161,6 +162,7 @@ def validate_warehouse(data_root: str) -> None:
     type=int,
     help=f"Expiration duration for temporary credentials used for role. Defaults to {DEFAULT_CREDS_EXPIRY_DURATION_SECONDS} seconds if unspecified",
 )
+@telemetry.with_telemetry(capture_kwargs=["duration_seconds"])
 def create(
     warehouse: str,
     description: Optional[str],
@@ -313,6 +315,7 @@ def create(
     type=int,
     help=f"Expiration duration for temporary credentials used for role. Defaults to {DEFAULT_CREDS_EXPIRY_DURATION_SECONDS} seconds if unspecified",
 )
+@telemetry.with_telemetry(capture_kwargs=["duration_seconds"])
 def update(
     warehouse: str,
     data_root: str,
@@ -398,6 +401,7 @@ def update(
 
 
 @iceberg.command()
+@telemetry.with_telemetry()
 def list() -> None:
     """
     List iceberg warehouses
@@ -413,6 +417,7 @@ def list() -> None:
 @click.option(
     "-w", "--warehouse", required=True, type=str, help="The name of the warehouse"
 )
+@telemetry.with_telemetry()
 def get(warehouse: str) -> None:
     """Fetches the details of the specified iceberg warehouse"""
     client = get_default_graph()
@@ -442,6 +447,7 @@ def get(warehouse: str) -> None:
     is_flag=True,
     help="force the delete if set without confirmation",
 )
+@telemetry.with_telemetry(capture_kwargs=["dry_run", "force"])
 def delete(warehouse: str, dry_run: bool, force: bool) -> None:
     """
     Delete warehouse
@@ -470,11 +476,19 @@ def delete(warehouse: str, dry_run: bool, force: bool) -> None:
             # Do we really need this double-check?
             if "__typename" in entity and "urn" in entity:
                 if entity["__typename"] in ["Container", "Dataset"]:
+                    # add the Platform Resource URN to also be deleted for each dataset.
+                    # This is not user visible, so no need to show a name to the user and include it in the count. Each
+                    # instance corresponds to a dataset whose name is shown.
+                    if entity["__typename"] == "Dataset":
+                        resource_urn = platform_resource_urn(
+                            entity["properties"]["qualifiedName"]
+                        )
+                        urns_to_delete.append(resource_urn)
+
                     urns_to_delete.append(entity["urn"])
                     resource_names_to_be_deleted.append(
                         entity.get("name", entity.get("urn"))
                     )
-            # TODO: PlatformResource associated with datasets need to be deleted.
 
         if dry_run:
             click.echo(
@@ -485,25 +499,32 @@ def delete(warehouse: str, dry_run: bool, force: bool) -> None:
         else:
             if not force:
                 click.confirm(
-                    f"This will delete {warehouse} warehouse, credentials, and {len(urns_to_delete)} datasets and namespaces from DataHub. Do you want to continue?",
+                    f"This will delete {warehouse} warehouse, credentials, and {len(resource_names_to_be_deleted)} datasets and namespaces from DataHub. Do you want to continue?",
                     abort=True,
                 )
-            client.hard_delete_entity(urn)
-            client.hard_delete_entity(warehouse_aspect.clientId)
-            client.hard_delete_entity(warehouse_aspect.clientSecret)
 
+            # Delete the resources in the warehouse first, so that in case it is interrupted, the warehouse itself is
+            # still available to enumerate the resources in it that are not yet deleted.
             for urn_to_delete in urns_to_delete:
                 client.hard_delete_entity(urn_to_delete)
 
+            client.hard_delete_entity(urn)
+            client.hard_delete_entity(warehouse_aspect.clientId)
+            client.hard_delete_entity(warehouse_aspect.clientSecret)
+
             click.echo(
-                f"✅ Successfully deleted iceberg warehouse {warehouse} and associated credentials, {len(urns_to_delete)} datasets and namespaces"
+                f"✅ Successfully deleted iceberg warehouse {warehouse} and associated credentials, {len(resource_names_to_be_deleted)} datasets and namespaces"
             )
 
 
 def iceberg_data_platform_instance_urn(warehouse: str) -> str:
     return f"urn:li:dataPlatformInstance:({iceberg_data_platform()},{warehouse})"
 
 
+def platform_resource_urn(dataset_name: str) -> str:
+    return f"urn:li:platformResource:iceberg.{dataset_name}"
+
+
 def iceberg_data_platform() -> str:
     return "urn:li:dataPlatform:iceberg"
 
@@ -677,6 +698,9 @@ def get_related_entities_for_platform_instance(
                 ... on Dataset {
                   urn
                   name
+                  properties{
+                    qualifiedName
+                  }
                 }
               }
             }

metadata-ingestion/src/datahub/ingestion/source/dbt/dbt_common.py

@@ -357,6 +357,11 @@ class DBTCommonConfig(
         default=True,
         description="When enabled, includes the compiled code in the emitted metadata.",
     )
+    include_database_name: bool = Field(
+        default=True,
+        description="Whether to add database name to the table urn. "
+        "Set to False to skip it for engines like AWS Athena where it's not required.",
+    )
 
     @validator("target_platform")
     def validate_target_platform_value(cls, target_platform: str) -> str:

metadata-ingestion/src/datahub/ingestion/source/dbt/dbt_core.py

@@ -167,6 +167,7 @@ def extract_dbt_entities(
     use_identifiers: bool,
     tag_prefix: str,
     only_include_if_in_catalog: bool,
+    include_database_name: bool,
     report: DBTSourceReport,
 ) -> List[DBTNode]:
     sources_by_id = {x["unique_id"]: x for x in sources_results}
@@ -267,7 +268,7 @@ def extract_dbt_entities(
             dbt_name=key,
             dbt_adapter=manifest_adapter,
             dbt_package_name=manifest_node.get("package_name"),
-            database=manifest_node["database"],
+            database=manifest_node["database"] if include_database_name else None,
             schema=manifest_node["schema"],
             name=name,
             alias=manifest_node.get("alias"),
@@ -543,14 +544,15 @@ def loadManifestAndCatalog(
         all_catalog_entities = {**catalog_nodes, **catalog_sources}
 
         nodes = extract_dbt_entities(
-            all_manifest_entities,
-            all_catalog_entities,
-            sources_results,
-            manifest_adapter,
-            self.config.use_identifiers,
-            self.config.tag_prefix,
-            self.config.only_include_if_in_catalog,
-            self.report,
+            all_manifest_entities=all_manifest_entities,
+            all_catalog_entities=all_catalog_entities,
+            sources_results=sources_results,
+            manifest_adapter=manifest_adapter,
+            use_identifiers=self.config.use_identifiers,
+            tag_prefix=self.config.tag_prefix,
+            only_include_if_in_catalog=self.config.only_include_if_in_catalog,
+            include_database_name=self.config.include_database_name,
+            report=self.report,
         )
 
         return (

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

@@ -165,6 +165,10 @@ def report_dropped(self, name: str) -> None:
     SourceCapability.PLATFORM_INSTANCE,
     "By default, platform_instance will use the AWS account id",
 )
+@capability(
+    SourceCapability.CLASSIFICATION,
+    "Optionally enabled via `classification.enabled`",
+)
 class DynamoDBSource(StatefulIngestionSourceBase):
     """
     This plugin extracts the following:
@@ -244,6 +248,7 @@ def _process_table(
             name=dataset_name,
         )
         dataset_properties = DatasetPropertiesClass(
+            name=table_name,
             tags=[],
             customProperties={
                 "table.arn": table_info["TableArn"],

metadata-ingestion/tests/integration/dynamodb/dynamodb_default_platform_instance_mces_golden.json

@@ -36,7 +36,8 @@
                 },
                 {
                     "fieldPath": "city",
-                    "nullable": true,
+                    "nullable": false,
+                    "description": "Sort Key",
                     "type": {
                         "type": {
                             "com.linkedin.schema.StringType": {}
@@ -175,6 +176,7 @@
                 "table.arn": "arn:aws:dynamodb:us-west-2:123456789012:table/Location",
                 "table.totalItems": "1"
             },
+            "name": "Location",
             "tags": []
         }
     },

metadata-ingestion/tests/integration/dynamodb/dynamodb_platform_instance_mces_golden.json

@@ -36,7 +36,8 @@
                 },
                 {
                     "fieldPath": "city",
-                    "nullable": true,
+                    "nullable": false,
+                    "description": "Sort Key",
                     "type": {
                         "type": {
                             "com.linkedin.schema.StringType": {}
@@ -186,6 +187,7 @@
                 "table.arn": "arn:aws:dynamodb:us-west-2:123456789012:table/Location",
                 "table.totalItems": "1"
             },
+            "name": "Location",
             "tags": []
         }
     },

metadata-ingestion/tests/integration/dynamodb/test_dynamodb.py

@@ -29,9 +29,11 @@ def test_dynamodb(pytestconfig, tmp_path):
         TableName="Location",
         KeySchema=[
             {"AttributeName": "partitionKey", "KeyType": "HASH"},
+            {"AttributeName": "city", "KeyType": "RANGE"},
         ],
         AttributeDefinitions=[
             {"AttributeName": "partitionKey", "AttributeType": "S"},
+            {"AttributeName": "city", "AttributeType": "S"},
         ],
         ProvisionedThroughput={"ReadCapacityUnits": 10, "WriteCapacityUnits": 10},
     )