From a19a5a7a5b5b055867d69592a2425e7adfc01790 Mon Sep 17 00:00:00 2001
From: Ayush Kumar Saha <225407166+AP24110011695@users.noreply.github.com>
Date: Wed, 11 Nov 2026 03:39:39 +0530
Subject: [PATCH] Order -0.0 below 0.0 in data file metrics bounds

StatsAggregator.update_min/update_max feed DataFile lower/upper bounds,
but min(val, current) and max(val, current) keep the first argument on a
tie, so whichever zero arrives last wins: a late -0.0 lowers an existing
+0.0 upper bound, and a late +0.0 raises an existing -0.0 lower bound.
The serialized bounds then order above a value present in the file.

Mirror the fix that #4324 applied to partition field summaries: restore
the stored bound when a zero arrives that would invert the -0.0 < 0.0
ordering. The argument order here (val first) flips which direction
needs the guard compared to manifest.py.

Verified failing-first: the new tests fail on main for both arrival
orders and both float types, and pass with the guard.

Signed-off-by: Ayush Kumar Saha <225407166+AP24110011695@users.noreply.github.com>
---
 pyiceberg/io/pyarrow.py  | 18 ++++++++++++++++++
 tests/io/test_pyarrow.py | 24 ++++++++++++++++++++++++
 2 files changed, 42 insertions(+)

diff --git a/pyiceberg/io/pyarrow.py b/pyiceberg/io/pyarrow.py
index f62e53a911..5700e4d93f 100644
--- a/pyiceberg/io/pyarrow.py
+++ b/pyiceberg/io/pyarrow.py
@@ -2381,13 +2381,31 @@ def update_min(self, val: Any | None) -> None:
         if self.current_min is None:
             self.current_min = val
         elif val is not None:
+            previous = self.current_min
             self.current_min = min(val, self.current_min)
+            if (
+                isinstance(val, float)
+                and val == 0.0
+                and isinstance(previous, float)
+                and math.copysign(1.0, previous) < math.copysign(1.0, self.current_min)
+            ):
+                # min keeps whichever zero came first, but bounds must order -0.0 below 0.0
+                self.current_min = previous
 
     def update_max(self, val: Any | None) -> None:
         if self.current_max is None:
             self.current_max = val
         elif val is not None:
+            previous = self.current_max
             self.current_max = max(val, self.current_max)
+            if (
+                isinstance(val, float)
+                and val == 0.0
+                and isinstance(previous, float)
+                and math.copysign(1.0, previous) > math.copysign(1.0, self.current_max)
+            ):
+                # max keeps whichever zero came first, but bounds must order -0.0 below 0.0
+                self.current_max = previous
 
     def min_as_bytes(self) -> bytes | None:
         if self.current_min is None:
diff --git a/tests/io/test_pyarrow.py b/tests/io/test_pyarrow.py
index 419d130965..56cc916390 100644
--- a/tests/io/test_pyarrow.py
+++ b/tests/io/test_pyarrow.py
@@ -43,6 +43,7 @@
 from pyroaring import BitMap
 
 from pyiceberg.catalog.rest.scan_planning import RESTDataFile, RESTDeleteFile, RESTFileScanTask, RESTPositionDeleteFile
+from pyiceberg.conversions import to_bytes
 from pyiceberg.encryption.key_metadata import StandardKeyMetadata
 from pyiceberg.encryption.stream import _encrypt_ags1
 from pyiceberg.exceptions import ResolveError
@@ -2910,6 +2911,29 @@ def test_stats_aggregator_update_max(vals: list[Any], primitive_type: PrimitiveT
     assert stats.current_max == expected_result
 
 
+@pytest.mark.parametrize(
+    "vals, primitive_type",
+    [
+        ([0.0, -0.0], DoubleType()),
+        ([-0.0, 0.0], DoubleType()),
+        ([0.0, -0.0], FloatType()),
+        ([-0.0, 0.0], FloatType()),
+    ],
+)
+def test_stats_aggregator_orders_negative_zero_below_zero(vals: list[float], primitive_type: PrimitiveType) -> None:
+    # min/max keep whichever zero came first, but the serialized bounds must
+    # order -0.0 below 0.0, whichever order the values arrive in
+    stats_min = StatsAggregator(primitive_type, _primitive_to_physical(primitive_type))
+    stats_max = StatsAggregator(primitive_type, _primitive_to_physical(primitive_type))
+
+    for val in vals:
+        stats_min.update_min(val)
+        stats_max.update_max(val)
+
+    assert stats_min.min_as_bytes() == to_bytes(primitive_type, -0.0)
+    assert stats_max.max_as_bytes() == to_bytes(primitive_type, 0.0)
+
+
 @pytest.mark.parametrize(
     "iceberg_type, physical_type_string",
     [