zclllyybb commented on code in PR #65476:
URL: https://github.com/apache/doris/pull/65476#discussion_r3632149385
##########
fe/fe-core/src/main/java/org/apache/doris/dictionary/DictionaryManager.java:
##########
@@ -464,8 +491,7 @@ public void dataLoad(ConnectContext ctx, Dictionary
dictionary, boolean adaptive
lockRead();
boolean unlocked = false;
try {
- if (!dictionaryIds.containsKey(dictionary.getDbName())
- ||
!dictionaryIds.get(dictionary.getDbName()).containsKey(dictionary.getName())) {
+ if (!isCurrentDictionaryWithoutLock(dictionary)) {
Review Comment:
Fixed in 566e7aa2c88. The failure path no longer logs a version decrement.
Once a full load may have staged V+1, FE persists V+1 as a monotonic fence and
leaves the dictionary OUT_OF_DATE, so replay cannot see a post-drop decrement
and that version is never reused.
##########
fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertTargetDropRaceTest.java:
##########
@@ -0,0 +1,182 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.nereids.trees.plans.commands.insert;
+
+import org.apache.doris.analysis.UserIdentity;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.common.Config;
+import org.apache.doris.common.util.DebugPointUtil;
+import org.apache.doris.dictionary.Dictionary;
+import org.apache.doris.dictionary.Dictionary.DictionaryStatus;
+import org.apache.doris.nereids.StatementContext;
+import org.apache.doris.nereids.parser.NereidsParser;
+import org.apache.doris.nereids.trees.plans.PlanType;
+import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.qe.OriginStatement;
+import org.apache.doris.qe.StmtExecutor;
+import org.apache.doris.utframe.TestWithFeService;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.BooleanSupplier;
+
+class InsertTargetDropRaceTest extends TestWithFeService {
+ private static final String DICTIONARY_BLOCK_POINT =
"DictionaryManager.dataLoad.blockBeforePlan";
+ private boolean debugPointsEnabled;
+
+ @BeforeEach
+ void saveDebugPointConfig() {
+ debugPointsEnabled = Config.enable_debug_points;
+ }
+
+ @AfterEach
+ void clearDebugPoints() {
+ DebugPointUtil.clearDebugPoints();
+ Config.enable_debug_points = debugPointsEnabled;
+ }
+
+ @Test
+ void
normalInsertReturnsAnalysisErrorWhenDatabaseIsDroppedAfterTargetResolution()
throws Exception {
+ String dbName = "normal_insert_drop_race";
+ createDatabaseAndUse(dbName);
+ createTable("CREATE TABLE target_table (id INT) DISTRIBUTED BY
HASH(id) BUCKETS 1 "
+ + "PROPERTIES ('replication_num' = '1')");
+
+ String sql = "INSERT INTO " + dbName + ".target_table VALUES (1)";
+ InsertIntoTableCommand baseCommand = (InsertIntoTableCommand) new
NereidsParser().parseSingle(sql);
+ CountDownLatch targetResolved = new CountDownLatch(1);
+ CountDownLatch resumeInsert = new CountDownLatch(1);
+ AtomicBoolean blockOnce = new AtomicBoolean(true);
+ InsertIntoTableCommand command = new
InsertIntoTableCommand(baseCommand,
+ PlanType.INSERT_INTO_TABLE_COMMAND) {
+ @Override
+ protected InsertTarget getTarget(ConnectContext ctx, List<String>
qualifiedTargetTableName) {
+ InsertTarget target = super.getTarget(ctx,
qualifiedTargetTableName);
+ if (blockOnce.compareAndSet(true, false)) {
+ targetResolved.countDown();
+ await(resumeInsert);
+ }
+ return target;
+ }
+ };
+
+ ExecutorService executorService = Executors.newSingleThreadExecutor();
+ try {
+ Future<Throwable> result = executorService.submit(() ->
runInitPlan(command, sql, dbName));
+ Assertions.assertTrue(targetResolved.await(10, TimeUnit.SECONDS));
+ Env.getCurrentInternalCatalog().dropDb(dbName, false, true);
+ resumeInsert.countDown();
+
+ Throwable failure = result.get(10, TimeUnit.SECONDS);
+ Assertions.assertNotNull(failure);
+ Assertions.assertTrue(hasCause(failure,
org.apache.doris.nereids.exceptions.AnalysisException.class),
+ failure.toString());
+ Assertions.assertFalse(hasCause(failure,
NullPointerException.class), failure.toString());
+ } finally {
+ resumeInsert.countDown();
+ executorService.shutdownNow();
+ }
+ }
+
+ @Test
+ void dictionaryLoadStopsCleanlyWhenDatabaseIsDroppedBeforePlanning()
throws Exception {
+ Config.enable_debug_points = true;
+ DebugPointUtil.addDebugPoint(DICTIONARY_BLOCK_POINT);
+
+ String dbName = "dictionary_insert_drop_race";
+ createDatabaseAndUse(dbName);
+ createTable("CREATE TABLE source_table (id INT NOT NULL, city
VARCHAR(32) NOT NULL) "
+ + "DISTRIBUTED BY HASH(id) BUCKETS 1 PROPERTIES
('replication_num' = '1')");
+ executeNereidsSql("CREATE DICTIONARY dic1 USING source_table (city
KEY, id VALUE) "
+ + "LAYOUT(HASH_MAP) PROPERTIES ('data_lifetime' = '600')");
+
+ Dictionary dictionary =
Env.getCurrentEnv().getDictionaryManager().getDictionary(dbName, "dic1");
+ try {
+ await(() -> dictionary.getStatus() == DictionaryStatus.LOADING);
Review Comment:
Fixed. The FE race test now waits until the DictionaryManager debug point
itself has executed before dropping the database, so it exercises the
captured-target planning window instead of merely observing LOADING.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/AbstractInsertExecutor.java:
##########
@@ -107,8 +108,18 @@ public AbstractInsertExecutor(ConnectContext ctx, TableIf
table, String labelNam
*/
public AbstractInsertExecutor(ConnectContext ctx, TableIf table, String
labelName, NereidsPlanner planner,
Optional<InsertCommandContext> insertCtx, boolean emptyInsert,
long jobId, boolean needRegister) {
+ this(ctx, table.getDatabase(), table, labelName, planner, insertCtx,
emptyInsert, jobId, needRegister);
Review Comment:
Fixed. Executor construction now receives the database resolved together
with the validated table target; it no longer recovers a database through the
stale Table object. Target identity is checked under the database lock before
the executor and load job are created.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTableCommand.java:
##########
@@ -308,13 +333,17 @@ public void beforeComplete(AbstractInsertExecutor
insertExecutor, StmtExecutor e
}
// 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()
Review Comment:
Fixed. Rejected planning attempts restore group-commit state, and an
identity change now fails the statement instead of retargeting the same
StatementContext. Schema-only retries reset the per-attempt privilege state
before replanning.
##########
fe/fe-core/src/main/java/org/apache/doris/dictionary/DictionaryManager.java:
##########
@@ -464,8 +491,7 @@ public void dataLoad(ConnectContext ctx, Dictionary
dictionary, boolean adaptive
lockRead();
boolean unlocked = false;
try {
- if (!dictionaryIds.containsKey(dictionary.getDbName())
- ||
!dictionaryIds.get(dictionary.getDbName()).containsKey(dictionary.getName())) {
+ if (!isCurrentDictionaryWithoutLock(dictionary)) {
Review Comment:
Fixed. Cleanup computes the staged version from the load type: partial
refresh aborts the current version, while full refresh aborts the incremented
fenced version. Failed partial refreshes are marked OUT_OF_DATE so the next
refresh is full.
##########
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:
Fixed by removing identity retargeting from the retry loop. If the target
database or table ID changes, planning fails and the caller retries with a
fresh statement; only same-object schema changes retry in place, so cached
source and target relations cannot cross object generations.
##########
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.
--
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]
