diff --git a/src/freshdata/enterprise/entity_resolution.py b/src/freshdata/enterprise/entity_resolution.py index c768e5a..a2300a4 100644 --- a/src/freshdata/enterprise/entity_resolution.py +++ b/src/freshdata/enterprise/entity_resolution.py @@ -33,6 +33,7 @@ from __future__ import annotations import json +import logging import re from collections.abc import Iterable, Sequence from dataclasses import dataclass, field, replace @@ -50,6 +51,8 @@ EntityResolutionConfig, ) +logger = logging.getLogger("freshdata.enterprise.entity_resolution") + _PAIRS_SAMPLE = 50 _CLUSTERS_SAMPLE = 50 _PREVIEW_LEN = 24 @@ -546,7 +549,12 @@ def __str__(self) -> str: def _is_missing(v: Any) -> bool: - return v is None or (isinstance(v, float) and pd.isna(v)) + """True for ``None`` and every scalar pandas missing value (NaN, NaT, pd.NA).""" + if v is None: + return True + if isinstance(v, (str, bytes)): + return False + return bool(pd.api.types.is_scalar(v) and pd.isna(v)) def _compare(cmp: ComparisonLevel, a: Any, b: Any) -> float | None: @@ -714,7 +722,7 @@ def _parse_blocking_rules( for idx, rule in enumerate(config.blocking_rules): try: left_key, right_key = _parse_blocking(rule.sql) - except (ValueError, KeyError): + except (ValueError, KeyError, EntityResolutionError): continue parsed.append((blocking_rule_id(idx), left_key, right_key)) return parsed @@ -797,16 +805,42 @@ def _candidates_duckdb( # ===================================================================== _FUNC_RE = re.compile(r"^(\w+)\s*\((.*)\)$", re.DOTALL) +# An SQL identifier: unquoted (``name``) or double-quoted with ``""`` escapes. +_IDENT = r'(?:[^\W\d][\w$]*|"(?:[^"]|"")*")' +# A bare column reference, optionally table-prefixed (``l.col`` / ``r."a b"``). +_COLUMN_REF_RE = re.compile(rf"^(?:{_IDENT}\s*\.\s*)?({_IDENT})$") +# Quoted spans (identifiers or string literals) — masked before splitting. +_QUOTED_RE = re.compile(r'"(?:[^"]|"")*"|\'(?:[^\']|\'\')*\'') +_AND_RE = re.compile(r"\band\b", re.IGNORECASE) +_EQ_RE = re.compile(r"==?") + + +def _quote_identifier(name: object) -> str: + """Double-quote *name* as an SQL identifier, escaping embedded quotes.""" + return '"' + str(name).replace('"', '""') + '"' + + +def _unquote_identifier(ident: str) -> str: + if len(ident) >= 2 and ident[0] == ident[-1] == '"': + return ident[1:-1].replace('""', '"') + return ident + +def _make_expr(expr: str, refs: list[str] | None = None) -> Callable[[dict[str, Any]], Any]: + """Compile a tiny SQL expression subset to a record→value function. -def _make_expr(expr: str) -> Callable[[dict[str, Any]], Any]: - """Compile a tiny SQL expression subset to a record→value function.""" + Supported: column references (``l.col``, ``r."quoted col"``, ``col``) and + ``lower``/``upper``/``trim``/``left``/``right``/``substr`` over them. Any + other expression (operators, literals, other functions) raises + :class:`EntityResolutionError`. Referenced column names are appended to + *refs* when given. + """ expr = expr.strip() m = _FUNC_RE.match(expr) if m: func = m.group(1).lower() args = _split_args(m.group(2)) - inner = _make_expr(args[0]) + inner = _make_expr(args[0] if args else "", refs) if func == "lower": return lambda rec: _safe_str(inner(rec)).lower() if func == "upper": @@ -826,9 +860,16 @@ def _make_expr(expr: str) -> Callable[[dict[str, Any]], Any]: raise EntityResolutionError( f"unsupported SQL function {func!r} in pandas blocking; use the duckdb backend" ) - # bare column reference, possibly prefixed (l.col / r.col) - col = expr.split(".", 1)[1] if "." in expr else expr - col = col.strip().strip('"') + # bare column reference, possibly prefixed (l.col / r."col") + ref = _COLUMN_REF_RE.match(expr) + if ref is None: + raise EntityResolutionError( + f"unsupported expression {expr!r} in pandas blocking; only column references " + "and lower/upper/trim/left/right/substr are supported; use the duckdb backend" + ) + col = _unquote_identifier(ref.group(1)) + if refs is not None: + refs.append(col) return lambda rec: rec.get(col) @@ -863,22 +904,38 @@ def _split_args(text: str) -> list[str]: return [a.strip() for a in args] +def _split_outside_quotes(text: str, pattern: re.Pattern[str]) -> list[str]: + """Split *text* on *pattern*, ignoring matches inside quoted spans.""" + masked = _QUOTED_RE.sub(lambda m: "\0" * len(m.group()), text) + parts: list[str] = [] + start = 0 + for m in pattern.finditer(masked): + parts.append(text[start : m.start()]) + start = m.end() + parts.append(text[start:]) + return parts + + def _parse_blocking( - sql: str, + sql: str, refs: list[str] | None = None ) -> tuple[Callable[[dict[str, Any]], Any], Callable[[dict[str, Any]], Any]]: - """Parse ``a = b [and c = d ...]`` into (left_key_fn, right_key_fn).""" - predicates = re.split(r"\band\b", sql, flags=re.IGNORECASE) + """Parse ``a = b [and c = d ...]`` into (left_key_fn, right_key_fn). + + Each predicate must be exactly `` = `` (see :func:`_make_expr`); + ``OR``, ``<``/``<=``/``>``/``>=``/``!=``/``<>``, literals and other SQL raise + :class:`EntityResolutionError` instead of silently matching nothing. + """ left_fns: list[Callable[[dict[str, Any]], Any]] = [] right_fns: list[Callable[[dict[str, Any]], Any]] = [] - for pred in predicates: - if "=" not in pred: + for pred in _split_outside_quotes(sql, _AND_RE): + sides = _split_outside_quotes(pred, _EQ_RE) + if len(sides) != 2 or sides[0].rstrip()[-1:] in ("<", ">", "!"): raise EntityResolutionError( - f"pandas blocking only supports equality predicates, got {pred!r}; " + f"pandas blocking only supports equality predicates, got {pred.strip()!r}; " "use the duckdb backend for richer SQL" ) - lhs, rhs = pred.split("=", 1) - left_fns.append(_make_expr(lhs)) - right_fns.append(_make_expr(rhs)) + left_fns.append(_make_expr(sides[0], refs)) + right_fns.append(_make_expr(sides[1], refs)) def left_key(rec: dict[str, Any]) -> tuple[Any, ...] | None: vals = tuple(fn(rec) for fn in left_fns) @@ -900,7 +957,16 @@ def _candidates_pandas( n = len(records) pairs: set[tuple[int, int]] = set() for rule in config.blocking_rules: - left_key, right_key = _parse_blocking(rule.sql) + refs: list[str] = [] + left_key, right_key = _parse_blocking(rule.sql, refs) + missing = sorted({c for c in refs if c not in frame.columns}) + if missing: + logger.warning( + "blocking rule %r references column(s) %s not in the frame; " + "it produces no candidate pairs", + rule.sql, + missing, + ) buckets: dict[Any, list[int]] = defaultdict(list) right_keys = [right_key(records[j]) for j in range(n)] for j, rk in enumerate(right_keys): @@ -1140,7 +1206,12 @@ def link_entities( backend=backend, pairs=pairs, clusters=[c for c in clusters if c.size > 1], - runtime_metadata={"link_type": config.link_type}, + runtime_metadata={ + "link_type": config.link_type, + "match_threshold": config.match_threshold, + "clerical_review_threshold": config.clerical_review_threshold, + "scoring": "rule_weighted_probabilistic_linkage", + }, ) out = from_pandas(resolved, left_df) return (out, report) if return_report else out @@ -1177,10 +1248,12 @@ def _link_config( """Build an EntityResolutionConfig from keys + strategy for exact/fuzzy linking.""" rules = _coerce_blocking(blocking) if not rules: + # Quote identifiers so keys like "first name" / "e-mail" stay valid SQL. + quoted = [_quote_identifier(k) for k in keys] if strategy == "exact": - sql = " AND ".join(f"l.{k} = r.{k}" for k in keys) + sql = " AND ".join(f"l.{q} = r.{q}" for q in quoted) else: # fuzzy: block on the first key to bound the candidate space - sql = f"l.{keys[0]} = r.{keys[0]}" + sql = f"l.{quoted[0]} = r.{quoted[0]}" rules = (BlockingRule(sql=sql, description=f"{strategy} block on {list(keys)}"),) comparisons: list[ComparisonLevel] = [] @@ -1269,7 +1342,12 @@ def _external_report( backend="external", pairs=pairs, clusters=[], - runtime_metadata={"link_type": "link_only", "strategy": "external"}, + runtime_metadata={ + "link_type": "link_only", + "strategy": "external", + "match_threshold": match_threshold, + "clerical_review_threshold": review_threshold, + }, ) diff --git a/tests/test_enterprise_entity_resolution.py b/tests/test_enterprise_entity_resolution.py index de162f1..bbd6b00 100644 --- a/tests/test_enterprise_entity_resolution.py +++ b/tests/test_enterprise_entity_resolution.py @@ -12,6 +12,7 @@ from freshdata.enterprise.config import BlockingRule, ComparisonLevel, EntityResolutionConfig from freshdata.enterprise.entity_resolution import ( EntityResolutionError, + _is_missing, jaro_winkler, levenshtein, levenshtein_similarity, @@ -306,3 +307,240 @@ def test_scales_blocking_only_touches_candidates(): ) _out, report = resolve_entities(df, config=cfg) assert report.n_candidate_pairs == 250 # each email shared by exactly 2 rows + + +# --- regressions: blocking parser, missing values, link metadata ------------- + + +def test_duckdb_backend_accepts_non_equality_blocking_sql(): + # #236: the pandas parser must not reject valid DuckDB SQL on the duckdb backend. + pytest.importorskip("duckdb") + df = pd.DataFrame({"id": [1, 2, 3], "name": ["jonathan", "jonathon", "bob"]}) + cfg = EntityResolutionConfig( + backend="duckdb", + comparisons=(ComparisonLevel("name", "jaro_winkler"),), + blocking_rules=( + BlockingRule("jaro_winkler_similarity(l.name, r.name) > 0.8"), + BlockingRule("l.id = r.id + 99"), + ), + ) + _out, report = resolve_entities(df, config=cfg) + assert report.backend == "duckdb" + assert report.n_candidate_pairs == 1 + # Rules the pandas parser cannot read are simply not attributed. + assert report.pairs[0].blocking_rule_ids == () + + +def test_duckdb_backend_still_attributes_equality_rules(): + pytest.importorskip("duckdb") + cfg = _config( + "duckdb", + blocking_rules=( + BlockingRule("l.dob < r.dob"), + BlockingRule("lower(l.email) = lower(r.email)"), + ), + ) + _out, report = resolve_entities(_people(), config=cfg) + pair = next(p for p in report.pairs if {p.left_id, p.right_id} == {1, 2}) + assert pair.blocking_rule_ids == ("block_001",) + + +@pytest.mark.parametrize( + "sql", + [ + "l.id <= r.id", + "l.id >= r.id", + "l.email != r.email", + "l.email <> r.email", + "l.email = r.email OR l.phone = r.phone", + "l.email = r.email or l.phone = r.phone", + "l.phone = '555'", + "l.id = 1", + "l.id = r.id + 1", + "l.email between r.email and r.phone", + "(l.email = r.email)", + "lower() = lower()", + ], +) +def test_pandas_blocking_rejects_non_equality_sql(sql): + # #237: these used to be accepted and silently produce zero candidate pairs. + df = pd.DataFrame({"id": [1, 2], "email": ["a@x.com", "b@x.com"], "phone": ["555", "555"]}) + cfg = EntityResolutionConfig( + backend="pandas", + blocking_rules=(BlockingRule(sql),), + comparisons=(ComparisonLevel("phone"),), + ) + with pytest.raises(EntityResolutionError, match="equality|unsupported"): + resolve_entities(df, config=cfg) + + +_VALID_EQUALITY_RULES = [ + "l.dob = r.dob", + "lower(l.email) = lower(r.email)", + "upper(l.email)=upper(r.email)", + "l.dob = r.dob and substr(lower(l.name), 1, 3) = substr(lower(r.name), 1, 3)", + "l.dob = r.dob AND right(l.email, 5) = right(r.email, 5)", + "lower(left(l.name,4)) = lower(left(r.name,4))", + "trim(l.dob) = trim(r.dob)", + "l.dob == r.dob", + ' l."dob" = r."dob"\n AND l."email" = r."email" ', +] + + +@pytest.mark.parametrize("sql", _VALID_EQUALITY_RULES) +def test_pandas_blocking_equality_rules_match_duckdb(sql): + # #237: the stricter parser must keep every valid equality rule working. + pytest.importorskip("duckdb") + df = _people() + cfg_p = _config("pandas", blocking_rules=(BlockingRule(sql),)) + cfg_d = _config("duckdb", blocking_rules=(BlockingRule(sql),)) + _o, rp = resolve_entities(df, config=cfg_p) + _o, rd = resolve_entities(df, config=cfg_d) + assert rp.n_candidate_pairs > 0 + assert {(p.left_id, p.right_id) for p in rp.pairs} == { + (p.left_id, p.right_id) for p in rd.pairs + } + + +def test_pandas_blocking_quoted_identifiers(): + df = pd.DataFrame( + { + "id": [1, 2, 3], + "rock and roll": ["x", "x", "y"], + 'say "hi"': ["a", "a", "a"], + "a=b": ["k", "k", "k"], + } + ) + sql = ( + 'l."rock and roll" = r."rock and roll" and l."say ""hi""" = r."say ""hi""" ' + 'and l."a=b" = r."a=b"' + ) + cfg = _config( + "pandas", + blocking_rules=(BlockingRule(sql),), + comparisons=(ComparisonLevel("a=b", "exact"),), + ) + _out, report = resolve_entities(df, config=cfg) + assert [(p.left_id, p.right_id) for p in report.pairs] == [(1, 2)] + assert report.pairs[0].blocking_rule_ids == ("block_000",) + + +def test_pandas_blocking_warns_on_unknown_column(caplog): + df = _people() + cfg = _config("pandas", blocking_rules=(BlockingRule("l.emial = r.emial"),)) + with caplog.at_level("WARNING", logger="freshdata.enterprise.entity_resolution"): + _out, report = resolve_entities(df, config=cfg) + assert report.n_candidate_pairs == 0 + assert "emial" in caplog.text + + +@pytest.mark.parametrize( + "value", [None, float("nan"), np.nan, pd.NaT, pd.NA, np.datetime64("NaT", "ns")] +) +def test_is_missing_recognises_scalar_missing_values(value): + assert _is_missing(value) + + +@pytest.mark.parametrize( + "value", ["", "NaT", "nan", b"", 0, 0.0, pd.Timestamp("2020-01-01"), [None]] +) +def test_is_missing_keeps_real_values(value): + assert not _is_missing(value) + + +def test_missing_datetimes_do_not_agree(): + # #238: two NaT values used to score 1.0 and merge the records. + df = pd.DataFrame( + {"id": [1, 2], "zip": ["10001", "10001"], "dob": pd.to_datetime([None, None])} + ) + cfg = EntityResolutionConfig( + backend="pandas", + blocking_rules=(BlockingRule("l.zip = r.zip"),), + comparisons=(ComparisonLevel("dob", "exact"),), + ) + frame, report = resolve_entities(df, config=cfg) + pair = report.pairs[0] + assert pair.comparison_vector == {"dob": 0.0} + assert pair.decision == "non_match" + assert frame["cluster_id"].nunique() == 2 + assert report.clusters == [] + + +@pytest.mark.parametrize("kind", ["exact", "date_distance"]) +def test_missing_datetime_on_one_side_is_missing(kind): + df = pd.DataFrame( + {"id": [1, 2], "zip": ["1", "1"], "dob": pd.to_datetime(["2020-01-01", None])} + ) + cfg = EntityResolutionConfig( + backend="pandas", + blocking_rules=(BlockingRule("l.zip = r.zip"),), + comparisons=(ComparisonLevel("dob", kind, threshold=5.0),), + ) + _frame, report = resolve_entities(df, config=cfg) + assert report.pairs[0].comparison_vector == {"dob": 0.0} + assert "missing" in report.pairs[0].explanation[0].rationale + + +def test_pd_na_values_do_not_agree(): + df = pd.DataFrame( + { + "id": [1, 2], + "zip": ["1", "1"], + "code": pd.array([None, None], dtype="string"), + "n": pd.array([None, None], dtype="Int64"), + } + ) + cfg = EntityResolutionConfig( + backend="pandas", + blocking_rules=(BlockingRule("l.zip = r.zip"),), + comparisons=(ComparisonLevel("code", "exact"), ComparisonLevel("n", "exact")), + ) + _frame, report = resolve_entities(df, config=cfg) + assert report.pairs[0].comparison_vector == {"code": 0.0, "n": 0.0} + assert report.pairs[0].decision == "non_match" + + +@pytest.mark.parametrize( + "column", + [ + pd.to_datetime([None, None]), + pd.array([None, None], dtype="string"), + pd.array([None, None], dtype="Int64"), + ], +) +def test_missing_blocking_keys_do_not_block_together(column): + # #238: NaT / pd.NA blocking keys must not form candidate pairs. + df = pd.DataFrame({"id": [1, 2], "key": column, "name": ["a", "a"]}) + cfg = EntityResolutionConfig( + backend="pandas", + blocking_rules=(BlockingRule("l.key = r.key"),), + comparisons=(ComparisonLevel("name", "exact"),), + ) + _frame, report = resolve_entities(df, config=cfg) + assert report.n_candidate_pairs == 0 + + +def test_link_entities_records_thresholds_for_review_queue(): + # #271: link_entities must record thresholds like resolve_entities does. + base = "a" * 30 + left = pd.DataFrame({"id": ["l1", "l2"], "e": ["x", "y"], "n": [base, base]}) + right = pd.DataFrame( + {"id": ["r1", "r2"], "e": ["x", "y"], "n": ["bb" + base[2:], "bbb" + base[3:]]} + ) + cfg = EntityResolutionConfig( + backend="pandas", + blocking_rules=(BlockingRule("l.e = r.e"),), + comparisons=(ComparisonLevel("n", "levenshtein"),), + match_threshold=0.95, + clerical_review_threshold=0.9, + ) + _o, linked = link_entities(left, right, config=cfg) + _o, resolved = resolve_entities(pd.concat([left, right], ignore_index=True), config=cfg) + assert linked.runtime_metadata == resolved.runtime_metadata + assert linked.runtime_metadata["match_threshold"] == 0.95 + assert linked.runtime_metadata["clerical_review_threshold"] == 0.9 + + def order(rep): + return [round(i.score, 3) for i in fd.build_review_queue(rep).items] + + assert order(linked) == order(resolved) == [0.933, 0.9] diff --git a/tests/test_link.py b/tests/test_link.py index 731767f..6864b8b 100644 --- a/tests/test_link.py +++ b/tests/test_link.py @@ -123,3 +123,67 @@ def test_does_not_mutate_inputs(left, right): fd.link(left, right, keys=["name"], strategy="fuzzy", blocking="l.city = r.city") pd.testing.assert_frame_equal(left, lbefore) pd.testing.assert_frame_equal(right, rbefore) + + +_ODD_KEYS = ["first name", "e-mail", 'say "hi"', "select"] + + +@pytest.mark.parametrize("backend", ["pandas", "duckdb"]) +@pytest.mark.parametrize("key", _ODD_KEYS) +def test_exact_link_with_awkward_key_names(backend, key): + # #266: default blocking SQL must quote identifiers for the duckdb backend. + if backend == "duckdb": + pytest.importorskip("duckdb") + lf = pd.DataFrame({key: ["ann", "bob"], "other": [1, 2]}) + rf = pd.DataFrame({key: ["ann", "cat"], "other": [3, 4]}) + rep = fd.link(lf, rf, keys=[key], backend=backend) + assert rep.backend == backend + assert rep.n_candidate_pairs == 1 + assert rep.n_matches == 1 + + +@pytest.mark.parametrize("backend", ["pandas", "duckdb"]) +def test_multi_key_and_fuzzy_link_with_spaced_keys(backend): + if backend == "duckdb": + pytest.importorskip("duckdb") + lf = pd.DataFrame({"first name": ["Alice", "Bob"], "home city": ["NYC", "LA"]}) + rf = pd.DataFrame({"first name": ["Alice", "Bobby"], "home city": ["NYC", "LA"]}) + exact = fd.link(lf, rf, keys=["first name", "home city"], backend=backend) + assert exact.n_matches == 1 + fuzzy = fd.link( + lf, rf, keys=["home city", "first name"], strategy="fuzzy", threshold=0.8, backend=backend + ) + assert fuzzy.n_candidate_pairs == 2 + assert fuzzy.n_matches >= 1 + + +def test_link_report_records_thresholds(left, right): + # #271: review queues read thresholds from runtime_metadata. + rep = fd.link( + left, + right, + keys=["name"], + strategy="fuzzy", + threshold=0.8, + blocking="l.city = r.city", + review_threshold=0.6, + ) + assert rep.runtime_metadata["match_threshold"] == 0.8 + assert rep.runtime_metadata["clerical_review_threshold"] == 0.6 + + +def test_external_link_report_records_thresholds(left, right): + def adapter(lf, rf, keys): + return [{"left_index": 0, "right_index": 0, "score": 0.9}] + + rep = fd.link( + left, + right, + keys=["name"], + strategy="external", + adapter=adapter, + threshold=0.95, + review_threshold=0.7, + ) + assert rep.runtime_metadata["match_threshold"] == 0.95 + assert rep.runtime_metadata["clerical_review_threshold"] == 0.7