This is an automated email from the ASF dual-hosted git repository. XiaoHongbo-Hope pushed a commit to branch release-2.1 in repository https://gitbox.apache.org/repos/asf/paimon.git
commit ae04d95ffd9a9d2a34af92b2ed5b35782e20e957 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:
