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 7ebc701c32a1b9..569aae8e43fae2 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.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 void dropDbDictionaries(String dbName) { // Log the drop operation if (dbDictIds != null) { for (Map.Entry entry : dbDictIds.entrySet()) { + idToDictionary.remove(entry.getValue()); Env.getCurrentEnv().getEditLog().logDropDictionary(dbName, entry.getKey()); } // also drop all name mapping records. @@ -282,6 +285,19 @@ private boolean hasDictionaryWithoutLock(String dbName, String dictName) { return dbDictIds != null && dbDictIds.containsKey(dictName); } + public boolean isCurrentDictionary(Dictionary dictionary) { + lockRead(); + try { + return isCurrentDictionaryWithoutLock(dictionary); + } finally { + unlockRead(); + } + } + + private boolean isCurrentDictionaryWithoutLock(Dictionary dictionary) { + return idToDictionary.get(dictionary.getId()) == dictionary; + } + public Map getDictionaries(String dbName) { lockRead(); try { @@ -412,6 +428,16 @@ public void dataLoad(ConnectContext ctx, Dictionary dictionary, boolean adaptive + dictionary.getStatus().name()); } + Database database = Env.getCurrentInternalCatalog().getDbNullable(dictionary.getDbName()); + if (database == null || !isCurrentDictionary(dictionary)) { + dictionary.trySetStatus(oldStatus); + throw new AnalysisException("Dictionary " + dictionary.getName() + " has been dropped"); + } + + 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 +459,8 @@ public void dataLoad(ConnectContext ctx, Dictionary dictionary, boolean adaptive 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 { @@ -442,6 +469,7 @@ public void dataLoad(ConnectContext ctx, Dictionary dictionary, boolean adaptive "OLAP_SCAN_PARTITION_PRUNE,PRUNE_EMPTY_PARTITION"); command.run(ctx, executor); } catch (Exception e) { + abortStagedVersion(ctx, dictionary); // wait next shedule. dictionary.trySetStatus(oldStatus); dictionary.setLastUpdateResult(e.getMessage()); @@ -449,23 +477,24 @@ public void dataLoad(ConnectContext ctx, Dictionary dictionary, boolean adaptive } // some insert failed won't throw but only set error status. if (ctx.getState().getErrorCode() != null && ctx.getState().getErrorMessage() != null) { + String errorMessage = ctx.getState().getErrorMessage(); + abortStagedVersion(ctx, dictionary); dictionary.trySetStatus(oldStatus); - dictionary.setLastUpdateResult(ctx.getState().getErrorMessage()); + dictionary.setLastUpdateResult(errorMessage); // for must failed refresh, we can skip it at next time. this mark is tricky but we have to do now. - if (ctx.getState().getErrorMessage().contains("[INVALID_DICT_MARK]")) { + if (errorMessage.contains("[INVALID_DICT_MARK]")) { LOG.warn("Dictionary {} load failed with src version {}, mark it invalid", dictionary.getName(), ctx.getStatementContext().getDictionaryUsedSrcVersion()); dictionary.updateLatestInvalidVersion(ctx.getStatementContext().getDictionaryUsedSrcVersion()); } - throw new RuntimeException(ctx.getState().getErrorMessage()); + throw new RuntimeException(errorMessage); } // because of deleting does NOT conflict with loading, we should check dictionary's existance again! lockRead(); boolean unlocked = false; try { - if (!dictionaryIds.containsKey(dictionary.getDbName()) - || !dictionaryIds.get(dictionary.getDbName()).containsKey(dictionary.getName())) { + if (!isCurrentDictionaryWithoutLock(dictionary)) { unlockRead(); unlocked = true; @@ -495,16 +524,32 @@ public void dataLoad(ConnectContext ctx, Dictionary dictionary, boolean adaptive } } + while (DebugPointUtil.isEnable("DictionaryManager.dataLoad.blockBeforeCommit")) { + Thread.sleep(10); + } + // 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); + long stagedVersion = dictionary.getVersion(); + CommitResult commitResult = commitNowVersion(ctx, dictionary); + if (!commitResult.allSucceeded) { + // Keep the identity check and rollback journal ordered with DROP. + lockRead(); + try { + if (isCurrentDictionaryWithoutLock(dictionary)) { + if (!commitResult.mayHaveCommitted + && !ctx.getStatementContext().isPartialLoadDictionary()) { + dictionary.decreaseVersion(); + Env.getCurrentEnv().getEditLog().logDictionaryDecVersion(dictionary); + } + dictionary.trySetStatus(commitResult.mayHaveCommitted + ? DictionaryStatus.OUT_OF_DATE : oldStatus); + } + } finally { + unlockRead(); } - dictionary.trySetStatus(oldStatus); - abortSpecificVersion(ctx, dictionary, dictionary.getVersion() + 1); + abortSpecificVersion(ctx, dictionary, stagedVersion); throw new RuntimeException("Dictionary " + dictionary.getName() + " commit version " - + (dictionary.getVersion() + 1) + " failed"); + + stagedVersion + " failed"); } // commit succeed. update metadata. @@ -524,12 +569,27 @@ public void dataLoad(ConnectContext ctx, Dictionary dictionary, boolean adaptive dictionary.getVersion(), ctx.getStatementContext().getDictionaryUsedSrcVersion()); } - private boolean commitNowVersion(ConnectContext ctx, Dictionary dictionary) { + private static class CommitResult { + private final boolean allSucceeded; + private final boolean mayHaveCommitted; + + private CommitResult(boolean allSucceeded, boolean mayHaveCommitted) { + this.allSucceeded = allSucceeded; + this.mayHaveCommitted = mayHaveCommitted; + } + } + + private CommitResult commitNowVersion(ConnectContext ctx, Dictionary dictionary) { + if (DebugPointUtil.isEnable("DictionaryManager.commitNowVersion.forceFailure")) { + return new CommitResult(false, false); + } + // use the same BEs when we get before start loading. List beList = ctx.getStatementContext().getUsedBackendsDistributing(); List> futureList = new ArrayList<>(); boolean allSucceed = true; + boolean mayHaveCommitted = false; try { for (Backend be : beList) { if (!be.isAlive()) { @@ -557,6 +617,8 @@ private boolean commitNowVersion(ConnectContext ctx, Dictionary dictionary) { LOG.warn("Failed to commit dictionary " + dictionary.getId() + " on be " + be.getAddress() + " because " + status.getErrorMsg()); allSucceed = false; + } else { + mayHaveCommitted = true; } } else { LOG.warn("Failed to commit dictionary " + dictionary.getId() + " on be " + be.getAddress()); @@ -567,13 +629,29 @@ private boolean commitNowVersion(ConnectContext ctx, Dictionary dictionary) { dictionary.setLastUpdateResult("commit failed: " + e.getMessage()); LOG.warn("Failed to commit dictionary " + dictionary.getId(), e); allSucceed = false; + // A timed-out RPC can have committed remotely even when FE did not receive the response. + mayHaveCommitted = mayHaveCommitted || !futureList.isEmpty(); } - return allSucceed; + return new CommitResult(allSucceed, mayHaveCommitted); + } + + private void abortStagedVersion(ConnectContext ctx, Dictionary dictionary) { + if (ctx.getStatementContext() == null) { + return; + } + long stagedVersion = ctx.getStatementContext().isPartialLoadDictionary() + ? dictionary.getVersion() : dictionary.getVersion() + 1; + abortSpecificVersion(ctx, dictionary, stagedVersion); } // abort could to all BE. swallow any failures. private void abortSpecificVersion(ConnectContext ctx, Dictionary dictionary, long versionId) { // use the same BEs when we get before start loading. + if (ctx.getStatementContext() == null + || ctx.getStatementContext().getUsedBackendsDistributing() == null + || ctx.getStatementContext().getUsedBackendsDistributing().isEmpty()) { + return; + } List beList = ctx.getStatementContext().getUsedBackendsDistributing(); List> futureList = new ArrayList<>(); 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 5d6cc853b10866..399b4b691f25ce 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 @@ public class UnboundDictionarySink extends UnboundLogicalSink 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 UnboundDictionarySink(Dictionary dictionary, CHILD_TYPE child, boolean ad dictionary.getColumnNames(), // colNames from dictionary DMLCommandType.INSERT, // dmlCommandType child); + this.database = Objects.requireNonNull(database, "database should not be null"); this.dictionary = dictionary; this.allowAdaptiveLoad = adaptiveLoad; } + public Database getDatabase() { + return database; + } + public Dictionary getDictionary() { return dictionary; } @@ -76,7 +83,7 @@ public boolean allowAdaptiveLoad() { @Override public Plan withChildren(List 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 R accept(PlanVisitor visitor, C context) { @Override public Plan withGroupExpression(Optional groupExpression) { - return new UnboundDictionarySink<>(dictionary, child(), allowAdaptiveLoad); + return new UnboundDictionarySink<>(database, dictionary, child(), allowAdaptiveLoad); } @Override public Plan withGroupExprLogicalPropChildren(Optional groupExpression, Optional logicalProperties, List 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 ff0cfc71264a12..499519ad95eb7d 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; @@ -161,8 +162,8 @@ public static LogicalSink createUnboundTableSinkMaybeOverwrite(L /** * create unbound sink for dictionary sink */ - public static UnboundDictionarySink createUnboundDictionarySink(Dictionary dictionary, - LogicalPlan child, boolean adaptiveLoad) { - return new UnboundDictionarySink<>(dictionary, child, adaptiveLoad); + public static UnboundDictionarySink 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 8c580171b21151..f270957e522023 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 @@ -990,9 +990,8 @@ private static Map getConnectorColumnToOutput( private Plan bindDictionarySink(MatchingContext> ctx) { UnboundDictionarySink sink = ctx.root; - Pair 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 @@ -1118,19 +1117,6 @@ private Pair bind(CascadesContext c throw new AnalysisException("the target table of insert into is not a plugin-driven connector table"); } - private Pair bind(CascadesContext cascadesContext, - UnboundDictionarySink 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 bindPartitionIds(OlapTable table, List 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 add2ab967898f8..9beba9223fb4da 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 @@ -47,6 +47,7 @@ import org.apache.logging.log4j.Logger; import java.util.List; +import java.util.Objects; import java.util.Optional; import java.util.concurrent.CopyOnWriteArrayList; @@ -107,8 +108,18 @@ public AbstractInsertExecutor(ConnectContext ctx, TableIf table, String labelNam */ public AbstractInsertExecutor(ConnectContext ctx, TableIf table, String labelName, NereidsPlanner planner, Optional insertCtx, boolean emptyInsert, long jobId, boolean needRegister) { + this(ctx, table.getDatabase(), table, labelName, planner, insertCtx, emptyInsert, jobId, needRegister); + } + + /** + * Constructor for targets whose owning database cannot be recovered from the table during concurrent metadata + * changes. + */ + public AbstractInsertExecutor(ConnectContext ctx, DatabaseIf database, TableIf table, String labelName, + NereidsPlanner planner, Optional 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 d29a3350ba34ad..f045a4c19b4166 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 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 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 45615f62c5b17d..bbae0063ab3c1e 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,7 +18,7 @@ package org.apache.doris.nereids.trees.plans.commands.insert; import org.apache.doris.analysis.RedirectStatus; -import org.apache.doris.catalog.TableIf; +import org.apache.doris.catalog.Database; import org.apache.doris.dictionary.Dictionary; import org.apache.doris.nereids.analyzer.UnboundDictionarySink; import org.apache.doris.nereids.analyzer.UnboundTableSink; @@ -35,6 +35,7 @@ * logic of InsertIntoTableCommand to maximize code reuse. */ public class InsertIntoDictionaryCommand extends InsertIntoTableCommand { + private final Database database; private final Dictionary dictionary; /** @@ -45,9 +46,10 @@ public class InsertIntoDictionaryCommand extends InsertIntoTableCommand { * @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 +59,7 @@ public InsertIntoDictionaryCommand(InsertIntoTableCommand baseCommand, Dictionar } UnboundTableSink sink = (UnboundTableSink) logicalQuery; - UnboundDictionarySink newSink = UnboundTableSinkCreator.createUnboundDictionarySink(dictionary, + UnboundDictionarySink newSink = UnboundTableSinkCreator.createUnboundDictionarySink(database, dictionary, (LogicalPlan) sink.child(0), adaptiveLoad); setLogicalQuery(newSink); setOriginLogicalQuery(newSink); @@ -70,8 +72,11 @@ public RedirectStatus toRedirectStatus() { } @Override - protected TableIf getTargetTableIf(ConnectContext ctx, List qualifiedTargetTableName) { - return dictionary; + protected InsertTarget getTarget(ConnectContext ctx, List qualifiedTargetTableName) { + if (!ctx.getEnv().getDictionaryManager().isCurrentDictionary(dictionary)) { + throw new AnalysisException("Dictionary " + dictionary.getName() + " has been dropped"); + } + return new InsertTarget(database, dictionary); } public Dictionary getDictionary() { 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 1372cce62e2ab9..aef41ff48c605f 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 @@ -19,6 +19,7 @@ import org.apache.doris.analysis.RedirectStatus; import org.apache.doris.analysis.StmtType; +import org.apache.doris.catalog.DatabaseIf; import org.apache.doris.catalog.Env; import org.apache.doris.catalog.OlapTable; import org.apache.doris.catalog.TableIf; @@ -26,6 +27,7 @@ import org.apache.doris.common.Config; import org.apache.doris.common.ErrorCode; import org.apache.doris.common.ErrorReport; +import org.apache.doris.common.Pair; import org.apache.doris.common.UserException; import org.apache.doris.common.profile.ProfileManager.ProfileType; import org.apache.doris.common.util.DebugUtil; @@ -135,6 +137,25 @@ public class InsertIntoTableCommand extends Command implements NeedAuditEncrypti private InsertExecutorListener insertExecutorListener; + /** A database and table resolved by the same catalog lookup. */ + protected static class InsertTarget { + private final DatabaseIf database; + private final TableIf table; + + protected InsertTarget(DatabaseIf database, TableIf table) { + this.database = Objects.requireNonNull(database, "database should not be null"); + this.table = Objects.requireNonNull(table, "table should not be null"); + } + + public DatabaseIf getDatabase() { + return database; + } + + public TableIf getTable() { + return table; + } + } + public InsertIntoTableCommand(LogicalPlan logicalQuery, Optional labelName, Optional insertCtx, Optional cte) { this(PlanType.INSERT_INTO_TABLE_COMMAND, logicalQuery, labelName, insertCtx, cte, true, Optional.empty()); @@ -235,9 +256,11 @@ public void runWithUpdateInfo(ConnectContext ctx, StmtExecutor executor, } // may be overridden - protected TableIf getTargetTableIf( + protected InsertTarget getTarget( ConnectContext ctx, List qualifiedTargetTableName) { - return RelationUtil.getTable(qualifiedTargetTableName, ctx.getEnv(), Optional.empty()); + Pair, TableIf> target = RelationUtil.getDbAndTable( + qualifiedTargetTableName, ctx.getEnv(), Optional.empty()); + return new InsertTarget(target.first, target.second); } public AbstractInsertExecutor initPlan(ConnectContext ctx, StmtExecutor executor) throws Exception { @@ -256,77 +279,105 @@ public AbstractInsertExecutor initPlan(ConnectContext ctx, StmtExecutor stmtExec boolean needBeginTransaction) throws Exception { List qualifiedTargetTableName = InsertUtils.getTargetTableQualified(originLogicalQuery, ctx); - AbstractInsertExecutor insertExecutor; + AbstractInsertExecutor insertExecutor = null; int retryTimes = 0; ctx.getStatementContext().setIsInsert(true); while (++retryTimes < Math.max(ctx.getSessionVariable().dmlPlanRetryTimes, 3)) { - TableIf targetTableIf = getTargetTableIf(ctx, qualifiedTargetTableName); + boolean groupCommitBeforeAttempt = ctx.isGroupCommit(); + Backend groupCommitBackendBeforeAttempt = + ctx.getStatementContext().getGroupCommitMergeBackend(); + insertExecutor = null; + InsertTarget target = getTarget(ctx, qualifiedTargetTableName); + DatabaseIf targetDatabase = target.getDatabase(); + TableIf targetTableIf = target.getTable(); + // Retargeting a StatementContext would mix the new target with its cached relation metadata. + for (TableIf cachedTargetTable + : ctx.getStatementContext().getInsertTargetTables().values()) { + if (cachedTargetTable.getId() != targetTableIf.getId()) { + LOG.warn("insert target changed between planning attempts. query id is {}. table id changed " + + "from {} to {}", + DebugUtil.printId(ctx.queryId()), cachedTargetTable.getId(), targetTableIf.getId()); + throw new AnalysisException("Insert target changed while planning. Please retry the statement"); + } + } // 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; + ExecutorFactory executorFactory; try { // use originLogicalQuery to build logicalQuery again. - buildResult = initPlanOnce(ctx, stmtExecutor, targetTableIf); + executorFactory = initPlanOnce(ctx, stmtExecutor, targetTableIf); } catch (Throwable e) { Throwables.throwIfInstanceOf(e, RuntimeException.class); throw new IllegalStateException(e.getMessage(), e); } - insertExecutor = buildResult.executor; - parsedPlan = Optional.ofNullable(buildResult.planner.getParsedPlan()); - Plan analyzedPlan = buildResult.planner.getAnalyzedPlan(); + parsedPlan = Optional.ofNullable(executorFactory.planner.getParsedPlan()); + Plan analyzedPlan = executorFactory.planner.getAnalyzedPlan(); lineagePlan = Optional.ofNullable(analyzedPlan); - if (!needBeginTransaction) { - return insertExecutor; - } - List infos = StreamConsumptionInfoExtractor.extract(analyzedPlan); + List infos = needBeginTransaction + ? StreamConsumptionInfoExtractor.extract(analyzedPlan) : Lists.newArrayList(); if (!infos.isEmpty()) { if (!Config.enable_feature_binlog) { throw new AnalysisException("Insert plan with Table stream failed." + " should enable binlog feature in FE config."); } - // put offset into executor - insertExecutor.setStreamUpdateInfos(infos); - insertExecutor.registerListener(new InsertExecutorListener() { - @Override - public void beforeComplete(AbstractInsertExecutor insertExecutor, StmtExecutor executor, - long jobId) throws Exception { - TransactionState transactionState = Env.getCurrentGlobalTransactionMgr() - .getTransactionState(insertExecutor.getDatabase().getId(), - insertExecutor.getTxnId()); - transactionState.setStreamUpdateInfos(insertExecutor.getStreamUpdateInfos()); - } - }); } // lock after plan and check does table's schema changed to ensure we lock table order by id. - TableIf newestTargetTableIf = getTargetTableIf(ctx, qualifiedTargetTableName); + InsertTarget newestTarget = getTarget(ctx, qualifiedTargetTableName); + DatabaseIf newestTargetDatabase = newestTarget.getDatabase(); + TableIf newestTargetTableIf = newestTarget.getTable(); newestTargetTableIf.readLock(); try { - if (targetTableIf.getId() != newestTargetTableIf.getId()) { - LOG.warn("insert plan failed {} times. query id is {}. table id changed from {} to {}", - retryTimes, DebugUtil.printId(ctx.queryId()), - targetTableIf.getId(), newestTargetTableIf.getId()); - newestTargetTableIf.readUnlock(); - continue; + if (targetDatabase.getId() != newestTargetDatabase.getId() + || targetTableIf.getId() != newestTargetTableIf.getId()) { + LOG.warn("insert target changed while planning. query id is {}. target id changed from {}.{} " + + "to {}.{}", + DebugUtil.printId(ctx.queryId()), + targetDatabase.getId(), targetTableIf.getId(), + newestTargetDatabase.getId(), newestTargetTableIf.getId()); + ctx.setGroupCommit(groupCommitBeforeAttempt); + ctx.getStatementContext().setGroupCommitMergeBackend(groupCommitBackendBeforeAttempt); + throw new AnalysisException("Insert target changed while planning. Please retry the statement"); } // Use the schema saved during planning as the schema of the original target table. if (!ctx.getStatementContext().getInsertTargetSchema().equals(newestTargetTableIf.getFullSchema())) { LOG.warn("insert plan failed {} times. query id is {}. table schema changed from {} to {}", retryTimes, DebugUtil.printId(ctx.queryId()), ctx.getStatementContext().getInsertTargetSchema(), newestTargetTableIf.getFullSchema()); + ctx.setGroupCommit(groupCommitBeforeAttempt); + ctx.getStatementContext().setGroupCommitMergeBackend(groupCommitBackendBeforeAttempt); + ctx.getStatementContext().setPrivChecked(false); newestTargetTableIf.readUnlock(); continue; } - if (!insertExecutor.isEmptyInsert()) { + + BuildInsertExecutorResult buildResult = executorFactory.build(); + insertExecutor = buildResult.executor; + applyInsertPlanStatistic(buildResult.planner); + if (!infos.isEmpty()) { + // put offset into executor + insertExecutor.setStreamUpdateInfos(infos); + insertExecutor.registerListener(new InsertExecutorListener() { + @Override + public void beforeComplete(AbstractInsertExecutor insertExecutor, StmtExecutor executor, + long jobId) throws Exception { + TransactionState transactionState = Env.getCurrentGlobalTransactionMgr() + .getTransactionState(insertExecutor.getDatabase().getId(), + insertExecutor.getTxnId()); + transactionState.setStreamUpdateInfos(insertExecutor.getStreamUpdateInfos()); + } + }); + } + if (needBeginTransaction && !insertExecutor.isEmptyInsert()) { insertExecutor.beginTransaction(); insertExecutor.finalizeSink( buildResult.planner.getFragments().get(0), buildResult.dataSink, @@ -343,6 +394,9 @@ public void beforeComplete(AbstractInsertExecutor insertExecutor, StmtExecutor e Throwables.throwIfInstanceOf(e, RuntimeException.class); throw new IllegalStateException(e.getMessage(), e); } + if (!needBeginTransaction) { + return insertExecutor; + } stmtExecutor.setProfileType(ProfileType.LOAD); // We exposed @StmtExecutor#cancel as a unified entry point for statement interruption, // so we need to set this here @@ -364,7 +418,7 @@ protected boolean needAuthCheck(TableIf targetTableIf) { return true; } - private BuildInsertExecutorResult initPlanOnce(ConnectContext ctx, + private ExecutorFactory initPlanOnce(ConnectContext ctx, StmtExecutor stmtExecutor, TableIf targetTableIf) throws Throwable { targetTableIf.readLock(); try { @@ -577,10 +631,11 @@ ExecutorFactory selectInsertExecutorFactory( } else if (physicalSink instanceof PhysicalDictionarySink) { boolean emptyInsert = childIsEmptyRelation(physicalSink); Dictionary dictionary = (Dictionary) targetTableIf; + DatabaseIf database = ((PhysicalDictionarySink) physicalSink).getDatabase(); // 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. @@ -598,7 +653,7 @@ ExecutorFactory selectInsertExecutorFactory( } } - private BuildInsertExecutorResult planInsertExecutor( + private ExecutorFactory planInsertExecutor( ConnectContext ctx, StmtExecutor stmtExecutor, LogicalPlanAdapter logicalPlanAdapter, TableIf targetTableIf) throws Throwable { LogicalPlan logicalPlan = logicalPlanAdapter.getLogicalPlan(); @@ -638,15 +693,10 @@ protected void doDistribute(boolean canUseNereidsDistributePlanner, ExplainLevel planner.getPhysicalPlan().treeString()); } - // step 4 - BuildInsertExecutorResult build = executorFactoryRef.get().build(); - - // apply insert plan Statistic - applyInsertPlanStatistic(planner); - return build; + return executorFactoryRef.get(); } - private void applyInsertPlanStatistic(FastInsertIntoValuesPlanner planner) { + private void applyInsertPlanStatistic(NereidsPlanner planner) { LoadJob loadJob = Env.getCurrentEnv().getLoadManager().getLoadJob(getJobId()); if (loadJob == null) { return; @@ -695,14 +745,17 @@ public boolean isExternalTableSink() { * get the target table of the insert command */ public TableIf getTable(ConnectContext ctx) throws Exception { - TableIf targetTableIf = InsertUtils.getTargetTable(originLogicalQuery, ctx); + List qualifiedTargetTableName = InsertUtils.getTargetTableQualified(originLogicalQuery, ctx); + InsertTarget target = getTarget(ctx, qualifiedTargetTableName); + DatabaseIf targetDatabase = target.getDatabase(); + TableIf targetTableIf = target.getTable(); if (!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())); } return targetTableIf; diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/GroupCommitPlanner.java b/fe/fe-core/src/main/java/org/apache/doris/planner/GroupCommitPlanner.java index 1a9874056711d7..74a23a7590a385 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/planner/GroupCommitPlanner.java +++ b/fe/fe-core/src/main/java/org/apache/doris/planner/GroupCommitPlanner.java @@ -38,6 +38,7 @@ import org.apache.doris.nereids.load.NereidsStreamLoadTask; import org.apache.doris.nereids.trees.expressions.literal.Literal; import org.apache.doris.nereids.trees.plans.commands.PrepareCommand; +import org.apache.doris.nereids.trees.plans.commands.insert.AbstractInsertExecutor; import org.apache.doris.nereids.trees.plans.commands.insert.InsertIntoTableCommand; import org.apache.doris.proto.InternalService; import org.apache.doris.proto.InternalService.PGroupCommitInsertRequest; @@ -193,26 +194,34 @@ public static void executeGroupCommitInsert(ConnectContext ctx, PreparedStatemen StatementContext statementContext) throws Exception { PrepareCommand prepareCommand = preparedStmtCtx.command; InsertIntoTableCommand command = (InsertIntoTableCommand) (prepareCommand.getLogicalPlan()); - OlapTable table = (OlapTable) command.getTable(ctx); for (int retry = 0; retry < MAX_RETRY; retry++) { - if (Env.getCurrentEnv().getGroupCommitManager().isBlock(table.getId())) { - String msg = "insert table " + table.getId() + SCHEMA_CHANGE; - LOG.info(msg); - throw new DdlException(msg); - } boolean reuse = false; GroupCommitPlanner groupCommitPlanner; - if (preparedStmtCtx.groupCommitPlanner.isPresent() - && table.getId() == preparedStmtCtx.groupCommitPlanner.get().table.getId() - && table.getBaseSchemaVersion() == preparedStmtCtx.groupCommitPlanner.get().baseSchemaVersion) { - groupCommitPlanner = preparedStmtCtx.groupCommitPlanner.get(); - reuse = true; + if (preparedStmtCtx.groupCommitPlanner.isPresent()) { + OlapTable currentTable = (OlapTable) command.getTable(ctx); + GroupCommitPlanner cachedPlanner = preparedStmtCtx.groupCommitPlanner.get(); + if (currentTable.getId() == cachedPlanner.table.getId() + && currentTable.getBaseSchemaVersion() == cachedPlanner.baseSchemaVersion) { + checkGroupCommitBlocked(currentTable); + groupCommitPlanner = cachedPlanner; + reuse = true; + } else { + preparedStmtCtx.groupCommitPlanner = Optional.empty(); + groupCommitPlanner = null; + } } else { + groupCommitPlanner = null; + } + if (groupCommitPlanner == null) { // call nereids planner to check to sql - command.initPlan(ctx, new StmtExecutor(new ConnectContext(), ""), false); + AbstractInsertExecutor insertExecutor = command.initPlan( + ctx, new StmtExecutor(new ConnectContext(), ""), false); + Database database = (Database) insertExecutor.getDatabase(); + OlapTable table = (OlapTable) insertExecutor.getTable(); + checkGroupCommitBlocked(table); List targetColumnNames = command.getTargetColumns(); groupCommitPlanner = EnvFactory.getInstance() - .createGroupCommitPlanner((Database) table.getDatabase(), table, + .createGroupCommitPlanner(database, table, targetColumnNames, ctx.queryId(), ConnectContext.get().getSessionVariable().getGroupCommit()); // TODO use planner column size @@ -240,6 +249,14 @@ public static void executeGroupCommitInsert(ConnectContext ctx, PreparedStatemen } } + private static void checkGroupCommitBlocked(OlapTable table) throws DdlException { + if (Env.getCurrentEnv().getGroupCommitManager().isBlock(table.getId())) { + String msg = "insert table " + table.getId() + SCHEMA_CHANGE; + LOG.info(msg); + throw new DdlException(msg); + } + } + // return private Pair handleResponse(ConnectContext ctx, boolean canRetry, boolean reuse, PGroupCommitInsertResponse response) throws DdlException { diff --git a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertTargetDropRaceTest.java b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertTargetDropRaceTest.java new file mode 100644 index 00000000000000..51a747e10a6008 --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertTargetDropRaceTest.java @@ -0,0 +1,616 @@ +// 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.Env; +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.dictionary.Dictionary.DictionaryStatus; +import org.apache.doris.load.loadv2.JobState; +import org.apache.doris.nereids.StatementContext; +import org.apache.doris.nereids.parser.NereidsParser; +import org.apache.doris.nereids.trees.plans.PlanType; +import org.apache.doris.nereids.trees.plans.commands.refresh.RefreshDictionaryCommand; +import org.apache.doris.persist.CreateDictionaryPersistInfo; +import org.apache.doris.proto.InternalService; +import org.apache.doris.proto.Types; +import org.apache.doris.qe.ConnectContext; +import org.apache.doris.qe.OriginStatement; +import org.apache.doris.qe.StmtExecutor; +import org.apache.doris.rpc.BackendServiceProxy; +import org.apache.doris.system.Backend; +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 org.mockito.MockedConstruction; +import org.mockito.MockedStatic; +import org.mockito.Mockito; + +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; +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.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.BooleanSupplier; +import java.util.function.Consumer; + +class InsertTargetDropRaceTest extends TestWithFeService { + private static final String DICTIONARY_BLOCK_POINT = "DictionaryManager.dataLoad.blockBeforePlan"; + private static final String DICTIONARY_COMMIT_BLOCK_POINT = "DictionaryManager.dataLoad.blockBeforeCommit"; + private static final String DICTIONARY_COMMIT_FAILURE_POINT = + "DictionaryManager.commitNowVersion.forceFailure"; + private boolean debugPointsEnabled; + + @BeforeEach + void saveDebugPointConfig() { + debugPointsEnabled = Config.enable_debug_points; + } + + @AfterEach + void clearDebugPoints() { + DebugPointUtil.clearDebugPoints(); + Config.enable_debug_points = debugPointsEnabled; + } + + @Test + void normalInsertReturnsAnalysisErrorWhenDatabaseIsDroppedAfterTargetResolution() throws Exception { + String dbName = "normal_insert_drop_race"; + createDatabaseAndUse(dbName); + createTable("CREATE TABLE target_table (id INT) DISTRIBUTED BY HASH(id) BUCKETS 1 " + + "PROPERTIES ('replication_num' = '1')"); + + String sql = "INSERT INTO " + dbName + ".target_table VALUES (1)"; + InsertIntoTableCommand baseCommand = (InsertIntoTableCommand) new NereidsParser().parseSingle(sql); + CountDownLatch targetResolved = new CountDownLatch(1); + CountDownLatch resumeInsert = new CountDownLatch(1); + AtomicBoolean blockOnce = new AtomicBoolean(true); + InsertIntoTableCommand command = new InsertIntoTableCommand(baseCommand, + PlanType.INSERT_INTO_TABLE_COMMAND) { + @Override + protected InsertTarget getTarget(ConnectContext ctx, List qualifiedTargetTableName) { + InsertTarget target = super.getTarget(ctx, qualifiedTargetTableName); + if (blockOnce.compareAndSet(true, false)) { + targetResolved.countDown(); + await(resumeInsert); + } + return target; + } + }; + + ExecutorService executorService = Executors.newSingleThreadExecutor(); + try { + Future result = executorService.submit(() -> runInitPlan(command, sql, dbName)); + Assertions.assertTrue(targetResolved.await(10, TimeUnit.SECONDS)); + Env.getCurrentInternalCatalog().dropDb(dbName, false, true); + resumeInsert.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.assertFalse(hasCause(failure, NullPointerException.class), failure.toString()); + } finally { + resumeInsert.countDown(); + executorService.shutdownNow(); + } + } + + @Test + void dictionaryLoadStopsCleanlyWhenDatabaseIsDroppedBeforePlanning() throws Exception { + Config.enable_debug_points = true; + DebugPoint blockPoint = addBlockingDebugPoint(DICTIONARY_BLOCK_POINT); + + String dbName = "dictionary_insert_drop_race"; + createDatabaseAndUse(dbName); + createTable("CREATE TABLE source_table (id INT NOT NULL, city VARCHAR(32) NOT NULL) " + + "DISTRIBUTED BY HASH(id) BUCKETS 1 PROPERTIES ('replication_num' = '1')"); + executeNereidsSql("CREATE DICTIONARY dic1 USING source_table (city KEY, id VALUE) " + + "LAYOUT(HASH_MAP) PROPERTIES ('data_lifetime' = '600')"); + + Dictionary dictionary = Env.getCurrentEnv().getDictionaryManager().getDictionary(dbName, "dic1"); + try { + await(() -> blockPoint.executeNum.get() > 0); + Env.getCurrentInternalCatalog().dropDb(dbName, false, true); + + createDatabaseAndUse(dbName); + createTable("CREATE TABLE source_table (id INT NOT NULL, city VARCHAR(32) NOT NULL) " + + "DISTRIBUTED BY HASH(id) BUCKETS 1 PROPERTIES ('replication_num' = '1')"); + executeNereidsSql("CREATE DICTIONARY dic1 USING 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 { + DebugPointUtil.removeDebugPoint(DICTIONARY_BLOCK_POINT); + } + + 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); + } + + @Test + void droppedDictionaryDoesNotJournalVersionRollbackAfterCommitFailure() throws Exception { + Config.enable_debug_points = true; + String dbName = "dictionary_commit_drop_race"; + Dictionary dictionary = createDictionaryWithoutAutoLoad(dbName); + + DebugPoint blockPoint = addBlockingDebugPoint(DICTIONARY_COMMIT_BLOCK_POINT); + DebugPointUtil.addDebugPoint(DICTIONARY_COMMIT_FAILURE_POINT); + ExecutorService executorService = Executors.newSingleThreadExecutor(); + try (MockedConstruction ignored = Mockito.mockConstruction( + InsertIntoDictionaryCommand.class, (mock, context) -> Mockito.doAnswer(invocation -> { + ConnectContext loadContext = invocation.getArgument(0); + StatementContext statementContext = new StatementContext( + loadContext, new OriginStatement("", 0)); + statementContext.setUsedBackendsDistributing(List.of()); + loadContext.setStatementContext(statementContext); + return null; + }).when(mock).run(Mockito.any(), Mockito.any()))) { + Future dropResult = executorService.submit(() -> { + await(() -> blockPoint.executeNum.get() > 0); + Env.getCurrentInternalCatalog().dropDb(dbName, false, true); + DebugPointUtil.removeDebugPoint(DICTIONARY_COMMIT_BLOCK_POINT); + return null; + }); + + long originalVersion = dictionary.getVersion(); + Throwable failure = runRefresh(dbName); + dropResult.get(10, TimeUnit.SECONDS); + + Assertions.assertNotNull(failure); + Assertions.assertTrue(failure.toString().contains("commit version"), failure.toString()); + Assertions.assertEquals(originalVersion + 1, dictionary.getVersion()); + Assertions.assertFalse(Env.getCurrentEnv().getDictionaryManager().isCurrentDictionary(dictionary)); + } finally { + DebugPointUtil.removeDebugPoint(DICTIONARY_BLOCK_POINT); + DebugPointUtil.removeDebugPoint(DICTIONARY_COMMIT_BLOCK_POINT); + DebugPointUtil.removeDebugPoint(DICTIONARY_COMMIT_FAILURE_POINT); + executorService.shutdownNow(); + } + } + + @Test + void partialDictionaryCommitFailureAbortsStagedVersion() throws Exception { + Config.enable_debug_points = true; + String dbName = "partial_dictionary_commit_failure"; + Dictionary dictionary = createDictionaryWithoutAutoLoad(dbName); + long stagedVersion = dictionary.getVersion(); + + Backend backend = new Backend(1, "127.0.0.1", 9050); + backend.setBrpcPort(8060); + backend.setAlive(true); + BackendServiceProxy backendServiceProxy = Mockito.mock(BackendServiceProxy.class); + DebugPointUtil.addDebugPoint(DICTIONARY_COMMIT_FAILURE_POINT); + try (MockedConstruction ignored = Mockito.mockConstruction( + InsertIntoDictionaryCommand.class, (mock, context) -> Mockito.doAnswer(invocation -> { + ConnectContext loadContext = invocation.getArgument(0); + StatementContext statementContext = new StatementContext( + loadContext, new OriginStatement("", 0)); + statementContext.setPartialLoadDictionary(true); + statementContext.setUsedBackendsDistributing(List.of(backend)); + loadContext.setStatementContext(statementContext); + return null; + }).when(mock).run(Mockito.any(), Mockito.any())); + MockedStatic backendServiceProxyStatic = + Mockito.mockStatic(BackendServiceProxy.class)) { + backendServiceProxyStatic.when(BackendServiceProxy::getInstance).thenReturn(backendServiceProxy); + Mockito.when(backendServiceProxy.abortDictionaryAsync( + Mockito.any(), Mockito.anyInt(), Mockito.any())).thenReturn( + CompletableFuture.completedFuture( + InternalService.PAbortRefreshDictionaryResponse.getDefaultInstance())); + + Throwable failure = runRefresh(dbName); + + Assertions.assertNotNull(failure); + Assertions.assertTrue(failure.toString().contains("commit version " + stagedVersion), failure.toString()); + Assertions.assertEquals(stagedVersion, dictionary.getVersion()); + Mockito.verify(backendServiceProxy).abortDictionaryAsync( + Mockito.any(), Mockito.anyInt(), + Mockito.argThat(request -> request.getVersionId() == stagedVersion)); + } finally { + DebugPointUtil.removeDebugPoint(DICTIONARY_COMMIT_FAILURE_POINT); + Env.getCurrentInternalCatalog().dropDb(dbName, false, true); + } + } + + @Test + void mixedCommitKeepsFeAtPossiblyCommittedVersion() throws Exception { + Config.enable_debug_points = true; + String dbName = "mixed_dictionary_commit_result"; + Dictionary dictionary = createDictionaryWithoutAutoLoad(dbName); + long originalVersion = dictionary.getVersion(); + Backend firstBackend = newBackend(1); + Backend secondBackend = newBackend(2); + BackendServiceProxy backendServiceProxy = Mockito.mock(BackendServiceProxy.class); + InternalService.PCommitRefreshDictionaryResponse ok = + InternalService.PCommitRefreshDictionaryResponse.newBuilder() + .setStatus(Types.PStatus.newBuilder().setStatusCode(0)).build(); + InternalService.PCommitRefreshDictionaryResponse failed = + InternalService.PCommitRefreshDictionaryResponse.newBuilder() + .setStatus(Types.PStatus.newBuilder().setStatusCode(1)).build(); + InternalService.PAbortRefreshDictionaryResponse aborted = + InternalService.PAbortRefreshDictionaryResponse.newBuilder() + .setStatus(Types.PStatus.newBuilder().setStatusCode(0)).build(); + + try (MockedConstruction ignored = Mockito.mockConstruction( + InsertIntoDictionaryCommand.class, (mock, context) -> Mockito.doAnswer(invocation -> { + ConnectContext loadContext = invocation.getArgument(0); + StatementContext statementContext = new StatementContext( + loadContext, new OriginStatement("", 0)); + statementContext.setUsedBackendsDistributing(List.of(firstBackend, secondBackend)); + loadContext.setStatementContext(statementContext); + return null; + }).when(mock).run(Mockito.any(), Mockito.any())); + MockedStatic backendServiceProxyStatic = + Mockito.mockStatic(BackendServiceProxy.class)) { + backendServiceProxyStatic.when(BackendServiceProxy::getInstance).thenReturn(backendServiceProxy); + Mockito.when(backendServiceProxy.commitDictionaryAsync( + Mockito.any(), Mockito.anyInt(), Mockito.any())).thenReturn( + CompletableFuture.completedFuture(ok), CompletableFuture.completedFuture(failed)); + Mockito.when(backendServiceProxy.abortDictionaryAsync( + Mockito.any(), Mockito.anyInt(), Mockito.any())).thenReturn( + CompletableFuture.completedFuture(aborted)); + + Throwable failure = runRefresh(dbName); + + Assertions.assertNotNull(failure); + Assertions.assertEquals(originalVersion + 1, dictionary.getVersion()); + Assertions.assertEquals(DictionaryStatus.OUT_OF_DATE, dictionary.getStatus()); + Mockito.verify(backendServiceProxy, Mockito.times(2)).abortDictionaryAsync( + Mockito.any(), Mockito.anyInt(), + Mockito.argThat(request -> request.getVersionId() == originalVersion + 1)); + } finally { + Env.getCurrentInternalCatalog().dropDb(dbName, false, true); + } + } + + @Test + void loadFailureAbortsStagedDictionaryVersion() throws Exception { + Config.enable_debug_points = true; + String dbName = "dictionary_load_failure_abort"; + Dictionary dictionary = createDictionaryWithoutAutoLoad(dbName); + long originalVersion = dictionary.getVersion(); + Backend backend = newBackend(1); + BackendServiceProxy backendServiceProxy = Mockito.mock(BackendServiceProxy.class); + InternalService.PAbortRefreshDictionaryResponse aborted = + InternalService.PAbortRefreshDictionaryResponse.newBuilder() + .setStatus(Types.PStatus.newBuilder().setStatusCode(0)).build(); + + try (MockedConstruction ignored = Mockito.mockConstruction( + InsertIntoDictionaryCommand.class, (mock, context) -> Mockito.doAnswer(invocation -> { + ConnectContext loadContext = invocation.getArgument(0); + StatementContext statementContext = new StatementContext( + loadContext, new OriginStatement("", 0)); + statementContext.setUsedBackendsDistributing(List.of(backend)); + loadContext.setStatementContext(statementContext); + throw new RuntimeException("injected load failure"); + }).when(mock).run(Mockito.any(), Mockito.any())); + MockedStatic backendServiceProxyStatic = + Mockito.mockStatic(BackendServiceProxy.class)) { + backendServiceProxyStatic.when(BackendServiceProxy::getInstance).thenReturn(backendServiceProxy); + Mockito.when(backendServiceProxy.abortDictionaryAsync( + Mockito.any(), Mockito.anyInt(), Mockito.any())).thenReturn( + CompletableFuture.completedFuture(aborted)); + + Throwable failure = runRefresh(dbName); + + Assertions.assertNotNull(failure); + Assertions.assertTrue(failure.toString().contains("injected load failure"), failure.toString()); + Assertions.assertEquals(originalVersion, dictionary.getVersion()); + Mockito.verify(backendServiceProxy).abortDictionaryAsync( + Mockito.any(), Mockito.anyInt(), + Mockito.argThat(request -> request.getVersionId() == originalVersion + 1)); + } finally { + Env.getCurrentInternalCatalog().dropDb(dbName, false, true); + } + } + + @Test + void targetReplacementFailsBeforeExecutorCreation() throws Exception { + String dbName = "normal_insert_replaced_target"; + createDatabaseAndUse(dbName); + createTable("CREATE TABLE source_table (id INT) DISTRIBUTED BY HASH(id) BUCKETS 1 " + + "PROPERTIES ('replication_num' = '1')"); + createTable("CREATE TABLE target_table (id INT) DISTRIBUTED BY HASH(id) BUCKETS 1 " + + "PROPERTIES ('replication_num' = '1')"); + long oldDbId = Env.getCurrentInternalCatalog().getDbOrMetaException(dbName).getId(); + + String sql = "INSERT INTO " + dbName + ".target_table SELECT id FROM " + + dbName + ".source_table"; + CountDownLatch beforeValidation = new CountDownLatch(1); + CountDownLatch resumeValidation = new CountDownLatch(1); + InsertIntoTableCommand command = commandBlockingBeforeTargetLookup( + sql, 2, beforeValidation, resumeValidation, ctx -> { }); + AtomicReference planningContext = new AtomicReference<>(); + + ExecutorService executorService = Executors.newSingleThreadExecutor(); + try { + Future result = executorService.submit( + () -> runPlan(command, sql, dbName, false, planningContext::set)); + await(beforeValidation, result); + Env.getCurrentInternalCatalog().dropDb(dbName, false, true); + createDatabaseAndUse(dbName); + createTable("CREATE TABLE source_table (id INT) DISTRIBUTED BY HASH(id) BUCKETS 1 " + + "PROPERTIES ('replication_num' = '1')"); + createTable("CREATE TABLE target_table (id INT) DISTRIBUTED BY HASH(id) BUCKETS 1 " + + "PROPERTIES ('replication_num' = '1')"); + resumeValidation.countDown(); + + ExecutionException failure = Assertions.assertThrows(ExecutionException.class, + () -> result.get(10, TimeUnit.SECONDS)); + Assertions.assertTrue(hasCause(failure, org.apache.doris.nereids.exceptions.AnalysisException.class), + failure.toString()); + Assertions.assertTrue(failure.toString().contains("Insert target changed while planning"), + failure.toString()); + Assertions.assertFalse(planningContext.get().isGroupCommit()); + Assertions.assertNull(planningContext.get().getStatementContext().getGroupCommitMergeBackend()); + long currentDbId = Env.getCurrentInternalCatalog().getDbOrMetaException(dbName).getId(); + Assertions.assertNotEquals(oldDbId, currentDbId); + Assertions.assertEquals(0, Env.getCurrentEnv().getLoadManager().getLoadJobNum(JobState.PENDING, oldDbId)); + Assertions.assertEquals(0, + Env.getCurrentEnv().getLoadManager().getLoadJobNum(JobState.PENDING, currentDbId)); + } finally { + resumeValidation.countDown(); + executorService.shutdownNow(); + } + } + + @Test + void preparedPlanningRejectsCachedReplacedTarget() throws Exception { + String dbName = "repeated_plan_replaced_target"; + createDatabaseAndUse(dbName); + createTable("CREATE TABLE target_table (id INT) DISTRIBUTED BY HASH(id) BUCKETS 1 " + + "PROPERTIES ('replication_num' = '1')"); + long oldDbId = Env.getCurrentInternalCatalog().getDbOrMetaException(dbName).getId(); + String sql = "INSERT INTO " + dbName + ".target_table VALUES (1)"; + InsertIntoTableCommand command = (InsertIntoTableCommand) new NereidsParser().parseSingle(sql); + ConnectContext ctx = createCtx(UserIdentity.ROOT, "127.0.0.1"); + ctx.setDatabase(dbName); + ctx.setQueryId(new TUniqueId(1, System.nanoTime())); + ctx.setStatementContext(new StatementContext(ctx, new OriginStatement(sql, 0))); + + try { + command.initPlan(ctx, new StmtExecutor(new ConnectContext(), ""), false); + Assertions.assertFalse(ctx.getStatementContext().getInsertTargetTables().isEmpty()); + Env.getCurrentInternalCatalog().dropDb(dbName, false, true); + connectContext.setThreadLocalInfo(); + createDatabaseAndUse(dbName); + createTable("CREATE TABLE target_table (id INT) DISTRIBUTED BY HASH(id) BUCKETS 1 " + + "PROPERTIES ('replication_num' = '1')"); + ctx.setThreadLocalInfo(); + + org.apache.doris.nereids.exceptions.AnalysisException failure = Assertions.assertThrows( + org.apache.doris.nereids.exceptions.AnalysisException.class, + () -> command.initPlan(ctx, new StmtExecutor(new ConnectContext(), ""), false)); + Assertions.assertTrue(failure.getMessage().contains("Insert target changed while planning"), + failure.toString()); + long currentDbId = Env.getCurrentInternalCatalog().getDbOrMetaException(dbName).getId(); + Assertions.assertEquals(0, Env.getCurrentEnv().getLoadManager().getLoadJobNum(JobState.PENDING, oldDbId)); + Assertions.assertEquals(0, + Env.getCurrentEnv().getLoadManager().getLoadJobNum(JobState.PENDING, currentDbId)); + } finally { + connectContext.setThreadLocalInfo(); + } + } + + @Test + void schemaRetryRestoresAttemptState() throws Exception { + String dbName = "group_commit_target_retry"; + createDatabaseAndUse(dbName); + createTable("CREATE TABLE target_table (id INT) DISTRIBUTED BY HASH(id) BUCKETS 1 PROPERTIES (" + + "'replication_num' = '1', 'light_schema_change' = 'true')"); + + String sql = "INSERT INTO " + dbName + ".target_table VALUES (1)"; + AtomicInteger lookups = new AtomicInteger(); + AtomicBoolean oldAttemptUsedGroupCommit = new AtomicBoolean(false); + AtomicBoolean stateRestoredBeforeRetry = new AtomicBoolean(false); + AtomicBoolean privilegesResetBeforeRetry = new AtomicBoolean(false); + InsertIntoTableCommand baseCommand = (InsertIntoTableCommand) new NereidsParser().parseSingle(sql); + InsertIntoTableCommand command = new InsertIntoTableCommand(baseCommand, + PlanType.INSERT_INTO_TABLE_COMMAND) { + @Override + protected InsertTarget getTarget(ConnectContext ctx, List qualifiedTargetTableName) { + int lookup = lookups.incrementAndGet(); + if (lookup == 2) { + oldAttemptUsedGroupCommit.set(ctx.isGroupCommit()); + ctx.getStatementContext().setPrivChecked(true); + ctx.getStatementContext().getInsertTargetSchema().clear(); + } else if (lookup == 3) { + stateRestoredBeforeRetry.set(!ctx.isGroupCommit() + && ctx.getStatementContext().getGroupCommitMergeBackend() == null); + privilegesResetBeforeRetry.set(!ctx.getStatementContext().isPrivChecked()); + } + return super.getTarget(ctx, qualifiedTargetTableName); + } + }; + + PlanResult planResult = runPlan( + command, sql, dbName, false, ctx -> ctx.getSessionVariable().groupCommit = "async_mode"); + Assertions.assertTrue(oldAttemptUsedGroupCommit.get()); + Assertions.assertTrue(stateRestoredBeforeRetry.get()); + Assertions.assertTrue(privilegesResetBeforeRetry.get()); + Assertions.assertTrue(planResult.context.isGroupCommit()); + Assertions.assertEquals(OlapGroupCommitInsertExecutor.class, planResult.executor.getClass()); + } + + private Throwable runInitPlan(InsertIntoTableCommand command, String sql, String dbName) { + try { + ConnectContext ctx = createCtx(UserIdentity.ROOT, "127.0.0.1"); + ctx.setDatabase(dbName); + ctx.setQueryId(new TUniqueId(1, System.nanoTime())); + StatementContext statementContext = new StatementContext(ctx, new OriginStatement(sql, 0)); + ctx.setStatementContext(statementContext); + command.initPlan(ctx, new StmtExecutor(ctx, sql)); + return null; + } catch (Throwable t) { + return t; + } + } + + private PlanResult runPlan(InsertIntoTableCommand command, String sql, String dbName, + boolean needBeginTransaction, Consumer configureContext) throws Exception { + try { + ConnectContext ctx = createCtx(UserIdentity.ROOT, "127.0.0.1"); + ctx.setDatabase(dbName); + ctx.setQueryId(new TUniqueId(1, System.nanoTime())); + configureContext.accept(ctx); + StatementContext statementContext = new StatementContext(ctx, new OriginStatement(sql, 0)); + ctx.setStatementContext(statementContext); + AbstractInsertExecutor executor = command.initPlan( + ctx, new StmtExecutor(ctx, sql), needBeginTransaction); + return new PlanResult(ctx, executor); + } finally { + connectContext.setThreadLocalInfo(); + } + } + + private Throwable runRefresh(String dbName) { + try { + String sql = "REFRESH DICTIONARY " + dbName + ".dic1"; + ConnectContext ctx = createCtx(UserIdentity.ROOT, "127.0.0.1"); + ctx.setDatabase(dbName); + ctx.setQueryId(new TUniqueId(1, System.nanoTime())); + StatementContext statementContext = new StatementContext(ctx, new OriginStatement(sql, 0)); + ctx.setStatementContext(statementContext); + RefreshDictionaryCommand command = (RefreshDictionaryCommand) new NereidsParser().parseSingle(sql); + command.run(ctx, new StmtExecutor(ctx, sql)); + return null; + } catch (Throwable t) { + return t; + } finally { + connectContext.setThreadLocalInfo(); + } + } + + private Dictionary createDictionaryWithoutAutoLoad(String dbName) throws Exception { + DebugPoint initialLoadBlockPoint = addBlockingDebugPoint(DICTIONARY_BLOCK_POINT); + createDatabaseAndUse(dbName); + createTable("CREATE TABLE source_table (id INT NOT NULL, city VARCHAR(32) NOT NULL) " + + "DISTRIBUTED BY HASH(id) BUCKETS 1 PROPERTIES ('replication_num' = '1')"); + executeNereidsSql("CREATE DICTIONARY dic1 USING source_table (city KEY, id VALUE) " + + "LAYOUT(HASH_MAP) PROPERTIES ('data_lifetime' = '600')"); + + Dictionary dictionary = Env.getCurrentEnv().getDictionaryManager().getDictionary(dbName, "dic1"); + try { + await(() -> initialLoadBlockPoint.executeNum.get() > 0); + Env.getCurrentEnv().getDictionaryManager().dropDictionary(null, dbName, "dic1", false); + } finally { + DebugPointUtil.removeDebugPoint(DICTIONARY_BLOCK_POINT); + } + await(() -> dictionary.getStatus() == DictionaryStatus.OUT_OF_DATE + && !dictionary.getLastUpdateResult().isEmpty()); + // Re-register without scheduling another automatic load; the test thread owns the refresh below. + Env.getCurrentEnv().getDictionaryManager().replayCreateDictionary( + new CreateDictionaryPersistInfo(dictionary)); + return dictionary; + } + + private InsertIntoTableCommand commandBlockingBeforeTargetLookup(String sql, int lookupNumber, + CountDownLatch reached, CountDownLatch resume, Consumer onReached) { + InsertIntoTableCommand baseCommand = (InsertIntoTableCommand) new NereidsParser().parseSingle(sql); + AtomicInteger lookups = new AtomicInteger(); + return new InsertIntoTableCommand(baseCommand, PlanType.INSERT_INTO_TABLE_COMMAND) { + @Override + protected InsertTarget getTarget(ConnectContext ctx, List qualifiedTargetTableName) { + if (lookups.incrementAndGet() == lookupNumber) { + onReached.accept(ctx); + reached.countDown(); + await(resume); + } + return super.getTarget(ctx, qualifiedTargetTableName); + } + }; + } + + private static DebugPoint addBlockingDebugPoint(String name) { + DebugPoint debugPoint = new DebugPoint(); + debugPoint.executeLimit = Integer.MAX_VALUE; + DebugPointUtil.addDebugPoint(name, debugPoint); + return debugPoint; + } + + private static Backend newBackend(long id) { + Backend backend = new Backend(id, "127.0.0.1", 9050); + backend.setBrpcPort((int) (8060 + id)); + backend.setAlive(true); + return backend; + } + + private static class PlanResult { + private final ConnectContext context; + private final AbstractInsertExecutor executor; + + private PlanResult(ConnectContext context, AbstractInsertExecutor executor) { + this.context = context; + this.executor = executor; + } + } + + private static boolean hasCause(Throwable throwable, Class causeClass) { + for (Throwable cause = throwable; cause != null; cause = cause.getCause()) { + if (causeClass.isInstance(cause)) { + return true; + } + } + return false; + } + + private static void await(CountDownLatch latch, Future result) throws Exception { + if (!latch.await(10, TimeUnit.SECONDS)) { + if (result.isDone()) { + result.get(); + } + Assertions.fail("Timed out before the planning thread reached the race point"); + } + } + + 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()); + } + + private static void await(CountDownLatch latch) { + try { + if (!latch.await(10, TimeUnit.SECONDS)) { + throw new IllegalStateException("Timed out waiting to resume insert"); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IllegalStateException(e); + } + } +} diff --git a/fe/fe-core/src/test/java/org/apache/doris/planner/GroupCommitPlannerTargetRaceTest.java b/fe/fe-core/src/test/java/org/apache/doris/planner/GroupCommitPlannerTargetRaceTest.java new file mode 100644 index 00000000000000..abe816846a5dc5 --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/planner/GroupCommitPlannerTargetRaceTest.java @@ -0,0 +1,107 @@ +// 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.planner; + +import org.apache.doris.catalog.Database; +import org.apache.doris.catalog.Env; +import org.apache.doris.catalog.EnvFactory; +import org.apache.doris.catalog.OlapTable; +import org.apache.doris.common.DdlException; +import org.apache.doris.nereids.StatementContext; +import org.apache.doris.nereids.parser.NereidsParser; +import org.apache.doris.nereids.trees.expressions.literal.IntegerLiteral; +import org.apache.doris.nereids.trees.plans.PlanType; +import org.apache.doris.nereids.trees.plans.commands.PrepareCommand; +import org.apache.doris.nereids.trees.plans.commands.insert.AbstractInsertExecutor; +import org.apache.doris.nereids.trees.plans.commands.insert.InsertIntoTableCommand; +import org.apache.doris.qe.ConnectContext; +import org.apache.doris.qe.OriginStatement; +import org.apache.doris.qe.PreparedStatementContext; +import org.apache.doris.qe.StmtExecutor; +import org.apache.doris.thrift.TUniqueId; +import org.apache.doris.utframe.TestWithFeService; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; +import org.mockito.MockedStatic; +import org.mockito.Mockito; + +import java.util.List; + +class GroupCommitPlannerTargetRaceTest extends TestWithFeService { + + @Test + void replanUsesTargetValidatedByInsertPlan() throws Exception { + String originalDbName = "group_commit_original_target"; + createDatabaseAndUse(originalDbName); + createTable("CREATE TABLE target_table (id INT, value INT) DISTRIBUTED BY HASH(id) BUCKETS 1 " + + "PROPERTIES ('replication_num' = '1')"); + String validatedDbName = "group_commit_validated_target"; + createDatabaseAndUse(validatedDbName); + createTable("CREATE TABLE target_table (id INT, value INT) DISTRIBUTED BY HASH(id) BUCKETS 1 " + + "PROPERTIES ('replication_num' = '1')"); + Database validatedDatabase = Env.getCurrentInternalCatalog().getDbOrMetaException(validatedDbName); + OlapTable validatedTable = (OlapTable) validatedDatabase.getTableOrMetaException("target_table"); + + String sql = "INSERT INTO " + originalDbName + ".target_table VALUES (1, 2)"; + InsertIntoTableCommand baseCommand = (InsertIntoTableCommand) new NereidsParser().parseSingle(sql); + InsertIntoTableCommand command = new InsertIntoTableCommand(baseCommand, + PlanType.INSERT_INTO_TABLE_COMMAND) { + @Override + public AbstractInsertExecutor initPlan(ConnectContext ctx, StmtExecutor stmtExecutor, + boolean needBeginTransaction) throws Exception { + AbstractInsertExecutor executor = Mockito.mock(AbstractInsertExecutor.class); + Mockito.when(executor.getDatabase()).thenReturn(validatedDatabase); + Mockito.when(executor.getTable()).thenReturn(validatedTable); + return executor; + } + }; + + connectContext.setDatabase(originalDbName); + connectContext.setQueryId(new TUniqueId(1, System.nanoTime())); + StatementContext statementContext = new StatementContext(connectContext, new OriginStatement(sql, 0)); + statementContext.getIdToPlaceholderRealExpr().put( + statementContext.getNextPlaceholderId(), new IntegerLiteral(1)); + connectContext.setStatementContext(statementContext); + connectContext.setThreadLocalInfo(); + + PrepareCommand prepareCommand = new PrepareCommand( + "1", command, List.of(), new OriginStatement(sql, 0)); + PreparedStatementContext preparedStatementContext = new PreparedStatementContext( + prepareCommand, connectContext, statementContext, sql); + EnvFactory envFactory = Mockito.mock(EnvFactory.class); + GroupCommitPlanner planner = Mockito.mock(GroupCommitPlanner.class); + Mockito.when(envFactory.createGroupCommitPlanner( + Mockito.any(), Mockito.any(), Mockito.any(), Mockito.any(), Mockito.any())).thenReturn(planner); + + try (MockedStatic envFactoryStatic = Mockito.mockStatic(EnvFactory.class)) { + envFactoryStatic.when(EnvFactory::getInstance).thenReturn(envFactory); + + Assertions.assertThrows(DdlException.class, () -> GroupCommitPlanner.executeGroupCommitInsert( + connectContext, preparedStatementContext, statementContext)); + + ArgumentCaptor databaseCaptor = ArgumentCaptor.forClass(Database.class); + ArgumentCaptor tableCaptor = ArgumentCaptor.forClass(OlapTable.class); + Mockito.verify(envFactory).createGroupCommitPlanner( + databaseCaptor.capture(), tableCaptor.capture(), Mockito.any(), Mockito.any(), Mockito.any()); + Assertions.assertEquals(validatedDatabase.getId(), databaseCaptor.getValue().getId()); + Assertions.assertEquals(validatedTable.getId(), tableCaptor.getValue().getId()); + } + } +} 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 951e2bef382969..64b64cbf59fb03 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" @@ -81,28 +81,39 @@ 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() + + long originalVersion = (sql "SHOW DICTIONARIES")[0][3].toLong() + def commitFailureFuture + try { + GetDebugPoint().enableDebugPointForAllFEs('DictionaryManager.dataLoad.blockBeforeCommit') + GetDebugPoint().enableDebugPointForAllFEs('DictionaryManager.commitNowVersion.forceFailure') + commitFailureFuture = thread { + sql "REFRESH DICTIONARY test_create_drop_sync.dic1" + } + awaitUntil(10) { + def dictionaries = sql "SHOW DICTIONARIES" + dictionaries.size() == 1 && dictionaries[0][3].toLong() == originalVersion + 1 + } + sql "DROP DATABASE test_create_drop_sync" + } finally { + GetDebugPoint().disableDebugPointForAllFEs('DictionaryManager.dataLoad.blockBeforeCommit') + GetDebugPoint().disableDebugPointForAllFEs('DictionaryManager.commitNowVersion.forceFailure') + } + + Exception commitFailure = null + assertNotNull(commitFailureFuture) + try { + commitFailureFuture.get() + } catch (Exception e) { + commitFailure = e + } + assertNotNull(commitFailure) + assertTrue(commitFailure.toString().contains("commit version ${originalVersion + 1} failed"), + commitFailure.toString()) + sql "CREATE DATABASE test_create_drop_sync" sql "USE test_create_drop_sync" dict_res = sql "SHOW DICTIONARIES" assertEquals(dict_res.size(), 0) - sql """ - CREATE TABLE source_table ( - id INT NOT NULL, - city VARCHAR(32) NOT NULL, - code VARCHAR(32) NOT NULL - ) ENGINE=OLAP - DISTRIBUTED BY HASH(id) BUCKETS 1 - PROPERTIES("replication_num" = "1") - """ - sql """ - create dictionary dic1 using source_table - ( - city KEY, - id VALUE - ) - LAYOUT(HASH_MAP) - properties('data_lifetime'='600'); - """ -} \ No newline at end of file +}