This is an automated email from the ASF dual-hosted git repository. XiaoHongbo-Hope pushed a commit to branch release-2.1 in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 2e8803c4981c7bbeea0bb9f9e573fab7e39e3786 Author: Yann Byron <[email protected]> AuthorDate: Thu Sep 17 11:39:01 2026 +0800 [python] Honor snapshot.ignore-empty-commit for empty appends (#9878) --- .../pypaimon/common/options/core_options.py | 13 ++++++ paimon-python/pypaimon/tests/table_commit_test.py | 47 ++++++++++++++++++++++ paimon-python/pypaimon/write/file_store_commit.py | 16 +++++--- paimon-python/pypaimon/write/table_commit.py | 3 +- 4 files changed, 72 insertions(+), 7 deletions(-) diff --git a/paimon-python/pypaimon/common/options/core_options.py b/paimon-python/pypaimon/common/options/core_options.py index 2fe0d039fd..0ac4610864 100644 --- a/paimon-python/pypaimon/common/options/core_options.py +++ b/paimon-python/pypaimon/common/options/core_options.py @@ -727,6 +727,16 @@ class CoreOptions: .with_description("The prefix for commit user.") ) + SNAPSHOT_IGNORE_EMPTY_COMMIT: ConfigOption[bool] = ( + ConfigOptions.key("snapshot.ignore-empty-commit") + .boolean_type() + .no_default_value() + .with_description( + "Whether to skip append commits without changes. " + "PyPaimon defaults to true; false allows tagging an empty table." + ) + ) + COMMIT_MAX_RETRIES: ConfigOption[int] = ( ConfigOptions.key("commit.max-retries") .int_type() @@ -1609,6 +1619,9 @@ class CoreOptions: weights.append(parsed) return weights + def snapshot_ignore_empty_commit(self) -> bool: + return self.options.get(CoreOptions.SNAPSHOT_IGNORE_EMPTY_COMMIT, True) + def commit_max_retries(self) -> int: return self.options.get(CoreOptions.COMMIT_MAX_RETRIES) diff --git a/paimon-python/pypaimon/tests/table_commit_test.py b/paimon-python/pypaimon/tests/table_commit_test.py index d607e73921..2c7079bcc7 100644 --- a/paimon-python/pypaimon/tests/table_commit_test.py +++ b/paimon-python/pypaimon/tests/table_commit_test.py @@ -16,6 +16,7 @@ # under the License. import unittest +from tempfile import TemporaryDirectory from unittest.mock import Mock from parameterized import parameterized @@ -27,6 +28,52 @@ from pypaimon.write.table_commit import BatchTableCommit, StreamTableCommit class TestTableCommit(unittest.TestCase): + def test_empty_append_snapshot_is_opt_in_and_can_be_tagged(self): + import pyarrow as pa + import pypaimon.multimodal as pmm + + with TemporaryDirectory(prefix="paimon-empty-commit-") as warehouse: + connection = pmm.connect(options={"warehouse": warehouse}) + schema = pa.schema([pa.field("feature", pa.string(), False)]) + table = connection.create_table("stat", schema=schema) + empty = pa.Table.from_pylist([], schema=schema) + table.add(empty) + snapshots = table.raw_table.snapshot_manager() + self.assertIsNone(snapshots.get_latest_snapshot()) + + def commit_empty(): + writable = table.raw_table.copy({ + "snapshot.ignore-empty-commit": "false", + }) + commit = writable.new_batch_write_builder().new_commit() + try: + commit.commit([], snapshot_properties={"source": "empty-stat"}) + finally: + commit.close() + + commit_empty() + snapshot = snapshots.get_latest_snapshot() + self.assertIsNotNone(snapshot) + self.assertEqual((1, 0, 0), ( + snapshot.id, snapshot.total_record_count, + snapshot.delta_record_count)) + self.assertEqual({"source": "empty-stat"}, snapshot.properties) + table.raw_table.create_tag("empty") + tagged = table.scan(tag_name="empty").to_arrow() + self.assertEqual(0, tagged.num_rows) + self.assertEqual(schema, tagged.schema) + + table.add([{"feature": "state_imu_body"}]) + table.add(empty) + self.assertEqual(2, snapshots.get_latest_snapshot().id) + commit_empty() + snapshot = snapshots.get_latest_snapshot() + self.assertEqual((3, 1, 0), ( + snapshot.id, snapshot.total_record_count, + snapshot.delta_record_count)) + self.assertEqual([{"feature": "state_imu_body"}], table.scan().to_list()) + self.assertEqual([], table.scan(tag_name="empty").to_list()) + def _create_commit(self, cls, overwrite_partition=None): commit = cls.__new__(cls) commit.table = Mock() diff --git a/paimon-python/pypaimon/write/file_store_commit.py b/paimon-python/pypaimon/write/file_store_commit.py index 10ac3e2b53..8c7cab1175 100644 --- a/paimon-python/pypaimon/write/file_store_commit.py +++ b/paimon-python/pypaimon/write/file_store_commit.py @@ -262,7 +262,8 @@ class FileStoreCommit: commit_identifier: int, snapshot_properties: Optional[Dict[str, str]] = None): """Commit the given commit messages in normal append mode.""" - if not commit_messages: + ignore_empty_commit = self.table.options.snapshot_ignore_empty_commit() + if not commit_messages and ignore_empty_commit: return # Extract the minimum check_from_snapshot from commit messages @@ -345,7 +346,8 @@ class FileStoreCommit: index_deletes=index_deletes, index_adds=index_adds, hash_index_base_snapshot=hash_index_base_snapshot, - snapshot_properties=snapshot_properties) + snapshot_properties=snapshot_properties, + allow_empty_commit=not ignore_empty_commit) def overwrite( self, @@ -505,7 +507,8 @@ class FileStoreCommit: detect_conflicts=False, allow_rollback=False, index_deletes=None, index_adds=None, changelog_entries=None, hash_index_base_snapshot=None, - snapshot_properties: Optional[Dict[str, str]] = None): + snapshot_properties: Optional[Dict[str, str]] = None, + allow_empty_commit=False): retry_count = 0 retry_result = None @@ -528,9 +531,10 @@ class FileStoreCommit: else commit_entries_plan(latest_snapshot) ) - # No entries to commit (e.g. drop_partitions with no matching - # data): skip an empty snapshot. - if not commit_entries and not index_deletes and not index_adds: + # Append can explicitly publish an empty snapshot for tagging. + # No-op overwrite/drop operations retain their existing behavior. + if (not allow_empty_commit and not commit_entries + and not index_deletes and not index_adds): break result = self._try_commit_once( diff --git a/paimon-python/pypaimon/write/table_commit.py b/paimon-python/pypaimon/write/table_commit.py index f6eb6748cd..35ee83f2fc 100644 --- a/paimon-python/pypaimon/write/table_commit.py +++ b/paimon-python/pypaimon/write/table_commit.py @@ -88,7 +88,8 @@ class TableCommit: overwrite_partition=self.overwrite_partition, **commit_kwargs) else: - if not non_empty_messages: + if (not non_empty_messages + and self.table.options.snapshot_ignore_empty_commit()): return logger.info( "Committing table %s, %d non-empty messages",
