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 9a302e7d08 [python][ray] Fix zero-block overwrite (#8632)
9a302e7d08 is described below
commit 9a302e7d082fdc2d849dc70723f9054be0db9c89
Author: QuakeWang <[email protected]>
AuthorDate: Wed Jul 15 18:14:32 2026 +0800
[python][ray] Fix zero-block overwrite (#8632)
Ray 2.53+ defers `Datasink.on_write_start()` until the first input
bundle. A zero-block dataset has no bundle, but `on_write_complete()`
still runs, leaving Paimon's writer builder uninitialized. Whole-table
and static-partition overwrites therefore fail before `commit([])`.
Initialize the writer builder on demand when completion needs to commit.
Zero-block append writes remain no-ops.
---
.../pypaimon/tests/ray_integration_test.py | 25 ++++++++++++++++
paimon-python/pypaimon/tests/ray_sink_test.py | 35 ++++++++++++++++++++++
paimon-python/pypaimon/write/ray_datasink.py | 4 +++
3 files changed, 64 insertions(+)
diff --git a/paimon-python/pypaimon/tests/ray_integration_test.py
b/paimon-python/pypaimon/tests/ray_integration_test.py
index c79b62e9fb..875a9a3c9e 100644
--- a/paimon-python/pypaimon/tests/ray_integration_test.py
+++ b/paimon-python/pypaimon/tests/ray_integration_test.py
@@ -396,6 +396,31 @@ class RayIntegrationTest(unittest.TestCase):
result = read_paimon(identifier, self.catalog_options)
self.assertEqual(result.count(), 0)
+ def test_write_paimon_zero_block_overwrite_unpartitioned(self):
+ """An overwrite with no Ray blocks clears an unpartitioned table."""
+ from pypaimon.ray import read_paimon, write_paimon
+
+ pa_schema = pa.schema([('id', pa.int64())])
+ identifier = 'default.test_write_zero_block_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]}, schema=pa_schema)
+ )
+ write_paimon(initial, identifier, self.catalog_options)
+
+ write_paimon(
+ ray.data.range(0),
+ identifier,
+ self.catalog_options,
+ overwrite=True,
+ )
+
+ result = read_paimon(identifier, self.catalog_options)
+ self.assertEqual(result.count(), 0)
+
def test_table_write_ray_builder_partition_overwrite(self):
"""Builder-level partition overwrite is honored by write_ray()."""
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 afc86b571e..5dc462304a 100644
--- a/paimon-python/pypaimon/tests/ray_sink_test.py
+++ b/paimon-python/pypaimon/tests/ray_sink_test.py
@@ -453,6 +453,41 @@ class RaySinkTest(unittest.TestCase):
datasink.on_write_complete(write_result)
self.assertEqual(len(datasink._pending_commit_messages), 1)
+ def test_on_write_complete_without_on_write_start(self):
+ from ray.data.datasource.datasink import WriteResult
+
+ write_result = WriteResult(
+ num_rows=0,
+ size_bytes=0,
+ write_returns=[],
+ )
+
+ for overwrite, static_partition in [
+ (True, None),
+ (False, {'dt': '2024-01-01'}),
+ ]:
+ with self.subTest(
+ overwrite=overwrite,
+ static_partition=static_partition,
+ ):
+ table = Mock()
+ table.identifier.get_full_name.return_value =
'test_db.test_table'
+ writer_builder = table.new_batch_write_builder.return_value
+ writer_builder.overwrite.return_value = writer_builder
+ table_commit = writer_builder.new_commit.return_value
+
+ datasink = PaimonDatasink(
+ table,
+ overwrite=overwrite,
+ static_partition=static_partition,
+ )
+ datasink.on_write_complete(write_result)
+
+ table.new_batch_write_builder.assert_called_once_with()
+
writer_builder.overwrite.assert_called_once_with(static_partition)
+ table_commit.commit.assert_called_once_with([])
+ table_commit.close.assert_called_once_with()
+
def test_table_write_ray_forwards_static_partition(self):
dataset = Mock()
table_write = TableWrite.__new__(TableWrite)
diff --git a/paimon-python/pypaimon/write/ray_datasink.py
b/paimon-python/pypaimon/write/ray_datasink.py
index 6f7a336190..37c753b4ca 100644
--- a/paimon-python/pypaimon/write/ray_datasink.py
+++ b/paimon-python/pypaimon/write/ray_datasink.py
@@ -188,6 +188,10 @@ class PaimonDatasink(_DatasinkBase):
self._pending_commit_messages = []
return
+ # Ray does not call on_write_start when the input has no blocks.
+ if self._writer_builder is None:
+ self.on_write_start()
+
logger.info(
f"Committing {len(non_empty_messages)} commit messages "
f"for table {self._table_name}"
