From 35e45415983d3a710d1293fb97f4a795f23aa6dc Mon Sep 17 00:00:00 2001
From: Steve Ahn <38049807+steveahnahn@users.noreply.github.com>
Date: Fri, 10 Jul 2026 16:06:36 -0700
Subject: [PATCH] Fix SQL check tolerance for negative expected values

---
 .../providers/common/sql/operators/sql.py     | 40 ++++++++-----------
 .../unit/common/sql/operators/test_sql.py     | 30 ++++++++++++++
 2 files changed, 46 insertions(+), 24 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 21c8b6502b0bb..7897ce19c155b 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
@@ -744,35 +744,27 @@ def _get_match(self, check_values, record, tolerance=None) -> bool:
         if record is None and self.accept_none:
             record = 0
         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
+
         if "geq_to" in check_values:
-            if tolerance is not None:
-                match_boolean = record >= check_values["geq_to"] * (1 - tolerance)
-            else:
-                match_boolean = record >= check_values["geq_to"]
+            match_boolean = record >= check_values["geq_to"] - _margin(check_values["geq_to"])
         elif "greater_than" in check_values:
-            if tolerance is not None:
-                match_boolean = record > check_values["greater_than"] * (1 - tolerance)
-            else:
-                match_boolean = record > check_values["greater_than"]
+            match_boolean = record > check_values["greater_than"] - _margin(check_values["greater_than"])
         if "leq_to" in check_values:
-            if tolerance is not None:
-                match_boolean = record <= check_values["leq_to"] * (1 + tolerance) and match_boolean
-            else:
-                match_boolean = record <= check_values["leq_to"] and match_boolean
+            match_boolean = (
+                record <= check_values["leq_to"] + _margin(check_values["leq_to"]) and match_boolean
+            )
         elif "less_than" in check_values:
-            if tolerance is not None:
-                match_boolean = record < check_values["less_than"] * (1 + tolerance) and match_boolean
-            else:
-                match_boolean = record < check_values["less_than"] and match_boolean
+            match_boolean = (
+                record < check_values["less_than"] + _margin(check_values["less_than"]) and match_boolean
+            )
         if "equal_to" in check_values:
-            if tolerance is not None:
-                match_boolean = (
-                    check_values["equal_to"] * (1 - tolerance)
-                    <= record
-                    <= check_values["equal_to"] * (1 + tolerance)
-                ) and match_boolean
-            else:
-                match_boolean = record == check_values["equal_to"] and match_boolean
+            expected = check_values["equal_to"]
+            margin = _margin(expected)
+            match_boolean = (expected - margin <= record <= expected + margin) 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 14380b717363c..2cee94a635796 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
@@ -1684,6 +1684,36 @@ def test_accepted_range_geq_leq(self):
         assert r.expected == ">=1, <=100"
         assert r.params == {"geq_to": 1, "leq_to": 100, "accept_none": True}
 
+    @pytest.mark.parametrize(
+        ("check_values", "record", "tolerance", "expected"),
+        [
+            # Negative thresholds: tolerance widens the bound outward, so a record equal to
+            # the threshold passes.
+            ({"geq_to": -1000}, -1000, 0.1, True),
+            ({"geq_to": -1000}, -1100, 0.1, True),
+            ({"geq_to": -1000}, -1101, 0.1, False),
+            ({"greater_than": -1000}, -1050, 0.1, True),
+            ({"greater_than": -1000}, -1100, 0.1, False),  # strict: equal to the widened bound fails
+            ({"leq_to": -10}, -10, 0.1, True),
+            ({"less_than": -10}, -11, 0.1, True),
+            ({"equal_to": -100}, -100, 0.1, True),
+            ({"equal_to": -100}, -110, 0.1, True),
+            ({"equal_to": -100}, -111, 0.1, False),
+            ({"geq_to": -100, "leq_to": -50}, -75, 0.1, True),
+            # Positive thresholds keep their existing bounds.
+            ({"geq_to": 1000}, 900, 0.1, True),
+            ({"geq_to": 1000}, 899, 0.1, False),
+            ({"equal_to": 100}, 110, 0.1, True),
+            ({"equal_to": 100}, 111, 0.1, False),
+            # No tolerance: exact bounds.
+            ({"geq_to": 10}, 10, None, True),
+            ({"equal_to": 5}, 6, None, False),
+        ],
+    )
+    def test_get_match_tolerance_handles_negative_thresholds(self, check_values, record, tolerance, expected):
+        op = self._make_operator({"col": {"min": {"geq_to": 1}}})
+        assert op._get_match(check_values, record, tolerance) == expected
+
     def test_multiple_checks_correct_names_and_order(self):
         op = self._make_operator(
             {