github-actions[bot] commented on code in PR #65476:
URL: https://github.com/apache/doris/pull/65476#discussion_r3567664093
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTableCommand.java:
##########
@@ -256,65 +279,62 @@ public AbstractInsertExecutor initPlan(ConnectContext
ctx, StmtExecutor stmtExec
boolean needBeginTransaction)
throws Exception {
List<String> qualifiedTargetTableName =
InsertUtils.getTargetTableQualified(originLogicalQuery, ctx);
- AbstractInsertExecutor insertExecutor;
+ AbstractInsertExecutor insertExecutor = null;
int retryTimes = 0;
ctx.getStatementContext().setIsInsert(true);
while (++retryTimes <
Math.max(ctx.getSessionVariable().dmlPlanRetryTimes, 3)) {
- TableIf targetTableIf = getTargetTableIf(ctx,
qualifiedTargetTableName);
+ boolean groupCommitBeforeAttempt = ctx.isGroupCommit();
+ Backend groupCommitBackendBeforeAttempt =
+ ctx.getStatementContext().getGroupCommitMergeBackend();
+ insertExecutor = null;
+ InsertTarget target = getTarget(ctx, qualifiedTargetTableName);
+ DatabaseIf<?> targetDatabase = target.getDatabase();
+ TableIf targetTableIf = target.getTable();
// check auth
if (needAuthCheck(targetTableIf) &&
!Env.getCurrentEnv().getAccessManager()
- .checkTblPriv(ConnectContext.get(),
targetTableIf.getDatabase().getCatalog().getName(),
- targetTableIf.getDatabase().getFullName(),
targetTableIf.getName(),
+ .checkTblPriv(ConnectContext.get(),
targetDatabase.getCatalog().getName(),
+ targetDatabase.getFullName(),
targetTableIf.getName(),
PrivPredicate.LOAD)) {
ErrorReport.reportAnalysisException(ErrorCode.ERR_TABLEACCESS_DENIED_ERROR,
"LOAD",
ConnectContext.get().getQualifiedUser(),
ConnectContext.get().getRemoteIP(),
- targetTableIf.getDatabase().getFullName()
+ targetDatabase.getFullName()
+ "." +
Util.getTempTableDisplayName(targetTableIf.getName()));
}
- BuildInsertExecutorResult buildResult;
+ ExecutorFactory executorFactory;
try {
// use originLogicalQuery to build logicalQuery again.
- buildResult = initPlanOnce(ctx, stmtExecutor, targetTableIf);
+ executorFactory = initPlanOnce(ctx, stmtExecutor,
targetTableIf);
} catch (Throwable e) {
Throwables.throwIfInstanceOf(e, RuntimeException.class);
throw new IllegalStateException(e.getMessage(), e);
}
- insertExecutor = buildResult.executor;
- parsedPlan =
Optional.ofNullable(buildResult.planner.getParsedPlan());
- Plan analyzedPlan = buildResult.planner.getAnalyzedPlan();
+ parsedPlan =
Optional.ofNullable(executorFactory.planner.getParsedPlan());
+ Plan analyzedPlan = executorFactory.planner.getAnalyzedPlan();
lineagePlan = Optional.ofNullable(analyzedPlan);
- if (!needBeginTransaction) {
- return insertExecutor;
- }
- List<TableStreamUpdateInfo> infos =
StreamConsumptionInfoExtractor.extract(analyzedPlan);
+ List<TableStreamUpdateInfo> infos = needBeginTransaction
+ ? StreamConsumptionInfoExtractor.extract(analyzedPlan) :
Lists.newArrayList();
if (!infos.isEmpty()) {
if (!Config.enable_feature_binlog) {
throw new AnalysisException("Insert plan with Table stream
failed."
+ " should enable binlog feature in FE config.");
}
- // put offset into executor
- insertExecutor.setStreamUpdateInfos(infos);
- insertExecutor.registerListener(new InsertExecutorListener() {
- @Override
- public void beforeComplete(AbstractInsertExecutor
insertExecutor, StmtExecutor executor,
- long jobId) throws Exception {
- TransactionState transactionState =
Env.getCurrentGlobalTransactionMgr()
-
.getTransactionState(insertExecutor.getDatabase().getId(),
- insertExecutor.getTxnId());
-
transactionState.setStreamUpdateInfos(insertExecutor.getStreamUpdateInfos());
- }
- });
}
// lock after plan and check does table's schema changed to ensure
we lock table order by id.
- TableIf newestTargetTableIf = getTargetTableIf(ctx,
qualifiedTargetTableName);
+ InsertTarget newestTarget = getTarget(ctx,
qualifiedTargetTableName);
+ DatabaseIf<?> newestTargetDatabase = newestTarget.getDatabase();
+ TableIf newestTargetTableIf = newestTarget.getTable();
newestTargetTableIf.readLock();
try {
- if (targetTableIf.getId() != newestTargetTableIf.getId()) {
- LOG.warn("insert plan failed {} times. query id is {}.
table id changed from {} to {}",
+ 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());
+ ctx.setGroupCommit(groupCommitBeforeAttempt);
Review Comment:
This retry path still preserves another failed-attempt `StatementContext`
field. The first planning attempt runs `CheckPrivileges` and sets
`stmtCtx.privChecked=true`; if this branch discards that attempt because the
target id/schema changed, the code restores only group-commit state before
`continue`. The accepted retry then reuses the same `StatementContext`, so
`CheckPrivileges.rewriteRoot()` returns before visiting source relations,
views, CTE producers, or TVFs, leaving only the explicit target LOAD check.
Please reset per-attempt privilege state here, or use a fresh per-attempt
`StatementContext` that preserves only intentional cross-attempt fields.
##########
fe/fe-core/src/main/java/org/apache/doris/dictionary/DictionaryManager.java:
##########
@@ -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)) {
Review Comment:
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.
##########
regression-test/suites/dictionary_p0/test_create_drop_sync.groovy:
##########
@@ -105,4 +132,40 @@ suite('test_create_drop_sync') {
LAYOUT(HASH_MAP)
properties('data_lifetime'='600');
"""
-}
\ No newline at end of file
+
+ waitAllDictionariesReady()
+ long originalVersion = (sql "SHOW DICTIONARIES")[0][3].toLong()
+ def commitFailureFuture
+ try {
+
GetDebugPoint().enableDebugPointForAllFEs('DictionaryManager.dataLoad.blockBeforeCommit')
Review Comment:
This cleanup is split across two blocks, so
`DictionaryManager.commitNowVersion.forceFailure` can leak if the setup fails
before `commitFailureFuture.get()` is reached. For example, `awaitUntil(10)`
throws on timeout and the first `finally` disables only `blockBeforeCommit`,
leaving the force-failure debug point active on every FE. Since that debug
point makes every later dictionary commit return false, please disable both
debug points in the same outer `finally` that covers their
`enableDebugPointForAllFEs` calls.
##########
fe/fe-core/src/main/java/org/apache/doris/dictionary/DictionaryManager.java:
##########
@@ -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(
Review Comment:
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.
--
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]
