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 1651043c91 [python] Enable native nested updates, temporal upserts and
descending sequences (#10242)
1651043c91 is described below
commit 1651043c91c4779729b16f7f3c7724d6b036f622
Author: Jingsong Lee <[email protected]>
AuthorDate: Sun Sep 27 22:57:16 2026 +0800
[python] Enable native nested updates, temporal upserts and descending
sequences (#10242)
---
.../pypaimon/tests/native_nested_update_test.py | 301 +++++++++++++++++++++
.../pypaimon/tests/native_sequence_test.py | 121 +++++++++
.../tests/native_upsert_capabilities_test.py | 150 ++++++++++
paimon-python/pypaimon/tests/native_write_test.py | 2 +-
paimon-python/pypaimon/write/native_update.py | 23 +-
paimon-python/pypaimon/write/native_write.py | 9 +-
paimon-python/pypaimon/write/table_update.py | 2 +-
.../pypaimon/write/table_upsert_by_key.py | 3 +-
8 files changed, 584 insertions(+), 27 deletions(-)
diff --git a/paimon-python/pypaimon/tests/native_nested_update_test.py
b/paimon-python/pypaimon/tests/native_nested_update_test.py
new file mode 100644
index 0000000000..db691e94f6
--- /dev/null
+++ b/paimon-python/pypaimon/tests/native_nested_update_test.py
@@ -0,0 +1,301 @@
+# 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.
+
+"""Nested Native update dispatch and parity with the Python implementation."""
+
+from contextlib import ExitStack
+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.row.generic_row import GenericRow
+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 = [
+ (pa.list_(pa.struct([('name', pa.string()), ('amount', pa.int32())])),
+ [[{'name': 'a', 'amount': 1}, None], [], None, [{'name': None, 'amount':
4}]]),
+ (pa.map_(pa.field('source_key', pa.string(), nullable=False),
+ pa.field('source_value', pa.list_(pa.int32()))),
+ [[('a', [1, None]), ('b', [])], [], None, [('c', None)]]),
+ (pa.struct([('items', pa.list_(pa.int32())), ('raw', pa.binary())]),
+ [{'items': [1, None], 'raw': b'abc'}, {'items': [], 'raw': None}, None,
+ {'items': None, 'raw': b''}]),
+]
+
+
+def _table(tmp_path, data_type, native):
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path)})
+ catalog.create_database('default', True)
+ schema = pa.schema([('id', pa.int32()), ('value', data_type),
('untouched', pa.int32())])
+ catalog.create_table('default.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('default.t')
+ builder = table.new_batch_write_builder()
+ writer, commit = builder.new_write(), builder.new_commit()
+ try:
+ writer.write_arrow(pa.table({'id': [0, 1, 2, 3], 'value': [None] * 4,
+ 'untouched': [10, 11, 12, 13]},
schema=schema))
+ commit.commit(writer.prepare_commit())
+ finally:
+ writer.close()
+ commit.close()
+ return table, schema
+
+
+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'),
+ (TableUpsertByKey, '_upsert_row_partition')]:
+ stack.enter_context(patch.object(cls, method,
side_effect=AssertionError('Python fallback')))
+ return stack
+
+
+def _commit(builder, messages, stream):
+ commit = builder.new_commit()
+ try:
+ commit.commit(messages, 7) if stream else commit.commit(messages)
+ finally:
+ commit.close()
+
+
+def _check_rows(table, values):
+ for native in (False, True):
+ copy = table.copy({'scan.native-plan.enabled': str(native).lower(),
+ 'read.native.enabled': str(native).lower()})
+ builder = copy.new_read_builder()
+ read = builder.new_read()
+ with ExitStack() as stack:
+ if native:
+ stack.enter_context(patch.object(read, '_create_split_read',
+
side_effect=AssertionError('Python read fallback')))
+ actual =
read.to_arrow(builder.new_scan().plan().splits()).sort_by('id')
+ assert actual['value'].to_pylist() == values
+ assert actual['untouched'].to_pylist() == list(range(10, 10 +
len(values)))
+
+
[email protected]('native', [False, True])
[email protected]('stream', [False, True])
[email protected]('data_type,values', CASES)
[email protected]('operation', [
+ 'row_id', 'incremental', 'predicate_array', 'predicate_callable',
'predicate_scalar',
+])
+def test_nested_updates_use_core(tmp_path, native, stream, data_type, values,
operation):
+ table, _ = _table(tmp_path, data_type, native)
+ builder = table.new_stream_write_builder() if stream else
table.new_batch_write_builder()
+ update = builder.new_update().with_update_type(['value'])
+ # Slices retain nonzero offsets in both the top level and nested buffers.
+ array = pa.array([values[0]] + values, type=data_type).slice(1)
+ data = pa.table({'_ROW_ID': [0, 1, 2, 3], 'value': array})
+ with _native_only(native):
+ if operation == 'row_id':
+ messages = (update.update_by_arrow_with_row_id(data, 7) if stream
+ else update.update_by_arrow_with_row_id(data))
+ elif operation == 'incremental':
+ writer = update.new_update_by_row_id(7) if stream else
update.new_update_by_row_id()
+ messages = writer.update_columns(data, ['value'])
+ else:
+ predicate = PredicateBuilder(table.fields).greater_or_equal('id',
0)
+ if operation == 'predicate_callable':
+ def assignment(matched):
+ assert matched.column_names == ['id', '_ROW_ID']
+ return pa.array([values[i] for i in
matched['id'].to_pylist()], type=data_type)
+ elif operation == 'predicate_scalar':
+ assignment = values[0]
+ else:
+ assignment = pa.chunked_array([array.slice(0, 1),
array.slice(1)])
+ kwargs = {'read_columns': ['id']} if operation ==
'predicate_callable' else {}
+ if stream:
+ messages = update.update_by_predicate(predicate, {'value':
assignment}, 7, **kwargs)
+ else:
+ messages = update.update_by_predicate(predicate, {'value':
assignment}, **kwargs)
+ _commit(builder, messages, stream)
+ _check_rows(table, [values[0]] * 4 if operation == 'predicate_scalar' else
values)
+
+
[email protected]('native', [False, True])
[email protected]('stream', [False, True])
[email protected]('data_type,values', CASES)
[email protected]('row_input', [False, True])
+def test_nested_upsert_updates_and_appends(tmp_path, native, stream,
data_type, values, row_input):
+ table, schema = _table(tmp_path, data_type, native)
+ builder = table.new_stream_write_builder() if stream else
table.new_batch_write_builder()
+ update = builder.new_update().with_update_type(['value'])
+ ids = [0, 0, 1, 2, 3, 4]
+ source_values = [None] + values + [values[0]]
+ data = pa.table({'id': ids, 'value': source_values,
+ 'untouched': [99, 99, 99, 99, 99, 14]}, schema=schema)
+ with _native_only(native):
+ if row_input:
+ rows = [GenericRow([row[name] for name in schema.names],
table.fields)
+ for row in data.to_pylist()]
+ messages = (update.upsert_by_key(rows, ['id'], 7) if stream
+ else update.upsert_by_key(rows, ['id']))
+ else:
+ # Column order is unrelated to the schema/updated-column order.
+ data = data.select(['untouched', 'value', 'id'])
+ messages = (update.upsert_by_arrow_with_key(data, ['id'], 7) if
stream
+ else update.upsert_by_arrow_with_key(data, ['id']))
+ _commit(builder, messages, stream)
+ _check_rows(table, values + [values[0]])
+
+
[email protected]('native', [False, True])
[email protected]('input_values,target', [
+ (pa.array([{'a': 1.5, 'b': 2.0}, None]), pa.map_(pa.string(), pa.int32())),
+ (pa.array([{'a': 1, 'b': None}, None]), pa.map_(pa.string(), pa.int32())),
+ (pa.array([{}, None]), pa.map_(pa.string(), pa.int32())),
+ (pa.array([[[1.0, 2.5], [3.0, 4.0]], None, []]), pa.map_(pa.int32(),
pa.int32())),
+ (pa.array([[[1, 2, 3]], None]), pa.map_(pa.int32(), pa.int32())),
+ (pa.array([[1.5, None], [], None]), pa.list_(pa.int32())),
+ (pa.array([{'a': 1, 'b': 1.5}]), pa.struct([('a', pa.string()), ('b',
pa.int32())])),
+ (pa.array([{'m': {'a': 1, 'b': None}}, None]),
+ pa.struct([('m', pa.map_(pa.string(), pa.int32()))])),
+ (pa.array([{'m': [[1, 2]]}]), pa.struct([('m', pa.map_(pa.int32(),
pa.int32()))])),
+ (pa.array([[{'a': 1, 'b': None}], None]), pa.list_(pa.map_(pa.string(),
pa.int32()))),
+ (pa.array([[[[1, 2]]]]), pa.list_(pa.map_(pa.int32(), pa.int32()))),
+ (pa.array([[('a', 1), ('b', 2)], None], type=pa.map_(pa.string(),
pa.int32())),
+ pa.struct([('a', pa.int32()), ('b', pa.int32())])),
+ (pa.array([[('a', 1.5), ('b', 2.5), ('ignored', float('nan'))], [('a',
3.5)]],
+ type=pa.map_(pa.string(), pa.float64())),
+ pa.struct([('a', pa.int32()), ('b', pa.int32())])),
+ (pa.array([[('b', 1), ('a', 2)]], type=pa.map_(pa.string(), pa.int32())),
+ pa.struct([('a', pa.int32()), ('b', pa.int32())])),
+ (pa.StructArray.from_arrays([
+ pa.DictionaryArray.from_arrays(pa.array([0, 1], type=pa.int8()),
+ pa.array([1, None], type=pa.int32()))],
names=['a']),
+ pa.map_(pa.string(), pa.int32())),
+])
+def test_nested_row_id_coercion_matches_python(tmp_path, native, input_values,
target):
+ table, _ = _table(tmp_path, target, native)
+ builder = table.new_batch_write_builder()
+ update = builder.new_update().with_update_type(['value'])
+ data = pa.table({'_ROW_ID': list(range(len(input_values))), 'value':
input_values})
+ before = set(tmp_path.rglob('*.parquet'))
+ try:
+ expected = TableUpdateByRowId._coerce_column(input_values,
target).to_pylist()
+ except (ValueError, TypeError, pa.ArrowException):
+ with _native_only(native), pytest.raises((ValueError, TypeError,
pa.ArrowException)):
+ update.update_by_arrow_with_row_id(data)
+ assert set(tmp_path.rglob('*.parquet')) == before
+ expected = [None] * len(input_values)
+ else:
+ with _native_only(native):
+ messages = update.update_by_arrow_with_row_id(data)
+ _commit(builder, messages, False)
+ _check_rows(table, expected + [None] * (4 - len(input_values)))
+
+
[email protected]('native', [False, True])
[email protected]('operation', ['row_id', 'predicate'])
+def test_nested_struct_reordering_uses_operation_cast_rules(tmp_path, native,
operation):
+ target = pa.struct([('b', pa.int32()), ('a', pa.int32())])
+ values = pa.array([{'a': 1, 'b': 2}])
+ table, _ = _table(tmp_path, target, native)
+ builder = table.new_batch_write_builder()
+ update = builder.new_update().with_update_type(['value'])
+ predicate = PredicateBuilder(table.fields).equal('id', 0)
+ before = set(tmp_path.rglob('*.parquet'))
+ with _native_only(native):
+ if operation == 'predicate':
+ # Safe assignment casts reject reordering; row-ID coercion retries
+ # the whole input as dictionaries and can match ROW fields by name.
+ with pytest.raises((ValueError, pa.ArrowException)):
+ update.update_by_predicate(predicate, {'value': values})
+ assert set(tmp_path.rglob('*.parquet')) == before
+ expected = [None] * 4
+ else:
+ messages = update.update_by_arrow_with_row_id(pa.table({'_ROW_ID':
[0], 'value': values}))
+ _commit(builder, messages, False)
+ expected = [{'b': 2, 'a': 1}, None, None, None]
+ _check_rows(table, expected)
+
+
[email protected]('native', [False, True])
[email protected]('layout', ['list_view', 'large_list_view'])
+def test_view_pairs_update_map(tmp_path, native, layout):
+ if not hasattr(pa, layout):
+ pytest.skip('List view arrays require a newer PyArrow version')
+ target = pa.map_(pa.int32(), pa.int32())
+ table, _ = _table(tmp_path, target, native)
+ values = pa.array([[[99, 99, 99]], [[1, 2], [3, None]], None, []],
+ type=getattr(pa, layout)(pa.list_(pa.int64()))).slice(1)
+ builder = table.new_batch_write_builder()
+ with _native_only(native):
+ messages = builder.new_update().update_by_arrow_with_row_id(
+ pa.table({'_ROW_ID': [0, 1, 2], 'value': values}))
+ _commit(builder, messages, False)
+ _check_rows(table, [[(1, 2), (3, None)], None, [], None])
+
+
[email protected]('native', [False, True])
[email protected]('data_type,values', CASES)
[email protected]('overlap', [False, True])
+def test_nested_grouped_updates_abort_overlap(tmp_path, native, data_type,
values, overlap):
+ table, _ = _table(tmp_path, data_type, native)
+ builder = table.new_batch_write_builder()
+ update = builder.new_update().with_update_type(['value'])
+ data = pa.table({'_ROW_ID': [0, 1, 2, 3], 'value': pa.array(values,
type=data_type)})
+ before = set(tmp_path.rglob('*.parquet'))
+ with _native_only(native):
+ if overlap:
+ with pytest.raises(ValueError):
+ update.update_by_arrow_batches_with_row_id(iter([data, data]))
+ assert set(tmp_path.rglob('*.parquet')) == before
+ else:
+ messages =
update.update_by_arrow_batches_with_row_id(iter([data.slice(0, 0), data]))
+ _commit(builder, messages, False)
+ _check_rows(table, [None] * 4 if overlap else values)
+
+
[email protected]('native', [False, True])
[email protected]('key_type', ['string', 'large_string', 'string_view',
'binary',
+ 'large_binary', 'binary_view',
'fixed_binary'])
+def test_map_field_name_encodings(tmp_path, native, key_type):
+ if key_type == 'fixed_binary':
+ key_type = pa.binary(1)
+ elif hasattr(pa, key_type):
+ key_type = getattr(pa, key_type)()
+ else:
+ pytest.skip('View arrays require a newer PyArrow version')
+ target = pa.struct([('a', pa.int32()), ('b', pa.int32())])
+ table, _ = _table(tmp_path, target, native)
+ values = pa.array([[('a', 1.5), ('b', 2.5), ('x', float('nan'))], [('a',
3.5)]],
+ type=pa.map_(key_type, pa.float64()))
+ if not native:
+ try:
+ values.take(pa.array([0], type=pa.int32()))
+ except pa.ArrowNotImplementedError:
+ pytest.skip('This PyArrow version cannot filter MAPs with view
keys')
+ builder = table.new_batch_write_builder()
+ with _native_only(native):
+ messages = builder.new_update().update_by_arrow_with_row_id(
+ pa.table({'_ROW_ID': [0, 1], 'value': values}))
+ _commit(builder, messages, False)
+ _check_rows(table, [{'a': 1, 'b': 2}, {'a': 3, 'b': None}, None, None])
diff --git a/paimon-python/pypaimon/tests/native_sequence_test.py
b/paimon-python/pypaimon/tests/native_sequence_test.py
new file mode 100644
index 0000000000..20058bcc12
--- /dev/null
+++ b/paimon-python/pypaimon/tests/native_sequence_test.py
@@ -0,0 +1,121 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Java-compatible user sequence ordering across Python and Native files."""
+
+from contextlib import ExitStack
+from decimal import Decimal
+import datetime
+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.write.native_write import NativeTableWrite
+
+
+pytestmark = pytest.mark.native_plan
+
+
+def _table(tmp_path, sequence_type, native, order, engine):
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path)})
+ catalog.create_database('default', True)
+ schema = pa.schema([('id', pa.int32()), ('seq', sequence_type),
+ ('seq2', pa.int32()), ('value', pa.string())])
+ options = {'bucket': '1', 'sequence.field': 'seq,seq2',
+ 'sequence.field.sort-order': order, 'merge-engine': engine,
+ 'write.native.enabled': str(native).lower()}
+ if engine == 'aggregation':
+ options['fields.value.aggregate-function'] = 'last_value'
+ catalog.create_table('default.t', Schema.from_pyarrow_schema(
+ schema, primary_keys=['id'], options=options), False)
+ return catalog.get_table('default.t'), schema
+
+
+def _write(table, groups, native, streaming):
+ builder = table.new_stream_write_builder() if streaming else
table.new_batch_write_builder()
+ commit = builder.new_commit()
+ try:
+ for identifier, group in enumerate(groups, 1):
+ writer = builder.new_write()
+ if native:
+ assert isinstance(writer, NativeTableWrite)
+ try:
+ for chunk in group:
+ writer.write_arrow(chunk)
+ messages = (writer.prepare_commit(identifier) if streaming
+ else writer.prepare_commit())
+ if streaming:
+ commit.commit(messages, identifier)
+ else:
+ commit.commit(messages)
+ commit.close()
+ commit = builder.new_commit()
+ if native:
+ assert writer._python_writer is None
+ finally:
+ writer.close()
+ finally:
+ commit.close()
+
+
+def _read(table, native, predicate=None, projection=None):
+ copy = table.copy({'read.native.enabled': str(native).lower(),
+ 'scan.native-plan.enabled': str(native).lower()})
+ builder = copy.new_read_builder()
+ if predicate is not None:
+ builder.with_filter(predicate)
+ if projection is not None:
+ builder.with_projection(projection)
+ splits = builder.new_scan().plan().splits()
+ read = builder.new_read()
+ with ExitStack() as stack:
+ if native:
+ stack.enter_context(patch.object(read, '_create_split_read',
+
side_effect=AssertionError('Python read fallback')))
+ return read.to_arrow(splits).sort_by('id').to_pylist()
+
+
[email protected]('native', [False, True])
[email protected]('streaming', [False, True])
[email protected]('separate_commits', [False, True])
[email protected]('engine', ['deduplicate', 'partial-update',
'aggregation'])
[email protected]('sequence_type,high,low', [
+ (pa.int64(), 100, 50), (pa.string(), 'z', 'a'), (pa.binary(), b'z', b'a'),
+ (pa.decimal128(10, 2), Decimal('100.50'), Decimal('50.25')),
+ (pa.timestamp('us'), datetime.datetime(2020, 1, 2),
datetime.datetime(2020, 1, 1)),
+ (pa.time32('ms'), datetime.time(12, 0), datetime.time(1, 0)),
+])
+def test_descending_sequence_dispatch(
+ tmp_path, native, streaming, separate_commits, engine, sequence_type,
high, low):
+ table, schema = _table(tmp_path, sequence_type, native, 'descending',
engine)
+ batches = []
+ for i, value in enumerate([high, low, None]):
+ batches.append(pa.table({'id': [1, 2, 3], 'seq': [value, high, None],
+ 'seq2': [0, [2, 1, None][i], None],
+ 'value': [str(i)] * 3}, schema=schema))
+ _write(table, [[batch] for batch in batches] if separate_commits else
[batches],
+ native, streaming)
+ expected = [{'id': 1, 'value': '1'}, {'id': 2, 'value': '1'}, {'id': 3,
'value': '2'}]
+ for native_read in (False, True):
+ # A later NULL must not beat non-NULL, the second sequence field breaks
+ # ties, and automatic sequence still resolves entirely equal tuples.
+ assert _read(table, native_read, projection=['id', 'value']) ==
expected
+ predicate = PredicateBuilder(table.fields).equal('value', '1')
+ assert _read(table, native_read, predicate, ['id', 'value']) ==
expected[:2]
diff --git a/paimon-python/pypaimon/tests/native_upsert_capabilities_test.py
b/paimon-python/pypaimon/tests/native_upsert_capabilities_test.py
new file mode 100644
index 0000000000..888edf112e
--- /dev/null
+++ b/paimon-python/pypaimon/tests/native_upsert_capabilities_test.py
@@ -0,0 +1,150 @@
+# 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.
+
+"""Native upserts preserve temporal keys and read only input partitions."""
+
+from contextlib import ExitStack
+from unittest.mock import patch
+
+import pyarrow as pa
+import pytest
+
+from pypaimon import CatalogFactory, Schema
+from pypaimon.schema.data_types import PyarrowFieldParser
+from pypaimon.table.row.generic_row import GenericRow
+from pypaimon.write.table_upsert_by_key import TableUpsertByKey
+
+
+pytestmark = pytest.mark.native_plan
+
+
[email protected]('native', [False, True])
[email protected]('streaming', [False, True])
[email protected]('row_input', [False, True])
[email protected]('key_type', [
+ pa.time32('ms'), pa.timestamp('ms'), pa.timestamp('us'),
pa.timestamp('ns'),
+ pa.timestamp('us', 'UTC'), pa.timestamp('ns', 'UTC'),
+])
+def test_temporal_upsert_keys(tmp_path, native, streaming, row_input,
key_type):
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path)})
+ catalog.create_database('default', True)
+ schema = pa.schema([('key', key_type), ('tag', pa.int32()), ('value',
pa.string())])
+ catalog.create_table('default.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('default.t')
+ schema = PyarrowFieldParser.from_paimon_schema(table.fields)
+ key_type = schema.field('key').type
+ # Adjacent storage-unit values must stay distinct, including nanoseconds.
+ keys = pa.array([10, 10, None, 11], type=key_type)
+ seed = pa.table({'key': keys, 'tag': [0, 1, 2, 3], 'value': ['old'] * 4},
schema=schema)
+ batch_builder = table.new_batch_write_builder()
+ writer, commit = batch_builder.new_write(), batch_builder.new_commit()
+ try:
+ writer.write_arrow(seed)
+ commit.commit(writer.prepare_commit())
+ finally:
+ writer.close()
+ commit.close()
+ source = pa.table({'key': pa.array([None, 10, 12, 10], type=key_type),
+ 'tag': [99, 99, 4, 99],
+ 'value': ['null update', 'discarded', 'new', 'final']},
schema=schema)
+ builder = table.new_stream_write_builder() if streaming else
table.new_batch_write_builder()
+ update = builder.new_update().with_update_type(['value'])
+ with ExitStack() as stack:
+ if native:
+ for method in ('_upsert_partition', '_upsert_row_partition'):
+ stack.enter_context(patch.object(TableUpsertByKey, method,
+
side_effect=AssertionError('Python upsert fallback')))
+ if row_input:
+ rows = [GenericRow([row[field.name] for field in table.fields],
table.fields)
+ for row in source.to_pylist()]
+ messages = (update.upsert_by_key(rows, ['key'], 7) if streaming
+ else update.upsert_by_key(rows, ['key']))
+ else:
+ source = source.select(['value', 'key', 'tag'])
+ messages = (update.upsert_by_arrow_with_key(source, ['key'], 7) if
streaming
+ else update.upsert_by_arrow_with_key(source, ['key']))
+ commit = builder.new_commit()
+ try:
+ commit.commit(messages, 7) if streaming else commit.commit(messages)
+ finally:
+ commit.close()
+ for native_read in (False, True):
+ copy = table.copy({'read.native.enabled': str(native_read).lower(),
+ 'scan.native-plan.enabled':
str(native_read).lower()})
+ reader = copy.new_read_builder()
+ result =
reader.new_read().to_arrow(reader.new_scan().plan().splits()).sort_by('tag')
+ assert result['value'].to_pylist() == ['final', 'final', 'null
update', 'old', 'new']
+ assert result['tag'].to_pylist() == [0, 1, 2, 3, 4]
+ assert result['key'].cast(pa.int32() if pa.types.is_time32(key_type)
+ else pa.int64()).to_pylist() == [10, 10,
None, 11, 12]
+
+
[email protected]('native', [False, True])
+def test_upsert_reads_only_exact_input_partitions(tmp_path, native):
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path)})
+ catalog.create_database('default', True)
+ schema = pa.schema([('id', pa.int32()), ('pt', pa.string()), ('sub',
pa.int32()),
+ ('value', pa.int32())])
+ catalog.create_table('default.t', Schema.from_pyarrow_schema(
+ schema, partition_keys=['pt', 'sub'], options={
+ 'row-tracking.enabled': 'true', 'data-evolution.enabled': 'true',
+ 'write.native.enabled': str(native).lower(),
+ }), False)
+ table = catalog.get_table('default.t')
+ data = pa.table({'id': [1] * 5, 'pt': ['a', 'a', 'b', None, None],
+ 'sub': [1, 2, 1, 1, 2], 'value': [10] * 5}, schema=schema)
+ builder = table.new_batch_write_builder()
+ writer, commit = builder.new_write(), builder.new_commit()
+ try:
+ writer.write_arrow(data)
+ commit.commit(writer.prepare_commit())
+ finally:
+ writer.close()
+ commit.close()
+
+ def relevant(path):
+ return any(part in str(path) for part in (
+ '/pt=a/sub=1/', '/pt=b/sub=2/',
'/pt=__DEFAULT_PARTITION__/sub=1/'))
+
+ removed = 0
+ for path in tmp_path.rglob('*.parquet'):
+ if not relevant(path):
+ path.unlink()
+ removed += 1
+ assert removed == 3
+ source = pa.table({'id': [1] * 4, 'pt': ['a', 'b', None, 'a'],
+ 'sub': [1, 2, 1, 1], 'value': [15, 30, 40, 20]},
schema=schema)
+ with ExitStack() as stack:
+ if native:
+ stack.enter_context(patch.object(TableUpsertByKey,
'_upsert_partition',
+
side_effect=AssertionError('Python upsert fallback')))
+ messages =
builder.new_update().with_update_type(['value']).upsert_by_arrow_with_key(
+ source, ['id'])
+ commit = builder.new_commit()
+ try:
+ commit.commit(messages)
+ finally:
+ commit.close()
+ # Read only the partitions intentionally left accessible. The missing
+ # unrelated files prove that matching did not scan their data.
+ read = table.new_read_builder()
+ splits = [split for split in read.new_scan().plan().splits()
+ if
relevant(table.path_factory().bucket_path(tuple(split.partition.values),
split.bucket))]
+ assert sorted(read.new_read().to_arrow(splits)['value'].to_pylist()) ==
[20, 30, 40]
diff --git a/paimon-python/pypaimon/tests/native_write_test.py
b/paimon-python/pypaimon/tests/native_write_test.py
index 56bd5bb8a7..ba853082c0 100644
--- a/paimon-python/pypaimon/tests/native_write_test.py
+++ b/paimon-python/pypaimon/tests/native_write_test.py
@@ -79,7 +79,7 @@ def
test_sequence_validation_precedes_native_selection(tmp_path, streaming, sequ
@pytest.mark.parametrize('type_,order,supported', [
(pa.int64(), 'ascending', True),
- (pa.int64(), 'descending', False),
+ (pa.int64(), 'descending', True),
(pa.float32(), 'ascending', False),
(pa.float64(), 'ascending', False),
])
diff --git a/paimon-python/pypaimon/write/native_update.py
b/paimon-python/pypaimon/write/native_update.py
index d4c47367b0..8ecab84802 100644
--- a/paimon-python/pypaimon/write/native_update.py
+++ b/paimon-python/pypaimon/write/native_update.py
@@ -78,18 +78,9 @@ def _native_update_paths_supported(table):
return True
-def _native_update_table(table, columns=None):
- schema = PyarrowFieldParser.from_paimon_schema(table.table_schema.fields)
- selected = schema.names if columns is None else columns
- if any(pa.types.is_nested(schema.field(name).type)
- for name in selected if name in schema.names):
- return None
- return _native_row_id_table(table)
-
-
def create_native_update(table, commit_user, columns):
"""Use the public core updater for direct and grouped row-ID updates."""
- native_table = _native_update_table(table, columns)
+ native_table = _native_row_id_table(table)
if native_table is None:
return None
writer = (native_table.new_batch_write_builder()
@@ -102,7 +93,7 @@ def create_native_update(table, commit_user, columns):
def create_native_update_by_row_id(table, commit_user, commit_identifier):
"""Create a core updater sharing one snapshot across incremental calls."""
- native_table = _native_update_table(table)
+ native_table = _native_row_id_table(table)
if native_table is None:
return None
if commit_identifier == BATCH_COMMIT_IDENTIFIER:
@@ -120,6 +111,7 @@ def _supported_upsert_key_type(data_type):
pa.types.is_boolean, pa.types.is_integer, pa.types.is_string,
pa.types.is_large_string, pa.types.is_binary, pa.types.is_large_binary,
pa.types.is_fixed_size_binary, pa.types.is_date, pa.types.is_decimal,
+ pa.types.is_time, pa.types.is_timestamp,
))
@@ -130,8 +122,7 @@ def create_native_upsert(table, commit_user, data, keys,
columns):
return None
fields = table.table_schema.fields
schema = PyarrowFieldParser.from_paimon_schema(fields)
- if (any(not _supported_upsert_key_type(schema.field(key).type) for key in
set(keys + table.partition_keys))
- or any(pa.types.is_nested(schema.field(name).type) for name in
columns)):
+ if any(not _supported_upsert_key_type(schema.field(key).type) for key in
set(keys + table.partition_keys)):
return None
if not isinstance(data, pa.Table):
# Missing fields retain their row-object semantics on the fallback
@@ -154,12 +145,8 @@ def create_native_upsert(table, commit_user, data, keys,
columns):
return NativeTableUpsert(table, writer, keys, data)
-def create_native_predicate_update(table, commit_user, columns, predicate):
+def create_native_predicate_update(table, commit_user, predicate):
"""Prepare a public core operation before any assignment can run."""
- schema = PyarrowFieldParser.from_paimon_schema(table.table_schema.fields)
- if any(pa.types.is_nested(schema.field(name).type)
- for name in columns if name in schema.names):
- return None
native_table = _native_row_id_table(table)
if native_table is None:
return None
diff --git a/paimon-python/pypaimon/write/native_write.py
b/paimon-python/pypaimon/write/native_write.py
index 96f2abb9e8..227a1b905c 100644
--- a/paimon-python/pypaimon/write/native_write.py
+++ b/paimon-python/pypaimon/write/native_write.py
@@ -63,12 +63,9 @@ def create_native_write(table, commit_user,
static_partition=None, stream=False)
"""Return a native writer if the table can use the filesystem write
path."""
schema = PyarrowFieldParser.from_paimon_schema(table.table_schema.fields)
sequence_fields = table.options.sequence_field()
- if table.is_primary_key_table and sequence_fields:
- # The native writer currently sorts sequence fields ascending and
- # does not implement Java's NaN/signed-zero ordering.
- if (not table.options.sequence_field_sort_order_is_ascending()
- or any(pa.types.is_floating(schema.field(name).type) for name
in sequence_fields)):
- return None
+ if (table.is_primary_key_table and any(
+ pa.types.is_floating(schema.field(name).type) for name in
sequence_fields)):
+ return None
if (not native_write_available()
# Rust does not produce the optional random-access .row sidecars.
or (table.options.data_evolution_enabled()
diff --git a/paimon-python/pypaimon/write/table_update.py
b/paimon-python/pypaimon/write/table_update.py
index 5c5958cb1e..81e3717ed1 100644
--- a/paimon-python/pypaimon/write/table_update.py
+++ b/paimon-python/pypaimon/write/table_update.py
@@ -306,7 +306,7 @@ class TableUpdate:
from pypaimon.write.native_update import
create_native_predicate_update
try:
native = create_native_predicate_update(
- self.table, self.commit_user, list(assignments.keys()),
predicate)
+ self.table, self.commit_user, predicate)
except Exception as error:
logger.debug('Native predicate update preparation failed: %s',
error)
else:
diff --git a/paimon-python/pypaimon/write/table_upsert_by_key.py
b/paimon-python/pypaimon/write/table_upsert_by_key.py
index 3ffb6ad30e..f3b4064048 100644
--- a/paimon-python/pypaimon/write/table_upsert_by_key.py
+++ b/paimon-python/pypaimon/write/table_upsert_by_key.py
@@ -524,7 +524,8 @@ class TableUpsertByKey:
if partition_spec:
predicate_builder = read_builder.new_predicate_builder()
sub_predicates = [
- predicate_builder.equal(k, v)
+ (predicate_builder.is_null(k) if v is None
+ else predicate_builder.equal(k, v))
for k, v in partition_spec.items()
]
partition_predicate =
predicate_builder.and_predicates(sub_predicates)
