github-actions[bot] commented on code in PR #65476:
URL: https://github.com/apache/doris/pull/65476#discussion_r3630030730
##########
fe/fe-core/src/main/java/org/apache/doris/dictionary/DictionaryManager.java:
##########
@@ -556,45 +643,109 @@ 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);
}
- // abort could to all BE. swallow any failures.
- private void abortSpecificVersion(ConnectContext ctx, Dictionary
dictionary, long versionId) {
+ 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();
+
Env.getCurrentEnv().getEditLog().logDictionaryIncVersion(dictionary);
+ versionFenced = true;
+ }
+ } finally {
+ unlockRead();
+ }
+ }
+
+ AbortResult abortResult = abortSpecificVersion(ctx, dictionary,
stagedVersion);
+ lockRead();
+ try {
+ if (isCurrentDictionary(dictionary)) {
+ if (versionFenced && !abortResult.outcomeUnknown) {
+ dictionary.decreaseVersion();
Review Comment:
This rollback is unsafe even when every abort RPC returns OK, because the
acknowledgement does not fence the asynchronously cancelled sink work.
`Coordinator.cancelInternal()` sends fragment cancellation asynchronously and
immediately releases its latches, so a BE can handle the abort before its
dictionary sink reaches EOS, return OK because no refreshing entry exists, and
then stage the old attempt at `V+1` afterward. FE now decrements to `V`, so a
retry reuses `V+1`; if the retry stages fresh data first and the old EOS
arrives next, `refresh_dict()` unconditionally overwrites the retry's entry and
the retry commit publishes the old snapshot while FE records the new source
version. Please keep the generation fenced until remote work is known
quiescent, or add a BE-side abort tombstone/generation check that rejects a
late stage, and cover the abort-before-EOS ordering in a lifecycle test.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTableCommand.java:
##########
@@ -354,6 +411,37 @@ 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();
Review Comment:
This fallback locks only the newest external wrapper; it never proves that
it is the same target generation used to build the plan. External
database/table cache invalidation rebuilds objects with deterministic
`Util.genIdByName` IDs, so a same-name/same-schema replacement passes the
ID/schema checks above. For remote Doris, the first wrapper has already
produced a `RemoteOlapTable` containing the old partition/tablet graph (only
its local table ID is rewritten), while this code locks the replacement wrapper
and the executor/data sink still retain the old graph. A concurrent remote
drop/recreate can therefore begin the transaction by the replacement name but
route it with stale tablet metadata. Please validate an external cache
generation/current object and reject or replan when it changed, and add a
same-schema external refresh/recreate race test.
##########
fe/fe-core/src/main/java/org/apache/doris/planner/GroupCommitPlanner.java:
##########
@@ -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;
+ OlapTable currentTable = (OlapTable) command.getTable(ctx);
+ checkGroupCommitBlocked(currentTable);
Review Comment:
This check is only a point-in-time read; prepared group commit never reaches
`OlapGroupCommitInsertExecutor.beforeExec()` for another admission check. A
schema change can block the table after this line, let `waitWalFinished()`
observe zero WAL, and then this method can select a backend and submit its
cached pre-change plan after that drain decision. Cache misses have the same
gap because `initPlan(false)` releases its final table lock before
`GroupCommitPlanner` is built and submitted, and DROP can use that window too.
Please couple prepared request admission/in-flight accounting atomically with
the block-and-drain protocol for both cache hits and misses (another unlocked
lookup/check will still race), and add deterministic races after cache
validation and after `initPlan(false)` returns.
--
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]
