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 8cb0b754a2 [python] Honor snapshot.ignore-empty-commit for empty
appends (#9878)
8cb0b754a2 is described below
commit 8cb0b754a22e72c496a0e47967d5bdde31818f49
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",
