fix(executor): semantic_deduplicate via MinHash + LSH (was O(n²), hung run 8afd8d39)

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) |

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", []):

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)")