latest_collector_tstamp to latest_selected_tstamp

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 %}
-    , collector_tstamp as latest_collector_tstamp
+    , {{ var('snowplow__partition_key') }} as latest_selected_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_collector_tstamp desc) = 1
+    row_number() over (partition by {{ snake_user_id }} order by latest_selected_tstamp desc) = 1
 {% endmacro %}
 
 
@@ -117,7 +117,7 @@ with defined_user_id as (
                                         ) -%}
             {{ user_id_cont_coal[0] }} as {{ snake_user_id }}
         {%- endif %}
-        , collector_tstamp as latest_collector_tstamp
+        , {{ var('snowplow__partition_key') }} as latest_selected_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_collector_tstamp desc) as rn
+        , row_number() over (partition by {{ snake_user_id }} order by latest_selected_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 %}
-        , collector_tstamp as latest_collector_tstamp
+        ,  {{ var('snowplow__partition_key') }} as latest_selected_tstamp
         {% if target.type in ['databricks', 'spark'] -%}
-            , DATE(collector_tstamp) as latest_collector_tstamp_date
+            , DATE( {{ var('snowplow__partition_key') }}) as latest_selected_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_collector_tstamp desc) as rn
+    , row_number() over (partition by {{ snake_user_id }} order by latest_selected_tstamp desc) as rn
 from
     defined_user_id a
 where

utils/snowplow_normalize_model_gen.py

@@ -323,9 +323,9 @@
     tags = "snowplow_normalize_incremental",
     materialized = "incremental",
     unique_key = "{user_alias}",
-    upsert_date_key = "latest_collector_tstamp",
+    upsert_date_key = "latest_selected_tstamp",
     partition_by = snowplow_utils.get_value_by_target_type(bigquery_val={{
-      "field": "latest_collector_tstamp",
+      "field": "latest_selected_tstamp",
       "data_type": "timestamp"
     }}, databricks_val='latest_collector_tstamp_date'),
     sql_header=snowplow_utils.set_query_tag(var('snowplow__query_tag', 'snowplow_dbt')),