This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 11905d8680 [python] Write TIMESTAMP(0) columns to Avro as
timestamp-millis (#10193)
11905d8680 is described below
commit 11905d8680c00cbadc5ef2df83634ab2088b11f3
Author: jackylee <[email protected]>
AuthorDate: Sat Oct 3 17:09:55 2026 +0800
[python] Write TIMESTAMP(0) columns to Avro as timestamp-millis (#10193)
---
paimon-python/pypaimon/filesystem/local_file_io.py | 25 +++++++-
paimon-python/pypaimon/schema/data_types.py | 6 +-
paimon-python/pypaimon/tests/data_types_test.py | 75 ++++++++++++++++++++++
3 files changed, 101 insertions(+), 5 deletions(-)
diff --git a/paimon-python/pypaimon/filesystem/local_file_io.py
b/paimon-python/pypaimon/filesystem/local_file_io.py
index e3530c1951..45bb5421d3 100644
--- a/paimon-python/pypaimon/filesystem/local_file_io.py
+++ b/paimon-python/pypaimon/filesystem/local_file_io.py
@@ -35,6 +35,27 @@ from pypaimon.schema.data_types import DataField,
AtomicType, PyarrowFieldParser
from pypaimon.write.blob_format_writer import BlobFormatWriter
+def _normalize_naive_datetimes(value):
+ """Recursively attach UTC to naive datetimes, including ones nested inside
+ ROW (dict) and ARRAY/MAP (list/tuple) values.
+
+ fastavro converts a naive datetime using the host timezone, which corrupts
+ the stored instant on a non-UTC host; Paimon timestamps are UTC-based.
+ ``write_avro`` previously normalized only top-level column values, so a
+ naive datetime inside a struct or list was written shifted by the host
+ offset. Normalizing recursively keeps nested timestamps correct.
+ """
+ if isinstance(value, datetime):
+ if value.tzinfo is None:
+ return value.replace(tzinfo=timezone.utc)
+ return value
+ if isinstance(value, dict):
+ return {key: _normalize_naive_datetimes(val) for key, val in
value.items()}
+ if isinstance(value, (list, tuple)):
+ return type(value)(_normalize_naive_datetimes(val) for val in value)
+ return value
+
+
def _file_uri_path(parsed, windows=None) -> str:
"""Decode a file URI path, preserving Windows drives and UNC hosts."""
if windows is None:
@@ -358,9 +379,7 @@ class LocalFileIO(FileIO):
record = {}
for col in records_dict.keys():
value = records_dict[col][i]
- if isinstance(value, datetime) and value.tzinfo is None:
- value = value.replace(tzinfo=timezone.utc)
- record[col] = value
+ record[col] = _normalize_naive_datetimes(value)
yield record
records = record_generator()
diff --git a/paimon-python/pypaimon/schema/data_types.py
b/paimon-python/pypaimon/schema/data_types.py
index c73223b883..f4b176fe6f 100755
--- a/paimon-python/pypaimon/schema/data_types.py
+++ b/paimon-python/pypaimon/schema/data_types.py
@@ -900,14 +900,16 @@ class PyarrowFieldParser:
elif pyarrow.types.is_timestamp(field_type):
unit = field_type.unit
if field_type.tz is None:
- if unit == 'ms':
+ if unit in ('s', 'ms'):
+ # Avro's coarsest timestamp is millis; seconds map to it
+ # losslessly, matching Java AvroSchemaConverter
(precision<=3).
return {"type": "long", "logicalType": "timestamp-millis"}
elif unit == 'us':
return {"type": "long", "logicalType": "timestamp-micros"}
else:
raise ValueError(f"Avro does not support pyarrow timestamp
with unit {unit}.")
else:
- if unit == 'ms':
+ if unit in ('s', 'ms'):
return {"type": "long", "logicalType":
"local-timestamp-millis"}
elif unit == 'us':
return {"type": "long", "logicalType":
"local-timestamp-micros"}
diff --git a/paimon-python/pypaimon/tests/data_types_test.py
b/paimon-python/pypaimon/tests/data_types_test.py
index 8ccef3a165..8d0957fac0 100755
--- a/paimon-python/pypaimon/tests/data_types_test.py
+++ b/paimon-python/pypaimon/tests/data_types_test.py
@@ -336,3 +336,78 @@ class DataTypesTest(unittest.TestCase):
paimon_type = PyarrowFieldParser.to_paimon_type(pa.time32('ms'),
nullable=True)
self.assertEqual(paimon_type.type, "TIME(0)")
+
+ def test_avro_timestamp_seconds_maps_and_roundtrips(self):
+ import datetime
+ import os
+ import tempfile
+
+ import fastavro
+
+ from pypaimon.filesystem.local_file_io import LocalFileIO
+
+ # TIMESTAMP(0) -> pyarrow 's'; Avro's coarsest timestamp is millis,
which
+ # holds seconds losslessly and matches Java AvroSchemaConverter
(precision<=3).
+ self.assertEqual(
+ PyarrowFieldParser.to_avro_type(pa.timestamp('s'), 'ts', 'r'),
+ {"type": "long", "logicalType": "timestamp-millis"})
+ self.assertEqual(
+ PyarrowFieldParser.to_avro_type(pa.timestamp('s', tz='UTC'), 'ts',
'r'),
+ {"type": "long", "logicalType": "local-timestamp-millis"})
+ # nanos stays rejected: Avro has no nanos logical type (matches Java
precision>6).
+ with self.assertRaises(ValueError):
+ PyarrowFieldParser.to_avro_type(pa.timestamp('ns'), 'ts', 'r')
+
+ # A second-granularity value round-trips unchanged (no seconds/millis
mixup).
+ ts = datetime.datetime(2024, 1, 2, 3, 4, 5)
+ table = pa.table({"ts": pa.array([ts], pa.timestamp('s'))})
+ with tempfile.TemporaryDirectory() as d:
+ path = os.path.join(d, "data.avro")
+ LocalFileIO().write_avro(path, table)
+ with open(path, 'rb') as f:
+ rows = list(fastavro.reader(f))
+ self.assertEqual(rows[0]["ts"].replace(tzinfo=None), ts)
+
+ def test_avro_nested_timestamp_roundtrips_on_non_utc_host(self):
+ # A naive TIMESTAMP(0) nested in a ROW or ARRAY must survive a write on
+ # a non-UTC host. fastavro converts a naive datetime using the host
+ # timezone, so without recursive UTC normalization the nested values
+ # were persisted shifted by the host offset (reproduced by the review
+ # under TZ=Asia/Shanghai). The top-level value was already normalized.
+ import datetime
+ import os
+ import tempfile
+ import time
+
+ import fastavro
+
+ from pypaimon.filesystem.local_file_io import LocalFileIO
+
+ if not hasattr(time, "tzset"):
+ self.skipTest("time.tzset is unavailable on this platform")
+
+ ts = datetime.datetime(2024, 1, 2, 3, 4, 5)
+ table = pa.table({
+ "ts": pa.array([ts], pa.timestamp('s')),
+ "r": pa.array([{"ts": ts}], pa.struct([("ts",
pa.timestamp('s'))])),
+ "arr": pa.array([[ts]], pa.list_(pa.timestamp('s'))),
+ })
+ previous_tz = os.environ.get("TZ")
+ os.environ["TZ"] = "Asia/Shanghai"
+ time.tzset()
+ try:
+ with tempfile.TemporaryDirectory() as d:
+ path = os.path.join(d, "data.avro")
+ LocalFileIO().write_avro(path, table)
+ with open(path, 'rb') as f:
+ rows = list(fastavro.reader(f))
+ finally:
+ if previous_tz is None:
+ os.environ.pop("TZ", None)
+ else:
+ os.environ["TZ"] = previous_tz
+ time.tzset()
+
+ self.assertEqual(rows[0]["ts"].replace(tzinfo=None), ts)
+ self.assertEqual(rows[0]["r"]["ts"].replace(tzinfo=None), ts)
+ self.assertEqual(rows[0]["arr"][0].replace(tzinfo=None), ts)
