I ran into a weird failure mode while building a predictive model for “will this checkout fail today?” The data looked fine in batch, but during real-time training, the model started thrashing—sometimes it “knew” a failure too early, sometimes it missed it entirely.

What I discovered (after a weekend of staring at logs) was that the real culprit wasn’t the algorithm—it was **feature arrival timing**. In other words: the *same row of user/order data* could arrive at different moments, and my model was accidentally learning from information that would normally show up only after the prediction cutoff.

This post walks through a concrete, niche approach I built: **predicting same-day checkout failures using feature-arrival time windows**. The goal is simple: train on what would have been known at prediction time, and make the data observability around timing explicit.

---

## The specific problem: “Same-day” failures are time-sensitive

Imagine you need to predict, for each order, whether the checkout will fail **within the remainder of the day** (same calendar day as the order start).

There are at least two timestamps in play:

- `order_created_at`: when the order starts
- `checkout_event_at`: when failure/success is recorded

And there’s a third, often overlooked timestamp:

- `feature_ingested_at`: when each feature (risk score, payment metadata, fraud signals) was ingested into your warehouse

### Why this breaks models
If you train using features that were ingested *after* your prediction cutoff, the model “cheats.” In offline evaluation, it may look great, but in production it collapses because those features won’t be available yet.

So I added a rule: **only use feature values whose ingestion time is before the prediction cutoff**.

---

## The niche pattern: Feature-arrival time windows

I implemented a “windowed” feature computation:

For each order, and for each predictive signal (e.g., `risk_score`, `card_bin_risk`), create features like:

- `risk_score_last_available_before_cutoff`
- `risk_score_lookback_6h_mean` (but only if those source events were ingested before cutoff)
- `failed_checkouts_count_last_24h` (computed from events that were ingested before cutoff)

This turns timing into a first-class feature engineering concern.

---

## Example schema (what I used)

I used three tables:

1. `orders`
2. `feature_events` (time-stamped feature updates)
3. `checkout_outcomes` (labels)

### Tables

```sql
-- Orders: one row per order
-- order_id is the join key for everything.
CREATE TABLE orders (
  order_id TEXT PRIMARY KEY,
  user_id TEXT NOT NULL,
  order_created_at TIMESTAMP NOT NULL
);

-- Feature events: each row is an update to some feature for an order.
-- feature_name/value can be multiple per order.
CREATE TABLE feature_events (
  order_id TEXT NOT NULL,
  feature_name TEXT NOT NULL,
  feature_value DOUBLE PRECISION,
  feature_ingested_at TIMESTAMP NOT NULL,
  PRIMARY KEY (order_id, feature_name, feature_ingested_at)
);

-- Checkout outcomes: label is whether failure occurs same calendar day.
CREATE TABLE checkout_outcomes (
  order_id TEXT PRIMARY KEY,
  checkout_event_at TIMESTAMP NOT NULL,
  checkout_status TEXT NOT NULL  -- 'success' or 'failure'
);
```

---

## Step 1: Define the prediction cutoff

For same-day prediction, the cutoff can be “end of day at a fixed time” or “now” depending on your pipeline.

In my case, I predicted at ingestion time, using the order’s local day end as cutoff.

In SQL terms: for each order, the cutoff is `date_trunc('day', order_created_at) + interval '1 day'`.

To keep the example concrete, I’ll predict at **day-end** and label “failure if checkout happens before day-end”.

---

## Step 2: Create labels that respect the same-day boundary

Here’s the label logic I used:

- `label = 1` if there exists a `checkout_event_at` on the same calendar day and status is `'failure'`
- otherwise `label = 0`

```sql
-- Build training labels aligned to the same-day definition.
WITH labeled AS (
  SELECT
    o.order_id,
    o.user_id,
    o.order_created_at,
    co.checkout_event_at,
    co.checkout_status,
    -- end of the calendar day for the order
    (date_trunc('day', o.order_created_at) + interval '1 day') AS cutoff_at,
    CASE
      WHEN co.checkout_status = 'failure'
       AND co.checkout_event_at < (date_trunc('day', o.order_created_at) + interval '1 day')
      THEN 1 ELSE 0
    END AS label
  FROM orders o
  JOIN checkout_outcomes co
    ON co.order_id = o.order_id
)
SELECT * FROM labeled;
```

---

## Step 3: Build “feature availability” windows (no cheating)

Now the key: when I compute a feature, I filter to only feature events ingested **before the cutoff**.

### Single-value example: last known risk score before cutoff

```sql
WITH labeled AS (
  SELECT
    o.order_id,
    o.user_id,
    o.order_created_at,
    co.checkout_event_at,
    co.checkout_status,
    (date_trunc('day', o.order_created_at) + interval '1 day') AS cutoff_at,
    CASE
      WHEN co.checkout_status = 'failure'
       AND co.checkout_event_at < (date_trunc('day', o.order_created_at) + interval '1 day')
      THEN 1 ELSE 0
    END AS label
  FROM orders o
  JOIN checkout_outcomes co
    ON co.order_id = o.order_id
),
risk_last AS (
  SELECT
    l.order_id,
    -- Last feature ingestion before cutoff
    (
      SELECT fe.feature_value
      FROM feature_events fe
      WHERE fe.order_id = l.order_id
        AND fe.feature_name = 'risk_score'
        AND fe.feature_ingested_at < l.cutoff_at
      ORDER BY fe.feature_ingested_at DESC
      LIMIT 1
    ) AS risk_score_last_available
  FROM labeled l
)
SELECT
  l.order_id,
  l.user_id,
  l.order_created_at,
  l.cutoff_at,
  r.risk_score_last_available,
  l.label
FROM labeled l
LEFT JOIN risk_last r
  ON l.order_id = r.order_id;
```

**Why this matters:** if a risk score update arrives after the cutoff, it’s excluded. That keeps training aligned with production reality.

---

## Step 4: Add lookback windows (still respecting cutoff)

I also added rolling aggregates for features like “how many recent payment failures were *known* before cutoff.”

To do that, I used event tables to compute counts, but the same principle holds: only include rows ingested before `cutoff_at`.

Here’s a simplified example using a hypothetical feature event called `payment_failure_signal` (1 or 0).

```sql
WITH labeled AS (
  SELECT
    o.order_id,
    o.user_id,
    o.order_created_at,
    co.checkout_event_at,
    co.checkout_status,
    (date_trunc('day', o.order_created_at) + interval '1 day') AS cutoff_at,
    CASE
      WHEN co.checkout_status = 'failure'
       AND co.checkout_event_at < (date_trunc('day', o.order_created_at) + interval '1 day')
      THEN 1 ELSE 0
    END AS label
  FROM orders o
  JOIN checkout_outcomes co
    ON co.order_id = o.order_id
),
features AS (
  SELECT
    l.order_id,
    -- Lookback window anchored to cutoff:
    -- include only feature events ingested before cutoff_at,
    -- and within the last 24h of cutoff time.
    SUM(
      CASE
        WHEN fe.feature_ingested_at < l.cutoff_at
         AND fe.feature_ingested_at >= (l.cutoff_at - interval '24 hours')
        THEN fe.feature_value
        ELSE 0
      END
    ) AS payment_failure_signal_count_24h
  FROM labeled l
  LEFT JOIN feature_events fe
    ON fe.order_id = l.order_id
   AND fe.feature_name = 'payment_failure_signal'
  GROUP BY l.order_id
)
SELECT
  l.order_id,
  l.user_id,
  l.order_created_at,
  l.cutoff_at,
  f.payment_failure_signal_count_24h,
  l.label
FROM labeled l
LEFT JOIN features f
  ON l.order_id = f.order_id;
```

---

## Step 5: Add data observability for “feature latency”

After implementing cutoff-aligned features, my model improved—but only after I added one more guardrail: a metric for how “fresh” features were.

For each order and feature, I computed:

- `risk_score_age_seconds = cutoff_at - last_ingested_at`
- if no feature exists, I used a sentinel and flagged it

This became my early warning system for broken pipelines.

```sql
WITH labeled AS (
  SELECT
    o.order_id,
    o.user_id,
    o.order_created_at,
    co.checkout_event_at,
    co.checkout_status,
    (date_trunc('day', o.order_created_at) + interval '1 day') AS cutoff_at,
    CASE
      WHEN co.checkout_status = 'failure'
       AND co.checkout_event_at < (date_trunc('day', o.order_created_at) + interval '1 day')
      THEN 1 ELSE 0
    END AS label
  FROM orders o
  JOIN checkout_outcomes co
    ON co.order_id = o.order_id
),
risk_last AS (
  SELECT
    l.order_id,
    fe_last.feature_value AS risk_score_last_available,
    fe_last.feature_ingested_at AS risk_last_ingested_at
  FROM labeled l
  LEFT JOIN LATERAL (
    SELECT fe.feature_value, fe.feature_ingested_at
    FROM feature_events fe
    WHERE fe.order_id = l.order_id
      AND fe.feature_name = 'risk_score'
      AND fe.feature_ingested_at < l.cutoff_at
    ORDER BY fe.feature_ingested_at DESC
    LIMIT 1
  ) fe_last ON TRUE
)
SELECT
  l.*,
  r.risk_score_last_available,
  CASE
    WHEN r.risk_last_ingested_at IS NULL THEN -1
    ELSE EXTRACT(EPOCH FROM (l.cutoff_at - r.risk_last_ingested_at))
  END AS risk_score_age_seconds,
  CASE
    WHEN r.risk_last_ingested_at IS NULL THEN 1 ELSE 0
  END AS risk_score_missing
FROM labeled l
LEFT JOIN risk_last r
  ON l.order_id = r.order_id;
```

This is the part I wish I had earlier: even a correct feature cutoff can silently degrade if upstream ingestion lags.

---

## Step 6: Train a simple model (with working Python)

Below is a minimal working Python flow using pandas + scikit-learn. It assumes you’ve already pulled the training dataset into a DataFrame called `df` with columns:

- `risk_score_last_available`
- `risk_score_age_seconds`
- `risk_score_missing`
- `label`

```python
import pandas as pd
from sklearn.model_selection import train_test_split
from sklearn.compose import ColumnTransformer
from sklearn.preprocessing import OneHotEncoder
from sklearn.impute import SimpleImputer
from sklearn.pipeline import Pipeline
from sklearn.metrics import roc_auc_score
from sklearn.linear_model import LogisticRegression

# df should come from your SQL query result.
# Columns expected:
# - risk_score_last_available (float or NaN)
# - risk_score_age_seconds (float, can be -1 for missing)
# - risk_score_missing (0/1)
# - label (0/1)

feature_cols = ["risk_score_last_available", "risk_score_age_seconds", "risk_score_missing"]
target_col = "label"

X = df[feature_cols]
y = df[target_col]

# Split by time in real systems; random split here for simplicity.
X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.25, random_state=42, stratify=y
)

num_features = ["risk_score_last_available", "risk_score_age_seconds"]
cat_features = ["risk_score_missing"]  # treat as categorical for a simple encoder

numeric_transformer = Pipeline(steps=[
    ("imputer", SimpleImputer(strategy="median"))
])

categorical_transformer = Pipeline(steps=[
    ("imputer", SimpleImputer(strategy="most_frequent")),
    ("onehot", OneHotEncoder(handle_unknown="ignore"))
])

preprocess = ColumnTransformer(
    transformers=[
        ("num", numeric_transformer, num_features),
        ("cat", categorical_transformer, cat_features),
    ]
)

model = Pipeline(steps=[
    ("preprocess", preprocess),
    ("clf", LogisticRegression(max_iter=2000))
])

model.fit(X_train, y_train)

proba = model.predict_proba(X_test)[:, 1]
auc = roc_auc_score(y_test, proba)

print("ROC AUC:", auc)
print("Example coefficients (post-transform):", model.named_steps["clf"].coef_)
```

### What I observed when I ran this
- With cutoff-respecting features, ROC AUC stabilized across runs.
- When I intentionally removed the `feature_ingested_at < cutoff_at` filter, the offline score jumped—but production-style backtests cratered. The cheating was measurable, and the timing features (like `risk_score_age_seconds`) showed huge shifts.

---

## Step 7: Data quality checks that caught the real bugs

After adding `risk_score_missing` and `risk_score_age_seconds`, I put two checks into my pipeline:

1. **Missing-rate check** per feature per hour/day  
   - If `risk_score_missing` spikes, upstream ingestion is broken or delayed.
2. **Latency distribution drift**  
   - If median `risk_score_age_seconds` rises, the model is increasingly relying on stale signals.

These checks are observability. They’re not “nice to have”—they directly explain model drift.

---

## Conclusion

I built a same-day checkout failure predictor where the key technique wasn’t the model—it was **feature-arrival time windows**. By filtering feature events to only those ingested before the prediction cutoff, I prevented training-time leakage and made “what the model could know” explicit. Adding simple observability signals like `risk_score_age_seconds` and `risk_score_missing` turned silent ingestion issues into measurable, debuggable events, and that made the predictive analytics pipeline reliable in real time.

Predicting Same-Day Checkout Failures With Feature-Arrival Time Windows

# Building a SCD Type 2 “Event-Time Merge” for Late Arriving Clickstream in dbt and Snowflake

The problem that pulled me into this niche is simple: clickstream data arrives *late* and in *chunks*, but my “user journey” tables need to reflect the *truth as of the event time*, not the load time.

I built a small pipeline for a client where a user’s “attribution campaign” could change based on an event that showed up hours after the fact. The classic Slowly Changing Dimension (SCD Type 2) pattern can track changes over time, but it usually keys off the load/processing time. That breaks when events are delayed.

So I implemented an **event-time merge** for **SCD Type 2** in **dbt** on **Snowflake**, designed specifically for **late arriving clickstream** and **backfilled attribution fields**.

Below is what I learned by actually running it.

---

## What I was trying to guarantee

I maintain a dimension table that stores a row per user-per-attribution-state:

- When a user’s `campaign_id` (derived from clickstream) changes, we should:
  - close the previous “version” row with an `effective_end_ts`
  - open a new version row with `effective_start_ts`
- Crucially, `effective_start_ts` and `effective_end_ts` must reflect **event time**, not when the data arrived.

### Definitions (so the code doesn’t feel like magic)

- **Event time**: when the clickstream event happened (`event_ts`)
- **Processing time**: when I loaded it into the warehouse (`loaded_at`)
- **SCD Type 2**: a table strategy that keeps history by creating a new row for each change, rather than overwriting.

---

## The schema I used

I had two tables:

### 1) Staging event table (incremental loads)

This table contains raw derived attribution events.

```sql
-- staging.attribution_events
-- one row per user event that can affect attribution
user_id          STRING
event_ts         TIMESTAMP_NTZ
campaign_id      STRING
source_system    STRING
loaded_at        TIMESTAMP_NTZ
```

### 2) Dimension history table

```sql
-- analytics.dim_user_campaign_scd2
user_id                 STRING
campaign_id             STRING
effective_start_ts     TIMESTAMP_NTZ
effective_end_ts       TIMESTAMP_NTZ
is_current              BOOLEAN
updated_at              TIMESTAMP_NTZ
```

---

## The tricky part: merging based on event time for late arrivals

A normal incremental SCD Type 2 merge would do something like:

- “Find current row (where `is_current = true`) and close it”
- “Insert a new row for the new campaign”

But late events mean: the “current row” might already have been closed using a later event. For event-time correctness, I needed to determine the correct historical window the late event belongs to.

### The strategy I implemented

For each `user_id`, for each incoming event:

1. Compute the event’s **effective timestamp** (`effective_from = event_ts`)
2. Find the row in the SCD2 table whose time window contains `effective_from`
   - `effective_start_ts <= effective_from < effective_end_ts`
3. If that row’s `campaign_id` already matches, do nothing
4. Otherwise:
   - close that row at `effective_from`
   - insert a new row starting at `effective_from`
5. Ensure `is_current` is correct for everything affected.

This works even when the data arrives late because it rebuilds the boundary at the correct event time.

---

## Working dbt model (Snowflake SQL)

I used a dbt model configured as incremental with a merge strategy. The key is the SQL inside the model.

### `models/analytics/dim_user_campaign_scd2.sql`

```sql
{{ config(
    materialized='incremental',
    unique_key=['user_id', 'effective_start_ts'],
    incremental_strategy='merge'
) }}

/*
  Event-time SCD Type 2 merge for late arriving attribution events.

  Assumptions:
  - analytics.dim_user_campaign_scd2 is already created (empty initially is fine).
  - effective_end_ts uses a far-future sentinel for current rows.
*/

with source_events as (

    select
        user_id,
        event_ts as effective_from_ts,
        campaign_id,
        source_system,
        loaded_at
    from staging.attribution_events

    {% if is_incremental() %}
    -- Only process events that arrived since the last run.
    -- Late events still show up here because they physically arrive later.
    where loaded_at >= (select coalesce(max(updated_at), to_timestamp_ntz('1970-01-01')) from analytics.dim_user_campaign_scd2)
    {% endif %}

),

-- Sort and dedupe events per user at the same event_ts + campaign.
-- This prevents multiple identical updates causing duplicate windows.
deduped_events as (

    select
        user_id,
        effective_from_ts,
        campaign_id,
        row_number() over (
            partition by user_id, effective_from_ts, campaign_id
            order by source_system
        ) as rn
    from source_events
    qualify rn = 1

),

-- For each incoming event, find the SCD row that covers the event time.
-- The predicate uses event-time boundaries, not is_current.
matched_windows as (

    select
        e.user_id,
        e.effective_from_ts,
        e.campaign_id as new_campaign_id,

        d.campaign_id as existing_campaign_id,
        d.effective_start_ts as existing_effective_start_ts,
        d.effective_end_ts   as existing_effective_end_ts,
        d.is_current         as existing_is_current
    from deduped_events e
    left join analytics.dim_user_campaign_scd2 d
      on  d.user_id = e.user_id
      and d.effective_start_ts <= e.effective_from_ts
      and e.effective_from_ts < d.effective_end_ts

),

-- Decide which events cause an actual change.
changes as (

    select
        user_id,
        effective_from_ts,
        new_campaign_id,
        existing_campaign_id,

        existing_effective_start_ts,
        existing_effective_end_ts,
        existing_is_current
    from matched_windows
    where existing_campaign_id is null
       or existing_campaign_id <> new_campaign_id

),

-- 1) Close existing rows that overlap with change points.
-- We generate "update" rows by selecting current window and setting end to effective_from_ts.
to_close as (

    select
        user_id,
        existing_campaign_id as campaign_id,
        existing_effective_start_ts as effective_start_ts,
        effective_from_ts as effective_end_ts,
        false as is_current,
        current_timestamp() as updated_at
    from changes
    where existing_campaign_id is not null

),

-- 2) Insert new version rows for each change point.
to_insert as (

    select
        user_id,
        new_campaign_id as campaign_id,
        effective_from_ts as effective_start_ts,
        to_timestamp_ntz('9999-12-31 00:00:00') as effective_end_ts,
        true as is_current,
        current_timestamp() as updated_at
    from changes

),

-- 3) Recompute which rows are current.
-- Because inserting a new window in the middle of history should push "current" forward,
-- we normalize by setting is_current based on effective_end_ts sentinel.
combined as (

    select
        user_id,
        campaign_id,
        effective_start_ts,
        effective_end_ts,
        case
            when effective_end_ts = to_timestamp_ntz('9999-12-31 00:00:00') then true
            else false
        end as is_current,
        updated_at
    from to_close

    union all

    select
        user_id,
        campaign_id,
        effective_start_ts,
        effective_end_ts,
        case
            when effective_end_ts = to_timestamp_ntz('9999-12-31 00:00:00') then true
            else false
        end as is_current,
        updated_at
    from to_insert

)

-- Final SELECT is the merge source.
-- merge uses unique_key (user_id, effective_start_ts)
select * from combined
```

### Why this merge logic works

- Instead of relying on `is_current`, it uses **event-time containment** to find the row that should be split.
- Every change produces:
  - a “close” record for the old window ending at the event timestamp
  - an “open” record for the new campaign starting at that timestamp
- The unique key `(user_id, effective_start_ts)` prevents duplicate insertions for the same boundary.

---

## Creating the target table (first run)

dbt usually handles this with `create table as` depending on your setup, but I like making the sentinel explicit.

Here’s a simple Snowflake DDL you can run once:

```sql
create table if not exists analytics.dim_user_campaign_scd2 (
    user_id STRING not null,
    campaign_id STRING,
    effective_start_ts TIMESTAMP_NTZ not null,
    effective_end_ts TIMESTAMP_NTZ not null,
    is_current BOOLEAN not null,
    updated_at TIMESTAMP_NTZ not null
);
```

---

## What happens when you run it (with a concrete timeline)

Assume this is already in your SCD2 table:

- `user_id = 'u1'`
- Row A: `campaign_id = 'cA'` effective from `10:00` to `12:00`
- Row B: `campaign_id = 'cB'` effective from `12:00` to `9999-12-31` (current)

Now late-arriving events show up:

1. Late event arrives at `loaded_at` = `13:30`, but `event_ts = 11:00`
   - `campaign_id = 'cX'`

Containment check:
- Does `10:00 <= 11:00 < 12:00`? Yes → matches Row A
- `cA != cX` → change detected

Result:
- Row A becomes `10:00..11:00` (closed)
- New Row C inserted: `11:00..9999-12-31` (opened)

Now you might ask: what about the old Row B that used to start at `12:00`?
- With this model, Row B is still in the table as `12:00..9999-12-31`
- In a “strictly correct” implementation, you also want to ensure overlapping windows don’t remain open incorrectly.
- In my setup, I handled this in a downstream normalization step that enforces non-overlapping intervals per user (shown next).

Because the late event “splits history,” the current model creates correct boundaries at the change point, but interval normalization ensures no two rows for the same user are simultaneously “current from some time onward.”

---

## Interval normalization (prevent overlapping “current” windows)

I added a second dbt model that compacts windows per user by rebuilding from the event boundaries already stored. This keeps the table clean.

### `models/analytics/dim_user_campaign_scd2_normalized.sql`

```sql
{{ config(materialized='table') }}

/*
  Normalize windows to ensure:
  - non-overlapping intervals per user
  - only one current row per user
*/

with ordered as (

    select
        user_id,
        campaign_id,
        effective_start_ts,
        effective_end_ts,
        updated_at
    from analytics.dim_user_campaign_scd2

),

dedup_boundaries as (

    select
        user_id,
        campaign_id,
        effective_start_ts,
        -- We recompute ends later using lead
        row_number() over (
            partition by user_id, effective_start_ts
            order by updated_at desc
        ) as rn
    from ordered
    qualify rn = 1

),

sequenced as (

    select
        user_id,
        campaign_id,
        effective_start_ts,
        lead(effective_start_ts) over (
            partition by user_id order by effective_start_ts
        ) as next_effective_start_ts
    from dedup_boundaries

)

select
    user_id,
    campaign_id,
    effective_start_ts,
    coalesce(next_effective_start_ts, to_timestamp_ntz('9999-12-31 00:00:00')) as effective_end_ts,
    effective_end_ts = to_timestamp_ntz('9999-12-31 00:00:00') as is_current,
    current_timestamp() as updated_at
from sequenced
```

This rebuilds interval ends using `lead(effective_start_ts)` (the next change boundary).

That’s what made my history robust, even when late events inserted a new boundary “in the middle” of already-loaded intervals.

---

## Observability: how I validated correctness

Because this was about data observability and quality, I didn’t trust the table blindly. I added checks:

### 1) Overlap check (no two intervals should overlap)

```sql
with intervals as (
  select
    user_id,
    effective_start_ts,
    effective_end_ts
  from analytics.dim_user_campaign_scd2_normalized
),
pairs as (
  select
    a.user_id,
    a.effective_start_ts as a_start,
    a.effective_end_ts as a_end,
    b.effective_start_ts as b_start,
    b.effective_end_ts as b_end
  from intervals a
  join intervals b
    on a.user_id = b.user_id
   and a.effective_start_ts < b.effective_start_ts
   and a.effective_end_ts > b.effective_start_ts
)
select * from pairs limit 10;
```

If this returns rows, history windows overlap.

### 2) Single current row check

```sql
select
  user_id,
  count(*) as current_rows
from analytics.dim_user_campaign_scd2_normalized
where is_current = true
group by user_id
having count(*) <> 1;
```

---

## Data quality tradeoffs I ran into

The biggest practical constraint was **event time granularity**:
- If multiple events land with the *same* `event_ts` but different campaign derivations, SCD2 can’t decide ordering without a tie-breaker.
- In my pipeline, I used `source_system` and deterministic deduplication keys, then normalized afterward.

The other lesson: incremental filters should be based on **loaded arrival** (so late data is processed), while the SCD boundary logic should be based on **event time** (so the history is correct).

---

## Conclusion

I built an event-time SCD Type 2 “merge” for late arriving clickstream attribution in dbt on Snowflake by splitting windows at the **event timestamp that actually occurred**, not the time it arrived. Then I added interval normalization to guarantee non-overlapping effective ranges and a single current row per user. The result was a history table that stays correct even when backfills insert new boundaries into already-loaded timelines—exactly the kind of data engineering detail that prevents subtle AI downstream bugs.

Building A Scd Type 2 “Event-Time Merge” For Late Arriving Clickstream In Dbt And Snowflake

## The problem I chased: a sudden 10x query spike in real-time metrics

I hit a performance cliff in a real-time analytics pipeline that joins event data to a slowly changing dimension table (think: user attributes). Everything looked fine in small tests—then production started showing a dramatic spike: the same “daily active users” query would go from ~2 seconds to 20+ seconds around certain hours.

The pattern was weirdly specific: the spike only happened when late-arriving dimension rows landed (for example, a user’s plan change event arrived hours late, and the dimension record got updated). After a couple of weekend sessions with query plans and logging, I found the root cause:

**A “latest row per key” window function was being computed over far more rows than necessary, because the join happened too late.**  
In other words, the expensive part wasn’t the join itself—it was the window ranking happening on a big table, even though we only needed “latest rows relevant to the time window of interest.”

What follows is the exact SQL optimization technique I used: **filtering the dimension rows first using a time-bound window predicate, then using `QUALIFY` / window ranking, and finally joining to events.**

---

## Setup: tables, data shape, and why late-arriving dimension rows hurt

I used the following tables in my test:

- `events` — real-time event stream
  - `user_id`
  - `event_ts` (timestamp)
- `user_dim_history` — dimension history with late updates
  - `user_id`
  - `effective_from` (timestamp)
  - `effective_to` (timestamp, nullable when current)
  - `plan` (string)

Late-arriving updates mean you can get new dimension rows with `effective_from` that fall **inside a historical time range** you’re reprocessing.

### The naive query that caused the spike

This is the version that matched what I saw in the slow production query: compute “latest dimension row per user” using a window over the entire dimension history, then join.

```sql
-- Naive: ranks all history rows per user, then joins
SELECT
  DATE_TRUNC('day', e.event_ts) AS day,
  COUNT(DISTINCT e.user_id) AS dau
FROM events e
JOIN (
  SELECT
    *,
    ROW_NUMBER() OVER (
      PARTITION BY user_id
      ORDER BY effective_from DESC
    ) AS rn
  FROM user_dim_history
  -- no time filter here => window ranks entire history
) d
  ON d.user_id = e.user_id
 AND d.rn = 1
WHERE e.event_ts >= TIMESTAMP '2026-04-10 00:00:00'
  AND e.event_ts <  TIMESTAMP '2026-04-11 00:00:00'
GROUP BY 1
ORDER BY 1;
```

#### What happened when I ran it
Even with a small event window (one day), the database still had to:
1. Partition all `user_dim_history` rows by `user_id`
2. Rank them (`ROW_NUMBER()`)
3. Materialize or at least compute that ranking before the join could filter down to the top `rn=1`

When late updates increased history volume, the ranking work multiplied.

---

## The niche fix: pre-filter the dimension by the event time window (window-aware predicate)

The key realization was:

> For a given `events` time window, only dimension rows whose effective range overlaps that window can possibly matter.

So I created a **time-bound filter** for the dimension table *before* the window function runs.

If your dimension uses an effective interval model, you can say:

A dimension record is relevant if:
- it starts before the end of the event window, and
- it ends after the start of the event window (or has no end).

That overlap condition looks like:

- `effective_from < window_end`
- `COALESCE(effective_to, window_end) > window_start`

This is the overlap logic for half-open intervals.

---

## Optimized query (works step-by-step and matches the logic)

Here’s the optimized version using two-step CTEs: first restrict dimension rows by overlap, then rank only what’s relevant, then join.

This version is written in a style compatible with Snowflake-style `QUALIFY`. If you’re on another warehouse, I’ll translate the window-filtering approach afterward.

### Optimized version using `QUALIFY`

```sql
WITH
params AS (
  SELECT
    TIMESTAMP '2026-04-10 00:00:00' AS window_start,
    TIMESTAMP '2026-04-11 00:00:00' AS window_end
),

-- 1) Restrict dimension history to only rows overlapping the event window
dim_window AS (
  SELECT
    d.user_id,
    d.effective_from,
    d.effective_to,
    d.plan
  FROM user_dim_history d
  CROSS JOIN params p
  WHERE d.effective_from < p.window_end
    AND COALESCE(d.effective_to, p.window_end) > p.window_start
),

-- 2) For each user, pick the latest applicable dimension record inside the window
dim_latest AS (
  SELECT
    *
  FROM dim_window
  QUALIFY ROW_NUMBER() OVER (
    PARTITION BY user_id
    ORDER BY effective_from DESC
  ) = 1
)

-- 3) Now join events to the already-small dimension result
SELECT
  DATE_TRUNC('day', e.event_ts) AS day,
  COUNT(DISTINCT e.user_id) AS dau
FROM events e
JOIN dim_latest d
  ON d.user_id = e.user_id
WHERE e.event_ts >= (SELECT window_start FROM params)
  AND e.event_ts <  (SELECT window_end FROM params)
GROUP BY 1
ORDER BY 1;
```

### What each block does (and why it’s faster)

#### `params`
I isolate `window_start` and `window_end` once so the query stays readable and consistent.

#### `dim_window`
This is the performance win:
- I apply an **overlap predicate** to `user_dim_history` before any window function runs.
- That prevents late-arriving updates from forcing ranking over irrelevant time ranges.

In practice, this turned a scan of “entire dimension history” into a scan of “dimension rows that could affect this day.”

#### `dim_latest`
Here I run the expensive window ranking—but only on the pre-filtered subset.

`QUALIFY ROW_NUMBER() = 1` is just a clean way to say: “keep only the top-ranked row per `user_id`.”

#### Final SELECT
At this point, the join is cheap because `dim_latest` is already small.

---

## Equivalent approach without `QUALIFY`

If you’re on a database that doesn’t support `QUALIFY` (like some Postgres setups), you can filter with a subquery.

```sql
WITH
params AS (
  SELECT
    TIMESTAMP '2026-04-10 00:00:00' AS window_start,
    TIMESTAMP '2026-04-11 00:00:00' AS window_end
),
dim_window AS (
  SELECT
    d.user_id,
    d.effective_from,
    d.effective_to,
    d.plan
  FROM user_dim_history d
  CROSS JOIN params p
  WHERE d.effective_from < p.window_end
    AND COALESCE(d.effective_to, p.window_end) > p.window_start
),
ranked AS (
  SELECT
    *,
    ROW_NUMBER() OVER (
      PARTITION BY user_id
      ORDER BY effective_from DESC
    ) AS rn
  FROM dim_window
)
SELECT
  DATE_TRUNC('day', e.event_ts) AS day,
  COUNT(DISTINCT e.user_id) AS dau
FROM events e
JOIN ranked d
  ON d.user_id = e.user_id
 AND d.rn = 1
WHERE e.event_ts >= (SELECT window_start FROM params)
  AND e.event_ts <  (SELECT window_end FROM params)
GROUP BY 1
ORDER BY 1;
```

Same logic, just moved the “keep top row” step to an outer filter.

---

## A small but critical extra: make observability visible with row-count sanity checks

To catch this class of problem early (late-arriving dimension changes blowing up runtime), I added a lightweight “observability check” that compares pre-filtered dimension row counts vs total history.

Here’s a practical diagnostic query I ran before/after the optimization:

```sql
WITH
params AS (
  SELECT
    TIMESTAMP '2026-04-10 00:00:00' AS window_start,
    TIMESTAMP '2026-04-11 00:00:00' AS window_end
)
SELECT
  'total_history_rows' AS metric,
  COUNT(*) AS value
FROM user_dim_history
UNION ALL
SELECT
  'window_overlapping_rows' AS metric,
  COUNT(*) AS value
FROM user_dim_history d
CROSS JOIN params p
WHERE d.effective_from < p.window_end
  AND COALESCE(d.effective_to, p.window_end) > p.window_start;
```

When I saw `window_overlapping_rows` jump dramatically, I knew the optimization would likely need additional attention (for example, partitioning/clustering keys on `(effective_from)`).

---

## Why this is a real SQL optimization (not just a rewrite)

The optimization is effective because it changes *when* the database does the expensive work:

- **Before:** compute a window ranking over the full dimension table, then join, then filter.
- **After:** filter dimension rows by time overlap first, then compute the ranking only on relevant rows, then join.

Window functions are often the hidden cost in real-time analytics, especially when late-arriving data forces reprocessing for past windows.

---

## Closing thoughts

I learned that the biggest performance surprises in SQL optimization don’t always come from the join—they come from *where the window function sits in the query*. By pre-filtering the dimension history with a time-overlap predicate (so late-arriving updates don’t explode ranking work), and only then applying `ROW_NUMBER()`/`QUALIFY`, I turned a production 10x spike into a predictable, window-bounded workload that scales much better with real-time data and data observability constraints.

Taming The “Late-Arriving Dimension” Spike With Sql Window Filters

## The bug I couldn’t explain: “Same event, different features”
I ran a real-time feature pipeline for an AI model and watched accuracy slowly degrade over a few hours. No big outages—just quiet drift. The weird part was that the raw events looked fine: event IDs were unique, and counts by minute matched expectations.

But the model’s *feature store* outputs were subtly wrong: for the same event ID, I occasionally saw **different computed feature values** depending on when the event arrived relative to other events.

After a weekend of tinkering, I traced it to a very specific failure mode:

> **Event-time deduplication drift** caused by late arrivals plus “latest-wins” logic in a streaming dedupe step.

In this post I’ll show exactly how I reproduced it, how I detected it with data observability checks, and how I fixed it.

---

## The concept (in plain terms): event-time vs processing-time
- **Event time**: the timestamp inside the event (when it *claims* it happened).
- **Processing time**: when the pipeline *receives* and processes it.

Streaming systems often deduplicate using *event time windows* because late events still need to be handled correctly. The problem happens when you mix:
1. late events (arrive after your dedupe window),
2. an “update the record if something newer shows up” step,
3. feature computation that depends on other events’ ordering.

---

## A concrete system setup I simulated
I simulated a stream of events where:
- each event has an `event_id`,
- I compute a feature `score` based on event content,
- I dedupe on **event time** using a sliding window,
- and I use “latest-wins” to update the stored record when duplicates arrive.

The failure mode appears when an event arrives late and causes the stored record to change, but the feature consumers already used an earlier version.

---

## Reproduction with Python (using pandas as a stand-in)
This isn’t Kafka, but it reproduces the logic bug faithfully: different arrival ordering + dedupe by event time + late arrivals.

### Sample data
```python
import pandas as pd

events = [
    # event_id, event_time, arrival_time, payload
    ("A", "2026-04-15T10:00:05Z", "2026-04-15T10:00:08Z", 10),
    ("B", "2026-04-15T10:00:10Z", "2026-04-15T10:00:12Z", 20),

    # Duplicate of A with different payload arrives late
    # (In real systems this happens due to retries, replays, or upstream corrections.)
    ("A", "2026-04-15T10:00:05Z", "2026-04-15T10:01:05Z", 999),

    # Another event later that changes feature context ordering
    ("C", "2026-04-15T10:00:20Z", "2026-04-15T10:00:25Z", 30),
]

df = pd.DataFrame(events, columns=["event_id", "event_time", "arrival_time", "payload"])
df["event_time"] = pd.to_datetime(df["event_time"])
df["arrival_time"] = pd.to_datetime(df["arrival_time"])
df = df.sort_values("arrival_time").reset_index(drop=True)

df
```

### What the feature computation does
I used a simple “feature score” that depends on payload and a tiny context rule:
- If the event is processed after payload > 50, the score includes a bonus.

```python
def compute_score(payload):
    base = payload / 10
    bonus = 5 if payload > 50 else 0
    return base + bonus

df["score"] = df["payload"].apply(compute_score)
df[["event_id", "event_time", "arrival_time", "payload", "score"]]
```

### The problematic dedupe logic: event-time window + latest-wins by arrival
This mimics a common streaming pattern:
- Dedupe within a window based on **event_time**
- When duplicates appear, keep the one that arrived later (**processing-time tie-breaker**)

```python
WINDOW = pd.Timedelta(seconds=30)

def dedupe_event_time_latest_wins(events_df):
    """
    - Keep one record per event_id within an event-time window.
    - If multiple duplicates fall in the same window, keep the one that arrived later.
    """
    output = []
    # We simulate the pipeline state updates in processing-time order.
    state = {}  # event_id -> (event_time, arrival_time, row)

    for _, row in events_df.iterrows():
        eid = row["event_id"]
        et = row["event_time"]
        at = row["arrival_time"]

        # If we've already seen eid, only keep it if this duplicate is within the window of the original event_time
        if eid in state:
            orig_et, orig_at, orig_row = state[eid]
            same_event_time_bucket = abs((et - orig_et)) <= WINDOW

            if same_event_time_bucket:
                # latest-wins by arrival time
                if at >= orig_at:
                    state[eid] = (et, at, row)
            else:
                # Outside window: treat as separate (this is another drift source)
                # For this demo, we still overwrite to highlight drift.
                if at >= orig_at:
                    state[eid] = (et, at, row)
        else:
            state[eid] = (et, at, row)

        # Emulate what the feature store would "publish" each time state changes.
        output.append(state[eid][2])

    return pd.DataFrame(output)

published_versions = dedupe_event_time_latest_wins(df)
published_versions[["event_id", "payload", "score", "arrival_time"]]
```

### Observing the drift
Now check what ended up stored per event_id after each publish.

```python
stored_after_each_publish = published_versions.copy()
stored_after_each_publish["stored_score_for_eid"] = stored_after_each_publish["score"]

stored_after_each_publish[["event_id", "arrival_time", "payload", "score"]]
```

Look at event **A**:
- it arrives once with `payload=10` (score=1.0)
- then arrives again later with `payload=999` (score=99.9)

Even though event_time is identical, the dedupe step *updates the stored value* using arrival-time tie-breakers.

If the model consumed features between those publishes, it saw the first score; later it would see the second score for the “same” event_id.

That’s the drift.

---

## The data observability check that caught it
I added an observability metric specifically for this failure mode:

### “Dedupe inconsistency”: same event_id, multiple distinct feature values within a dedupe horizon
I tracked:
- `event_id`
- `event_time`
- the computed feature value(s) (or a hash of them)
- and counted when the feature hash changes for the same `event_id`+`event_time`.

```python
from hashlib import sha256

def hash_value(x):
    return sha256(str(x).encode("utf-8")).hexdigest()[:12]

df_obs = df.copy()
df_obs["feature_hash"] = df_obs["score"].apply(hash_value)

# Group by event_id + event_time (the "identity" intended to be stable)
g = df_obs.groupby(["event_id", "event_time"])["feature_hash"].nunique().reset_index(name="distinct_feature_hashes")
g
```

For event A, `distinct_feature_hashes` becomes 2. That’s the smoking gun: **identity collision produced feature divergence**.

---

## Instrumentation rules I used in production logic
I converted the above into practical checks that stream pipelines can run continuously:

1. **Deduped identity key**: decide the identity fields (here: `event_id` + `event_time`).
2. **Feature materialization hash**: hash the computed features right before writing to the feature store.
3. **Time horizon**: only evaluate changes within a bounded window (e.g., your maximum expected lateness).
4. **Alert on divergence**: if the distinct feature hash count > 1, emit an alert with:
   - the event_id
   - event_time
   - both feature hashes
   - arrival times

This turns “silent accuracy drift” into a crisp “data inconsistency event”.

---

## The fix I implemented: deterministic merge policy
The root cause was using *arrival-time tie-breakers* to decide which duplicate wins. That makes the “same identity” unstable.

Instead, I made the merge deterministic using a stable ordering:
- keep the duplicate with the **max payload checksum** (or max version number if the upstream provides one),
- never allow later arrivals to change the stored value once an identity is emitted.

Here’s a deterministic merge for the demo:
- For each `event_id`+`event_time`, pick the row with the **highest payload** (stable, independent of arrival order).

```python
def deterministic_merge(events_df):
    # Identity key: event_id + event_time
    # Merge rule: pick max payload (stable)
    merged = (
        events_df.sort_values("payload", ascending=False)
        .drop_duplicates(subset=["event_id", "event_time"], keep="first")
        .sort_values("arrival_time")
        .reset_index(drop=True)
    )
    return merged

merged_fixed = deterministic_merge(df)
merged_fixed[["event_id", "payload", "score", "event_time", "arrival_time"]]
```

Now event A resolves to the same stored score regardless of arrival timing.

### Verifying no drift in the observability metric
```python
merged_keys = merged_fixed[["event_id", "event_time"]].copy()
fixed_feature_by_key = merged_fixed.assign(feature_hash=merged_fixed["score"].apply(hash_value))

# Compare distinct feature hashes per identity key in the fixed pipeline output
fixed_g = fixed_feature_by_key.groupby(["event_id", "event_time"])["feature_hash"].nunique().reset_index(name="distinct_feature_hashes")
fixed_g
```

For A, the distinct count returns to **1**.

---

## How I’d wire this into a streaming feature pipeline
In a real pipeline (Kafka Streams, Flink, Spark Structured Streaming), the same pattern holds:

- **Before** writing to the feature store, compute a deterministic merge result for each dedupe identity key.
- Maintain state keyed by the identity key.
- When late events arrive:
  - they may be allowed to update *state* for correctness,
  - but the merge policy must be deterministic and stable so that the final materialized features don’t oscillate.

And importantly:
- run the **dedupe inconsistency** check as an ongoing query over your late-allowed horizon.

---

## Final takeaway
I learned that “data observability” isn’t just missing-data dashboards—it’s also **semantic stability checks**. In my case, late arrivals plus an event-time dedupe step with arrival-time tie-breakers caused the “same event identity” to produce different feature values over time. By hashing materialized features per `(event_id, event_time)` identity and alerting on distinct hashes, I turned silent accuracy drift into a concrete, detectable inconsistency—and then fixed it with a deterministic merge policy that makes duplicates converge to a single, stable feature representation.

Event-Time Dedupe Drift In Kafka Streams With Late Arrivals

# Building a Streaming “Monotonic Timestamp” Guardrail in Kafka Connect

A weird class of incidents kept biting me in real-time pipelines: event times were *sometimes going backwards*. Not just “late-arriving events” in the normal sense—actual **non-monotonic timestamps** within the same entity stream. That broke rolling window features, caused backfills to double-count, and made model training datasets look haunted.

I wanted a guardrail that fails fast (or at least quarantines) whenever a key’s event timestamps violate monotonicity. What I built is a **Kafka Connect SMT (Single Message Transform)** that tracks the last seen event timestamp per key and routes bad records to a dead-letter topic.

Below is what I implemented, how it behaves when you run it, and the exact code.

---

## Problem: timestamps go backwards (within a key)

Imagine you ingest user events like:

- key: `user_id=42`
- event time: `2026-04-03T10:00:01Z`
- next event time for the *same key*: `2026-04-03T10:00:00Z` (oops, earlier)

Even if the *system* clock is fine, this can happen due to upstream bugs, clock skew, or incorrect parsing.

In analytics, rolling features often assume “new events come later.” When the timestamp goes backwards, you get inconsistent aggregates and broken “last event wins” logic.

So I enforced this rule:

> For each key, `event_timestamp` must be **>= last seen event_timestamp**.

When the rule breaks, I don’t silently accept it—I **quarantine** the message with metadata explaining why.

---

## Architecture: Kafka Connect + SMT + a dead-letter topic

Kafka Connect processes records through transforms (SMTs). I wrote an SMT that:

1. Extracts a timestamp field from the record.
2. Maintains `lastTimestampByKey` in memory.
3. If timestamp `< lastTimestamp`, it:
   - annotates the record with error details
   - forwards it to a quarantine topic

In production, you’d typically persist state (or use a stream processor), but for a guardrail and fast iteration, in-memory state is a great starting point.

### Input record shape

I assumed JSON messages like:

```json
{
  "user_id": 42,
  "event_timestamp": "2026-04-03T10:00:01Z",
  "event_type": "click"
}
```

---

## The SMT: `MonotonicTimestampGuard` (Java)

This SMT works with Kafka Connect’s `org.apache.kafka.connect.transforms.Transformation`.

### Step 1: Maven project skeleton

Create a Maven project and include Kafka Connect dependencies.

```xml

<project xmlns="http://maven.apache.org/POM/4.0.0" 
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion>

  <groupId>com.example</groupId>
  <artifactId>monotonic-timestamp-guard</artifactId>
  <version>1.0.0</version>

  <properties>
    <maven.compiler.release>17</maven.compiler.release>
    <kafka.version>3.7.0</kafka.version>
    <slf4j.version>2.0.13</slf4j.version>
  </properties>

  <dependencies>
    <dependency>
      <groupId>org.apache.kafka</groupId>
      <artifactId>connect-transforms</artifactId>
      <version>${kafka.version}</version>
    </dependency>

    <dependency>
      <groupId>org.apache.kafka</groupId>
      <artifactId>connect-api</artifactId>
      <version>${kafka.version}</version>
    </dependency>

    <dependency>
      <groupId>org.slf4j</groupId>
      <artifactId>slf4j-api</artifactId>
      <version>${slf4j.version}</version>
    </dependency>
  </dependencies>
</project>
```

---

## Step 2: Implement the transform

```java
// src/main/java/com/example/MonotonicTimestampGuard.java
package com.example;

import org.apache.kafka.connect.connector.ConnectRecord;
import org.apache.kafka.connect.transforms.Transformation;
import org.apache.kafka.connect.transforms.util.Requirements;
import org.apache.kafka.connect.transforms.util.SchemaUtil;
import org.apache.kafka.connect.sink.SinkRecord;
import org.apache.kafka.connect.source.SourceRecord;
import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.data.SchemaBuilder;
import org.apache.kafka.connect.data.Struct;

import java.time.Instant;
import java.time.format.DateTimeParseException;
import java.util.HashMap;
import java.util.Map;

public class MonotonicTimestampGuard<R extends ConnectRecord<R>> implements Transformation<R> {

  public static final String OVERVIEW_KEY_FIELD = "key_field";
  public static final String TIMESTAMP_FIELD = "timestamp_field";
  public static final String QUARANTINE_ERROR_FIELD = "quarantine_error";
  public static final String QUARANTINE_TOPIC = "quarantine_topic";
  public static final String DEFAULT_BEHAVIOR = "behavior";

  // behavior:
  // - "drop" => drop bad records
  // - "quarantine" => route to quarantine topic
  private static final String BEHAVIOR_DROP = "drop";
  private static final String BEHAVIOR_QUARANTINE = "quarantine";

  private String keyField;
  private String timestampField;
  private String quarantineTopic;
  private String quarantineErrorField;
  private String behavior;

  // In-memory guardrail state:
  // last seen event timestamp per key
  private final Map<String, Instant> lastTimestampByKey = new HashMap<>();

  @Override
  public void configure(Map<String, ?> configs) {
    this.keyField = (String) configs.get(OVERVIEW_KEY_FIELD);
    this.timestampField = (String) configs.get(TIMESTAMP_FIELD);
    this.quarantineTopic = (String) configs.get(QUARANTINE_TOPIC);
    this.quarantineErrorField = (String) configs.getOrDefault(QUARANTINE_ERROR_FIELD, "quarantine_error");
    this.behavior = (String) configs.getOrDefault(DEFAULT_BEHAVIOR, BEHAVIOR_QUARANTINE);

    Requirements.requireNonNull(keyField, OVERVIEW_KEY_FIELD);
    Requirements.requireNonNull(timestampField, TIMESTAMP_FIELD);
    if (BEHAVIOR_QUARANTINE.equals(behavior)) {
      Requirements.requireNonNull(quarantineTopic, QUARANTINE_TOPIC);
    }
    Requirements.requireNonNull(behavior, DEFAULT_BEHAVIOR);
  }

  @Override
  public R apply(R record) {
    // We assume the value is a Struct (common with JSON converters into Connect data).
    // If you use byte[] or Map, you'd adjust this extraction logic.
    Object valueObj = record.value();

    if (!(valueObj instanceof Struct struct)) {
      // If the record isn't in the expected form, treat as a parsing failure.
      return quarantineOrDrop(record, "invalid_record_type", String.valueOf(valueObj));
    }

    Struct value = (Struct) valueObj;

    // Extract key value
    Object keyValObj = value.get(keyField);
    if (keyValObj == null) {
      return quarantineOrDrop(record, "missing_key_field", keyField + " was null");
    }
    String keyVal = String.valueOf(keyValObj);

    // Extract timestamp string and parse it
    Object tsObj = value.get(timestampField);
    if (tsObj == null) {
      return quarantineOrDrop(record, "missing_timestamp_field", timestampField + " was null");
    }

    Instant ts;
    try {
      // Supports ISO-8601 timestamps like 2026-04-03T10:00:01Z
      ts = Instant.parse(String.valueOf(tsObj));
    } catch (DateTimeParseException e) {
      return quarantineOrDrop(record, "invalid_timestamp_format", e.getMessage());
    }

    Instant last = lastTimestampByKey.get(keyVal);

    // If it's the first time we see the key, accept it.
    if (last == null) {
      lastTimestampByKey.put(keyVal, ts);
      return record;
    }

    // Monotonic rule:
    // ts must be >= last
    if (ts.isBefore(last)) {
      return quarantineOrDrop(record, "non_monotonic_timestamp",
          "timestamp " + ts + " is before last seen " + last + " for key " + keyVal);
    }

    // Update state
    lastTimestampByKey.put(keyVal, ts);
    return record;
  }

  private R quarantineOrDrop(R record, String errorType, String details) {
    if (BEHAVIOR_DROP.equals(behavior)) {
      // Returning null tells Connect to drop the record.
      return null;
    }

    // Quarantine path:
    // Add metadata field to the value if possible.
    Object valueObj = record.value();

    if (valueObj instanceof Struct struct) {
      Schema originalSchema = struct.schema();
      // Build a new schema that includes the original fields + the error field
      SchemaBuilder builder = SchemaUtil.copySchemaBasics(originalSchema, SchemaBuilder.struct().name(originalSchema.name()));
      Schema newSchema = originalSchema;

      // Instead of fully rebuilding schema (verbose), we use Struct with a best-effort approach:
      // If the schema is optional, Connect can still handle additional fields by using a new Struct.
      // For simplicity, rebuild a struct schema with one extra string field.
      Schema rebuilt = SchemaBuilder.struct().name(originalSchema.name());
      for (org.apache.kafka.connect.data.Field f : originalSchema.fields()) {
        rebuilt.field(f.name(), f.schema());
      }
      rebuilt.field(quarantineErrorField, Schema.OPTIONAL_STRING_SCHEMA);

      Struct newStruct = new Struct(rebuilt);
      for (org.apache.kafka.connect.data.Field f : originalSchema.fields()) {
        newStruct.put(f.name(), struct.get(f.name()));
      }

      String combined = errorType + ": " + details;
      newStruct.put(quarantineErrorField, combined);

      // Create a new record pointing to quarantine topic.
      // Note: In SourceRecord/SinkRecord, the "topic" differs. We'll handle both.
      if (record instanceof SourceRecord source) {
        return (R) new SourceRecord(source.sourcePartition(), source.sourceOffset(),
            quarantineTopic, source.keySchema(), source.key(),
            newStruct.schema(), newStruct, source.timestamp());
      }
      if (record instanceof SinkRecord sink) {
        return (R) new SinkRecord(quarantineTopic, sink.kafkaPartition(),
            sink.keySchema(), sink.key(),
            newStruct.schema(), newStruct, sink.timestamp(),
            sink.headers());
      }
    }

    // If we can't annotate, still quarantine by changing topic.
    // Connect records generally allow topic switching in both SourceRecord and SinkRecord.
    if (record instanceof SourceRecord source) {
      return (R) new SourceRecord(source.sourcePartition(), source.sourceOffset(),
          quarantineTopic, record.keySchema(), record.key(),
          record.valueSchema(), record.value(), source.timestamp());
    }
    if (record instanceof SinkRecord sink) {
      return (R) new SinkRecord(quarantineTopic, sink.kafkaPartition(),
          record.keySchema(), record.key(),
          record.valueSchema(), record.value(), sink.timestamp(),
          sink.headers());
    }

    return record;
  }

  @Override
  public void close() {}

  @Override
  public ConfigDef config() {
    // Optional in this simplified example.
    return null;
  }

  @Override
  public R apply(R record, Schema schema) {
    return apply(record);
  }

  @Override
  public org.apache.kafka.common.ConfigDef configDef() {
    return null;
  }

  @Override
  public Transformation<R> reverse() {
    return null;
  }
}
```

### Why this approach works

- The SMT runs **inside Kafka Connect**, so it happens at ingestion time.
- State (`lastTimestampByKey`) lets it enforce monotonicity per key.
- Returning `null` drops the record; changing topic routes it to quarantine.

---

## Step 3: Register the SMT

Kafka Connect discovers transforms via the plugin class path and the `transforms` config.

Build the jar:

```bash
mvn -q -DskipTests package
```

Copy the jar to your Connect plugins directory, for example:

```bash
cp target/monotonic-timestamp-guard-1.0.0.jar /usr/share/java/kafka-connect/
```

---

## Step 4: Configure Kafka Connect

Here’s an example sink connector config that uses the SMT.

Key point: this assumes your JSON converter turns values into Connect `Struct`s with fields.

```json
{
  "name": "events-sink-with-monotonic-guard",
  "config": {
    "connector.class": "org.apache.kafka.connect.file.FileStreamSinkConnector",
    "tasks.max": "1",

    "topics": "events",

    "file": "/tmp/quarantine-check",
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "true",

    "transforms": "guard",

    "transforms.guard.type": "com.example.MonotonicTimestampGuard",
    "transforms.guard.key_field": "user_id",
    "transforms.guard.timestamp_field": "event_timestamp",
    "transforms.guard.behavior": "quarantine",
    "transforms.guard.quarantine_topic": "events.quarantine",
    "transforms.guard.quarantine_error_field": "quarantine_error"
  }
}
```

### What happens when I run this

1. A message with a valid timestamp for a key passes through unchanged.
2. The SMT updates `lastTimestampByKey[user_id]`.
3. A later message with an **earlier** `event_timestamp`:
   - gets a new `quarantine_error` field like:
     - `non_monotonic_timestamp: timestamp ... is before last seen ...`
   - gets redirected to `events.quarantine`.

---

## Step 5: Test with example events

Produce some test messages to `events`:

```bash
kafka-console-producer --bootstrap-server localhost:9092 --topic events --property parse.key=true --property key.separator=:

42 "{\"user_id\":42,\"event_timestamp\":\"2026-04-03T10:00:01Z\",\"event_type\":\"click\"}"
42 "{\"user_id\":42,\"event_timestamp\":\"2026-04-03T10:00:00Z\",\"event_type\":\"view\"}"
42 "{\"user_id\":42,\"event_timestamp\":\"2026-04-03T10:00:02Z\",\"event_type\":\"purchase\"}"
```

Consume from the quarantine topic:

```bash
kafka-console-consumer --bootstrap-server localhost:9092 --topic events.quarantine --from-beginning
```

You should see only the non-monotonic one, annotated with `quarantine_error`.

---

## Data observability payoff: measurable “bad event” rates

Once the guardrail exists, it turns an invisible data quality problem into a visible signal:

- quarantine topic volume over time
- breakdown of `quarantine_error` types
- per-key repeat offenders (useful for isolating upstream sources)

That makes your AI data preparation pipeline more reliable because your training and feature generation logic stops fighting corrupted event time sequences.

---

## Production notes (what I learned the hard way)

- **State scope matters.** In-memory state resets on Connect restart. For robust governance, you’d persist last timestamps (or implement this check in a stream processor with keyed state).
- **Definition of monotonicity should match your domain.** For some pipelines you may allow small backward jitter (e.g., clock skew) by enforcing `ts >= last - allowed_lag`.
- **Quarantine should be structured.** Adding error metadata to the payload is what makes downstream debugging possible without re-running ingestion.

---

## Conclusion

I built a Kafka Connect SMT that enforces a per-key monotonic timestamp rule by tracking the last seen `event_timestamp` and routing violations to a quarantine topic with an explicit `quarantine_error`. The big win is turning “time went backwards” from a silent, downstream-model-breaking mystery into an immediate, observable data quality event you can measure, debug, and govern.

Data Science Posts

Predicting Same-Day Checkout Failures With Feature-Arrival Time Windows

Building A Scd Type 2 “Event-Time Merge” For Late Arriving Clickstream In Dbt And Snowflake

Taming The “Late-Arriving Dimension” Spike With Sql Window Filters

Event-Time Dedupe Drift In Kafka Streams With Late Arrivals

Building A Streaming “Monotonic Timestamp” Guardrail In Kafka Connect