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 cfa3612f5f [python] Clean up normal files when composite writer
metadata creation fails (#9917)
cfa3612f5f is described below
commit cfa3612f5f60ab8f5970c9bcb9c0a9bfcad96bfa
Author: zhigang <[email protected]>
AuthorDate: Thu Sep 17 16:48:00 2026 +0800
[python] Clean up normal files when composite writer metadata creation
fails (#9917)
---
.../tests/write/writer_file_cleanup_test.py | 58 ++++++++++++
.../pypaimon/write/writer/data_vector_writer.py | 103 +++++++++++----------
.../write/writer/dedicated_format_writer.py | 51 +++++-----
3 files changed, 140 insertions(+), 72 deletions(-)
diff --git a/paimon-python/pypaimon/tests/write/writer_file_cleanup_test.py
b/paimon-python/pypaimon/tests/write/writer_file_cleanup_test.py
new file mode 100644
index 0000000000..8286ea989c
--- /dev/null
+++ b/paimon-python/pypaimon/tests/write/writer_file_cleanup_test.py
@@ -0,0 +1,58 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Failed normal-file cleanup for composite data writers."""
+
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+import pyarrow as pa
+
+from pypaimon import CatalogFactory, Schema
+from pypaimon.write.writer.data_vector_writer import DataVectorWriter
+from pypaimon.write.writer.dedicated_format_writer import DedicatedFormatWriter
+
+
+class WriterFileCleanupTest(unittest.TestCase):
+ def test_metadata_failure_deletes_written_file(self):
+ for writer_type in (DataVectorWriter, DedicatedFormatWriter):
+ with self.subTest(writer=writer_type.__name__),
tempfile.TemporaryDirectory() as tmp:
+ catalog = CatalogFactory.create({'warehouse': tmp})
+ catalog.create_database('db', False)
+ data = pa.table({'id': [1]})
+ schema = data.schema
+ if writer_type is DedicatedFormatWriter:
+ schema = schema.append(pa.field('payload',
pa.large_binary()))
+ catalog.create_table('db.t',
Schema.from_pyarrow_schema(schema, options={
+ 'file.format': 'parquet',
+ 'data-evolution.enabled': 'true',
+ 'row-tracking.enabled': 'true',
+ }), False)
+ table = catalog.get_table('db.t')
+ writer = writer_type(table, (), 0, 0, table.options,
write_cols=['id'])
+ writer.write(data.to_batches()[0])
+
+ def fail_after_output(path):
+ self.assertTrue(Path(path).is_file())
+ raise OSError('cannot read file size')
+
+ with patch.object(writer.file_io, 'get_file_size',
side_effect=fail_after_output):
+ with self.assertRaisesRegex(OSError, 'cannot read file
size'):
+ writer.prepare_commit()
+ self.assertEqual(list(Path(tmp).rglob('*.parquet')), [])
diff --git a/paimon-python/pypaimon/write/writer/data_vector_writer.py
b/paimon-python/pypaimon/write/writer/data_vector_writer.py
index 4825b22af1..63f5cbc136 100644
--- a/paimon-python/pypaimon/write/writer/data_vector_writer.py
+++ b/paimon-python/pypaimon/write/writer/data_vector_writer.py
@@ -266,55 +266,60 @@ class DataVectorWriter(DataWriter):
file_name =
f"{CoreOptions.data_file_prefix(self.options)}{uuid.uuid4()}-0.{self.file_format}"
file_path = self._generate_file_path(file_name)
- if self.file_format == CoreOptions.FILE_FORMAT_PARQUET:
- shredding_stats = self._write_parquet_data(file_path, data)
- elif self.file_format == CoreOptions.FILE_FORMAT_ORC:
- self.file_io.write_orc(file_path, data,
compression=self.compression, zstd_level=self.zstd_level)
- elif self.file_format == CoreOptions.FILE_FORMAT_AVRO:
- self.file_io.write_avro(file_path, data,
compression=self.compression, zstd_level=self.zstd_level)
- elif self.file_format == CoreOptions.FILE_FORMAT_LANCE:
- self.file_io.write_lance(file_path, data)
- elif self.file_format == CoreOptions.FILE_FORMAT_VORTEX:
- self.file_io.write_vortex(file_path, data)
- elif self.file_format == CoreOptions.FILE_FORMAT_MOSAIC:
- self.file_io.write_mosaic(file_path, data,
options=self.mosaic_writer_options)
- elif self.file_format == CoreOptions.FILE_FORMAT_ROW:
- self.file_io.write_row(file_path, data, zstd_level=self.zstd_level)
- else:
- raise ValueError(f"Unsupported file format: {self.file_format}")
-
- is_external_path = self.external_path_provider is not None
- external_path_str = file_path if is_external_path else None
-
- metadata_stats_enabled = self.options.metadata_stats_enabled()
- stats_columns = self.normal_columns if metadata_stats_enabled else []
- value_stats = self._collect_value_stats(data, stats_columns)
-
- min_seq, max_seq = self._append_file_sequence_range(data.num_rows)
-
- meta = DataFileMeta.create(
- file_name=file_name,
- file_size=self.file_io.get_file_size(file_path),
- row_count=data.num_rows,
- min_key=GenericRow([], []),
- max_key=GenericRow([], []),
- key_stats=SimpleStats.empty_stats(),
- value_stats=value_stats,
- min_sequence_number=min_seq,
- max_sequence_number=max_seq,
- schema_id=self.table.table_schema.id,
- level=0,
- extra_files=[],
- creation_time=Timestamp.now(),
- delete_row_count=0,
- file_source=0,
- value_stats_cols=[column.name for column in stats_columns],
- external_path=external_path_str,
- file_path=file_path,
- write_cols=self.write_cols,
- )
- self._map_shared_shredding.file_completed(shredding_stats)
- return meta
+ # Until metadata is returned, no caller can track this file for abort.
+ try:
+ if self.file_format == CoreOptions.FILE_FORMAT_PARQUET:
+ shredding_stats = self._write_parquet_data(file_path, data)
+ elif self.file_format == CoreOptions.FILE_FORMAT_ORC:
+ self.file_io.write_orc(file_path, data,
compression=self.compression, zstd_level=self.zstd_level)
+ elif self.file_format == CoreOptions.FILE_FORMAT_AVRO:
+ self.file_io.write_avro(file_path, data,
compression=self.compression, zstd_level=self.zstd_level)
+ elif self.file_format == CoreOptions.FILE_FORMAT_LANCE:
+ self.file_io.write_lance(file_path, data)
+ elif self.file_format == CoreOptions.FILE_FORMAT_VORTEX:
+ self.file_io.write_vortex(file_path, data)
+ elif self.file_format == CoreOptions.FILE_FORMAT_MOSAIC:
+ self.file_io.write_mosaic(file_path, data,
options=self.mosaic_writer_options)
+ elif self.file_format == CoreOptions.FILE_FORMAT_ROW:
+ self.file_io.write_row(file_path, data,
zstd_level=self.zstd_level)
+ else:
+ raise ValueError(f"Unsupported file format:
{self.file_format}")
+
+ is_external_path = self.external_path_provider is not None
+ external_path_str = file_path if is_external_path else None
+
+ metadata_stats_enabled = self.options.metadata_stats_enabled()
+ stats_columns = self.normal_columns if metadata_stats_enabled else
[]
+ value_stats = self._collect_value_stats(data, stats_columns)
+
+ min_seq, max_seq = self._append_file_sequence_range(data.num_rows)
+
+ meta = DataFileMeta.create(
+ file_name=file_name,
+ file_size=self.file_io.get_file_size(file_path),
+ row_count=data.num_rows,
+ min_key=GenericRow([], []),
+ max_key=GenericRow([], []),
+ key_stats=SimpleStats.empty_stats(),
+ value_stats=value_stats,
+ min_sequence_number=min_seq,
+ max_sequence_number=max_seq,
+ schema_id=self.table.table_schema.id,
+ level=0,
+ extra_files=[],
+ creation_time=Timestamp.now(),
+ delete_row_count=0,
+ file_source=0,
+ value_stats_cols=[column.name for column in stats_columns],
+ external_path=external_path_str,
+ file_path=file_path,
+ write_cols=self.write_cols,
+ )
+ self._map_shared_shredding.file_completed(shredding_stats)
+ return meta
+ except Exception:
+ self.file_io.delete_quietly(file_path)
+ raise
def _validate_consistency(
self, normal_meta: DataFileMeta, vector_metas: List[DataFileMeta]):
diff --git a/paimon-python/pypaimon/write/writer/dedicated_format_writer.py
b/paimon-python/pypaimon/write/writer/dedicated_format_writer.py
index f79369d1cd..22dc87be0e 100644
--- a/paimon-python/pypaimon/write/writer/dedicated_format_writer.py
+++ b/paimon-python/pypaimon/write/writer/dedicated_format_writer.py
@@ -708,31 +708,36 @@ class DedicatedFormatWriter(DataWriter):
file_name =
f"{CoreOptions.data_file_prefix(self.options)}{uuid.uuid4()}-0.{self.file_format}"
file_path = self._generate_file_path(file_name)
- # Write file based on format
- if self.file_format == CoreOptions.FILE_FORMAT_PARQUET:
- shredding_stats = self._write_parquet_data(file_path, data)
- elif self.file_format == CoreOptions.FILE_FORMAT_ORC:
- self.file_io.write_orc(file_path, data,
compression=self.compression, zstd_level=self.zstd_level)
- elif self.file_format == CoreOptions.FILE_FORMAT_AVRO:
- self.file_io.write_avro(file_path, data,
compression=self.compression, zstd_level=self.zstd_level)
- elif self.file_format == CoreOptions.FILE_FORMAT_LANCE:
- self.file_io.write_lance(file_path, data)
- elif self.file_format == CoreOptions.FILE_FORMAT_VORTEX:
- self.file_io.write_vortex(file_path, data)
- elif self.file_format == CoreOptions.FILE_FORMAT_MOSAIC:
- self.file_io.write_mosaic(file_path, data,
options=self.mosaic_writer_options)
- elif self.file_format == CoreOptions.FILE_FORMAT_ROW:
- self.file_io.write_row(file_path, data, zstd_level=self.zstd_level)
- else:
- raise ValueError(f"Unsupported file format: {self.file_format}")
+ # Until metadata is returned, no caller can track this file for abort.
+ try:
+ # Write file based on format
+ if self.file_format == CoreOptions.FILE_FORMAT_PARQUET:
+ shredding_stats = self._write_parquet_data(file_path, data)
+ elif self.file_format == CoreOptions.FILE_FORMAT_ORC:
+ self.file_io.write_orc(file_path, data,
compression=self.compression, zstd_level=self.zstd_level)
+ elif self.file_format == CoreOptions.FILE_FORMAT_AVRO:
+ self.file_io.write_avro(file_path, data,
compression=self.compression, zstd_level=self.zstd_level)
+ elif self.file_format == CoreOptions.FILE_FORMAT_LANCE:
+ self.file_io.write_lance(file_path, data)
+ elif self.file_format == CoreOptions.FILE_FORMAT_VORTEX:
+ self.file_io.write_vortex(file_path, data)
+ elif self.file_format == CoreOptions.FILE_FORMAT_MOSAIC:
+ self.file_io.write_mosaic(file_path, data,
options=self.mosaic_writer_options)
+ elif self.file_format == CoreOptions.FILE_FORMAT_ROW:
+ self.file_io.write_row(file_path, data,
zstd_level=self.zstd_level)
+ else:
+ raise ValueError(f"Unsupported file format:
{self.file_format}")
- # Determine if this is an external path
- is_external_path = self.external_path_provider is not None
- external_path_str = file_path if is_external_path else None
+ # Determine if this is an external path
+ is_external_path = self.external_path_provider is not None
+ external_path_str = file_path if is_external_path else None
- meta = self._create_data_file_meta(file_name, file_path, data,
external_path_str)
- self._map_shared_shredding.file_completed(shredding_stats)
- return meta
+ meta = self._create_data_file_meta(file_name, file_path, data,
external_path_str)
+ self._map_shared_shredding.file_completed(shredding_stats)
+ return meta
+ except Exception:
+ self.file_io.delete_quietly(file_path)
+ raise
def _create_data_file_meta(self, file_name: str, file_path: str, data:
pa.Table,
external_path: Optional[str] = None) ->
DataFileMeta:
