Fix SQLColumnCheckOperator crash on non-numeric column bounds

providers/common/sql/src/airflow/providers/common/sql/operators/sql.py

@@ -746,25 +746,30 @@ def _get_match(self, check_values, record, tolerance=None) -> bool:
         match_boolean = True
 
         # abs() so the tolerance margin widens the bound outward for negative expected values too.
-        def _margin(expected):
-            return abs(expected) * tolerance if tolerance is not None else 0
+        # Without a tolerance the bound is compared as-is: a min/max check may be declared on a
+        # date or text column, where arithmetic on the bound raises TypeError.
+        def _lower(expected):
+            return expected - abs(expected) * tolerance if tolerance is not None else expected
+
+        def _upper(expected):
+            return expected + abs(expected) * tolerance if tolerance is not None else expected
 
         if "geq_to" in check_values:
-            match_boolean = record >= check_values["geq_to"] - _margin(check_values["geq_to"])
+            match_boolean = record >= _lower(check_values["geq_to"])
         elif "greater_than" in check_values:
-            match_boolean = record > check_values["greater_than"] - _margin(check_values["greater_than"])
+            match_boolean = record > _lower(check_values["greater_than"])
         if "leq_to" in check_values:
-            match_boolean = (
-                record <= check_values["leq_to"] + _margin(check_values["leq_to"]) and match_boolean
-            )
+            match_boolean = record <= _upper(check_values["leq_to"]) and match_boolean
         elif "less_than" in check_values:
-            match_boolean = (
-                record < check_values["less_than"] + _margin(check_values["less_than"]) and match_boolean
-            )
+            match_boolean = record < _upper(check_values["less_than"]) and match_boolean
         if "equal_to" in check_values:
             expected = check_values["equal_to"]
-            margin = _margin(expected)
-            match_boolean = (expected - margin <= record <= expected + margin) and match_boolean
+            if tolerance is None:
+                # Equality, not a degenerate range: a NULL result must fail the check rather
+                # than raise on an ordering comparison against None.
+                match_boolean = record == expected and match_boolean
+            else:
+                match_boolean = (_lower(expected) <= record <= _upper(expected)) and match_boolean
         return match_boolean
 
     def _column_mapping_validation(self, check, check_values):

providers/common/sql/tests/unit/common/sql/operators/test_sql.py

@@ -1721,6 +1721,36 @@ def test_get_match_tolerance_handles_negative_thresholds(self, check_values, rec
         op = self._make_operator({"col": {"min": {"geq_to": 1}}})
         assert op._get_match(check_values, record, tolerance) == expected
 
+    @pytest.mark.parametrize(
+        ("check_values", "record", "expected"),
+        [
+            ({"geq_to": "2020-01-01"}, "2021-06-30", True),
+            ({"geq_to": "2020-01-01"}, "2019-12-31", False),
+            ({"greater_than": "2020-01-01"}, "2020-01-01", False),
+            ({"leq_to": "2020-01-01"}, "2019-12-31", True),
+            ({"less_than": "2020-01-01"}, "2020-01-01", False),
+            ({"equal_to": "abc"}, "abc", True),
+            ({"equal_to": "abc"}, "abcd", False),
+            (
+                {"geq_to": datetime.date(2020, 1, 1), "leq_to": datetime.date(2020, 12, 31)},
+                datetime.date(2020, 6, 30),
+                True,
+            ),
+            (
+                {"geq_to": datetime.date(2020, 1, 1), "leq_to": datetime.date(2020, 12, 31)},
+                datetime.date(2021, 1, 1),
+                False,
+            ),
+        ],
+    )
+    def test_get_match_compares_non_numeric_bounds_without_tolerance(self, check_values, record, expected):
+        op = self._make_operator({"col": {"min": {"geq_to": 1}}})
+        assert op._get_match(check_values, record) == expected
+
+    def test_get_match_equal_to_fails_cleanly_on_none_record(self):
+        op = self._make_operator({"col": {"null_check": {"equal_to": 0}}}, accept_none=False)
+        assert op._get_match({"equal_to": 0}, None) is False
+
     def test_multiple_checks_correct_names_and_order(self):
         op = self._make_operator(
             {