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 5a6739c6c1 [Python] Align native mutations, partial writes and commit
semantics (#10648)
5a6739c6c1 is described below
commit 5a6739c6c10e96052bac678eb245742ad92bc556
Author: Jingsong Lee <[email protected]>
AuthorDate: Mon Oct 19 16:28:16 2026 +0800
[Python] Align native mutations, partial writes and commit semantics
(#10648)
---
docs/docs/pypaimon/writing.md | 18 +-
.../pypaimon/manifest/index_manifest_file.py | 29 +-
.../pypaimon/table/data_evolution_merge_into.py | 4 +
.../pypaimon/tests/index_manifest_write_test.py | 28 +-
.../tests/native_authorized_update_test.py | 584 +++++++++++++++++++++
.../pypaimon/tests/native_commit_semantics_test.py | 244 +++++++++
.../tests/native_overwrite_indexes_test.py | 357 +++++++++++++
.../tests/native_upsert_encoded_keys_test.py | 228 ++++++++
.../pypaimon/tests/native_write_routing_test.py | 21 +-
.../pypaimon/tests/native_write_type_test.py | 254 +++++++++
.../pypaimon/tests/write/table_write_test.py | 6 +-
.../write/commit/overwrite_changes_provider.py | 42 +-
paimon-python/pypaimon/write/file_store_commit.py | 70 +--
paimon-python/pypaimon/write/native_merge_into.py | 5 +-
paimon-python/pypaimon/write/native_update.py | 23 +-
paimon-python/pypaimon/write/native_write.py | 27 +-
paimon-python/pypaimon/write/table_update.py | 11 +
paimon-python/pypaimon/write/table_write.py | 15 +-
18 files changed, 1879 insertions(+), 87 deletions(-)
diff --git a/docs/docs/pypaimon/writing.md b/docs/docs/pypaimon/writing.md
index c74deca7c8..33144b6dd4 100644
--- a/docs/docs/pypaimon/writing.md
+++ b/docs/docs/pypaimon/writing.md
@@ -79,12 +79,20 @@ to use Rust for Parquet/BLOB writes and
`commit.native.enabled=true` to use
Rust for REST snapshot publication. Batch and stream writers retain the same
prepare-and-commit lifecycle. Native commits also support overwrite,
`truncate_table()`, `truncate_partitions()` and commit callbacks.
+Overwrite and truncate record an OVERWRITE snapshot even when no files match,
+and remove index metadata for the replaced partitions. Physical files remain
+available to historical snapshots. Changelog-only commits also publish a
snapshot.
Standard REST tables with `query-auth.enabled=true` can write and commit
without requesting SELECT permission. The REST server still authorizes snapshot
publication. Row filters and column masks continue to apply to explicit reads;
internal sequence-number and HASH-index restoration reads manifest metadata.
-Predicate updates and deletes retain their existing read authorization checks.
+For query-authorized targets, Native row-ID updates, predicate updates/deletes,
+upserts, shard updates and MERGE require an unrestricted whole-table SELECT
grant
+before reading target rows or evaluating assignments. A row filter, column mask
+or denied grant rejects the operation; ordinary writes still require no SELECT
+grant. MERGE table sources obtain their own grants independently from the
target,
+including historical snapshots and branches.
Query-authorized cross-partition index bootstrap remains unsupported.
`write_ray()` uses the shared Python coordinator and lets workers select Native
@@ -93,6 +101,14 @@ unless an explicit bucket plan was supplied. A complete
primary-key
`with_write_type()` leaves the writer unchanged; primary-key column subsets
remain unsupported.
+Append `with_write_type()` preserves the selected column order. Both the full
+table schema and the selected schema are accepted as Arrow input; only selected
+columns are written. Full Arrow inputs and named Rows can carry partition and
+hash columns used only for routing, independently of the physical write type.
+Reads restore omitted partition columns from split metadata. Selected-schema
+inputs must include the routing columns they need; `write_arrow(data,
bucket=...)`
+also accepts a bucket assigned upstream.
+
Native commit selection requires a standard REST catalog and a reconstructible
built-in FileIO. Other catalogs and custom Python FileIO implementations retain
the Python path. Binary partition keys require `partition.legacy-name=false`
for
diff --git a/paimon-python/pypaimon/manifest/index_manifest_file.py
b/paimon-python/pypaimon/manifest/index_manifest_file.py
index ee78f08d4a..a19c196f50 100644
--- a/paimon-python/pypaimon/manifest/index_manifest_file.py
+++ b/paimon-python/pypaimon/manifest/index_manifest_file.py
@@ -252,8 +252,10 @@ class IndexManifestFile:
if not adds and not deletes:
return previous_name
previous = self.read(previous_name) if previous_name else []
- delete_names = {e.index_file.file_name for e in deletes}
- survivors = [e for e in previous if e.index_file.file_name not in
delete_names]
+ _validate_global_index_deletes(previous, deletes)
+ delete_files = {(e.index_file.index_type, e.index_file.file_name) for
e in deletes}
+ survivors = [e for e in previous
+ if (e.index_file.index_type, e.index_file.file_name) not
in delete_files]
_validate_retained_global_index_files(survivors, adds)
combined = _combine_hash_indexes(survivors, adds, deletes)
if not combined:
@@ -313,6 +315,29 @@ class IndexManifestFile:
}
+def _validate_global_index_deletes(previous_entries, deleted_entries) -> None:
+ # Java combines global indexes by index type and file name. Verify each
+ # removal against the current entry, including metadata-only reassignment.
+ # HASH and DV keep their existing bucket/index-specific deletion rules.
+ current = {(entry.index_file.index_type, entry.index_file.file_name): entry
+ for entry in previous_entries if
_is_global_index(entry.index_file.index_type)}
+ for deleted in deleted_entries:
+ meta = deleted.index_file
+ if not _is_global_index(meta.index_type):
+ continue
+ previous = current.pop((meta.index_type, meta.file_name), None)
+ if previous is None:
+ raise RuntimeError(
+ 'Trying to delete global index file %s which does not exist.'
% meta.file_name)
+ previous_meta = previous.index_file
+ if (previous.partition != deleted.partition or previous.bucket !=
deleted.bucket
+ or previous_meta != meta or previous_meta.dv_ranges !=
meta.dv_ranges
+ or previous_meta.external_path != meta.external_path
+ or previous_meta.global_index_meta != meta.global_index_meta):
+ raise RuntimeError(
+ 'Index file %s metadata does not match the current entry.' %
meta.file_name)
+
+
def _validate_retained_global_index_files(
retained_entries: List[IndexManifestEntry],
added_entries: List[IndexManifestEntry],
diff --git a/paimon-python/pypaimon/table/data_evolution_merge_into.py
b/paimon-python/pypaimon/table/data_evolution_merge_into.py
index d64294af60..91ca996634 100644
--- a/paimon-python/pypaimon/table/data_evolution_merge_into.py
+++ b/paimon-python/pypaimon/table/data_evolution_merge_into.py
@@ -106,6 +106,10 @@ def merge_into(
if native is not None:
return native.prepare_commit()
+ # Empty-target INSERT skips the target scan, but still needs the same
+ # unrestricted whole-table grant before any user callback or staging.
+ from pypaimon.write.table_update import _authorize_update
+ _authorize_update(target_table)
base_snapshot = target_table.snapshot_manager().get_latest_snapshot()
source_table, matched_specs, not_matched_specs, ctx = _prepare(
target_table,
diff --git a/paimon-python/pypaimon/tests/index_manifest_write_test.py
b/paimon-python/pypaimon/tests/index_manifest_write_test.py
index e6e428f6b5..b3ec41b500 100644
--- a/paimon-python/pypaimon/tests/index_manifest_write_test.py
+++ b/paimon-python/pypaimon/tests/index_manifest_write_test.py
@@ -109,12 +109,32 @@ class IndexManifestWriteTest(unittest.TestCase):
survivors = {e.index_file.file_name for e in imf.read(new_name)}
self.assertEqual({'idx-b'}, survivors)
- def test_combine_unknown_delete_is_noop_on_content(self):
+ def test_combine_unknown_global_index_delete_is_rejected(self):
imf = IndexManifestFile(self._table())
previous = imf.write([self._entry('idx-a', 1)])
- new_name = imf.combine_deletes(previous, [self._entry('idx-zzz', 9)])
- survivors = {e.index_file.file_name for e in imf.read(new_name)}
- self.assertEqual({'idx-a'}, survivors)
+ with self.assertRaisesRegex(RuntimeError, 'does not exist'):
+ imf.combine_deletes(previous, [self._entry('idx-zzz', 9)])
+ self.assertEqual(['idx-a'], [e.index_file.file_name for e in
imf.read(previous)])
+
+ def
test_combine_global_index_delete_retains_other_type_with_same_name(self):
+ imf = IndexManifestFile(self._table())
+ btree = self._entry('shared-name', 1)
+ bitmap = self._entry('shared-name', 1)
+ bitmap.index_file.index_type = 'bitmap'
+ previous = imf.write([btree, bitmap])
+ current = imf.combine_deletes(previous, [btree])
+ self.assertEqual(['bitmap'], [e.index_file.index_type for e in
imf.read(current)])
+
+ def
test_combine_hash_index_delete_keeps_bucketed_stale_delete_semantics(self):
+ imf = IndexManifestFile(self._table())
+ current = self._entry('current-hash', 1)
+ current.index_file.index_type = 'HASH'
+ current.index_file.global_index_meta = None
+ previous = imf.write([current])
+ stale = self._entry('old-hash', 1)
+ stale.index_file.index_type = 'HASH'
+ stale.index_file.global_index_meta = None
+ self.assertIsNone(imf.combine_deletes(previous, [stale]))
def test_combine_empty_deletes_returns_previous(self):
imf = IndexManifestFile(self._table())
diff --git a/paimon-python/pypaimon/tests/native_authorized_update_test.py
b/paimon-python/pypaimon/tests/native_authorized_update_test.py
new file mode 100644
index 0000000000..1f9c95d02a
--- /dev/null
+++ b/paimon-python/pypaimon/tests/native_authorized_update_test.py
@@ -0,0 +1,584 @@
+# 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.
+
+"""Read-based Native mutations require unrestricted REST table
authorization."""
+
+import json
+from contextlib import ExitStack
+from pathlib import Path
+from urllib.parse import unquote, urlparse
+from unittest.mock import Mock, patch
+
+import pyarrow as pa
+import pyarrow.compute as pc
+import pytest
+
+from pypaimon import Schema
+from pypaimon.api.api_response import AuthTableQueryResponse, ErrorResponse
+from pypaimon.read.native_plan import native_runtime_available
+from pypaimon.table.data_evolution_merge_into import WhenMatched,
WhenNotMatched, source_col
+from pypaimon.table.row.generic_row import GenericRow
+from pypaimon.tests import native_plan_rest_test
+from pypaimon.write.native_update import NativeShardTableUpdator,
_native_row_id_table
+
+
+rest_catalog = native_plan_rest_test.rest_catalog
+pytestmark = [pytest.mark.python_write, pytest.mark.python_commit]
+native = pytest.mark.skipif(not native_runtime_available(),
reason='pypaimon-rust not installed')
+_ENGINES = [False, pytest.param(True, marks=native)]
+_SCHEMA = pa.schema([('id', pa.int32()), ('value', pa.int32()), ('name',
pa.string())])
+_ROWS = [dict(id=1, value=10, name='a'), dict(id=2, value=20, name='b'),
dict(id=3, value=30, name='c')]
+_OPERATIONS = ['row-id', 'grouped', 'incremental', 'predicate', 'upsert',
'row-upsert',
+ 'row-delete', 'predicate-delete', 'shard', 'merge']
+
+
+def _files(table, pattern='*'):
+ root = Path(unquote(urlparse(table.table_path).path))
+ assert root.is_dir(), table.table_path
+ return set(root.rglob(pattern))
+
+
+def _append(table, rows):
+ builder = table.new_batch_write_builder()
+ writer, commit = builder.new_write(), builder.new_commit()
+ try:
+ writer.write_arrow(pa.Table.from_pylist(rows, schema=_SCHEMA))
+ commit.commit(writer.prepare_commit())
+ finally:
+ writer.close()
+ commit.close()
+
+
+def _table(catalog, use_native=False, deletion_vectors=True,
partitioned=False, seed=True):
+ catalog.create_table('default.t', Schema.from_pyarrow_schema(
+ _SCHEMA, partition_keys=['name'] if partitioned else [], options={
+ 'query-auth.enabled': 'true', 'row-tracking.enabled': 'true',
'data-evolution.enabled': 'true',
+ 'deletion-vectors.enabled': str(deletion_vectors).lower(),
+ 'write.native.enabled': 'false', 'commit.native.enabled': 'false',
+ 'read.native.enabled': 'false', 'scan.native-plan.enabled':
'false',
+ }), False)
+ table = catalog.get_table('default.t')
+ # Separate anchor files permit grouped updates of the same column.
+ if seed:
+ _append(table, _ROWS[:2])
+ _append(table, _ROWS[2:])
+ return table.copy({'write.native.enabled': str(use_native).lower()})
+
+
+def _rows(table):
+ builder = table.new_read_builder()
+ return
builder.new_read().to_arrow(builder.new_scan().plan().splits()).sort_by('id').to_pylist()
+
+
+def _native_only(stack, use_native):
+ if use_native:
+ for method in (
+
'pypaimon.write.table_update_by_row_id.TableUpdateByRowId._load_existing_files_info',
+
'pypaimon.write.table_upsert_by_key.TableUpsertByKey._upsert_partition',
+
'pypaimon.write.table_delete.TableDeleteByRowId._scan_anchor_ranges',
+ 'pypaimon.write.table_update.ShardTableUpdator',
+ 'pypaimon.table.data_evolution_merge_into._build_tables',
+ 'pypaimon.read.table_read.TableRead.to_arrow'):
+ stack.enter_context(patch(method,
side_effect=AssertionError('Python mutation fallback')))
+
+
+def _operate(update, operation, stream=False, assignment=None):
+ kwargs = {'commit_identifier': 27} if stream else {}
+ if operation == 'row-id':
+ return update.update_by_arrow_with_row_id(pa.table({'_ROW_ID': [0],
'value': [11]}), **kwargs)
+ if operation == 'grouped':
+ return update.update_by_arrow_batches_with_row_id(iter([
+ pa.table({'_ROW_ID': [0], 'value': [11]}), pa.table({'_ROW_ID':
[2], 'value': [33]})]))
+ if operation == 'incremental':
+ updater = update.new_update_by_row_id(**kwargs)
+ updater.update_columns(pa.table({'_ROW_ID': [0], 'value': [11]}),
['value'])
+ return updater.update_columns(pa.table({'_ROW_ID': [1], 'name':
['B']}), ['name'])
+ predicate = update.new_predicate_builder().equal('id', 1)
+ if operation == 'predicate':
+ assignment = assignment or (lambda batch: pc.add(batch['value'], 1))
+ return update.update_by_predicate(predicate, {'value': assignment},
read_columns=['value'], **kwargs)
+ if operation in ('upsert', 'row-upsert'):
+ rows = [dict(id=1, value=11, name='A'), dict(id=4, value=40, name='d')]
+ if operation == 'row-upsert':
+ return update.upsert_by_key([
+ GenericRow([row['id'], row['value'], row['name']],
update.table.fields) for row in rows
+ ], ['id'], **kwargs)
+ return update.upsert_by_arrow_with_key(pa.Table.from_pylist(rows,
schema=_SCHEMA), ['id'], **kwargs)
+ if operation == 'row-delete':
+ return update.delete_by_row_id([0, 0], **kwargs)
+ if operation == 'predicate-delete':
+ return update.delete_by_predicate(predicate, **kwargs)
+ if operation == 'shard':
+ update.with_update_type(['value'])
+ update.with_read_projection(['id'])
+ updater = update.new_shard_updator(0, 1)
+ try:
+ with updater.arrow_reader() as reader:
+ for batch in reader:
+
updater.update_by_arrow_batch(pa.record_batch([pc.multiply(batch['id'], 100)],
names=['value']))
+ return updater.prepare_commit()
+ finally:
+ if isinstance(updater, NativeShardTableUpdator):
+ updater.close()
+ if operation == 'merge':
+ source = pa.Table.from_pylist([dict(id=1, value=11, name='A'),
dict(id=4, value=40, name='d')],
+ schema=_SCHEMA)
+ assignment = assignment or (lambda batch: pc.add(batch['s.value'], 1))
+ return update.merge_into(source, on=['id'],
when_matched=[WhenMatched.update({'value': assignment})],
+ when_not_matched=[WhenNotMatched('*')],
**kwargs)
+ raise AssertionError(operation)
+
+
+def _expected(operation):
+ rows = [dict(row) for row in _ROWS]
+ if operation in ('row-delete', 'predicate-delete'):
+ return rows[1:]
+ if operation == 'shard':
+ return [dict(row, value=row['id'] * 100) for row in rows]
+ rows[0]['value'] = 12 if operation == 'merge' else 11
+ if operation == 'grouped':
+ rows[2]['value'] = 33
+ if operation == 'incremental':
+ rows[1]['name'] = 'B'
+ if operation in ('upsert', 'row-upsert', 'merge'):
+ rows.append(dict(id=4, value=40, name='d'))
+ if operation != 'merge':
+ rows[0]['name'] = 'A'
+ return rows
+
+
[email protected]('use_native', _ENGINES)
[email protected]('operation', _OPERATIONS)
+def test_unrestricted_rest_updates_share_python_semantics(rest_catalog,
use_native, operation):
+ catalog, server = rest_catalog
+ table = _table(catalog, use_native)
+ builder = table.new_batch_write_builder()
+ with patch.object(server, '_table_query_auth_handle',
wraps=server._table_query_auth_handle) as authorize:
+ with ExitStack() as stack:
+ _native_only(stack, use_native)
+ messages = _operate(builder.new_update(), operation)
+ assert messages
+ builder.new_commit().commit(messages)
+ assert authorize.call_count >= 1
+ assert any(json.loads(call.args[1]).get('select') is None for call in
authorize.call_args_list)
+ assert _rows(table) == _expected(operation)
+
+
[email protected]('use_native', _ENGINES)
[email protected]('operation', ['update', 'delete'])
[email protected]('policy', ['whole', 'denied', 'filter', 'mask'])
[email protected]('metadata', ['_ROW_ID', '_SEQUENCE_NUMBER'])
+def test_metadata_predicate_mutations_require_whole_table_authorization(
+ rest_catalog, use_native, operation, policy, metadata):
+ catalog, server = rest_catalog
+ table = _table(catalog, use_native)
+ read = table.new_read_builder().with_projection(_SCHEMA.names + [metadata])
+ predicate = read.new_predicate_builder().equal(metadata, 0 if metadata ==
'_ROW_ID' else 1)
+ builder = table.new_batch_write_builder()
+ assignment = Mock(side_effect=lambda batch: pc.add(batch['value'], 1))
+ before = _files(table)
+ snapshot_id = table.snapshot_manager().get_latest_snapshot().id
+
+ def mutate():
+ update = builder.new_update()
+ if operation == 'delete':
+ return update.delete_by_predicate(predicate)
+ return update.update_by_predicate(predicate, {'value': assignment},
read_columns=['value'])
+
+ with ExitStack() as stack:
+ authorize = stack.enter_context(
+ patch.object(server, '_table_query_auth_handle',
wraps=server._table_query_auth_handle)
+ if policy == 'whole' else _restricted(server, table, policy))
+ _native_only(stack, use_native)
+ if policy == 'whole':
+ messages = mutate()
+ assert messages
+ else:
+ with pytest.raises(Exception,
match='[Pp]ermission|denied|unrestricted|restrict|[Mm]ask|query.auth'):
+ mutate()
+ assignment.assert_not_called()
+ assert _files(table) == before
+ assert table.snapshot_manager().get_latest_snapshot().id ==
snapshot_id
+ assert authorize.call_count >= 1
+ assert all(json.loads(call.args[1]).get('select') is None for call in
authorize.call_args_list)
+ if policy == 'whole':
+ assert assignment.call_count == (1 if operation == 'update' else 0)
+ builder.new_commit().commit(messages)
+ matched = 1 if metadata == '_ROW_ID' else 2
+ expected = ([dict(row, value=row['value'] + 1) for row in
_ROWS[:matched]] + _ROWS[matched:]
+ if operation == 'update' else _ROWS[matched:])
+ assert _rows(table) == expected
+
+
[email protected]('use_native', _ENGINES)
[email protected]('operation', ['row-id', 'incremental', 'predicate',
'upsert', 'row-delete', 'merge'])
+def test_stream_rest_updates_keep_commit_identifier(rest_catalog, use_native,
operation):
+ catalog, _ = rest_catalog
+ table = _table(catalog, use_native)
+ builder = table.new_stream_write_builder()
+ with ExitStack() as stack:
+ _native_only(stack, use_native)
+ messages = _operate(builder.new_update(), operation, stream=True)
+ commit = builder.new_commit()
+ try:
+ commit.commit(messages, 27)
+ finally:
+ commit.close()
+ assert table.snapshot_manager().get_latest_snapshot().commit_identifier ==
27
+ assert _rows(table) == _expected(operation)
+
+
+def _restricted(server, table, policy):
+ if policy == 'denied':
+ return patch.object(server, '_table_query_auth_handle',
return_value=server._mock_response(
+ ErrorResponse('TABLE', 't', 'SELECT denied', 403), 403))
+ field = {'name': 'id', 'type': 'INT', 'index': 0}
+ response = (AuthTableQueryResponse(filter=[json.dumps({
+ 'kind': 'LEAF', 'transform': {'name': 'FIELD_REF', 'fieldRef': field},
+ 'function': 'EQUAL', 'literals': [1]})]) if policy == 'filter' else
+ AuthTableQueryResponse(column_masking={'name': json.dumps({'name':
'CONCAT', 'inputs': ['masked']})}))
+ server.table_query_auth_store[table.identifier.get_full_name()] = response
+ return patch.object(server, '_table_query_auth_handle',
wraps=server._table_query_auth_handle)
+
+
[email protected]('use_native', _ENGINES)
[email protected]('policy', ['denied', 'filter', 'mask'])
[email protected]('operation', _OPERATIONS)
+def test_restricted_target_refuses_before_callable_or_staging(rest_catalog,
use_native, policy, operation):
+ catalog, server = rest_catalog
+ table = _table(catalog, use_native)
+ before = _files(table)
+ snapshot_id = table.snapshot_manager().get_latest_snapshot().id
+ assignment = Mock(side_effect=AssertionError('assignment ran before
authorization'))
+ with _restricted(server, table, policy) as authorize:
+ with ExitStack() as stack:
+ _native_only(stack, use_native)
+ with pytest.raises(Exception,
match='[Pp]ermission|denied|unrestricted|restrict|[Mm]ask|query.auth'):
+ _operate(table.new_batch_write_builder().new_update(),
operation, assignment=assignment)
+ assert authorize.call_count >= 1
+ assignment.assert_not_called()
+ assert _files(table) == before
+ assert table.snapshot_manager().get_latest_snapshot().id == snapshot_id
+
+
[email protected]('use_native', _ENGINES)
[email protected]('operation', ['row-delete', 'predicate-delete'])
+def test_authorized_delete_still_requires_deletion_vectors(rest_catalog,
use_native, operation):
+ catalog, _ = rest_catalog
+ table = _table(catalog, use_native, deletion_vectors=False)
+ before = _files(table)
+ with pytest.raises(NotImplementedError, match='deletion-vectors.enabled'):
+ _operate(table.new_batch_write_builder().new_update(), operation)
+ assert _files(table) == before
+
+
+def
test_guarded_native_update_resolves_the_rest_table_instead_of_switching_to_python(rest_catalog):
+ catalog, _ = rest_catalog
+ table = _table(catalog, True)
+ resolved = object()
+ with patch('pypaimon.write.native_update.native_runtime_available',
return_value=True), \
+ patch('pypaimon.write.native_update.create_native_write_table',
return_value=resolved) as create:
+ assert _native_row_id_table(table) is resolved
+ create.assert_called_once_with(table)
+
+
[email protected]('use_native', _ENGINES)
[email protected]('policy', ['whole', 'denied', 'filter', 'mask'])
+def
test_partition_delete_optimization_keeps_whole_table_authorization(rest_catalog,
use_native, policy):
+ catalog, server = rest_catalog
+ table = _table(catalog, use_native, partitioned=True)
+ builder = table.new_batch_write_builder()
+ predicate = builder.new_update().new_predicate_builder().equal('name', 'a')
+ before = _files(table)
+ with ExitStack() as stack:
+ authorize = stack.enter_context(
+ patch.object(server, '_table_query_auth_handle',
wraps=server._table_query_auth_handle)
+ if policy == 'whole' else _restricted(server, table, policy))
+ _native_only(stack, use_native)
+ if policy == 'whole':
+ messages = builder.new_update().delete_by_predicate(predicate)
+ assert messages
+ else:
+ with pytest.raises(Exception,
match='[Pp]ermission|denied|unrestricted|restrict|[Mm]ask|query.auth'):
+ builder.new_update().delete_by_predicate(predicate)
+ assert _files(table) == before
+ assert authorize.call_count >= 1
+ if policy == 'whole':
+ builder.new_commit().commit(messages)
+ assert _rows(table) == _ROWS[1:]
+
+
[email protected]('use_native', _ENGINES)
[email protected]('policy', ['denied', 'filter', 'mask'])
+def test_empty_target_merge_authorizes_before_insert_callable(rest_catalog,
use_native, policy):
+ catalog, server = rest_catalog
+ table = _table(catalog, use_native, seed=False)
+ assignment = Mock(side_effect=AssertionError('insert ran before target
authorization'))
+ source = pa.Table.from_pylist(_ROWS[:1], schema=_SCHEMA)
+ before = _files(table)
+ with _restricted(server, table, policy) as authorize:
+ with ExitStack() as stack:
+ _native_only(stack, use_native)
+ with pytest.raises(Exception,
match='[Pp]ermission|denied|unrestricted|restrict|[Mm]ask|query.auth'):
+ table.new_batch_write_builder().new_update().merge_into(
+ source, on=['id'], when_not_matched=[
+ WhenNotMatched({'value': assignment, 'name':
'insert'})])
+ assert authorize.call_count >= 1
+ assignment.assert_not_called()
+ assert _files(table) == before
+ assert table.snapshot_manager().get_latest_snapshot() is None
+
+
+def test_custom_query_authorization_stays_on_python_path(rest_catalog):
+ catalog, _ = rest_catalog
+ table = _table(catalog, True)
+ custom_auth = Mock()
+ with patch.object(table.catalog_environment, 'table_query_auth',
return_value=custom_auth), \
+ patch('pypaimon.write.native_update.native_runtime_available',
return_value=True), \
+ patch('pypaimon.write.native_update.create_native_write_table') as
create:
+ assert _native_row_id_table(table) is None
+ create.assert_not_called()
+ custom_auth.assert_not_called()
+
+
[email protected]('use_native', _ENGINES)
+def
test_unrestricted_empty_target_merge_evaluates_insert_in_rust_or_python(rest_catalog,
use_native):
+ catalog, _ = rest_catalog
+ table = _table(catalog, use_native, seed=False)
+ builder = table.new_batch_write_builder()
+ source = pa.Table.from_pylist(_ROWS[:1], schema=_SCHEMA)
+ assignment = Mock(side_effect=lambda batch: pc.add(batch['s.value'], 1))
+ with ExitStack() as stack:
+ _native_only(stack, use_native)
+ messages = builder.new_update().merge_into(
+ source, on=['id'], when_not_matched=[WhenNotMatched({'value':
assignment, 'name': source_col('name')})])
+ assert assignment.call_count == 1
+ builder.new_commit().commit(messages)
+ assert _rows(table) == [dict(id=1, value=11, name='a')]
+
+
[email protected]('use_native', _ENGINES)
[email protected]('restricted_side,policy', [
+ ('source', 'whole'), ('source', 'denied'), ('source', 'filter'),
('source', 'mask'),
+ ('target', 'denied'), ('target', 'filter'), ('target', 'mask'),
+])
+def test_merge_keeps_source_and_target_grants_separate(rest_catalog,
use_native, restricted_side, policy):
+ catalog, server = rest_catalog
+ table = _table(catalog, use_native)
+ catalog.create_table('default.source', Schema.from_pyarrow_schema(_SCHEMA,
options={
+ 'query-auth.enabled': 'true', 'write.native.enabled': 'false',
'commit.native.enabled': 'false',
+ 'read.native.enabled': 'false', 'scan.native-plan.enabled': 'false',
+ }), False)
+ source = catalog.get_table('default.source')
+ _append(source, _ROWS[:1])
+ assignment = Mock(side_effect=lambda batch: pc.add(batch['s.value'], 1))
+ before = _files(table)
+ original = server._table_query_auth_handle
+ guarded = source if restricted_side == 'source' else table
+
+ def authorization(method, data, identifier):
+ if policy == 'denied' and identifier == guarded.identifier:
+ return server._mock_response(ErrorResponse('TABLE',
restricted_side, 'SELECT denied', 403), 403)
+ return original(method, data, identifier)
+
+ builder = table.new_batch_write_builder()
+ with ExitStack() as stack:
+ if policy in ('filter', 'mask'):
+ stack.enter_context(_restricted(server, guarded, policy))
+ authorize = stack.enter_context(
+ patch.object(server, '_table_query_auth_handle',
side_effect=authorization))
+ _native_only(stack, use_native)
+ if policy == 'whole':
+ messages = builder.new_update().merge_into(
+ source, on=['id'], when_matched=[WhenMatched.update({'value':
assignment})])
+ assert messages
+ else:
+ with pytest.raises(Exception,
match='[Pp]ermission|denied|unrestricted|restrict|[Mm]ask|query.auth'):
+ builder.new_update().merge_into(
+ source, on=['id'],
when_matched=[WhenMatched.update({'value': assignment})])
+ assignment.assert_not_called()
+ assert _files(table) == before
+ assert any(call.args[2] == guarded.identifier for call in
authorize.call_args_list)
+ if policy == 'whole':
+ assert assignment.call_count == 1
+ builder.new_commit().commit(messages)
+ assert _rows(table) == [dict(id=1, value=11, name='a')] + _ROWS[1:]
+
+
[email protected]('use_native', _ENGINES)
+def
test_authorized_update_keeps_physical_positions_after_deletion_vectors(rest_catalog,
use_native):
+ catalog, _ = rest_catalog
+ table = _table(catalog, use_native)
+ # Rows 1 and 2 share an anchor; deleting the first physical slot must not
+ # shift the second slot when a column file is subsequently materialized.
+ for operation in ('delete', 'update'):
+ builder = table.new_batch_write_builder()
+ with ExitStack() as stack:
+ _native_only(stack, use_native)
+ update = builder.new_update()
+ messages = (update.delete_by_row_id([0]) if operation == 'delete'
else
+
update.update_by_arrow_with_row_id(pa.table({'_ROW_ID': [1], 'value': [22]})))
+ builder.new_commit().commit(messages)
+ assert _rows(table) == [dict(id=2, value=22, name='b'), _ROWS[2]]
+
+
+def _blob_table(catalog, use_native):
+ from pypaimon.schema.data_types import AtomicType, DataField,
PyarrowFieldParser
+
+ fields = Schema.from_pyarrow_schema(_SCHEMA).fields + [DataField(3,
'payload', AtomicType('BLOB'))]
+ catalog.create_table('default.t', Schema(fields, options={
+ 'query-auth.enabled': 'true', 'row-tracking.enabled': 'true',
'data-evolution.enabled': 'true',
+ 'write.native.enabled': 'false', 'commit.native.enabled': 'false',
+ 'read.native.enabled': 'false', 'scan.native-plan.enabled': 'false',
+ 'blob.target-file-size': '32 B',
+ }), False)
+ table = catalog.get_table('default.t')
+ schema = PyarrowFieldParser.from_paimon_schema(fields)
+ builder = table.new_batch_write_builder()
+ writer, commit = builder.new_write(), builder.new_commit()
+ original = [dict(row, payload=payload) for row, payload in zip(_ROWS,
[b'a' * 40, b'b' * 60, None])]
+ try:
+ writer.write_arrow(pa.Table.from_pylist(original, schema=schema))
+ commit.commit(writer.prepare_commit())
+ finally:
+ writer.close()
+ commit.close()
+ return table.copy({'write.native.enabled': str(use_native).lower()}),
original
+
+
[email protected]('use_native', _ENGINES)
[email protected]('column', ['value', 'payload'])
[email protected]('policy', ['whole', 'mask'])
+def test_authorized_blob_row_id_updates_preserve_untouched_payload_references(
+ rest_catalog, use_native, column, policy):
+ from pypaimon.table.row.blob import BlobDescriptor
+
+ catalog, server = rest_catalog
+ table, original = _blob_table(catalog, use_native)
+ descriptors = _rows(table.copy({'blob-as-descriptor': 'true'}))
+ assert BlobDescriptor.is_blob_descriptor(descriptors[0]['payload'])
+ assert BlobDescriptor.is_blob_descriptor(descriptors[1]['payload'])
+ data = pa.table({'_ROW_ID': [0], column: pa.array(
+ [b'changed' * 15] if column == 'payload' else [111],
+ type=pa.large_binary() if column == 'payload' else pa.int32())})
+ builder = table.new_batch_write_builder()
+ before = _files(table)
+ with ExitStack() as stack:
+ if policy == 'mask':
+ stack.enter_context(_restricted(server, table, policy))
+ _native_only(stack, use_native)
+ if policy == 'mask':
+ with pytest.raises(Exception,
match='[Pp]ermission|denied|unrestricted|restrict|[Mm]ask|query.auth'):
+ builder.new_update().update_by_arrow_with_row_id(data)
+ assert _files(table) == before
+ return
+ messages = builder.new_update().update_by_arrow_with_row_id(data)
+ builder.new_commit().commit(messages)
+ expected = [dict(row) for row in original]
+ expected[0][column] = b'changed' * 15 if column == 'payload' else 111
+ assert _rows(table) == expected
+ after = _rows(table.copy({'blob-as-descriptor': 'true'}))
+ assert after[1]['payload'] == descriptors[1]['payload']
+ if column == 'value':
+ assert [row['payload'] for row in after] == [row['payload'] for row in
descriptors]
+ else:
+ assert after[0]['payload'] != descriptors[0]['payload']
+
+
+def _merge_source_view(catalog, view):
+ catalog.create_table('default.source', Schema.from_pyarrow_schema(_SCHEMA,
options={
+ 'query-auth.enabled': 'true', 'write.native.enabled': 'false',
'commit.native.enabled': 'false',
+ 'read.native.enabled': 'false', 'scan.native-plan.enabled': 'false',
+ }), False)
+ source = catalog.get_table('default.source')
+ captured_empty = source._copy_with_snapshot(None)
+ _append(source, [dict(id=1, value=11, name='source-old')])
+ first = source.snapshot_manager().get_latest_snapshot()
+ if view == 'snapshot-id':
+ selected = source.copy({'scan.snapshot-id': str(first.id)})
+ elif view == 'captured':
+ selected = source._copy_with_snapshot(first)
+ elif view == 'captured-empty':
+ selected = captured_empty
+ else:
+ assert view == 'loaded-branch'
+ catalog.create_tag(source.identifier, 'first', first.id)
+ catalog.create_branch(source.identifier, 'dev', tag_name='first')
+ selected = catalog.get_table('default.source$branch_dev')
+ _append(source, [dict(id=4, value=40, name='source-latest')])
+ return selected
+
+
[email protected]('use_native', _ENGINES)
[email protected]('view', ['snapshot-id', 'captured', 'captured-empty',
'loaded-branch'])
+def test_authorized_merge_keeps_selected_source_view(rest_catalog, use_native,
view):
+ from urllib.parse import unquote
+
+ catalog, server = rest_catalog
+ table = _table(catalog, use_native)
+ source = _merge_source_view(catalog, view)
+ builder = table.new_batch_write_builder()
+ with patch.object(server, '_route_request', wraps=server._route_request)
as requests:
+ with ExitStack() as stack:
+ _native_only(stack, use_native)
+ messages = builder.new_update().merge_into(
+ source, on=['id'], when_matched=[WhenMatched.update({'value':
source_col('value')})],
+ when_not_matched=[WhenNotMatched('*')])
+ expected_path = '/source$branch_dev/auth' if view == 'loaded-branch'
else '/source/auth'
+ assert any(unquote(call.args[1]).endswith(expected_path) for call in
requests.call_args_list)
+ if view == 'captured-empty':
+ assert messages == []
+ assert _rows(table) == _ROWS
+ else:
+ assert messages
+ builder.new_commit().commit(messages)
+ assert _rows(table) == [dict(id=1, value=11, name='a')] + _ROWS[1:]
+
+
[email protected]('use_native', _ENGINES)
[email protected]('policy', ['denied', 'mask'])
+def test_authorized_merge_branch_source_checks_its_own_grant(rest_catalog,
use_native, policy):
+ from urllib.parse import unquote
+
+ catalog, server = rest_catalog
+ table = _table(catalog, use_native)
+ source = _merge_source_view(catalog, 'loaded-branch')
+ before = _files(table)
+ assignment = Mock(side_effect=AssertionError('branch source was read
without authorization'))
+ original = server._table_query_auth_handle
+
+ def authorize(method, data, identifier):
+ # RESTCatalogServer resolves branch metadata before passing the base
+ # identifier here. Verify the branch-bearing HTTP path separately.
+ if identifier.get_full_name() == 'default.source':
+ if policy == 'denied':
+ return server._mock_response(ErrorResponse('TABLE', 'source',
'SELECT denied', 403), 403)
+ return
server._mock_response(AuthTableQueryResponse(column_masking={
+ 'name': json.dumps({'name': 'CONCAT', 'inputs':
['masked']})}), 200)
+ return original(method, data, identifier)
+
+ with patch.object(server, '_table_query_auth_handle',
side_effect=authorize), \
+ patch.object(server, '_route_request',
wraps=server._route_request) as requests:
+ with ExitStack() as stack:
+ _native_only(stack, use_native)
+ with pytest.raises(Exception,
match='[Pp]ermission|denied|unrestricted|restrict|[Mm]ask|query.auth'):
+ table.new_batch_write_builder().new_update().merge_into(
+ source, on=['id'],
when_matched=[WhenMatched.update({'value': assignment})])
+ assert any(unquote(call.args[1]).endswith('/source$branch_dev/auth')
+ for call in requests.call_args_list)
+ assignment.assert_not_called()
+ assert _files(table) == before
diff --git a/paimon-python/pypaimon/tests/native_commit_semantics_test.py
b/paimon-python/pypaimon/tests/native_commit_semantics_test.py
new file mode 100644
index 0000000000..8ee8e12c6c
--- /dev/null
+++ b/paimon-python/pypaimon/tests/native_commit_semantics_test.py
@@ -0,0 +1,244 @@
+# 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.
+
+"""REST commits preserve Java's empty-overwrite and changelog-only
semantics."""
+
+from contextlib import contextmanager
+from unittest.mock import Mock, patch
+
+import pyarrow as pa
+import pytest
+
+from pypaimon import Schema
+from pypaimon.manifest.manifest_file_manager import ManifestFileManager
+from pypaimon.manifest.manifest_list_manager import ManifestListManager
+from pypaimon.read.native_plan import native_runtime_available
+from pypaimon.snapshot.snapshot import BATCH_COMMIT_IDENTIFIER
+from pypaimon.tests import native_plan_rest_test
+from pypaimon.write.commit_message import CommitMessage
+
+
+rest_catalog = native_plan_rest_test.rest_catalog
+pytestmark = [pytest.mark.python_write, pytest.mark.python_commit]
+_ENGINES = [False, pytest.param(True, marks=pytest.mark.skipif(
+ not native_runtime_available(), reason='pypaimon-rust not installed'))]
+_SCHEMA = pa.schema([('id', pa.int64()), ('pt', pa.string()), ('region',
pa.string())])
+_ROWS = [dict(id=1, pt='a', region='us'), dict(id=2, pt='b', region='eu')]
+
+
+def _table(catalog, use_native, ignore_empty, primary_key=False):
+ catalog.create_table('default.t', Schema.from_pyarrow_schema(
+ _SCHEMA, primary_keys=['id'] if primary_key else [],
+ partition_keys=[] if primary_key else ['pt', 'region'], options={
+ 'query-auth.enabled': 'true', 'bucket': '1' if primary_key else
'-1',
+ 'changelog-producer': 'input' if primary_key else 'none',
+ 'write.native.enabled': 'false', 'read.native.enabled': 'false',
+ 'scan.native-plan.enabled': 'false',
+ 'commit.native.enabled': str(use_native).lower(),
+ 'snapshot.ignore-empty-commit': str(ignore_empty).lower(),
+ 'commit.min-retry-wait': '0 ms', 'commit.max-retry-wait': '0 ms',
+ }), False)
+ return catalog.get_table('default.t')
+
+
+def _prepare(builder, rows, identifier=None):
+ writer = builder.new_write()
+ try:
+ writer.write_arrow(pa.Table.from_pylist(rows, schema=_SCHEMA))
+ return writer.prepare_commit() if identifier is None else
writer.prepare_commit(identifier)
+ finally:
+ writer.close()
+
+
+@contextmanager
+def _publication(server, commit, use_native, lose_response=False):
+ original = server._table_commit_handle
+ attempts = []
+
+ def publish(*args, **kwargs):
+ response = original(*args, **kwargs)
+ attempts.append(response[1])
+ assert response[1] == 200
+ # The snapshot is published but the client cannot decode its response.
+ return ('not-json', 200) if lose_response else response
+
+ with patch.object(server, '_table_commit_handle', side_effect=publish), \
+ patch.object(server, '_table_query_auth_handle',
+ side_effect=AssertionError('publication must not
request SELECT')), \
+ patch.object(commit.file_store_commit, 'abort',
side_effect=AssertionError('implicit abort')):
+ if use_native:
+ with patch.object(commit.file_store_commit, '_try_commit',
+ side_effect=AssertionError('native commit fell
back')):
+ yield attempts
+ else:
+ yield attempts
+
+
[email protected]('use_native', _ENGINES)
[email protected]('ignore_empty', [False, True])
[email protected]('target', ['empty-table', 'empty-partition',
'missing-partition', 'partial-spec'])
+def test_truncate_records_empty_overwrite_and_notifies(rest_catalog,
use_native, ignore_empty, target):
+ catalog, server = rest_catalog
+ table = _table(catalog, use_native, ignore_empty)
+ seeded = target in ('missing-partition', 'partial-spec')
+ if seeded:
+ builder = table.copy({'commit.native.enabled':
'false'}).new_batch_write_builder()
+ seed_commit = builder.new_commit()
+ try:
+ seed_commit.commit(_prepare(builder, _ROWS))
+ finally:
+ seed_commit.close()
+ commit = table.new_batch_write_builder().new_commit()
+ callback = Mock()
+ commit.add_commit_callback(callback)
+ try:
+ with _publication(server, commit, use_native) as attempts:
+ if target == 'empty-table':
+ commit.truncate_table()
+ else:
+ spec = {'pt': 'absent'}
+ if target != 'partial-spec':
+ spec['region'] = 'us'
+ commit.truncate_partitions([spec])
+ assert attempts == [200]
+ snapshot = table.snapshot_manager().get_latest_snapshot()
+ assert snapshot.id == (2 if seeded else 1)
+ assert snapshot.commit_kind == 'OVERWRITE'
+ assert snapshot.commit_identifier == BATCH_COMMIT_IDENTIFIER
+ assert snapshot.total_record_count == (2 if seeded else 0)
+ assert snapshot.delta_record_count == 0
+ callback.call.assert_called_once()
+ context = callback.call.call_args[0][0]
+ assert context.snapshot == snapshot
+ assert context.identifier == BATCH_COMMIT_IDENTIFIER
+ assert context.commit_entries == []
+ finally:
+ commit.close()
+ callback.close.assert_called_once()
+
+
[email protected]('use_native', _ENGINES)
[email protected]('ignore_empty', [False, True])
[email protected]('stream', [False, True])
[email protected]('lose_response', [False, True])
+def test_changelog_only_commit_keeps_snapshot_and_recovers_once(
+ rest_catalog, use_native, ignore_empty, stream, lose_response):
+ catalog, server = rest_catalog
+ table = _table(catalog, use_native, ignore_empty, primary_key=True)
+ builder = table.new_stream_write_builder() if stream else
table.new_batch_write_builder()
+ prepared = _prepare(builder, _ROWS[:1], 7 if stream else None)
+ assert sum(len(message.changelog_files) for message in prepared) == 1
+ changes = [CommitMessage(message.partition, message.bucket, [],
+ changelog_files=message.changelog_files) for
message in prepared]
+ paths = [file.file_path for message in prepared for file in
message.new_files + message.changelog_files]
+ commit = builder.new_commit()
+ callback = Mock()
+ commit.add_commit_callback(callback)
+ try:
+ with _publication(server, commit, use_native, lose_response) as
attempts:
+ args = (changes, 7) if stream else (changes,)
+ commit.commit(*args, snapshot_properties={'source':
'changelog-only'})
+ assert attempts == [200], 'lost response recovery must not publish a
second snapshot'
+ snapshot = table.snapshot_manager().get_latest_snapshot()
+ assert snapshot.id == 1
+ assert snapshot.commit_kind == 'APPEND'
+ assert snapshot.commit_identifier == (7 if stream else
BATCH_COMMIT_IDENTIFIER)
+ assert snapshot.total_record_count == snapshot.delta_record_count == 0
+ assert snapshot.changelog_record_count == 1
+ assert snapshot.changelog_manifest_list
+ assert snapshot.properties == {'source': 'changelog-only'}
+ manifests = ManifestListManager(table).read_changelog(snapshot)
+ entries = [entry for manifest in manifests for entry in
ManifestFileManager(table).read(
+ manifest.file_name, drop_stats=False,
file_size=manifest.file_size)]
+ assert [entry.file.file_name for entry in entries] == [
+ file.file_name for message in changes for file in
message.changelog_files]
+ callback.call.assert_called_once()
+ context = callback.call.call_args[0][0]
+ assert context.snapshot == snapshot
+ assert context.identifier == snapshot.commit_identifier
+ assert context.commit_entries == [], 'changelogs are not data-file
deltas'
+ assert all(table.file_io.exists(path) for path in paths)
+
+ if stream:
+ with _publication(server, commit, use_native) as empty_attempts:
+ commit.commit([], 8)
+ latest = table.snapshot_manager().get_latest_snapshot()
+ if ignore_empty:
+ assert empty_attempts == []
+ assert latest.id == 1
+ callback.call.assert_called_once()
+ else:
+ assert empty_attempts == [200]
+ assert latest.id == 2
+ assert latest.commit_identifier == 8
+ assert latest.properties is None
+ assert latest.changelog_manifest_list is None
+ assert latest.changelog_record_count is None
+ assert callback.call.call_count == 2
+ else:
+ with pytest.raises(RuntimeError, match='one-time'):
+ commit.commit([])
+ assert table.snapshot_manager().get_latest_snapshot().id == 1
+ finally:
+ commit.close()
+ callback.close.assert_called_once()
+
+
[email protected]('use_native', _ENGINES)
[email protected]('operation', ['truncate-table',
'truncate-partitions', 'changelog-only'])
+def
test_zero_data_snapshot_callback_error_preserves_confirmed_publication(rest_catalog,
use_native, operation):
+ catalog, server = rest_catalog
+ table = _table(catalog, use_native, True, primary_key=operation ==
'changelog-only')
+ builder = table.new_batch_write_builder()
+ paths = []
+ changes = []
+ if operation == 'changelog-only':
+ prepared = _prepare(builder, _ROWS[:1])
+ changes = [CommitMessage(message.partition, message.bucket, [],
+ changelog_files=message.changelog_files) for
message in prepared]
+ assert sum(len(message.changelog_files) for message in changes) == 1
+ paths = [file.file_path for message in prepared for file in
message.new_files + message.changelog_files]
+ commit = builder.new_commit()
+ failure = LookupError('notification failed after publication')
+ callback = Mock()
+ callback.call.side_effect = failure
+ follower = Mock()
+ commit.add_commit_callback(callback)
+ commit.add_commit_callback(follower)
+ try:
+ with _publication(server, commit, use_native, lose_response=True) as
attempts:
+ with pytest.raises(LookupError) as error:
+ if operation == 'truncate-table':
+ commit.truncate_table()
+ elif operation == 'truncate-partitions':
+ commit.truncate_partitions([{'pt': 'absent'}])
+ else:
+ commit.commit(changes)
+ assert error.value is failure
+ assert attempts == [200]
+ snapshot = table.snapshot_manager().get_latest_snapshot()
+ assert snapshot.id == 1
+ assert snapshot.commit_kind == ('APPEND' if operation ==
'changelog-only' else 'OVERWRITE')
+ assert snapshot.total_record_count == snapshot.delta_record_count == 0
+ callback.call.assert_called_once()
+ assert callback.call.call_args[0][0].snapshot == snapshot
+ follower.call.assert_not_called()
+ assert all(table.file_io.exists(path) for path in paths)
+ finally:
+ commit.close()
+ callback.close.assert_called_once()
+ follower.close.assert_called_once()
diff --git a/paimon-python/pypaimon/tests/native_overwrite_indexes_test.py
b/paimon-python/pypaimon/tests/native_overwrite_indexes_test.py
new file mode 100644
index 0000000000..2e228de3f3
--- /dev/null
+++ b/paimon-python/pypaimon/tests/native_overwrite_indexes_test.py
@@ -0,0 +1,357 @@
+# 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.
+
+"""Overwrite retires matching index metadata without deleting historical
files."""
+
+import json
+from contextlib import ExitStack
+from copy import deepcopy
+from unittest.mock import Mock, patch
+
+import pyarrow as pa
+import pytest
+
+from pypaimon import Schema
+from pypaimon.globalindex.create_global_index import GlobalIndexBuilder
+from pypaimon.globalindex.drop_global_index import GlobalIndexDropper
+from pypaimon.index.index_file_handler import IndexFileHandler
+from pypaimon.read.native_plan import native_runtime_available
+from pypaimon.tests import native_plan_rest_test
+from pypaimon.tests.native_commit_semantics_test import _publication
+
+
+rest_catalog = native_plan_rest_test.rest_catalog
+pytestmark = [pytest.mark.python_write, pytest.mark.python_commit]
+_ENGINES = [False, pytest.param(True, marks=pytest.mark.skipif(
+ not native_runtime_available(), reason='pypaimon-rust not installed'))]
+_SCHEMA = pa.schema([('id', pa.int64()), ('pt', pa.string()), ('region',
pa.string())])
+_ROWS = [dict(id=1, pt='a', region='us'), dict(id=2, pt='a', region='us'),
+ dict(id=3, pt='b', region='eu'), dict(id=4, pt='b', region='eu')]
+
+
+def _write(table, rows):
+ builder = table.new_batch_write_builder()
+ writer, commit = builder.new_write(), builder.new_commit()
+ try:
+ writer.write_arrow(pa.Table.from_pylist(rows, schema=_SCHEMA))
+ commit.commit(writer.prepare_commit())
+ finally:
+ writer.close()
+ commit.close()
+
+
+def _indexed_table(catalog, index_type):
+ options = {
+ 'bucket': '-1', 'dynamic-bucket.target-row-num': '100',
+ 'dynamic-partition-overwrite': 'false',
+ 'write.native.enabled': 'false', 'commit.native.enabled': 'false',
+ 'read.native.enabled': 'false', 'scan.native-plan.enabled': 'false',
+ 'commit.min-retry-wait': '0 ms', 'commit.max-retry-wait': '0 ms',
+ }
+ if index_type != 'HASH':
+ options.update({'data-evolution.enabled': 'true',
'row-tracking.enabled': 'true',
+ 'deletion-vectors.enabled': 'true' if index_type ==
'DELETION_VECTORS' else 'false'})
+ catalog.create_table('default.t', Schema.from_pyarrow_schema(
+ _SCHEMA, partition_keys=['pt', 'region'],
+ primary_keys=['id', 'pt', 'region'] if index_type == 'HASH' else [],
options=options), False)
+ table = catalog.get_table('default.t')
+ _write(table, _ROWS)
+ if index_type == 'DELETION_VECTORS':
+ _delete(table, [0, 2])
+ elif index_type == 'btree':
+ assert table.create_global_index('id') == 2
+ entries =
IndexFileHandler(table).scan(table.snapshot_manager().get_latest_snapshot())
+ assert {entry.index_file.index_type for entry in entries} == {index_type}
+ assert {entry.partition.values[0] for entry in entries} == {'a', 'b'}
+ return table
+
+
+def _delete(table, row_ids):
+ builder = table.new_batch_write_builder()
+ commit = builder.new_commit()
+ try:
+ commit.commit(builder.new_update().delete_by_row_id(row_ids))
+ finally:
+ commit.close()
+
+
+def _rows(table, snapshot=None):
+ if snapshot is not None:
+ table = table._copy_with_snapshot(snapshot)
+ builder = table.new_read_builder()
+ return
builder.new_read().to_arrow(builder.new_scan().plan().splits()).sort_by('id').to_pylist()
+
+
+def _index_paths(table, entries):
+ factory = table.path_factory()
+ return [entry.index_file.external_path or (
+ factory.bucket_index_path(tuple(entry.partition.values), entry.bucket,
entry.index_file)
+ if entry.index_file.index_type in ('HASH', 'DELETION_VECTORS') else
+
factory.global_index_path_factory().to_path(entry.index_file.file_name)) for
entry in entries]
+
+
+def _native_only(commit, use_native):
+ return (patch.object(commit.file_store_commit, '_try_commit',
side_effect=AssertionError('Python fallback'))
+ if use_native else ExitStack())
+
+
[email protected]('use_native', _ENGINES)
[email protected]('index_type', ['HASH', 'DELETION_VECTORS', 'btree'])
[email protected]('operation', ['truncate-table',
'truncate-partitions', 'overwrite'])
+def test_overwrite_retires_only_matching_index_metadata(rest_catalog,
use_native, index_type, operation):
+ catalog, _ = rest_catalog
+ table = _indexed_table(catalog, index_type)
+ before = table.snapshot_manager().get_latest_snapshot()
+ original_rows = _rows(table)
+ original_indexes = IndexFileHandler(table).scan(before)
+ paths = _index_paths(table, original_indexes)
+ table = table.copy({'commit.native.enabled': str(use_native).lower()})
+ builder = table.new_batch_write_builder()
+ if operation == 'overwrite':
+ builder.overwrite({'pt': 'a'})
+ commit = builder.new_commit()
+ writer = None
+ try:
+ with _native_only(commit, use_native):
+ if operation == 'truncate-table':
+ commit.truncate_table()
+ elif operation == 'truncate-partitions':
+ commit.truncate_partitions([{'pt': 'a'}])
+ else:
+ writer = builder.new_write()
+ writer.write_arrow(pa.Table.from_pylist([dict(id=10, pt='a',
region='us')], schema=_SCHEMA))
+ commit.commit(writer.prepare_commit())
+ latest = table.snapshot_manager().get_latest_snapshot()
+ remaining = IndexFileHandler(table).scan(latest)
+ old_names = {entry.index_file.file_name for entry in original_indexes
if entry.partition.values[0] == 'a'}
+ assert not old_names.intersection(entry.index_file.file_name for entry
in remaining)
+ if operation == 'truncate-table':
+ assert latest.index_manifest is None
+ expected = []
+ else:
+ retained_names = {entry.index_file.file_name for entry in
remaining if entry.partition.values[0] == 'b'}
+ assert retained_names == {entry.index_file.file_name for entry in
original_indexes
+ if entry.partition.values[0] == 'b'}
+ expected = [row for row in original_rows if row['pt'] == 'b']
+ if operation == 'overwrite':
+ expected = [dict(id=10, pt='a', region='us')] + expected
+ assert _rows(table) == sorted(expected, key=lambda row: row['id'])
+ assert _rows(table, before) == original_rows
+ assert all(table.file_io.exists(path) for path in paths)
+ assert table.file_io.exists(table.table_path.rstrip('/') +
'/manifest/' + before.index_manifest)
+ finally:
+ if writer is not None:
+ writer.close()
+ commit.close()
+
+
[email protected]('use_native', _ENGINES)
[email protected]('index_type', ['HASH', 'DELETION_VECTORS'])
+def
test_truncate_retry_replans_indexes_added_by_concurrent_writer(rest_catalog,
use_native, index_type):
+ catalog, server = rest_catalog
+ table = _indexed_table(catalog, index_type)
+ original = table.snapshot_manager().get_latest_snapshot()
+ original_rows = _rows(table)
+ builder = table.copy({'commit.native.enabled':
str(use_native).lower()}).new_batch_write_builder()
+ commit = builder.new_commit()
+ publish = server._table_commit_handle
+ concurrent = []
+ attempts = []
+
+ def change_before_first_compare_and_swap(method, data, identifier,
branch=None):
+ request = json.loads(data)
+ if request['snapshot']['commitUser'] == builder.commit_user:
+ attempts.append(request['snapshot']['id'])
+ if len(attempts) == 1:
+ # A direct storage writer runs inside the single-threaded mock
+ # REST server, avoiding a recursive HTTP request to that
server.
+ storage = server._get_file_table(identifier, branch)
+ if index_type == 'HASH':
+ _write(storage, [dict(id=5, pt='a', region='us')])
+ else:
+ _delete(storage, [1])
+ snapshot = storage.snapshot_manager().get_latest_snapshot()
+ concurrent.extend(IndexFileHandler(storage).scan(snapshot))
+ return publish(method, data, identifier, branch)
+
+ try:
+ with patch.object(server, '_table_commit_handle',
side_effect=change_before_first_compare_and_swap), \
+ _native_only(commit, use_native):
+ commit.truncate_partitions([{'pt': 'a'}])
+ assert attempts == [original.id + 1, original.id + 2]
+ latest = table.snapshot_manager().get_latest_snapshot()
+ remaining = IndexFileHandler(table).scan(latest)
+ assert {entry.partition.values[0] for entry in remaining} == {'b'}
+ assert {entry.index_file.file_name for entry in remaining} == {
+ entry.index_file.file_name for entry in concurrent if
entry.partition.values[0] == 'b'}
+ assert all(table.file_io.exists(path) for path in _index_paths(table,
concurrent))
+ assert _rows(table) == [row for row in original_rows if row['pt'] ==
'b']
+ assert _rows(table, original) == original_rows
+ finally:
+ commit.close()
+
+
[email protected]('use_native', _ENGINES)
[email protected]('index_change', ['add', 'delete'])
[email protected]('with_data', [False, True])
+def test_dynamic_overwrite_targets_only_partitions_with_data_changes(
+ rest_catalog, use_native, index_change, with_data):
+ catalog, _ = rest_catalog
+ table = _indexed_table(catalog, 'btree')
+ before = table.snapshot_manager().get_latest_snapshot()
+ before_indexes = IndexFileHandler(table).scan(before)
+ if index_change == 'delete':
+ index_messages = GlobalIndexDropper(table, 'id', partitions={'pt':
'b'}).build()
+ else:
+ # Prepared index files are legitimate outputs of the existing builder;
+ # an otherwise empty dynamic overwrite must leave them uncommitted.
+ index_messages = GlobalIndexBuilder(
+ table, 'id', index_type='bitmap', partitions={'pt': 'b'}).build()
+ assert index_messages and all(not message.new_files for message in
index_messages)
+ index_paths = _index_paths(
+ table, [entry for message in index_messages for entry in
message.index_adds + message.index_deletes])
+ table = table.copy({'dynamic-partition-overwrite': 'true',
+ 'commit.native.enabled': str(use_native).lower()})
+ builder = table.new_batch_write_builder().overwrite()
+ writer, commit = builder.new_write(), builder.new_commit()
+ callback = Mock()
+ commit.add_commit_callback(callback)
+ try:
+ messages = []
+ if with_data:
+ writer.write_arrow(pa.Table.from_pylist([dict(id=10, pt='a',
region='us')], schema=_SCHEMA))
+ messages.extend(writer.prepare_commit())
+ messages.extend(index_messages)
+ with _native_only(commit, use_native):
+ commit.commit(messages)
+ latest = table.snapshot_manager().get_latest_snapshot()
+ if with_data:
+ assert latest.id == before.id + 1
+ assert _rows(table) == [dict(id=3, pt='b', region='eu'),
dict(id=4, pt='b', region='eu'),
+ dict(id=10, pt='a', region='us')]
+ callback.call.assert_called_once()
+ assert all(tuple(entry.partition.values) == ('a', 'us')
+ for entry in
callback.call.call_args[0][0].commit_entries)
+ remaining = IndexFileHandler(table).scan(latest)
+ if index_change == 'delete':
+ assert remaining == []
+ else:
+ assert {entry.index_file.index_type for entry in remaining} ==
{'btree', 'bitmap'}
+ assert {entry.partition.values[0] for entry in remaining} ==
{'b'}
+ else:
+ assert latest == before
+ assert IndexFileHandler(table).scan(latest) == before_indexes
+ assert _rows(table) == _ROWS
+ callback.call.assert_not_called()
+ assert all(table.file_io.exists(path) for path in index_paths)
+ assert _rows(table, before) == _ROWS
+ finally:
+ writer.close()
+ commit.close()
+
+
[email protected]('use_native', _ENGINES)
[email protected]('mismatch', ['missing', 'duplicate', 'type',
'partition', 'bucket', 'size', 'row-count',
+ 'external-path', 'row-range',
'field-id', 'extra-fields', 'empty-extra-fields',
+ 'index-meta', 'source-meta'])
+def test_stale_global_index_delete_fails_before_publication(rest_catalog,
use_native, mismatch):
+ catalog, _ = rest_catalog
+ table = _indexed_table(catalog, 'btree')
+ changes = GlobalIndexDropper(table, 'id', partitions={'pt': 'a'}).build()
+ paths = _index_paths(table, [entry for message in changes for entry in
message.index_deletes])
+ if mismatch == 'missing':
+ first = table.new_batch_write_builder().new_commit()
+ try:
+ first.commit(deepcopy(changes))
+ finally:
+ first.close()
+ else:
+ entry = changes[0].index_deletes[0]
+ meta = entry.index_file
+ if mismatch == 'duplicate':
+ changes[0].index_deletes.append(deepcopy(entry))
+ elif mismatch == 'type':
+ meta.index_type = 'bitmap'
+ elif mismatch == 'partition':
+ entry.partition = deepcopy(entry.partition)
+ entry.partition.values[0] = 'b'
+ changes[0].partition = tuple(entry.partition.values)
+ elif mismatch == 'bucket':
+ entry.bucket += 1
+ changes[0].bucket = entry.bucket
+ elif mismatch == 'size':
+ meta.file_size += 1
+ elif mismatch == 'row-count':
+ meta.row_count += 1
+ elif mismatch == 'external-path':
+ meta.external_path = paths[0] + '.other'
+ elif mismatch == 'row-range':
+ meta.global_index_meta.row_range_start += 1
+ elif mismatch == 'field-id':
+ meta.global_index_meta.index_field_id += 1
+ elif mismatch == 'extra-fields':
+ meta.global_index_meta.extra_field_ids = [1]
+ elif mismatch == 'empty-extra-fields':
+ meta.global_index_meta.extra_field_ids = []
+ elif mismatch == 'index-meta':
+ meta.global_index_meta.index_meta = b'different'
+ elif mismatch == 'source-meta':
+ meta.global_index_meta.source_meta = b'different'
+ before = table.snapshot_manager().get_latest_snapshot()
+ before_indexes = IndexFileHandler(table).scan(before)
+ table = table.copy({'commit.native.enabled': str(use_native).lower()})
+ commit = table.new_batch_write_builder().new_commit()
+ callback = Mock()
+ commit.add_commit_callback(callback)
+ try:
+ with _native_only(commit, use_native), pytest.raises(
+ (RuntimeError, ValueError), match='does not exist|metadata
does not match'):
+ commit.commit(changes)
+ assert table.snapshot_manager().get_latest_snapshot() == before
+ assert IndexFileHandler(table).scan(before) == before_indexes
+ callback.call.assert_not_called()
+ assert all(table.file_io.exists(path) for path in paths)
+ assert _rows(table) == _ROWS
+ finally:
+ commit.close()
+
+
[email protected]('use_native', _ENGINES)
+def
test_global_index_delete_recovers_lost_response_before_rechecking_current_files(rest_catalog,
use_native):
+ catalog, server = rest_catalog
+ table = _indexed_table(catalog, 'btree')
+ before = table.snapshot_manager().get_latest_snapshot()
+ changes = GlobalIndexDropper(table, 'id', partitions={'pt': 'a'}).build()
+ paths = _index_paths(table, [entry for message in changes for entry in
message.index_deletes])
+ commit = table.copy({'commit.native.enabled':
str(use_native).lower()}).new_batch_write_builder().new_commit()
+ callback = Mock()
+ commit.add_commit_callback(callback)
+ try:
+ with _publication(server, commit, use_native, lose_response=True) as
attempts:
+ commit.commit(changes)
+ assert attempts == [200]
+ latest = table.snapshot_manager().get_latest_snapshot()
+ assert latest.id == before.id + 1
+ assert {entry.partition.values[0] for entry in
IndexFileHandler(table).scan(latest)} == {'b'}
+ callback.call.assert_called_once()
+ context = callback.call.call_args[0][0]
+ assert context.snapshot == latest
+ assert context.commit_entries == []
+ assert all(table.file_io.exists(path) for path in paths)
+ assert _rows(table, before) == _rows(table) == _ROWS
+ finally:
+ commit.close()
diff --git a/paimon-python/pypaimon/tests/native_upsert_encoded_keys_test.py
b/paimon-python/pypaimon/tests/native_upsert_encoded_keys_test.py
new file mode 100644
index 0000000000..f6e2e9e8b8
--- /dev/null
+++ b/paimon-python/pypaimon/tests/native_upsert_encoded_keys_test.py
@@ -0,0 +1,228 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Existing upsert keys compare logical values, independently of encoding."""
+
+from contextlib import ExitStack
+from pathlib import Path
+from urllib.parse import unquote, urlparse
+from unittest.mock import patch
+
+import pyarrow as pa
+import pytest
+
+from pypaimon import Schema
+from pypaimon.read.native_plan import native_runtime_available
+from pypaimon.write.native_update import _native_upsert_key_type_compatible
+from pypaimon.write.table_upsert_by_key import TableUpsertByKey
+
+
+pytestmark = [pytest.mark.python_write, pytest.mark.python_commit]
+_ENGINES = [False, pytest.param(True, marks=pytest.mark.skipif(
+ not native_runtime_available(), reason='pypaimon-rust is not installed'))]
+_KINDS = [
+ 'dense_union', 'sparse_union', 'dictionary_dense_union',
'dictionary_sparse_union',
+ pytest.param('dictionary_string_view', marks=pytest.mark.skipif(
+ not hasattr(pa, 'string_view'), reason='PyArrow view arrays are
unavailable')),
+ pytest.param('dictionary_binary_view', marks=pytest.mark.skipif(
+ not hasattr(pa, 'binary_view'), reason='PyArrow view arrays are
unavailable')),
+ pytest.param('dictionary_run_end', marks=pytest.mark.skipif(
+ not hasattr(pa, 'RunEndEncodedArray'), reason='PyArrow run-end arrays
are unavailable')),
+]
+
+
+def _files(table, pattern='*'):
+ root = Path(unquote(urlparse(table.table_path).path))
+ assert root.is_dir(), table.table_path
+ return set(root.rglob(pattern))
+
+
+def _union(dense):
+ # The first logical value overflows INT and the final child never occurs.
+ # Slicing excludes the first value; inactive sparse children still contain
+ # overflows and text which must never participate in key conversion.
+ tags = pa.array([42, 7, 42, 7, 42], type=pa.int8())
+ fields = ['small', 'large', 'unused']
+ if dense:
+ children = [pa.array([0, 1, 2], type=pa.int32()),
+ pa.array([2 ** 63 - 1, 1, None], type=pa.int64()),
pa.array(['unused'])]
+ return pa.UnionArray.from_dense(tags, pa.array([0, 1, 1, 2, 2],
type=pa.int32()),
+ children, field_names=fields,
type_codes=[7, 42, 99])
+ children = [pa.array([0, 1, 0, 2, 0], type=pa.int32()),
+ pa.array([2 ** 63 - 1, 2 ** 63 - 1, 1, 2 ** 63 - 1, None],
type=pa.int64()),
+ pa.array(['unused'] * 5)]
+ return pa.UnionArray.from_sparse(tags, children, field_names=fields,
type_codes=[7, 42, 99])
+
+
+def _keys(kind):
+ if 'union' in kind:
+ values = _union('sparse' not in kind).slice(1, 4)
+ if kind.startswith('dictionary'):
+ values = pa.DictionaryArray.from_arrays(pa.array([0, None, 1, 2],
type=pa.int8()), values)
+ else:
+ values = values.take(pa.array([0, 3, 1, 2], type=pa.int32()))
+ return pa.int32(), values, [1, 2]
+ if kind == 'dictionary_run_end':
+ values = pa.RunEndEncodedArray.from_arrays(
+ pa.array([2, 4, 5, 6], type=pa.int16()), pa.array([999, 1, 2,
None], type=pa.int32()))
+ values = values.slice(2, 4)
+ values = pa.DictionaryArray.from_arrays(pa.array([0, None, 1, 2],
type=pa.int8()), values)
+ return pa.int32(), values, [1, 2]
+ binary = kind == 'dictionary_binary_view'
+ first, second = (b'\xff\0', b'longer-than-inline-view') if binary else
('a', 'longer-than-inline-view')
+ view = pa.binary_view() if binary else pa.string_view()
+ values = pa.array([first, second, None, first], type=view)
+ values = pa.DictionaryArray.from_arrays(pa.array([0, None, 3, 1],
type=pa.int8()), values)
+ return pa.binary() if binary else pa.string(), values, [first, second]
+
+
+def _table(catalog, key_type, keys, native, guarded):
+ schema = pa.schema([('key', key_type), ('p', pa.string()), ('value',
pa.int32()), ('keep', pa.string())])
+ catalog.create_table('default.t', Schema.from_pyarrow_schema(schema,
partition_keys=['p'], options={
+ 'row-tracking.enabled': 'true', 'data-evolution.enabled': 'true',
+ 'query-auth.enabled': str(guarded).lower(), 'write.native.enabled':
'false',
+ 'commit.native.enabled': 'false', 'read.native.enabled': 'false',
'scan.native-plan.enabled': 'false',
+ }), False)
+ table = catalog.get_table('default.t')
+ rows = [dict(key=keys[0], p='a', value=10, keep='first'),
+ dict(key=keys[0], p='a', value=11, keep='duplicate'),
+ dict(key=keys[1], p='a', value=20, keep='second'),
+ dict(key=None, p='a', value=30, keep='null'),
+ dict(key=keys[0], p='b', value=40, keep='other-partition')]
+ builder = table.new_batch_write_builder()
+ writer, commit = builder.new_write(), builder.new_commit()
+ try:
+ writer.write_arrow(pa.Table.from_pylist(rows, schema=schema))
+ commit.commit(writer.prepare_commit())
+ finally:
+ writer.close()
+ commit.close()
+ return table.copy({'write.native.enabled': str(native).lower()})
+
+
+def _native_only(stack, native):
+ if native:
+ stack.enter_context(patch.object(TableUpsertByKey, '_upsert_partition',
+ side_effect=AssertionError('Python
Arrow upsert fallback')))
+ stack.enter_context(patch.object(TableUpsertByKey,
'_upsert_row_partition',
+ side_effect=AssertionError('Python
Row upsert fallback')))
+
+
+def _read(table):
+ read = table.new_read_builder()
+ return
read.new_read().to_arrow(read.new_scan().plan().splits()).sort_by('keep').to_pylist()
+
+
[email protected]('native', _ENGINES)
[email protected]('guarded', [False, True])
[email protected]('stream', [False, True])
[email protected]('kind', _KINDS)
+def test_encoded_upsert_keys_preserve_last_wins_nulls_and_partitions(
+ native_rest_catalog, native, guarded, stream, kind):
+ key_type, keys, seed = _keys(kind)
+ table = _table(native_rest_catalog, key_type, seed, native, guarded)
+ values = pa.array([90, 91, 92, 93], type=pa.int32())
+ # Duplicate source keys cross Arrow batch boundaries, while target keys
+ # fan out to duplicate existing rows within one partition only.
+ source = pa.table({'value': values, 'p': ['a'] * 4, 'key': keys})
+ source = pa.concat_tables([source.slice(0, 2), source.slice(2)])
+ builder = table.new_stream_write_builder() if stream else
table.new_batch_write_builder()
+ with ExitStack() as stack:
+ _native_only(stack, native)
+ update = builder.new_update().with_update_type(['value'])
+ messages = (update.upsert_by_arrow_with_key(source, ['key'], 17) if
stream
+ else update.upsert_by_arrow_with_key(source, ['key']))
+ commit = builder.new_commit()
+ try:
+ commit.commit(messages, 17) if stream else commit.commit(messages)
+ finally:
+ commit.close()
+ assert _read(table) == [
+ dict(key=seed[0], p='a', value=92, keep='duplicate'),
+ dict(key=seed[0], p='a', value=92, keep='first'),
+ dict(key=None, p='a', value=91, keep='null'),
+ dict(key=seed[0], p='b', value=40, keep='other-partition'),
+ dict(key=seed[1], p='a', value=93, keep='second'),
+ ]
+
+
[email protected]('kind', _KINDS)
+def test_native_key_gate_qualifies_existing_encoded_upsert_layouts(kind):
+ key_type, keys, _ = _keys(kind)
+ assert _native_upsert_key_type_compatible(keys.type, key_type)
+ if kind in ('dictionary_string_view', 'dictionary_binary_view',
'dictionary_run_end'):
+ assert not _native_upsert_key_type_compatible(keys.dictionary.type,
key_type), (
+ 'Bare views/run-end arrays have no PyArrow take kernel and remain
outside this bridge')
+
+
[email protected]('native', _ENGINES)
[email protected]('invalid', [2 ** 63 - 1, 1.5, '1'])
+def test_active_union_key_cast_failure_never_matches_or_stages(
+ native_rest_catalog, native, invalid):
+ table = _table(native_rest_catalog, pa.int32(), [1, 2], native, False)
+ child = pa.array([invalid])
+ keys = pa.UnionArray.from_dense(pa.array([7], type=pa.int8()),
pa.array([0], type=pa.int32()),
+ [child, pa.array([1], type=pa.int32())],
type_codes=[7, 42])
+ source = pa.table({'key': keys, 'p': ['a'], 'value': pa.array([99],
type=pa.int32())})
+ before = _files(table, '*.parquet')
+ snapshot = table.snapshot_manager().get_latest_snapshot().id
+ expected = _read(table)
+ with ExitStack() as stack:
+ _native_only(stack, native)
+ with pytest.raises((ValueError, RuntimeError, pa.ArrowInvalid)):
+
table.new_batch_write_builder().new_update().with_update_type(['value']).upsert_by_arrow_with_key(
+ source, ['key'])
+ assert table.snapshot_manager().get_latest_snapshot().id == snapshot
+ assert _files(table, '*.parquet') == before
+ assert _read(table) == expected
+
+
[email protected]('native', _ENGINES)
[email protected]('dense', [False, True])
[email protected]('dictionary', [False, True])
[email protected]('child_type', [pa.string(), pa.struct([('unused',
pa.string())])])
+def test_typed_null_union_children_match_existing_null_keys(
+ native_rest_catalog, native, dense, dictionary, child_type):
+ table = _table(native_rest_catalog, pa.int32(), [1, 2], native, True)
+ tags = pa.array([7, 42], type=pa.int8())
+ if dense:
+ keys = pa.UnionArray.from_dense(
+ tags, pa.array([0, 0], type=pa.int32()),
+ [pa.array([1], type=pa.int32()), pa.nulls(1, type=child_type)],
type_codes=[7, 42])
+ else:
+ inactive = 'not an integer' if pa.types.is_string(child_type) else
{'unused': 'not an integer'}
+ keys = pa.UnionArray.from_sparse(
+ tags, [pa.array([1, 0], type=pa.int32()), pa.array([inactive,
None], type=child_type)],
+ type_codes=[7, 42])
+ if dictionary:
+ keys = pa.DictionaryArray.from_arrays(pa.array([0, 1],
type=pa.int8()), keys)
+ source = pa.table({'key': keys, 'p': ['a', 'a'], 'value': pa.array([90,
91], type=pa.int32())})
+ source = pa.concat_tables([source.slice(0, 1), source.slice(1)])
+ builder = table.new_batch_write_builder()
+ with ExitStack() as stack:
+ _native_only(stack, native)
+ messages =
builder.new_update().with_update_type(['value']).upsert_by_arrow_with_key(source,
['key'])
+ commit = builder.new_commit()
+ try:
+ commit.commit(messages)
+ finally:
+ commit.close()
+ assert _read(table) == [
+ dict(key=1, p='a', value=90, keep='duplicate'), dict(key=1, p='a',
value=90, keep='first'),
+ dict(key=None, p='a', value=91, keep='null'), dict(key=1, p='b',
value=40, keep='other-partition'),
+ dict(key=2, p='a', value=20, keep='second'),
+ ]
diff --git a/paimon-python/pypaimon/tests/native_write_routing_test.py
b/paimon-python/pypaimon/tests/native_write_routing_test.py
index b2871d506f..e8c5534079 100644
--- a/paimon-python/pypaimon/tests/native_write_routing_test.py
+++ b/paimon-python/pypaimon/tests/native_write_routing_test.py
@@ -159,6 +159,21 @@ def
test_pk_complete_write_type_keeps_table_order_and_native_writer(tmp_path, na
assert writer._python_writer is None
[email protected]('full_input', [False, True])
+def test_append_write_type_preserves_input_routing_columns_for_core(tmp_path,
full_input):
+ underlying = Mock()
+ writer = _writer(_table(tmp_path), underlying).with_write_type(['value'])
+ data = pa.Table.from_pylist([{'id': 1, 'pt': 'a', 'value': 'x'}],
schema=SCHEMA)
+ if not full_input:
+ data = data.select(['value'])
+ writer.write_arrow(data)
+ actual, bucket = underlying.write_arrow.call_args[0]
+ assert actual.schema.names == data.schema.names
+ assert actual.to_pylist() == data.to_pylist()
+ assert bucket is None
+ assert writer._python_writer is None
+
+
def
test_standard_rest_query_auth_write_table_keeps_native_publication(rest_catalog):
catalog, _ = rest_catalog
table = _table(None, catalog=catalog, guarded=True)
@@ -178,15 +193,15 @@ def
test_standard_rest_query_auth_write_table_keeps_native_publication(rest_cata
assert schema['options']['query-auth.enabled'] == 'true'
-def test_read_based_updates_keep_existing_guarded_selection(rest_catalog):
+def
test_read_based_updates_delegate_standard_rest_authorization_to_core(rest_catalog):
from pypaimon.write.native_update import _native_row_id_table
catalog, _ = rest_catalog
table = _table(None, 'data-evolution', catalog, use_native=True,
guarded=True)
with patch('pypaimon.write.native_update.native_runtime_available',
return_value=True), \
patch('pypaimon.write.native_update.create_native_write_table') as
create:
- assert _native_row_id_table(table) is None
- create.assert_not_called()
+ assert _native_row_id_table(table) is create.return_value
+ create.assert_called_once_with(table)
def _rows(table):
diff --git a/paimon-python/pypaimon/tests/native_write_type_test.py
b/paimon-python/pypaimon/tests/native_write_type_test.py
new file mode 100644
index 0000000000..561a153dcc
--- /dev/null
+++ b/paimon-python/pypaimon/tests/native_write_type_test.py
@@ -0,0 +1,254 @@
+# 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.
+
+"""Selected append input columns and upstream buckets retain Java semantics."""
+
+from contextlib import ExitStack
+from pathlib import Path
+from urllib.parse import unquote, urlparse
+from unittest.mock import Mock, patch
+
+import pyarrow as pa
+import pytest
+
+from pypaimon import Schema
+from pypaimon.read.native_plan import native_runtime_available
+from pypaimon.table.row.generic_row import GenericRow
+from pypaimon.tests import native_plan_rest_test
+from pypaimon.write.native_write import NativeTableWrite
+
+
+rest_catalog = native_plan_rest_test.rest_catalog
+pytestmark = [pytest.mark.python_write, pytest.mark.python_commit]
+native = pytest.mark.skipif(not native_runtime_available(),
reason='pypaimon-rust not installed')
+_ENGINES = [False, pytest.param(True, marks=native)]
+_SCHEMA = pa.schema([('id', pa.int64()), ('pt', pa.int64()), ('value',
pa.int64())])
+
+
+def _files(table, pattern='*'):
+ root = Path(unquote(urlparse(table.table_path).path))
+ assert root.is_dir(), table.table_path
+ return set(root.rglob(pattern))
+
+
+def _table(catalog, use_native, partitioned=False, bucketed=False, buckets=4,
schema=_SCHEMA):
+ options = {'write.native.enabled': str(use_native).lower(),
'commit.native.enabled': 'false',
+ 'read.native.enabled': 'false', 'scan.native-plan.enabled':
'false'}
+ if bucketed:
+ options.update({'bucket': str(buckets), 'bucket-key': 'id'})
+ catalog.create_table('default.t', Schema.from_pyarrow_schema(
+ schema, partition_keys=['pt'] if partitioned else [],
options=options), False)
+ return catalog.get_table('default.t')
+
+
+def _rows(table):
+ builder = table.new_read_builder()
+ return
builder.new_read().to_arrow(builder.new_scan().plan().splits()).sort_by('value').to_pylist()
+
+
+def _native_only(stack, writer, use_native):
+ assert isinstance(writer, NativeTableWrite) == use_native
+ if use_native:
+ stack.enter_context(patch.object(writer, '_switch_to_python',
side_effect=AssertionError('Python fallback')))
+
+
+def _finish(builder, writer):
+ messages = writer.prepare_commit()
+ commit = builder.new_commit()
+ try:
+ commit.commit(messages)
+ finally:
+ commit.close()
+ return messages
+
+
[email protected]('use_native', _ENGINES)
[email protected]('partitioned', [False, True])
[email protected]('full_input', [False, True])
+def test_upstream_bucket_subset_omits_hash_columns_without_leaking_full_input(
+ rest_catalog, use_native, partitioned, full_input):
+ catalog, _ = rest_catalog
+ table = _table(catalog, use_native, partitioned, bucketed=True)
+ columns = ['value', 'pt'] if partitioned else ['value']
+ data = pa.Table.from_pylist([dict(id=1, pt=7, value=42)], schema=_SCHEMA)
+ data = data if full_input else data.select(columns).to_batches()[0]
+ builder = table.new_batch_write_builder()
+ writer = builder.new_write()
+ try:
+ with ExitStack() as stack:
+ _native_only(stack, writer, use_native)
+ writer.with_write_type(columns)
+ with pytest.raises(ValueError, match='missing.*not in table
schema|Column.*missing'):
+ writer.with_write_type(['missing'])
+ before = _files(table)
+ # Java does not route zero rows, even when the selected input
+ # could only be written with an upstream bucket for nonempty data.
+ writer.write_arrow(data.slice(0, 0))
+ assert _files(table) == before
+ if not full_input:
+ with pytest.raises(ValueError, match="routing column 'id'"):
+ writer.write_arrow(data)
+ assert _files(table) == before
+ writer.write_arrow(data, bucket=2)
+ messages = _finish(builder, writer)
+ assert [message.bucket for message in messages] == [2]
+ assert all(file.write_cols == columns for message in messages for file
in message.new_files)
+ assert _rows(table) == [dict(id=None, pt=7 if partitioned else None,
value=42)]
+ finally:
+ writer.close()
+
+
[email protected]('use_native', _ENGINES)
+def
test_partial_upstream_groups_reject_invalid_buckets_and_mixed_partitions(rest_catalog,
use_native):
+ catalog, _ = rest_catalog
+ table = _table(catalog, use_native, partitioned=True, bucketed=True)
+ builder = table.new_batch_write_builder()
+ writer = builder.new_write()
+ data = pa.table({'value': [42, 43], 'pt': [7, 8]})
+ try:
+ with ExitStack() as stack:
+ _native_only(stack, writer, use_native)
+ writer.with_write_type(['value', 'pt'])
+ before = _files(table)
+ for bucket in (-1, 4):
+ with pytest.raises(ValueError, match='Bucket id'):
+ writer.write_arrow(data, bucket=bucket)
+ with pytest.raises(ValueError, match='multiple partitions'):
+ writer.write_arrow(data, bucket=2)
+ assert _files(table) == before
+ writer.write_arrow(data.slice(0, 1), bucket=2)
+ _finish(builder, writer)
+ assert _rows(table) == [dict(id=None, pt=7, value=42)]
+ finally:
+ writer.close()
+
+
[email protected]('use_native', _ENGINES)
[email protected]('partitioned', [False, True])
[email protected]('full_input', [False, True])
+def
test_complete_reordered_append_type_keeps_field_identity_and_copies_column_names(
+ rest_catalog, use_native, partitioned, full_input):
+ catalog, _ = rest_catalog
+ table = _table(catalog, use_native, partitioned)
+ columns = ['value', 'pt', 'id']
+ expected_columns = list(columns)
+ data = pa.Table.from_pylist([dict(id=1, pt=7, value=42)], schema=_SCHEMA)
+ data = data if full_input else data.select(columns)
+ builder = table.new_batch_write_builder()
+ writer = builder.new_write()
+ try:
+ with ExitStack() as stack:
+ _native_only(stack, writer, use_native)
+ writer.with_write_type(columns)
+ columns.reverse()
+ writer.write_arrow(data)
+ messages = _finish(builder, writer)
+ assert all(file.write_cols == expected_columns for message in messages
for file in message.new_files)
+ assert _rows(table) == [dict(id=1, pt=7, value=42)]
+ finally:
+ writer.close()
+
+
[email protected]('use_native', _ENGINES)
[email protected]('buckets', [1, 4])
[email protected]('columns', [('value', 'pt'), ('value',)])
+def test_full_input_routes_before_projecting_unwritten_routing_fields(
+ rest_catalog, use_native, buckets, columns):
+ catalog, _ = rest_catalog
+ table = _table(catalog, use_native, partitioned=True, bucketed=True,
buckets=buckets)
+ data = pa.Table.from_pylist([
+ dict(id=1, pt=7, value=42), dict(id=None, pt=None, value=43),
+ ], schema=_SCHEMA)
+ partitions, bucket_ids =
table.create_row_key_extractor().extract_partition_bucket_batch(data.to_batches()[0])
+ expected_buckets = dict(zip(partitions, bucket_ids))
+ builder = table.new_batch_write_builder()
+ writer = builder.new_write()
+ try:
+ with ExitStack() as stack:
+ _native_only(stack, writer, use_native)
+ writer.with_write_type(list(columns))
+ before = _files(table)
+ writer.write_arrow(data.slice(0, 0))
+ assert _files(table) == before
+ writer.write_arrow(data)
+ messages = _finish(builder, writer)
+ assert {tuple(message.partition): message.bucket for message in
messages} == expected_buckets
+ assert all(file.write_cols == list(columns) for message in messages
for file in message.new_files)
+ assert _rows(table) == [dict(id=None, pt=7, value=42), dict(id=None,
pt=None, value=43)]
+ finally:
+ writer.close()
+
+
[email protected]('use_native', _ENGINES)
[email protected]('partial_row', [False, True])
[email protected]('unwritten_blob', [False, True])
+def test_rows_route_with_unwritten_columns_without_requiring_other_fields(
+ rest_catalog, use_native, partial_row, unwritten_blob):
+ from pypaimon.table.row.blob import Blob
+
+ catalog, _ = rest_catalog
+ schema = _SCHEMA.append(pa.field('unwritten', pa.large_binary() if
unwritten_blob else pa.int64()))
+ table = _table(catalog, use_native, partitioned=True, bucketed=True,
schema=schema)
+ fields = {field.name: field for field in table.fields}
+ names = ['value', 'pt', 'id'] if partial_row else list(schema.names)
+ blob = Mock(spec=Blob)
+ blob.new_input_stream.side_effect = AssertionError('unwritten Blob was
opened')
+ values = dict(id=1, pt=7, value=42, unwritten=blob if unwritten_blob else
999)
+ row = GenericRow([values[name] for name in names], [fields[name] for name
in names])
+ full_input = pa.Table.from_pylist([dict(values, unwritten=None)],
schema=schema).to_batches()[0]
+ _, buckets =
table.create_row_key_extractor().extract_partition_bucket_batch(full_input)
+ builder = table.new_batch_write_builder()
+ writer = builder.new_write()
+ try:
+ with ExitStack() as stack:
+ _native_only(stack, writer, use_native)
+ writer.with_write_type(['value'])
+ writer.write_row(row)
+ messages = _finish(builder, writer)
+ assert [(tuple(message.partition), message.bucket) for message in
messages] == [((7,), buckets[0])]
+ assert all(file.write_cols == ['value'] for message in messages for
file in message.new_files)
+ assert _rows(table) == [dict(id=None, pt=7, value=42, unwritten=None)]
+ blob.new_input_stream.assert_not_called()
+ blob.to_descriptor.assert_not_called()
+ finally:
+ writer.close()
+
+
[email protected]('use_native', _ENGINES)
+def
test_upstream_bucket_partial_overwrite_preserves_other_partitions(rest_catalog,
use_native):
+ catalog, _ = rest_catalog
+ table = _table(catalog, use_native, partitioned=True, bucketed=True)
+ seed = table.new_batch_write_builder()
+ writer = seed.new_write()
+ try:
+ for row in [dict(id=1, pt=7, value=10), dict(id=2, pt=8, value=20)]:
+ writer.write_arrow(pa.Table.from_pylist([row], schema=_SCHEMA),
bucket=0)
+ _finish(seed, writer)
+ finally:
+ writer.close()
+ builder = table.new_batch_write_builder().overwrite({'pt': 7})
+ writer = builder.new_write()
+ try:
+ with ExitStack() as stack:
+ _native_only(stack, writer, use_native)
+ writer.with_write_type(['value', 'pt'])
+ writer.write_arrow(pa.table({'value': [42], 'pt': [7]}), bucket=2)
+ messages = _finish(builder, writer)
+ assert [message.bucket for message in messages] == [2]
+ assert _rows(table) == [dict(id=2, pt=8, value=20), dict(id=None,
pt=7, value=42)]
+ finally:
+ writer.close()
diff --git a/paimon-python/pypaimon/tests/write/table_write_test.py
b/paimon-python/pypaimon/tests/write/table_write_test.py
index 072d050489..47ddaef1f5 100644
--- a/paimon-python/pypaimon/tests/write/table_write_test.py
+++ b/paimon-python/pypaimon/tests/write/table_write_test.py
@@ -233,7 +233,7 @@ class TableWriteTest(unittest.TestCase):
table_write.table_pyarrow_schema = schema
table_write._validate_pyarrow_schema = Mock()
table_write.row_key_extractor = Mock()
- table_write.file_store_write = Mock()
+ table_write.file_store_write = Mock(write_cols=None)
table_write.row_key_extractor.extract_partition_bucket_groups.side_effect = (
lambda data: TableWriteTest._groups_from_rows(
partitions, buckets, data.num_rows))
@@ -308,7 +308,7 @@ class TableWriteTest(unittest.TestCase):
table_write.table = Mock(is_primary_key_table=False)
table_write.table_pyarrow_schema = data.schema
table_write._validate_pyarrow_schema = Mock()
- table_write.file_store_write = Mock()
+ table_write.file_store_write = Mock(write_cols=None)
table_write.row_key_extractor = Mock()
table_write.row_key_extractor.extract_partition_bucket_groups.return_value = [
(('p1',), 0, pa.array([0, 2, 3], type=pa.int64())),
@@ -334,7 +334,7 @@ class TableWriteTest(unittest.TestCase):
table_write.table = Mock(is_primary_key_table=False)
table_write.table_pyarrow_schema = data.schema
table_write._validate_pyarrow_schema = Mock()
- table_write.file_store_write = Mock()
+ table_write.file_store_write = Mock(write_cols=None)
table_write.row_key_extractor = Mock()
table_write.row_key_extractor.extract_partition_bucket_groups.return_value = [
(('p0',), 0, pa.array([0], type=pa.int64())),
diff --git a/paimon-python/pypaimon/write/commit/overwrite_changes_provider.py
b/paimon-python/pypaimon/write/commit/overwrite_changes_provider.py
index 31e969ab7b..acd88833f5 100644
--- a/paimon-python/pypaimon/write/commit/overwrite_changes_provider.py
+++ b/paimon-python/pypaimon/write/commit/overwrite_changes_provider.py
@@ -15,8 +15,11 @@
# specific language governing permissions and limitations
# under the License.
+from dataclasses import dataclass
from typing import List, Optional
+from pypaimon.manifest.index_manifest_entry import IndexManifestEntry
+from pypaimon.manifest.index_manifest_file import IndexManifestFile
from pypaimon.manifest.manifest_file_manager import ManifestFileManager
from pypaimon.manifest.schema.file_entry import FileEntry
from pypaimon.manifest.schema.manifest_entry import ManifestEntry
@@ -25,10 +28,19 @@ from pypaimon.snapshot.snapshot import Snapshot
from pypaimon.table.row.generic_row import GenericRow
+@dataclass
+class OverwriteChanges:
+ table_files: List[ManifestEntry]
+ index_adds: List[IndexManifestEntry]
+ index_deletes: List[IndexManifestEntry]
+
+
class OverwriteChangesProvider:
- """Builds the commit entries (DELETE existing + ADD new) for an OVERWRITE,
- caching the existing files of the target partitions across commit retries
- to avoid repeated full scans.
+ """Return file and index changes (DELETE existing + ADD new) for OVERWRITE.
+
+ Existing data files of the target partitions are cached across commit
+ retries to avoid repeated full scans. Index entries come from the latest
+ snapshot on every attempt, including changes in partitions without data.
On retry, if the latest snapshot advanced, the cached state is updated by
applying the target-partition delta manifests. Missing snapshots or
@@ -51,10 +63,10 @@ class OverwriteChangesProvider:
self.delta_probe_count = 0
self.delta_apply_count = 0
- def provide(self, latest_snapshot: Optional[Snapshot]) ->
List[ManifestEntry]:
+ def provide(self, latest_snapshot: Optional[Snapshot]) -> OverwriteChanges:
if latest_snapshot is None:
# Empty table: nothing existing to delete, just add the new files.
- return self._build_result([])
+ return self._build_result([], None)
if self._cached_snapshot is None:
self._cached_entries =
self._full_scan_manifest_entries(latest_snapshot)
@@ -69,7 +81,7 @@ class OverwriteChangesProvider:
self._cached_snapshot = latest_snapshot
# cached_snapshot.id == latest_snapshot.id -> reuse cache as-is
- return self._build_result(self._cached_entries)
+ return self._build_result(self._cached_entries, latest_snapshot)
def _full_scan_manifest_entries(self, latest_snapshot: Snapshot) ->
List[ManifestEntry]:
self.full_scan_count += 1
@@ -119,8 +131,11 @@ class OverwriteChangesProvider:
entries.append(entry)
return entries
- def _build_result(self, existing_entries: List[ManifestEntry]) ->
List[ManifestEntry]:
+ def _build_result(self, existing_entries: List[ManifestEntry],
+ latest_snapshot: Optional[Snapshot]) -> OverwriteChanges:
entries = []
+ index_adds = []
+ index_deletes = []
# Existing files of the target partitions become DELETE entries. Build
# fresh entries so the cached (kind=0) entries are never mutated.
for entry in existing_entries:
@@ -131,8 +146,19 @@ class OverwriteChangesProvider:
total_buckets=entry.total_buckets,
file=entry.file,
))
+ # Resolve every index type from this attempt's snapshot. A concurrent
+ # writer can replace or add indexes while the data-entry cache remains
+ # reusable; deleting only the first attempt's HASH files would miss it.
+ if latest_snapshot is not None and latest_snapshot.index_manifest:
+ for entry in
IndexManifestFile(self.table).read(latest_snapshot.index_manifest):
+ if self.partition_filter is None or
self.partition_filter.test(entry.partition):
+ index_deletes.append(IndexManifestEntry(
+ kind=1, partition=entry.partition, bucket=entry.bucket,
+ index_file=entry.index_file))
# New files being written by this overwrite.
for msg in self.commit_messages:
+ index_adds.extend(msg.index_adds)
+ index_deletes.extend(msg.index_deletes)
partition = GenericRow(list(msg.partition),
self.table.partition_keys_fields)
total_buckets = (
msg.total_buckets
@@ -147,4 +173,4 @@ class OverwriteChangesProvider:
total_buckets=total_buckets,
file=file,
))
- return entries
+ return OverwriteChanges(entries, index_adds, index_deletes)
diff --git a/paimon-python/pypaimon/write/file_store_commit.py
b/paimon-python/pypaimon/write/file_store_commit.py
index d5390088e3..8ba69b936c 100644
--- a/paimon-python/pypaimon/write/file_store_commit.py
+++ b/paimon-python/pypaimon/write/file_store_commit.py
@@ -46,7 +46,7 @@ from pypaimon.write.commit.row_id_conflict_rewriter import (
RowIdConflictRewriter,
RowIdRewriteResult,
)
-from pypaimon.write.commit.overwrite_changes_provider import
OverwriteChangesProvider
+from pypaimon.write.commit.overwrite_changes_provider import OverwriteChanges,
OverwriteChangesProvider
from pypaimon.table.special_fields import SpecialFields
from pypaimon.write.commit_callback import CommitCallback,
CommitCallbackContext
from pypaimon.write.commit_message import CommitMessage
@@ -298,24 +298,18 @@ class FileStoreCommit:
# Partition filter is built from dynamic or static partition according
to options.
if len(self.table.partition_keys) > 0 and
self.table.options.dynamic_partition_overwrite():
- if not commit_messages:
- # In dynamic mode, if there are no changes to commit, no data
will be deleted
+ data_messages = [message for message in commit_messages
+ if message.new_files or message.deleted_files]
+ if not data_messages:
+ # Java derives dynamic overwrite targets from table-file
+ # changes. Index/changelog-only messages do not replace data.
skip_overwrite = True
else:
- partition_filter =
self._create_dynamic_partition_filter(commit_messages)
+ partition_filter =
self._create_dynamic_partition_filter(data_messages)
else:
partition_filter =
self._create_static_partition_filter(overwrite_partition, commit_messages)
- index_adds = [
- entry for message in commit_messages for entry in
message.index_adds
- ]
- index_deletes = [
- entry for message in commit_messages for entry in
message.index_deletes
- ]
if not skip_overwrite:
- index_deletes = self._overwrite_hash_index_deletes(
- partition_filter, index_deletes
- )
provider = self._overwrite_changes_provider(partition_filter,
commit_messages)
self._try_commit(
commit_kind="OVERWRITE",
@@ -324,40 +318,11 @@ class FileStoreCommit:
changelog_entries=[],
detect_conflicts=True,
allow_rollback=False,
- index_deletes=index_deletes,
- index_adds=index_adds,
snapshot_properties=snapshot_properties,
# Java records static/table overwrite even when no files match.
allow_empty_commit=True,
)
- def _overwrite_hash_index_deletes(self, partition_filter, deletes):
- """Delete HASH indexes for every partition replaced by overwrite."""
- from pypaimon.index.dynamic_bucket import HASH_INDEX
- from pypaimon.index.index_file_handler import IndexFileHandler
- from pypaimon.manifest.index_manifest_entry import IndexManifestEntry
- from pypaimon.table.bucket_mode import BucketMode
-
- if self.table.bucket_mode() != BucketMode.HASH_DYNAMIC:
- return deletes
-
- by_file_name = {entry.index_file.file_name: entry for entry in deletes}
- snapshot = self.snapshot_manager.get_latest_snapshot()
- for entry in IndexFileHandler(self.table).scan(snapshot):
- if entry.index_file.index_type != HASH_INDEX:
- continue
- if partition_filter is not None and not partition_filter.test(
- entry.partition
- ):
- continue
- by_file_name[entry.index_file.file_name] = IndexManifestEntry(
- kind=1,
- partition=entry.partition,
- bucket=entry.bucket,
- index_file=entry.index_file,
- )
- return list(by_file_name.values())
-
def drop_partitions(self, partitions: List[Dict[str, str]],
commit_identifier: int) -> None:
if not partitions:
raise ValueError("Partitions list cannot be empty.")
@@ -404,6 +369,8 @@ class FileStoreCommit:
commit_entries_plan=provider.provide,
detect_conflicts=True,
allow_rollback=False,
+ # Java records a partition overwrite even when no files match.
+ allow_empty_commit=True,
)
def truncate_table(self, commit_identifier: int) -> None:
@@ -415,6 +382,7 @@ class FileStoreCommit:
commit_entries_plan=provider.provide,
detect_conflicts=True,
allow_rollback=False,
+ allow_empty_commit=True,
)
def _try_commit(self, commit_kind, commit_identifier, commit_entries_plan,
@@ -438,15 +406,23 @@ class FileStoreCommit:
commit_kind,
notify_callbacks=True):
break
- commit_entries = (
+ changes = (
rewritten_commit_entries
if rewritten_commit_entries is not None
else commit_entries_plan(latest_snapshot)
)
+ if isinstance(changes, OverwriteChanges):
+ commit_entries = changes.table_files
+ attempt_index_adds = changes.index_adds
+ attempt_index_deletes = changes.index_deletes
+ else:
+ commit_entries = changes
+ attempt_index_adds = index_adds
+ attempt_index_deletes = index_deletes
# Callers opt in when the operation records an empty snapshot.
- if (not allow_empty_commit and not commit_entries
- and not index_deletes and not index_adds):
+ if (not allow_empty_commit and not commit_entries and not
changelog_entries
+ and not attempt_index_deletes and not attempt_index_adds):
break
result = self._try_commit_once(
@@ -458,8 +434,8 @@ class FileStoreCommit:
latest_snapshot=latest_snapshot,
detect_conflicts=detect_conflicts,
allow_rollback=allow_rollback,
- index_deletes=index_deletes,
- index_adds=index_adds,
+ index_deletes=attempt_index_deletes,
+ index_adds=attempt_index_adds,
commit_result_may_be_uncertain=commit_result_may_be_uncertain,
snapshot_properties=snapshot_properties,
)
diff --git a/paimon-python/pypaimon/write/native_merge_into.py
b/paimon-python/pypaimon/write/native_merge_into.py
index 3338ccefd3..717950763f 100644
--- a/paimon-python/pypaimon/write/native_merge_into.py
+++ b/paimon-python/pypaimon/write/native_merge_into.py
@@ -82,11 +82,8 @@ def create_native_merge_into(table, source, on, matched,
not_matched, commit_use
return None
if source.options.query_auth_enabled:
from pypaimon.read.native_plan import
_native_query_auth_supported, _native_table
- from pypaimon.snapshot.time_travel_util import SCAN_KEYS
from pypaimon.write.native_commit import _rest_catalog_supported
- if (not _rest_catalog_supported(source) or not
_native_query_auth_supported(source)
- or source.current_branch() != 'main' or '_read_snapshot'
in vars(source)
- or any(source.table_schema.options.get(key) is not None
for key in SCAN_KEYS)):
+ if not _rest_catalog_supported(source) or not
_native_query_auth_supported(source):
return None
# A source is a read, not a write. Core obtains a fresh REST grant
# and requires it to be unrestricted, as Python plan_all_files
does.
diff --git a/paimon-python/pypaimon/write/native_update.py
b/paimon-python/pypaimon/write/native_update.py
index 610eaedd54..186f5f8f53 100644
--- a/paimon-python/pypaimon/write/native_update.py
+++ b/paimon-python/pypaimon/write/native_update.py
@@ -21,7 +21,7 @@ import json
import pyarrow as pa
-from pypaimon.read.native_plan import native_runtime_available
+from pypaimon.read.native_plan import _native_query_auth_supported,
native_runtime_available
from pypaimon.schema.arrow_schema import arrow_schemas_compatible
from pypaimon.schema.data_types import MapType, PyarrowFieldParser, RowType,
is_blob_file_field
from pypaimon.common.options.core_options import ChangelogProducer, CoreOptions
@@ -45,13 +45,11 @@ def _native_update_table(table):
"""Resolve common native update prerequisites before preparing changes."""
if (type(table) is not FileStoreTable
or not table.options.native_write_enabled()
- # These operations read target rows before staging changes. Pure
- # write/commit support must not select an unsupported guarded read.
- or table.options.query_auth_enabled
or not _native_update_formats_supported(table)
or table.options.changelog_producer() != ChangelogProducer.NONE
or any(table.options.options.contains_key(key) for key in
SCAN_KEYS)
- or not native_runtime_available()):
+ or not native_runtime_available()
+ or not _native_query_auth_supported(table)):
return None
schema = PyarrowFieldParser.from_paimon_schema(table.table_schema.fields)
if not _native_partition_types_supported(
@@ -167,7 +165,22 @@ def _dictionary_value_type(data_type):
def _native_upsert_key_type_compatible(source, target):
"""Qualify layouts; exact values and conversion bounds belong to core."""
+ dictionary = pa.types.is_dictionary(source)
source = _dictionary_value_type(source)
+ if dictionary:
+ # Python's take kernel can select dictionary indices even when it
+ # cannot take the underlying view/run-end values directly. These
+ # layouts are therefore valid existing matched-upsert inputs.
+ while getattr(pa.types, 'is_run_end_encoded', lambda _: False)(source):
+ source = _dictionary_value_type(source.value_type)
+ if getattr(pa.types, 'is_string_view', lambda _: False)(source):
+ source = pa.string()
+ elif getattr(pa.types, 'is_binary_view', lambda _: False)(source):
+ source = pa.binary()
+ if pa.types.is_union(source):
+ # A Union can contain unused children of unrelated types. Qualify a
+ # possible match here; core validates only the active values exactly.
+ return any(_native_upsert_key_type_compatible(field.type, target) for
field in source)
if pa.types.is_null(source):
return True
diff --git a/paimon-python/pypaimon/write/native_write.py
b/paimon-python/pypaimon/write/native_write.py
index d7aef5ebfd..267694d695 100644
--- a/paimon-python/pypaimon/write/native_write.py
+++ b/paimon-python/pypaimon/write/native_write.py
@@ -225,19 +225,16 @@ class NativeTableWrite:
data = normalize_arrow_strings(data)
target = self._table_schema if data.schema.names ==
self._table_schema.names else self._schema
data = normalize_arrow_write_nullability(data, target)
- if self._write_cols is not None and arrow_schemas_compatible(
- data.schema, self._table_schema,
check_top_level_nullability=False,
- allow_binary_compatibility=True):
- # Python accepts either the table schema or the selected schema.
- # Project the input before handing its actual write type to core.
- data = data.select(self._write_cols)
- if not arrow_schemas_compatible(
- data.schema, self._schema, check_top_level_nullability=False,
- allow_binary_compatibility=True):
+ compatible = dict(check_top_level_nullability=False,
allow_binary_compatibility=True)
+ if not (arrow_schemas_compatible(data.schema, self._schema,
**compatible)
+ or (self._write_cols is not None
+ and arrow_schemas_compatible(data.schema,
self._table_schema, **compatible))):
raise ValueError(
"Input schema isn't consistent with table schema and write
cols. "
f"Input schema is: {data.schema} Table schema is:
{self._schema} "
f"Write cols is: {self._write_cols}")
+ # Full input can carry routing columns outside the physical write type.
+ # Core routes rows first and writes only the selected columns.
return data
def write_pandas(self, dataframe):
@@ -275,6 +272,18 @@ class NativeTableWrite:
from pypaimon.write.row_kind import skip_write_row, with_row_kind
if skip_write_row(self.table, values, row.get_row_kind()):
return
+ if self._write_cols is not None:
+ routing = list(self.table.partition_keys)
+ if ((self.table.bucket_mode() == BucketMode.HASH_FIXED and
self.table.options.bucket() > 1)
+ or isinstance(self, NativePostponeFixedBucketTableWrite)):
+ routing.extend(self.table.table_schema.bucket_keys)
+ require_columns(values, routing, 'write_row')
+ if not set(routing).issubset(names):
+ retained = set(names) | set(routing)
+ names = self._table_schema.names
+ # Only routing and written values need transport. In
particular,
+ # an unwritten Blob object must not be opened or encoded.
+ values = {name: values.get(name) if name in retained else None
for name in names}
try:
data = (self._blob_rows.to_batch(self.table, [values], names)
if self._blob_rows is not None else
diff --git a/paimon-python/pypaimon/write/table_update.py
b/paimon-python/pypaimon/write/table_update.py
index 1599de5ec9..ec881bf65e 100644
--- a/paimon-python/pypaimon/write/table_update.py
+++ b/paimon-python/pypaimon/write/table_update.py
@@ -57,6 +57,16 @@ from pypaimon.write.writer.append_only_data_writer import
AppendOnlyDataWriter
logger = logging.getLogger(__name__)
+def _authorize_update(table):
+ """Require the same whole-table grant as plan_all_files for fast paths."""
+ from pypaimon.catalog.catalog_exception import TableNoPermissionException
+ from pypaimon.read.table_scan import authorize
+
+ query_auth = table.catalog_environment.table_query_auth(table.options,
table.identifier)
+ if authorize(table, query_auth, None) is not None:
+ raise TableNoPermissionException(table.identifier)
+
+
def _filter_by_whole_file_shard(splits: List[DataSplit], sub_task_id: int,
total_tasks: int) -> List[DataSplit]:
list_ranges = []
for split in splits:
@@ -568,6 +578,7 @@ class TableUpdate:
def _delete_by_partition_filter(
self, partition_filter: Predicate) -> List[CommitMessage]:
+ _authorize_update(self.table)
snapshot = self.table.snapshot_manager().get_latest_snapshot()
if snapshot is None:
return []
diff --git a/paimon-python/pypaimon/write/table_write.py
b/paimon-python/pypaimon/write/table_write.py
index 63b175e168..65c9996326 100644
--- a/paimon-python/pypaimon/write/table_write.py
+++ b/paimon-python/pypaimon/write/table_write.py
@@ -96,6 +96,8 @@ class TableWrite:
if bucket is not None:
return self._write_precomputed_bucket(data, bucket)
data = self._prepare_arrow_data(data)
+ if data.num_rows == 0:
+ return
for partition, bucket, row_indices in \
self.row_key_extractor.extract_partition_bucket_groups(data):
@@ -129,6 +131,11 @@ class TableWrite:
self.file_store_write.roll_before_group_if_needed(row_count)
def _write_partition_bucket_batch(self, partition, bucket, data):
+ write_cols = self.file_store_write.write_cols
+ if write_cols is not None:
+ # Keep routing columns available for full-schema inputs, then
+ # write only the columns and order recorded in the file metadata.
+ data = data.select(write_cols)
self.file_store_write.write(partition, bucket, data)
def _write_precomputed_bucket(self, data, bucket):
@@ -202,10 +209,16 @@ class TableWrite:
return self.write_arrow_batch(record_batch)
def with_write_type(self, write_cols: List[str]):
+ write_cols = list(write_cols)
for col in write_cols:
if col not in self.table_pyarrow_schema.names:
raise ValueError(f"Column {col} is not in table schema.")
- if len(write_cols) == len(self.table_pyarrow_schema.names):
+ full_names = self.table_pyarrow_schema.names
+ # Java preserves the declared append column order in writeCols. PK
+ # writes still require the complete table layout in the Python buffer.
+ full_pk_type = (self.table.is_primary_key_table and len(write_cols) ==
len(full_names)
+ and set(write_cols) == set(full_names))
+ if write_cols == full_names or full_pk_type:
write_cols = None
self.file_store_write.write_cols = write_cols
return self
