From 9b06968d2e87ffcd1b4d2add340f313805f55f9a Mon Sep 17 00:00:00 2001 From: Kevin Costner <120246174+kevincostner17@users.noreply.github.com> Date: Tue, 15 Sep 2026 02:54:41 +0530 Subject: [PATCH] fix(er): honour DuckDB blocking SQL, strict pandas blocking, NaT/NA as missing, quoted link keys - resolve_entities/link_entities(backend="duckdb") no longer reject valid DuckDB blocking SQL (e.g. jaro_winkler_similarity(...) > 0.8). The pandas parser still runs to attribute blocking_rule_ids, but its EntityResolutionError is now caught alongside ValueError/KeyError, so rules it cannot read are skipped for attribution as documented. - The pandas blocking parser is now strict. Predicates are split on AND and on "="/"==" outside quoted identifiers and string literals. Each side must be a column reference (l.col, r."quoted col", col) or lower/upper/trim/left/right/substr over one. OR, <, <=, >, >=, !=, <>, literals and arithmetic now raise EntityResolutionError instead of resolving to missing keys and silently producing zero candidate pairs. Quoted identifiers unescape "" and may contain spaces, "=" or "and". A rule that references a column absent from the frame logs a warning. - _is_missing treats every scalar pandas missing value (NaN, NaT, pd.NA, numpy NaT) as missing, not only None and float NaN. Two NaT values no longer score as full agreement and merge records, and NaT/pd.NA blocking keys no longer form candidate pairs. - fd.link builds its default blocking SQL with double-quoted identifiers (embedded quotes escaped as ""), so keys such as "first name" or "e-mail" work on the DuckDB backend. - link_entities records match_threshold, clerical_review_threshold and scoring in runtime_metadata, as resolve_entities does. The external fd.link report records its thresholds too. build_review_queue therefore sorts linkage queues around the configured midpoint instead of 0.75. Closes #236 Closes #237 Closes #238 Closes #266 Closes #271 --- src/freshdata/enterprise/entity_resolution.py | 122 +++++++-- tests/test_enterprise_entity_resolution.py | 238 ++++++++++++++++++ tests/test_link.py | 64 +++++ 3 files changed, 402 insertions(+), 22 deletions(-) diff --git a/src/freshdata/enterprise/entity_resolution.py b/src/freshdata/enterprise/entity_resolution.py index c768e5a4..a2300a47 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 de162f12..bbd6b00a 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 731767f5..6864b8ba 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