· 8 years ago · Feb 06, 2018, 08:12 PM
1---- Select new rows then rows of new users/rows of existing users, then segment existing users by whether or not their new
2---- rows are a new session or not
3--
4---- 1. Select newly added rows
5---- 2. Select rows of new users *
6---- 3. Select rows of existing users
7---- A. Select rows that are a new session *
8---- B. Select rows that are extending a previous session *
9---- a. Delete this session, recalculate
10---- C. Select rows that start before the end of the existing sessions *
11---- a. Delete session info for sessions including and after first new row time
12--
13----
14
15
16---------------------------------------------------------------------------------------------------
17-- Create sessions and sessions_info tables
18---------------------------------------------------------------------------------------------------
19
20CREATE TABLE IF NOT EXISTS test.tracks_sessions_info
21(
22 "session_id" VARCHAR(534) ENCODE zstd
23 ,"anonymous_id" VARCHAR(512) ENCODE zstd
24 ,"session_start" TIMESTAMP WITHOUT TIME ZONE ENCODE zstd
25 ,"session_end" TIMESTAMP WITHOUT TIME ZONE ENCODE zstd
26 ,"last_event_time" TIMESTAMP WITHOUT TIME ZONE ENCODE zstd
27 ,"num_events" INTEGER ENCODE zstd
28 ,"session_time_seconds" INTEGER ENCODE zstd
29 ,"session_seq_number" BIGINT ENCODE zstd
30 ,"next_session_start" TIMESTAMP WITHOUT TIME ZONE ENCODE zstd
31 ,"context_campaign_source" VARCHAR(512) ENCODE zstd
32 ,"context_campaign_medium" VARCHAR(512) ENCODE zstd
33 ,"context_campaign_name" VARCHAR(512) ENCODE zstd
34 ,"context_page_referrer" VARCHAR(512) ENCODE zstd
35)
36 DISTSTYLE KEY
37 DISTKEY ("anonymous_id")
38 SORTKEY (
39 "anonymous_id"
40 , "session_start"
41)
42;
43
44CREATE TABLE IF NOT EXISTS test.tracks_sessions
45(
46 "id" VARCHAR(512) ENCODE zstd
47 ,"anonymous_id" VARCHAR(512) ENCODE zstd
48 ,"session_id" VARCHAR(534) ENCODE zstd
49 ,"session_num" BIGINT ENCODE zstd
50 ,"received_at" TIMESTAMP WITHOUT TIME ZONE ENCODE zstd
51 ,"tablename" VARCHAR(50) ENCODE zstd
52)
53 DISTSTYLE KEY
54 DISTKEY ("id")
55 SORTKEY (
56 "received_at"
57)
58;
59
60
61---------------------------------------------------------------------------------------------------
62-- Load session IDs to delete to temp table
63---- This is all sessions that are extended by or after newly added rows which need to be
64---- recalculated
65---------------------------------------------------------------------------------------------------
66
67BEGIN;
68
69CREATE TEMP TABLE sessions_to_delete_temp
70(
71 "session_id" VARCHAR
72)
73DISTSTYLE ALL
74SORTKEY ("session_id")
75;
76
77INSERT INTO sessions_to_delete_temp
78
79WITH all_source AS (
80 SELECT id,
81 anonymous_id,
82 received_at,
83 'tracks' AS tablename
84 FROM test.tracks
85
86 UNION ALL
87
88 SELECT id,
89 anonymous_id,
90 received_at,
91 'pages' AS tablename
92 FROM test.pages
93),
94
95-- All newly added rows, which are not yet in the sessions table
96new_rows AS (
97 SELECT e.*
98 FROM all_source e
99 LEFT OUTER JOIN test.tracks_sessions s ON e.id = s.id
100 AND e.tablename = s.tablename
101 WHERE s.id IS NULL
102 ORDER BY received_at ASC
103 LIMIT 60000000
104),
105
106-- Newly added rows for returning users
107existing_users_rows AS (
108 SELECT *
109 FROM new_rows
110 WHERE new_rows.anonymous_id IN (SELECT anonymous_id
111 FROM test.tracks_sessions_info
112 GROUP BY 1)
113 ),
114
115-- Calculate the first timestamp of a user's new rows and the last timestamp of their existing rows
116existing_users_event_info AS (
117 SELECT eur.anonymous_id,
118 min(eur.received_at) AS first_new_event_time,
119 max(tsi.last_event_time) AS last_old_event_time
120 FROM existing_users_rows eur JOIN test.tracks_sessions_info tsi
121 ON eur.anonymous_id = tsi.anonymous_id
122 GROUP BY eur.anonymous_id
123),
124
125-- Grab all session_ids where new events occur before the session's last event + 120 seconds
126-- This is all sessions that need to be recalculated - they are either extended or intersected
127-- by new events
128sessions_to_delete AS (
129 SELECT tsi.session_id
130 FROM existing_users_event_info euei JOIN test.tracks_sessions_info tsi
131 ON euei.anonymous_id = tsi.anonymous_id
132 WHERE euei.first_new_event_time <= tsi.last_event_time + INTERVAL '120 seconds'
133 GROUP BY 1
134 )
135SELECT session_id
136FROM sessions_to_delete
137ORDER BY 1;
138
139
140---------------------------------------------------------------------------------------------------
141-- Delete sessions from sessions and sessions_info table
142---------------------------------------------------------------------------------------------------
143
144DELETE FROM test.tracks_sessions
145USING sessions_to_delete_temp
146WHERE test.tracks_sessions.session_id = sessions_to_delete_temp.session_id;
147
148DELETE FROM test.tracks_sessions_info
149USING sessions_to_delete_temp
150WHERE test.tracks_sessions_info.session_id = sessions_to_delete_temp.session_id;
151
152
153
154---------------------------------------------------------------------------------------------------
155-- Calculate new sessions_info
156---------------------------------------------------------------------------------------------------
157
158-- First create the new sessions info rows
159INSERT INTO test.tracks_sessions_info
160WITH all_source AS (
161 SELECT id,
162 anonymous_id,
163 received_at,
164 'tracks' AS tablename
165 FROM test.tracks
166
167 UNION ALL
168
169 SELECT id,
170 anonymous_id,
171 received_at,
172 'pages' AS tablename
173 FROM test.pages
174),
175
176-- 1
177new_rows AS (
178 SELECT e.*
179 FROM all_source e
180 LEFT OUTER JOIN test.tracks_sessions s ON e.id = s.id
181 AND e.tablename = s.tablename
182 WHERE s.id IS NULL
183 ORDER BY received_at ASC
184 LIMIT 60000000
185 ),
186
187-- 2
188new_users_rows AS (
189 SELECT *
190 FROM new_rows
191 WHERE new_rows.anonymous_id NOT IN (SELECT anonymous_id
192 FROM test.tracks_sessions_info
193 GROUP BY 1)
194 ),
195
196-- 3
197existing_users_rows AS (
198 SELECT *
199 FROM new_rows
200 WHERE new_rows.anonymous_id IN (SELECT anonymous_id
201 FROM test.tracks_sessions_info
202 GROUP BY 1)
203 ),
204
205-- Explicitly calculate event time bounds for existing users
206-- The timestamp of their earliest newly added event, and the timestamp of their latest existing event
207existing_users_event_info AS (
208 SELECT eur.anonymous_id,
209 min(eur.received_at) AS first_new_event_time,
210 max(tsi.last_event_time) AS last_old_event_time,
211 max(tsi.session_seq_number) AS last_session_number,
212 max(tsi.session_seq_number) || '-' || eur.anonymous_id AS last_session_id
213 FROM existing_users_rows eur JOIN test.tracks_sessions_info tsi
214 ON eur.anonymous_id = tsi.anonymous_id
215 GROUP BY eur.anonymous_id
216 ),
217
218-- 3 A
219new_session_rows AS (
220 SELECT eur.*
221 FROM existing_users_rows eur JOIN existing_users_event_info euei
222 ON eur.anonymous_id = euei.anonymous_id
223 WHERE DATEDIFF('milliseconds', euei.last_old_event_time, euei.first_new_event_time) > 120000
224 ),
225lag AS (
226 -- For each event, calculates the amount of seconds that has passed since the previous event per user (idle_time)
227 -- and appends a session number offset, where the session_seq_number will start from
228 -- Sessions for users with no previous events/sessions
229 SELECT nur.id AS event_id,
230 nur.anonymous_id AS anonymous_id,
231 nur.received_at AS event_time,
232 DATEDIFF('milliseconds', LAG(nur.received_at)
233 OVER (PARTITION BY nur.anonymous_id ORDER BY nur.received_at),
234 nur.received_at) AS idle_time,
235 0 AS session_number_offset
236 FROM new_users_rows nur
237
238 UNION ALL
239
240 -- Sessions for users with previous events/sessions
241 SELECT nsr.id AS event_id,
242 nsr.anonymous_id AS anonymous_id,
243 nsr.received_at AS event_time,
244 DATEDIFF('milliseconds', LAG(nsr.received_at)
245 OVER (PARTITION BY nsr.anonymous_id ORDER BY nsr.received_at),
246 nsr.received_at) AS idle_time,
247 euei.last_session_number AS session_number_offset
248 FROM new_session_rows nsr JOIN existing_users_event_info euei
249 ON nsr.anonymous_id = euei.anonymous_id
250 )
251
252SELECT lag.session_number_offset + ROW_NUMBER() OVER (PARTITION BY lag.anonymous_id ORDER BY lag.event_time)
253 || '-' || lag.anonymous_id AS session_id,
254 lag.anonymous_id,
255 lag.event_time AS session_start,
256 NULL::timestamp AS session_end,
257 NULL::timestamp AS last_event_time,
258 NULL::integer AS num_events,
259 NULL::integer AS session_time_seconds,
260 lag.session_number_offset + ROW_NUMBER() OVER (PARTITION BY lag.anonymous_id ORDER BY lag.event_time)
261 AS session_seq_number,
262 -- Set the next session start time to either the next new session time, or way in the future
263 COALESCE(LEAD(lag.event_time) OVER (PARTITION BY lag.anonymous_id ORDER BY lag.event_time),
264 '3000-01-01'::timestamp) AS next_session_start,
265 NULL::varchar AS context_campaign_source,
266 NULL::varchar AS context_campaign_medium,
267 NULL::varchar AS context_campaign_name,
268 NULL::varchar AS context_page_referrer
269FROM lag
270WHERE (lag.idle_time > 120000 OR lag.idle_time IS NULL)
271ORDER BY 2, 3
272;
273
274---------------------------------------------------------------------------------------------------
275-- Ensure sessions that used to be the last user session have next_session_time properly set
276-- (next_session_time = '3000-01-01'::timestamp AND session_end IS NOT NULL
277---------------------------------------------------------------------------------------------------
278
279UPDATE test.tracks_sessions_info
280SET next_session_start = comp.next_session_start
281FROM (
282 SELECT s.session_id AS session_id,
283 COALESCE(LEAD(s.session_start) OVER (PARTITION BY s.anonymous_id ORDER BY s.session_seq_number),
284 '3000-01-01'::timestamp) AS next_session_start
285 FROM test.tracks_sessions_info s
286 ) comp
287WHERE comp.session_id = test.tracks_sessions_info.session_id
288 AND test.tracks_sessions_info.session_end IS NOT NULL
289 AND test.tracks_sessions_info.next_session_start = '3000-01-01'::timestamp
290 AND test.tracks_sessions_info.next_session_start != comp.next_session_start;
291
292---------------------------------------------------------------------------------------------------
293-- Calculate new sessions
294---------------------------------------------------------------------------------------------------
295
296INSERT INTO test.tracks_sessions
297WITH all_source AS (
298 SELECT id,
299 anonymous_id,
300 received_at,
301 'tracks' AS tablename
302 FROM test.tracks
303
304 UNION ALL
305
306 SELECT id,
307 anonymous_id,
308 received_at,
309 'pages' AS tablename
310 FROM test.pages
311),
312
313-- 1
314new_rows AS (
315 SELECT e.*
316 FROM all_source e
317 LEFT OUTER JOIN test.tracks_sessions s ON e.id = s.id
318 AND e.tablename = s.tablename
319 WHERE s.id IS NULL
320 ORDER BY received_at ASC
321 LIMIT 60000000
322)
323SELECT e.id,
324 e.anonymous_id,
325 s.session_id,
326 s.session_seq_number AS session_num,
327 e.received_at,
328 e.tablename
329FROM new_rows e INNER JOIN test.tracks_sessions_info s
330 ON e.anonymous_id = s.anonymous_id
331 AND e.received_at >= s.session_start
332 AND e.received_at < s.next_session_start;
333
334
335
336
337
338---------------------------------------------------------------------------------------------------
339-- Update sessions_info table with session end time, number events in session
340---------------------------------------------------------------------------------------------------
341
342
343UPDATE test.tracks_sessions_info
344SET session_end = comp.session_end,
345 last_event_time = comp.last_event_time,
346 num_events = comp.num_events,
347 session_time_seconds = DATEDIFF('seconds', session_start, comp.session_end),
348 context_campaign_source = comp.context_campaign_source,
349 context_campaign_medium = comp.context_campaign_medium,
350 context_campaign_name = comp.context_campaign_name
351FROM
352 (SELECT s.session_id AS session_id,
353 LEAST(MAX(e.received_at) + INTERVAL '15 seconds',
354 MIN(s.next_session_start)) AS session_end,
355 MAX(e.received_at) AS last_event_time,
356 COUNT(DISTINCT e.id) AS num_events,
357 MAX(tpp.context_campaign_source) AS context_campaign_source,
358 MAX(tpp.context_campaign_medium) AS context_campaign_medium,
359 MAX(tpp.context_campaign_name) AS context_campaign_name,
360 MAX(tpp.context_page_referrer) AS context_page_referrer
361 FROM test.tracks_sessions_info s
362 LEFT JOIN test.tracks_sessions e ON s.session_id = e.session_id
363 --LEFT JOIN test.tracks_pages tp ON tp.id = e.id AND tp.tablename = e.tablename
364 LEFT JOIN (
365 SELECT tp.id AS id,
366 tp.tablename AS tablename,
367 FIRST_VALUE(tp.context_campaign_source ignore nulls)
368 OVER (PARTITION BY ts.session_id ORDER BY tp.received_at ASC rows between unbounded preceding and unbounded following)
369 AS context_campaign_source,
370 FIRST_VALUE(tp.context_campaign_medium ignore nulls)
371 OVER (PARTITION BY ts.session_id ORDER BY tp.received_at ASC rows between unbounded preceding and unbounded following)
372 AS context_campaign_medium,
373 FIRST_VALUE(tp.context_campaign_name ignore nulls)
374 OVER (PARTITION BY ts.session_id ORDER BY tp.received_at ASC rows between unbounded preceding and unbounded following)
375 AS context_campaign_name,
376 FIRST_VALUE(tp.context_page_referrer ignore nulls)
377 OVER (PARTITION BY ts.session_id ORDER BY tp.received_at ASC rows between unbounded preceding and unbounded following)
378 AS context_page_referrer
379 FROM test.tracks_pages tp JOIN test.tracks_sessions ts ON tp.id = ts.id
380 ) tpp on tpp.id = e.id AND tpp.tablename = e.tablename
381 GROUP BY s.session_id) comp
382WHERE comp.session_id = test.tracks_sessions_info.session_id
383 AND test.tracks_sessions_info.session_end IS NULL;
384
385
386
387COMMIT;