From 352a6336b0ee94cdb23bfca38a4081ad981c60bd Mon Sep 17 00:00:00 2001
From: mborodii-prog <mborodii@binariks.com>
Date: Tue, 25 Aug 2026 13:44:39 +0300
Subject: [PATCH 1/4] Add robust similarity metrics and normalization to
 compare.text

---
 requirements.txt                       |   3 +
 tests/recipes/wrangles/test_compare.py | 413 +++++++++++++++++++++++++
 wrangles/compare.py                    |  86 ++++-
 wrangles/recipe_wrangles/compare.py    |  69 ++++-
 4 files changed, 548 insertions(+), 23 deletions(-)

diff --git a/requirements.txt b/requirements.txt
index 546d3c9e3..f0b013c5b 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -8,6 +8,9 @@ numexpr
 polars==1.33.0
 pyarrow
 
+# Fuzzy text similarity
+rapidfuzz>=3.0,<4.0
+
 # Recipe engine & templating
 jinja2
 lorem
diff --git a/tests/recipes/wrangles/test_compare.py b/tests/recipes/wrangles/test_compare.py
index 47b50b6c7..50830f90a 100644
--- a/tests/recipes/wrangles/test_compare.py
+++ b/tests/recipes/wrangles/test_compare.py
@@ -1,5 +1,7 @@
 import wrangles
 import pandas as pd
+import numpy as np
+import logging
 
 
 class TestCompareList:
@@ -908,3 +910,414 @@ def test_compare_text_case_insensitive_overlap(self):
             dataframe=data,
         )
         assert df["output"][1] == "ANOTHER LINE THAT IS ALSO IN *****CA*S*"
+
+    def test_compare_text_case_sensitive_deprecation_warning(self, caplog):
+        """
+        Supplying case_sensitive should emit a deprecation warning; omitting it should not.
+        """
+        data = pd.DataFrame({"col1": ["Mario"], "col2": ["mario"]})
+
+        recipe_with_param = """
+        wrangles:
+        - compare.text:
+            input:
+            - col1
+            - col2
+            output: output
+            method: difference
+            case_sensitive: true
+        """
+        with caplog.at_level(logging.WARNING):
+            caplog.clear()
+            wrangles.recipe.run(recipe=recipe_with_param, dataframe=data)
+        assert any("case_sensitive" in record.message for record in caplog.records)
+
+        recipe_without_param = """
+        wrangles:
+        - compare.text:
+            input:
+            - col1
+            - col2
+            output: output
+            method: difference
+        """
+        with caplog.at_level(logging.WARNING):
+            caplog.clear()
+            wrangles.recipe.run(recipe=recipe_without_param, dataframe=data)
+        assert not any("case_sensitive" in record.message for record in caplog.records)
+
+    def test_compare_text_case_sensitive_true_false_omitted_equivalent(self):
+        """
+        case_sensitive: true, false, and omitted must all produce identical results,
+        since it is deprecated and ignored - comparisons are always case-insensitive.
+        """
+        data = pd.DataFrame(
+            {
+                "col1": ["THIS IS IN ALL CAPS"],
+                "col2": ["this is in all lowercase"],
+            }
+        )
+        outputs = {}
+        for label, case_sensitive_line in [
+            ("true", "case_sensitive: true"),
+            ("false", "case_sensitive: false"),
+            ("omitted", ""),
+        ]:
+            recipe = f"""
+            wrangles:
+            - compare.text:
+                input:
+                - col1
+                - col2
+                output: output
+                method: difference
+                {case_sensitive_line}
+            """
+            df = wrangles.recipe.run(recipe=recipe, dataframe=data)
+            outputs[label] = df["output"][0]
+
+        assert outputs["true"] == outputs["false"] == outputs["omitted"] == "lowercase"
+
+    def test_compare_text_default_preserves_title_case(self):
+        """
+        Regression: forcing case-insensitive matching by default must not lowercase
+        legacy difference/intersection output - original casing should be preserved.
+        """
+        data = pd.DataFrame(
+            {
+                "col1": ["Mario Oak Wood White Marble Top Bookshelf"],
+                "col2": ["Mario Pine Wood Black Marble Bottom Bookshelf"],
+            }
+        )
+        recipe = """
+        wrangles:
+        - compare.text:
+            input:
+            - col1
+            - col2
+            output: output
+            method: difference
+        """
+        df = wrangles.recipe.run(recipe=recipe, dataframe=data)
+        assert df["output"][0] == "Pine Black Bottom"
+
+    def test_compare_text_overlap_quoted_decimal_places(self):
+        """
+        Regression: a quoted decimal_places value must not crash overlap.
+        """
+        data = pd.DataFrame({"col1": ["Mario"], "col2": ["Martio"]})
+        recipe = """
+        wrangles:
+        - compare.text:
+            input:
+            - col1
+            - col2
+            output: output
+            method: overlap
+            include_ratio: true
+            decimal_places: "2"
+        """
+        df = wrangles.recipe.run(recipe=recipe, dataframe=data)
+        assert df["output"][0] == ["Mar*io", 0.91]
+
+
+class TestCompareTextSimilarity:
+    """
+    Test compare.text method: similarity
+    """
+
+    def _run(self, data, metric=None, decimal_places=None, extra=""):
+        metric_line = f"metric: {metric}" if metric else ""
+        decimal_places_line = (
+            f"decimal_places: {decimal_places}" if decimal_places is not None else ""
+        )
+        recipe = f"""
+        wrangles:
+        - compare.text:
+            input:
+            - col1
+            - col2
+            output: output
+            method: similarity
+            {metric_line}
+            {decimal_places_line}
+            {extra}
+        """
+        return wrangles.recipe.run(recipe=recipe, dataframe=data)
+
+    def test_default_metric_is_token_sort(self):
+        """
+        Omitting metric should behave the same as explicitly requesting token_sort.
+        """
+        data = pd.DataFrame(
+            {
+                "col1": ["M8 stainless bolt"],
+                "col2": ["M8 stainless bolt 20mm"],
+            }
+        )
+        df_default = self._run(data)
+        df_token_sort = self._run(data, metric="token_sort")
+        assert df_default["output"][0] == df_token_sort["output"][0]
+
+    def test_token_sort_reordered_tokens_scores_one(self):
+        """
+        token_sort ignores token order.
+        """
+        data = pd.DataFrame(
+            {
+                "col1": ["M8 stainless bolt 20mm"],
+                "col2": ["20mm bolt stainless M8"],
+            }
+        )
+        df = self._run(data, metric="token_sort")
+        assert df["output"][0] == 1.0
+
+    def test_token_sort_penalizes_missing_and_extra_tokens(self):
+        """
+        token_sort scores below 1.0 when tokens are missing/extra.
+        """
+        data = pd.DataFrame(
+            {
+                "col1": ["M8 stainless bolt"],
+                "col2": ["M8 stainless bolt 20mm"],
+            }
+        )
+        df = self._run(data, metric="token_sort")
+        assert 0.0 <= df["output"][0] < 1.0
+
+    def test_token_set_subset_containment_scores_one(self):
+        """
+        token_set can score 1.0 when a shorter token set is fully contained in a longer one.
+        """
+        data = pd.DataFrame(
+            {
+                "col1": ["M8 stainless bolt"],
+                "col2": ["M8 stainless bolt 20mm"],
+            }
+        )
+        df = self._run(data, metric="token_set")
+        assert df["output"][0] == 1.0
+
+    def test_conflicting_attribute_reduces_order_independent_scores(self):
+        """
+        A conflicting attribute (red vs blue) must reduce both order-independent metrics.
+        """
+        data = pd.DataFrame(
+            {
+                "col1": ["red steel bolt"],
+                "col2": ["blue steel bolt"],
+            }
+        )
+        df_token_sort = self._run(data, metric="token_sort")
+        df_token_set = self._run(data, metric="token_set")
+        assert df_token_sort["output"][0] < 1.0
+        assert df_token_set["output"][0] < 1.0
+
+    def test_damerau_levenshtein_adjacent_transposition(self):
+        """
+        damerau_levenshtein should recognize an adjacent transposition as a single edit.
+        """
+        data = pd.DataFrame({"col1": ["smtih"], "col2": ["smith"]})
+        df = self._run(data, metric="damerau_levenshtein")
+        # a single transposition on a 5-character word -> normalized similarity 0.8
+        assert df["output"][0] == 0.8
+
+    def test_damerau_levenshtein_insertion_deletion_substitution(self):
+        """
+        damerau_levenshtein should score each single-edit-distance operation
+        (insertion, deletion, substitution) as one edit, distinct from a no-op match.
+        """
+        data = pd.DataFrame(
+            {
+                "col1": ["bolt", "bolt", "bolt"],
+                "col2": ["boltx", "bol", "belt"],
+            },
+            index=["insertion", "deletion", "substitution"],
+        )
+        df = self._run(data, metric="damerau_levenshtein")
+        # a single insertion on a 4/5-character word -> normalized similarity 0.8
+        assert df["output"]["insertion"] == 0.8
+        # a single deletion on a 4-character word -> normalized similarity 0.75
+        assert df["output"]["deletion"] == 0.75
+        # a single substitution on a 4-character word -> normalized similarity 0.75
+        assert df["output"]["substitution"] == 0.75
+
+    def test_duplicate_tokens_token_sort_vs_token_set(self):
+        """
+        token_sort retains duplicate tokens (penalized); token_set is duplicate-insensitive.
+        """
+        data = pd.DataFrame(
+            {
+                "col1": ["bolt bolt steel"],
+                "col2": ["bolt steel"],
+            }
+        )
+        df_token_sort = self._run(data, metric="token_sort")
+        df_token_set = self._run(data, metric="token_set")
+        assert df_token_sort["output"][0] < 1.0
+        assert df_token_set["output"][0] == 1.0
+
+    def test_anagram_does_not_score_perfect(self):
+        """
+        Character anagrams with no matching tokens must not receive a perfect score.
+        """
+        data = pd.DataFrame({"col1": ["tide"], "col2": ["diet"]})
+        for metric in ["token_sort", "damerau_levenshtein", "token_set"]:
+            df = self._run(data, metric=metric)
+            assert df["output"][0] < 1.0
+
+    def test_punctuation_does_not_equate_identifiers(self):
+        """
+        Normalization must not decide that AB-12 and AB12 are equivalent.
+        """
+        data = pd.DataFrame({"col1": ["AB-12"], "col2": ["AB12"]})
+        for metric in ["token_sort", "damerau_levenshtein", "token_set"]:
+            df = self._run(data, metric=metric)
+            assert df["output"][0] < 1.0
+
+    def test_unicode_and_case_normalization_exact_match(self):
+        """
+        Unicode compatibility normalization and case folding should make these equal.
+        """
+        data = pd.DataFrame(
+            {
+                "col1": ["Mario Oak Wood"],
+                "col2": ["mario   oak-wood"],
+            }
+        )
+        for metric in ["token_sort", "damerau_levenshtein", "token_set"]:
+            df = self._run(data, metric=metric)
+            assert df["output"][0] == 1.0
+
+    def test_symmetry(self):
+        """
+        Swapping the two inputs must not change the score, for every metric.
+        """
+        data_ab = pd.DataFrame(
+            {"col1": ["red steel bolt M8"], "col2": ["M8 steel bolt blue"]}
+        )
+        data_ba = pd.DataFrame(
+            {"col1": ["M8 steel bolt blue"], "col2": ["red steel bolt M8"]}
+        )
+        for metric in ["token_sort", "damerau_levenshtein", "token_set"]:
+            df_ab = self._run(data_ab, metric=metric)
+            df_ba = self._run(data_ba, metric=metric)
+            assert df_ab["output"][0] == df_ba["output"][0]
+
+    def test_scores_are_bounded_between_zero_and_one(self):
+        """
+        Every produced score must be between 0.0 and 1.0 inclusive.
+        """
+        data = pd.DataFrame(
+            {
+                "col1": ["Mario", "red steel bolt", "M8 stainless bolt 20mm", "smtih", "tide"],
+                "col2": ["Luigi", "blue steel bolt", "20mm bolt stainless M8", "smith", "diet"],
+            }
+        )
+        for metric in ["token_sort", "damerau_levenshtein", "token_set"]:
+            df = self._run(data, metric=metric)
+            assert all(0.0 <= score <= 1.0 for score in df["output"].tolist())
+
+    def test_missing_values_return_null_not_string(self):
+        """
+        wrangles.compare.similarity() (the library function) should return None
+        for null/blank inputs, not the strings "nan"/"None", and should never
+        stringify a missing value before checking for it.
+        """
+        from wrangles import compare as compare_lib
+
+        results = compare_lib.similarity(
+            input=[
+                ["Mario", "Mario"],
+                [None, "Luigi"],
+                [np.nan, "Peach"],
+                ["Bowser", None],
+                ["", ""],
+            ],
+            metric="token_sort",
+        )
+        assert results[0] == 1.0
+        assert results[1] is None
+        assert results[2] is None
+        assert results[3] is None
+        assert results[4] is None
+
+    def test_missing_values_via_recipe_are_not_stringified(self):
+        """
+        Regression: missing values must never surface as the literal strings
+        "nan"/"None" in the output (the recipe engine blanks nulls to '' after
+        every wrangle, so that - not "nan"/"None" - is the expected surface value).
+        """
+        data = pd.DataFrame(
+            {
+                "col1": ["Mario", None, np.nan, "Bowser"],
+                "col2": ["Mario", "Luigi", "Peach", None],
+            }
+        )
+        df = self._run(data, metric="token_sort")
+        assert df["output"][0] == 1.0
+        for value in df["output"][1:]:
+            assert value not in ("nan", "None")
+
+    def test_long_repetitive_string_returns_bounded_score(self):
+        """
+        Long repetitive strings should still return a single bounded score.
+        """
+        data = pd.DataFrame(
+            {
+                "col1": [" ".join(["bolt"] * 500)],
+                "col2": [" ".join(["bolt"] * 499 + ["nut"])],
+            }
+        )
+        df = self._run(data, metric="token_sort")
+        assert 0.0 <= df["output"][0] <= 1.0
+
+    def test_similarity_requires_exactly_two_columns(self):
+        """
+        method: similarity requires exactly two input columns.
+        """
+        data = pd.DataFrame({"col1": ["a"], "col2": ["b"], "col3": ["c"]})
+        recipe = """
+        wrangles:
+        - compare.text:
+            input:
+            - col1
+            - col2
+            - col3
+            output: output
+            method: similarity
+        """
+        try:
+            wrangles.recipe.run(recipe=recipe, dataframe=data)
+            assert False, "Should raise an error if more than two columns are passed"
+        except Exception:
+            pass
+
+    def test_similarity_invalid_metric_raises(self):
+        """
+        An unknown metric should raise an error.
+        """
+        data = pd.DataFrame({"col1": ["a"], "col2": ["b"]})
+        recipe = """
+        wrangles:
+        - compare.text:
+            input:
+            - col1
+            - col2
+            output: output
+            method: similarity
+            metric: not_a_real_metric
+        """
+        try:
+            wrangles.recipe.run(recipe=recipe, dataframe=data)
+            assert False, "Should raise an error for an invalid metric"
+        except Exception:
+            pass
+
+    def test_quoted_decimal_places(self):
+        """
+        Regression: a quoted decimal_places value must not crash similarity.
+        """
+        data = pd.DataFrame({"col1": ["smtih"], "col2": ["smith"]})
+        df = self._run(data, metric="damerau_levenshtein", decimal_places='"2"')
+        assert df["output"][0] == 0.8
diff --git a/wrangles/compare.py b/wrangles/compare.py
index 38a840263..2e415bdd3 100644
--- a/wrangles/compare.py
+++ b/wrangles/compare.py
@@ -8,6 +8,8 @@
 import unicodedata
 from typing import Tuple
 
+from rapidfuzz import fuzz as _fuzz, distance as _distance
+
 def normalize_alphanum(text: str) -> str:
     """
     Normalizes text using Python's built-in unicodedata library.
@@ -142,15 +144,14 @@ def contrast(input: list, type: str ='difference', char: str = ' ', case_sensiti
     _logging.debug(f": Comparing {len(input)} records :: type :: {type}, case_sensitive :: {case_sensitive}")
     results = []
     for row in input:
-            
+
         if not row:
             return ""
 
-        # Generate ordered words for each string
-        if not case_sensitive and type != 'intersection':
-            ordered_words_list = [_ordered_words(x.lower(), char) for x in row]
-        else:
-            ordered_words_list = [_ordered_words(x, char) for x in row]
+        # Always keep original casing for the word lists - case-insensitive
+        # matching (when requested) is applied only at comparison time below,
+        # so the output preserves the source casing either way.
+        ordered_words_list = [_ordered_words(x, char) for x in row]
 
         # Initialize intersection with the words of the first string
         common_words = _OrderedDict(ordered_words_list[0])
@@ -177,7 +178,11 @@ def contrast(input: list, type: str ='difference', char: str = ' ', case_sensiti
                     if word not in all_words_flat:
                         all_words_flat[word] = None
 
-            difference = " ".join(k for k in all_words_flat if k not in common_words)
+            if not case_sensitive:
+                common_words_lower = set(w.lower() for w in common_words)
+                difference = " ".join(k for k in all_words_flat if k.lower() not in common_words_lower)
+            else:
+                difference = " ".join(k for k in all_words_flat if k not in common_words)
             results.append(difference)
 
     return results
@@ -230,10 +235,10 @@ def overlap(
         if matcher.ratio() == 1.0:
             if include_ratio:
                 results.append(
-                    [exact_match if exact_match else a_str, 1]
+                    [exact_match if exact_match else a_original, 1]
                 )
             else:
-                results.append(exact_match if exact_match else a_str)
+                results.append(exact_match if exact_match else a_original)
             continue
 
         result = []
@@ -271,6 +276,69 @@ def overlap(
 
     return results
 
+def normalize_similarity_text(text) -> str:
+    """
+    Standard preprocessing pipeline for method: similarity.
+
+    Applies Unicode compatibility normalization, Unicode-aware case folding,
+    converts punctuation/separators to spaces (without merging adjacent
+    tokens), and collapses whitespace. Numbers and units are preserved as-is,
+    so e.g. "AB-12" and "AB12" normalize to different token sequences.
+    """
+    text = unicodedata.normalize('NFKC', str(text))
+    text = text.casefold()
+    text = ''.join(ch if ch.isalnum() else ' ' for ch in text)
+    return ' '.join(text.split())
+
+
+def _is_missing(value) -> bool:
+    if value is None:
+        return True
+    if isinstance(value, float) and value != value:
+        return True
+    return False
+
+
+def similarity(input: list, metric: str = 'token_sort', decimal_places: int = 3) -> list:
+    """
+    Compute a symmetric, 0.0-1.0 bounded similarity score between two strings.
+
+    :param input: 2D list of value pairs to compare. [[a, b], [a1, b1], ...]
+    :param metric: 'token_sort', 'damerau_levenshtein', or 'token_set'
+    :param decimal_places: Number of decimal places to round the score to
+    """
+    _logging.debug(f": Computing {metric} similarity for {len(input)} record pairs")
+    results = []
+    for row in input:
+        a_raw, b_raw = row[0], row[1]
+
+        if _is_missing(a_raw) or _is_missing(b_raw):
+            results.append(None)
+            continue
+
+        a_norm = normalize_similarity_text(a_raw)
+        b_norm = normalize_similarity_text(b_raw)
+
+        if not a_norm or not b_norm:
+            results.append(None)
+            continue
+
+        if metric == 'token_sort':
+            score = _fuzz.token_sort_ratio(a_norm, b_norm) / 100
+        elif metric == 'token_set':
+            score = _fuzz.token_set_ratio(a_norm, b_norm) / 100
+        elif metric == 'damerau_levenshtein':
+            score = _distance.DamerauLevenshtein.normalized_similarity(a_norm, b_norm)
+        else:
+            raise ValueError(
+                "metric must be one of 'token_sort', 'damerau_levenshtein', 'token_set'"
+            )
+
+        results.append(round(score, decimal_places))
+
+    return results
+
+
 def deduplicate(result, enabled=False, ignore_case=False):
     _logging.debug(f": Deduplicating {len(result)} items :: ignore_case :: {ignore_case}")
     if not enabled:
diff --git a/wrangles/recipe_wrangles/compare.py b/wrangles/recipe_wrangles/compare.py
index 879d09563..b29adce1a 100644
--- a/wrangles/recipe_wrangles/compare.py
+++ b/wrangles/recipe_wrangles/compare.py
@@ -120,11 +120,13 @@ def text(
     empty_a: str = None,
     empty_b: str = None,
     all_empty: str = None,
-    case_sensitive: bool = True,
+    case_sensitive: bool = None,
+    # similarity parameters
+    metric: str = "token_sort",
 ) -> _pd.DataFrame:
     """
     type: object
-    description: Compare two strings and return the intersection or difference, or use overlap to find the matching characters between the two strings.
+    description: Compare two strings and return the intersection or difference, use overlap to find the matching characters between the two strings, or use similarity to get a numeric similarity score.
     required:
       - input
       - output
@@ -138,11 +140,12 @@ def text(
         description: The column to output the results to
       method:
         type: string
-        description: The type of comparison to perform (difference, intersection, overlap)
+        description: The type of comparison to perform (difference, intersection, overlap, similarity)
         enum:
           - difference
           - intersection
           - overlap
+          - similarity
     allOf:
       - if:
           properties:
@@ -155,7 +158,7 @@ def text(
               description: "(Optional) The character to split the strings on. Default is a space"
             case_sensitive:
               type: boolean
-              description: "(Optional) Whether the comparison is case sensitive. Default is True"
+              description: "(Optional, Deprecated) Ignored - comparisons are always case-insensitive. Retained only for backward compatibility with existing recipes."
       - if:
           properties:
             method:
@@ -167,7 +170,7 @@ def text(
               description: "(Optional) The character to split the strings on. Default is a space"
             case_sensitive:
               type: boolean
-              description: "(Optional) Whether the comparison is case sensitive. Default is True"
+              description: "(Optional, Deprecated) Ignored - comparisons are always case-insensitive. Retained only for backward compatibility with existing recipes."
       - if:
           properties:
             method:
@@ -179,7 +182,7 @@ def text(
               description: "(Optional) Character to use for non-matching characters"
             include_ratio:
               type: boolean
-              description: "(Optional) Include the ratio of matching characters"
+              description: "(Optional) Include the ratio of matching characters. This is the legacy difflib.SequenceMatcher score, not the similarity score from method: similarity"
             decimal_places:
               type: integer
               description: "(Optional) Number of decimal places to round the ratio to"
@@ -197,15 +200,40 @@ def text(
               description: "(Optional) Value to use for both inputs"
             case_sensitive:
               type: boolean
-              description: "(Optional) Whether the comparison is case sensitive. Default is True"
+              description: "(Optional, Deprecated) Ignored - comparisons are always case-insensitive. Retained only for backward compatibility with existing recipes."
+      - if:
+          properties:
+            method:
+              const: similarity
+        then:
+          properties:
+            metric:
+              type: string
+              description: "(Optional) The similarity metric to use. token_sort ignores token order but keeps duplicates and penalizes missing/extra content. damerau_levenshtein is a sequential character similarity that recognizes adjacent transpositions. token_set ignores token order and duplicates, and can score 1.0 when a shorter token set is fully contained in a longer one. Default is token_sort"
+              enum:
+                - token_sort
+                - damerau_levenshtein
+                - token_set
+            decimal_places:
+              type: integer
+              description: "(Optional) Number of decimal places to round the score to. Default is 3"
 
     """
     _logging.debug(f": Comparing text strings :: input :: {input}")
-    if method not in ["difference", "intersection", "overlap"]:
+    if method not in ["difference", "intersection", "overlap", "similarity"]:
         raise ValueError(
-            "Method must be one of 'overlap', 'difference' or 'intersection'"
+            "Method must be one of 'difference', 'intersection', 'overlap' or 'similarity'"
+        )
+
+    if case_sensitive is not None:
+        _logging.warning(
+            "compare.text: 'case_sensitive' is deprecated and ignored - "
+            "comparisons are always case-insensitive."
         )
 
+    if isinstance(decimal_places, str):
+        decimal_places = int(decimal_places)
+
     if method == "difference" or method == "intersection":
         # ensure that input is at least a list of two columns
         if not isinstance(input, list) or len(input) < 2:
@@ -215,13 +243,10 @@ def text(
             input=df[input].astype(str).values.tolist(),
             type=method,
             char=char,
-            case_sensitive=case_sensitive,
+            case_sensitive=False,
         )
 
     if method == "overlap":
-        if isinstance(decimal_places, str):
-            int(decimal_places)
-
         # ensure that input is a list of two columns
         if not isinstance(input, list) or len(input) != 2:
             raise ValueError("Input must be a list of two columns")
@@ -235,7 +260,23 @@ def text(
             empty_a=empty_a,
             empty_b=empty_b,
             all_empty=all_empty,
-            case_sensitive=case_sensitive,
+            case_sensitive=False,
+        )
+
+    if method == "similarity":
+        # ensure that input is a list of exactly two columns
+        if not isinstance(input, list) or len(input) != 2:
+            raise ValueError("Input must be a list of two columns")
+
+        if metric not in ["token_sort", "damerau_levenshtein", "token_set"]:
+            raise ValueError(
+                "metric must be one of 'token_sort', 'damerau_levenshtein', 'token_set'"
+            )
+
+        df[output] = _compare.similarity(
+            input=df[input].values.tolist(),
+            metric=metric,
+            decimal_places=decimal_places,
         )
 
     return df

From 6306882d4fff4c4a0266fa415f23f770c3d89942 Mon Sep 17 00:00:00 2001
From: mborodii-prog <mborodii@binariks.com>
Date: Tue, 25 Aug 2026 14:19:20 +0300
Subject: [PATCH 2/4] fix tests

---
 tests/recipes/test_recipes.py | 72 +++++++++++++++++++++++++++++++++--
 tests/test_wrangles.py        |  6 ++-
 2 files changed, 73 insertions(+), 5 deletions(-)

diff --git a/tests/recipes/test_recipes.py b/tests/recipes/test_recipes.py
index b2e09ef59..380972620 100644
--- a/tests/recipes/test_recipes.py
+++ b/tests/recipes/test_recipes.py
@@ -127,11 +127,36 @@ def test_recipe_by_version_tag():
     )
 
 
-def test_recipe_by_production_version():
+def test_recipe_by_production_version(mocker):
     """
     Test running a recipe using a model ID and production version
     """
+    mocker.patch(
+        "wrangles.data.model",
+        return_value={
+            "purpose": "recipe",
+            "production_version_id": "production-version-id"
+        }
+    )
+    model_content = mocker.patch(
+        "wrangles.data.model_content",
+        return_value={
+            "recipe": """
+            read:
+              - test:
+                  rows: 20
+                  values:
+                    header: value1
+            """
+        }
+    )
+
     df = wrangles.recipe.run("a6bac9e7-2388-4347")
+
+    model_content.assert_called_once_with(
+        "a6bac9e7-2388-4347",
+        "production-version-id"
+    )
     assert (
         len(df) == 20 and
         list(df.columns) == ["header"]
@@ -184,21 +209,62 @@ def test_recipe_by_production_semantic_version_falls_back_to_latest(
     assert "No production version exists, defaulting to latest version" in caplog.text
 
 
-def test_recipe_by_version_latest():
+def test_recipe_by_version_latest(mocker):
     """
     Test running a recipe using a model ID and latest version
     """
+    mocker.patch(
+        "wrangles.data.model",
+        return_value={
+            "purpose": "recipe",
+            "production_version_id": "production-version-id"
+        }
+    )
+    model_content = mocker.patch(
+        "wrangles.data.model_content",
+        return_value={
+            "recipe": """
+            read:
+              - test:
+                  rows: 10
+                  values:
+                    header: value1
+            """
+        }
+    )
+
     df = wrangles.recipe.run("a6bac9e7-2388-4347:latest")
+
+    model_content.assert_called_once_with("a6bac9e7-2388-4347", None)
     assert (
         len(df) == 10 and
         list(df.columns) == ["header"]
     )
 
-def test_recipe_by_latest_version():
+def test_recipe_by_latest_version(mocker):
     """
     Test running a recipe using a model ID and latest version
     """
+    mocker.patch(
+        "wrangles.data.model",
+        return_value={"purpose": "recipe"}
+    )
+    model_content = mocker.patch(
+        "wrangles.data.model_content",
+        return_value={
+            "recipe": """
+            read:
+              - test:
+                  rows: 15
+                  values:
+                    header: value1
+            """
+        }
+    )
+
     df = wrangles.recipe.run("02fc0c63-1294-415b")
+
+    model_content.assert_called_once_with("02fc0c63-1294-415b", None)
     assert (
         len(df) == 15 and
         list(df.columns) == ["header"]
diff --git a/tests/test_wrangles.py b/tests/test_wrangles.py
index 1276e2505..91617523d 100644
--- a/tests/test_wrangles.py
+++ b/tests/test_wrangles.py
@@ -913,10 +913,12 @@ def test_compare_overlap_empty_strings():
 
 def test_compare_overlap_case_insensitive():
     """
-    Test compare.overlap with case_sensitive=False
+    Test compare.overlap with case_sensitive=False.
+    Matching is case-insensitive, but the original casing of the first
+    input is preserved in the output rather than being forced to lowercase.
     """
     result = wrangles.compare.overlap([['HELLO', 'hello']], case_sensitive=False)
-    assert result == ['hello']
+    assert result == ['HELLO']
 
 def test_compare_overlap_custom_non_match():
     """

From 418e8b2ff3f2d07f4ced129b7de7f6bb1a1dcafa Mon Sep 17 00:00:00 2001
From: mborodii-prog <mborodii@binariks.com>
Date: Wed, 26 Aug 2026 11:40:34 +0300
Subject: [PATCH 3/4] Update test_recipes.py

---
 tests/recipes/test_recipes.py | 86 ++++-------------------------------
 1 file changed, 10 insertions(+), 76 deletions(-)

diff --git a/tests/recipes/test_recipes.py b/tests/recipes/test_recipes.py
index 380972620..172d9fe57 100644
--- a/tests/recipes/test_recipes.py
+++ b/tests/recipes/test_recipes.py
@@ -127,36 +127,11 @@ def test_recipe_by_version_tag():
     )
 
 
-def test_recipe_by_production_version(mocker):
+def test_recipe_by_production_version():
     """
     Test running a recipe using a model ID and production version
     """
-    mocker.patch(
-        "wrangles.data.model",
-        return_value={
-            "purpose": "recipe",
-            "production_version_id": "production-version-id"
-        }
-    )
-    model_content = mocker.patch(
-        "wrangles.data.model_content",
-        return_value={
-            "recipe": """
-            read:
-              - test:
-                  rows: 20
-                  values:
-                    header: value1
-            """
-        }
-    )
-
-    df = wrangles.recipe.run("a6bac9e7-2388-4347")
-
-    model_content.assert_called_once_with(
-        "a6bac9e7-2388-4347",
-        "production-version-id"
-    )
+    df = wrangles.recipe.run("e954717c-fb9c-4c47")
     assert (
         len(df) == 20 and
         list(df.columns) == ["header"]
@@ -179,10 +154,10 @@ def test_recipe_by_production_semantic_version(mocker):
         return_value={"recipe": "{}"}
     )
 
-    wrangles.recipe.run("a6bac9e7-2388-4347:production")
+    wrangles.recipe.run("e954717c-fb9c-4c47:production")
 
     model_content.assert_called_once_with(
-        "a6bac9e7-2388-4347",
+        "e954717c-fb9c-4c47",
         "production-version-id"
     )
 
@@ -203,68 +178,27 @@ def test_recipe_by_production_semantic_version_falls_back_to_latest(
         return_value={"recipe": "{}"}
     )
 
-    wrangles.recipe.run("a6bac9e7-2388-4347:production")
+    wrangles.recipe.run("e954717c-fb9c-4c47:production")
 
-    model_content.assert_called_once_with("a6bac9e7-2388-4347", None)
+    model_content.assert_called_once_with("e954717c-fb9c-4c47", None)
     assert "No production version exists, defaulting to latest version" in caplog.text
 
 
-def test_recipe_by_version_latest(mocker):
+def test_recipe_by_version_latest():
     """
     Test running a recipe using a model ID and latest version
     """
-    mocker.patch(
-        "wrangles.data.model",
-        return_value={
-            "purpose": "recipe",
-            "production_version_id": "production-version-id"
-        }
-    )
-    model_content = mocker.patch(
-        "wrangles.data.model_content",
-        return_value={
-            "recipe": """
-            read:
-              - test:
-                  rows: 10
-                  values:
-                    header: value1
-            """
-        }
-    )
-
-    df = wrangles.recipe.run("a6bac9e7-2388-4347:latest")
-
-    model_content.assert_called_once_with("a6bac9e7-2388-4347", None)
+    df = wrangles.recipe.run("e954717c-fb9c-4c47:latest")
     assert (
         len(df) == 10 and
         list(df.columns) == ["header"]
     )
 
-def test_recipe_by_latest_version(mocker):
+def test_recipe_by_latest_version():
     """
     Test running a recipe using a model ID and latest version
     """
-    mocker.patch(
-        "wrangles.data.model",
-        return_value={"purpose": "recipe"}
-    )
-    model_content = mocker.patch(
-        "wrangles.data.model_content",
-        return_value={
-            "recipe": """
-            read:
-              - test:
-                  rows: 15
-                  values:
-                    header: value1
-            """
-        }
-    )
-
-    df = wrangles.recipe.run("02fc0c63-1294-415b")
-
-    model_content.assert_called_once_with("02fc0c63-1294-415b", None)
+    df = wrangles.recipe.run("1b41d016-7129-4b66")
     assert (
         len(df) == 15 and
         list(df.columns) == ["header"]

From 2f2b0690bd85abcafab192648cc1db9b30ad94ce Mon Sep 17 00:00:00 2001
From: mborodii-prog <mborodii@binariks.com>
Date: Wed, 26 Aug 2026 17:57:09 +0300
Subject: [PATCH 4/4] Restore case_sensitive toggle, split overlap ratio
 output, add per-metric schema descriptions

---
 tests/recipes/wrangles/test_compare.py | 150 +++++++++++++++++--------
 wrangles/recipe_wrangles/compare.py    |  56 +++++----
 2 files changed, 140 insertions(+), 66 deletions(-)

diff --git a/tests/recipes/wrangles/test_compare.py b/tests/recipes/wrangles/test_compare.py
index 50830f90a..822b08dd9 100644
--- a/tests/recipes/wrangles/test_compare.py
+++ b/tests/recipes/wrangles/test_compare.py
@@ -1,7 +1,6 @@
 import wrangles
 import pandas as pd
 import numpy as np
-import logging
 
 
 class TestCompareList:
@@ -652,7 +651,9 @@ def test_compare_text_overlap_empty_values(self):
             input:
             - col1
             - col2
-            output: output
+            output:
+            - output_mask
+            - output_ratio
             method: overlap
             non_match_char: '@'
             include_ratio: True
@@ -666,13 +667,14 @@ def test_compare_text_overlap_empty_values(self):
             recipe=recipe,
             dataframe=data,
         )
-        assert df["output"].values.tolist() == [
-            ["@@@@@Mario", 0.67],
-            ["@@@@@Luigi", 0.67],
-            ["Empty A", 0],
-            ["Empty B", 0],
-            ["Both Empty", 0],
+        assert df["output_mask"].values.tolist() == [
+            "@@@@@Mario",
+            "@@@@@Luigi",
+            "Empty A",
+            "Empty B",
+            "Both Empty",
         ]
+        assert df["output_ratio"].values.tolist() == [0.67, 0.67, 0, 0, 0]
 
     def test_compare_text_overlap_include_ratio(self):
         """
@@ -701,7 +703,9 @@ def test_compare_text_overlap_include_ratio(self):
             input:
             - col1
             - col2
-            output: output
+            output:
+            - output_mask
+            - output_ratio
             method: overlap
             include_ratio: True
         """
@@ -710,12 +714,8 @@ def test_compare_text_overlap_include_ratio(self):
             dataframe=data,
         )
 
-        assert df["output"].values.tolist() == [
-            ["Mario", 1],
-            ["Luigi", 1],
-            ["Mar*io", 0.909],
-            ["Lui*gi", 0.909],
-        ]
+        assert df["output_mask"].values.tolist() == ["Mario", "Luigi", "Mar*io", "Lui*gi"]
+        assert df["output_ratio"].values.tolist() == [1, 1, 0.909, 0.909]
 
     def test_compare_overlap_default_settings(self):
         """
@@ -911,13 +911,13 @@ def test_compare_text_case_insensitive_overlap(self):
         )
         assert df["output"][1] == "ANOTHER LINE THAT IS ALSO IN *****CA*S*"
 
-    def test_compare_text_case_sensitive_deprecation_warning(self, caplog):
+    def test_compare_text_case_sensitive_true_enables_case_sensitive_matching(self):
         """
-        Supplying case_sensitive should emit a deprecation warning; omitting it should not.
+        case_sensitive: true restores real case-sensitive matching, distinct
+        from the case-insensitive default.
         """
         data = pd.DataFrame({"col1": ["Mario"], "col2": ["mario"]})
-
-        recipe_with_param = """
+        recipe = """
         wrangles:
         - compare.text:
             input:
@@ -927,29 +927,13 @@ def test_compare_text_case_sensitive_deprecation_warning(self, caplog):
             method: difference
             case_sensitive: true
         """
-        with caplog.at_level(logging.WARNING):
-            caplog.clear()
-            wrangles.recipe.run(recipe=recipe_with_param, dataframe=data)
-        assert any("case_sensitive" in record.message for record in caplog.records)
-
-        recipe_without_param = """
-        wrangles:
-        - compare.text:
-            input:
-            - col1
-            - col2
-            output: output
-            method: difference
-        """
-        with caplog.at_level(logging.WARNING):
-            caplog.clear()
-            wrangles.recipe.run(recipe=recipe_without_param, dataframe=data)
-        assert not any("case_sensitive" in record.message for record in caplog.records)
+        df = wrangles.recipe.run(recipe=recipe, dataframe=data)
+        assert df["output"][0] == "mario"
 
-    def test_compare_text_case_sensitive_true_false_omitted_equivalent(self):
+    def test_compare_text_case_sensitive_false_and_omitted_equivalent(self):
         """
-        case_sensitive: true, false, and omitted must all produce identical results,
-        since it is deprecated and ignored - comparisons are always case-insensitive.
+        case_sensitive: false and omitting it entirely must produce identical
+        (case-insensitive) results, since False is the default.
         """
         data = pd.DataFrame(
             {
@@ -959,7 +943,6 @@ def test_compare_text_case_sensitive_true_false_omitted_equivalent(self):
         )
         outputs = {}
         for label, case_sensitive_line in [
-            ("true", "case_sensitive: true"),
             ("false", "case_sensitive: false"),
             ("omitted", ""),
         ]:
@@ -976,7 +959,42 @@ def test_compare_text_case_sensitive_true_false_omitted_equivalent(self):
             df = wrangles.recipe.run(recipe=recipe, dataframe=data)
             outputs[label] = df["output"][0]
 
-        assert outputs["true"] == outputs["false"] == outputs["omitted"] == "lowercase"
+        assert outputs["false"] == outputs["omitted"] == "lowercase"
+
+    def test_compare_text_case_sensitive_true_differs_from_default(self):
+        """
+        case_sensitive: true must produce a different result than the
+        case-insensitive default when casing differs between inputs.
+        """
+        data = pd.DataFrame(
+            {
+                "col1": ["THIS IS IN ALL CAPS"],
+                "col2": ["this is in all lowercase"],
+            }
+        )
+        recipe_true = """
+        wrangles:
+        - compare.text:
+            input:
+            - col1
+            - col2
+            output: output
+            method: difference
+            case_sensitive: true
+        """
+        recipe_default = """
+        wrangles:
+        - compare.text:
+            input:
+            - col1
+            - col2
+            output: output
+            method: difference
+        """
+        df_true = wrangles.recipe.run(recipe=recipe_true, dataframe=data)
+        df_default = wrangles.recipe.run(recipe=recipe_default, dataframe=data)
+        assert df_true["output"][0] != df_default["output"][0]
+        assert df_default["output"][0] == "lowercase"
 
     def test_compare_text_default_preserves_title_case(self):
         """
@@ -1012,13 +1030,57 @@ def test_compare_text_overlap_quoted_decimal_places(self):
             input:
             - col1
             - col2
-            output: output
+            output:
+            - output_mask
+            - output_ratio
             method: overlap
             include_ratio: true
             decimal_places: "2"
         """
         df = wrangles.recipe.run(recipe=recipe, dataframe=data)
-        assert df["output"][0] == ["Mar*io", 0.91]
+        assert df["output_mask"][0] == "Mar*io"
+        assert df["output_ratio"][0] == 0.91
+
+    def test_compare_text_overlap_include_ratio_requires_two_outputs(self):
+        """
+        include_ratio: true must raise a clear error if output is not a
+        list of exactly two column names.
+        """
+        data = pd.DataFrame({"col1": ["Mario"], "col2": ["Martio"]})
+
+        for bad_output in ["output", "\n            - only_one_column"]:
+            recipe = f"""
+            wrangles:
+            - compare.text:
+                input:
+                - col1
+                - col2
+                output:{bad_output}
+                method: overlap
+                include_ratio: true
+            """
+            try:
+                wrangles.recipe.run(recipe=recipe, dataframe=data)
+                assert False, "Should raise an error if output is not a list of two columns"
+            except Exception:
+                pass
+
+    def test_compare_text_overlap_without_include_ratio_still_accepts_single_output(self):
+        """
+        Without include_ratio, output can still be a single column name as before.
+        """
+        data = pd.DataFrame({"col1": ["Mario"], "col2": ["Martio"]})
+        recipe = """
+        wrangles:
+        - compare.text:
+            input:
+            - col1
+            - col2
+            output: output
+            method: overlap
+        """
+        df = wrangles.recipe.run(recipe=recipe, dataframe=data)
+        assert df["output"][0] == "Mar*io"
 
 
 class TestCompareTextSimilarity:
diff --git a/wrangles/recipe_wrangles/compare.py b/wrangles/recipe_wrangles/compare.py
index b29adce1a..ba4f7e67d 100644
--- a/wrangles/recipe_wrangles/compare.py
+++ b/wrangles/recipe_wrangles/compare.py
@@ -4,6 +4,7 @@
 
 import logging as _logging
 import pandas as _pd
+from typing import Union as _Union
 from .. import compare as _compare
 
 
@@ -108,7 +109,7 @@ def compare_lists(row):
 def text(
     df: _pd.DataFrame,
     input: list,
-    output: str,
+    output: _Union[str, list],
     method: str = "difference",
     # Overlap parameters
     char: str = " ",
@@ -120,7 +121,7 @@ def text(
     empty_a: str = None,
     empty_b: str = None,
     all_empty: str = None,
-    case_sensitive: bool = None,
+    case_sensitive: bool = False,
     # similarity parameters
     metric: str = "token_sort",
 ) -> _pd.DataFrame:
@@ -136,8 +137,10 @@ def text(
         type: array
         description: the columns to compare. First column is the base column
       output:
-        type: string
-        description: The column to output the results to
+        type:
+          - string
+          - array
+        description: The column to output the results to. Must be a list of two column names [mask_column, ratio_column] when method is overlap and include_ratio is true; otherwise a single column name.
       method:
         type: string
         description: The type of comparison to perform (difference, intersection, overlap, similarity)
@@ -158,7 +161,7 @@ def text(
               description: "(Optional) The character to split the strings on. Default is a space"
             case_sensitive:
               type: boolean
-              description: "(Optional, Deprecated) Ignored - comparisons are always case-insensitive. Retained only for backward compatibility with existing recipes."
+              description: "(Optional) Whether the comparison is case sensitive. Default is False"
       - if:
           properties:
             method:
@@ -170,7 +173,7 @@ def text(
               description: "(Optional) The character to split the strings on. Default is a space"
             case_sensitive:
               type: boolean
-              description: "(Optional, Deprecated) Ignored - comparisons are always case-insensitive. Retained only for backward compatibility with existing recipes."
+              description: "(Optional) Whether the comparison is case sensitive. Default is False"
       - if:
           properties:
             method:
@@ -182,7 +185,7 @@ def text(
               description: "(Optional) Character to use for non-matching characters"
             include_ratio:
               type: boolean
-              description: "(Optional) Include the ratio of matching characters. This is the legacy difflib.SequenceMatcher score, not the similarity score from method: similarity"
+              description: "(Optional) Include the ratio of matching characters. This is the legacy difflib.SequenceMatcher score, not the similarity score from method: similarity. When true, output must be a list of two column names: [mask_column, ratio_column]"
             decimal_places:
               type: integer
               description: "(Optional) Number of decimal places to round the ratio to"
@@ -200,7 +203,7 @@ def text(
               description: "(Optional) Value to use for both inputs"
             case_sensitive:
               type: boolean
-              description: "(Optional, Deprecated) Ignored - comparisons are always case-insensitive. Retained only for backward compatibility with existing recipes."
+              description: "(Optional) Whether the comparison is case sensitive. Default is False"
       - if:
           properties:
             method:
@@ -209,11 +212,14 @@ def text(
           properties:
             metric:
               type: string
-              description: "(Optional) The similarity metric to use. token_sort ignores token order but keeps duplicates and penalizes missing/extra content. damerau_levenshtein is a sequential character similarity that recognizes adjacent transpositions. token_set ignores token order and duplicates, and can score 1.0 when a shorter token set is fully contained in a longer one. Default is token_sort"
-              enum:
-                - token_sort
-                - damerau_levenshtein
-                - token_set
+              description: "(Optional) The similarity metric to use. Default is token_sort"
+              oneOf:
+                - const: token_sort
+                  description: "Ignores token order but keeps duplicate tokens, penalizing missing or extra content. Best general-purpose choice for comparing full descriptions where word order may differ."
+                - const: damerau_levenshtein
+                  description: "Sequential character-edit similarity that recognizes adjacent transpositions (e.g. smtih vs smith) as a single edit. Best for short, order-sensitive strings like part numbers or codes."
+                - const: token_set
+                  description: "Ignores token order and duplicate tokens. A shorter token set fully contained in a longer one can score 1.0. Best when one description is expected to be a subset of the other."
             decimal_places:
               type: integer
               description: "(Optional) Number of decimal places to round the score to. Default is 3"
@@ -225,12 +231,6 @@ def text(
             "Method must be one of 'difference', 'intersection', 'overlap' or 'similarity'"
         )
 
-    if case_sensitive is not None:
-        _logging.warning(
-            "compare.text: 'case_sensitive' is deprecated and ignored - "
-            "comparisons are always case-insensitive."
-        )
-
     if isinstance(decimal_places, str):
         decimal_places = int(decimal_places)
 
@@ -243,7 +243,7 @@ def text(
             input=df[input].astype(str).values.tolist(),
             type=method,
             char=char,
-            case_sensitive=False,
+            case_sensitive=case_sensitive,
         )
 
     if method == "overlap":
@@ -251,7 +251,13 @@ def text(
         if not isinstance(input, list) or len(input) != 2:
             raise ValueError("Input must be a list of two columns")
 
-        df[output] = _compare.overlap(
+        if include_ratio and (not isinstance(output, list) or len(output) != 2):
+            raise ValueError(
+                "output must be a list of two columns when include_ratio is true "
+                "(e.g. [mask_column, ratio_column])"
+            )
+
+        results = _compare.overlap(
             input=df[input].astype(str).values.tolist(),
             non_match_char=non_match_char,
             include_ratio=include_ratio,
@@ -260,9 +266,15 @@ def text(
             empty_a=empty_a,
             empty_b=empty_b,
             all_empty=all_empty,
-            case_sensitive=False,
+            case_sensitive=case_sensitive,
         )
 
+        if include_ratio:
+            df[output[0]] = [row[0] for row in results]
+            df[output[1]] = [row[1] for row in results]
+        else:
+            df[output] = results
+
     if method == "similarity":
         # ensure that input is a list of exactly two columns
         if not isinstance(input, list) or len(input) != 2: