Write each row once when an identity partition has both -0.0 and 0.0

pyiceberg/io/pyarrow.py

@@ -3218,6 +3218,10 @@ def _partition_value_filter(partition_field_name: str, value: Any) -> pc.Express
         return pc.field(partition_field_name).is_null()
     elif isinstance(value, float) and math.isnan(value):
         return pc.is_nan(pc.field(partition_field_name))
+    elif isinstance(value, float) and value == 0.0:
+        # -0.0 and 0.0 are separate partitions but compare equal, so also match the sign (1 / -0.0 is -inf)
+        field = pc.field(partition_field_name)
+        return (field == value) & (pc.divide(pc.scalar(1.0), field) == math.copysign(math.inf, value))
     return pc.field(partition_field_name) == value
 
 

tests/catalog/test_catalog_behaviors.py

@@ -1345,6 +1345,30 @@ def test_append_nan_to_identity_partitioned_table(catalog: Catalog) -> None:
     assert sorted(tbl.scan(row_filter="value is nan").to_arrow()["id"].to_pylist()) == [2, 4]
 
 
+def test_append_signed_zero_to_identity_partitioned_table(catalog: Catalog) -> None:
+    catalog.create_namespace("default")
+    identifier = f"default.append_signed_zero_identity_partition_{catalog.name}"
+    iceberg_schema = Schema(
+        NestedField(1, "id", IntegerType(), required=False),
+        NestedField(2, "value", DoubleType(), required=False),
+    )
+    partition_spec = PartitionSpec(
+        PartitionField(source_id=2, field_id=1000, transform=IdentityTransform(), name="value"),
+    )
+    tbl = catalog.create_table(identifier=identifier, schema=iceberg_schema, partition_spec=partition_spec)
+
+    tbl.append(
+        pa.Table.from_pydict(
+            {"id": [1, 2, 3], "value": [-0.0, 0.0, 1.0]},
+            schema=schema_to_pyarrow(iceberg_schema),
+        )
+    )
+
+    # Each row is written once, although -0.0 and 0.0 compare equal
+    assert sorted(tbl.scan().to_arrow()["id"].to_pylist()) == [1, 2, 3]
+    assert sorted(tbl.scan(row_filter="value = 0").to_arrow()["id"].to_pylist()) == [1, 2]
+
+
 def test_scan_integer_column_with_decimal_literal(catalog: Catalog) -> None:
     catalog.create_namespace("default")
     identifier = f"default.scan_integer_decimal_literal_{catalog.name}"

tests/io/test_pyarrow.py

@@ -3287,6 +3287,27 @@ def test_determine_partitions_identity_nan() -> None:
     assert rows_by_partition == {"1.0": [1], "nan": [2, 4], "None": [3]}
 
 
+def test_determine_partitions_identity_signed_zero() -> None:
+    for field_type in (DoubleType(), FloatType()):
+        schema = Schema(
+            NestedField(id=1, name="id", field_type=IntegerType(), required=False),
+            NestedField(id=2, name="value", field_type=field_type, required=False),
+        )
+        spec = PartitionSpec(PartitionField(source_id=2, field_id=1000, transform=IdentityTransform(), name="value"))
+        arrow_table = pa.Table.from_pydict(
+            {"id": [1, 2, 3, 4], "value": [-0.0, 0.0, 1.0, -0.0]},
+            schema=schema.as_arrow(),
+        )
+
+        partitions = list(_determine_partitions(spec, schema, arrow_table))
+
+        # -0.0 == 0.0, but they are separate partitions, so each row must land in exactly one of them
+        rows_by_partition = {
+            repr(p.partition_key.partition[0]): sorted(p.arrow_table_partition["id"].to_pylist()) for p in partitions
+        }
+        assert rows_by_partition == {"-0.0": [1, 4], "0.0": [2], "1.0": [3]}, field_type
+
+
 @pytest.mark.parametrize("transform", [IdentityTransform(), BucketTransform(4), TruncateTransform(1)])
 def test_determine_partitions_with_view_types(transform: Transform[Any, Any]) -> None:
     schema = Schema(