[python] Align temporal string schema evolution with Java and Native

paimon-python/pypaimon/casting/row_to_string.py

@@ -38,6 +38,7 @@
 
 from typing import Any, Optional
 
+from pypaimon.casting.temporal_string import format_time, format_timestamp
 from pypaimon.table.row.generic_row import _parse_type_precision_scale
 
 _UNSUPPORTED_TYPES = ("FLOAT", "REAL", "DOUBLE", "TIMESTAMP_LTZ",
@@ -107,39 +108,9 @@ def _is_unsupported(data_type) -> bool:
 
 
 def _format_timestamp(value, precision: int) -> str:
-    """Format as ``yyyy-MM-dd HH:mm:ss[.fraction]``.
-
-    The fraction is padded to nine digits and then stripped of trailing zeros
-    down to ``precision`` digits, as ``DateTimeUtils.formatTimestamp`` does.
-    Python datetimes are microsecond resolution, so the last three digits of a
-    TIMESTAMP(7..9) value are always zero.
-    """
-    text = "{:04d}-{:02d}-{:02d} {:02d}:{:02d}:{:02d}".format(
-        value.year, value.month, value.day,
-        value.hour, value.minute, value.second)
-    fraction = "{:09d}".format(value.microsecond * 1000)
-    while len(fraction) > precision and fraction.endswith("0"):
-        fraction = fraction[:-1]
-    return (text + "." + fraction) if fraction else text
+    return format_timestamp(value, precision)
 
 
 def _format_time(value, precision: int) -> str:
-    """Format as ``HH:mm:ss[.fraction]``.
-
-    Digit by digit off the millisecond part, exactly as
-    ``DateTimeUtils.formatTimestampMillis`` does: it stops only once nothing
-    but zeros is left, so a truncated non zero tail keeps the zero before it
-    (``.10`` for 101 ms at precision 2, not ``.1``).
-    """
-    text = "{:02d}:{:02d}:{:02d}".format(value.hour, value.minute, value.second)
-    if precision <= 0:
-        return text
-    millis = value.microsecond // 1000
-    digits = []
-    while precision > 0:
-        digits.append(str(millis // 100))
-        millis = millis % 100 * 10
-        if millis == 0:
-            break
-        precision -= 1
-    return text + "." + "".join(digits)
+    millis = ((value.hour * 60 + value.minute) * 60 + value.second) * 1000 + value.microsecond // 1000
+    return format_time(millis, precision)

paimon-python/pypaimon/casting/schema_evolution.py

@@ -22,6 +22,7 @@
 
 import pyarrow as pa
 
+from pypaimon.casting.temporal_string import temporal_to_string
 from pypaimon.table.row.generic_row import _parse_type_precision_scale
 
 
@@ -122,6 +123,10 @@ def cast_array_for_schema_evolution(array, target_type, source_paimon_type=None,
     and saturate to int/long before byte/short narrowing.
     DECIMAL scale reduction uses HALF_UP and returns NULL on precision overflow.
     """
+    if pa.types.is_string(target_type):
+        strings = temporal_to_string(array, source_paimon_type, target_paimon_type)
+        if strings is not None:
+            return strings
     timestamp = _truncate_timestamp(array, target_type, source_paimon_type, target_paimon_type)
     if timestamp is not None:
         return timestamp

paimon-python/pypaimon/casting/temporal_string.py

@@ -0,0 +1,143 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Java DateTimeUtils text shared by schema evolution and row rendering."""
+
+import time
+
+import pyarrow as pa
+
+from pypaimon.table.row.generic_row import _parse_type_precision_scale
+
+
+def _epoch_day_to_date(days):
+    # March-based Gregorian eras, with floor division also for pre-epoch days.
+    days += 719468
+    era, day_of_era = divmod(days, 146097)
+    year_of_era = (day_of_era - day_of_era // 1460 + day_of_era // 36524 - day_of_era // 146096) // 365
+    year = year_of_era + era * 400
+    day_of_year = day_of_era - (365 * year_of_era + year_of_era // 4 - year_of_era // 100)
+    march_month = (5 * day_of_year + 2) // 153
+    day = day_of_year - (153 * march_month + 2) // 5 + 1
+    month = march_month + (3 if march_month < 10 else -9)
+    return year + (month <= 2), month, day
+
+
+def _local_offset(seconds):
+    # Avoid converting the date itself through datetime's 1..9999 year range.
+    local = time.localtime(seconds)
+    offset = getattr(local, 'tm_gmtoff', None)
+    if offset is not None:
+        return offset
+    return -(time.altzone if local.tm_isdst > 0 else time.timezone)
+
+
+def format_timestamp(value, precision, nanos=None):
+    """Keep significant fraction digits, with declared precision as a minimum."""
+    return format_timestamp_parts(value.year, value.month, value.day, value.hour, value.minute,
+                                  value.second, value.microsecond * 1000 if nanos is None else nanos, precision)
+
+
+def format_timestamp_parts(year, month, day, hour, minute, second, nanos, precision):
+    # DateTimeUtils.int4 writes four individual digits, even for years outside
+    # 0000..9999. Preserve Java's signed remainder rather than Python's modulo.
+    digits = []
+    for divisor in (1000, 100, 10, 1):
+        part = abs(year) // divisor
+        part = -(part % 10) if year < 0 else part % 10
+        digits.append(chr(ord('0') + part))
+    text = "{}-{:02d}-{:02d} {:02d}:{:02d}:{:02d}".format(
+        ''.join(digits), month, day, hour, minute, second)
+    fraction = "{:09d}".format(nanos)
+    while len(fraction) > precision and fraction.endswith("0"):
+        fraction = fraction[:-1]
+    return text + "." + fraction if fraction else text
+
+
+def format_time(millis, precision):
+    """Java TIME has millisecond storage, independent of declared precision."""
+    if millis < 0:
+        millis %= 86400000
+    seconds, fraction = divmod(millis, 1000)
+    hours, seconds = divmod(seconds, 3600)
+    minutes, seconds = divmod(seconds, 60)
+    text = "{:02d}:{:02d}:{:02d}".format(hours % 100, minutes, seconds)
+    if precision <= 0:
+        return text
+    digits = []
+    while precision > 0:
+        digits.append(str(fraction // 100))
+        fraction = fraction % 100 * 10
+        if fraction == 0:
+            break
+        precision -= 1
+    return text + "." + "".join(digits)
+
+
+def temporal_to_string(array, source_type=None, target_type=None):
+    """Return None for non-temporal input; preserve NULLs, slices and nanos.
+
+    LTZ uses the process local zone, as Java TimeZone.getDefault() does.
+    Character length is applied after formatting, using the target Paimon type.
+    """
+    timestamp = pa.types.is_timestamp(array.type)
+    if not timestamp and not pa.types.is_time(array.type):
+        return None
+    unit = array.type.unit
+    precision = (_parse_type_precision_scale(source_type)[0] if source_type is not None
+                 else {'s': 0, 'ms': 3, 'us': 6, 'ns': 9}[unit])
+    if timestamp:
+        raw = array.cast(pa.int64()).to_pylist()
+        ticks = {'s': 1, 'ms': 1000, 'us': 1000000, 'ns': 1000000000}[unit]
+        source_name = str(source_type).upper() if source_type is not None else ''
+        local_zone = (array.type.tz is not None if source_type is None else
+                      source_name.startswith('TIMESTAMP_LTZ') or 'WITH LOCAL TIME ZONE' in source_name)
+    else:
+        raw = array.cast(pa.time32('ms'), safe=False).cast(pa.int32()).to_pylist()
+    name = str(target_type).upper() if target_type is not None else 'STRING'
+    length = (_parse_type_precision_scale(target_type)[0] or 1) if name.startswith(('CHAR', 'VARCHAR')) else None
+    values = []
+    for value in raw:
+        if value is None:
+            values.append(None)
+            continue
+        if timestamp:
+            millis = value * 1000 if unit == 's' else value // (ticks // 1000)
+            # Arrow seconds may exceed Java's signed millisecond storage.
+            millis = (millis + (1 << 63)) % (1 << 64) - (1 << 63)
+            seconds = millis // 1000
+            if local_zone:
+                millis += _local_offset(seconds) * 1000
+                millis = (millis + (1 << 63)) % (1 << 64) - (1 << 63)
+                seconds = millis // 1000
+            sub_milli = value % (ticks // 1000) * (1000000000 // ticks) if ticks > 1000 else 0
+            nanos = millis % 1000 * 1000000 + sub_milli
+            day, second_of_day = divmod(seconds, 86400)
+            day = (day + (1 << 31)) % (1 << 32) - (1 << 31)
+            year, month, day = _epoch_day_to_date(day)
+            hour, second_of_day = divmod(second_of_day, 3600)
+            minute, second = divmod(second_of_day, 60)
+            text = format_timestamp_parts(year, month, day, hour, minute, second,
+                                          nanos, precision)
+        else:
+            text = format_time(value, precision)
+        if length is not None:
+            text = text[:length]
+            if name.startswith('CHAR'):
+                text = text.ljust(length)
+        values.append(text)
+    return pa.array(values, type=pa.string())

paimon-python/pypaimon/read/reader/data_file_batch_reader.py

@@ -60,7 +60,7 @@ def _to_string_values(array, data_type) -> list:
     """Render *array* as a list of per-row strings (None for NULL rows)."""
     if isinstance(data_type, (RowType, ArrayType, MapType)):
         return _constructed_to_string_array(array, data_type).to_pylist()
-    return array.cast(pa.string(), safe=False).to_pylist()
+    return cast_array_for_schema_evolution(array, pa.string(), data_type, AtomicType('STRING')).to_pylist()
 
 
 def _constructed_to_string_array(array, file_type):

paimon-python/pypaimon/tests/native_schema_evolution_cast_test.py

@@ -18,6 +18,7 @@
 """Read existing Python schema changes through Native using REST tables."""
 
 from contextlib import ExitStack
+from datetime import datetime, timezone
 from decimal import Decimal
 from unittest.mock import patch
 
@@ -26,7 +27,7 @@
 
 from pypaimon import Schema
 from pypaimon.read.native_plan import native_runtime_available
-from pypaimon.schema.data_types import AtomicType, PyarrowFieldParser
+from pypaimon.schema.data_types import AtomicType, DataField, MapType, PyarrowFieldParser, RowType
 from pypaimon.schema.schema_change import SchemaChange
 from pypaimon.tests import native_plan_rest_test
 
@@ -48,15 +49,18 @@ def _write(table, data):
         commit.close()
 
 
-def _table(catalog, mode, arrow_type, values):
+def _table(catalog, mode, arrow_type, values, paimon_type=None):
     options = {'file.format': 'parquet', 'read.batch-size': '2'}
     if mode == 'pk':
         options['bucket'] = '1'
     elif mode == 'de':
         options.update({'data-evolution.enabled': 'true', 'row-tracking.enabled': 'true'})
     schema = pa.schema([('id', pa.int64()), ('v', arrow_type)])
-    catalog.create_table('default.t', Schema.from_pyarrow_schema(
-        schema, primary_keys=['id'] if mode == 'pk' else [], options=options), False)
+    table_schema = Schema.from_pyarrow_schema(
+        schema, primary_keys=['id'] if mode == 'pk' else [], options=options)
+    if paimon_type is not None:
+        table_schema.fields[1].type = paimon_type
+    catalog.create_table('default.t', table_schema, False)
     table = catalog.get_table('default.t')
     _write(table, pa.table({'id': list(range(len(values))), 'v': pa.array(values, type=arrow_type)}))
 
@@ -81,6 +85,66 @@ def _read(table, native, predicate=None, projection=None):
     return result.sort_by([('id', 'ascending')])
 
 
+_TEMPORAL_STRING_CASES = [
+    (pa.time32('ms'), [0, 100, 101, 12345678, None],
+     ['00:00:00.0', '00:00:00.1', '00:00:00.101', '03:25:45.678', None]),
+    (pa.timestamp('ms'), [0, 100, -1, 12345678, None],
+     ['1970-01-01 00:00:00.000', '1970-01-01 00:00:00.100',
+      '1969-12-31 23:59:59.999', '1970-01-01 03:25:45.678', None]),
+    (pa.timestamp('ns'), [0, 123456789, -1, 1000000000, None],
+     ['1970-01-01 00:00:00.000000000', '1970-01-01 00:00:00.123456789',
+      '1969-12-31 23:59:59.999999999', '1970-01-01 00:00:01.000000000', None]),
+]
+
+
+@pytest.mark.parametrize('mode', ['append', 'pk', 'de'])
+@pytest.mark.parametrize('source_type,values,expected', _TEMPORAL_STRING_CASES)
+@pytest.mark.parametrize('target_name', ['STRING', 'VARCHAR(12)', 'CHAR(35)'])
+def test_temporal_character_strings_read_old_rest_files(
+        rest_catalog, mode, source_type, values, expected, target_name):
+    _check_temporal_string_reads(rest_catalog, mode, source_type, values, expected, target_name)
+
+
+def _check_temporal_string_reads(rest_catalog, mode, source_type, values, expected, target_name):
+    catalog, _ = rest_catalog
+    _table(catalog, mode, source_type, values,
+           AtomicType('TIME(3)') if pa.types.is_time(source_type) else None)
+    catalog.alter_table('default.t', [SchemaChange.update_column_type('v', AtomicType(target_name))], False)
+    table = catalog.get_table('default.t')
+    if target_name == 'VARCHAR(12)':
+        expected = [value[:12] if value is not None else None for value in expected]
+    elif target_name == 'CHAR(35)':
+        expected = [value.ljust(35) if value is not None else None for value in expected]
+    predicate = table.new_read_builder().new_predicate_builder().equal('v', expected[0])
+    for native in (False, True):
+        assert _read(table, native).column('v').to_pylist() == expected
+        matches = [index for index, value in enumerate(expected) if value == expected[0]]
+        assert _read(table, native, predicate, ['id']).column('id').to_pylist() == matches
+
+
+@pytest.mark.parametrize('mode', ['append', 'de'])
+@pytest.mark.parametrize('target_name', ['STRING', 'VARCHAR(12)', 'CHAR(35)'])
+def test_wide_timestamp_strings_read_old_rest_files(rest_catalog, mode, target_name):
+    # The existing Python PK writer materializes datetimes while merging rows,
+    # so it cannot create this input; append and DE retain the raw Arrow ticks.
+    _check_temporal_string_reads(rest_catalog, mode, pa.timestamp('ms'),
+                                 [253402300800000, -62167219200000, 2 ** 63 - 1, -(2 ** 63), None],
+                                 ['0000-01-01 00:00:00.000', '0000-01-01 00:00:00.000',
+                                  '/+-.-04-06 07:12:55.807', '5471-09-26 16:47:04.192', None], target_name)
+
+
+@pytest.mark.parametrize('mode', ['append', 'pk', 'de'])
+def test_ltz_string_uses_local_zone_without_zone_suffix(rest_catalog, mode):
+    catalog, _ = rest_catalog
+    # Check a winter and a summer instant: the system zone may observe DST.
+    instants = [datetime(2024, 1, 1, tzinfo=timezone.utc), datetime(2024, 7, 1, tzinfo=timezone.utc)]
+    expected = [value.astimezone().strftime('%Y-%m-%d %H:%M:%S') + '.000' for value in instants] + [None]
+    _table(catalog, mode, pa.timestamp('ms', 'UTC'), instants + [None])
+    catalog.alter_table('default.t', [SchemaChange.update_column_type('v', AtomicType('STRING'))], False)
+    for native in (False, True):
+        assert _read(catalog.get_table('default.t'), native).column('v').to_pylist() == expected
+
+
 _NUMERIC_CASES = [
     (pa.int64(), 'TINYINT', [128, -129, 2 ** 63 - 1, -(2 ** 63), None], [-128, 127, -1, 0, None]),
     (pa.int64(), 'SMALLINT', [32768, -32769, 2 ** 63 - 1, None], [-32768, 32767, -1, None]),
@@ -335,3 +399,49 @@ def test_nested_epoch_seconds_evolution_and_projection(rest_catalog, mode, integ
         assert selected.column('m_positive').cast(pa.int64()).to_pylist() == [42000, None, None]
         assert selected.column('m_negative').cast(pa.int64()).to_pylist() == [-42000, None, None]
         assert selected.column('m_missing').null_count == 3
+
+
+_NESTED_TEMPORAL_STRING_CASES = [
+    (pa.list_(pa.timestamp('ns')), None, [[0, -1, None], None, [], [1000000000]],
+     ['[1970-01-01 00:00:00.000000000, 1969-12-31 23:59:59.999999999, null]', None,
+      '[]', '[1970-01-01 00:00:01.000000000]']),
+    (pa.map_(pa.string(), pa.time32('ms')), MapType(True, AtomicType('STRING'), AtomicType('TIME(3)')),
+     [[('x', 0), ('y', 101), ('n', None)], None, [], [('p', 100)]],
+     ['{x -> 00:00:00.0, y -> 00:00:00.101, n -> null}', None, '{}', '{p -> 00:00:00.1}']),
+    (pa.struct([('t', pa.time32('ms')), ('ts', pa.timestamp('ns'))]),
+     RowType(True, [DataField(2, 't', AtomicType('TIME(3)')), DataField(3, 'ts', AtomicType('TIMESTAMP(9)'))]),
+     [{'t': 0, 'ts': -1}, None, {'t': None, 'ts': None}, {'t': 100, 'ts': 0}],
+     ['{00:00:00.0, 1969-12-31 23:59:59.999999999}', None, '{null, null}',
+      '{00:00:00.1, 1970-01-01 00:00:00.000000000}']),
+]
+
+
+@pytest.mark.parametrize('mode', ['append', 'pk', 'de'])
+@pytest.mark.parametrize('target_name', ['STRING', 'CHAR(8)'])
+@pytest.mark.parametrize('arrow_type,paimon_type,values,expected', _NESTED_TEMPORAL_STRING_CASES)
+def test_constructed_string_evolution_uses_temporal_child_casts(
+        rest_catalog, mode, target_name, arrow_type, paimon_type, values, expected):
+    catalog, _ = rest_catalog
+    _table(catalog, mode, arrow_type, values, paimon_type)
+    catalog.alter_table('default.t', [SchemaChange.update_column_type('v', AtomicType(target_name))], False)
+    table = catalog.get_table('default.t')
+    # Java's constructed cast does not trim/pad the overall result. Its scalar
+    # children use STRING, retaining their own declared temporal precision.
+    for native in (False, True):
+        assert _read(table, native).column('v').to_pylist() == expected
+
+
+@pytest.mark.parametrize('mode', ['append', 'de'])
+@pytest.mark.parametrize('target_name', ['STRING', 'VARCHAR(12)', 'CHAR(35)'])
+def test_wide_ltz_string_does_not_fail_a_mixed_batch(rest_catalog, mode, target_name):
+    catalog, _ = rest_catalog
+    values = [0, 2 ** 63 - 1, None, -(2 ** 63), 2 ** 63 - 1 - 181 * 86400000]
+    _table(catalog, mode, pa.timestamp('ms', 'UTC'), values)
+    catalog.alter_table('default.t', [SchemaChange.update_column_type('v', AtomicType(target_name))], False)
+    table = catalog.get_table('default.t')
+    # The Python executor supports the complete signed millisecond range. Core
+    # unit tests check its matching Java strings for UTC, Shanghai and DST zones.
+    expected = _read(table, False).column('v').to_pylist()
+    assert expected[2] is None
+    assert all(value is not None for i, value in enumerate(expected) if i != 2)
+    assert _read(table, True).column('v').to_pylist() == expected