diff --git a/.github/workflows/ci-python.yml b/.github/workflows/ci-python.yml index c28becd9f552..797bed8bf616 100644 --- a/.github/workflows/ci-python.yml +++ b/.github/workflows/ci-python.yml @@ -214,6 +214,12 @@ jobs: source = json.loads(distribution.read_text('direct_url.json')) print('pypaimon-rust', distribution.version, 'commit', source['vcs_info']['commit_id']) + # Fail before the suite when Rust main lacks the required core bridge. + from pypaimon_rust.data import ( + cast_array_for_schema_evolution, + normalize_array_for_write, + read_arrow_schema, + ) PY - name: Run Python tests with native planning, reading, writing, updating and committing @@ -239,7 +245,8 @@ jobs: TZ="$zone" python -m pytest -q \ paimon-python/pypaimon/tests/native_cast_completion_test.py \ paimon-python/pypaimon/tests/native_temporal_zone_cast_test.py \ - -k 'TIMESTAMP_LTZ or temporal' + paimon-python/pypaimon/tests/native_schema_cast_test.py \ + -k 'TIMESTAMP or temporal' done torch: diff --git a/paimon-python/pypaimon/casting/data_type_casts.py b/paimon-python/pypaimon/casting/data_type_casts.py index 849be57e541e..027f3293c729 100644 --- a/paimon-python/pypaimon/casting/data_type_casts.py +++ b/paimon-python/pypaimon/casting/data_type_casts.py @@ -216,6 +216,11 @@ def can_execute_cast(source_type, target_type) -> bool: # Same root: identity, or a same-shape constructed type whose value is # rebuilt by the read path's field-id alignment rather than a value cast. if source_root == target_root: + if source_root == TIME: + from pypaimon.table.row.generic_row import _parse_type_precision_scale + # Compare parsed precision: TIME and TIME(0) describe the same type. + return (_parse_type_precision_scale(source_type.type)[0] + == _parse_type_precision_scale(target_type.type)[0]) return True # Constructed -> character string is rendered by the read path's custom # ``_constructed_to_string_array`` (see DataFileBatchReader), not a cast. @@ -254,7 +259,7 @@ def _schema_evolution_cast_supported(source_type, target_type) -> bool: pa.nulls(1, type=source_pa), target_pa, source_type, target_type) ok = True except (pa.lib.ArrowNotImplementedError, pa.lib.ArrowInvalid, - pa.lib.ArrowTypeError): + pa.lib.ArrowTypeError, ValueError, NotImplementedError): ok = False _EXECUTABLE_CAST_CACHE[cache_key] = ok return ok diff --git a/paimon-python/pypaimon/casting/schema_evolution.py b/paimon-python/pypaimon/casting/schema_evolution.py index 1116aeaacd51..1b6c76175dc9 100644 --- a/paimon-python/pypaimon/casting/schema_evolution.py +++ b/paimon-python/pypaimon/casting/schema_evolution.py @@ -18,6 +18,7 @@ """Java cast rules shared by old-file readers.""" import math +import json import pyarrow as pa @@ -69,9 +70,16 @@ def _truncate_timestamp(array, target_type, source_paimon_type, target_paimon_ty 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()] + values = [] + for value in array.cast(pa.int64()).to_pylist(): + if value is None: + values.append(None) + continue + nanos = value * nanos_per_tick[array.type.unit] // quantum * quantum + millis, remainder = divmod(nanos, 1000000) + # Java stores signed-long milliseconds before converting storage units. + millis = _narrow_integer(millis, 64) + values.append((millis * 1000000 + remainder) // nanos_per_tick[target_type.unit]) # 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. @@ -114,6 +122,33 @@ def _integer_to_timestamp(array, target_type): return pa.array(values, type=pa.int64()).cast(target_type) +def _native_schema_cast(array, target_type, source_paimon_type, target_paimon_type): + if source_paimon_type is None or target_paimon_type is None: + return None + try: + from pypaimon_rust.data import cast_array_for_schema_evolution as native_cast + except ImportError: + return None + source = json.dumps(source_paimon_type.to_dict()) + target = json.dumps(target_paimon_type.to_dict()) + if isinstance(array, pa.ChunkedArray): + chunks = array.chunks or [pa.array([], type=array.type)] + # Runtime cast results may be NULL despite a logical NOT NULL child. + return pa.chunked_array([native_cast(chunk, source, target) for chunk in chunks]) + return native_cast(array, source, target) + + +def read_arrow_schema(fields): + """Use core runtime nullability while retaining logical write constraints.""" + try: + from pypaimon_rust.data import read_arrow_schema as native_schema + except ImportError: + from pypaimon.schema.data_types import PyarrowFieldParser + return PyarrowFieldParser.from_paimon_schema(fields) + schema = native_schema(json.dumps([field.to_dict() for field in fields])) + return pa.schema([field.remove_metadata() for field in schema]) + + 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. @@ -126,6 +161,9 @@ def cast_array_for_schema_evolution(array, target_type, source_paimon_type=None, precision overflow after rounding. LTZ conversions use the process local zone, including Java gap/overlap rules. """ + native = _native_schema_cast(array, target_type, source_paimon_type, target_paimon_type) + if native is not None: + return native if pa.types.is_string(target_type): strings = temporal_to_string(array, source_paimon_type, target_paimon_type) if strings is not None: 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 e9d8066a3977..3e5c2d6e3a4b 100644 --- a/paimon-python/pypaimon/read/reader/data_file_batch_reader.py +++ b/paimon-python/pypaimon/read/reader/data_file_batch_reader.py @@ -21,7 +21,7 @@ import pyarrow.compute as pc from pyarrow import RecordBatch -from pypaimon.casting.schema_evolution import cast_array_for_schema_evolution +from pypaimon.casting.schema_evolution import cast_array_for_schema_evolution, read_arrow_schema from pypaimon.common.file_io import FileIO from pypaimon.utils.arrow_utils import zero_column_batch from pypaimon.data.map_shared_shredding import ( @@ -68,6 +68,10 @@ def _constructed_to_string_array(array, file_type): ROW -> ``{v1, v2}``, ARRAY -> ``[e1, e2]``, MAP -> ``{k1 -> v1, k2 -> v2}``. Sub-values are rendered recursively; a NULL sub-value renders as the literal ``null`` while a NULL container row stays NULL.""" + from pypaimon.casting.schema_evolution import _native_schema_cast + native = _native_schema_cast(array, pa.string(), file_type, AtomicType('STRING')) + if native is not None: + return native array = _unslice(array) valid = pc.is_valid(array).to_pylist() out = [] @@ -172,7 +176,7 @@ def __init__(self, format_reader: RecordBatchReader, index_mapping: List[int], p self._file_data_field_map = { field.name: field for field in source_fields } - self.schema_map = {field.name: field for field in PyarrowFieldParser.from_paimon_schema(fields)} + self.schema_map = {field.name: field for field in read_arrow_schema(fields)} self.row_tracking_enabled = row_tracking_enabled self.first_row_id = first_row_id self.row_id_offsets = row_id_offsets @@ -264,6 +268,11 @@ def _align_array_by_id(self, array, file_type, target_type): """Return *array* converted to *target_type*, matching ROW sub-fields by field id (reorder, pad missing with NULL, follow renames, cast changed types) recursively, transparently through ARRAY/MAP wrappers.""" + from pypaimon.casting.schema_evolution import _native_schema_cast + native = _native_schema_cast( + array, PyarrowFieldParser.from_paimon_type(target_type), file_type, target_type) + if native is not None: + return native if isinstance(target_type, RowType) and isinstance(file_type, RowType): n = len(array) file_id_to_pos = {f.id: i for i, f in enumerate(file_type.fields)} diff --git a/paimon-python/pypaimon/read/table_read.py b/paimon-python/pypaimon/read/table_read.py index 537c2ae7b2f9..6e0b26362f16 100644 --- a/paimon-python/pypaimon/read/table_read.py +++ b/paimon-python/pypaimon/read/table_read.py @@ -25,6 +25,8 @@ import pandas import pyarrow +from pypaimon.casting.schema_evolution import read_arrow_schema + from pypaimon.common.options.core_options import CoreOptions from pypaimon.common.predicate import Predicate from pypaimon.common.predicate_json_parser import extract_referenced_fields @@ -46,7 +48,7 @@ MergeFileSplitRead, RawFileSplitRead, SplitRead, deferred_blob_field_names) from pypaimon.schema.data_types import ( - DataField, MapType, PyarrowFieldParser, is_blob_file_field, is_map_blob_type) + DataField, MapType, is_blob_file_field, is_map_blob_type) from pypaimon.table.row.offset_row import OffsetRow from pypaimon.utils.arrow_utils import zero_column_batch @@ -464,7 +466,7 @@ def _reader_adapter(self): return projected if predicate_field_names(self.predicate) & extra_names else adapter def _output_arrow_schema(self, splits=None): - schema = PyarrowFieldParser.from_paimon_schema(self.read_type) + schema = read_arrow_schema(self.read_type) masks = set() if splits: for split in splits: @@ -851,7 +853,12 @@ def _prepare_native_batch(self, batch, schema): def _flatten_native_nested_batch(self, batch, schema): _, paths = self._reader_adapter() arrays = [extract_array(batch, path) for path in paths] - fields = list(PyarrowFieldParser.from_paimon_schema(self._adapter_read_type)) + fields = list(read_arrow_schema(self._adapter_read_type)) + # Authorization can mask a NOT NULL field. Keep the captured source + # nullability when flattening, including whole-parent/MAP projections. + projected = output_schema(batch.schema, OutputProjection([ + (field.name, path) for field, path in zip(fields, paths)])) + fields = [field.with_nullable(field.nullable or projected[i].nullable) for i, field in enumerate(fields)] if self.include_row_kind: arrays.insert(0, batch.column(ROW_KIND_COLUMN)) fields.insert(0, batch.schema.field(ROW_KIND_COLUMN)) @@ -1159,12 +1166,13 @@ def _convert_rows_to_output_batches( yield from self._convert_rows_to_arrow_batches_with_row_kind( row_tuples, row_kinds, schema) return - physical_schema = PyarrowFieldParser.from_paimon_schema(self._adapter_read_type) + physical_schema = read_arrow_schema(self._adapter_read_type) if self.include_row_kind: physical_schema = self._add_row_kind_to_schema(physical_schema) for batch in self._convert_rows_to_arrow_batches_with_row_kind( row_tuples, row_kinds, physical_schema): - yield self._project_batch_to_output(batch) + yield self._try_to_pad_batch_by_schema( + self._project_batch_to_output(batch), schema, allow_type_cast=True) def _convert_rows_to_arrow_batches_with_row_kind( self, @@ -1665,8 +1673,8 @@ def _project_batch_to_output(self, batch: pyarrow.RecordBatch) -> pyarrow.Record # Project the already masked schema, retaining captured nullability. adapted_projection = OutputProjection([(alias, path) for alias, path in outputs]) actual = output_schema(batch.schema, adapted_projection) - declared = output_schema(PyarrowFieldParser.from_paimon_schema(self.read_type), self.output_projection) - fields = [field.with_nullable(actual[i].nullable) for i, field in enumerate(declared)] + declared = output_schema(read_arrow_schema(self.read_type), self.output_projection) + fields = [field.with_nullable(field.nullable or actual[i].nullable) for i, field in enumerate(declared)] if self.include_row_kind and ROW_KIND_COLUMN in batch.schema.names: arrays.insert(0, batch.column(ROW_KIND_COLUMN)) fields.insert(0, batch.schema.field(ROW_KIND_COLUMN)) @@ -1873,7 +1881,7 @@ def __authed_reader(self, split, auth_result, blob_parallelism=1, needs_convert_back = False if not isinstance(reader, RecordBatchReader): - schema = PyarrowFieldParser.from_paimon_schema(effective_read_type) + schema = read_arrow_schema(effective_read_type) reader = RecordReaderToBatchAdapter(reader, schema, include_row_kind=self.include_row_kind) if getattr(reader, 'blob_field_indices', None) is None: from pypaimon.read.reader.field_indices import ( diff --git a/paimon-python/pypaimon/schema/arrow_schema.py b/paimon-python/pypaimon/schema/arrow_schema.py index 98a232033b1f..010b1ae2d8cb 100644 --- a/paimon-python/pypaimon/schema/arrow_schema.py +++ b/paimon-python/pypaimon/schema/arrow_schema.py @@ -20,14 +20,18 @@ import pyarrow as pa -def _normalize_string_type(data_type): +def _normalize_string_type(data_type, ignore_child_nullability=False): + def child(field): + field = field.with_type(_normalize_string_type(field.type, ignore_child_nullability)) + return field.with_nullable(True) if ignore_child_nullability else field + if pa.types.is_large_string(data_type): return pa.string() if pa.types.is_struct(data_type): - return pa.struct([field.with_type(_normalize_string_type(field.type)) for field in data_type]) + return pa.struct([child(field) for field in data_type]) if (pa.types.is_list(data_type) or pa.types.is_large_list(data_type) or pa.types.is_fixed_size_list(data_type)): - field = data_type.value_field.with_type(_normalize_string_type(data_type.value_type)) + field = child(data_type.value_field) if pa.types.is_large_list(data_type): return pa.large_list(field) if pa.types.is_fixed_size_list(data_type): @@ -38,13 +42,14 @@ def _normalize_string_type(data_type): # includes the sorting flag. Preserve it along with the child fields. sorted_keys = data_type == pa.map_(data_type.key_field, data_type.item_field, keys_sorted=True) return pa.map_( - data_type.key_field.with_type(_normalize_string_type(data_type.key_type)), - data_type.item_field.with_type(_normalize_string_type(data_type.item_type)), + data_type.key_field.with_type(_normalize_string_type(data_type.key_type, ignore_child_nullability)), + child(data_type.item_field), keys_sorted=sorted_keys) return data_type -def arrow_schemas_compatible(source, target, check_top_level_nullability=True, allow_binary_compatibility=False): +def arrow_schemas_compatible(source, target, check_top_level_nullability=True, allow_binary_compatibility=False, + check_nested_nullability=True): """Compare ordered fields, ignoring metadata and STRING offset width. The writer opts into its existing top-level nullability and binary / @@ -55,7 +60,9 @@ def arrow_schemas_compatible(source, target, check_top_level_nullability=True, a for actual, expected in zip(source, target): if check_top_level_nullability and actual.nullable != expected.nullable: return False - if actual.type == expected.type or _normalize_string_type(actual.type) == _normalize_string_type(expected.type): + if (actual.type == expected.type + or _normalize_string_type(actual.type, not check_nested_nullability) + == _normalize_string_type(expected.type, not check_nested_nullability)): continue if allow_binary_compatibility and all( pa.types.is_binary(t) or pa.types.is_fixed_size_binary(t) @@ -65,6 +72,32 @@ def arrow_schemas_compatible(source, target, check_top_level_nullability=True, a return True +def normalize_arrow_write_nullability(data, target): + """Validate read-result NULLs in core before restoring writer child fields.""" + try: + from pypaimon_rust.data import normalize_array_for_write + except ImportError: + return data + options = dict(check_top_level_nullability=False, allow_binary_compatibility=True) + if (arrow_schemas_compatible(data.schema, target, **options) + or not arrow_schemas_compatible(data.schema, target, check_nested_nullability=False, **options)): + return data + arrays, fields = [], [] + for column, field, expected in zip(data.columns, data.schema, target): + if column.type != expected.type: + if isinstance(column, pa.ChunkedArray): + column = pa.chunked_array( + [normalize_array_for_write(chunk, expected.type) for chunk in column.chunks], type=expected.type) + else: + column = normalize_array_for_write(column, expected.type) + arrays.append(column) + fields.append(field.with_type(column.type)) + schema = pa.schema(fields, metadata=data.schema.metadata) + if isinstance(data, pa.RecordBatch): + return pa.RecordBatch.from_arrays(arrays, schema=schema) + return pa.Table.from_arrays(arrays, schema=schema) + + def normalize_arrow_strings(data): """Safely normalize large_string input before routing or buffering rows. diff --git a/paimon-python/pypaimon/tests/native_constructed_evolution_test.py b/paimon-python/pypaimon/tests/native_constructed_evolution_test.py new file mode 100644 index 000000000000..2eab7af72686 --- /dev/null +++ b/paimon-python/pypaimon/tests/native_constructed_evolution_test.py @@ -0,0 +1,96 @@ +# 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 constructed schema changes must read physical Arrow lists natively.""" + +import os +from contextlib import ExitStack +from unittest.mock import patch +from urllib.parse import unquote, urlparse + +import pyarrow as pa +import pyarrow.parquet as pq +import pytest + +from pypaimon.schema.data_types import AtomicType +from pypaimon.schema.schema_change import SchemaChange +from pypaimon.tests import native_schema_evolution_cast_test + +rest_catalog = native_schema_evolution_cast_test.rest_catalog +pytestmark = native_schema_evolution_cast_test.pytestmark + +_LIST = pa.list_(pa.int32()) +_LARGE_LIST = pa.large_list(pa.int32()) +_CASES = [ + (_LIST, _LARGE_LIST, [[9], [1, None], [], None, [2, 3]], 'v', + ['[9]', '[1, null]', '[]', None, '[2, 3]']), + (pa.struct([('a', _LIST)]), pa.struct([('a', _LARGE_LIST)]), + [{'a': None}, None, {'a': [1, None]}, {'a': []}], 'v', + ['{null}', None, '{[1, null]}', '{[]}']), + (pa.map_(pa.string(), _LIST), pa.map_(pa.string(), _LARGE_LIST), + [[('one', [1, None]), ('empty', []), ('missing', None)], None, []], 'v', + ['{one -> [1, null], empty -> [], missing -> null}', None, '{}']), + (pa.struct([('a', _LIST)]), pa.struct([('a', _LARGE_LIST)]), + [{'a': None}, None, {'a': [1, None]}, {'a': []}], ['v', 'a'], + [{'a': None}, None, {'a': '[1, null]'}, {'a': '[]'}]), +] + + +@pytest.mark.parametrize('mode', ['append', 'pk', 'de']) +@pytest.mark.parametrize('target', ['STRING', 'CHAR(2)']) +@pytest.mark.parametrize('source,physical,values,path,expected', _CASES, + ids=['array', 'row', 'map', 'row-child']) +def test_large_list_constructed_strings_use_native_read( + rest_catalog, mode, target, source, physical, values, path, expected): + catalog, _ = rest_catalog + native_schema_evolution_cast_test._table(catalog, mode, source, values) + # The original nested schema must also plan through Rust, before changing + # only the logical type and preserving the old physical data files. + original = catalog.get_table('default.t').copy({'scan.native-plan.enabled': 'true'}) + original_plan = original.new_read_builder().new_scan().plan() + assert sum(file.row_count for split in original_plan.splits() for file in split.files) == len(values) + catalog.alter_table('default.t', [SchemaChange.update_column_type(path, AtomicType(target))], False) + table = catalog.get_table('default.t') + planning = table.copy({'scan.native-plan.enabled': 'false', 'read.native.enabled': 'false'}) + splits = planning.new_read_builder().new_scan().plan().splits() + assert splits + for split in splits: + for meta in split.files: + location = meta.file_path + if location.startswith('file:'): + location = unquote(urlparse(location).path) + data = pq.read_table(location) + index = data.schema.get_field_index('v') + field = data.schema.field(index) + data = data.set_column(index, pa.field( + field.name, physical, nullable=field.nullable, metadata=field.metadata), + data.column(index).cast(physical)) + pq.write_table(data, location, row_group_size=2) + meta.file_size = os.stat(location).st_size + if not pa.types.is_map(physical): + assert pq.read_schema(location).field('v').type == physical + # PyArrow restores LargeList at the top level and inside ROW, + # while its Parquet reader normalizes MAP values to regular List. + for native in (False, True): + reader = table.copy({'read.native.enabled': str(native).lower()}).new_read_builder().new_read() + guard = patch('pypaimon.read.table_read.TableRead._create_split_read', + side_effect=AssertionError('Native read fell back')) if native else ExitStack() + with guard: + result = reader.to_arrow(splits).sort_by([('id', 'ascending')]) + # Java's constructed casts use STRING for children and ignore the + # outer CHAR length. NULL containers and NULL elements stay distinct. + assert result.column('v').to_pylist() == expected diff --git a/paimon-python/pypaimon/tests/native_read_type_test.py b/paimon-python/pypaimon/tests/native_read_type_test.py index f15f1c0f4326..a1625f85d984 100644 --- a/paimon-python/pypaimon/tests/native_read_type_test.py +++ b/paimon-python/pypaimon/tests/native_read_type_test.py @@ -396,6 +396,27 @@ def test_whole_parent_and_child_apply_authorization_before_extraction( assert sorted(row[1] for row in rows if row[1] is not None) == [20] +@pytest.mark.parametrize('native', [False, True]) +def test_map_projection_preserves_masked_key_nullability(tmp_path, native): + from pypaimon.catalog.filesystem_catalog import FileSystemCatalog + from pypaimon.catalog.table_query_auth import TableQueryAuthResult + + table, _ = _table(tmp_path, 'pk', True) + table = table.copy({'read.native.enabled': str(native).lower(), 'query-auth.enabled': 'true'}) + builder = table.new_read_builder().with_projection(['attrs', "attrs['overflow']", 'id']) + auth = TableQueryAuthResult(None, {'id': json.dumps({'name': 'NULL'})}) + # A logical NOT NULL key becomes nullable after masking. Test the captured + # batch schema independently of the core's conservative read nullability. + with patch.object(FileSystemCatalog, 'auth_table_query', return_value=auth), \ + patch('pypaimon.read.table_read.read_arrow_schema', PyarrowFieldParser.from_paimon_schema): + splits = builder.new_scan().plan().splits() + result = builder.new_read().to_arrow_batch_reader(splits).read_all() + assert result.num_rows == 3 + assert result.schema.field('id').nullable + assert result.column('id').to_pylist() == [None] * 3 + assert sorted(value for value in result.column('attrs_overflow').to_pylist() if value is not None) == [20] + + def test_clipped_variant_reads_only_selected_paths_for_all_json_kinds(tmp_path): table, _ = _table(tmp_path, 'append', False, variant_shredded=True) schema = PyarrowFieldParser.from_paimon_schema(table.fields) diff --git a/paimon-python/pypaimon/tests/native_schema_cast_test.py b/paimon-python/pypaimon/tests/native_schema_cast_test.py new file mode 100644 index 000000000000..1a1e90adae5b --- /dev/null +++ b/paimon-python/pypaimon/tests/native_schema_cast_test.py @@ -0,0 +1,243 @@ +# 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. + +"""Python schema casts transport Arrow arrays into the core Java executors.""" + +from decimal import Decimal +from unittest.mock import patch + +import pyarrow as pa +import pytest + +from pypaimon.casting.schema_evolution import cast_array_for_schema_evolution +from pypaimon.schema.data_types import (ArrayType, AtomicType, DataField, DataTypeParser, + MapType, PyarrowFieldParser, RowType) +from pypaimon.tests import native_schema_evolution_cast_test + +rest_catalog = native_schema_evolution_cast_test.rest_catalog +pytestmark = native_schema_evolution_cast_test.pytestmark + +_CASES = [ + ('STRING', 'FLOAT', [' 1.25 ', '0x1.8p2', '1.25f', None], [1.25, 6.0, 1.25, None]), + ('STRING', 'DOUBLE', [' 1.25 ', '0x1.8p2', '1.25D', None], [1.25, 6.0, 1.25, None]), + ('FLOAT', 'STRING', [0.0001, 1e23, 16777216, None], ['1.0E-4', '1.0E23', '1.6777216E7', None]), + ('DOUBLE', 'STRING', [0.0001, 1e23, None], ['1.0E-4', '9.999999999999999E22', None]), + ('STRING', 'CHAR(3)', ['πŸ™‚δΈ­', 'a', 'abcdef', None], ['πŸ™‚δΈ­ ', 'a ', 'abc', None]), + ('STRING', 'VARCHAR(2)', ['πŸ™‚δΈ­x', 'a', None], ['πŸ™‚δΈ­', 'a', None]), + ('STRING', 'BINARY(3)', ['ab', 'abcd', None], [b'ab\x00', b'abc', None]), + ('STRING', 'VARBINARY(1)', ['πŸ™‚', None], [b'\xf0', None]), + ('INT', 'DECIMAL(4,2)', [1, 100, None], [Decimal('1.00'), None, None]), + ('STRING', 'DATE', ['42', '2001-02-28', None], [42, 11381, None]), + ('STRING', 'TIMESTAMP(1)', ['1970-01-01 00:00:00.123456', None], [100, None]), + ('TIMESTAMP(3)', 'TIMESTAMP(1)', [-(1 << 63), -1, 123, None], + [9223372036854775716, -100, 100, None]), + ('INT', 'TIMESTAMP(6)', [1, -1, None], [1000000, -1000000, None]), + ('INTEGER', 'BIGINT', [1, -1, None], [1, -1, None]), + ('DECIMAL', 'STRING', [Decimal('12'), None], ['12', None]), + ('TIME', 'STRING', [0, 1000, None], ['00:00:00', '00:00:01', None]), + ('TIMESTAMP', 'STRING', [0, 100000, None], + ['1970-01-01 00:00:00.000000', '1970-01-01 00:00:00.100000', None]), + ('TIMESTAMP_LTZ', 'TIMESTAMP_LTZ(3)', [0, 100000, None], [0, 100, None]), +] + + +def _type(name): + logical = DataTypeParser.parse_data_type(name) + return logical, PyarrowFieldParser.from_paimon_type(logical) + + +@pytest.mark.parametrize('source,target,values,expected', _CASES, + ids=lambda value: value if isinstance(value, str) else None) +@pytest.mark.parametrize('chunked', [False, True]) +def test_python_schema_casts_use_rust(source, target, values, expected, chunked): + source_type, source_arrow = _type(source) + target_type, target_arrow = _type(target) + array = pa.array(values, type=source_arrow) + if chunked: + array = pa.chunked_array([array.slice(0, 1), array.slice(1), pa.array([], type=source_arrow)]) + with patch('pypaimon.casting.schema_evolution.cast_decimal', side_effect=AssertionError('Python scalar cast')), \ + patch('pypaimon.casting.schema_evolution.cast_temporal', + side_effect=AssertionError('Python temporal cast')): + result = cast_array_for_schema_evolution(array, target_arrow, source_type, target_type) + if chunked: + assert isinstance(result, pa.ChunkedArray) + result = result.combine_chunks() + assert result.equals(pa.array(expected, type=target_arrow)) + + +@pytest.mark.parametrize('mode', ['append', 'pk', 'de']) +@pytest.mark.parametrize('source,target,values,expected', _CASES, + ids=lambda value: value if isinstance(value, str) else None) +def test_existing_scalar_evolution_uses_native_rest_reader(rest_catalog, mode, source, target, values, expected): + if mode == 'pk' and source == 'TIMESTAMP(3)' and target == 'TIMESTAMP(1)': + pytest.skip('The existing Python PK writer cannot materialize the extreme timestamp input') + catalog, _ = rest_catalog + source_type, source_arrow = _type(source) + _, target_arrow = _type(target) + native_schema_evolution_cast_test._table(catalog, mode, source_arrow, values, source_type) + from pypaimon.schema.schema_change import SchemaChange + catalog.alter_table('default.t', [SchemaChange.update_column_type('v', AtomicType(target))], False) + result = native_schema_evolution_cast_test._read(catalog.get_table('default.t'), True) + assert result.column('v').combine_chunks().equals(pa.array(expected, type=target_arrow)) + + +@pytest.mark.parametrize('target', ['INT', 'FLOAT', 'DOUBLE', 'DATE', 'TIMESTAMP(3)']) +def test_invalid_scalar_text_is_an_error_instead_of_null(target): + logical, arrow = _type(target) + with pytest.raises((ValueError, RuntimeError)): + cast_array_for_schema_evolution(pa.array(['bad']), arrow, AtomicType('STRING'), logical) + + +def test_empty_chunked_input_retains_the_requested_type(): + result = cast_array_for_schema_evolution( + pa.chunked_array([], type=pa.string()), pa.int32(), AtomicType('STRING'), AtomicType('INT')) + assert result.type == pa.int32() and len(result) == 0 + + +def test_time_precision_changes_are_not_executable(): + from pypaimon.casting.data_type_casts import can_execute_cast + assert not can_execute_cast(AtomicType('TIME(3)'), AtomicType('TIME(1)')) + with pytest.raises(NotImplementedError): + cast_array_for_schema_evolution( + pa.array([], type=pa.time32('ms')), pa.time32('ms'), AtomicType('TIME(3)'), AtomicType('TIME(1)')) + + +@pytest.mark.parametrize('source,target', [('TIME', 'TIME(0)'), ('TIME(3)', 'TIME( 3 )'), ('TIME(3)', 'TIME(03)')]) +def test_time_equivalent_precision_spellings_are_identity(source, target): + from pypaimon.casting.data_type_casts import can_execute_cast + assert can_execute_cast(AtomicType(source), AtomicType(target)) + data = pa.array([100, None], type=pa.time32('ms')) + assert cast_array_for_schema_evolution(data, data.type, AtomicType(source), AtomicType(target)).equals(data) + + +def _decimal_container(kind, precision): + leaf = AtomicType('DECIMAL({})'.format(precision), False) + if kind == 'array': + return ArrayType(True, leaf), ['v', 'element'] + if kind == 'row': + return RowType(True, [DataField(2, 'd', leaf)]), ['v', 'd'] + if kind == 'map': + return MapType(True, AtomicType('STRING', False), leaf), ['v', 'value'] + return leaf, 'v' + + +_DECIMAL_CONTAINERS = [ + ('leaf', [Decimal('9.995'), Decimal('1.000')], [None, Decimal('1.00')]), + ('array', [[Decimal('9.995'), Decimal('1.000')], None, []], [[None, Decimal('1.00')], None, []]), + ('row', [{'d': Decimal('9.995')}, None, {'d': Decimal('1.000')}], + [{'d': None}, None, {'d': Decimal('1.00')}]), + ('map', [[('a', Decimal('9.995')), ('b', Decimal('1.000'))], None, []], + [[('a', None), ('b', Decimal('1.00'))], None, []]), +] + + +@pytest.mark.parametrize('kind,values,expected', _DECIMAL_CONTAINERS) +@pytest.mark.parametrize('chunked', [False, True]) +def test_not_null_casts_can_return_java_runtime_nulls(kind, values, expected, chunked): + source, _ = _decimal_container(kind, '6,3') + target, _ = _decimal_container(kind, '3,2') + data = pa.array(values, type=PyarrowFieldParser.from_paimon_type(source)) + if chunked: + data = pa.chunked_array([data.slice(0, 1), data.slice(1)]) + result = cast_array_for_schema_evolution(data, PyarrowFieldParser.from_paimon_type(target), source, target) + assert result.to_pylist() == expected + result.validate(full=True) + + +@pytest.mark.parametrize('mode', ['append', 'pk', 'de']) +@pytest.mark.parametrize('kind,values,expected', _DECIMAL_CONTAINERS) +def test_not_null_decimal_evolution_uses_native_rest_reader(rest_catalog, mode, kind, values, expected): + catalog, _ = rest_catalog + source, path = _decimal_container(kind, '6,3') + native_schema_evolution_cast_test._table( + catalog, mode, PyarrowFieldParser.from_paimon_type(source), values, source) + from pypaimon.schema.schema_change import SchemaChange + catalog.alter_table('default.t', [SchemaChange.update_column_type(path, AtomicType('DECIMAL(3,2)'), True)], False) + table = catalog.get_table('default.t') + for native in [False, True]: + result = native_schema_evolution_cast_test._read(table, native) + assert result.column('v').to_pylist() == expected + # Read metadata permits cast-produced NULLs; persisted constraints are intact. + logical = catalog.get_table('default.t').fields[1].type + if kind == 'array': + logical = logical.element + elif kind == 'row': + logical = logical.fields[0].type + elif kind == 'map': + logical = logical.value + assert not logical.nullable + + +@pytest.mark.parametrize('target,expected', [ + ('DECIMAL(10,2)', [Decimal('1.24'), None]), ('INT', [1, None]), ('STRING', ['1.235', None]), +]) +def test_decimal256_physical_storage_uses_the_logical_decimal_type(target, expected): + logical, arrow = _type(target) + data = pa.array([Decimal('1.235'), None], type=pa.decimal256(20, 3)) + result = cast_array_for_schema_evolution(data, arrow, AtomicType('DECIMAL(20,3)'), logical) + assert result.equals(pa.array(expected, type=arrow)) + + +@pytest.mark.parametrize('chunked', [False, True]) +def test_large_list_rows_align_by_field_id_in_the_shared_core(chunked): + source_row = RowType(True, [DataField(1, 'old', AtomicType('STRING')), + DataField(2, 'dropped', AtomicType('INT'))]) + target_row = RowType(True, [DataField(2, 'dropped', AtomicType('BIGINT')), + DataField(1, 'renamed', AtomicType('INT')), + DataField(3, 'added', AtomicType('INT'))]) + source, target = ArrayType(True, source_row), ArrayType(True, target_row) + physical = pa.large_list(pa.struct([('old', pa.large_string()), ('dropped', pa.int32())])) + data = pa.array([[{'old': 'bad', 'dropped': 0}], + [{'old': '1.9', 'dropped': 7}, None], None, [], + [{'old': 'bad', 'dropped': 0}]], type=physical).slice(1, 3) + if chunked: + data = pa.chunked_array([data.slice(0, 1), data.slice(1)]) + result = cast_array_for_schema_evolution(data, PyarrowFieldParser.from_paimon_type(target), source, target) + assert result.to_pylist() == [[{'dropped': 7, 'renamed': 1, 'added': None}, None], None, []] + + +@pytest.mark.parametrize('native_write', [False, True]) +@pytest.mark.parametrize('kind,values,invalid', [ + ('array', [[Decimal('1.000')], None, []], [[None]]), + ('row', [{'d': Decimal('1.000')}, None], [{'d': None}]), + ('map', [[('a', Decimal('1.000'))], None, []], [[('a', None)]]), +]) +def test_read_results_write_back_with_not_null_constraints(rest_catalog, native_write, kind, values, invalid): + catalog, _ = rest_catalog + logical, _ = _decimal_container(kind, '6,3') + native_schema_evolution_cast_test._table( + catalog, 'append', PyarrowFieldParser.from_paimon_type(logical), values, logical) + table = catalog.get_table('default.t') + data = native_schema_evolution_cast_test._read(table, True) + table = table.copy({'write.native.enabled': str(native_write).lower()}) + builder = table.new_batch_write_builder() + writer, commit = builder.new_write(), builder.new_commit() + from pypaimon.write.native_write import NativeTableWrite + assert isinstance(writer, NativeTableWrite) == native_write + try: + writer.write_arrow(data) + commit.commit(writer.prepare_commit()) + bad = pa.Table.from_arrays( + [pa.array([100], type=pa.int64()), pa.array(invalid, type=data.schema.field('v').type)], + schema=data.schema) + with pytest.raises(ValueError): + writer.write_arrow(bad) + finally: + writer.close() + commit.close() + output = native_schema_evolution_cast_test._read(catalog.get_table('default.t'), True) + assert output.num_rows == 2 * len(values) diff --git a/paimon-python/pypaimon/tests/py36/filter_record_batch_reader_test.py b/paimon-python/pypaimon/tests/py36/filter_record_batch_reader_test.py new file mode 100644 index 000000000000..fbeb43a94c77 --- /dev/null +++ b/paimon-python/pypaimon/tests/py36/filter_record_batch_reader_test.py @@ -0,0 +1,64 @@ +# 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 unittest + +import pyarrow as pa + +from pypaimon.common.predicate import Predicate +from pypaimon.read.reader.filter_record_batch_reader import FilterRecordBatchReader +from pypaimon.read.reader.iface.record_batch_reader import RecordBatchReader +from pypaimon.schema.data_types import AtomicType, DataField + + +class _Batches(RecordBatchReader): + def __init__(self, batches): + self.batches = iter(batches) + + def read_arrow_batch(self, start_idx=None, end_idx=None): + return next(self.batches, None) + + def close(self): + pass + + +class FilterRecordBatchReaderTest(unittest.TestCase): + def test_row_filter_uses_logical_field_order_and_preserves_metadata(self): + schema = pa.schema([ + ('_row_kind', pa.string()), ('id', pa.int64()), ('behavior', pa.string()), + ], metadata={b'source': b'row-filter'}) + batch = pa.RecordBatch.from_arrays([ + pa.array(['+I', '-U', '+U', '-D']), pa.array([1, 2, 3, 4]), + pa.array(['apple', None, 'pear', 'apricot']), + ], schema=schema) + # The empty first result must not terminate the reader. + source = _Batches([batch.slice(2, 1), batch]) + predicate = Predicate('or', None, None, [ + Predicate('startsWith', 0, 'behavior', ['a']), + Predicate('isNull', 0, 'behavior', []), + ]) + fields = [DataField(0, 'behavior', AtomicType('STRING')), DataField(1, 'id', AtomicType('BIGINT'))] + reader = FilterRecordBatchReader(source, predicate, schema_fields=fields) + try: + result = reader.read_arrow_batch() + self.assertTrue(result.schema.equals(schema, check_metadata=True)) + self.assertEqual(result.column(0).to_pylist(), ['+I', '-U', '-D']) + self.assertEqual(result.column(1).to_pylist(), [1, 2, 4]) + self.assertEqual(result.column(2).to_pylist(), ['apple', None, 'apricot']) + self.assertIsNone(reader.read_arrow_batch()) + finally: + reader.close() diff --git a/paimon-python/pypaimon/tests/schema_evolution_cast_test.py b/paimon-python/pypaimon/tests/schema_evolution_cast_test.py index 363986c77d80..f6ce193f57d1 100644 --- a/paimon-python/pypaimon/tests/schema_evolution_cast_test.py +++ b/paimon-python/pypaimon/tests/schema_evolution_cast_test.py @@ -28,6 +28,13 @@ from pypaimon.schema.data_types import AtomicType +@pytest.fixture(autouse=True) +def python_fallback_casts(monkeypatch): + # These tests exercise libc/Python fallbacks, including mocked localtime + # and changes to TZ. Native casts have separate process-zone regressions. + monkeypatch.setattr('pypaimon.casting.schema_evolution._native_schema_cast', lambda *args: None) + + @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+ diff --git a/paimon-python/pypaimon/tests/schema_evolution_nested_read_test.py b/paimon-python/pypaimon/tests/schema_evolution_nested_read_test.py index 899634fafed0..8bbbd072d0a0 100644 --- a/paimon-python/pypaimon/tests/schema_evolution_nested_read_test.py +++ b/paimon-python/pypaimon/tests/schema_evolution_nested_read_test.py @@ -777,11 +777,17 @@ def test_can_execute_cast_decimal_precision(self): # A numeric -> decimal cast has a PyArrow kernel but is only executable # when the target precision can hold the source's range at the target # scale (INT needs >= 12 at scale 2, BIGINT >= 21). An empty-array probe - # misses this; can_execute_cast must reject the too-small targets so the - # read path does not later fail with ArrowInvalid. + # misses this. The Python fallback must reject those targets; the core + # executor supports them and returns NULL for per-value Java overflow. + try: + from pypaimon_rust.data import cast_array_for_schema_evolution + except ImportError: + native = False + else: + native = callable(cast_array_for_schema_evolution) for src, dst in [('INT', 'DECIMAL(10, 2)'), ('BIGINT', 'DECIMAL(10, 2)'), ('BIGINT', 'DECIMAL(20, 2)')]: - self.assertFalse(can_execute_cast(AtomicType(src), AtomicType(dst)), + self.assertEqual(can_execute_cast(AtomicType(src), AtomicType(dst)), native, '{} -> {}'.format(src, dst)) for src, dst in [('INT', 'DECIMAL(12, 2)'), ('BIGINT', 'DECIMAL(21, 2)'), ('INT', 'BIGINT'), ('DOUBLE', 'INT'), ('INT', 'STRING')]: diff --git a/paimon-python/pypaimon/write/native_write.py b/paimon-python/pypaimon/write/native_write.py index 0897ca8e95db..e737c6249722 100644 --- a/paimon-python/pypaimon/write/native_write.py +++ b/paimon-python/pypaimon/write/native_write.py @@ -21,7 +21,8 @@ import pyarrow as pa from pypaimon.common.options.core_options import CoreOptions -from pypaimon.schema.arrow_schema import arrow_schemas_compatible, normalize_arrow_strings +from pypaimon.schema.arrow_schema import ( + arrow_schemas_compatible, normalize_arrow_strings, normalize_arrow_write_nullability) from pypaimon.schema.data_types import PyarrowFieldParser, is_blob_file_field from pypaimon.table.bucket_mode import BucketMode from pypaimon.write.native_commit import ( @@ -226,6 +227,9 @@ def _write_native_batch(self, data, bucket=None): self._native_writer.write_arrow(data, bucket) def _prepare_native_arrow_data(self, data): + data = normalize_arrow_strings(data) + target = self._table_schema if data.schema.names == self._table_schema.names else self._schema + data = normalize_arrow_write_nullability(data, target) if self._write_cols is not None and arrow_schemas_compatible( data.schema, self._table_schema, check_top_level_nullability=False, allow_binary_compatibility=True): @@ -239,7 +243,7 @@ def _prepare_native_arrow_data(self, data): "Input schema isn't consistent with table schema and write cols. " f"Input schema is: {data.schema} Table schema is: {self._schema} " f"Write cols is: {self._write_cols}") - return normalize_arrow_strings(data) + return data def write_pandas(self, dataframe): if self._python_writer is not None: diff --git a/paimon-python/pypaimon/write/table_write.py b/paimon-python/pypaimon/write/table_write.py index 44909ea45a11..c78ea6080152 100644 --- a/paimon-python/pypaimon/write/table_write.py +++ b/paimon-python/pypaimon/write/table_write.py @@ -20,7 +20,8 @@ import pyarrow as pa from pypaimon.common.options.core_options import ChangelogProducer -from pypaimon.schema.arrow_schema import arrow_schemas_compatible, normalize_arrow_strings +from pypaimon.schema.arrow_schema import ( + arrow_schemas_compatible, normalize_arrow_strings, normalize_arrow_write_nullability) from pypaimon.schema.data_types import PyarrowFieldParser from pypaimon.snapshot.snapshot import BATCH_COMMIT_IDENTIFIER from pypaimon.table.row.blob import BlobConsumer @@ -72,7 +73,7 @@ def _create_row_key_extractor(self, static_partition): def write_arrow(self, data, bucket: Optional[int] = None): """Write Arrow data, optionally to a bucket assigned upstream.""" - self._validate_pyarrow_schema(data.schema) + data = self._prepare_arrow_schema(data) if isinstance(data, pa.RecordBatch): return self.write_arrow_batch(data, bucket) for batch in data.to_batches(): @@ -308,8 +309,7 @@ def _release_prepared_indexes(self) -> None: release() def _prepare_arrow_data(self, data): - self._validate_pyarrow_schema(data.schema) - data = normalize_arrow_strings(data) + data = self._prepare_arrow_schema(data) # Schema-only writers can normalize input without table options. table = getattr(self, 'table', None) if table is None: @@ -318,6 +318,18 @@ def _prepare_arrow_data(self, data): from pypaimon.write.row_kind import filter_write_batch return filter_write_batch(table, data) + def _prepare_arrow_schema(self, data): + data = normalize_arrow_strings(data) + target = self.table_pyarrow_schema + if data.schema.names != target.names and self.file_store_write.write_cols is not None: + target = self._write_cols_pyarrow_schema(self.file_store_write.write_cols) + validated = normalize_arrow_write_nullability(data, target) + self._validate_pyarrow_schema(validated.schema) + # Core checks only visible NOT NULL children. PyArrow's Parquet writer + # also rejects hidden NULL children under NULL ROW parents when their + # Arrow field is nonnullable, so retain the input's transport metadata. + return data + def _normalize_arrow_write_schema(self, data): # Compatible input batches may differ in top-level nullability or # metadata. Buffered writes and row-to-Arrow conversion must use the