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 d564417598 [python][daft] Abort failed sink writes (#8454)
d564417598 is described below
commit d564417598dc90f5220358b9fbe8b0a9fa97ab0c
Author: QuakeWang <[email protected]>
AuthorDate: Sat Jul 4 21:11:01 2026 +0800
[python][daft] Abort failed sink writes (#8454)
Daft `PaimonDataSink.write()` previously always called
`table_write.close()` in a `finally` block. `close()` may flush and
close writers, but it does not remove uncommitted files. If a later
micropartition failed after an earlier one had produced data files, the
failed write could leave orphan files.
This changes the worker write path to close only after
`prepare_commit()` succeeds, and to call `table_write.abort()` on
exceptions before re-raising. It also adds a regression test for a valid
micropartition followed by a schema-mismatched micropartition, verifying
no data files remain.
---
paimon-python/pypaimon/daft/daft_datasink.py | 10 ++++-
.../pypaimon/tests/daft/daft_sink_test.py | 49 ++++++++++++++++++++++
2 files changed, 58 insertions(+), 1 deletion(-)
diff --git a/paimon-python/pypaimon/daft/daft_datasink.py
b/paimon-python/pypaimon/daft/daft_datasink.py
index 7e6b871f06..747ccd2bc4 100644
--- a/paimon-python/pypaimon/daft/daft_datasink.py
+++ b/paimon-python/pypaimon/daft/daft_datasink.py
@@ -18,6 +18,7 @@
from __future__ import annotations
+import logging
from typing import TYPE_CHECKING, Any
import pyarrow as pa
@@ -32,6 +33,8 @@ if TYPE_CHECKING:
from pypaimon.table.file_store_table import FileStoreTable
+logger = logging.getLogger(__name__)
+
_PaimonIdentifier = tuple[str, str, str | None]
@@ -243,8 +246,13 @@ class PaimonDataSink(DataSink[list[Any]]):
total_rows += batch.num_rows
total_bytes += batch.nbytes
commit_messages = table_write.prepare_commit()
- finally:
table_write.close()
+ except Exception:
+ try:
+ table_write.abort()
+ except Exception:
+ logger.warning("Failed to abort Daft Paimon table write.",
exc_info=True)
+ raise
yield WriteResult(
result=list(commit_messages),
diff --git a/paimon-python/pypaimon/tests/daft/daft_sink_test.py
b/paimon-python/pypaimon/tests/daft/daft_sink_test.py
index 34f434a2b1..a2b058c116 100644
--- a/paimon-python/pypaimon/tests/daft/daft_sink_test.py
+++ b/paimon-python/pypaimon/tests/daft/daft_sink_test.py
@@ -25,6 +25,8 @@ reader to ensure correctness.
from __future__ import annotations
+import os
+
import pyarrow as pa
import pytest
@@ -35,6 +37,7 @@ from pypaimon.daft.daft_compat import
file_range_position_field, has_file_range_
from pypaimon.daft.daft_catalog import PaimonTable
from pypaimon.daft.daft_datasink import PaimonDataSink
from pypaimon.daft.daft_paimon import _read_table, _write_table
+from daft.recordbatch.micropartition import MicroPartition
requires_blob = pytest.mark.skipif(not has_file_range_reads(), reason="BLOB
support requires daft >= 0.7.11")
@@ -59,6 +62,16 @@ def _write_to_paimon(table, arrow_table, mode="append",
overwrite_partition=None
table_commit.close()
+def _table_data_files(table):
+ data_files = []
+ for dirpath, _, filenames in os.walk(table.table_path):
+ for filename in filenames:
+ if filename.endswith((".parquet", ".orc", ".avro")):
+ path = os.path.join(dirpath, filename)
+ data_files.append(os.path.relpath(path, table.table_path))
+ return sorted(data_files)
+
+
def _create_id_dt_table(catalog, table_name: str):
schema = pypaimon.Schema.from_pyarrow_schema(
pa.schema([
@@ -337,6 +350,42 @@ def
test_write_paimon_rejects_extra_columns(local_paimon_catalog):
_write_table(df, table)
+def
test_write_paimon_aborts_data_files_after_later_micropartition_fails(local_paimon_catalog):
+ """A failed write task must not leave files from earlier
micropartitions."""
+ catalog, _ = local_paimon_catalog
+ schema = pypaimon.Schema.from_pyarrow_schema(
+ pa.schema([
+ pa.field("id", pa.int64()),
+ pa.field("name", pa.string()),
+ ]),
+ options={
+ "bucket": "1",
+ "file.format": "parquet",
+ "target-file-size": "1kb",
+ },
+ )
+ catalog.create_table("test_db.abort_failed_write", schema,
ignore_if_exists=False)
+ table = catalog.get_table("test_db.abort_failed_write")
+
+ valid = MicroPartition.from_arrow(
+ pa.table({
+ "id": pa.array(list(range(128)), type=pa.int64()),
+ "name": pa.array([f"name-{i:03d}" for i in range(128)],
type=pa.string()),
+ })
+ )
+ invalid = MicroPartition.from_arrow(
+ pa.table({
+ "id": pa.array([999], type=pa.int64()),
+ "extra": pa.array(["bad"], type=pa.string()),
+ })
+ )
+
+ with pytest.raises(ValueError, match="Paimon write schema mismatch"):
+ list(PaimonDataSink(table).write(iter([valid, invalid])))
+
+ assert _table_data_files(table) == []
+
+
def test_write_paimon_pk_table(pk_table):
"""Writing to a PK table should work and be readable back."""
table, _ = pk_table
