[remove datanode] Add Remove DataNode SQL (#14678)

* add remove sql * Update iotdb-system.properties * fix IT * add table model sql * add IT * resolve conflicts * simplified location * fix comments
apache · Jan 23, 2025 · 7f3b281 · 7f3b281
1 parent cb4c05c
commit 7f3b281
Show file tree

Hide file tree

Showing 23 changed files with 518 additions and 27 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,8 @@
 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.relational.it.query.old.aligned.TableUtils;
 import org.apache.iotdb.rpc.TSStatusCode;
 
 import org.apache.thrift.TException;
@@ -42,6 +44,7 @@
 
 import java.sql.Connection;
 import java.sql.ResultSet;
+import java.sql.SQLException;
 import java.sql.Statement;
 import java.util.HashSet;
 import java.util.List;
@@ -57,7 +60,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 +93,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 +104,8 @@ public void successTest(
         removeDataNodeNum,
         dataRegionPerDataNode,
         true,
-        rejoinRemovedDataNode);
+        rejoinRemovedDataNode,
+        model);
   }
 
   public void failTest(
@@ -110,7 +115,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 +126,8 @@ public void failTest(
         removeDataNodeNum,
         dataRegionPerDataNode,
         false,
-        rejoinRemovedDataNode);
+        rejoinRemovedDataNode,
+        model);
   }
 
   public void testRemoveDataNode(
@@ -131,7 +138,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()
@@ -143,13 +151,18 @@ public void testRemoveDataNode(
             dataRegionPerDataNode * dataNodeNum / dataReplicateFactor);
     EnvFactory.getEnv().initClusterEnvironment(configNodeNum, dataNodeNum);
 
-    try (final Connection connection = makeItCloseQuietly(EnvFactory.getEnv().getConnection());
+    try (final Connection connection = makeItCloseQuietly(getConnectionWithSQLType(model));
         final Statement statement = makeItCloseQuietly(connection.createStatement());
         SyncConfigNodeIServiceClient client =
             (SyncConfigNodeIServiceClient) EnvFactory.getEnv().getLeaderConfigNodeConnection()) {
 
-      // Insert data
-      statement.execute(INSERTION1);
+      if (SQLModel.TABLE_MODEL_SQL.equals(model)) {
+        // Insert data in table model
+        TableUtils.insertData();
+      } else {
+        // Insert data in tree model
+        statement.execute(TREE_MODEL_INSERTION);
+      }
 
       Map<Integer, Set<Integer>> regionMap = getDataRegionMap(statement);
       regionMap.forEach(
@@ -187,21 +200,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 +394,24 @@ 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();
+  }
+
+  public Connection getConnectionWithSQLType(SQLModel model) throws SQLException {
+    if (SQLModel.TABLE_MODEL_SQL.equals(model)) {
+      return EnvFactory.getEnv().getTableConnection();
+    } else {
+      return EnvFactory.getEnv().getConnection();
+    }
+  }
 }
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,34 @@
 @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);
+  }
+
+  @Test
+  public void success1C4DTestUseTableSQL() throws Exception {
+    successTest(2, 3, 1, 4, 1, 2, true, SQLModel.TABLE_MODEL_SQL);
+  }
+
+  @Test
+  public void fail1C3DTestUseTableSQL() throws Exception {
+    failTest(2, 3, 1, 3, 1, 2, false, SQLModel.TABLE_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,28 @@
+/*
+ * 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,
+
+  TABLE_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
@@ -63,7 +63,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
@@ -550,6 +550,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/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/Coordinator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/Coordinator.java
@@ -72,6 +72,7 @@
 import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Flush;
 import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.KillQuery;
 import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.PipeStatement;
+import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.RemoveDataNode;
 import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.SetConfiguration;
 import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.SetProperties;
 import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowAINodes;
@@ -407,6 +408,7 @@ private IQueryExecution createQueryExecutionForTableModel(
         || statement instanceof StartRepairData
         || statement instanceof StopRepairData
         || statement instanceof PipeStatement
+        || statement instanceof RemoveDataNode
         || statement instanceof SubscriptionStatement
         || statement instanceof ShowCurrentSqlDialect
         || statement instanceof ShowCurrentUser

diff --git a/...in/java/org/apache/iotdb/db/queryengine/plan/execution/config/TableConfigTaskVisitor.java b/...in/java/org/apache/iotdb/db/queryengine/plan/execution/config/TableConfigTaskVisitor.java
@@ -36,6 +36,7 @@
 import org.apache.iotdb.db.queryengine.plan.execution.config.metadata.CreatePipePluginTask;
 import org.apache.iotdb.db.queryengine.plan.execution.config.metadata.DropFunctionTask;
 import org.apache.iotdb.db.queryengine.plan.execution.config.metadata.DropPipePluginTask;
+import org.apache.iotdb.db.queryengine.plan.execution.config.metadata.RemoveDataNodeTask;
 import org.apache.iotdb.db.queryengine.plan.execution.config.metadata.ShowClusterIdTask;
 import org.apache.iotdb.db.queryengine.plan.execution.config.metadata.ShowClusterTask;
 import org.apache.iotdb.db.queryengine.plan.execution.config.metadata.ShowFunctionsTask;
@@ -120,6 +121,7 @@
 import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Node;
 import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Property;
 import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.QualifiedName;
+import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.RemoveDataNode;
 import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.RenameColumn;
 import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.RenameTable;
 import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.SetConfiguration;
@@ -151,6 +153,7 @@
 import org.apache.iotdb.db.queryengine.plan.relational.sql.rewrite.StatementRewrite;
 import org.apache.iotdb.db.queryengine.plan.relational.type.TypeNotFoundException;
 import org.apache.iotdb.db.queryengine.plan.statement.metadata.DatabaseSchemaStatement;
+import org.apache.iotdb.db.queryengine.plan.statement.metadata.RemoveDataNodeStatement;
 import org.apache.iotdb.db.queryengine.plan.statement.metadata.ShowClusterStatement;
 import org.apache.iotdb.db.queryengine.plan.statement.metadata.ShowRegionStatement;
 import org.apache.iotdb.db.queryengine.plan.statement.sys.FlushStatement;
@@ -351,6 +354,18 @@ protected IConfigTask visitShowRegions(
     return new ShowRegionTask(treeStatement, true);
   }
 
+  @Override
+  protected IConfigTask visitRemoveDataNode(
+      final RemoveDataNode removeDataNode, final MPPQueryContext context) {
+    context.setQueryType(QueryType.WRITE);
+    accessControl.checkUserHasMaintainPrivilege(context.getSession().getUserName());
+    // As the implementation is identical, we'll simply translate to the
+    // corresponding tree-model variant and execute that.
+    final RemoveDataNodeStatement treeStatement =
+        new RemoveDataNodeStatement(removeDataNode.getNodeIds());
+    return new RemoveDataNodeTask(treeStatement);
+  }
+
   @Override
   protected IConfigTask visitShowDataNodes(
       final ShowDataNodes showDataNodesStatement, final MPPQueryContext context) {

diff --git a/...ain/java/org/apache/iotdb/db/queryengine/plan/execution/config/TreeConfigTaskVisitor.java b/...ain/java/org/apache/iotdb/db/queryengine/plan/execution/config/TreeConfigTaskVisitor.java
@@ -40,6 +40,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.ShowAINodesTask;
 import org.apache.iotdb.db.queryengine.plan.execution.config.metadata.ShowClusterDetailsTask;
@@ -117,6 +118,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;
@@ -652,6 +654,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) {