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 cd47e65487 [Python] Align supported temporal schema evolution with
Java (#10524)
cd47e65487 is described below
commit cd47e65487d6726d9d1d9e0a4334dbf1cdf475f3
Author: Jingsong Lee <[email protected]>
AuthorDate: Tue Oct 13 11:01:43 2026 +0800
[Python] Align supported temporal schema evolution with Java (#10524)
---
paimon-python/pypaimon/casting/schema_evolution.py | 5 +
paimon-python/pypaimon/casting/temporal_cast.py | 95 +++++++++
paimon-python/pypaimon/casting/temporal_string.py | 25 +--
paimon-python/pypaimon/casting/temporal_utils.py | 108 +++++++++++
.../tests/native_temporal_zone_cast_test.py | 213 +++++++++++++++++++++
.../pypaimon/tests/schema_evolution_cast_test.py | 180 +++++++++++++++++
6 files changed, 602 insertions(+), 24 deletions(-)
diff --git a/paimon-python/pypaimon/casting/schema_evolution.py
b/paimon-python/pypaimon/casting/schema_evolution.py
index 3aabc54371..afa3650627 100644
--- a/paimon-python/pypaimon/casting/schema_evolution.py
+++ b/paimon-python/pypaimon/casting/schema_evolution.py
@@ -22,6 +22,7 @@ import math
import pyarrow as pa
+from pypaimon.casting.temporal_cast import cast_temporal
from pypaimon.casting.temporal_string import temporal_to_string
from pypaimon.table.row.generic_row import _parse_type_precision_scale
@@ -122,11 +123,15 @@ def cast_array_for_schema_evolution(array, target_type,
source_paimon_type=None,
Numeric narrowing preserves low bits, while floating-point values truncate
and saturate to int/long before byte/short narrowing.
DECIMAL scale reduction uses HALF_UP and returns NULL on precision
overflow.
+ LTZ conversions use the process local zone, including Java gap/overlap
rules.
"""
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
+ converted = cast_temporal(array, target_type, source_paimon_type,
target_paimon_type)
+ if converted is not None:
+ return converted
timestamp = _truncate_timestamp(array, target_type, source_paimon_type,
target_paimon_type)
if timestamp is not None:
return timestamp
diff --git a/paimon-python/pypaimon/casting/temporal_cast.py
b/paimon-python/pypaimon/casting/temporal_cast.py
new file mode 100644
index 0000000000..5700100446
--- /dev/null
+++ b/paimon-python/pypaimon/casting/temporal_cast.py
@@ -0,0 +1,95 @@
+# 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 local-zone conversions used by existing schema-evolution casts."""
+
+import pyarrow as pa
+
+from pypaimon.casting.temporal_utils import (
+ _epoch_day_to_java_date, _java_date_to_epoch_day,
+ _local_offset, _local_to_utc_seconds, _signed,
+)
+from pypaimon.table.row.generic_row import _parse_type_precision_scale
+
+
+def cast_temporal(array, target, source_type=None, target_type=None):
+ timestamp = pa.types.is_timestamp(array.type)
+ source_name = str(source_type).upper() if source_type is not None else ''
+ ltz = timestamp and (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)
+ target_ts = pa.types.is_timestamp(target)
+ cross = timestamp and target_ts and ltz != (target.tz is not None)
+ date = pa.types.is_date32(array.type) and target_ts and target.tz is not
None
+ numeric = array.type == pa.int64() and target_ts and target.tz is not None
+ if not (cross or date or numeric or timestamp and
(pa.types.is_date32(target) or pa.types.is_time(target))):
+ return None
+ raw = array.cast(pa.int32() if date else pa.int64()).to_pylist()
+ unit = array.type.unit if timestamp else 's'
+ source_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
else 0)
+ target_precision = (_parse_type_precision_scale(target_type)[0] if
target_type is not None else
+ {'s': 0, 'ms': 3, 'us': 6, 'ns': 9}[target.unit] if
target_ts else 0)
+ values = []
+ for value in raw:
+ if value is None:
+ values.append(None)
+ continue
+ if date:
+ millis, sub_milli = _java_date_to_epoch_day(value) * 86400000, 0
+ elif unit == 's':
+ millis, sub_milli = _signed(value * 1000, 64), 0
+ elif unit == 'ms':
+ millis, sub_milli = value, 0
+ elif unit == 'us':
+ millis, sub_milli = value // 1000, value % 1000 * 1000
+ else:
+ millis, sub_milli = value // 1000000, value % 1000000
+ original = millis
+ if cross and not ltz:
+ millis = _signed(millis // 86400000, 32) * 86400000 + millis %
86400000
+ if date or cross and not ltz:
+ millis = _local_to_utc_seconds(millis // 1000) * 1000 + millis %
1000
+ elif ltz or numeric:
+ millis = _signed(millis + _local_offset(millis // 1000) * 1000, 64)
+ if pa.types.is_date32(target):
+ value = (_epoch_day_to_java_date((original +
_local_offset(original // 1000) * 1000) // 86400000)
+ if ltz else _signed(millis // 86400000, 32))
+ elif pa.types.is_time(target):
+ value = ((original + _local_offset(original // 1000) * 1000) %
86400000
+ if ltz else millis % 86400000)
+ else:
+ if source_precision > target_precision:
+ if target_precision <= 3:
+ quantum = 10 ** (3 - target_precision)
+ millis, sub_milli = _signed(millis // quantum * quantum,
64), 0
+ else:
+ quantum = 10 ** (9 - target_precision)
+ sub_milli = sub_milli // quantum * quantum
+ if target.unit == 's':
+ value = millis // 1000
+ elif target.unit == 'ms':
+ value = millis
+ elif target.unit == 'us':
+ value = millis * 1000 + sub_milli // 1000
+ else:
+ value = millis * 1000000 + sub_milli
+ if not -(1 << 63) <= value < (1 << 63):
+ value = None
+ values.append(value)
+ if pa.types.is_time(target):
+ return pa.array(values,
type=pa.int32()).cast(pa.time32('ms')).cast(target, safe=False)
+ return pa.array(values, type=pa.int32() if pa.types.is_date32(target) else
pa.int64()).cast(target)
diff --git a/paimon-python/pypaimon/casting/temporal_string.py
b/paimon-python/pypaimon/casting/temporal_string.py
index 5cf6e5034c..a47dba3a9c 100644
--- a/paimon-python/pypaimon/casting/temporal_string.py
+++ b/paimon-python/pypaimon/casting/temporal_string.py
@@ -17,35 +17,12 @@
"""Java DateTimeUtils text shared by schema evolution and row rendering."""
-import time
-
import pyarrow as pa
+from pypaimon.casting.temporal_utils import _epoch_day_to_date, _local_offset
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,
diff --git a/paimon-python/pypaimon/casting/temporal_utils.py
b/paimon-python/pypaimon/casting/temporal_utils.py
new file mode 100644
index 0000000000..7a923093d1
--- /dev/null
+++ b/paimon-python/pypaimon/casting/temporal_utils.py
@@ -0,0 +1,108 @@
+# 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.
+
+"""Integer calendar and process-local zone helpers for Java temporal casts."""
+
+import time
+
+
+_GREGORIAN_CYCLE_SECONDS = 146097 * 86400
+# Match the finite zone lookup window used by the Native temporal helpers.
+_MIN_ZONE_SECONDS = -377705023201
+_MAX_ZONE_SECONDS = 253402207200
+
+
+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):
+ # libc can lose DST in very large years. Java's final transition rules
+ # repeat every Gregorian 400-year cycle, including month/day/weekday/time.
+ # Query an equivalent date in 2370..2770, after recorded transitions.
+ if seconds > _MAX_ZONE_SECONDS:
+ seconds = seconds % _GREGORIAN_CYCLE_SECONDS + _GREGORIAN_CYCLE_SECONDS
+ elif seconds < _MIN_ZONE_SECONDS:
+ # Before recorded history Java uses its first wall offset.
+ seconds = _MIN_ZONE_SECONDS
+ 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 _signed(value, bits):
+ return (value + (1 << (bits - 1))) % (1 << bits) - (1 << (bits - 1))
+
+
+def _div(value, divisor):
+ # Java integer division truncates towards zero.
+ return -(abs(value) // divisor) if value < 0 else value // divisor
+
+
+def _rem(value, divisor):
+ return value - _div(value, divisor) * divisor
+
+
+def _java_date_to_epoch_day(date):
+ j = _signed(date + 2440588 + 32044, 32)
+ g, dg = _div(j, 146097), _rem(j, 146097)
+ c = _div((_div(dg, 36524) + 1) * 3, 4)
+ dc = dg - c * 36524
+ b, db = _div(dc, 1461), _rem(dc, 1461)
+ a = _div((_div(db, 365) + 1) * 3, 4)
+ da = db - a * 365
+ y = g * 400 + c * 100 + b * 4 + a
+ m = _div(da * 5 + 308, 153) - 2
+ d = da - _div((m + 4) * 153, 5) + 122
+ year, month, day = y - 4800 + _div(m + 2, 12), _rem(m + 2, 12) + 1, d + 1
+ leap = year % 400 == 0 or year % 4 == 0 and year % 100 != 0
+ days = 29 if month == 2 and leap else 28 if month == 2 else 30 if month in
(4, 6, 9, 11) else 31
+ if not 1 <= month <= 12 or not 1 <= day <= days:
+ raise ValueError('Invalid Java DATE conversion: {}-{}-{}'.format(year,
month, day))
+ year -= month <= 2
+ era, yoe = divmod(year, 400)
+ march = month + (-3 if month > 2 else 9)
+ return era * 146097 + yoe * 365 + yoe // 4 - yoe // 100 + (153 * march +
2) // 5 + day - 1 - 719468
+
+
+def _epoch_day_to_java_date(day):
+ year, month, day = _epoch_day_to_date(day)
+ a = (14 - month) // 12
+ y, m = year + 4800 - a, month + 12 * a - 3
+ return _signed(day + (153 * m + 2) // 5 + 365 * y + _div(y, 4) - _div(y,
100)
+ + _div(y, 400) - 32045 - 2440588, 32)
+
+
+def _local_to_utc_seconds(seconds):
+ # Inspect both sides of a possible transition. In an overlap both offsets
+ # are valid: Java chooses the earlier instant. In a gap neither is valid:
+ # applying the earlier offset shifts the wall clock forward through the
gap.
+ offsets = {_local_offset(seconds + delta) for delta in (-172800, 0,
172800)}
+ candidates = [seconds - offset for offset in offsets]
+ valid = [utc for utc in candidates if utc + _local_offset(utc) == seconds]
+ return min(valid) if valid else max(candidates)
diff --git a/paimon-python/pypaimon/tests/native_temporal_zone_cast_test.py
b/paimon-python/pypaimon/tests/native_temporal_zone_cast_test.py
new file mode 100644
index 0000000000..5e772e5634
--- /dev/null
+++ b/paimon-python/pypaimon/tests/native_temporal_zone_cast_test.py
@@ -0,0 +1,213 @@
+# 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.
+
+"""Existing temporal changes read natively from old REST Parquet files."""
+
+import os
+
+import pyarrow as pa
+import pytest
+
+from pypaimon import Schema
+from pypaimon.read.native_plan import native_runtime_available
+from pypaimon.schema.data_types import AtomicType, PyarrowFieldParser
+from pypaimon.schema.schema_change import SchemaChange
+from pypaimon.tests import native_plan_rest_test
+from pypaimon.tests.native_schema_evolution_cast_test import _read, _table,
_write
+
+rest_catalog = native_plan_rest_test.rest_catalog
+pytestmark = [pytest.mark.native_plan, pytest.mark.skipif(
+ not native_runtime_available(), reason='Rust reader required')]
+
+# Golden outputs from Java CastExecutors; UTC is also exercised by CI.
+_JAVA_CASTS = {
+ 'America/New_York': [
+ [18000000, 17999900, 1710055800100, 1730611800100, 1710070200100,
1730633400100, None],
+ [-18000000, -18000100, 1710019800100, 1730583000100, 1710041400100,
1730597400100, None],
+ [-1, -1, 19791, 20029, 19792, 20030, None],
+ [68400000, 68399999, 77400123, 77400123, 12600123, 5400123, None],
+ ],
+ 'Asia/Shanghai': [
+ [-28800000, -28800100, 1710009000100, 1730568600100, 1710027000100,
1730586600100, None],
+ [28800000, 28799900, 1710066600100, 1730626200100, 1710084600100,
1730644200100, None],
+ [0, 0, 19792, 20030, 19792, 20030, None],
+ [28800000, 28799999, 37800123, 34200123, 55800123, 52200123, None],
+ ],
+ 'UTC': [
+ [0, -100, 1710037800100, 1730597400100, 1710055800100, 1730615400100,
None],
+ [0, -100, 1710037800100, 1730597400100, 1710055800100, 1730615400100,
None],
+ [0, -1, 19792, 20030, 19792, 20030, None],
+ [0, 86399999, 9000123, 5400123, 27000123, 23400123, None],
+ ],
+}
+_JAVA_WIDE = {
+ 'America/New_York': [
+ [-53757322487462200, 53757322519624100],
+ [9223372036840375800, 9223372036837013800],
+ [-622191233, 622191233],
+ [11575807, 42662192],
+ ],
+ 'Asia/Shanghai': [
+ [-53757322534367200, 53757322476424100],
+ [-9223372036825975900, -9223372036825632900],
+ [-622191233, 622191234],
+ [54775807, 3167192],
+ ],
+ 'UTC': [
+ [-53757322505224200, 53757322505224100],
+ [9223372036854775800, 9223372036854775716],
+ [-622191233, 622191233],
+ [25975807, 60424192],
+ ],
+}
+
+_VALUES = [0, -1, 1710037800123, 1730597400123, 1710055800123, 1730615400123,
None]
+
+
+def _zone():
+ # Run non-UTC cases in separate processes so Native's cached system zone
+ # and Python's libc zone are initialized under the same environment.
+ name = os.environ.get('TZ', 'UTC')
+ if name not in _JAVA_CASTS:
+ pytest.skip('Java golden values are defined for UTC, New York and
Shanghai')
+ return name
+
+
+def _case(operation):
+ date_values = {'UTC': [0, -86400000, 1717200000000, 1325203200000, None],
+ 'Asia/Shanghai': [-28800000, -115200000, 1717171200000,
1325174400000, None],
+ 'America/New_York': [18000000, -68400000, 1717214400000,
1325221200000, None]}
+ zone = _zone()
+ if operation < 4:
+ source, target = [
+ (pa.timestamp('ms'), 'TIMESTAMP_LTZ(1)'),
+ (pa.timestamp('ms', 'UTC'), 'TIMESTAMP(1)'),
+ (pa.timestamp('ms', 'UTC'), 'DATE'),
+ (pa.timestamp('ms', 'UTC'), 'TIME(1)'),
+ ][operation]
+ return source, _VALUES, target, _JAVA_CASTS[zone][operation]
+ if operation == 4:
+ return pa.date32(), [0, -1, 19875, 15338, None], 'TIMESTAMP_LTZ(3)',
date_values[zone]
+ offset = {'UTC': 0, 'Asia/Shanghai': 28800000, 'America/New_York':
-18000000}[zone]
+ return pa.int64(), [0, -1, 42, None], 'TIMESTAMP_LTZ(3)', [offset, offset
- 1000, offset + 42000, None]
+
+
+def _raw(column):
+ kind = column.type
+ return column.cast(pa.int32() if pa.types.is_date(kind) or
pa.types.is_time(kind) else pa.int64()).to_pylist()
+
+
[email protected]('mode', ['append', 'pk', 'de'])
[email protected]('operation', range(6))
+def test_existing_temporal_changes_read_old_rest_files(rest_catalog, mode,
operation):
+ source, values, target, expected = _case(operation)
+ catalog, _ = rest_catalog
+ _table(catalog, mode, source, values)
+ catalog.alter_table('default.t', [SchemaChange.update_column_type('v',
AtomicType(target))], False)
+ table = catalog.get_table('default.t')
+ for native in (False, True):
+ result = _read(table, native)
+ assert _raw(result.column('v')) == expected
+ assert result.column('id').to_pylist() == list(range(len(values)))
+ target_arrow = PyarrowFieldParser.from_paimon_type(AtomicType(target))
+ _write(table, pa.table({'id': [len(values)], 'v': pa.array([expected[0]],
type=target_arrow)}))
+ table = catalog.get_table('default.t')
+ literal = pa.scalar(expected[0], type=target_arrow).as_py()
+ predicate = table.new_read_builder().new_predicate_builder().equal('v',
literal)
+ matched_ids = [index for index, value in enumerate(expected) if value ==
expected[0]] + [len(values)]
+ for native in (False, True):
+ assert _raw(_read(table, native).column('v')) == expected +
[expected[0]]
+ assert _read(table, native, predicate,
['id']).column('id').to_pylist() == matched_ids
+
+
[email protected]('mode', ['append', 'pk', 'de'])
+def
test_local_zone_casts_reach_row_array_and_selected_map_values(rest_catalog,
mode):
+ zone = _zone()
+ catalog, _ = rest_catalog
+ schema = pa.schema([
+ ('id', pa.int64()),
+ ('r', pa.struct([('t', pa.timestamp('ms')), ('d', pa.date32())])),
+ ('a', pa.list_(pa.timestamp('ms', 'UTC'))),
+ ('m', pa.map_(pa.string(), pa.timestamp('ms', 'UTC'))),
+ ])
+ options = {'read.batch-size': '2', 'bucket': '1'} if mode == 'pk' else
{'read.batch-size': '2'}
+ if mode == 'de':
+ options.update({'data-evolution.enabled': 'true',
'row-tracking.enabled': 'true'})
+ catalog.create_table('default.t', Schema.from_pyarrow_schema(
+ schema, primary_keys=['id'] if mode == 'pk' else [], options=options),
False)
+ _write(catalog.get_table('default.t'), pa.Table.from_pylist([
+ dict(id=0, r={'t': 0, 'd': 0}, a=[0, -1, None], m=[('first', 0),
('last', -1)]),
+ dict(id=1, r=None, a=None, m=None),
+ dict(id=2, r={'t': _VALUES[2], 'd': 19875}, a=[], m=[]),
+ ], schema=schema))
+ catalog.alter_table('default.t', [
+ SchemaChange.update_column_type(['r', 't'],
AtomicType('TIMESTAMP_LTZ(1)')),
+ SchemaChange.update_column_type(['r', 'd'],
AtomicType('TIMESTAMP_LTZ(3)')),
+ SchemaChange.update_column_type(['a', 'element'], AtomicType('DATE')),
+ SchemaChange.update_column_type(['m', 'value'], AtomicType('TIME(1)')),
+ ], False)
+ table = catalog.get_table('default.t')
+ expected_t = [_JAVA_CASTS[zone][0][0], None, _JAVA_CASTS[zone][0][2]]
+ expected_d = [_case(4)[3][0], None, _case(4)[3][2]]
+ expected_a = pa.array([[_JAVA_CASTS[zone][2][0], _JAVA_CASTS[zone][2][1],
None], None, []],
+ type=pa.list_(pa.date32())).to_pylist()
+ for native in (False, True):
+ result = _read(table, native)
+ assert _raw(result.column('r').flatten()[0]) == expected_t
+ assert _raw(result.column('r').flatten()[1]) == expected_d
+ assert result.column('a').to_pylist() == expected_a
+ leaves = _read(table, native, projection=['id', 'r.t', 'r.d'])
+ assert _raw(leaves.column('r_t')) == expected_t
+ assert _raw(leaves.column('r_d')) == expected_d
+ selected = _read(table, native, projection=['id', "m['first']",
"m['last']", "m['missing']"])
+ assert _raw(selected.column('m_first')) == [_JAVA_CASTS[zone][3][0],
None, None]
+ assert _raw(selected.column('m_last')) == [_JAVA_CASTS[zone][3][1],
None, None]
+ assert selected.column('m_missing').null_count == 3
+
+
[email protected]('mode', ['append', 'de'])
[email protected]('operation', range(4))
+def test_wide_temporal_casts_preserve_mixed_rest_batches(rest_catalog, mode,
operation):
+ zone = _zone()
+ source, _, target, ordinary = _case(operation)
+ catalog, _ = rest_catalog
+ _table(catalog, mode, source, [0, 2 ** 63 - 1, None, -(2 ** 63)])
+ catalog.alter_table('default.t', [SchemaChange.update_column_type('v',
AtomicType(target))], False)
+ table = catalog.get_table('default.t')
+ expected = [ordinary[0], _JAVA_WIDE[zone][operation][0], None,
_JAVA_WIDE[zone][operation][1]]
+ for native in (False, True):
+ result = _read(table, native).column('v')
+ assert _raw(result) == expected
+ assert result.null_count == 1
+
+
[email protected]('mode', ['append', 'de'])
+def test_local_timestamp_can_cross_library_bounds_in_rest_reads(rest_catalog,
mode):
+ zone = _zone()
+ values = [-377705023201000, -377705023200000, 253402207199000,
253402207200000, None]
+ expected = {
+ 'UTC': values,
+ 'Asia/Shanghai': [-377705052344000, -377705052343000, 253402178399000,
253402178400000, None],
+ 'America/New_York': [-377705005439000, -377705005438000,
253402225199000, 253402225200000, None],
+ }[zone]
+ catalog, _ = rest_catalog
+ _table(catalog, mode, pa.timestamp('ms'), values)
+ catalog.alter_table('default.t', [SchemaChange.update_column_type('v',
AtomicType('TIMESTAMP_LTZ(3)'))], False)
+ for native in (False, True):
+ result = _read(catalog.get_table('default.t'), native).column('v')
+ assert _raw(result) == expected
+ assert result.null_count == 1
diff --git a/paimon-python/pypaimon/tests/schema_evolution_cast_test.py
b/paimon-python/pypaimon/tests/schema_evolution_cast_test.py
index 1b52f967d8..f3280ec9f8 100644
--- a/paimon-python/pypaimon/tests/schema_evolution_cast_test.py
+++ b/paimon-python/pypaimon/tests/schema_evolution_cast_test.py
@@ -16,6 +16,9 @@
# under the License.
from decimal import Decimal
+import os
+import time
+from types import SimpleNamespace
import pyarrow as pa
import pytest
@@ -24,6 +27,36 @@ from pypaimon.casting.schema_evolution import
cast_array_for_schema_evolution
from pypaimon.schema.data_types import AtomicType
[email protected]
+def local_zone():
+ if not hasattr(time, 'tzset'):
+ pytest.skip('Changing the process local zone requires tzset')
+ original = os.environ.get('TZ')
+
+ def set_zone(name):
+ os.environ['TZ'] = name
+ time.tzset()
+
+ yield set_zone
+ if original is None:
+ os.environ.pop('TZ', None)
+ else:
+ os.environ['TZ'] = original
+ time.tzset()
+
+
[email protected]('source_type,target_type,expected', [
+ (pa.timestamp('ms', 'UTC'), pa.timestamp('ms'), [28800000, 28799999,
None]),
+ (pa.timestamp('ms'), pa.timestamp('ms', 'UTC'), [-28800000, -28800001,
None]),
+ (pa.date32(), pa.timestamp('ms', 'UTC'), [-28800000, -115200000, None]),
+])
+def test_temporal_cast_uses_java_process_local_zone(local_zone, source_type,
target_type, expected):
+ local_zone('Asia/Shanghai')
+ source = pa.array([0, -1, None], type=source_type)
+ result = cast_array_for_schema_evolution(source, target_type)
+ assert result.cast(pa.int64()).to_pylist() == expected
+
+
@pytest.mark.parametrize('source_type,values,target_type,expected', [
(pa.float32(), [128.9, -129.9, 1e30, -1e30, None], pa.int8(), [-128, 127,
-1, 0, None]),
(pa.float64(), [32768.9, -32769.9, 1e30, -1e30, None], pa.int16(),
[-32768, 32767, -1, 0, None]),
@@ -163,3 +196,150 @@ def
test_timestamp_seconds_wrapping_keeps_wrapped_millisecond_fraction():
assert cast_array_for_schema_evolution(
source, pa.string(), AtomicType('TIMESTAMP(3)'),
AtomicType('STRING')).to_pylist() == [
'5471-09-26 16:47:04.384', None]
+
+
+_JAVA_ZONE_CASTS = [
+ ('America/New_York', [
+ [18000000, 17999900, 1710055800100, 1730611800100, 1710070200100,
1730633400100,
+ -53757322487462200, 53757322519624100],
+ [-18000000, -18000100, 1710019800100, 1730583000100, 1710041400100,
1730597400100,
+ 9223372036840375800, 9223372036837013800],
+ [-1, -1, 19791, 20029, 19792, 20030,
+ -622191233, 622191233],
+ [68400000, 68399999, 77400123, 77400123, 12600123, 5400123,
+ 11575807, 42662192],
+ ]),
+ ('Asia/Shanghai', [
+ [-28800000, -28800100, 1710009000100, 1730568600100, 1710027000100,
1730586600100,
+ -53757322534367200, 53757322476424100],
+ [28800000, 28799900, 1710066600100, 1730626200100, 1710084600100,
1730644200100,
+ -9223372036825975900, -9223372036825632900],
+ [0, 0, 19792, 20030, 19792, 20030,
+ -622191233, 622191234],
+ [28800000, 28799999, 37800123, 34200123, 55800123, 52200123,
+ 54775807, 3167192],
+ ]),
+]
+
+
[email protected]('zone,expected', _JAVA_ZONE_CASTS)
[email protected]('operation', range(4))
[email protected]('chunked', [False, True])
+def test_temporal_zone_casts_match_java_boundaries(local_zone, zone, expected,
operation, chunked):
+ local_zone(zone)
+ from_type, target, from_name, target_name = [
+ (pa.timestamp('ms'), pa.timestamp('ms', 'UTC'), 'TIMESTAMP(9)',
'TIMESTAMP_LTZ(1)'),
+ (pa.timestamp('ms', 'UTC'), pa.timestamp('ms'), 'TIMESTAMP_LTZ(9)',
'TIMESTAMP(1)'),
+ (pa.timestamp('ms', 'UTC'), pa.date32(), 'TIMESTAMP_LTZ(9)', 'DATE'),
+ (pa.timestamp('ms', 'UTC'), pa.time32('ms'), 'TIMESTAMP_LTZ(9)',
'TIME(1)'),
+ ][operation]
+ values = [0, -1, 1710037800123, 1730597400123, 1710055800123,
1730615400123,
+ 9223372036854775807, -9223372036854775808]
+ source = pa.array([None] + values + [None, None], type=from_type).slice(1,
9)
+ if chunked:
+ source = pa.chunked_array([source.slice(0, 1), source.slice(1, 3),
source.slice(4)])
+ result = cast_array_for_schema_evolution(source, target,
AtomicType(from_name), AtomicType(target_name))
+ if operation in (2, 3):
+ result = result.cast(pa.int32())
+ else:
+ result = result.cast(pa.int64())
+ assert result.to_pylist() == expected[operation] + [None]
+
+
[email protected]('zone,expected', [
+ ('Asia/Shanghai', [-28800000, -115200000, 1717171200000, 1325174400000,
None]),
+ ('America/New_York', [18000000, -68400000, 1717214400000, 1325221200000,
None]),
+ ('Pacific/Apia', [39600000, -46800000, 1717153200000, 1325239200000,
None]),
+])
+def test_date_to_ltz_resolves_java_midnight(local_zone, zone, expected):
+ local_zone(zone)
+ source = pa.array([0, -1, 19875, 15338, None], type=pa.date32())
+ assert cast_array_for_schema_evolution(source, pa.timestamp('ms',
'UTC')).cast(pa.int64()).to_pylist() == expected
+
+
[email protected]('value', [-(2 ** 31), 2 ** 31 - 1])
+def test_date_to_ltz_rejects_invalid_java_julian_date(local_zone, value):
+ local_zone('UTC')
+ with pytest.raises(ValueError, match='Invalid Java DATE'):
+ cast_array_for_schema_evolution(pa.array([value], type=pa.date32()),
pa.timestamp('ms', 'UTC'))
+
+
[email protected]('precision,target,expected', [
+ (6, pa.timestamp('us', 'UTC'), [1710055800123456, 17999999999, None]),
+ (1, pa.timestamp('ms', 'UTC'), [1710055800100, 17999900, None]),
+])
+def test_cross_zone_cast_preserves_nanos_and_declared_precision(local_zone,
precision, target, expected):
+ local_zone('America/New_York')
+ source = pa.array([1710037800123456789, -1, None], type=pa.timestamp('ns'))
+ result = cast_array_for_schema_evolution(
+ source, target, AtomicType('TIMESTAMP(9)'),
AtomicType('TIMESTAMP_LTZ(%d)' % precision))
+ assert result.cast(pa.int64()).to_pylist() == expected
+
+
+def test_bigint_to_ltz_keeps_java_epoch_seconds_and_long_wrapping(local_zone):
+ local_zone('Asia/Shanghai')
+ source = pa.array([0, -1, 2 ** 63 - 1, -(2 ** 63), None], type=pa.int64())
+ assert cast_array_for_schema_evolution(source, pa.timestamp('ms',
'UTC')).cast(pa.int64()).to_pylist() == [
+ 28800000, 28799000, 28799000, 28800000, None]
+
+
+def test_ltz_semantics_follow_declared_paimon_type(local_zone):
+ local_zone('Asia/Shanghai')
+ source = pa.array([0, -1, None], type=pa.timestamp('ms'))
+ result = cast_array_for_schema_evolution(
+ source, pa.timestamp('ms'), AtomicType('TIMESTAMP_LTZ(3)'),
AtomicType('TIMESTAMP(3)'))
+ assert result.cast(pa.int64()).to_pylist() == [28800000, 28799999, None]
+
+
[email protected]('zone,values,expected', [
+ ('Asia/Shanghai', [-377705023201000, -377705023200000, None],
+ [-377705052344000, -377705052343000, None]),
+ ('America/New_York', [253402207199000, 253402207200000, None],
+ [253402225199000, 253402225200000, None]),
+])
+def test_local_wall_time_can_cross_date_library_bounds(local_zone, zone,
values, expected):
+ local_zone(zone)
+ source = pa.array(values, type=pa.timestamp('ms'))
+ result = cast_array_for_schema_evolution(source, pa.timestamp('ms', 'UTC'))
+ assert result.cast(pa.int64()).to_pylist() == expected
+
+
[email protected]('platform', ['standard_offset', 'range_error'])
[email protected]('target,expected', [
+ (pa.timestamp('ms'), [
+ -18000000, 9223372036840375807, 9223372021198375807,
9223372036830600000,
+ 9223372036837013808, 253402282800000, None]),
+ (pa.time32('ms'), [68400000, 11575807, 7975807, 1800000, 42662192,
68400000, None]),
+ (pa.date32(), [-1, -622191233, -622191414, -622191233, 622191233, 2932896,
None]),
+ (pa.string(), [
+ '1969-12-31 19:00:00.000', '/+-.-04-06 03:12:55.807', '/+---10-08
02:12:55.807',
+ '/+-.-04-06 00:30:00.000', '/+-.-04-06 02:16:53.808', '9999-12-31
19:00:00.000', None]),
+])
+def test_wide_ltz_casts_keep_java_dst_when_platform_localtime_cannot(
+ local_zone, monkeypatch, platform, target, expected):
+ local_zone('America/New_York')
+ libc_localtime = time.localtime
+
+ def limited_localtime(seconds=None):
+ if seconds is None:
+ return libc_localtime()
+ if seconds > 253402207200:
+ if platform == 'standard_offset':
+ # Some libc implementations lose the far-future DST rule.
+ return SimpleNamespace(tm_gmtoff=-18000, tm_isdst=0)
+ raise OverflowError('Platform cannot convert this future year')
+ if seconds < -377705023201 and platform == 'range_error':
+ raise OverflowError('Platform cannot convert this ancient year')
+ return libc_localtime(seconds)
+
+ monkeypatch.setattr(time, 'localtime', limited_localtime)
+ values = [0, 9223372036854775807, 9223372021216375807, 9223372036845000000,
+ -9223372036854775808, 253402300800000, None]
+ source = pa.array([None] + values + [None], type=pa.timestamp('ms',
'UTC')).slice(1, len(values))
+ source = pa.chunked_array([source.slice(0, 3), source.slice(3)])
+ result = cast_array_for_schema_evolution(source, target)
+ if not pa.types.is_string(target):
+ result = result.cast(pa.int64() if pa.types.is_timestamp(target) else
pa.int32())
+ # Golden values from real Java CastExecutors, including a local midnight
+ # whose DATE changes if a standard offset replaces the correct DST offset.
+ assert result.to_pylist() == expected
