You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by ju...@apache.org on 2024/09/27 09:49:56 UTC
(paimon) branch master updated: [core] Check existence before expire: more check. (#4272)
This is an automated email from the ASF dual-hosted git repository.
junhao 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 66cc214b4 [core] Check existence before expire: more check. (#4272)
66cc214b4 is described below
commit 66cc214b4261b43caa50d795779d7b9096119f9f
Author: YeJunHao <41...@users.noreply.github.com>
AuthorDate: Fri Sep 27 17:49:51 2024 +0800
[core] Check existence before expire: more check. (#4272)
---
.../apache/paimon/table/ExpireSnapshotsImpl.java | 22 ++++++++++-
.../paimon/operation/ExpireSnapshotsTest.java | 46 ++++++++++++++++++++++
2 files changed, 66 insertions(+), 2 deletions(-)
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/ExpireSnapshotsImpl.java b/paimon-core/src/main/java/org/apache/paimon/table/ExpireSnapshotsImpl.java
index bc04cc4ca..eb163f8fa 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/ExpireSnapshotsImpl.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/ExpireSnapshotsImpl.java
@@ -35,6 +35,7 @@ import org.slf4j.LoggerFactory;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.UncheckedIOException;
+import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.function.Predicate;
@@ -213,8 +214,25 @@ public class ExpireSnapshotsImpl implements ExpireSnapshots {
List<Snapshot> skippingSnapshots =
SnapshotManager.findOverlappedSnapshots(
taggedSnapshots, beginInclusiveId, endExclusiveId);
- skippingSnapshots.add(snapshotManager.snapshot(endExclusiveId));
- Set<String> skippingSet = snapshotDeletion.manifestSkippingSet(skippingSnapshots);
+
+ try {
+ skippingSnapshots.add(snapshotManager.tryGetSnapshot(endExclusiveId));
+ } catch (FileNotFoundException e) {
+ // the end exclusive snapshot is gone
+ // there is no need to proceed
+ return 0;
+ }
+
+ Set<String> skippingSet = new HashSet<>();
+ try {
+ skippingSet.addAll(snapshotDeletion.manifestSkippingSet(skippingSnapshots));
+ } catch (Exception e) {
+ // maybe snapshot been deleted by other jobs.
+ if (e.getCause() == null || !(e.getCause() instanceof FileNotFoundException)) {
+ throw e;
+ }
+ }
+
for (long id = beginInclusiveId; id < endExclusiveId; id++) {
if (LOG.isDebugEnabled()) {
LOG.debug("Ready to delete manifests in snapshot #" + id);
diff --git a/paimon-core/src/test/java/org/apache/paimon/operation/ExpireSnapshotsTest.java b/paimon-core/src/test/java/org/apache/paimon/operation/ExpireSnapshotsTest.java
index bd32e48e5..16a2ab56b 100644
--- a/paimon-core/src/test/java/org/apache/paimon/operation/ExpireSnapshotsTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/operation/ExpireSnapshotsTest.java
@@ -42,6 +42,7 @@ import org.apache.paimon.utils.RecordWriter;
import org.apache.paimon.utils.SnapshotManager;
import org.apache.paimon.utils.TagManager;
+import org.assertj.core.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
@@ -56,6 +57,7 @@ import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Random;
import java.util.Set;
import java.util.concurrent.ThreadLocalRandom;
import java.util.stream.Collectors;
@@ -292,6 +294,50 @@ public class ExpireSnapshotsTest {
store.assertCleaned();
}
+ @Test
+ public void testExpireEmptySnapshot() throws Exception {
+ Random random = new Random();
+
+ List<KeyValue> allData = new ArrayList<>();
+ List<Integer> snapshotPositions = new ArrayList<>();
+ commit(100, allData, snapshotPositions);
+ int latestSnapshotId = requireNonNull(snapshotManager.latestSnapshotId()).intValue();
+
+ List<Thread> s = new ArrayList<>();
+ s.add(
+ new Thread(
+ () -> {
+ final ExpireSnapshotsImpl expire =
+ (ExpireSnapshotsImpl) store.newExpire(1, Integer.MAX_VALUE, 1);
+ expire.expireUntil(89, latestSnapshotId);
+ }));
+ for (int i = 0; i < 10; i++) {
+ final ExpireSnapshotsImpl expire =
+ (ExpireSnapshotsImpl) store.newExpire(1, Integer.MAX_VALUE, 1);
+ s.add(
+ new Thread(
+ () -> {
+ int start = random.nextInt(latestSnapshotId - 10);
+ int end = start + random.nextInt(10);
+ expire.expireUntil(start, end);
+ }));
+ }
+
+ Assertions.assertThatCode(
+ () -> {
+ s.forEach(Thread::start);
+ s.forEach(
+ tt -> {
+ try {
+ tt.join();
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ });
+ })
+ .doesNotThrowAnyException();
+ }
+
@Test
public void testExpireWithNumber() throws Exception {
ExpireSnapshots expire = store.newExpire(1, 3, Long.MAX_VALUE);