[To dev/1.3][remove datanode] Add Remove DataNode SQL #14761

apache · Jan 23, 2025 · 1c1fa79 · 1c1fa79
1 parent 632026c
commit 1c1fa79
Show file tree

Hide file tree

Showing 17 changed files with 378 additions and 23 deletions.
diff --git a/...st/java/org/apache/iotdb/confignode/it/removedatanode/IoTDBRemoveDataNodeITFramework.java b/...st/java/org/apache/iotdb/confignode/it/removedatanode/IoTDBRemoveDataNodeITFramework.java
@@ -29,6 +29,7 @@
 import org.apache.iotdb.it.env.EnvFactory;
 import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper;
 import org.apache.iotdb.itbase.exception.InconsistentDataException;
+import org.apache.iotdb.jdbc.IoTDBSQLException;
 import org.apache.iotdb.rpc.TSStatusCode;
 
 import org.apache.thrift.TException;
@@ -57,7 +58,7 @@
 public class IoTDBRemoveDataNodeITFramework {
   private static final Logger LOGGER =
       LoggerFactory.getLogger(IoTDBRemoveDataNodeITFramework.class);
-  private static final String INSERTION1 =
+  private static final String TREE_MODEL_INSERTION =
       "INSERT INTO root.sg.d1(timestamp,speed,temperature) values(100, 1, 2)";
 
   private static final String SHOW_REGIONS = "show regions";
@@ -90,7 +91,8 @@ public void successTest(
       final int dataNodeNum,
       final int removeDataNodeNum,
       final int dataRegionPerDataNode,
-      final boolean rejoinRemovedDataNode)
+      final boolean rejoinRemovedDataNode,
+      final SQLModel model)
       throws Exception {
     testRemoveDataNode(
         dataReplicateFactor,
@@ -100,7 +102,8 @@ public void successTest(
         removeDataNodeNum,
         dataRegionPerDataNode,
         true,
-        rejoinRemovedDataNode);
+        rejoinRemovedDataNode,
+        model);
   }
 
   public void failTest(
@@ -110,7 +113,8 @@ public void failTest(
       final int dataNodeNum,
       final int removeDataNodeNum,
       final int dataRegionPerDataNode,
-      final boolean rejoinRemovedDataNode)
+      final boolean rejoinRemovedDataNode,
+      final SQLModel model)
       throws Exception {
     testRemoveDataNode(
         dataReplicateFactor,
@@ -120,7 +124,8 @@ public void failTest(
         removeDataNodeNum,
         dataRegionPerDataNode,
         false,
-        rejoinRemovedDataNode);
+        rejoinRemovedDataNode,
+        model);
   }
 
   public void testRemoveDataNode(
@@ -131,7 +136,8 @@ public void testRemoveDataNode(
       final int removeDataNodeNum,
       final int dataRegionPerDataNode,
       final boolean expectRemoveSuccess,
-      final boolean rejoinRemovedDataNode)
+      final boolean rejoinRemovedDataNode,
+      final SQLModel model)
       throws Exception {
     // Set up the environment
     EnvFactory.getEnv()
@@ -148,8 +154,8 @@ public void testRemoveDataNode(
         SyncConfigNodeIServiceClient client =
             (SyncConfigNodeIServiceClient) EnvFactory.getEnv().getLeaderConfigNodeConnection()) {
 
-      // Insert data
-      statement.execute(INSERTION1);
+      // Insert data in tree model
+      statement.execute(TREE_MODEL_INSERTION);
 
       Map<Integer, Set<Integer>> regionMap = getDataRegionMap(statement);
       regionMap.forEach(
@@ -187,21 +193,39 @@ public void testRemoveDataNode(
               .map(TDataNodeConfiguration::getLocation)
               .filter(location -> removeDataNodes.contains(location.getDataNodeId()))
               .collect(Collectors.toList());
-      TDataNodeRemoveReq removeReq = new TDataNodeRemoveReq(removeDataNodeLocations);
+      if (SQLModel.NOT_USE_SQL.equals(model)) {
+        TDataNodeRemoveReq removeReq = new TDataNodeRemoveReq(removeDataNodeLocations);
+
+        // Remove data nodes
+        TDataNodeRemoveResp removeResp = clientRef.get().removeDataNode(removeReq);
+        LOGGER.info("Submit Remove DataNodes result {} ", removeResp);
+        if (removeResp.getStatus().getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+          if (expectRemoveSuccess) {
+            LOGGER.error("Submit Remove DataNodes fail");
+            Assert.fail();
+          } else {
+            LOGGER.info("Submit Remove DataNodes fail, as expected.");
+            return;
+          }
+        }
+        LOGGER.info("Submit Remove DataNodes request: {}", removeReq);
 
-      // Remove data nodes
-      TDataNodeRemoveResp removeResp = clientRef.get().removeDataNode(removeReq);
-      LOGGER.info("Submit Remove DataNodes result {} ", removeResp);
-      if (removeResp.getStatus().getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
-        if (expectRemoveSuccess) {
-          LOGGER.error("Submit Remove DataNodes fail");
-          Assert.fail();
-        } else {
-          LOGGER.info("Submit Remove DataNodes fail, as expected");
-          return;
+      } else {
+        String removeDataNodeSQL = generateRemoveString(removeDataNodes);
+        LOGGER.info("Remove DataNodes SQL: {}", removeDataNodeSQL);
+        try {
+          statement.execute(removeDataNodeSQL);
+        } catch (IoTDBSQLException e) {
+          if (expectRemoveSuccess) {
+            LOGGER.error("Remove DataNodes SQL execute fail: {}", e.getMessage());
+            Assert.fail();
+          } else {
+            LOGGER.info("Submit Remove DataNodes fail, as expected");
+            return;
+          }
         }
+        LOGGER.info("Remove DataNodes SQL submit successfully.");
       }
-      LOGGER.info("Submit Remove DataNodes request: {}", removeReq);
 
       // Wait until success
       boolean removeSuccess = false;
@@ -363,4 +387,16 @@ public void restartDataNodes(List<DataNodeWrapper> dataNodeWrappers) {
               LOGGER.info("Node {} restarted.", nodeWrapper.getId());
             });
   }
+
+  public static String generateRemoveString(Set<Integer> dataNodes) {
+    StringBuilder sb = new StringBuilder("remove datanode ");
+
+    for (Integer node : dataNodes) {
+      sb.append(node).append(", ");
+    }
+
+    sb.setLength(sb.length() - 2);
+
+    return sb.toString();
+  }
 }
diff --git a/.../test/java/org/apache/iotdb/confignode/it/removedatanode/IoTDBRemoveDataNodeNormalIT.java b/.../test/java/org/apache/iotdb/confignode/it/removedatanode/IoTDBRemoveDataNodeNormalIT.java
@@ -29,13 +29,24 @@
 @Category({ClusterIT.class})
 @RunWith(IoTDBTestRunner.class)
 public class IoTDBRemoveDataNodeNormalIT extends IoTDBRemoveDataNodeITFramework {
+
   @Test
   public void success1C4DTest() throws Exception {
-    successTest(2, 3, 1, 4, 1, 2, true);
+    successTest(2, 3, 1, 4, 1, 2, true, SQLModel.NOT_USE_SQL);
   }
 
   @Test
   public void fail1C3DTest() throws Exception {
-    failTest(2, 3, 1, 3, 1, 2, false);
+    failTest(2, 3, 1, 3, 1, 2, false, SQLModel.NOT_USE_SQL);
+  }
+
+  @Test
+  public void success1C4DTestUseSQL() throws Exception {
+    successTest(2, 3, 1, 4, 1, 2, true, SQLModel.TREE_MODEL_SQL);
+  }
+
+  @Test
+  public void fail1C3DTestUseSQL() throws Exception {
+    failTest(2, 3, 1, 3, 1, 2, false, SQLModel.TREE_MODEL_SQL);
   }
 }
diff --git a/integration-test/src/test/java/org/apache/iotdb/confignode/it/removedatanode/SQLModel.java b/integration-test/src/test/java/org/apache/iotdb/confignode/it/removedatanode/SQLModel.java
@@ -0,0 +1,26 @@
+/*
+ * 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.confignode.it.removedatanode;
+
+public enum SQLModel {
+  NOT_USE_SQL,
+
+  TREE_MODEL_SQL,
+}
diff --git a/iotdb-core/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/IdentifierParser.g4 b/iotdb-core/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/IdentifierParser.g4
@@ -78,6 +78,7 @@ keyWords
     | DATA_REGION_GROUP_NUM
     | DATABASE
     | DATABASES
+    | DATANODE
     | DATANODEID
     | DATANODES
     | DATASET

diff --git a/iotdb-core/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/IoTDBSqlParser.g4 b/iotdb-core/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/IoTDBSqlParser.g4
@@ -65,7 +65,7 @@ ddlStatement
     // Cluster
     | showVariables | showCluster | showRegions | showDataNodes | showConfigNodes | showClusterId
     | getRegionId | getTimeSlotList | countTimeSlotList | getSeriesSlotList
-    | migrateRegion | reconstructRegion | extendRegion | removeRegion
+    | migrateRegion | reconstructRegion | extendRegion | removeRegion  | removeDataNode
     | verifyConnection
     // AINode
     | showAINodes | createModel | dropModel | showModels | callInference
@@ -554,6 +554,11 @@ verifyConnection
     : VERIFY CONNECTION (DETAILS)?
     ;
 
+// ---- Remove DataNode
+removeDataNode
+    : REMOVE DATANODE dataNodeId=INTEGER_LITERAL (COMMA dataNodeId=INTEGER_LITERAL)*
+    ;
+
 // Pipe Task =========================================================================================
 createPipe
     : CREATE PIPE  (IF NOT EXISTS)? pipeName=identifier

diff --git a/iotdb-core/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/SqlLexer.g4 b/iotdb-core/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/SqlLexer.g4
@@ -217,6 +217,10 @@ DATABASES
     : D A T A B A S E S
     ;
 
+DATANODE
+    : D A T A N O D E
+    ;
+
 DATANODEID
     : D A T A N O D E I D
     ;

diff --git a/...rc/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/ConfigTaskVisitor.java b/...rc/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/ConfigTaskVisitor.java
@@ -37,6 +37,7 @@
 import org.apache.iotdb.db.queryengine.plan.execution.config.metadata.GetRegionIdTask;
 import org.apache.iotdb.db.queryengine.plan.execution.config.metadata.GetSeriesSlotListTask;
 import org.apache.iotdb.db.queryengine.plan.execution.config.metadata.GetTimeSlotListTask;
+import org.apache.iotdb.db.queryengine.plan.execution.config.metadata.RemoveDataNodeTask;
 import org.apache.iotdb.db.queryengine.plan.execution.config.metadata.SetTTLTask;
 import org.apache.iotdb.db.queryengine.plan.execution.config.metadata.ShowClusterDetailsTask;
 import org.apache.iotdb.db.queryengine.plan.execution.config.metadata.ShowClusterIdTask;
@@ -114,6 +115,7 @@
 import org.apache.iotdb.db.queryengine.plan.statement.metadata.GetRegionIdStatement;
 import org.apache.iotdb.db.queryengine.plan.statement.metadata.GetSeriesSlotListStatement;
 import org.apache.iotdb.db.queryengine.plan.statement.metadata.GetTimeSlotListStatement;
+import org.apache.iotdb.db.queryengine.plan.statement.metadata.RemoveDataNodeStatement;
 import org.apache.iotdb.db.queryengine.plan.statement.metadata.SetTTLStatement;
 import org.apache.iotdb.db.queryengine.plan.statement.metadata.ShowClusterIdStatement;
 import org.apache.iotdb.db.queryengine.plan.statement.metadata.ShowClusterStatement;
@@ -607,6 +609,12 @@ public IConfigTask visitRemoveRegion(
     return new RemoveRegionTask(removeRegionStatement);
   }
 
+  @Override
+  public IConfigTask visitRemoveDataNode(
+      RemoveDataNodeStatement removeDataNodeStatement, MPPQueryContext context) {
+    return new RemoveDataNodeTask(removeDataNodeStatement);
+  }
+
   @Override
   public IConfigTask visitCreateContinuousQuery(
       CreateContinuousQueryStatement createContinuousQueryStatement, MPPQueryContext context) {

diff --git a/...apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java b/...apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java
@@ -19,6 +19,8 @@
 
 package org.apache.iotdb.db.queryengine.plan.execution.config.executor;
 
+import org.apache.iotdb.common.rpc.thrift.TDataNodeConfiguration;
+import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
 import org.apache.iotdb.common.rpc.thrift.TFlushReq;
 import org.apache.iotdb.common.rpc.thrift.TSStatus;
 import org.apache.iotdb.common.rpc.thrift.TSetConfigurationReq;
@@ -69,6 +71,8 @@
 import org.apache.iotdb.confignode.rpc.thrift.TCreatePipeReq;
 import org.apache.iotdb.confignode.rpc.thrift.TCreateTopicReq;
 import org.apache.iotdb.confignode.rpc.thrift.TCreateTriggerReq;
+import org.apache.iotdb.confignode.rpc.thrift.TDataNodeRemoveReq;
+import org.apache.iotdb.confignode.rpc.thrift.TDataNodeRemoveResp;
 import org.apache.iotdb.confignode.rpc.thrift.TDatabaseSchema;
 import org.apache.iotdb.confignode.rpc.thrift.TDeactivateSchemaTemplateReq;
 import org.apache.iotdb.confignode.rpc.thrift.TDeleteDatabasesReq;
@@ -184,6 +188,7 @@
 import org.apache.iotdb.db.queryengine.plan.statement.metadata.GetRegionIdStatement;
 import org.apache.iotdb.db.queryengine.plan.statement.metadata.GetSeriesSlotListStatement;
 import org.apache.iotdb.db.queryengine.plan.statement.metadata.GetTimeSlotListStatement;
+import org.apache.iotdb.db.queryengine.plan.statement.metadata.RemoveDataNodeStatement;
 import org.apache.iotdb.db.queryengine.plan.statement.metadata.SetTTLStatement;
 import org.apache.iotdb.db.queryengine.plan.statement.metadata.ShowClusterStatement;
 import org.apache.iotdb.db.queryengine.plan.statement.metadata.ShowDataNodesStatement;
@@ -272,8 +277,10 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.TreeMap;
 import java.util.stream.Collectors;
 
@@ -2664,6 +2671,70 @@ public SettableFuture<ConfigTaskResult> migrateRegion(
     return future;
   }
 
+  @Override
+  public SettableFuture<ConfigTaskResult> removeDataNode(
+      final RemoveDataNodeStatement removeDataNodeStatement) {
+    final SettableFuture<ConfigTaskResult> future = SettableFuture.create();
+    try (ConfigNodeClient configNodeClient =
+        CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) {
+      Set<Integer> nodeIds = removeDataNodeStatement.getNodeIds();
+
+      Set<Integer> validNodeIds =
+          configNodeClient.getDataNodeConfiguration(-1).getDataNodeConfigurationMap().keySet();
+
+      Set<Integer> invalidNodeIds = new HashSet<>(nodeIds);
+      invalidNodeIds.removeAll(validNodeIds);
+
+      if (!invalidNodeIds.isEmpty()) {
+        LOGGER.info("Cannot remove invalid nodeIds:{}", invalidNodeIds);
+        nodeIds.removeAll(invalidNodeIds);
+      }
+
+      LOGGER.info("Starting to remove DataNode with nodeIds: {}", nodeIds);
+
+      final Set<Integer> finalNodeIds = nodeIds;
+      List<TDataNodeLocation> removeDataNodeLocations =
+          configNodeClient
+              .getDataNodeConfiguration(-1)
+              .getDataNodeConfigurationMap()
+              .values()
+              .stream()
+              .map(TDataNodeConfiguration::getLocation)
+              .filter(location -> finalNodeIds.contains(location.getDataNodeId()))
+              .collect(Collectors.toList());
+
+      List<String> simplifiedLocations = new ArrayList<>();
+      for (TDataNodeLocation dataNodeLocation : removeDataNodeLocations) {
+        simplifiedLocations.add(
+            dataNodeLocation.getDataNodeId()
+                + "@"
+                + dataNodeLocation.getInternalEndPoint().getIp());
+      }
+
+      LOGGER.info("Start to remove datanode, removed DataNodes endpoint: {}", simplifiedLocations);
+      TDataNodeRemoveReq removeReq = new TDataNodeRemoveReq(removeDataNodeLocations);
+      TDataNodeRemoveResp removeResp = configNodeClient.removeDataNode(removeReq);
+      LOGGER.info("Submit Remove DataNodes result {} ", removeResp);
+      if (removeResp.getStatus().getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+        future.setException(
+            new IoTDBException(
+                removeResp.getStatus().toString(), removeResp.getStatus().getCode()));
+        return future;
+      } else {
+        LOGGER.info(
+            "Submit remove-datanode request successfully, but the process may fail. "
+                + "more details are shown in the logs of confignode-leader and removed-datanode, "
+                + "and after the process of removing datanode ends successfully, "
+                + "you are supposed to delete directory and data of the removed-datanode manually");
+        future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
+      }
+    } catch (Exception e) {
+      future.setException(e);
+    }
+
+    return future;
+  }
+
   @Override
   public SettableFuture<ConfigTaskResult> reconstructRegion(
       ReconstructRegionStatement reconstructRegionStatement) {

diff --git a/...a/org/apache/iotdb/db/queryengine/plan/execution/config/executor/IConfigTaskExecutor.java b/...a/org/apache/iotdb/db/queryengine/plan/execution/config/executor/IConfigTaskExecutor.java
@@ -39,6 +39,7 @@
 import org.apache.iotdb.db.queryengine.plan.statement.metadata.GetRegionIdStatement;
 import org.apache.iotdb.db.queryengine.plan.statement.metadata.GetSeriesSlotListStatement;
 import org.apache.iotdb.db.queryengine.plan.statement.metadata.GetTimeSlotListStatement;
+import org.apache.iotdb.db.queryengine.plan.statement.metadata.RemoveDataNodeStatement;
 import org.apache.iotdb.db.queryengine.plan.statement.metadata.SetTTLStatement;
 import org.apache.iotdb.db.queryengine.plan.statement.metadata.ShowClusterStatement;
 import org.apache.iotdb.db.queryengine.plan.statement.metadata.ShowDataNodesStatement;
@@ -233,6 +234,8 @@ SettableFuture<ConfigTaskResult> reconstructRegion(
 
   SettableFuture<ConfigTaskResult> removeRegion(RemoveRegionStatement removeRegionStatement);
 
+  SettableFuture<ConfigTaskResult> removeDataNode(RemoveDataNodeStatement removeDataNodeStatement);
+
   SettableFuture<ConfigTaskResult> createContinuousQuery(
       CreateContinuousQueryStatement createContinuousQueryStatement, MPPQueryContext context);