Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
122 changes: 108 additions & 14 deletions src/freshdata/learning/merge.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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)
Expand All @@ -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
Expand Down Expand Up @@ -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
8 changes: 8 additions & 0 deletions src/freshdata/learning/privacy.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
31 changes: 29 additions & 2 deletions src/freshdata/learning/replay.py
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@

import pandas as pd

from .._util import is_text_dtype
from .profile import LearningProfile, load_profile

__all__ = [
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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]))
Expand Down
108 changes: 108 additions & 0 deletions tests/learning/test_merge_diff.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,8 @@

from __future__ import annotations

import copy

import pytest

from freshdata.learning import learn, load_profile, save_profile
Expand Down Expand Up @@ -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")
Expand Down
Loading
Loading