diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/CatalogRecycleBin.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/CatalogRecycleBin.java index 50e901064a4f0f..bebd85b07f8738 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/catalog/CatalogRecycleBin.java +++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/CatalogRecycleBin.java @@ -91,6 +91,18 @@ private void writeUnlock() { private ConcurrentHashMap idToPartition; private ConcurrentHashMap idToRecycleTime; + private static final class ExpiredCandidate { + private final long id; + private final T recycleInfo; + private final long recycleTime; + + ExpiredCandidate(long id, T recycleInfo, long recycleTime) { + this.id = id; + this.recycleInfo = recycleInfo; + this.recycleTime = recycleTime; + } + } + // Caches below to avoid calculate meta with same name every demon run cycle. // When the meta is updated, these caches should be updated too. No need to // persist these caches because they can be recalculated when FE restarting. @@ -303,17 +315,25 @@ private boolean isExpire(long id, long currentTimeMs) { && latency > Config.catalog_trash_expire_second * 1000L; } + private boolean isSameGenerationAndExpired(ExpiredCandidate candidate, + Map recycleInfoMap, long currentTimeMs) { + return recycleInfoMap.get(candidate.id) == candidate.recycleInfo + && idToRecycleTime.get(candidate.id) == candidate.recycleTime + && isExpire(candidate.id, currentTimeMs); + } + private void eraseDatabase(long currentTimeMs, int keepNum) { int eraseNum = 0; StopWatch watch = StopWatch.createStarted(); try { - // 1. collect expired database IDs under read lock - List expiredIds = new ArrayList<>(); + // 1. collect expired database generations under read lock + List> expiredCandidates = new ArrayList<>(); readLock(); try { for (Map.Entry entry : idToDatabase.entrySet()) { if (isExpire(entry.getKey(), currentTimeMs)) { - expiredIds.add(entry.getKey()); + expiredCandidates.add(new ExpiredCandidate<>(entry.getKey(), entry.getValue(), + idToRecycleTime.get(entry.getKey()))); } } } finally { @@ -321,13 +341,14 @@ private void eraseDatabase(long currentTimeMs, int keepNum) { } // 2. erase each expired database one at a time - for (Long dbId : expiredIds) { + for (ExpiredCandidate candidate : expiredCandidates) { writeLock(); try { - RecycleDatabaseInfo dbInfo = idToDatabase.remove(dbId); - if (dbInfo == null) { + long dbId = candidate.id; + if (!isSameGenerationAndExpired(candidate, idToDatabase, currentTimeMs)) { continue; } + RecycleDatabaseInfo dbInfo = idToDatabase.remove(dbId); Database db = dbInfo.getDb(); idToRecycleTime.remove(dbId); @@ -469,13 +490,14 @@ private void eraseTable(long currentTimeMs, int keepNum) { int eraseNum = 0; StopWatch watch = StopWatch.createStarted(); try { - // 1. collect expired table IDs under read lock - List expiredIds = new ArrayList<>(); + // 1. collect expired table generations under read lock + List> expiredCandidates = new ArrayList<>(); readLock(); try { for (Map.Entry entry : idToTable.entrySet()) { if (isExpire(entry.getKey(), currentTimeMs)) { - expiredIds.add(entry.getKey()); + expiredCandidates.add(new ExpiredCandidate<>(entry.getKey(), entry.getValue(), + idToRecycleTime.get(entry.getKey()))); } } } finally { @@ -483,13 +505,14 @@ private void eraseTable(long currentTimeMs, int keepNum) { } // 2. erase each expired table one at a time - for (Long tableId : expiredIds) { + for (ExpiredCandidate candidate : expiredCandidates) { writeLock(); try { - RecycleTableInfo tableInfo = idToTable.get(tableId); - if (tableInfo == null) { + long tableId = candidate.id; + if (!isSameGenerationAndExpired(candidate, idToTable, currentTimeMs)) { continue; } + RecycleTableInfo tableInfo = idToTable.get(tableId); Table table = tableInfo.getTable(); try { Env.getCurrentInternalCatalog().beforeEraseTable(tableInfo.dbId, table, false); @@ -614,13 +637,14 @@ private void erasePartition(long currentTimeMs, int keepNum) { int eraseNum = 0; StopWatch watch = StopWatch.createStarted(); try { - // 1. collect expired partition IDs under read lock - List expiredIds = new ArrayList<>(); + // 1. collect expired partition generations under read lock + List> expiredCandidates = new ArrayList<>(); readLock(); try { for (Map.Entry entry : idToPartition.entrySet()) { if (isExpire(entry.getKey(), currentTimeMs)) { - expiredIds.add(entry.getKey()); + expiredCandidates.add(new ExpiredCandidate<>(entry.getKey(), entry.getValue(), + idToRecycleTime.get(entry.getKey()))); } } } finally { @@ -628,13 +652,14 @@ private void erasePartition(long currentTimeMs, int keepNum) { } // 2. erase each expired partition one at a time (microbatch) - for (Long partitionId : expiredIds) { + for (ExpiredCandidate candidate : expiredCandidates) { writeLock(); try { - RecyclePartitionInfo partitionInfo = idToPartition.remove(partitionId); - if (partitionInfo == null) { + long partitionId = candidate.id; + if (!isSameGenerationAndExpired(candidate, idToPartition, currentTimeMs)) { continue; } + RecyclePartitionInfo partitionInfo = idToPartition.remove(partitionId); Partition partition = partitionInfo.getPartition(); Env.getCurrentEnv().onErasePartition(partition); idToRecycleTime.remove(partitionId); diff --git a/fe/fe-core/src/test/java/org/apache/doris/catalog/CatalogRecycleBinTest.java b/fe/fe-core/src/test/java/org/apache/doris/catalog/CatalogRecycleBinTest.java index a06e88ef4cdcb2..835109da6961b8 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/catalog/CatalogRecycleBinTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/catalog/CatalogRecycleBinTest.java @@ -25,6 +25,7 @@ import org.apache.doris.common.DdlException; import org.apache.doris.common.FeConstants; import org.apache.doris.common.Pair; +import org.apache.doris.common.jmockit.Deencapsulation; import org.apache.doris.common.util.URI; import org.apache.doris.nereids.trees.expressions.functions.FunctionBuilder; import org.apache.doris.thrift.TInvertedIndexFileStorageFormat; @@ -52,9 +53,14 @@ import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.locks.ReentrantReadWriteLock; import java.util.stream.Collectors; public class CatalogRecycleBinTest extends TestWithFeService { + private interface ThrowingRunnable { + void run() throws Exception; + } + private static final long ROW_BINLOG_INDEX_ID = 10001L; private static final long ROW_BINLOG_TABLET_ID = 10002L; @@ -1139,4 +1145,127 @@ public void testMicrobatchEraseReleasesLockBetweenItems() throws Exception { CatalogTestUtil.testTableId1, 9000)); } } + + @Test + public void testExpiredDatabaseSnapshotDoesNotEraseNewGeneration() throws Exception { + CatalogRecycleBin recycleBin = new CatalogRecycleBin(); + Database database = new Database(CatalogTestUtil.testDbId1, CatalogTestUtil.testDb1); + Assertions.assertTrue(recycleBin.recycleDatabase(database, Sets.newHashSet(), Sets.newHashSet(), + false, false, 0)); + recycleBin.setRecycleTimeByIdForReplay(database.getId(), 0L); + + runEraseAfterReplacingGeneration(recycleBin, "eraseDatabase", () -> { + Database recovered = recycleBin.recoverDatabase(CatalogTestUtil.testDb1, database.getId()); + Assertions.assertTrue(recycleBin.recycleDatabase(recovered, Sets.newHashSet(), Sets.newHashSet(), + false, false, 0)); + }); + + Assertions.assertTrue(recycleBin.isRecycleDatabase(database.getId())); + Assertions.assertTrue(recycleBin.getRecycleTimeById(database.getId()) > 0); + } + + @Test + public void testExpiredTableSnapshotDoesNotEraseNewGeneration() throws Exception { + CatalogRecycleBin recycleBin = new CatalogRecycleBin(); + Database database = createSimpleDatabase(); + OlapTable table = (OlapTable) database.getTable(CatalogTestUtil.testTableId1).get(); + Assertions.assertTrue(recycleBin.recycleTable(database.getId(), table, false, false, 0)); + recycleBin.setRecycleTimeByIdForReplay(table.getId(), 0L); + + runEraseAfterReplacingGeneration(recycleBin, "eraseTable", () -> { + Assertions.assertTrue(recycleBin.recoverTable(database, table.getName(), table.getId(), null)); + Assertions.assertTrue(recycleBin.recycleTable(database.getId(), table, false, false, 0)); + }); + + Assertions.assertTrue(recycleBin.isRecycleTable(database.getId(), table.getId())); + Assertions.assertTrue(recycleBin.getRecycleTimeById(table.getId()) > 0); + } + + @Test + public void testExpiredPartitionSnapshotDoesNotEraseNewGeneration() throws Exception { + CatalogRecycleBin recycleBin = new CatalogRecycleBin(); + Database database = createSimpleDatabase(); + OlapTable table = (OlapTable) database.getTable(CatalogTestUtil.testTableId1).get(); + Partition partition = table.getPartition(CatalogTestUtil.testPartitionId1); + recyclePartition(recycleBin, database, table, partition); + recycleBin.setRecycleTimeByIdForReplay(partition.getId(), 0L); + + runEraseAfterReplacingGeneration(recycleBin, "erasePartition", () -> { + recycleBin.recoverPartition(database.getId(), table, partition.getName(), partition.getId(), null); + recyclePartition(recycleBin, database, table, partition); + }); + + Assertions.assertTrue(recycleBin.isRecyclePartition(database.getId(), table.getId(), partition.getId())); + Assertions.assertTrue(recycleBin.getRecycleTimeById(partition.getId()) > 0); + } + + private Database createSimpleDatabase() { + return CatalogTestUtil.createSimpleDb( + CatalogTestUtil.testDbId1, + CatalogTestUtil.testTableId1, + CatalogTestUtil.testPartitionId1, + CatalogTestUtil.testIndexId1, + CatalogTestUtil.testTabletId1, + CatalogTestUtil.testStartVersion); + } + + private void recyclePartition(CatalogRecycleBin recycleBin, Database database, + OlapTable table, Partition partition) { + Assertions.assertTrue(recycleBin.recyclePartition(database.getId(), table.getId(), table.getName(), + partition, null, null, new DataProperty(TStorageMedium.HDD), + new ReplicaAllocation((short) 3), false, false)); + } + + private void runEraseAfterReplacingGeneration(CatalogRecycleBin recycleBin, + String eraseMethod, ThrowingRunnable replaceGeneration) throws Exception { + CountDownLatch snapshotCollected = new CountDownLatch(1); + CountDownLatch continueErase = new CountDownLatch(1); + ReentrantReadWriteLock originalLock = Deencapsulation.getField(recycleBin, "lock"); + ReentrantReadWriteLock testLock = new ReentrantReadWriteLock(true) { + private final ReadLock readLock = new ReadLock(this) { + @Override + public void unlock() { + super.unlock(); + if (Thread.currentThread().getName().equals("stale-candidate-erase") + && snapshotCollected.getCount() != 0) { + snapshotCollected.countDown(); + try { + Assertions.assertTrue(continueErase.await(30, TimeUnit.SECONDS), + "Timed out waiting to resume erase"); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new AssertionError(e); + } + } + } + }; + + @Override + public ReadLock readLock() { + return readLock; + } + }; + Deencapsulation.setField(recycleBin, "lock", testLock); + ExecutorService executor = Executors.newSingleThreadExecutor(); + try { + Future erase = executor.submit(() -> { + Thread.currentThread().setName("stale-candidate-erase"); + Deencapsulation.invoke(recycleBin, eraseMethod, System.currentTimeMillis(), -1); + }); + Assertions.assertTrue(snapshotCollected.await(10, TimeUnit.SECONDS), + "Expired snapshot was not collected"); + replaceGeneration.run(); + continueErase.countDown(); + erase.get(10, TimeUnit.SECONDS); + } finally { + continueErase.countDown(); + executor.shutdown(); + try { + Assertions.assertTrue(executor.awaitTermination(30, TimeUnit.SECONDS), + "Erase worker did not finish"); + } finally { + Deencapsulation.setField(recycleBin, "lock", originalLock); + } + } + } }