-
Notifications
You must be signed in to change notification settings - Fork 19
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
data: migrate int_events__dependencies to sqlmesh (#2927)
* This includes porting the deps.dev macros from dbt to sqlmesh syntax
- Loading branch information
Showing
4 changed files
with
313 additions
and
2 deletions.
There are no files selected for viewing
184 changes: 184 additions & 0 deletions
184
warehouse/metrics_mesh/macros/deps_dev/deps_dev_artifact_details.py
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,184 @@ | ||
|
|
||
| from sqlglot import expressions as exp | ||
| from sqlmesh import macro | ||
| from sqlmesh.core.macros import MacroEvaluator | ||
|
|
||
|
|
||
| @macro() | ||
| def deps_parse_namespace(evaluator: MacroEvaluator, event_source: exp.Expression, artifact_name: exp.Expression): | ||
| """ | ||
| Macro to parse the namespace from the artifact name based on the event source. | ||
| Arguments: | ||
| - event_source: The event source of the artifact. | ||
| - artifact_name: The name of the artifact. | ||
| Returns the namespace based on event source rules. | ||
| """ | ||
|
|
||
| name = exp.Case( | ||
| ifs=[ | ||
| exp.If( | ||
| this=exp.And( | ||
| this=exp.EQ( | ||
| this=event_source, | ||
| expression=exp.Literal.string('NPM'), | ||
| ), | ||
| expression=exp.GT( | ||
| this=exp.StrPosition( | ||
| this=artifact_name, | ||
| substr=exp.Literal.string('/'), | ||
| ), | ||
| expression=exp.Literal.number(0), | ||
| ), | ||
| ), | ||
| true=exp.SplitPart( | ||
| this=exp.SplitPart( | ||
| this=artifact_name, | ||
| delimiter=exp.Literal.string('/'), | ||
| part_index=exp.Literal.number(1), | ||
| ), | ||
| delimiter=exp.Literal.string('@'), | ||
| part_index=exp.Literal.number(2), | ||
| ), | ||
| ), | ||
| exp.If( | ||
| this=exp.And( | ||
| this=exp.EQ( | ||
| this=event_source, | ||
| expression=exp.Literal.string('GO'), | ||
| ), | ||
| expression=exp.GT( | ||
| this=exp.StrPosition( | ||
| this=artifact_name, | ||
| substr=exp.Literal.string('/'), | ||
| ), | ||
| expression=exp.Literal.number(0), | ||
| ), | ||
| ), | ||
| true=exp.SplitPart( | ||
| this=artifact_name, | ||
| delimiter=exp.Literal.string('/'), | ||
| part_index=exp.Literal.number(2), | ||
| ), | ||
| ), | ||
| exp.If( | ||
| this=exp.EQ( | ||
| this=event_source, | ||
| expression=exp.Literal.string('MAVEN'), | ||
| ), | ||
| true=exp.SplitPart( | ||
| this=artifact_name, | ||
| delimiter=exp.Literal.string(':'), | ||
| part_index=exp.Literal.number(1), | ||
| ), | ||
| ), | ||
| exp.If( | ||
| this=exp.And( | ||
| this=exp.EQ( | ||
| this=event_source, | ||
| expression=exp.Literal.string('NUGET'), | ||
| ), | ||
| expression=exp.GT( | ||
| this=exp.StrPosition( | ||
| this=artifact_name, | ||
| substr=exp.Literal.string('.'), | ||
| ), | ||
| expression=exp.Literal.number(0), | ||
| ), | ||
| ), | ||
| true=exp.SplitPart( | ||
| this=artifact_name, | ||
| delimiter=exp.Literal.string('.'), | ||
| part_index=exp.Literal.number(1), | ||
| ), | ||
| ), | ||
| ], | ||
| default=artifact_name, | ||
| ) | ||
| return name | ||
|
|
||
| @macro() | ||
| def deps_parse_name(evaluator: MacroEvaluator, event_source: exp.Expression, artifact_name: exp.Expression): | ||
| """ | ||
| Macro to parse the name from the artifact name based on the event source. | ||
| Arguments: | ||
| - event_source: The event source of the artifact. | ||
| - artifact_name: The name of the artifact. | ||
| Returns the name based on event source rules. | ||
| """ | ||
| name = exp.Case( | ||
| ifs=[ | ||
| exp.If( | ||
| this=exp.And( | ||
| this=exp.EQ( | ||
| this=event_source, | ||
| expression=exp.Literal.string('NPM'), | ||
| ), | ||
| expression=exp.GT( | ||
| this=exp.StrPosition( | ||
| this=artifact_name, | ||
| substr=exp.Literal.string('/'), | ||
| ), | ||
| expression=exp.Literal.number(0), | ||
| ), | ||
| ), | ||
| true=exp.SplitPart( | ||
| this=artifact_name, | ||
| delimiter=exp.Literal.string('/'), | ||
| part_index=exp.Literal.number(2), | ||
| ), | ||
| ), | ||
| exp.If( | ||
| this=exp.And( | ||
| this=exp.EQ( | ||
| this=event_source, | ||
| expression=exp.Literal.string('GO'), | ||
| ), | ||
| expression=exp.GT( | ||
| this=exp.StrPosition( | ||
| this=artifact_name, | ||
| substr=exp.Literal.string('/'), | ||
| ), | ||
| expression=exp.Literal.number(0), | ||
| ), | ||
| ), | ||
| true=exp.SplitPart( | ||
| this=artifact_name, | ||
| delimiter=exp.Literal.string('/'), | ||
| part_index=exp.Literal.number(3), | ||
| ), | ||
| ), | ||
| exp.If( | ||
| this=exp.EQ( | ||
| this=event_source, | ||
| expression=exp.Literal.string('MAVEN'), | ||
| ), | ||
| true=exp.SplitPart( | ||
| this=artifact_name, | ||
| delimiter=exp.Literal.string(':'), | ||
| part_index=exp.Literal.number(2), | ||
| ), | ||
| ), | ||
| exp.If( | ||
| this=exp.And( | ||
| this=exp.EQ( | ||
| this=event_source, | ||
| expression=exp.Literal.string('NUGET'), | ||
| ), | ||
| expression=exp.GT( | ||
| this=exp.StrPosition( | ||
| this=artifact_name, | ||
| substr=exp.Literal.string('.'), | ||
| ), | ||
| expression=exp.Literal.number(0), | ||
| ), | ||
| ), | ||
| true=exp.RegexpReplace( | ||
| this=artifact_name, | ||
| expression=exp.Literal.string(r'^[^.]+\.'), | ||
| replacement=exp.Literal.string(''), | ||
| ), | ||
| ), | ||
| ], | ||
| default=artifact_name, | ||
| ) | ||
| return name |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
127 changes: 127 additions & 0 deletions
127
warehouse/metrics_mesh/models/intermediate/events/int_events__dependencies.sql
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,127 @@ | ||
| MODEL ( | ||
| name metrics.int_events__dependencies, | ||
| dialect trino, | ||
| kind INCREMENTAL_BY_TIME_RANGE ( | ||
| time_column time, | ||
| batch_size 90, | ||
| batch_concurrency 1 | ||
| ), | ||
| start '2015-01-01', | ||
| cron '@daily', | ||
| partitioned_by (DAY("time"), "event_type"), | ||
| grain (time, event_type, event_source, from_artifact_id, to_artifact_id) | ||
| ); | ||
|
|
||
| @DEF(event_source_name, 'DEPS_DEV'); | ||
|
|
||
| with artifacts as ( | ||
| select artifact_name | ||
| from metrics.int_all_artifacts | ||
| where artifact_source = 'NPM' | ||
| ), | ||
|
|
||
| snapshots as ( | ||
| select | ||
| SnapshotAt as time, | ||
| System as from_artifact_type, | ||
| Name as from_artifact_name, | ||
| Version as from_artifact_version, | ||
| Dependency.Name as to_artifact_name, | ||
| Dependency.System as to_artifact_type, | ||
| Dependency.Version as to_artifact_version, | ||
| LAG(Dependency.Name) over ( | ||
| partition by System, Name, Dependency.Name, Version, Dependency.Version | ||
| order by SnapshotAt | ||
| ) as previous_to_artifact_name | ||
| from @oso_source('bigquery.oso.stg_deps_dev__dependencies') | ||
| where | ||
| MinimumDepth = 1 | ||
| and Dependency.Name in (select artifact_name from artifacts) | ||
| ), | ||
|
|
||
| intermediate as ( | ||
| select | ||
| time, | ||
| case | ||
| when previous_to_artifact_name is null then 'ADD_DEPENDENCY' | ||
| when | ||
| to_artifact_name is not null and to_artifact_name <> previous_to_artifact_name | ||
| then 'REMOVE_DEPENDENCY' | ||
| else 'NO_CHANGE' | ||
| end as event_type, | ||
| @event_source_name as event_source, | ||
| @deps_parse_name(to_artifact_type, to_artifact_name) as to_artifact_name, | ||
| @deps_parse_namespace(to_artifact_type, to_artifact_name) as to_artifact_namespace, | ||
| to_artifact_type, | ||
| @deps_parse_name(from_artifact_type, from_artifact_name) as from_artifact_name, | ||
| @deps_parse_namespace(from_artifact_type, from_artifact_name) as from_artifact_namespace, | ||
| from_artifact_type, | ||
| 1.0 as amount | ||
| from snapshots | ||
| ), | ||
|
|
||
| artifact_ids as ( | ||
| select | ||
| time, | ||
| event_type, | ||
| event_source, | ||
| @oso_id(event_source, to_artifact_namespace, to_artifact_name) as to_artifact_id, | ||
| to_artifact_name, | ||
| to_artifact_namespace, | ||
| to_artifact_type, | ||
| @oso_id(event_source, to_artifact_type) as to_artifact_source_id, | ||
| @oso_id(event_source, from_artifact_namespace, from_artifact_name) as from_artifact_id, | ||
| from_artifact_name, | ||
| from_artifact_namespace, | ||
| from_artifact_type, | ||
| @oso_id(event_source, from_artifact_type) as from_artifact_source_id, | ||
| amount | ||
| from intermediate | ||
| where event_type <> 'NO_CHANGE' | ||
| ), | ||
|
|
||
| changes as ( | ||
| select | ||
| time, | ||
| event_type, | ||
| event_source, | ||
| to_artifact_id, | ||
| to_artifact_name, | ||
| to_artifact_namespace, | ||
| to_artifact_type, | ||
| to_artifact_source_id, | ||
| from_artifact_id, | ||
| from_artifact_name, | ||
| from_artifact_namespace, | ||
| from_artifact_type, | ||
| from_artifact_source_id, | ||
| amount, | ||
| @oso_id( | ||
| event_source, | ||
| time, | ||
| to_artifact_id, | ||
| to_artifact_type, | ||
| from_artifact_id, | ||
| from_artifact_type, | ||
| event_type | ||
| ) as event_source_id | ||
| from artifact_ids | ||
| ) | ||
|
|
||
| select | ||
| time, | ||
| to_artifact_id, | ||
| from_artifact_id, | ||
| UPPER(event_type) as event_type, | ||
| CAST(event_source_id as STRING) as event_source_id, | ||
| UPPER(event_source) as event_source, | ||
| LOWER(to_artifact_name) as to_artifact_name, | ||
| LOWER(to_artifact_namespace) as to_artifact_namespace, | ||
| UPPER(to_artifact_type) as to_artifact_type, | ||
| LOWER(to_artifact_source_id) as to_artifact_source_id, | ||
| LOWER(from_artifact_name) as from_artifact_name, | ||
| LOWER(from_artifact_namespace) as from_artifact_namespace, | ||
| UPPER(from_artifact_type) as from_artifact_type, | ||
| LOWER(from_artifact_source_id) as from_artifact_source_id, | ||
| CAST(amount as DOUBLE) as amount | ||
| from changes |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters