From 3e1fa8b1e80a21d5704b4a810d6b11184f6b8017 Mon Sep 17 00:00:00 2001 From: Kevin Costner <120246174+kevincostner17@users.noreply.github.com> Date: Tue, 15 Sep 2026 21:54:43 +0530 Subject: [PATCH 1/3] fix(learning): treat text dtypes as compatible in profile replay The drift gate only treated object as text, so a profile learned on a categorical or string column and replayed on the same values as object (or the reverse) reported mild drift and dropped that column's value maps and hints. A text categorical replayed on numeric data was also not flagged. Compare text-ness with _util.is_text_dtype on the frame's dtype and a name-based equivalent for the stored learned dtype. object, string, Arrow string and text categoricals are interchangeable; text vs non-text is still reported as drift. --- src/freshdata/learning/replay.py | 31 +++++++++- tests/learning/test_replay.py | 101 +++++++++++++++++++++++++++++++ 2 files changed, 130 insertions(+), 2 deletions(-) diff --git a/src/freshdata/learning/replay.py b/src/freshdata/learning/replay.py index 44705dd0..65b24710 100644 --- a/src/freshdata/learning/replay.py +++ b/src/freshdata/learning/replay.py @@ -27,6 +27,7 @@ import pandas as pd +from .._util import is_text_dtype from .profile import LearningProfile, load_profile __all__ = [ @@ -100,6 +101,31 @@ def _profile_schema(profile: LearningProfile) -> dict[str, str]: return {} +#: Arrow text types, as spelled before the ``[pyarrow]`` suffix of a dtype name. +_ARROW_TEXT_TYPES = ("string", "large_string", "string_view") + + +def _is_text_dtype_name(name: str) -> bool: + """:func:`is_text_dtype` for a stored dtype *name* (profiles keep ``str(dtype)``). + + A learned ``"category"`` carries no categories, so it counts as text, as + it does in ``CleaningMemory`` signatures; a frame's categorical is judged + by its real categories. + """ + if name in ("category", "str"): # "str": the pandas 3 default string dtype + return True + try: + return is_text_dtype(pd.api.types.pandas_dtype(name)) + except (TypeError, ValueError, ImportError, NotImplementedError): + pass + base, _, backend = name.partition("[") + if backend != "pyarrow]": + return False + if base.startswith("dictionary set[str]: columns = {r.column for r in profile.rules if r.column} columns.update(profile.value_maps) @@ -136,8 +162,9 @@ def check_profile_drift(df: pd.DataFrame, profile: LearningProfile) -> ProfileRe learned = schema.get(column) if learned is None: continue - actual = str(df[column].dtype) - if learned != actual and (learned == "object") != (actual == "object"): + dtype = df[column].dtype + actual = str(dtype) + if learned != actual and _is_text_dtype_name(learned) != is_text_dtype(dtype): dtype_incompatible.append(f"{column} ({learned} -> {actual})") if dtype_incompatible: reasons.append("dtype changed for: " + ", ".join(dtype_incompatible[:5])) diff --git a/tests/learning/test_replay.py b/tests/learning/test_replay.py index 964d4cf0..8b804d24 100644 --- a/tests/learning/test_replay.py +++ b/tests/learning/test_replay.py @@ -7,6 +7,7 @@ from __future__ import annotations +import copy import dataclasses from types import SimpleNamespace @@ -156,6 +157,106 @@ def test_dtype_flip_marks_column_incompatible(self, orders_profile, new_batch): assert "status" not in gate.compatible_columns +def _as_text_dtype(frame: pd.DataFrame, dtype: object) -> pd.DataFrame: + text_columns = [c for c in frame.columns if frame[c].dtype == object] + return frame.astype(dict.fromkeys(text_columns, dtype)) + + +#: pandas 1.x Arrow string arrays cannot go through the cleaning pipeline at all +#: (regex and datetime comparisons raise), with or without a profile, so there +#: only the drift gate is checked for them. +_ARROW_PIPELINE_OK = int(pd.__version__.split(".")[0]) >= 2 + + +def _text_dtypes() -> list: + dtypes = [ + pytest.param("category", True, id="category"), + pytest.param("string", True, id="string"), + ] + try: + import pyarrow as pa # noqa: PLC0415 - optional dependency + except ImportError: + return dtypes + arrow_ok = _ARROW_PIPELINE_OK + dtypes.append(pytest.param(pd.StringDtype("pyarrow"), arrow_ok, id="string-pyarrow")) + if hasattr(pd, "ArrowDtype"): + large_string = pd.ArrowDtype(pa.large_string()) + dtypes.append(pytest.param(large_string, arrow_ok, id="arrow-large-string")) + return dtypes + + +def _learned_with_schema(profile: LearningProfile, frame: pd.DataFrame) -> LearningProfile: + """Copy of *profile* whose recorded source schema is *frame*'s dtypes.""" + copied = dataclasses.replace(profile, audit_info=copy.deepcopy(profile.audit_info)) + copied.audit_info.alignment["source_schema"] = {str(c): str(frame[c].dtype) for c in frame} + return copied + + +class TestTextDtypeReplay: + """Text dtypes are interchangeable for replay; text vs numeric is still drift.""" + + @pytest.mark.parametrize(("dtype", "pipeline"), _text_dtypes()) + def test_learned_on_object_replays_on_other_text_dtype( + self, orders_profile, new_batch, dtype, pipeline + ): + frame = _as_text_dtype(new_batch, dtype) + gate = check_profile_drift(frame, orders_profile) + assert gate.severity == "none", gate.reasons + assert set(gate.compatible_columns) == {str(c) for c in frame.columns} + if not pipeline: + return + + cleaned, report = fd.clean( + frame, profile=orders_profile, semantic_mode="auto", return_report=True + ) + assert report.profile_replay["severity"] == "none" + assert not any("partially replayed" in w for w in report.warnings) + assert cleaned["status"].astype(object).tolist() == ["delivered", "shipped", "shipped"] + assert _profile_actions(report) + + @pytest.mark.parametrize(("dtype", "pipeline"), _text_dtypes()) + def test_learned_on_text_dtype_replays_on_object( + self, orders_profile, orders_pair, new_batch, dtype, pipeline + ): + messy, clean = orders_pair + messy = _as_text_dtype(messy, dtype) + if pipeline: + profile = learn(messy, clean, key="order_id", min_support=2) + else: + profile = _learned_with_schema(orders_profile, messy) + assert "status" in profile.value_maps + assert profile.audit_info.alignment["source_schema"]["status"] != "object" + + gate = check_profile_drift(new_batch, profile) + assert gate.severity == "none", gate.reasons + assert "status" in gate.compatible_columns + + cleaned, report = fd.clean( + new_batch, profile=profile, semantic_mode="auto", return_report=True + ) + assert report.profile_replay["severity"] == "none" + assert not any("partially replayed" in w for w in report.warnings) + assert list(cleaned["status"]) == ["delivered", "shipped", "shipped"] + + @pytest.mark.parametrize("dtype", ["object", "category"]) + def test_text_to_numeric_still_drifts(self, orders_pair, new_batch, dtype): + messy, clean = orders_pair + profile = learn(_as_text_dtype(messy, dtype), clean, key="order_id", min_support=2) + numeric = new_batch.copy() + numeric["status"] = [1, 2, 3] + gate = check_profile_drift(numeric, profile) + assert gate.severity == "mild" + assert "status" not in gate.compatible_columns + assert any("status" in r for r in gate.reasons) + + def test_numeric_categorical_is_not_text(self, orders_profile, new_batch): + frame = new_batch.copy() + frame["status"] = pd.Series([1, 2, 3], dtype="category") + gate = check_profile_drift(frame, orders_profile) + assert gate.severity == "mild" + assert "status" not in gate.compatible_columns + + class TestFolding: def test_fold_respects_existing_options(self, orders_profile): gate = ProfileReplayGate( From 6e2c0dd74ec526cdf152d7f439493fe8bded5ca4 Mon Sep 17 00:00:00 2001 From: Kevin Costner <120246174+kevincostner17@users.noreply.github.com> Date: Tue, 15 Sep 2026 21:54:43 +0530 Subject: [PATCH 2/3] fix(learning): merge list and scalar memory value patterns union_min_precision and error_on_conflict assumed every entry of the embedded memory's value_patterns was a {raw: clean} mapping and raised ValueError when memory held the semantic_repairs list. Union list patterns with de-duplication. A semantic repair both sides propose differently is dropped and recorded under union_min_precision and raises ProfileMergeError under error_on_conflict; differing scalar patterns are handled the same way. prefer_self/prefer_other still copy the preferred side's memory. --- src/freshdata/learning/merge.py | 122 ++++++++++++++++++++++++++---- tests/learning/test_merge_diff.py | 108 ++++++++++++++++++++++++++ 2 files changed, 216 insertions(+), 14 deletions(-) diff --git a/src/freshdata/learning/merge.py b/src/freshdata/learning/merge.py index 50723178..b8d662c3 100644 --- a/src/freshdata/learning/merge.py +++ b/src/freshdata/learning/merge.py @@ -15,6 +15,7 @@ from __future__ import annotations import copy +import json from collections.abc import Mapping from dataclasses import dataclass, field from datetime import datetime, timezone @@ -247,6 +248,11 @@ def merge_profiles(self_profile: Any, other_profile: Any, *, strategy: Strategy) ) conflicts.extend(map_conflicts_found) + memory, memory_conflicts = _merge_memory_checked( + self_profile.memory, other_profile.memory, strategy + ) + conflicts.extend(memory_conflicts) + if strategy == "error_on_conflict" and conflicts: raise ProfileMergeError( "profiles conflict; refusing to merge:\n - " + "\n - ".join(conflicts) @@ -255,7 +261,6 @@ def merge_profiles(self_profile: Any, other_profile: Any, *, strategy: Strategy) notes.extend(f"conflict ({_resolution_word(strategy)}): {c}" for c in conflicts) examples = _merge_examples(self_profile.examples, other_profile.examples) - memory = _merge_memory(self_profile.memory, other_profile.memory, strategy) privacy_mode = _merged_privacy(self_profile, other_profile, strategy) base = other_profile if strategy == "prefer_other" else self_profile @@ -469,21 +474,110 @@ def _merge_memory(a: Any, b: Any, strategy: str) -> Any: or any later change to the merged profile, rewrite the parents' replay behaviour and saved bytes. """ + return _merge_memory_checked(a, b, strategy)[0] + + +def _merge_memory_checked(a: Any, b: Any, strategy: str) -> tuple[Any, list[str]]: + """:func:`_merge_memory` plus the conflicts found in non-mapping patterns. + + ``prefer_self``/``prefer_other`` copy the preferred side's memory whole, so + they never conflict. The union keeps self's memory and folds in the other + side's patterns: per-column ``{raw: clean}`` mappings drop a disagreeing + raw value (the value-map merge already reports those), while list patterns + such as ``semantic_repairs`` are unioned and de-duplicated, and a repair + both sides propose differently is dropped and reported as a conflict. + """ if a is None: - return copy.deepcopy(b) + return copy.deepcopy(b), [] if b is None or strategy == "prefer_self": - return copy.deepcopy(a) + return copy.deepcopy(a), [] if strategy == "prefer_other": - return copy.deepcopy(b) - # union: keep self's memory but fold in non-conflicting value patterns. + return copy.deepcopy(b), [] merged = copy.deepcopy(a) - merged_patterns = {c: dict(p) for c, p in merged.value_patterns.items()} - for column, patterns in b.value_patterns.items(): - target = merged_patterns.setdefault(column, {}) - for raw, clean in patterns.items(): - if raw in target and str(target[raw]) != str(clean): - del target[raw] # conflicting mapping: drop from replay - elif raw not in target: - target[raw] = copy.deepcopy(clean) + conflicts: list[str] = [] + merged_patterns: dict[str, Any] = dict(merged.value_patterns) + for key, theirs in b.value_patterns.items(): + if key not in merged_patterns: + merged_patterns[key] = copy.deepcopy(theirs) + continue + ours = merged_patterns[key] + if isinstance(ours, Mapping) and isinstance(theirs, Mapping): + target = dict(ours) + for raw, clean in theirs.items(): + if raw in target and str(target[raw]) != str(clean): + del target[raw] # conflicting mapping: drop from replay + elif raw not in target: + target[raw] = copy.deepcopy(clean) + merged_patterns[key] = target + elif isinstance(ours, list) and isinstance(theirs, list): + merged_patterns[key], found = _merge_pattern_lists(str(key), ours, theirs) + conflicts.extend(found) + elif _canonical(ours) != _canonical(theirs): + del merged_patterns[key] + conflicts.append( + f"memory value_patterns '{key}': {type(ours).__name__} vs " + f"{type(theirs).__name__} values differ" + ) merged.value_patterns = merged_patterns - return merged + return merged, conflicts + + +_REPAIR_IDENTITY_FIELDS = ("column", "issue_type", "expert", "raw_value") + + +def _canonical(value: object) -> str: + try: + return json.dumps(value, sort_keys=True, default=repr) + except (TypeError, ValueError): # mixed-type keys cannot be sorted + return repr(value) + + +def _repair_identity(entry: object) -> tuple[str, str] | None: + """``(identity, proposal)`` for a repair record, or None for any other entry.""" + if not isinstance(entry, Mapping) or "proposed_value" not in entry: + return None + identity = _canonical({f: entry.get(f) for f in _REPAIR_IDENTITY_FIELDS}) + proposal = _canonical([entry.get("proposed_value"), entry.get("proposed_type", "str")]) + return identity, proposal + + +def _merge_pattern_lists( + key: str, ours: list[Any], theirs: list[Any] +) -> tuple[list[Any], list[str]]: + """Union of two list patterns (e.g. ``semantic_repairs``), first copy wins. + + Repair records are the same entry when column, issue type, expert and raw + value match. When the two sides propose different values for one entry, + it is dropped from both and reported; other entries de-duplicate by value. + """ + proposals: tuple[dict[str, set[str]], dict[str, set[str]]] = ({}, {}) + labels: dict[str, str] = {} + for side, entries in zip(proposals, (ours, theirs)): + for entry in entries: + found = _repair_identity(entry) + if found is None: + continue + side.setdefault(found[0], set()).add(found[1]) + labels.setdefault( + found[0], f"column {entry.get('column')!r}, issue {entry.get('issue_type')!r}" + ) + conflicting = { + identity + for identity in proposals[0].keys() & proposals[1].keys() + if proposals[0][identity] != proposals[1][identity] + } + conflicts = [ + f"memory value_patterns '{key}': {labels[identity]} proposes different values" + for identity in sorted(conflicting, key=lambda i: labels[i]) + ] + merged: list[Any] = [] + seen: set[str] = set() + for entry in (*ours, *theirs): + found = _repair_identity(entry) + if found is not None and found[0] in conflicting: + continue + marker = _canonical(found) if found is not None else _canonical(entry) + if marker not in seen: + seen.add(marker) + merged.append(copy.deepcopy(entry)) + return merged, conflicts diff --git a/tests/learning/test_merge_diff.py b/tests/learning/test_merge_diff.py index c27568c6..aaa46f2a 100644 --- a/tests/learning/test_merge_diff.py +++ b/tests/learning/test_merge_diff.py @@ -2,6 +2,8 @@ from __future__ import annotations +import copy + import pytest from freshdata.learning import learn, load_profile, save_profile @@ -93,6 +95,112 @@ def test_unknown_strategy_rejected(self, base_profile): merge_profiles(base_profile, base_profile, strategy="bogus") +_STRATEGIES = ("union_min_precision", "prefer_self", "prefer_other", "error_on_conflict") + + +def _repair(raw_value, proposed_value, **extra): + return { + "column": "status", + "issue_type": "case", + "expert": "case_normalization", + "raw_value": raw_value, + "proposed_value": proposed_value, + "proposed_type": "str", + **extra, + } + + +def _with_patterns(profile, **patterns): + """Deep copy of *profile* whose embedded memory carries extra value patterns.""" + copied = copy.deepcopy(profile) + assert copied.memory is not None + copied.memory.value_patterns.update(copy.deepcopy(patterns)) + return copied + + +def _notes(profile): + return list(profile.audit_info.notes) + + +class TestMergeMemoryListPatterns: + """``semantic_repairs`` and other non-mapping memory patterns merge without crashing.""" + + @pytest.mark.parametrize("strategy", _STRATEGIES) + def test_semantic_repairs_union_deduplicated(self, base_profile, strategy): + shared, ours, theirs = _repair("SHIPPED", "shipped"), _repair("A", "a"), _repair("B", "b") + a = _with_patterns(base_profile, semantic_repairs=[shared, ours]) + b = _with_patterns(base_profile, semantic_repairs=[dict(shared, confidence=0.5), theirs]) + + merged = merge_profiles(a, b, strategy=strategy) + + repairs = merged.memory.value_patterns["semantic_repairs"] + expected = { + "prefer_self": [shared, ours], + "prefer_other": [dict(shared, confidence=0.5), theirs], + }.get(strategy, [shared, ours, theirs]) + assert repairs == expected + assert not any("semantic_repairs" in n for n in _notes(merged)) + # The parents are untouched. + assert a.memory.value_patterns["semantic_repairs"] == [shared, ours] + assert repairs is not a.memory.value_patterns["semantic_repairs"] + + @pytest.mark.parametrize("strategy", _STRATEGIES) + def test_semantic_repairs_conflict(self, base_profile, strategy): + agreed = _repair("SHIPPED", "shipped") + ours, theirs = _repair("Deliverd", "delivered"), _repair("Deliverd", "returned") + a = _with_patterns(base_profile, semantic_repairs=[ours, agreed]) + b = _with_patterns(base_profile, semantic_repairs=[theirs, agreed]) + + if strategy == "error_on_conflict": + with pytest.raises(ProfileMergeError, match="semantic_repairs"): + merge_profiles(a, b, strategy=strategy) + return + merged = merge_profiles(a, b, strategy=strategy) + proposals = { + r["raw_value"]: r["proposed_value"] + for r in merged.memory.value_patterns["semantic_repairs"] + } + if strategy == "union_min_precision": + assert proposals == {"SHIPPED": "shipped"} + assert any("semantic_repairs" in n for n in _notes(merged)) + else: + winner = "delivered" if strategy == "prefer_self" else "returned" + assert proposals == {"Deliverd": winner, "SHIPPED": "shipped"} + + @pytest.mark.parametrize("strategy", _STRATEGIES) + def test_semantic_repairs_from_saved_memory(self, base_profile, strategy, tmp_path): + a = _with_patterns(base_profile, semantic_repairs=[_repair("SHIPPED", "shipped")]) + b = _with_patterns(base_profile, semantic_repairs=[_repair("Deliverd", "delivered")]) + path_a, path_b = tmp_path / "a.fdprofile", tmp_path / "b.fdprofile" + save_profile(a, path_a) + save_profile(b, path_b) + + merged = merge_profiles(load_profile(path_a), load_profile(path_b), strategy=strategy) + + raw_values = [r["raw_value"] for r in merged.memory.value_patterns["semantic_repairs"]] + expected = {"prefer_self": ["SHIPPED"], "prefer_other": ["Deliverd"]} + assert raw_values == expected.get(strategy, ["SHIPPED", "Deliverd"]) + + @pytest.mark.parametrize("strategy", _STRATEGIES) + def test_other_non_mapping_patterns(self, base_profile, strategy): + a = _with_patterns(base_profile, tags=["x", "y"], version=1, note="same") + b = _with_patterns(base_profile, tags=["y", "z"], version=2, note="same") + + if strategy == "error_on_conflict": + with pytest.raises(ProfileMergeError, match="version"): + merge_profiles(a, b, strategy=strategy) + return + patterns = merge_profiles(a, b, strategy=strategy).memory.value_patterns + if strategy == "union_min_precision": + assert patterns["tags"] == ["x", "y", "z"] + assert "version" not in patterns + assert patterns["note"] == "same" + else: + side = a if strategy == "prefer_self" else b + assert patterns["tags"] == side.memory.value_patterns["tags"] + assert patterns["version"] == side.memory.value_patterns["version"] + + class TestMergedProfileIntegrity: def test_merged_id_differs_and_roundtrips(self, base_profile, conflicting_profile, tmp_path): merged = merge_profiles(base_profile, conflicting_profile, strategy="union_min_precision") From 875d0baa8262ad4bb2275437efcec5eafc5761fd Mon Sep 17 00:00:00 2001 From: Kevin Costner <120246174+kevincostner17@users.noreply.github.com> Date: Tue, 15 Sep 2026 21:54:43 +0530 Subject: [PATCH 3/3] docs(learning): note that masked tokens are per profile The masking salt comes from the frame signature (row count, dtypes and a head sample), so the same values can mask to different tokens in different profiles and merged profiles cannot match each other's masked entries. --- src/freshdata/learning/privacy.py | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/src/freshdata/learning/privacy.py b/src/freshdata/learning/privacy.py index 0463741b..e44443e6 100644 --- a/src/freshdata/learning/privacy.py +++ b/src/freshdata/learning/privacy.py @@ -14,6 +14,14 @@ Masking is deterministic per profile: the salt is derived from the training pair's dataset signature so re-learning the same pair yields identical tokens (and therefore identical profile hashes). + +Tokens are per profile, not global. The dataset signature covers the messy +frame's row count, column names and dtypes, and a sample of its first rows, +so the same values learned from a frame that differs in any of those (for +example a ``category`` column instead of ``object``, or more rows) get +different tokens. Masked tokens are therefore only comparable within one +profile: a merged profile cannot match one parent's masked entries against +the other's, so they are neither combined nor reported as conflicts. """ from __future__ import annotations