From 69d39244b5301e5612e9122e3031f08fa26da24c Mon Sep 17 00:00:00 2001 From: zhaochangle Date: Sat, 11 Jul 2026 02:11:30 +0800 Subject: [PATCH 1/4] fix: stabilize dictionary insert target during database drop Resolve insert targets as a database-table pair so authorization and retry validation do not recover the database through a stale table object. Carry the dictionary owner database through Nereids sink binding and executor creation. Remove dropped dictionary IDs from the canonical map and validate exact dictionary identity before and after loading to reject same-name recreation. Add deterministic FE unit races for normal inserts and dictionary loads, plus a debug-point regression covering concurrent REFRESH DICTIONARY and DROP DATABASE. Tests: ./run-fe-ut.sh --run org.apache.doris.nereids.trees.plans.commands.insert.InsertTargetDropRaceTest Tests: ./run-regression-test.sh --run -f regression-test/suites/dictionary_p0/test_create_drop_sync.groovy Tests: ./build.sh --fe Tests: ./build.sh --be --- .../doris/dictionary/DictionaryManager.java | 32 ++- .../analyzer/UnboundDictionarySink.java | 17 +- .../analyzer/UnboundTableSinkCreator.java | 7 +- .../nereids/rules/analysis/BindSink.java | 18 +- .../insert/AbstractInsertExecutor.java | 13 +- .../insert/DictionaryInsertExecutor.java | 8 +- .../insert/InsertIntoDictionaryCommand.java | 15 +- .../insert/InsertIntoTableCommand.java | 63 ++++-- .../insert/InsertTargetDropRaceTest.java | 182 ++++++++++++++++++ .../test_create_drop_sync.groovy | 35 +++- 10 files changed, 335 insertions(+), 55 deletions(-) create mode 100644 fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertTargetDropRaceTest.java 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..ddf41cd4dd0d36 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 { @@ -464,8 +491,7 @@ public void dataLoad(ConnectContext ctx, Dictionary dictionary, boolean adaptive lockRead(); boolean unlocked = false; try { - if (!dictionaryIds.containsKey(dictionary.getDbName()) - || !dictionaryIds.get(dictionary.getDbName()).containsKey(dictionary.getName())) { + if (!isCurrentDictionaryWithoutLock(dictionary)) { unlockRead(); unlocked = true; 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..8e5fe0df9efab1 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 { @@ -260,15 +283,17 @@ public AbstractInsertExecutor initPlan(ConnectContext ctx, StmtExecutor stmtExec int retryTimes = 0; ctx.getStatementContext().setIsInsert(true); while (++retryTimes < Math.max(ctx.getSessionVariable().dmlPlanRetryTimes, 3)) { - TableIf targetTableIf = getTargetTableIf(ctx, qualifiedTargetTableName); + InsertTarget target = getTarget(ctx, qualifiedTargetTableName); + DatabaseIf targetDatabase = target.getDatabase(); + TableIf targetTableIf = target.getTable(); // 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; @@ -308,13 +333,17 @@ public void beforeComplete(AbstractInsertExecutor insertExecutor, StmtExecutor e } // 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 {}", + if (targetDatabase.getId() != newestTargetDatabase.getId() + || targetTableIf.getId() != newestTargetTableIf.getId()) { + LOG.warn("insert plan failed {} times. query id is {}. target id changed from {}.{} to {}.{}", retryTimes, DebugUtil.printId(ctx.queryId()), - targetTableIf.getId(), newestTargetTableIf.getId()); + targetDatabase.getId(), targetTableIf.getId(), + newestTargetDatabase.getId(), newestTargetTableIf.getId()); newestTargetTableIf.readUnlock(); continue; } @@ -577,10 +606,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. @@ -695,14 +725,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/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..cd9ecb36b7fb6c --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertTargetDropRaceTest.java @@ -0,0 +1,182 @@ +// 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.dictionary.Dictionary; +import org.apache.doris.dictionary.Dictionary.DictionaryStatus; +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.qe.ConnectContext; +import org.apache.doris.qe.OriginStatement; +import org.apache.doris.qe.StmtExecutor; +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 InsertTargetDropRaceTest extends TestWithFeService { + private static final String DICTIONARY_BLOCK_POINT = "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 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; + DebugPointUtil.addDebugPoint(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(() -> dictionary.getStatus() == DictionaryStatus.LOADING); + 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); + } + + private Throwable runInitPlan(InsertIntoTableCommand command, String sql, String dbName) { + try { + ConnectContext ctx = createCtx(UserIdentity.ROOT, "127.0.0.1"); + ctx.setDatabase(dbName); + 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 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(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/regression-test/suites/dictionary_p0/test_create_drop_sync.groovy b/regression-test/suites/dictionary_p0/test_create_drop_sync.groovy index 951e2bef382969..d28af10f200846 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,8 +81,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 no stale dictionary metadata remains. sql "CREATE DATABASE test_create_drop_sync" sql "USE test_create_drop_sync" dict_res = sql "SHOW DICTIONARIES" @@ -105,4 +132,4 @@ suite('test_create_drop_sync') { LAYOUT(HASH_MAP) properties('data_lifetime'='600'); """ -} \ No newline at end of file +} From e4709e3d002c2b1bad131241d58b66b4678f712b Mon Sep 17 00:00:00 2001 From: zhaochangle Date: Sun, 12 Jul 2026 22:36:31 +0800 Subject: [PATCH 2/4] [fix](dictionary) Make refresh cleanup and insert retries race-safe ### What problem does this PR solve? Issue Number: None Related PR: #65476 Problem Summary: A dictionary commit failure could append a version-decrement journal after the dictionary had already been dropped, which breaks edit-log replay. Partial refresh failures also aborted the next version instead of the version staged on BEs. Separately, insert retries constructed executors and registered load jobs before confirming that the planned database and table were still current, and a discarded group-commit attempt could leak derived state into the next attempt. Serialize the dictionary identity check and full-refresh rollback journal with DROP, capture and abort the exact staged version, and delay insert executor construction until target ID and schema validation succeeds. Restore both group-commit mode and its selected merge backend when an attempt is discarded. ### Release note Fix concurrent dictionary refresh/drop cleanup and prevent discarded insert plans from creating load jobs or leaking group-commit state. ### Check List (For Author) - Test: Unit Test and Regression test - InsertTargetDropRaceTest: 6 tests passed - InsertIntoTableCommandTableStreamTest: 4 tests passed - dictionary_p0/test_create_drop_sync.groovy: passed - ./build.sh --fe: passed - Behavior changed: Yes. Failed refresh cleanup is replay-safe and insert retry side effects occur only for a validated target. - Does this need documentation: No --- .../doris/dictionary/DictionaryManager.java | 26 +- .../insert/InsertIntoTableCommand.java | 73 +++-- .../insert/InsertTargetDropRaceTest.java | 282 +++++++++++++++++- .../test_create_drop_sync.groovy | 36 +++ 4 files changed, 378 insertions(+), 39 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 ddf41cd4dd0d36..f04ad273b63172 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 @@ -521,16 +521,28 @@ 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. + long stagedVersion = dictionary.getVersion(); if (!commitNowVersion(ctx, dictionary)) { - if (!ctx.getStatementContext().isPartialLoadDictionary()) { - dictionary.decreaseVersion(); - Env.getCurrentEnv().getEditLog().logDictionaryDecVersion(dictionary); + // Keep the identity check and rollback journal ordered with DROP. + lockRead(); + try { + if (isCurrentDictionaryWithoutLock(dictionary) + && !ctx.getStatementContext().isPartialLoadDictionary()) { + dictionary.decreaseVersion(); + Env.getCurrentEnv().getEditLog().logDictionaryDecVersion(dictionary); + } + } 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. @@ -551,6 +563,10 @@ public void dataLoad(ConnectContext ctx, Dictionary dictionary, boolean adaptive } private boolean commitNowVersion(ConnectContext ctx, Dictionary dictionary) { + if (DebugPointUtil.isEnable("DictionaryManager.commitNowVersion.forceFailure")) { + return false; + } + // use the same BEs when we get before start loading. List beList = ctx.getStatementContext().getUsedBackendsDistributing(); 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 8e5fe0df9efab1..5f4fc3573fde02 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 @@ -279,10 +279,14 @@ 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)) { + boolean groupCommitBeforeAttempt = ctx.isGroupCommit(); + Backend groupCommitBackendBeforeAttempt = + ctx.getStatementContext().getGroupCommitMergeBackend(); + insertExecutor = null; InsertTarget target = getTarget(ctx, qualifiedTargetTableName); DatabaseIf targetDatabase = target.getDatabase(); TableIf targetTableIf = target.getTable(); @@ -296,40 +300,25 @@ public AbstractInsertExecutor initPlan(ConnectContext ctx, StmtExecutor stmtExec 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. @@ -344,6 +333,8 @@ public void beforeComplete(AbstractInsertExecutor insertExecutor, StmtExecutor e retryTimes, DebugUtil.printId(ctx.queryId()), targetDatabase.getId(), targetTableIf.getId(), newestTargetDatabase.getId(), newestTargetTableIf.getId()); + ctx.setGroupCommit(groupCommitBeforeAttempt); + ctx.getStatementContext().setGroupCommitMergeBackend(groupCommitBackendBeforeAttempt); newestTargetTableIf.readUnlock(); continue; } @@ -352,10 +343,30 @@ public void beforeComplete(AbstractInsertExecutor insertExecutor, StmtExecutor e 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); 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, @@ -372,6 +383,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 @@ -393,7 +407,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 { @@ -628,7 +642,7 @@ ExecutorFactory selectInsertExecutorFactory( } } - private BuildInsertExecutorResult planInsertExecutor( + private ExecutorFactory planInsertExecutor( ConnectContext ctx, StmtExecutor stmtExecutor, LogicalPlanAdapter logicalPlanAdapter, TableIf targetTableIf) throws Throwable { LogicalPlan logicalPlan = logicalPlanAdapter.getLogicalPlan(); @@ -668,15 +682,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; 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 index cd9ecb36b7fb6c..646daf180ef44c 100644 --- 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 @@ -21,32 +21,49 @@ 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.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.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.function.Consumer; import java.util.function.BooleanSupplier; 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 @@ -106,7 +123,7 @@ protected InsertTarget getTarget(ConnectContext ctx, List qualifiedTarge @Test void dictionaryLoadStopsCleanlyWhenDatabaseIsDroppedBeforePlanning() throws Exception { Config.enable_debug_points = true; - DebugPointUtil.addDebugPoint(DICTIONARY_BLOCK_POINT); + DebugPoint blockPoint = addBlockingDebugPoint(DICTIONARY_BLOCK_POINT); String dbName = "dictionary_insert_drop_race"; createDatabaseAndUse(dbName); @@ -117,7 +134,7 @@ void dictionaryLoadStopsCleanlyWhenDatabaseIsDroppedBeforePlanning() throws Exce Dictionary dictionary = Env.getCurrentEnv().getDictionaryManager().getDictionary(dbName, "dic1"); try { - await(() -> dictionary.getStatus() == DictionaryStatus.LOADING); + await(() -> blockPoint.executeNum.get() > 0); Env.getCurrentInternalCatalog().dropDb(dbName, false, true); createDatabaseAndUse(dbName); @@ -139,10 +156,175 @@ void dictionaryLoadStopsCleanlyWhenDatabaseIsDroppedBeforePlanning() throws Exce 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 targetRetryBuildsExecutorOnlyForCurrentDatabase() throws Exception { + String sourceDb = "normal_insert_retry_source"; + createDatabaseAndUse(sourceDb); + createTable("CREATE TABLE source_table (id INT) DISTRIBUTED BY HASH(id) BUCKETS 1 " + + "PROPERTIES ('replication_num' = '1')"); + + String targetDb = "normal_insert_retry_target"; + createDatabaseAndUse(targetDb); + createTable("CREATE TABLE target_table (id INT) DISTRIBUTED BY HASH(id) BUCKETS 1 " + + "PROPERTIES ('replication_num' = '1')"); + long oldDbId = Env.getCurrentInternalCatalog().getDbOrMetaException(targetDb).getId(); + + String sql = "INSERT INTO " + targetDb + ".target_table SELECT id FROM " + + sourceDb + ".source_table"; + CountDownLatch beforeValidation = new CountDownLatch(1); + CountDownLatch resumeValidation = new CountDownLatch(1); + InsertIntoTableCommand command = commandBlockingBeforeTargetLookup( + sql, 2, beforeValidation, resumeValidation, ctx -> { }); + + ExecutorService executorService = Executors.newSingleThreadExecutor(); + try { + Future result = executorService.submit( + () -> runPlan(command, sql, targetDb, false, ctx -> { })); + await(beforeValidation, result); + Env.getCurrentInternalCatalog().dropDb(targetDb, false, true); + createDatabaseAndUse(targetDb); + createTable("CREATE TABLE target_table (id INT) DISTRIBUTED BY HASH(id) BUCKETS 1 " + + "PROPERTIES ('replication_num' = '1')"); + resumeValidation.countDown(); + + PlanResult planResult = result.get(10, TimeUnit.SECONDS); + long currentDbId = Env.getCurrentInternalCatalog().getDbOrMetaException(targetDb).getId(); + Assertions.assertNotEquals(oldDbId, currentDbId); + Assertions.assertEquals(currentDbId, planResult.executor.getDatabase().getId()); + Assertions.assertEquals(0, Env.getCurrentEnv().getLoadManager().getLoadJobNum(JobState.PENDING, oldDbId)); + } finally { + resumeValidation.countDown(); + executorService.shutdownNow(); + } + } + + @Test + void targetRetryRecomputesDerivedGroupCommitState() 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)"; + CountDownLatch beforeValidation = new CountDownLatch(1); + CountDownLatch resumeValidation = new CountDownLatch(1); + AtomicBoolean oldAttemptUsedGroupCommit = new AtomicBoolean(false); + InsertIntoTableCommand command = commandBlockingBeforeTargetLookup( + sql, 2, beforeValidation, resumeValidation, + ctx -> oldAttemptUsedGroupCommit.set(ctx.isGroupCommit())); + + ExecutorService executorService = Executors.newSingleThreadExecutor(); + try { + Future result = executorService.submit(() -> runPlan( + command, sql, dbName, false, ctx -> ctx.getSessionVariable().groupCommit = "async_mode")); + await(beforeValidation, result); + Assertions.assertTrue(oldAttemptUsedGroupCommit.get()); + Env.getCurrentInternalCatalog().dropDb(dbName, false, true); + createDatabaseAndUse(dbName); + createTable("CREATE TABLE target_table (id INT) DISTRIBUTED BY HASH(id) BUCKETS 1 PROPERTIES (" + + "'replication_num' = '1', 'light_schema_change' = 'false')"); + resumeValidation.countDown(); + + PlanResult planResult = result.get(10, TimeUnit.SECONDS); + Assertions.assertFalse(planResult.context.isGroupCommit()); + Assertions.assertNull(planResult.context.getStatementContext().getGroupCommitMergeBackend()); + Assertions.assertEquals(OlapInsertExecutor.class, planResult.executor.getClass()); + } finally { + resumeValidation.countDown(); + executorService.shutdownNow(); + } + } + 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)); @@ -152,6 +334,93 @@ private Throwable runInitPlan(InsertIntoTableCommand command, String sql, String } } + private PlanResult runPlan(InsertIntoTableCommand command, String sql, String dbName, + boolean needBeginTransaction, Consumer configureContext) throws Exception { + 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); + } + + 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 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)) { @@ -161,6 +430,15 @@ private static boolean hasCause(Throwable throwable, Class 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) { 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 d28af10f200846..bff34789946b2a 100644 --- a/regression-test/suites/dictionary_p0/test_create_drop_sync.groovy +++ b/regression-test/suites/dictionary_p0/test_create_drop_sync.groovy @@ -132,4 +132,40 @@ suite('test_create_drop_sync', 'nonConcurrent') { LAYOUT(HASH_MAP) properties('data_lifetime'='600'); """ + + 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') + } + + Exception commitFailure = null + assertNotNull(commitFailureFuture) + try { + commitFailureFuture.get() + } catch (Exception e) { + commitFailure = e + } finally { + GetDebugPoint().disableDebugPointForAllFEs('DictionaryManager.commitNowVersion.forceFailure') + } + 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) } From 4411858e49e95feefb24ea0fa1a25ac8fb8f36e8 Mon Sep 17 00:00:00 2001 From: zhaochangle Date: Mon, 13 Jul 2026 08:15:19 +0800 Subject: [PATCH 3/4] [fix](fe) Sort dictionary race test imports ### What problem does this PR solve? Issue Number: None Related PR: #65476 Problem Summary: The FE Code Style Checker rejected the dictionary race test because BooleanSupplier appeared after Consumer instead of following lexicographical import order. Sort the imports according to the repository CheckStyle rule. ### Release note None ### Check List (For Author) - Test: mvn clean checkstyle:check - Behavior changed: No - Does this need documentation: No --- .../trees/plans/commands/insert/InsertTargetDropRaceTest.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 index 646daf180ef44c..697a7bf350160c 100644 --- 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 @@ -56,8 +56,8 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; -import java.util.function.Consumer; import java.util.function.BooleanSupplier; +import java.util.function.Consumer; class InsertTargetDropRaceTest extends TestWithFeService { private static final String DICTIONARY_BLOCK_POINT = "DictionaryManager.dataLoad.blockBeforePlan"; From 5550b6ebbd4bd95ea99f0b86431e542f381a4f8d Mon Sep 17 00:00:00 2001 From: zhaochangle Date: Mon, 13 Jul 2026 17:32:55 +0800 Subject: [PATCH 4/4] [fix](dictionary) Close insert and refresh retry races Reject insert target generation changes instead of retrying with cached StatementContext relations, and reset privilege/group-commit state for same-table schema retries. Build prepared group-commit plans from the target validated by initPlan, and invalidate stale cached planners by table ID and schema version. Abort staged dictionary data on pre-commit failures. Track uncertain or mixed BE commits so FE keeps a possibly committed version OUT_OF_DATE instead of journaling an unsafe rollback. Add FE race tests for ordinary and prepared insert paths, mixed dictionary commits and staged cleanup, plus targeted dictionary regression coverage. --- .../doris/dictionary/DictionaryManager.java | 60 ++++- .../insert/InsertIntoTableCommand.java | 19 +- .../doris/planner/GroupCommitPlanner.java | 43 ++- .../insert/InsertTargetDropRaceTest.java | 252 ++++++++++++++---- .../GroupCommitPlannerTargetRaceTest.java | 107 ++++++++ .../test_create_drop_sync.groovy | 54 +--- 6 files changed, 405 insertions(+), 130 deletions(-) create mode 100644 fe/fe-core/src/test/java/org/apache/doris/planner/GroupCommitPlannerTargetRaceTest.java 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 f04ad273b63172..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 @@ -469,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()); @@ -476,15 +477,17 @@ 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! @@ -527,19 +530,23 @@ public void dataLoad(ConnectContext ctx, Dictionary dictionary, boolean adaptive // commit and check the result. not modify metadata so dont need lock. long stagedVersion = dictionary.getVersion(); - if (!commitNowVersion(ctx, dictionary)) { + CommitResult commitResult = commitNowVersion(ctx, dictionary); + if (!commitResult.allSucceeded) { // Keep the identity check and rollback journal ordered with DROP. lockRead(); try { - if (isCurrentDictionaryWithoutLock(dictionary) - && !ctx.getStatementContext().isPartialLoadDictionary()) { - dictionary.decreaseVersion(); - Env.getCurrentEnv().getEditLog().logDictionaryDecVersion(dictionary); + 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, stagedVersion); throw new RuntimeException("Dictionary " + dictionary.getName() + " commit version " + stagedVersion + " failed"); @@ -562,9 +569,19 @@ 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 false; + return new CommitResult(false, false); } // use the same BEs when we get before start loading. @@ -572,6 +589,7 @@ private boolean commitNowVersion(ConnectContext ctx, Dictionary dictionary) { List> futureList = new ArrayList<>(); boolean allSucceed = true; + boolean mayHaveCommitted = false; try { for (Backend be : beList) { if (!be.isAlive()) { @@ -599,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()); @@ -609,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/trees/plans/commands/insert/InsertIntoTableCommand.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTableCommand.java index 5f4fc3573fde02..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 @@ -290,6 +290,16 @@ public AbstractInsertExecutor initPlan(ConnectContext ctx, StmtExecutor stmtExec 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(), targetDatabase.getCatalog().getName(), @@ -329,14 +339,14 @@ public AbstractInsertExecutor initPlan(ConnectContext ctx, StmtExecutor stmtExec try { if (targetDatabase.getId() != newestTargetDatabase.getId() || targetTableIf.getId() != newestTargetTableIf.getId()) { - LOG.warn("insert plan failed {} times. query id is {}. target id changed from {}.{} to {}.{}", - retryTimes, DebugUtil.printId(ctx.queryId()), + 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); - newestTargetTableIf.readUnlock(); - continue; + 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())) { @@ -345,6 +355,7 @@ public AbstractInsertExecutor initPlan(ConnectContext ctx, StmtExecutor stmtExec ctx.getStatementContext().getInsertTargetSchema(), newestTargetTableIf.getFullSchema()); ctx.setGroupCommit(groupCommitBeforeAttempt); ctx.getStatementContext().setGroupCommitMergeBackend(groupCommitBackendBeforeAttempt); + ctx.getStatementContext().setPrivChecked(false); newestTargetTableIf.readUnlock(); continue; } 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 index 697a7bf350160c..51a747e10a6008 100644 --- 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 @@ -31,6 +31,7 @@ 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; @@ -50,12 +51,14 @@ 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; @@ -242,41 +245,141 @@ void partialDictionaryCommitFailureAbortsStagedVersion() throws Exception { } @Test - void targetRetryBuildsExecutorOnlyForCurrentDatabase() throws Exception { - String sourceDb = "normal_insert_retry_source"; - createDatabaseAndUse(sourceDb); + 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')"); - - String targetDb = "normal_insert_retry_target"; - createDatabaseAndUse(targetDb); createTable("CREATE TABLE target_table (id INT) DISTRIBUTED BY HASH(id) BUCKETS 1 " + "PROPERTIES ('replication_num' = '1')"); - long oldDbId = Env.getCurrentInternalCatalog().getDbOrMetaException(targetDb).getId(); + long oldDbId = Env.getCurrentInternalCatalog().getDbOrMetaException(dbName).getId(); - String sql = "INSERT INTO " + targetDb + ".target_table SELECT id FROM " - + sourceDb + ".source_table"; + 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, targetDb, false, ctx -> { })); + () -> runPlan(command, sql, dbName, false, planningContext::set)); await(beforeValidation, result); - Env.getCurrentInternalCatalog().dropDb(targetDb, false, true); - createDatabaseAndUse(targetDb); + 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(); - PlanResult planResult = result.get(10, TimeUnit.SECONDS); - long currentDbId = Env.getCurrentInternalCatalog().getDbOrMetaException(targetDb).getId(); + 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(currentDbId, planResult.executor.getDatabase().getId()); 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(); @@ -284,42 +387,83 @@ void targetRetryBuildsExecutorOnlyForCurrentDatabase() throws Exception { } @Test - void targetRetryRecomputesDerivedGroupCommitState() throws Exception { - String dbName = "group_commit_target_retry"; + 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', 'light_schema_change' = 'true')"); - + 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)"; - CountDownLatch beforeValidation = new CountDownLatch(1); - CountDownLatch resumeValidation = new CountDownLatch(1); - AtomicBoolean oldAttemptUsedGroupCommit = new AtomicBoolean(false); - InsertIntoTableCommand command = commandBlockingBeforeTargetLookup( - sql, 2, beforeValidation, resumeValidation, - ctx -> oldAttemptUsedGroupCommit.set(ctx.isGroupCommit())); + 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))); - ExecutorService executorService = Executors.newSingleThreadExecutor(); try { - Future result = executorService.submit(() -> runPlan( - command, sql, dbName, false, ctx -> ctx.getSessionVariable().groupCommit = "async_mode")); - await(beforeValidation, result); - Assertions.assertTrue(oldAttemptUsedGroupCommit.get()); + 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', 'light_schema_change' = 'false')"); - resumeValidation.countDown(); + createTable("CREATE TABLE target_table (id INT) DISTRIBUTED BY HASH(id) BUCKETS 1 " + + "PROPERTIES ('replication_num' = '1')"); + ctx.setThreadLocalInfo(); - PlanResult planResult = result.get(10, TimeUnit.SECONDS); - Assertions.assertFalse(planResult.context.isGroupCommit()); - Assertions.assertNull(planResult.context.getStatementContext().getGroupCommitMergeBackend()); - Assertions.assertEquals(OlapInsertExecutor.class, planResult.executor.getClass()); + 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 { - resumeValidation.countDown(); - executorService.shutdownNow(); + 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"); @@ -336,14 +480,19 @@ private Throwable runInitPlan(InsertIntoTableCommand command, String sql, String private PlanResult runPlan(InsertIntoTableCommand command, String sql, String dbName, boolean needBeginTransaction, Consumer configureContext) throws Exception { - 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); + 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) { @@ -411,6 +560,13 @@ private static DebugPoint addBlockingDebugPoint(String name) { 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; 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 bff34789946b2a..64b64cbf59fb03 100644 --- a/regression-test/suites/dictionary_p0/test_create_drop_sync.groovy +++ b/regression-test/suites/dictionary_p0/test_create_drop_sync.groovy @@ -83,57 +83,6 @@ suite('test_create_drop_sync', 'nonConcurrent') { 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 no stale dictionary metadata remains. - 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'); - """ - - waitAllDictionariesReady() long originalVersion = (sql "SHOW DICTIONARIES")[0][3].toLong() def commitFailureFuture try { @@ -149,6 +98,7 @@ suite('test_create_drop_sync', 'nonConcurrent') { sql "DROP DATABASE test_create_drop_sync" } finally { GetDebugPoint().disableDebugPointForAllFEs('DictionaryManager.dataLoad.blockBeforeCommit') + GetDebugPoint().disableDebugPointForAllFEs('DictionaryManager.commitNowVersion.forceFailure') } Exception commitFailure = null @@ -157,8 +107,6 @@ suite('test_create_drop_sync', 'nonConcurrent') { commitFailureFuture.get() } catch (Exception e) { commitFailure = e - } finally { - GetDebugPoint().disableDebugPointForAllFEs('DictionaryManager.commitNowVersion.forceFailure') } assertNotNull(commitFailure) assertTrue(commitFailure.toString().contains("commit version ${originalVersion + 1} failed"),