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 d257057b28 [Python] Align schema evolution casts and pruning with Java
and Native (#10502)
d257057b28 is described below
commit d257057b2852380cbe5f763f0edcb0c14a259fd8
Author: Jingsong Lee <[email protected]>
AuthorDate: Sun Oct 11 17:47:57 2026 +0800
[Python] Align schema evolution casts and pruning with Java and Native
(#10502)
---
paimon-python/pypaimon/casting/schema_evolution.py | 109 +++++++++++
.../pypaimon/manifest/simple_stats_evolutions.py | 49 ++++-
.../pypaimon/read/reader/data_file_batch_reader.py | 34 +---
.../pypaimon/read/scanner/file_scanner.py | 10 +-
.../tests/manifest/simple_stats_evolutions_test.py | 42 ++++
.../tests/native_schema_evolution_cast_test.py | 216 +++++++++++++++++++++
.../pypaimon/tests/schema_evolution_cast_test.py | 56 ++++++
7 files changed, 480 insertions(+), 36 deletions(-)
diff --git a/paimon-python/pypaimon/casting/schema_evolution.py
b/paimon-python/pypaimon/casting/schema_evolution.py
new file mode 100644
index 0000000000..62793ec79c
--- /dev/null
+++ b/paimon-python/pypaimon/casting/schema_evolution.py
@@ -0,0 +1,109 @@
+# 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 cast rules shared by old-file readers."""
+
+from decimal import Decimal, localcontext, ROUND_HALF_UP
+import math
+
+import pyarrow as pa
+
+from pypaimon.table.row.generic_row import _parse_type_precision_scale
+
+
+def _narrow_integer(value, bits):
+ half = 1 << (bits - 1)
+ return (value + half) % (half * 2) - half
+
+
+def _numeric_to_integer(array, target_type):
+ bits = target_type.bit_width
+ floating = pa.types.is_floating(array.type)
+ intermediate_bits = 64 if bits == 64 else 32
+ lower = -(1 << (intermediate_bits - 1))
+ upper = (1 << (intermediate_bits - 1)) - 1
+ values = []
+ for value in array.to_pylist():
+ if value is None:
+ values.append(None)
+ continue
+ # Number.byteValue/shortValue first convert floats to int; Java's
+ # float-to-int/long conversion saturates before the final narrowing.
+ if floating:
+ # Java's floating-point narrowing maps NaN to zero.
+ if math.isnan(value):
+ value = 0
+ else:
+ value = lower if value <= lower else upper if value >= upper
else int(value)
+ else:
+ # BigDecimal.longValue truncates towards zero and retains low bits.
+ value = int(value)
+ values.append(_narrow_integer(value, bits))
+ return pa.array(values, type=target_type)
+
+
+def _truncate_timestamp(array, target_type, source_paimon_type,
target_paimon_type):
+ if (not pa.types.is_timestamp(array.type) or not
pa.types.is_timestamp(target_type)
+ or array.type.tz != target_type.tz
+ or source_paimon_type is None or target_paimon_type is None):
+ return None
+ source_precision = _parse_type_precision_scale(source_paimon_type)[0]
+ target_precision = _parse_type_precision_scale(target_paimon_type)[0]
+ if source_precision <= target_precision:
+ return None
+ nanos_per_tick = {'s': 1000000000, 'ms': 1000000, 'us': 1000, 'ns': 1}
+ quantum = 10 ** (9 - target_precision)
+ values = [None if value is None else
+ (value * nanos_per_tick[array.type.unit] // quantum * quantum)
// nanos_per_tick[target_type.unit]
+ for value in array.cast(pa.int64()).to_pylist()]
+ # Flooring a nanosecond boundary can exceed Arrow's int64 storage even
+ # though Java's timestamp representation can hold it. Match Native's
+ # lenient overflow handling when rebuilding the Arrow result.
+ values = [value if value is None or -(1 << 63) <= value < (1 << 63) else
None for value in values]
+ return pa.array(values, type=pa.int64()).cast(target_type)
+
+
+def cast_array_for_schema_evolution(array, target_type,
source_paimon_type=None, target_paimon_type=None):
+ """Materialize existing type changes using Java numeric and timestamp
casts.
+
+ Declared Paimon types are needed for precision changes within one Arrow
+ timestamp unit. 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.
+ """
+ timestamp = _truncate_timestamp(array, target_type, source_paimon_type,
target_paimon_type)
+ if timestamp is not None:
+ return timestamp
+ if array.type == target_type:
+ return array
+ if (pa.types.is_signed_integer(target_type)
+ and (pa.types.is_floating(array.type) or
pa.types.is_decimal(array.type))):
+ return _numeric_to_integer(array, target_type)
+ if (pa.types.is_decimal(array.type) and pa.types.is_decimal(target_type)
+ and array.type.scale > target_type.scale):
+ values = []
+ with localcontext() as context:
+ context.prec = max(array.type.precision, target_type.precision) + 1
+ quantum = Decimal(1).scaleb(-target_type.scale)
+ for value in array.to_pylist():
+ if value is None:
+ values.append(None)
+ continue
+ rounded = value.quantize(quantum, rounding=ROUND_HALF_UP)
+ values.append(rounded if len(rounded.as_tuple().digits) <=
target_type.precision else None)
+ return pa.array(values, type=target_type)
+ return array.cast(target_type, safe=False)
diff --git a/paimon-python/pypaimon/manifest/simple_stats_evolutions.py
b/paimon-python/pypaimon/manifest/simple_stats_evolutions.py
index 0035852c1f..5f59e402c2 100644
--- a/paimon-python/pypaimon/manifest/simple_stats_evolutions.py
+++ b/paimon-python/pypaimon/manifest/simple_stats_evolutions.py
@@ -17,8 +17,10 @@
from typing import Callable, Dict, List, Optional
+from pypaimon.common.predicate import Predicate
+from pypaimon.common.predicate_builder import PredicateBuilder
from pypaimon.manifest.simple_stats_evolution import SimpleStatsEvolution
-from pypaimon.schema.data_types import DataField
+from pypaimon.schema.data_types import AtomicType, DataField
class SimpleStatsEvolutions:
@@ -47,6 +49,51 @@ class SimpleStatsEvolutions:
self.evolutions[data_schema_id] = evolution
return evolution
+ def filter_unsafe_filter(self, data_schema_id: int,
+ predicate: Optional[Predicate]) ->
Optional[Predicate]:
+ """Keep evolution-safe conjuncts, like Java's filterUnsafeFilter.
+
+ Integer narrowing can wrap and precision changes can truncate values.
+ Comparing current literals to those files' original bounds is unsafe.
+ An OR must stay intact: dropping one of its branches could lose rows.
+ """
+ if predicate is None or data_schema_id == self.table_schema_id:
+ return predicate
+ source_fields = {field.id: field for field in
self.schema_fields(data_schema_id)}
+
+ def safe(part):
+ if part.method in ('and', 'or'):
+ return all(safe(child) for child in part.literals)
+ if part.index is None or not 0 <= part.index <
len(self.table_fields):
+ return False
+ target = self.table_fields[part.index]
+ source = source_fields.get(target.id)
+ if source is None:
+ # Missing fields remain NULL in evolved stats.
+ return True
+ source_type, target_type = source.type, target.type
+ if isinstance(source_type, AtomicType) and isinstance(target_type,
AtomicType):
+ if source_type.type == target_type.type:
+ return True
+ widths = {'TINYINT': 8, 'SMALLINT': 16, 'INT': 32, 'BIGINT':
64}
+ source_width = widths.get(source_type.type)
+ target_width = widths.get(target_type.type)
+ if source_width and target_width and source_width <
target_width:
+ # Java also rejects literals that cannot be represented in
+ # the old type. Stats values need no conversion in Python.
+ bound = 1 << (source_width - 1)
+ return all(value is not None and -bound <= value < bound
+ for value in (part.literals or []))
+ return False
+ return source_type == target_type
+
+ def conjuncts(part):
+ if part.method == 'and':
+ return [child for item in part.literals for child in
conjuncts(item)]
+ return [part]
+
+ return PredicateBuilder.and_predicates([part for part in
conjuncts(predicate) if safe(part)])
+
@staticmethod
def _create_index_cast_mapping(table_fields: List[DataField],
data_fields: List[DataField]) -> Dict[str,
Optional[List[int]]]:
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 b50842299f..dcd8c0a967 100644
--- a/paimon-python/pypaimon/read/reader/data_file_batch_reader.py
+++ b/paimon-python/pypaimon/read/reader/data_file_batch_reader.py
@@ -15,13 +15,13 @@
# specific language governing permissions and limitations
# under the License.
-from decimal import Decimal, localcontext, ROUND_HALF_UP
from typing import List, Optional, Tuple
import pyarrow as pa
import pyarrow.compute as pc
from pyarrow import RecordBatch
+from pypaimon.casting.schema_evolution import cast_array_for_schema_evolution
from pypaimon.common.file_io import FileIO
from pypaimon.utils.arrow_utils import zero_column_batch
from pypaimon.data.map_shared_shredding import (
@@ -44,32 +44,6 @@ def _is_character_string_type(data_type) -> bool:
return t == 'STRING' or t.startswith('VARCHAR') or t.startswith('CHAR')
-def cast_array_for_schema_evolution(array, target_type):
- """Cast old-file values using Java's DECIMAL rounding and overflow rules.
-
- PyArrow's unsafe DECIMAL scale reduction truncates and can retain values
- outside the target precision. Java rounds HALF_UP and returns NULL for
- values that do not fit after rounding. Other conversions retain the
- existing unsafe-cast behavior (such as DOUBLE -> INT truncation).
- """
- if (pa.types.is_decimal(array.type) and pa.types.is_decimal(target_type)
- and array.type.scale > target_type.scale):
- values = []
- with localcontext() as context:
- # Rounding a maximal value can carry into one extra digit.
- context.prec = max(array.type.precision, target_type.precision) + 1
- quantum = Decimal(1).scaleb(-target_type.scale)
- for value in array.to_pylist():
- if value is None:
- values.append(None)
- continue
- rounded = value.quantize(quantum, rounding=ROUND_HALF_UP)
- values.append(rounded if len(rounded.as_tuple().digits)
- <= target_type.precision else None)
- return pa.array(values, type=target_type)
- return array.cast(target_type, safe=False)
-
-
def _unslice(array):
"""Re-materialize a sliced array so offsets/buffers start at zero.
@@ -335,9 +309,7 @@ class DataFileBatchReader(RecordBatchReader):
return _constructed_to_string_array(array, file_type)
# Leaf / non-nested: cast to the target type when it differs.
target_pa_type = PyarrowFieldParser.from_paimon_type(target_type)
- if array.type != target_pa_type:
- return cast_array_for_schema_evolution(array, target_pa_type)
- return array
+ return cast_array_for_schema_evolution(array, target_pa_type,
file_type, target_type)
def read_arrow_batch(self, start_idx=None, end_idx=None) ->
Optional[RecordBatch]:
if isinstance(self.format_reader, FormatBlobReader):
@@ -423,7 +395,7 @@ class DataFileBatchReader(RecordBatchReader):
common (non-evolution) path zero-copy.
DECIMAL scale reductions use Java's HALF_UP and overflow-to-NULL
- semantics; other conversions retain the existing unsafe Arrow cast.
+ semantics; numeric narrowing and timestamp precision also follow Java
casts.
Evolution legality is the writer's concern (``DataTypeCasts``).
"""
out_arrays = []
diff --git a/paimon-python/pypaimon/read/scanner/file_scanner.py
b/paimon-python/pypaimon/read/scanner/file_scanner.py
index a7eb538520..7a60208a41 100755
--- a/paimon-python/pypaimon/read/scanner/file_scanner.py
+++ b/paimon-python/pypaimon/read/scanner/file_scanner.py
@@ -890,6 +890,8 @@ class FileScanner:
stats.buckets_after_pruning.add((partition_key, entry.bucket))
# Get SimpleStatsEvolution for this schema
evolution =
self.simple_stats_evolutions.get_or_create(entry.file.schema_id)
+ stats_predicate = self.simple_stats_evolutions.filter_unsafe_filter(
+ entry.file.schema_id, self.predicate_for_stats)
# Apply evolution to stats
if self.table.is_primary_key_table:
@@ -903,7 +905,7 @@ class FileScanner:
return False
# Java enables value filtering only when batch scans exclude L0.
# With L0 present, pruning an update can expose an older value.
- if self.skip_level0 and self.predicate_for_stats:
+ if self.skip_level0 and stats_predicate:
if entry.file.value_stats_cols is None and
entry.file.write_cols is not None:
stats_fields = entry.file.write_cols
else:
@@ -913,7 +915,7 @@ class FileScanner:
entry.file.row_count,
stats_fields
)
- if not self.predicate_for_stats.test_by_simple_stats(
+ if not stats_predicate.test_by_simple_stats(
evolved_stats,
entry.file.row_count
):
@@ -922,7 +924,7 @@ class FileScanner:
stats.entries_after_stats += 1
return True
else:
- if not self.predicate or self.predicate_for_stats is None:
+ if not self.predicate or stats_predicate is None:
if stats is not None:
stats.entries_after_stats += 1
return True
@@ -938,7 +940,7 @@ class FileScanner:
entry.file.row_count,
stats_fields
)
- kept = self.predicate_for_stats.test_by_simple_stats(
+ kept = stats_predicate.test_by_simple_stats(
evolved_stats,
entry.file.row_count
)
diff --git
a/paimon-python/pypaimon/tests/manifest/simple_stats_evolutions_test.py
b/paimon-python/pypaimon/tests/manifest/simple_stats_evolutions_test.py
index 9f134a96cf..8831bd0deb 100644
--- a/paimon-python/pypaimon/tests/manifest/simple_stats_evolutions_test.py
+++ b/paimon-python/pypaimon/tests/manifest/simple_stats_evolutions_test.py
@@ -43,6 +43,48 @@ class SimpleStatsEvolutionsTest(unittest.TestCase):
evolution = evolutions.get_or_create(0)
self.assertIsNone(evolution.index_mapping)
+ def test_unsafe_cast_filters_keep_safe_conjuncts_and_preserve_or(self):
+ source = self._make_fields([(0, 'old_name', 'BIGINT'), (1, 'id',
'INT')])
+ target = self._make_fields([(1, 'id', 'INT'), (0, 'new_name',
'TINYINT')])
+ schemas = {0: source, 1: target}
+ evolutions = SimpleStatsEvolutions(lambda sid: schemas[sid], 1)
+ narrowed = Predicate('equal', 1, 'new_name', [-128])
+ unchanged = Predicate('equal', 0, 'id', [1])
+ self.assertIsNone(evolutions.filter_unsafe_filter(0, narrowed))
+ conjunction = Predicate('and', None, None, [narrowed, unchanged])
+ self.assertEqual(evolutions.filter_unsafe_filter(0, conjunction),
unchanged)
+ disjunction = Predicate('or', None, None, [narrowed, unchanged])
+ self.assertIsNone(evolutions.filter_unsafe_filter(0, disjunction))
+ self.assertIs(evolutions.filter_unsafe_filter(1, narrowed), narrowed)
+
+ def test_integer_widening_only_keeps_representable_literals(self):
+ source = self._make_fields([(0, 'v', 'TINYINT')])
+ target = self._make_fields([(0, 'v', 'BIGINT')])
+ schemas = {0: source, 1: target}
+ evolutions = SimpleStatsEvolutions(lambda sid: schemas[sid], 1)
+ for value in (-128, 127):
+ predicate = Predicate('equal', 0, 'v', [value])
+ self.assertEqual(evolutions.filter_unsafe_filter(0, predicate),
predicate)
+ for value in (-129, 128):
+ self.assertIsNone(evolutions.filter_unsafe_filter(0,
Predicate('equal', 0, 'v', [value])))
+ for method in ('in', 'notIn'):
+ self.assertIsNone(evolutions.filter_unsafe_filter(0,
Predicate(method, 0, 'v', [1, None])))
+ predicate = Predicate('isNull', 0, 'v')
+ self.assertEqual(evolutions.filter_unsafe_filter(0, predicate),
predicate)
+
+ def test_timestamp_precision_and_decimal_casts_disable_stats_filter(self):
+ for old_type, new_type in [('TIMESTAMP(3)', 'TIMESTAMP(1)'),
+ ('TIMESTAMP_LTZ(3)', 'TIMESTAMP_LTZ(1)'),
+ ('DECIMAL(10,2)', 'DECIMAL(10,1)'),
+ ('DOUBLE', 'INT')]:
+ source = self._make_fields([(0, 'v', old_type)])
+ target = self._make_fields([(0, 'v', new_type)])
+ schemas = {0: source, 1: target}
+ evolutions = SimpleStatsEvolutions(lambda sid: schemas[sid], 1)
+ for method in ('equal', 'notEqual', 'isNull', 'isNotNull'):
+ with self.subTest(old_type=old_type, new_type=new_type,
method=method):
+ self.assertIsNone(evolutions.filter_unsafe_filter(0,
Predicate(method, 0, 'v', [1])))
+
def test_predicate_with_incomplete_projected_stats(self):
fields = self._make_fields([(0, 'a', 'INT'), (1, 'b', 'INT')])
evolution = SimpleStatsEvolutions(lambda _: fields, 0).get_or_create(0)
diff --git a/paimon-python/pypaimon/tests/native_schema_evolution_cast_test.py
b/paimon-python/pypaimon/tests/native_schema_evolution_cast_test.py
new file mode 100644
index 0000000000..4f6d7b6400
--- /dev/null
+++ b/paimon-python/pypaimon/tests/native_schema_evolution_cast_test.py
@@ -0,0 +1,216 @@
+# 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.
+
+"""Read existing Python schema changes through Native using REST tables."""
+
+from contextlib import ExitStack
+from decimal import Decimal
+from unittest.mock import patch
+
+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
+
+rest_catalog = native_plan_rest_test.rest_catalog
+
+
+pytestmark = [pytest.mark.native_plan, pytest.mark.skipif(
+ not native_runtime_available(), reason='Rust reader required')]
+
+
+def _write(table, data):
+ builder = table.new_batch_write_builder()
+ writer, commit = builder.new_write(), builder.new_commit()
+ try:
+ writer.write_arrow(data)
+ commit.commit(writer.prepare_commit())
+ finally:
+ writer.close()
+ commit.close()
+
+
+def _table(catalog, mode, arrow_type, values):
+ 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 = catalog.get_table('default.t')
+ _write(table, pa.table({'id': list(range(len(values))), 'v':
pa.array(values, type=arrow_type)}))
+
+
+def _read(table, native, predicate=None, projection=None):
+ table = table.copy({'read.native.enabled': str(native).lower(),
+ 'scan.native-plan.enabled': str(native).lower()})
+ builder = table.new_read_builder()
+ if predicate is not None:
+ builder.with_filter(predicate)
+ if projection is not None:
+ builder.with_projection(projection)
+ scan = builder.new_scan()
+ with ExitStack() as stack:
+ if native:
+ stack.enter_context(patch.object(scan.file_scanner, 'scan',
+
side_effect=AssertionError('Native plan fell back')))
+
stack.enter_context(patch('pypaimon.read.table_read.TableRead._create_split_read',
+ side_effect=AssertionError('Native read
fell back')))
+ plan = scan.plan()
+ result = builder.new_read().to_arrow(plan.splits())
+ return result.sort_by([('id', 'ascending')])
+
+
+_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]),
+ (pa.int64(), 'INT', [2 ** 31, -(2 ** 31) - 1, 2 ** 63 - 1, None],
+ [-(2 ** 31), 2 ** 31 - 1, -1, None]),
+ (pa.float64(), 'TINYINT', [128.9, -129.9, 1e30, -1e30, None], [-128, 127,
-1, 0, None]),
+ (pa.float64(), 'INT', [42.9, -42.9, 1e30, -1e30, None],
+ [42, -42, 2 ** 31 - 1, -(2 ** 31), None]),
+ (pa.float64(), 'BIGINT', [42.9, -42.9, 1e30, -1e30, None],
+ [42, -42, 2 ** 63 - 1, -(2 ** 63), None]),
+ (pa.decimal128(38, 2), 'BIGINT',
+ [Decimal('42.99'), Decimal('-42.99'), Decimal('9223372036854775808.00'),
None],
+ [42, -42, -(2 ** 63), None]),
+ (pa.decimal128(10, 2), 'TINYINT', [Decimal('128.99'), Decimal('-129.99'),
None], [-128, 127, None]),
+]
+
+
[email protected]('source_type,target_name,values,expected',
_NUMERIC_CASES)
+def test_single_value_numeric_stats_do_not_prune_converted_match(
+ rest_catalog, source_type, target_name, values, expected):
+ catalog, _ = rest_catalog
+ _table(catalog, 'append', source_type, values[:1])
+ catalog.alter_table('default.t', [SchemaChange.update_column_type('v',
AtomicType(target_name))], False)
+ table = catalog.get_table('default.t')
+ predicate = table.new_read_builder().new_predicate_builder().equal('v',
expected[0])
+ for native in (False, True):
+ assert _read(table, native, predicate,
['id']).column('id').to_pylist() == [0]
+
+
[email protected]('zone', [None, 'UTC'])
+def
test_single_value_timestamp_stats_do_not_prune_truncated_match(rest_catalog,
zone):
+ catalog, _ = rest_catalog
+ _table(catalog, 'append', pa.timestamp('ms', zone), [123])
+ target_name = 'TIMESTAMP_LTZ(1)' if zone else 'TIMESTAMP(1)'
+ catalog.alter_table('default.t', [SchemaChange.update_column_type('v',
AtomicType(target_name))], False)
+ table = catalog.get_table('default.t')
+ literal = pa.scalar(100, type=pa.timestamp('ms', zone)).as_py()
+ predicate = table.new_read_builder().new_predicate_builder().equal('v',
literal)
+ for native in (False, True):
+ assert _read(table, native, predicate,
['id']).column('id').to_pylist() == [0]
+
+
[email protected]('mode', ['append', 'pk', 'de'])
[email protected]('source_type,target_name,values,expected',
_NUMERIC_CASES,
+ ids=['long-byte', 'long-short', 'long-int',
'double-byte', 'double-int',
+ 'double-long', 'decimal-long', 'decimal-byte'])
+def test_numeric_narrowing_reads_old_and_new_rest_files(
+ rest_catalog, mode, source_type, target_name, values, expected):
+ catalog, _ = rest_catalog
+ _table(catalog, mode, source_type, values)
+ catalog.alter_table('default.t', [SchemaChange.update_column_type('v',
AtomicType(target_name))], False)
+ table = catalog.get_table('default.t')
+ for native in (False, True):
+ assert _read(table, native).column('v').to_pylist() == expected
+ target_type = PyarrowFieldParser.from_paimon_type(AtomicType(target_name))
+ _write(table, pa.table({'id': [len(values)], 'v': pa.array([7],
type=target_type)}))
+ table = catalog.get_table('default.t')
+ 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 + [7]
+ matches = {native: _read(table, native, predicate,
['id']).column('id').to_pylist()
+ for native in (False, True)}
+ assert matches == {False: [0], True: [0]}
+
+
[email protected]('mode', ['append', 'pk', 'de'])
[email protected]('precision', [0, 1, 2, 4, 5, 7, 8])
[email protected]('zone', [None, 'UTC'])
+def test_timestamp_declared_precision_reads_old_rest_files(rest_catalog, mode,
precision, zone):
+ catalog, _ = rest_catalog
+ raw = [123456789, -1, -123456789, 1000000000, None]
+ _table(catalog, mode, pa.timestamp('ns', zone), raw)
+ type_name = 'TIMESTAMP_LTZ' if zone else 'TIMESTAMP'
+ catalog.alter_table('default.t', [SchemaChange.update_column_type(
+ 'v', AtomicType('%s(%d)' % (type_name, precision)))], False)
+ table = catalog.get_table('default.t')
+ target_type =
PyarrowFieldParser.from_paimon_type(table.field_dict['v'].type)
+ ticks_per_second = {'s': 1, 'ms': 1000, 'us': 1000000, 'ns':
1000000000}[target_type.unit]
+ quantum = 10 ** (9 - precision)
+ expected = [None if value is None else (value // quantum * quantum) //
(1000000000 // ticks_per_second)
+ for value in raw]
+ for native in (False, True):
+ result = _read(table, native).column('v')
+ assert result.type == target_type
+ assert result.cast(pa.int64()).to_pylist() == expected
+ if precision <= 6:
+ literal = pa.scalar(expected[0],
type=pa.int64()).cast(target_type).as_py()
+ predicate =
table.new_read_builder().new_predicate_builder().equal('v', literal)
+ assert _read(table, native, predicate,
['id']).column('id').to_pylist() == [0]
+
+
[email protected]('mode', ['append', 'pk', 'de'])
+def
test_nested_narrowing_preserves_null_containers_and_leaf_projection(rest_catalog,
mode):
+ catalog, _ = rest_catalog
+ schema = pa.schema([
+ ('id', pa.int64()),
+ ('r', pa.struct([('n', pa.float64()), ('t', pa.timestamp('ns'))])),
+ ('a', pa.list_(pa.int64())),
+ ('m', pa.map_(pa.string(), pa.timestamp('ns'))),
+ ])
+ options = {'file.format': 'parquet', 'read.batch-size': '1'}
+ if mode == 'pk':
+ options['bucket'] = '1'
+ elif 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([
+ {'id': 0, 'r': {'n': 128.9, 't': 123456789}, 'a': [128, None, -129],
+ 'm': [('one', 123456789), ('two', -1)]},
+ {'id': 1, 'r': None, 'a': None, 'm': None},
+ {'id': 2, 'r': {'n': -129.9, 't': -1}, 'a': [], 'm': []},
+ ], schema=schema))
+ catalog.alter_table('default.t', [
+ SchemaChange.update_column_type(['r', 'n'], AtomicType('TINYINT')),
+ SchemaChange.update_column_type(['r', 't'],
AtomicType('TIMESTAMP(1)')),
+ SchemaChange.update_column_type(['a', 'element'],
AtomicType('TINYINT')),
+ SchemaChange.update_column_type(['m', 'value'],
AtomicType('TIMESTAMP(1)')),
+ ], False)
+ table = catalog.get_table('default.t')
+ expected_a = [[-128, None, 127], None, []]
+ expected_m = pa.array([[('one', 100), ('two', -100)], None, []],
+ type=pa.map_(pa.string(),
pa.timestamp('ms'))).to_pylist()
+ for native in (False, True):
+ actual = _read(table, native)
+ assert actual.column('a').to_pylist() == expected_a
+ assert actual.column('m').to_pylist() == expected_m
+ # ROW field ids belong to the catalog; assert values independently of
them.
+ assert actual.column('r').flatten()[0].to_pylist() == [-128, None, 127]
+ assert actual.column('r').flatten()[1].cast(pa.int64()).to_pylist() ==
[100, None, -100]
+ leaves = _read(table, native, projection=['id', 'r.n', 'r.t'])
+ assert leaves.column('r_n').to_pylist() == [-128, None, 127]
+ assert leaves.column('r_t').cast(pa.int64()).to_pylist() == [100,
None, -100]
diff --git a/paimon-python/pypaimon/tests/schema_evolution_cast_test.py
b/paimon-python/pypaimon/tests/schema_evolution_cast_test.py
new file mode 100644
index 0000000000..1b6e4f817b
--- /dev/null
+++ b/paimon-python/pypaimon/tests/schema_evolution_cast_test.py
@@ -0,0 +1,56 @@
+# 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.
+
+from decimal import Decimal
+
+import pyarrow as pa
+import pytest
+
+from pypaimon.casting.schema_evolution import cast_array_for_schema_evolution
+from pypaimon.schema.data_types import AtomicType
+
+
[email protected]('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]),
+])
[email protected]('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
+
+
[email protected]('zone', [None, 'UTC'])
[email protected]('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]
