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 e711c1c277 [python] Extend Native upsert keys and preserve decimal
index predicates (#10635)
e711c1c277 is described below
commit e711c1c277cca259df4be7e64afb1ce9991192ea
Author: Jingsong Lee <[email protected]>
AuthorDate: Sun Oct 18 08:47:04 2026 +0800
[python] Extend Native upsert keys and preserve decimal index predicates
(#10635)
---
.../pypaimon/globalindex/global_index_evaluator.py | 26 ++-
.../pypaimon/tests/global_index_evaluator_test.py | 32 ++++
.../tests/native_decimal_predicate_test.py | 203 +++++++++++++++++++++
.../tests/native_upsert_key_normalization_test.py | 176 +++++++++++++++++-
paimon-python/pypaimon/write/native_update.py | 14 +-
5 files changed, 445 insertions(+), 6 deletions(-)
diff --git a/paimon-python/pypaimon/globalindex/global_index_evaluator.py
b/paimon-python/pypaimon/globalindex/global_index_evaluator.py
index fe786e604b..7bb5e4eb65 100644
--- a/paimon-python/pypaimon/globalindex/global_index_evaluator.py
+++ b/paimon-python/pypaimon/globalindex/global_index_evaluator.py
@@ -18,14 +18,16 @@
"""Global index evaluator for filtering data using global indexes."""
import threading
+import struct
from collections import deque
from concurrent.futures import Future
from typing import Callable, Collection, Dict, FrozenSet, List, NamedTuple,
Optional
-from pypaimon.globalindex.global_index_reader import GlobalIndexReader,
FieldRef
+from pypaimon.globalindex.global_index_reader import GlobalIndexReader,
FieldRef, _completed_future
from pypaimon.globalindex.global_index_result import GlobalIndexResult
+from pypaimon.globalindex.key_serializer import create_serializer
from pypaimon.common.predicate import Predicate
-from pypaimon.schema.data_types import DataField
+from pypaimon.schema.data_types import AtomicType, DataField
class GlobalIndexEvaluation(NamedTuple):
@@ -82,6 +84,9 @@ class GlobalIndexEvaluator:
f.set_result(None)
return f
+ if not self._decimal_literals_supported(field, predicate):
+ return _completed_future(None)
+
field_id = field.id
readers = self._index_readers_cache.get(field_id)
if readers is None:
@@ -121,6 +126,23 @@ class GlobalIndexEvaluator:
return all_done
+ @staticmethod
+ def _decimal_literals_supported(field: DataField, predicate: Predicate) ->
bool:
+ if not (isinstance(field.type, AtomicType)
+ and field.type.type.upper().startswith(('DECIMAL',
'NUMERIC'))):
+ return True
+ # Like Native, decline a physical index lookup if conversion would
+ # round a comparison literal. The original predicate must then filter
+ # the data; a rounded bitmap lookup can silently discard matching rows.
+ serializer = create_serializer(field.type)
+ compare = serializer.create_comparator()
+ try:
+ return all(literal is None or compare(
+ serializer.deserialize(serializer.serialize(literal)),
literal) == 0
+ for literal in (predicate.literals or []))
+ except (ArithmeticError, ValueError, struct.error):
+ return False
+
def _combine_reader_results(
self, reader_futures: List[Future], field_id: int,
) -> Optional[GlobalIndexEvaluation]:
diff --git a/paimon-python/pypaimon/tests/global_index_evaluator_test.py
b/paimon-python/pypaimon/tests/global_index_evaluator_test.py
index d979275ec0..8c3daf6bd8 100644
--- a/paimon-python/pypaimon/tests/global_index_evaluator_test.py
+++ b/paimon-python/pypaimon/tests/global_index_evaluator_test.py
@@ -18,6 +18,7 @@
import threading
import time
import unittest
+from decimal import Decimal, localcontext
from concurrent.futures import ThreadPoolExecutor
from pypaimon.common.predicate import Predicate
@@ -69,6 +70,37 @@ def _result_of(*row_ids):
class GlobalIndexEvaluatorTest(unittest.TestCase):
+ def test_decimal_literals_cannot_round_to_an_index_key(self):
+ for precision in [10, 38]:
+ field = DataField(0, 'amount', AtomicType('DECIMAL(%d,2)' %
precision))
+ evaluator = GlobalIndexEvaluator([field], lambda field:
self.fail('Unsafe index lookup'))
+ try:
+ for method, values in [('equal', ['1.235']), ('notEqual',
['-1.235']),
+ ('in', ['1.23', '1.235']), ('notIn',
['1.235']),
+ ('between', ['1.23', '1.235'])]:
+ literals = [Decimal(value) for value in values]
+ predicate = Predicate(method, 0, 'amount', literals)
+ with localcontext() as context:
+ context.prec = 3
+ self.assertIsNone(evaluator.evaluate(predicate))
+ self.assertEqual(predicate.literals, literals)
+ finally:
+ evaluator.close()
+
+ def test_exact_decimal_literals_still_use_the_index(self):
+ field = DataField(0, 'amount', AtomicType('DECIMAL(10,2)'))
+ expected = _result_of(1, 3)
+ evaluator = GlobalIndexEvaluator([field], lambda field:
[StubGlobalIndexReader(expected)])
+ try:
+ for value in ['1.2300', '-1.23', '0E+100']:
+ predicate = Predicate('equal', 0, 'amount', [Decimal(value)])
+ with localcontext() as context:
+ context.prec = 3
+ self.assertIs(evaluator.evaluate(predicate), expected)
+ self.assertIsNone(evaluator.evaluate(Predicate('equal', 0,
'amount', [Decimal('1E+37')])))
+ finally:
+ evaluator.close()
+
def test_single_field_sequential(self):
fields = _make_fields()
expected = GlobalIndexResult.from_range(Range(1, 3))
diff --git a/paimon-python/pypaimon/tests/native_decimal_predicate_test.py
b/paimon-python/pypaimon/tests/native_decimal_predicate_test.py
new file mode 100644
index 0000000000..691fa783ce
--- /dev/null
+++ b/paimon-python/pypaimon/tests/native_decimal_predicate_test.py
@@ -0,0 +1,203 @@
+# 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.
+
+"""DECIMAL predicates compare values without rounding literals to column
scale."""
+
+from contextlib import ExitStack
+from decimal import Decimal, localcontext
+from unittest.mock import patch
+
+import pyarrow as pa
+import pytest
+
+from pypaimon import CatalogFactory, Schema
+from pypaimon.common.predicate import Predicate
+from pypaimon.globalindex.create_global_index import GlobalIndexBuilder
+from pypaimon.write.table_update import TableUpdate
+
+
+pytestmark = pytest.mark.native_plan
+
+_VALUES = [Decimal(value) for value in ['-1.24', '-1.23', '0.00', '1.23',
'1.24']] + [None]
+_PREDICATES = [
+ ('equal', ['1.235'], []),
+ ('notEqual', ['1.235'], [0, 1, 2, 3, 4]),
+ ('lessThan', ['1.235'], [0, 1, 2, 3]),
+ ('lessOrEqual', ['1.235'], [0, 1, 2, 3]),
+ ('greaterThan', ['1.235'], [4]),
+ ('greaterOrEqual', ['1.235'], [4]),
+ ('in', ['1.235', '1.240', None], [4]),
+ ('notIn', ['1.235', '1.240'], [0, 1, 2, 3]),
+ ('notIn', ['1.235', None], []),
+ ('between', ['-1.235', '1.235'], [1, 2, 3]),
+ ('notBetween', ['-1.235', '1.235'], [0, 4]),
+ ('lessThan', ['123456789'], [0, 1, 2, 3, 4]),
+ ('equal', ['1E+37'], []),
+ ('greaterThan', ['1E-38'], [3, 4]),
+ ('lessThan', ['0E+100'], [0, 1]),
+ ('lessThan', ['-1.23500000000000000000000000000000000000000'], [0]),
+]
+
+
+def _predicate(method, literals):
+ return Predicate(method, 1, 'amount', [Decimal(value) if value is not None
else None for value in literals])
+
+
+def _table(tmp_path, native=True, evolution=False, indexed=None, precision=10):
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path)})
+ catalog.create_database('default', True)
+ schema = pa.schema([('id', pa.int32()), ('amount',
pa.decimal128(precision, 2)), ('value', pa.int32())])
+ evolution = evolution or indexed is not None
+ options = {'write.native.enabled': str(native).lower(),
+ 'row-tracking.enabled': str(evolution).lower(),
+ 'deletion-vectors.enabled': str(evolution).lower(),
+ 'data-evolution.enabled': str(evolution).lower()}
+ if indexed is not None:
+ options.update({'global-index.enabled': 'true',
'global-index.search-mode': 'full'})
+ catalog.create_table('default.t', Schema.from_pyarrow_schema(schema,
options=options), False)
+ table = catalog.get_table('default.t')
+ builder = table.new_batch_write_builder()
+ writer, commit = builder.new_write(), builder.new_commit()
+ try:
+ # Separate files make incorrect min/max pruning observable too.
+ for start in range(0, len(_VALUES), 2):
+ writer.write_arrow(pa.table({'id': [start, start + 1], 'amount':
_VALUES[start:start + 2],
+ 'value': [0, 0]}, schema=schema))
+ commit.commit(writer.prepare_commit())
+ writer.close()
+ commit.close()
+ if start + 2 < len(_VALUES):
+ writer, commit = builder.new_write(), builder.new_commit()
+ finally:
+ writer.close()
+ commit.close()
+ if indexed is not None:
+ messages = GlobalIndexBuilder(table, 'amount',
index_type=indexed).build()
+ assert messages
+ commit = table.new_batch_write_builder().new_commit()
+ try:
+ commit.commit(messages)
+ finally:
+ commit.close()
+ return table
+
+
+def _read(table, predicate, native_plan, native_read, projection=('id',)):
+ table = table.copy({'scan.native-plan.enabled': str(native_plan).lower(),
+ 'read.native.enabled': str(native_read).lower()})
+ builder = table.new_read_builder().with_projection(list(projection))
+ if predicate is not None:
+ builder.with_filter(predicate)
+ plan = builder.new_scan().plan()
+ if native_plan:
+ assert plan.stats().native_planned
+ reader = builder.new_read()
+ with ExitStack() as stack:
+ if native_read:
+ stack.enter_context(patch.object(reader, '_native_fallback',
+
side_effect=AssertionError('Python read fallback')))
+ stack.enter_context(patch.object(reader, '_create_split_read',
+
side_effect=AssertionError('Python split reader')))
+ rows = reader.to_arrow(plan.splits(), parallelism=1).to_pylist()
+ return sorted(rows, key=lambda row: row['id'])
+
+
[email protected]('native_plan,native_read', [(False, False), (True,
False), (False, True), (True, True)])
[email protected]('indexed', [None, 'btree', 'bitmap'])
[email protected]('precision', [10, 38], ids=['compact', 'wide'])
+def test_decimal_predicates_preserve_literal_value_in_scan_and_read(
+ tmp_path, native_plan, native_read, indexed, precision):
+ table = _table(tmp_path, indexed=indexed, precision=precision)
+ # Literal construction and conversion must not depend on Decimal context.
+ with localcontext() as context:
+ context.prec = 3
+ for method, literals, expected in _PREDICATES:
+ assert _read(table, _predicate(method, literals), native_plan,
native_read) == [
+ {'id': i} for i in expected], (method, literals)
+ fine = _predicate('lessThan', ['1.235'])
+ exact = Predicate('equal', 1, 'amount', [Decimal('1.24')])
+ for method, expected in [('and', []), ('or', list(range(5)))]:
+ combined = Predicate(method, None, None, [fine, exact])
+ assert _read(table, combined, native_plan, native_read) == [{'id':
i} for i in expected]
+
+
[email protected]('native', [False, True], ids=['python', 'native'])
[email protected]('stream', [False, True], ids=['batch', 'stream'])
[email protected]('operation', ['literal', 'callable', 'delete'])
+def
test_decimal_predicates_select_the_same_rows_for_updates_and_deletes(tmp_path,
native, stream, operation):
+ table = _table(tmp_path, native=native, evolution=True)
+ builder = table.new_stream_write_builder() if stream else
table.new_batch_write_builder()
+ update = builder.new_update()
+ predicate = _predicate('between', ['-1.235', '1.235'])
+ with ExitStack() as stack:
+ if native:
+ for method in ['_build_predicate_update_table',
'_matched_delete_row_ids']:
+ stack.enter_context(patch.object(TableUpdate, method,
+
side_effect=AssertionError('Python matching fallback')))
+ if operation == 'delete':
+ messages = update.delete_by_predicate(predicate, 19) if stream
else update.delete_by_predicate(predicate)
+ else:
+ if operation == 'callable':
+ def assignment(rows):
+ assert rows.column_names == ['id', '_ROW_ID']
+ return pa.array([i + 10 for i in rows['id'].to_pylist()],
type=pa.int32())
+ kwargs = {'read_columns': ['id']}
+ else:
+ assignment, kwargs = 42, {}
+ messages = (update.update_by_predicate(predicate, {'value':
assignment}, 19, **kwargs) if stream
+ else update.update_by_predicate(predicate, {'value':
assignment}, **kwargs))
+ commit = builder.new_commit()
+ try:
+ commit.commit(messages, 19) if stream else commit.commit(messages)
+ finally:
+ commit.close()
+ ids = [0, 4, 5] if operation == 'delete' else list(range(6))
+ expected = [{'id': i, 'amount': _VALUES[i], 'value': (
+ (i + 10 if operation == 'callable' else 42) if i in (1, 2, 3) else 0)}
for i in ids]
+ for read_native in [False, True]:
+ assert _read(table, None, read_native, read_native, ('id', 'amount',
'value')) == expected
+
+
[email protected]('engine', ['deduplicate', 'partial-update'])
[email protected]('native_plan,native_read', [(False, False), (True,
False), (False, True), (True, True)])
+def test_decimal_filter_observes_merged_primary_key_values(tmp_path, engine,
native_plan, native_read):
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path)})
+ catalog.create_database('default', True)
+ schema = pa.schema([('id', pa.int32()), ('amount', pa.decimal128(10, 2)),
('value', pa.int32())])
+ catalog.create_table('default.t', Schema.from_pyarrow_schema(schema,
primary_keys=['id'], options={
+ 'bucket': '1', 'merge-engine': engine, 'write.native.enabled': 'true',
+ }), False)
+ table = catalog.get_table('default.t')
+ batches = [
+ pa.table({'id': range(6), 'amount': _VALUES, 'value': [0] * 6},
schema=schema),
+ # Filtering either file before merging can resurrect id=3's obsolete
+ # value or discard id=4's new value on the other side of 1.235.
+ pa.table({'id': [3, 4], 'amount': [Decimal('1.24'), Decimal('1.23')],
'value': [42, 43]}, schema=schema),
+ ]
+ for batch in batches:
+ builder = table.new_batch_write_builder()
+ writer, commit = builder.new_write(), builder.new_commit()
+ try:
+ writer.write_arrow(batch)
+ commit.commit(writer.prepare_commit())
+ finally:
+ writer.close()
+ commit.close()
+ for method, literals, expected in _PREDICATES:
+ expected = sorted(4 if i == 3 else 3 if i == 4 else i for i in
expected)
+ rows = _read(table, _predicate(method, literals), native_plan,
native_read, ('id', 'value'))
+ assert rows == [{'id': i, 'value': 42 if i == 3 else 43 if i == 4 else
0} for i in expected]
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 004c653269..50bcde062e 100644
--- a/paimon-python/pypaimon/tests/native_upsert_key_normalization_test.py
+++ b/paimon-python/pypaimon/tests/native_upsert_key_normalization_test.py
@@ -18,6 +18,10 @@
"""Logical upsert keys must not depend on Arrow widths or dictionaries."""
import datetime
+import mmap
+import os
+import struct
+import sys
from contextlib import ExitStack
from decimal import Decimal
from unittest.mock import patch
@@ -37,6 +41,12 @@ _KEYS = [
(pa.int16(), pa.uint32(), 1, 2),
(pa.int32(), pa.int64(), 1, 2),
(pa.int64(), pa.uint64(), 1, 2),
+ (pa.float32(), pa.float64(), 1.5, -2.25),
+ (pa.float64(), pa.float32(), 1.5, -2.25),
+ (pa.float64(), pa.dictionary(pa.int8(), pa.float32()), 1.5, -2.25),
+ (pa.binary(), pa.large_binary(), b'\xff\0', b'other'),
+ (pa.large_binary(), pa.binary(), b'\xff\0', b'other'),
+ (pa.binary(), pa.dictionary(pa.int8(), pa.large_binary()), b'\xff\0',
b'other'),
(pa.decimal128(10, 2), pa.decimal128(15, 4), Decimal('1.25'),
Decimal('-2.50')),
(pa.date32(), pa.date64(), datetime.date(1969, 12, 31),
datetime.date(1970, 1, 2)),
(pa.time32('ms'), pa.time64('us'), datetime.time(1, 2, 3, 400000),
datetime.time(4, 5, 6)),
@@ -52,6 +62,32 @@ _KEYS = [
(pa.binary(), pa.dictionary(pa.int8(), pa.binary()), b'key', b'other'),
]
+_NUMERIC_KEYS = [
+ (pa.int32(), pa.float64(), 1, 1.0, 2),
+ (pa.int64(), pa.float64(), 2 ** 53 + 2, float(2 ** 53 + 2), 2 ** 53 + 4),
+ (pa.float32(), pa.int64(), 1.0, 1, 2.0),
+ (pa.float64(), pa.int64(), float(2 ** 53 + 2), 2 ** 53 + 2, float(2 ** 53
+ 4)),
+ (pa.decimal128(10, 2), pa.int64(), Decimal('1.00'), 1, Decimal('2.00')),
+ (pa.int32(), pa.decimal128(15, 4), 1, Decimal('1.0000'), 2),
+ (pa.decimal128(10, 2), pa.float64(), Decimal('1.50'), 1.5,
Decimal('2.25')),
+ (pa.float64(), pa.decimal128(10, 2), 1.5, Decimal('1.50'), 2.25),
+ (pa.decimal128(38, 38), pa.float64(), Decimal('0'), 0.0, Decimal('1E-38')),
+ (pa.float32(), pa.decimal256(40, 4), 1.5, Decimal('1.5000'), 2.25),
+ (pa.int64(), pa.dictionary(pa.int8(), pa.float64()), 1, 1.0, 2),
+ (pa.float64(), pa.dictionary(pa.int8(), pa.decimal128(10, 2)), 1.5,
Decimal('1.50'), 2.25),
+ (pa.float64(), pa.decimal256(76, 55), 0.1, Decimal.from_float(0.1), 0.25),
+ (pa.float64(), pa.decimal256(76, 70), 2.0 ** -70, Decimal.from_float(2.0
** -70), 2.0 ** -69),
+ (pa.float64(), pa.decimal256(76, -20), 1e20, Decimal('1E+20'), 2e20),
+ (pa.float32(), pa.uint32(), float(2 ** 24 + 2), 2 ** 24 + 2, float(2 ** 24
+ 4)),
+ (pa.float64(), pa.uint64(), float(2 ** 64 - 2048), 2 ** 64 - 2048, 1.0),
+ (pa.decimal128(38, 38), pa.int64(), Decimal('0'), 0, Decimal('1E-38')),
+]
+
+for _width in (32, 64):
+ _factory = getattr(pa, 'decimal{}'.format(_width), None)
+ if _factory is not None:
+ _NUMERIC_KEYS.append((pa.float64(), _factory(9, 2), 1.5,
Decimal('1.50'), 2.25))
+
def _array(values, data_type):
if pa.types.is_dictionary(data_type):
@@ -128,9 +164,98 @@ def
test_upsert_normalizes_match_keys_without_changing_row_identity(
@pytest.mark.parametrize('native', [False, True], ids=['python', 'native'])
@pytest.mark.parametrize('stream', [False, True], ids=['batch', 'stream'])
-def
test_partition_key_dictionaries_and_integer_widths_share_one_identity(tmp_path,
native, stream):
- table, _ = _table(tmp_path, native, pa.int32(), 1, 2, partitioned=True)
- source = pa.table({'key': pa.array([1, 2, 1], type=pa.int64()),
[email protected]('key_type,input_type,key,input_key,other',
_NUMERIC_KEYS)
+def test_numeric_keys_match_by_exact_value_across_arrow_types(
+ tmp_path, native, stream, key_type, input_type, key, input_key, other):
+ table, _ = _table(tmp_path, native, key_type, key, other)
+ before = next(_rows(table))
+ chunks = [pa.table({'key': _array([value], input_type), 'value':
pa.array([number], type=pa.int32())})
+ for value, number in [(input_key, 1), (None, 9), (input_key,
42)]]
+ _upsert(table, pa.concat_tables(chunks), native, stream)
+ for rows in _rows(table):
+ assert [row['value'] for row in rows] == [42, 42, 0, 9]
+ assert [{k: v for k, v in row.items() if k != 'value'} for row in
rows] == [
+ {k: v for k, v in row.items() if k != 'value'} for row in before]
+
+
[email protected]('native', [False, True], ids=['python', 'native'])
[email protected]('key_type,input_type,key,invalid,other', [
+ (pa.int32(), pa.float64(), 1, 1.00000001, 2),
+ (pa.int64(), pa.float64(), 2 ** 63 - 1, float(2 ** 63), 2),
+ (pa.float64(), pa.int64(), float(2 ** 53), 2 ** 53 + 1, 2.0),
+ (pa.decimal128(10, 2), pa.float64(), Decimal('0.10'), 0.1,
Decimal('0.20')),
+ (pa.float64(), pa.decimal128(10, 2), 0.1, Decimal('0.10'), 0.2),
+])
+def test_numeric_key_conversion_cannot_round_into_an_existing_row(
+ tmp_path, native, key_type, input_type, key, invalid, other):
+ table, _ = _table(tmp_path, native, key_type, key, other)
+ before = next(_rows(table))
+ source = pa.table({'key': _array([invalid], input_type), 'value':
pa.array([42], type=pa.int32())})
+ with pytest.raises((ValueError, TypeError, NotImplementedError)):
+ _upsert(table, source, native, False)
+ for rows in _rows(table):
+ assert rows == before
+
+
[email protected]('native', [False, True], ids=['python', 'native'])
[email protected]('key_type,input_type,key,input_key,other',
_NUMERIC_KEYS[:8])
+def test_mixed_numeric_upsert_validates_original_unmatched_rows_before_updates(
+ tmp_path, native, key_type, input_type, key, input_key, other):
+ table, _ = _table(tmp_path, native, key_type, key, other)
+ before = next(_rows(table))
+ source = pa.table({'key': _array([input_key, input_key + 4], input_type),
+ 'value': pa.array([42, 99], type=pa.int32())})
+ with pytest.raises((ValueError, TypeError, NotImplementedError)):
+ _upsert(table, source, native, False)
+ for rows in _rows(table):
+ assert rows == before
+
+
[email protected]('native', [False, True], ids=['python', 'native'])
[email protected]('stream', [False, True], ids=['batch', 'stream'])
[email protected]('key_type,input_type,key,input_key,other',
_NUMERIC_KEYS[:3] + _NUMERIC_KEYS[4:8])
+def test_full_row_numeric_upsert_preserves_row_identity_and_payload_casts(
+ tmp_path, native, stream, key_type, input_type, key, input_key, other):
+ table, _ = _table(tmp_path, native, key_type, key, other)
+ before = next(_rows(table))
+ source = pa.table({'key': _array([input_key], input_type), 'pt':
['updated'],
+ 'value': pa.array([42], type=pa.int32()), 'keep':
pa.array([99], type=pa.int32())})
+ _upsert(table, source, native, stream, columns=None)
+ for rows in _rows(table):
+ by_id = {row['_ROW_ID']: row for row in rows}
+ for original in before:
+ expected = dict(original)
+ if original['key'] == key:
+ expected.update(pt='updated', value=42, keep=99)
+ assert by_id[original['_ROW_ID']] == expected
+
+
[email protected]('native', [False, True], ids=['python', 'native'])
+def test_exact_match_does_not_relax_updated_key_assignment_range(tmp_path,
native):
+ key = float(2 ** 53 + 2)
+ table, _ = _table(tmp_path, native, pa.float64(), key, float(2 ** 53 + 4))
+ before = next(_rows(table))
+ source = pa.table({'key': pa.array([2 ** 53 + 2], type=pa.int64()), 'pt':
['updated'],
+ 'value': pa.array([42], type=pa.int32()), 'keep':
pa.array([99], type=pa.int32())})
+ # Matching only needs an exact numeric identity. Updating the key payload
+ # still uses the ordinary integer-to-double assignment range (<= 2^53).
+ with pytest.raises((ValueError, TypeError, NotImplementedError)):
+ _upsert(table, source, native, False, columns=None)
+ for rows in _rows(table):
+ assert rows == before
+
+
[email protected]('native', [False, True], ids=['python', 'native'])
[email protected]('stream', [False, True], ids=['batch', 'stream'])
[email protected]('key_type,input_type,key,other', [
+ (pa.int32(), pa.int64(), 1, 2),
+ (pa.float64(), pa.float32(), 1.5, -2.25),
+ (pa.binary(), pa.large_binary(), b'key', b'other'),
+])
+def test_partition_dictionaries_and_equivalent_key_layouts_share_one_identity(
+ tmp_path, native, stream, key_type, input_type, key, other):
+ table, _ = _table(tmp_path, native, key_type, key, other, partitioned=True)
+ source = pa.table({'key': _array([key, other, key], input_type),
'pt': _array(['a', 'b', 'a'], pa.dictionary(pa.int8(),
pa.string())),
'value': pa.array([1, 7, 42], type=pa.int32())})
before = next(_rows(table))
@@ -143,6 +268,10 @@ def
test_partition_key_dictionaries_and_integer_widths_share_one_identity(tmp_pa
@pytest.mark.parametrize('native', [False, True], ids=['python', 'native'])
@pytest.mark.parametrize('key_type,input_type,key,other,new_key', [
(pa.int32(), pa.int64(), 1, 2, 3),
+ (pa.float32(), pa.float64(), 1.5, -2.25, 3.5),
+ (pa.float64(), pa.float32(), 1.5, -2.25, 3.5),
+ (pa.binary(), pa.large_binary(), b'key', b'other', b'new'),
+ (pa.large_binary(), pa.binary(), b'key', b'other', b'new'),
(pa.decimal128(10, 2), pa.decimal128(15, 4), Decimal('1.00'),
Decimal('2.00'), Decimal('3.00')),
(pa.timestamp('ms'), pa.timestamp('us'), datetime.datetime(2024, 1, 1),
datetime.datetime(2024, 1, 2), datetime.datetime(2024, 1, 3)),
@@ -161,6 +290,7 @@ def test_new_rows_retain_original_key_write_type_validation(
@pytest.mark.parametrize('native', [False, True], ids=['python', 'native'])
@pytest.mark.parametrize('key_type,input_type,key,other,invalid', [
(pa.int8(), pa.int64(), 1, 2, 256),
+ (pa.float32(), pa.float64(), 1.0, 2.0, 1.00000001),
(pa.int64(), pa.uint64(), 1, 2, 2 ** 63),
(pa.decimal128(10, 2), pa.decimal128(15, 4), Decimal('1.25'),
Decimal('2.50'), Decimal('1.2501')),
(pa.timestamp('ms'), pa.timestamp('us'), datetime.datetime(2024, 1, 1),
@@ -230,3 +360,43 @@ def
test_decimal_zero_and_null_keys_rescale_across_the_full_scale_range(tmp_path
_upsert(table, source, native, stream)
for rows in _rows(table):
assert [row['value'] for row in rows] == [42, 42, 0, 9]
+
+
[email protected](os.name != 'posix' or sys.maxsize <= 2 ** 32,
reason='Requires 64-bit POSIX mmap')
[email protected]('native', [False, True], ids=['python', 'native'])
+def test_large_binary_key_slice_after_two_gib_matches_visible_bytes(tmp_path,
native):
+ table, _ = _table(tmp_path, native, pa.binary(), b'ab', b'other')
+ before = next(_rows(table))
+ start = 2 ** 31 + 17
+ # Reserve address space without allocating or reading a 2 GiB prefix.
+ memory = mmap.mmap(-1, start + 2)
+ memory[start:start + 2] = b'ab'
+ keys = pa.Array.from_buffers(pa.large_binary(), 1, [
+ None, pa.py_buffer(struct.pack('<qq', start, start + 2)),
pa.py_buffer(memory)])
+ source = pa.table({'key': keys, 'value': pa.array([42], type=pa.int32())})
+ _upsert(table, source, native, False)
+ for rows in _rows(table):
+ assert [row['value'] for row in rows] == [42, 42, 0, 0]
+ assert [{k: v for k, v in row.items() if k != 'value'} for row in
rows] == [
+ {k: v for k, v in row.items() if k != 'value'} for row in before]
+
+
[email protected](os.name != 'posix' or sys.maxsize <= 2 ** 32,
reason='Requires 64-bit POSIX mmap')
[email protected]('native', [False, True], ids=['python', 'native'])
[email protected]('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')
+ 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)
+ 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):
+ assert [row['value'] for row in rows] == ([42, 42, 0, 9] if mixed else
[0, 0, 0, 9])
+ assert [{k: v for k, v in row.items() if k != 'value'} for row in
rows] == [
+ {k: v for k, v in row.items() if k != 'value'} for row in before]
diff --git a/paimon-python/pypaimon/write/native_update.py
b/paimon-python/pypaimon/write/native_update.py
index dbf0b50d17..abd129958e 100644
--- a/paimon-python/pypaimon/write/native_update.py
+++ b/paimon-python/pypaimon/write/native_update.py
@@ -169,9 +169,21 @@ def _native_upsert_key_type_compatible(source, target):
source = _dictionary_value_type(source)
if pa.types.is_null(source):
return True
- for family in (pa.types.is_integer, pa.types.is_decimal, pa.types.is_date,
pa.types.is_time):
+
+ def numeric(data_type):
+ return (pa.types.is_integer(data_type) or
pa.types.is_floating(data_type)
+ or pa.types.is_decimal(data_type))
+ if numeric(source) and numeric(target):
+ return True
+ for family in (pa.types.is_date, pa.types.is_time):
if family(source) and family(target):
return True
+
+ def binary(data_type):
+ return (pa.types.is_binary(data_type) or
pa.types.is_large_binary(data_type)
+ or pa.types.is_fixed_size_binary(data_type))
+ if binary(source) and binary(target):
+ return True
if pa.types.is_timestamp(source) and pa.types.is_timestamp(target):
return (source.tz is None) == (target.tz is None)
return arrow_schemas_compatible(
