fix(learning): text dtypes compatible in replay; merge list memory patterns (#419)

* 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.

* 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.

* 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.

Author: kevincostner17

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

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

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<values="):
+        base = base[len("dictionary<values=") :].split(",", 1)[0]
+    return base in _ARROW_TEXT_TYPES
+
+
 def _referenced_columns(profile: LearningProfile) -> 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]))

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