diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 79765e82..740a9f00 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -95,8 +95,8 @@ jobs: # Without this step the coverage rule never reaches the dataset code. # devenv.nix already supplies pytest (see the comment beside # python3.withPackages), so only the CI step was missing. - - name: Python tests (pytest, generate_dataset) - run: devenv shell -- pytest generate_dataset -q --junitxml=junit-generate-dataset.xml + - name: Python tests (pytest, generate_dataset, token_history) + run: devenv shell -- pytest generate_dataset token_history -q --junitxml=junit-generate-dataset.xml # `devenv shell -- pytest` does not flush its summary line, so the printed # count is not trustworthy. Read the counts from the XML instead, and fail diff --git a/token_history/README.md b/token_history/README.md new file mode 100644 index 00000000..13bcde44 --- /dev/null +++ b/token_history/README.md @@ -0,0 +1,82 @@ +# token_history — every token that ever existed + +The tip dataset (`generate_dataset`) holds only the tokens alive at HEAD. This step replays the tokenized history (`-cregit.git`) and writes one row for every token that ever existed: its birth commit, the first commit that removed it, and its intervals on the mainline (the first-parent history of HEAD). + +**Scope: `.c` and `.h` paths only** (`MASK` in `token_history.py`). The tip dataset also has other file types; on Linux, 503 files (`.rs`, `.cpp`, `.cc`, 1.05 M rows) are not in the history. + +## Stages + +| Stage | Script | Cost (Linux, Sep 2026) | Output | +| --- | --- | --- | --- | +| 1. Replay | `token_history.py` | 50.4 h with 8 jobs; 122,910 paths, 407.9 M token rows | `part-*.tsv` (tokens), `runs-*.tsv` (move candidates) | +| 2. Build | `build.py --mode ` | load 12 min; link pass 2.3 h (`moves`) or 1.9 h (`renames`); chain resolution 5–10 min; write 5 min | `history-.parquet`, `links-.tsv` | +| 3. Attribute | `attribute.py` | not timed | `commits.parquet`: person and firm of every cregit commit | + +On the io_uring pilot (96 paths) stage 1 takes 2–3 min and stage 2 under 10 s. + +Stage 2 reads only the files of stage 1. Keep them: every mode, and a change to `moves.py`, rebuilds from them without a new replay, within one limit: stage 1 keeps only runs of at least `--run-min-alnum` (100) alphanumeric characters. So with a `--min-alnum` below 100, a removed or added run shorter than 100 still never links, and `renames` never links a file with fewer than 100. A change to `replay.py` or to `--run-min-alnum` needs a new stage 1. + +On the kernel, stage 2 stays inside `--memory` (default 6 GB): the links go to disk, and the parquet is written in `--chunks` (16) chunks of paths, so its row order is not defined. `--reuse` keeps the tables of `OUT/build.duckdb` and a finished `links-.tsv`, so a build that stopped after the link pass restarts at the chain resolution. The analysis (`msr-code/analysis/rq_history.py`) converts the links TSV to parquet itself. + +## Move modes (stage 2) + +| Mode | Links | Matches | +| --- | --- | --- | +| `off` | none | the history of each path alone | +| `renames` | a deleted file to a file created in the same commit | close to plain `git blame` with rename detection | +| `moves` (default) | any exact run of ≥ `--min-alnum` alphanumeric characters that one commit removes and adds | keeps the first identity of moved code | + +`moves` is not `git blame -C100`, and it does **not** follow more moves. It links only an exact run that the **same commit** removes and adds, each removed token once, so it misses a **copy** whose source stays in place; `-C100` follows copies, but drops a move that its diff breaks into short pieces. On a 1/32 sample of the Linux tip (5.49 M tokens, 30 Sep 2026), where the two disagree on the birth time, `-C100` gives the **older** commit in about 87% of the cases (981,149 tokens against 145,019). On the whole Linux tip, the origin commit agrees with the `-C100` tip dataset on 80.46% of the tokens for `moves` and 81.13% for `renames` (`validate.py`). + +## Run + +```sh +# once per repository: changed-path Bloom filters (a path-limited log goes from ~21 s to ~4 s on Linux) +git -C X-cregit.git commit-graph write --reachable --changed-paths + +# stage 1 (resumable: a new run skips the paths in paths-done.txt) +python3 token_history.py --repo X-cregit.git --out OUT --jobs 8 --ping 60 [--paths FILE | -- PATHSPEC] + +# progress at any time +./history_status.sh OUT + +# stage 2, once per mode, any time later +python3 build.py --out OUT --mode moves +python3 build.py --out OUT --mode renames + +# check the tip against the tip dataset +python3 validate.py --history OUT/history-moves.parquet --dataset X-dataset.parquet + +# person and firm of every cregit commit, with the joins of the tip dataset +python3 attribute.py --cregit-db X-cregit.db --persons-db X-persons.db \ + --firm-map affiliation.merged.csv --firm-canonical firm_canonical.csv \ + --out OUT/commits.parquet --dataset X-dataset.parquet +``` + +`validate.py` compares only the files that the history has (`.c` and `.h`). + +Stage 1 writes `history-progress.json` and a log line every `--ping` seconds. A `git` command that fails stops the run with `GitError`, so a run cannot end with no data and 0 errors. Any other error of a path goes to `errors.txt` with its traceback, and the run continues. + +## Columns of `history-.parquet` + +| Column | Meaning | +| --- | --- | +| `file_path`, `token_id` | the token; `token_id` counts per path | +| `token` | the cregit line, `type\|value` | +| `born_sha` | the cregit commit that added the line | +| `born_in_merge` | 1 when no parent of a merge had the line | +| `copy_of` | the token this line duplicates, when a merge kept one token twice | +| `died_sha` | the first non-merge commit that removed it | +| `tip_index` | its line in the file at HEAD, when alive there (joins `token_index` of the tip dataset) | +| `mainline_in_sha`, `mainline_in_ts` | the first-parent commit (and committer time) where it entered the mainline | +| `mainline_out_sha`, `mainline_out_ts` | where it left the mainline; empty while alive | +| `origin_path`, `origin_token_id`, `origin_born_sha` | its first identity after the move links of the mode | +| `move_mode` | the mode of the build | + +A token that left the mainline and came back has one row per interval. A token alive at any time T is a row with `mainline_in_ts <= T` and `T < mainline_out_ts` (or no `mainline_out_ts`). + +## Tests + +```sh +devenv shell -- pytest token_history -q +``` diff --git a/token_history/attribute.py b/token_history/attribute.py new file mode 100644 index 00000000..282b7aae --- /dev/null +++ b/token_history/attribute.py @@ -0,0 +1,93 @@ +#!/usr/bin/env python3 +"""Author, person and firm of every cregit commit, as the tip dataset has them. +Same columns and joins as generate_dataset phase 2, so a history token joins +its born_sha to them; --dataset prints how many commits agree with it.""" +import argparse +import json +import os +import sys + +import duckdb + +sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), + "..", "generate_dataset")) +from generate_dataset import firm_sql, sql_literal # noqa: E402 + + +def parse_args(): + ap = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + ap.add_argument("--cregit-db", required=True) + ap.add_argument("--persons-db", required=True) + ap.add_argument("--firm-map", required=True) + ap.add_argument("--firm-canonical") + ap.add_argument("--out", required=True) + ap.add_argument("--dataset") + ap.add_argument("--memory", default="4GB") + return ap.parse_args() + + +def connect(args): + con = duckdb.connect() + con.execute(f"SET memory_limit='{args.memory}'") + con.execute("INSTALL sqlite_scanner; LOAD sqlite_scanner;") + con.execute(f"CALL sqlite_attach({sql_literal(args.cregit_db)})") + con.execute(f"CALL sqlite_attach({sql_literal(args.persons_db)})") + return con + + +def write_commits(con, args): + firm_select, firm_join = firm_sql(args.firm_map, args.firm_canonical) + con.execute(f""" + COPY ( + SELECT + c.cid AS cregit_commit_sha, + coalesce(m.originalcid, c.cid) AS original_commit_sha, + c.autname AS author_name, + c.autemail AS author_email, + c.autdate AS author_date, + c.comdate AS committer_date, + e.personid, + coalesce(p.personname, e.personid) AS person_name, + e.emailaddr AS person_email, + e.domain AS person_domain, +{firm_select} + coalesce(m.repo, '') AS repo_tag + FROM commits c + LEFT JOIN commitmap m ON c.cid = m.cid + LEFT JOIN emails e ON (c.autname = e.emailname + AND c.autemail = e.emailaddr) + LEFT JOIN persons p ON e.personid = p.personid +{firm_join} ORDER BY c.cid + ) TO {sql_literal(args.out)} (FORMAT parquet, COMPRESSION zstd)""") + + +def compare_with_dataset(con, commits, dataset): + return dict(zip(["dataset_commits", "same_person", "same_firm", + "same_original"], con.execute(f""" + WITH d AS ( + SELECT DISTINCT cregit_commit_sha, original_commit_sha, + personid, firm + FROM read_parquet({sql_literal(dataset)})) + SELECT count(*), + count(*) FILTER (WHERE a.personid IS NOT DISTINCT FROM d.personid), + count(*) FILTER (WHERE a.firm IS NOT DISTINCT FROM d.firm), + count(*) FILTER (WHERE a.original_commit_sha + = d.original_commit_sha) + FROM d LEFT JOIN read_parquet({sql_literal(commits)}) a + USING (cregit_commit_sha)""").fetchone())) + + +def main(): + args = parse_args() + con = connect(args) + write_commits(con, args) + out = dict(commits=con.execute( + f"SELECT count(*), count(DISTINCT cregit_commit_sha) " + f"FROM read_parquet({sql_literal(args.out)})").fetchone()) + if args.dataset: + out.update(compare_with_dataset(con, args.out, args.dataset)) + print(json.dumps(out, indent=1)) + + +if __name__ == "__main__": + main() diff --git a/token_history/build.py b/token_history/build.py new file mode 100644 index 00000000..01955c2c --- /dev/null +++ b/token_history/build.py @@ -0,0 +1,265 @@ +#!/usr/bin/env python3 +"""Build history-.parquet from the parts that token_history.py wrote.""" +import argparse +import glob +import os +import shutil +import time +from itertools import groupby +from operator import itemgetter + +import duckdb + +from moves import MODES, link +from token_history import COLUMNS, RUN_COLUMNS, SEP, log + +TYPES = dict(token_id="BIGINT", born_in_merge="INTEGER", copy_of="BIGINT", + tip_index="BIGINT", origin_token_id="BIGINT", + mainline_in_ts="BIGINT", mainline_out_ts="BIGINT", + whole_file="INTEGER") + + +# A whole-file run of a large generated header is one line of several MB. +MAX_LINE = 16_000_000 + + +def read_tsv(con, name, pattern, columns): + """Load TSV parts. A run that stopped after the rows of a path, before + paths-done.txt, writes the path again on resume, in a new part file: + keep each path from one file only, the one with the most rows.""" + cols = ", ".join(f"'{c}': '{TYPES.get(c, 'VARCHAR')}'" for c in columns) + keep_text = ", force_not_null=['token']" if "token" in columns else "" + con.execute(f""" + CREATE TABLE {name} AS SELECT * FROM read_csv( + '{pattern}', delim='{SEP}', header=false, quote='', escape='', + nullstr='', auto_detect=false, max_line_size={MAX_LINE}{keep_text}, + filename=true, columns={{{cols}}})""") + con.execute(f""" + DELETE FROM {name} t USING ( + SELECT file_path, arg_max(filename, n) AS keep + FROM (SELECT file_path, filename, count(*) AS n + FROM {name} GROUP BY ALL) + GROUP BY file_path HAVING count(*) > 1) d + WHERE t.file_path = d.file_path AND t.filename <> d.keep""") + con.execute(f"ALTER TABLE {name} DROP COLUMN filename") + + +def resolve_origins(con): + """Point each link to the first identity of its chain by pointer jumping: + about log2(n) rounds for a chain of n moves. A move always points to an + older token, so chains end.""" + rounds = 0 + while True: + con.execute(""" + CREATE OR REPLACE TABLE next_links AS + SELECT l.file_path, l.token_id, + coalesce(u.origin_path, l.origin_path) AS origin_path, + coalesce(u.origin_token_id, l.origin_token_id) + AS origin_token_id, + u.file_path IS NOT NULL AS jumped + FROM links l LEFT JOIN links u + ON u.file_path = l.origin_path + AND u.token_id = l.origin_token_id""") + jumped = con.execute("SELECT count(*) FILTER (WHERE jumped) " + "FROM next_links").fetchone()[0] + con.execute("CREATE OR REPLACE TABLE links AS SELECT file_path, " + "token_id, origin_path, origin_token_id FROM next_links") + con.execute("DROP TABLE next_links") + rounds += 1 + if not jumped: + return rounds + + +def in_chunk(column, k, chunks): + return f"hash({column}) % {chunks} = {k}" + + +def write_history(con, target, mode, chunks): + """Tokens with their resolved origin, in chunks of paths: on a whole kernel + one join of every token with every link does not fit in memory. The chunk + parquets are then copied into one (no row order is kept).""" + fill_origins(con, chunks) + parts = os.path.join(os.path.dirname(target), + f".{os.path.basename(target)}.parts") + shutil.rmtree(parts, ignore_errors=True) + os.makedirs(parts) + for k in range(chunks): + copy_chunk(con, os.path.join(parts, f"{k:03d}.parquet"), mode, k, + chunks) + log(f"build: write chunk {k + 1}/{chunks}") + con.execute(f"COPY (SELECT * FROM read_parquet('{parts}/*.parquet')) " + f"TO '{target}' (FORMAT parquet, COMPRESSION zstd)") + shutil.rmtree(parts) + con.execute("DROP TABLE origins") + + +def fill_origins(con, chunks): + con.execute("CREATE OR REPLACE TABLE origins (file_path VARCHAR, " + "token_id BIGINT, origin_path VARCHAR, origin_token_id BIGINT, " + "origin_born_sha VARCHAR)") + # A token has one born_sha on all its rows, so the DISTINCT runs on the + # join output (a few rows per link), not on the tokens of the chunk. + for k in range(chunks): + con.execute(f""" + INSERT INTO origins + SELECT DISTINCT l.file_path, l.token_id, l.origin_path, + l.origin_token_id, o.born_sha + FROM (SELECT * FROM links + WHERE {in_chunk('origin_path', k, chunks)}) l + JOIN (SELECT file_path, token_id, born_sha FROM tokens + WHERE {in_chunk('file_path', k, chunks)}) o + ON o.file_path = l.origin_path + AND o.token_id = l.origin_token_id""") + log(f"build: origins chunk {k + 1}/{chunks}") + + +def copy_chunk(con, part, mode, k, chunks): + con.execute(f""" + COPY ( + SELECT t.*, + coalesce(o.origin_path, t.file_path) AS origin_path, + coalesce(o.origin_token_id, t.token_id) AS origin_token_id, + coalesce(o.origin_born_sha, t.born_sha) AS origin_born_sha, + '{mode}' AS move_mode + FROM (SELECT * FROM tokens + WHERE {in_chunk('file_path', k, chunks)}) t + LEFT JOIN (SELECT * FROM origins + WHERE {in_chunk('file_path', k, chunks)}) o + USING (file_path, token_id) + ) TO '{part}' (FORMAT parquet, COMPRESSION zstd)""") + + +def commit_runs(con): + """Yield (sha, runs, texts) for commits that have both kinds of run.""" + cur = con.execute(""" + WITH r AS ( + SELECT sha, kind, whole_file, file_path, token_ids, + -- a fixed order: with repeated code, the order of the + -- runs decides which copy a moved token links to + row_number() OVER ( + ORDER BY sha, kind, file_path, + string_split(token_ids, ',')[1]::BIGINT) + AS run_no + FROM runs + WHERE sha IN (SELECT sha FROM runs GROUP BY sha + HAVING count(DISTINCT kind) = 2)), + t AS ( + SELECT r.*, unnest(string_split(token_ids, ','))::BIGINT AS tid, + generate_subscripts(string_split(token_ids, ','), 1) AS pos + FROM r) + SELECT t.sha, t.run_no, t.kind, t.whole_file, t.file_path, t.tid, + tok.token, t.pos + FROM t JOIN tokens tok + ON tok.file_path = t.file_path AND tok.token_id = t.tid + ORDER BY t.sha, t.run_no, t.pos""") + # tokens has a row per mainline interval: keep the first row of each + # place (the sort puts them together), a DISTINCT that needs no memory + places = (next(rows) for _, rows in + groupby(fetch_rows(cur), key=itemgetter(0, 1, 7))) + for sha, rows in groupby(places, key=itemgetter(0)): + runs, texts = [], {} + for _, run in groupby(rows, key=itemgetter(1)): + run = list(run) + _, _, kind, whole, path = run[0][:5] + runs.append((kind, bool(whole), path, [r[5] for r in run])) + texts.update(((path, r[5]), r[6]) for r in run) + yield sha, runs, texts + + +def fetch_rows(cur): + while batch := cur.fetchmany(100_000): + yield from batch + + +def load(con, out): + if not glob.glob(os.path.join(out, "part-*.tsv")): + raise SystemExit(f"build: no part-*.tsv in {out}: run token_history.py") + read_tsv(con, "tokens", os.path.join(out, "part-*.tsv"), COLUMNS) + if glob.glob(os.path.join(out, "runs-*.tsv")): + read_tsv(con, "runs", os.path.join(out, "runs-*.tsv"), RUN_COLUMNS) + else: + con.execute("CREATE TABLE runs (sha VARCHAR, kind VARCHAR, " + "whole_file INTEGER, file_path VARCHAR, token_ids VARCHAR)") + + +def write_links(con, links_file, mode, min_alnum): + """Links go to disk commit by commit: on a whole kernel they do not fit + in memory. A born token has one birth commit, so it has one link.""" + linked = commits = 0 + with open(links_file + ".part", "w", encoding="utf-8") as f: + f.write(f"{SEP}{SEP}{SEP}\n") + if mode != "off": + for sha, runs, texts in commit_runs(con): + for (p, t), (op, ot) in link(runs, texts.__getitem__, mode, + min_alnum).items(): + f.write(f"{p}{SEP}{t}{SEP}{op}{SEP}{ot}\n") + linked += 1 + commits += 1 + if commits % 20_000 == 0: + log(f"build: {commits:,} commits, {linked:,} links") + os.replace(links_file + ".part", links_file) + log(f"build: mode {mode}, {linked:,} tokens linked in {commits:,} commits") + + +def parse_args(): + ap = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + ap.add_argument("--out", required=True) + ap.add_argument("--mode", choices=MODES, default="moves") + ap.add_argument("--min-alnum", type=int, default=100) + ap.add_argument("--memory", default="6GB") + ap.add_argument("--threads", type=int, default=4) + ap.add_argument("--chunks", type=int, default=16, + help="path chunks for the final write") + ap.add_argument("--reuse", action="store_true", + help="keep the loaded tables and a finished links file") + return ap.parse_args() + + +def connect(args): + con = duckdb.connect(os.path.join(args.out, "build.duckdb")) + con.execute(f"SET memory_limit='{args.memory}'") + con.execute(f"SET temp_directory='{os.path.join(args.out, 'tmp')}'") + con.execute(f"SET threads={args.threads}") + con.execute("SET preserve_insertion_order=false") + return con + + +def load_tables(con, out, reuse): + """Load tokens and runs unless reuse finds both; True when it did.""" + tables = {n for n, in con.execute("SELECT table_name FROM " + "duckdb_tables()").fetchall()} + reuse = reuse and {"tokens", "runs"} <= tables + for name in ("links",) if reuse else ("tokens", "runs", "links"): + con.execute(f"DROP TABLE IF EXISTS {name}") + if not reuse: + load(con, out) + log(f"build: {con.execute('SELECT count(*) FROM tokens').fetchone()[0]:,}" + f" token rows, {con.execute('SELECT count(*) FROM runs').fetchone()[0]:,}" + " runs") + return reuse + + +def load_links(con, args, reuse): + links_file = os.path.join(args.out, f"links-{args.mode}.tsv") + if reuse and os.path.exists(links_file): + log(f"build: reuse {links_file}") + else: + write_links(con, links_file, args.mode, args.min_alnum) + read_tsv(con, "links", links_file, + ["file_path", "token_id", "origin_path", "origin_token_id"]) + con.execute("DELETE FROM links WHERE file_path IS NULL") + + +def main(): + args = parse_args() + t0 = time.time() + con = connect(args) + load_links(con, args, load_tables(con, args.out, args.reuse)) + log(f"build: move chains resolved in {resolve_origins(con)} rounds") + target = os.path.join(args.out, f"history-{args.mode}.parquet") + write_history(con, target, args.mode, args.chunks) + log(f"build: wrote {target} in {time.time() - t0:.0f} s") + + +if __name__ == "__main__": + main() diff --git a/token_history/history_status.sh b/token_history/history_status.sh new file mode 100755 index 00000000..ab170a49 --- /dev/null +++ b/token_history/history_status.sh @@ -0,0 +1,21 @@ +#!/usr/bin/env bash +# Print the progress of a token_history run: history_status.sh +set -euo pipefail +f="${1:?usage: history_status.sh }/history-progress.json" +[ -f "$f" ] || { echo "no progress file at $f"; exit 1; } +python3 - "$f" <<'PY' +import json, sys, time +p = json.load(open(sys.argv[1])) +pct = 100 * p["paths_done"] / max(p["paths_total"], 1) +eta = p.get("eta_s") +print(f"state {p['state']} (updated {p['updated']})") +print(f"paths {p['paths_done']:,} / {p['paths_total']:,} ({pct:.1f}%)") +print(f"tokens {p['tokens']:,} in {p['rows']:,} rows") +print(f"rate {p['paths_per_min']} paths/min, elapsed {p['elapsed_s'] / 3600:.1f} h") +print(f"eta {'?' if eta is None else f'{eta / 3600:.1f} h'}") +print(f"flags missing parents {p['missing_parents']}, unaligned merges " + f"{p['unaligned_merges']}, unmapped mainline {p['unmapped_mainline']}, " + f"errors {p.get('errors', 0)}") +for s, path in p["slowest"]: + print(f"slow {s:7.1f} s {path}") +PY diff --git a/token_history/moves.py b/token_history/moves.py new file mode 100644 index 00000000..aad3ebb7 --- /dev/null +++ b/token_history/moves.py @@ -0,0 +1,91 @@ +"""Link tokens that moved between files, so they keep their first identity: +within one commit, a run of born tokens that equals a run of died tokens is a +move. The README describes the modes.""" +from collections import defaultdict + +from replay import alnum + +MODES = ("off", "renames", "moves") +ANCHOR = 4 +MAX_CANDIDATES = 64 + + +def matching_blocks(a, b): + """Maximal equal blocks (i, j, size) of b in a, grown from k-token anchors + that prefer to continue the previous block. Linear in practice, where + difflib is quadratic on runs full of repeated tokens.""" + k = min(ANCHOR, len(a), len(b)) + if not k: + return [] + index = defaultdict(list) + for i in range(len(a) - k + 1): + cands = index[tuple(a[i:i + k])] + if len(cands) < MAX_CANDIDATES: + cands.append(i) + blocks, j, next_i = [], 0, None + while j <= len(b) - k: + cands = index.get(tuple(b[j:j + k])) + if not cands: + j += 1 + continue + i = next_i if next_i in cands else cands[0] + n = k + while i + n < len(a) and j + n < len(b) and a[i + n] == b[j + n]: + n += 1 + blocks.append((i, j, n)) + j, next_i = j + n, i + n + return blocks + + +SEPARATOR = object() + + +def link(runs, text_of, mode, min_alnum): + """{(path, born id): (path, died id)} for one commit's runs. + runs: [(kind, whole_file, path, ids)]; text_of((path, id)): its text.""" + if mode == "off": + return {} + eligible = [r for r in runs if mode == "moves" or r[1]] + died = died_sequence(eligible, text_of) + if not died[0]: + return {} + return longest_first([ + block for kind, _, path, ids in eligible if kind == "born" + for block in born_blocks(path, ids, died, text_of, mode, min_alnum)]) + + +def born_blocks(path, ids, died, text_of, mode, min_alnum): + """(size, died keys, born keys) of each move into one born run.""" + died_text, died_key = died + born_text = [text_of((path, t)) for t in ids] + return [(n, died_key[i:i + n], [(path, t) for t in ids[j:j + n]]) + for i, j, n in matching_blocks(died_text, born_text) + if is_move(mode, min_alnum, died_key[i][0], path, + died_text[i:i + n])] + + +def died_sequence(runs, text_of): + """The texts and keys of every died run, a SEPARATOR between two runs.""" + texts, keys = [], [] + for kind, _, path, ids in runs: + if kind == "died": + texts += [text_of((path, t)) for t in ids] + [SEPARATOR] + keys += [(path, t) for t in ids] + [None] + return texts, keys + + +def is_move(mode, min_alnum, died_path, born_path, texts): + if mode == "renames" and died_path == born_path: + return False + return mode != "moves" or alnum(texts) >= min_alnum + + +def longest_first(blocks): + """Link the longest blocks first, each died and born token once.""" + links, used = {}, set() + for _, died, born in sorted(blocks, key=lambda x: -x[0]): + for d, b in zip(died, born): + if d not in used and b not in links: + links[b] = d + used.add(d) + return links diff --git a/token_history/replay.py b/token_history/replay.py new file mode 100644 index 00000000..6b4b63b9 --- /dev/null +++ b/token_history/replay.py @@ -0,0 +1,355 @@ +"""Replay one path of a cregit repository, one token per line. A line keeps +its identity while diffs leave it unchanged; at a merge it takes the identity +of the first parent that has it unchanged, as `git blame` does.""" +import functools +import re +import subprocess +from collections import Counter +from dataclasses import dataclass, field + +HEADER = "\x01" +NULL_BLOB = "0" * 40 +HUNK = re.compile(r"^@@ -(\d+)(?:,(\d+))? \+(\d+)(?:,(\d+))? @@") +RAW = re.compile(r"^:\d+ \d+ ([0-9a-f]{40}) ([0-9a-f]{40}) ") + + +@dataclass +class Hunk: + old_start: int + old_len: int + added: list = field(default_factory=list) + + +@dataclass +class Diff: + old_blob: str = NULL_BLOB + hunks: list = field(default_factory=list) + + +@dataclass +class Commit: + sha: str + parents: list + ct: int + blob: str = None + diffs: list = field(default_factory=list) + + +def path_log(repo, path): + return subprocess.run( + ["git", "-C", repo, "log", "--full-history", "--simplify-merges", + "--reverse", "--topo-order", "--parents", "-m", "-p", "-U0", "--raw", + "--no-abbrev", "--no-renames", "--no-color", + f"--format={HEADER}%H %P%x09%ct", "--", path], + check=True, capture_output=True, text=True, errors="replace").stdout + + +def parse_log(text): + """Commits in topological order; a merge holds one diff per parent.""" + by_sha, cur, diff = {}, None, None + for line in text.split("\n"): + if line.startswith(HEADER): + head, ct = line[1:].split("\t") + sha, *parents = head.split() + cur = by_sha.setdefault(sha, Commit(sha, parents, int(ct))) + diff = Diff() + cur.diffs.append(diff) + elif cur is not None: + parse_diff_line(line, cur, diff) + return list(by_sha.values()) + + +def parse_diff_line(line, commit, diff): + if (m := RAW.match(line)): + diff.old_blob, commit.blob = m.group(1), m.group(2) + elif (m := HUNK.match(line)): + old_len = 1 if m.group(2) is None else int(m.group(2)) + diff.hunks.append(Hunk(int(m.group(1)), old_len)) + elif diff.hunks and line.startswith("+") and not line.startswith("+++"): + diff.hunks[-1].added.append(line[1:]) + + +class GitError(RuntimeError): + """git exited with an error: the caller must not go on with no data.""" + + +def git_out(repo, *args): + """The output of one git command; GitError on a non-zero exit.""" + proc = subprocess.run(["git", "-C", repo, *args], capture_output=True, + text=True, errors="replace") + if proc.returncode: + raise GitError(f"git {' '.join(args[:3])} exited {proc.returncode}: " + f"{proc.stderr.strip()[:300]}") + return proc.stdout + + +def diff_hunks(repo, a, b): + """The hunks from blob a to blob b. A null blob gives no hunks: git + cannot diff it, and the replay treats that case as no change.""" + if NULL_BLOB in (a, b): + return [] + return parse_hunks(git_out(repo, "diff", "-U0", "--no-color", a, b)) + + +def parse_hunks(diff_text): + return parse_log(f"{HEADER}x\t0\n{diff_text}")[0].diffs[0].hunks + + +def align(parent_len, hunks): + """For each child line: the parent line index it keeps, or its new text.""" + child, pos = [], 0 + for h in hunks: + start = h.old_start - 1 if h.old_len else h.old_start + child.extend(range(pos, start)) + child.extend(("+", text) for text in h.added) + pos = start + h.old_len + child.extend(range(pos, parent_len)) + return child + + +@dataclass +class Token: + text: str + born: str + born_in_merge: bool + died: str = None + copy_of: int = None + + +def alnum(texts): + return sum(ch.isalnum() for text in texts for ch in text) + + +def unique(items): + return list(dict.fromkeys(items)) + + +def adjacent_runs(positioned): + """Split [(position, token)] into runs of consecutive positions.""" + run, last = [], None + for pos, t in positioned: + if run and pos != last + 1: + yield run + run = [] + run.append(t) + last = pos + if run: + yield run + + +def kept_id(i, line, sources): + """The id that child line i keeps: from the base parent, else from the + first other parent that has the line unchanged; None for a new line.""" + if not isinstance(line, tuple): + return sources[0][0][line] + return next((ids[m[i]] for ids, m in sources[1:] + if i < len(m) and not isinstance(m[i], tuple)), None) + + +class PathReplay: + """Token identities along the history of one path. real_parents, blob_at + and diff_blobs let a merge try its real parents in their real order, as + `git blame` does, when the simplified history dropped one of them.""" + + def __init__(self, commits, real_parents=None, blob_at=None, + diff_blobs=None, run_min_alnum=None): + self.commits = commits + self.tokens = [] + self.state = {} + self.blob_nodes = {} + self.order = {c.sha: i for i, c in enumerate(commits)} + self.real_parents = real_parents + self.blob_at = blob_at and functools.cache(blob_at) + self.diff_blobs = diff_blobs + self.run_min_alnum = run_min_alnum + self.runs = [] + self.missing_parents = 0 + self.unaligned_merges = 0 + + def born(self, text, commit, in_merge): + self.tokens.append(Token(text, commit.sha, in_merge)) + return len(self.tokens) - 1 + + def copy(self, t): + """A second line with the same identity: a new token, same birth.""" + orig = self.tokens[t] + self.tokens.append(Token(orig.text, orig.born, orig.born_in_merge, + copy_of=t)) + return len(self.tokens) - 1 + + def parent_ids(self, sha): + if sha not in self.state: + self.missing_parents += 1 + return [] + return self.state[sha] + + def blob_of(self, sha): + i = self.order.get(sha) + return NULL_BLOB if i is None else self.commits[i].blob + + def parents_of(self, c): + return self.real_parents(c.sha) if self.real_parents else c.parents + + def merge_blobs(self, c, real): + blobs = unique(self.blob_of(p) for p in c.parents) + if real == c.parents: + return blobs + real_blobs = unique(self.blob_at(p) for p in real) + return blobs if real_blobs == blobs else real_blobs + + def ids_of(self, blob, c): + for p in c.parents: + if self.blob_of(p) == blob: + return self.parent_ids(p) + return self.state_of_blob(blob, c.ct) + + def first_parent_source(self, c): + ids = self.parent_ids(c.parents[0]) if c.parents else [] + return [(ids, align(len(ids), c.diffs[0].hunks))] + + def sources(self, c): + """(parent ids, alignment) per parent, in the order blame tries them.""" + real = self.parents_of(c) + if len(real) <= 1 or not c.parents: + return self.first_parent_source(c) + blobs = self.merge_blobs(c, real) + ids = self.ids_of(c.blob, c) if c.blob in blobs else None + if ids is not None: + return [(ids, align(len(ids), []))] + return self.merge_sources(c, blobs) or self.first_parent_source(c) + + def merge_sources(self, c, blobs): + by_blob = {d.old_blob: d.hunks for d in c.diffs} + out = [self.merge_source(c, blob, by_blob) + for blob in blobs if blob != NULL_BLOB] + return [s for s in out if s] + + def merge_source(self, c, blob, by_blob): + ids = self.ids_of(blob, c) + hunks = None if ids is None else self.hunks_to_child(c, blob, by_blob) + if hunks is None: + self.unaligned_merges += 1 + return None + return ids, align(len(ids), hunks) + + def hunks_to_child(self, c, blob, by_blob): + if blob in by_blob: + return by_blob[blob] + if blob == c.blob: + return [] + return self.diff_blobs(blob, c.blob) if self.diff_blobs else None + + def step(self, c): + sources = self.sources(c) + in_merge = len(sources) > 1 or len(c.parents) > 1 + base_ids, base = sources[0] + ids, used = [], set() + for i, line in enumerate(base): + kept = kept_id(i, line, sources) + if kept is None: + kept = self.born(line[1], c, in_merge) + elif kept in used: + kept = self.copy(kept) + used.add(kept) + ids.append(kept) + if not in_merge: + self.mark_died(c, base_ids, set(ids)) + self.add_runs(c, "born", [(i, t) for i, t in enumerate(ids) + if self.tokens[t].born == c.sha]) + self.state[c.sha] = ids + + def mark_died(self, c, base_ids, alive): + gone = [(j, t) for j, t in enumerate(base_ids) if t not in alive] + for _, t in gone: + if self.tokens[t].died is None: + self.tokens[t].died = c.sha + self.add_runs(c, "died", gone) + + def resolve_blobs(self): + for c in self.commits: + if c.blob is None: + c.blob = self.blob_of(c.parents[0]) if c.parents else NULL_BLOB + + def latest_node(self, blob, before_ct): + nodes = [s for s in self.blob_nodes.get(blob, ()) + if self.commits[self.order[s]].ct <= before_ct] + nodes = nodes or self.blob_nodes.get(blob, ()) + return max(nodes, key=self.order.get) if nodes else None + + def add_runs(self, c, kind, positioned): + """Record runs of adjacent tokens that are long enough to move. + positioned: [(position in the file, token id)], in file order.""" + if self.run_min_alnum is None: + return + whole = self.is_whole_file(c, kind) + for run in adjacent_runs(positioned): + if alnum(self.tokens[t].text for t in run) >= self.run_min_alnum: + self.runs.append((c.sha, kind, whole, run)) + + def is_whole_file(self, c, kind): + if kind == "died": + return c.blob == NULL_BLOB + return all(self.blob_of(p) == NULL_BLOB for p in c.parents) + + def run(self, mainline=None): + """Replay every commit. With mainline changes, free unneeded states.""" + self.resolve_blobs() + for c in self.commits: + self.blob_nodes.setdefault(c.blob, []).append(c.sha) + keep = None if mainline is None else self.states_to_keep(mainline) + pending = Counter(p for c in self.commits for p in c.parents) + for c in self.commits: + self.step(c) + if keep is not None: + self.free_parents(c, pending, keep) + return self + + def free_parents(self, c, pending, keep): + for p in c.parents: + pending[p] -= 1 + if not pending[p] and p not in keep: + self.state.pop(p, None) + + def states_to_keep(self, mainline): + """The states that a mainline change or a real-parent lookup reads.""" + keep = {self.latest_node(b, ct) for _, ct, b in mainline} + for c in self.commits: + real = self.parents_of(c) + if len(real) > 1 and real != c.parents: + keep |= {self.latest_node(self.blob_at(p), c.ct) + for p in real} + return keep + + def state_of_blob(self, blob, before_ct): + """The latest node with this blob, committed at or before a time.""" + if blob == NULL_BLOB: + return [] + node = self.latest_node(blob, before_ct) + return None if node is None else self.state.get(node) + + def tip_positions(self, changes): + """{token: position} in the path's file at the last mainline change.""" + if not changes: + return {} + sha, ct, blob = changes[-1] + ids = self.state_of_blob(blob, ct) or [] + return {t: i for i, t in enumerate(ids)} + + def mainline_intervals(self, changes): + """{token: [(in_sha, in_ct, out_sha, out_ct)]} on the mainline. + changes: [(sha, ct, blob)] of the first-parent commits that changed + this path, oldest first.""" + intervals, alive, unmapped = {}, set(), 0 + for sha, ct, blob in changes: + ids = self.state_of_blob(blob, ct) + if ids is None: + unmapped += 1 + continue + now = set(ids) + for t in now - alive: + intervals.setdefault(t, []).append([sha, ct, None, None]) + for t in alive - now: + intervals[t][-1][2:] = [sha, ct] + alive = now + self.unmapped_mainline = unmapped + return intervals diff --git a/token_history/test_build.py b/token_history/test_build.py new file mode 100644 index 00000000..e2b2ef5b --- /dev/null +++ b/token_history/test_build.py @@ -0,0 +1,51 @@ +import duckdb +import pytest + +from build import commit_runs, load, resolve_origins +from token_history import SEP + + +def test_a_token_with_two_mainline_intervals_is_one_place_in_its_run(): + con = duckdb.connect() + con.execute("CREATE TABLE tokens (file_path VARCHAR, token_id BIGINT, " + "token VARCHAR, mainline_in_sha VARCHAR)") + con.execute("INSERT INTO tokens VALUES ('a.c', 0, 'x', 'm1'), " + "('a.c', 1, 'y', 'm1'), ('a.c', 1, 'y', 'm3'), " + "('a.c', 2, 'z', 'm1')") + con.execute("CREATE TABLE runs (sha VARCHAR, kind VARCHAR, " + "whole_file INTEGER, file_path VARCHAR, token_ids VARCHAR)") + con.execute("INSERT INTO runs VALUES ('c1', 'born', 0, 'a.c', '0,1,2'), " + "('c1', 'died', 0, 'a.c', '0,1,2')") + (sha, runs, texts), = list(commit_runs(con)) + assert sha == "c1" + assert [ids for _, _, _, ids in runs] == [[0, 1, 2], [0, 1, 2]] + assert texts[("a.c", 1)] == "y" + + +def test_a_path_written_again_by_a_resume_is_loaded_once(tmp_path): + # the resume wrote a.c again in part-2, after a change of the replay + def row(path, born): + return SEP.join([path, "0", "x", born, "0", "", "", "", "m", "1", "", + ""]) + "\n" + (tmp_path / "part-1.tsv").write_text(row("a.c", "b1") + row("b.c", "b1")) + (tmp_path / "part-2.tsv").write_text(row("a.c", "b2")) + con = duckdb.connect() + load(con, str(tmp_path)) + assert con.execute("SELECT count(*), count(DISTINCT (file_path, token_id))" + " FROM tokens").fetchone() == (2, 2) + assert len(con.execute("SELECT * FROM tokens").description) == 12 + + +def test_no_part_files_is_a_clear_error(tmp_path): + with pytest.raises(SystemExit, match="no part"): + load(duckdb.connect(), str(tmp_path)) + + +def test_origins_follow_chains(): + con = duckdb.connect() + con.execute("CREATE TABLE links (file_path VARCHAR, token_id BIGINT, " + "origin_path VARCHAR, origin_token_id BIGINT)") + con.execute("INSERT INTO links VALUES ('c', 1, 'b', 1), ('b', 1, 'a', 1)") + resolve_origins(con) + assert sorted(con.execute("FROM links").fetchall()) == [ + ("b", 1, "a", 1), ("c", 1, "a", 1)] diff --git a/token_history/test_moves.py b/token_history/test_moves.py new file mode 100644 index 00000000..2599b6ad --- /dev/null +++ b/token_history/test_moves.py @@ -0,0 +1,58 @@ +import pytest + +from moves import link, matching_blocks + + +RUN = ["int", "alpha_long_identifier", ";", "beta_long_identifier"] + + +def runs_and_text(moved_whole=True): + tx = {} + for i, t in enumerate(RUN): + tx[("old.c", 10 + i)] = t + tx[("new.c", i)] = t + runs = [("died", moved_whole, "old.c", [10, 11, 12, 13]), + ("born", moved_whole, "new.c", [0, 1, 2, 3])] + return runs, tx.__getitem__ + + +EACH_TOKEN = {("new.c", i): ("old.c", 10 + i) for i in range(4)} + + +@pytest.mark.parametrize("mode, whole, min_alnum, links", [ + ("off", True, 10, {}), + ("moves", False, 10, EACH_TOKEN), + ("moves", False, 10_000, {}), + ("renames", False, 10, {}), + ("renames", True, 10, EACH_TOKEN), +], ids=["off links nothing", "moves link each token", + "short runs do not move", "renames need a deleted file", + "renames link a whole file"]) +def test_link_modes(mode, whole, min_alnum, links): + runs, tx = runs_and_text(moved_whole=whole) + assert link(runs, tx, mode, min_alnum) == links + + +def test_matching_blocks_on_a_long_repetitive_run(): + import time + a = [";", "}", "x", ";"] * 20000 + ["unique_tail_token"] * 10 + b = ["new"] + a + t = time.time() + blocks = matching_blocks(a, b) + assert time.time() - t < 5 + assert sum(n for _, _, n in blocks) == len(a) + assert blocks[0][:2] == (0, 1) + + +def test_a_block_never_spans_two_died_runs(): + a = [f"a_token_{i}" for i in range(4)] + b = [f"b_token_{i}" for i in range(4)] + tx = {("a.c", i): t for i, t in enumerate(a)} + tx |= {("b.c", i): t for i, t in enumerate(b)} + tx |= {("n.c", i): t for i, t in enumerate(a + b)} + runs = [("died", False, "a.c", [0, 1, 2, 3]), + ("died", False, "b.c", [0, 1, 2, 3]), + ("born", False, "n.c", list(range(8)))] + got = link(runs, tx.__getitem__, "moves", 5) + assert [got[("n.c", i)] for i in range(8)] == ( + [("a.c", i) for i in range(4)] + [("b.c", i) for i in range(4)]) diff --git a/token_history/test_replay.py b/token_history/test_replay.py new file mode 100644 index 00000000..5f699420 --- /dev/null +++ b/token_history/test_replay.py @@ -0,0 +1,282 @@ +import os +import re +import subprocess + +import pytest + +from replay import NULL_BLOB, PathReplay, align, Hunk, parse_log, path_log +from token_history import GitSource, mainline_changes, merge_parents + + +def git(repo, *args, date=None, check=True): + env = dict(os.environ, GIT_AUTHOR_NAME="t", GIT_AUTHOR_EMAIL="t@x", + GIT_COMMITTER_NAME="t", GIT_COMMITTER_EMAIL="t@x") + if date: + env["GIT_AUTHOR_DATE"] = env["GIT_COMMITTER_DATE"] = f"{date} +0000" + return subprocess.run(["git", "-C", str(repo), *args], check=check, + capture_output=True, text=True, env=env).stdout + + +def init_repo(path): + path.mkdir() + git(path, "init", "-q", "-b", "main") + return path + + +def write(repo, path, lines): + (repo / path).write_text("".join(f"{x}\n" for x in lines)) + + +def commit(repo, msg, date): + git(repo, "add", "-A") + git(repo, "commit", "-q", "-m", msg, date=date) + return git(repo, "rev-parse", "HEAD").strip() + + +def blame(repo, path): + out = git(repo, "blame", "--line-porcelain", path) + return [line[:40] for line in out.splitlines() + if re.match(r"^[0-9a-f]{40} \d+ \d+", line)] + + +@pytest.fixture +def repo(tmp_path): + r = init_repo(tmp_path / "r") + write(r, "a.c", ["int", "a", ";", "int", "b", ";"]) + commit(r, "c1", "2020-01-01T00:00:00") + write(r, "a.c", ["int", "a", ";", "long", "b", ";", "x"]) + commit(r, "c2", "2020-02-01T00:00:00") + git(r, "checkout", "-q", "-b", "side") + write(r, "a.c", ["int", "a", ";", "long", "b", ";", "x", "side1"]) + commit(r, "s1", "2020-03-01T00:00:00") + git(r, "checkout", "-q", "main") + write(r, "a.c", ["main0", "int", "a", ";", "long", "b", ";", "x"]) + commit(r, "m1", "2020-04-01T00:00:00") + git(r, "merge", "-q", "--no-commit", "side") + write(r, "a.c", ["main0", "int", "a", ";", "long", "b", ";", "x", + "side1", "fix"]) + commit(r, "merge", "2020-05-01T00:00:00") + write(r, "a.c", ["main0", "a", ";", "long", "b", ";", "x", "side1", "fix"]) + commit(r, "c3", "2020-06-01T00:00:00") + return r + + +def replay(repo, path): + return PathReplay(parse_log(path_log(str(repo), path))).run() + + +def real_replay(repo, path, mainline=None): + """The replay of the runner, with the real parents and blobs of git.""" + return GitSource(str(repo), merge_parents(str(repo))).replay( + path, mainline, None) + + +def tip_born(rp, repo, path): + tip = git(repo, "rev-parse", f"HEAD:{path}").strip() + ids = rp.state_of_blob(tip, 2**62) + return [rp.tokens[t].born for t in ids], [rp.tokens[t].text for t in ids] + + +def test_align_insert_delete(): + assert align(3, [Hunk(2, 1, ["y"])]) == [0, ("+", "y"), 2] + assert align(2, [Hunk(1, 0, ["z"])]) == [0, ("+", "z"), 1] + assert align(2, [Hunk(0, 0, ["z"])]) == [("+", "z"), 0, 1] + + +def test_tip_matches_git_blame(repo): + rp = replay(repo, "a.c") + born, texts = tip_born(rp, repo, "a.c") + assert texts == (repo / "a.c").read_text().split() + assert born == blame(repo, "a.c") + assert rp.missing_parents == 0 and rp.unaligned_merges == 0 + + +def test_merge_resolution_line_is_born_in_merge(repo): + rp = replay(repo, "a.c") + merge = git(repo, "rev-parse", "HEAD~1").strip() + fix = next(t for t in rp.tokens if t.text == "fix") + assert fix.born == merge and fix.born_in_merge + + +def test_deleted_token_keeps_its_death(repo): + rp = replay(repo, "a.c") + c3 = git(repo, "rev-parse", "HEAD").strip() + first_int = next(t for t in rp.tokens if t.text == "int") + assert first_int.died == c3 + + +def test_deleted_file_and_recreated(repo): + git(repo, "rm", "-q", "a.c") + commit(repo, "del", "2020-07-01T00:00:00") + write(repo, "a.c", ["int", "new"]) + recreate = commit(repo, "re", "2020-08-01T00:00:00") + rp = replay(repo, "a.c") + born, _ = tip_born(rp, repo, "a.c") + assert born == [recreate, recreate] == blame(repo, "a.c") + assert all(t.died for t in rp.tokens if t.born != recreate) + + +def test_mainline_intervals(repo): + rp = replay(repo, "a.c") + changes = mainline_changes(str(repo), [])["a.c"] + iv = rp.mainline_intervals(changes) + side1 = next(i for i, t in enumerate(rp.tokens) if t.text == "side1") + merge = git(repo, "rev-parse", "HEAD~1").strip() + assert iv[side1][0][0] == merge and iv[side1][0][2] is None + first_int = next(i for i, t in enumerate(rp.tokens) if t.text == "int") + assert iv[first_int][0][2] == git(repo, "rev-parse", "HEAD").strip() + assert NULL_BLOB not in {c[2] for c in changes} + + +def test_merge_of_equal_parents_keeps_the_blob(repo): + git(repo, "checkout", "-q", "-b", "twin") + write(repo, "b.c", ["other"]) + commit(repo, "twin only touches b.c", "2020-07-01T00:00:00") + git(repo, "checkout", "-q", "main") + write(repo, "c.c", ["main only touches c.c"]) + commit(repo, "main only touches c.c", "2020-07-02T00:00:00") + git(repo, "merge", "-q", "--no-edit", "twin", date="2020-07-03T00:00:00") + write(repo, "a.c", ["main0", "a", ";", "long", "b", ";", "x", "side1", + "fix", "after"]) + commit(repo, "after merge", "2020-07-04T00:00:00") + rp = replay(repo, "a.c") + born, _ = tip_born(rp, repo, "a.c") + assert born == blame(repo, "a.c") + assert rp.unaligned_merges == 0 + + +def test_merge_follows_real_parent_order(repo): + base = ["main0", "a", ";", "long", "b", ";", "x", "side1", "fix"] + git(repo, "checkout", "-q", "-b", "redo") + write(repo, "a.c", [t for t in base if t != "long"]) + commit(repo, "drop long", "2020-07-01T00:00:00") + write(repo, "a.c", base + ["tail"]) + commit(repo, "re-add long", "2020-07-02T00:00:00") + git(repo, "checkout", "-q", "main") + git(repo, "merge", "-q", "--no-ff", "--no-commit", "redo") + write(repo, "a.c", base + ["tail", "merged"]) + commit(repo, "merge redo", "2020-07-03T00:00:00") + rp = real_replay(repo, "a.c") + born, _ = tip_born(rp, repo, "a.c") + assert born == blame(repo, "a.c") + + +def test_merge_equal_to_second_parent_takes_it_whole(repo): + base = ["main0", "a", ";", "long", "b", ";", "x", "side1", "fix"] + git(repo, "checkout", "-q", "-b", "same") + write(repo, "a.c", base + ["s"]) + commit(repo, "side adds s", "2020-07-01T00:00:00") + git(repo, "checkout", "-q", "main") + write(repo, "a.c", [t for t in base if t != "long"]) + commit(repo, "main drops long", "2020-07-02T00:00:00") + write(repo, "a.c", base) + commit(repo, "main re-adds long", "2020-07-03T00:00:00") + write(repo, "c.c", ["other"]) + commit(repo, "main touches c.c", "2020-07-04T00:00:00") + git(repo, "merge", "-q", "--no-ff", "--no-commit", "same") + write(repo, "a.c", base + ["s"]) + commit(repo, "merge same", "2020-07-05T00:00:00") + rp = real_replay(repo, "a.c") + born, _ = tip_born(rp, repo, "a.c") + assert born == blame(repo, "a.c") + + +def test_freeing_states_changes_nothing(repo): + test_merge_follows_real_parent_order(repo) + changes = mainline_changes(str(repo), [])["a.c"] + full = real_replay(repo, "a.c") + lean = real_replay(repo, "a.c", changes) + assert len(lean.state) < len(full.state) + assert lean.tokens == full.tokens + assert lean.mainline_intervals(changes) == full.mainline_intervals(changes) + + +def test_runs_mark_a_whole_file_move(repo): + body = ["int", "alpha_long_identifier", ";", "int", "beta_long_identifier", + ";", "return", "gamma_long_identifier", ";"] + write(repo, "m.c", body) + commit(repo, "add m.c", "2020-07-01T00:00:00") + git(repo, "mv", "m.c", "n.c") + moved = commit(repo, "move m.c to n.c", "2020-07-02T00:00:00") + runs = {} + for path in ("m.c", "n.c"): + rp = PathReplay(parse_log(path_log(str(repo), path)), + run_min_alnum=20).run() + runs[path] = [(sha, kind, whole, [rp.tokens[t].text for t in ids]) + for sha, kind, whole, ids in rp.runs if sha == moved] + assert runs["m.c"] == [(moved, "died", True, body)] + assert runs["n.c"] == [(moved, "born", True, body)] + + +def test_a_line_kept_from_two_parents_becomes_a_copy(tmp_path): + r = init_repo(tmp_path / "dup") + write(r, "d.c", ["A"]) + first = commit(r, "base", "2021-01-01T00:00:00") + git(r, "checkout", "-q", "-b", "keep") + write(r, "d.c", ["Q", "A"]) + commit(r, "keep adds Q before A", "2021-01-02T00:00:00") + git(r, "checkout", "-q", "main") + write(r, "d.c", ["A", "P"]) + commit(r, "main adds P after A", "2021-01-03T00:00:00") + git(r, "merge", "-q", "--no-ff", "--no-commit", "keep", check=False) + assert (r / ".git" / "MERGE_HEAD").exists() + write(r, "d.c", ["A", "P", "Q", "A"]) + commit(r, "merge keeps both", "2021-01-04T00:00:00") + rp = real_replay(r, "d.c") + for ids in rp.state.values(): + assert len(ids) == len(set(ids)) + born, _ = tip_born(rp, r, "d.c") + assert born == blame(r, "d.c") + assert any(t.copy_of is not None and t.born == first for t in rp.tokens) + + +def test_merge_that_brings_a_new_file_to_main(tmp_path): + r = init_repo(tmp_path / "newfile") + write(r, "other.c", ["o"]) + commit(r, "base", "2022-01-01T00:00:00") + git(r, "checkout", "-q", "-b", "topic") + write(r, "n.c", ["a", "b", "c"]) + commit(r, "topic adds n.c", "2022-01-02T00:00:00") + git(r, "checkout", "-q", "main") + write(r, "other.c", ["o", "p"]) + commit(r, "main changes other.c", "2022-01-03T00:00:00") + git(r, "merge", "-q", "--no-ff", "--no-commit", "topic") + write(r, "n.c", ["a", "b", "c", "fixed in merge"]) + commit(r, "merge topic with a fix", "2022-01-04T00:00:00") + write(r, "n.c", ["a", "b", "c", "fixed in merge", "later"]) + commit(r, "later", "2022-01-05T00:00:00") + rp = real_replay(r, "n.c") + born, _ = tip_born(rp, r, "n.c") + assert born == blame(r, "n.c") + + +def test_path_log_drops_a_merge_that_brings_no_change(tmp_path): + # --simplify-merges: a side branch that never touches the path adds no + # merge to its log, so no merge is replayed for nothing + r = init_repo(tmp_path / "r") + write(r, "a.c", ["a"]) + commit(r, "base", "2020-01-01T00:00:00") + git(r, "checkout", "-q", "-b", "side") + write(r, "o.c", ["o"]) + commit(r, "side", "2020-01-02T00:00:00") + git(r, "checkout", "-q", "main") + write(r, "a.c", ["a", "m"]) + commit(r, "main", "2020-01-03T00:00:00") + git(r, "merge", "-q", "--no-ff", "-m", "merge", "side") + assert [len(c.parents) for c in parse_log(path_log(str(r), "a.c"))] \ + == [0, 1] + + +def test_a_commit_with_no_diff_keeps_the_parent_blob(): + # git log can list a commit with no diff for the path; it has the + # parent's content, not an empty file + a, b = "a" * 40, "b" * 40 + blob = "1" * 40 + text = (f"\x01{a} \t1\n:000000 100644 {NULL_BLOB} {blob} A\ta.c\n" + "@@ -0,0 +1,2 @@\n+x\n+y\n" + f"\x01{b} {a}\t2\n") + commits = parse_log(text) + rp = PathReplay(commits) + rp.run() + assert commits[1].blob == blob + assert rp.state[b] == rp.state[a] diff --git a/token_history/test_token_history.py b/token_history/test_token_history.py new file mode 100644 index 00000000..5bfd16f0 --- /dev/null +++ b/token_history/test_token_history.py @@ -0,0 +1,67 @@ +import argparse + +import pytest + +from replay import NULL_BLOB, GitError, diff_hunks +from test_replay import commit, git, init_repo, write +from token_history import list_paths, mainline_changes, merge_parents + + +@pytest.fixture +def not_a_repo(tmp_path): + d = tmp_path / "empty" + d.mkdir() + return str(d) + + +def test_git_failures_stop_the_run(not_a_repo): + with pytest.raises(GitError): + merge_parents(not_a_repo) + with pytest.raises(GitError, match="not a git repository"): + mainline_changes(not_a_repo, []) + with pytest.raises(GitError): + list_paths(argparse.Namespace(paths=None, repo=not_a_repo, + pathspec=[])) + + +def test_the_null_blob_diffs_to_no_hunks(tmp_path): + r = init_repo(tmp_path / "r") + write(r, "a.c", ["x", "y"]) + commit(r, "a", "2020-01-01T00:00:00") + blob = git(r, "rev-parse", "HEAD:a.c").strip() + assert diff_hunks(str(r), blob, NULL_BLOB) == [] + assert diff_hunks(str(r), NULL_BLOB, blob) == [] + with pytest.raises(GitError): + diff_hunks(str(r), blob, "1" * 40) + + +def test_a_good_repo_still_works(tmp_path): + r = init_repo(tmp_path / "r") + write(r, "a.c", ["x"]) + sha = commit(r, "a", "2020-01-01T00:00:00") + assert merge_parents(str(r)) == {} + assert list(mainline_changes(str(r), [])) == ["a.c"] + assert mainline_changes(str(r), [])["a.c"][0][0] == sha + assert list_paths(argparse.Namespace(paths=None, repo=str(r), + pathspec=[])) == ["a.c"] + (tmp_path / "paths").write_text("a.c\nREADME\n") + assert list_paths(argparse.Namespace(paths=tmp_path / "paths")) == ["a.c"] + + +def test_an_error_in_one_path_is_a_record_not_a_stop(monkeypatch): + import token_history + + def broken(path): + raise ValueError("bad path") + monkeypatch.setattr(token_history, "replay_path", broken) + r = token_history.work("a.c") + assert r["path"] == "a.c" and "ValueError" in r["error"] + + +def test_a_dead_cat_file_stops_the_run(not_a_repo, monkeypatch): + import token_history + src = token_history.GitSource(not_a_repo, {}) + monkeypatch.setattr(token_history, "replay_path", + lambda path: src.blob_at("HEAD")) + with pytest.raises(GitError): + token_history.work("a.c") diff --git a/token_history/token_history.py b/token_history/token_history.py new file mode 100755 index 00000000..79e8b40e --- /dev/null +++ b/token_history/token_history.py @@ -0,0 +1,254 @@ +#!/usr/bin/env python3 +"""Extract every token that ever existed in a cregit repository. + +One row per token and mainline interval; the README describes the columns.""" +import argparse +import json +import multiprocessing as mp +import os +import re +import subprocess +import time +import traceback +from datetime import datetime + +from replay import (HEADER, NULL_BLOB, GitError, PathReplay, diff_hunks, + git_out, parse_log, path_log) + +MASK = re.compile(r"\.[ch]$") +SEP = "\x1f" +COLUMNS = ["file_path", "token_id", "token", "born_sha", "born_in_merge", + "copy_of", "died_sha", "tip_index", "mainline_in_sha", "mainline_in_ts", + "mainline_out_sha", "mainline_out_ts"] +RUN_COLUMNS = ["sha", "kind", "whole_file", "file_path", "token_ids"] + + +def log(msg): + print(f"[{datetime.now():%Y-%m-%d %H:%M:%S}] {msg}", flush=True) + + +def merge_parents(repo): + out = git_out(repo, "rev-list", "--merges", "--parents", "HEAD") + return {sha: parents for sha, *parents in map(str.split, out.splitlines())} + + +def mainline_changes(repo, pathspec): + """{path: [(sha, ct, new blob)]} along the first-parent history.""" + proc = subprocess.Popen( + ["git", "-C", repo, "log", "--first-parent", "--reverse", "-m", + "--raw", "--no-abbrev", "--no-renames", f"--format={HEADER}%H %ct", + "HEAD", "--", *pathspec], + stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, + errors="replace") + changes, sha, ct = {}, None, 0 + for line in proc.stdout: + if line.startswith(HEADER): + sha, ct = line[1:].split() + ct = int(ct) + elif line.startswith(":"): + meta, path = line.rstrip("\n").split("\t", 1) + if MASK.search(path): + changes.setdefault(path, []).append((sha, ct, meta.split()[3])) + if proc.wait(): + raise GitError(f"git log --first-parent exited {proc.returncode}: " + f"{proc.stderr.read().strip()[:300]}") + return changes + + +class GitSource: + """Real parents from a shared map, blobs from one cat-file per worker.""" + + def __init__(self, repo, merges): + self.repo, self.merges = repo, merges + self.cat = subprocess.Popen( + ["git", "-C", repo, "cat-file", "--batch-check=%(objectname)"], + stdin=subprocess.PIPE, stdout=subprocess.PIPE, text=True) + self.path = None + + def real_parents(self, sha): + return self.merges.get(sha) or self.rewritten[sha] + + def blob_at(self, sha): + try: + self.cat.stdin.write(f"{sha}:{self.path}\n") + self.cat.stdin.flush() + line = self.cat.stdout.readline() + except BrokenPipeError: + line = "" + if not line: # cat-file died: every later path would fail too + raise GitError(f"git cat-file exited {self.cat.wait()}") + out = line.split()[0] + return out if len(out) == 40 else NULL_BLOB + + def replay(self, path, mainline, run_min_alnum): + self.path = path + commits = parse_log(path_log(self.repo, path)) + self.rewritten = {c.sha: c.parents for c in commits} + return PathReplay(commits, real_parents=self.real_parents, + blob_at=self.blob_at, + diff_blobs=lambda a, b: diff_hunks(self.repo, a, b), + run_min_alnum=run_min_alnum).run(mainline) + + +def rows(path, rp, intervals, tip): + for i, t in enumerate(rp.tokens): + base = [path, i, t.text, t.born, int(t.born_in_merge), + "" if t.copy_of is None else t.copy_of, t.died or "", + tip.get(i, "")] + for span in intervals.get(i) or [["", "", "", ""]]: + yield base + ["" if v is None else v for v in span] + + +WORKER = {} + + +def init_worker(repo, merges, changes, out_dir, run_min_alnum): + # unique per worker, even when a later run reuses a pid + tag = f"{os.getpid()}-{time.time_ns()}" + WORKER.update( + src=GitSource(repo, merges), changes=changes, + run_min_alnum=run_min_alnum, + out=open(os.path.join(out_dir, f"part-{tag}.tsv"), "a", + encoding="utf-8"), + runs=open(os.path.join(out_dir, f"runs-{tag}.tsv"), "a", + encoding="utf-8")) + + +def clean(value): + return str(value).replace(SEP, " ").replace("\n", " ") + + +def work(path): + try: + return replay_path(path) + except GitError: + raise # git failed, not the path: stop the run + except Exception: + return dict(path=path, error=traceback.format_exc()) + + +def replay_path(path): + t0 = time.time() + mainline = WORKER["changes"].get(path, []) + rp = WORKER["src"].replay(path, mainline, WORKER["run_min_alnum"]) + lines = [SEP.join(map(clean, r)) + "\n" + for r in rows(path, rp, rp.mainline_intervals(mainline), + rp.tip_positions(mainline))] + runs = [SEP.join([sha, kind, str(int(whole)), clean(path), + ",".join(map(str, ids))]) + "\n" + for sha, kind, whole, ids in rp.runs] + WORKER["out"].write("".join(lines)) + WORKER["runs"].write("".join(runs)) + WORKER["out"].flush() + WORKER["runs"].flush() + return dict(path=path, tokens=len(rp.tokens), rows=len(lines), + commits=len(rp.commits), seconds=round(time.time() - t0, 2), + missing_parents=rp.missing_parents, + unaligned_merges=rp.unaligned_merges, + unmapped_mainline=rp.unmapped_mainline) + + +class Progress: + def __init__(self, path, total, ping): + self.path, self.total, self.ping = path, total, ping + self.start = self.last = time.time() + self.done, self.tokens, self.rows = 0, 0, 0 + self.flags = dict(missing_parents=0, unaligned_merges=0, + unmapped_mainline=0, errors=0) + self.slowest = [] + + def add(self, r): + self.done += 1 + if "error" in r: + self.flags["errors"] += 1 + r = dict(r, tokens=0, rows=0, seconds=0) + self.tokens += r["tokens"] + self.rows += r["rows"] + for k in self.flags: + self.flags[k] += r.get(k, 0) + self.slowest = sorted(self.slowest + [(r["seconds"], r["path"])])[-5:] + if time.time() - self.last >= self.ping or self.done == self.total: + self.write() + + def write(self, state="running"): + self.last = time.time() + elapsed = self.last - self.start + rate = self.done / elapsed if elapsed else 0 + eta = (self.total - self.done) / rate if rate else None + snap = dict(state=state, paths_done=self.done, + paths_total=self.total, tokens=self.tokens, + rows=self.rows, elapsed_s=round(elapsed), + paths_per_min=round(60 * rate, 1), + eta_s=None if eta is None else round(eta), + updated=datetime.now().isoformat(timespec="seconds"), + **self.flags, slowest=self.slowest[::-1]) + tmp = self.path + ".tmp" + with open(tmp, "w") as f: + json.dump(snap, f, indent=1) + os.replace(tmp, self.path) + eta_txt = "?" if eta is None else f"{eta / 3600:.1f} h" + log(f"token_history: {self.done}/{self.total} paths, " + f"{self.tokens:,} tokens, {60 * rate:.1f} paths/min, ETA {eta_txt}") + + +def list_paths(args): + out = open(args.paths).read() if args.paths else git_out( + args.repo, "log", "--diff-filter=A", "--name-only", "--no-renames", + "-m", "--format=", "HEAD", "--", *args.pathspec) + return sorted({p for p in out.splitlines() if MASK.search(p)}) + + +def parse_args(): + ap = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + ap.add_argument("--repo", required=True) + ap.add_argument("--out", required=True) + ap.add_argument("--paths") + ap.add_argument("--jobs", type=int, default=8) + ap.add_argument("--ping", type=int, default=60) + ap.add_argument("--run-min-alnum", type=int, default=100, + help="shortest run that the move pass may link") + ap.add_argument("pathspec", nargs="*") + return ap.parse_args() + + +def read_done(done_file): + if not os.path.exists(done_file): + return set() + return set(open(done_file).read().split("\n")) + + +def record(r, out_dir, done_out): + if "error" in r: + with open(os.path.join(out_dir, "errors.txt"), "a") as err: + err.write(f"== {r['path']}\n{r['error']}\n") + return + done_out.write(r["path"] + "\n") + done_out.flush() + + +def main(): + args = parse_args() + os.makedirs(args.out, exist_ok=True) + done_file = os.path.join(args.out, "paths-done.txt") + done = read_done(done_file) + paths = [p for p in list_paths(args) if p not in done] + log(f"token_history: {len(paths)} paths to replay ({len(done)} done)") + merges = merge_parents(args.repo) + changes = mainline_changes(args.repo, args.pathspec) + log(f"token_history: {len(merges):,} merges, mainline changes for " + f"{len(changes):,} paths") + progress = Progress(os.path.join(args.out, "history-progress.json"), + len(paths), args.ping) + progress.write("starting") + with open(done_file, "a") as done_out, mp.Pool( + args.jobs, init_worker, + (args.repo, merges, changes, args.out, + args.run_min_alnum)) as pool: + for r in pool.imap_unordered(work, paths): + record(r, args.out, done_out) + progress.add(r) + progress.write("done") + + +if __name__ == "__main__": + main() diff --git a/token_history/validate.py b/token_history/validate.py new file mode 100755 index 00000000..03362204 --- /dev/null +++ b/token_history/validate.py @@ -0,0 +1,50 @@ +#!/usr/bin/env python3 +"""Compare the tip of a history parquet with the tip dataset parquet. +Every tip token must be a dataset row at the same file, token_index and text; +the share whose origin commit is its cregit_commit_sha is the agreement.""" +import argparse +import json + +import duckdb + + +def main(): + ap = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + ap.add_argument("--history", required=True) + ap.add_argument("--dataset", required=True) + args = ap.parse_args() + con = duckdb.connect() + con.execute("SET memory_limit='6GB'") + con.execute(f""" + CREATE VIEW h AS SELECT * FROM read_parquet('{args.history}') + WHERE tip_index IS NOT NULL AND mainline_out_sha IS NULL""") + con.execute(f""" + CREATE VIEW d AS SELECT file_path, token_index, cregit_commit_sha, + token_type, token_value + FROM read_parquet('{args.dataset}') + WHERE file_path IN (SELECT DISTINCT file_path FROM h)""") + row = con.execute(""" + SELECT + (SELECT count(*) FROM h) AS history_tip, + (SELECT count(*) FROM d) AS dataset_rows, + count(*) AS joined, + count(*) FILTER (WHERE h.token IN (d.token_type, d.token_value, + d.token_type || '|' || d.token_value)) + AS same_text, + count(*) FILTER (WHERE h.origin_born_sha = d.cregit_commit_sha) + AS same_origin, + count(*) FILTER (WHERE h.born_sha = d.cregit_commit_sha) AS same_born + FROM h JOIN d ON d.file_path = h.file_path + AND d.token_index = h.tip_index""").fetchone() + names = ["history_tip", "dataset_rows", "joined", "same_text", + "same_origin", "same_born"] + out = dict(zip(names, row)) + out["origin_agreement_pct"] = round(100 * out["same_origin"] + / max(out["joined"], 1), 2) + out["files"] = con.execute( + "SELECT count(DISTINCT file_path) FROM h").fetchone()[0] + print(json.dumps(out, indent=1)) + + +if __name__ == "__main__": + main()