You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ne...@apache.org on 2022/08/03 05:15:03 UTC
[iotdb] branch master updated: [IOTDB-3758] support merge in new cluster (#6866)
This is an automated email from the ASF dual-hosted git repository.
neuyilan 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 0bc694c320 [IOTDB-3758] support merge in new cluster (#6866)
0bc694c320 is described below
commit 0bc694c320359c23f35bccb0d5ac312b3cca5784
Author: 任宇华 <79...@users.noreply.github.com>
AuthorDate: Wed Aug 3 13:14:58 2022 +0800
[IOTDB-3758] support merge in new cluster (#6866)
---
.../org/apache/iotdb/db/qp/sql/IoTDBSqlParser.g4 | 4 +-
.../confignode/client/DataNodeRequestType.java | 4 +-
.../async/datanode/AsyncDataNodeClientPool.java | 18 +-
.../client/async/handlers/MergeHandler.java | 83 +++++++++
.../iotdb/confignode/manager/ConfigManager.java | 11 +-
.../apache/iotdb/confignode/manager/IManager.java | 5 +-
.../iotdb/confignode/manager/NodeManager.java | 17 +-
.../thrift/ConfigNodeRPCServiceProcessor.java | 8 +-
.../Maintenance-Tools/Maintenance-Command.md | 14 +-
.../Maintenance-Tools/Maintenance-Command.md | 14 +-
.../apache/iotdb/db/it/IoTDBFlushQueryMergeIT.java | 193 +++++++++++++++++++++
.../apache/iotdb/db/client/ConfigNodeClient.java | 19 +-
.../apache/iotdb/db/engine/StorageEngineV2.java | 13 ++
.../iotdb/db/localconfignode/LocalConfigNode.java | 10 ++
.../plan/execution/config/ConfigTaskVisitor.java | 29 ++++
.../config/executor/ClusterConfigTaskExecutor.java | 44 +++--
.../config/executor/IConfigTaskExecutor.java | 5 +-
.../executor/StandaloneConfigTaskExecutor.java | 22 ++-
.../{ => metadata}/CountStorageGroupTask.java | 4 +-
.../config/{ => metadata}/CreateFunctionTask.java | 4 +-
.../{ => metadata}/DeleteStorageGroupTask.java | 4 +-
.../config/{ => metadata}/DropFunctionTask.java | 4 +-
.../config/{ => metadata}/SetStorageGroupTask.java | 4 +-
.../config/{ => metadata}/SetTTLTask.java | 4 +-
.../config/{ => metadata}/ShowClusterTask.java | 4 +-
.../config/{ => metadata}/ShowConfigNodesTask.java | 4 +-
.../config/{ => metadata}/ShowDataNodesTask.java | 4 +-
.../config/{ => metadata}/ShowFunctionsTask.java | 4 +-
.../config/{ => metadata}/ShowRegionTask.java | 4 +-
.../{ => metadata}/ShowStorageGroupTask.java | 4 +-
.../config/{ => metadata}/ShowTTLTask.java | 4 +-
.../config/{ => metadata}/UnSetTTLTask.java | 2 +-
.../template}/CreateSchemaTemplateTask.java | 4 +-
.../template}/SetSchemaTemplateTask.java | 4 +-
.../template}/ShowNodesInSchemaTemplateTask.java | 4 +-
.../template}/ShowPathSetTemplateTask.java | 4 +-
.../template}/ShowSchemaTemplateTask.java | 4 +-
.../execution/config/{ => sys}/AuthorizerTask.java | 4 +-
.../execution/config/{ => sys}/ClearCacheTask.java | 6 +-
.../plan/execution/config/{ => sys}/FlushTask.java | 4 +-
.../{ClearCacheTask.java => sys/MergeTask.java} | 29 ++--
.../iotdb/db/mpp/plan/parser/ASTVisitor.java | 58 +++++--
.../db/mpp/plan/statement/StatementVisitor.java | 5 +
.../db/mpp/plan/statement/sys/MergeStatement.java | 62 +++++++
.../impl/DataNodeInternalRPCServiceImpl.java | 18 +-
thrift-commons/src/main/thrift/common.thrift | 4 -
.../src/main/thrift/confignode.thrift | 17 +-
thrift/src/main/thrift/datanode.thrift | 13 +-
48 files changed, 702 insertions(+), 105 deletions(-)
diff --git a/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/IoTDBSqlParser.g4 b/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/IoTDBSqlParser.g4
index 410f32d388..dfd3ece92a 100644
--- a/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/IoTDBSqlParser.g4
+++ b/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/IoTDBSqlParser.g4
@@ -629,12 +629,12 @@ usernameWithRoot
// Merge
merge
- : MERGE
+ : MERGE (ON (LOCAL | CLUSTER))?
;
// Full Merge
fullMerge
- : FULL MERGE
+ : FULL MERGE (ON (LOCAL | CLUSTER))?
;
// Flush
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/client/DataNodeRequestType.java b/confignode/src/main/java/org/apache/iotdb/confignode/client/DataNodeRequestType.java
index 82e6974176..8eccbfe2d0 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/client/DataNodeRequestType.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/client/DataNodeRequestType.java
@@ -37,5 +37,7 @@ public enum DataNodeRequestType {
UPDATE_REGION_ROUTE_MAP,
BROADCAST_LATEST_CONFIG_NODE_GROUP,
UPDATE_TEMPLATE,
- CLEAR_CACHE
+ CLEAR_CACHE,
+ MERGE,
+ FULL_MERGE
}
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/client/async/datanode/AsyncDataNodeClientPool.java b/confignode/src/main/java/org/apache/iotdb/confignode/client/async/datanode/AsyncDataNodeClientPool.java
index 813a25d468..4348a6b68b 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/client/async/datanode/AsyncDataNodeClientPool.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/client/async/datanode/AsyncDataNodeClientPool.java
@@ -18,7 +18,6 @@
*/
package org.apache.iotdb.confignode.client.async.datanode;
-import org.apache.iotdb.common.rpc.thrift.TClearCacheReq;
import org.apache.iotdb.common.rpc.thrift.TConfigNodeLocation;
import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
import org.apache.iotdb.common.rpc.thrift.TEndPoint;
@@ -36,6 +35,7 @@ import org.apache.iotdb.confignode.client.async.handlers.CreateRegionHandler;
import org.apache.iotdb.confignode.client.async.handlers.DataNodeHeartbeatHandler;
import org.apache.iotdb.confignode.client.async.handlers.FlushHandler;
import org.apache.iotdb.confignode.client.async.handlers.FunctionManagementHandler;
+import org.apache.iotdb.confignode.client.async.handlers.MergeHandler;
import org.apache.iotdb.confignode.client.async.handlers.SetTTLHandler;
import org.apache.iotdb.confignode.client.async.handlers.UpdateConfigNodeGroupHandler;
import org.apache.iotdb.confignode.client.async.handlers.UpdateRegionRouteMapHandler;
@@ -108,6 +108,16 @@ public class AsyncDataNodeClientPool {
dataNodeLocationMap,
dataNodeResponseStatus);
break;
+ case FULL_MERGE:
+ case MERGE:
+ handler =
+ new MergeHandler(
+ countDownLatch,
+ requestType,
+ targetDataNode,
+ dataNodeLocationMap,
+ dataNodeResponseStatus);
+ break;
case FLUSH:
handler =
new FlushHandler(
@@ -177,11 +187,15 @@ public class AsyncDataNodeClientPool {
case DROP_FUNCTION:
client.dropFunction((TDropFunctionRequest) req, (FunctionManagementHandler) handler);
break;
+ case MERGE:
+ case FULL_MERGE:
+ client.merge((MergeHandler) handler);
+ break;
case FLUSH:
client.flush((TFlushReq) req, (FlushHandler) handler);
break;
case CLEAR_CACHE:
- client.clearCache((TClearCacheReq) req, (ClearCacheHandler) handler);
+ client.clearCache((ClearCacheHandler) handler);
break;
case UPDATE_REGION_ROUTE_MAP:
client.updateRegionCache((TRegionRouteReq) req, (UpdateRegionRouteMapHandler) handler);
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/MergeHandler.java b/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/MergeHandler.java
new file mode 100644
index 0000000000..5ed81310de
--- /dev/null
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/MergeHandler.java
@@ -0,0 +1,83 @@
+/*
+ * 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.client.async.handlers;
+
+import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.confignode.client.DataNodeRequestType;
+import org.apache.iotdb.rpc.RpcUtils;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.apache.thrift.async.AsyncMethodCallback;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+
+public class MergeHandler extends AbstractRetryHandler implements AsyncMethodCallback<TSStatus> {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(MergeHandler.class);
+
+ private final List<TSStatus> dataNodeResponseStatus;
+
+ public MergeHandler(
+ CountDownLatch countDownLatch,
+ DataNodeRequestType dataNodeRequestType,
+ TDataNodeLocation targetDataNode,
+ Map<Integer, TDataNodeLocation> dataNodeLocationMap,
+ List<TSStatus> dataNodeResponseStatus) {
+ super(countDownLatch, dataNodeRequestType, targetDataNode, dataNodeLocationMap);
+ this.dataNodeResponseStatus = dataNodeResponseStatus;
+ }
+
+ @Override
+ public void onComplete(TSStatus response) {
+ if (response.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ dataNodeResponseStatus.add(response);
+ dataNodeLocationMap.remove(targetDataNode.getDataNodeId());
+ LOGGER.info("Successfully {} on DataNode: {}", dataNodeRequestType, targetDataNode);
+ } else {
+ LOGGER.error(
+ "Failed to {} on DataNode {}, {}",
+ dataNodeRequestType,
+ dataNodeLocationMap.get(targetDataNode.getDataNodeId()),
+ response);
+ }
+ countDownLatch.countDown();
+ }
+
+ @Override
+ public void onError(Exception exception) {
+ countDownLatch.countDown();
+ dataNodeResponseStatus.add(
+ new TSStatus(
+ RpcUtils.getStatus(
+ TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode(),
+ dataNodeRequestType
+ + " error on DataNode: {id="
+ + targetDataNode.getDataNodeId()
+ + ", internalEndPoint="
+ + targetDataNode.getInternalEndPoint()
+ + "}"
+ + exception.getMessage())));
+ }
+}
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java b/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java
index d329f4ac26..28747e1a82 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java
@@ -19,7 +19,6 @@
package org.apache.iotdb.confignode.manager;
-import org.apache.iotdb.common.rpc.thrift.TClearCacheReq;
import org.apache.iotdb.common.rpc.thrift.TConfigNodeLocation;
import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
import org.apache.iotdb.common.rpc.thrift.TDataNodeConfiguration;
@@ -74,6 +73,7 @@ import org.apache.iotdb.confignode.persistence.UDFInfo;
import org.apache.iotdb.confignode.persistence.executor.ConfigPlanExecutor;
import org.apache.iotdb.confignode.persistence.partition.PartitionInfo;
import org.apache.iotdb.confignode.persistence.schema.ClusterSchemaInfo;
+import org.apache.iotdb.confignode.rpc.thrift.TClearCacheReq;
import org.apache.iotdb.confignode.rpc.thrift.TConfigNodeRegisterReq;
import org.apache.iotdb.confignode.rpc.thrift.TCreateSchemaTemplateReq;
import org.apache.iotdb.confignode.rpc.thrift.TDataNodeInfo;
@@ -82,6 +82,7 @@ import org.apache.iotdb.confignode.rpc.thrift.TDataPartitionTableResp;
import org.apache.iotdb.confignode.rpc.thrift.TGetAllTemplatesResp;
import org.apache.iotdb.confignode.rpc.thrift.TGetPathsSetTemplatesResp;
import org.apache.iotdb.confignode.rpc.thrift.TGetTemplateResp;
+import org.apache.iotdb.confignode.rpc.thrift.TMergeReq;
import org.apache.iotdb.confignode.rpc.thrift.TPermissionInfoResp;
import org.apache.iotdb.confignode.rpc.thrift.TRegionInfo;
import org.apache.iotdb.confignode.rpc.thrift.TRegionRouteMapResp;
@@ -796,6 +797,14 @@ public class ConfigManager implements IManager {
: status;
}
+ @Override
+ public TSStatus merge(TMergeReq req) {
+ TSStatus status = confirmLeader();
+ return status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()
+ ? RpcUtils.squashResponseStatusList(nodeManager.merge(req))
+ : status;
+ }
+
@Override
public TSStatus flush(TFlushReq req) {
TSStatus status = confirmLeader();
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/manager/IManager.java b/confignode/src/main/java/org/apache/iotdb/confignode/manager/IManager.java
index 41eeec7b93..a08abd4529 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/manager/IManager.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/manager/IManager.java
@@ -18,7 +18,6 @@
*/
package org.apache.iotdb.confignode.manager;
-import org.apache.iotdb.common.rpc.thrift.TClearCacheReq;
import org.apache.iotdb.common.rpc.thrift.TConfigNodeLocation;
import org.apache.iotdb.common.rpc.thrift.TFlushReq;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
@@ -39,11 +38,13 @@ import org.apache.iotdb.confignode.consensus.request.write.SetStorageGroupPlan;
import org.apache.iotdb.confignode.consensus.request.write.SetTTLPlan;
import org.apache.iotdb.confignode.consensus.request.write.SetTimePartitionIntervalPlan;
import org.apache.iotdb.confignode.manager.load.LoadManager;
+import org.apache.iotdb.confignode.rpc.thrift.TClearCacheReq;
import org.apache.iotdb.confignode.rpc.thrift.TConfigNodeRegisterReq;
import org.apache.iotdb.confignode.rpc.thrift.TCreateSchemaTemplateReq;
import org.apache.iotdb.confignode.rpc.thrift.TGetAllTemplatesResp;
import org.apache.iotdb.confignode.rpc.thrift.TGetPathsSetTemplatesResp;
import org.apache.iotdb.confignode.rpc.thrift.TGetTemplateResp;
+import org.apache.iotdb.confignode.rpc.thrift.TMergeReq;
import org.apache.iotdb.confignode.rpc.thrift.TPermissionInfoResp;
import org.apache.iotdb.confignode.rpc.thrift.TRegionRouteMapResp;
import org.apache.iotdb.confignode.rpc.thrift.TSchemaNodeManagementResp;
@@ -268,6 +269,8 @@ public interface IManager {
TSStatus dropFunction(String udfName);
+ TSStatus merge(TMergeReq req);
+
TSStatus flush(TFlushReq req);
TSStatus clearCache(TClearCacheReq req);
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/manager/NodeManager.java b/confignode/src/main/java/org/apache/iotdb/confignode/manager/NodeManager.java
index 353c65e0cd..3e8d430ac0 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/manager/NodeManager.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/manager/NodeManager.java
@@ -18,7 +18,6 @@
*/
package org.apache.iotdb.confignode.manager;
-import org.apache.iotdb.common.rpc.thrift.TClearCacheReq;
import org.apache.iotdb.common.rpc.thrift.TConfigNodeLocation;
import org.apache.iotdb.common.rpc.thrift.TDataNodeConfiguration;
import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
@@ -40,9 +39,11 @@ import org.apache.iotdb.confignode.consensus.response.DataNodeToStatusResp;
import org.apache.iotdb.confignode.manager.load.LoadManager;
import org.apache.iotdb.confignode.persistence.NodeInfo;
import org.apache.iotdb.confignode.procedure.env.DataNodeRemoveHandler;
+import org.apache.iotdb.confignode.rpc.thrift.TClearCacheReq;
import org.apache.iotdb.confignode.rpc.thrift.TConfigNodeInfo;
import org.apache.iotdb.confignode.rpc.thrift.TDataNodeInfo;
import org.apache.iotdb.confignode.rpc.thrift.TGlobalConfig;
+import org.apache.iotdb.confignode.rpc.thrift.TMergeReq;
import org.apache.iotdb.consensus.common.DataSet;
import org.apache.iotdb.consensus.common.Peer;
import org.apache.iotdb.consensus.common.response.ConsensusGenericResponse;
@@ -353,12 +354,22 @@ public class NodeManager {
+ ".");
}
- public List<TSStatus> flush(TFlushReq req) {
+ public List<TSStatus> merge(TMergeReq req) {
Map<Integer, TDataNodeLocation> dataNodeLocationMap =
configManager.getNodeManager().getRegisteredDataNodeLocations(req.dataNodeId);
List<TSStatus> dataNodeResponseStatus =
Collections.synchronizedList(new ArrayList<>(dataNodeLocationMap.size()));
+ AsyncDataNodeClientPool.getInstance()
+ .sendAsyncRequestToDataNodeWithRetry(
+ null, dataNodeLocationMap, DataNodeRequestType.MERGE, dataNodeResponseStatus);
+ return dataNodeResponseStatus;
+ }
+ public List<TSStatus> flush(TFlushReq req) {
+ Map<Integer, TDataNodeLocation> dataNodeLocationMap =
+ configManager.getNodeManager().getRegisteredDataNodeLocations(req.dataNodeId);
+ List<TSStatus> dataNodeResponseStatus =
+ Collections.synchronizedList(new ArrayList<>(dataNodeLocationMap.size()));
AsyncDataNodeClientPool.getInstance()
.sendAsyncRequestToDataNodeWithRetry(
req, dataNodeLocationMap, DataNodeRequestType.FLUSH, dataNodeResponseStatus);
@@ -372,7 +383,7 @@ public class NodeManager {
Collections.synchronizedList(new ArrayList<>(dataNodeLocationMap.size()));
AsyncDataNodeClientPool.getInstance()
.sendAsyncRequestToDataNodeWithRetry(
- req, dataNodeLocationMap, DataNodeRequestType.CLEAR_CACHE, dataNodeResponseStatus);
+ null, dataNodeLocationMap, DataNodeRequestType.CLEAR_CACHE, dataNodeResponseStatus);
return dataNodeResponseStatus;
}
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java b/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java
index 6c5aa24a89..4b13c383e7 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java
@@ -18,7 +18,6 @@
*/
package org.apache.iotdb.confignode.service.thrift;
-import org.apache.iotdb.common.rpc.thrift.TClearCacheReq;
import org.apache.iotdb.common.rpc.thrift.TConfigNodeLocation;
import org.apache.iotdb.common.rpc.thrift.TFlushReq;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
@@ -63,6 +62,7 @@ import org.apache.iotdb.confignode.rpc.thrift.TAddConsensusGroupReq;
import org.apache.iotdb.confignode.rpc.thrift.TAuthorizerReq;
import org.apache.iotdb.confignode.rpc.thrift.TAuthorizerResp;
import org.apache.iotdb.confignode.rpc.thrift.TCheckUserPrivilegesReq;
+import org.apache.iotdb.confignode.rpc.thrift.TClearCacheReq;
import org.apache.iotdb.confignode.rpc.thrift.TConfigNodeRegisterReq;
import org.apache.iotdb.confignode.rpc.thrift.TCountStorageGroupResp;
import org.apache.iotdb.confignode.rpc.thrift.TCreateFunctionReq;
@@ -82,6 +82,7 @@ import org.apache.iotdb.confignode.rpc.thrift.TGetAllTemplatesResp;
import org.apache.iotdb.confignode.rpc.thrift.TGetPathsSetTemplatesResp;
import org.apache.iotdb.confignode.rpc.thrift.TGetTemplateResp;
import org.apache.iotdb.confignode.rpc.thrift.TLoginReq;
+import org.apache.iotdb.confignode.rpc.thrift.TMergeReq;
import org.apache.iotdb.confignode.rpc.thrift.TPermissionInfoResp;
import org.apache.iotdb.confignode.rpc.thrift.TRegionMigrateResultReportReq;
import org.apache.iotdb.confignode.rpc.thrift.TRegionRouteMapResp;
@@ -510,6 +511,11 @@ public class ConfigNodeRPCServiceProcessor implements IConfigNodeRPCService.Ifac
return configManager.dropFunction(req.getUdfName());
}
+ @Override
+ public TSStatus merge(TMergeReq req) throws TException {
+ return configManager.merge(req);
+ }
+
@Override
public TSStatus flush(TFlushReq req) throws TException {
if (req.storageGroups != null) {
diff --git a/docs/UserGuide/Maintenance-Tools/Maintenance-Command.md b/docs/UserGuide/Maintenance-Tools/Maintenance-Command.md
index 3b3087e218..323e3df90a 100644
--- a/docs/UserGuide/Maintenance-Tools/Maintenance-Command.md
+++ b/docs/UserGuide/Maintenance-Tools/Maintenance-Command.md
@@ -22,7 +22,7 @@
# Maintenance Command
## FLUSH
-Persist all the data points in the memory table of the storage group to the disk, and seal the data file.
+Persist all the data points in the memory table of the storage group to the disk, and seal the data file.In cluster mode, we provide commands to persist the specified storage group cache of a single node and persist the specified storage group cache of the cluster.
Note: This command does not need to be invoked manually by the client. IoTDB has WAL to ensure data security
and IoTDB will flush when appropriate.
@@ -30,8 +30,11 @@ Frequently call flush can result in small data files that degrade query performa
```sql
IoTDB> FLUSH
+IoTDB> FLUSH ON LOCAL
+IoTDB> FLUSH ON CLUSTER
IoTDB> FLUSH root.ln
-IoTDB> FLUSH root.sg1,root.sg2
+IoTDB> FLUSH root.sg1,root.sg2 ON LOCAL
+IoTDB> FLUSH root.sg1,root.sg2 ON CLUSTER
```
## MERGE
@@ -45,6 +48,13 @@ Execute Level Compaction and unsequence Compaction task. Currently IoTDB support
IoTDB> MERGE
IoTDB> FULL MERGE
```
+At the same time, manually trigger the compaction process of data files are supported for individual nodes or the entire cluster in cluster mode:
+```sql
+IoTDB> MERGE ON LOCAL
+IoTDB> MERGE ON CLUSTER
+IoTDB> FULL MERGE ON LOCAL
+IoTDB> FULL MERGE ON CLUSTER
+```
## CLEAR CACHE
diff --git a/docs/zh/UserGuide/Maintenance-Tools/Maintenance-Command.md b/docs/zh/UserGuide/Maintenance-Tools/Maintenance-Command.md
index d983f05c06..4a31815b3d 100644
--- a/docs/zh/UserGuide/Maintenance-Tools/Maintenance-Command.md
+++ b/docs/zh/UserGuide/Maintenance-Tools/Maintenance-Command.md
@@ -22,15 +22,18 @@
## FLUSH
-将指定存储组的内存缓存区 Memory Table 的数据持久化到磁盘上，并将数据文件封口。
+将指定存储组的内存缓存区 Memory Table 的数据持久化到磁盘上，并将数据文件封口。在集群模式下，我们提供了持久化单个节点的指定存储组的缓存、持久化整个集群指定存储组的缓存命令。
注意：此命令客户端不需要手动调用，IoTDB 有 wal 保证数据安全，IoTDB 会选择合适的时机进行 flush。
如果频繁调用 flush 会导致数据文件很小，降低查询性能。
```sql
IoTDB> FLUSH
+IoTDB> FLUSH ON LOCAL
+IoTDB> FLUSH ON CLUSTER
IoTDB> FLUSH root.ln
-IoTDB> FLUSH root.sg1,root.sg2
+IoTDB> FLUSH root.sg1,root.sg2 ON LOCAL
+IoTDB> FLUSH root.sg1,root.sg2 ON CLUSTER
```
## MERGE
@@ -44,6 +47,13 @@ IoTDB> FLUSH root.sg1,root.sg2
IoTDB> MERGE
IoTDB> FULL MERGE
```
+同时，在集群模式中支持对单个节点或整个集群手动触发数据文件的合并:
+```sql
+IoTDB> MERGE ON LOCAL
+IoTDB> MERGE ON CLUSTER
+IoTDB> FULL MERGE ON LOCAL
+IoTDB> FULL MERGE ON CLUSTER
+```
## CLEAR CACHE
diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBFlushQueryMergeIT.java b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBFlushQueryMergeIT.java
new file mode 100644
index 0000000000..d11cde5c29
--- /dev/null
+++ b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBFlushQueryMergeIT.java
@@ -0,0 +1,193 @@
+/*
+ * 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.db.it;
+
+import org.apache.iotdb.it.env.EnvFactory;
+import org.apache.iotdb.it.framework.IoTDBTestRunner;
+import org.apache.iotdb.itbase.category.ClusterIT;
+import org.apache.iotdb.itbase.category.LocalStandaloneIT;
+
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Ignore;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.Locale;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+
+@RunWith(IoTDBTestRunner.class)
+@Category({LocalStandaloneIT.class, ClusterIT.class})
+public class IoTDBFlushQueryMergeIT {
+
+ private static String[] sqls =
+ new String[] {
+ "SET STORAGE GROUP TO root.vehicle.d0",
+ "CREATE TIMESERIES root.vehicle.d0.s0 WITH DATATYPE=INT32, ENCODING=RLE",
+ "insert into root.vehicle.d0(timestamp,s0) values(1,101)",
+ "insert into root.vehicle.d0(timestamp,s0) values(2,198)",
+ "insert into root.vehicle.d0(timestamp,s0) values(100,99)",
+ "insert into root.vehicle.d0(timestamp,s0) values(101,99)",
+ "insert into root.vehicle.d0(timestamp,s0) values(102,80)",
+ "insert into root.vehicle.d0(timestamp,s0) values(103,99)",
+ "insert into root.vehicle.d0(timestamp,s0) values(104,90)",
+ "insert into root.vehicle.d0(timestamp,s0) values(105,99)",
+ "insert into root.vehicle.d0(timestamp,s0) values(106,99)",
+ "flush",
+ "insert into root.vehicle.d0(timestamp,s0) values(2,10000)",
+ "insert into root.vehicle.d0(timestamp,s0) values(50,10000)",
+ "insert into root.vehicle.d0(timestamp,s0) values(1000,22222)",
+ };
+
+ @BeforeClass
+ public static void setUp() throws Exception {
+ Locale.setDefault(Locale.ENGLISH);
+ EnvFactory.getEnv().initBeforeClass();
+ insertData();
+ }
+
+ @AfterClass
+ public static void tearDown() throws Exception {
+ EnvFactory.getEnv().cleanAfterClass();
+ }
+
+ private static void insertData() {
+ try (Connection connection = EnvFactory.getEnv().getConnection();
+ Statement statement = connection.createStatement()) {
+ for (String sql : sqls) {
+ statement.execute(sql);
+ }
+ } catch (Exception e) {
+ Assert.fail("insertData failed.");
+ }
+ }
+
+ @Test
+ public void selectAllSQLTest() {
+
+ try (Connection connection = EnvFactory.getEnv().getConnection();
+ Statement statement = connection.createStatement()) {
+ try (ResultSet resultSet = statement.executeQuery("SELECT * FROM root.**"); ) {
+ int cnt = 0;
+ while (resultSet.next()) {
+ cnt++;
+ }
+ }
+ statement.execute("merge");
+ } catch (Exception e) {
+ Assert.fail("selectAllSQLTest failed.");
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testFlushGivenGroup() {
+ String insertTemplate =
+ "INSERT INTO root.group%d(timestamp, s1, s2, s3) VALUES (%d, %d, %f, %s)";
+ try (Connection connection = EnvFactory.getEnv().getConnection();
+ Statement statement = connection.createStatement()) {
+ statement.execute("SET STORAGE GROUP TO root.group1");
+ statement.execute("SET STORAGE GROUP TO root.group2");
+ statement.execute("SET STORAGE GROUP TO root.group3");
+
+ for (int i = 1; i <= 3; i++) {
+ for (int j = 10; j < 20; j++) {
+ statement.execute(String.format(insertTemplate, i, j, j, j * 0.1, j));
+ }
+ }
+ statement.execute("FLUSH");
+
+ for (int i = 1; i <= 3; i++) {
+ for (int j = 0; j < 10; j++) {
+ statement.execute(String.format(insertTemplate, i, j, j, j * 0.1, j));
+ }
+ }
+ statement.execute("FLUSH root.group1");
+ statement.execute("FLUSH root.group2,root.group3");
+
+ for (int i = 1; i <= 3; i++) {
+ for (int j = 0; j < 30; j++) {
+ statement.execute(String.format(insertTemplate, i, j, j, j * 0.1, j));
+ }
+ }
+ statement.execute("FLUSH root.group1 TRUE");
+ statement.execute("FLUSH root.group2,root.group3 FALSE");
+
+ int i = 0;
+ try (ResultSet resultSet =
+ statement.executeQuery("SELECT * FROM root.group1,root.group2,root" + ".group3")) {
+ while (resultSet.next()) {
+ i++;
+ }
+ }
+ assertEquals(30, i);
+
+ } catch (Exception e) {
+ Assert.fail("testFlushGivenGroup failed.");
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testFlushGivenGroupNoData() {
+ try (Connection connection = EnvFactory.getEnv().getConnection();
+ Statement statement = connection.createStatement()) {
+ statement.execute("SET STORAGE GROUP TO root.nodatagroup1");
+ statement.execute("SET STORAGE GROUP TO root.nodatagroup2");
+ statement.execute("SET STORAGE GROUP TO root.nodatagroup3");
+ statement.execute("FLUSH root.nodatagroup1");
+ statement.execute("FLUSH root.nodatagroup2");
+ statement.execute("FLUSH root.nodatagroup3");
+ statement.execute("FLUSH root.nodatagroup1, root.nodatagroup2");
+ } catch (Exception e) {
+ Assert.fail("testFlushGivenGroupNoData failed.");
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ @Ignore
+ public void testFlushNotExistGroupNoData() {
+ try (Connection connection = EnvFactory.getEnv().getConnection();
+ Statement statement = connection.createStatement()) {
+ statement.execute("SET STORAGE GROUP TO root.noexist.nodatagroup1");
+ try {
+ statement.execute(
+ "FLUSH root.noexist.nodatagroup1,root.notExistGroup1,root.notExistGroup2");
+ } catch (SQLException sqe) {
+ String expectedMsg =
+ "322: 322: storageGroup root.notExistGroup1,root.notExistGroup2 does not exist";
+ sqe.printStackTrace();
+ Assert.assertTrue(sqe.getMessage().contains(expectedMsg));
+ }
+ } catch (Exception e) {
+ Assert.fail("testFlushNotExistGroupNoData failed.");
+ fail(e.getMessage());
+ }
+ }
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/client/ConfigNodeClient.java b/server/src/main/java/org/apache/iotdb/db/client/ConfigNodeClient.java
index 447e276dc8..9354a2aa14 100644
--- a/server/src/main/java/org/apache/iotdb/db/client/ConfigNodeClient.java
+++ b/server/src/main/java/org/apache/iotdb/db/client/ConfigNodeClient.java
@@ -19,7 +19,6 @@
package org.apache.iotdb.db.client;
-import org.apache.iotdb.common.rpc.thrift.TClearCacheReq;
import org.apache.iotdb.common.rpc.thrift.TConfigNodeLocation;
import org.apache.iotdb.common.rpc.thrift.TEndPoint;
import org.apache.iotdb.common.rpc.thrift.TFlushReq;
@@ -37,6 +36,7 @@ import org.apache.iotdb.confignode.rpc.thrift.TAddConsensusGroupReq;
import org.apache.iotdb.confignode.rpc.thrift.TAuthorizerReq;
import org.apache.iotdb.confignode.rpc.thrift.TAuthorizerResp;
import org.apache.iotdb.confignode.rpc.thrift.TCheckUserPrivilegesReq;
+import org.apache.iotdb.confignode.rpc.thrift.TClearCacheReq;
import org.apache.iotdb.confignode.rpc.thrift.TConfigNodeRegisterReq;
import org.apache.iotdb.confignode.rpc.thrift.TCountStorageGroupResp;
import org.apache.iotdb.confignode.rpc.thrift.TCreateFunctionReq;
@@ -56,6 +56,7 @@ import org.apache.iotdb.confignode.rpc.thrift.TGetAllTemplatesResp;
import org.apache.iotdb.confignode.rpc.thrift.TGetPathsSetTemplatesResp;
import org.apache.iotdb.confignode.rpc.thrift.TGetTemplateResp;
import org.apache.iotdb.confignode.rpc.thrift.TLoginReq;
+import org.apache.iotdb.confignode.rpc.thrift.TMergeReq;
import org.apache.iotdb.confignode.rpc.thrift.TPermissionInfoResp;
import org.apache.iotdb.confignode.rpc.thrift.TRegionMigrateResultReportReq;
import org.apache.iotdb.confignode.rpc.thrift.TRegionRouteMapResp;
@@ -748,6 +749,22 @@ public class ConfigNodeClient
throw new TException(MSG_RECONNECTION_FAIL);
}
+ @Override
+ public TSStatus merge(TMergeReq req) throws TException {
+ for (int i = 0; i < RETRY_NUM; i++) {
+ try {
+ TSStatus status = client.merge(req);
+ if (!updateConfigNodeLeader(status)) {
+ return status;
+ }
+ } catch (TException e) {
+ configLeader = null;
+ }
+ reconnect();
+ }
+ throw new TException(MSG_RECONNECTION_FAIL);
+ }
+
@Override
public TSStatus flush(TFlushReq req) throws TException {
for (int i = 0; i < RETRY_NUM; i++) {
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/StorageEngineV2.java b/server/src/main/java/org/apache/iotdb/db/engine/StorageEngineV2.java
index efb859e1ba..2c51263a1e 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/StorageEngineV2.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/StorageEngineV2.java
@@ -43,6 +43,7 @@ import org.apache.iotdb.db.engine.flush.TsFileFlushPolicy.DirectFlushPolicy;
import org.apache.iotdb.db.engine.storagegroup.DataRegion;
import org.apache.iotdb.db.engine.storagegroup.TsFileProcessor;
import org.apache.iotdb.db.exception.DataRegionException;
+import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.exception.TsFileProcessorException;
import org.apache.iotdb.db.exception.WriteProcessRejectException;
import org.apache.iotdb.db.exception.runtime.StorageEngineFailureException;
@@ -547,6 +548,18 @@ public class StorageEngineV2 implements IService {
}
}
+ /**
+ * merge all storage groups.
+ *
+ * @throws StorageEngineException StorageEngineException
+ */
+ public void mergeAll() throws StorageEngineException {
+ if (IoTDBDescriptor.getInstance().getConfig().isReadOnly()) {
+ throw new StorageEngineException("Current system mode is read only, does not support merge");
+ }
+ dataRegionMap.values().forEach(DataRegion::compact);
+ }
+
public TSStatus operateFlush(TFlushReq req) {
if (req.storageGroups == null) {
StorageEngineV2.getInstance().syncCloseAllProcessor();
diff --git a/server/src/main/java/org/apache/iotdb/db/localconfignode/LocalConfigNode.java b/server/src/main/java/org/apache/iotdb/db/localconfignode/LocalConfigNode.java
index 7fca93226b..e965551040 100644
--- a/server/src/main/java/org/apache/iotdb/db/localconfignode/LocalConfigNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/localconfignode/LocalConfigNode.java
@@ -56,6 +56,7 @@ import org.apache.iotdb.db.engine.cache.ChunkCache;
import org.apache.iotdb.db.engine.cache.TimeSeriesMetadataCache;
import org.apache.iotdb.db.engine.storagegroup.DataRegion;
import org.apache.iotdb.db.exception.DataRegionException;
+import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.exception.metadata.PathNotExistException;
import org.apache.iotdb.db.exception.metadata.StorageGroupAlreadySetException;
import org.apache.iotdb.db.exception.metadata.StorageGroupNotSetException;
@@ -1304,6 +1305,15 @@ public class LocalConfigNode {
return iAuthorizer.checkUserPrivileges(username, path, permission);
}
+ public TSStatus executeMergeOperation() {
+ try {
+ storageEngine.mergeAll();
+ } catch (StorageEngineException e) {
+ return RpcUtils.getStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR, e.getMessage());
+ }
+ return RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS);
+ }
+
public TSStatus executeFlushOperation(TFlushReq tFlushReq) {
return storageEngine.operateFlush(tFlushReq);
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ConfigTaskVisitor.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ConfigTaskVisitor.java
index c5c4b2f822..0ef41c3612 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ConfigTaskVisitor.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ConfigTaskVisitor.java
@@ -19,6 +19,29 @@
package org.apache.iotdb.db.mpp.plan.execution.config;
+import org.apache.iotdb.db.mpp.plan.execution.config.metadata.CountStorageGroupTask;
+import org.apache.iotdb.db.mpp.plan.execution.config.metadata.CreateFunctionTask;
+import org.apache.iotdb.db.mpp.plan.execution.config.metadata.DeleteStorageGroupTask;
+import org.apache.iotdb.db.mpp.plan.execution.config.metadata.DropFunctionTask;
+import org.apache.iotdb.db.mpp.plan.execution.config.metadata.SetStorageGroupTask;
+import org.apache.iotdb.db.mpp.plan.execution.config.metadata.SetTTLTask;
+import org.apache.iotdb.db.mpp.plan.execution.config.metadata.ShowClusterTask;
+import org.apache.iotdb.db.mpp.plan.execution.config.metadata.ShowConfigNodesTask;
+import org.apache.iotdb.db.mpp.plan.execution.config.metadata.ShowDataNodesTask;
+import org.apache.iotdb.db.mpp.plan.execution.config.metadata.ShowFunctionsTask;
+import org.apache.iotdb.db.mpp.plan.execution.config.metadata.ShowRegionTask;
+import org.apache.iotdb.db.mpp.plan.execution.config.metadata.ShowStorageGroupTask;
+import org.apache.iotdb.db.mpp.plan.execution.config.metadata.ShowTTLTask;
+import org.apache.iotdb.db.mpp.plan.execution.config.metadata.UnSetTTLTask;
+import org.apache.iotdb.db.mpp.plan.execution.config.metadata.template.CreateSchemaTemplateTask;
+import org.apache.iotdb.db.mpp.plan.execution.config.metadata.template.SetSchemaTemplateTask;
+import org.apache.iotdb.db.mpp.plan.execution.config.metadata.template.ShowNodesInSchemaTemplateTask;
+import org.apache.iotdb.db.mpp.plan.execution.config.metadata.template.ShowPathSetTemplateTask;
+import org.apache.iotdb.db.mpp.plan.execution.config.metadata.template.ShowSchemaTemplateTask;
+import org.apache.iotdb.db.mpp.plan.execution.config.sys.AuthorizerTask;
+import org.apache.iotdb.db.mpp.plan.execution.config.sys.ClearCacheTask;
+import org.apache.iotdb.db.mpp.plan.execution.config.sys.FlushTask;
+import org.apache.iotdb.db.mpp.plan.execution.config.sys.MergeTask;
import org.apache.iotdb.db.mpp.plan.statement.Statement;
import org.apache.iotdb.db.mpp.plan.statement.StatementNode;
import org.apache.iotdb.db.mpp.plan.statement.StatementVisitor;
@@ -44,6 +67,7 @@ import org.apache.iotdb.db.mpp.plan.statement.metadata.template.ShowSchemaTempla
import org.apache.iotdb.db.mpp.plan.statement.sys.AuthorStatement;
import org.apache.iotdb.db.mpp.plan.statement.sys.ClearCacheStatement;
import org.apache.iotdb.db.mpp.plan.statement.sys.FlushStatement;
+import org.apache.iotdb.db.mpp.plan.statement.sys.MergeStatement;
import org.apache.iotdb.tsfile.exception.NotImplementedException;
public class ConfigTaskVisitor
@@ -115,6 +139,11 @@ public class ConfigTaskVisitor
return new CreateFunctionTask(createFunctionStatement);
}
+ @Override
+ public IConfigTask visitMerge(MergeStatement mergeStatement, TaskContext context) {
+ return new MergeTask(mergeStatement);
+ }
+
@Override
public IConfigTask visitFlush(FlushStatement flushStatement, TaskContext context) {
return new FlushTask(flushStatement);
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/executor/ClusterConfigTaskExecutor.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/executor/ClusterConfigTaskExecutor.java
index ee22753c99..5e85c95f66 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/executor/ClusterConfigTaskExecutor.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/executor/ClusterConfigTaskExecutor.java
@@ -19,7 +19,6 @@
package org.apache.iotdb.db.mpp.plan.execution.config.executor;
-import org.apache.iotdb.common.rpc.thrift.TClearCacheReq;
import org.apache.iotdb.common.rpc.thrift.TFlushReq;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.common.rpc.thrift.TSetTTLReq;
@@ -27,11 +26,13 @@ import org.apache.iotdb.commons.client.IClientManager;
import org.apache.iotdb.commons.consensus.PartitionRegionId;
import org.apache.iotdb.commons.exception.IoTDBException;
import org.apache.iotdb.commons.path.PartialPath;
+import org.apache.iotdb.confignode.rpc.thrift.TClearCacheReq;
import org.apache.iotdb.confignode.rpc.thrift.TCountStorageGroupResp;
import org.apache.iotdb.confignode.rpc.thrift.TCreateFunctionReq;
import org.apache.iotdb.confignode.rpc.thrift.TDeleteStorageGroupsReq;
import org.apache.iotdb.confignode.rpc.thrift.TDropFunctionReq;
import org.apache.iotdb.confignode.rpc.thrift.TGetTemplateResp;
+import org.apache.iotdb.confignode.rpc.thrift.TMergeReq;
import org.apache.iotdb.confignode.rpc.thrift.TSetStorageGroupReq;
import org.apache.iotdb.confignode.rpc.thrift.TShowClusterResp;
import org.apache.iotdb.confignode.rpc.thrift.TShowConfigNodesResp;
@@ -46,17 +47,17 @@ import org.apache.iotdb.db.client.DataNodeClientPoolFactory;
import org.apache.iotdb.db.metadata.template.ClusterTemplateManager;
import org.apache.iotdb.db.metadata.template.Template;
import org.apache.iotdb.db.mpp.plan.execution.config.ConfigTaskResult;
-import org.apache.iotdb.db.mpp.plan.execution.config.CountStorageGroupTask;
-import org.apache.iotdb.db.mpp.plan.execution.config.SetStorageGroupTask;
-import org.apache.iotdb.db.mpp.plan.execution.config.ShowClusterTask;
-import org.apache.iotdb.db.mpp.plan.execution.config.ShowConfigNodesTask;
-import org.apache.iotdb.db.mpp.plan.execution.config.ShowDataNodesTask;
-import org.apache.iotdb.db.mpp.plan.execution.config.ShowNodesInSchemaTemplateTask;
-import org.apache.iotdb.db.mpp.plan.execution.config.ShowPathSetTemplateTask;
-import org.apache.iotdb.db.mpp.plan.execution.config.ShowRegionTask;
-import org.apache.iotdb.db.mpp.plan.execution.config.ShowSchemaTemplateTask;
-import org.apache.iotdb.db.mpp.plan.execution.config.ShowStorageGroupTask;
-import org.apache.iotdb.db.mpp.plan.execution.config.ShowTTLTask;
+import org.apache.iotdb.db.mpp.plan.execution.config.metadata.CountStorageGroupTask;
+import org.apache.iotdb.db.mpp.plan.execution.config.metadata.SetStorageGroupTask;
+import org.apache.iotdb.db.mpp.plan.execution.config.metadata.ShowClusterTask;
+import org.apache.iotdb.db.mpp.plan.execution.config.metadata.ShowConfigNodesTask;
+import org.apache.iotdb.db.mpp.plan.execution.config.metadata.ShowDataNodesTask;
+import org.apache.iotdb.db.mpp.plan.execution.config.metadata.ShowRegionTask;
+import org.apache.iotdb.db.mpp.plan.execution.config.metadata.ShowStorageGroupTask;
+import org.apache.iotdb.db.mpp.plan.execution.config.metadata.ShowTTLTask;
+import org.apache.iotdb.db.mpp.plan.execution.config.metadata.template.ShowNodesInSchemaTemplateTask;
+import org.apache.iotdb.db.mpp.plan.execution.config.metadata.template.ShowPathSetTemplateTask;
+import org.apache.iotdb.db.mpp.plan.execution.config.metadata.template.ShowSchemaTemplateTask;
import org.apache.iotdb.db.mpp.plan.statement.metadata.CountStorageGroupStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.DeleteStorageGroupStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.SetStorageGroupStatement;
@@ -268,6 +269,25 @@ public class ClusterConfigTaskExecutor implements IConfigTaskExecutor {
return future;
}
+ @Override
+ public SettableFuture<ConfigTaskResult> merge(TMergeReq tMergeReq) {
+ SettableFuture<ConfigTaskResult> future = SettableFuture.create();
+ try (ConfigNodeClient client =
+ CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.partitionRegionId)) {
+ // Send request to some API server
+ TSStatus tsStatus = client.merge(tMergeReq);
+ // Get response or throw exception
+ if (tsStatus.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
+ } else {
+ future.setException(new StatementExecutionException(tsStatus));
+ }
+ } catch (IOException | TException e) {
+ future.setException(e);
+ }
+ return future;
+ }
+
@Override
public SettableFuture<ConfigTaskResult> flush(TFlushReq tFlushReq) {
SettableFuture<ConfigTaskResult> future = SettableFuture.create();
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/executor/IConfigTaskExecutor.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/executor/IConfigTaskExecutor.java
index 8a36ea40ac..5e13984c90 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/executor/IConfigTaskExecutor.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/executor/IConfigTaskExecutor.java
@@ -19,8 +19,9 @@
package org.apache.iotdb.db.mpp.plan.execution.config.executor;
-import org.apache.iotdb.common.rpc.thrift.TClearCacheReq;
import org.apache.iotdb.common.rpc.thrift.TFlushReq;
+import org.apache.iotdb.confignode.rpc.thrift.TClearCacheReq;
+import org.apache.iotdb.confignode.rpc.thrift.TMergeReq;
import org.apache.iotdb.db.mpp.plan.execution.config.ConfigTaskResult;
import org.apache.iotdb.db.mpp.plan.statement.metadata.CountStorageGroupStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.DeleteStorageGroupStatement;
@@ -61,6 +62,8 @@ public interface IConfigTaskExecutor {
SettableFuture<ConfigTaskResult> setTTL(SetTTLStatement setTTLStatement, String taskName);
+ SettableFuture<ConfigTaskResult> merge(TMergeReq mergeReq);
+
SettableFuture<ConfigTaskResult> flush(TFlushReq tFlushReq);
SettableFuture<ConfigTaskResult> clearCache(TClearCacheReq tClearCacheReq);
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/executor/StandaloneConfigTaskExecutor.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/executor/StandaloneConfigTaskExecutor.java
index acfac53c2f..68f19d09af 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/executor/StandaloneConfigTaskExecutor.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/executor/StandaloneConfigTaskExecutor.java
@@ -19,7 +19,6 @@
package org.apache.iotdb.db.mpp.plan.execution.config.executor;
-import org.apache.iotdb.common.rpc.thrift.TClearCacheReq;
import org.apache.iotdb.common.rpc.thrift.TFlushReq;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.exception.IoTDBException;
@@ -27,13 +26,15 @@ import org.apache.iotdb.commons.exception.MetadataException;
import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.commons.udf.service.UDFExecutableManager;
import org.apache.iotdb.commons.udf.service.UDFRegistrationService;
+import org.apache.iotdb.confignode.rpc.thrift.TClearCacheReq;
+import org.apache.iotdb.confignode.rpc.thrift.TMergeReq;
import org.apache.iotdb.confignode.rpc.thrift.TStorageGroupSchema;
import org.apache.iotdb.db.localconfignode.LocalConfigNode;
import org.apache.iotdb.db.metadata.mnode.IStorageGroupMNode;
import org.apache.iotdb.db.mpp.plan.execution.config.ConfigTaskResult;
-import org.apache.iotdb.db.mpp.plan.execution.config.CountStorageGroupTask;
-import org.apache.iotdb.db.mpp.plan.execution.config.ShowStorageGroupTask;
-import org.apache.iotdb.db.mpp.plan.execution.config.ShowTTLTask;
+import org.apache.iotdb.db.mpp.plan.execution.config.metadata.CountStorageGroupTask;
+import org.apache.iotdb.db.mpp.plan.execution.config.metadata.ShowStorageGroupTask;
+import org.apache.iotdb.db.mpp.plan.execution.config.metadata.ShowTTLTask;
import org.apache.iotdb.db.mpp.plan.statement.metadata.CountStorageGroupStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.DeleteStorageGroupStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.SetStorageGroupStatement;
@@ -223,6 +224,19 @@ public class StandaloneConfigTaskExecutor implements IConfigTaskExecutor {
return future;
}
+ @Override
+ public SettableFuture<ConfigTaskResult> merge(TMergeReq mergeReq) {
+ SettableFuture<ConfigTaskResult> future = SettableFuture.create();
+ LocalConfigNode localConfigNode = LocalConfigNode.getInstance();
+ TSStatus tsStatus = localConfigNode.executeMergeOperation();
+ if (tsStatus.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
+ } else {
+ future.setException(new StatementExecutionException(tsStatus));
+ }
+ return future;
+ }
+
@Override
public SettableFuture<ConfigTaskResult> flush(TFlushReq tFlushReq) {
SettableFuture<ConfigTaskResult> future = SettableFuture.create();
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/CountStorageGroupTask.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/CountStorageGroupTask.java
similarity index 92%
rename from server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/CountStorageGroupTask.java
rename to server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/CountStorageGroupTask.java
index 0aa26a7336..35eeb3f527 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/CountStorageGroupTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/CountStorageGroupTask.java
@@ -17,11 +17,13 @@
* under the License.
*/
-package org.apache.iotdb.db.mpp.plan.execution.config;
+package org.apache.iotdb.db.mpp.plan.execution.config.metadata;
import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.db.mpp.common.header.ColumnHeader;
import org.apache.iotdb.db.mpp.common.header.DatasetHeader;
+import org.apache.iotdb.db.mpp.plan.execution.config.ConfigTaskResult;
+import org.apache.iotdb.db.mpp.plan.execution.config.IConfigTask;
import org.apache.iotdb.db.mpp.plan.execution.config.executor.IConfigTaskExecutor;
import org.apache.iotdb.db.mpp.plan.statement.metadata.CountStorageGroupStatement;
import org.apache.iotdb.rpc.TSStatusCode;
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/CreateFunctionTask.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/CreateFunctionTask.java
similarity index 89%
rename from server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/CreateFunctionTask.java
rename to server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/CreateFunctionTask.java
index a0c9cfc96f..104e3b92e3 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/CreateFunctionTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/CreateFunctionTask.java
@@ -17,8 +17,10 @@
* under the License.
*/
-package org.apache.iotdb.db.mpp.plan.execution.config;
+package org.apache.iotdb.db.mpp.plan.execution.config.metadata;
+import org.apache.iotdb.db.mpp.plan.execution.config.ConfigTaskResult;
+import org.apache.iotdb.db.mpp.plan.execution.config.IConfigTask;
import org.apache.iotdb.db.mpp.plan.execution.config.executor.IConfigTaskExecutor;
import org.apache.iotdb.db.mpp.plan.statement.metadata.CreateFunctionStatement;
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/DeleteStorageGroupTask.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/DeleteStorageGroupTask.java
similarity index 89%
rename from server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/DeleteStorageGroupTask.java
rename to server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/DeleteStorageGroupTask.java
index 3da9564ff9..ef24acd871 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/DeleteStorageGroupTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/DeleteStorageGroupTask.java
@@ -17,8 +17,10 @@
* under the License.
*/
-package org.apache.iotdb.db.mpp.plan.execution.config;
+package org.apache.iotdb.db.mpp.plan.execution.config.metadata;
+import org.apache.iotdb.db.mpp.plan.execution.config.ConfigTaskResult;
+import org.apache.iotdb.db.mpp.plan.execution.config.IConfigTask;
import org.apache.iotdb.db.mpp.plan.execution.config.executor.IConfigTaskExecutor;
import org.apache.iotdb.db.mpp.plan.statement.metadata.DeleteStorageGroupStatement;
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/DropFunctionTask.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/DropFunctionTask.java
similarity index 87%
rename from server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/DropFunctionTask.java
rename to server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/DropFunctionTask.java
index 458f73c550..c81900139a 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/DropFunctionTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/DropFunctionTask.java
@@ -17,8 +17,10 @@
* under the License.
*/
-package org.apache.iotdb.db.mpp.plan.execution.config;
+package org.apache.iotdb.db.mpp.plan.execution.config.metadata;
+import org.apache.iotdb.db.mpp.plan.execution.config.ConfigTaskResult;
+import org.apache.iotdb.db.mpp.plan.execution.config.IConfigTask;
import org.apache.iotdb.db.mpp.plan.execution.config.executor.IConfigTaskExecutor;
import org.apache.iotdb.db.mpp.plan.statement.metadata.DropFunctionStatement;
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/SetStorageGroupTask.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/SetStorageGroupTask.java
similarity index 93%
rename from server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/SetStorageGroupTask.java
rename to server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/SetStorageGroupTask.java
index 19b68e16c2..ad6efea84c 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/SetStorageGroupTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/SetStorageGroupTask.java
@@ -17,9 +17,11 @@
* under the License.
*/
-package org.apache.iotdb.db.mpp.plan.execution.config;
+package org.apache.iotdb.db.mpp.plan.execution.config.metadata;
import org.apache.iotdb.confignode.rpc.thrift.TStorageGroupSchema;
+import org.apache.iotdb.db.mpp.plan.execution.config.ConfigTaskResult;
+import org.apache.iotdb.db.mpp.plan.execution.config.IConfigTask;
import org.apache.iotdb.db.mpp.plan.execution.config.executor.IConfigTaskExecutor;
import org.apache.iotdb.db.mpp.plan.statement.metadata.SetStorageGroupStatement;
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/SetTTLTask.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/SetTTLTask.java
similarity index 88%
rename from server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/SetTTLTask.java
rename to server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/SetTTLTask.java
index 3f968b8736..10e63e55d3 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/SetTTLTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/SetTTLTask.java
@@ -17,8 +17,10 @@
* under the License.
*/
-package org.apache.iotdb.db.mpp.plan.execution.config;
+package org.apache.iotdb.db.mpp.plan.execution.config.metadata;
+import org.apache.iotdb.db.mpp.plan.execution.config.ConfigTaskResult;
+import org.apache.iotdb.db.mpp.plan.execution.config.IConfigTask;
import org.apache.iotdb.db.mpp.plan.execution.config.executor.IConfigTaskExecutor;
import org.apache.iotdb.db.mpp.plan.statement.metadata.SetTTLStatement;
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ShowClusterTask.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/ShowClusterTask.java
similarity index 94%
rename from server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ShowClusterTask.java
rename to server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/ShowClusterTask.java
index 176b3420c4..257e3d824b 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ShowClusterTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/ShowClusterTask.java
@@ -17,11 +17,13 @@
* under the License.
*/
-package org.apache.iotdb.db.mpp.plan.execution.config;
+package org.apache.iotdb.db.mpp.plan.execution.config.metadata;
import org.apache.iotdb.confignode.rpc.thrift.TShowClusterResp;
import org.apache.iotdb.db.mpp.common.header.DatasetHeader;
import org.apache.iotdb.db.mpp.common.header.HeaderConstant;
+import org.apache.iotdb.db.mpp.plan.execution.config.ConfigTaskResult;
+import org.apache.iotdb.db.mpp.plan.execution.config.IConfigTask;
import org.apache.iotdb.db.mpp.plan.execution.config.executor.IConfigTaskExecutor;
import org.apache.iotdb.db.mpp.plan.statement.metadata.ShowClusterStatement;
import org.apache.iotdb.rpc.TSStatusCode;
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ShowConfigNodesTask.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/ShowConfigNodesTask.java
similarity index 93%
rename from server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ShowConfigNodesTask.java
rename to server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/ShowConfigNodesTask.java
index 341cebe9a7..ad82988420 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ShowConfigNodesTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/ShowConfigNodesTask.java
@@ -17,12 +17,14 @@
* under the License.
*/
-package org.apache.iotdb.db.mpp.plan.execution.config;
+package org.apache.iotdb.db.mpp.plan.execution.config.metadata;
import org.apache.iotdb.confignode.rpc.thrift.TConfigNodeInfo;
import org.apache.iotdb.confignode.rpc.thrift.TShowConfigNodesResp;
import org.apache.iotdb.db.mpp.common.header.DatasetHeader;
import org.apache.iotdb.db.mpp.common.header.HeaderConstant;
+import org.apache.iotdb.db.mpp.plan.execution.config.ConfigTaskResult;
+import org.apache.iotdb.db.mpp.plan.execution.config.IConfigTask;
import org.apache.iotdb.db.mpp.plan.execution.config.executor.IConfigTaskExecutor;
import org.apache.iotdb.db.mpp.plan.statement.metadata.ShowConfigNodesStatement;
import org.apache.iotdb.rpc.TSStatusCode;
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ShowDataNodesTask.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/ShowDataNodesTask.java
similarity index 94%
rename from server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ShowDataNodesTask.java
rename to server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/ShowDataNodesTask.java
index 514d9cfe69..76fc64de91 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ShowDataNodesTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/ShowDataNodesTask.java
@@ -17,12 +17,14 @@
* under the License.
*/
-package org.apache.iotdb.db.mpp.plan.execution.config;
+package org.apache.iotdb.db.mpp.plan.execution.config.metadata;
import org.apache.iotdb.confignode.rpc.thrift.TDataNodeInfo;
import org.apache.iotdb.confignode.rpc.thrift.TShowDataNodesResp;
import org.apache.iotdb.db.mpp.common.header.DatasetHeader;
import org.apache.iotdb.db.mpp.common.header.HeaderConstant;
+import org.apache.iotdb.db.mpp.plan.execution.config.ConfigTaskResult;
+import org.apache.iotdb.db.mpp.plan.execution.config.IConfigTask;
import org.apache.iotdb.db.mpp.plan.execution.config.executor.IConfigTaskExecutor;
import org.apache.iotdb.db.mpp.plan.statement.metadata.ShowDataNodesStatement;
import org.apache.iotdb.rpc.TSStatusCode;
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ShowFunctionsTask.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/ShowFunctionsTask.java
similarity index 96%
rename from server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ShowFunctionsTask.java
rename to server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/ShowFunctionsTask.java
index 886a58fee4..714739d010 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ShowFunctionsTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/ShowFunctionsTask.java
@@ -17,13 +17,15 @@
* under the License.
*/
-package org.apache.iotdb.db.mpp.plan.execution.config;
+package org.apache.iotdb.db.mpp.plan.execution.config.metadata;
import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.commons.udf.builtin.BuiltinAggregationFunction;
import org.apache.iotdb.commons.udf.service.UDFRegistrationInformation;
import org.apache.iotdb.commons.udf.service.UDFRegistrationService;
import org.apache.iotdb.db.mpp.common.header.HeaderConstant;
+import org.apache.iotdb.db.mpp.plan.execution.config.ConfigTaskResult;
+import org.apache.iotdb.db.mpp.plan.execution.config.IConfigTask;
import org.apache.iotdb.db.mpp.plan.execution.config.executor.IConfigTaskExecutor;
import org.apache.iotdb.db.query.dataset.ListDataSet;
import org.apache.iotdb.rpc.TSStatusCode;
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ShowRegionTask.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/ShowRegionTask.java
similarity index 95%
rename from server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ShowRegionTask.java
rename to server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/ShowRegionTask.java
index 2c84953129..64f25eaf7c 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ShowRegionTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/ShowRegionTask.java
@@ -17,13 +17,15 @@
* under the License.
*/
-package org.apache.iotdb.db.mpp.plan.execution.config;
+package org.apache.iotdb.db.mpp.plan.execution.config.metadata;
import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
import org.apache.iotdb.confignode.rpc.thrift.TRegionInfo;
import org.apache.iotdb.confignode.rpc.thrift.TShowRegionResp;
import org.apache.iotdb.db.mpp.common.header.DatasetHeader;
import org.apache.iotdb.db.mpp.common.header.HeaderConstant;
+import org.apache.iotdb.db.mpp.plan.execution.config.ConfigTaskResult;
+import org.apache.iotdb.db.mpp.plan.execution.config.IConfigTask;
import org.apache.iotdb.db.mpp.plan.execution.config.executor.IConfigTaskExecutor;
import org.apache.iotdb.db.mpp.plan.statement.metadata.ShowRegionStatement;
import org.apache.iotdb.rpc.TSStatusCode;
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ShowStorageGroupTask.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/ShowStorageGroupTask.java
similarity index 94%
rename from server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ShowStorageGroupTask.java
rename to server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/ShowStorageGroupTask.java
index dc31a5628a..be486f3a9d 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ShowStorageGroupTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/ShowStorageGroupTask.java
@@ -17,11 +17,13 @@
* under the License.
*/
-package org.apache.iotdb.db.mpp.plan.execution.config;
+package org.apache.iotdb.db.mpp.plan.execution.config.metadata;
import org.apache.iotdb.confignode.rpc.thrift.TStorageGroupSchema;
import org.apache.iotdb.db.mpp.common.header.DatasetHeader;
import org.apache.iotdb.db.mpp.common.header.HeaderConstant;
+import org.apache.iotdb.db.mpp.plan.execution.config.ConfigTaskResult;
+import org.apache.iotdb.db.mpp.plan.execution.config.IConfigTask;
import org.apache.iotdb.db.mpp.plan.execution.config.executor.IConfigTaskExecutor;
import org.apache.iotdb.db.mpp.plan.statement.metadata.ShowStorageGroupStatement;
import org.apache.iotdb.rpc.TSStatusCode;
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ShowTTLTask.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/ShowTTLTask.java
similarity index 92%
rename from server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ShowTTLTask.java
rename to server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/ShowTTLTask.java
index f9aa46c423..3e83fb1e07 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ShowTTLTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/ShowTTLTask.java
@@ -17,10 +17,12 @@
* under the License.
*/
-package org.apache.iotdb.db.mpp.plan.execution.config;
+package org.apache.iotdb.db.mpp.plan.execution.config.metadata;
import org.apache.iotdb.db.mpp.common.header.DatasetHeader;
import org.apache.iotdb.db.mpp.common.header.HeaderConstant;
+import org.apache.iotdb.db.mpp.plan.execution.config.ConfigTaskResult;
+import org.apache.iotdb.db.mpp.plan.execution.config.IConfigTask;
import org.apache.iotdb.db.mpp.plan.execution.config.executor.IConfigTaskExecutor;
import org.apache.iotdb.db.mpp.plan.statement.metadata.ShowTTLStatement;
import org.apache.iotdb.rpc.TSStatusCode;
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/UnSetTTLTask.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/UnSetTTLTask.java
similarity index 94%
rename from server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/UnSetTTLTask.java
rename to server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/UnSetTTLTask.java
index 2046ef8625..e77d9f8219 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/UnSetTTLTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/UnSetTTLTask.java
@@ -17,7 +17,7 @@
* under the License.
*/
-package org.apache.iotdb.db.mpp.plan.execution.config;
+package org.apache.iotdb.db.mpp.plan.execution.config.metadata;
import org.apache.iotdb.db.mpp.plan.statement.metadata.UnSetTTLStatement;
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/CreateSchemaTemplateTask.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/template/CreateSchemaTemplateTask.java
similarity index 88%
rename from server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/CreateSchemaTemplateTask.java
rename to server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/template/CreateSchemaTemplateTask.java
index 231f7de5e9..a85b1e6625 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/CreateSchemaTemplateTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/template/CreateSchemaTemplateTask.java
@@ -17,8 +17,10 @@
* under the License.
*/
-package org.apache.iotdb.db.mpp.plan.execution.config;
+package org.apache.iotdb.db.mpp.plan.execution.config.metadata.template;
+import org.apache.iotdb.db.mpp.plan.execution.config.ConfigTaskResult;
+import org.apache.iotdb.db.mpp.plan.execution.config.IConfigTask;
import org.apache.iotdb.db.mpp.plan.execution.config.executor.IConfigTaskExecutor;
import org.apache.iotdb.db.mpp.plan.statement.metadata.template.CreateSchemaTemplateStatement;
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/SetSchemaTemplateTask.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/template/SetSchemaTemplateTask.java
similarity index 88%
rename from server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/SetSchemaTemplateTask.java
rename to server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/template/SetSchemaTemplateTask.java
index 101f2264d4..e695d61f86 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/SetSchemaTemplateTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/template/SetSchemaTemplateTask.java
@@ -17,8 +17,10 @@
* under the License.
*/
-package org.apache.iotdb.db.mpp.plan.execution.config;
+package org.apache.iotdb.db.mpp.plan.execution.config.metadata.template;
+import org.apache.iotdb.db.mpp.plan.execution.config.ConfigTaskResult;
+import org.apache.iotdb.db.mpp.plan.execution.config.IConfigTask;
import org.apache.iotdb.db.mpp.plan.execution.config.executor.IConfigTaskExecutor;
import org.apache.iotdb.db.mpp.plan.statement.metadata.template.SetSchemaTemplateStatement;
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ShowNodesInSchemaTemplateTask.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/template/ShowNodesInSchemaTemplateTask.java
similarity index 94%
rename from server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ShowNodesInSchemaTemplateTask.java
rename to server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/template/ShowNodesInSchemaTemplateTask.java
index 7faaec9372..348571163f 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ShowNodesInSchemaTemplateTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/template/ShowNodesInSchemaTemplateTask.java
@@ -17,11 +17,13 @@
* under the License.
*/
-package org.apache.iotdb.db.mpp.plan.execution.config;
+package org.apache.iotdb.db.mpp.plan.execution.config.metadata.template;
import org.apache.iotdb.db.metadata.template.Template;
import org.apache.iotdb.db.mpp.common.header.DatasetHeader;
import org.apache.iotdb.db.mpp.common.header.HeaderConstant;
+import org.apache.iotdb.db.mpp.plan.execution.config.ConfigTaskResult;
+import org.apache.iotdb.db.mpp.plan.execution.config.IConfigTask;
import org.apache.iotdb.db.mpp.plan.execution.config.executor.IConfigTaskExecutor;
import org.apache.iotdb.db.mpp.plan.statement.metadata.template.ShowNodesInSchemaTemplateStatement;
import org.apache.iotdb.rpc.TSStatusCode;
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ShowPathSetTemplateTask.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/template/ShowPathSetTemplateTask.java
similarity index 92%
rename from server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ShowPathSetTemplateTask.java
rename to server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/template/ShowPathSetTemplateTask.java
index f1ba3d66e1..b21091b13f 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ShowPathSetTemplateTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/template/ShowPathSetTemplateTask.java
@@ -17,11 +17,13 @@
* under the License.
*/
-package org.apache.iotdb.db.mpp.plan.execution.config;
+package org.apache.iotdb.db.mpp.plan.execution.config.metadata.template;
import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.db.mpp.common.header.DatasetHeader;
import org.apache.iotdb.db.mpp.common.header.HeaderConstant;
+import org.apache.iotdb.db.mpp.plan.execution.config.ConfigTaskResult;
+import org.apache.iotdb.db.mpp.plan.execution.config.IConfigTask;
import org.apache.iotdb.db.mpp.plan.execution.config.executor.IConfigTaskExecutor;
import org.apache.iotdb.db.mpp.plan.statement.metadata.template.ShowPathSetTemplateStatement;
import org.apache.iotdb.rpc.TSStatusCode;
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ShowSchemaTemplateTask.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/template/ShowSchemaTemplateTask.java
similarity index 93%
rename from server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ShowSchemaTemplateTask.java
rename to server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/template/ShowSchemaTemplateTask.java
index 9ceca337a9..5d5d8e9c50 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ShowSchemaTemplateTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/metadata/template/ShowSchemaTemplateTask.java
@@ -17,11 +17,13 @@
* under the License.
*/
-package org.apache.iotdb.db.mpp.plan.execution.config;
+package org.apache.iotdb.db.mpp.plan.execution.config.metadata.template;
import org.apache.iotdb.db.metadata.template.Template;
import org.apache.iotdb.db.mpp.common.header.DatasetHeader;
import org.apache.iotdb.db.mpp.common.header.HeaderConstant;
+import org.apache.iotdb.db.mpp.plan.execution.config.ConfigTaskResult;
+import org.apache.iotdb.db.mpp.plan.execution.config.IConfigTask;
import org.apache.iotdb.db.mpp.plan.execution.config.executor.IConfigTaskExecutor;
import org.apache.iotdb.db.mpp.plan.statement.metadata.template.ShowSchemaTemplateStatement;
import org.apache.iotdb.rpc.TSStatusCode;
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/AuthorizerTask.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/sys/AuthorizerTask.java
similarity index 90%
rename from server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/AuthorizerTask.java
rename to server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/sys/AuthorizerTask.java
index f289431746..546fd59eac 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/AuthorizerTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/sys/AuthorizerTask.java
@@ -17,10 +17,12 @@
* under the License.
*/
-package org.apache.iotdb.db.mpp.plan.execution.config;
+package org.apache.iotdb.db.mpp.plan.execution.config.sys;
import org.apache.iotdb.db.auth.AuthorizerManager;
import org.apache.iotdb.db.mpp.plan.analyze.QueryType;
+import org.apache.iotdb.db.mpp.plan.execution.config.ConfigTaskResult;
+import org.apache.iotdb.db.mpp.plan.execution.config.IConfigTask;
import org.apache.iotdb.db.mpp.plan.execution.config.executor.IConfigTaskExecutor;
import org.apache.iotdb.db.mpp.plan.statement.sys.AuthorStatement;
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ClearCacheTask.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/sys/ClearCacheTask.java
similarity index 88%
copy from server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ClearCacheTask.java
copy to server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/sys/ClearCacheTask.java
index 69e80d462a..8d735a2f71 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ClearCacheTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/sys/ClearCacheTask.java
@@ -17,11 +17,13 @@
* under the License.
*/
-package org.apache.iotdb.db.mpp.plan.execution.config;
+package org.apache.iotdb.db.mpp.plan.execution.config.sys;
-import org.apache.iotdb.common.rpc.thrift.TClearCacheReq;
+import org.apache.iotdb.confignode.rpc.thrift.TClearCacheReq;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.mpp.plan.execution.config.ConfigTaskResult;
+import org.apache.iotdb.db.mpp.plan.execution.config.IConfigTask;
import org.apache.iotdb.db.mpp.plan.execution.config.executor.IConfigTaskExecutor;
import org.apache.iotdb.db.mpp.plan.statement.sys.ClearCacheStatement;
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/FlushTask.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/sys/FlushTask.java
similarity index 92%
rename from server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/FlushTask.java
rename to server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/sys/FlushTask.java
index 4fe27a20cd..4309d1bdbf 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/FlushTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/sys/FlushTask.java
@@ -17,12 +17,14 @@
* under the License.
*/
-package org.apache.iotdb.db.mpp.plan.execution.config;
+package org.apache.iotdb.db.mpp.plan.execution.config.sys;
import org.apache.iotdb.common.rpc.thrift.TFlushReq;
import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.mpp.plan.execution.config.ConfigTaskResult;
+import org.apache.iotdb.db.mpp.plan.execution.config.IConfigTask;
import org.apache.iotdb.db.mpp.plan.execution.config.executor.IConfigTaskExecutor;
import org.apache.iotdb.db.mpp.plan.statement.sys.FlushStatement;
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ClearCacheTask.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/sys/MergeTask.java
similarity index 61%
rename from server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ClearCacheTask.java
rename to server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/sys/MergeTask.java
index 69e80d462a..05e2695e95 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ClearCacheTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/sys/MergeTask.java
@@ -17,37 +17,36 @@
* under the License.
*/
-package org.apache.iotdb.db.mpp.plan.execution.config;
+package org.apache.iotdb.db.mpp.plan.execution.config.sys;
-import org.apache.iotdb.common.rpc.thrift.TClearCacheReq;
+import org.apache.iotdb.confignode.rpc.thrift.TMergeReq;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.mpp.plan.execution.config.ConfigTaskResult;
+import org.apache.iotdb.db.mpp.plan.execution.config.IConfigTask;
import org.apache.iotdb.db.mpp.plan.execution.config.executor.IConfigTaskExecutor;
-import org.apache.iotdb.db.mpp.plan.statement.sys.ClearCacheStatement;
+import org.apache.iotdb.db.mpp.plan.statement.sys.MergeStatement;
import com.google.common.util.concurrent.ListenableFuture;
-public class ClearCacheTask implements IConfigTask {
+public class MergeTask implements IConfigTask {
- private ClearCacheStatement clearCacheStatement;
+ private MergeStatement mergeStatement;
- public ClearCacheTask(ClearCacheStatement clearCacheStatement) {
- this.clearCacheStatement = clearCacheStatement;
+ public MergeTask(MergeStatement mergeStatement) {
+ this.mergeStatement = mergeStatement;
}
@Override
public ListenableFuture<ConfigTaskResult> execute(IConfigTaskExecutor configTaskExecutor)
throws InterruptedException {
- TClearCacheReq tClearCacheReq = new TClearCacheReq();
-
+ TMergeReq mergeReq = new TMergeReq();
IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
- if (clearCacheStatement.isCluster()) {
- tClearCacheReq.setDataNodeId(-1);
+ if (mergeStatement.isCluster()) {
+ mergeReq.setDataNodeId(-1);
} else {
- tClearCacheReq.setDataNodeId(config.getDataNodeId());
+ mergeReq.setDataNodeId(config.getDataNodeId());
}
- // If the action is executed successfully, return the Future.
- // If your operation is async, you can return the corresponding future directly.
- return configTaskExecutor.clearCache(tClearCacheReq);
+ return configTaskExecutor.merge(mergeReq);
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/parser/ASTVisitor.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/parser/ASTVisitor.java
index 2f01be859a..e11ada738b 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/parser/ASTVisitor.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/parser/ASTVisitor.java
@@ -118,6 +118,7 @@ import org.apache.iotdb.db.mpp.plan.statement.sys.AuthorStatement;
import org.apache.iotdb.db.mpp.plan.statement.sys.ClearCacheStatement;
import org.apache.iotdb.db.mpp.plan.statement.sys.ExplainStatement;
import org.apache.iotdb.db.mpp.plan.statement.sys.FlushStatement;
+import org.apache.iotdb.db.mpp.plan.statement.sys.MergeStatement;
import org.apache.iotdb.db.mpp.plan.statement.sys.ShowVersionStatement;
import org.apache.iotdb.db.qp.constant.SQLConstant;
import org.apache.iotdb.db.qp.logical.sys.AuthorOperator;
@@ -1611,9 +1612,7 @@ public class ASTVisitor extends IoTDBSqlParserBaseVisitor<Statement> {
throw new SQLParserException("Invalid prefix path");
}
nodeNameList =
- ctx.prefixPath().stream()
- .map(prefixPath -> parsePrefixPath(prefixPath))
- .collect(Collectors.toList());
+ ctx.prefixPath().stream().map(this::parsePrefixPath).collect(Collectors.toList());
}
authorStatement.setNodeNameList(nodeNameList);
return authorStatement;
@@ -1627,9 +1626,7 @@ public class ASTVisitor extends IoTDBSqlParserBaseVisitor<Statement> {
authorStatement.setRoleName(parseIdentifier(ctx.roleName.getText()));
authorStatement.setPrivilegeList(parsePrivilege(ctx.privileges()));
List<PartialPath> nodeNameList =
- ctx.prefixPath().stream()
- .map(prefixPath -> parsePrefixPath(prefixPath))
- .collect(Collectors.toList());
+ ctx.prefixPath().stream().map(this::parsePrefixPath).collect(Collectors.toList());
authorStatement.setNodeNameList(nodeNameList);
return authorStatement;
}
@@ -1663,9 +1660,7 @@ public class ASTVisitor extends IoTDBSqlParserBaseVisitor<Statement> {
throw new SQLParserException("Invalid prefix path");
}
nodeNameList =
- ctx.prefixPath().stream()
- .map(prefixPath -> parsePrefixPath(prefixPath))
- .collect(Collectors.toList());
+ ctx.prefixPath().stream().map(this::parsePrefixPath).collect(Collectors.toList());
}
authorStatement.setNodeNameList(nodeNameList);
return authorStatement;
@@ -1679,9 +1674,7 @@ public class ASTVisitor extends IoTDBSqlParserBaseVisitor<Statement> {
authorStatement.setRoleName(parseIdentifier(ctx.roleName.getText()));
authorStatement.setPrivilegeList(parsePrivilege(ctx.privileges()));
List<PartialPath> nodeNameList =
- ctx.prefixPath().stream()
- .map(prefixPath -> parsePrefixPath(prefixPath))
- .collect(Collectors.toList());
+ ctx.prefixPath().stream().map(this::parsePrefixPath).collect(Collectors.toList());
authorStatement.setNodeNameList(nodeNameList);
return authorStatement;
}
@@ -1737,9 +1730,7 @@ public class ASTVisitor extends IoTDBSqlParserBaseVisitor<Statement> {
new AuthorStatement(AuthorOperator.AuthorType.LIST_USER_PRIVILEGE);
authorStatement.setUserName(parseIdentifier(ctx.userName.getText()));
List<PartialPath> nodeNameList =
- ctx.prefixPath().stream()
- .map(prefixPath -> parsePrefixPath(prefixPath))
- .collect(Collectors.toList());
+ ctx.prefixPath().stream().map(this::parsePrefixPath).collect(Collectors.toList());
authorStatement.setNodeNameList(nodeNameList);
return authorStatement;
}
@@ -1752,9 +1743,7 @@ public class ASTVisitor extends IoTDBSqlParserBaseVisitor<Statement> {
new AuthorStatement(AuthorOperator.AuthorType.LIST_ROLE_PRIVILEGE);
authorStatement.setRoleName(parseIdentifier(ctx.roleName.getText()));
List<PartialPath> nodeNameList =
- ctx.prefixPath().stream()
- .map(prefixPath -> parsePrefixPath(prefixPath))
- .collect(Collectors.toList());
+ ctx.prefixPath().stream().map(this::parsePrefixPath).collect(Collectors.toList());
authorStatement.setNodeNameList(nodeNameList);
return authorStatement;
}
@@ -2325,6 +2314,35 @@ public class ASTVisitor extends IoTDBSqlParserBaseVisitor<Statement> {
}
}
+ // Merge
+ @Override
+ public Statement visitMerge(IoTDBSqlParser.MergeContext ctx) {
+ MergeStatement mergeStatement = new MergeStatement(StatementType.MERGE);
+ if (ctx.CLUSTER() != null && !IoTDBDescriptor.getInstance().getConfig().isClusterMode()) {
+ throw new SemanticException("MERGE ON CLUSTER is not supported in standalone mode");
+ }
+ if (ctx.LOCAL() != null) {
+ mergeStatement.setCluster(false);
+ } else {
+ mergeStatement.setCluster(true);
+ }
+ return mergeStatement;
+ }
+
+ @Override
+ public Statement visitFullMerge(IoTDBSqlParser.FullMergeContext ctx) {
+ MergeStatement mergeStatement = new MergeStatement(StatementType.FULL_MERGE);
+ if (ctx.CLUSTER() != null && !IoTDBDescriptor.getInstance().getConfig().isClusterMode()) {
+ throw new SemanticException("FULL MERGE ON CLUSTER is not supported in standalone mode");
+ }
+ if (ctx.LOCAL() != null) {
+ mergeStatement.setCluster(false);
+ } else {
+ mergeStatement.setCluster(true);
+ }
+ return mergeStatement;
+ }
+
// Flush
@Override
@@ -2357,7 +2375,9 @@ public class ASTVisitor extends IoTDBSqlParserBaseVisitor<Statement> {
@Override
public Statement visitClearCache(IoTDBSqlParser.ClearCacheContext ctx) {
ClearCacheStatement clearCacheStatement = new ClearCacheStatement(StatementType.CLEAR_CACHE);
-
+ if (ctx.CLUSTER() != null && !IoTDBDescriptor.getInstance().getConfig().isClusterMode()) {
+ throw new SemanticException("CLEAR CACHE ON CLUSTER is not supported in standalone mode");
+ }
if (ctx.LOCAL() != null) {
clearCacheStatement.setCluster(false);
} else {
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/StatementVisitor.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/StatementVisitor.java
index 918881984e..ca040dc46e 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/StatementVisitor.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/StatementVisitor.java
@@ -67,6 +67,7 @@ import org.apache.iotdb.db.mpp.plan.statement.sys.AuthorStatement;
import org.apache.iotdb.db.mpp.plan.statement.sys.ClearCacheStatement;
import org.apache.iotdb.db.mpp.plan.statement.sys.ExplainStatement;
import org.apache.iotdb.db.mpp.plan.statement.sys.FlushStatement;
+import org.apache.iotdb.db.mpp.plan.statement.sys.MergeStatement;
import org.apache.iotdb.db.mpp.plan.statement.sys.ShowVersionStatement;
/**
@@ -254,6 +255,10 @@ public abstract class StatementVisitor<R, C> {
return visitStatement(deleteDataStatement, context);
}
+ public R visitMerge(MergeStatement mergeStatement, C context) {
+ return visitStatement(mergeStatement, context);
+ }
+
public R visitFlush(FlushStatement flushStatement, C context) {
return visitStatement(flushStatement, context);
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/sys/MergeStatement.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/sys/MergeStatement.java
new file mode 100644
index 0000000000..7c5a4c5bc3
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/sys/MergeStatement.java
@@ -0,0 +1,62 @@
+/*
+ * 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.db.mpp.plan.statement.sys;
+
+import org.apache.iotdb.commons.path.PartialPath;
+import org.apache.iotdb.db.mpp.plan.analyze.QueryType;
+import org.apache.iotdb.db.mpp.plan.constant.StatementType;
+import org.apache.iotdb.db.mpp.plan.statement.IConfigStatement;
+import org.apache.iotdb.db.mpp.plan.statement.Statement;
+import org.apache.iotdb.db.mpp.plan.statement.StatementVisitor;
+
+import java.util.Collections;
+import java.util.List;
+
+public class MergeStatement extends Statement implements IConfigStatement {
+
+ private boolean isCluster;
+
+ public MergeStatement(StatementType mergeType) {
+ this.statementType = mergeType;
+ }
+
+ public boolean isCluster() {
+ return isCluster;
+ }
+
+ public void setCluster(boolean cluster) {
+ isCluster = cluster;
+ }
+
+ @Override
+ public QueryType getQueryType() {
+ return QueryType.WRITE;
+ }
+
+ @Override
+ public List<PartialPath> getPaths() {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public <R, C> R accept(StatementVisitor<R, C> visitor, C context) {
+ return visitor.visitMerge(this, context);
+ }
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/service/thrift/impl/DataNodeInternalRPCServiceImpl.java b/server/src/main/java/org/apache/iotdb/db/service/thrift/impl/DataNodeInternalRPCServiceImpl.java
index 31b8728813..b2bdbcc4de 100644
--- a/server/src/main/java/org/apache/iotdb/db/service/thrift/impl/DataNodeInternalRPCServiceImpl.java
+++ b/server/src/main/java/org/apache/iotdb/db/service/thrift/impl/DataNodeInternalRPCServiceImpl.java
@@ -19,7 +19,6 @@
package org.apache.iotdb.db.service.thrift.impl;
-import org.apache.iotdb.common.rpc.thrift.TClearCacheReq;
import org.apache.iotdb.common.rpc.thrift.TConfigNodeLocation;
import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
@@ -51,6 +50,7 @@ import org.apache.iotdb.db.engine.cache.BloomFilterCache;
import org.apache.iotdb.db.engine.cache.ChunkCache;
import org.apache.iotdb.db.engine.cache.TimeSeriesMetadataCache;
import org.apache.iotdb.db.exception.DataRegionException;
+import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.exception.sql.SemanticException;
import org.apache.iotdb.db.metadata.cache.DataNodeSchemaCache;
import org.apache.iotdb.db.metadata.schemaregion.SchemaEngine;
@@ -483,13 +483,23 @@ public class DataNodeInternalRPCServiceImpl implements IDataNodeRPCService.Iface
return RpcUtils.getStatus(TSStatusCode.INVALIDATE_PERMISSION_CACHE_ERROR);
}
+ @Override
+ public TSStatus merge() throws TException {
+ try {
+ storageEngine.mergeAll();
+ } catch (StorageEngineException e) {
+ return RpcUtils.getStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR, e.getMessage());
+ }
+ return RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS);
+ }
+
@Override
public TSStatus flush(TFlushReq req) throws TException {
- return StorageEngineV2.getInstance().operateFlush(req);
+ return storageEngine.operateFlush(req);
}
@Override
- public TSStatus clearCache(TClearCacheReq req) throws TException {
+ public TSStatus clearCache() throws TException {
ChunkCache.getInstance().clear();
TimeSeriesMetadataCache.getInstance().clear();
BloomFilterCache.getInstance().clear();
@@ -498,7 +508,7 @@ public class DataNodeInternalRPCServiceImpl implements IDataNodeRPCService.Iface
@Override
public TSStatus setTTL(TSetTTLReq req) throws TException {
- return StorageEngineV2.getInstance().setTTL(req);
+ return storageEngine.setTTL(req);
}
@Override
diff --git a/thrift-commons/src/main/thrift/common.thrift b/thrift-commons/src/main/thrift/common.thrift
index 5098d9f9ee..049d3c5269 100644
--- a/thrift-commons/src/main/thrift/common.thrift
+++ b/thrift-commons/src/main/thrift/common.thrift
@@ -102,10 +102,6 @@ struct TFlushReq {
3: optional i32 dataNodeId
}
-struct TClearCacheReq {
- 1: optional i32 dataNodeId
-}
-
// for node management
struct TSchemaNode {
1: required string nodeName
diff --git a/thrift-confignode/src/main/thrift/confignode.thrift b/thrift-confignode/src/main/thrift/confignode.thrift
index c8fb44fd2e..7c19b090cf 100644
--- a/thrift-confignode/src/main/thrift/confignode.thrift
+++ b/thrift-confignode/src/main/thrift/confignode.thrift
@@ -335,6 +335,15 @@ struct TGetPathsSetTemplatesResp {
2: optional list<string> pathList
}
+// Maintenance Tools
+struct TMergeReq {
+ 1: optional i32 dataNodeId
+}
+
+struct TClearCacheReq {
+ 1: optional i32 dataNodeId
+}
+
service IConfigNodeRPCService {
/* DataNode */
@@ -425,13 +434,13 @@ service IConfigNodeRPCService {
common.TSStatus dropFunction(TDropFunctionReq req)
- /* Flush */
+ /* Maintenance Tools */
- common.TSStatus flush(common.TFlushReq req)
+ common.TSStatus merge(TMergeReq req)
- /* ClearCache */
+ common.TSStatus flush(common.TFlushReq req)
- common.TSStatus clearCache(common.TClearCacheReq req)
+ common.TSStatus clearCache(TClearCacheReq req)
/* Cluster Tools */
diff --git a/thrift/src/main/thrift/datanode.thrift b/thrift/src/main/thrift/datanode.thrift
index 8e62898e25..5f436c12e3 100644
--- a/thrift/src/main/thrift/datanode.thrift
+++ b/thrift/src/main/thrift/datanode.thrift
@@ -330,12 +330,19 @@ service IDataNodeRPCService {
*/
common.TSStatus invalidatePermissionCache(TInvalidatePermissionCacheReq req)
- common.TSStatus flush(common.TFlushReq req)
+ /* Maintenance Tools */
- common.TSStatus setTTL(common.TSetTTLReq req)
+ common.TSStatus merge()
+
+ common.TSStatus flush(common.TFlushReq req)
- common.TSStatus clearCache(common.TClearCacheReq req)
+ common.TSStatus clearCache()
+ /**
+ * Config node will Set the TTL for the storage group on a list of data nodes.
+ */
+ common.TSStatus setTTL(common.TSetTTLReq req)
+
/**
* configNode will notify all DataNodes when the capacity of the ConfigNodeGroup is expanded or reduced
*