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 fb5b3436a3 [python] Minor compact manifest files after commit (#8240)
fb5b3436a3 is described below
commit fb5b3436a33be53d69172e4422754d517b36e772
Author: Jingsong Lee <[email protected]>
AuthorDate: Mon Jun 15 16:36:31 2026 +0800
[python] Minor compact manifest files after commit (#8240)
Add Python-side minor manifest compaction after commits, matching the
minor path of Java ManifestFileMerger while intentionally avoiding full
compaction and manifest sort rewrite. This prevents committed Python
tables from accumulating many small base manifest files.
---
.../pypaimon/common/options/core_options.py | 25 +++++
.../pypaimon/manifest/manifest_file_manager.py | 61 +++++++++++-
.../pypaimon/manifest/manifest_file_merger.py | 102 ++++++++++++++++++++
.../pypaimon/tests/file_store_commit_test.py | 2 +
.../manifest/manifest_entry_identifier_test.py | 47 ++++++++++
.../pypaimon/tests/write/table_write_test.py | 51 ++++++++++
paimon-python/pypaimon/write/file_store_commit.py | 103 ++++++---------------
7 files changed, 317 insertions(+), 74 deletions(-)
diff --git a/paimon-python/pypaimon/common/options/core_options.py
b/paimon-python/pypaimon/common/options/core_options.py
index e84ea6c4e1..e2df232193 100644
--- a/paimon-python/pypaimon/common/options/core_options.py
+++ b/paimon-python/pypaimon/common/options/core_options.py
@@ -180,6 +180,23 @@ class CoreOptions:
.with_description("The parallelism for scanning manifest files.")
)
+ MANIFEST_TARGET_FILE_SIZE: ConfigOption[MemorySize] = (
+ ConfigOptions.key("manifest.target-file-size")
+ .memory_type()
+ .default_value(MemorySize.of_mebi_bytes(8))
+ .with_description("Suggested file size of a manifest file.")
+ )
+
+ MANIFEST_MERGE_MIN_COUNT: ConfigOption[int] = (
+ ConfigOptions.key("manifest.merge-min-count")
+ .int_type()
+ .default_value(30)
+ .with_description(
+ "To avoid frequent manifest merges, this parameter specifies the
minimum number "
+ "of ManifestFileMeta to merge."
+ )
+ )
+
# File format options
FILE_FORMAT: ConfigOption[str] = (
ConfigOptions.key("file.format")
@@ -778,6 +795,14 @@ class CoreOptions:
def scan_manifest_parallelism(self, default=None):
return self.options.get(CoreOptions.SCAN_MANIFEST_PARALLELISM, default)
+ def manifest_target_size(self, default=None):
+ if default is not None and not isinstance(default, MemorySize):
+ default = MemorySize.of_bytes(default) if isinstance(default, int)
else MemorySize.parse(default)
+ return self.options.get(CoreOptions.MANIFEST_TARGET_FILE_SIZE,
default).get_bytes()
+
+ def manifest_merge_min_count(self, default=None):
+ return self.options.get(CoreOptions.MANIFEST_MERGE_MIN_COUNT, default)
+
def file_format(self, default=None):
return self.options.get(CoreOptions.FILE_FORMAT, default)
diff --git a/paimon-python/pypaimon/manifest/manifest_file_manager.py
b/paimon-python/pypaimon/manifest/manifest_file_manager.py
index 6a0fd7dd6a..2d97516829 100644
--- a/paimon-python/pypaimon/manifest/manifest_file_manager.py
+++ b/paimon-python/pypaimon/manifest/manifest_file_manager.py
@@ -28,7 +28,8 @@ from pypaimon.manifest.schema.manifest_entry import
(MANIFEST_ENTRY_SCHEMA,
ManifestEntry)
from pypaimon.manifest.schema.manifest_file_meta import ManifestFileMeta
from pypaimon.manifest.schema.simple_stats import SimpleStats
-from pypaimon.table.row.generic_row import (GenericRowDeserializer,
+from pypaimon.table.row.generic_row import (GenericRow,
+ GenericRowDeserializer,
GenericRowSerializer)
from pypaimon.table.row.binary_row import BinaryRow
@@ -254,3 +255,61 @@ class ManifestFileManager:
except Exception as e:
self.file_io.delete_quietly(manifest_path)
raise RuntimeError(f"Failed to write manifest file: {e}") from e
+
+ def write_with_meta(self, file_name, entries: List[ManifestEntry]) ->
ManifestFileMeta:
+ self.write(file_name, entries)
+ added_file_count = 0
+ deleted_file_count = 0
+ schema_id = None
+ for entry in entries:
+ if entry.kind == 0:
+ added_file_count += 1
+ else:
+ deleted_file_count += 1
+ schema_id = entry.file.schema_id if schema_id is None else
max(schema_id, entry.file.schema_id)
+ if schema_id is None:
+ schema_id = self.table.table_schema.id
+
+ partition_columns = list(zip(*(entry.partition.values for entry in
entries))) if entries else []
+ partition_null_counts = [sum(1 for value in col if value is None) for
col in partition_columns]
+ partition_min_stats = [
+ min((v for v in col if v is not None), default=None) for col in
partition_columns
+ ]
+ partition_max_stats = [
+ max((v for v in col if v is not None), default=None) for col in
partition_columns
+ ]
+
+ min_row_id = None
+ max_row_id = None
+ for entry in entries:
+ if entry.file.first_row_id is None:
+ min_row_id = None
+ max_row_id = None
+ break
+ file_range = entry.file.row_id_range()
+ if min_row_id is None or file_range.from_ < min_row_id:
+ min_row_id = file_range.from_
+ if max_row_id is None or file_range.to > max_row_id:
+ max_row_id = file_range.to
+
+ manifest_file_path = f"{self.manifest_path}/{file_name}"
+ return ManifestFileMeta(
+ file_name=file_name,
+ file_size=self.table.file_io.get_file_size(manifest_file_path),
+ num_added_files=added_file_count,
+ num_deleted_files=deleted_file_count,
+ partition_stats=SimpleStats(
+ min_values=GenericRow(
+ values=partition_min_stats,
+ fields=self.table.partition_keys_fields
+ ),
+ max_values=GenericRow(
+ values=partition_max_stats,
+ fields=self.table.partition_keys_fields
+ ),
+ null_counts=partition_null_counts,
+ ),
+ schema_id=schema_id,
+ min_row_id=min_row_id,
+ max_row_id=max_row_id,
+ )
diff --git a/paimon-python/pypaimon/manifest/manifest_file_merger.py
b/paimon-python/pypaimon/manifest/manifest_file_merger.py
new file mode 100644
index 0000000000..47cc66b72e
--- /dev/null
+++ b/paimon-python/pypaimon/manifest/manifest_file_merger.py
@@ -0,0 +1,102 @@
+# 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.
+
+import uuid
+from typing import List, Tuple
+
+from pypaimon.manifest.schema.file_entry import FileEntry
+from pypaimon.manifest.schema.manifest_file_meta import ManifestFileMeta
+
+
+class ManifestFileMerger:
+ """Minor manifest compaction for Python commits.
+
+ This intentionally implements only the minor compaction path from Java
+ ManifestFileMerger. It does not do full compaction or manifest sort
rewrite.
+ """
+
+ def __init__(self, manifest_file_manager, suggested_meta_size: int,
+ suggested_min_meta_count: int):
+ self.manifest_file_manager = manifest_file_manager
+ self.suggested_meta_size = suggested_meta_size
+ self.suggested_min_meta_count = suggested_min_meta_count
+
+ def merge(self, manifest_files: List[ManifestFileMeta]) ->
Tuple[List[ManifestFileMeta],
+
List[ManifestFileMeta]]:
+ new_files = []
+ try:
+ return self._try_minor_compaction(manifest_files, new_files),
new_files
+ except Exception:
+ self._delete_manifests(new_files)
+ raise
+
+ def _try_minor_compaction(self, manifest_files: List[ManifestFileMeta],
+ new_files: List[ManifestFileMeta]) ->
List[ManifestFileMeta]:
+ result = []
+ candidates = []
+ total_size = 0
+
+ for manifest in manifest_files:
+ total_size += manifest.file_size
+ candidates.append(manifest)
+ if total_size >= self.suggested_meta_size:
+ self._merge_candidates(candidates, result, new_files)
+ candidates = []
+ total_size = 0
+
+ if len(candidates) >= self.suggested_min_meta_count:
+ self._merge_candidates(candidates, result, new_files)
+ else:
+ result.extend(candidates)
+
+ return result
+
+ def _merge_candidates(self, candidates: List[ManifestFileMeta],
+ result: List[ManifestFileMeta],
+ new_files: List[ManifestFileMeta]):
+ if len(candidates) == 1:
+ result.append(candidates[0])
+ return
+
+ entries = []
+ for manifest in candidates:
+ entries.extend(
+ self.manifest_file_manager.read(
+ manifest.file_name,
+ drop_stats=False,
+ )
+ )
+
+ merged_entries = FileEntry.merge_entries(entries)
+ if not merged_entries:
+ return
+
+ manifest_file = "manifest-{}-0".format(str(uuid.uuid4()))
+ merged_meta = self.manifest_file_manager.write_with_meta(
+ manifest_file,
+ merged_entries,
+ )
+ result.append(merged_meta)
+ new_files.append(merged_meta)
+
+ 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)
diff --git a/paimon-python/pypaimon/tests/file_store_commit_test.py
b/paimon-python/pypaimon/tests/file_store_commit_test.py
index 8b6b892bae..e3ff8c547f 100644
--- a/paimon-python/pypaimon/tests/file_store_commit_test.py
+++ b/paimon-python/pypaimon/tests/file_store_commit_test.py
@@ -40,6 +40,8 @@ class TestFileStoreCommit(unittest.TestCase):
self.mock_table.current_branch.return_value = 'main'
self.mock_table.table_path = '/test/table/path'
self.mock_table.file_io = Mock()
+ self.mock_table.options.manifest_target_size.return_value = 8 * 1024 *
1024
+ self.mock_table.options.manifest_merge_min_count.return_value = 30
# Mock snapshot commit
self.mock_snapshot_commit = Mock()
diff --git
a/paimon-python/pypaimon/tests/manifest/manifest_entry_identifier_test.py
b/paimon-python/pypaimon/tests/manifest/manifest_entry_identifier_test.py
index 9aa6522246..9638f04707 100644
--- a/paimon-python/pypaimon/tests/manifest/manifest_entry_identifier_test.py
+++ b/paimon-python/pypaimon/tests/manifest/manifest_entry_identifier_test.py
@@ -24,6 +24,7 @@ from pypaimon.common.identifier import Identifier
from pypaimon.common.options import Options
from pypaimon.common.options.config import CatalogOptions
from pypaimon.manifest.manifest_file_manager import ManifestFileManager
+from pypaimon.manifest.manifest_file_merger import ManifestFileMerger
from pypaimon.manifest.schema.data_file_meta import DataFileMeta
from pypaimon.manifest.schema.manifest_entry import ManifestEntry
from pypaimon.manifest.schema.manifest_file_meta import ManifestFileMeta
@@ -152,6 +153,52 @@ class ManifestEntryIdentifierTest(unittest.TestCase):
len(final_entries), 0,
"ADD and DELETE entries with same identifier should both be
removed")
+ def test_minor_compaction_cancels_add_delete_matching_same_file(self):
+ partition = GenericRow([], [])
+ add_entry = ManifestEntry(
+ kind=0,
+ partition=partition,
+ bucket=0,
+ total_buckets=1,
+ file=self._create_file_meta("data-1.parquet", level=0)
+ )
+ delete_entry = ManifestEntry(
+ kind=1,
+ partition=partition,
+ bucket=0,
+ total_buckets=1,
+ file=self._create_file_meta("data-1.parquet", level=0)
+ )
+
+ manifest_file_1 = ManifestFileMeta(
+ file_name="manifest-minor-1.avro",
+ file_size=1024,
+ num_added_files=1,
+ num_deleted_files=0,
+ partition_stats=SimpleStats.empty_stats(),
+ schema_id=0
+ )
+ manifest_file_2 = ManifestFileMeta(
+ file_name="manifest-minor-2.avro",
+ file_size=1024,
+ num_added_files=0,
+ num_deleted_files=1,
+ partition_stats=SimpleStats.empty_stats(),
+ schema_id=0
+ )
+ self.manifest_file_manager.write(manifest_file_1.file_name,
[add_entry])
+ self.manifest_file_manager.write(manifest_file_2.file_name,
[delete_entry])
+
+ merger = ManifestFileMerger(
+ self.manifest_file_manager,
+ suggested_meta_size=8 * 1024 * 1024,
+ suggested_min_meta_count=2,
+ )
+ merged_files, new_files = merger.merge([manifest_file_1,
manifest_file_2])
+
+ self.assertEqual(merged_files, [])
+ self.assertEqual(new_files, [])
+
def test_add_delete_different_levels(self):
"""
Test that entries with different levels are NOT matched.
diff --git a/paimon-python/pypaimon/tests/write/table_write_test.py
b/paimon-python/pypaimon/tests/write/table_write_test.py
index 7c31778c17..c22ebcd52c 100644
--- a/paimon-python/pypaimon/tests/write/table_write_test.py
+++ b/paimon-python/pypaimon/tests/write/table_write_test.py
@@ -29,6 +29,7 @@ from parameterized import parameterized
from pypaimon.common.json_util import JSON
from pypaimon.common.options.core_options import CoreOptions
+from pypaimon.manifest.manifest_list_manager import ManifestListManager
from pypaimon.write.writer.append_only_data_writer import AppendOnlyDataWriter
@@ -147,6 +148,56 @@ class TableWriteTest(unittest.TestCase):
actual = table_read.to_arrow(splits).sort_by('user_id')
self.assertEqual(self.expected, actual)
+ def test_commit_minor_compacts_manifest_files(self):
+ schema = Schema.from_pyarrow_schema(
+ self.pa_schema,
+ partition_keys=['dt'],
+ options={'manifest.merge-min-count': '2'},
+ )
+ self.catalog.create_table('default.test_minor_manifest_compaction',
schema, False)
+ table =
self.catalog.get_table('default.test_minor_manifest_compaction')
+
+ expected_data = {
+ 'user_id': [],
+ 'item_id': [],
+ 'behavior': [],
+ 'dt': [],
+ }
+ for i in range(3):
+ row = {
+ 'user_id': [i + 1],
+ 'item_id': [1000 + i],
+ 'behavior': ['click'],
+ 'dt': ['p1'],
+ }
+ for key, values in row.items():
+ expected_data[key].extend(values)
+
+ write_builder = table.new_batch_write_builder()
+ table_write = write_builder.new_write()
+ table_commit = write_builder.new_commit()
+ table_write.write_arrow(pa.Table.from_pydict(row,
schema=self.pa_schema))
+ table_commit.commit(table_write.prepare_commit())
+ table_write.close()
+ table_commit.close()
+
+ snapshot = table.snapshot_manager().get_latest_snapshot()
+ manifest_list_manager = ManifestListManager(table)
+ base_manifests =
manifest_list_manager.read(snapshot.base_manifest_list)
+ delta_manifests =
manifest_list_manager.read(snapshot.delta_manifest_list)
+
+ self.assertEqual(len(base_manifests), 1)
+ self.assertEqual(base_manifests[0].num_added_files, 2)
+ self.assertEqual(base_manifests[0].num_deleted_files, 0)
+ self.assertEqual(len(delta_manifests), 1)
+
+ expected = pa.Table.from_pydict(expected_data, schema=self.pa_schema)
+ read_builder = table.new_read_builder()
+ table_read = read_builder.new_read()
+ splits = read_builder.new_scan().plan().splits()
+ actual = table_read.to_arrow(splits).sort_by('user_id')
+ self.assertEqual(expected, actual)
+
def test_multi_prepare_commit_pk(self):
schema = Schema.from_pyarrow_schema(self.pa_schema,
partition_keys=['dt'], primary_keys=['user_id', 'dt'],
options={'bucket': '2'})
diff --git a/paimon-python/pypaimon/write/file_store_commit.py
b/paimon-python/pypaimon/write/file_store_commit.py
index 49f12cfac7..a2e41a1217 100644
--- a/paimon-python/pypaimon/write/file_store_commit.py
+++ b/paimon-python/pypaimon/write/file_store_commit.py
@@ -24,12 +24,12 @@ from typing import Dict, List, Optional
from pypaimon.common.options.core_options import CoreOptions
from pypaimon.common.predicate_builder import PredicateBuilder
from pypaimon.manifest.manifest_file_manager import ManifestFileManager
+from pypaimon.manifest.manifest_file_merger import ManifestFileMerger
from pypaimon.manifest.manifest_list_manager import ManifestListManager
from pypaimon.manifest.schema.data_file_meta import DataFileMeta
from pypaimon.manifest.schema.manifest_entry import ManifestEntry
from pypaimon.manifest.schema.manifest_file_meta import ManifestFileMeta
-from pypaimon.manifest.schema.simple_stats import SimpleStats
from pypaimon.read.scanner.file_scanner import FileScanner
from pypaimon.snapshot.snapshot import Snapshot
from pypaimon.snapshot.snapshot_commit import (PartitionStatistics,
@@ -92,8 +92,13 @@ class FileStoreCommit:
self.manifest_file_manager = ManifestFileManager(table)
self.manifest_list_manager = ManifestListManager(table)
- self.manifest_target_size = 8 * 1024 * 1024
- self.manifest_merge_min_count = 30
+ self.manifest_target_size = table.options.manifest_target_size()
+ self.manifest_merge_min_count =
table.options.manifest_merge_min_count()
+ self.manifest_file_merger = ManifestFileMerger(
+ self.manifest_file_manager,
+ self.manifest_target_size,
+ self.manifest_merge_min_count,
+ )
self.commit_max_retries = table.options.commit_max_retries()
self.commit_timeout = table.options.commit_timeout()
@@ -393,6 +398,7 @@ class FileStoreCommit:
changelog_manifest_list_name = None
changelog_manifest_list_size = None
changelog_record_count = None
+ new_manifest_files_for_abort = []
try:
new_manifest_file_meta = self._write_manifest_file(commit_entries,
new_manifest_file)
self.manifest_list_manager.write(delta_manifest_list,
[new_manifest_file_meta])
@@ -421,7 +427,9 @@ class FileStoreCommit:
total_record_count += previous_record_count
else:
existing_manifest_files = []
- self.manifest_list_manager.write(base_manifest_list,
existing_manifest_files)
+ merged_manifest_files, new_manifest_files_for_abort =
self.manifest_file_merger.merge(
+ existing_manifest_files)
+ self.manifest_list_manager.write(base_manifest_list,
merged_manifest_files)
delta_record_count = 0
for entry in commit_entries:
@@ -464,7 +472,8 @@ class FileStoreCommit:
statistics = self._generate_partition_statistics(commit_entries)
except Exception as e:
self._cleanup_preparation_failure(delta_manifest_list,
base_manifest_list,
- new_index_manifest,
changelog_manifest_list_name)
+ new_index_manifest,
changelog_manifest_list_name,
+ new_manifest_files_for_abort)
logger.warning(f"Exception occurs when preparing snapshot: {e}",
exc_info=True)
raise RuntimeError(f"Failed to prepare snapshot: {e}")
@@ -476,16 +485,13 @@ class FileStoreCommit:
commit_time_s = (int(time.time() * 1000) - start_millis) /
1000
logger.warning(
"Atomic commit failed for snapshot #%d by user %s "
- "with identifier %s and kind %s after %.0f seconds. "
- "Clean up and try again.",
+ "with identifier %s and kind %s after %.0f seconds.
Try again.",
new_snapshot_id,
self.commit_user,
commit_identifier,
commit_kind,
commit_time_s,
)
- self._cleanup_preparation_failure(delta_manifest_list,
base_manifest_list,
- new_index_manifest,
changelog_manifest_list_name)
return RetryResult(latest_snapshot, None)
except Exception as e:
# Commit exception, not sure about the situation and should not
clean up the files
@@ -514,65 +520,7 @@ class FileStoreCommit:
return SuccessResult()
def _write_manifest_file(self, commit_entries, new_manifest_file):
- # Write new manifest file
- self.manifest_file_manager.write(new_manifest_file, commit_entries)
-
- # Calculate file count & record count statistics
- added_file_count = 0
- deleted_file_count = 0
- for entry in commit_entries:
- if entry.kind == 0:
- added_file_count += 1
- else:
- deleted_file_count += 1
-
- # Calculate partition statistics
- partition_columns = list(zip(*(entry.partition.values for entry in
commit_entries)))
- partition_null_counts = [sum(1 for value in col if value is None) for
col in partition_columns]
- partition_min_stats = [
- min((v for v in col if v is not None), default=None) for col in
partition_columns
- ]
- partition_max_stats = [
- max((v for v in col if v is not None), default=None) for col in
partition_columns
- ]
-
- # Calculate min_row_id and max_row_id from commit_entries
- min_row_id = None
- max_row_id = None
- for entry in commit_entries:
- if entry.file.first_row_id is None:
- # If any file has first_row_id as None, set both min_row_id
and max_row_id to None
- min_row_id = None
- max_row_id = None
- break
- file_range = entry.file.row_id_range()
- if min_row_id is None or file_range.from_ < min_row_id:
- min_row_id = file_range.from_
- if max_row_id is None or file_range.to > max_row_id:
- max_row_id = file_range.to
-
- # return new ManifestFileMeta
- manifest_file_path =
f"{self.manifest_file_manager.manifest_path}/{new_manifest_file}"
- return ManifestFileMeta(
- file_name=new_manifest_file,
- file_size=self.table.file_io.get_file_size(manifest_file_path),
- num_added_files=added_file_count,
- num_deleted_files=deleted_file_count,
- partition_stats=SimpleStats(
- min_values=GenericRow(
- values=partition_min_stats,
- fields=self.table.partition_keys_fields
- ),
- max_values=GenericRow(
- values=partition_max_stats,
- fields=self.table.partition_keys_fields
- ),
- null_counts=partition_null_counts,
- ),
- schema_id=self.table.table_schema.id,
- min_row_id=min_row_id,
- max_row_id=max_row_id,
- )
+ return self.manifest_file_manager.write_with_meta(new_manifest_file,
commit_entries)
def _is_duplicate_commit(self, retry_result, latest_snapshot,
commit_identifier, commit_kind) -> bool:
if retry_result is not None and latest_snapshot is not None:
@@ -682,7 +630,8 @@ class FileStoreCommit:
delta_manifest_list: Optional[str],
base_manifest_list: Optional[str],
index_manifest: Optional[str] = None,
- changelog_manifest_list: Optional[str] =
None):
+ changelog_manifest_list: Optional[str] =
None,
+ base_manifest_files_to_delete:
Optional[List[ManifestFileMeta]] = None):
try:
manifest_path = self.manifest_list_manager.manifest_path
@@ -690,14 +639,22 @@ class FileStoreCommit:
self.table.file_io.delete_quietly(f"{manifest_path}/{index_manifest}")
if delta_manifest_list:
- manifest_files =
self.manifest_list_manager.read(delta_manifest_list)
- for manifest_meta in manifest_files:
- manifest_file_path =
f"{self.manifest_file_manager.manifest_path}/{manifest_meta.file_name}"
- self.table.file_io.delete_quietly(manifest_file_path)
+ try:
+ manifest_files =
self.manifest_list_manager.read(delta_manifest_list)
+ for manifest_meta in manifest_files:
+ manifest_file_path =
f"{self.manifest_file_manager.manifest_path}/{manifest_meta.file_name}"
+ self.table.file_io.delete_quietly(manifest_file_path)
+ except Exception:
+ pass
delta_path = f"{manifest_path}/{delta_manifest_list}"
self.table.file_io.delete_quietly(delta_path)
if base_manifest_list:
+ if base_manifest_files_to_delete:
+ for manifest_meta in base_manifest_files_to_delete:
+ manifest_file_path = (
+
f"{self.manifest_file_manager.manifest_path}/{manifest_meta.file_name}")
+ self.table.file_io.delete_quietly(manifest_file_path)
base_path = f"{manifest_path}/{base_manifest_list}"
self.table.file_io.delete_quietly(base_path)
