attribution-lab

A marketing-attribution pipeline at realistic scale: 42.4M real e-commerce events loaded into range-partitioned PostgreSQL, indexed and pre-aggregated for dashboard-grade query latency, with first-touch / last-touch / linear attribution computed in a single window-function pass.

Built as a working study of the data problems behind e-commerce attribution platforms: high-volume append-only ingest, time-partitioned storage, query optimization on multi-GB tables, and recomputable attribution models over immutable raw events.

Architecture

2019-Oct.csv (42.4M rows, REES46)
      |
      |  psql \copy  (COPY protocol, unlogged staging, synchronous_commit=off)
      v
staging (UNLOGGED)
      |
      |  INSERT ... SELECT: type casts, ''->NULL, channel simulation (hash bucketing)
      v
events  (range-partitioned by month, BRIN + composite + partial indexes)
      |
      +--> daily_channel_rollup  (incremental ON CONFLICT refresh)   -> dashboard queries
      |
      +--> attribution SQL (window functions, 30-day lookback)       -> first/last/linear revenue by channel

Dataset

eCommerce behavior data from multi category store (REES46 Marketing Platform, via Kaggle). October 2019 file: 42,448,764 events (view / cart / purchase) with product, brand, price, user, and session fields.

The dataset has no marketing-channel labels, so channels are simulated deterministically: abs(hashtext(user_session)) % 100 mapped to weighted buckets (google_ads 25%, meta 20%, organic 20%, email 15%, tiktok 10%, direct 10%; null sessions get 'direct'). A session always maps to one channel. The engineering (ingest, partitioning, indexing, window functions, rollups) is real; the channel labels are synthetic, so the attribution numbers demonstrate the mechanism rather than any marketing insight.

Load pipeline and throughput

Stage	Rows	Time	Throughput
\copy CSV into unlogged staging	42,448,764	45.4 s	~935,000 rows/s
Transform + insert into partitioned events (casts, NULL mapping, channel hash)	42,448,764	140.1 s	~303,000 rows/s
Full pipeline, raw CSV to typed/partitioned/enriched	42,448,764	~3 min	~229,000 rows/s

Notes: staging is UNLOGGED (no WAL) and the session ran with synchronous_commit = off. The transform is slower than the copy because the target table pays for WAL, per-row casts and hashing, and partition routing. The table was loaded bare on purpose; indexes are added afterward so their cost and benefit can be measured rather than assumed (see below).

Scale context: a production attribution platform ingesting 20M events/day averages ~230 events/s. This single-machine pipeline sustains roughly 1,000x that, which is why the hard scaling problems in this domain are query latency and maintenance windows, not raw ingest.

Schema decisions

CREATE TABLE events (
  event_id    BIGINT GENERATED ALWAYS AS IDENTITY,
  occurred_at TIMESTAMPTZ NOT NULL,
  event_type  TEXT        NOT NULL,
  product_id  BIGINT,
  brand       TEXT,
  price       NUMERIC(10,2),
  user_id     BIGINT      NOT NULL,
  session_id  UUID,
  channel     TEXT        NOT NULL
) PARTITION BY RANGE (occurred_at);

• Range-partitioned by month. Buys partition pruning on time-windowed queries, instant retention via DROP PARTITION, and per-partition indexes that fit in cache.
• Partition bounds are written with explicit UTC offsets ('2019-10-01 00:00:00+00'). A bare date literal is cast using the session timezone; on a US/Eastern machine that shifts the boundary by 4 hours and the first 4 hours of the month have no partition to land in. Found during the data probe rather than in production.
• No primary key. This is an append-only event stream, and on a partitioned table a PK must include the partition key, so it would cost index maintenance on every insert for a uniqueness guarantee nothing queries.
• Nullable session_id. The probe found empty sessions in the raw data; the loader maps '' to NULL, matching COPY CSV semantics (an unquoted empty field is NULL).
• text over varchar(n): identical performance in Postgres, and the length caps buy nothing.

Data probe (before any pipeline code)

docs/probe-notes.md documents a ~1,000-row head sample and a ~1,000-row spread sample pushed through Postgres's own parsers before the schema was finalized. Findings that shaped the design:

• event_time format 2019-10-01 00:00:00 UTC casts cleanly to timestamptz.
• NULL rates (spread sample): brand ~13%, category ~30%, session ~0% but nonzero across 42M rows.
• The count(col) vs count(*) distinction matters. An early census using WHERE col = '' reported zero missing values because COPY had already turned empties into NULLs, and three-valued logic silently excludes NULLs from equality filters.
• Head-sample bias is real. The first 1,000 rows (first 17 minutes of Oct 1) showed zero missing brands and an inverted purchase/cart ratio; the spread sample corrected both.
• The timezone bug, quantified: a query written with bare date literals (>= '2019-10-01') on a US/Eastern session shifts the window by 4 hours, silently dropping 1,171 purchases (742,849 vs 741,678, measured as the row-count difference between the two captured plans) and breaking partition pruning (the November partition shows up in the plan because the shifted window genuinely overlaps it). Fix: ALTER DATABASE ... SET timezone TO 'UTC' plus explicit +00 offsets in range predicates.

Query performance experiment

Reference query: daily revenue by channel for October (purchases only, ~1.75% of events).

Configuration	Time	Buffers (8KB pages)	What the plan shows
Bare table (no indexes)	1,026 ms	577,266 (~4.5 GB)	Parallel seq scan of all 42.4M rows to keep 742,849 purchases; external-merge sort spill (work_mem)
With indexes (below)	232 ms	3,882 (~30 MB)	Index Only Scan on the covering partial index, Heap Fetches: 0; hash aggregate in 609 kB
Rollup table	0.078 ms	2	Seq scan of a 186-row table, which is optimal; small tables need no index

Plans preserved in docs/explain-{before,mid,after}.txt. The covering index cuts I/O ~148x, the rollup ~290,000x; wall clock goes from 1,026 ms to 0.078 ms. Time improves less than I/O on this machine because laptop NVMe (~4.5 GB/s) hides I/O cost; on production cloud disks the same buffer reduction is worth far more.

Index choices, with sizes and build times measured on the 42.4M-row partition:

Index	Size	Build	Why
BRIN on occurred_at	192 kB	1.9 s	Append-only table means physical order tracks time order; one min/max summary per 128-page range (577,266 pages / 128 = ~4,510 summaries = ~192 kB, roughly 6,600x smaller than a B-tree)
B-tree (user_id, occurred_at)	1,276 MB	37.7 s	Per-user journey scans; equality column leads, range column trails
Partial covering (occurred_at) INCLUDE (channel, price) WHERE event_type='purchase'	30 MB	2.4 s	Purchases are 1.75% of rows; the INCLUDE payload lets the revenue query run as an index-only scan and never touch the 4.5 GB heap

VACUUM ANALYZE after the load is required, not hygiene: ANALYZE gives the planner statistics, and VACUUM builds the visibility map, without which the "index-only" scan pays a heap visit per row.

The rollup is refreshed incrementally (INSERT ... ON CONFLICT (day, channel) DO UPDATE), so dashboards never touch raw events; raw stays as the immutable source of truth for reprocessing.

Attribution models

Three models computed in one pass over journeys (touchpoints within a 30-day lookback before each purchase) using window functions:

ROW_NUMBER() OVER (PARTITION BY order ORDER BY occurred_at)        -- first-touch rank
ROW_NUMBER() OVER (PARTITION BY order ORDER BY occurred_at DESC)   -- last-touch rank
COUNT(*)     OVER (PARTITION BY order)                             -- linear denominator
...
SUM(revenue) FILTER (WHERE first_rank = 1)   AS first_touch_rev,
SUM(revenue) FILTER (WHERE last_rank  = 1)   AS last_touch_rev,
SUM(revenue / n_touches)                     AS linear_rev

Touchpoints are sessions (visits), not pageviews: 42.4M events collapse to 9,244,770 sessions (3.1 per user, ~4.6 events per session). Building the sessions table also surfaced a data quirk a 1,000-row probe could not: a PRIMARY KEY (session_id) failed because some session ids appear under more than one user id (an artifact of the source data's user identification). The PK is (user_id, session_id) instead, and a shared session counts as a touch in each user's journey.

Results (October, 30-day lookback, 11.9 s over 742,849 purchases):

channel	first_touch_rev	last_touch_rev	linear_rev	orders (first touch)
google_ads	56,804,326.89	57,408,099.18	57,440,323.76	184,953
meta	45,342,065.62	46,001,833.72	45,983,225.71	148,976
organic	45,933,994.53	45,981,130.31	45,862,007.31	148,244
email	34,918,735.70	34,659,581.38	34,622,421.21	111,075
tiktok	23,547,277.58	23,115,274.61	23,167,293.15	74,719
direct	23,411,101.95	22,791,583.07	22,882,231.12	74,882

Validation:

• Conservation: each model's column sums to 229,957,502.27, the total attributed revenue, to the cent (each order's price is handed out exactly once per model; linear's fractions re-add to the whole).
• Coverage: 742,849 of 742,849 purchases attributed (100%). Every purchase event carries a session, so every order has at least one touch; zero purchases have a NULL session.
• Weight recovery: linear shares come out 25.0 / 20.0 / 19.9 / 15.1 / 10.1 / 9.9 percent, matching the simulation weights almost exactly. That is the expected result for randomly assigned channels and validates the mechanism; on real pixel data the models diverge, and that divergence is the product.

Because raw events are immutable and attribution is derived, changing the lookback window (7 vs 30 vs 90 days) is a reprocess, not a migration. That is the architectural reason to keep raw events around even after rollups exist.

What I'd do at 10x scale

1. Measure first: pg_stat_statements to find the workload's actual top costs before touching anything.
2. Partition maintenance becomes policy: automated monthly partition creation, retention by DROP PARTITION.
3. Rollups own the dashboards: incremental aggregates per (day, channel, tenant); raw events only for reprocessing and audit.
4. Read replicas for the read-heavy dashboard workload; PgBouncer in transaction mode in front of everything.
5. Tier cold partitions to object storage or a columnar warehouse once they age past the maximum lookback window; shard by tenant id (every query in a multi-tenant attribution product is per-store) only when the cheaper rungs are exhausted.

Repo layout

sql/        schema, load, indexes, rollup + refresh, attribution queries
docs/       probe-notes.md, explain-before/mid/after.txt
cmd/        Go: rollup query client (pgx)
data/       (gitignored) raw CSVs