You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ge...@apache.org on 2020/05/01 17:22:21 UTC
[incubator-iotdb] branch move_flush created (now 964030b)
This is an automated email from the ASF dual-hosted git repository.
geniuspig pushed a change to branch move_flush
in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git.
at 964030b fix bugs.
This branch includes the following new commits:
new 529d282 move flush
new 964030b fix bugs.
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[incubator-iotdb] 02/02: fix bugs.
Posted by ge...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
geniuspig pushed a commit to branch move_flush
in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git
commit 964030b5cf14be825499197c573530e93e4a2b21
Author: zhutianci <zh...@gmail.com>
AuthorDate: Sat May 2 01:21:53 2020 +0800
fix bugs.
---
server/src/main/antlr4/org/apache/iotdb/db/qp/strategy/SqlBase.g4 | 2 +-
server/src/main/java/org/apache/iotdb/db/mqtt/PublishHandler.java | 7 +++++--
server/src/main/java/org/apache/iotdb/db/qp/Planner.java | 2 ++
.../main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java | 1 +
.../src/main/java/org/apache/iotdb/db/qp/logical/RootOperator.java | 1 -
.../java/org/apache/iotdb/db/qp/logical/sys/MergeOperator.java | 1 +
.../src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java | 2 +-
7 files changed, 11 insertions(+), 5 deletions(-)
diff --git a/server/src/main/antlr4/org/apache/iotdb/db/qp/strategy/SqlBase.g4 b/server/src/main/antlr4/org/apache/iotdb/db/qp/strategy/SqlBase.g4
index f21fbc1..e75f4d1 100644
--- a/server/src/main/antlr4/org/apache/iotdb/db/qp/strategy/SqlBase.g4
+++ b/server/src/main/antlr4/org/apache/iotdb/db/qp/strategy/SqlBase.g4
@@ -37,7 +37,7 @@ statement
| CREATE INDEX ON fullPath USING function=ID indexWithClause? whereClause? #createIndex //not support yet
| DROP INDEX function=ID ON fullPath #dropIndex //not support yet
| MERGE #merge
- | FLUSH (COMMA prefixPath)* (ID)?#flush //ID is true or false
+ | FLUSH prefixPath? (COMMA prefixPath)* (ID)?#flush //ID is true or false
| FULL MERGE #fullMerge
| CREATE USER userName=ID password=STRING_LITERAL #createUser
| ALTER USER userName=(ROOT|ID) SET PASSWORD password=STRING_LITERAL #alterUser
diff --git a/server/src/main/java/org/apache/iotdb/db/mqtt/PublishHandler.java b/server/src/main/java/org/apache/iotdb/db/mqtt/PublishHandler.java
index 57cb3aa..fcbaef2 100644
--- a/server/src/main/java/org/apache/iotdb/db/mqtt/PublishHandler.java
+++ b/server/src/main/java/org/apache/iotdb/db/mqtt/PublishHandler.java
@@ -23,6 +23,8 @@ import io.netty.buffer.ByteBuf;
import io.netty.handler.codec.mqtt.MqttQoS;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.exception.StorageEngineException;
+import org.apache.iotdb.db.exception.metadata.StorageGroupNotSetException;
import org.apache.iotdb.db.exception.query.QueryProcessException;
import org.apache.iotdb.db.qp.executor.IPlanExecutor;
import org.apache.iotdb.db.qp.executor.PlanExecutor;
@@ -92,7 +94,7 @@ public class PublishHandler extends AbstractInterceptHandler {
boolean status;
try {
status = executeNonQuery(plan);
- } catch (QueryProcessException e) {
+ } catch (QueryProcessException | StorageGroupNotSetException | StorageEngineException e ) {
throw new RuntimeException(e);
}
@@ -100,7 +102,8 @@ public class PublishHandler extends AbstractInterceptHandler {
}
}
- private boolean executeNonQuery(PhysicalPlan plan) throws QueryProcessException {
+ private boolean executeNonQuery(PhysicalPlan plan)
+ throws QueryProcessException, StorageGroupNotSetException, StorageEngineException {
if (IoTDBDescriptor.getInstance().getConfig().isReadOnly()) {
throw new QueryProcessException(
"Current system mode is read-only, does not support non-query operation");
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/Planner.java b/server/src/main/java/org/apache/iotdb/db/qp/Planner.java
index 15a3a14..fce62a7 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/Planner.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/Planner.java
@@ -95,6 +95,8 @@ public class Planner {
case LOAD_FILES:
case REMOVE_FILE:
case MOVE_FILE:
+ case FLUSH:
+ case MERGE:
return operator;
case QUERY:
case UPDATE:
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java b/server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java
index 3e4f8b8..2183923 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java
@@ -239,6 +239,7 @@ public class PlanExecutor implements IPlanExecutor {
return true;
case MERGE:
operateMerge((MergePlan) plan);
+ return true;
default:
throw new UnsupportedOperationException(
String.format("operation %s is not supported", plan.getOperatorType()));
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/logical/RootOperator.java b/server/src/main/java/org/apache/iotdb/db/qp/logical/RootOperator.java
index edb44ed..adfa8c7 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/logical/RootOperator.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/logical/RootOperator.java
@@ -26,6 +26,5 @@ public abstract class RootOperator extends Operator {
public RootOperator(int tokenIntType) {
super(tokenIntType);
- operatorType = OperatorType.MERGE;
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/logical/sys/MergeOperator.java b/server/src/main/java/org/apache/iotdb/db/qp/logical/sys/MergeOperator.java
index 407019e..16b9a5f 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/logical/sys/MergeOperator.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/logical/sys/MergeOperator.java
@@ -24,5 +24,6 @@ public class MergeOperator extends RootOperator {
public MergeOperator(int tokenIntType) {
super(tokenIntType);
+ operatorType = OperatorType.MERGE;
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java b/server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java
index 58f15f8..4903884 100644
--- a/server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java
+++ b/server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java
@@ -898,7 +898,7 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext {
}
private boolean executeNonQuery(PhysicalPlan plan)
- throws QueryProcessException, StorageGroupNotSetException {
+ throws QueryProcessException, StorageGroupNotSetException, StorageEngineException {
if (IoTDBDescriptor.getInstance().getConfig().isReadOnly()) {
throw new QueryProcessException(
"Current system mode is read-only, does not support non-query operation");
[incubator-iotdb] 01/02: move flush
Posted by ge...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
geniuspig pushed a commit to branch move_flush
in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git
commit 529d28296d3d067945e80ce314e0d2ca58824cf9
Author: zhutianci <zh...@gmail.com>
AuthorDate: Fri May 1 17:52:40 2020 +0800
move flush
---
.../org/apache/iotdb/db/qp/strategy/SqlBase.g4 | 7 ++-
.../apache/iotdb/db/qp/constant/SQLConstant.java | 4 ++
.../apache/iotdb/db/qp/executor/IPlanExecutor.java | 4 +-
.../apache/iotdb/db/qp/executor/PlanExecutor.java | 43 +++++++++++++-
.../org/apache/iotdb/db/qp/logical/Operator.java | 2 +-
.../apache/iotdb/db/qp/logical/RootOperator.java | 2 +-
.../{RootOperator.java => sys/FlushOperator.java} | 37 +++++++++---
.../{RootOperator.java => sys/MergeOperator.java} | 15 ++---
.../sys/FlushPlan.java} | 30 +++++++---
.../sys/MergePlan.java} | 26 ++++++---
.../iotdb/db/qp/strategy/LogicalGenerator.java | 36 ++++++++++++
.../iotdb/db/qp/strategy/PhysicalGenerator.java | 9 +++
.../org/apache/iotdb/db/service/TSServiceImpl.java | 66 +---------------------
13 files changed, 177 insertions(+), 104 deletions(-)
diff --git a/server/src/main/antlr4/org/apache/iotdb/db/qp/strategy/SqlBase.g4 b/server/src/main/antlr4/org/apache/iotdb/db/qp/strategy/SqlBase.g4
index c181d38..f21fbc1 100644
--- a/server/src/main/antlr4/org/apache/iotdb/db/qp/strategy/SqlBase.g4
+++ b/server/src/main/antlr4/org/apache/iotdb/db/qp/strategy/SqlBase.g4
@@ -36,7 +36,9 @@ statement
| DESCRIBE prefixPath #describePath // not support yet
| CREATE INDEX ON fullPath USING function=ID indexWithClause? whereClause? #createIndex //not support yet
| DROP INDEX function=ID ON fullPath #dropIndex //not support yet
- | MERGE #merge //not support yet
+ | MERGE #merge
+ | FLUSH (COMMA prefixPath)* (ID)?#flush //ID is true or false
+ | FULL MERGE #fullMerge
| CREATE USER userName=ID password=STRING_LITERAL #createUser
| ALTER USER userName=(ROOT|ID) SET PASSWORD password=STRING_LITERAL #alterUser
| DROP USER userName=ID #dropUser
@@ -817,6 +819,9 @@ RENAME
: R E N A M E
;
+FULL
+ : F U L L
+ ;
//============================
// End of the keywords list
//============================
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/constant/SQLConstant.java b/server/src/main/java/org/apache/iotdb/db/qp/constant/SQLConstant.java
index a3771a8..e7258a8 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/constant/SQLConstant.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/constant/SQLConstant.java
@@ -141,6 +141,10 @@ public class SQLConstant {
public static final int TOK_METADATA_ALTER = 80;
+ public static final int TOK_FLUSH = 81;
+ public static final int TOK_MERGE = 82;
+ public static final int TOK_FULL_MERGE = 83;
+
public static final Map<Integer, String> tokenSymbol = new HashMap<>();
public static final Map<Integer, String> tokenNames = new HashMap<>();
public static final Map<Integer, Integer> reverseWords = new HashMap<>();
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/executor/IPlanExecutor.java b/server/src/main/java/org/apache/iotdb/db/qp/executor/IPlanExecutor.java
index addf757..7e70ed2 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/executor/IPlanExecutor.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/executor/IPlanExecutor.java
@@ -22,6 +22,7 @@ import java.io.IOException;
import java.sql.SQLException;
import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.exception.metadata.MetadataException;
+import org.apache.iotdb.db.exception.metadata.StorageGroupNotSetException;
import org.apache.iotdb.db.exception.query.QueryProcessException;
import org.apache.iotdb.db.qp.physical.PhysicalPlan;
import org.apache.iotdb.db.qp.physical.crud.InsertTabletPlan;
@@ -51,7 +52,8 @@ public interface IPlanExecutor {
*
* @param plan Physical Non-Query Plan
*/
- boolean processNonQuery(PhysicalPlan plan) throws QueryProcessException;
+ boolean processNonQuery(PhysicalPlan plan)
+ throws QueryProcessException, StorageGroupNotSetException, StorageEngineException;
/**
* execute update command and return whether the operator is successful.
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java b/server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java
index 92f363e..3e4f8b8 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java
@@ -69,12 +69,14 @@ import org.apache.iotdb.db.exception.metadata.DeleteFailedException;
import org.apache.iotdb.db.exception.metadata.MetadataException;
import org.apache.iotdb.db.exception.metadata.PathAlreadyExistException;
import org.apache.iotdb.db.exception.metadata.PathNotExistException;
+import org.apache.iotdb.db.exception.metadata.StorageGroupNotSetException;
import org.apache.iotdb.db.exception.query.QueryProcessException;
import org.apache.iotdb.db.metadata.MManager;
import org.apache.iotdb.db.metadata.mnode.InternalMNode;
import org.apache.iotdb.db.metadata.mnode.LeafMNode;
import org.apache.iotdb.db.metadata.mnode.MNode;
import org.apache.iotdb.db.metadata.mnode.StorageGroupMNode;
+import org.apache.iotdb.db.qp.logical.Operator.OperatorType;
import org.apache.iotdb.db.qp.logical.sys.AuthorOperator;
import org.apache.iotdb.db.qp.logical.sys.AuthorOperator.AuthorType;
import org.apache.iotdb.db.qp.physical.PhysicalPlan;
@@ -97,6 +99,8 @@ import org.apache.iotdb.db.qp.physical.sys.CreateTimeSeriesPlan;
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;
+import org.apache.iotdb.db.qp.physical.sys.FlushPlan;
+import org.apache.iotdb.db.qp.physical.sys.MergePlan;
import org.apache.iotdb.db.qp.physical.sys.OperateFilePlan;
import org.apache.iotdb.db.qp.physical.sys.SetStorageGroupPlan;
import org.apache.iotdb.db.qp.physical.sys.SetTTLPlan;
@@ -173,7 +177,8 @@ public class PlanExecutor implements IPlanExecutor {
}
@Override
- public boolean processNonQuery(PhysicalPlan plan) throws QueryProcessException {
+ public boolean processNonQuery(PhysicalPlan plan)
+ throws QueryProcessException, StorageGroupNotSetException, StorageEngineException {
switch (plan.getOperatorType()) {
case DELETE:
delete((DeletePlan) plan);
@@ -229,12 +234,43 @@ public class PlanExecutor implements IPlanExecutor {
case MOVE_FILE:
operateMoveFile((OperateFilePlan) plan);
return true;
+ case FLUSH:
+ operateFlush((FlushPlan) plan);
+ return true;
+ case MERGE:
+ operateMerge((MergePlan) plan);
default:
throw new UnsupportedOperationException(
String.format("operation %s is not supported", plan.getOperatorType()));
}
}
+ private void operateMerge(MergePlan plan) throws StorageEngineException {
+ if(plan.getOperatorType() == OperatorType.FULL_MERGE) {
+ StorageEngine.getInstance().mergeAll(true);
+ } else {
+ StorageEngine.getInstance()
+ .mergeAll(IoTDBDescriptor.getInstance().getConfig().isForceFullMerge());
+ }
+ }
+
+ private void operateFlush(FlushPlan plan) throws StorageGroupNotSetException {
+ if(plan.getPaths() == null) {
+ StorageEngine.getInstance().syncCloseAllProcessor();
+ } else {
+ if(plan.isSeq() == null) {
+ for (Path storageGroup : plan.getPaths()) {
+ StorageEngine.getInstance().asyncCloseProcessor(storageGroup.toString(), true);
+ StorageEngine.getInstance().asyncCloseProcessor(storageGroup.toString(), false);
+ }
+ } else {
+ for (Path storageGroup : plan.getPaths()) {
+ StorageEngine.getInstance().asyncCloseProcessor(storageGroup.toString(), plan.isSeq());
+ }
+ }
+ }
+ }
+
protected QueryDataSet processDataQuery(QueryPlan queryPlan, QueryContext context)
throws StorageEngineException, QueryFilterOptimizationException, QueryProcessException,
IOException {
@@ -1035,7 +1071,7 @@ public class PlanExecutor implements IPlanExecutor {
if (!failedNames.isEmpty()) {
throw new DeleteFailedException(String.join(",", failedNames));
}
- } catch (MetadataException e) {
+ } catch (MetadataException | StorageEngineException e) {
throw new QueryProcessException(e);
}
return true;
@@ -1113,7 +1149,8 @@ public class PlanExecutor implements IPlanExecutor {
*
* @param pathList deleted paths
*/
- private void deleteDataOfTimeSeries(List<Path> pathList) throws QueryProcessException {
+ private void deleteDataOfTimeSeries(List<Path> pathList)
+ throws QueryProcessException, StorageGroupNotSetException, StorageEngineException {
for (Path p : pathList) {
DeletePlan deletePlan = new DeletePlan();
deletePlan.addPath(p);
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/logical/Operator.java b/server/src/main/java/org/apache/iotdb/db/qp/logical/Operator.java
index 37c1d10..50ccba6 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/logical/Operator.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/logical/Operator.java
@@ -75,6 +75,6 @@ public abstract class Operator {
LIST_USER_PRIVILEGE, LIST_ROLE_PRIVILEGE, LIST_USER_ROLES, LIST_ROLE_USERS,
GRANT_WATERMARK_EMBEDDING, REVOKE_WATERMARK_EMBEDDING,
TTL, DELETE_STORAGE_GROUP, LOAD_CONFIGURATION, SHOW, LOAD_FILES, REMOVE_FILE, MOVE_FILE, LAST, GROUP_BY_FILL,
- ALTER_TIMESERIES
+ ALTER_TIMESERIES, FLUSH, MERGE, FULL_MERGE
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/logical/RootOperator.java b/server/src/main/java/org/apache/iotdb/db/qp/logical/RootOperator.java
index 98da176..edb44ed 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/logical/RootOperator.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/logical/RootOperator.java
@@ -26,6 +26,6 @@ public abstract class RootOperator extends Operator {
public RootOperator(int tokenIntType) {
super(tokenIntType);
+ operatorType = OperatorType.MERGE;
}
-
}
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/logical/RootOperator.java b/server/src/main/java/org/apache/iotdb/db/qp/logical/sys/FlushOperator.java
similarity index 55%
copy from server/src/main/java/org/apache/iotdb/db/qp/logical/RootOperator.java
copy to server/src/main/java/org/apache/iotdb/db/qp/logical/sys/FlushOperator.java
index 98da176..ef9c370 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/logical/RootOperator.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/logical/sys/FlushOperator.java
@@ -16,16 +16,37 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.iotdb.db.qp.logical;
+package org.apache.iotdb.db.qp.logical.sys;
-/**
- * RootOperator indicates the operator that could be executed as a entire command. RootOperator
- * consists of SFWOperator, like INSERT/UPDATE/DELETE, and other Operators.
- */
-public abstract class RootOperator extends Operator {
+import java.util.List;
+import org.apache.iotdb.db.qp.logical.RootOperator;
+import org.apache.iotdb.tsfile.read.common.Path;
- public RootOperator(int tokenIntType) {
- super(tokenIntType);
+public class FlushOperator extends RootOperator {
+
+ public List<Path> getStorageGroupList() {
+ return storageGroupList;
}
+ public void setStorageGroupList(
+ List<Path> storageGroupList) {
+ this.storageGroupList = storageGroupList;
+ }
+
+ private List<Path> storageGroupList;
+
+ public Boolean isSeq() {
+ return isSeq;
+ }
+
+ public void setSeq(boolean seq) {
+ isSeq = seq;
+ }
+
+ private Boolean isSeq;
+
+ public FlushOperator(int tokenIntType) {
+ super(tokenIntType);
+ operatorType = OperatorType.FLUSH;
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/logical/RootOperator.java b/server/src/main/java/org/apache/iotdb/db/qp/logical/sys/MergeOperator.java
similarity index 72%
copy from server/src/main/java/org/apache/iotdb/db/qp/logical/RootOperator.java
copy to server/src/main/java/org/apache/iotdb/db/qp/logical/sys/MergeOperator.java
index 98da176..407019e 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/logical/RootOperator.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/logical/sys/MergeOperator.java
@@ -1,3 +1,7 @@
+package org.apache.iotdb.db.qp.logical.sys;
+
+import org.apache.iotdb.db.qp.logical.RootOperator;
+
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
@@ -16,16 +20,9 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.iotdb.db.qp.logical;
+public class MergeOperator extends RootOperator {
-/**
- * RootOperator indicates the operator that could be executed as a entire command. RootOperator
- * consists of SFWOperator, like INSERT/UPDATE/DELETE, and other Operators.
- */
-public abstract class RootOperator extends Operator {
-
- public RootOperator(int tokenIntType) {
+ public MergeOperator(int tokenIntType) {
super(tokenIntType);
}
-
}
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/logical/RootOperator.java b/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/FlushPlan.java
similarity index 56%
copy from server/src/main/java/org/apache/iotdb/db/qp/logical/RootOperator.java
copy to server/src/main/java/org/apache/iotdb/db/qp/physical/sys/FlushPlan.java
index 98da176..d21c26e 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/logical/RootOperator.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/FlushPlan.java
@@ -16,16 +16,30 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.iotdb.db.qp.logical;
+package org.apache.iotdb.db.qp.physical.sys;
-/**
- * RootOperator indicates the operator that could be executed as a entire command. RootOperator
- * consists of SFWOperator, like INSERT/UPDATE/DELETE, and other Operators.
- */
-public abstract class RootOperator extends Operator {
+import java.util.List;
+import org.apache.iotdb.db.qp.logical.Operator.OperatorType;
+import org.apache.iotdb.db.qp.physical.PhysicalPlan;
+import org.apache.iotdb.tsfile.read.common.Path;
+
+public class FlushPlan extends PhysicalPlan {
+ private List<Path> storeGroups;
+
+ public Boolean isSeq() {
+ return isSeq;
+ }
+
+ private Boolean isSeq;
- public RootOperator(int tokenIntType) {
- super(tokenIntType);
+ public FlushPlan(Boolean isSeq, List<Path> storeGroups) {
+ super(false, OperatorType.FLUSH);
+ this.storeGroups = storeGroups;
+ this.isSeq = isSeq;
}
+ @Override
+ public List<Path> getPaths() {
+ return storeGroups;
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/logical/RootOperator.java b/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/MergePlan.java
similarity index 60%
copy from server/src/main/java/org/apache/iotdb/db/qp/logical/RootOperator.java
copy to server/src/main/java/org/apache/iotdb/db/qp/physical/sys/MergePlan.java
index 98da176..d443002 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/logical/RootOperator.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/MergePlan.java
@@ -16,16 +16,26 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.iotdb.db.qp.logical;
+package org.apache.iotdb.db.qp.physical.sys;
-/**
- * RootOperator indicates the operator that could be executed as a entire command. RootOperator
- * consists of SFWOperator, like INSERT/UPDATE/DELETE, and other Operators.
- */
-public abstract class RootOperator extends Operator {
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.iotdb.db.qp.logical.Operator.OperatorType;
+import org.apache.iotdb.db.qp.physical.PhysicalPlan;
+import org.apache.iotdb.tsfile.read.common.Path;
+
+public class MergePlan extends PhysicalPlan {
- public RootOperator(int tokenIntType) {
- super(tokenIntType);
+ public MergePlan(OperatorType operatorType) {
+ super(false, operatorType);
}
+ public MergePlan() {
+ super(false, OperatorType.MERGE);
+ }
+
+ @Override
+ public List<Path> getPaths() {
+ return new ArrayList<>();
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/strategy/LogicalGenerator.java b/server/src/main/java/org/apache/iotdb/db/qp/strategy/LogicalGenerator.java
index 561ffe0..8647ab2 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/strategy/LogicalGenerator.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/strategy/LogicalGenerator.java
@@ -18,6 +18,7 @@
*/
package org.apache.iotdb.db.qp.strategy;
+import org.antlr.v4.runtime.misc.ParseCancellationException;
import org.antlr.v4.runtime.tree.TerminalNode;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.exception.runtime.SQLParserException;
@@ -81,6 +82,41 @@ public class LogicalGenerator extends SqlBaseBaseListener {
}
@Override
+ public void enterFlush(FlushContext ctx) {
+ super.enterFlush(ctx);
+ FlushOperator flushOperator = new FlushOperator(SQLConstant.TOK_FLUSH);
+ if(ctx.ID() != null) {
+ if(ctx.ID().getText().equalsIgnoreCase("true")
+ || ctx.ID().getText().equalsIgnoreCase("false")) {
+ flushOperator.setSeq(Boolean.parseBoolean(ctx.ID().getText()));
+ } else {
+ throw new ParseCancellationException("Should be true or false");
+ }
+ }
+ if(ctx.prefixPath(0) != null) {
+ List<Path> storageGroups = new ArrayList<>();
+ for(PrefixPathContext prefixPathContext : ctx.prefixPath()) {
+ storageGroups.add(parsePrefixPath(prefixPathContext));
+ }
+ flushOperator.setStorageGroupList(storageGroups);
+ }
+
+ initializedOperator = flushOperator;
+ }
+
+ @Override
+ public void enterMerge(MergeContext ctx) {
+ super.enterMerge(ctx);
+ initializedOperator = new MergeOperator(SQLConstant.TOK_MERGE);
+ }
+
+ @Override
+ public void enterFullMerge(FullMergeContext ctx) {
+ super.enterFullMerge(ctx);
+ initializedOperator = new MergeOperator(SQLConstant.TOK_FULL_MERGE);
+ }
+
+ @Override
public void enterCountNodes(CountNodesContext ctx) {
super.enterCountNodes(ctx);
initializedOperator = new CountOperator(SQLConstant.TOK_COUNT_NODES,
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/strategy/PhysicalGenerator.java b/server/src/main/java/org/apache/iotdb/db/qp/strategy/PhysicalGenerator.java
index 72e66b4..1ed5af4 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/strategy/PhysicalGenerator.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/strategy/PhysicalGenerator.java
@@ -126,6 +126,15 @@ public class PhysicalGenerator {
insert.getTime(),
insert.getMeasurementList(),
insert.getValueList());
+ case MERGE:
+ if(operator.getTokenIntType() == SQLConstant.TOK_FULL_MERGE) {
+ return new MergePlan(OperatorType.FULL_MERGE);
+ } else {
+ return new MergePlan();
+ }
+ case FLUSH:
+ FlushOperator flushOperator = (FlushOperator) operator;
+ return new FlushPlan(flushOperator.isSeq(), flushOperator.getStorageGroupList());
case QUERY:
QueryOperator query = (QueryOperator) operator;
return transformQuery(query);
diff --git a/server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java b/server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java
index f8a36c2..58f15f8 100644
--- a/server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java
+++ b/server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java
@@ -28,7 +28,6 @@ import org.apache.iotdb.db.conf.IoTDBConstant;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.cost.statistic.Measurement;
import org.apache.iotdb.db.cost.statistic.Operation;
-import org.apache.iotdb.db.engine.StorageEngine;
import org.apache.iotdb.db.exception.QueryInBatchStatementException;
import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.exception.metadata.MetadataException;
@@ -322,60 +321,6 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext {
return MManager.getInstance().getAllTimeseriesName(path);
}
- /**
- * Judge whether the statement is ADMIN COMMAND and if true, execute it.
- *
- * @param statement command
- * @return true if the statement is ADMIN COMMAND
- */
- private boolean execAdminCommand(String statement, long sessionId) throws StorageEngineException {
- if (!"root".equals(sessionIdUsernameMap.get(sessionId))) {
- return false;
- }
- if (statement == null) {
- return false;
- }
- statement = statement.toLowerCase();
- if (statement.startsWith("flush")) {
- try {
- execFlush(statement);
- } catch (StorageGroupNotSetException e) {
- throw new StorageEngineException(e);
- }
- return true;
- }
- switch (statement) {
- case "merge":
- StorageEngine.getInstance()
- .mergeAll(IoTDBDescriptor.getInstance().getConfig().isForceFullMerge());
- return true;
- case "full merge":
- StorageEngine.getInstance().mergeAll(true);
- return true;
- default:
- return false;
- }
- }
-
- private void execFlush(String statement) throws StorageGroupNotSetException {
- String[] args = statement.split("\\s+");
- if (args.length == 1) {
- StorageEngine.getInstance().syncCloseAllProcessor();
- } else if (args.length == 2) {
- String[] storageGroups = args[1].split(",");
- for (String storageGroup : storageGroups) {
- StorageEngine.getInstance().asyncCloseProcessor(storageGroup, true);
- StorageEngine.getInstance().asyncCloseProcessor(storageGroup, false);
- }
- } else {
- String[] storageGroups = args[1].split(",");
- boolean isSeq = Boolean.parseBoolean(args[2]);
- for (String storageGroup : storageGroups) {
- StorageEngine.getInstance().asyncCloseProcessor(storageGroup, isSeq);
- }
- }
- }
-
@Override
public TSExecuteBatchStatementResp executeBatchStatement(TSExecuteBatchStatementReq req) {
long t1 = System.currentTimeMillis();
@@ -465,10 +410,6 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext {
}
String statement = req.getStatement();
- if (execAdminCommand(statement, req.getSessionId())) {
- return RpcUtils.getTSExecuteStatementResp(
- RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS, "ADMIN_COMMAND_SUCCESS"));
- }
PhysicalPlan physicalPlan =
processor.parseSQLToPhysicalPlan(statement, sessionIdZoneIdMap.get(req.getSessionId()));
if (physicalPlan.isQuery()) {
@@ -489,10 +430,6 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext {
return RpcUtils.getTSExecuteStatementResp(
RpcUtils.getStatus(TSStatusCode.QUERY_PROCESS_ERROR,
"Meet error in query process: " + e.getMessage()));
- } catch (StorageEngineException e) {
- logger.info(ERROR_PARSING_SQL, e.getMessage());
- return RpcUtils.getTSExecuteStatementResp(
- RpcUtils.getStatus(TSStatusCode.READ_ONLY_SYSTEM_ERROR, e.getMessage()));
} catch (Exception e) {
logger.error("{}: server Internal Error: ", IoTDBConstant.GLOBAL_DB_NAME, e);
return RpcUtils.getTSExecuteStatementResp(TSStatusCode.INTERNAL_SERVER_ERROR, e.getMessage());
@@ -960,7 +897,8 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext {
return resp;
}
- private boolean executeNonQuery(PhysicalPlan plan) throws QueryProcessException {
+ private boolean executeNonQuery(PhysicalPlan plan)
+ throws QueryProcessException, StorageGroupNotSetException {
if (IoTDBDescriptor.getInstance().getConfig().isReadOnly()) {
throw new QueryProcessException(
"Current system mode is read-only, does not support non-query operation");