From 3b73e4f7d284403cb950a5c825595572634610ad Mon Sep 17 00:00:00 2001
From: Shahar Epstein <60007259+shahar1@users.noreply.github.com>
Date: Sat, 1 Aug 2026 10:30:24 +0300
Subject: [PATCH] Fix SQLColumnCheckOperator crash on non-numeric column bounds

The tolerance rewrite made every bound go through arithmetic, including
when no tolerance is configured. A min/max check declared on a date or
text column has a bound that cannot be subtracted from, so checks that
compared cleanly before now fail the task with a TypeError.

An equal_to check also stopped reporting a NULL result as a failed check
and started raising instead, because a degenerate range still orders the
record against the bound.
---
 .../providers/common/sql/operators/sql.py     | 29 ++++++++++--------
 .../unit/common/sql/operators/test_sql.py     | 30 +++++++++++++++++++
 2 files changed, 47 insertions(+), 12 deletions(-)

diff --git a/providers/common/sql/src/airflow/providers/common/sql/operators/sql.py b/providers/common/sql/src/airflow/providers/common/sql/operators/sql.py
index 9c2da9d6d7e9b..01148ae4507f6 100644
--- a/providers/common/sql/src/airflow/providers/common/sql/operators/sql.py
+++ b/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):
diff --git a/providers/common/sql/tests/unit/common/sql/operators/test_sql.py b/providers/common/sql/tests/unit/common/sql/operators/test_sql.py
index 6c6fa9a405ae1..fe3bc8cb265dc 100644
--- a/providers/common/sql/tests/unit/common/sql/operators/test_sql.py
+++ b/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(
             {