working for big query but only when the table already exists

snowplow · Nov 15, 2024 · c469819 · c469819
1 parent 938a2ef
commit c469819
Show file tree

Hide file tree

Showing 2 changed files with 13 additions and 13 deletions.
diff --git a/macros/users_table.sql b/macros/users_table.sql
@@ -31,7 +31,7 @@ select
     {% elif user_id_context != '' %}
         {{ '_'.join(user_id_context.split('_')[:-2]) }}[0]:{{user_id_field}}::string as {{ snake_user_id }}
     {%- endif %}
-    , {{ var('snowplow__partition_key') }} as latest_selected_tstamp
+    , collector_tstamp as latest_collector_tstamp
     -- Flat columns from event table
     {% if flat_cols|length > 0 %}
         {%- for col in flat_cols -%}
@@ -63,7 +63,7 @@ from
 where
     {{ snake_user_id }} is not null
 qualify
-    row_number() over (partition by {{ snake_user_id }} order by latest_selected_tstamp desc) = 1
+    row_number() over (partition by {{ snake_user_id }} order by latest_collector_tstamp desc) = 1
 {% endmacro %}
 
 
@@ -117,7 +117,7 @@ with defined_user_id as (
                                         ) -%}
             {{ user_id_cont_coal[0] }} as {{ snake_user_id }}
         {%- endif %}
-        , {{ var('snowplow__partition_key') }} as latest_selected_tstamp
+        , collector_tstamp as latest_collector_tstamp
         -- Flat columns from event table
         {% if flat_cols|length > 0 %}
             {%- for col in flat_cols -%}
@@ -151,7 +151,7 @@ with defined_user_id as (
 users_ordering as (
     select
         a.*
-        , row_number() over (partition by {{ snake_user_id }} order by latest_selected_tstamp desc) as rn
+        , row_number() over (partition by {{ snake_user_id }} order by latest_collector_tstamp desc) as rn
     from
         defined_user_id a
     where
@@ -201,9 +201,9 @@ with defined_user_id as (
         {% elif user_id_context != '' %}
             {{ '_'.join(user_id_context.split('_')[:-2]) }}[0].{{ user_id_field }} as {{ snake_user_id }}
         {%- endif %}
-        ,  {{ var('snowplow__partition_key') }} as latest_selected_tstamp
+        , collector_tstamp as latest_collector_tstamp
         {% if target.type in ['databricks', 'spark'] -%}
-            , DATE( {{ var('snowplow__partition_key') }}) as latest_selected_tstamp_date
+            , DATE(collector_tstamp) as latest_collector_tstamp_date
         {%- endif %}
         -- Flat columns from event table
         {% if flat_cols|length > 0 %}
@@ -233,7 +233,7 @@ with defined_user_id as (
 users_ordering as (
 select
     a.*
-    , row_number() over (partition by {{ snake_user_id }} order by latest_selected_tstamp desc) as rn
+    , row_number() over (partition by {{ snake_user_id }} order by latest_collector_tstamp desc) as rn
 from
     defined_user_id a
 where

diff --git a/utils/snowplow_normalize_model_gen.py b/utils/snowplow_normalize_model_gen.py
@@ -195,9 +195,9 @@
     tags = "snowplow_normalize_incremental",
     materialized = "incremental",
     unique_key = "event_id",
-    upsert_date_key = var('snowplow__partition_key'),
+    upsert_date_key = "collector_tstamp",
     partition_by = snowplow_utils.get_value_by_target_type(bigquery_val={{
-      "field": var('snowplow__partition_key'),
+      "field":  var("snowplow__partition_key"),
       "data_type": "timestamp"
     }}, databricks_val='collector_tstamp_date'),
     sql_header=snowplow_utils.set_query_tag(var('snowplow__query_tag', 'snowplow_dbt')),
@@ -251,9 +251,9 @@
     tags = "snowplow_normalize_incremental",
     materialized = "incremental",
     unique_key = "unique_id",
-    upsert_date_key = var('snowplow__partition_key'),
+    upsert_date_key = "collector_tstamp",
     partition_by = snowplow_utils.get_value_by_target_type(bigquery_val={{
-      "field": var('snowplow__partition_key'),
+      "field":  var("snowplow__partition_key"),
       "data_type": "timestamp"
     }}, databricks_val='collector_tstamp_date'),
     sql_header=snowplow_utils.set_query_tag(var('snowplow__query_tag', 'snowplow_dbt')),
@@ -323,9 +323,9 @@
     tags = "snowplow_normalize_incremental",
     materialized = "incremental",
     unique_key = "{user_alias}",
-    upsert_date_key = "latest_selected_tstamp",
+    upsert_date_key = "latest_collector_tstamp",
     partition_by = snowplow_utils.get_value_by_target_type(bigquery_val={{
-      "field": "latest_selected_tstamp",
+      "field": "latest_collector_tstamp",
       "data_type": "timestamp"
     }}, databricks_val='latest_collector_tstamp_date'),
     sql_header=snowplow_utils.set_query_tag(var('snowplow__query_tag', 'snowplow_dbt')),