yujunjun/datahub
2
0
Fork 0
mirror of https://github.com/datahub-project/datahub.git synced 2025-12-28 02:17:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(ingest): check ordering in SqlParsingAggregator tests (#11735)

Browse Source
This commit is contained in:
Harshal Sheth 2024-10-29 09:50:37 -07:00 committed by GitHub
parent 91fbd12f84
commit 6316e10d48
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
9 changed files with 271 additions and 262 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
metadata-ingestion/src/datahub/sql_parsing/sql_parsing_aggregator.py
View File

@ -134,7 +134,7 @@ class QueryMetadata:
upstreams: List[UrnStr] # this is direct upstreams, which may be temp tables
column_lineage: List[ColumnLineageInfo]
column_usage: Dict[UrnStr, Set[UrnStr]]
column_usage: Dict[UrnStr, Set[UrnStr]] # TODO: Change to an OrderedSet
confidence_score: float
used_temp_tables: bool = True
@ -1426,7 +1426,7 @@ class SqlParsingAggregator(Closeable):
for upstream in query.upstreams:
query_subject_urns.add(upstream)
if self.generate_query_subject_fields:
for column in query.column_usage.get(upstream, []):
for column in sorted(query.column_usage.get(upstream, [])):
query_subject_urns.add(
builder.make_schema_field_urn(upstream, column)
)

4
metadata-ingestion/tests/test_helpers/mce_helpers.py
View File

@ -82,6 +82,7 @@ def check_golden_file(
golden_path: Union[str, os.PathLike],
ignore_paths: Sequence[str] = (),
ignore_paths_v2: Sequence[str] = (),
ignore_order: bool = True,
) -> None:
update_golden = pytestconfig.getoption("--update-golden-files")
copy_output = pytestconfig.getoption("--copy-output-files")
@ -92,6 +93,7 @@ def check_golden_file(
copy_output=copy_output,
ignore_paths=ignore_paths,
ignore_paths_v2=ignore_paths_v2,
ignore_order=ignore_order,
)
@ -100,6 +102,7 @@ def check_goldens_stream(
outputs: List,
golden_path: Union[str, os.PathLike],
ignore_paths: Sequence[str] = (),
ignore_order: bool = True,
) -> None:
with tempfile.NamedTemporaryFile() as f:
write_metadata_file(pathlib.Path(f.name), outputs)
@ -109,6 +112,7 @@ def check_goldens_stream(
output_path=f.name,
golden_path=golden_path,
ignore_paths=ignore_paths,
ignore_order=ignore_order,
)

40
metadata-ingestion/tests/unit/sql_parsing/aggregator_goldens/test_add_known_query_lineage.json
View File

@ -85,26 +85,6 @@
}
}
},
{
"entityType": "dataset",
"entityUrn": "urn:li:dataset:(urn:li:dataPlatform:redshift,dev.public.foo,PROD)",
"changeType": "UPSERT",
"aspectName": "operation",
"aspect": {
"json": {
"timestampMillis": 1707182625000,
"partitionSpec": {
"type": "FULL_TABLE",
"partition": "FULL_TABLE_SNAPSHOT"
},
"operationType": "INSERT",
"customProperties": {
"query_urn": "urn:li:query:6ed1d12fbf2ccc8138ceec08cc35b981030d6d004bfad9743c7afd84260fa63f"
},
"lastUpdatedTimestamp": 20000
}
}
},
{
"entityType": "query",
"entityUrn": "urn:li:query:6ed1d12fbf2ccc8138ceec08cc35b981030d6d004bfad9743c7afd84260fa63f",
@ -142,5 +122,25 @@
"platform": "urn:li:dataPlatform:redshift"
}
}
},
{
"entityType": "dataset",
"entityUrn": "urn:li:dataset:(urn:li:dataPlatform:redshift,dev.public.foo,PROD)",
"changeType": "UPSERT",
"aspectName": "operation",
"aspect": {
"json": {
"timestampMillis": 1707182625000,
"partitionSpec": {
"partition": "FULL_TABLE_SNAPSHOT",
"type": "FULL_TABLE"
},
"operationType": "INSERT",
"customProperties": {
"query_urn": "urn:li:query:6ed1d12fbf2ccc8138ceec08cc35b981030d6d004bfad9743c7afd84260fa63f"
},
"lastUpdatedTimestamp": 20000
}
}
}
]

4
metadata-ingestion/tests/unit/sql_parsing/aggregator_goldens/test_column_lineage_deduplication.json
View File

@ -164,10 +164,10 @@
"entity": "urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:redshift,dev.public.bar,PROD),a)"
},
{
"entity": "urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:redshift,dev.public.bar,PROD),c)"
"entity": "urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:redshift,dev.public.bar,PROD),b)"
},
{
"entity": "urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:redshift,dev.public.bar,PROD),b)"
"entity": "urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:redshift,dev.public.bar,PROD),c)"
},
{
"entity": "urn:li:dataset:(urn:li:dataPlatform:redshift,dev.public.foo,PROD)"

194
metadata-ingestion/tests/unit/sql_parsing/aggregator_goldens/test_table_rename.json
View File

@ -227,59 +227,6 @@
}
}
},
{
"entityType": "query",
"entityUrn": "urn:li:query:234a2904c367a6cc02d76cf358cd86937ec9e14af03e5539b5edb0b6df5db3dc",
"changeType": "UPSERT",
"aspectName": "queryProperties",
"aspect": {
"json": {
"statement": {
"value": "CREATE TABLE foo_staging AS\nSELECT\n a,\n b\nFROM foo_dep",
"language": "SQL"
},
"source": "SYSTEM",
"created": {
"time": 0,
"actor": "urn:li:corpuser:_ingestion"
},
"lastModified": {
"time": 1707182625000,
"actor": "urn:li:corpuser:_ingestion"
}
}
}
},
{
"entityType": "query",
"entityUrn": "urn:li:query:234a2904c367a6cc02d76cf358cd86937ec9e14af03e5539b5edb0b6df5db3dc",
"changeType": "UPSERT",
"aspectName": "querySubjects",
"aspect": {
"json": {
"subjects": [
{
"entity": "urn:li:dataset:(urn:li:dataPlatform:redshift,dev.public.foo_dep,PROD)"
},
{
"entity": "urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:redshift,dev.public.foo_dep,PROD),b)"
},
{
"entity": "urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:redshift,dev.public.foo_dep,PROD),a)"
},
{
"entity": "urn:li:dataset:(urn:li:dataPlatform:redshift,dev.public.foo_staging,PROD)"
},
{
"entity": "urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:redshift,dev.public.foo_staging,PROD),a)"
},
{
"entity": "urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:redshift,dev.public.foo_staging,PROD),b)"
}
]
}
}
},
{
"entityType": "query",
"entityUrn": "urn:li:query:a30d42497a737321ece461fa17344c3ba3588fdee736016acb59a00cec955a0c",
@ -291,17 +238,6 @@
}
}
},
{
"entityType": "query",
"entityUrn": "urn:li:query:234a2904c367a6cc02d76cf358cd86937ec9e14af03e5539b5edb0b6df5db3dc",
"changeType": "UPSERT",
"aspectName": "dataPlatformInstance",
"aspect": {
"json": {
"platform": "urn:li:dataPlatform:redshift"
}
}
},
{
"entityType": "dataset",
"entityUrn": "urn:li:dataset:(urn:li:dataPlatform:redshift,dev.public.foo_downstream,PROD)",
@ -353,6 +289,70 @@
}
}
},
{
"entityType": "query",
"entityUrn": "urn:li:query:e4b3b60ab99e0f0bc1629ea82a5d7705a30dbd98a3923d599b39fb68624ea58d",
"changeType": "UPSERT",
"aspectName": "queryProperties",
"aspect": {
"json": {
"statement": {
"value": "CREATE TABLE foo_downstream AS\nSELECT\n a,\n b\nFROM foo_staging",
"language": "SQL"
},
"source": "SYSTEM",
"created": {
"time": 0,
"actor": "urn:li:corpuser:_ingestion"
},
"lastModified": {
"time": 1707182625000,
"actor": "urn:li:corpuser:_ingestion"
}
}
}
},
{
"entityType": "query",
"entityUrn": "urn:li:query:e4b3b60ab99e0f0bc1629ea82a5d7705a30dbd98a3923d599b39fb68624ea58d",
"changeType": "UPSERT",
"aspectName": "querySubjects",
"aspect": {
"json": {
"subjects": [
{
"entity": "urn:li:dataset:(urn:li:dataPlatform:redshift,dev.public.foo_staging,PROD)"
},
{
"entity": "urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:redshift,dev.public.foo_staging,PROD),a)"
},
{
"entity": "urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:redshift,dev.public.foo_staging,PROD),b)"
},
{
"entity": "urn:li:dataset:(urn:li:dataPlatform:redshift,dev.public.foo_downstream,PROD)"
},
{
"entity": "urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:redshift,dev.public.foo_downstream,PROD),a)"
},
{
"entity": "urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:redshift,dev.public.foo_downstream,PROD),b)"
}
]
}
}
},
{
"entityType": "query",
"entityUrn": "urn:li:query:e4b3b60ab99e0f0bc1629ea82a5d7705a30dbd98a3923d599b39fb68624ea58d",
"changeType": "UPSERT",
"aspectName": "dataPlatformInstance",
"aspect": {
"json": {
"platform": "urn:li:dataPlatform:redshift"
}
}
},
{
"entityType": "dataset",
"entityUrn": "urn:li:dataset:(urn:li:dataPlatform:redshift,dev.public.foo_staging,PROD)",
@ -406,13 +406,13 @@
},
{
"entityType": "query",
"entityUrn": "urn:li:query:e4b3b60ab99e0f0bc1629ea82a5d7705a30dbd98a3923d599b39fb68624ea58d",
"entityUrn": "urn:li:query:234a2904c367a6cc02d76cf358cd86937ec9e14af03e5539b5edb0b6df5db3dc",
"changeType": "UPSERT",
"aspectName": "queryProperties",
"aspect": {
"json": {
"statement": {
"value": "CREATE TABLE foo_downstream AS\nSELECT\n a,\n b\nFROM foo_staging",
"value": "CREATE TABLE foo_staging AS\nSELECT\n a,\n b\nFROM foo_dep",
"language": "SQL"
},
"source": "SYSTEM",
@ -429,7 +429,37 @@
},
{
"entityType": "query",
"entityUrn": "urn:li:query:e4b3b60ab99e0f0bc1629ea82a5d7705a30dbd98a3923d599b39fb68624ea58d",
"entityUrn": "urn:li:query:234a2904c367a6cc02d76cf358cd86937ec9e14af03e5539b5edb0b6df5db3dc",
"changeType": "UPSERT",
"aspectName": "querySubjects",
"aspect": {
"json": {
"subjects": [
{
"entity": "urn:li:dataset:(urn:li:dataPlatform:redshift,dev.public.foo_dep,PROD)"
},
{
"entity": "urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:redshift,dev.public.foo_dep,PROD),a)"
},
{
"entity": "urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:redshift,dev.public.foo_dep,PROD),b)"
},
{
"entity": "urn:li:dataset:(urn:li:dataPlatform:redshift,dev.public.foo_staging,PROD)"
},
{
"entity": "urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:redshift,dev.public.foo_staging,PROD),a)"
},
{
"entity": "urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:redshift,dev.public.foo_staging,PROD),b)"
}
]
}
}
},
{
"entityType": "query",
"entityUrn": "urn:li:query:234a2904c367a6cc02d76cf358cd86937ec9e14af03e5539b5edb0b6df5db3dc",
"changeType": "UPSERT",
"aspectName": "dataPlatformInstance",
"aspect": {
@ -437,35 +467,5 @@
"platform": "urn:li:dataPlatform:redshift"
}
}
},
{
"entityType": "query",
"entityUrn": "urn:li:query:e4b3b60ab99e0f0bc1629ea82a5d7705a30dbd98a3923d599b39fb68624ea58d",
"changeType": "UPSERT",
"aspectName": "querySubjects",
"aspect": {
"json": {
"subjects": [
{
"entity": "urn:li:dataset:(urn:li:dataPlatform:redshift,dev.public.foo_staging,PROD)"
},
{
"entity": "urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:redshift,dev.public.foo_staging,PROD),b)"
},
{
"entity": "urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:redshift,dev.public.foo_staging,PROD),a)"
},
{
"entity": "urn:li:dataset:(urn:li:dataPlatform:redshift,dev.public.foo_downstream,PROD)"
},
{
"entity": "urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:redshift,dev.public.foo_downstream,PROD),a)"
},
{
"entity": "urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:redshift,dev.public.foo_downstream,PROD),b)"
}
]
}
}
}
]

4
metadata-ingestion/tests/unit/sql_parsing/aggregator_goldens/test_table_rename_with_temp.json
View File

@ -85,10 +85,10 @@
"entity": "urn:li:dataset:(urn:li:dataPlatform:redshift,dev.public.baz,PROD)"
},
{
"entity": "urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:redshift,dev.public.baz,PROD),b)"
"entity": "urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:redshift,dev.public.baz,PROD),a)"
},
{
"entity": "urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:redshift,dev.public.baz,PROD),a)"
"entity": "urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:redshift,dev.public.baz,PROD),b)"
},
{
"entity": "urn:li:dataset:(urn:li:dataPlatform:redshift,dev.public.bar,PROD)"

240
metadata-ingestion/tests/unit/sql_parsing/aggregator_goldens/test_table_swap.json
View File

@ -177,24 +177,6 @@
}
}
},
{
"entityType": "query",
"entityUrn": "urn:li:query:6f71602f39d01a39b3f8bd411c74c5ac08dc4b90bc3d49b257089acb19fa8559",
"changeType": "UPSERT",
"aspectName": "querySubjects",
"aspect": {
"json": {
"subjects": [
{
"entity": "urn:li:dataset:(urn:li:dataPlatform:snowflake,dev.public.person_info_swap,PROD)"
},
{
"entity": "urn:li:dataset:(urn:li:dataPlatform:snowflake,dev.public.person_info_backup,PROD)"
}
]
}
}
},
{
"entityType": "query",
"entityUrn": "urn:li:query:3865108263e5f0670e6506f5747392f8315a72039cbfde1c4be4dd9a71bdd500",
@ -220,24 +202,28 @@
},
{
"entityType": "query",
"entityUrn": "urn:li:query:6f71602f39d01a39b3f8bd411c74c5ac08dc4b90bc3d49b257089acb19fa8559",
"entityUrn": "urn:li:query:3865108263e5f0670e6506f5747392f8315a72039cbfde1c4be4dd9a71bdd500",
"changeType": "UPSERT",
"aspectName": "queryProperties",
"aspectName": "querySubjects",
"aspect": {
"json": {
"statement": {
"value": "CREATE TABLE person_info_backup AS\nSELECT\n *\nFROM person_info_swap",
"language": "SQL"
},
"source": "SYSTEM",
"created": {
"time": 0,
"actor": "urn:li:corpuser:_ingestion"
},
"lastModified": {
"time": 1707182625000,
"actor": "urn:li:corpuser:_ingestion"
}
"subjects": [
{
"entity": "urn:li:dataset:(urn:li:dataPlatform:snowflake,dev.public.person_info_swap,PROD)"
},
{
"entity": "urn:li:dataset:(urn:li:dataPlatform:snowflake,dev.public.person_info,PROD)"
},
{
"entity": "urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:snowflake,dev.public.person_info,PROD),a)"
},
{
"entity": "urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:snowflake,dev.public.person_info,PROD),b)"
},
{
"entity": "urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:snowflake,dev.public.person_info,PROD),c)"
}
]
}
}
},
@ -279,7 +265,30 @@
},
{
"entityType": "query",
"entityUrn": "urn:li:query:3865108263e5f0670e6506f5747392f8315a72039cbfde1c4be4dd9a71bdd500",
"entityUrn": "urn:li:query:6f71602f39d01a39b3f8bd411c74c5ac08dc4b90bc3d49b257089acb19fa8559",
"changeType": "UPSERT",
"aspectName": "queryProperties",
"aspect": {
"json": {
"statement": {
"value": "CREATE TABLE person_info_backup AS\nSELECT\n *\nFROM person_info_swap",
"language": "SQL"
},
"source": "SYSTEM",
"created": {
"time": 0,
"actor": "urn:li:corpuser:_ingestion"
},
"lastModified": {
"time": 1707182625000,
"actor": "urn:li:corpuser:_ingestion"
}
}
}
},
{
"entityType": "query",
"entityUrn": "urn:li:query:6f71602f39d01a39b3f8bd411c74c5ac08dc4b90bc3d49b257089acb19fa8559",
"changeType": "UPSERT",
"aspectName": "querySubjects",
"aspect": {
@ -289,16 +298,7 @@
"entity": "urn:li:dataset:(urn:li:dataPlatform:snowflake,dev.public.person_info_swap,PROD)"
},
{
"entity": "urn:li:dataset:(urn:li:dataPlatform:snowflake,dev.public.person_info,PROD)"
},
{
"entity": "urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:snowflake,dev.public.person_info,PROD),a)"
},
{
"entity": "urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:snowflake,dev.public.person_info,PROD),b)"
},
{
"entity": "urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:snowflake,dev.public.person_info,PROD),c)"
"entity": "urn:li:dataset:(urn:li:dataPlatform:snowflake,dev.public.person_info_backup,PROD)"
}
]
}
@ -340,24 +340,6 @@
}
}
},
{
"entityType": "query",
"entityUrn": "urn:li:query:481d0392ffeffdafd198d94e0a9f778dd722b60daa47083a32800b99ea21f86f",
"changeType": "UPSERT",
"aspectName": "querySubjects",
"aspect": {
"json": {
"subjects": [
{
"entity": "urn:li:dataset:(urn:li:dataPlatform:snowflake,dev.public.person_info_incremental,PROD)"
},
{
"entity": "urn:li:dataset:(urn:li:dataPlatform:snowflake,dev.public.person_info_swap,PROD)"
}
]
}
}
},
{
"entityType": "query",
"entityUrn": "urn:li:query:4b1fad909083e1ed5c47c146bd01247ed4d6295d175c34f9065b8fc6000fc7ae",
@ -381,6 +363,35 @@
}
}
},
{
"entityType": "query",
"entityUrn": "urn:li:query:4b1fad909083e1ed5c47c146bd01247ed4d6295d175c34f9065b8fc6000fc7ae",
"changeType": "UPSERT",
"aspectName": "querySubjects",
"aspect": {
"json": {
"subjects": [
{
"entity": "urn:li:dataset:(urn:li:dataPlatform:snowflake,dev.public.person_info_dep,PROD)"
},
{
"entity": "urn:li:dataset:(urn:li:dataPlatform:snowflake,dev.public.person_info_incremental,PROD)"
}
]
}
}
},
{
"entityType": "query",
"entityUrn": "urn:li:query:4b1fad909083e1ed5c47c146bd01247ed4d6295d175c34f9065b8fc6000fc7ae",
"changeType": "UPSERT",
"aspectName": "dataPlatformInstance",
"aspect": {
"json": {
"platform": "urn:li:dataPlatform:snowflake"
}
}
},
{
"entityType": "dataset",
"entityUrn": "urn:li:dataset:(urn:li:dataPlatform:snowflake,dev.public.person_info_swap,PROD)",
@ -421,22 +432,56 @@
},
{
"entityType": "query",
"entityUrn": "urn:li:query:4b1fad909083e1ed5c47c146bd01247ed4d6295d175c34f9065b8fc6000fc7ae",
"entityUrn": "urn:li:query:d29a1c8ed6d4d77efb290260234e5eee56f98311a5631d0a12213798077d1a68",
"changeType": "UPSERT",
"aspectName": "queryProperties",
"aspect": {
"json": {
"statement": {
"value": "ALTER TABLE dev.public.person_info SWAP WITH dev.public.person_info_swap",
"language": "SQL"
},
"source": "SYSTEM",
"created": {
"time": 0,
"actor": "urn:li:corpuser:_ingestion"
},
"lastModified": {
"time": 1707182625000,
"actor": "urn:li:corpuser:_ingestion"
}
}
}
},
{
"entityType": "query",
"entityUrn": "urn:li:query:d29a1c8ed6d4d77efb290260234e5eee56f98311a5631d0a12213798077d1a68",
"changeType": "UPSERT",
"aspectName": "querySubjects",
"aspect": {
"json": {
"subjects": [
{
"entity": "urn:li:dataset:(urn:li:dataPlatform:snowflake,dev.public.person_info_dep,PROD)"
"entity": "urn:li:dataset:(urn:li:dataPlatform:snowflake,dev.public.person_info,PROD)"
},
{
"entity": "urn:li:dataset:(urn:li:dataPlatform:snowflake,dev.public.person_info_incremental,PROD)"
"entity": "urn:li:dataset:(urn:li:dataPlatform:snowflake,dev.public.person_info_swap,PROD)"
}
]
}
}
},
{
"entityType": "query",
"entityUrn": "urn:li:query:d29a1c8ed6d4d77efb290260234e5eee56f98311a5631d0a12213798077d1a68",
"changeType": "UPSERT",
"aspectName": "dataPlatformInstance",
"aspect": {
"json": {
"platform": "urn:li:dataPlatform:snowflake"
}
}
},
{
"entityType": "query",
"entityUrn": "urn:li:query:481d0392ffeffdafd198d94e0a9f778dd722b60daa47083a32800b99ea21f86f",
@ -462,35 +507,19 @@
},
{
"entityType": "query",
"entityUrn": "urn:li:query:4b1fad909083e1ed5c47c146bd01247ed4d6295d175c34f9065b8fc6000fc7ae",
"entityUrn": "urn:li:query:481d0392ffeffdafd198d94e0a9f778dd722b60daa47083a32800b99ea21f86f",
"changeType": "UPSERT",
"aspectName": "dataPlatformInstance",
"aspectName": "querySubjects",
"aspect": {
"json": {
"platform": "urn:li:dataPlatform:snowflake"
}
}
},
{
"entityType": "query",
"entityUrn": "urn:li:query:d29a1c8ed6d4d77efb290260234e5eee56f98311a5631d0a12213798077d1a68",
"changeType": "UPSERT",
"aspectName": "queryProperties",
"aspect": {
"json": {
"statement": {
"value": "ALTER TABLE dev.public.person_info SWAP WITH dev.public.person_info_swap",
"language": "SQL"
},
"source": "SYSTEM",
"created": {
"time": 0,
"actor": "urn:li:corpuser:_ingestion"
},
"lastModified": {
"time": 1707182625000,
"actor": "urn:li:corpuser:_ingestion"
}
"subjects": [
{
"entity": "urn:li:dataset:(urn:li:dataPlatform:snowflake,dev.public.person_info_incremental,PROD)"
},
{
"entity": "urn:li:dataset:(urn:li:dataPlatform:snowflake,dev.public.person_info_swap,PROD)"
}
]
}
}
},
@ -504,34 +533,5 @@
"platform": "urn:li:dataPlatform:snowflake"
}
}
},
{
"entityType": "query",
"entityUrn": "urn:li:query:d29a1c8ed6d4d77efb290260234e5eee56f98311a5631d0a12213798077d1a68",
"changeType": "UPSERT",
"aspectName": "dataPlatformInstance",
"aspect": {
"json": {
"platform": "urn:li:dataPlatform:snowflake"
}
}
},
{
"entityType": "query",
"entityUrn": "urn:li:query:d29a1c8ed6d4d77efb290260234e5eee56f98311a5631d0a12213798077d1a68",
"changeType": "UPSERT",
"aspectName": "querySubjects",
"aspect": {
"json": {
"subjects": [
{
"entity": "urn:li:dataset:(urn:li:dataPlatform:snowflake,dev.public.person_info,PROD)"
},
{
"entity": "urn:li:dataset:(urn:li:dataPlatform:snowflake,dev.public.person_info_swap,PROD)"
}
]
}
}
}
]

4
metadata-ingestion/tests/unit/sql_parsing/aggregator_goldens/test_temp_table.json
View File

@ -281,10 +281,10 @@
"json": {
"subjects": [
{
"entity": "urn:li:dataset:(urn:li:dataPlatform:redshift,dev.public.foo_session3,PROD)"
"entity": "urn:li:dataset:(urn:li:dataPlatform:redshift,dev.public.foo,PROD)"
},
{
"entity": "urn:li:dataset:(urn:li:dataPlatform:redshift,dev.public.foo,PROD)"
"entity": "urn:li:dataset:(urn:li:dataPlatform:redshift,dev.public.foo_session3,PROD)"
}
]
}

39
metadata-ingestion/tests/unit/sql_parsing/test_sql_aggregator.py
View File

@ -1,3 +1,4 @@
import functools
import os
import pathlib
from datetime import datetime, timezone
@ -31,6 +32,10 @@ from tests.test_helpers.click_helpers import run_datahub_cmd
RESOURCE_DIR = pathlib.Path(__file__).parent / "aggregator_goldens"
FROZEN_TIME = "2024-02-06T01:23:45Z"
check_goldens_stream = functools.partial(
mce_helpers.check_goldens_stream, ignore_order=False
)
def _ts(ts: int) -> datetime:
return datetime.fromtimestamp(ts, tz=timezone.utc)
@ -56,7 +61,7 @@ def test_basic_lineage(pytestconfig: pytest.Config, tmp_path: pathlib.Path) -> N
mcps = list(aggregator.gen_metadata())
mce_helpers.check_goldens_stream(
check_goldens_stream(
pytestconfig,
outputs=mcps,
golden_path=RESOURCE_DIR / "test_basic_lineage.json",
@ -108,7 +113,7 @@ def test_overlapping_inserts(pytestconfig: pytest.Config) -> None:
mcps = list(aggregator.gen_metadata())
mce_helpers.check_goldens_stream(
check_goldens_stream(
pytestconfig,
outputs=mcps,
golden_path=RESOURCE_DIR / "test_overlapping_inserts.json",
@ -167,7 +172,7 @@ def test_temp_table(pytestconfig: pytest.Config) -> None:
mcps = list(aggregator.gen_metadata())
mce_helpers.check_goldens_stream(
check_goldens_stream(
pytestconfig,
outputs=mcps,
golden_path=RESOURCE_DIR / "test_temp_table.json",
@ -229,7 +234,7 @@ def test_multistep_temp_table(pytestconfig: pytest.Config) -> None:
)
== 4
)
mce_helpers.check_goldens_stream(
check_goldens_stream(
pytestconfig,
outputs=mcps,
golden_path=RESOURCE_DIR / "test_multistep_temp_table.json",
@ -305,7 +310,7 @@ def test_overlapping_inserts_from_temp_tables(pytestconfig: pytest.Config) -> No
assert len(report.queries_with_non_authoritative_session) == 1
mcps = list(aggregator.gen_metadata())
mce_helpers.check_goldens_stream(
check_goldens_stream(
pytestconfig,
outputs=mcps,
golden_path=RESOURCE_DIR / "test_overlapping_inserts_from_temp_tables.json",
@ -354,7 +359,7 @@ def test_aggregate_operations(pytestconfig: pytest.Config) -> None:
mcps = list(aggregator.gen_metadata())
mce_helpers.check_goldens_stream(
check_goldens_stream(
pytestconfig,
outputs=mcps,
golden_path=RESOURCE_DIR / "test_aggregate_operations.json",
@ -392,7 +397,7 @@ def test_view_lineage(pytestconfig: pytest.Config) -> None:
mcps = list(aggregator.gen_metadata())
mce_helpers.check_goldens_stream(
check_goldens_stream(
pytestconfig,
outputs=mcps,
golden_path=RESOURCE_DIR / "test_view_lineage.json",
@ -423,7 +428,7 @@ def test_known_lineage_mapping(pytestconfig: pytest.Config) -> None:
mcps = list(aggregator.gen_metadata())
mce_helpers.check_goldens_stream(
check_goldens_stream(
pytestconfig,
outputs=mcps,
golden_path=RESOURCE_DIR / "test_known_lineage_mapping.json",
@ -461,7 +466,7 @@ def test_column_lineage_deduplication(pytestconfig: pytest.Config) -> None:
# not get any credit for a and b, as they are already covered by query 2,
# which came later and hence has higher precedence.
mce_helpers.check_goldens_stream(
check_goldens_stream(
pytestconfig,
outputs=mcps,
golden_path=RESOURCE_DIR / "test_column_lineage_deduplication.json",
@ -506,7 +511,7 @@ def test_add_known_query_lineage(pytestconfig: pytest.Config) -> None:
mcps = list(aggregator.gen_metadata())
mce_helpers.check_goldens_stream(
check_goldens_stream(
pytestconfig,
outputs=mcps,
golden_path=RESOURCE_DIR / "test_add_known_query_lineage.json",
@ -564,7 +569,7 @@ def test_table_rename(pytestconfig: pytest.Config) -> None:
mcps = list(aggregator.gen_metadata())
mce_helpers.check_goldens_stream(
check_goldens_stream(
pytestconfig,
outputs=mcps,
golden_path=RESOURCE_DIR / "test_table_rename.json",
@ -624,7 +629,7 @@ def test_table_rename_with_temp(pytestconfig: pytest.Config) -> None:
mcps = list(aggregator.gen_metadata())
mce_helpers.check_goldens_stream(
check_goldens_stream(
pytestconfig,
outputs=mcps,
golden_path=RESOURCE_DIR / "test_table_rename_with_temp.json",
@ -711,7 +716,7 @@ def test_table_swap(pytestconfig: pytest.Config) -> None:
mcps = list(aggregator.gen_metadata())
mce_helpers.check_goldens_stream(
check_goldens_stream(
pytestconfig,
outputs=mcps,
golden_path=RESOURCE_DIR / "test_table_swap.json",
@ -881,7 +886,7 @@ def test_table_swap_with_temp(pytestconfig: pytest.Config) -> None:
mcps = list(aggregator.gen_metadata())
mce_helpers.check_goldens_stream(
check_goldens_stream(
pytestconfig,
outputs=mcps,
golden_path=RESOURCE_DIR / "test_table_swap_with_temp.json",
@ -908,7 +913,7 @@ def test_create_table_query_mcps(pytestconfig: pytest.Config) -> None:
mcps = list(aggregator.gen_metadata())
mce_helpers.check_goldens_stream(
check_goldens_stream(
pytestconfig,
outputs=mcps,
golden_path=RESOURCE_DIR / "test_create_table_query_mcps.json",
@ -943,7 +948,7 @@ def test_table_lineage_via_temp_table_disordered_add(
mcps = list(aggregator.gen_metadata())
mce_helpers.check_goldens_stream(
check_goldens_stream(
pytestconfig,
outputs=mcps,
golden_path=RESOURCE_DIR
@ -993,7 +998,7 @@ def test_basic_usage(pytestconfig: pytest.Config) -> None:
mcps = list(aggregator.gen_metadata())
mce_helpers.check_goldens_stream(
check_goldens_stream(
pytestconfig,
outputs=mcps,
golden_path=RESOURCE_DIR / "test_basic_usage.json",