diff --git a/README.md b/README.md index 0be5086..22602cb 100644 --- a/README.md +++ b/README.md @@ -187,7 +187,7 @@ The Bedrock model id and the token prices used for metering are in `src/lib/ai-c |---|---| | profiler | `DATABASE_URL`, `APP_URL`, `WEBHOOK_SECRET`, `SENTRY_DSN` (optional), `AWS_REGION` | | ai-trigger | `APP_URL`, `WEBHOOK_SECRET` | -| executor | `DB_SECRET_ARN`, `S3_RAW_BUCKET`, `S3_PROCESSED_BUCKET`, `SNS_DRIFT_TOPIC_ARN` (optional), `EXECUTOR_MAX_ATTEMPTS` (optional, default 3), `SENTRY_DSN` (optional), `AWS_REGION` | +| executor | `DB_SECRET_ARN`, `S3_RAW_BUCKET`, `S3_PROCESSED_BUCKET`, `SNS_DRIFT_TOPIC_ARN` (optional), `EXECUTOR_MAX_ATTEMPTS` (optional, default 3), `SEMANTIC_DEDUP_MAX_ROWS` (optional, default 500000), `EXECUTOR_RESERVE_S` (optional, default 120: seconds of the invocation `semantic_deduplicate` leaves for writing the output, at most 25% of it), `SENTRY_DSN` (optional), `AWS_REGION` | | drift | `DB_SECRET_ARN`, `AWS_REGION` | Python dependencies are pinned in `lambdas/*/requirements.txt`. At runtime pandas/numpy come from the AWS SDK for pandas (`AWSSDKPandas-Python312`) layer. The code is tested on pandas 2.2.3 and 3.0.6 because the deployed layer version is not recorded here. @@ -265,7 +265,7 @@ Report vulnerabilities as described in [SECURITY.md](SECURITY.md). - **Downloads are served as stored.** Deliverables produced before the sidecar split may still contain `__orig_*` columns. **Export As** and the training export strip them; the native download does not. Re-run those pipelines to regenerate them. - **Excel blanks are not counted as nulls** by the profiler (cells are read as empty strings). Before/after scores are still computed the same way. - **AI spend cap coverage.** The cap is checked before `suggest-transforms` only. Committee calls are metered but not blocked, and chat-builder calls are neither metered nor blocked. -- **No upload size limit**, and `semantic_deduplicate` is O(n²). Large files can exhaust the executor Lambda. +- **No upload size limit.** Large files can still exhaust the executor Lambda's memory or time. `semantic_deduplicate` (MinHash + LSH, a few seconds for 50k short texts) is skipped with a reason instead of running past the Lambda deadline, and above `SEMANTIC_DEDUP_MAX_ROWS` rows; other rules have no such guard. - **Schema drift alerts** store their diff inside `column_definitions`, so the next alert reports a phantom `_diff` column. - **Templates** copy approved rules from all recent runs of a pipeline, not only the latest run. - **`middleware.ts`** uses the file convention that Next.js 16 deprecated in favour of `proxy.ts`. It still works; the rename is pending a test against real Clerk keys. @@ -285,7 +285,7 @@ The model can only suggest rules the executor implements (a contract test compar |---|---| | `trim_whitespace` | Strip leading/trailing whitespace in text columns | | `deduplicate` | Drop exact duplicate rows (row-loss guard 20%) | -| `semantic_deduplicate` | Drop near-duplicate rows (MinHash) | +| `semantic_deduplicate` | Drop near-duplicate rows: MinHash Jaccard estimate ≥ `threshold` (default 0.8) over lowercased word sets, keeping the first; LSH banding, deterministic hashing | | `fill_nulls` | Fill nulls (mean / median / mode / constant); original kept in the audit file | | `drop_nulls` | With a column: drop rows where that column is null. Without: drop rows with at least `ceil(threshold × columns)` nulls | | `type_cast` | Cast to float / int / datetime / str. Not applied if values would be lost (e.g. `2.5` → int) | diff --git a/lambdas/executor/handler.py b/lambdas/executor/handler.py index 221059a..7391396 100644 --- a/lambdas/executor/handler.py +++ b/lambdas/executor/handler.py @@ -17,6 +17,7 @@ import re import hashlib import math +import time import boto3 import psycopg2 import pandas as pd @@ -719,6 +720,177 @@ def _normalize_column(df: pd.DataFrame, col, params: dict) -> None: df[col] = new_col +# ── semantic_deduplicate: MinHash + LSH banding ────────────────────────────── +# +# A row is a near-duplicate when the MinHash estimate of the Jaccard similarity of its +# lowercased whitespace-token set with an earlier KEPT row is >= threshold; the first +# occurrence is kept. The old implementation compared every row with every kept row +# (O(n^2) in Python: hours for ~57k rows, so the run outlived the 15-minute Lambda timeout). +# LSH banding only verifies rows that share at least one band bucket, so the cost is +# O(n * num_perm) to build signatures plus the (small) number of candidate pairs. +# +# Hashing is deterministic: tokens are hashed with blake2b and the permutations are fixed +# universal hashes, so results no longer depend on PYTHONHASHSEED (built-in hash()). + +SEMANTIC_DEDUP_MAX_ROWS = int(os.environ.get("SEMANTIC_DEDUP_MAX_ROWS", "500000")) +SEMANTIC_DEDUP_MAX_PERM = 512 +# With no Lambda context (local runs, tests), cap the rule at this many seconds. +SEMANTIC_DEDUP_DEFAULT_BUDGET_S = float(os.environ.get("SEMANTIC_DEDUP_BUDGET_S", "600")) +# Time left in the invocation that the rule never uses (other rules, writing the output, +# DB updates): min(EXECUTOR_RESERVE_S, 25% of the invocation's remaining time). The rule is +# skipped, not half-applied, if it would run into it. +LAMBDA_RESERVE_S = float(os.environ.get("EXECUTOR_RESERVE_S", "120")) +_MERSENNE_61 = (1 << 61) - 1 +_MASK_32 = np.uint64(0xFFFFFFFF) + +_invocation_deadline: float | None = None # time.monotonic(); set by handler(), reserve already taken off + + +def set_invocation_deadline(context) -> None: + """Record when long-running rules must stop in this invocation (None without a Lambda context).""" + global _invocation_deadline + try: + remaining_s = float(context.get_remaining_time_in_millis()) / 1000.0 + except Exception: + _invocation_deadline = None + return + reserve = min(LAMBDA_RESERVE_S, 0.25 * remaining_s) + _invocation_deadline = time.monotonic() + remaining_s - reserve + + +def _rule_deadline(budget_s: float = SEMANTIC_DEDUP_DEFAULT_BUDGET_S) -> tuple[float, str]: + """Monotonic time by which a long-running rule must finish, and where the limit comes from.""" + now = time.monotonic() + if _invocation_deadline is not None: + return _invocation_deadline, "the Lambda time limit" + return now + budget_s, f"the {budget_s:.0f}s time budget" + + +def _lsh_params(threshold: float, num_perm: int) -> tuple[int, int]: + """(bands, rows_per_band) for LSH over num_perm MinHash values. + + A pair with Jaccard s becomes a candidate with probability 1 - (1 - s^r)^b. Pick the + largest r (most selective buckets, fewest candidates) whose b = num_perm // r bands + still make a pair AT the threshold a candidate with probability >= 99%, so LSH itself + misses at most ~1% of pairs right at the threshold and fewer above it (MinHash + estimation noise near the threshold is larger than that). Candidates are always + verified against the full signature, so a smaller r only costs time, never accuracy. + """ + t = min(max(threshold, 0.01), 0.999) + best = (num_perm, 1) + for r in range(1, num_perm + 1): + b = num_perm // r + if b < 1: + break + if 1 - (1 - t ** r) ** b >= 0.99: + best = (b, r) + return best + + +def _perm_coeffs(num_perm: int) -> tuple[np.ndarray, np.ndarray]: + """Fixed (a, b) coefficients for h(x) = ((a*x + b) mod 2^61-1) mod 2^32, from blake2b.""" + a = np.empty(num_perm, dtype=np.uint64) + b = np.empty(num_perm, dtype=np.uint64) + for i in range(num_perm): + d = hashlib.blake2b(f"cleanstack-minhash-{i}".encode(), digest_size=8).digest() + a[i] = (int.from_bytes(d[:4], "little") | 1) # odd, < 2^32 + b[i] = int.from_bytes(d[4:], "little") # < 2^32 + return a, b + + +def _minhash_signatures(texts: list, num_perm: int, deadline: float, limit_desc: str) -> np.ndarray: + """(n, num_perm) uint32 MinHash signatures of each text's lowercased token set.""" + n = len(texts) + token_ids: dict = {} + token_hashes: list = [] + flat: list = [] + starts = np.empty(n, dtype=np.int64) + for i, text in enumerate(texts): + starts[i] = len(flat) + for tok in set(text.lower().split()) or {""}: + tid = token_ids.get(tok) + if tid is None: + tid = token_ids[tok] = len(token_hashes) + token_hashes.append(int.from_bytes( + hashlib.blake2b(tok.encode("utf-8", "surrogatepass"), digest_size=4).digest(), "little")) + flat.append(tid) + if time.monotonic() > deadline: + raise RuleSkipped(f"semantic_deduplicate skipped: tokenizing {n:,} rows already reached {limit_desc}") + + tok_hash = np.asarray(token_hashes, dtype=np.uint64) + flat_ids = np.asarray(flat, dtype=np.int64) + a, b = _perm_coeffs(num_perm) + mersenne = np.uint64(_MERSENNE_61) + sigs = np.empty((n, num_perm), dtype=np.uint32) + # Chunk rows so each (num_perm x tokens) uint64 intermediate stays around 16 MB. + max_tokens = max(1, (16 << 20) // (8 * num_perm)) + row = 0 + while row < n: + end_tok_target = starts[row] + max_tokens + row_end = int(np.searchsorted(starts, end_tok_target, side="right")) + row_end = min(max(row_end, row + 1), n) + t0 = starts[row] + t1 = starts[row_end] if row_end < n else len(flat_ids) + x = tok_hash[flat_ids[t0:t1]] # (k,) + # a, b, x < 2^32, so a*x + b < 2^64: exact in uint64, no overflow. + hv = ((a[:, None] * x[None, :] + b[:, None]) % mersenne) & _MASK_32 # (num_perm, k) + sigs[row:row_end] = np.minimum.reduceat(hv, starts[row:row_end] - t0, axis=1).T + row = row_end + if time.monotonic() > deadline: + raise RuleSkipped( + f"semantic_deduplicate skipped: hashing {n:,} rows would exceed {limit_desc}") + return sigs + + +def _semantic_dedup_keep(texts: list, threshold: float, num_perm: int) -> list: + """Row positions to keep (first occurrence of each near-duplicate group), in order.""" + n = len(texts) + if n > SEMANTIC_DEDUP_MAX_ROWS: + raise RuleSkipped( + f"semantic_deduplicate skipped: {n:,} rows exceeds the {SEMANTIC_DEDUP_MAX_ROWS:,}-row limit " + "for one executor run") + if not 0 < threshold <= 1: + raise RuleSkipped(f"threshold must be in (0, 1], got {threshold}") + if not 1 <= num_perm <= SEMANTIC_DEDUP_MAX_PERM: + raise RuleSkipped(f"num_perm must be between 1 and {SEMANTIC_DEDUP_MAX_PERM}, got {num_perm}") + if n == 0: + return [] + deadline, limit_desc = _rule_deadline() + if time.monotonic() >= deadline: + raise RuleSkipped(f"semantic_deduplicate skipped: not enough time left before {limit_desc}") + + sigs = _minhash_signatures(texts, num_perm, deadline, limit_desc) + bands, r = _lsh_params(threshold, num_perm) + min_equal = math.ceil(threshold * num_perm - 1e-9) # equal slots needed: estimate >= threshold + + # Bucket key for (row, band k): the band's r signature values as bytes. + banded = np.ascontiguousarray(sigs[:, :bands * r]) + width = 4 * r # bytes per band (uint32 values) + buckets = [dict() for _ in range(bands)] + keep: list = [] + for i in range(n): + if i & 1023 == 0 and time.monotonic() > deadline: + raise RuleSkipped( + f"semantic_deduplicate skipped: comparing {n:,} rows would exceed {limit_desc} " + f"(stopped after {i:,} rows; too many similar candidate rows)") + row_bytes = banded[i].tobytes() + keys = [row_bytes[k * width:(k + 1) * width] for k in range(bands)] + cands: set = set() + for k in range(bands): + hit = buckets[k].get(keys[k]) + if hit: + cands.update(hit) + if cands: + idx = np.fromiter(cands, dtype=np.int64, count=len(cands)) + equal = np.count_nonzero(sigs[idx] == sigs[i], axis=1) + if (equal >= min_equal).any(): + continue # near-duplicate of an earlier kept row + keep.append(i) + for k in range(bands): + buckets[k].setdefault(keys[k], []).append(i) + return keep + + def _apply_rule(df: pd.DataFrame, rtype: str, col, params: dict) -> pd.DataFrame: """Apply one rule. Returns the new frame; raises RuleSkipped when it cannot be applied.""" if rtype == "drop_nulls": @@ -752,19 +924,7 @@ def _apply_rule(df: pd.DataFrame, rtype: str, col, params: dict) -> pd.DataFrame raise RuleSkipped("no text column to compare") threshold = float(params.get("threshold", 0.8)) num_perm = int(params.get("num_perm", 64)) - - def _minhash_sig(text: str, n: int) -> list: - tokens = set(text.lower().split()) or {""} - return [min((hash((seed, t)) & 0x7FFFFFFF) for t in tokens) for seed in range(n)] - - texts = df[target_col].astype(str).tolist() - sigs = [_minhash_sig(t, num_perm) for t in texts] - keep, kept_sigs = [], [] - for i, sig in enumerate(sigs): - is_dup = any(sum(a == b for a, b in zip(sig, ks)) / num_perm >= threshold for ks in kept_sigs) - if not is_dup: - keep.append(i) - kept_sigs.append(sig) + keep = _semantic_dedup_keep(df[target_col].astype(str).tolist(), threshold, num_perm) return df.iloc[keep].reset_index(drop=True) if rtype == "type_cast": @@ -1464,6 +1624,7 @@ def handler(event, context): If any record needs a retry the invocation raises, so SQS redelivers the batch; records that already finished are skipped on redelivery by the conditional claim. """ + set_invocation_deadline(context) retry_errors = [] results = [] for record in event.get("Records", []): diff --git a/lambdas/tests/conftest.py b/lambdas/tests/conftest.py index c93dd16..b1eee51 100644 --- a/lambdas/tests/conftest.py +++ b/lambdas/tests/conftest.py @@ -41,3 +41,7 @@ def drift(): def rule(rule_type, column=None, rule_id=None, **params): """Build a transform_rules row the way the executor reads it from Postgres.""" return {"id": rule_id, "rule_type": rule_type, "column_name": column, "parameters": params} + + +def pytest_configure(config): + config.addinivalue_line("markers", "perf: performance bound with a generous limit (run in CI)") diff --git a/lambdas/tests/test_semantic_dedup.py b/lambdas/tests/test_semantic_dedup.py new file mode 100644 index 0000000..68db436 --- /dev/null +++ b/lambdas/tests/test_semantic_dedup.py @@ -0,0 +1,215 @@ +"""semantic_deduplicate: MinHash + LSH banding (was an O(n^2) pure-Python scan). + +Run 8afd8d39 (56,745 tweets, num_perm 120, threshold 0.9) stayed on Running until the +15-minute Lambda timeout because every row was compared with every kept row. +""" +import math +import os +import random +import subprocess +import sys +import time + +import numpy as np +import pandas as pd +import pytest + +from conftest import LAMBDAS, rule + + +@pytest.fixture(autouse=True) +def _no_invocation_deadline(executor): + executor.set_invocation_deadline(None) + yield + executor.set_invocation_deadline(None) + + +def _synthetic(n, seed=7, vocab_size=20000, dup_rate=0.1): + """Short tweet-like texts; dup_rate of them are an earlier row reshuffled / re-cased.""" + rng = random.Random(seed) + vocab = [f"w{i}" for i in range(vocab_size)] + rows = [] + for _ in range(n): + if rows and rng.random() < dup_rate: + src = rows[rng.randrange(len(rows))].split() + rng.shuffle(src) + rows.append(" ".join(w.upper() if rng.random() < 0.3 else w for w in src)) + else: + rows.append(" ".join(rng.choice(vocab) for _ in range(rng.randint(8, 20)))) + return rows + + +def _apply(executor, df, column="text", **params): + results = [] + out = executor.apply_transforms(df.copy(), [rule("semantic_deduplicate", column, **params)], results) + return out, results[0] + + +# ------------------------------------------------------------------ behaviour +def test_drops_near_duplicates_and_keeps_first_occurrence(executor): + base = " ".join(f"tok{i}" for i in range(40)) + df = pd.DataFrame({ + "text": [base, "something else entirely here", base.upper(), base + " extra", "something else entirely here"], + "id": [1, 2, 3, 4, 5], + }) + out, res = _apply(executor, df, threshold=0.8, num_perm=128) + assert res["applied"] is True + # 3 = same token set (case-insensitive), 4 = Jaccard 40/41, 5 = exact copy of 2 + assert out["id"].tolist() == [1, 2] + assert out.index.tolist() == [0, 1] + + +def test_keeps_rows_below_the_threshold(executor): + a = " ".join(f"a{i}" for i in range(10)) + b = " ".join([f"a{i}" for i in range(5)] + [f"b{i}" for i in range(5)]) # Jaccard 5/15 + df = pd.DataFrame({"text": [a, b, "unrelated words only"]}) + out, _ = _apply(executor, df, threshold=0.8, num_perm=64) + assert len(out) == 3 + + +def test_picks_the_first_text_column_when_none_is_given(executor): + df = pd.DataFrame({"n": [1, 2, 3], "text": ["hello big world", "HELLO world big", "bye"]}) + out, res = _apply(executor, df, column=None, threshold=0.9) + assert res["applied"] is True + assert out["n"].tolist() == [1, 3] + + +def test_skips_without_a_text_column(executor): + df = pd.DataFrame({"n": [1, 2, 2]}) + out, res = _apply(executor, df, column=None) + assert res["applied"] is False and res["reason"] == "no text column to compare" + assert out.equals(df) + + +@pytest.mark.parametrize("params", [{"threshold": 0}, {"threshold": 1.5}, {"num_perm": 0}, {"num_perm": 10_000}]) +def test_rejects_invalid_parameters(executor, params): + df = pd.DataFrame({"text": ["a b", "a b"]}) + out, res = _apply(executor, df, **params) + assert res["applied"] is False + assert out.equals(df) + + +def test_lsh_matches_brute_force_pairwise_comparison(executor): + """Banding must not change the result: same signatures, same verdicts as comparing every pair.""" + rng = random.Random(11) + vocab = [f"v{i}" for i in range(400)] + rows = [] + for _ in range(1500): + if rows and rng.random() < 0.35: + src = rows[rng.randrange(len(rows))].split() + src[rng.randrange(len(src))] = rng.choice(vocab) # Jaccard around the threshold + if rng.random() < 0.5: + src.append(rng.choice(vocab)) + rows.append(" ".join(src)) + else: + rows.append(" ".join(rng.choice(vocab) for _ in range(rng.randint(10, 25)))) + for threshold, num_perm in [(0.9, 120), (0.8, 64), (0.6, 32)]: + deadline, desc = executor._rule_deadline() + sigs = executor._minhash_signatures(rows, num_perm, deadline, desc) + need = math.ceil(threshold * num_perm - 1e-9) + brute = [] + for i in range(len(rows)): + if brute and (np.count_nonzero(sigs[brute] == sigs[i], axis=1) >= need).any(): + continue + brute.append(i) + assert executor._semantic_dedup_keep(rows, threshold, num_perm) == brute, (threshold, num_perm) + + +@pytest.mark.parametrize("threshold,num_perm", [(0.9, 120), (0.8, 64), (0.8, 128), (0.5, 64), (0.95, 256), (0.7, 16)]) +def test_lsh_params_catch_pairs_at_the_threshold(executor, threshold, num_perm): + bands, r = executor._lsh_params(threshold, num_perm) + assert 1 <= bands * r <= num_perm + assert 1 - (1 - threshold ** r) ** bands >= 0.99 + + +# --------------------------------------------------------------- determinism +_DET_ROWS = ["the quick brown fox", "Brown fox the QUICK", "lazy dog", ""] +# sha256 of the (4, 64) uint32 signatures of _DET_ROWS. Fixed forever: a change means +# semantic_deduplicate results changed for every existing pipeline. +_DET_DIGEST = "6256df0f7b30047f8ff9c5c22e27d1cf58fc8005dd465aca107e5b40c7a5519e" + + +def _det_digest(executor): + import hashlib + deadline, desc = executor._rule_deadline() + return hashlib.sha256(executor._minhash_signatures(_DET_ROWS, 64, deadline, desc).tobytes()).hexdigest() + + +def test_signatures_are_pinned(executor): + assert _det_digest(executor) == _DET_DIGEST + assert executor._semantic_dedup_keep(_DET_ROWS, 0.8, 64) == [0, 2, 3] + + +def test_signatures_do_not_depend_on_pythonhashseed(executor): + """The old code used built-in hash(), which PYTHONHASHSEED randomizes per process.""" + script = ( + "import sys, hashlib; sys.path.insert(0, sys.argv[1]);" + "from conftest import load_handler; ex = load_handler('executor');" + f"rows = {_DET_ROWS!r}; d, l = ex._rule_deadline();" + "print(hashlib.sha256(ex._minhash_signatures(rows, 64, d, l).tobytes()).hexdigest())" + ) + env = {**os.environ, "PYTHONHASHSEED": "12345"} + out = subprocess.run([sys.executable, "-c", script, str(LAMBDAS / "tests")], env=env, + capture_output=True, text=True, check=True).stdout.strip() + assert out == _DET_DIGEST + + +def test_same_input_gives_the_same_rows(executor): + rows = _synthetic(3000, seed=5) + assert executor._semantic_dedup_keep(rows, 0.9, 120) == executor._semantic_dedup_keep(list(rows), 0.9, 120) + + +# -------------------------------------------------------------------- guards +def test_row_cap_skips_instead_of_hanging(executor, monkeypatch): + monkeypatch.setattr(executor, "SEMANTIC_DEDUP_MAX_ROWS", 10) + df = pd.DataFrame({"text": ["same words"] * 11}) + out, res = _apply(executor, df) + assert res["applied"] is False + assert "exceeds the 10-row limit" in res["reason"] + assert out.equals(df) + + +class _Ctx: + def __init__(self, ms): + self.ms = ms + + def get_remaining_time_in_millis(self): + return self.ms + + +def test_skips_when_the_lambda_deadline_is_too_close(executor): + executor.set_invocation_deadline(_Ctx(1)) # 1 ms left + df = pd.DataFrame({"text": _synthetic(500)}) + out, res = _apply(executor, df, threshold=0.9, num_perm=120) + assert res["applied"] is False + assert "Lambda time limit" in res["reason"] + assert out.equals(df) + + +def test_deadline_from_context_keeps_a_reserve(executor): + executor.set_invocation_deadline(_Ctx(900_000)) # a fresh 15-minute invocation + deadline, desc = executor._rule_deadline() + assert desc == "the Lambda time limit" + assert 900 - executor.LAMBDA_RESERVE_S - 5 < deadline - time.monotonic() <= 900 - executor.LAMBDA_RESERVE_S + executor.set_invocation_deadline(object()) # no get_remaining_time_in_millis + assert executor._rule_deadline()[1].endswith("time budget") + + +def test_handler_records_the_invocation_deadline(executor, monkeypatch): + seen = [] + monkeypatch.setattr(executor, "process_run", lambda run_id, receive_count=1: seen.append(executor._invocation_deadline) or {"ok": True}) + executor.handler({"Records": [{"body": '{"run_id": "r1"}'}]}, _Ctx(60_000)) + assert seen and seen[0] is not None and seen[0] > time.monotonic() + + +# ---------------------------------------------------------------------- perf +@pytest.mark.perf +def test_50k_short_texts_finish_in_seconds(executor): + """The O(n^2) version needed ~33 s for 4k rows and ~1.5 h for 50k; LSH takes a few seconds.""" + df = pd.DataFrame({"text": _synthetic(50_000), "id": range(50_000)}) + start = time.perf_counter() + out, res = _apply(executor, df, threshold=0.9, num_perm=120) + elapsed = time.perf_counter() - start + assert res["applied"] is True + assert 0.85 * len(df) < len(out) < len(df) # ~10% of rows are reshuffled copies + assert elapsed < 60, f"semantic_deduplicate took {elapsed:.1f}s for 50k rows"