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 bcdd0d6f69 [python] Fix empty Ray overwrite commit (#8041)
bcdd0d6f69 is described below
commit bcdd0d6f6953e71406a234e26fc132c7c65f7167
Author: QuakeWang <[email protected]>
AuthorDate: Sun May 31 10:16:48 2026 +0800
[python] Fix empty Ray overwrite commit (#8041)
`PaimonDatasink.on_write_complete()` previously returned early when all
Ray write tasks produced empty commit messages. This made
`write_paimon(empty_ds, table, overwrite=True)` a silent no-op, so
existing data was kept even though Paimon overwrite commit semantics
require empty overwrite commits to reach `TableCommit`.
This PR keeps the empty-message fast path only for append writes. For
overwrite writes, Ray now calls `TableCommit.commit([])`, allowing
`FileStoreCommit.overwrite()` to delete the target range for
static/unpartitioned overwrite and preserve dynamic-partition empty
overwrite no-op semantics.
---
.../pypaimon/tests/ray_integration_test.py | 27 ++++++++++++++++++++++
paimon-python/pypaimon/tests/ray_sink_test.py | 16 +++++++++++++
paimon-python/pypaimon/write/ray_datasink.py | 2 +-
3 files changed, 44 insertions(+), 1 deletion(-)
diff --git a/paimon-python/pypaimon/tests/ray_integration_test.py
b/paimon-python/pypaimon/tests/ray_integration_test.py
index 10464f1653..2ad95f610b 100644
--- a/paimon-python/pypaimon/tests/ray_integration_test.py
+++ b/paimon-python/pypaimon/tests/ray_integration_test.py
@@ -306,6 +306,33 @@ class RayIntegrationTest(unittest.TestCase):
df = result.to_pandas()
self.assertEqual(list(df['id']), [3])
+ def test_write_paimon_empty_overwrite_unpartitioned(self):
+ """write_paimon(overwrite=True) with empty data clears an
unpartitioned table."""
+ from pypaimon.ray import read_paimon, write_paimon
+
+ pa_schema = pa.schema([
+ ('id', pa.int32()),
+ ('val', pa.int64()),
+ ])
+ identifier = 'default.test_write_empty_overwrite_unpartitioned'
+ catalog = CatalogFactory.create(self.catalog_options)
+ schema = Schema.from_pyarrow_schema(pa_schema)
+ catalog.create_table(identifier, schema, False)
+
+ initial = ray.data.from_arrow(
+ pa.Table.from_pydict({'id': [1, 2], 'val': [10, 20]},
schema=pa_schema)
+ )
+ write_paimon(initial, identifier, self.catalog_options)
+ self.assertEqual(read_paimon(identifier,
self.catalog_options).count(), 2)
+
+ empty = ray.data.from_arrow(
+ pa.Table.from_pydict({'id': [], 'val': []}, schema=pa_schema)
+ )
+ write_paimon(empty, identifier, self.catalog_options, overwrite=True)
+
+ result = read_paimon(identifier, self.catalog_options)
+ self.assertEqual(result.count(), 0)
+
def test_read_paimon_primary_key(self):
"""read_paimon() merges PK rows correctly after an upsert."""
from pypaimon.ray import read_paimon
diff --git a/paimon-python/pypaimon/tests/ray_sink_test.py
b/paimon-python/pypaimon/tests/ray_sink_test.py
index fab77ecf87..ca51b05d1b 100644
--- a/paimon-python/pypaimon/tests/ray_sink_test.py
+++ b/paimon-python/pypaimon/tests/ray_sink_test.py
@@ -225,6 +225,22 @@ class RaySinkTest(unittest.TestCase):
)
datasink.on_write_complete(write_result)
+ # Empty overwrite must still reach TableCommit so overwrite semantics
+ # can delete the target range.
+ datasink = PaimonDatasink(self.table, overwrite=True)
+ datasink.on_write_start()
+ write_result = WriteResult(
+ num_rows=0,
+ size_bytes=0,
+ write_returns=[[], []]
+ )
+ mock_commit = Mock()
+ datasink._writer_builder.new_commit = Mock(return_value=mock_commit)
+ datasink.on_write_complete(write_result)
+
+ mock_commit.commit.assert_called_once_with([])
+ mock_commit.close.assert_called_once()
+
# Test with messages and filtering empty messages
datasink = PaimonDatasink(self.table, overwrite=False)
datasink.on_write_start()
diff --git a/paimon-python/pypaimon/write/ray_datasink.py
b/paimon-python/pypaimon/write/ray_datasink.py
index a01387b327..989ec42dc4 100644
--- a/paimon-python/pypaimon/write/ray_datasink.py
+++ b/paimon-python/pypaimon/write/ray_datasink.py
@@ -163,7 +163,7 @@ class PaimonDatasink(_DatasinkBase):
self._pending_commit_messages = non_empty_messages
- if not non_empty_messages:
+ if not non_empty_messages and not self.overwrite:
logger.info("No data to commit (all commit messages are
empty)")
self._pending_commit_messages = []
return
