[iotdb] branch master updated: [IOTDB-1416] Maintanance command support in cluster mode (#3480)

[wa...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

wangchao316 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git


The following commit(s) were added to refs/heads/master by this push:
     new 6615778  [IOTDB-1416] Maintanance command support in cluster mode (#3480)
6615778 is described below

commit 6615778fb1a0f79749b2d1dc92c3a96af9ff9f3d
Author: Hang Ji <55...@users.noreply.github.com>
AuthorDate: Tue Jul 13 15:56:44 2021 +0800

    [IOTDB-1416] Maintanance command support in cluster mode (#3480)
    
    [IOTDB-1416] Maintanance command support in cluster mode (#3480)
---
 .../apache/iotdb/cluster/utils/PartitionUtils.java |  8 ++++--
 .../cluster/log/applier/DataLogApplierTest.java    | 18 +++++++++++++
 .../cluster/log/applier/MetaLogApplierTest.java    | 10 ++++++++
 .../apache/iotdb/db/qp/physical/PhysicalPlan.java  | 17 +++++++++++-
 .../iotdb/db/qp/physical/sys/ClearCachePlan.java   | 17 ++++++++++++
 .../db/qp/physical/sys/CreateSnapshotPlan.java     | 17 ++++++++++++
 .../apache/iotdb/db/qp/physical/sys/MergePlan.java | 17 ++++++++++++
 .../test/java/org/apache/iotdb/db/sql/Cases.java   | 30 ++++++++++++++++++++++
 8 files changed, 131 insertions(+), 3 deletions(-)

diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/utils/PartitionUtils.java b/cluster/src/main/java/org/apache/iotdb/cluster/utils/PartitionUtils.java
index 6c4cf9d..2bf9885 100644
--- a/cluster/src/main/java/org/apache/iotdb/cluster/utils/PartitionUtils.java
+++ b/cluster/src/main/java/org/apache/iotdb/cluster/utils/PartitionUtils.java
@@ -28,6 +28,8 @@ import org.apache.iotdb.db.qp.physical.crud.DeletePlan;
 import org.apache.iotdb.db.qp.physical.crud.InsertTabletPlan;
 import org.apache.iotdb.db.qp.physical.crud.SetDeviceTemplatePlan;
 import org.apache.iotdb.db.qp.physical.sys.AuthorPlan;
+import org.apache.iotdb.db.qp.physical.sys.ClearCachePlan;
+import org.apache.iotdb.db.qp.physical.sys.CreateSnapshotPlan;
 import org.apache.iotdb.db.qp.physical.sys.DataAuthPlan;
 import org.apache.iotdb.db.qp.physical.sys.DeleteStorageGroupPlan;
 import org.apache.iotdb.db.qp.physical.sys.DeleteTimeSeriesPlan;
@@ -87,7 +89,8 @@ public class PartitionUtils {
         || plan instanceof DeleteStorageGroupPlan
         // DataAuthPlan is global because all nodes must have all user info
         || plan instanceof DataAuthPlan
-        || plan instanceof CreateTemplatePlan;
+        || plan instanceof CreateTemplatePlan
+        || plan instanceof CreateSnapshotPlan;
   }
 
   /**
@@ -103,7 +106,8 @@ public class PartitionUtils {
         || plan instanceof DeleteTimeSeriesPlan
         || plan instanceof MergePlan
         || plan instanceof FlushPlan
-        || plan instanceof SetDeviceTemplatePlan;
+        || plan instanceof SetDeviceTemplatePlan
+        || plan instanceof ClearCachePlan;
   }
 
   public static int calculateStorageGroupSlotByTime(
diff --git a/cluster/src/test/java/org/apache/iotdb/cluster/log/applier/DataLogApplierTest.java b/cluster/src/test/java/org/apache/iotdb/cluster/log/applier/DataLogApplierTest.java
index c1c23ef..cd90c7b 100644
--- a/cluster/src/test/java/org/apache/iotdb/cluster/log/applier/DataLogApplierTest.java
+++ b/cluster/src/test/java/org/apache/iotdb/cluster/log/applier/DataLogApplierTest.java
@@ -59,8 +59,10 @@ import org.apache.iotdb.db.metadata.PartialPath;
 import org.apache.iotdb.db.metadata.mnode.MeasurementMNode;
 import org.apache.iotdb.db.qp.physical.crud.DeletePlan;
 import org.apache.iotdb.db.qp.physical.crud.InsertRowPlan;
+import org.apache.iotdb.db.qp.physical.sys.ClearCachePlan;
 import org.apache.iotdb.db.qp.physical.sys.CreateMultiTimeSeriesPlan;
 import org.apache.iotdb.db.qp.physical.sys.FlushPlan;
+import org.apache.iotdb.db.qp.physical.sys.MergePlan;
 import org.apache.iotdb.db.service.IoTDB;
 import org.apache.iotdb.tsfile.exception.filter.QueryFilterOptimizationException;
 import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType;
@@ -374,4 +376,20 @@ public class DataLogApplierTest extends IoTDBTest {
     assertTrue(IoTDB.metaManager.getAllStorageGroupPaths().contains(new PartialPath("root.sg2")));
     assertNull(log.getException());
   }
+
+  @Test
+  public void testApplyClearCache() {
+    ClearCachePlan clearCachePlan = new ClearCachePlan();
+    PhysicalPlanLog physicalPlanLog = new PhysicalPlanLog(clearCachePlan);
+    applier.apply(physicalPlanLog);
+    assertNull(physicalPlanLog.getException());
+  }
+
+  @Test
+  public void testApplyMerge() {
+    MergePlan mergePlan = new MergePlan();
+    PhysicalPlanLog physicalPlanLog = new PhysicalPlanLog(mergePlan);
+    applier.apply(physicalPlanLog);
+    assertNull(physicalPlanLog.getException());
+  }
 }
diff --git a/cluster/src/test/java/org/apache/iotdb/cluster/log/applier/MetaLogApplierTest.java b/cluster/src/test/java/org/apache/iotdb/cluster/log/applier/MetaLogApplierTest.java
index 013403e..d6d6d02 100644
--- a/cluster/src/test/java/org/apache/iotdb/cluster/log/applier/MetaLogApplierTest.java
+++ b/cluster/src/test/java/org/apache/iotdb/cluster/log/applier/MetaLogApplierTest.java
@@ -32,6 +32,7 @@ import org.apache.iotdb.cluster.utils.Constants;
 import org.apache.iotdb.db.exception.StorageEngineException;
 import org.apache.iotdb.db.exception.metadata.MetadataException;
 import org.apache.iotdb.db.metadata.PartialPath;
+import org.apache.iotdb.db.qp.physical.sys.CreateSnapshotPlan;
 import org.apache.iotdb.db.qp.physical.sys.CreateTimeSeriesPlan;
 import org.apache.iotdb.db.qp.physical.sys.SetStorageGroupPlan;
 import org.apache.iotdb.db.service.IoTDB;
@@ -50,6 +51,7 @@ import java.util.Set;
 import static junit.framework.TestCase.assertEquals;
 import static junit.framework.TestCase.assertFalse;
 import static junit.framework.TestCase.assertTrue;
+import static org.junit.Assert.assertNull;
 
 public class MetaLogApplierTest extends IoTDBTest {
 
@@ -132,4 +134,12 @@ public class MetaLogApplierTest extends IoTDBTest {
         TSDataType.DOUBLE,
         IoTDB.metaManager.getSeriesType(new PartialPath("root" + ".applyMeta.s1")));
   }
+
+  @Test
+  public void testApplyCreateSnapshot() {
+    CreateSnapshotPlan createSnapshotPlan = new CreateSnapshotPlan();
+    PhysicalPlanLog physicalPlanLog = new PhysicalPlanLog(createSnapshotPlan);
+    applier.apply(physicalPlanLog);
+    assertNull(physicalPlanLog.getException());
+  }
 }
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/physical/PhysicalPlan.java b/server/src/main/java/org/apache/iotdb/db/qp/physical/PhysicalPlan.java
index 97d5daa..163ed8a 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/physical/PhysicalPlan.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/physical/PhysicalPlan.java
@@ -35,10 +35,12 @@ import org.apache.iotdb.db.qp.physical.sys.AuthorPlan;
 import org.apache.iotdb.db.qp.physical.sys.AutoCreateDeviceMNodePlan;
 import org.apache.iotdb.db.qp.physical.sys.ChangeAliasPlan;
 import org.apache.iotdb.db.qp.physical.sys.ChangeTagOffsetPlan;
+import org.apache.iotdb.db.qp.physical.sys.ClearCachePlan;
 import org.apache.iotdb.db.qp.physical.sys.CreateAlignedTimeSeriesPlan;
 import org.apache.iotdb.db.qp.physical.sys.CreateContinuousQueryPlan;
 import org.apache.iotdb.db.qp.physical.sys.CreateIndexPlan;
 import org.apache.iotdb.db.qp.physical.sys.CreateMultiTimeSeriesPlan;
+import org.apache.iotdb.db.qp.physical.sys.CreateSnapshotPlan;
 import org.apache.iotdb.db.qp.physical.sys.CreateTimeSeriesPlan;
 import org.apache.iotdb.db.qp.physical.sys.CreateTriggerPlan;
 import org.apache.iotdb.db.qp.physical.sys.DataAuthPlan;
@@ -52,6 +54,7 @@ import org.apache.iotdb.db.qp.physical.sys.LoadConfigurationPlan;
 import org.apache.iotdb.db.qp.physical.sys.LogPlan;
 import org.apache.iotdb.db.qp.physical.sys.MNodePlan;
 import org.apache.iotdb.db.qp.physical.sys.MeasurementMNodePlan;
+import org.apache.iotdb.db.qp.physical.sys.MergePlan;
 import org.apache.iotdb.db.qp.physical.sys.SetStorageGroupPlan;
 import org.apache.iotdb.db.qp.physical.sys.SetTTLPlan;
 import org.apache.iotdb.db.qp.physical.sys.SetUsingDeviceTemplatePlan;
@@ -380,6 +383,15 @@ public abstract class PhysicalPlan {
         case DROP_CONTINUOUS_QUERY:
           plan = new DropContinuousQueryPlan();
           break;
+        case MERGE:
+          plan = new MergePlan();
+          break;
+        case CREATE_SNAPSHOT:
+          plan = new CreateSnapshotPlan();
+          break;
+        case CLEARCACHE:
+          plan = new ClearCachePlan();
+          break;
         default:
           throw new IOException("unrecognized log type " + type);
       }
@@ -438,7 +450,10 @@ public abstract class PhysicalPlan {
     STOP_TRIGGER,
     CREATE_CONTINUOUS_QUERY,
     DROP_CONTINUOUS_QUERY,
-    SHOW_CONTINUOUS_QUERIES
+    SHOW_CONTINUOUS_QUERIES,
+    MERGE,
+    CREATE_SNAPSHOT,
+    CLEARCACHE
   }
 
   public long getIndex() {
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/ClearCachePlan.java b/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/ClearCachePlan.java
index 71c5d24..4f71861 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/ClearCachePlan.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/ClearCachePlan.java
@@ -18,10 +18,14 @@
  */
 package org.apache.iotdb.db.qp.physical.sys;
 
+import org.apache.iotdb.db.exception.metadata.IllegalPathException;
 import org.apache.iotdb.db.metadata.PartialPath;
 import org.apache.iotdb.db.qp.logical.Operator.OperatorType;
 import org.apache.iotdb.db.qp.physical.PhysicalPlan;
 
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
 import java.util.Collections;
 import java.util.List;
 
@@ -35,4 +39,17 @@ public class ClearCachePlan extends PhysicalPlan {
   public List<PartialPath> getPaths() {
     return Collections.emptyList();
   }
+
+  @Override
+  public void serialize(DataOutputStream stream) throws IOException {
+    stream.writeByte((byte) PhysicalPlanType.CLEARCACHE.ordinal());
+  }
+
+  @Override
+  public void serialize(ByteBuffer buffer) {
+    buffer.put((byte) PhysicalPlanType.CLEARCACHE.ordinal());
+  }
+
+  @Override
+  public void deserialize(ByteBuffer buffer) throws IllegalPathException {}
 }
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/CreateSnapshotPlan.java b/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/CreateSnapshotPlan.java
index 399c3ce..8d826d2 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/CreateSnapshotPlan.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/CreateSnapshotPlan.java
@@ -19,10 +19,14 @@
 
 package org.apache.iotdb.db.qp.physical.sys;
 
+import org.apache.iotdb.db.exception.metadata.IllegalPathException;
 import org.apache.iotdb.db.metadata.PartialPath;
 import org.apache.iotdb.db.qp.logical.Operator.OperatorType;
 import org.apache.iotdb.db.qp.physical.PhysicalPlan;
 
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
 import java.util.Collections;
 import java.util.List;
 
@@ -36,4 +40,17 @@ public class CreateSnapshotPlan extends PhysicalPlan {
   public List<PartialPath> getPaths() {
     return Collections.emptyList();
   }
+
+  @Override
+  public void serialize(DataOutputStream stream) throws IOException {
+    stream.writeByte((byte) PhysicalPlanType.CREATE_SNAPSHOT.ordinal());
+  }
+
+  @Override
+  public void serialize(ByteBuffer buffer) {
+    buffer.put((byte) PhysicalPlanType.CREATE_SNAPSHOT.ordinal());
+  }
+
+  @Override
+  public void deserialize(ByteBuffer buffer) throws IllegalPathException {}
 }
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/MergePlan.java b/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/MergePlan.java
index 30e9043..848c1e7 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/MergePlan.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/MergePlan.java
@@ -18,10 +18,14 @@
  */
 package org.apache.iotdb.db.qp.physical.sys;
 
+import org.apache.iotdb.db.exception.metadata.IllegalPathException;
 import org.apache.iotdb.db.metadata.PartialPath;
 import org.apache.iotdb.db.qp.logical.Operator.OperatorType;
 import org.apache.iotdb.db.qp.physical.PhysicalPlan;
 
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
 import java.util.Collections;
 import java.util.List;
 
@@ -39,4 +43,17 @@ public class MergePlan extends PhysicalPlan {
   public List<PartialPath> getPaths() {
     return Collections.emptyList();
   }
+
+  @Override
+  public void serialize(DataOutputStream stream) throws IOException {
+    stream.writeByte((byte) PhysicalPlanType.MERGE.ordinal());
+  }
+
+  @Override
+  public void serialize(ByteBuffer buffer) {
+    buffer.put((byte) PhysicalPlanType.MERGE.ordinal());
+  }
+
+  @Override
+  public void deserialize(ByteBuffer buffer) throws IllegalPathException {}
 }
diff --git a/testcontainer/src/test/java/org/apache/iotdb/db/sql/Cases.java b/testcontainer/src/test/java/org/apache/iotdb/db/sql/Cases.java
index dc8f487..b703698 100644
--- a/testcontainer/src/test/java/org/apache/iotdb/db/sql/Cases.java
+++ b/testcontainer/src/test/java/org/apache/iotdb/db/sql/Cases.java
@@ -406,4 +406,34 @@ public abstract class Cases {
       }
     }
   }
+
+  @Test
+  public void testApplyClearCache() {
+    String sql = "CLEAR CACHE";
+    try {
+      writeStatement.execute(sql);
+    } catch (SQLException e) {
+      Assert.assertNull(e);
+    }
+  }
+
+  @Test
+  public void testApplyMerge() {
+    String sql = "MERGE";
+    try {
+      writeStatement.execute(sql);
+    } catch (SQLException e) {
+      Assert.assertNull(e);
+    }
+  }
+
+  @Test
+  public void testCreateSnapshot() {
+    String sql = "CREATE SNAPSHOT FOR SCHEMA";
+    try {
+      writeStatement.execute(sql);
+    } catch (SQLException e) {
+      Assert.assertNull(e);
+    }
+  }
 }