Aliasing in a polling pipeline

# /// script
# requires-python = ">=3.10"
# ///
# Copyright (c) 2026 Aayush Sharma · SPDX-License-Identifier: MIT
# https://thedataengineering.com/lab-code/LICENSE.txt
"""Aliasing in a polling pipeline.

An orders table with updated_at in SQLite; one order's status changes several
times a second while a poller selects updated_at > watermark every few
seconds. Then one row is deleted. Time is simulated, so the run is instant
and exact: each change and each poll carries its own timestamp.
"""
import argparse, json, sqlite3

p = argparse.ArgumentParser(description=__doc__.splitlines()[0])
p.add_argument("--changes-per-second", type=float, default=3.0, help="Status changes per second on the busy order.")
p.add_argument("--seconds", type=int, default=60, help="Length of the run.")
p.add_argument("--poll-every", type=float, default=5.0, help="Seconds between polls.")
p.add_argument("--json", help="Also write the results to this JSON file.")
args = p.parse_args()

STATUSES = ["placed", "paid", "packed", "shipped", "out for delivery", "delivered"]
db = sqlite3.connect(":memory:")
db.execute("CREATE TABLE orders (order_id INTEGER PRIMARY KEY, status TEXT, updated_at REAL)")
db.executemany("INSERT INTO orders VALUES (?, 'placed', 0)", [(1,), (2,)])

# The timeline: status changes on order 1, polls, and one delete of order 2 half-way
events = [(i / args.changes_per_second, "change") for i in range(1, int(args.seconds * args.changes_per_second) + 1)]
events += [(k * args.poll_every, "poll") for k in range(1, int(args.seconds / args.poll_every) + 1)]
events += [(args.seconds / 2 + 0.01, "delete")]
events.sort(key=lambda e: (e[0], e[1] != "change"))       # a change at the same instant lands before the poll

watermark, n_changes, seen, polls = 0.0, 0, [], []
for t, kind in events:
    if kind == "change":
        n_changes += 1
        db.execute("UPDATE orders SET status = ?, updated_at = ? WHERE order_id = 1", (STATUSES[n_changes % len(STATUSES)], t))
    elif kind == "delete":
        db.execute("DELETE FROM orders WHERE order_id = 2")
    else:
        rows = db.execute("SELECT order_id, status, updated_at FROM orders WHERE updated_at > ?", (watermark,)).fetchall()
        for oid, status, ts in rows:
            seen.append({"t": t, "order_id": oid, "status": status})
            watermark = max(watermark, ts)
        polls.append({"poll_at": t, "rows_returned": len(rows)})

observed = sum(1 for s in seen if s["order_id"] == 1)
delete_seen = any(s["order_id"] == 2 for s in seen)
print(f"{n_changes} status changes on order 1 in {args.seconds} s; {len(polls)} polls every {args.poll_every:g} s")
print(f"changes observed by the poller: {observed} of {n_changes} ({observed / n_changes:.0%}); "
      f"distinct statuses seen: {len({s['status'] for s in seen if s['order_id'] == 1})} of {len(STATUSES)}")
print(f"order 2 deleted at {args.seconds / 2:g} s: {'seen' if delete_seen else 'never seen'} by the poller")

if args.json:
    with open(args.json, "w") as f:
        json.dump({"tables": {"polls": polls, "summary": [{
            "changes": n_changes, "observed": observed, "observed_pct": observed / n_changes * 100,
            "polls": len(polls), "delete_seen": "yes" if delete_seen else "no"}]}}, f)