Fix SQL value checks for negative expected values

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

@@ -1279,12 +1279,14 @@ def _to_float(self, records):
     def _get_string_matches(self, records, pass_value_conv):
         return [str(record) == pass_value_conv for record in records]
 
+    def _get_tolerance_bounds(self, numeric_pass_value_conv):
+        margin = abs(numeric_pass_value_conv) * self.tol
+        return numeric_pass_value_conv - margin, numeric_pass_value_conv + margin
+
     def _get_numeric_matches(self, numeric_records, numeric_pass_value_conv):
         if self.has_tolerance:
-            return [
-                numeric_pass_value_conv * (1 - self.tol) <= record <= numeric_pass_value_conv * (1 + self.tol)
-                for record in numeric_records
-            ]
+            lower_bound, upper_bound = self._get_tolerance_bounds(numeric_pass_value_conv)
+            return [lower_bound <= record <= upper_bound for record in numeric_records]
 
         return [record == numeric_pass_value_conv for record in numeric_records]
 
@@ -1308,7 +1310,8 @@ def _build_check_results(self, records) -> list[SQLCheckResult]:
 
             pass_value_conv = _convert_to_float_if_possible(self.pass_value)
             if isinstance(pass_value_conv, float) and isinstance(self.tol, float):
-                expected_str = f">= {pass_value_conv * (1 - self.tol)}, <= {pass_value_conv * (1 + self.tol)}"
+                lower_bound, upper_bound = self._get_tolerance_bounds(pass_value_conv)
+                expected_str = f">= {lower_bound}, <= {upper_bound}"
                 check_type = "accepted_range"
 
             return [

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

@@ -876,6 +876,12 @@ def test_execute_fail(self, mock_get_db_hook):
         with pytest.raises(AirflowException, match="Tolerance:100.0%"):
             operator.execute(context=MagicMock())
 
+    @pytest.mark.parametrize("record", [-110, -100, -90])
+    def test_negative_pass_value_with_tolerance(self, record):
+        operator = self._construct_operator("select value from tab1 limit 1;", -100, 0.1)
+
+        operator.check_value([record])
+
 
 class TestIntervalCheckOperator:
     def _construct_operator(self, table, metric_thresholds, ratio_formula, ignore_zero):
@@ -2044,6 +2050,16 @@ def test_numeric_tolerance_produces_accepted_range(self):
         assert r.expected == ">= 4.5, <= 5.5"
         assert r.params == {"pass_value": "5", "tolerance": 0.1}
 
+    def test_negative_numeric_tolerance_produces_accepted_range(self):
+        op = self._make_operator(pass_value=-100, tolerance=0.1)
+        results = op._build_check_results([-100])
+        assert len(results) == 1
+        r = results[0]
+        assert r.success is True
+        assert r.check_type == "accepted_range"
+        assert r.expected == ">= -110.0, <= -90.0"
+        assert r.params == {"pass_value": "-100", "tolerance": 0.1}
+
     def test_non_numeric_pass_value_is_accepted_values(self):
         op = self._make_operator(pass_value="hello")
         results = op._build_check_results(["hello"])