zclllyybb commented on code in PR #65476:
URL: https://github.com/apache/doris/pull/65476#discussion_r3632151870
##########
regression-test/suites/dictionary_p0/test_create_drop_sync.groovy:
##########
@@ -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) {
Review Comment:
Fixed. The exact captured-database race is now covered by a deterministic FE
unit test that waits for execution of
DictionaryManager.dataLoad.blockBeforePlan. The regression test was changed to
cover the separate post-staging commit/drop race with blockBeforeCommit.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTableCommand.java:
##########
@@ -343,6 +383,9 @@ public void beforeComplete(AbstractInsertExecutor
insertExecutor, StmtExecutor e
Throwables.throwIfInstanceOf(e, RuntimeException.class);
throw new IllegalStateException(e.getMessage(), e);
}
+ if (!needBeginTransaction) {
Review Comment:
Fixed. Prepared group commit now acquires a target admission containing the
exact current Database and OlapTable, builds or validates the cached planner
against that table object and base schema version, and holds admission through
the BE request.
##########
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:
Fixed. Identity replacement no longer retries with the same
StatementContext. The remaining same-object schema retry explicitly clears
privChecked, so CheckPrivileges revisits the accepted attempt.
##########
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:
Fixed. A failed full commit with any participating BE advances and persists
the FE version as a fence and leaves the dictionary OUT_OF_DATE. FE does not
decrement or reuse that version even when commit or abort responses are mixed
or lost.
##########
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:
Fixed. Exceptions and error-state exits after dictionary sink execution now
abort the exact staged version before returning. The cleanup also records
unknown abort outcomes and applies the same monotonic full-load fence.
--
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]
