Merge branch 'master' of https://github.com/apache/iotdb into rel/0.14.0-preview2

Author: HTHou

RELEASE_NOTES.md

@@ -29,10 +29,8 @@ The 0.14.0-preview2 version only contains the new cluster version.
 * Support Between expression
 * Support order by timeseries in last query
 * Support hot configuration of data_dirs
-* Support set system mode to readonly, running and error
 * Support schema template
 * Support executeBatchStatement in JDBC
-* Support clear cache
 * Support new UDF 'change_points'
 
 ## Improvements

antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/IdentifierParser.g4

@@ -75,6 +75,7 @@ keyWords
     | EVERY
     | EXPLAIN
     | FILL
+    | FILE
     | FLUSH
     | FOR
     | FROM
@@ -169,6 +170,7 @@ keyWords
     | UNSET
     | UPDATE
     | UPSERT
+    | URI
     | USER
     | USING
     | VALUES

antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/IoTDBSqlParser.g4

@@ -126,7 +126,7 @@ uri
 
 // Create Trigger
 createTrigger
-    : CREATE triggerType? TRIGGER triggerName=identifier triggerEventClause ON prefixPath AS className=STRING_LITERAL triggerAttributeClause?
+    : CREATE triggerType? TRIGGER triggerName=identifier triggerEventClause ON prefixPath AS className=STRING_LITERAL jarLocation? triggerAttributeClause?
     ;
 
 triggerType
@@ -137,6 +137,10 @@ triggerEventClause
     : (BEFORE | AFTER) (INSERT | DELETE)
     ;
 
+jarLocation
+    : USING ((FILE fileName=STRING_LITERAL) | URI uri)
+    ;
+
 triggerAttributeClause
     : WITH LR_BRACKET triggerAttribute (COMMA triggerAttribute)* RR_BRACKET
     ;

antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/SqlLexer.g4

@@ -215,6 +215,10 @@ FILL
     : F I L L
     ;
 
+FILE
+    : F I L E
+    ;
+
 FLUSH
     : F L U S H
     ;
@@ -603,6 +607,10 @@ UPSERT
     : U P S E R T
     ;
 
+URI
+    : U R I
+    ;
+
 USER
     : U S E R
     ;

confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/CreateRegionGroupsPlan.java

@@ -60,6 +60,16 @@ public void addRegionGroup(String storageGroup, TRegionReplicaSet regionReplicaS
         .add(regionReplicaSet);
   }
 
+  public void serializeForProcedure(DataOutputStream stream) throws IOException {
+    this.serializeImpl(stream);
+  }
+
+  public void deserializeForProcedure(ByteBuffer buffer) throws IOException {
+    // to remove the ordinal of ConfigPhysicalPlanType
+    buffer.getInt();
+    this.deserializeImpl(buffer);
+  }
+
   @Override
   protected void serializeImpl(DataOutputStream stream) throws IOException {
     stream.writeInt(ConfigPhysicalPlanType.CreateRegionGroups.ordinal());

confignode/src/main/java/org/apache/iotdb/confignode/persistence/UDFInfo.java

@@ -20,10 +20,10 @@
 package org.apache.iotdb.confignode.persistence;
 
 import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.executable.ExecutableResource;
 import org.apache.iotdb.commons.snapshot.SnapshotProcessor;
 import org.apache.iotdb.commons.udf.service.UDFClassLoader;
 import org.apache.iotdb.commons.udf.service.UDFExecutableManager;
-import org.apache.iotdb.commons.udf.service.UDFExecutableResource;
 import org.apache.iotdb.commons.udf.service.UDFRegistrationService;
 import org.apache.iotdb.confignode.conf.ConfigNodeConfig;
 import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor;
@@ -78,7 +78,7 @@ private void fetchExecutablesAndCheckInstantiation(String className) throws Exce
 
   private void fetchExecutablesAndCheckInstantiation(String className, List<String> uris)
       throws Exception {
-    final UDFExecutableResource resource = udfExecutableManager.request(uris);
+    final ExecutableResource resource = udfExecutableManager.request(uris);
     try (UDFClassLoader temporaryUdfClassLoader = new UDFClassLoader(resource.getResourceDir())) {
       Class.forName(className, true, temporaryUdfClassLoader)
           .getDeclaredConstructor()

confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/CreateRegionGroupsProcedure.java

@@ -20,26 +20,50 @@
 
 import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
 import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
+import org.apache.iotdb.commons.utils.ThriftCommonsSerDeUtils;
 import org.apache.iotdb.confignode.consensus.request.write.CreateRegionGroupsPlan;
 import org.apache.iotdb.confignode.consensus.request.write.DeleteRegionGroupsPlan;
 import org.apache.iotdb.confignode.procedure.StateMachineProcedure;
 import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv;
 import org.apache.iotdb.confignode.procedure.exception.ProcedureException;
 import org.apache.iotdb.confignode.procedure.state.CreateRegionGroupsState;
+import org.apache.iotdb.confignode.procedure.store.ProcedureFactory;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.HashMap;
 import java.util.Map;
+import java.util.Objects;
 
 public class CreateRegionGroupsProcedure
     extends StateMachineProcedure<ConfigNodeProcedureEnv, CreateRegionGroupsState> {
 
-  private final CreateRegionGroupsPlan createRegionGroupsPlan;
-  // Map<TConsensusGroupId, Failed RegionReplicas>
-  private Map<TConsensusGroupId, TRegionReplicaSet> failedRegions;
+  private static final Logger LOGGER = LoggerFactory.getLogger(CreateRegionGroupsProcedure.class);
+
+  private CreateRegionGroupsPlan createRegionGroupsPlan = new CreateRegionGroupsPlan();
+
+  /** key: TConsensusGroupId value: Failed RegionReplicas */
+  private Map<TConsensusGroupId, TRegionReplicaSet> failedRegions = new HashMap<>();
+
+  public CreateRegionGroupsProcedure() {
+    super();
+  }
 
   public CreateRegionGroupsProcedure(CreateRegionGroupsPlan createRegionGroupsPlan) {
     this.createRegionGroupsPlan = createRegionGroupsPlan;
   }
 
+  public CreateRegionGroupsProcedure(
+      CreateRegionGroupsPlan createRegionGroupsPlan,
+      Map<TConsensusGroupId, TRegionReplicaSet> failedRegions) {
+    this.createRegionGroupsPlan = createRegionGroupsPlan;
+    this.failedRegions = failedRegions;
+  }
+
   @Override
   protected Flow executeFromState(ConfigNodeProcedureEnv env, CreateRegionGroupsState state) {
     switch (state) {
@@ -130,6 +154,59 @@ protected int getStateId(CreateRegionGroupsState createRegionGroupsState) {
 
   @Override
   protected CreateRegionGroupsState getInitialState() {
-    return CreateRegionGroupsState.CREATE_REGION_GROUPS;
+    return CreateRegionGroupsState.CREATE_REGION_GROUPS_PREPARE;
+  }
+
+  @Override
+  public void serialize(DataOutputStream stream) throws IOException {
+    // must serialize CREATE_REGION_GROUPS.ordinal() firstly
+    stream.writeInt(ProcedureFactory.ProcedureType.CREATE_REGION_GROUPS.ordinal());
+    super.serialize(stream);
+    createRegionGroupsPlan.serializeForProcedure(stream);
+    stream.writeInt(failedRegions.size());
+    failedRegions.forEach(
+        (groupId, replica) -> {
+          ThriftCommonsSerDeUtils.serializeTConsensusGroupId(groupId, stream);
+          ThriftCommonsSerDeUtils.serializeTRegionReplicaSet(replica, stream);
+        });
+  }
+
+  @Override
+  public void deserialize(ByteBuffer byteBuffer) {
+    super.deserialize(byteBuffer);
+    try {
+      createRegionGroupsPlan.deserializeForProcedure(byteBuffer);
+      failedRegions.clear();
+      int failedRegionsSize = byteBuffer.getInt();
+      while (failedRegionsSize-- > 0) {
+        TConsensusGroupId groupId =
+            ThriftCommonsSerDeUtils.deserializeTConsensusGroupId(byteBuffer);
+        TRegionReplicaSet replica =
+            ThriftCommonsSerDeUtils.deserializeTRegionReplicaSet(byteBuffer);
+        failedRegions.put(groupId, replica);
+      }
+    } catch (Exception e) {
+      LOGGER.error("Deserialize meets error in CreateRegionGroupsProcedure", e);
+      throw new RuntimeException(e);
+    }
+  }
+
+  @Override
+  public boolean equals(Object that) {
+    if (that instanceof CreateRegionGroupsProcedure) {
+      CreateRegionGroupsProcedure thatProc = (CreateRegionGroupsProcedure) that;
+      return thatProc.getProcId() == this.getProcId()
+          && thatProc.getState() == this.getState()
+          && thatProc.createRegionGroupsPlan.equals(this.createRegionGroupsPlan)
+          && thatProc.failedRegions.equals(this.failedRegions);
+    }
+    return false;
+  }
+
+  @Override
+  public int hashCode() {
+    int result = createRegionGroupsPlan.hashCode();
+    result = 31 * result + Objects.hash(failedRegions);
+    return result;
   }
 }

confignode/src/main/java/org/apache/iotdb/confignode/procedure/state/RemoveConfigNodeState.java

@@ -20,8 +20,7 @@
 package org.apache.iotdb.confignode.procedure.state;
 
 public enum RemoveConfigNodeState {
-  REMOVE_CONSENSUS_GROUP,
   REMOVE_PEER,
-
+  REMOVE_CONSENSUS_GROUP,
   STOP_CONFIG_NODE
 }

confignode/src/main/java/org/apache/iotdb/confignode/procedure/store/ProcedureFactory.java

@@ -27,15 +27,21 @@
 import org.apache.iotdb.confignode.procedure.impl.RemoveConfigNodeProcedure;
 import org.apache.iotdb.confignode.procedure.impl.RemoveDataNodeProcedure;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import java.io.IOException;
 import java.nio.ByteBuffer;
 
 public class ProcedureFactory implements IProcedureFactory {
 
+  private static final Logger LOGGER = LoggerFactory.getLogger(ProcedureFactory.class);
+
   @Override
   public Procedure create(ByteBuffer buffer) throws IOException {
     int typeNum = buffer.getInt();
     if (typeNum >= ProcedureType.values().length) {
+      LOGGER.error("unrecognized log type " + typeNum);
       throw new IOException("unrecognized log type " + typeNum);
     }
     ProcedureType type = ProcedureType.values()[typeNum];
@@ -56,7 +62,11 @@ public Procedure create(ByteBuffer buffer) throws IOException {
       case REGION_MIGRATE_PROCEDURE:
         procedure = new RegionMigrateProcedure();
         break;
+      case CREATE_REGION_GROUPS:
+        procedure = new CreateRegionGroupsProcedure();
+        break;
       default:
+        LOGGER.error("unknown Procedure type: " + typeNum);
         throw new IOException("unknown Procedure type: " + typeNum);
     }
     procedure.deserialize(buffer);

confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCService.java

@@ -72,7 +72,7 @@ public void initThriftServiceThread() throws IllegalAccessException {
           new ThriftServiceThread(
               processor,
               getID().getName(),
-              ThreadName.CONFIG_NODE_RPC_CLIENT.getName(),
+              ThreadName.CONFIGNODE_RPC_PROCESSOR.getName(),
               getBindIP(),
               getBindPort(),
               configConf.getRpcMaxConcurrentClientNum(),
@@ -82,15 +82,15 @@ public void initThriftServiceThread() throws IllegalAccessException {
     } catch (RPCServiceException e) {
       throw new IllegalAccessException(e.getMessage());
     }
-    thriftServiceThread.setName(ThreadName.CONFIG_NODE_RPC_SERVER.getName());
+    thriftServiceThread.setName(ThreadName.CONFIGNODE_RPC_SERVICE.getName());
     MetricService.getInstance()
         .getOrCreateAutoGauge(
             Metric.THRIFT_ACTIVE_THREADS.toString(),
             MetricLevel.CORE,
             thriftServiceThread,
             AbstractThriftServiceThread::getActiveThreadCount,
             Tag.NAME.toString(),
-            ThreadName.CONFIG_NODE_RPC_SERVER.getName());
+            ThreadName.CONFIGNODE_RPC_SERVICE.getName());
   }
 
   @Override