add assertions and fix user uniqueness

Author: Dan Lee

dataform.json

@@ -1,6 +1,6 @@
 {
-    "defaultSchema": "dataform",
-    "assertionSchema": "dataform_assertions",
+    "defaultSchema": "segment_dataform_package",
+    "assertionSchema": "segment_dataform_package",
     "warehouse": "bigquery",
     "gcloudProjectId": "tada-analytics"
 }

includes/page_events.js

@@ -11,7 +11,7 @@ module.exports = (params) => {
   }).query(ctx => `
 
 -- format page calls into a format suitable to join with track calls
-select
+select distinct
   pages.timestamp,
   user_id,
   anonymous_id,

includes/sessions.js

@@ -8,6 +8,9 @@ module.exports = (params) => {
   const customTrackFieldsObj = params.customTrackFields.reduce((acc, item) => ({...acc, [item]: item }), {});
 
   return publish("segment_sessions", {
+    assertions: {
+      uniqueKey: ["session_id"]
+    },
     description: "Sessions contain a combined view of tracks and pages from segment. Each session is a period of sustained activity, with a new session starting after a 30min+ period of inactivity. Each session contains a repeated field of records which are either tracks or pages. Common fields are extracted out into the top level and type specific fields are kept within two structs: records.track and records.page",
     columns: {
       session_id: "Unique identifier of the session",
@@ -18,7 +21,7 @@ module.exports = (params) => {
     ...params.defaultConfig
   }).query(ctx => `
 
-/* TODO: optimise this code to make it work, or enable only for incremental builds
+/* TODO: optimise this code to make it work, or enable only for incremental builds */
 with first_and_last_page_values as (
 select distinct
   session_id,
@@ -29,6 +32,7 @@ select distinct
         ignore_nulls: true,
         partition_fields: "session_id",
         order_fields: 'sessionized_pages.timestamp asc',
+        frame_clause: "rows between unbounded preceding and unbounded following",
       })} as first_${value}`).join(",\n  ")},
   ${Object.entries({...segmentCommon.PAGE_FIELDS, ...customPageFieldsObj}).map(
       ([key, value]) => `${crossdb.windowFunction({
@@ -37,10 +41,12 @@ select distinct
         ignore_nulls: true,
         partition_fields: "session_id",
         order_fields: 'sessionized_pages.timestamp asc',
+        frame_clause: "rows between unbounded preceding and unbounded following",
       })} as last_${value}`).join(",\n  ")}
   from
     ${ctx.ref(params.defaultConfig.schema, "segment_sessionized_pages")} as sessionized_pages
   )
+/*
 */
 
 select
@@ -56,12 +62,13 @@ select
   count(segment_sessionized_events.page_id) as total_pages,
   ${crossdb.timestampDiff("millisecond", "min(segment_sessionized_events.timestamp)", "max(segment_sessionized_events.timestamp)")} as duration_millis
   ${ctx.when(global.session.config.warehouse == "bigquery", `) as stats`)},
-/* See TODO at start of script
+/* See TODO at start of script */
   -- first values in the session for page fields
   ${ctx.when(global.session.config.warehouse == "bigquery", `struct(\n  `)}
   ${Object.entries({...segmentCommon.PAGE_FIELDS, ...customPageFieldsObj}).map(
       ([key, value]) => `first_and_last_page_values.first_${value}`).join(",\n  ")}
   ${ctx.when(global.session.config.warehouse == "bigquery", `) as first_page_values`)},
+  /*
   -- last values in the session for page fields
   ${ctx.when(global.session.config.warehouse == "bigquery", `struct(\n  `)}
   ${Object.entries({...segmentCommon.PAGE_FIELDS, ...customPageFieldsObj}).map(
@@ -88,21 +95,21 @@ select
   ) as records`)}
 from
   ${ctx.ref(params.defaultConfig.schema, "segment_sessionized_events")} as segment_sessionized_events
-/* See TODO at start of script
+/* See TODO at start of script */
   left join first_and_last_page_values
     using(session_id)
-*/
+/**/
   ${ctx.when(global.session.config.warehouse == "bigquery", `
   left join ${ctx.ref(params.defaultConfig.schema, "segment_sessionized_pages")} as segment_sessionized_pages
     using(page_id)
   left join ${ctx.ref(params.defaultConfig.schema, "segment_sessionized_tracks")} as segment_sessionized_tracks
     using(track_id)`)}
 group by
   session_id, session_index, user_id 
-/* See TODO at start of script
+/* See TODO at start of script */
   ${Object.entries({...segmentCommon.PAGE_FIELDS, ...customPageFieldsObj}).map(
       ([key, value]) => `, first_${value}`).join(" ")}
-  ${Object.entries({...segmentCommon.PAGE_FIELDS, ...customPageFieldsObj}).map(
+ /* ${Object.entries({...segmentCommon.PAGE_FIELDS, ...customPageFieldsObj}).map(
       ([key, value]) => `, last_${value}`).join(" ")}
 */
 `)

includes/track_events.js

@@ -11,7 +11,7 @@ module.exports = (params) => {
   }).query(ctx => `
 
 -- format track calls into a format suitable to join with page calls
-select
+select distinct
   tracks.timestamp,
   user_id,
   anonymous_id,

includes/users.js

@@ -8,6 +8,9 @@ let USER = `coalesce(
 
 module.exports = (params) => {
   return publish("segment_users", {
+    assertions: {
+      uniqueKey: ["user_id"]
+    },
     description: "Users aggregates all identifies calls to give a table with one row per user_id. Identify calls without only an anonymous_id are mapped to the user where possible.",
     columns: {
       user_id: "Unique identifier of the user",
@@ -24,14 +27,16 @@ select distinct
         ignore_nulls: true,
         partition_fields: USER,
         order_fields: "identifies.timestamp asc",
+        frame_clause: "rows between unbounded preceding and unbounded following",
       })} as first_seen_at
   ${params.customUserFields.length ? `,` : ``}
-  ${params.customUserFields.map(f=> `${crossdb.windowFunction({
+  ${params.customUserFields.map(f => `${crossdb.windowFunction({
         func: "first_value",
         value: f,
         ignore_nulls: true,
         partition_fields: USER,
         order_fields: "identifies.timestamp desc",
+        frame_clause: "rows between unbounded preceding and unbounded following",
       })} as ${f}`).join(",\n  ")}
 from
   ${ctx.ref(params.defaultConfig.schema, "segment_user_map")} as segment_user_anonymous_map