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 cca01af0fd [core] Reuse manifest merge results on commit retry (#8229)
cca01af0fd is described below
commit cca01af0fdda277171212a27c16ba2311850428d
Author: YeJunHao <[email protected]>
AuthorDate: Sat Jun 20 13:11:20 2026 +0800
[core] Reuse manifest merge results on commit retry (#8229)
---
.../java/org/apache/paimon/utils/ListUtils.java | 34 ++
.../org/apache/paimon/utils/ListUtilsTest.java | 87 ++++
.../paimon/operation/FileStoreCommitImpl.java | 72 ++-
.../paimon/operation/commit/RetryCommitResult.java | 31 +-
.../paimon/operation/FileStoreCommitTest.java | 564 ++++++++++++++++++++-
5 files changed, 779 insertions(+), 9 deletions(-)
diff --git a/paimon-common/src/main/java/org/apache/paimon/utils/ListUtils.java
b/paimon-common/src/main/java/org/apache/paimon/utils/ListUtils.java
index bd35c3798c..dc9c9233cf 100644
--- a/paimon-common/src/main/java/org/apache/paimon/utils/ListUtils.java
+++ b/paimon-common/src/main/java/org/apache/paimon/utils/ListUtils.java
@@ -18,10 +18,13 @@
package org.apache.paimon.utils;
+import javax.annotation.Nullable;
+
import java.util.ArrayList;
import java.util.Collection;
import java.util.Iterator;
import java.util.List;
+import java.util.Objects;
import java.util.concurrent.ThreadLocalRandom;
import static org.apache.paimon.utils.Preconditions.checkArgument;
@@ -53,4 +56,35 @@ public class ListUtils {
result.addAll(list2);
return result;
}
+
+ /**
+ * Replace the first continuous occurrence of {@code replaced} in {@code
list} with {@code
+ * replacement}. Return null if {@code replaced} does not appear as a
continuous sub-list.
+ */
+ @Nullable
+ public static <E> List<E> tryReplace(
+ List<? extends E> list, List<?> replaced, List<? extends E>
replacement) {
+ checkArgument(!replaced.isEmpty(), "Cannot replace an empty list.");
+
+ for (int start = 0; start <= list.size() - replaced.size(); start++) {
+ boolean found = true;
+ for (int i = 0; i < replaced.size(); i++) {
+ if (!Objects.equals(list.get(start + i), replaced.get(i))) {
+ found = false;
+ break;
+ }
+ }
+
+ if (found) {
+ ArrayList<E> result =
+ new ArrayList<>(list.size() - replaced.size() +
replacement.size());
+ result.addAll(list.subList(0, start));
+ result.addAll(replacement);
+ result.addAll(list.subList(start + replaced.size(),
list.size()));
+ return result;
+ }
+ }
+
+ return null;
+ }
}
diff --git
a/paimon-common/src/test/java/org/apache/paimon/utils/ListUtilsTest.java
b/paimon-common/src/test/java/org/apache/paimon/utils/ListUtilsTest.java
new file mode 100644
index 0000000000..44068d254d
--- /dev/null
+++ b/paimon-common/src/test/java/org/apache/paimon/utils/ListUtilsTest.java
@@ -0,0 +1,87 @@
+/*
+ * 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.
+ */
+
+package org.apache.paimon.utils;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Test for {@link ListUtils}. */
+public class ListUtilsTest {
+
+ @Test
+ public void testTryReplace() {
+ assertThat(
+ ListUtils.tryReplace(
+ Arrays.asList("a", "b", "c", "d", "e", "f"),
+ Arrays.asList("b", "c", "d"),
+ Arrays.asList("n", "w")))
+ .containsExactly("a", "n", "w", "e", "f");
+
+ assertThat(
+ ListUtils.tryReplace(
+ Arrays.asList("b", "c", "d"),
+ Arrays.asList("b", "c", "d"),
+ Collections.singletonList("n")))
+ .containsExactly("n");
+
+ assertThat(
+ ListUtils.tryReplace(
+ Arrays.asList("a", "b", "c"),
+ Collections.singletonList("b"),
+ Collections.emptyList()))
+ .containsExactly("a", "c");
+ }
+
+ @Test
+ public void testTryReplaceRequiresContinuousMatch() {
+ assertThat(
+ ListUtils.tryReplace(
+ Arrays.asList("a", "b", "x", "c", "d"),
+ Arrays.asList("b", "c", "d"),
+ Collections.singletonList("n")))
+ .isNull();
+ }
+
+ @Test
+ public void testTryReplaceRequiresSameOrder() {
+ assertThat(
+ ListUtils.tryReplace(
+ Arrays.asList("a", "d", "c", "b", "e"),
+ Arrays.asList("b", "c", "d"),
+ Collections.singletonList("n")))
+ .isNull();
+ }
+
+ @Test
+ public void testTryReplaceRequiresNonEmptyReplacedList() {
+ assertThatThrownBy(
+ () ->
+ ListUtils.tryReplace(
+ Collections.singletonList("a"),
+ Collections.emptyList(),
+ Collections.singletonList("n")))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Cannot replace an empty list");
+ }
+}
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 a5cc109614..47442f1b78 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
@@ -49,6 +49,7 @@ import org.apache.paimon.operation.commit.ConflictDetection;
import org.apache.paimon.operation.commit.ManifestEntryChanges;
import org.apache.paimon.operation.commit.RetryCommitResult;
import
org.apache.paimon.operation.commit.RetryCommitResult.CommitFailRetryResult;
+import
org.apache.paimon.operation.commit.RetryCommitResult.ManifestMergeResult;
import org.apache.paimon.operation.commit.RowIdColumnConflictChecker;
import
org.apache.paimon.operation.commit.RowTrackingCommitUtils.RowTrackingAssigned;
import org.apache.paimon.operation.commit.StrictModeChecker;
@@ -74,6 +75,7 @@ import org.apache.paimon.utils.DataFilePathFactories;
import org.apache.paimon.utils.FileStorePathFactory;
import org.apache.paimon.utils.IOUtils;
import org.apache.paimon.utils.InternalRowPartitionComputer;
+import org.apache.paimon.utils.ListUtils;
import org.apache.paimon.utils.Pair;
import org.apache.paimon.utils.RetryWaiter;
import org.apache.paimon.utils.SnapshotManager;
@@ -979,6 +981,7 @@ public class FileStoreCommitImpl implements FileStoreCommit
{
String indexManifest = null;
List<ManifestFileMeta> mergeBeforeManifests = new ArrayList<>();
List<ManifestFileMeta> mergeAfterManifests = new ArrayList<>();
+ boolean skipManifestMergeOnRetry = false;
long nextRowIdStart = firstRowIdStart;
try {
long previousTotalRecordCount = 0L;
@@ -998,9 +1001,19 @@ public class FileStoreCommitImpl implements
FileStoreCommit {
}
// try to merge old manifest files to create base manifest list
- mergeAfterManifests =
- ManifestFileMerger.merge(
- mergeBeforeManifests, manifestFile, partitionType,
options);
+ ManifestMergeReuse manifestMergeReuse =
+ tryReuseManifestMergeResult(retryResult,
mergeBeforeManifests);
+ skipManifestMergeOnRetry = manifestMergeReuse == null &&
retryResult != null;
+ if (manifestMergeReuse != null) {
+ mergeBeforeManifests = manifestMergeReuse.preservedManifests;
+ mergeAfterManifests = manifestMergeReuse.mergeAfterManifests;
+ } else if (skipManifestMergeOnRetry) {
+ mergeAfterManifests = mergeBeforeManifests;
+ } else {
+ mergeAfterManifests =
+ ManifestFileMerger.merge(
+ mergeBeforeManifests, manifestFile,
partitionType, options);
+ }
baseManifestList = manifestList.write(mergeAfterManifests);
if (options.rowTrackingEnabled()) {
@@ -1120,7 +1133,7 @@ public class FileStoreCommitImpl implements
FileStoreCommit {
} catch (Exception e) {
// commit exception, not sure about the situation and should not
clean up the files
LOG.warn("Retry commit for exception.", e);
- return RetryCommitResult.forCommitFail(latestSnapshot,
baseDataFiles, e);
+ return RetryCommitResult.forCommitFail(latestSnapshot,
baseDataFiles, e, null);
}
if (!success) {
@@ -1134,7 +1147,13 @@ public class FileStoreCommitImpl implements
FileStoreCommit {
identifier,
commitKind.name(),
commitTime);
- return RetryCommitResult.forCommitFail(latestSnapshot,
baseDataFiles, null);
+ return RetryCommitResult.forCommitFail(
+ latestSnapshot,
+ baseDataFiles,
+ null,
+ skipManifestMergeOnRetry
+ ? null
+ : new ManifestMergeResult(mergeBeforeManifests,
mergeAfterManifests));
}
LOG.info(
@@ -1156,6 +1175,49 @@ public class FileStoreCommitImpl implements
FileStoreCommit {
return new SuccessCommitResult();
}
+ @Nullable
+ private ManifestMergeReuse tryReuseManifestMergeResult(
+ @Nullable RetryCommitResult retryResult, List<ManifestFileMeta>
currentManifests) {
+ if (!(retryResult instanceof CommitFailRetryResult)) {
+ return null;
+ }
+
+ CommitFailRetryResult commitFailRetry = (CommitFailRetryResult)
retryResult;
+ ManifestMergeResult previous = commitFailRetry.manifestMergeResult;
+ if (previous == null) {
+ return null;
+ }
+ if (previous.mergeBeforeManifests.isEmpty()) {
+ return currentManifests.isEmpty()
+ ? new ManifestMergeReuse(currentManifests,
previous.mergeAfterManifests)
+ : null;
+ }
+
+ List<ManifestFileMeta> mergeAfterManifests =
+ ListUtils.tryReplace(
+ currentManifests,
+ previous.mergeBeforeManifests,
+ previous.mergeAfterManifests);
+ if (mergeAfterManifests == null) {
+ return null;
+ }
+
+ return new ManifestMergeReuse(currentManifests, mergeAfterManifests);
+ }
+
+ private static class ManifestMergeReuse {
+
+ private final List<ManifestFileMeta> preservedManifests;
+ private final List<ManifestFileMeta> mergeAfterManifests;
+
+ private ManifestMergeReuse(
+ List<ManifestFileMeta> preservedManifests,
+ List<ManifestFileMeta> mergeAfterManifests) {
+ this.preservedManifests = preservedManifests;
+ this.mergeAfterManifests = mergeAfterManifests;
+ }
+ }
+
public boolean replaceManifestList(
Snapshot latest,
long totalRecordCount,
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/commit/RetryCommitResult.java
b/paimon-core/src/main/java/org/apache/paimon/operation/commit/RetryCommitResult.java
index b9e0ab2a2e..717df209ce 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/commit/RetryCommitResult.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/commit/RetryCommitResult.java
@@ -19,10 +19,13 @@
package org.apache.paimon.operation.commit;
import org.apache.paimon.Snapshot;
+import org.apache.paimon.manifest.ManifestFileMeta;
import org.apache.paimon.manifest.SimpleFileEntry;
import javax.annotation.Nullable;
+import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
/** Need to retry commit of {@link CommitResult}. */
@@ -35,8 +38,11 @@ public abstract class RetryCommitResult implements
CommitResult {
}
public static RetryCommitResult forCommitFail(
- Snapshot snapshot, List<SimpleFileEntry> baseDataFiles, Exception
exception) {
- return new CommitFailRetryResult(snapshot, baseDataFiles, exception);
+ Snapshot snapshot,
+ List<SimpleFileEntry> baseDataFiles,
+ Exception exception,
+ @Nullable ManifestMergeResult manifestMergeResult) {
+ return new CommitFailRetryResult(snapshot, baseDataFiles, exception,
manifestMergeResult);
}
public static RetryCommitResult forRollback(Exception exception) {
@@ -53,14 +59,33 @@ public abstract class RetryCommitResult implements
CommitResult {
public final @Nullable Snapshot latestSnapshot;
public final @Nullable List<SimpleFileEntry> baseDataFiles;
+ public final @Nullable ManifestMergeResult manifestMergeResult;
private CommitFailRetryResult(
@Nullable Snapshot latestSnapshot,
@Nullable List<SimpleFileEntry> baseDataFiles,
- Exception exception) {
+ Exception exception,
+ @Nullable ManifestMergeResult manifestMergeResult) {
super(exception);
this.latestSnapshot = latestSnapshot;
this.baseDataFiles = baseDataFiles;
+ this.manifestMergeResult = manifestMergeResult;
+ }
+ }
+
+ /** Manifest merge result which can be reused by commit retry. */
+ public static class ManifestMergeResult {
+
+ public final List<ManifestFileMeta> mergeBeforeManifests;
+ public final List<ManifestFileMeta> mergeAfterManifests;
+
+ public ManifestMergeResult(
+ List<ManifestFileMeta> mergeBeforeManifests,
+ List<ManifestFileMeta> mergeAfterManifests) {
+ this.mergeBeforeManifests =
+ Collections.unmodifiableList(new
ArrayList<>(mergeBeforeManifests));
+ this.mergeAfterManifests =
+ Collections.unmodifiableList(new
ArrayList<>(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 bed2a3863c..5a386947c1 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
@@ -43,8 +43,10 @@ import org.apache.paimon.manifest.ManifestCommittable;
import org.apache.paimon.manifest.ManifestEntry;
import org.apache.paimon.manifest.ManifestFile;
import org.apache.paimon.manifest.ManifestFileMeta;
+import org.apache.paimon.manifest.ManifestList;
import org.apache.paimon.mergetree.compact.DeduplicateMergeFunction;
import org.apache.paimon.operation.commit.ConflictDetection;
+import org.apache.paimon.operation.commit.ManifestEntryChanges;
import org.apache.paimon.operation.commit.RetryCommitResult;
import org.apache.paimon.predicate.PredicateBuilder;
import org.apache.paimon.schema.Schema;
@@ -61,6 +63,7 @@ import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowKind;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.FailingFileIO;
+import org.apache.paimon.utils.Pair;
import org.apache.paimon.utils.SnapshotManager;
import org.apache.paimon.utils.TraceableFileIO;
@@ -1132,7 +1135,8 @@ public class FileStoreCommitTest {
null);
// Compact
commit.tryCommitOnce(
- RetryCommitResult.forCommitFail(firstLatest,
Collections.emptyList(), null),
+ RetryCommitResult.forCommitFail(
+ firstLatest, Collections.emptyList(), null, null),
Collections.emptyList(),
Collections.emptyList(),
Collections.emptyList(),
@@ -1183,6 +1187,393 @@ public class FileStoreCommitTest {
assertThat(store.readKvsFromSnapshot(latestSnapshot.id())).hasSize(1);
}
+ @Test
+ public void
testCommitRetryReusePreviousManifestMergeResultWhenBeforeStillExists()
+ throws Exception {
+ TestFileStore store =
+ createStore(
+ false,
+ Collections.singletonMap(
+
CoreOptions.MANIFEST_FULL_COMPACTION_FILE_SIZE.key(), "1"));
+
+ store.commitData(
+ Collections.singletonList(gen.nextInsert("20211110", 8, 1L,
null, "a")),
+ gen::getPartition,
+ kv -> 0);
+ store.commitData(
+ Collections.singletonList(gen.nextInsert("20211111", 9, 2L,
null, "b")),
+ gen::getPartition,
+ kv -> 0);
+
+ AtomicReference<ManifestCommittable> committableRef = new
AtomicReference<>();
+ store.commitDataImpl(
+ Collections.singletonList(gen.nextInsert("20211110", 9, 3L,
null, "c")),
+ gen::getPartition,
+ value -> 0,
+ false,
+ 23L,
+ null,
+ Collections.emptyList(),
+ (commit, committable) -> committableRef.set(committable));
+
+ ConflictingSnapshotCommit snapshotCommit =
+ new ConflictingSnapshotCommit(
+ new RenamingSnapshotCommit(store.snapshotManager(),
Lock.empty()),
+ store.snapshotManager(),
+ store.manifestListFactory().create(),
+ store.manifestFileFactory().create(),
+ false,
+ conflictAttempts(Collections.emptyList()));
+ try (FileStoreCommitImpl commit =
+ newCommitWithSnapshotCommit(store, "retry-reuse-merge",
snapshotCommit)) {
+ commit.commit(checkNotNull(committableRef.get()), false);
+ }
+
+ Snapshot latestSnapshot =
checkNotNull(store.snapshotManager().latestSnapshot());
+ List<ManifestFileMeta> finalBaseManifests =
+ store.manifestListFactory()
+ .create()
+ .read(
+ latestSnapshot.baseManifestList(),
+ latestSnapshot.baseManifestListSize());
+ assertThat(finalBaseManifests)
+
.containsExactlyElementsOf(snapshotCommit.firstAttemptBaseManifests());
+ assertThat(store.readKvsFromSnapshot(latestSnapshot.id())).hasSize(3);
+ }
+
+ @Test
+ public void
testCommitRetrySkipsManifestMergeWhenBeforeExistsNonContiguously()
+ throws Exception {
+ TestFileStore store =
+ createStore(
+ false,
+ Collections.singletonMap(
+
CoreOptions.MANIFEST_FULL_COMPACTION_FILE_SIZE.key(), "1"));
+
+ store.commitData(
+ Collections.singletonList(gen.nextInsert("20211110", 8, 1L,
null, "a")),
+ gen::getPartition,
+ kv -> 0);
+ store.commitData(
+ Collections.singletonList(gen.nextInsert("20211111", 9, 2L,
null, "b")),
+ gen::getPartition,
+ kv -> 0);
+
+ AtomicReference<ManifestCommittable> conflictCommittableRef = new
AtomicReference<>();
+ store.commitDataImpl(
+ Collections.singletonList(gen.nextInsert("20211112", 10, 4L,
null, "d")),
+ gen::getPartition,
+ value -> 0,
+ false,
+ 22L,
+ null,
+ Collections.emptyList(),
+ (commit, committable) ->
conflictCommittableRef.set(committable));
+ List<ManifestEntry> conflictDeltaFiles =
+ tableFilesFrom(checkNotNull(conflictCommittableRef.get()),
store.options());
+
+ AtomicReference<ManifestCommittable> committableRef = new
AtomicReference<>();
+ store.commitDataImpl(
+ Collections.singletonList(gen.nextInsert("20211110", 9, 3L,
null, "c")),
+ gen::getPartition,
+ value -> 0,
+ false,
+ 23L,
+ null,
+ Collections.emptyList(),
+ (commit, committable) -> committableRef.set(committable));
+
+ ConflictingSnapshotCommit snapshotCommit =
+ new ConflictingSnapshotCommit(
+ new RenamingSnapshotCommit(store.snapshotManager(),
Lock.empty()),
+ store.snapshotManager(),
+ store.manifestListFactory().create(),
+ store.manifestFileFactory().create(),
+ false,
+ conflictAttempts(conflictDeltaFiles));
+ try (FileStoreCommitImpl commit =
+ newCommitWithSnapshotCommit(store,
"retry-skip-non-contiguous", snapshotCommit)) {
+ commit.commit(checkNotNull(committableRef.get()), false);
+ }
+
+ Snapshot latestSnapshot =
checkNotNull(store.snapshotManager().latestSnapshot());
+ List<ManifestFileMeta> finalBaseManifests =
+ store.manifestListFactory()
+ .create()
+ .read(
+ latestSnapshot.baseManifestList(),
+ latestSnapshot.baseManifestListSize());
+ assertThat(finalBaseManifests)
+
.containsExactlyElementsOf(snapshotCommit.conflictBaseManifests());
+ assertThat(store.readKvsFromSnapshot(latestSnapshot.id())).hasSize(4);
+ }
+
+ @Test
+ public void
testCommitRetrySkipsManifestMergeWhenPreviousMergeCannotBeReused()
+ throws Exception {
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.MANIFEST_FULL_COMPACTION_FILE_SIZE.key(), "1");
+ TestFileStore store = createStore(false, options);
+
+ store.commitData(
+ Collections.singletonList(gen.nextInsert("20211110", 8, 1L,
null, "a")),
+ gen::getPartition,
+ kv -> 0);
+ store.commitData(
+ Collections.singletonList(gen.nextInsert("20211111", 9, 2L,
null, "b")),
+ gen::getPartition,
+ kv -> 0);
+
+ Snapshot latestBeforeRetry =
checkNotNull(store.snapshotManager().latestSnapshot());
+
assertThat(store.manifestListFactory().create().readDataManifests(latestBeforeRetry))
+ .hasSize(2);
+
+ AtomicReference<ManifestCommittable> committableRef = new
AtomicReference<>();
+ store.commitDataImpl(
+ Collections.singletonList(gen.nextInsert("20211110", 9, 3L,
null, "c")),
+ gen::getPartition,
+ value -> 0,
+ false,
+ 23L,
+ null,
+ Collections.emptyList(),
+ (commit, committable) -> committableRef.set(committable));
+
+ ConflictingSnapshotCommit snapshotCommit =
+ new ConflictingSnapshotCommit(
+ new RenamingSnapshotCommit(store.snapshotManager(),
Lock.empty()),
+ store.snapshotManager(),
+ store.manifestListFactory().create(),
+ store.manifestFileFactory().create(),
+ true,
+ conflictAttempts(Collections.emptyList()));
+ try (FileStoreCommitImpl commit =
+ newCommitWithSnapshotCommit(store, "retry-reuse-merge",
snapshotCommit)) {
+ commit.commit(checkNotNull(committableRef.get()), false);
+ }
+
+ Snapshot latestSnapshot =
checkNotNull(store.snapshotManager().latestSnapshot());
+ List<ManifestFileMeta> finalBaseManifests =
+ store.manifestListFactory()
+ .create()
+ .read(
+ latestSnapshot.baseManifestList(),
+ latestSnapshot.baseManifestListSize());
+ assertThat(snapshotCommit.conflictBaseManifests()).hasSize(2);
+ assertThat(finalBaseManifests)
+
.containsExactlyElementsOf(snapshotCommit.conflictBaseManifests());
+
assertThat(finalBaseManifests).isNotEqualTo(snapshotCommit.firstAttemptBaseManifests());
+ assertThat(store.readKvsFromSnapshot(latestSnapshot.id())).hasSize(3);
+ }
+
+ @Test
+ public void testCommitRetrySkipsManifestMergeAcrossMultipleRetries()
throws Exception {
+ TestFileStore store =
+ createStore(
+ false,
+ Collections.singletonMap(
+
CoreOptions.MANIFEST_FULL_COMPACTION_FILE_SIZE.key(), "1"));
+
+ store.commitData(
+ Collections.singletonList(gen.nextInsert("20211110", 8, 1L,
null, "a")),
+ gen::getPartition,
+ kv -> 0);
+ store.commitData(
+ Collections.singletonList(gen.nextInsert("20211111", 9, 2L,
null, "b")),
+ gen::getPartition,
+ kv -> 0);
+
+ AtomicReference<ManifestCommittable> firstConflictCommittableRef = new
AtomicReference<>();
+ store.commitDataImpl(
+ Collections.singletonList(gen.nextInsert("20211112", 10, 4L,
null, "d")),
+ gen::getPartition,
+ value -> 0,
+ false,
+ 22L,
+ null,
+ Collections.emptyList(),
+ (commit, committable) ->
firstConflictCommittableRef.set(committable));
+ List<ManifestEntry> firstConflictDeltaFiles =
+
tableFilesFrom(checkNotNull(firstConflictCommittableRef.get()),
store.options());
+
+ AtomicReference<ManifestCommittable> secondConflictCommittableRef =
new AtomicReference<>();
+ store.commitDataImpl(
+ Collections.singletonList(gen.nextInsert("20211113", 11, 5L,
null, "e")),
+ gen::getPartition,
+ value -> 0,
+ false,
+ 24L,
+ null,
+ Collections.emptyList(),
+ (commit, committable) ->
secondConflictCommittableRef.set(committable));
+ List<ManifestEntry> secondConflictDeltaFiles =
+
tableFilesFrom(checkNotNull(secondConflictCommittableRef.get()),
store.options());
+
+ AtomicReference<ManifestCommittable> committableRef = new
AtomicReference<>();
+ store.commitDataImpl(
+ Collections.singletonList(gen.nextInsert("20211110", 9, 3L,
null, "c")),
+ gen::getPartition,
+ value -> 0,
+ false,
+ 23L,
+ null,
+ Collections.emptyList(),
+ (commit, committable) -> committableRef.set(committable));
+
+ ConflictingSnapshotCommit snapshotCommit =
+ new ConflictingSnapshotCommit(
+ new RenamingSnapshotCommit(store.snapshotManager(),
Lock.empty()),
+ store.snapshotManager(),
+ store.manifestListFactory().create(),
+ store.manifestFileFactory().create(),
+ false,
+ conflictAttempts(firstConflictDeltaFiles,
secondConflictDeltaFiles));
+ try (FileStoreCommitImpl commit =
+ newCommitWithSnapshotCommit(store, "retry-reuse-multiple",
snapshotCommit)) {
+ commit.commit(checkNotNull(committableRef.get()), false);
+ }
+
+ Snapshot latestSnapshot =
checkNotNull(store.snapshotManager().latestSnapshot());
+ List<ManifestFileMeta> finalBaseManifests =
+ store.manifestListFactory()
+ .create()
+ .read(
+ latestSnapshot.baseManifestList(),
+ latestSnapshot.baseManifestListSize());
+ assertThat(finalBaseManifests)
+
.containsExactlyElementsOf(snapshotCommit.conflictBaseManifests());
+ assertThat(store.readKvsFromSnapshot(latestSnapshot.id())).hasSize(5);
+ }
+
+ @Test
+ public void testCommitRetryFromEmptyTableWithConcurrentFirstSnapshot()
throws Exception {
+ TestFileStore store =
+ createStore(
+ false,
+ Collections.singletonMap(
+
CoreOptions.MANIFEST_FULL_COMPACTION_FILE_SIZE.key(), "1"));
+
+ AtomicReference<ManifestCommittable> conflictCommittableRef = new
AtomicReference<>();
+ store.commitDataImpl(
+ Collections.singletonList(gen.nextInsert("20211112", 10, 4L,
null, "d")),
+ gen::getPartition,
+ value -> 0,
+ false,
+ 22L,
+ null,
+ Collections.emptyList(),
+ (commit, committable) ->
conflictCommittableRef.set(committable));
+ List<ManifestEntry> conflictDeltaFiles =
+ tableFilesFrom(checkNotNull(conflictCommittableRef.get()),
store.options());
+
+ AtomicReference<ManifestCommittable> committableRef = new
AtomicReference<>();
+ store.commitDataImpl(
+ Collections.singletonList(gen.nextInsert("20211110", 9, 3L,
null, "c")),
+ gen::getPartition,
+ value -> 0,
+ false,
+ 23L,
+ null,
+ Collections.emptyList(),
+ (commit, committable) -> committableRef.set(committable));
+
+ ConflictingSnapshotCommit snapshotCommit =
+ new ConflictingSnapshotCommit(
+ new RenamingSnapshotCommit(store.snapshotManager(),
Lock.empty()),
+ store.snapshotManager(),
+ store.manifestListFactory().create(),
+ store.manifestFileFactory().create(),
+ false,
+ conflictAttempts(conflictDeltaFiles));
+ try (FileStoreCommitImpl commit =
+ newCommitWithSnapshotCommit(store, "retry-reuse-empty-table",
snapshotCommit)) {
+ commit.commit(checkNotNull(committableRef.get()), false);
+ }
+
+ Snapshot latestSnapshot =
checkNotNull(store.snapshotManager().latestSnapshot());
+ List<ManifestFileMeta> finalBaseManifests =
+ store.manifestListFactory()
+ .create()
+ .read(
+ latestSnapshot.baseManifestList(),
+ latestSnapshot.baseManifestListSize());
+ assertThat(finalBaseManifests)
+
.containsExactlyElementsOf(snapshotCommit.conflictDeltaManifests());
+ assertThat(store.readKvsFromSnapshot(latestSnapshot.id())).hasSize(2);
+ }
+
+ @Test
+ public void testCommitRetrySkipsManifestMergePreservesDeleteOrder() throws
Exception {
+ TestFileStore store =
+ createStore(
+ false,
+ Collections.singletonMap(
+
CoreOptions.MANIFEST_FULL_COMPACTION_FILE_SIZE.key(), "1"));
+
+ store.commitData(
+ Collections.singletonList(gen.nextInsert("20211110", 8, 1L,
null, "a")),
+ gen::getPartition,
+ kv -> 0);
+ store.commitData(
+ Collections.singletonList(gen.nextInsert("20211111", 9, 2L,
null, "b")),
+ gen::getPartition,
+ kv -> 0);
+
+ Snapshot latestBeforeRetry =
checkNotNull(store.snapshotManager().latestSnapshot());
+ List<ManifestFileMeta> manifestsBeforeRetry =
+
store.manifestListFactory().create().readDataManifests(latestBeforeRetry);
+ assertThat(manifestsBeforeRetry).hasSize(2);
+ ManifestEntry firstEntry =
+ store.manifestFileFactory()
+ .create()
+ .read(manifestsBeforeRetry.get(0).fileName())
+ .get(0);
+ ManifestEntry deleteFirstEntry =
+ ManifestEntry.create(
+ FileKind.DELETE,
+ firstEntry.partition(),
+ firstEntry.bucket(),
+ firstEntry.totalBuckets(),
+ firstEntry.file());
+
+ AtomicReference<ManifestCommittable> committableRef = new
AtomicReference<>();
+ store.commitDataImpl(
+ Collections.singletonList(gen.nextInsert("20211110", 9, 3L,
null, "c")),
+ gen::getPartition,
+ value -> 0,
+ false,
+ 23L,
+ null,
+ Collections.emptyList(),
+ (commit, committable) -> committableRef.set(committable));
+
+ ConflictingSnapshotCommit snapshotCommit =
+ new ConflictingSnapshotCommit(
+ new RenamingSnapshotCommit(store.snapshotManager(),
Lock.empty()),
+ store.snapshotManager(),
+ store.manifestListFactory().create(),
+ store.manifestFileFactory().create(),
+ false,
+
conflictAttempts(Collections.singletonList(deleteFirstEntry)));
+ try (FileStoreCommitImpl commit =
+ newCommitWithSnapshotCommit(store, "retry-reuse-delete-order",
snapshotCommit)) {
+ commit.commit(checkNotNull(committableRef.get()), false);
+ }
+
+ Snapshot latestSnapshot =
checkNotNull(store.snapshotManager().latestSnapshot());
+ List<ManifestFileMeta> finalBaseManifests =
+ store.manifestListFactory()
+ .create()
+ .read(
+ latestSnapshot.baseManifestList(),
+ latestSnapshot.baseManifestListSize());
+ assertThat(finalBaseManifests)
+
.containsExactlyElementsOf(snapshotCommit.conflictBaseManifests());
+ assertThat(store.readKvsFromSnapshot(latestSnapshot.id()))
+ .extracting(kv -> kv.value().getString(6).toString())
+ .containsExactlyInAnyOrder("b", "c");
+ }
+
private FileStoreCommitImpl newCommitWithSnapshotCommit(
TestFileStore store, String commitUser, SnapshotCommit
snapshotCommit) {
return newCommitWithSnapshotCommit(
@@ -1281,6 +1672,177 @@ public class FileStoreCommitTest {
return committable;
}
+ private static List<ManifestEntry> tableFilesFrom(
+ ManifestCommittable committable, CoreOptions options) {
+ ManifestEntryChanges changes = new
ManifestEntryChanges(options.bucket());
+ committable.fileCommittables().forEach(changes::collect);
+ return new ArrayList<>(changes.appendTableFiles);
+ }
+
+ @SafeVarargs
+ private static List<List<ManifestEntry>>
conflictAttempts(List<ManifestEntry>... attempts) {
+ return Arrays.asList(attempts);
+ }
+
+ private static class ConflictingSnapshotCommit implements SnapshotCommit {
+
+ private final SnapshotCommit delegate;
+ private final SnapshotManager snapshotManager;
+ private final ManifestList manifestList;
+ private final ManifestFile manifestFile;
+ private final boolean mergeConflictManifests;
+ private final List<List<ManifestEntry>> conflictDeltaFilesByAttempt;
+ private int commitAttempt = 0;
+ private final List<List<ManifestFileMeta>> attemptBaseManifests;
+ private final List<List<ManifestFileMeta>>
conflictDeltaManifestsByAttempt;
+ private List<ManifestFileMeta> conflictBaseManifests;
+
+ private ConflictingSnapshotCommit(
+ SnapshotCommit delegate,
+ SnapshotManager snapshotManager,
+ ManifestList manifestList,
+ ManifestFile manifestFile,
+ boolean mergeConflictManifests,
+ List<List<ManifestEntry>> conflictDeltaFilesByAttempt) {
+ this.delegate = delegate;
+ this.snapshotManager = snapshotManager;
+ this.manifestList = manifestList;
+ this.manifestFile = manifestFile;
+ this.mergeConflictManifests = mergeConflictManifests;
+ this.conflictDeltaFilesByAttempt = conflictDeltaFilesByAttempt;
+ this.attemptBaseManifests = new ArrayList<>();
+ this.conflictDeltaManifestsByAttempt = new ArrayList<>();
+ }
+
+ @Override
+ public boolean commit(
+ Snapshot snapshot,
+ String branch,
+ List<org.apache.paimon.partition.PartitionStatistics>
statistics)
+ throws Exception {
+ if (commitAttempt >= conflictDeltaFilesByAttempt.size()) {
+ return delegate.commit(snapshot, branch, statistics);
+ }
+
+ List<ManifestEntry> conflictDeltaFiles =
conflictDeltaFilesByAttempt.get(commitAttempt);
+ commitAttempt++;
+ attemptBaseManifests.add(
+ manifestList.read(
+ snapshot.baseManifestList(),
snapshot.baseManifestListSize()));
+
+ Snapshot previousSnapshot =
+ snapshot.id() == Snapshot.FIRST_SNAPSHOT_ID
+ ? null
+ : snapshotManager.snapshot(snapshot.id() - 1);
+ List<ManifestFileMeta> previousManifests =
+ previousSnapshot == null
+ ? Collections.emptyList()
+ : manifestList.readDataManifests(previousSnapshot);
+ conflictBaseManifests =
+ mergeConflictManifests
+ ? rewriteManifests(previousManifests)
+ : previousManifests;
+ List<ManifestFileMeta> conflictNewManifests =
+ conflictDeltaFiles.isEmpty()
+ ? Collections.emptyList()
+ : manifestFile.write(conflictDeltaFiles);
+ conflictDeltaManifestsByAttempt.add(conflictNewManifests);
+ boolean putConflictNewManifestsInBase =
+ !mergeConflictManifests
+ && !conflictNewManifests.isEmpty()
+ && !previousManifests.isEmpty();
+ if (putConflictNewManifestsInBase) {
+ conflictBaseManifests = new ArrayList<>();
+ conflictBaseManifests.add(previousManifests.get(0));
+ conflictBaseManifests.addAll(conflictNewManifests);
+ conflictBaseManifests.addAll(
+ previousManifests.subList(1,
previousManifests.size()));
+ }
+ Pair<String, Long> conflictBaseManifestList =
manifestList.write(conflictBaseManifests);
+ long conflictDeltaRecordCount =
+ conflictDeltaFiles.stream()
+ .mapToLong(
+ entry ->
+ entry.kind() == FileKind.ADD
+ ? entry.file().rowCount()
+ : -entry.file().rowCount())
+ .sum();
+ Pair<String, Long> conflictDeltaManifestList =
+ manifestList.write(
+ putConflictNewManifestsInBase
+ ? Collections.emptyList()
+ : conflictNewManifests);
+ Snapshot conflictSnapshot =
+ new Snapshot(
+ snapshot.id(),
+ previousSnapshot == null
+ ? snapshot.schemaId()
+ : previousSnapshot.schemaId(),
+ conflictBaseManifestList.getLeft(),
+ conflictBaseManifestList.getRight(),
+ conflictDeltaManifestList.getLeft(),
+ conflictDeltaManifestList.getRight(),
+ null,
+ null,
+ previousSnapshot == null ? null :
previousSnapshot.indexManifest(),
+ "conflict-user",
+ Long.MAX_VALUE,
+ Snapshot.CommitKind.ANALYZE,
+ System.currentTimeMillis(),
+ (previousSnapshot == null ? 0L :
previousSnapshot.totalRecordCount())
+ + conflictDeltaRecordCount,
+ conflictDeltaRecordCount,
+ null,
+ previousSnapshot == null ? null :
previousSnapshot.watermark(),
+ previousSnapshot == null ? null :
previousSnapshot.statistics(),
+ previousSnapshot == null ? null :
previousSnapshot.properties(),
+ previousSnapshot == null ? null :
previousSnapshot.nextRowId());
+ assertThat(delegate.commit(conflictSnapshot, branch,
Collections.emptyList())).isTrue();
+ return false;
+ }
+
+ private List<ManifestFileMeta> rewriteManifests(List<ManifestFileMeta>
manifests) {
+ if (manifests.isEmpty()) {
+ return Collections.emptyList();
+ }
+
+ List<ManifestFileMeta> rewrittenManifests = new ArrayList<>();
+ for (ManifestFileMeta manifest : manifests) {
+ rewrittenManifests.addAll(
+
manifestFile.write(manifestFile.read(manifest.fileName())));
+ }
+ return rewrittenManifests;
+ }
+
+ private List<ManifestFileMeta> firstAttemptBaseManifests() {
+ return attemptBaseManifests(0);
+ }
+
+ private List<ManifestFileMeta> attemptBaseManifests(int attempt) {
+ assertThat(attemptBaseManifests).hasSizeGreaterThan(attempt);
+ return attemptBaseManifests.get(attempt);
+ }
+
+ private List<ManifestFileMeta> conflictBaseManifests() {
+ return checkNotNull(conflictBaseManifests);
+ }
+
+ private List<ManifestFileMeta> conflictDeltaManifests() {
+ assertThat(conflictDeltaManifestsByAttempt).isNotEmpty();
+ return
conflictDeltaManifestsByAttempt.get(conflictDeltaManifestsByAttempt.size() - 1);
+ }
+
+ private List<ManifestFileMeta> conflictDeltaManifests(int attempt) {
+
assertThat(conflictDeltaManifestsByAttempt).hasSizeGreaterThan(attempt);
+ return conflictDeltaManifestsByAttempt.get(attempt);
+ }
+
+ @Override
+ public void close() throws Exception {
+ delegate.close();
+ }
+ }
+
private static class FalseSuccessSnapshotCommit implements SnapshotCommit {
private final SnapshotCommit delegate;
