Dedup window versus replay

# /// script
# requires-python = ">=3.10"
# ///
# Copyright (c) 2026 Aayush Sharma · SPDX-License-Identifier: MIT
# https://thedataengineering.com/lab-code/LICENSE.txt
"""Dedup window against replay.

A consumer that dedups with event_id -> time and a 60-second TTL, fed 1,000
events and the same 1,000 again 5 minutes later; then a sink that overwrites
the whole partition for the date (idempotency by replacement).
"""
import argparse, json

p = argparse.ArgumentParser(description=__doc__.splitlines()[0])
p.add_argument("--events", type=int, default=1000, help="Events.")
p.add_argument("--ttl", type=float, default=60, help="Dedup TTL in seconds.")
p.add_argument("--replay-after", type=float, default=300, help="Seconds before the replay.")
p.add_argument("--json", help="Also write the results to this JSON file.")
args = p.parse_args()

events = [(f"e{i}", "2025-10-20", i % 7) for i in range(args.events)]
def feed(at):
    return [(at + i * 0.01, e) for i, e in enumerate(events)]

# 1. TTL dedup into an append-only sink
seen, sink = {}, []
for t, (eid, day, v) in feed(0) + feed(args.replay_after):
    if eid in seen and t - seen[eid] < args.ttl: continue
    seen[eid] = t; sink.append((eid, day, v))
twice = len(sink) - len({e for e, _, _ in sink})

# 2. Replace the partition for the date
partitions = {}
for at in (0, args.replay_after):
    batch = [e for _, e in feed(at)]
    partitions["2025-10-20"] = sorted(batch)                     # overwrite, not append
same = len(partitions["2025-10-20"]) == args.events
print(f"TTL dedup ({args.ttl:g} s), replay after {args.replay_after:g} s: {twice:,} of {args.events:,} events applied twice")
print(f"overwrite the date's partition: {len(partitions['2025-10-20']):,} rows after the replay (identical: {same})")

if args.json:
    with open(args.json, "w") as f:
        json.dump({"tables": {"sinks": [{"sink": "append + TTL dedup", "rows_after_replay": len(sink), "applied_twice": twice},
                                        {"sink": "overwrite the partition", "rows_after_replay": len(partitions['2025-10-20']), "applied_twice": 0}]}}, f)