Reject duplicate schema region IDs during recovery (#18754)

Author: Caideyipi

iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeSchemaMessages.java

@@ -44,6 +44,9 @@ public final class DataNodeSchemaMessages {
   public static final String PEER_IS_SHUTTING_DOWN = "Peer is shutting down now.";
   public static final String SCHEMA_REGION_DUPLICATED =
       "SchemaRegion [%s] is duplicated between [%s] and [%s], and the former one has been recovered.";
+  public static final String
+      EXCEPTION_CANNOT_RECOVER_DUPLICATED_SCHEMAREGION_ARG_FOUND_IN_DATABASES_ARG_AND_ARG_D9F60E05 =
+          "Cannot recover duplicated SchemaRegion [%s] found in databases [%s] and [%s].";
 
   // ======================== MemSchemaEngineStatistics ========================
 

iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeSchemaMessages.java

@@ -42,6 +42,9 @@ public final class DataNodeSchemaMessages {
   public static final String PEER_IS_SHUTTING_DOWN = "节点正在关闭中。";
   public static final String SCHEMA_REGION_DUPLICATED =
       "SchemaRegion [%s] 在 [%s] 和 [%s] 之间重复，前者已被恢复。";
+  public static final String
+      EXCEPTION_CANNOT_RECOVER_DUPLICATED_SCHEMAREGION_ARG_FOUND_IN_DATABASES_ARG_AND_ARG_D9F60E05 =
+          "无法恢复重复的 SchemaRegion [%s]，其同时存在于数据库 [%s] 和 [%s] 中。";
 
   // ======================== MemSchemaEngineStatistics 相关消息 ========================
 

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/SchemaEngine.java

@@ -184,6 +184,7 @@ public static Map<String, List<SchemaRegionId>> getLocalSchemaRegionInfo() {
   private void initSchemaRegion() {
     // recover SchemaRegion concurrently
     final Map<String, List<SchemaRegionId>> localSchemaRegionInfo = getLocalSchemaRegionInfo();
+    validateNoDuplicatedSchemaRegionId(localSchemaRegionInfo);
     final ExecutorService schemaRegionRecoverPools =
         IoTDBThreadPoolFactory.newFixedThreadPool(
             Runtime.getRuntime().availableProcessors(),
@@ -207,6 +208,27 @@ private void initSchemaRegion() {
     schemaRegionRecoverPools.shutdown();
   }
 
+  static void validateNoDuplicatedSchemaRegionId(
+      final Map<String, List<SchemaRegionId>> localSchemaRegionInfo) {
+    final Map<SchemaRegionId, String> databaseBySchemaRegionId = new HashMap<>();
+    localSchemaRegionInfo.forEach(
+        (database, schemaRegionIds) -> {
+          for (final SchemaRegionId schemaRegionId : schemaRegionIds) {
+            final String existingDatabase =
+                databaseBySchemaRegionId.putIfAbsent(schemaRegionId, database);
+            if (existingDatabase != null) {
+              throw new IllegalStateException(
+                  String.format(
+                      DataNodeSchemaMessages
+                          .EXCEPTION_CANNOT_RECOVER_DUPLICATED_SCHEMAREGION_ARG_FOUND_IN_DATABASES_ARG_AND_ARG_D9F60E05,
+                      schemaRegionId,
+                      existingDatabase,
+                      database));
+            }
+          }
+        });
+  }
+
   private void initSchemaEngineStatistics() {
     if (CommonDescriptor.getInstance().getConfig().getSchemaEngineMode().equals("Memory")) {
       schemaEngineStatistics = new MemSchemaEngineStatistics();
@@ -306,7 +328,6 @@ private Callable<ISchemaRegion> recoverSchemaRegionTask(
     return () -> {
       long timeRecord = System.currentTimeMillis();
       try {
-        // TODO: handle duplicated regionId across different database
         ISchemaRegion schemaRegion =
             createSchemaRegionWithoutExistenceCheck(storageGroup, schemaRegionId);
         timeRecord = System.currentTimeMillis() - timeRecord;

iotdb-core/datanode/src/test/java/org/apache/iotdb/db/schemaengine/SchemaEngineTest.java

@@ -0,0 +1,57 @@
+/*
+ * 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.iotdb.db.schemaengine;
+
+import org.apache.iotdb.commons.consensus.SchemaRegionId;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class SchemaEngineTest {
+
+  @Test
+  public void uniqueSchemaRegionIdsPassRecoveryValidation() {
+    final Map<String, List<SchemaRegionId>> localSchemaRegionInfo = new HashMap<>();
+    localSchemaRegionInfo.put(
+        "root.db1", Arrays.asList(new SchemaRegionId(0), new SchemaRegionId(1)));
+    localSchemaRegionInfo.put("root.db2", Collections.singletonList(new SchemaRegionId(2)));
+
+    SchemaEngine.validateNoDuplicatedSchemaRegionId(localSchemaRegionInfo);
+  }
+
+  @Test
+  public void duplicatedSchemaRegionIdIsRejectedBeforeRecovery() {
+    final Map<String, List<SchemaRegionId>> localSchemaRegionInfo = new HashMap<>();
+    localSchemaRegionInfo.put("root.db1", Collections.singletonList(new SchemaRegionId(0)));
+    localSchemaRegionInfo.put("root.db2", Collections.singletonList(new SchemaRegionId(0)));
+
+    final IllegalStateException exception =
+        Assert.assertThrows(
+            IllegalStateException.class,
+            () -> SchemaEngine.validateNoDuplicatedSchemaRegionId(localSchemaRegionInfo));
+    Assert.assertTrue(exception.getMessage().contains("root.db1"));
+    Assert.assertTrue(exception.getMessage().contains("root.db2"));
+  }
+}