Skip to content

[fix](dictionary) stabilize insert target during database drop #65476

New issue

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

Sign up for GitHub

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

Already on GitHub? Sign in to your account

Open
zclllyybb wants to merge 3 commits into
base: master
Choose a base branch
from
Open

[fix](dictionary) stabilize insert target during database drop #65476

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
69d3924
fix: stabilize dictionary insert target during database drop
zclllyybb Jul 10, 2026
e4709e3
[fix](dictionary) Make refresh cleanup and insert retries race-safe
zclllyybb Jul 12, 2026
4411858
[fix](fe) Sort dictionary race test imports
zclllyybb Jul 13, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
58 changes: 50 additions & 8 deletions fe/fe-core/src/main/java/org/apache/doris/dictionary/DictionaryManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand Down Expand Up @@ -267,6 +269,7 @@ public void dropDbDictionaries(String dbName) {
// Log the drop operation
if (dbDictIds != null) {
for (Map.Entry<String, Long> entry : dbDictIds.entrySet()) {
idToDictionary.remove(entry.getValue());
Env.getCurrentEnv().getEditLog().logDropDictionary(dbName, entry.getKey());
}
// also drop all name mapping records.
Expand All @@ -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<String, Dictionary> getDictionaries(String dbName) {
lockRead();
try {
Expand Down Expand Up @@ -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());
Expand All @@ -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(

@github-actions github-actions Bot Jul 13, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The pre-commit failure paths still do not clean up staged dictionary data on BEs. A dictionary sink stages the refresh at EOS via DictionaryFactory::refresh_dict(), but if one BE has staged data and command.run() later throws, or leaves an error in ctx.getState(), the following catch/error branches only restore FE status and rethrow. They never call abortSpecificVersion(), and the later dropped-dictionary/commit-failure abort blocks are not reached. Because BE status collection and delete_dict() only look at committed dictionaries, the _refreshing_dict_map entry can remain invisible until another refresh overwrites it. Please abort the staged version in both pre-commit error paths when the selected BEs are known, using the current version for partial loads and version + 1 for full loads.

baseCommand, database, dictionary, adaptiveLoad);

// run with sync by status.
try {
Expand Down Expand Up @@ -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)) {

@github-actions github-actions Bot Jul 10, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This identity check is still too early for the failure path below. After this block releases the dictionary-manager read lock, a concurrent DROP DICTIONARY/DROP DATABASE can remove this same dictionary and log OP_DROP_DICTIONARY; if commitNowVersion() then returns false, lines 527-528 still call logDictionaryDecVersion(dictionary) for an object that is no longer current. On replay the drop removes the dictionary, and replayDecreaseVersion() calls getDictionary(dbName, dictName), so the later decrement entry cannot be applied. Please recheck/hold the dictionary identity before logging the decrement, and skip the rollback journal entry once the dictionary has already been dropped.

@github-actions github-actions Bot Jul 12, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

There is another cleanup mismatch in this same post-load section for partial refreshes. When statementContext.isPartialLoadDictionary() is true, the sink staged dictionary.getVersion() on the selected BEs, not version + 1; DictionarySink.toThrift() only uses version + 1 for full refreshes. But if commitNowVersion() fails, this branch skips the full-load decrement and still calls abortSpecificVersion(..., dictionary.getVersion() + 1). BE abort only erases the refreshing dictionary when the version matches exactly, so a failed partial load can leave the staged current-version refresh in _refreshing_dict_map. Please abort dictionary.getVersion() for partial loads and dictionary.getVersion() + 1 for full loads.

unlockRead();
unlocked = true;

Expand Down Expand Up @@ -495,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)) {

@github-actions github-actions Bot Jul 13, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This rollback still assumes commit failure means no BE made the staged version visible, but commitNowVersion() treats mixed results as a plain false. If one BE returns OK and a later BE fails or times out, the OK BE has already moved stagedVersion into _dict_id_to_dict_map and erased _refreshing_dict_map; the subsequent abortSpecificVersion(..., stagedVersion) cannot undo it because BE abort only removes refreshing entries. FE then decrements back to the old version and future dict_get plans carry that old version, while the committed BE requires the staged version exactly. Please handle mixed commit results explicitly, for example by not rolling FE metadata back once any BE has committed, or by adding a BE protocol that can restore the previous committed 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.
Expand All @@ -525,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<Backend> beList = ctx.getStatementContext().getUsedBackendsDistributing();

Expand Down
17 changes: 12 additions & 5 deletions fe/fe-core/src/main/java/org/apache/doris/nereids/analyzer/UnboundDictionarySink.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -45,26 +46,32 @@
public class UnboundDictionarySink<CHILD_TYPE extends Plan> extends UnboundLogicalSink<CHILD_TYPE>
implements Unbound, Sink, BlockFuncDepsPropagation {

private final Database database;
private final Dictionary dictionary;
private final boolean allowAdaptiveLoad;

/**
* create unbound sink for dictionary sink
*/
public UnboundDictionarySink(Dictionary dictionary, CHILD_TYPE child, boolean adaptiveLoad) {
public UnboundDictionarySink(Database database, Dictionary dictionary, CHILD_TYPE child, boolean adaptiveLoad) {
// all the empty arguments is like UnboundTableSink
super(ImmutableList.copyOf(dictionary.getNameWithFullQualifiers().split("\\.")), // nameParts
super(ImmutableList.of(database.getCatalog().getName(), database.getFullName(), dictionary.getName()),
PlanType.LOGICAL_UNBOUND_DICTIONARY_SINK, // type
ImmutableList.of(), // outputExprs
Optional.empty(), // groupExpression
Optional.empty(), // logicalProperties
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;
}
Expand All @@ -76,7 +83,7 @@ public boolean allowAdaptiveLoad() {
@Override
public Plan withChildren(List<Plan> children) {
Preconditions.checkArgument(children.size() == 1, "UnboundDictionarySink only accepts one child");
return new UnboundDictionarySink<>(dictionary, children.get(0), allowAdaptiveLoad);
return new UnboundDictionarySink<>(database, dictionary, children.get(0), allowAdaptiveLoad);
}

@Override
Expand All @@ -91,13 +98,13 @@ public <R, C> R accept(PlanVisitor<R, C> visitor, C context) {

@Override
public Plan withGroupExpression(Optional<GroupExpression> groupExpression) {
return new UnboundDictionarySink<>(dictionary, child(), allowAdaptiveLoad);
return new UnboundDictionarySink<>(database, dictionary, child(), allowAdaptiveLoad);
}

@Override
public Plan withGroupExprLogicalPropChildren(Optional<GroupExpression> groupExpression,
Optional<LogicalProperties> logicalProperties, List<Plan> children) {
return new UnboundDictionarySink<>(dictionary, children.get(0), allowAdaptiveLoad);
return new UnboundDictionarySink<>(database, dictionary, children.get(0), allowAdaptiveLoad);
}

@Override
Expand Down
7 changes: 4 additions & 3 deletions fe/fe-core/src/main/java/org/apache/doris/nereids/analyzer/UnboundTableSinkCreator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -161,8 +162,8 @@ public static LogicalSink<? extends Plan> createUnboundTableSinkMaybeOverwrite(L
/**
* create unbound sink for dictionary sink
*/
public static UnboundDictionarySink<? extends Plan> createUnboundDictionarySink(Dictionary dictionary,
LogicalPlan child, boolean adaptiveLoad) {
return new UnboundDictionarySink<>(dictionary, child, adaptiveLoad);
public static UnboundDictionarySink<? extends Plan> createUnboundDictionarySink(Database database,
Dictionary dictionary, LogicalPlan child, boolean adaptiveLoad) {
return new UnboundDictionarySink<>(database, dictionary, child, adaptiveLoad);
}
}
18 changes: 2 additions & 16 deletions fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindSink.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -990,9 +990,8 @@ private static Map<String, NamedExpression> getConnectorColumnToOutput(

private Plan bindDictionarySink(MatchingContext<UnboundDictionarySink<Plan>> ctx) {
UnboundDictionarySink<?> sink = ctx.root;
Pair<Database, Dictionary> pair = bind(ctx.cascadesContext, sink);
Database database = pair.first;
Dictionary dictionary = pair.second;
Database database = sink.getDatabase();
Dictionary dictionary = sink.getDictionary();
LogicalPlan child = ((LogicalPlan) sink.child());

// 1. bind target columns: from sink's column names to target tables' Columns
Expand Down Expand Up @@ -1118,19 +1117,6 @@ private Pair<ExternalDatabase, PluginDrivenExternalTable> bind(CascadesContext c
throw new AnalysisException("the target table of insert into is not a plugin-driven connector table");
}

private Pair<Database, Dictionary> bind(CascadesContext cascadesContext,
UnboundDictionarySink<? extends Plan> sink) {
Dictionary dictionary = sink.getDictionary();
Database db;
try {
db = cascadesContext.getConnectContext().getEnv().getInternalCatalog()
.getDbOrAnalysisException(dictionary.getDatabase().getName());
} catch (org.apache.doris.common.AnalysisException e) {
throw new AnalysisException(e.getMessage());
}
return Pair.of(db, dictionary);
}

private List<Long> bindPartitionIds(OlapTable table, List<String> partitions, boolean temp) {
return partitions.isEmpty()
? ImmutableList.of()
Expand Down
13 changes: 12 additions & 1 deletion ...ain/java/org/apache/doris/nereids/trees/plans/commands/insert/AbstractInsertExecutor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand Down Expand Up @@ -107,8 +108,18 @@ public AbstractInsertExecutor(ConnectContext ctx, TableIf table, String labelNam
*/
public AbstractInsertExecutor(ConnectContext ctx, TableIf table, String labelName, NereidsPlanner planner,
Optional<InsertCommandContext> insertCtx, boolean emptyInsert, long jobId, boolean needRegister) {
this(ctx, table.getDatabase(), table, labelName, planner, insertCtx, emptyInsert, jobId, needRegister);

@github-actions github-actions Bot Jul 12, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This overload still lets normal insert retry attempts recover the database from the stale Table object before the new database/table ID recheck runs. In initPlan(), ExecutorFactory.build() constructs the executor before the second getTarget() comparison; the OLAP path still passes only the original targetTableIf, so this constructor calls table.getDatabase() and, with needRegister, immediately adds an InsertLoadJob. If the original database/table was dropped and recreated while planning, Table.getDatabase() resolves the old table's qualifiedDbName against the current catalog and can register a load job under the replacement database even though the retry check later sees the ID mismatch and continues. Please either pass the captured InsertTarget database into the normal executor constructors too, or delay executor construction/load-job registration until after the target identity check has proven this planned target is still current.

}

/**
* 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<InsertCommandContext> insertCtx, boolean emptyInsert, long jobId,
boolean needRegister) {
this.ctx = ctx;
this.database = table.getDatabase();
this.database = Objects.requireNonNull(database, "database should not be null");
this.insertLoadJob = new InsertLoadJob(database.getId(), labelName, jobId);
if (needRegister) {
ctx.getEnv().getLoadManager().addLoadJob(insertLoadJob);
Expand Down
8 changes: 5 additions & 3 deletions ...n/java/org/apache/doris/nereids/trees/plans/commands/insert/DictionaryInsertExecutor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -42,9 +43,10 @@ public class DictionaryInsertExecutor extends AbstractInsertExecutor {
/**
* constructor
*/
public DictionaryInsertExecutor(ConnectContext ctx, Dictionary dictionary, String labelName, NereidsPlanner planner,
Optional<InsertCommandContext> insertCtx, boolean emptyInsert, long jobId) {
super(ctx, dictionary, labelName, planner, insertCtx, emptyInsert, jobId);
public DictionaryInsertExecutor(ConnectContext ctx, DatabaseIf<?> database, Dictionary dictionary,
String labelName, NereidsPlanner planner, Optional<InsertCommandContext> insertCtx, boolean emptyInsert,
long jobId) {
super(ctx, database, dictionary, labelName, planner, insertCtx, emptyInsert, jobId, false);
}

@Override
Expand Down
15 changes: 10 additions & 5 deletions ...ava/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoDictionaryCommand.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -35,6 +35,7 @@
* logic of InsertIntoTableCommand to maximize code reuse.
*/
public class InsertIntoDictionaryCommand extends InsertIntoTableCommand {
private final Database database;
private final Dictionary dictionary;

/**
Expand All @@ -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
Expand All @@ -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);
Expand All @@ -70,8 +72,11 @@ public RedirectStatus toRedirectStatus() {
}

@Override
protected TableIf getTargetTableIf(ConnectContext ctx, List<String> qualifiedTargetTableName) {
return dictionary;
protected InsertTarget getTarget(ConnectContext ctx, List<String> qualifiedTargetTableName) {
if (!ctx.getEnv().getDictionaryManager().isCurrentDictionary(dictionary)) {
throw new AnalysisException("Dictionary " + dictionary.getName() + " has been dropped");
}
return new InsertTarget(database, dictionary);
}

public Dictionary getDictionary() {
Expand Down
Loading
Loading