From 2e8ed3f27d0cc53ee79b515ce5c2811ac2eec450 Mon Sep 17 00:00:00 2001 From: JingsongLi Date: Sat, 17 Oct 2026 21:37:14 +0800 Subject: [PATCH 1/3] Cover Native updates from Arrow map containers --- .../native_update_container_casts_test.py | 190 ++++++++++++++++++ 1 file changed, 190 insertions(+) create mode 100644 paimon-python/pypaimon/tests/native_update_container_casts_test.py diff --git a/paimon-python/pypaimon/tests/native_update_container_casts_test.py b/paimon-python/pypaimon/tests/native_update_container_casts_test.py new file mode 100644 index 000000000000..d19a0ff01235 --- /dev/null +++ b/paimon-python/pypaimon/tests/native_update_container_casts_test.py @@ -0,0 +1,190 @@ +# 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. + +"""Equivalent Arrow container inputs must work on the existing update APIs.""" + +from contextlib import ExitStack +from decimal import Decimal +from unittest.mock import patch + +import pyarrow as pa +import pytest + +from pypaimon import CatalogFactory, Schema +from pypaimon.common.predicate_builder import PredicateBuilder +from pypaimon.table.data_evolution_merge_into import WhenMatched, source_col +from pypaimon.write.table_update import TableUpdate +from pypaimon.write.table_update_by_row_id import TableUpdateByRowId +from pypaimon.write.table_upsert_by_key import TableUpsertByKey + + +pytestmark = pytest.mark.native_plan + +_CASES = [ + pytest.param(pa.int32(), pa.int64(), [1, None, 4], [1, None, 4], id='integer'), + pytest.param(pa.string(), pa.int32(), ['1', None, '4'], [1, None, 4], id='text'), + pytest.param(pa.decimal128(8, 3), pa.decimal128(10, 2), + [Decimal('1.250'), None, Decimal('-2.500')], + [Decimal('1.25'), None, Decimal('-2.50')], id='decimal'), + pytest.param(pa.large_binary(), pa.binary(), [b'\xff\0a', None, b''], [b'\xff\0a', None, b''], id='binary'), +] +_OPERATIONS = ['row_id', 'incremental', 'predicate_array', 'predicate_callable', 'upsert', 'merge'] + + +def _layout(source_value_type, target_value_type, source_values, expected_values, shape): + source_type = pa.map_(pa.field('source_key', pa.string(), nullable=False), + pa.field('source_value', source_value_type)) + target_type = pa.list_(pa.struct([('label', pa.string()), ('amount', target_value_type)])) + values = [[('a', source_values[0]), ('b', source_values[1])], None, [], [('z', source_values[2])]] + expected = [[{'label': 'a', 'amount': expected_values[0]}, + {'label': 'b', 'amount': expected_values[1]}], None, [], + [{'label': 'z', 'amount': expected_values[2]}]] + if shape == 'row': + source_type = pa.struct([('items', source_type), ('extra', pa.int32())]) + target_type = pa.struct([('items', target_type), ('extra', pa.int32())]) + values = [{'items': value, 'extra': i} if i != 1 else None for i, value in enumerate(values)] + expected = [{'items': value, 'extra': i} if i != 1 else None for i, value in enumerate(expected)] + elif shape == 'array': + source_type, target_type = pa.list_(source_type), pa.list_(target_type) + values = [[values[0], None, []], None, [], [values[3]]] + expected = [[expected[0], None, []], None, [], [expected[3]]] + elif shape == 'map': + source_type, target_type = pa.map_(pa.string(), source_type), pa.map_(pa.string(), target_type) + values = [[('items', values[0]), ('null', None)], None, [], [('last', values[3])]] + expected = [[('items', expected[0]), ('null', None)], None, [], [('last', expected[3])]] + # The slice retains nonzero MAP offsets at one or more nesting levels. + source = pa.array([values[0]] + values + [values[3]], type=source_type).slice(1, 4) + return source, target_type, expected + + +def _table(tmp_path, value_type, native): + catalog = CatalogFactory.create({'warehouse': str(tmp_path)}) + catalog.create_database('db', True) + schema = pa.schema([('id', pa.int32()), ('value', value_type), ('keep', pa.int32())]) + catalog.create_table('db.t', Schema.from_pyarrow_schema(schema, options={ + 'row-tracking.enabled': 'true', 'data-evolution.enabled': 'true', + 'write.native.enabled': str(native).lower(), + }), False) + table = catalog.get_table('db.t') + builder = table.new_batch_write_builder() + writer, commit = builder.new_write(), builder.new_commit() + try: + writer.write_arrow(pa.table({'id': list(range(5)), 'value': [None] * 5, + 'keep': list(range(10, 15))}, schema=schema)) + commit.commit(writer.prepare_commit()) + finally: + writer.close() + commit.close() + return table + + +def _native_only(native): + stack = ExitStack() + if native: + for cls, method in [(TableUpdateByRowId, '_load_existing_files_info'), + (TableUpdate, '_build_predicate_update_table'), + (TableUpsertByKey, '_upsert_partition')]: + stack.enter_context(patch.object(cls, method, side_effect=AssertionError('Python update fallback'))) + stack.enter_context(patch('pypaimon.table.data_evolution_merge_into._build_tables', + side_effect=AssertionError('Python MERGE fallback'))) + return stack + + +def _apply(table, source, native, stream, operation): + builder = table.new_stream_write_builder() if stream else table.new_batch_write_builder() + update = builder.new_update().with_update_type(['value']) + # Include empty chunks and boundaries that differ from the target files. + chunks = pa.chunked_array([source.slice(0, 0), source.slice(0, 1), source.slice(1, 2), + source.slice(3, 0), source.slice(3)]) + ids = pa.array(range(len(source)), type=pa.int32()) + data = pa.table({'value': chunks, '_ROW_ID': pa.array(range(len(source)), type=pa.int64())}) + with _native_only(native): + if operation == 'row_id': + messages = (update.update_by_arrow_with_row_id(data, 31) if stream + else update.update_by_arrow_with_row_id(data)) + elif operation == 'incremental': + writer = update.new_update_by_row_id(31) if stream else update.new_update_by_row_id() + messages = writer.update_columns(data, ['value']) + elif operation.startswith('predicate'): + kwargs = {} + if operation == 'predicate_callable': + def assignment(rows): + assert rows.column_names == ['id', '_ROW_ID'] + # Return a different chunk layout from the selected batches. + return chunks.take(rows['id']) + kwargs['read_columns'] = ['id'] + else: + assignment = chunks + predicate = PredicateBuilder(table.fields).less_than('id', len(source)) + messages = (update.update_by_predicate(predicate, {'value': assignment}, 31, **kwargs) if stream + else update.update_by_predicate(predicate, {'value': assignment}, **kwargs)) + elif operation == 'upsert': + data = pa.table({'id': ids, 'value': chunks}) + messages = update.upsert_by_arrow_with_key(data, ['id'], 31) if stream else \ + update.upsert_by_arrow_with_key(data, ['id']) + else: + data = pa.table({'id': ids, 'value': chunks}) + kwargs = {'commit_identifier': 31} if stream else {} + messages = update.merge_into(data, on=['id'], when_matched=[ + WhenMatched.update({'value': source_col('value')})], **kwargs) + commit = builder.new_commit() + try: + commit.commit(messages, 31) if stream else commit.commit(messages) + finally: + commit.close() + + +def _check(table, expected): + for native in (False, True): + view = table.copy({'scan.native-plan.enabled': str(native).lower(), + 'read.native.enabled': str(native).lower()}) + builder = view.new_read_builder().with_projection(['id', 'value', 'keep', '_ROW_ID']) + reader = builder.new_read() + with ExitStack() as stack: + if native: + stack.enter_context(patch.object(reader, '_create_split_read', + side_effect=AssertionError('Python read fallback'))) + rows = reader.to_arrow(builder.new_scan().plan().splits()).sort_by('id') + assert rows['value'].to_pylist() == expected + [None] * (5 - len(expected)) + assert rows['keep'].to_pylist() == list(range(10, 15)) + assert rows['_ROW_ID'].to_pylist() == list(range(5)) + + +@pytest.mark.parametrize('native', [False, True], ids=['python', 'native']) +@pytest.mark.parametrize('stream', [False, True], ids=['batch', 'stream']) +@pytest.mark.parametrize('operation', _OPERATIONS) +@pytest.mark.parametrize('shape', ['root', 'row', 'array', 'map']) +@pytest.mark.parametrize('source_type,target_type,values,expected', _CASES) +def test_map_inputs_update_array_rows(tmp_path, native, stream, operation, shape, + source_type, target_type, values, expected): + source, target_type, expected = _layout(source_type, target_type, values, expected, shape) + table = _table(tmp_path, target_type, native) + _apply(table, source, native, stream, operation) + _check(table, expected) + + +@pytest.mark.parametrize('native', [False, True], ids=['python', 'native']) +@pytest.mark.parametrize('operation', _OPERATIONS) +@pytest.mark.parametrize('shape', ['root', 'row', 'array']) +def test_invalid_map_child_cannot_change_existing_rows(tmp_path, native, operation, shape): + source, target_type, _ = _layout(pa.int64(), pa.int32(), [1, None, 2 ** 40], [1, None, 0], shape) + table = _table(tmp_path, target_type, native) + before = table.snapshot_manager().get_latest_snapshot().id + with pytest.raises((ValueError, TypeError, OverflowError)): + _apply(table, source, native, False, operation) + assert table.snapshot_manager().get_latest_snapshot().id == before + _check(table, [None] * 4) From f9fef33625598b753e67afc1936d7701c77cb9b9 Mon Sep 17 00:00:00 2001 From: JingsongLi Date: Sat, 17 Oct 2026 22:30:06 +0800 Subject: [PATCH 2/3] Cover encoded and scalar Native update inputs --- .../native_update_encoded_inputs_test.py | 329 ++++++++++++++++++ 1 file changed, 329 insertions(+) create mode 100644 paimon-python/pypaimon/tests/native_update_encoded_inputs_test.py diff --git a/paimon-python/pypaimon/tests/native_update_encoded_inputs_test.py b/paimon-python/pypaimon/tests/native_update_encoded_inputs_test.py new file mode 100644 index 000000000000..78f87236b20f --- /dev/null +++ b/paimon-python/pypaimon/tests/native_update_encoded_inputs_test.py @@ -0,0 +1,329 @@ +# 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. + +"""Arrow encodings used by existing update APIs must retain their values.""" + +from decimal import Decimal + +import numpy as np +import pyarrow as pa +import pytest + +from pypaimon.tests.native_update_container_casts_test import ( + _OPERATIONS, _apply, _check, _native_only, _table, +) + + +pytestmark = pytest.mark.native_plan + + +def _wrap(source, target, expected, shape): + if shape == 'row': + source = pa.StructArray.from_arrays([source], names=['nested']) + target = pa.struct([('nested', target)]) + expected = [{'nested': value} for value in expected] + elif shape == 'array': + source = pa.ListArray.from_arrays(pa.array(range(len(source) + 1), type=pa.int32()), source) + target = pa.list_(target) + expected = [[value] for value in expected] + elif shape == 'map': + keys = [str(i) for i in range(len(source))] + source = pa.MapArray.from_arrays(pa.array(range(len(source) + 1), type=pa.int32()), + pa.array(keys), source) + target = pa.map_(pa.string(), target) + expected = [[(key, value)] for key, value in zip(keys, expected)] + return source, target, expected + + +def _scalar_case(kind): + if kind == 'fixed_binary': + source = pa.array([b'old', b'a\0b', None, '中'.encode(), b'xyz', b'end'], type=pa.binary(3)).slice(1, 4) + return source, pa.string(), ['a\0b', None, '中', 'xyz'] + if kind == 'empty_fixed_binary': + return pa.array([b'', None, b'', b''], type=pa.binary(0)), pa.string(), ['', None, '', ''] + if kind == 'half_float': + values = np.array([0.1, 0.0, -3.25, 2 ** -24], dtype=np.float16) + source = pa.array(values, mask=np.array([False, True, False, False])) + return source, pa.string(), ['0.0999755859375', None, '-3.25', '5.960464477539063e-8'] + coefficient = '1508079088755887087797858531882816082' + value = Decimal(coefficient[:-1] + '.' + coefficient[-1]) + factory = pa.decimal256 if kind.startswith('decimal256') else pa.decimal128 + source = pa.array([value, None, value.copy_negate(), Decimal('1.0')], type=factory(38, 1)) + expected = [1.508079088755887e35, None, -1.508079088755887e35, 1.0] + target = pa.float64() + if kind.endswith('_float'): + target = pa.float32() + expected = [None if item is None else float(np.float32(item)) for item in expected] + return source, target, expected + + +@pytest.mark.parametrize('native', [False, True], ids=['python', 'native']) +@pytest.mark.parametrize('stream', [False, True], ids=['batch', 'stream']) +@pytest.mark.parametrize('operation', _OPERATIONS) +@pytest.mark.parametrize('shape', ['root', 'row', 'array', 'map']) +@pytest.mark.parametrize('kind', ['fixed_binary', 'empty_fixed_binary', 'half_float', + 'decimal_double', 'decimal_float', 'decimal256_double', 'decimal256_float']) +def test_scalar_input_encodings_keep_values(tmp_path, native, stream, operation, shape, kind): + source, target, expected = _wrap(*_scalar_case(kind), shape) + table = _table(tmp_path, target, native) + _apply(table, source, native, stream, operation) + _check(table, expected) + + +def _numeric_union(dense, kind='number'): + if dense: + left, right = [-1, None, 2], [3, 1, 2 ** 40] + else: + left = [-1, 0, 2, 0, None, 0] + right = [2 ** 40, 1, 2 ** 40, 3, 2 ** 40, 2 ** 40] + if kind == 'array': + left_type, right_type, target = pa.list_(pa.int32()), pa.large_list(pa.int64()), pa.list_(pa.int64()) + left = [None if value is None else [value] for value in left] + right = [[value] for value in right] + expected = [[1], [2], [3], None] + elif kind == 'row': + left_type, right_type, target = pa.struct([('n', pa.int32())]), pa.struct([('n', pa.int64())]), \ + pa.struct([('n', pa.int64())]) + left = [None if value is None else {'n': value} for value in left] + right = [{'n': value} for value in right] + expected = [{'n': 1}, {'n': 2}, {'n': 3}, None] + else: + left_type, right_type, target = pa.int32(), pa.int64(), pa.int32() + expected = [1, 2, 3, None] + children = [pa.array(left, type=left_type), pa.array(right, type=right_type), + pa.array(['incompatible inactive child'] * (1 if dense else 6))] + ids = pa.array([7, 42, 7, 42, 7, 42], type=pa.int8()) + kwargs = dict(field_names=['small', 'wide', 'unused'], type_codes=[7, 42, 99]) + union = (pa.UnionArray.from_dense(ids, pa.array([0, 1, 2, 0, 1, 2], type=pa.int32()), children, **kwargs) + if dense else pa.UnionArray.from_sparse(ids, children, **kwargs)) + return union.slice(1, 4), target, expected + + +@pytest.mark.parametrize('native', [False, True], ids=['python', 'native']) +@pytest.mark.parametrize('stream', [False, True], ids=['batch', 'stream']) +@pytest.mark.parametrize('operation', ['row_id', 'incremental', 'upsert']) +@pytest.mark.parametrize('shape', ['root', 'row', 'array', 'map']) +@pytest.mark.parametrize('dense', [False, True], ids=['sparse', 'dense']) +@pytest.mark.parametrize('kind', ['number', 'array', 'row']) +def test_union_inputs_select_only_active_values(tmp_path, native, stream, operation, shape, dense, kind): + source, target, expected = _wrap(*_numeric_union(dense, kind), shape) + table = _table(tmp_path, target, native) + _apply(table, source, native, stream, operation) + _check(table, expected) + + +@pytest.mark.parametrize('native', [False, True], ids=['python', 'native']) +@pytest.mark.parametrize('stream', [False, True], ids=['batch', 'stream']) +@pytest.mark.parametrize('operation', ['row_id', 'incremental', 'upsert']) +@pytest.mark.parametrize('shape', ['root', 'row', 'array', 'map']) +@pytest.mark.parametrize('dense', [False, True], ids=['sparse', 'dense']) +def test_dictionary_union_keeps_null_indices_and_active_nulls(tmp_path, native, stream, operation, shape, dense): + source, target, expected = _numeric_union(dense) + source = pa.DictionaryArray.from_arrays(pa.array([3, None, 0, 1], type=pa.int8()), source) + expected = [expected[3], None, expected[0], expected[1]] + source, target, expected = _wrap(source, target, expected, shape) + table = _table(tmp_path, target, native) + _apply(table, source, native, stream, operation) + _check(table, expected) + + +@pytest.mark.parametrize('native', [False, True], ids=['python', 'native']) +@pytest.mark.parametrize('stream', [False, True], ids=['batch', 'stream']) +@pytest.mark.parametrize('operation', ['row_id', 'incremental', 'upsert']) +@pytest.mark.parametrize('shape,dense', [('row', False), ('row', True), ('fixed_array', True)], + ids=['sparse-row', 'dense-row', 'dense-fixed-array']) +@pytest.mark.parametrize('encoded', [False, True], ids=['union', 'dictionary_union']) +def test_union_children_below_null_parents_are_not_converted( + tmp_path, native, stream, operation, shape, encoded, dense): + ids = pa.array([42, 7, 7, 42], type=pa.int8()) + if dense: + source = pa.UnionArray.from_dense( + ids, pa.array([0, 0, 1, 1], type=pa.int32()), + [pa.array([1, 2], type=pa.int32()), pa.array(['hidden', 'also hidden'])], type_codes=[7, 42]) + else: + source = pa.UnionArray.from_sparse( + ids, [pa.array([0, 1, 2, 0], type=pa.int32()), pa.array(['hidden'] * 4)], type_codes=[7, 42]) + if encoded: + source = pa.DictionaryArray.from_arrays(pa.array([0, 1, 2, 3], type=pa.int8()), source) + nulls = pa.array([True, False, False, True]) + if shape == 'row': + source = pa.StructArray.from_arrays([source], names=['n'], mask=nulls) + target, expected = pa.struct([('n', pa.int32())]), [None, {'n': 1}, {'n': 2}, None] + else: + source = pa.FixedSizeListArray.from_arrays(source, 1, mask=nulls) + target, expected = pa.list_(pa.int32()), [None, [1], [2], None] + table = _table(tmp_path, target, native) + _apply(table, source, native, stream, operation) + _check(table, expected) + + +@pytest.mark.parametrize('native', [False, True], ids=['python', 'native']) +@pytest.mark.parametrize('stream', [False, True], ids=['batch', 'stream']) +@pytest.mark.parametrize('operation', ['row_id', 'incremental', 'upsert']) +@pytest.mark.parametrize('dense', [False, True], ids=['sparse', 'dense']) +def test_map_union_values_do_not_fill_missing_row_fields(tmp_path, native, stream, operation, dense): + ids = pa.array([7] * 4, type=pa.int8()) + children = [pa.array([1, 2, 3, 4], type=pa.int32())] + values = (pa.UnionArray.from_dense(ids, pa.array(range(4), type=pa.int32()), children, type_codes=[7]) + if dense else pa.UnionArray.from_sparse(ids, children, type_codes=[7])) + source = pa.MapArray.from_arrays(pa.array(range(5), type=pa.int32()), pa.array(['n'] * 4), values) + target = pa.struct([('n', pa.int32()), ('v', pa.string())]) + table = _table(tmp_path, target, native) + _apply(table, source, native, stream, operation) + _check(table, [{'n': i, 'v': None} for i in range(1, 5)]) + + +@pytest.mark.parametrize('native', [False, True], ids=['python', 'native']) +@pytest.mark.parametrize('operation', ['row_id', 'incremental', 'upsert']) +def test_union_import_aligns_foreign_children_and_row_siblings(tmp_path, native, operation): + raw = bytearray(b'x' + np.array([1, 2, 3, 4], dtype=' Date: Sat, 17 Oct 2026 22:51:13 +0800 Subject: [PATCH 3/3] Cover sliced nested Arrow imports and wide decimal sources --- .../native_update_encoded_inputs_test.py | 53 ++++++++++++++++++- 1 file changed, 51 insertions(+), 2 deletions(-) diff --git a/paimon-python/pypaimon/tests/native_update_encoded_inputs_test.py b/paimon-python/pypaimon/tests/native_update_encoded_inputs_test.py index 78f87236b20f..6aa7434896b7 100644 --- a/paimon-python/pypaimon/tests/native_update_encoded_inputs_test.py +++ b/paimon-python/pypaimon/tests/native_update_encoded_inputs_test.py @@ -62,7 +62,8 @@ def _scalar_case(kind): coefficient = '1508079088755887087797858531882816082' value = Decimal(coefficient[:-1] + '.' + coefficient[-1]) factory = pa.decimal256 if kind.startswith('decimal256') else pa.decimal128 - source = pa.array([value, None, value.copy_negate(), Decimal('1.0')], type=factory(38, 1)) + precision = 76 if kind.startswith('decimal256_wide') else 38 + source = pa.array([value, None, value.copy_negate(), Decimal('1.0')], type=factory(precision, 1)) expected = [1.508079088755887e35, None, -1.508079088755887e35, 1.0] target = pa.float64() if kind.endswith('_float'): @@ -76,7 +77,8 @@ def _scalar_case(kind): @pytest.mark.parametrize('operation', _OPERATIONS) @pytest.mark.parametrize('shape', ['root', 'row', 'array', 'map']) @pytest.mark.parametrize('kind', ['fixed_binary', 'empty_fixed_binary', 'half_float', - 'decimal_double', 'decimal_float', 'decimal256_double', 'decimal256_float']) + 'decimal_double', 'decimal_float', 'decimal256_double', 'decimal256_float', + 'decimal256_wide_double', 'decimal256_wide_float']) def test_scalar_input_encodings_keep_values(tmp_path, native, stream, operation, shape, kind): source, target, expected = _wrap(*_scalar_case(kind), shape) table = _table(tmp_path, target, native) @@ -202,6 +204,53 @@ def test_union_import_aligns_foreign_children_and_row_siblings(tmp_path, native, _check(table, [{'n': i, 'sibling': i} for i in range(1, 5)]) +@pytest.mark.parametrize('native', [False, True], ids=['python', 'native']) +@pytest.mark.parametrize('stream', [False, True], ids=['batch', 'stream']) +@pytest.mark.parametrize('operation', ['row_id', 'incremental', 'upsert']) +@pytest.mark.parametrize('shape', ['nested_sibling', 'nested_child', 'masked_struct', 'masked_fixed_array']) +def test_sparse_union_import_normalizes_all_parent_coordinates(tmp_path, native, stream, operation, shape): + ids = pa.array([7] * 6, type=pa.int8()) + numbers = pa.array([None, 1, 2, None, 4, None], type=pa.int32()) + nested_type = pa.struct([('a', pa.struct([('b', pa.int32())]))]) + nested = pa.array([{'a': {'b': i}} for i in range(6)], type=nested_type) + union = pa.UnionArray.from_sparse(ids, [pa.array(range(6), type=pa.int32())], type_codes=[7]) + if shape == 'nested_child': + source = pa.UnionArray.from_sparse(ids, [nested], type_codes=[7]) + target = nested_type + expected = nested.slice(1, 4).to_pylist() + else: + if shape == 'nested_sibling': + ordinary, ordinary_type = nested, nested_type + elif shape == 'masked_struct': + ordinary = pa.StructArray.from_arrays( + [numbers], fields=[pa.field('n', pa.int32(), nullable=False)], mask=numbers.is_null()) + ordinary_type = pa.struct([('n', pa.int32())]) + else: + ordinary = pa.FixedSizeListArray.from_arrays( + numbers, type=pa.list_(pa.field('item', pa.int32(), nullable=False), 1), mask=numbers.is_null()) + ordinary_type = pa.list_(pa.int32()) + source = pa.StructArray.from_arrays([union, ordinary], names=['u', 'ordinary']) + target = pa.struct([('u', pa.int32()), ('ordinary', ordinary_type)]) + expected = [{'u': i, 'ordinary': item} for i, item in enumerate(ordinary.to_pylist()[1:5], 1)] + source = source.slice(1, 4) + source.validate(full=True) + table = _table(tmp_path, target, native) + _apply(table, source, native, stream, operation) + _check(table, expected) + + +@pytest.mark.parametrize('native', [False, True], ids=['python', 'native']) +@pytest.mark.parametrize('operation', _OPERATIONS) +@pytest.mark.parametrize('precision,scale', [(39, 1), (76, 39)]) +def test_wide_decimal_floating_input_does_not_require_decimal128(tmp_path, native, operation, precision, scale): + coefficient = '1508079088755887087797858531882816082' + '0' * (scale - 1) + value = Decimal(coefficient + 'e-' + str(scale)) + source = pa.array([value, None, value.copy_negate(), Decimal('1.0')], type=pa.decimal256(precision, scale)) + table = _table(tmp_path, pa.float64(), native) + _apply(table, source, native, False, operation) + _check(table, [1.508079088755887e35, None, -1.508079088755887e35, 1.0]) + + @pytest.mark.parametrize('native', [False, True], ids=['python', 'native']) @pytest.mark.parametrize('operation', ['predicate_array', 'predicate_callable', 'merge']) @pytest.mark.parametrize('shape', ['root', 'row', 'array', 'map'])