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 ab980922d0 [core][python] Add option to skip manifest merge in
write-only mode (#9727)
ab980922d0 is described below
commit ab980922d05b8f908d06b7f9ac40a84fdd6a406d
Author: Jingsong Lee <[email protected]>
AuthorDate: Fri Sep 11 11:27:10 2026 +0800
[core][python] Add option to skip manifest merge in write-only mode (#9727)
---
docs/docs/maintenance/dedicated-compaction.mdx | 2 +-
docs/docs/pypaimon/writing.md | 13 +++++
docs/generated/core_configuration.html | 8 ++-
.../main/java/org/apache/paimon/CoreOptions.java | 16 ++++-
.../paimon/operation/FileStoreCommitImpl.java | 16 +++--
.../paimon/operation/FileStoreCommitTest.java | 68 +++++++++++++++++++++-
.../pypaimon/common/options/core_options.py | 26 +++++++++
.../pypaimon/tests/file_store_commit_test.py | 32 ++++++++++
.../pypaimon/tests/write/table_write_test.py | 44 +++++++++++---
paimon-python/pypaimon/write/file_store_commit.py | 18 +++---
10 files changed, 217 insertions(+), 26 deletions(-)
diff --git a/docs/docs/maintenance/dedicated-compaction.mdx
b/docs/docs/maintenance/dedicated-compaction.mdx
index d1fee22e9a..7b190a7710 100644
--- a/docs/docs/maintenance/dedicated-compaction.mdx
+++ b/docs/docs/maintenance/dedicated-compaction.mdx
@@ -93,7 +93,7 @@ To skip compactions in writers, set the following table
property to `true`.
<td>No</td>
<td style={{wordWrap: "break-word"}}>false</td>
<td>Boolean</td>
- <td>If set to true, compactions and snapshot expiration will be skipped.
This option is used along with dedicated compact jobs.</td>
+ <td>If set to true, compactions and snapshot expiration will be skipped.
This option is used along with dedicated compact jobs. Set
manifest.merge.skip-on-write-only=true to also skip automatic manifest merging;
it defaults to false.</td>
</tr>
</tbody>
</table>
diff --git a/docs/docs/pypaimon/writing.md b/docs/docs/pypaimon/writing.md
index cccddf955d..d42f5c4b06 100644
--- a/docs/docs/pypaimon/writing.md
+++ b/docs/docs/pypaimon/writing.md
@@ -83,6 +83,19 @@ write_builder = table.new_batch_write_builder().overwrite()
write_builder = table.new_batch_write_builder().overwrite({'dt': '2024-01-01'})
```
+### Manifest Merging
+
+`manifest.merge.skip-on-write-only` defaults to `false` in both Python and
Java,
+so commits keep their automatic manifest merging behavior. Set both this option
+and `write-only` to `true` to retain existing manifest files during commit and
+avoid the cost of reading and rewriting them. This option has no effect when
+`write-only=false`, which is also the default.
+
+Python supports minor manifest compaction, using `manifest.merge-min-count` and
+`manifest.target-file-size`. Python does not support manifest sort rewrite.
+In Java, skipping automatic manifest merging also skips automatic manifest sort
+rewrite; explicit manifest compaction remains available.
+
### Commit Callback
You can register `CommitCallback` instances on a `TableCommit` to be notified
after each successful
diff --git a/docs/generated/core_configuration.html
b/docs/generated/core_configuration.html
index bd231c88c0..ac57300175 100644
--- a/docs/generated/core_configuration.html
+++ b/docs/generated/core_configuration.html
@@ -1107,6 +1107,12 @@ Mainly to resolve data skew on primary keys. We
recommend starting with 64 mb wh
<td>Boolean</td>
<td>Whether to enable block-aware ordinary manifest merging. When
disabled, ordinary manifest compaction uses the legacy full-entry merger.</td>
</tr>
+ <tr>
+ <td><h5>manifest.merge.skip-on-write-only</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Whether to skip automatic manifest merging during commit when
write-only is true. This also skips automatic manifest sort rewrite. Explicit
manifest compaction is not affected.</td>
+ </tr>
<tr>
<td><h5>manifest.target-file-size</h5></td>
<td style="word-wrap: break-word;">8 mb</td>
@@ -1991,7 +1997,7 @@ The size strategy ranges by the data size allocated to
each sorting task, which
<td><h5>write-only</h5></td>
<td style="word-wrap: break-word;">false</td>
<td>Boolean</td>
- <td>If set to true, compactions and snapshot expiration will be
skipped. This option is used along with dedicated compact jobs.</td>
+ <td>If set to true, compactions and snapshot expiration will be
skipped. This option is used along with dedicated compact jobs. Automatic
manifest merging is also skipped when manifest.merge.skip-on-write-only is
true.</td>
</tr>
<tr>
<td><h5>write.batch-memory</h5></td>
diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
index 41b338cb9e..7b0665c502 100644
--- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
@@ -529,6 +529,15 @@ public class CoreOptions implements Serializable {
.withDescription(
"The size threshold for triggering full compaction
of manifest.");
+ public static final ConfigOption<Boolean>
MANIFEST_MERGE_SKIP_ON_WRITE_ONLY =
+ key("manifest.merge.skip-on-write-only")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Whether to skip automatic manifest merging during
commit when write-only is true."
+ + " This also skips automatic manifest
sort rewrite."
+ + " Explicit manifest compaction is not
affected.");
+
public static final ConfigOption<Integer> MANIFEST_MERGE_MIN_COUNT =
key("manifest.merge-min-count")
.intType()
@@ -771,7 +780,8 @@ public class CoreOptions implements Serializable {
.withFallbackKeys("write.compaction-skip")
.withDescription(
"If set to true, compactions and snapshot
expiration will be skipped. "
- + "This option is used along with
dedicated compact jobs.");
+ + "This option is used along with
dedicated compact jobs. "
+ + "Automatic manifest merging is also
skipped when manifest.merge.skip-on-write-only is true.");
public static final ConfigOption<MemorySize> SOURCE_SPLIT_TARGET_SIZE =
key("source.split.target-size")
@@ -3529,6 +3539,10 @@ public class CoreOptions implements Serializable {
.build();
}
+ public boolean manifestMergeSkipOnWriteOnly() {
+ return options.get(MANIFEST_MERGE_SKIP_ON_WRITE_ONLY);
+ }
+
public int manifestMergeMinCount() {
return options.get(MANIFEST_MERGE_MIN_COUNT);
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
index 2d9c94ec72..2325d85ff7 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
@@ -1103,7 +1103,7 @@ public class FileStoreCommitImpl implements
FileStoreCommit {
String indexManifest = null;
List<ManifestFileMeta> mergeBeforeManifests = new ArrayList<>();
List<ManifestFileMeta> mergeAfterManifests = new ArrayList<>();
- boolean skipManifestMergeOnRetry = false;
+ boolean skipManifestMerge = false;
long nextRowIdStart = firstRowIdStart;
try {
long previousTotalRecordCount = 0L;
@@ -1128,13 +1128,19 @@ public class FileStoreCommitImpl implements
FileStoreCommit {
mergeAfterManifests = emptyList();
oldIndexManifest = null;
} else {
+ boolean skipManifestMergeForWriteOnly =
+ options.writeOnly() &&
options.manifestMergeSkipOnWriteOnly();
ManifestMergeReuse manifestMergeReuse =
- tryReuseManifestMergeResult(retryResult,
mergeBeforeManifests);
- skipManifestMergeOnRetry = manifestMergeReuse == null &&
retryResult != null;
+ skipManifestMergeForWriteOnly
+ ? null
+ : tryReuseManifestMergeResult(retryResult,
mergeBeforeManifests);
+ skipManifestMerge =
+ skipManifestMergeForWriteOnly
+ || (manifestMergeReuse == null && retryResult
!= null);
if (manifestMergeReuse != null) {
mergeBeforeManifests =
manifestMergeReuse.preservedManifests;
mergeAfterManifests =
manifestMergeReuse.mergeAfterManifests;
- } else if (skipManifestMergeOnRetry) {
+ } else if (skipManifestMerge) {
mergeAfterManifests = mergeBeforeManifests;
} else {
mergeAfterManifests =
@@ -1291,7 +1297,7 @@ public class FileStoreCommitImpl implements
FileStoreCommit {
latestSnapshot,
baseDataFiles,
null,
- skipManifestMergeOnRetry
+ skipManifestMerge
? null
: new ManifestMergeResult(mergeBeforeManifests,
mergeAfterManifests));
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java
b/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java
index 008cf3131f..c96b7b7111 100644
---
a/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java
@@ -1336,9 +1336,71 @@ public class FileStoreCommitTest {
.isEqualTo(store.toKvMap(Collections.singletonList(original)));
}
- @Test
- public void testManifestCompact() throws Exception {
- TestFileStore store = createStore(false);
+ @ParameterizedTest
+ @CsvSource({
+ "false,false,default",
+ "false,true,default",
+ "true,false,default",
+ "true,true,default",
+ "false,false,true",
+ "false,true,true",
+ "true,false,true",
+ "true,true,true",
+ "false,false,false",
+ "false,true,false",
+ "true,false,false",
+ "true,true,false"
+ })
+ public void testCommitManifestMerge(
+ boolean sortEnabled, boolean writeOnly, String skipOnWriteOnly)
throws Exception {
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.MANIFEST_MERGE_MIN_COUNT.key(), "2");
+ options.put(CoreOptions.MANIFEST_SORT_ENABLED.key(),
String.valueOf(sortEnabled));
+ options.put(CoreOptions.WRITE_ONLY.key(), String.valueOf(writeOnly));
+ if (!"default".equals(skipOnWriteOnly)) {
+ options.put(CoreOptions.MANIFEST_MERGE_SKIP_ON_WRITE_ONLY.key(),
skipOnWriteOnly);
+ }
+ TestFileStore store = createStore(false, options);
+ // Override the randomized manifest size in TestFileStore to keep both
files under budget.
+
store.options().toConfiguration().set(CoreOptions.MANIFEST_TARGET_FILE_SIZE.key(),
"8 mb");
+ List<KeyValue> expected = new ArrayList<>();
+ List<ManifestFileMeta> previousManifests = Collections.emptyList();
+ for (int i = 0; i < 3; i++) {
+ KeyValue kv = gen.nextInsert("20211110", 8, (long) i, null,
"value-" + i);
+ expected.add(kv);
+ Snapshot snapshot =
+ store.commitData(Collections.singletonList(kv),
gen::getPartition, value -> 0)
+ .get(0);
+ if (i == 1) {
+ previousManifests =
+
store.manifestListFactory().create().readDataManifests(snapshot);
+ }
+ }
+
+ Snapshot latest = store.snapshotManager().latestSnapshot();
+ List<ManifestFileMeta> baseManifests =
+ store.manifestListFactory()
+ .create()
+ .read(latest.baseManifestList(),
latest.baseManifestListSize());
+ if (writeOnly && "true".equals(skipOnWriteOnly)) {
+
assertThat(baseManifests).hasSize(2).containsExactlyElementsOf(previousManifests);
+ } else {
+ assertThat(baseManifests).hasSize(1);
+ assertThat(baseManifests.get(0).numAddedFiles()).isEqualTo(2);
+ }
+ assertThat(store.toKvMap(store.readKvsFromSnapshot(latest.id())))
+ .isEqualTo(store.toKvMap(expected));
+ }
+
+ @ParameterizedTest
+ @CsvSource({"true,false", "false,false", "true,true", "false,true"})
+ public void testManifestCompact(boolean skipOnWriteOnly, boolean
writeOnly) throws Exception {
+ Map<String, String> options = new HashMap<>();
+ options.put(
+ CoreOptions.MANIFEST_MERGE_SKIP_ON_WRITE_ONLY.key(),
+ String.valueOf(skipOnWriteOnly));
+ options.put(CoreOptions.WRITE_ONLY.key(), String.valueOf(writeOnly));
+ TestFileStore store = createStore(false, options);
List<KeyValue> keyValues = generateDataList(1);
BinaryRow partition = gen.getPartition(keyValues.get(0));
diff --git a/paimon-python/pypaimon/common/options/core_options.py
b/paimon-python/pypaimon/common/options/core_options.py
index 0e3ddb9bab..903697a94f 100644
--- a/paimon-python/pypaimon/common/options/core_options.py
+++ b/paimon-python/pypaimon/common/options/core_options.py
@@ -280,6 +280,16 @@ class CoreOptions:
)
)
+ WRITE_ONLY: ConfigOption[bool] = (
+ ConfigOptions.key("write-only")
+ .boolean_type()
+ .default_value(False)
+ .with_description(
+ "Whether to use write-only mode. Automatic manifest merging is
skipped "
+ "when both this option and manifest.merge.skip-on-write-only are
true."
+ )
+ )
+
SCAN_MANIFEST_PARALLELISM: ConfigOption[int] = (
ConfigOptions.key("scan.manifest.parallelism")
.int_type()
@@ -301,6 +311,16 @@ class CoreOptions:
.with_description("Suggested file size of a manifest file.")
)
+ MANIFEST_MERGE_SKIP_ON_WRITE_ONLY: ConfigOption[bool] = (
+ ConfigOptions.key("manifest.merge.skip-on-write-only")
+ .boolean_type()
+ .default_value(False)
+ .with_description(
+ "Whether to skip automatic manifest merging during commit when
write-only is true. "
+ "Python only supports minor manifest compaction, without manifest
sort rewrite."
+ )
+ )
+
MANIFEST_MERGE_MIN_COUNT: ConfigOption[int] = (
ConfigOptions.key("manifest.merge-min-count")
.int_type()
@@ -1188,6 +1208,9 @@ class CoreOptions:
CoreOptions.POSTPONE_TARGET_SIZE_PER_BUCKET, default
).get_bytes()
+ def write_only(self, default=None):
+ return self.options.get(CoreOptions.WRITE_ONLY, default)
+
def scan_manifest_parallelism(self, default=None):
return self.options.get(CoreOptions.SCAN_MANIFEST_PARALLELISM, default)
@@ -1199,6 +1222,9 @@ class CoreOptions:
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_skip_on_write_only(self, default=None):
+ return self.options.get(CoreOptions.MANIFEST_MERGE_SKIP_ON_WRITE_ONLY,
default)
+
def manifest_merge_min_count(self, default=None):
return self.options.get(CoreOptions.MANIFEST_MERGE_MIN_COUNT, default)
diff --git a/paimon-python/pypaimon/tests/file_store_commit_test.py
b/paimon-python/pypaimon/tests/file_store_commit_test.py
index df1992cc71..4a2f9d2450 100644
--- a/paimon-python/pypaimon/tests/file_store_commit_test.py
+++ b/paimon-python/pypaimon/tests/file_store_commit_test.py
@@ -69,6 +69,8 @@ class TestFileStoreCommitRowTracking(unittest.TestCase):
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
+ self.mock_table.options.write_only.return_value = False
+ self.mock_table.options.manifest_merge_skip_on_write_only.return_value
= False
self.mock_snapshot_commit = Mock()
def _create_file_store_commit(self):
@@ -272,6 +274,8 @@ class TestFileStoreCommit(unittest.TestCase):
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
+ self.mock_table.options.write_only.return_value = False
+ self.mock_table.options.manifest_merge_skip_on_write_only.return_value
= False
# Mock snapshot commit
self.mock_snapshot_commit = Mock()
@@ -515,6 +519,34 @@ class TestFileStoreCommit(unittest.TestCase):
)
file_store_commit.manifest_file_merger.merge.assert_called_once()
+ def test_disabled_manifest_merge_preserves_manifests_on_retry(
+ self, mock_manifest_list_manager, mock_manifest_file_manager):
+ options = CoreOptions(Options({
+ 'write-only': 'true',
+ 'manifest.merge.skip-on-write-only': 'true',
+ }))
+ self.mock_table.options.write_only.side_effect = options.write_only
+ self.mock_table.options.manifest_merge_skip_on_write_only.side_effect
= (
+ options.manifest_merge_skip_on_write_only)
+ current = [self._manifest_meta('before-a'),
self._manifest_meta('before-b')]
+ first_commit, retry_result = self._run_manifest_commit_attempt(
+ commit_result=False, existing_manifests=current)
+
+ self.assertIsInstance(retry_result, CommitFailRetryResult)
+ self.assertIsNone(retry_result.manifest_merge_result)
+ first_commit.manifest_file_merger.merge.assert_not_called()
+ self.assertEqual(
+ current,
first_commit.manifest_list_manager.write.call_args_list[-1].args[1])
+
+ current.append(self._manifest_meta('concurrent'))
+ retry_commit, result = self._run_manifest_commit_attempt(
+ commit_result=True, retry_result=retry_result,
existing_manifests=current)
+
+ self.assertTrue(result.is_success())
+ retry_commit.manifest_file_merger.merge.assert_not_called()
+ self.assertEqual(
+ current,
retry_commit.manifest_list_manager.write.call_args_list[-1].args[1])
+
def test_atomic_commit_exception_does_not_retain_manifest_merge_result(
self, mock_manifest_list_manager, mock_manifest_file_manager):
failure = TimeoutError('lost commit response')
diff --git a/paimon-python/pypaimon/tests/write/table_write_test.py
b/paimon-python/pypaimon/tests/write/table_write_test.py
index aa46603d7d..04e555a73c 100644
--- a/paimon-python/pypaimon/tests/write/table_write_test.py
+++ b/paimon-python/pypaimon/tests/write/table_write_test.py
@@ -575,14 +575,33 @@ 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):
+ @parameterized.expand([
+ ('default', None, None, True),
+ ('not_write_only_default', 'false', None, True),
+ ('write_only_default', 'true', None, True),
+ ('skip_enabled', None, 'true', True),
+ ('not_write_only_skip_enabled', 'false', 'true', True),
+ ('write_only_skip_enabled', 'true', 'true', False),
+ ('skip_disabled', None, 'false', True),
+ ('not_write_only_skip_disabled', 'false', 'false', True),
+ ('write_only_skip_disabled', 'true', 'false', True),
+ ])
+ def test_commit_manifest_merge(self, name, write_only, skip_on_write_only,
merge_enabled):
+ options = {'manifest.merge-min-count': '2'}
+ if write_only is not None:
+ options['write-only'] = write_only
+ if skip_on_write_only is not None:
+ options['manifest.merge.skip-on-write-only'] = skip_on_write_only
schema = Schema.from_pyarrow_schema(
self.pa_schema,
partition_keys=['dt'],
- options={'manifest.merge-min-count': '2'},
+ options=options,
)
- self.catalog.create_table('default.test_minor_manifest_compaction',
schema, False)
- table =
self.catalog.get_table('default.test_minor_manifest_compaction')
+ identifier = 'default.test_manifest_merge_' + name
+ self.catalog.create_table(identifier, schema, False)
+ table = self.catalog.get_table(identifier)
+ manifest_list_manager = ManifestListManager(table)
+ previous_manifests = []
expected_data = {
'user_id': [],
@@ -607,15 +626,24 @@ class TableWriteTest(unittest.TestCase):
table_commit.commit(table_write.prepare_commit())
table_write.close()
table_commit.close()
+ if i == 1:
+ previous_manifests = manifest_list_manager.read_all(
+ table.snapshot_manager().get_latest_snapshot())
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)
+ if merge_enabled:
+ self.assertEqual(len(base_manifests), 1)
+ self.assertEqual(base_manifests[0].num_added_files, 2)
+ self.assertEqual(base_manifests[0].num_deleted_files, 0)
+ else:
+ self.assertEqual(len(base_manifests), 2)
+ self.assertEqual(
+ [manifest.file_name for manifest in previous_manifests],
+ [manifest.file_name for manifest in base_manifests],
+ )
self.assertEqual(len(delta_manifests), 1)
expected = pa.Table.from_pydict(expected_data, schema=self.pa_schema)
diff --git a/paimon-python/pypaimon/write/file_store_commit.py
b/paimon-python/pypaimon/write/file_store_commit.py
index 917f4286cd..66e88f2f5a 100644
--- a/paimon-python/pypaimon/write/file_store_commit.py
+++ b/paimon-python/pypaimon/write/file_store_commit.py
@@ -222,6 +222,8 @@ class FileStoreCommit:
self.manifest_list_manager = ManifestListManager(table)
self.manifest_target_size = table.options.manifest_target_size()
+ self.skip_manifest_merge_on_write_only = (
+ table.options.write_only() and
table.options.manifest_merge_skip_on_write_only())
self.manifest_merge_min_count =
table.options.manifest_merge_min_count()
self.manifest_file_merger = ManifestFileMerger(
self.manifest_file_manager,
@@ -734,7 +736,7 @@ class FileStoreCommit:
merge_before_manifests = []
merge_after_manifests = []
merge_new_files = []
- skip_manifest_merge_on_retry = False
+ skip_manifest_merge = False
try:
new_manifest_file_metas =
self._write_manifest_files(commit_entries, new_manifest_file)
self.manifest_list_manager.write(delta_manifest_list,
new_manifest_file_metas)
@@ -763,10 +765,12 @@ class FileStoreCommit:
if previous_record_count:
total_record_count += previous_record_count
- reused_manifests = _try_reuse_manifest_merge_result(
- retry_result, merge_before_manifests)
- skip_manifest_merge_on_retry = (
- reused_manifests is None and retry_result is not None)
+ reused_manifests = (
+ _try_reuse_manifest_merge_result(retry_result,
merge_before_manifests)
+ if not self.skip_manifest_merge_on_write_only else None)
+ skip_manifest_merge = (
+ self.skip_manifest_merge_on_write_only
+ or (reused_manifests is None and retry_result is not None))
if reused_manifests is not None:
merge_after_manifests = reused_manifests
old_names = {
@@ -776,7 +780,7 @@ class FileStoreCommit:
manifest for manifest in merge_after_manifests
if manifest.file_name not in old_names
]
- elif skip_manifest_merge_on_retry:
+ elif skip_manifest_merge:
merge_after_manifests = merge_before_manifests
else:
merge_after_manifests, merge_new_files = (
@@ -861,7 +865,7 @@ class FileStoreCommit:
)
manifest_merge_result = (
None
- if skip_manifest_merge_on_retry
+ if skip_manifest_merge
else ManifestMergeResult(
merge_before_manifests,
merge_after_manifests,
