From 970e7700a8e688bac77e6866b5341a9b29c32313 Mon Sep 17 00:00:00 2001 From: Yann Date: Wed, 16 Sep 2026 14:59:05 +0800 Subject: [PATCH] feat(python): support explicit empty append snapshots Honor snapshot.ignore-empty-commit=false so an empty table can be versioned with a tag. Preserve default empty-append skipping and no-op overwrite/drop behavior. Validation: real empty-tag readback and commit/callback/overwrite tests. Co-Authored-By: Codex AI-Model: gpt-6 Co-Authored-By: Codex AI-Contributed/Feature: 32/32 AI-Contributed/UT: 47/47 --- .../pypaimon/common/options/core_options.py | 13 +++++ .../pypaimon/tests/table_commit_test.py | 47 +++++++++++++++++++ .../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 6586e051a5d4..2b1deb26d869 100644 --- a/paimon-python/pypaimon/common/options/core_options.py +++ b/paimon-python/pypaimon/common/options/core_options.py @@ -723,6 +723,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() @@ -1605,6 +1615,9 @@ def data_file_external_paths_weights(self, default=None): 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 d607e73921cf..2c7079bcc7bb 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 @@ 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 10ac3e2b53c3..8c7cab1175a8 100644 --- a/paimon-python/pypaimon/write/file_store_commit.py +++ b/paimon-python/pypaimon/write/file_store_commit.py @@ -262,7 +262,8 @@ def commit( 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 @@ def commit( 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 @@ def _try_commit(self, commit_kind, commit_identifier, commit_entries_plan, 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 @@ def _try_commit(self, commit_kind, commit_identifier, commit_entries_plan, 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 f6eb6748cd0a..35ee83f2fccf 100644 --- a/paimon-python/pypaimon/write/table_commit.py +++ b/paimon-python/pypaimon/write/table_commit.py @@ -88,7 +88,8 @@ def _commit( 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",