github-actions[bot] commented on code in PR #65476:
URL: https://github.com/apache/doris/pull/65476#discussion_r3567496468
##########
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:
This wait still does not prove the refresh thread reached
`DictionaryManager.dataLoad.blockBeforePlan`. `dataLoad()` sets `LOADING`
before the database/current-dictionary check and before the debug-point loop,
so this test can observe `LOADING`, drop the database, and have the refresh
fail at the early dropped-dictionary check with the same message asserted
below. In that case the regression passes without exercising the
captured-database planning path it is meant to cover. Please synchronize on a
signal reached after the FE debug point is actually entered, or keep this exact
race coverage in a harness that can inspect `DebugPoint.executeNum`.
##########
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:
This `needBeginTransaction=false` validation still leaves prepared
group-commit full-prepare with a stale target.
`GroupCommitPlanner.executeGroupCommitInsert()` resolves `OlapTable table =
command.getTable(ctx)` before calling `command.initPlan(..., false)`, but after
this branch returns it ignores the returned executor and builds/caches the
stream-load planner from that earlier `table` and `table.getDatabase()`. If the
database/table is dropped and recreated between those calls, `initPlan(false)`
can retry and validate the replacement target while the cached
`GroupCommitPlanner` is built from the dropped table object's schema/id. Please
have the full-prepare caller derive the database/table from the target that
survived `initPlan(false)` validation, or re-resolve and check the target ids
again after validation before constructing or reusing the cached planner.
##########
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);
+
ctx.getStatementContext().setGroupCommitMergeBackend(groupCommitBackendBeforeAttempt);
newestTargetTableIf.readUnlock();
continue;
Review Comment:
This retry still keeps the failed attempt's `StatementContext` relation
caches. Each attempt builds a new `CascadesContext` from the same
`ctx.getStatementContext()`, and `CollectRelation`/`BindRelation` populate
`tables` and `insertTargetTables` through
`getAndCacheTable(...).computeIfAbsent`; `StatementContext.close()` only
releases locks, not those maps. After `INSERT INTO db.target SELECT ... FROM
db.source` plans once, a drop/recreate can make this branch detect the old
target id and `continue`, but the next attempt can still bind `db.source` and
even `insertTargetSchema` from the stale cached objects while `BindSink`
re-resolves the replacement target. If the schemas match, the new id/schema
validation can pass with old source metadata in the plan. Please clear/recreate
the per-attempt relation state before retrying, or plan retries with a fresh
`StatementContext` while preserving only intentionally shared statement state.
--
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]
