Updating milestones in place

Description

Lesson 3 built fact_sessions in full with a single aggregate query — the equivalent of a total recalculation, as if Kiosko processed the whole week of events all at once, at the end. In production, events don't arrive that way: they arrive one at a time, in the order they happen, and the pipeline that maintains fact_sessions processes them as they arrive, without waiting to have the full week available. This lesson rebuilds the same table, event by event, in real chronological order — and verifies, with a direct comparison, that it produces exactly the same result as lesson 3's aggregate recalculation.

Connection to the module. This lesson demonstrates that lesson 2's minimal mechanism (INSERT on the first milestone, UPDATE on the following ones) scales, with no change in logic, to Kiosko's full 32 events — and that the result is identical, row by row, to what lesson 3's aggregate approach produced. This equivalence is the foundation of lesson 5, which applies the same "process incrementally" idea to a different pattern.

An analogy: the list of cubbyholes, checked one at a time

Imagine Kiosko had a physical board with one cubbyhole per active session, and an employee reviewing the queue of events arriving from the system, one after another, in the order they happened. When the first event of a new session arrives — always a page_view — the employee opens a new cubbyhole for that session. When an event arrives for a session that already has a cubbyhole, the employee doesn't open a new one: they write the data over the existing cubbyhole. The employee never needs to know, up front, how many events they're going to receive in total, or in what exact order each session's milestones are going to arrive — they only need to know, for each event that arrives, whether the session already has a cubbyhole or not.

That's exactly what this lesson builds in code: a loop that reviews each event once, decides whether the session already exists in fact_sessions, and acts accordingly — INSERT if it doesn't exist, UPDATE if it already does.

Worked example: processing 32 events, one at a time, in chronological order

This lesson's central algorithm has a simple rule: for each event, if the session doesn't exist yet in fact_sessions, a new row gets inserted (always with a page_view, the funnel's first milestone); if the session already exists, the column corresponding to the event type gets updated, without touching the rest of the row.

# fact_sessions_incremental.py -- continues on top of RAW_EVENTS and store_for_session (lesson 3)
from datetime import datetime
import duckdb

con = duckdb.connect()
con.execute("""
    CREATE TABLE fact_sessions (
        session_id VARCHAR PRIMARY KEY,
        store_id VARCHAR,
        session_date DATE,
        view_ts TIMESTAMP,
        add_to_cart_ts TIMESTAMP,
        purchase_ts TIMESTAMP,
        is_converted BOOLEAN
    )
""")

# CHRONOLOGICAL order -- exactly how the events would arrive in production,
# not the order they're declared in events.py
events_sorted = sorted(RAW_EVENTS, key=lambda r: r[3])

inserts, updates = 0, 0
for event_id, event_type, session_id, event_ts_str in events_sorted:
    event_ts = datetime.fromisoformat(event_ts_str)
    exists = con.execute(
        "SELECT 1 FROM fact_sessions WHERE session_id = ?", [session_id]
    ).fetchone()

    if exists is None:
        # First event of this session -- ALWAYS page_view in this dataset -- INSERT
        con.execute(
            "INSERT INTO fact_sessions VALUES (?, ?, ?, ?, NULL, NULL, false)",
            [session_id, store_for_session(session_id), event_ts.date(), event_ts],
        )
        inserts += 1
    elif event_type == "add_to_cart":
        # The session already exists -- UPDATE in place, never a new row
        con.execute(
            "UPDATE fact_sessions SET add_to_cart_ts = ? WHERE session_id = ?",
            [event_ts, session_id],
        )
        updates += 1
    elif event_type == "purchase":
        con.execute(
            "UPDATE fact_sessions SET purchase_ts = ?, is_converted = true WHERE session_id = ?",
            [event_ts, session_id],
        )
        updates += 1

print(f"events processed: {len(events_sorted)} (INSERT: {inserts}, UPDATE: {updates})")
print(f"final rows in fact_sessions: {con.sql('SELECT COUNT(*) FROM fact_sessions').fetchone()[0]}\n")
con.sql("SELECT * FROM fact_sessions ORDER BY session_id").show(max_width=300)

What to expect.

events processed: 32 (INSERT: 17, UPDATE: 15)
final rows in fact_sessions: 17

┌────────────┬──────────┬──────────────┬─────────────────────┬─────────────────────┬─────────────────────┬──────────────┐
│ session_id │ store_id │ session_date │       view_ts       │   add_to_cart_ts    │     purchase_ts     │ is_converted │
│  varchar   │ varchar  │     date     │      timestamp      │      timestamp      │      timestamp      │   boolean    │
├────────────┼──────────┼──────────────┼─────────────────────┼─────────────────────┼─────────────────────┼──────────────┤
│ SESS-01    │ S01      │ 2026-08-03   │ 2026-08-03 08:00:12 │ 2026-08-03 08:02:45 │ 2026-08-03 08:03:10 │ true         │
│ SESS-02    │ S02      │ 2026-08-03   │ 2026-08-03 08:05:00 │ NULL                │ NULL                │ false        │
│ SESS-03    │ S03      │ 2026-08-04   │ 2026-08-04 08:10:00 │ 2026-08-04 08:12:30 │ 2026-08-04 08:13:05 │ true         │
│ SESS-04    │ S01      │ 2026-08-04   │ 2026-08-04 08:20:00 │ NULL                │ NULL                │ false        │
│ SESS-05    │ S02      │ 2026-08-04   │ 2026-08-04 08:45:00 │ NULL                │ NULL                │ false        │
│ SESS-06    │ S03      │ 2026-08-05   │ 2026-08-05 08:00:00 │ NULL                │ NULL                │ false        │
│ SESS-07    │ S01      │ 2026-08-05   │ 2026-08-05 08:15:00 │ 2026-08-05 08:16:20 │ NULL                │ false        │
│ SESS-08    │ S02      │ 2026-08-06   │ 2026-08-06 08:05:00 │ 2026-08-06 08:07:15 │ 2026-08-06 08:08:00 │ true         │
│ SESS-09    │ S03      │ 2026-08-06   │ 2026-08-06 08:30:00 │ NULL                │ NULL                │ false        │
│ SESS-10    │ S01      │ 2026-08-07   │ 2026-08-07 08:00:00 │ 2026-08-07 08:03:10 │ 2026-08-07 08:04:00 │ true         │
│ SESS-11    │ S02      │ 2026-08-07   │ 2026-08-07 08:20:00 │ 2026-08-07 08:22:00 │ NULL                │ false        │
│ SESS-12    │ S03      │ 2026-08-07   │ 2026-08-07 08:50:00 │ NULL                │ NULL                │ false        │
│ SESS-13    │ S01      │ 2026-08-08   │ 2026-08-08 07:55:00 │ 2026-08-08 07:58:00 │ 2026-08-08 07:59:10 │ true         │
│ SESS-14    │ S02      │ 2026-08-08   │ 2026-08-08 08:10:00 │ 2026-08-08 08:12:45 │ 2026-08-08 08:13:30 │ true         │
│ SESS-15    │ S03      │ 2026-08-08   │ 2026-08-08 08:40:00 │ NULL                │ NULL                │ false        │
│ SESS-16    │ S01      │ 2026-08-09   │ 2026-08-09 09:00:00 │ NULL                │ NULL                │ false        │
│ SESS-17    │ S02      │ 2026-08-09   │ 2026-08-09 09:20:00 │ 2026-08-09 09:22:00 │ NULL                │ false        │
└────────────┴──────────┴──────────────┴─────────────────────┴─────────────────────┴─────────────────────┴──────────────┘

Stop on the first line: INSERT: 17, UPDATE: 15. Seventeen new sessions — one for each page_view that opens a session — and fifteen updates — the 9 add_to_cart plus the 6 purchase that followed one of those sessions. 17 + 15 = 32, the exact total number of events, with none lost or counted twice. And the final result — seventeen rows, with exactly the same values in every column — is identical to the table lesson 3 built with MAX(CASE WHEN ...).

Verifying the equivalence between the two approaches

The two tables "looking the same" isn't enough evidence — this lesson confirms it by comparing them row by row:

# fact_sessions_batch.py -- rebuilds lesson 3's version, under a different table name
con.execute("""
    CREATE TABLE fact_sessions_batch AS
    SELECT
        e.session_id, m.store_id, MIN(CAST(e.event_ts AS DATE)) AS session_date,
        MAX(CASE WHEN e.event_type = 'page_view'   THEN e.event_ts END) AS view_ts,
        MAX(CASE WHEN e.event_type = 'add_to_cart' THEN e.event_ts END) AS add_to_cart_ts,
        MAX(CASE WHEN e.event_type = 'purchase'    THEN e.event_ts END) AS purchase_ts,
        MAX(CASE WHEN e.event_type = 'purchase' THEN true ELSE false END) AS is_converted
    FROM events e JOIN session_store_map m ON e.session_id = m.session_id
    GROUP BY e.session_id, m.store_id
""")

mismatches = con.sql("""
    SELECT COUNT(*) FROM fact_sessions incr
    JOIN fact_sessions_batch batch ON incr.session_id = batch.session_id
    WHERE incr.store_id != batch.store_id
       OR incr.view_ts != batch.view_ts
       OR incr.add_to_cart_ts IS DISTINCT FROM batch.add_to_cart_ts
       OR incr.purchase_ts IS DISTINCT FROM batch.purchase_ts
       OR incr.is_converted != batch.is_converted
""").fetchone()[0]

print(f"rows where the incremental version differs from the aggregate version: {mismatches}")
assert mismatches == 0, "the two versions of fact_sessions don't match"
print("Verification OK: the event-by-event approach produces EXACTLY the same table as MAX(CASE WHEN...)")

What to expect.

rows where the incremental version differs from the aggregate version: 0
Verification OK: the event-by-event approach produces EXACTLY the same table as MAX(CASE WHEN...)

Zero differences. This confirms something important about the accumulating snapshot pattern: the mechanical order in which you build the table — all at once with an aggregation, or one event at a time with INSERT/UPDATE — doesn't change the result, as long as the set of input events is the same. What does change is which of the two approaches looks more like how a real production system works: a pipeline that receives events continuously (for example, one that reads from a message queue or an event CDC) uses this lesson's INSERT/UPDATE pattern, processing each event as it arrives, without waiting to have a full day's worth of data.

Diagram: two paths, the same destination

flowchart TD
    subgraph batch["Lesson 3: aggregate approach"]
        A["32 complete events\n(the whole week available)"] --> B["MAX(CASE WHEN...)\nGROUP BY session_id"]
        B --> C["fact_sessions_batch\n17 rows"]
    end

    subgraph incremental["Lesson 4: incremental approach"]
        D["Event 1 arrives"] --> E{"does the session\nalready exist?"}
        E -->|"No"| F["INSERT\nnew incomplete row"]
        E -->|"Yes"| G["UPDATE\nthe milestone column"]
        F --> H["Next event..."]
        G --> H
        H -.->|"repeats 32 times"| D
        H --> I["fact_sessions\n17 rows"]
    end

    C -.->|"0 differences, verified"| I

Going deeper: why events' chronological order matters here, and didn't in lesson 3

In lesson 3, the order in which events appears in the table doesn't affect the result — MAX(CASE WHEN ...) looks at all of a session's events at once, regardless of what sequence the engine processes them in internally. In this lesson, order does matter, for a concrete reason: the algorithm decides whether to do INSERT or UPDATE by looking at whether the session already exists at that point in processing — a decision that depends on which events already got processed before. If this lesson processed events out of chronological order (for example, a purchase before its corresponding page_view), the algorithm would attempt an UPDATE on a session that doesn't exist yet, and would find no row to update — the purchase would get silently lost, with no visible error.

This is a real limitation, not an academic detail: in a production system, events can arrive out of order for reasons of network delay or message-queue partitioning. The complete solution to that problem — what to do when a milestone arrives "before" the one that should have preceded it — is exactly the late-arriving dimensions problem module 5 already named for dim_product_scd, applied here to facts instead of dimensions; solving it thoroughly is out of scope for this guide (it belongs to streaming-with-kafka-and-flink-guide, where the real-time event arrival order is in fact handled with streaming-system guarantees). This lesson assumes, like this guide's entire executable thread, that events are processed in the order they happened.

Common mistakes

Processing events in the order it appears in the file, without sorting by event_ts. What happens: someone iterates over RAW_EVENTS directly, in the order it's declared in Python, without applying sorted(..., key=lambda r: r[3]) first. Why it happens: RAW_EVENTS is already declared, visually, in an order that looks chronological — grouped by day — so re-sorting seems unnecessary. How to spot it: if your result differs from this lesson's on any session, check whether you're relying on declaration order instead of explicitly sorting by event_ts — declaration order in a Python file is never a formal guarantee of chronological order, even though it happens to match in this particular case. How to fix it: always sort explicitly by the time column before processing events incrementally — sorted(events, key=lambda r: r[3]), as this lesson does, never trust the source file's order.

Using INSERT for the second and third events of a session too. What happens: someone, while porting this lesson's algorithm, forgets the if exists is None and uses INSERT ... ON CONFLICT or similar for every event, expecting the primary-key conflict to silently "fix" the problem. Why it happens: INSERT ... ON CONFLICT DO UPDATE is a common pattern in other contexts, and it can look like a valid shortcut here. How to spot it: if your final row count isn't 17, or your INSERT/UPDATE count doesn't add up to exactly 32, something in the decision logic is wrong. How to fix it: this lesson's pattern is deliberately explicit — SELECT to check existence, then INSERT or UPDATE depending on the result — because it makes the accumulating snapshot's central decision visible in the code itself: is this a new process, or one that already exists?

Thinking the number of UPDATEs (15) should equal the number of sessions with more than one event. What happens: someone expects 15 UPDATE to correspond to "15 sessions that had more than one event," when it actually corresponds to "15 events that weren't the first one of their session." Why it happens: it's easy to confuse an event count with a session count when both numbers appear side by side. How to spot it: if you count how many sessions have add_to_cart_ts or purchase_ts not null (9 sessions reached the cart, some of those also bought), you're not going to get to 15 — you're going to get a smaller number, because some sessions generate two UPDATEs (one for add_to_cart, another for purchase), not one. How to fix it: 15 UPDATE is a count of non-initial events, not of sessions — it counts each add_to_cart and each purchase separately, even if both belong to the same session (like SESS-01, SESS-03, SESS-08, SESS-10, SESS-13, SESS-14, which each generate two UPDATEs).

Exercises

Exercise 1 — Count how many sessions generated exactly two UPDATEs. Without re-running the full script, use fact_sessions as already built to count how many sessions have both add_to_cart_ts and purchase_ts filled (those are exactly the ones that generated two UPDATEs each during incremental processing).

print(con.sql("""
    SELECT COUNT(*) AS sessions_with_two_updates
    FROM fact_sessions
    WHERE add_to_cart_ts IS NOT NULL AND purchase_ts IS NOT NULL
"""))

┌───────────────────────────┐
│ sessions_with_two_updates │
│           int64           │
├───────────────────────────┤
│                          6 │
└───────────────────────────┘

Exercise 2 — Simulate an out-of-order event and observe what happens. Add, at the end of events_sorted (without re-sorting), a fake purchase event for a session SESS-99 that never had a prior page_view. Run the processing loop on that single event and describe what happens.

con.execute("DELETE FROM fact_sessions WHERE session_id = 'SESS-99'")  # in case it already existed
exists = con.execute("SELECT 1 FROM fact_sessions WHERE session_id = 'SESS-99'").fetchone()
print(f"SESS-99 exists before processing the orphan event: {exists}")

if exists is None:
    # the event is 'purchase', but this lesson's code only knows how to INSERT
    # new sessions when the FIRST event is a page_view -- an orphan purchase
    # doesn't fit any branch of the if/elif, and gets silently lost
    print("This event doesn't match any branch of the algorithm: it gets discarded with no visible error.")

SESS-99 exists before processing the orphan event: None
This event doesn't match any branch of the algorithm: it gets discarded with no visible error.

Exercise 3 — Explain, in your own words, why this lesson didn't need to change a single column of fact_sessions's schema relative to lesson 3. In 2-3 sentences, explain exactly what changed between lessons 3 and 4, and what stayed identical.

Summary and next step

This lesson rebuilt fact_sessions by processing Kiosko's 32 events one at a time, in chronological order — INSERT when a session starts (17 times), UPDATE when it advances (15 times) — and verified, by comparing row by row, that the result is identical to what lesson 3's aggregate approach produced: zero differences. This confirms that lesson 2's minimal INSERT+UPDATE mechanism scales without changes to the full session population, and that it's the mechanism closest to how a real production pipeline would maintain this table — receiving events continuously, without waiting to have all the data available up front.

Before moving on you should be able to: explain why events' chronological order does matter in this lesson although it didn't in lesson 3; describe what would happen if an event arrived out of order; and calculate how many INSERTs and UPDATEs a set of events is going to generate, without running the code.

With the first pattern — accumulating snapshot — complete and verified two different ways, lesson 5 changes topic: it introduces Zach Wilson's cumulative table design, a pattern that doesn't describe a process with an end (like a session that ends in a purchase or an abandonment), but a continuous series of daily activity per store, where each new row gets built from the previous day's row.

Resources

• Kimball Group — "Accumulating Snapshot Fact Table" — the pattern this lesson verifies with a second construction mechanism. kimballgroup.com/data-warehouse-business-intelligence-resources/kimball-techniques/dimensional-modeling-techniques/accumulating-snapshot-fact-table. In English.
• Kimball Group — "Late Arriving Dimension" — the technique related to the out-of-order events problem this lesson names in its deep dive, already introduced in module 5 for dimensions. kimballgroup.com/data-warehouse-business-intelligence-resources/kimball-techniques/dimensional-modeling-techniques/late-arriving-dimension. In English.
• DuckDB — official Python client documentation, the interface that ran every incremental INSERT/UPDATE in this lesson. duckdb.org/docs/current/clients/python/overview. In English.