Core: Fix time bucket partition writes

pyiceberg/partitioning.py

@@ -546,8 +546,10 @@ def _(type: IcebergType, value: int | date | None) -> int | None:
 
 
 @_to_partition_representation.register(TimeType)
-def _(type: IcebergType, value: time | None) -> int | None:
-    return time_to_micros(value) if value is not None else None
+def _(type: IcebergType, value: int | time | None) -> int | None:
+    if value is None or isinstance(value, int):
+        return value
+    return time_to_micros(value)
 
 
 @_to_partition_representation.register(UUIDType)

tests/io/test_pyarrow.py

@@ -22,7 +22,7 @@
 import uuid
 import warnings
 from collections.abc import Iterator
-from datetime import date, datetime, timezone
+from datetime import date, datetime, time, timezone
 from pathlib import Path
 from typing import Any
 from unittest.mock import MagicMock, patch
@@ -93,7 +93,7 @@
 from pyiceberg.table import FileScanTask, TableProperties, WriteTask
 from pyiceberg.table.metadata import TableMetadataV2
 from pyiceberg.table.name_mapping import create_mapping_from_schema
-from pyiceberg.transforms import HourTransform, IdentityTransform
+from pyiceberg.transforms import BucketTransform, HourTransform, IdentityTransform
 from pyiceberg.typedef import UTF8, Properties, Record, TableVersion
 from pyiceberg.types import (
     BinaryType,
@@ -2784,6 +2784,40 @@ def test_partition_for_demo() -> None:
     )
 
 
+@pytest.mark.parametrize("format_version", [1, 2])
+def test_append_time_bucket_partition(tmp_path: Path, format_version: int) -> None:
+    schema = Schema(NestedField(1, "event_time", TimeType(), required=False))
+    spec = PartitionSpec(PartitionField(1, 1000, BucketTransform(16), "event_time_bucket"))
+    data = pa.table(
+        {"event_time": [time(0), time(12, 34, 56, 123456), time(23, 59, 59, 999999), None, time(0)]},
+        schema=schema.as_arrow(),
+    )
+
+    with InMemoryCatalog("test", warehouse=tmp_path.as_uri()) as catalog:
+        catalog.create_namespace("default")
+        table = catalog.create_table(
+            "default.events", schema=schema, partition_spec=spec, properties={"format-version": str(format_version)}
+        )
+        table.append(data)
+
+        rows = table.scan().to_arrow()["event_time"].to_pylist()
+        assert rows.count(None) == 1
+        assert sorted(value for value in rows if value is not None) == sorted(
+            value for value in data["event_time"].to_pylist() if value is not None
+        )
+
+        files = table.inspect.data_files().to_pylist()
+        assert {file["partition"]["event_time_bucket"]: file["record_count"] for file in files} == {
+            12: 2,
+            11: 1,
+            8: 1,
+            None: 1,
+        }
+        for file in files:
+            bucket = file["partition"]["event_time_bucket"]
+            assert f"/event_time_bucket={bucket if bucket is not None else 'null'}/" in file["file_path"]
+
+
 def test_partition_for_nested_field() -> None:
     schema = Schema(
         NestedField(id=1, name="foo", field_type=StringType(), required=True),

tests/test_transforms.py

@@ -17,7 +17,7 @@
 # under the License.
 # pylint: disable=eval-used,protected-access,redefined-outer-name
 from collections.abc import Callable
-from datetime import date, datetime
+from datetime import date, datetime, time
 from decimal import Decimal
 from typing import Annotated, Any
 from uuid import UUID
@@ -1643,6 +1643,22 @@ def test_to_partition_representation_timestamps(source_type: PrimitiveType, valu
     assert _to_partition_representation(source_type, value) == expected
 
 
+@pytest.mark.parametrize(
+    "value, expected",
+    [
+        pytest.param(None, None, id="none"),
+        pytest.param(time(0), 0, id="midnight"),
+        pytest.param(time(12, 34, 56, 123456), 45_296_123_456, id="microseconds"),
+        pytest.param(time(23, 59, 59, 999999), 86_399_999_999, id="end_of_day"),
+        pytest.param(0, 0, id="zero_int_passthrough"),
+        pytest.param(12, 12, id="bucket_int_passthrough"),
+        pytest.param(86_399_999_999, 86_399_999_999, id="time_int_passthrough"),
+    ],
+)
+def test_to_partition_representation_time(value: int | time | None, expected: int | None) -> None:
+    assert _to_partition_representation(TimeType(), value) == expected
+
+
 def test_to_partition_representation_unrecognized_type_raises() -> None:
     with pytest.raises(ValueError, match="Type not recognized"):
         _to_partition_representation(TimestampType(), "not-a-datetime-or-int")