|
22 | 22 | import uuid |
23 | 23 | import warnings |
24 | 24 | from collections.abc import Iterator |
25 | | -from datetime import date, datetime, timezone |
| 25 | +from datetime import date, datetime, time, timezone |
26 | 26 | from pathlib import Path |
27 | 27 | from typing import Any |
28 | 28 | from unittest.mock import MagicMock, patch |
|
93 | 93 | from pyiceberg.table import FileScanTask, TableProperties, WriteTask |
94 | 94 | from pyiceberg.table.metadata import TableMetadataV2 |
95 | 95 | from pyiceberg.table.name_mapping import create_mapping_from_schema |
96 | | -from pyiceberg.transforms import HourTransform, IdentityTransform |
| 96 | +from pyiceberg.transforms import BucketTransform, HourTransform, IdentityTransform |
97 | 97 | from pyiceberg.typedef import UTF8, Properties, Record, TableVersion |
98 | 98 | from pyiceberg.types import ( |
99 | 99 | BinaryType, |
@@ -2784,6 +2784,40 @@ def test_partition_for_demo() -> None: |
2784 | 2784 | ) |
2785 | 2785 |
|
2786 | 2786 |
|
| 2787 | +@pytest.mark.parametrize("format_version", [1, 2]) |
| 2788 | +def test_append_time_bucket_partition(tmp_path: Path, format_version: int) -> None: |
| 2789 | + schema = Schema(NestedField(1, "event_time", TimeType(), required=False)) |
| 2790 | + spec = PartitionSpec(PartitionField(1, 1000, BucketTransform(16), "event_time_bucket")) |
| 2791 | + data = pa.table( |
| 2792 | + {"event_time": [time(0), time(12, 34, 56, 123456), time(23, 59, 59, 999999), None, time(0)]}, |
| 2793 | + schema=schema.as_arrow(), |
| 2794 | + ) |
| 2795 | + |
| 2796 | + with InMemoryCatalog("test", warehouse=tmp_path.as_uri()) as catalog: |
| 2797 | + catalog.create_namespace("default") |
| 2798 | + table = catalog.create_table( |
| 2799 | + "default.events", schema=schema, partition_spec=spec, properties={"format-version": str(format_version)} |
| 2800 | + ) |
| 2801 | + table.append(data) |
| 2802 | + |
| 2803 | + rows = table.scan().to_arrow()["event_time"].to_pylist() |
| 2804 | + assert rows.count(None) == 1 |
| 2805 | + assert sorted(value for value in rows if value is not None) == sorted( |
| 2806 | + value for value in data["event_time"].to_pylist() if value is not None |
| 2807 | + ) |
| 2808 | + |
| 2809 | + files = table.inspect.data_files().to_pylist() |
| 2810 | + assert {file["partition"]["event_time_bucket"]: file["record_count"] for file in files} == { |
| 2811 | + 12: 2, |
| 2812 | + 11: 1, |
| 2813 | + 8: 1, |
| 2814 | + None: 1, |
| 2815 | + } |
| 2816 | + for file in files: |
| 2817 | + bucket = file["partition"]["event_time_bucket"] |
| 2818 | + assert f"/event_time_bucket={bucket if bucket is not None else 'null'}/" in file["file_path"] |
| 2819 | + |
| 2820 | + |
2787 | 2821 | def test_partition_for_nested_field() -> None: |
2788 | 2822 | schema = Schema( |
2789 | 2823 | NestedField(id=1, name="foo", field_type=StringType(), required=True), |
|
0 commit comments