Add dbt cloud compatibility (#38)

* Add dbt cloud compatibility

* Add full refresh to CI

Co-authored-by: Niall Woodward <[email protected]>

.circleci/config.yml

@@ -31,6 +31,7 @@ jobs:
             dbt run-operation create_artifact_resources
             dbt seed --full-refresh
             dbt run-operation upload_dbt_artifacts --args '{filenames: [manifest, run_results]}'
+            dbt run --full-refresh
             dbt run
             dbt run-operation upload_dbt_artifacts --args '{filenames: [manifest, run_results]}'
             dbt test

models/fct_dbt__critical_path.sql

@@ -16,7 +16,8 @@ latest_id as (
     -- Find the latest full, incremental execution
 
     select
-        any_value(command_invocation_id) as command_invocation_id
+        any_value(command_invocation_id) as command_invocation_id,
+        any_value(dbt_cloud_run_id) as dbt_cloud_run_id
     from latest_executions
 
 ),
@@ -29,7 +30,9 @@ latest_models as (
         models.depends_on_nodes,
         models.model_materialization
     from latest_id
-    left join models on latest_id.command_invocation_id = models.command_invocation_id
+    left join models on
+        latest_id.command_invocation_id = models.command_invocation_id
+        or latest_id.dbt_cloud_run_id = models.dbt_cloud_run_id
 
 
 ),

models/fct_dbt__latest_full_model_executions.sql

@@ -27,7 +27,9 @@ joined as (
     select
         model_executions.*
     from latest_full
-    left join model_executions on model_executions.command_invocation_id = latest_full.command_invocation_id
+    left join model_executions on
+        model_executions.command_invocation_id = latest_full.command_invocation_id
+        or model_executions.dbt_cloud_run_id = latest_full.dbt_cloud_run_id
 
 ),
 
@@ -36,6 +38,7 @@ fields as (
     select
         artifact_generated_at,
         command_invocation_id,
+        dbt_cloud_run_id,
         compile_started_at,
         query_completed_at,
         total_node_runtime,

models/incremental/dim_dbt__exposures.sql

@@ -1,8 +1,8 @@
-{{ 
+{{
   config(
-    materialized='incremental', 
+    materialized='incremental',
     unique_key='manifest_model_id'
-    ) 
+    )
 }}
 
 with dbt_models as (
@@ -25,9 +25,10 @@ dbt_models_incremental as (
 
 fields as (
 
-     select 
+     select
         t.manifest_model_id,
         t.command_invocation_id,
+        t.dbt_cloud_run_id,
         t.artifact_generated_at,
         t.node_id,
         t.name,
@@ -36,8 +37,8 @@ fields as (
         t.maturity,
         f.value::string as output_feeds,
         t.package_name
-    from dbt_models_incremental t,
-    lateral flatten(input => depends_on_nodes) f
+    from dbt_models_incremental as t,
+    lateral flatten(input => depends_on_nodes) as f
 
 )
 

models/incremental/dim_dbt__models.sql

@@ -23,6 +23,7 @@ fields as (
     select
         manifest_model_id,
         command_invocation_id,
+        dbt_cloud_run_id,
         artifact_generated_at,
         node_id,
         model_database,

models/incremental/dim_dbt__sources.sql

@@ -23,6 +23,7 @@ fields as (
     select
         manifest_source_id,
         command_invocation_id,
+        dbt_cloud_run_id,
         artifact_generated_at,
         node_id,
         name,

models/incremental/fct_dbt__model_executions.sql

@@ -34,10 +34,12 @@ model_executions_with_materialization as (
         models.model_schema,
         models.name
     from model_executions_incremental
-    left join models on (
-        model_executions_incremental.command_invocation_id = models.command_invocation_id
+    left join models on
+        (
+            model_executions_incremental.command_invocation_id = models.command_invocation_id
+            or model_executions_incremental.dbt_cloud_run_id = models.dbt_cloud_run_id
+        )
         and model_executions_incremental.node_id = models.node_id
-    )
 
 ),
 
@@ -46,6 +48,7 @@ fields as (
     select
         model_execution_id,
         command_invocation_id,
+        dbt_cloud_run_id,
         artifact_generated_at,
         was_full_refresh,
         node_id,

models/incremental/fct_dbt__run_results.sql

@@ -26,6 +26,7 @@ fields as (
     select
         artifact_generated_at,
         command_invocation_id,
+        dbt_cloud_run_id,
         dbt_version,
         elapsed_time,
         execution_command,

models/incremental/fct_dbt__test_executions.sql

@@ -24,6 +24,7 @@ fields as (
     select
         model_execution_id,
         command_invocation_id,
+        dbt_cloud_run_id,
         artifact_generated_at,
         was_full_refresh,
         node_id,

models/incremental/int_dbt__model_executions.sql

@@ -24,6 +24,7 @@ fields as (
     select
         model_execution_id,
         command_invocation_id,
+        dbt_cloud_run_id,
         artifact_generated_at,
         was_full_refresh,
         node_id,

models/staging/stg_dbt__artifacts.sql

@@ -9,6 +9,7 @@ fields as (
 
     select
         data:metadata:invocation_id::string as command_invocation_id,
+        data:metadata:env:DBT_CLOUD_RUN_ID::int as dbt_cloud_run_id,
         generated_at,
         path,
         artifact_type,
@@ -17,7 +18,7 @@ fields as (
 
 ),
 
-duduped as (
+deduped as (
 
     select
         *,
@@ -34,11 +35,12 @@ artifacts as (
 
     select
         command_invocation_id,
+        dbt_cloud_run_id,
         generated_at,
         path,
         artifact_type,
         data
-    from duduped
+    from deduped
 
 )
 

models/staging/stg_dbt__exposures.sql

@@ -17,6 +17,7 @@ flatten as (
 
     select
         command_invocation_id,
+        dbt_cloud_run_id,
         generated_at as artifact_generated_at,
         node.key as node_id,
         node.value:name::string as name,
@@ -36,6 +37,7 @@ surrogate_key as (
     select
         {{ dbt_utils.surrogate_key(['command_invocation_id', 'node_id']) }} as manifest_model_id,
         command_invocation_id,
+        dbt_cloud_run_id,
         artifact_generated_at,
         node_id,
         name,

models/staging/stg_dbt__model_executions.sql

@@ -24,7 +24,8 @@ dbt_run as (
 fields as (
 
     select
-        data:metadata:invocation_id::string as command_invocation_id,
+        command_invocation_id,
+        dbt_cloud_run_id,
         generated_at as artifact_generated_at,
         coalesce(data:args:full_refresh, 'false')::boolean as was_full_refresh,
         result.value:unique_id::string as node_id,
@@ -36,11 +37,11 @@ fields as (
 
         -- The second item in the timing array is `execute`.
         result.value:timing[1]:completed_at::timestamp_ntz as query_completed_at,
-        
+
         -- Confusingly, this does not match the delta of the above two timestamps.
         -- should we calculate it instead?
         coalesce(result.value:execution_time::float, 0) as total_node_runtime,
-        
+
         result.value:adapter_response:rows_affected::int as rows_affected
     from dbt_run,
     lateral flatten(input => data:results) as result
@@ -52,6 +53,7 @@ surrogate_key as (
     select
         {{ dbt_utils.surrogate_key(['command_invocation_id', 'node_id']) }} as model_execution_id,
         command_invocation_id,
+        dbt_cloud_run_id,
         artifact_generated_at,
         was_full_refresh,
         node_id,

models/staging/stg_dbt__models.sql

@@ -17,6 +17,7 @@ flatten as (
 
     select
         command_invocation_id,
+        dbt_cloud_run_id,
         generated_at as artifact_generated_at,
         node.key as node_id,
         node.value:database::string as model_database,
@@ -38,6 +39,7 @@ surrogate_key as (
     select
         {{ dbt_utils.surrogate_key(['command_invocation_id', 'node_id']) }} as manifest_model_id,
         command_invocation_id,
+        dbt_cloud_run_id,
         artifact_generated_at,
         node_id,
         model_database,

models/staging/stg_dbt__run_results.sql

@@ -25,7 +25,8 @@ fields as (
 
     select
         generated_at as artifact_generated_at,
-        data:metadata:invocation_id::string as command_invocation_id,
+        command_invocation_id,
+        dbt_cloud_run_id,
         data:metadata:dbt_version::string as dbt_version,
         data:metadata:env as env,
         data:elapsed_time::float as elapsed_time,

models/staging/stg_dbt__sources.sql

@@ -17,6 +17,7 @@ flatten as (
 
     select
         command_invocation_id,
+        dbt_cloud_run_id,
         generated_at as artifact_generated_at,
         node.key as node_id,
         node.value:name::string as name,
@@ -36,6 +37,7 @@ surrogate_key as (
     select
         {{ dbt_utils.surrogate_key(['command_invocation_id', 'node_id']) }} as manifest_source_id,
         command_invocation_id,
+        dbt_cloud_run_id,
         artifact_generated_at,
         node_id,
         name,

models/staging/stg_dbt__test_executions.sql

@@ -24,7 +24,8 @@ dbt_run as (
 fields as (
 
     select
-        data:metadata:invocation_id::string as command_invocation_id,
+        command_invocation_id,
+        dbt_cloud_run_id,
         generated_at as artifact_generated_at,
         coalesce(data:args:full_refresh, 'false')::boolean as was_full_refresh,
         result.value:unique_id::string as node_id,
@@ -37,11 +38,11 @@ fields as (
 
         -- The second item in the timing array is `execute`.
         result.value:timing[1]:completed_at::timestamp_ntz as query_completed_at,
-        
+
         -- Confusingly, this does not match the delta of the above two timestamps.
         -- should we calculate it instead?
         coalesce(result.value:execution_time::float, 0) as total_node_runtime,
-        
+
         result.value:adapter_response:rows_affected::int as rows_affected
     from dbt_run,
     lateral flatten(input => data:results) as result
@@ -53,6 +54,7 @@ surrogate_key as (
     select
         {{ dbt_utils.surrogate_key(['command_invocation_id', 'node_id']) }} as model_execution_id,
         command_invocation_id,
+        dbt_cloud_run_id,
         artifact_generated_at,
         was_full_refresh,
         node_id,