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") + } + } +}