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 cd95f04b1a [Python] Commit snapshot properties through Native (#10505)
cd95f04b1a is described below
commit cd95f04b1a3b8e198caf120acf6e9f6a5f28bde2
Author: Jingsong Lee <[email protected]>
AuthorDate: Sun Oct 11 19:49:45 2026 +0800
[Python] Commit snapshot properties through Native (#10505)
---
paimon-python/pypaimon/tests/native_commit_test.py | 7 +-
.../tests/native_snapshot_properties_test.py | 160 +++++++++++++++++++++
paimon-python/pypaimon/write/table_commit.py | 30 ++--
3 files changed, 179 insertions(+), 18 deletions(-)
diff --git a/paimon-python/pypaimon/tests/native_commit_test.py
b/paimon-python/pypaimon/tests/native_commit_test.py
index 3b62aa0ac5..31d9b63a19 100644
--- a/paimon-python/pypaimon/tests/native_commit_test.py
+++ b/paimon-python/pypaimon/tests/native_commit_test.py
@@ -441,16 +441,17 @@ def
test_publication_response_loss_does_not_duplicate_or_delete_files(
@pytest.mark.parametrize('properties', [{}, {'source': 'python'}])
@pytest.mark.parametrize('overwrite', [False, True])
-def test_snapshot_properties_select_python_before_native(tmp_path, properties,
overwrite):
+def test_non_rest_snapshot_properties_keep_python_commit(tmp_path, properties,
overwrite):
table = _table(tmp_path)
builder = table.new_batch_write_builder()
if overwrite:
builder.overwrite()
commit = builder.new_commit()
with patch('pypaimon.write.native_commit.create_native_commit',
- side_effect=AssertionError('must not initialize native')) as
create:
+ wraps=create_native_commit) as create:
commit.commit(_prepare(builder, [{'id': 1, 'pt': None}]),
snapshot_properties=properties)
- create.assert_not_called()
+ create.assert_called_once()
+ assert commit._native_commit is None
assert table.snapshot_manager().get_latest_snapshot().properties ==
(properties or None)
commit.close()
diff --git a/paimon-python/pypaimon/tests/native_snapshot_properties_test.py
b/paimon-python/pypaimon/tests/native_snapshot_properties_test.py
new file mode 100644
index 0000000000..9080e1947d
--- /dev/null
+++ b/paimon-python/pypaimon/tests/native_snapshot_properties_test.py
@@ -0,0 +1,160 @@
+# 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.
+
+from unittest.mock import Mock, patch
+
+import pytest
+
+from pypaimon.tests import native_commit_test
+
+pytestmark = pytest.mark.native_plan
+
+
+@native_commit_test.requires_native
[email protected]('mode', ['append', 'pk', 'de'])
[email protected]('properties', [None, {}, {'source': 'capture',
'label': '中文'}])
[email protected]('overwrite', [False, True])
+def test_rest_native_snapshot_properties(tmp_path, native_rest_catalog, mode,
properties, overwrite):
+ table = native_commit_test._table(tmp_path, mode,
catalog=native_rest_catalog)
+ native_commit_test._seed(table)
+ builder = table.new_batch_write_builder()
+ if overwrite:
+ builder.overwrite()
+ messages = native_commit_test._prepare(builder, [{'id': 10, 'pt': 'a'}])
+ commit = builder.new_commit()
+ try:
+ with native_commit_test._must_not_fallback(commit, 'overwrite' if
overwrite else 'commit'):
+ commit.commit(messages, snapshot_properties=properties)
+ snapshot = table.snapshot_manager().get_latest_snapshot()
+ assert snapshot.properties == (properties or None)
+ expected = ([{'id': 10, 'pt': 'a'}] if mode == 'pk' else [
+ {'id': 2, 'pt': 'b'}, {'id': 3, 'pt': None}, {'id': 10, 'pt':
'a'}]) if overwrite else [
+ {'id': 1, 'pt': 'a'}, {'id': 2, 'pt': 'b'}, {'id': 3, 'pt': None},
{'id': 10, 'pt': 'a'}]
+ assert native_commit_test._rows(table) == expected
+ finally:
+ commit.close()
+
+
+@native_commit_test.requires_native
[email protected]('mode', ['append', 'pk', 'de'])
+def test_stream_properties_do_not_leak_between_checkpoints(tmp_path,
native_rest_catalog, mode):
+ table = native_commit_test._table(tmp_path, mode,
catalog=native_rest_catalog)
+ builder = table.new_stream_write_builder()
+ commit = builder.new_commit()
+ properties = {'checkpoint': 'first'}
+ try:
+ with native_commit_test._must_not_fallback(commit):
+ for identifier, metadata in enumerate([properties, None, {},
{'checkpoint': 'fourth'}], 1):
+ messages = native_commit_test._prepare(builder, [{'id':
identifier, 'pt': 'a'}], identifier)
+ commit.commit(messages, identifier,
snapshot_properties=metadata)
+ snapshot = table.snapshot_manager().get_latest_snapshot()
+ assert snapshot.commit_identifier == identifier
+ assert snapshot.properties == (metadata or None)
+ properties['checkpoint'] = 'changed'
+ assert table.snapshot_manager().get_snapshot_by_id(1).properties ==
{'checkpoint': 'first'}
+ assert native_commit_test._rows(table) == [{'id': i, 'pt': 'a'} for i
in range(1, 5)]
+ finally:
+ commit.close()
+
+
+@native_commit_test.requires_native
[email protected]('ignore_empty', [False, True])
+def test_empty_append_properties_follow_empty_commit_option(tmp_path,
native_rest_catalog, ignore_empty):
+ table = native_commit_test._table(tmp_path,
catalog=native_rest_catalog).copy(
+ {'snapshot.ignore-empty-commit': str(ignore_empty).lower()})
+ commit = table.new_batch_write_builder().new_commit()
+ try:
+ with native_commit_test._must_not_fallback(commit):
+ commit.commit([], snapshot_properties={'source': 'empty'})
+ snapshot = table.snapshot_manager().get_latest_snapshot()
+ if ignore_empty:
+ assert snapshot is None
+ else:
+ assert snapshot.properties == {'source': 'empty'}
+ assert snapshot.total_record_count == 0
+ assert snapshot.delta_record_count == 0
+ finally:
+ commit.close()
+
+
+@native_commit_test.requires_native
[email protected]('mode', ['append', 'de'])
+def test_empty_static_overwrite_records_properties(tmp_path,
native_rest_catalog, mode):
+ table = native_commit_test._table(tmp_path, mode,
catalog=native_rest_catalog).copy(
+ {'dynamic-partition-overwrite': 'false'})
+ native_commit_test._seed(table)
+ builder = table.new_batch_write_builder().overwrite({'pt': 'a'})
+ commit = builder.new_commit()
+ try:
+ with native_commit_test._must_not_fallback(commit, 'overwrite'):
+ commit.commit([], snapshot_properties={'reason': 'empty
overwrite'})
+ snapshot = table.snapshot_manager().get_latest_snapshot()
+ assert snapshot.properties == {'reason': 'empty overwrite'}
+ assert snapshot.commit_kind == 'OVERWRITE'
+ assert snapshot.delta_record_count == -1
+ assert native_commit_test._rows(table) == [{'id': 2, 'pt': 'b'},
{'id': 3, 'pt': None}]
+ finally:
+ commit.close()
+
+
+@native_commit_test.requires_native
+def test_empty_dynamic_overwrite_does_not_publish_properties(tmp_path,
native_rest_catalog):
+ table = native_commit_test._table(tmp_path,
catalog=native_rest_catalog).copy(
+ {'dynamic-partition-overwrite': 'true'})
+ native_commit_test._seed(table)
+ commit = table.new_batch_write_builder().overwrite().new_commit()
+ try:
+ with native_commit_test._must_not_fallback(commit, 'overwrite'):
+ commit.commit([], snapshot_properties={'source': 'unused'})
+ snapshot = table.snapshot_manager().get_latest_snapshot()
+ assert snapshot.id == 1
+ assert snapshot.properties is None
+ assert len(native_commit_test._rows(table)) == 3
+ finally:
+ commit.close()
+
+
+@native_commit_test.requires_native
[email protected]('overwrite', [False, True])
+def test_properties_survive_publication_response_loss(tmp_path,
native_rest_catalog, overwrite):
+ table = native_commit_test._table(tmp_path, catalog=native_rest_catalog)
+ builder = table.new_batch_write_builder()
+ if overwrite:
+ builder.overwrite({'pt': 'a'})
+ messages = native_commit_test._prepare(builder, [{'id': 1, 'pt': 'a'}])
+ native = native_commit_test.create_native_commit(table,
builder.commit_user, builder.static_partition)
+ proxy = Mock(wraps=native)
+
+ def publish_then_fail(*args, **kwargs):
+ native.commit(*args, **kwargs)
+ raise OSError('response lost')
+
+ proxy.commit.side_effect = publish_then_fail
+ commit = builder.new_commit()
+ try:
+ with patch('pypaimon.write.native_commit.create_native_commit',
return_value=proxy), \
+ native_commit_test._must_not_fallback(commit, 'overwrite' if
overwrite else 'commit'), \
+ patch.object(commit.file_store_commit, 'abort',
side_effect=AssertionError('abort')):
+ with pytest.raises(OSError, match='response lost'):
+ commit.commit(messages, snapshot_properties={'source':
'published'})
+ snapshot = table.snapshot_manager().get_latest_snapshot()
+ assert snapshot.id == 1
+ assert snapshot.properties == {'source': 'published'}
+ assert native_commit_test._rows(table) == [{'id': 1, 'pt': 'a'}]
+ proxy.abort.assert_not_called()
+ finally:
+ commit.close()
diff --git a/paimon-python/pypaimon/write/table_commit.py
b/paimon-python/pypaimon/write/table_commit.py
index 894d616fba..fb0ac3db6d 100644
--- a/paimon-python/pypaimon/write/table_commit.py
+++ b/paimon-python/pypaimon/write/table_commit.py
@@ -74,6 +74,8 @@ class TableCommit:
}
if snapshot_properties is not None:
commit_kwargs["snapshot_properties"] = snapshot_properties
+ native_kwargs = ({} if snapshot_properties is None else
+ {"snapshot_properties": snapshot_properties})
# Never abort files in response to a commit exception. Preserving
# possible orphan files is safer than deleting files which another
@@ -86,13 +88,12 @@ class TableCommit:
"Committing overwrite to table %s, %d non-empty messages",
self.table.identifier, len(non_empty_messages)
)
- if snapshot_properties is None:
- prepared = self._prepare_native_commit(non_empty_messages)
- if prepared is not None:
- native, messages = prepared
- # Keep publication failures outside the preparation
fallback.
- native.commit(messages)
- return
+ prepared = self._prepare_native_commit(non_empty_messages)
+ if prepared is not None:
+ native, messages = prepared
+ # Keep publication failures outside the preparation fallback.
+ native.commit(messages, **native_kwargs)
+ return
self.file_store_commit.overwrite(
overwrite_partition=self.overwrite_partition,
**commit_kwargs)
@@ -104,14 +105,13 @@ class TableCommit:
"Committing table %s, %d non-empty messages",
self.table.identifier, len(non_empty_messages)
)
- if snapshot_properties is None:
- prepared = self._prepare_native_commit(non_empty_messages)
- if prepared is not None:
- native, messages = prepared
- # Mutation is deliberately outside the fallback boundary:
- # an exception can mean the snapshot was already published.
- native.commit(commit_identifier, messages)
- return
+ prepared = self._prepare_native_commit(non_empty_messages)
+ if prepared is not None:
+ native, messages = prepared
+ # Mutation is deliberately outside the fallback boundary:
+ # an exception can mean the snapshot was already published.
+ native.commit(commit_identifier, messages, **native_kwargs)
+ return
self.file_store_commit.commit(**commit_kwargs)
def _prepare_native_commit(self, messages):
