Per-value checks ride on the parse

# /// script
# requires-python = ">=3.10"
# ///
# Copyright (c) 2026 Aayush Sharma · SPDX-License-Identifier: MIT
# https://thedataengineering.com/lab-code/LICENSE.txt
"""Per-value checks ride on the parse.

A CSV of 5 million rows with an integer, a decimal and a date. Timed:
(a) parsing only, (b) parsing plus type, range and not-null checks in the
same pass, (c) parsing, then a separate second pass over the parsed rows.
"""
import argparse, csv, datetime, json, os, random, tempfile, time

p = argparse.ArgumentParser(description=__doc__.splitlines()[0])
p.add_argument("--rows", type=int, default=5_000_000, help="Rows in the CSV.")
p.add_argument("--path", default=os.path.join(tempfile.gettempdir(), "checks-lab.csv"), help="Where the CSV is written.")
p.add_argument("--seed", type=int, default=0, help="Random seed.")
p.add_argument("--json", help="Also write the results to this JSON file.")
args = p.parse_args()
import atexit, shutil
atexit.register(lambda: shutil.rmtree(args.path, ignore_errors=True) if os.path.isdir(args.path) else os.path.exists(args.path) and os.remove(args.path))   # leave nothing behind
random.seed(args.seed)

if not os.path.exists(args.path):
    with open(args.path, "w", newline="") as f:
        w = csv.writer(f); w.writerow(["qty", "price", "day"])
        base = datetime.date(2025, 1, 1)
        for i in range(args.rows):
            w.writerow([random.randint(0, 120), f"{random.uniform(0.5, 120000):.2f}", (base + datetime.timedelta(days=i % 365)).isoformat()])

def ok(q, p, d): return d is not None and 1 <= q <= 100 and 0 < p < 100000

def parse_only():
    n = 0
    with open(args.path) as f:
        for r in csv.DictReader(f):
            q = int(r["qty"]); p = float(r["price"]); d = datetime.date.fromisoformat(r["day"]); n += 1
    return n

def parse_and_check():
    n = bad = 0
    with open(args.path) as f:
        for r in csv.DictReader(f):
            q = int(r["qty"]); p = float(r["price"]); d = datetime.date.fromisoformat(r["day"]); n += 1
            if not ok(q, p, d): bad += 1
    return n, bad

def parse_then_second_pass():
    rows = []
    with open(args.path) as f:
        for r in csv.DictReader(f):
            rows.append((int(r["qty"]), float(r["price"]), datetime.date.fromisoformat(r["day"])))
    t = time.perf_counter()
    bad = sum(1 for q, p, d in rows if not ok(q, p, d))
    return time.perf_counter() - t, bad

t = time.perf_counter(); parse_only(); a = time.perf_counter() - t
t = time.perf_counter(); _, bad = parse_and_check(); b = time.perf_counter() - t
second, _ = parse_then_second_pass()
rows = [{"variant": "(a) parse only", "seconds": a, "vs_parse_pct": 100.0},
        {"variant": "(b) parse + checks in the same pass", "seconds": b, "vs_parse_pct": b / a * 100},
        {"variant": "(c) parse, then a separate checking pass", "seconds": a + second, "vs_parse_pct": (a + second) / a * 100}]
for r in rows: print(f"{r['variant']:<44} {r['seconds']:6.2f} s  ({r['vs_parse_pct']:5.1f}% of parse only)")
print(f"rows failing the checks: {bad:,}; the separate pass alone took {second:.2f} s over rows already in memory")

if args.json:
    with open(args.json, "w") as f:
        json.dump({"tables": {"variants": rows, "summary": [{"inline_extra_pct": (b / a - 1) * 100, "second_pass_s": second,
                   "second_pass_pct": second / a * 100, "bad": bad}]}}, f)