Order -0.0 below 0.0 in data file metrics bounds

pyiceberg/io/pyarrow.py

@@ -2381,13 +2381,19 @@ def update_min(self, val: Any | None) -> None:
         if self.current_min is None:
             self.current_min = val
         elif val is not None:
-            self.current_min = min(val, self.current_min)
+            self.current_min = min(self.current_min, val)
+            if isinstance(val, float) and val == 0.0 and math.copysign(1.0, val) < 0 and self.current_min == 0.0:
+                # min keeps the existing zero on a tie, but bounds must order -0.0 below 0.0
+                self.current_min = val
 
     def update_max(self, val: Any | None) -> None:
         if self.current_max is None:
             self.current_max = val
         elif val is not None:
-            self.current_max = max(val, self.current_max)
+            self.current_max = max(self.current_max, val)
+            if isinstance(val, float) and val == 0.0 and math.copysign(1.0, val) > 0 and self.current_max == 0.0:
+                # max keeps the existing zero on a tie, but bounds must order -0.0 below 0.0
+                self.current_max = val
 
     def min_as_bytes(self) -> bytes | None:
         if self.current_min is None:

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",
     [