paimon-python/pypaimon/tests/schema_evolution_cast_test.py

# 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.

import builtins
from decimal import Decimal, DefaultContext, Inexact, localcontext, Rounded
import os
import struct
import time
from types import SimpleNamespace

import pyarrow as pa
import pytest

from pypaimon.schema.arrow_schema import read_arrow_schema
from pypaimon.execution.python.casting.schema_evolution import cast_array_for_schema_evolution
from pypaimon.schema.data_types import ArrayType, AtomicType, DataField, MapType, PyarrowFieldParser, RowType


@pytest.mark.parametrize('chunked', [False, True])
def test_schema_helpers_do_not_load_native_execution(monkeypatch, chunked):
    original_import = builtins.__import__

    def python_import(name, *args, **kwargs):
        if name.startswith(('pypaimon.execution.native', 'pypaimon_rust')):
            raise AssertionError('Python schema evolution imported Native execution: ' + name)
        return original_import(name, *args, **kwargs)

    monkeypatch.setattr(builtins, '__import__', python_import)
    values = pa.array([1.9, -2.5, None])
    if chunked:
        values = pa.chunked_array([values.slice(0, 1), values.slice(1)])
    result = cast_array_for_schema_evolution(values, pa.int32(), AtomicType('DOUBLE'), AtomicType('INT'))
    assert result.to_pylist() == [1, -2, None]

    fields = [DataField(1, 'rows', ArrayType(False, RowType(True, [
        DataField(2, 'id', AtomicType('INT', False)),
    ])), description='nested schema')]
    schema = read_arrow_schema(fields)
    assert schema.field('rows').nullable
    assert schema.field('rows').metadata is None
    assert schema.field('rows').type.value_field.nullable
    assert schema.field('rows').type.value_type.field('id').nullable


def test_read_schema_preserves_runtime_nulls_without_relaxing_write_constraints():
    from pypaimon.read.table_read import TableRead

    decimal = AtomicType('DECIMAL(3,2)', False)
    row = RowType(False, [DataField(2, 'amount', decimal)])
    fields = [DataField(1, 'rows', ArrayType(False, row)),
              DataField(3, 'amounts', MapType(False, AtomicType('STRING', False), row)),
              DataField(4, 'variant', AtomicType('VARIANT', False)),
              DataField(5, 'row_keys', MapType(False, row, ArrayType(False, decimal)))]
    schema = read_arrow_schema(fields)
    rows, amounts, variant = schema.field('rows'), schema.field('amounts'), schema.field('variant')
    assert rows.nullable and rows.type.value_field.nullable
    assert rows.type.value_type.field('amount').nullable
    assert amounts.nullable and not amounts.type.key_field.nullable
    assert amounts.type.item_field.nullable and amounts.type.item_type.field('amount').nullable
    row_keys = schema.field('row_keys').type
    assert not row_keys.key_field.nullable and row_keys.key_type.field('amount').nullable
    assert row_keys.item_field.nullable and row_keys.item_type.value_field.nullable
    # These VARIANT children describe its physical representation, not logical
    # ROW fields. Nullable read values do not relax the required binary buffers.
    assert variant.nullable and not any(field.nullable for field in variant.type)
    values = [[{'amount': None}], None, []]
    rows_array = pa.array(values, type=rows.type)
    rows_array.validate(full=True)
    batch = pa.record_batch([rows_array], names=['rows'])
    aligned = TableRead._try_to_pad_batch_by_schema(batch, pa.schema([rows]))
    assert aligned.schema.equals(pa.schema([rows]))
    assert aligned.column(0).to_pylist() == values
    maps = [[('a', {'amount': None}), ('b', None)], None, []]
    pa.array(maps, type=amounts.type).validate(full=True)

    write_schema = PyarrowFieldParser.from_paimon_schema(fields)
    assert not write_schema.field('rows').nullable
    assert not write_schema.field('rows').type.value_field.nullable
    assert not write_schema.field('rows').type.value_type.field('amount').nullable
    assert not write_schema.field('amounts').type.item_field.nullable


@pytest.mark.parametrize('chunked', [False, True])
def test_decimal_evolution_uses_supported_jvm_floating_decimal_digits(chunked):
    # Java 8/11/17 retain exact digits in this integral fast path. JDK 19+
    # shortest text changes the integer by four, before scale is applied.
    value = struct.unpack('>d', bytes.fromhex('c381e88f38c0c8fd'))[0]
    source = pa.array([None, value, -value, 1e23, None]).slice(1, 4)
    if chunked:
        source = pa.chunked_array([source.slice(0, 1), source.slice(1)])
    result = cast_array_for_schema_evolution(source, pa.decimal128(38, 20))
    assert result.to_pylist() == [
        Decimal('-161304845894623136.00000000000000000000'),
        Decimal('161304845894623136.00000000000000000000'), None, None,
    ]
    # For the power-of-ten boundary, the old converter excludes the upper
    # interval endpoint and preserves 9.999999999999999E22, rather than 1E23.
    result = cast_array_for_schema_evolution(pa.array([1e23]), pa.decimal128(38, 6))
    assert result.to_pylist() == [Decimal('99999999999999990000000.000000')]


@pytest.mark.parametrize('chunked', [False, True])
@pytest.mark.parametrize('source_type', [pa.int8(), pa.int16(), pa.int32(), pa.int64()])
def test_integer_decimal_evolution_applies_precision_per_value(chunked, source_type):
    # Java NumericPrimitiveToDecimalCastRule accepts a narrow target even when
    # the source type's full range would overflow it. Only overflowing rows
    # become NULL; PyArrow's integer cast instead rejects the whole target type.
    values = pa.array([None, 1, -1, 99, -99, 100, -100, None], type=source_type)
    source = values.slice(1, 7)
    if chunked:
        source = pa.chunked_array([source.slice(0, 2), source.slice(2, 0), source.slice(2)])
    actual = cast_array_for_schema_evolution(source, pa.decimal128(4, 2))
    assert actual.to_pylist() == [
        Decimal('1.00'), Decimal('-1.00'), Decimal('99.00'), Decimal('-99.00'),
        None, None, None,
    ]


@pytest.mark.parametrize('chunked', [False, True])
@pytest.mark.parametrize('source_type,values,precision,scale,expected', [
    (pa.float64(), [1.005, -1.005, 2.675, -2.675, 0.145, 999.995, 1e30, None], 5, 2,
     ['1.01', '-1.01', '2.68', '-2.68', '0.15', None, None, None]),
    (pa.float32(), [1.005, -1.005, 2.675, 0.145, 999.995, 1e30, None], 5, 2,
     ['1.00', '-1.00', '2.67', '0.14', '999.99', None, None]),
    (pa.float64(), [1.005, 1.2345678901234567, 2.2250738585072014e-308, None], 38, 20,
     ['1.00500000000000000000', '1.23456789012345670000', '0.00000000000000000000', None]),
    (pa.float32(), [1.005, 1.2345679, None], 38, 20,
     ['1.00499999523162840000', '1.23456788063049320000', None]),
    (pa.string(), ['1.005', '-1.005', '+.005', '2.675', '1e-2', '999.995', '1e100', None], 5, 2,
     ['1.01', '-1.01', '0.01', '2.68', '0.01', None, None, None]),
    (pa.decimal128(10, 2), [Decimal('9.99'), Decimal('10.00'), Decimal('-10.00'), None], 3, 2,
     ['9.99', None, None, None]),
    (pa.decimal128(10, 2), [Decimal('9.99'), Decimal('10.00'), Decimal('-10.00'), None], 4, 3,
     ['9.990', None, None, None]),
    (pa.decimal128(6, 3), [Decimal('9.994'), Decimal('9.995'), Decimal('-9.995'), None], 3, 2,
     ['9.99', None, None, None]),
])
def test_decimal_evolution_matches_java_rounding_and_precision(
        chunked, source_type, values, precision, scale, expected):
    # Golden values from CastExecutors: FLOAT is promoted to double before
    # BigDecimal.valueOf; overflow is NULL after HALF_UP, never wrapping.
    source = pa.array([None] + values + [None], type=source_type).slice(1, len(values))
    if chunked:
        source = pa.chunked_array([source.slice(0, 2), source.slice(2, 0), source.slice(2)])
    target = pa.decimal128(precision, scale)
    actual = cast_array_for_schema_evolution(source, target)
    assert actual.type == target
    assert actual.to_pylist() == [None if value is None else Decimal(value) for value in expected]


@pytest.mark.parametrize('text', [
    '', ' 1.2 ', '1_2', '1.2.3', '1e', 'Infinity', '0e-2147483648',
    '0.1e-2147483647', '1e2147483647', '1e-2147483647', '1e9999999999999999999',
])
def test_decimal_evolution_rejects_invalid_text(text):
    with pytest.raises(pa.ArrowInvalid):
        cast_array_for_schema_evolution(pa.array([None, text]), pa.decimal128(5, 2))


def test_decimal_evolution_handles_long_mantissas_and_java_zero_exponents():
    values = ['1.005' + '0' * 200, '0.' + '0' * 200 + '1',
              '0e2147483648', '0.0e2147483648', '0e2147483647']
    result = cast_array_for_schema_evolution(pa.array(values), pa.decimal128(5, 2))
    assert result.to_pylist() == [Decimal('1.01')] + [Decimal('0.00')] * 4


@pytest.mark.parametrize('source_type,values', [
    (pa.float64(), [1.005, 10000, None]),
    (pa.string(), ['1.005', '10000', None]),
    (pa.decimal128(10, 3), [Decimal('1.005'), Decimal('10000'), None]),
])
def test_decimal_evolution_is_independent_of_caller_decimal_context(source_type, values):
    source = pa.array(values, type=source_type)
    with localcontext() as context:
        context.prec = 2
        context.Emax = 2
        context.traps[Inexact] = True
        context.traps[Rounded] = True
        result = cast_array_for_schema_evolution(source, pa.decimal128(10, 2))
    assert result.to_pylist() == [Decimal('1.01'), Decimal('10000.00'), None]


def test_decimal_evolution_is_independent_of_default_decimal_context(monkeypatch):
    monkeypatch.setattr(DefaultContext, 'Emax', 2)
    monkeypatch.setattr(DefaultContext, 'Emin', -2)
    monkeypatch.setattr(DefaultContext, 'clamp', 1)
    result = cast_array_for_schema_evolution(pa.array(['1.005', '10000', None]), pa.decimal128(10, 2))
    assert result.to_pylist() == [Decimal('1.01'), Decimal('10000.00'), None]


@pytest.fixture
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()


@pytest.mark.parametrize('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]),
    (pa.float64(), [42.9, -42.9, 1e30, -1e30, None], pa.int32(),
     [42, -42, 2 ** 31 - 1, -(2 ** 31), None]),
    (pa.float64(), [42.9, -42.9, 1e30, -1e30, None], pa.int64(),
     [42, -42, 2 ** 63 - 1, -(2 ** 63), None]),
    (pa.decimal128(38, 2),
     [Decimal('999999999999999999999999999999999999.99'), Decimal('-42.99'), None],
     pa.int64(), [-5527149226598858753, -42, None]),
])
@pytest.mark.parametrize('chunked', [False, True])
def test_java_numeric_narrowing_of_sliced_and_chunked_arrays(source_type, values, target_type, expected, chunked):
    source = pa.array([None] + values + [None], type=source_type).slice(1, len(values))
    if chunked:
        source = pa.chunked_array([source.slice(0, 2), source.slice(2)])
    output = cast_array_for_schema_evolution(source, target_type)
    assert output.type == target_type
    assert output.to_pylist() == expected


@pytest.mark.parametrize('zone', [None, 'UTC'])
@pytest.mark.parametrize('precision', [7, 8])
def test_timestamp_floor_overflow_returns_null(zone, precision):
    source = pa.array([-(2 ** 63), -(2 ** 63) + 1000, -1, None], type=pa.timestamp('ns', zone))
    name = 'TIMESTAMP_LTZ' if zone else 'TIMESTAMP'
    output = cast_array_for_schema_evolution(
        source, source.type, AtomicType(name + '(9)'), AtomicType('%s(%d)' % (name, precision)))
    quantum = 10 ** (9 - precision)
    assert output.cast(pa.int64()).to_pylist() == [None, (-(2 ** 63) + 1000) // quantum * quantum, -quantum, None]


@pytest.mark.parametrize('unit,ticks', [('s', 1), ('ms', 1000), ('us', 1000000), ('ns', 1000000000)])
@pytest.mark.parametrize('zone', [None, 'UTC'])
@pytest.mark.parametrize('chunked', [False, True])
def test_timestamp_to_bigint_uses_epoch_seconds_flooring(unit, ticks, zone, chunked):
    fraction = 0 if unit == 's' else 1
    values = [ticks + fraction, -1, -ticks - fraction, 0,
              (2 ** 31 + 2) * ticks, -(2 ** 31 + 2) * ticks, None]
    expected = [1, -1, -1 if unit == 's' else -2, 0, 2 ** 31 + 2, -(2 ** 31 + 2), None]
    source = pa.array([None] + values + [None], type=pa.timestamp(unit, zone)).slice(1, len(values))
    if chunked:
        source = pa.chunked_array([source.slice(0, 2), source.slice(2, 0), source.slice(2)])
    result = cast_array_for_schema_evolution(source, pa.int64())
    assert result.type == pa.int64()
    assert result.to_pylist() == expected


@pytest.mark.parametrize('unit,ticks', [('s', 1), ('ms', 1000), ('us', 1000000), ('ns', 1000000000)])
@pytest.mark.parametrize('chunked', [False, True])
@pytest.mark.parametrize('source_type', [pa.int32(), pa.int64()])
def test_integer_to_timestamp_uses_epoch_seconds_and_java_millis_wrapping(unit, ticks, chunked, source_type):
    values = [42, -42, 1700000000, None]
    expected = [42 * ticks, -42 * ticks, 1700000000 * ticks, None]
    if source_type == pa.int64():
        values.extend([2 ** 63 - 1, -(2 ** 63), 2 ** 63 - 2])
        expected.extend([-ticks, 0, -2 * ticks])
    else:
        values.extend([2 ** 31 - 1, -(2 ** 31), 0])
        expected.extend([(2 ** 31 - 1) * ticks, -(2 ** 31) * ticks, 0])
    source = pa.array([None] + values + [None], type=source_type).slice(1, len(values))
    if chunked:
        source = pa.chunked_array([source.slice(0, 1), source.slice(1)])
    result = cast_array_for_schema_evolution(source, pa.timestamp(unit))
    assert result.type == pa.timestamp(unit)
    assert result.cast(pa.int64()).to_pylist() == expected


@pytest.mark.parametrize('zone,expected', [
    (None, [9223372036854775, -9223372036854776, None]),
    ('UTC', [-53757322505225, 53757322505224, None]),
])
def test_timestamp_to_bigint_matches_java_extreme_millis_round_trip(zone, expected):
    # Golden outputs from Java CastExecutors, including LTZ's int epoch day.
    source = pa.array([2 ** 63 - 1, -(2 ** 63), None], type=pa.timestamp('ms', zone))
    assert cast_array_for_schema_evolution(source, pa.int64()).to_pylist() == expected


@pytest.mark.parametrize('unit,expected', [
    ('s', [9223372036854775, -9223372036854775, None]),
    ('ms', [9223372036854775000, -9223372036854775000, None]),
    ('us', [None, None, None]),
    ('ns', [None, None, None]),
])
def test_bigint_to_timestamp_keeps_lenient_arrow_storage_overflow(unit, expected):
    source = pa.array([9223372036854775, -9223372036854775, None], type=pa.int64())
    result = cast_array_for_schema_evolution(source, pa.timestamp(unit))
    assert result.cast(pa.int64()).to_pylist() == expected


@pytest.mark.parametrize('precision,whole,nonzero', [
    (0, '1970-01-01 00:00:00', '1970-01-01 00:00:00.123456789'),
    (1, '1970-01-01 00:00:00.0', '1970-01-01 00:00:00.123456789'),
    (3, '1970-01-01 00:00:00.000', '1970-01-01 00:00:00.123456789'),
    (6, '1970-01-01 00:00:00.000000', '1970-01-01 00:00:00.123456789'),
    (9, '1970-01-01 00:00:00.000000000', '1970-01-01 00:00:00.123456789'),
])
@pytest.mark.parametrize('chunked', [False, True])
def test_java_timestamp_string_preserves_nanos_and_declared_precision(precision, whole, nonzero, chunked):
    source = pa.array([None, 0, 123456789, -1, None, 0], type=pa.timestamp('ns')).slice(1, 4)
    if chunked:
        source = pa.chunked_array([source.slice(0, 2), source.slice(2)])
    result = cast_array_for_schema_evolution(
        source, pa.string(), AtomicType('TIMESTAMP(%d)' % precision), AtomicType('STRING'))
    assert result.to_pylist() == [whole, nonzero, '1969-12-31 23:59:59.999999999', None]


@pytest.mark.parametrize('precision,expected', [
    (0, ['00:00:00', '00:00:00', '00:00:00', '23:59:59']),
    (1, ['00:00:00.0', '00:00:00.1', '00:00:00.1', '23:59:59.9']),
    (2, ['00:00:00.0', '00:00:00.1', '00:00:00.10', '23:59:59.99']),
    (3, ['00:00:00.0', '00:00:00.1', '00:00:00.101', '23:59:59.999']),
    (9, ['00:00:00.0', '00:00:00.1', '00:00:00.101', '23:59:59.999']),
])
def test_java_time_string_fraction_and_negative_millis(precision, expected):
    source = pa.array([None, 0, 100, 101, -1, None, 0], type=pa.time32('ms')).slice(1, 5)
    result = cast_array_for_schema_evolution(
        source, pa.string(), AtomicType('TIME(%d)' % precision), AtomicType('STRING'))
    assert result.to_pylist() == expected + [None]


@pytest.mark.parametrize('raw,expected', [
    (253402300800000, '0000-01-01 00:00:00.000'),  # year 10000: Java int4
    (-62167219200000, '0000-01-01 00:00:00.000'),  # year 0
    (-62198755200000, '000/-01-01 00:00:00.000'),  # year -1: signed remainder
    (2 ** 63 - 1, '/+-.-04-06 07:12:55.807'),  # signed epoch-day narrowing
    (-(2 ** 63), '5471-09-26 16:47:04.192'),
])
def test_timestamp_string_keeps_arrow_dates_outside_python_datetime_range(raw, expected):
    source = pa.array([raw, None], type=pa.timestamp('ms'))
    assert cast_array_for_schema_evolution(
        source, pa.string(), AtomicType('TIMESTAMP(3)'), AtomicType('STRING')).to_pylist() == [expected, None]


def test_timestamp_seconds_wrapping_keeps_wrapped_millisecond_fraction():
    source = pa.array([9223372036854776, None], type=pa.timestamp('s'))
    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],
    ]),
]


@pytest.mark.parametrize('zone,expected', _JAVA_ZONE_CASTS)
@pytest.mark.parametrize('operation', range(4))
@pytest.mark.parametrize('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]


@pytest.mark.parametrize('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


@pytest.mark.parametrize('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'))


@pytest.mark.parametrize('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]


@pytest.mark.parametrize('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


@pytest.mark.parametrize('platform', ['standard_offset', 'range_error'])
@pytest.mark.parametrize('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