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 219fe04001 [python] Expand native row-level update input coverage
(#10636)
219fe04001 is described below
commit 219fe0400113bb5a7d52a3fa466e15536f6dc9ab
Author: Jingsong Lee <[email protected]>
AuthorDate: Sun Oct 18 10:36:41 2026 +0800
[python] Expand native row-level update input coverage (#10636)
---
.../tests/native_decimal_integer_input_test.py | 129 +++++++++++++++++++++
.../tests/native_update_binary_text_test.py | 105 +++++++++++++++++
.../pypaimon/tests/native_update_map_pairs_test.py | 108 +++++++++++++++++
.../tests/native_upsert_key_normalization_test.py | 28 +++--
4 files changed, 362 insertions(+), 8 deletions(-)
diff --git a/paimon-python/pypaimon/tests/native_decimal_integer_input_test.py
b/paimon-python/pypaimon/tests/native_decimal_integer_input_test.py
new file mode 100644
index 0000000000..2846462b1c
--- /dev/null
+++ b/paimon-python/pypaimon/tests/native_decimal_integer_input_test.py
@@ -0,0 +1,129 @@
+# 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.
+
+"""External DECIMAL inputs retain integer values across native update entry
points."""
+
+from decimal import Decimal
+
+import pyarrow as pa
+import pytest
+
+from pypaimon.tests.native_update_container_casts_test import _OPERATIONS,
_apply, _check, _table
+from pypaimon.tests.native_update_encoded_inputs_test import _wrap
+
+
+pytestmark = pytest.mark.native_plan
+
+
[email protected]('native', [False, True], ids=['python', 'native'])
[email protected]('stream', [False, True], ids=['batch', 'stream'])
[email protected]('operation', _OPERATIONS)
[email protected]('shape', ['root', 'row', 'array', 'map'])
[email protected]('target', [pa.int32(), pa.int64()])
[email protected]('factory,precision,scale', [
+ (pa.decimal128, 38, -128), (pa.decimal256, 76, -90),
+])
+def test_zero_decimal_updates_do_not_overflow_scale_powers(
+ tmp_path, native, stream, operation, shape, target, factory,
precision, scale):
+ # A zero can use any valid external scale without evaluating
10**abs(scale).
+ zero = Decimal(0)
+ source = pa.array([zero, None, zero, zero], type=factory(precision, scale))
+ source, target, expected = _wrap(source, target, [0, None, 0, 0], shape)
+ table = _table(tmp_path, target, native)
+ _apply(table, source, native, stream, operation)
+ _check(table, expected)
+
+
[email protected]('native', [False, True], ids=['python', 'native'])
[email protected]('stream', [False, True], ids=['batch', 'stream'])
[email protected]('operation', _OPERATIONS)
[email protected]('shape', ['root', 'row', 'array', 'map'])
[email protected]('encoded', [False, True], ids=['plain', 'dictionary'])
+def test_wide_decimal_integer_values_are_reduced_before_narrowing(
+ tmp_path, native, stream, operation, shape, encoded):
+ source = pa.array([Decimal('123.' + '0' * 60), None,
+ Decimal('-456.' + '0' * 60), Decimal('0.' + '0' * 60)],
+ type=pa.decimal256(76, 60))
+ if encoded:
+ source = source.dictionary_encode()
+ source, target, expected = _wrap(source, pa.int32(), [123, None, -456, 0],
shape)
+ table = _table(tmp_path, target, native)
+ _apply(table, source, native, stream, operation)
+ _check(table, expected)
+
+
[email protected]('native', [False, True], ids=['python', 'native'])
[email protected]('stream', [False, True], ids=['batch', 'stream'])
[email protected]('operation', ['row_id', 'incremental', 'upsert'])
[email protected]('shape', ['root', 'row', 'array', 'map'])
[email protected]('factory,precision', [(pa.decimal128, 38),
(pa.decimal256, 76)])
+def test_row_updates_truncate_decimal_fractions_toward_zero(
+ tmp_path, native, stream, operation, shape, factory, precision):
+ source = pa.array([Decimal('-128.99'), None, Decimal('127.99'),
Decimal('-0.99')],
+ type=factory(precision, 2))
+ source, target, expected = _wrap(source, pa.int8(), [-128, None, 127, 0],
shape)
+ table = _table(tmp_path, target, native)
+ _apply(table, source, native, stream, operation)
+ _check(table, expected)
+
+
[email protected]('native', [False, True], ids=['python', 'native'])
[email protected]('stream', [False, True], ids=['batch', 'stream'])
[email protected]('operation', ['predicate_array',
'predicate_callable', 'merge'])
[email protected]('shape', ['root', 'row', 'array', 'map'])
+def test_evaluated_assignments_reject_decimal_fraction_loss(tmp_path, native,
stream, operation, shape):
+ source = pa.array([Decimal('1.00'), None, Decimal('-2.01'),
Decimal('3.00')],
+ type=pa.decimal256(76, 2))
+ source, target, _ = _wrap(source, pa.int32(), [1, None, -2, 3], shape)
+ table = _table(tmp_path, target, native)
+ snapshot = table.snapshot_manager().get_latest_snapshot().id
+ with pytest.raises((ValueError, TypeError)):
+ _apply(table, source, native, stream, operation)
+ assert table.snapshot_manager().get_latest_snapshot().id == snapshot
+ _check(table, [None] * 4)
+
+
[email protected]('native', [False, True], ids=['python', 'native'])
[email protected]('stream', [False, True], ids=['batch', 'stream'])
[email protected]('operation', _OPERATIONS)
[email protected]('target,minimum,maximum', [
+ (pa.int8(), -(2 ** 7), 2 ** 7 - 1),
+ (pa.int16(), -(2 ** 15), 2 ** 15 - 1),
+ (pa.int32(), -(2 ** 31), 2 ** 31 - 1),
+ (pa.int64(), -(2 ** 63), 2 ** 63 - 1),
+])
+def test_decimal_integer_boundaries_are_written_without_wrapping(
+ tmp_path, native, stream, operation, target, minimum, maximum):
+ source = pa.array([Decimal(minimum), None, Decimal(maximum), Decimal(0)],
+ type=pa.decimal256(76, 2))
+ table = _table(tmp_path, target, native)
+ _apply(table, source, native, stream, operation)
+ _check(table, [minimum, None, maximum, 0])
+
+
[email protected]('native', [False, True], ids=['python', 'native'])
[email protected]('operation', ['row_id', 'predicate_array', 'upsert',
'merge'])
[email protected]('value', ['-129', '128', '1E+30'])
+def test_decimal_integer_overflow_keeps_snapshot_unchanged(tmp_path, native,
operation, value):
+ scale = -30 if value.endswith('30') else 0
+ source = pa.array([Decimal(0), None, Decimal(value), Decimal(0)],
type=pa.decimal128(38, scale))
+ table = _table(tmp_path, pa.int8(), native)
+ snapshot = table.snapshot_manager().get_latest_snapshot().id
+ with pytest.raises((ValueError, TypeError)):
+ _apply(table, source, native, False, operation)
+ assert table.snapshot_manager().get_latest_snapshot().id == snapshot
+ _check(table, [None] * 4)
diff --git a/paimon-python/pypaimon/tests/native_update_binary_text_test.py
b/paimon-python/pypaimon/tests/native_update_binary_text_test.py
new file mode 100644
index 0000000000..7a3ff3d520
--- /dev/null
+++ b/paimon-python/pypaimon/tests/native_update_binary_text_test.py
@@ -0,0 +1,105 @@
+# 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.
+
+"""NULL binary storage is not an input to text or numeric parsing."""
+
+import struct
+from decimal import Decimal
+
+import pyarrow as pa
+import pytest
+
+from pypaimon.tests.native_update_container_casts_test import _OPERATIONS,
_apply, _check, _table
+from pypaimon.tests.native_update_encoded_inputs_test import _wrap
+
+
+pytestmark = pytest.mark.native_plan
+
+
+def _binary_input(kind, large):
+ targets = {'string': pa.string(), 'integer': pa.int32(), 'float':
pa.float64(),
+ 'decimal': pa.decimal128(10, 2), 'boolean': pa.bool_()}
+ left, right = (b'true', b'false') if kind == 'boolean' else (b'42', b'-7')
+ parts = [b'unused', b'\xff\xff', left, b'\xc3', right, b'unused']
+ offsets = [0]
+ for part in parts:
+ offsets.append(offsets[-1] + len(part))
+ source = pa.Array.from_buffers(
+ pa.large_binary() if large else pa.binary(), len(parts),
+ [pa.py_buffer(b'\x35'),
+ pa.py_buffer(struct.pack('<' + str(len(offsets)) + ('q' if large else
'i'), *offsets)),
+ pa.py_buffer(b''.join(parts))],
+ ).slice(1, 4)
+ expected = [None, '42', None, '-7']
+ if kind == 'integer':
+ expected = [None, 42, None, -7]
+ elif kind == 'float':
+ expected = [None, 42.0, None, -7.0]
+ elif kind == 'decimal':
+ expected = [None, Decimal('42.00'), None, Decimal('-7.00')]
+ elif kind == 'boolean':
+ expected = [None, True, None, False]
+ return source, targets[kind], expected
+
+
[email protected]('native', [False, True], ids=['python', 'native'])
[email protected]('stream', [False, True], ids=['batch', 'stream'])
[email protected]('operation', _OPERATIONS)
[email protected]('shape', ['root', 'row', 'array', 'map'])
[email protected]('kind', ['string', 'integer', 'float', 'decimal',
'boolean'])
[email protected]('large', [False, True], ids=['binary',
'large_binary'])
+def test_null_binary_bytes_do_not_participate_in_casts(tmp_path, native,
stream, operation, shape, kind, large):
+ source, target, expected = _wrap(*_binary_input(kind, large), shape)
+ table = _table(tmp_path, target, native)
+ _apply(table, source, native, stream, operation)
+ _check(table, expected)
+
+
[email protected]('native', [False, True], ids=['python', 'native'])
[email protected]('operation', _OPERATIONS)
[email protected]('kind', ['string', 'integer', 'float', 'decimal',
'boolean'])
+def test_dictionary_binary_text_uses_logical_validity(tmp_path, native,
operation, kind):
+ source, target, expected = _binary_input(kind, True)
+ indices = [3, None, 0, 1]
+ source = pa.DictionaryArray.from_arrays(pa.array(indices, type=pa.int8()),
source)
+ table = _table(tmp_path, target, native)
+ _apply(table, source, native, False, operation)
+ _check(table, [None if i is None else expected[i] for i in indices])
+
+
[email protected]('native', [False, True], ids=['python', 'native'])
[email protected]('operation', _OPERATIONS)
+def test_fixed_binary_text_skips_null_storage(tmp_path, native, operation):
+ source = pa.Array.from_buffers(pa.binary(2), 6, [
+ pa.py_buffer(b'\x35'),
pa.py_buffer(b'00\xff\xff42\xc3\xff-700')]).slice(1, 4)
+ table = _table(tmp_path, pa.string(), native)
+ _apply(table, source, native, False, operation)
+ _check(table, [None, '42', None, '-7'])
+
+
[email protected]('native', [False, True], ids=['python', 'native'])
[email protected]('operation', _OPERATIONS)
[email protected]('kind', ['string', 'integer', 'float', 'decimal',
'boolean'])
+def test_visible_invalid_binary_text_cannot_change_rows(tmp_path, native,
operation, kind):
+ source, target, _ = _binary_input(kind, False)
+ source = pa.array([source[1].as_py(), None, b'\xff', source[3].as_py()])
+ table = _table(tmp_path, target, native)
+ snapshot = table.snapshot_manager().get_latest_snapshot().id
+ with pytest.raises((ValueError, TypeError)):
+ _apply(table, source, native, False, operation)
+ assert table.snapshot_manager().get_latest_snapshot().id == snapshot
+ _check(table, [None] * 4)
diff --git a/paimon-python/pypaimon/tests/native_update_map_pairs_test.py
b/paimon-python/pypaimon/tests/native_update_map_pairs_test.py
new file mode 100644
index 0000000000..a79b7cd1ac
--- /dev/null
+++ b/paimon-python/pypaimon/tests/native_update_map_pairs_test.py
@@ -0,0 +1,108 @@
+# 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.
+
+"""External MAP pairs can construct ARRAY elements in existing row updates."""
+
+from decimal import Decimal
+
+import pyarrow as pa
+import pytest
+
+from pypaimon.tests.native_update_container_casts_test import _apply, _check,
_table
+from pypaimon.tests.native_update_encoded_inputs_test import _wrap
+
+
+pytestmark = pytest.mark.native_plan
+
+
+def _pairs(kind):
+ keys = [99, 1, 3, 99, 99, 5, 99]
+ values = [99, 2, None, 99, 99, -6, 99]
+ target = pa.int32()
+ key_type, value_type = pa.int64(), pa.int64()
+ if kind == 'float':
+ values = [1e100, 2.75, None, 1e100, 1e100, -6.75, 1e100]
+ value_type = pa.float64()
+ elif kind == 'decimal':
+ values = [Decimal(str(v)) if v is not None else None for v in values]
+ values[1], values[5] = Decimal('2.75'), Decimal('-6.75')
+ value_type = pa.decimal128(10, 2)
+ elif kind in ('string', 'binary'):
+ keys = ['padding', '一', '', 'hidden', 'hidden', 'z', 'padding']
+ values = ['padding', '二', None, 'hidden', 'hidden', 'a\0b', 'padding']
+ key_type, value_type, target = pa.string(), pa.large_string(),
pa.string()
+ if kind == 'binary':
+ keys = [v.encode() for v in keys]
+ values = [None if v is None else v.encode() for v in values]
+ key_type, value_type, target = pa.binary(), pa.large_binary(),
pa.binary()
+ elif kind == 'array':
+ keys = [[v] for v in keys]
+ values = [None if v is None else [v, None] for v in values]
+ key_type, value_type, target = pa.list_(pa.int32()),
pa.large_list(pa.int64()), pa.list_(pa.int32())
+ source = pa.MapArray.from_arrays(
+ pa.array([0, 1, 3, 5, 5, 6, 7], type=pa.int32()),
+ pa.array(keys, type=key_type), pa.array(values, type=value_type),
+ mask=pa.array([False, False, True, False, False, False]),
+ ).slice(1, 4)
+ expected = [[[keys[1], values[1]], [keys[2], values[2]]], None, [],
[[keys[5], values[5]]]]
+ if kind in ('float', 'decimal'):
+ expected[0][0][1], expected[3][0][1] = 2, -6
+ return source, pa.list_(pa.list_(target)), expected
+
+
[email protected]('native', [False, True], ids=['python', 'native'])
[email protected]('stream', [False, True], ids=['batch', 'stream'])
[email protected]('operation', ['row_id', 'incremental', 'upsert'])
[email protected]('shape', ['root', 'row', 'array', 'map'])
[email protected]('kind', ['integer', 'float', 'decimal', 'string',
'binary', 'array'])
[email protected]('encoded', [False, True], ids=['map',
'dictionary_map'])
+def test_map_pairs_construct_nested_arrays(tmp_path, native, stream,
operation, shape, kind, encoded):
+ source, target, expected = _pairs(kind)
+ if encoded:
+ source = pa.DictionaryArray.from_arrays(pa.array([0, 1, 2, 3],
type=pa.int8()), source)
+ source, target, expected = _wrap(source, target, expected, shape)
+ table = _table(tmp_path, target, native)
+ _apply(table, source, native, stream, operation)
+ _check(table, expected)
+
+
[email protected]('native', [False, True], ids=['python', 'native'])
[email protected]('operation', ['predicate_array',
'predicate_callable', 'merge'])
[email protected]('shape', ['root', 'row', 'array', 'map'])
+def test_map_pairs_keep_safe_assignment_boundary(tmp_path, native, operation,
shape):
+ source, target, expected = _wrap(*_pairs('integer'), shape)
+ table = _table(tmp_path, target, native)
+ snapshot = table.snapshot_manager().get_latest_snapshot().id
+ with pytest.raises((ValueError, TypeError, NotImplementedError)):
+ _apply(table, source, native, False, operation)
+ assert table.snapshot_manager().get_latest_snapshot().id == snapshot
+ _check(table, [None] * len(expected))
+
+
[email protected]('native', [False, True], ids=['python', 'native'])
[email protected]('operation', ['row_id', 'incremental', 'upsert'])
[email protected]('side', ['key', 'value'])
+def test_map_pair_integer_overflow_cannot_change_rows(tmp_path, native,
operation, side):
+ keys, values = [1, 2, 3, 4], [5, 6, 7, 8]
+ (keys if side == 'key' else values)[2] = 2 ** 40
+ source = pa.MapArray.from_arrays(pa.array(range(5), type=pa.int32()),
pa.array(keys), pa.array(values))
+ table = _table(tmp_path, pa.list_(pa.list_(pa.int32())), native)
+ snapshot = 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 == snapshot
+ _check(table, [None] * 4)
diff --git
a/paimon-python/pypaimon/tests/native_upsert_key_normalization_test.py
b/paimon-python/pypaimon/tests/native_upsert_key_normalization_test.py
index 50bcde062e..aa9c186e85 100644
--- a/paimon-python/pypaimon/tests/native_upsert_key_normalization_test.py
+++ b/paimon-python/pypaimon/tests/native_upsert_key_normalization_test.py
@@ -384,16 +384,28 @@ def
test_large_binary_key_slice_after_two_gib_matches_visible_bytes(tmp_path, na
@pytest.mark.skipif(os.name != 'posix' or sys.maxsize <= 2 ** 32,
reason='Requires 64-bit POSIX mmap')
@pytest.mark.parametrize('native', [False, True], ids=['python', 'native'])
@pytest.mark.parametrize('mixed', [False, True], ids=['all_null', 'mixed'])
-def test_binary_match_keys_ignore_storage_owned_by_null_values(tmp_path,
native, mixed):
- table, _ = _table(tmp_path, native, pa.binary(), b'ab', b'other')
[email protected]('kind', ['binary', 'string'])
[email protected]('encoded', [False, True], ids=['plain', 'dictionary'])
[email protected]('payload', [b'ab', '中文'.encode(), b''], ids=['ascii',
'utf8', 'empty'])
[email protected]('sliced', [False, True], ids=['zero_offset',
'sliced'])
+def test_byte_match_keys_ignore_storage_owned_by_null_values(tmp_path, native,
mixed, kind, encoded, payload, sliced):
+ text = kind == 'string'
+ key_type, input_type = (pa.string(), pa.large_string()) if text else
(pa.binary(), pa.large_binary())
+ table, _ = _table(tmp_path, native, key_type, payload.decode() if text
else payload, 'other' if text else b'other')
before = next(_rows(table))
start = 2 ** 31 + 17
- memory = mmap.mmap(-1, start + 2)
- memory[start:start + 2] = b'ab'
- offsets = [0, start, start + 2] if mixed else [0, start]
- keys = pa.Array.from_buffers(pa.large_binary(), len(offsets) - 1, [
- pa.py_buffer(b'\x02'), pa.array(offsets,
type=pa.int64()).buffers()[1], pa.py_buffer(memory)],
- null_count=1)
+ memory = mmap.mmap(-1, start + len(payload))
+ memory[start:start + len(payload)] = payload
+ offsets = [0, start, start + len(payload)] if mixed else [0, start]
+ length = len(offsets) - 1
+ if sliced:
+ offsets.insert(0, 0)
+ keys = pa.Array.from_buffers(input_type, length, [
+ pa.py_buffer(b'\x05' if sliced else b'\x02'),
+ pa.array(offsets, type=pa.int64()).buffers()[1], pa.py_buffer(memory)],
+ null_count=1, offset=int(sliced))
+ if encoded:
+ keys = pa.DictionaryArray.from_arrays(pa.array(range(len(keys)),
type=pa.int8()), keys)
source = pa.table({'key': keys, 'value': pa.array([9, 42] if mixed else
[9], type=pa.int32())})
_upsert(table, source, native, False)
for rows in _rows(table):
