github-actions[bot] commented on code in PR #65476:
URL: https://github.com/apache/doris/pull/65476#discussion_r3632436427
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTableCommand.java:
##########
@@ -354,6 +437,45 @@ public void beforeComplete(AbstractInsertExecutor
insertExecutor, StmtExecutor e
throw new AnalysisException("Insert plan failed. Could not get target
table lock.");
}
+ private void lockCurrentTarget(ConnectContext ctx, DatabaseIf<?> database,
TableIf table) {
+ if (!(database instanceof Database) || table instanceof Dictionary) {
+ table.readLock();
+ return;
+ }
+
+ // Internal DROP uses Database -> Table, so this order closes the
lookup-to-lock race without inversion.
+ database.readLock();
+ try {
+ boolean current = false;
+ table.readLock();
+ try {
+ Database internalDatabase = (Database) database;
+ current =
ctx.getEnv().getInternalCatalog().getDbNullable(internalDatabase.getId()) ==
internalDatabase
+ && internalDatabase.getTableNullable(table.getId()) ==
table;
+ } finally {
+ if (!current) {
+ table.readUnlock();
+ }
+ }
+ if (current) {
+ return;
+ }
+ LOG.warn("insert target changed before validation. query id is {}.
target id is {}.{}",
+ DebugUtil.printId(ctx.queryId()), database.getId(),
table.getId());
+ throw new AnalysisException("Insert target changed while planning.
Please retry the statement");
+ } finally {
+ database.readUnlock();
+ }
+ }
+
+ public GroupCommitTarget acquireGroupCommitTarget(ConnectContext ctx)
throws Exception {
+ OlapTable table = (OlapTable) getTable(ctx);
+ Database database = (Database) table.getDatabase();
Review Comment:
`getTable()` resolves and authorizes one database/table pair, but this
helper discards its database and calls `table.getDatabase()`, which is a fresh
catalog-by-name lookup. If `DROP DATABASE` completes after `getTable()`
returns, that lookup is null and `acquireGroupCommitAdmission()` immediately
dereferences it at `database.readLock()`, producing an NPE instead of the
intended retryable target-changed error. A same-name recreation also
temporarily pairs the old table with the replacement database. Please carry the
exact `InsertTarget` pair through authorization and admission, and add a
DROP-after-authorization test for this command-level path; the current
replacement test calls the manager directly and cannot catch this gap.
##########
fe/fe-core/src/main/java/org/apache/doris/dictionary/DictionaryManager.java:
##########
@@ -556,45 +645,106 @@ private boolean commitNowVersion(ConnectContext ctx,
Dictionary dictionary) {
if (status.getErrorCode() != TStatusCode.OK) {
LOG.warn("Failed to commit dictionary " +
dictionary.getId() + " on be " + be.getAddress()
+ " because " + status.getErrorMsg());
- allSucceed = false;
+ allSucceeded = false;
+ } else {
+ anySucceeded = true;
}
} else {
LOG.warn("Failed to commit dictionary " +
dictionary.getId() + " on be " + be.getAddress());
- allSucceed = false;
+ allSucceeded = false;
+ outcomeUnknown = true;
}
+ } catch (Exception e) {
+ dictionary.setLastUpdateResult("commit failed: " +
e.getMessage());
+ LOG.warn("Failed to commit dictionary {} on BE {}",
dictionary.getId(), be.getId(), e);
+ allSucceeded = false;
+ outcomeUnknown = true;
}
- } catch (Exception e) {
- dictionary.setLastUpdateResult("commit failed: " + e.getMessage());
- LOG.warn("Failed to commit dictionary " + dictionary.getId(), e);
- allSucceed = false;
}
- return allSucceed;
+ return new CommitResult(allSucceeded, anySucceeded, outcomeUnknown);
+ }
+
+ private void abortFailedLoad(ConnectContext ctx, Dictionary dictionary,
DictionaryStatus oldStatus) {
+ if (ctx.getStatementContext() == null) {
+ dictionary.trySetStatus(oldStatus);
+ return;
+ }
+ boolean partialLoad =
ctx.getStatementContext().isPartialLoadDictionary();
+ long stagedVersion = partialLoad
+ ? dictionary.getVersion() : dictionary.getVersion() + 1;
+ boolean versionFenced = false;
+ if (!partialLoad &&
ctx.getStatementContext().getUsedBackendsDistributing() != null
+ &&
!ctx.getStatementContext().getUsedBackendsDistributing().isEmpty()) {
+ // Persist the fence before abort so failover cannot reuse a
version targeted by a delayed RPC.
+ lockRead();
+ try {
+ if (isCurrentDictionary(dictionary)) {
+ dictionary.increaseVersion();
Review Comment:
This fence advances the same `Dictionary.version` that `dict_get` sends as
the committed query version. Starting from a healthy committed V, a full
refresh that fails after BE assignment increments FE to V+1 and aborts only the
refreshing entry; every BE can still hold committed V, but
`DictionaryFactory::get()` requires an exact match, so new queries ask for
absent V+1 and fail. For an `[INVALID_DICT_MARK]` failure, the source-version
marker can suppress the automatic retry and make the outage persistent. Please
keep a separate persisted attempt/high-watermark generation and leave the read
version at V until commit is known visible; add coverage that begins with
committed V and proves reads survive a failed refresh.
##########
fe/fe-core/src/main/java/org/apache/doris/dictionary/DictionaryManager.java:
##########
@@ -495,16 +512,41 @@ public void dataLoad(ConnectContext ctx, Dictionary
dictionary, boolean adaptive
}
}
+ while
(DebugPointUtil.isEnable("DictionaryManager.dataLoad.blockBeforeCommit")) {
Review Comment:
Reaching this post-load point is still the first time a full-refresh
generation has been journaled; fragments carrying V+1 were already dispatched
by `command.run()`. If the master dies during that call, replay retains V and
the new master can dispatch another V+1 while old-master fragments can still
reach EOS. Since BE `refresh_dict()` unconditionally replaces the single
refreshing slot for the dictionary ID, old EOS can overwrite newly staged S2
with S1 at the same V+1; the new FE then commits V+1 and records S2 even though
S1 was published (potentially on only a subset of BEs). Please reserve a
separate attempt generation before dispatch or transmit a persisted
master/attempt epoch that rejects stale EOS, and add an in-flight failover test
rather than only failing over after dictionaries are ready.
##########
fe/fe-core/src/main/java/org/apache/doris/load/GroupCommitManager.java:
##########
@@ -249,6 +356,10 @@ public Backend selectBackendForGroupCommit(long tableId,
ConnectContext context)
public long selectBackendForGroupCommitInternal(long tableId, String
cluster)
throws LoadException, DdlException {
+ // Followers execute group commit locally, so master-side BE selection
is their admission barrier.
+ synchronized (admissionLock) {
+ checkGroupCommitAdmission(tableId);
Review Comment:
This is only a point-in-time check, not the follower/HTTP admission barrier
the comment claims: it releases `admissionLock` without registering an active
request. A follower's eventual admission is in that FE's process-local manager,
and the HTTP redirect/plan-only routes can have no admission spanning the
returned plan/body handoff. Thus selection can succeed, then master DDL can
block, see no active request or WAL, switch/drop metadata, and finally the
paused request can reach the BE with its old plan. The added test covers only
block-before-selection. Please return/hold a master-visible lease until BE/WAL
acceptance (or execute through the master), and test selection-before-block for
follower ordinary/prepared and HTTP paths.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/OlapGroupCommitInsertExecutor.java:
##########
@@ -164,13 +168,14 @@ protected static void analyzeGroupCommit(ConnectContext
ctx, TableIf table, Logi
@Override
protected void beforeExec() {
- if
(Env.getCurrentEnv().getGroupCommitManager().isBlock(this.table.getId())) {
- String msg = "insert table " + this.table.getId() +
GroupCommitPlanner.SCHEMA_CHANGE;
- LOG.info(msg);
- throw new AnalysisException(msg);
- }
// this is used for old coordinator
this.coordinator.setGroupCommitBe(groupCommitBackend);
+ try {
+ groupCommitAdmission = Env.getCurrentEnv().getGroupCommitManager()
+ .acquireGroupCommitAdmission((Database) database,
(OlapTable) table);
Review Comment:
Ordinary local group commit reaches this admission only after `initPlan()`
has selected a BE, built the plan, and finalized the sink. If it pauses after
`initPlan()` returns, schema change/rollup can block, see no active request or
WAL, install S2, and unblock; this call then succeeds because admission
validates only the same database/table objects, while schema change mutates
that table object in place. The S1 plan is submitted after the very drain
intended to exclude it, and this path has no prepared-style generation replan.
Please acquire admission before ordinary group-commit planning/finalization, or
atomically validate a schema/block epoch here and replan, with a latch between
`initPlan()` and `beforeExec()`.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/OlapGroupCommitInsertExecutor.java:
##########
@@ -179,31 +184,47 @@ public void beginTransaction() {
@Override
protected void onComplete() {
- if (ctx.getState().getStateType() == MysqlStateType.ERR) {
- txnStatus = TransactionStatus.ABORTED;
- } else {
- txnStatus = TransactionStatus.PREPARE;
+ try {
+ if (ctx.getState().getStateType() == MysqlStateType.ERR) {
+ txnStatus = TransactionStatus.ABORTED;
+ } else {
+ txnStatus = TransactionStatus.PREPARE;
+ }
+ } finally {
+ releaseGroupCommitAdmission();
}
}
@Override
protected void onFail(Throwable t) {
- errMsg = t.getMessage() == null ? "unknown reason" : t.getMessage();
- String queryId = DebugUtil.printId(ctx.queryId());
- // if any throwable being thrown during insert operation, first we
should abort this txn
- LOG.warn("insert [{}] with query id {} failed, url={}", labelName,
queryId, coordinator.getTrackingUrl(), t);
- String firstErrorMsgPart = "";
- String urlPart = "";
- if (!Strings.isNullOrEmpty(coordinator.getFirstErrorMsg())) {
- firstErrorMsgPart =
StringUtils.abbreviate(coordinator.getFirstErrorMsg(),
- Config.first_error_msg_max_length);
- }
- if (!Strings.isNullOrEmpty(coordinator.getTrackingUrl())) {
- urlPart = coordinator.getTrackingUrl();
+ try {
+ errMsg = t.getMessage() == null ? "unknown reason" :
t.getMessage();
+ String queryId = DebugUtil.printId(ctx.queryId());
+ // if any throwable being thrown during insert operation, first we
should abort this txn
+ LOG.warn("insert [{}] with query id {} failed, url={}", labelName,
queryId,
+ coordinator.getTrackingUrl(), t);
+ String firstErrorMsgPart = "";
+ String urlPart = "";
+ if (!Strings.isNullOrEmpty(coordinator.getFirstErrorMsg())) {
+ firstErrorMsgPart =
StringUtils.abbreviate(coordinator.getFirstErrorMsg(),
+ Config.first_error_msg_max_length);
+ }
+ if (!Strings.isNullOrEmpty(coordinator.getTrackingUrl())) {
+ urlPart = coordinator.getTrackingUrl();
+ }
+
+ String finalErrorMsg = InsertUtils.getFinalErrorMsg(errMsg,
firstErrorMsgPart, urlPart);
+ ctx.getState().setError(ErrorCode.ERR_UNKNOWN_ERROR,
finalErrorMsg);
+ } finally {
+ releaseGroupCommitAdmission();
Review Comment:
This releases the admission before remote work is known quiescent.
`executeSingleInsert()` calls `onFail()` before `coordinator.close()`, and both
coordinator cancellation paths send their remote cancellation asynchronously.
After an exec timeout/lost response, DDL can therefore observe the active count
reach zero and no WAL, finish its schema/drop barrier, and only then have the
delayed old fragment reach its group-commit sink. The prepared direct-RPC
sibling has the same issue when `Future.get()` throws and closes its
try-with-resources admission. Please retain a fence until remote
cancellation/completion is acknowledged, or add a BE-side generation/tombstone
that rejects late old requests, with delayed-exec/lost-response races against
schema change and DROP.
##########
fe/fe-core/src/main/java/org/apache/doris/alter/MaterializedViewHandler.java:
##########
@@ -195,7 +195,10 @@ public void
processCreateMaterializedView(CreateMaterializedViewCommand createMv
OlapTable olapTable) throws DdlException, AnalysisException {
// wait wal delete
Env.getCurrentEnv().getGroupCommitManager().blockTable(olapTable.getId());
Review Comment:
With ref-counted blocking, this acquisition needs per-call ownership
tracking. Two concurrent CREATE MV/ADD ROLLUP requests can both increment
before taking the table lock. The winner submits a job and sets `ROLLUP`; the
loser then acquires the lock and fails `checkNormalStateForAlter()`, but its
`finally` skips `unblockTable()` because it infers ownership from the winner's
global `ROLLUP` state. When the winning job completes, it releases only one
reference, leaving the table blocked forever. The old Set semantics masked this
imbalance. Please release every failed call's own reference and transfer
ownership to a registered job explicitly, with a two-request test that verifies
the count reaches zero after the winner completes.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/InternalCatalog.java:
##########
@@ -1022,6 +1029,26 @@ private void dropTableInternal(Database db, Table table,
boolean isView, boolean
}
}
+ private List<Long> blockGroupCommitLoads(Collection<Table> tables) throws
DdlException {
+ List<Long> tableIds = tables.stream()
+ .filter(OlapTable.class::isInstance)
+ .map(Table::getId)
+ .collect(Collectors.toList());
+ // Rejecting an active load keeps DROP from waiting on remote work
while metadata locks are held.
+
tableIds.forEach(Env.getCurrentEnv().getGroupCommitManager()::blockTable);
+ for (long tableId : tableIds) {
+ if
(Env.getCurrentEnv().getGroupCommitManager().hasActiveGroupCommitLoad(tableId))
{
Review Comment:
This protects only the active-request interval, not an acknowledged async
request whose admission has closed but whose WAL is still queued. In async mode
the BE writes/registers WAL and can return success before the batched internal
transaction commits; at that point `hasActiveGroupCommitLoad()` is false while
the per-table WAL queue is nonempty. DROP then removes the table/database, and
the later `loadTxnCommit` fails and can strand the WAL in recovery. Unlike
schema change and rollup, this path never calls `waitWalFinished()`. Please
keep the block while draining both admissions and WAL outside metadata locks
and then revalidate, or reject DROP while either exists; cover both table and
database DROP with queued async WAL.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTableCommand.java:
##########
@@ -135,6 +139,50 @@ 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 static final class GroupCommitTarget implements AutoCloseable {
Review Comment:
The current head cannot pass the required FE style gate. CheckStyle reports
`MissingJavadocType` for this new public nested class and also reports the
unused local `OlapTable table` in
`GroupCommitPlannerTargetRaceTest.dropFailsWhileCacheMissPlanIsAdmitted()` at
line 129. Please add the required type Javadoc and remove/use that test
variable; run 29939849908 contains exactly these two errors.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
