From 267037c935bc6816baad8f7aaa1b9eaa39554199 Mon Sep 17 00:00:00 2001 From: zhaochangle Date: Thu, 6 Aug 2026 20:00:20 +0800 Subject: [PATCH] [fix](fe) Skip DEC journal when dictionary dropped during load commit ### What problem does this PR solve? Problem Summary: An async dictionary load task writes the INC version journal, then DROP deletes the dictionary, then a failed BE commit writes a DEC journal for the already dropped dictionary. Followers exit while replaying the DEC journal because the dictionary cannot be found by name anymore, and the cluster cannot recover since the bad journal is persistent. Root cause: the commit failure rollback runs outside the manager lock and does not check whether the dictionary is still current, so DROP can interleave between the INC journal and the rollback DEC journal. Replay of the DEC journal looks the dictionary up by name and throws when it is gone, which makes EditLog.loadJournal exit the FE. Fix: on commit failure, only persist the DEC journal while the dictionary is still the current one (identity check under the manager read lock), so the journal stream becomes CREATE -> INC -> DROP instead of CREATE -> INC -> DROP -> DEC. Replay of increase/decrease version journals is made idempotent and keyed by dictionary id, so a DEC journal of an already dropped or recreated dictionary is a no-op instead of a fatal error. ### Release note None ### Check List (For Author) - Test: Unit test DictionaryManagerTest (replay idempotency, journal order CREATE -> INC -> DROP -> DEC, ABA recreated same-name dictionary) and docker regression test test_dictionary_drop_while_load_commit_fail (deterministic race reproduction via debug points). - Behavior changed: No - Does this need documentation: No --- .../doris/dictionary/DictionaryManager.java | 50 ++++++-- .../dictionary/DictionaryManagerTest.java | 112 +++++++++++++++++ ...tionary_drop_while_load_commit_fail.groovy | 113 ++++++++++++++++++ 3 files changed, 264 insertions(+), 11 deletions(-) create mode 100644 fe/fe-core/src/test/java/org/apache/doris/dictionary/DictionaryManagerTest.java create mode 100644 regression-test/suites/dictionary_p0/test_dictionary_drop_while_load_commit_fail.groovy diff --git a/fe/fe-core/src/main/java/org/apache/doris/dictionary/DictionaryManager.java b/fe/fe-core/src/main/java/org/apache/doris/dictionary/DictionaryManager.java index fb57038afb1002..b389f4756c1efc 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/dictionary/DictionaryManager.java +++ b/fe/fe-core/src/main/java/org/apache/doris/dictionary/DictionaryManager.java @@ -536,11 +536,24 @@ public void dataLoad(ConnectContext ctx, Dictionary dictionary, boolean adaptive } } + // block here in test to simulate the race: INC journal written, commit not done yet. + while (DebugPointUtil.isEnable("DictionaryManager.afterIncJournal")) { + Thread.sleep(100); + } + // commit and check the result. not modify metadata so dont need lock. if (!commitNowVersion(ctx, dictionary)) { if (!ctx.getStatementContext().isPartialLoadDictionary()) { dictionary.decreaseVersion(); - Env.getCurrentEnv().getEditLog().logDictionaryDecVersion(dictionary); + // DROP may have removed the dictionary between the INC journal and this failed + // commit. A DEC journal for a dropped dictionary cannot be replayed by name, so + // only persist the rollback while the dictionary is still the current one. + if (isCurrentDictionary(database, dictionary)) { + Env.getCurrentEnv().getEditLog().logDictionaryDecVersion(dictionary); + } else { + LOG.warn("Dictionary {} has been dropped or replaced during commit, skip DEC journal", + dictionary.getName()); + } } dictionary.trySetStatus(oldStatus); abortSpecificVersion(ctx, dictionary, dictionary.getVersion() + 1); @@ -566,6 +579,9 @@ public void dataLoad(ConnectContext ctx, Dictionary dictionary, boolean adaptive } private boolean commitNowVersion(ConnectContext ctx, Dictionary dictionary) { + if (DebugPointUtil.isEnable("DictionaryManager.commitNowVersion.fail")) { + return false; + } // use the same BEs when we get before start loading. List beList = ctx.getStatementContext().getUsedBackendsDistributing(); @@ -861,21 +877,33 @@ public void replayDropDictionary(DropDictionaryPersistInfo info) { } public void replayIncreaseVersion(DictionaryIncreaseVersionInfo info) throws DdlException { - String dbName = info.getDictionary().getDbName(); - String dictName = info.getDictionary().getName(); - Dictionary dictionary = getDictionary(dbName, dictName); + long dictId = info.getDictionary().getId(); + Dictionary dictionary = getDictionary(dictId); + if (dictionary == null) { + LOG.warn("Dictionary with id {} does not exist when replaying increase version, skip", dictId); + return; + } dictionary.writeLock(); - dictionary.increaseVersion(); - dictionary.writeUnlock(); + try { + dictionary.increaseVersion(); + } finally { + dictionary.writeUnlock(); + } } public void replayDecreaseVersion(DictionaryDecreaseVersionInfo info) throws DdlException { - String dbName = info.getDictionary().getDbName(); - String dictName = info.getDictionary().getName(); - Dictionary dictionary = getDictionary(dbName, dictName); + long dictId = info.getDictionary().getId(); + Dictionary dictionary = getDictionary(dictId); + if (dictionary == null) { + LOG.warn("Dictionary with id {} does not exist when replaying decrease version, skip", dictId); + return; + } dictionary.writeLock(); - dictionary.decreaseVersion(); - dictionary.writeUnlock(); + try { + dictionary.decreaseVersion(); + } finally { + dictionary.writeUnlock(); + } } // Metadata serialization diff --git a/fe/fe-core/src/test/java/org/apache/doris/dictionary/DictionaryManagerTest.java b/fe/fe-core/src/test/java/org/apache/doris/dictionary/DictionaryManagerTest.java new file mode 100644 index 00000000000000..4fdb9a4ee38202 --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/dictionary/DictionaryManagerTest.java @@ -0,0 +1,112 @@ +// 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.dictionary; + +import org.apache.doris.persist.CreateDictionaryPersistInfo; +import org.apache.doris.persist.DictionaryDecreaseVersionInfo; +import org.apache.doris.persist.DictionaryIncreaseVersionInfo; +import org.apache.doris.persist.DropDictionaryPersistInfo; +import org.apache.doris.persist.gson.GsonUtils; + +import org.junit.Assert; +import org.junit.Test; + +/** + * Tests for dictionary version journal replay robustness. + * + * The crash in production: an async data load task writes the INC journal, then DROP removes the + * dictionary, then the failed commit writes a DEC journal for the already-dropped dictionary. + * Followers crash at replay because the dictionary cannot be found by name anymore. + * Replay must be idempotent and lookup by dictionary id. + */ +public class DictionaryManagerTest { + + private DictionaryManager createManager() { + return new DictionaryManager(); + } + + private Dictionary buildDictionary(long id, String dbName, String dictName, long version) { + String json = String.format( + "{\"clazz\":\"Dictionary\",\"id\":%d,\"name\":\"%s\",\"dbName\":\"%s\"," + + "\"sourceTableName\":\"src_%s\",\"version\":%d}", + id, dictName, dbName, dbName, version); + return GsonUtils.GSON.fromJson(json, Dictionary.class); + } + + @Test + public void testReplayDecreaseVersionMissingDictionary() throws Exception { + DictionaryManager manager = createManager(); + // dictionary never created on this FE + Dictionary dict = buildDictionary(1001, "db1", "dic1", 2); + manager.replayDecreaseVersion(new DictionaryDecreaseVersionInfo(dict)); + } + + @Test + public void testReplayIncreaseVersionMissingDictionary() throws Exception { + DictionaryManager manager = createManager(); + Dictionary dict = buildDictionary(1001, "db1", "dic1", 1); + manager.replayIncreaseVersion(new DictionaryIncreaseVersionInfo(dict)); + } + + @Test + public void testReplayDecreaseVersionAfterDrop() throws Exception { + DictionaryManager manager = createManager(); + Dictionary dict = buildDictionary(1001, "db1", "dic1", 2); + manager.replayCreateDictionary(new CreateDictionaryPersistInfo(dict)); + manager.replayDropDictionary(new DropDictionaryPersistInfo("db1", "dic1")); + + // journal order CREATE -> INC -> DROP -> DEC, DEC must be a no-op, not an exception + manager.replayDecreaseVersion(new DictionaryDecreaseVersionInfo(dict)); + Assert.assertNull(manager.getDictionary(1001)); + } + + @Test + public void testReplayDecreaseVersionAbA() throws Exception { + DictionaryManager manager = createManager(); + Dictionary oldDict = buildDictionary(1001, "db1", "dic1", 2); + manager.replayCreateDictionary(new CreateDictionaryPersistInfo(oldDict)); + manager.replayDropDictionary(new DropDictionaryPersistInfo("db1", "dic1")); + Dictionary newDict = buildDictionary(1002, "db1", "dic1", 1); + manager.replayCreateDictionary(new CreateDictionaryPersistInfo(newDict)); + + // DEC of the dropped dictionary must not affect the recreated same-name dictionary + manager.replayDecreaseVersion(new DictionaryDecreaseVersionInfo(oldDict)); + Assert.assertEquals(1, newDict.getVersion()); + Assert.assertEquals(1, manager.getDictionary(1002).getVersion()); + } + + @Test + public void testReplayDecreaseVersionNormal() throws Exception { + DictionaryManager manager = createManager(); + Dictionary dict = buildDictionary(1001, "db1", "dic1", 2); + manager.replayCreateDictionary(new CreateDictionaryPersistInfo(dict)); + + manager.replayDecreaseVersion(new DictionaryDecreaseVersionInfo(dict)); + Assert.assertEquals(1, manager.getDictionary(1001).getVersion()); + } + + @Test + public void testReplayIncreaseVersionNormal() throws Exception { + DictionaryManager manager = createManager(); + Dictionary dict = buildDictionary(1001, "db1", "dic1", 1); + manager.replayCreateDictionary(new CreateDictionaryPersistInfo(dict)); + + manager.replayIncreaseVersion(new DictionaryIncreaseVersionInfo(dict)); + Assert.assertEquals(2, manager.getDictionary(1001).getVersion()); + } +} diff --git a/regression-test/suites/dictionary_p0/test_dictionary_drop_while_load_commit_fail.groovy b/regression-test/suites/dictionary_p0/test_dictionary_drop_while_load_commit_fail.groovy new file mode 100644 index 00000000000000..77e117a4866a06 --- /dev/null +++ b/regression-test/suites/dictionary_p0/test_dictionary_drop_while_load_commit_fail.groovy @@ -0,0 +1,113 @@ +// 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. + +import org.apache.doris.regression.suite.ClusterOptions + +// Regress the race reported in DORIS-27820: an async dictionary load task writes the INC version +// journal, then DROP deletes the dictionary, then the failed BE commit writes a DEC journal for +// the already dropped dictionary, making all followers exit when replaying it. +// +// With the fix, the DEC journal is skipped when the dictionary was dropped during commit, and +// replay of DEC journals of dropped dictionaries is a no-op, so the whole journal stream +// (CREATE -> INC -> DROP) replays cleanly and FEs stay healthy after restart. +suite('test_dictionary_drop_while_load_commit_fail', 'docker') { + def options = new ClusterOptions() + options.cloudMode = false + options.feNum = 3 + options.beNum = 1 + options.enableDebugPoints() + + docker(options) { + sql "drop database if exists test_dictionary_drop_race" + sql "create database test_dictionary_drop_race" + sql "use test_dictionary_drop_race" + + sql """ + create table source_table( + k1 varchar(100) not null, + v1 int not null + ) + DISTRIBUTED BY HASH(`k1`) BUCKETS 1 + properties("replication_num" = "1"); + """ + sql "insert into source_table values ('k1', 1), ('k2', 2), ('k3', 3)" + + // Block the load task right after the INC journal is written, before BE commit. + GetDebugPoint().enableDebugPointForAllFEs("DictionaryManager.afterIncJournal") + try { + sql """ + create dictionary dict1 using source_table + ( + k1 KEY, + v1 VALUE + )LAYOUT(HASH_MAP) + properties('data_lifetime'='600'); + """ + + // wait until the load task is parked at the block point. + // status is LOADING before the block, and the task cannot pass the block, so once we + // observe LOADING for a grace period, INC journal is guaranteed already written. + boolean loading = false + for (int i = 0; i < 40; i++) { + def res = sql "SHOW DICTIONARIES" + if (res.size() == 1 && res[0][4] == "LOADING") { + loading = true + break + } + sleep(500) + } + assertTrue(loading) + sleep(1500) + + // DROP the dictionary while the load task is between INC journal and commit + sql "drop dictionary dict1" + def dictRes = sql "SHOW DICTIONARIES" + assertEquals(dictRes.size(), 0) + + // force the BE commit to fail, then release the blocked load task + GetDebugPoint().enableDebugPointForAllFEs("DictionaryManager.commitNowVersion.fail") + GetDebugPoint().disableDebugPointForAllFEs("DictionaryManager.afterIncJournal") + sleep(3000) + + // master must stay healthy: no DEC journal should have been written after DROP + assertTrue(cluster.getMasterFe().alive) + + // restart the master: it must replay the whole journal stream without exit. + // if a DEC journal for the dropped dictionary had been written, replay would throw + // and the FE would never come back alive. + def master = cluster.getMasterFe() + cluster.restartFrontends(master.index) + boolean hasRestart = false + for (int i = 0; i < 60; i++) { + if (cluster.getFeByIndex(master.index).alive) { + hasRestart = true + break + } + sleep(1000) + } + assertTrue(hasRestart) + + context.reconnectFe() + sql "use test_dictionary_drop_race" + def finalRes = sql "SHOW DICTIONARIES" + assertEquals(finalRes.size(), 0) + } finally { + GetDebugPoint().disableDebugPointForAllFEs("DictionaryManager.commitNowVersion.fail") + GetDebugPoint().disableDebugPointForAllFEs("DictionaryManager.afterIncJournal") + } + } +}