[fix](fe) Skip DEC journal when dictionary dropped during load commit

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<Backend> 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

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());
+    }
+}

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