This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new 8c21bda9fd4 branch-4.1: [fix](fe) Skip DEC journal when dictionary
dropped during load commit- #66552 (#66724)
8c21bda9fd4 is described below
commit 8c21bda9fd47d4331b9fd53194f8a09c5f9d1833
Author: linrrarity <[email protected]>
AuthorDate: Thu Aug 13 18:54:36 2026 +0800
branch-4.1: [fix](fe) Skip DEC journal when dictionary dropped during load
commit- #66552 (#66724)
cherry pick: https://github.com/apache/doris/pull/65476 and
https://github.com/apache/doris/pull/66552
---------
Co-authored-by: zclllyybb <[email protected]>
---
.../apache/doris/dictionary/DictionaryManager.java | 97 +++++++++--
.../nereids/analyzer/UnboundDictionarySink.java | 17 +-
.../nereids/analyzer/UnboundTableSinkCreator.java | 7 +-
.../doris/nereids/rules/analysis/BindSink.java | 18 +-
.../commands/insert/AbstractInsertExecutor.java | 12 +-
.../commands/insert/DictionaryInsertExecutor.java | 8 +-
.../insert/InsertIntoDictionaryCommand.java | 17 +-
.../commands/insert/InsertIntoTableCommand.java | 15 +-
.../doris/dictionary/DictionaryManagerTest.java | 112 ++++++++++++
.../insert/DictionaryInsertTargetDropRaceTest.java | 188 +++++++++++++++++++++
.../dictionary_p0/test_create_drop_sync.groovy | 36 +++-
...t_dictionary_drop_while_load_commit_fail.groovy | 113 +++++++++++++
12 files changed, 588 insertions(+), 52 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/dictionary/DictionaryManager.java
b/fe/fe-core/src/main/java/org/apache/doris/dictionary/DictionaryManager.java
index 7ebc701c32a..b389f4756c1 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/dictionary/DictionaryManager.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/dictionary/DictionaryManager.java
@@ -18,6 +18,7 @@
package org.apache.doris.dictionary;
import org.apache.doris.analysis.UserIdentity;
+import org.apache.doris.catalog.Database;
import org.apache.doris.catalog.Env;
import org.apache.doris.common.AnalysisException;
import org.apache.doris.common.ClientPool;
@@ -27,6 +28,7 @@ import org.apache.doris.common.DdlException;
import org.apache.doris.common.Status;
import org.apache.doris.common.io.Text;
import org.apache.doris.common.io.Writable;
+import org.apache.doris.common.util.DebugPointUtil;
import org.apache.doris.common.util.MasterDaemon;
import org.apache.doris.dictionary.Dictionary.DictionaryStatus;
import org.apache.doris.job.extensions.insert.InsertTask;
@@ -267,6 +269,7 @@ public class DictionaryManager extends MasterDaemon
implements Writable {
// Log the drop operation
if (dbDictIds != null) {
for (Map.Entry<String, Long> entry : dbDictIds.entrySet()) {
+ idToDictionary.remove(entry.getValue());
Env.getCurrentEnv().getEditLog().logDropDictionary(dbName,
entry.getKey());
}
// also drop all name mapping records.
@@ -282,6 +285,37 @@ public class DictionaryManager extends MasterDaemon
implements Writable {
return dbDictIds != null && dbDictIds.containsKey(dictName);
}
+ public boolean isCurrentDictionary(Database database, Dictionary
dictionary) {
+ lockRead();
+ try {
+ return isCurrentDictionaryWithoutLock(database, dictionary);
+ } finally {
+ unlockRead();
+ }
+ }
+
+ private boolean isCurrentDictionaryWithoutLock(Database database,
Dictionary dictionary) {
+ Map<String, Long> dbDictIds =
dictionaryIds.get(dictionary.getDbName());
+ Long dictionaryId = dbDictIds == null ? null :
dbDictIds.get(dictionary.getName());
+ return Env.getCurrentInternalCatalog().getDbNullable(database.getId())
== database
+ && dictionaryId != null
+ && dictionaryId == dictionary.getId()
+ && idToDictionary.get(dictionary.getId()) == dictionary;
+ }
+
+ private Database getCurrentDatabase(Dictionary dictionary) throws
AnalysisException {
+ lockRead();
+ try {
+ Database database =
Env.getCurrentInternalCatalog().getDbNullable(dictionary.getDbName());
+ if (database == null || !isCurrentDictionaryWithoutLock(database,
dictionary)) {
+ throw new AnalysisException("Dictionary " +
dictionary.getName() + " has been dropped");
+ }
+ return database;
+ } finally {
+ unlockRead();
+ }
+ }
+
public Map<String, Dictionary> getDictionaries(String dbName) {
lockRead();
try {
@@ -406,12 +440,19 @@ public class DictionaryManager extends MasterDaemon
implements Writable {
LOG.info("skip adaptive dataLoad of dictionary " +
dictionary.getName() + ". maybe last load finished.");
return;
}
+ // Resolve first so every LOADING refresh carries one exact owner
generation.
+ Database database = getCurrentDatabase(dictionary);
+
// use atomic status as a lock.
if (!dictionary.trySetStatus(Dictionary.DictionaryStatus.LOADING)) {
throw new AnalysisException("Dictionary " + dictionary.getName() +
" cannot load now, status is "
+ dictionary.getStatus().name());
}
+ while
(DebugPointUtil.isEnable("DictionaryManager.dataLoad.blockBeforePlan")) {
+ Thread.sleep(10);
+ }
+
if (ctx == null) { // for run with scheduler, not by command.
// priv check is done in relative(caller) command. so use ADMIN
here is ok.
ctx = InsertTask.makeConnectContext(UserIdentity.ADMIN,
dictionary.getDbName());
@@ -433,7 +474,8 @@ public class DictionaryManager extends MasterDaemon
implements Writable {
baseCommand.setJobId(DICTIONARY_JOB_ID);
}
- InsertIntoDictionaryCommand command = new
InsertIntoDictionaryCommand(baseCommand, dictionary, adaptiveLoad);
+ InsertIntoDictionaryCommand command = new InsertIntoDictionaryCommand(
+ baseCommand, database, dictionary, adaptiveLoad);
// run with sync by status.
try {
@@ -464,8 +506,7 @@ public class DictionaryManager extends MasterDaemon
implements Writable {
lockRead();
boolean unlocked = false;
try {
- if (!dictionaryIds.containsKey(dictionary.getDbName())
- ||
!dictionaryIds.get(dictionary.getDbName()).containsKey(dictionary.getName())) {
+ if (!isCurrentDictionaryWithoutLock(database, dictionary)) {
unlockRead();
unlocked = true;
@@ -495,11 +536,24 @@ public class DictionaryManager extends MasterDaemon
implements Writable {
}
}
+ // block here in test to simulate the race: INC journal written,
commit not done yet.
+ while (DebugPointUtil.isEnable("DictionaryManager.afterIncJournal")) {
+ Thread.sleep(100);
+ }
+
// commit and check the result. not modify metadata so dont need lock.
if (!commitNowVersion(ctx, dictionary)) {
if (!ctx.getStatementContext().isPartialLoadDictionary()) {
dictionary.decreaseVersion();
-
Env.getCurrentEnv().getEditLog().logDictionaryDecVersion(dictionary);
+ // DROP may have removed the dictionary between the INC
journal and this failed
+ // commit. A DEC journal for a dropped dictionary cannot be
replayed by name, so
+ // only persist the rollback while the dictionary is still the
current one.
+ if (isCurrentDictionary(database, dictionary)) {
+
Env.getCurrentEnv().getEditLog().logDictionaryDecVersion(dictionary);
+ } else {
+ LOG.warn("Dictionary {} has been dropped or replaced
during commit, skip DEC journal",
+ dictionary.getName());
+ }
}
dictionary.trySetStatus(oldStatus);
abortSpecificVersion(ctx, dictionary, dictionary.getVersion() + 1);
@@ -525,6 +579,9 @@ public class DictionaryManager extends MasterDaemon
implements Writable {
}
private boolean commitNowVersion(ConnectContext ctx, Dictionary
dictionary) {
+ if
(DebugPointUtil.isEnable("DictionaryManager.commitNowVersion.fail")) {
+ return false;
+ }
// use the same BEs when we get before start loading.
List<Backend> beList =
ctx.getStatementContext().getUsedBackendsDistributing();
@@ -820,21 +877,33 @@ public class DictionaryManager extends MasterDaemon
implements Writable {
}
public void replayIncreaseVersion(DictionaryIncreaseVersionInfo info)
throws DdlException {
- String dbName = info.getDictionary().getDbName();
- String dictName = info.getDictionary().getName();
- Dictionary dictionary = getDictionary(dbName, dictName);
+ long dictId = info.getDictionary().getId();
+ Dictionary dictionary = getDictionary(dictId);
+ if (dictionary == null) {
+ LOG.warn("Dictionary with id {} does not exist when replaying
increase version, skip", dictId);
+ return;
+ }
dictionary.writeLock();
- dictionary.increaseVersion();
- dictionary.writeUnlock();
+ try {
+ dictionary.increaseVersion();
+ } finally {
+ dictionary.writeUnlock();
+ }
}
public void replayDecreaseVersion(DictionaryDecreaseVersionInfo info)
throws DdlException {
- String dbName = info.getDictionary().getDbName();
- String dictName = info.getDictionary().getName();
- Dictionary dictionary = getDictionary(dbName, dictName);
+ long dictId = info.getDictionary().getId();
+ Dictionary dictionary = getDictionary(dictId);
+ if (dictionary == null) {
+ LOG.warn("Dictionary with id {} does not exist when replaying
decrease version, skip", dictId);
+ return;
+ }
dictionary.writeLock();
- dictionary.decreaseVersion();
- dictionary.writeUnlock();
+ try {
+ dictionary.decreaseVersion();
+ } finally {
+ dictionary.writeUnlock();
+ }
}
// Metadata serialization
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/analyzer/UnboundDictionarySink.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/analyzer/UnboundDictionarySink.java
index 5d6cc853b10..2750c5e837d 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/analyzer/UnboundDictionarySink.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/analyzer/UnboundDictionarySink.java
@@ -17,6 +17,7 @@
package org.apache.doris.nereids.analyzer;
+import org.apache.doris.catalog.Database;
import org.apache.doris.dictionary.Dictionary;
import org.apache.doris.nereids.exceptions.UnboundException;
import org.apache.doris.nereids.memo.GroupExpression;
@@ -45,15 +46,16 @@ import java.util.Optional;
public class UnboundDictionarySink<CHILD_TYPE extends Plan> extends
UnboundLogicalSink<CHILD_TYPE>
implements Unbound, Sink, BlockFuncDepsPropagation {
+ private final Database database;
private final Dictionary dictionary;
private final boolean allowAdaptiveLoad;
/**
* create unbound sink for dictionary sink
*/
- public UnboundDictionarySink(Dictionary dictionary, CHILD_TYPE child,
boolean adaptiveLoad) {
+ public UnboundDictionarySink(Database database, Dictionary dictionary,
CHILD_TYPE child, boolean adaptiveLoad) {
// all the empty arguments is like UnboundTableSink
-
super(ImmutableList.copyOf(dictionary.getNameWithFullQualifiers().split("\\.")),
// nameParts
+ super(ImmutableList.of(database.getCatalog().getName(),
database.getFullName(), dictionary.getName()),
PlanType.LOGICAL_UNBOUND_DICTIONARY_SINK, // type
ImmutableList.of(), // outputExprs
Optional.empty(), // groupExpression
@@ -61,10 +63,15 @@ public class UnboundDictionarySink<CHILD_TYPE extends Plan>
extends UnboundLogic
dictionary.getColumnNames(), // colNames from dictionary
DMLCommandType.INSERT, // dmlCommandType
child);
+ this.database = database;
this.dictionary = dictionary;
this.allowAdaptiveLoad = adaptiveLoad;
}
+ public Database getDatabase() {
+ return database;
+ }
+
public Dictionary getDictionary() {
return dictionary;
}
@@ -76,7 +83,7 @@ public class UnboundDictionarySink<CHILD_TYPE extends Plan>
extends UnboundLogic
@Override
public Plan withChildren(List<Plan> children) {
Preconditions.checkArgument(children.size() == 1,
"UnboundDictionarySink only accepts one child");
- return new UnboundDictionarySink<>(dictionary, children.get(0),
allowAdaptiveLoad);
+ return new UnboundDictionarySink<>(database, dictionary,
children.get(0), allowAdaptiveLoad);
}
@Override
@@ -91,13 +98,13 @@ public class UnboundDictionarySink<CHILD_TYPE extends Plan>
extends UnboundLogic
@Override
public Plan withGroupExpression(Optional<GroupExpression> groupExpression)
{
- return new UnboundDictionarySink<>(dictionary, child(),
allowAdaptiveLoad);
+ return new UnboundDictionarySink<>(database, dictionary, child(),
allowAdaptiveLoad);
}
@Override
public Plan withGroupExprLogicalPropChildren(Optional<GroupExpression>
groupExpression,
Optional<LogicalProperties> logicalProperties, List<Plan>
children) {
- return new UnboundDictionarySink<>(dictionary, children.get(0),
allowAdaptiveLoad);
+ return new UnboundDictionarySink<>(database, dictionary,
children.get(0), allowAdaptiveLoad);
}
@Override
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/analyzer/UnboundTableSinkCreator.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/analyzer/UnboundTableSinkCreator.java
index 1c5b5bf6064..c25b771daa3 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/analyzer/UnboundTableSinkCreator.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/analyzer/UnboundTableSinkCreator.java
@@ -17,6 +17,7 @@
package org.apache.doris.nereids.analyzer;
+import org.apache.doris.catalog.Database;
import org.apache.doris.catalog.Env;
import org.apache.doris.common.UserException;
import org.apache.doris.datasource.CatalogIf;
@@ -182,8 +183,8 @@ public class UnboundTableSinkCreator {
/**
* create unbound sink for dictionary sink
*/
- public static UnboundDictionarySink<? extends Plan>
createUnboundDictionarySink(Dictionary dictionary,
- LogicalPlan child, boolean adaptiveLoad) {
- return new UnboundDictionarySink<>(dictionary, child, adaptiveLoad);
+ public static UnboundDictionarySink<? extends Plan>
createUnboundDictionarySink(Database database,
+ Dictionary dictionary, LogicalPlan child, boolean adaptiveLoad) {
+ return new UnboundDictionarySink<>(database, dictionary, child,
adaptiveLoad);
}
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindSink.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindSink.java
index 17f7f7208a7..c6053e8b8c3 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindSink.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindSink.java
@@ -1128,9 +1128,8 @@ public class BindSink implements AnalysisRuleFactory {
private Plan
bindDictionarySink(MatchingContext<UnboundDictionarySink<Plan>> ctx) {
UnboundDictionarySink<?> sink = ctx.root;
- Pair<Database, Dictionary> pair = bind(ctx.cascadesContext, sink);
- Database database = pair.first;
- Dictionary dictionary = pair.second;
+ Database database = sink.getDatabase();
+ Dictionary dictionary = sink.getDictionary();
LogicalPlan child = ((LogicalPlan) sink.child());
// 1. bind target columns: from sink's column names to target tables'
Columns
@@ -1266,19 +1265,6 @@ public class BindSink implements AnalysisRuleFactory {
throw new AnalysisException("the target table of insert into is not an
jdbc table");
}
- private Pair<Database, Dictionary> bind(CascadesContext cascadesContext,
- UnboundDictionarySink<? extends Plan> sink) {
- Dictionary dictionary = sink.getDictionary();
- Database db;
- try {
- db =
cascadesContext.getConnectContext().getEnv().getInternalCatalog()
-
.getDbOrAnalysisException(dictionary.getDatabase().getName());
- } catch (org.apache.doris.common.AnalysisException e) {
- throw new AnalysisException(e.getMessage());
- }
- return Pair.of(db, dictionary);
- }
-
private List<Long> bindPartitionIds(OlapTable table, List<String>
partitions, boolean temp) {
return partitions.isEmpty()
? ImmutableList.of()
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/AbstractInsertExecutor.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/AbstractInsertExecutor.java
index 8c0ac00b961..b8c34291c93 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/AbstractInsertExecutor.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/AbstractInsertExecutor.java
@@ -46,6 +46,7 @@ import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import java.util.List;
+import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.CopyOnWriteArrayList;
@@ -105,8 +106,17 @@ public abstract class AbstractInsertExecutor {
*/
public AbstractInsertExecutor(ConnectContext ctx, TableIf table, String
labelName, NereidsPlanner planner,
Optional<InsertCommandContext> insertCtx, boolean emptyInsert,
long jobId, boolean needRegister) {
+ this(ctx, table.getDatabase(), table, labelName, planner, insertCtx,
emptyInsert, jobId, needRegister);
+ }
+
+ /**
+ * Dictionary loads must retain the owner resolved before a concurrent
database drop.
+ */
+ public AbstractInsertExecutor(ConnectContext ctx, DatabaseIf<?> database,
TableIf table, String labelName,
+ NereidsPlanner planner, Optional<InsertCommandContext> insertCtx,
boolean emptyInsert, long jobId,
+ boolean needRegister) {
this.ctx = ctx;
- this.database = table.getDatabase();
+ this.database = Objects.requireNonNull(database, "database should not
be null");
this.insertLoadJob = new InsertLoadJob(database.getId(), labelName,
jobId);
if (needRegister) {
ctx.getEnv().getLoadManager().addLoadJob(insertLoadJob);
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/DictionaryInsertExecutor.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/DictionaryInsertExecutor.java
index d29a3350ba3..f045a4c19b4 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/DictionaryInsertExecutor.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/DictionaryInsertExecutor.java
@@ -17,6 +17,7 @@
package org.apache.doris.nereids.trees.plans.commands.insert;
+import org.apache.doris.catalog.DatabaseIf;
import org.apache.doris.common.DdlException;
import org.apache.doris.common.UserException;
import org.apache.doris.common.util.DebugUtil;
@@ -42,9 +43,10 @@ public class DictionaryInsertExecutor extends
AbstractInsertExecutor {
/**
* constructor
*/
- public DictionaryInsertExecutor(ConnectContext ctx, Dictionary dictionary,
String labelName, NereidsPlanner planner,
- Optional<InsertCommandContext> insertCtx, boolean emptyInsert,
long jobId) {
- super(ctx, dictionary, labelName, planner, insertCtx, emptyInsert,
jobId);
+ public DictionaryInsertExecutor(ConnectContext ctx, DatabaseIf<?>
database, Dictionary dictionary,
+ String labelName, NereidsPlanner planner,
Optional<InsertCommandContext> insertCtx, boolean emptyInsert,
+ long jobId) {
+ super(ctx, database, dictionary, labelName, planner, insertCtx,
emptyInsert, jobId, false);
}
@Override
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoDictionaryCommand.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoDictionaryCommand.java
index 45615f62c5b..c589300c545 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoDictionaryCommand.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoDictionaryCommand.java
@@ -18,6 +18,8 @@
package org.apache.doris.nereids.trees.plans.commands.insert;
import org.apache.doris.analysis.RedirectStatus;
+import org.apache.doris.catalog.Database;
+import org.apache.doris.catalog.DatabaseIf;
import org.apache.doris.catalog.TableIf;
import org.apache.doris.dictionary.Dictionary;
import org.apache.doris.nereids.analyzer.UnboundDictionarySink;
@@ -35,19 +37,22 @@ import java.util.List;
* logic of InsertIntoTableCommand to maximize code reuse.
*/
public class InsertIntoDictionaryCommand extends InsertIntoTableCommand {
+ private final Database database;
private final Dictionary dictionary;
/**
* Constructor for InsertIntoDictionaryCommand.
*
* @param baseCommand The base InsertIntoTableCommand to copy from
+ * @param database The retained owner of the target dictionary
* @param dictionary The target dictionary to insert into
* @param adaptiveLoad see DictionaryManager.submitDataLoad
* @throws AnalysisException if the logical query is not a valid sink
*/
- public InsertIntoDictionaryCommand(InsertIntoTableCommand baseCommand,
Dictionary dictionary,
+ public InsertIntoDictionaryCommand(InsertIntoTableCommand baseCommand,
Database database, Dictionary dictionary,
boolean adaptiveLoad) {
super(baseCommand, PlanType.INSERT_INTO_DICTIONARY_COMMAND);
+ this.database = database;
this.dictionary = dictionary;
// Change sink type from olap table(need check) to dictionary
@@ -57,7 +62,7 @@ public class InsertIntoDictionaryCommand extends
InsertIntoTableCommand {
}
UnboundTableSink<?> sink = (UnboundTableSink<?>) logicalQuery;
- UnboundDictionarySink<?> newSink =
UnboundTableSinkCreator.createUnboundDictionarySink(dictionary,
+ UnboundDictionarySink<?> newSink =
UnboundTableSinkCreator.createUnboundDictionarySink(database, dictionary,
(LogicalPlan) sink.child(0), adaptiveLoad);
setLogicalQuery(newSink);
setOriginLogicalQuery(newSink);
@@ -71,9 +76,17 @@ public class InsertIntoDictionaryCommand extends
InsertIntoTableCommand {
@Override
protected TableIf getTargetTableIf(ConnectContext ctx, List<String>
qualifiedTargetTableName) {
+ if (!ctx.getEnv().getDictionaryManager().isCurrentDictionary(database,
dictionary)) {
+ throw new AnalysisException("Dictionary " + dictionary.getName() +
" has been dropped");
+ }
return dictionary;
}
+ @Override
+ protected DatabaseIf<?> getTargetDatabase(TableIf targetTable) {
+ return database;
+ }
+
public Dictionary getDictionary() {
return dictionary;
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTableCommand.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTableCommand.java
index 5db7a45f8a9..7618f7801e0 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTableCommand.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTableCommand.java
@@ -20,6 +20,7 @@ package org.apache.doris.nereids.trees.plans.commands.insert;
import org.apache.doris.analysis.RedirectStatus;
import org.apache.doris.analysis.StmtType;
import org.apache.doris.catalog.Column;
+import org.apache.doris.catalog.DatabaseIf;
import org.apache.doris.catalog.Env;
import org.apache.doris.catalog.OlapTable;
import org.apache.doris.catalog.TableIf;
@@ -239,6 +240,10 @@ public class InsertIntoTableCommand extends Command
implements NeedAuditEncrypti
return RelationUtil.getTable(qualifiedTargetTableName, ctx.getEnv(),
Optional.empty());
}
+ protected DatabaseIf<?> getTargetDatabase(TableIf targetTable) {
+ return targetTable.getDatabase();
+ }
+
public AbstractInsertExecutor initPlan(ConnectContext ctx, StmtExecutor
executor) throws Exception {
return initPlan(ctx, executor, true);
}
@@ -260,14 +265,15 @@ public class InsertIntoTableCommand extends Command
implements NeedAuditEncrypti
ctx.getStatementContext().setIsInsert(true);
while (++retryTimes <
Math.max(ctx.getSessionVariable().dmlPlanRetryTimes, 3)) {
TableIf targetTableIf = getTargetTableIf(ctx,
qualifiedTargetTableName);
+ DatabaseIf<?> targetDatabase = getTargetDatabase(targetTableIf);
// check auth
if (needAuthCheck(targetTableIf) &&
!Env.getCurrentEnv().getAccessManager()
- .checkTblPriv(ConnectContext.get(),
targetTableIf.getDatabase().getCatalog().getName(),
- targetTableIf.getDatabase().getFullName(),
targetTableIf.getName(),
+ .checkTblPriv(ConnectContext.get(),
targetDatabase.getCatalog().getName(),
+ targetDatabase.getFullName(),
targetTableIf.getName(),
PrivPredicate.LOAD)) {
ErrorReport.reportAnalysisException(ErrorCode.ERR_TABLEACCESS_DENIED_ERROR,
"LOAD",
ConnectContext.get().getQualifiedUser(),
ConnectContext.get().getRemoteIP(),
- targetTableIf.getDatabase().getFullName()
+ targetDatabase.getFullName()
+ "." +
Util.getTempTableDisplayName(targetTableIf.getName()));
}
BuildInsertExecutorResult buildResult;
@@ -559,10 +565,11 @@ public class InsertIntoTableCommand extends Command
implements NeedAuditEncrypti
} else if (physicalSink instanceof PhysicalDictionarySink) {
boolean emptyInsert = childIsEmptyRelation(physicalSink);
Dictionary dictionary = (Dictionary) targetTableIf;
+ DatabaseIf<?> database = getTargetDatabase(dictionary);
// insertCtx is not useful for dictionary. so keep it empty is
ok.
return ExecutorFactory.from(planner, dataSink, physicalSink,
() -> new DictionaryInsertExecutor(
- ctx, dictionary, label, planner, insertCtx,
emptyInsert, jobId));
+ ctx, database, dictionary, label, planner,
insertCtx, emptyInsert, jobId));
} else if (physicalSink instanceof PhysicalBlackholeSink) {
boolean emptyInsert = childIsEmptyRelation(physicalSink);
// insertCtx is not useful for blackhole. so keep it empty is
ok.
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/dictionary/DictionaryManagerTest.java
b/fe/fe-core/src/test/java/org/apache/doris/dictionary/DictionaryManagerTest.java
new file mode 100644
index 00000000000..4fdb9a4ee38
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/dictionary/DictionaryManagerTest.java
@@ -0,0 +1,112 @@
+// 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.doris.dictionary;
+
+import org.apache.doris.persist.CreateDictionaryPersistInfo;
+import org.apache.doris.persist.DictionaryDecreaseVersionInfo;
+import org.apache.doris.persist.DictionaryIncreaseVersionInfo;
+import org.apache.doris.persist.DropDictionaryPersistInfo;
+import org.apache.doris.persist.gson.GsonUtils;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+/**
+ * Tests for dictionary version journal replay robustness.
+ *
+ * The crash in production: an async data load task writes the INC journal,
then DROP removes the
+ * dictionary, then the failed commit writes a DEC journal for the
already-dropped dictionary.
+ * Followers crash at replay because the dictionary cannot be found by name
anymore.
+ * Replay must be idempotent and lookup by dictionary id.
+ */
+public class DictionaryManagerTest {
+
+ private DictionaryManager createManager() {
+ return new DictionaryManager();
+ }
+
+ private Dictionary buildDictionary(long id, String dbName, String
dictName, long version) {
+ String json = String.format(
+
"{\"clazz\":\"Dictionary\",\"id\":%d,\"name\":\"%s\",\"dbName\":\"%s\","
+ + "\"sourceTableName\":\"src_%s\",\"version\":%d}",
+ id, dictName, dbName, dbName, version);
+ return GsonUtils.GSON.fromJson(json, Dictionary.class);
+ }
+
+ @Test
+ public void testReplayDecreaseVersionMissingDictionary() throws Exception {
+ DictionaryManager manager = createManager();
+ // dictionary never created on this FE
+ Dictionary dict = buildDictionary(1001, "db1", "dic1", 2);
+ manager.replayDecreaseVersion(new DictionaryDecreaseVersionInfo(dict));
+ }
+
+ @Test
+ public void testReplayIncreaseVersionMissingDictionary() throws Exception {
+ DictionaryManager manager = createManager();
+ Dictionary dict = buildDictionary(1001, "db1", "dic1", 1);
+ manager.replayIncreaseVersion(new DictionaryIncreaseVersionInfo(dict));
+ }
+
+ @Test
+ public void testReplayDecreaseVersionAfterDrop() throws Exception {
+ DictionaryManager manager = createManager();
+ Dictionary dict = buildDictionary(1001, "db1", "dic1", 2);
+ manager.replayCreateDictionary(new CreateDictionaryPersistInfo(dict));
+ manager.replayDropDictionary(new DropDictionaryPersistInfo("db1",
"dic1"));
+
+ // journal order CREATE -> INC -> DROP -> DEC, DEC must be a no-op,
not an exception
+ manager.replayDecreaseVersion(new DictionaryDecreaseVersionInfo(dict));
+ Assert.assertNull(manager.getDictionary(1001));
+ }
+
+ @Test
+ public void testReplayDecreaseVersionAbA() throws Exception {
+ DictionaryManager manager = createManager();
+ Dictionary oldDict = buildDictionary(1001, "db1", "dic1", 2);
+ manager.replayCreateDictionary(new
CreateDictionaryPersistInfo(oldDict));
+ manager.replayDropDictionary(new DropDictionaryPersistInfo("db1",
"dic1"));
+ Dictionary newDict = buildDictionary(1002, "db1", "dic1", 1);
+ manager.replayCreateDictionary(new
CreateDictionaryPersistInfo(newDict));
+
+ // DEC of the dropped dictionary must not affect the recreated
same-name dictionary
+ manager.replayDecreaseVersion(new
DictionaryDecreaseVersionInfo(oldDict));
+ Assert.assertEquals(1, newDict.getVersion());
+ Assert.assertEquals(1, manager.getDictionary(1002).getVersion());
+ }
+
+ @Test
+ public void testReplayDecreaseVersionNormal() throws Exception {
+ DictionaryManager manager = createManager();
+ Dictionary dict = buildDictionary(1001, "db1", "dic1", 2);
+ manager.replayCreateDictionary(new CreateDictionaryPersistInfo(dict));
+
+ manager.replayDecreaseVersion(new DictionaryDecreaseVersionInfo(dict));
+ Assert.assertEquals(1, manager.getDictionary(1001).getVersion());
+ }
+
+ @Test
+ public void testReplayIncreaseVersionNormal() throws Exception {
+ DictionaryManager manager = createManager();
+ Dictionary dict = buildDictionary(1001, "db1", "dic1", 1);
+ manager.replayCreateDictionary(new CreateDictionaryPersistInfo(dict));
+
+ manager.replayIncreaseVersion(new DictionaryIncreaseVersionInfo(dict));
+ Assert.assertEquals(2, manager.getDictionary(1001).getVersion());
+ }
+}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/insert/DictionaryInsertTargetDropRaceTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/insert/DictionaryInsertTargetDropRaceTest.java
new file mode 100644
index 00000000000..df31a7b4f4e
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/insert/DictionaryInsertTargetDropRaceTest.java
@@ -0,0 +1,188 @@
+// 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.doris.nereids.trees.plans.commands.insert;
+
+import org.apache.doris.analysis.UserIdentity;
+import org.apache.doris.catalog.Database;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.TableIf;
+import org.apache.doris.common.Config;
+import org.apache.doris.common.util.DebugPointUtil;
+import org.apache.doris.common.util.DebugPointUtil.DebugPoint;
+import org.apache.doris.dictionary.Dictionary;
+import org.apache.doris.nereids.StatementContext;
+import org.apache.doris.nereids.parser.NereidsParser;
+import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.qe.OriginStatement;
+import org.apache.doris.qe.StmtExecutor;
+import org.apache.doris.thrift.TUniqueId;
+import org.apache.doris.utframe.TestWithFeService;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.BooleanSupplier;
+
+class DictionaryInsertTargetDropRaceTest extends TestWithFeService {
+ private static final String BLOCK_BEFORE_PLAN =
"DictionaryManager.dataLoad.blockBeforePlan";
+
+ private boolean debugPointsEnabled;
+
+ @BeforeEach
+ void saveDebugPointConfig() {
+ debugPointsEnabled = Config.enable_debug_points;
+ }
+
+ @AfterEach
+ void clearDebugPoints() {
+ DebugPointUtil.clearDebugPoints();
+ Config.enable_debug_points = debugPointsEnabled;
+ }
+
+ @Test
+ void rejectsDroppedDictionaryAfterResolvingItsOwner() throws Exception {
+ Config.enable_debug_points = true;
+ DebugPoint blockPoint = new DebugPoint();
+ blockPoint.executeLimit = Integer.MAX_VALUE;
+ DebugPointUtil.addDebugPoint(BLOCK_BEFORE_PLAN, blockPoint);
+
+ String sourceDbName = "dictionary_insert_drop_race_source";
+ createDatabaseAndUse(sourceDbName);
+ createSourceTable();
+
+ String dbName = "dictionary_insert_drop_race";
+ createDatabaseAndUse(dbName);
+ executeNereidsSql("CREATE DICTIONARY dic1 USING internal." +
sourceDbName
+ + ".source_table (city KEY, id VALUE) "
+ + "LAYOUT(HASH_MAP) PROPERTIES ('data_lifetime' = '600')");
+
+ Dictionary dictionary =
Env.getCurrentEnv().getDictionaryManager().getDictionary(dbName, "dic1");
+ Database database =
Env.getCurrentInternalCatalog().getDbOrDdlException(dbName);
+ String sql = "INSERT INTO " + dbName + ".dic1 SELECT * FROM "
+ + dictionary.getSourceCtlName() + "." +
dictionary.getSourceDbName() + "."
+ + dictionary.getSourceTableName();
+ InsertIntoTableCommand baseCommand = (InsertIntoTableCommand) new
NereidsParser().parseSingle(sql);
+ CountDownLatch targetValidated = new CountDownLatch(1);
+ CountDownLatch resumePlanning = new CountDownLatch(1);
+ AtomicBoolean blockOnce = new AtomicBoolean(true);
+ InsertIntoDictionaryCommand command = new InsertIntoDictionaryCommand(
+ baseCommand, database, dictionary, false) {
+ @Override
+ protected TableIf getTargetTableIf(ConnectContext ctx,
List<String> qualifiedTargetTableName) {
+ TableIf target = super.getTargetTableIf(ctx,
qualifiedTargetTableName);
+ if (blockOnce.compareAndSet(true, false)) {
+ targetValidated.countDown();
+ await(resumePlanning);
+ }
+ return target;
+ }
+ };
+
+ ExecutorService executorService = Executors.newSingleThreadExecutor();
+ try {
+ await(() -> blockPoint.executeNum.get() > 0);
+ Future<Throwable> result = executorService.submit(() ->
runInitPlan(command, sql, dbName));
+ Assertions.assertTrue(targetValidated.await(10, TimeUnit.SECONDS));
+ Env.getCurrentInternalCatalog().dropDb(dbName, false, true);
+ resumePlanning.countDown();
+
+ Throwable failure = result.get(10, TimeUnit.SECONDS);
+ Assertions.assertNotNull(failure);
+ Assertions.assertTrue(hasCause(failure,
org.apache.doris.nereids.exceptions.AnalysisException.class),
+ failure.toString());
+ Assertions.assertTrue(failure.toString().contains("Dictionary dic1
has been dropped"),
+ failure.toString());
+ Assertions.assertFalse(hasCause(failure,
NullPointerException.class), failure.toString());
+
Assertions.assertNull(Env.getCurrentEnv().getDictionaryManager().getDictionary(dictionary.getId()));
+
+ createDatabaseAndUse(dbName);
+ executeNereidsSql("CREATE DICTIONARY dic1 USING internal." +
sourceDbName
+ + ".source_table (city KEY, id VALUE) "
+ + "LAYOUT(HASH_MAP) PROPERTIES ('data_lifetime' = '600')");
+ Dictionary replacement =
Env.getCurrentEnv().getDictionaryManager().getDictionary(dbName, "dic1");
+ Assertions.assertNotEquals(dictionary.getId(),
replacement.getId());
+ } finally {
+ resumePlanning.countDown();
+ DebugPointUtil.removeDebugPoint(BLOCK_BEFORE_PLAN);
+ executorService.shutdownNow();
+ }
+
+ await(() -> !dictionary.getLastUpdateResult().isEmpty());
+ Assertions.assertTrue(dictionary.getLastUpdateResult().contains("has
been dropped"),
+ dictionary.getLastUpdateResult());
+
Assertions.assertFalse(dictionary.getLastUpdateResult().contains("Cannot
invoke"),
+ dictionary.getLastUpdateResult());
+ Env.getCurrentInternalCatalog().dropDb(dbName, false, true);
+ Env.getCurrentInternalCatalog().dropDb(sourceDbName, false, true);
+ }
+
+ private void createSourceTable() throws Exception {
+ createTable("CREATE TABLE source_table (id INT NOT NULL, city
VARCHAR(32) NOT NULL) "
+ + "DISTRIBUTED BY HASH(id) BUCKETS 1 PROPERTIES
('replication_num' = '1')");
+ }
+
+ private Throwable runInitPlan(InsertIntoDictionaryCommand command, String
sql, String dbName) {
+ try {
+ ConnectContext ctx = createCtx(UserIdentity.ROOT, "127.0.0.1");
+ ctx.setDatabase(dbName);
+ ctx.setQueryId(new TUniqueId(1, 2));
+ ctx.setStatementContext(new StatementContext(ctx, new
OriginStatement(sql, 0)));
+ command.initPlan(ctx, new StmtExecutor(ctx, sql));
+ return null;
+ } catch (Throwable t) {
+ return t;
+ }
+ }
+
+ private static boolean hasCause(Throwable throwable, Class<? extends
Throwable> causeClass) {
+ for (Throwable cause = throwable; cause != null; cause =
cause.getCause()) {
+ if (causeClass.isInstance(cause)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private static void await(CountDownLatch latch) {
+ try {
+ if (!latch.await(10, TimeUnit.SECONDS)) {
+ throw new IllegalStateException("Timed out waiting to resume
dictionary planning");
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IllegalStateException(e);
+ }
+ }
+
+ private static void await(BooleanSupplier condition) throws
InterruptedException {
+ long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10);
+ while (!condition.getAsBoolean() && System.nanoTime() < deadline) {
+ Thread.sleep(10);
+ }
+ Assertions.assertTrue(condition.getAsBoolean());
+ }
+}
diff --git a/regression-test/suites/dictionary_p0/test_create_drop_sync.groovy
b/regression-test/suites/dictionary_p0/test_create_drop_sync.groovy
index 951e2bef382..2bd885ec927 100644
--- a/regression-test/suites/dictionary_p0/test_create_drop_sync.groovy
+++ b/regression-test/suites/dictionary_p0/test_create_drop_sync.groovy
@@ -15,7 +15,7 @@
// specific language governing permissions and limitations
// under the License.
-suite('test_create_drop_sync') {
+suite('test_create_drop_sync', 'nonConcurrent') {
sql "DROP DATABASE IF EXISTS test_create_drop_sync"
sql "CREATE DATABASE test_create_drop_sync"
sql "USE test_create_drop_sync"
@@ -37,6 +37,7 @@ suite('test_create_drop_sync') {
DISTRIBUTED BY HASH(id) BUCKETS 1
PROPERTIES("replication_num" = "1")
"""
+ sql "INSERT INTO source_table VALUES (1, 'beijing', '001')"
// create dictionary
sql """
@@ -81,8 +82,35 @@ suite('test_create_drop_sync') {
properties('data_lifetime'='600');
"""
- // drop and recreate the database. check dic1 is dropped.
- sql "DROP DATABASE test_create_drop_sync"
+ waitAllDictionariesReady()
+
+ def refreshFuture
+ try {
+
GetDebugPoint().enableDebugPointForAllFEs('DictionaryManager.dataLoad.blockBeforePlan')
+ refreshFuture = thread {
+ sql "REFRESH DICTIONARY test_create_drop_sync.dic1"
+ }
+ awaitUntil(10) {
+ def dictionaries = sql "SHOW DICTIONARIES"
+ dictionaries.size() == 1 && dictionaries[0][4] == "LOADING"
+ }
+ sql "DROP DATABASE test_create_drop_sync"
+ } finally {
+
GetDebugPoint().disableDebugPointForAllFEs('DictionaryManager.dataLoad.blockBeforePlan')
+ }
+
+ Exception refreshFailure = null
+ assertNotNull(refreshFuture)
+ try {
+ refreshFuture.get()
+ } catch (Exception e) {
+ refreshFailure = e
+ }
+ assertNotNull(refreshFailure)
+ assertTrue(refreshFailure.toString().contains("Dictionary dic1 has been
dropped"), refreshFailure.toString())
+ assertFalse(refreshFailure.toString().contains("Cannot invoke"),
refreshFailure.toString())
+
+ // Recreate the database and verify that the dropped dictionary is not
retained.
sql "CREATE DATABASE test_create_drop_sync"
sql "USE test_create_drop_sync"
dict_res = sql "SHOW DICTIONARIES"
@@ -105,4 +133,4 @@ suite('test_create_drop_sync') {
LAYOUT(HASH_MAP)
properties('data_lifetime'='600');
"""
-}
\ No newline at end of file
+}
diff --git
a/regression-test/suites/dictionary_p0/test_dictionary_drop_while_load_commit_fail.groovy
b/regression-test/suites/dictionary_p0/test_dictionary_drop_while_load_commit_fail.groovy
new file mode 100644
index 00000000000..77e117a4866
--- /dev/null
+++
b/regression-test/suites/dictionary_p0/test_dictionary_drop_while_load_commit_fail.groovy
@@ -0,0 +1,113 @@
+// 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.
+
+import org.apache.doris.regression.suite.ClusterOptions
+
+// Regress the race reported in DORIS-27820: an async dictionary load task
writes the INC version
+// journal, then DROP deletes the dictionary, then the failed BE commit writes
a DEC journal for
+// the already dropped dictionary, making all followers exit when replaying it.
+//
+// With the fix, the DEC journal is skipped when the dictionary was dropped
during commit, and
+// replay of DEC journals of dropped dictionaries is a no-op, so the whole
journal stream
+// (CREATE -> INC -> DROP) replays cleanly and FEs stay healthy after restart.
+suite('test_dictionary_drop_while_load_commit_fail', 'docker') {
+ def options = new ClusterOptions()
+ options.cloudMode = false
+ options.feNum = 3
+ options.beNum = 1
+ options.enableDebugPoints()
+
+ docker(options) {
+ sql "drop database if exists test_dictionary_drop_race"
+ sql "create database test_dictionary_drop_race"
+ sql "use test_dictionary_drop_race"
+
+ sql """
+ create table source_table(
+ k1 varchar(100) not null,
+ v1 int not null
+ )
+ DISTRIBUTED BY HASH(`k1`) BUCKETS 1
+ properties("replication_num" = "1");
+ """
+ sql "insert into source_table values ('k1', 1), ('k2', 2), ('k3', 3)"
+
+ // Block the load task right after the INC journal is written, before
BE commit.
+
GetDebugPoint().enableDebugPointForAllFEs("DictionaryManager.afterIncJournal")
+ try {
+ sql """
+ create dictionary dict1 using source_table
+ (
+ k1 KEY,
+ v1 VALUE
+ )LAYOUT(HASH_MAP)
+ properties('data_lifetime'='600');
+ """
+
+ // wait until the load task is parked at the block point.
+ // status is LOADING before the block, and the task cannot pass
the block, so once we
+ // observe LOADING for a grace period, INC journal is guaranteed
already written.
+ boolean loading = false
+ for (int i = 0; i < 40; i++) {
+ def res = sql "SHOW DICTIONARIES"
+ if (res.size() == 1 && res[0][4] == "LOADING") {
+ loading = true
+ break
+ }
+ sleep(500)
+ }
+ assertTrue(loading)
+ sleep(1500)
+
+ // DROP the dictionary while the load task is between INC journal
and commit
+ sql "drop dictionary dict1"
+ def dictRes = sql "SHOW DICTIONARIES"
+ assertEquals(dictRes.size(), 0)
+
+ // force the BE commit to fail, then release the blocked load task
+
GetDebugPoint().enableDebugPointForAllFEs("DictionaryManager.commitNowVersion.fail")
+
GetDebugPoint().disableDebugPointForAllFEs("DictionaryManager.afterIncJournal")
+ sleep(3000)
+
+ // master must stay healthy: no DEC journal should have been
written after DROP
+ assertTrue(cluster.getMasterFe().alive)
+
+ // restart the master: it must replay the whole journal stream
without exit.
+ // if a DEC journal for the dropped dictionary had been written,
replay would throw
+ // and the FE would never come back alive.
+ def master = cluster.getMasterFe()
+ cluster.restartFrontends(master.index)
+ boolean hasRestart = false
+ for (int i = 0; i < 60; i++) {
+ if (cluster.getFeByIndex(master.index).alive) {
+ hasRestart = true
+ break
+ }
+ sleep(1000)
+ }
+ assertTrue(hasRestart)
+
+ context.reconnectFe()
+ sql "use test_dictionary_drop_race"
+ def finalRes = sql "SHOW DICTIONARIES"
+ assertEquals(finalRes.size(), 0)
+ } finally {
+
GetDebugPoint().disableDebugPointForAllFEs("DictionaryManager.commitNowVersion.fail")
+
GetDebugPoint().disableDebugPointForAllFEs("DictionaryManager.afterIncJournal")
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
