This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-4136-46669fad031d5b4e7a71c1df13940d17bbc9e71d in repository https://gitbox.apache.org/repos/asf/iceberg-python.git
commit e6940798b32f14fc7388eb36066e43457d678f6d Author: ump45nose <[email protected]> AuthorDate: Sat Oct 24 07:01:41 2026 +0000 Support reading Avro timestamp-nanos logical type (#4136) * Support reading Avro timestamp-nanos logical type * Use a loop instead of parametrize in timestamp-nanos conversion test Co-authored-by: Copilot App <[email protected]> --------- Co-authored-by: ump45nose <[email protected]> Co-authored-by: Kevin Liu <[email protected]> Co-authored-by: Kevin Liu <[email protected]> Co-authored-by: Copilot App <[email protected]> --- pyiceberg/utils/schema_conversion.py | 11 +++++++++-- tests/avro/test_file.py | 27 +++++++++++++++++++++++++++ tests/utils/test_schema_conversion.py | 25 +++++++++++++++++++++++++ 3 files changed, 61 insertions(+), 2 deletions(-) diff --git a/pyiceberg/utils/schema_conversion.py b/pyiceberg/utils/schema_conversion.py index 7e6343141..f19ff6026 100644 --- a/pyiceberg/utils/schema_conversion.py +++ b/pyiceberg/utils/schema_conversion.py @@ -48,7 +48,9 @@ from pyiceberg.types import ( PrimitiveType, StringType, StructType, + TimestampNanoType, TimestampType, + TimestamptzNanoType, TimestamptzType, TimeType, UnknownType, @@ -381,6 +383,11 @@ class AvroSchemaConversion: return TimestamptzType() else: return TimestampType() + elif logical_type == "timestamp-nanos": + if avro_logical_type.get("adjust-to-utc", False) is True: + return TimestamptzNanoType() + else: + return TimestampNanoType() elif (logical_type, physical_type) in LOGICAL_FIELD_TYPE_MAPPING: return LOGICAL_FIELD_TYPE_MAPPING[(logical_type, physical_type)] else: @@ -618,13 +625,13 @@ class ConvertSchemaToAvro(SchemaVisitorPerPrimitiveType[AvroType]): def visit_timestamp(self, timestamp_type: TimestampType) -> AvroType: return {"type": "long", "logicalType": "timestamp-micros", "adjust-to-utc": False} - def visit_timestamp_ns(self, timestamp_type: TimestampType) -> AvroType: + def visit_timestamp_ns(self, timestamp_ns_type: TimestampNanoType) -> AvroType: return {"type": "long", "logicalType": "timestamp-nanos", "adjust-to-utc": False} def visit_timestamptz(self, timestamptz_type: TimestamptzType) -> AvroType: return {"type": "long", "logicalType": "timestamp-micros", "adjust-to-utc": True} - def visit_timestamptz_ns(self, timestamptz_type: TimestamptzType) -> AvroType: + def visit_timestamptz_ns(self, timestamptz_ns_type: TimestamptzNanoType) -> AvroType: return {"type": "long", "logicalType": "timestamp-nanos", "adjust-to-utc": True} def visit_string(self, string_type: StringType) -> AvroType: diff --git a/tests/avro/test_file.py b/tests/avro/test_file.py index 175089c30..6cc190c33 100644 --- a/tests/avro/test_file.py +++ b/tests/avro/test_file.py @@ -51,7 +51,9 @@ from pyiceberg.types import ( LongType, NestedField, StringType, + TimestampNanoType, TimestampType, + TimestamptzNanoType, TimestamptzType, TimeType, UUIDType, @@ -455,6 +457,31 @@ def test_all_primitive_types(is_required: bool) -> None: assert record[idx] == avro_entry_read_with_fastavro[idx], f"Invalid {field} read with fastavro" +def test_timestamp_nanos_round_trip() -> None: + schema = Schema( + NestedField(field_id=1, name="ts_ns", field_type=TimestampNanoType(), required=True), + NestedField(field_id=2, name="tstz_ns", field_type=TimestamptzNanoType(), required=False), + schema_id=1, + ) + records = [ + Record(1677629965123456789, 1677629965123456789), + Record(-1, None), + ] + + with TemporaryDirectory() as tmpdir: + tmp_avro_file = tmpdir + "/timestamp_nanos.avro" + with avro.AvroOutputFile[Record](PyArrowFileIO().new_output(tmp_avro_file), schema, "timestamp_nanos") as out: + out.write_block(records) + + with avro.AvroFile[Record]( + PyArrowFileIO().new_input(tmp_avro_file), + schema, + {-1: Record}, + ) as avro_reader: + assert avro_reader.schema == schema + assert list(avro_reader) == records + + def manifest_entry(index: int) -> ManifestEntry: return ManifestEntry.from_args( status=ManifestEntryStatus.ADDED, diff --git a/tests/utils/test_schema_conversion.py b/tests/utils/test_schema_conversion.py index b110f92bd..a51b07075 100644 --- a/tests/utils/test_schema_conversion.py +++ b/tests/utils/test_schema_conversion.py @@ -33,7 +33,9 @@ from pyiceberg.types import ( NestedField, StringType, StructType, + TimestampNanoType, TimestampType, + TimestamptzNanoType, UnknownType, UUIDType, ) @@ -347,6 +349,29 @@ def test_convert_timestamp_micros_type() -> None: assert actual == TimestampType() +def test_convert_timestamp_nanos_type() -> None: + for adjust_to_utc, expected in [ + (None, TimestampNanoType()), + (False, TimestampNanoType()), + (True, TimestamptzNanoType()), + ]: + avro_logical_type: dict[str, Any] = {"type": "long", "logicalType": "timestamp-nanos"} + if adjust_to_utc is not None: + avro_logical_type["adjust-to-utc"] = adjust_to_utc + actual = AvroSchemaConversion()._convert_logical_type(avro_logical_type) + assert actual == expected, f"adjust-to-utc={adjust_to_utc}" + + +def test_timestamp_nanos_avro_schema_round_trip() -> None: + schema = Schema( + NestedField(field_id=1, name="ts_ns", field_type=TimestampNanoType(), required=True), + NestedField(field_id=2, name="tstz_ns", field_type=TimestamptzNanoType(), required=False), + ) + avro_schema = AvroSchemaConversion().iceberg_to_avro(schema) + assert isinstance(avro_schema, dict) + assert AvroSchemaConversion().avro_to_iceberg(avro_schema) == schema + + def test_unknown_logical_type() -> None: """Test raising a ValueError when converting an unknown logical type as part of an Avro schema conversion""" avro_logical_type = {"type": "bytes", "logicalType": "date"}
