diff --git a/docs/docs/concepts/spec/manifest.md b/docs/docs/concepts/spec/manifest.md index 6f72db906743..497f0e2d1fcb 100644 --- a/docs/docs/concepts/spec/manifest.md +++ b/docs/docs/concepts/spec/manifest.md @@ -90,12 +90,14 @@ Selected block bytes still share the manifest content cache without populating t whole-manifest entry cache with partial results. The low-level `build` method returns sidecar bytes without writing or publishing another file. -PyPaimon can read these sidecars and prune manifest blocks using partition, row-ID and bucket -filters. Its `manifest.sidecar.enabled` option inherits `manifest-sort.enabled` when unset. +PyPaimon generates sidecars for newly written manifests and can prune manifest blocks using +partition, row-ID and bucket filters. Its `manifest.sidecar.enabled` option controls reads and +writes and inherits `manifest-sort.enabled` when unset. Ordinary and rolling writes publish +`_EXTRA_FILES` references only after both the manifest and sidecar close successfully. Failed +writes, merges and commit cleanup remove the sidecars with their owning new manifests. Entry filters and ADD/DELETE reconciliation still apply after block selection. Missing or unusable sidecars fall back to full manifest reads; scans without pruning filters and -explain/statistics scans do not perform sidecar I/O. The standalone codec can build sidecar -bytes, but automatic Python writer publication and cleanup are not integrated yet. +explain/statistics scans do not perform sidecar I/O. Callers decide whether to invoke `build` and `read`; these utilities have no read/write switches. `build` and `Builder` accept `rowIdEnabled` and `bucketEnabled` arguments for independent diff --git a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFileMetaSerializer.java b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFileMetaSerializer.java index 30f74a305ade..b00135c8edfa 100644 --- a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFileMetaSerializer.java +++ b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFileMetaSerializer.java @@ -96,7 +96,9 @@ private ManifestFileMeta fromDataRow(InternalRow row) { row.isNullAt(9) ? null : row.getInt(9), row.isNullAt(10) ? null : row.getLong(10), row.isNullAt(11) ? null : row.getLong(11), - row.isNullAt(12) ? null : row.getInt(12), - row.isNullAt(13) ? null : fromStringArrayData(row.getArray(13))); + row.getFieldCount() <= 12 || row.isNullAt(12) ? null : row.getInt(12), + row.getFieldCount() <= 13 || row.isNullAt(13) + ? null + : fromStringArrayData(row.getArray(13))); } } diff --git a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/LegacyDataEvolutionRowIdReassigner.java b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/LegacyDataEvolutionRowIdReassigner.java index 641541499028..e113923f31cd 100644 --- a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/LegacyDataEvolutionRowIdReassigner.java +++ b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/LegacyDataEvolutionRowIdReassigner.java @@ -886,7 +886,8 @@ private List readPlanningManifestEntries( null, Filter.alwaysTrue(), entry -> partitionPredicate == null || partitionPredicate.test(entry.partition()), - ManifestEntry::copyWithoutStats); + ManifestEntry::copyWithoutStats, + null); } private Comparator entryComparator() { diff --git a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaSerializerTest.java b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaSerializerTest.java index 04f41aa8204f..0b2dc24baf09 100644 --- a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaSerializerTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaSerializerTest.java @@ -18,6 +18,7 @@ package org.apache.paimon.manifest; +import org.apache.paimon.data.GenericRow; import org.apache.paimon.utils.ObjectSerializer; import org.apache.paimon.utils.ObjectSerializerTestBase; @@ -86,6 +87,19 @@ void testExtraFiles() throws IOException { } } + @Test + void testOldRowWithoutExtraFilesField() { + ManifestFileMeta meta = object(); + ManifestFileMetaSerializer serializer = new ManifestFileMetaSerializer(); + GenericRow current = (GenericRow) serializer.toRow(meta); + GenericRow legacy = new GenericRow(current.getFieldCount() - 1); + for (int field = 0; field < legacy.getFieldCount(); field++) { + legacy.setField(field, current.getField(field)); + } + assertThat(serializer.fromRow(legacy)).isEqualTo(meta); + assertThat(serializer.fromRow(legacy).extraFiles()).isNull(); + } + @Override protected ObjectSerializer serializer() { return new ManifestFileMetaSerializer(); diff --git a/paimon-python/pypaimon/common/options/core_options.py b/paimon-python/pypaimon/common/options/core_options.py index 7df8ee0ef25f..4256a9dc41f4 100644 --- a/paimon-python/pypaimon/common/options/core_options.py +++ b/paimon-python/pypaimon/common/options/core_options.py @@ -305,7 +305,7 @@ class CoreOptions: ConfigOptions.key("manifest.sidecar.enabled") .boolean_type() .no_default_value() - .with_description("Enable sidecar pruning on reads. Defaults to manifest-sort.enabled when unset.") + .with_description("Enable manifest sidecar reads and writes. Defaults to manifest-sort.enabled when unset.") ) MANIFEST_SORT_ENABLED: ConfigOption[bool] = ( diff --git a/paimon-python/pypaimon/manifest/manifest_file_manager.py b/paimon-python/pypaimon/manifest/manifest_file_manager.py index 1d833f3d3516..6852099b4a30 100644 --- a/paimon-python/pypaimon/manifest/manifest_file_manager.py +++ b/paimon-python/pypaimon/manifest/manifest_file_manager.py @@ -32,7 +32,7 @@ from datetime import datetime from pypaimon.manifest.manifest_sidecar import ( - Query, read_sidecar, read_selected_bytes, + Settings, SUFFIX, Query, build_from_entries, read_sidecar, read_selected_bytes, ) from pypaimon.manifest.schema.data_file_meta import DataFileMeta from pypaimon.manifest.schema.manifest_entry import (MANIFEST_ENTRY_SCHEMA, @@ -342,12 +342,15 @@ def _get_value_stats_fields(self, file_dict: dict, file_schema) -> List: fields = [data_field_dict[col] for col in file_dict['_VALUE_STATS_COLS']] return fields - def write(self, file_name, entries: List[ManifestEntry]): + def _sidecar_settings(self): + return Settings.from_options(self.table.options) + + def write(self, file_name, entries: List[ManifestEntry]) -> ManifestFileMeta: buf = BytesIO() fastavro.writer( buf, MANIFEST_ENTRY_SCHEMA, self._to_avro_records(entries), codec=self._codec) - self._flush(file_name, buf.getvalue()) + return self._flush(file_name, buf.getvalue(), entries) def rolling_write(self, entries: List[ManifestEntry], suggested_file_size: int, @@ -371,10 +374,9 @@ def rolling_write(self, entries: List[ManifestEntry], writer.flush() avro_bytes = buf.getvalue() file_name = f"{name_prefix}-{len(result)}" - self._flush(file_name, avro_bytes) - written_files.append(file_name) - result.append(self._build_meta( - file_name, entries[chunk_start:i + 1], len(avro_bytes))) + meta = self._flush(file_name, avro_bytes, entries[chunk_start:i + 1]) + written_files.append(meta) + result.append(meta) chunk_start = i + 1 buf = BytesIO() writer = Writer( @@ -385,13 +387,12 @@ def rolling_write(self, entries: List[ManifestEntry], writer.flush() avro_bytes = buf.getvalue() file_name = f"{name_prefix}-{len(result)}" - self._flush(file_name, avro_bytes) - written_files.append(file_name) - result.append(self._build_meta( - file_name, entries[chunk_start:], len(avro_bytes))) - except Exception: - for fname in written_files: - self.file_io.delete_quietly(f"{self.manifest_path}/{fname}") + meta = self._flush(file_name, avro_bytes, entries[chunk_start:]) + written_files.append(meta) + result.append(meta) + except BaseException: + for meta in written_files: + self.delete(meta) raise return result @@ -438,17 +439,37 @@ def _to_avro_record(entry: ManifestEntry) -> dict: def _to_avro_records(self, entries: List[ManifestEntry]) -> List[dict]: return [self._to_avro_record(e) for e in entries] - def _flush(self, file_name: str, avro_bytes: bytes): + def delete(self, manifest: ManifestFileMeta): + self.file_io.delete_quietly(f"{self.manifest_path}/{manifest.file_name}") + for extra_file in manifest.extra_files or []: + self.file_io.delete_quietly(f"{self.manifest_path}/{extra_file}") + + def _flush(self, file_name: str, avro_bytes: bytes, entries: List[ManifestEntry]) -> ManifestFileMeta: manifest_path = f"{self.manifest_path}/{file_name}" + sidecar_file_name = None try: with self.file_io.new_output_stream(manifest_path) as output_stream: output_stream.write(avro_bytes) - except Exception as e: + settings = self._sidecar_settings() + if settings.enabled: + data = build_from_entries(avro_bytes, entries, settings) + if data is not None: + sidecar_file_name = file_name + SUFFIX + with self.file_io.new_output_stream(f"{self.manifest_path}/{sidecar_file_name}") as output_stream: + output_stream.write(data) + # Publish the reference only after both objects close successfully. + return self._build_meta(file_name, entries, len(avro_bytes), + [sidecar_file_name] if sidecar_file_name is not None else None) + except BaseException as e: self.file_io.delete_quietly(manifest_path) + if sidecar_file_name is not None: + self.file_io.delete_quietly(f"{self.manifest_path}/{sidecar_file_name}") + if not isinstance(e, Exception) or isinstance(e, InterruptedError): + raise raise RuntimeError(f"Failed to write manifest file: {e}") from e def _build_meta(self, file_name: str, entries: List[ManifestEntry], - file_size: int = None) -> ManifestFileMeta: + file_size: int = None, extra_files: Optional[List[str]] = None) -> ManifestFileMeta: added_file_count = 0 deleted_file_count = 0 schema_id = None @@ -480,7 +501,9 @@ def _build_meta(self, file_name: str, entries: List[ManifestEntry], min_row_id = None max_row_id = None for entry in entries: - if entry.file.first_row_id is None: + if (entry.file.first_row_id is None or entry.file.first_row_id < 0 + or entry.file.row_count <= 0 + or entry.file.row_count - 1 > (1 << 63) - 1 - entry.file.first_row_id): min_row_id = None max_row_id = None break @@ -516,5 +539,6 @@ def _build_meta(self, file_name: str, entries: List[ManifestEntry], max_level=max((e.file.level for e in entries), default=None), min_row_id=min_row_id, max_row_id=max_row_id, + extra_files=extra_files, total_buckets=total_buckets, ) diff --git a/paimon-python/pypaimon/manifest/manifest_file_merger.py b/paimon-python/pypaimon/manifest/manifest_file_merger.py index 821b12aef5e9..2f14f44e7736 100644 --- a/paimon-python/pypaimon/manifest/manifest_file_merger.py +++ b/paimon-python/pypaimon/manifest/manifest_file_merger.py @@ -93,8 +93,4 @@ def _merge_candidates(self, candidates: List[ManifestFileMeta], def _delete_manifests(self, manifests: List[ManifestFileMeta]): for manifest in manifests: - manifest_path = "{}/{}".format( - self.manifest_file_manager.manifest_path, - manifest.file_name, - ) - self.manifest_file_manager.file_io.delete_quietly(manifest_path) + self.manifest_file_manager.delete(manifest) diff --git a/paimon-python/pypaimon/tests/manifest/manifest_sidecar_test.py b/paimon-python/pypaimon/tests/manifest/manifest_sidecar_test.py index 405c4c92b14a..1b64d7b4d527 100644 --- a/paimon-python/pypaimon/tests/manifest/manifest_sidecar_test.py +++ b/paimon-python/pypaimon/tests/manifest/manifest_sidecar_test.py @@ -18,6 +18,7 @@ import os import unittest from concurrent.futures import CancelledError +from contextlib import contextmanager from copy import deepcopy from io import BytesIO from itertools import product @@ -442,30 +443,142 @@ def entry(self, name, first, count=10, kind=0): return ManifestEntry(kind, self._create_file_meta('unused').min_key, 0, 1, replace(self._create_file_meta(name), first_row_id=first, row_count=count)) - def test_enabling_sidecar_reads_does_not_change_python_writes(self): + def write_meta(self, name, entries): + manager = self.manifest_file_manager + return manager.write(name, entries) + + def test_explicit_disable_keeps_writes_plain_when_manifest_sort_is_enabled(self): + self.table.options.options.set(CoreOptions.MANIFEST_SIDECAR_ENABLED, False) manager = self.manifest_file_manager - entries = [self.entry('data.parquet', 100)] - self.assertTrue(self.table.options.manifest_sidecar_enabled()) - self.assertIsNone(manager.write('plain-writer', entries)) - self.assertFalse(Path(manager.manifest_path, 'plain-writer' + SUFFIX).exists()) - for meta in manager.rolling_write(entries, 1, 'rolling-writer'): + entries = [self.entry('file-%d.parquet' % i, i * 100) for i in range(3)] + metas = [manager.write('plain', entries)] + manager.rolling_write(entries, 1, 'plain-rolling') + for meta in metas: self.assertIsNone(meta.extra_files) self.assertFalse(Path(manager.manifest_path, meta.file_name + SUFFIX).exists()) + self.assertTrue(manager.read(meta.file_name)) + + @staticmethod + def failing_sidecar_output(original, target, phase, error): + @contextmanager + def output(path): + with original(path) as stream: + if path.endswith(target) and phase == 'write': + wrapped = Mock(wraps=stream) + + def partial_write(data): + stream.write(data[:3]) + raise error + + wrapped.write.side_effect = partial_write + yield wrapped + else: + yield stream + if path.endswith(target) and phase == 'close': + raise error + return output + + def test_sidecar_write_and_close_failures_do_not_publish_references(self): + manager = self.manifest_file_manager + kept = manager.write('kept', [self.entry('kept.parquet', 100)]) + original = manager.file_io.new_output_stream + for phase in ('write', 'close'): + for error in (OSError('failed'), InterruptedError('cancelled'), KeyboardInterrupt()): + with self.subTest(phase=phase, error=type(error).__name__): + stream = self.failing_sidecar_output(original, 'failed' + SUFFIX, phase, error) + expected = type(error) if isinstance(error, (InterruptedError, KeyboardInterrupt)) else RuntimeError + with patch.object(manager.file_io, 'new_output_stream', side_effect=stream), \ + patch.object(manager, '_build_meta', wraps=manager._build_meta) as publish: + with self.assertRaises(expected): + manager.write('failed', [self.entry('new.parquet', 200)]) + publish.assert_not_called() + self.assertFalse(Path(manager.manifest_path, 'failed').exists()) + self.assertFalse(Path(manager.manifest_path, 'failed' + SUFFIX).exists()) + self.assertTrue(Path(manager.manifest_path, kept.file_name).exists()) + self.assertTrue(Path(manager.manifest_path, sidecar_file_name(kept)).exists()) + + def test_later_rolling_failure_cleans_completed_manifest_sidecar_pairs(self): + manager = self.manifest_file_manager + manager.write('kept', [self.entry('kept.parquet', 100)]) + before = {p.name for p in Path(manager.manifest_path).iterdir()} + original = manager.file_io.new_output_stream + stream = self.failing_sidecar_output( + original, 'rolling-failed-1' + SUFFIX, 'close', OSError('second sidecar close failed')) + with patch.object(manager.file_io, 'new_output_stream', side_effect=stream): + with self.assertRaises(RuntimeError): + manager.rolling_write([self.entry('data-%d.parquet' % i, i * 100) for i in range(3)], + 1, 'rolling-failed') + self.assertEqual({p.name for p in Path(manager.manifest_path).iterdir()}, before) + + def test_commit_cleanup_removes_only_owned_manifest_sidecar_pairs(self): + from pypaimon.write.file_store_commit import FileStoreCommit - def write_meta(self, name, entries): - # Emulate an externally published sidecar; production Python writes are unchanged. manager = self.manifest_file_manager - manager.write(name, entries) - path = Path(manager.manifest_path, name) - avro_bytes = path.read_bytes() - meta = manager._build_meta(name, entries, len(avro_bytes)) - settings = Settings.from_options(self.table.options) - if settings.enabled: - data = manifest_sidecar.build_from_entries(avro_bytes, entries, settings) - sidecar_path = path.with_name(name + SUFFIX) - sidecar_path.write_bytes(data) - meta = replace(meta, extra_files=[sidecar_path.name]) - return meta + lists = ManifestListManager(self.table) + kept = manager.write('kept', [self.entry('kept.parquet', 100)]) + delta = manager.write('delta', [self.entry('delta.parquet', 200)]) + changelog = manager.write('changelog', [self.entry('changelog.parquet', 300)]) + merged = manager.write('merged', [self.entry('merged.parquet', 400)]) + lists.write('delta-list', [delta]) + lists.write('changelog-list', [changelog]) + lists.write('base-list', [kept, merged]) + committer = SimpleNamespace(table=self.table, manifest_file_manager=manager, manifest_list_manager=lists) + FileStoreCommit._clean_up_reuse_tmp_manifests(committer, 'delta-list', 'changelog-list') + FileStoreCommit._clean_up_no_reuse_tmp_manifests(committer, 'base-list', [merged]) + self.assertEqual({p.name for p in Path(manager.manifest_path).iterdir()}, + {kept.file_name, sidecar_file_name(kept)}) + + def test_invalid_row_id_ranges_keep_coarse_and_block_coverage_unknown(self): + for first, count in ((-1, 1), (100, 0), (MAX_ROW_ID, 2)): + with self.subTest(first=first, count=count): + manager = self.manifest_file_manager + meta = manager.write('invalid-%s-%s' % (first, count), [self.entry('data.parquet', first, count)]) + self.assertIsNone(meta.min_row_id) + self.assertIsNone(meta.max_row_id) + data = Path(manager.manifest_path, sidecar_file_name(meta)).read_bytes() + self.assertEqual(len(select(data, meta, [Range(50, 50)]).blocks), 1) + + def test_sidecars_do_not_depend_on_manifest_target_sizes(self): + for i, target in enumerate(('1 bytes', '1 gb', '9223372036854775807 bytes')): + with self.subTest(target=target): + self.table.options.options.set(CoreOptions.MANIFEST_TARGET_FILE_SIZE, target) + entry = self.entry('file.parquet', 100) + meta = self.write_meta(f'budget-{i}', [entry]) + self.assertEqual( + [e.file.file_name for e in self.manifest_file_manager.read(meta.file_name)], + [entry.file.file_name]) + self.assertIsNotNone(sidecar_file_name(meta)) + + def test_payloads_follow_table_metadata(self): + import pyarrow as pa + + for partitioned, data_evolution, bucket in product((False, True), (False, True), (-1, 4)): + with self.subTest(partitioned=partitioned, data_evolution=data_evolution, bucket=bucket): + name = f'default.sidecar_settings_{partitioned}_{data_evolution}_{bucket + 1}' + schema = Schema.from_pyarrow_schema( + pa.schema([('id', pa.int32()), ('value', pa.string())]), + partition_keys=['id'] if partitioned else [], + options={'manifest.sidecar.enabled': 'true', + 'data-evolution.enabled': str(data_evolution).lower(), 'bucket': str(bucket)}) + self.catalog.create_table(name, schema, False) + self.table = self.catalog.get_table(name) + manager = ManifestFileManager(self.table) + entry = replace(self.entry('file.parquet', 100), bucket=1, total_buckets=4, + partition=GenericRow([7] if partitioned else [], self.table.partition_keys_fields)) + metadata = manager.write('settings', [entry]) + with manager.file_io.new_input_stream( + manager.manifest_path + '/' + sidecar_file_name(metadata)) as stream: + data = stream.read() + encoded = manifest_sidecar._Buffer(data) + encoded.take(4) + encoded.uint() + encoded.take(encoded.uint()) + self.assertEqual(encoded.uint(), 1) + selected = select(data, metadata, None, bucket_filter=lambda bucket, total: bucket == 99) + self.assertEqual(len(selected.blocks), 1 if bucket == -1 else 0) + self.assertEqual( + len(select(data, metadata, [Range(99, 99)]).blocks), + 0 if data_evolution else 1) + self.assertEqual([e.file.file_name for e in manager.read(metadata.file_name)], ['file.parquet']) def test_bucket_point_lookup_with_rescale_and_delete_entries(self): self.table.options.options.set(CoreOptions.DATA_EVOLUTION_ENABLED, False) @@ -609,6 +722,10 @@ def test_explicit_reference_and_null_does_not_probe(self): self.assertEqual(len(actual), 1) self.assertEqual([call[0][0] for call in opened.call_args_list], [str(Path(manager.manifest_path, written.file_name))]) + manager.delete(indexed) + self.assertFalse(other_path.exists()) + self.assertFalse(explicit_path.exists()) + self.assertFalse(Path(manager.manifest_path, written.file_name).exists()) def test_manifest_list_index_reference_compatibility(self): indexed = self.write_meta('indexed', [self.entry('data.parquet', 100)]) @@ -799,3 +916,49 @@ def open_stream(path): self.assertEqual([call[0][0] for call in opened.call_args_list], [index_path, body_path]) if stream is not None: self.assertTrue(stream.closed) + + def test_rolling_merge_limits_and_abort_cleanup(self): + entries = [self.entry('file-%d' % i, i * 1000) for i in range(300)] + manager = self.manifest_file_manager + metas = manager.rolling_write(entries, 300, 'rolling') + self.assertGreater(len(metas), 1) + for meta in metas: + actual = manager.read(meta.file_name) + data = Path(manager.manifest_path, meta.file_name + SUFFIX).read_bytes() + for e in actual: + self.assertTrue(intersects(data, meta, [Range(e.file.first_row_id, e.file.first_row_id)], Settings())) + gap = actual[0].file.first_row_id + 10 + self.assertFalse(intersects(data, meta, [Range(gap, gap)], Settings())) + from pypaimon.manifest.manifest_file_merger import ManifestFileMerger + merger = ManifestFileMerger(manager, 1000000, 2) + outputs, new_files = merger.merge(metas) + self.assertTrue(new_files) + self.assertEqual( + sorted(e.file.file_name for meta in outputs for e in manager.read(meta.file_name)), + sorted(e.file.file_name for e in entries)) + for meta in outputs: + self.assertTrue(Path(manager.manifest_path, meta.file_name + SUFFIX).exists()) + for meta in metas: + self.assertIsNotNone(sidecar_file_name(meta)) + manager.delete(meta) + self.assertFalse(Path(manager.manifest_path, meta.file_name + SUFFIX).exists()) + original = self.table.file_io.new_output_stream + + def fail(path): + if path.endswith(SUFFIX): + raise OSError('sidecar write failed') + return original(path) + with patch.object(self.table.file_io, 'new_output_stream', side_effect=fail): + with self.assertRaises(RuntimeError): + manager.write('failed', entries[:1]) + self.assertFalse(Path(manager.manifest_path, 'failed').exists()) + self.assertFalse(Path(manager.manifest_path, 'failed' + SUFFIX).exists()) + unknown = manager.write('unknown', [self.entry('legacy', None)]) + self.assertIsNotNone(sidecar_file_name(unknown)) + data = Path(manager.manifest_path, sidecar_file_name(unknown)).read_bytes() + self.assertEqual(len(select(data, unknown, [Range(100, 100)]).blocks), 1) + huge = manager.write('huge', [self.entry('one', 0, MAX_ROW_ID), + self.entry('two', MAX_ROW_ID, 1)]) + self.assertIsNotNone(sidecar_file_name(huge)) + data = Path(manager.manifest_path, sidecar_file_name(huge)).read_bytes() + self.assertEqual(len(select(data, huge, [Range(50, 50)]).blocks), 1) diff --git a/paimon-python/pypaimon/write/file_store_commit.py b/paimon-python/pypaimon/write/file_store_commit.py index 27907f8741d1..97e2ac28717a 100644 --- a/paimon-python/pypaimon/write/file_store_commit.py +++ b/paimon-python/pypaimon/write/file_store_commit.py @@ -1154,8 +1154,7 @@ def _clean_up_reuse_tmp_manifests( if ml_name: try: for meta in self.manifest_list_manager.read(ml_name): - self.table.file_io.delete_quietly( - f"{self.manifest_file_manager.manifest_path}/{meta.file_name}") + self.manifest_file_manager.delete(meta) except Exception: pass self.table.file_io.delete_quietly(f"{manifest_path}/{ml_name}") @@ -1174,8 +1173,7 @@ def _clean_up_no_reuse_tmp_manifests( if base_manifest_list: self.table.file_io.delete_quietly(f"{manifest_path}/{base_manifest_list}") for meta in merge_new_files: - self.table.file_io.delete_quietly( - f"{self.manifest_file_manager.manifest_path}/{meta.file_name}") + self.manifest_file_manager.delete(meta) def abort(self, commit_messages: List[CommitMessage]): """Abort commit and delete files. Uses external_path if available to ensure proper scheme handling."""