# /// script
# requires-python = ">=3.10"
# dependencies = ["duckdb"]
# ///
# Copyright (c) 2026 Aayush Sharma · SPDX-License-Identifier: MIT
# https://thedataengineering.com/lab-code/LICENSE.txt
"""Find your own break-even point.

A full rebuild of a table against an incremental merge (DELETE the changed
keys, INSERT their new rows) for a growing changed fraction, in DuckDB (a
columnar engine) or SQLite (a row store keyed by id).
"""
import argparse, json, sqlite3, statistics, time
import duckdb

p = argparse.ArgumentParser(description=__doc__.splitlines()[0])
p.add_argument("--rows", type=int, default=5_000_000, help="Rows in the source table.")
p.add_argument("--fractions", type=lambda s: [float(v) for v in s.split(",")], default=[0.001, 0.005, 0.01, 0.05, 0.10, 0.20, 0.40, 0.70],
               help="Changed fractions to merge, comma-separated.")
p.add_argument("--engine", choices=["duckdb", "sqlite"], default="duckdb", help="Which engine holds the tables.")
p.add_argument("--repeats", type=int, default=3, help="Runs per measurement; the median is reported.")
p.add_argument("--json", help="Also write the results to this JSON file.")
args = p.parse_args()

if args.engine == "duckdb":
    con = duckdb.connect()
    con.execute(f"CREATE TABLE src AS SELECT i AS id, random() AS v FROM range({args.rows}) t(i)")
    REBUILD = "CREATE TABLE t2 AS SELECT * FROM src"
    COPY = "CREATE OR REPLACE TABLE t2 AS SELECT * FROM src"
    DELTA = "CREATE OR REPLACE TABLE delta AS SELECT id, random() AS v FROM src WHERE random() < {frac}"
    run = con.execute
else:                                                  # a row store: the target is keyed, as a merge needs
    con = sqlite3.connect(":memory:", isolation_level=None)
    con.execute("CREATE TABLE src (id INTEGER PRIMARY KEY, v REAL)")
    con.execute(f"WITH RECURSIVE r(i) AS (SELECT 0 UNION ALL SELECT i + 1 FROM r WHERE i < {args.rows - 1}) "
                "INSERT INTO src SELECT i, abs(random()) / 9.2e18 FROM r")
    REBUILD = "CREATE TABLE t2 (id INTEGER PRIMARY KEY, v REAL); INSERT INTO t2 SELECT * FROM src"
    COPY = "DROP TABLE IF EXISTS t2; CREATE TABLE t2 (id INTEGER PRIMARY KEY, v REAL); INSERT INTO t2 SELECT * FROM src"
    DELTA = ("DROP TABLE IF EXISTS delta; CREATE TABLE delta AS SELECT id, abs(random()) / 9.2e18 AS v FROM src "
             "WHERE abs(random()) / 9.2e18 < {frac}")
    run = lambda sql: con.executescript(sql) if ";" in sql else con.execute(sql)

def median(fn):
    runs = []
    for _ in range(args.repeats):
        prep = fn()
        t = time.perf_counter(); prep(); runs.append(time.perf_counter() - t)
    return statistics.median(runs)

# Full rebuild: a new table from the source
def rebuild():
    run("DROP TABLE IF EXISTS t2")
    return lambda: run(REBUILD)
full = median(rebuild)
print(f"{args.engine}: full rebuild of {args.rows:,} rows: {full:.3f} s")

# Incremental: a fresh copy, a delta of the changed rows, then DELETE + INSERT
rows = []
for frac in args.fractions:
    def incremental():
        run(COPY)
        run(DELTA.format(frac=frac))
        def merge():
            run("DELETE FROM t2 WHERE id IN (SELECT id FROM delta)")
            run("INSERT INTO t2 SELECT * FROM delta")
        return merge
    t = median(incremental)
    rows.append({"changed_pct": frac * 100, "incremental_s": t, "rebuild_s": full})
    print(f"changed {frac:6.1%}: incremental {t:.3f} s   ({'faster' if t < full else 'slower'} than the rebuild)")

cross = next((r["changed_pct"] for r in rows if r["incremental_s"] >= full), None)
if cross == rows[0]["changed_pct"]:
    print(f"incremental never won: even {cross:g}% changed is slower than rebuilding")
else:
    print(f"incremental stops winning at about {cross:g}% changed" if cross else "incremental won at every fraction tried")

if args.json:
    with open(args.json, "w") as f:
        json.dump({"tables": {"merge": rows, "rebuild": [{"label": "full rebuild", "seconds": full}],
                              "summary": [{"rebuild_s": full, "crossover_pct": cross if cross is not None else -1,
                                           "first_pct": rows[0]["changed_pct"], "first_s": rows[0]["incremental_s"]}]}}, f)