diff --git a/paimon-python/pypaimon/casting/row_to_string.py b/paimon-python/pypaimon/casting/row_to_string.py index e775ca457c3b..ddfa6f9e93ac 100644 --- a/paimon-python/pypaimon/casting/row_to_string.py +++ b/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) diff --git a/paimon-python/pypaimon/casting/schema_evolution.py b/paimon-python/pypaimon/casting/schema_evolution.py index b01cc99d7e44..3aabc54371e3 100644 --- a/paimon-python/pypaimon/casting/schema_evolution.py +++ b/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 diff --git a/paimon-python/pypaimon/casting/temporal_string.py b/paimon-python/pypaimon/casting/temporal_string.py new file mode 100644 index 000000000000..5cf6e5034c14 --- /dev/null +++ b/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()) diff --git a/paimon-python/pypaimon/read/reader/data_file_batch_reader.py b/paimon-python/pypaimon/read/reader/data_file_batch_reader.py index dcd8c0a9670f..e9d8066a3977 100644 --- a/paimon-python/pypaimon/read/reader/data_file_batch_reader.py +++ b/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): diff --git a/paimon-python/pypaimon/tests/native_schema_evolution_cast_test.py b/paimon-python/pypaimon/tests/native_schema_evolution_cast_test.py index f9107378179c..881983b779a3 100644 --- a/paimon-python/pypaimon/tests/native_schema_evolution_cast_test.py +++ b/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 diff --git a/paimon-python/pypaimon/tests/schema_evolution_cast_test.py b/paimon-python/pypaimon/tests/schema_evolution_cast_test.py index 97d3ecc3a4c6..1b52f967d8f0 100644 --- a/paimon-python/pypaimon/tests/schema_evolution_cast_test.py +++ b/paimon-python/pypaimon/tests/schema_evolution_cast_test.py @@ -112,3 +112,54 @@ 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]