[python] Honor snapshot.ignore-empty-commit for empty appends

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)
 

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()

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(

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",