Pairwise string distance comparison

CHANGELOG.md

@@ -7,6 +7,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## Unreleased
 
+### Added
+- Added new `PairwiseStringDistanceFunctionLevel` and `PairwiseStringDistanceFunctionAtThresholds`
+  for comparing array columns using a string similarity on each pair of values
+  ([#2517](https://github.com/moj-analytical-services/splink/pull/2517))
+
 ### Fixed
 
 - Various bugfixes for `debug_mode` ([#2481](https://github.com/moj-analytical-services/splink/pull/2481))

splink/comparison_library.py

@@ -16,6 +16,7 @@
     JaroWinklerAtThresholds,
     LevenshteinAtThresholds,
     NameComparison,
+    PairwiseStringDistanceFunctionAtThresholds,
     PostcodeComparison,
 )
 
@@ -28,6 +29,7 @@
     "JaroAtThresholds",
     "JaroWinklerAtThresholds",
     "DistanceFunctionAtThresholds",
+    "PairwiseStringDistanceFunctionAtThresholds",
     "AbsoluteTimeDifferenceAtThresholds",
     "AbsoluteDateDifferenceAtThresholds",
     "ArrayIntersectAtSizes",

splink/internals/comparison_level_library.py

@@ -610,6 +610,97 @@ def create_label_for_charts(self) -> str:
         )
 
 
+class PairwiseStringDistanceFunctionLevel(ComparisonLevelCreator):
+    def __init__(
+        self,
+        col_name: str | ColumnExpression,
+        distance_function_name: Literal[
+            "levenshtein", "damerau_levenshtein", "jaro_winkler", "jaro"
+        ],
+        distance_threshold: Union[int, float],
+    ):
+        """A comparison level using the *most similar* string distance
+        between any pair of values between arrays in an array column.
+
+        The function given by `distance_function_name` must be one of
+        "levenshtein," "damera_levenshtein," "jaro_winkler," or "jaro."
+
+        Args:
+            col_name (str | ColumnExpression): Input column name
+            distance_function_name (str): the name of the string distance function
+            distance_threshold (Union[int, float]): The threshold to use to assess
+                similarity
+        """
+
+        self.col_expression = ColumnExpression.instantiate_if_str(col_name)
+        self.distance_function_name = validate_categorical_parameter(
+            allowed_values=[
+                "levenshtein",
+                "damerau_levenshtein",
+                "jaro_winkler",
+                "jaro",
+            ],
+            parameter_value=distance_function_name,
+            level_name=self.__class__.__name__,
+            parameter_name="distance_function_name",
+        )
+        self.distance_threshold = validate_numeric_parameter(
+            lower_bound=0,
+            upper_bound=float("inf"),
+            parameter_value=distance_threshold,
+            level_name=self.__class__.__name__,
+            parameter_name="distance_threshold",
+        )
+
+    @unsupported_splink_dialects(["sqlite", "spark", "postgres", "athena"])
+    def create_sql(self, sql_dialect: SplinkDialect) -> str:
+        self.col_expression.sql_dialect = sql_dialect
+        col = self.col_expression
+        distance_function_name_transpiled = {
+            "levenshtein": sql_dialect.levenshtein_function_name,
+            "damerau_levenshtein": sql_dialect.damerau_levenshtein_function_name,
+            "jaro_winkler": sql_dialect.jaro_winkler_function_name,
+            "jaro": sql_dialect.jaro_function_name,
+        }[self.distance_function_name]
+
+        return f"""list_{self._aggregator()}(
+                    list_transform(
+                        flatten(
+                            list_transform(
+                                {col.name_l},
+                                x -> list_transform(
+                                    {col.name_r},
+                                    y -> [x,y]
+                                )
+                            )
+                        ),
+                        pair -> {distance_function_name_transpiled}(pair[1], pair[2])
+                    )
+                ) {self._comparator()} {self.distance_threshold}"""
+
+    def create_label_for_charts(self) -> str:
+        col = self.col_expression
+        return (
+            f"{self._aggregator().title()} `{self.distance_function_name}` "
+            f"distance of '{col.label} "
+            f"{self._comparator()} than {self.distance_threshold}'"
+        )
+
+    def _aggregator(self):
+        return "max" if self._higher_is_more_similar() else "min"
+
+    def _comparator(self):
+        return ">=" if self._higher_is_more_similar() else "<="
+
+    def _higher_is_more_similar(self):
+        return {
+            "levenshtein": False,
+            "damerau_levenshtein": False,
+            "jaro_winkler": True,
+            "jaro": True,
+        }[self.distance_function_name]
+
+
 DateMetricType = Literal["second", "minute", "hour", "day", "month", "year"]
 
 

splink/internals/comparison_library.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from typing import Any, Iterable, List, Optional, Union
+from typing import Any, Iterable, List, Literal, Optional, Union
 
 from splink.internals import comparison_level_library as cll
 from splink.internals.column_expression import ColumnExpression
@@ -345,6 +345,78 @@ def create_output_column_name(self) -> str:
         return self.col_expression.output_column_name
 
 
+class PairwiseStringDistanceFunctionAtThresholds(ComparisonCreator):
+    def __init__(
+        self,
+        col_name: str,
+        distance_function_name: Literal[
+            "levenshtein", "damerau_levenshtein", "jaro_winkler", "jaro"
+        ] = "levenshtein",
+        distance_threshold_or_thresholds: Union[Iterable[int | float], int | float] = [
+            1
+        ],
+    ):
+        """
+        Represents a comparison of the *most similar pair* of values
+        where the first value is in the array data in `col_name` for the first record
+        and the second value is in the array data in `col_name` for the second record.
+        The comparison has three or more levels:
+
+        - Exact match between any pair of values
+        - User-selected string distance function levels at specified thresholds
+        - ...
+        - Anything else
+
+        For example, with distance_threshold_or_thresholds = [1, 3]
+        and distance_function 'levenshtein' the levels are:
+
+        - Exact match between any pair of values
+        - Levenshtein distance between the most similar pair of values <= 1
+        - Levenshtein distance between the most similar pair of values <= 3
+        - Anything else
+
+        Args:
+            col_name (str): The name of the column to compare.
+            distance_function_name (str): the name of the string distance function.
+                Must be one of "levenshtein," "damera_levenshtein," "jaro_winkler,"
+                or "jaro."
+            distance_threshold_or_thresholds (Union[float, list], optional): The
+                threshold(s) to use for the distance function level(s).
+        """
+        thresholds_as_iterable = ensure_is_iterable(distance_threshold_or_thresholds)
+        self.thresholds = [*thresholds_as_iterable]
+        self.distance_function_name = distance_function_name
+        super().__init__(col_name)
+
+    def create_comparison_levels(self) -> List[ComparisonLevelCreator]:
+        return [
+            cll.NullLevel(self.col_expression),
+            # It is assumed that any string distance treats identical
+            # arrays as the most similar
+            cll.ArrayIntersectLevel(self.col_expression, min_intersection=1),
+            *[
+                cll.PairwiseStringDistanceFunctionLevel(
+                    self.col_expression,
+                    distance_threshold=threshold,
+                    distance_function_name=self.distance_function_name,
+                )
+                for threshold in self.thresholds
+            ],
+            cll.ElseLevel(),
+        ]
+
+    def create_description(self) -> str:
+        comma_separated_thresholds_string = ", ".join(map(str, self.thresholds))
+        plural = "s" if len(self.thresholds) > 1 else ""
+        return (
+            f"Pairwise {self.distance_function_name} distance at threshold{plural} "
+            f"{comma_separated_thresholds_string} vs. anything else"
+        )
+
+    def create_output_column_name(self) -> str:
+        return self.col_expression.output_column_name
+
+
 class AbsoluteTimeDifferenceAtThresholds(ComparisonCreator):
     def __init__(
         self,

tests/test_comparison_lib.py

@@ -65,6 +65,60 @@ def test_distance_function_comparison():
             assert sum(df_pred[f"gamma_{col}"] == gamma_val) == expected_count
 
 
+@mark_with_dialects_excluding("sqlite", "spark", "postgres", "athena")
+def test_pairwise_stringdistance_function_comparison(test_helpers, dialect):
+    helper = test_helpers[dialect]
+    db_api = helper.extra_linker_args()["db_api"]
+
+    data = [
+        {"unique_id": 1, "forename": ["Geof"]},
+        {"unique_id": 2, "forename": ["Cally"]},
+        {"unique_id": 3, "forename": ["Saly", "Barey"]},
+        {"unique_id": 4, "forename": ["Geoff", "Barry"]},
+        {"unique_id": 5, "forename": ["Carry"]},
+        {"unique_id": 6, "forename": ["Cally", "Sally"]},
+        {"unique_id": 7, "forename": ["Completely", "Different"]},
+    ]
+
+    df = pd.DataFrame(data)
+
+    settings = {
+        "link_type": "dedupe_only",
+        "comparisons": [
+            cl.PairwiseStringDistanceFunctionAtThresholds(
+                "forename",
+                "damerau_levenshtein",
+                distance_threshold_or_thresholds=[1, 2],
+            ),
+        ],
+    }
+
+    linker = Linker(df, settings, db_api=db_api)
+
+    df_pred = linker.inference.predict().as_pandas_dataframe()
+
+    expected_gamma_counts = {
+        "forename": {
+            # exact match: Cally (1)
+            3: 1,
+            # damerau_levenshtein <= 1 :
+            # Geof[f] (1) + Sal[l]y (1) + Bar[e/r]y (1) +
+            # [B/C]arry (1)
+            # -- note that [C/S]ally is not reached due to the exact match
+            2: 4,
+            # damerau_levenshtein <= 2 :
+            # Ca[ll/rr]y (2) + [S/C]al[l]y (1) + [B/C]ar[e/r]y (1)
+            1: 4,
+            # Else: (7 * 6) / 2 - 1 - 4 - 4
+            0: 12,
+        },
+    }
+
+    for col, expected_counts in expected_gamma_counts.items():
+        for gamma_val, expected_count in expected_counts.items():
+            assert sum(df_pred[f"gamma_{col}"] == gamma_val) == expected_count
+
+
 @mark_with_dialects_excluding()
 def test_set_to_lowercase(test_helpers, dialect):
     helper = test_helpers[dialect]