Skip to content

[fix](fe) Prevent stale recycle candidates from erasing new generations #67750

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wenzhenghu wants to merge 2 commits into
base: master
Choose a base branch
from
+172 −18
Open

[fix](fe) Prevent stale recycle candidates from erasing new generations #67750

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
0046e97
[fix](fe) Prevent stale recycle candidates from erasing new generations
wenzhenghu Sep 9, 2026
f0aba1d
Merge branch 'master' into wzh/upstream-fix-recycle-bin-aba
wenzhenghu Sep 10, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
61 changes: 43 additions & 18 deletions fe/fe-core/src/main/java/org/apache/doris/catalog/CatalogRecycleBin.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -91,6 +91,18 @@ private void writeUnlock() {
private ConcurrentHashMap<Long, RecyclePartitionInfo> idToPartition;
private ConcurrentHashMap<Long, Long> idToRecycleTime;

private static final class ExpiredCandidate<T> {
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.
Expand Down Expand Up @@ -303,31 +315,40 @@ private boolean isExpire(long id, long currentTimeMs) {
&& latency > Config.catalog_trash_expire_second * 1000L;
}

private <T> boolean isSameGenerationAndExpired(ExpiredCandidate<T> candidate,
Map<Long, T> 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<Long> expiredIds = new ArrayList<>();
// 1. collect expired database generations under read lock
List<ExpiredCandidate<RecycleDatabaseInfo>> expiredCandidates = new ArrayList<>();
readLock();
try {
for (Map.Entry<Long, RecycleDatabaseInfo> 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 {
readUnlock();
}

// 2. erase each expired database one at a time
for (Long dbId : expiredIds) {
for (ExpiredCandidate<RecycleDatabaseInfo> 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);

Expand Down Expand Up @@ -469,27 +490,29 @@ private void eraseTable(long currentTimeMs, int keepNum) {
int eraseNum = 0;
StopWatch watch = StopWatch.createStarted();
try {
// 1. collect expired table IDs under read lock
List<Long> expiredIds = new ArrayList<>();
// 1. collect expired table generations under read lock
List<ExpiredCandidate<RecycleTableInfo>> expiredCandidates = new ArrayList<>();
readLock();
try {
for (Map.Entry<Long, RecycleTableInfo> 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 {
readUnlock();
}

// 2. erase each expired table one at a time
for (Long tableId : expiredIds) {
for (ExpiredCandidate<RecycleTableInfo> 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);
Expand Down Expand Up @@ -614,27 +637,29 @@ private void erasePartition(long currentTimeMs, int keepNum) {
int eraseNum = 0;
StopWatch watch = StopWatch.createStarted();
try {
// 1. collect expired partition IDs under read lock
List<Long> expiredIds = new ArrayList<>();
// 1. collect expired partition generations under read lock
List<ExpiredCandidate<RecyclePartitionInfo>> expiredCandidates = new ArrayList<>();
readLock();
try {
for (Map.Entry<Long, RecyclePartitionInfo> 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 {
readUnlock();
}

// 2. erase each expired partition one at a time (microbatch)
for (Long partitionId : expiredIds) {
for (ExpiredCandidate<RecyclePartitionInfo> 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);
Expand Down
129 changes: 129 additions & 0 deletions fe/fe-core/src/test/java/org/apache/doris/catalog/CatalogRecycleBinTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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;

Expand Down Expand Up @@ -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);
}
}
}
}
Loading