diff --git a/README.md b/README.md index 8b73d46..fc1e041 100644 --- a/README.md +++ b/README.md @@ -14,6 +14,11 @@ usable in CI. See [usage and CI guidance](docs/USAGE_AND_CI.md) for the exact CSV contract, JSON output, exit-code handling and known blind spots. +CSV rows are processed incrementally using a training-text hash map and ID/group +sets, rather than keeping both full datasets in memory. See the +[algorithm and benchmark notes](docs/ALGORITHM.md) for complexity and a +reproducible comparison. + The comparison lowercases text, removes punctuation and collapses whitespace. It does not detect paraphrases, translations or copied content with substantial edits. In a real project, choose groups that match the deployment risk (such as diff --git a/benchmarks/streaming_audit.py b/benchmarks/streaming_audit.py new file mode 100644 index 0000000..7b92224 --- /dev/null +++ b/benchmarks/streaming_audit.py @@ -0,0 +1,88 @@ +"""Compare streaming and materialised audits on synthetic wide CSV files. + +Run from the repository root: python benchmarks/streaming_audit.py +This measures Python allocations, not process RSS. Dataset creation is excluded. +""" +import argparse +import csv +import gc +import json +import sys +import tempfile +import time +import tracemalloc +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) +from splitcheck import audit, normalise, read_rows + + +def materialised_audit(train, test): + train_rows = read_rows(train, "id", "text") + test_rows = read_rows(test, "id", "text") + index = {normalise(row["text"]): row["id"] for row in train_rows} + overlaps = [ + {"train_id": index[key], "test_id": row["id"]} + for row in test_rows + if (key := normalise(row["text"])) in index + ] + return { + "train_rows": len(train_rows), "test_rows": len(test_rows), + "exact_text_overlaps": overlaps, "shared_groups": [], + "passed": not overlaps, + } + + +def measure(function, train, test): + gc.collect() + tracemalloc.start() + started = time.perf_counter() + try: + result = function(train, test) + elapsed = time.perf_counter() - started + peak = tracemalloc.get_traced_memory()[1] + finally: + tracemalloc.stop() + return result, {"seconds": elapsed, "peak_python_bytes": peak} + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--rows", type=int, default=5000, help="rows per split") + parser.add_argument("--columns", type=int, default=8, help="unused payload columns") + parser.add_argument("--payload-size", type=int, default=256) + args = parser.parse_args() + if args.rows < 1 or args.columns < 0 or args.payload_size < 0: + parser.error("rows must be positive; columns and payload-size must be nonnegative") + with tempfile.TemporaryDirectory() as directory: + train, test = Path(directory) / "train.csv", Path(directory) / "test.csv" + fields = ["id", "text"] + [f"unused_{i}" for i in range(args.columns)] + for path, prefix in [(train, "train"), (test, "test")]: + with path.open("w", newline="", encoding="utf-8") as stream: + writer = csv.writer(stream) + writer.writerow(fields) + for i in range(args.rows): + writer.writerow( + [f"{prefix}-{i}", f"{prefix} message {i % 100}"] + + ["x" * args.payload_size] * args.columns + ) + original, baseline = measure(materialised_audit, train, test) + current, streaming = measure(audit, train, test) + if original != current: + raise AssertionError("streaming report differs from the reference") + print(json.dumps({ + "python": sys.version.split()[0], + "rows_per_split": args.rows, + "unused_columns": args.columns, + "bytes_per_payload_field": args.payload_size, + "materialised": baseline, + "streaming": streaming, + "peak_allocation_reduction_percent": ( + 100 * (1 - streaming["peak_python_bytes"] / baseline["peak_python_bytes"]) + ), + "reports_equal": True, + }, indent=2)) + + +if __name__ == "__main__": + main() diff --git a/docs/ALGORITHM.md b/docs/ALGORITHM.md new file mode 100644 index 0000000..b0c01cd --- /dev/null +++ b/docs/ALGORITHM.md @@ -0,0 +1,75 @@ +# Streaming leakage audit: algorithm and measurement + +## Data structures + +The audit reads the training CSV once, normalises each text, and stores its row +ID in a dictionary keyed by normalised text. Assigning an existing key preserves +the previous behaviour: the last matching training ID is used in the report. +An optional set records non-empty training groups. + +The test CSV is then read once. Dictionary membership identifies text overlap; +set membership identifies shared groups. Matches are appended in test-file +order, while a set deduplicates shared groups before sorting them for the report. +Unused CSV columns are discarded as iteration advances. + +The row iterator validates unique, non-empty IDs with a per-file set. It checks +the entire input, including rows after the first overlap. A late validation error +aborts the audit; the CLI does not print a partial JSON report. + +`read_rows()` remains available and returns a list for existing callers. The +`audit()` hot path uses the incremental iterator instead. + +## Complexity and limits + +For `n` training rows and `m` test rows, dictionary and set operations take +expected O(n + m) time, plus text normalisation and sorting `g` shared groups in +O(g log g). Hash-table operations are expected-time, not a worst-case guarantee. + +Memory is not constant: the training text index, training group set, per-file +ID-validation set and output overlaps must still fit in RAM. Its size depends on +unique text lengths, IDs, groups and number of matches. The change removes the +additional retention of both complete CSV row lists and their unused columns. +It is most useful for wide files with large payload fields that are not audited. + +## Reproduce the comparison + +From the repository root with Python 3.11 or newer: + +```bash +python benchmarks/streaming_audit.py --rows 5000 --columns 8 --payload-size 256 +python -m pytest +``` + +The benchmark generates two synthetic CSVs in a temporary directory, then +compares an equivalent materialised reference against the streaming audit. +It checks report equality before printing timings and peak Python allocations. +File generation is excluded from the measurement, and temporary files are cleaned +up afterwards. No production dataset or network access is used. + +`tracemalloc` measures tracked Python allocations, not total process RSS. Timings +include its overhead and depend on Python, hardware and filesystem caching; +this is a reproducible microbenchmark, not a production throughput guarantee. +Use `--columns 0` to compare narrow files, or increase `--rows` to study ID/index +growth. Keep memory claims tied to the exact reported workload. + +### Observed local allocation comparison + +A single local run on Python 3.14.5, Windows, on 13 October 2026 used 5,000 +rows per split and 256 characters per unused field: + +| Unused columns | Materialised peak bytes | Streaming peak bytes | Reduction | +| --- | --- | --- | --- | +| 8 | 28,316,228 | 1,088,533 | 96.2% | +| 0 | 3,775,481 | 1,075,891 | 71.5% | + +Both reports matched the reference exactly. This is a synthetic allocation +comparison, not a claim about real datasets, production throughput or process RSS. +The reference uses the same row validation but retains both full row lists. + +## Correctness checks + +Regression tests cover the streaming path, report equality with a materialised +reference across seeded random inputs, last-training-ID selection, output order, +empty groups, custom columns, UTF-8 BOM, missing columns, header-only files, +late duplicate IDs and incremental iterator validation. Existing example and CLI +semantics remain unchanged; this does not add semantic-duplicate detection. diff --git a/docs/USAGE_AND_CI.md b/docs/USAGE_AND_CI.md index 9f40f26..7094129 100644 --- a/docs/USAGE_AND_CI.md +++ b/docs/USAGE_AND_CI.md @@ -19,6 +19,10 @@ The file must contain at least one row and a header. It is read using `utf-8-sig`, which also accepts a UTF-8 byte-order mark. Change column names with `--id-column`, `--text-column` and `--group-column`. +The audit processes rows incrementally rather than retaining both complete CSVs. +It still stores the training text index, validation IDs and report findings; +see [Algorithm and measurement](ALGORITHM.md) for the memory boundary and benchmark. + Group values are compared as provided; there is **no normalisation of group labels**. For realistic leakage checks, apply a consistent grouping scheme upstream (for example the same campaign ID format in both files). diff --git a/splitcheck.py b/splitcheck.py index 0bea1ce..c74dec7 100644 --- a/splitcheck.py +++ b/splitcheck.py @@ -5,6 +5,8 @@ import csv import json import re +from collections.abc import Iterator +from itertools import chain from pathlib import Path @@ -12,35 +14,58 @@ def normalise(text: str) -> str: return " ".join(re.findall(r"[a-z0-9]+", text.casefold())) -def read_rows(path: Path, id_column: str, text_column: str) -> list[dict[str, str]]: +def _iter_rows( + path: Path, id_column: str, text_column: str, group_column=None +) -> Iterator[dict[str, str]]: + """Validate CSV rows incrementally without retaining unused columns.""" with path.open(newline="", encoding="utf-8-sig") as stream: - rows = list(csv.DictReader(stream)) - if not rows: - raise ValueError(f"{path.name} has no rows") - missing = {id_column, text_column} - set(rows[0]) - if missing: - raise ValueError(f"{path.name} is missing columns: {sorted(missing)}") - ids = [row[id_column].strip() for row in rows] - if any(not item for item in ids) or len(ids) != len(set(ids)): - raise ValueError(f"{path.name} IDs must be non-empty and unique") - return rows + reader = csv.DictReader(stream) + first = next(reader, None) + if first is None: + raise ValueError(f"{path.name} has no rows") + missing = {id_column, text_column} - set(first) + if missing: + raise ValueError(f"{path.name} is missing columns: {sorted(missing)}") + if group_column and group_column not in first: + raise ValueError(f"Both files need group column: {group_column}") + seen_ids = set() + for row in chain((first,), reader): + row_id = row[id_column].strip() + if not row_id or row_id in seen_ids: + raise ValueError(f"{path.name} IDs must be non-empty and unique") + seen_ids.add(row_id) + yield row + + +def read_rows(path: Path, id_column: str, text_column: str) -> list[dict[str, str]]: + """Return validated rows for callers that explicitly need materialisation.""" + return list(_iter_rows(path, id_column, text_column)) def audit(train: Path, test: Path, id_column="id", text_column="text", group_column=None) -> dict: - train_rows = read_rows(train, id_column, text_column) - test_rows = read_rows(test, id_column, text_column) - train_text = {normalise(row[text_column]): row[id_column] for row in train_rows} - overlaps = [ - {"train_id": train_text[key], "test_id": row[id_column]} - for row in test_rows if (key := normalise(row[text_column])) in train_text - ] - groups = [] - if group_column: - if group_column not in train_rows[0] or group_column not in test_rows[0]: - raise ValueError(f"Both files need group column: {group_column}") - train_groups = {row[group_column] for row in train_rows if row[group_column]} - groups = sorted(train_groups & {row[group_column] for row in test_rows if row[group_column]}) - return {"train_rows": len(train_rows), "test_rows": len(test_rows), + train_text = {} + train_groups = set() + train_count = 0 + for row in _iter_rows(train, id_column, text_column, group_column): + train_count += 1 + # Preserve the existing last-training-ID selection for duplicate texts. + train_text[normalise(row[text_column])] = row[id_column] + if group_column and row[group_column]: + train_groups.add(row[group_column]) + + overlaps = [] + shared_groups = set() + test_count = 0 + for row in _iter_rows(test, id_column, text_column, group_column): + test_count += 1 + key = normalise(row[text_column]) + if key in train_text: + overlaps.append({"train_id": train_text[key], "test_id": row[id_column]}) + if group_column and row[group_column] in train_groups: + shared_groups.add(row[group_column]) + + groups = sorted(shared_groups) + return {"train_rows": train_count, "test_rows": test_count, "exact_text_overlaps": overlaps, "shared_groups": groups, "passed": not overlaps and not groups} diff --git a/tests/test_iterator_validation.py b/tests/test_iterator_validation.py new file mode 100644 index 0000000..e7991c7 --- /dev/null +++ b/tests/test_iterator_validation.py @@ -0,0 +1,38 @@ +import csv + +import pytest + +from splitcheck import _iter_rows, audit + + +def write_rows(path, rows): + with path.open("w", newline="", encoding="utf-8") as stream: + writer = csv.DictWriter(stream, fieldnames=list(rows[0])) + writer.writeheader() + writer.writerows(rows) + + +def test_rows_are_validated_as_the_iterator_advances(tmp_path): + path = tmp_path / "data.csv" + write_rows(path, [{"id": "a", "text": "one"}, {"id": "a", "text": "two"}]) + rows = _iter_rows(path, "id", "text") + assert next(rows) == {"id": "a", "text": "one"} + with pytest.raises(ValueError, match="unique"): + next(rows) + + +@pytest.mark.parametrize("missing", ["id", "text"]) +def test_missing_required_column(tmp_path, missing): + path = tmp_path / "data.csv" + row = {"id": "a", "text": "one"} + del row[missing] + write_rows(path, [row]) + with pytest.raises(ValueError, match="missing columns"): + audit(path, path) + + +def test_whitespace_only_id(tmp_path): + path = tmp_path / "data.csv" + write_rows(path, [{"id": " ", "text": "one"}]) + with pytest.raises(ValueError, match="non-empty"): + audit(path, path) diff --git a/tests/test_streaming_audit.py b/tests/test_streaming_audit.py new file mode 100644 index 0000000..b10dd2a --- /dev/null +++ b/tests/test_streaming_audit.py @@ -0,0 +1,134 @@ +import csv +import random +import json +import subprocess +import sys + +import pytest + +import splitcheck + + +def write_rows(path, rows, fieldnames=None, encoding="utf-8"): + with path.open("w", newline="", encoding=encoding) as stream: + writer = csv.DictWriter(stream, fieldnames=fieldnames or list(rows[0])) + writer.writeheader() + writer.writerows(rows) + + +def materialised_reference(train_rows, test_rows, group_column=None): + index = {splitcheck.normalise(row["text"]): row["id"] for row in train_rows} + overlaps = [ + {"train_id": index[key], "test_id": row["id"]} + for row in test_rows + if (key := splitcheck.normalise(row["text"])) in index + ] + groups = [] + if group_column: + groups = sorted( + {row[group_column] for row in train_rows if row[group_column]} + & {row[group_column] for row in test_rows if row[group_column]} + ) + return { + "train_rows": len(train_rows), "test_rows": len(test_rows), + "exact_text_overlaps": overlaps, "shared_groups": groups, + "passed": not overlaps and not groups, + } + + +def test_audit_does_not_materialise_rows(tmp_path, monkeypatch): + train, test = tmp_path / "train.csv", tmp_path / "test.csv" + write_rows(train, [{"id": "t", "text": "Hello!"}]) + write_rows(test, [{"id": "e", "text": "hello"}]) + + def reject_materialisation(*args): + raise AssertionError("audit must not retain complete CSV row lists") + + monkeypatch.setattr(splitcheck, "read_rows", reject_materialisation) + assert splitcheck.audit(train, test)["exact_text_overlaps"] == [ + {"train_id": "t", "test_id": "e"} + ] + + +@pytest.mark.parametrize("group_column", [None, "group"]) +@pytest.mark.parametrize("seed", range(10)) +def test_matches_materialised_reference(tmp_path, group_column, seed): + rng = random.Random(seed) + texts = ["Hello!", "hello", "other text", "OTHER TEXT!", "", "123"] + train_rows = [ + {"id": f"t{i}", "text": rng.choice(texts), + "group": rng.choice(["", "a", "b", "c"]), "unused": "payload" * 100} + for i in range(rng.randint(1, 40)) + ] + test_rows = [ + {"id": f"e{i}", "text": rng.choice(texts), + "group": rng.choice(["", "b", "c", "d"]), "unused": "payload" * 100} + for i in range(rng.randint(1, 40)) + ] + train, test = tmp_path / "train.csv", tmp_path / "test.csv" + write_rows(train, train_rows) + write_rows(test, test_rows) + assert splitcheck.audit(train, test, group_column=group_column) == ( + materialised_reference(train_rows, test_rows, group_column) + ) + + +def test_custom_columns_and_bom(tmp_path): + train, test = tmp_path / "train.csv", tmp_path / "test.csv" + write_rows(train, [{"key": " t ", "body": "same", "source": "x"}], encoding="utf-8-sig") + write_rows(test, [{"key": "e", "body": "SAME!", "source": "x"}]) + result = splitcheck.audit(train, test, "key", "body", "source") + assert result["exact_text_overlaps"] == [{"train_id": " t ", "test_id": "e"}] + assert result["shared_groups"] == ["x"] + + +@pytest.mark.parametrize("bad_split", ["train", "test"]) +def test_late_duplicate_ids_abort_the_audit(tmp_path, bad_split): + train, test = tmp_path / "train.csv", tmp_path / "test.csv" + rows = [{"id": "first", "text": "same"}, {"id": "second", "text": "other"}] + write_rows(train, rows) + write_rows(test, rows) + write_rows(tmp_path / f"{bad_split}.csv", rows + [{"id": " first ", "text": "last"}]) + with pytest.raises(ValueError, match="non-empty and unique"): + splitcheck.audit(train, test) + + +@pytest.mark.parametrize("bad_split", ["train", "test"]) +def test_header_only_file_is_rejected(tmp_path, bad_split): + train, test = tmp_path / "train.csv", tmp_path / "test.csv" + write_rows(train, [{"id": "t", "text": "one"}]) + write_rows(test, [{"id": "e", "text": "two"}]) + write_rows(tmp_path / f"{bad_split}.csv", [], ["id", "text"]) + with pytest.raises(ValueError, match="has no rows"): + splitcheck.audit(train, test) + + +def test_missing_group_column_is_rejected(tmp_path): + train, test = tmp_path / "train.csv", tmp_path / "test.csv" + write_rows(train, [{"id": "t", "text": "one", "group": "a"}]) + write_rows(test, [{"id": "e", "text": "two"}]) + with pytest.raises(ValueError, match="Both files need group column"): + splitcheck.audit(train, test, group_column="group") + + +def test_read_rows_still_returns_a_list(tmp_path): + path = tmp_path / "data.csv" + rows = [{"id": "1", "text": "one"}, {"id": "2", "text": "two"}] + write_rows(path, rows) + assert splitcheck.read_rows(path, "id", "text") == rows + + +@pytest.mark.parametrize("overlap", [False, True]) +def test_cli_report_and_exit_code_are_preserved(tmp_path, overlap): + train, test = tmp_path / "train.csv", tmp_path / "test.csv" + train_rows = [{"id": "t", "text": "hello", "group": "a"}] + test_rows = [{"id": "e", "text": "HELLO!" if overlap else "other", "group": "b"}] + write_rows(train, train_rows) + write_rows(test, test_rows) + result = subprocess.run( + [sys.executable, splitcheck.__file__, str(train), str(test), "--group-column", "group"], + capture_output=True, text=True, + ) + assert result.returncode == int(overlap) + assert result.stderr == "" + assert json.loads(result.stdout) == materialised_reference(train_rows, test_rows, "group")