This is an automated email from the ASF dual-hosted git repository.
chaow 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 babb62a355 [IOTDB-3416] Abstract interface of ConfigTask Execution for
standalone IoTDB (#6260)
babb62a355 is described below
commit babb62a3558e0cd65473458ff7ab62427f701f8e
Author: 任宇华 <[email protected]>
AuthorDate: Tue Jun 14 15:50:01 2022 +0800
[IOTDB-3416] Abstract interface of ConfigTask Execution for standalone
IoTDB (#6260)
Co-authored-by: renyuhua <[email protected]>
---
.../org/apache/iotdb/commons/utils/AuthUtils.java | 22 ++
.../apache/iotdb/db/auth/AuthorizerManager.java | 18 +-
.../iotdb/db/auth/ClusterAuthorityFetcher.java | 51 +++-
.../apache/iotdb/db/auth/IAuthorityFetcher.java | 9 +-
.../iotdb/db/auth/StandaloneAuthorityFetcher.java | 20 +-
.../iotdb/db/localconfignode/LocalConfigNode.java | 74 ++---
.../org/apache/iotdb/db/mpp/plan/Coordinator.java | 8 +-
.../mpp/plan/execution/config/AuthorizerTask.java | 67 +----
.../mpp/plan/execution/config/ConfigExecution.java | 27 +-
.../execution/config/CountStorageGroupTask.java | 51 +---
.../plan/execution/config/CreateFunctionTask.java | 74 +----
.../execution/config/DeleteStorageGroupTask.java | 69 +----
.../plan/execution/config/DropFunctionTask.java | 65 +---
.../db/mpp/plan/execution/config/FlushTask.java | 41 +--
.../db/mpp/plan/execution/config/IConfigTask.java | 7 +-
.../plan/execution/config/SetStorageGroupTask.java | 68 +----
.../db/mpp/plan/execution/config/SetTTLTask.java | 58 +---
.../mpp/plan/execution/config/ShowClusterTask.java | 65 ++--
.../plan/execution/config/ShowFunctionsTask.java | 7 +-
.../execution/config/ShowStorageGroupTask.java | 64 +---
.../db/mpp/plan/execution/config/ShowTTLTask.java | 85 +-----
.../config/executor/ClusterConfigTaskExecutor.java | 328 +++++++++++++++++++++
.../config/executor/IConfigTaskExecutor.java | 61 ++++
.../executor/StandsloneConfigTaskExecutor.java | 265 +++++++++++++++++
.../iotdb/db/qp/physical/sys/AuthorPlan.java | 26 +-
.../db/mpp/execution/ConfigExecutionTest.java | 7 +-
26 files changed, 864 insertions(+), 773 deletions(-)
diff --git
a/node-commons/src/main/java/org/apache/iotdb/commons/utils/AuthUtils.java
b/node-commons/src/main/java/org/apache/iotdb/commons/utils/AuthUtils.java
index 52e1e661f3..b69e61c1f6 100644
--- a/node-commons/src/main/java/org/apache/iotdb/commons/utils/AuthUtils.java
+++ b/node-commons/src/main/java/org/apache/iotdb/commons/utils/AuthUtils.java
@@ -360,4 +360,26 @@ public class AuthUtils {
permissionInfoResp.setRoleInfo(roleInfo);
return permissionInfoResp;
}
+
+ public static Set<Integer> strToPermissions(String[] authorizationList)
throws AuthException {
+ Set<Integer> result = new HashSet<>();
+ if (authorizationList == null) {
+ return result;
+ }
+ for (String s : authorizationList) {
+ PrivilegeType[] types = PrivilegeType.values();
+ boolean legal = false;
+ for (PrivilegeType privilegeType : types) {
+ if (s.equalsIgnoreCase(privilegeType.name())) {
+ result.add(privilegeType.ordinal());
+ legal = true;
+ break;
+ }
+ }
+ if (!legal) {
+ throw new AuthException("No such privilege " + s);
+ }
+ }
+ return result;
+ }
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/auth/AuthorizerManager.java
b/server/src/main/java/org/apache/iotdb/db/auth/AuthorizerManager.java
index 5d42da83a5..d8114f01ca 100644
--- a/server/src/main/java/org/apache/iotdb/db/auth/AuthorizerManager.java
+++ b/server/src/main/java/org/apache/iotdb/db/auth/AuthorizerManager.java
@@ -25,12 +25,11 @@ import
org.apache.iotdb.commons.auth.authorizer.BasicAuthorizer;
import org.apache.iotdb.commons.auth.authorizer.IAuthorizer;
import org.apache.iotdb.commons.auth.entity.Role;
import org.apache.iotdb.commons.auth.entity.User;
-import org.apache.iotdb.confignode.rpc.thrift.TAuthorizerReq;
-import org.apache.iotdb.db.client.ConfigNodeClient;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
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.statement.sys.AuthorStatement;
import org.apache.iotdb.rpc.ConfigNodeConnectionException;
import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
@@ -388,29 +387,27 @@ public class AuthorizerManager implements IAuthorizer {
return ClusterAuthorityFetcher.getInstance().invalidateCache(username,
roleName);
}
- public SettableFuture<ConfigTaskResult> queryPermission(
- TAuthorizerReq authorizerReq, ConfigNodeClient configNodeClient) throws
TException {
+ public SettableFuture<ConfigTaskResult> queryPermission(AuthorStatement
authorStatement) {
authReadWriteLock.readLock().lock();
try {
- return authorityFetcher.queryPermission(authorizerReq, configNodeClient);
+ return authorityFetcher.queryPermission(authorStatement);
} finally {
authReadWriteLock.readLock().unlock();
}
}
- public SettableFuture<ConfigTaskResult> operatePermission(
- TAuthorizerReq authorizerReq, ConfigNodeClient configNodeClient) {
+ public SettableFuture<ConfigTaskResult> operatePermission(AuthorStatement
authorStatement) {
authReadWriteLock.writeLock().lock();
try {
- return authorityFetcher.operatePermission(authorizerReq,
configNodeClient);
+ return authorityFetcher.operatePermission(authorStatement);
} finally {
authReadWriteLock.writeLock().unlock();
}
}
/** build TSBlock */
- public SettableFuture<ConfigTaskResult> buildTSBlock(Map<String,
List<String>> authorizerInfo) {
- SettableFuture<ConfigTaskResult> future = SettableFuture.create();
+ public void buildTSBlock(
+ Map<String, List<String>> authorizerInfo,
SettableFuture<ConfigTaskResult> future) {
List<TSDataType> types = new ArrayList<>();
for (int i = 0; i < authorizerInfo.size(); i++) {
types.add(TSDataType.TEXT);
@@ -436,6 +433,5 @@ public class AuthorizerManager implements IAuthorizer {
DatasetHeader datasetHeader = new DatasetHeader(headerList, true);
future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS,
builder.build(), datasetHeader));
- return future;
}
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/auth/ClusterAuthorityFetcher.java
b/server/src/main/java/org/apache/iotdb/db/auth/ClusterAuthorityFetcher.java
index 0da77fde91..4a9883e4bc 100644
--- a/server/src/main/java/org/apache/iotdb/db/auth/ClusterAuthorityFetcher.java
+++ b/server/src/main/java/org/apache/iotdb/db/auth/ClusterAuthorityFetcher.java
@@ -37,6 +37,7 @@ import org.apache.iotdb.db.client.ConfigNodeInfo;
import org.apache.iotdb.db.client.DataNodeClientPoolFactory;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.mpp.plan.execution.config.ConfigTaskResult;
+import org.apache.iotdb.db.mpp.plan.statement.sys.AuthorStatement;
import org.apache.iotdb.db.qp.logical.sys.AuthorOperator;
import org.apache.iotdb.rpc.RpcUtils;
import org.apache.iotdb.rpc.StatementExecutionException;
@@ -75,9 +76,10 @@ public class ClusterAuthorityFetcher implements
IAuthorityFetcher {
.expireAfterAccess(conf.getConfig().getAuthorCacheExpireTime(),
TimeUnit.MINUTES)
.build();
- private static final IClientManager<PartitionRegionId, ConfigNodeClient>
configNodeClientManager =
- new IClientManager.Factory<PartitionRegionId, ConfigNodeClient>()
- .createClientManager(new
DataNodeClientPoolFactory.ConfigNodeClientPoolFactory());
+ private static final IClientManager<PartitionRegionId, ConfigNodeClient>
+ CONFIG_NODE_CLIENT_MANAGER =
+ new IClientManager.Factory<PartitionRegionId, ConfigNodeClient>()
+ .createClientManager(new
DataNodeClientPoolFactory.ConfigNodeClientPoolFactory());
private static final class ClusterAuthorityFetcherHolder {
private static final ClusterAuthorityFetcher INSTANCE = new
ClusterAuthorityFetcher();
@@ -129,10 +131,12 @@ public class ClusterAuthorityFetcher implements
IAuthorityFetcher {
}
@Override
- public SettableFuture<ConfigTaskResult> operatePermission(
- TAuthorizerReq authorizerReq, ConfigNodeClient configNodeClient) {
+ public SettableFuture<ConfigTaskResult> operatePermission(AuthorStatement
authorStatement) {
SettableFuture<ConfigTaskResult> future = SettableFuture.create();
- try {
+ try (ConfigNodeClient configNodeClient =
+
CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.partitionRegionId)) {
+ // Construct request using statement
+ TAuthorizerReq authorizerReq = statementToAuthorizerReq(authorStatement);
// Send request to some API server
TSStatus tsStatus = configNodeClient.operatePermission(authorizerReq);
// Get response or throw exception
@@ -147,9 +151,11 @@ public class ClusterAuthorityFetcher implements
IAuthorityFetcher {
} else {
future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
}
- } catch (TException e) {
+ } catch (TException | IOException e) {
logger.error("Failed to connect to config node.");
future.setException(e);
+ } catch (AuthException e) {
+ future.setException(e);
}
// If the action is executed successfully, return the Future.
// If your operation is async, you can return the corresponding future
directly.
@@ -157,11 +163,14 @@ public class ClusterAuthorityFetcher implements
IAuthorityFetcher {
}
@Override
- public SettableFuture<ConfigTaskResult> queryPermission(
- TAuthorizerReq authorizerReq, ConfigNodeClient configNodeClient) {
+ public SettableFuture<ConfigTaskResult> queryPermission(AuthorStatement
authorStatement) {
SettableFuture<ConfigTaskResult> future = SettableFuture.create();
TAuthorizerResp authorizerResp = new TAuthorizerResp();
- try {
+
+ try (ConfigNodeClient configNodeClient =
+
CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.partitionRegionId)) {
+ // Construct request using statement
+ TAuthorizerReq authorizerReq = statementToAuthorizerReq(authorStatement);
// Send request to some API server
authorizerResp = configNodeClient.queryPermission(authorizerReq);
// Get response or throw exception
@@ -174,13 +183,15 @@ public class ClusterAuthorityFetcher implements
IAuthorityFetcher {
authorizerResp.getStatus());
future.setException(new
StatementExecutionException(authorizerResp.getStatus()));
} else {
- future =
AuthorizerManager.getInstance().buildTSBlock(authorizerResp.getAuthorizerInfo());
+
AuthorizerManager.getInstance().buildTSBlock(authorizerResp.getAuthorizerInfo(),
future);
}
- } catch (TException e) {
+ } catch (TException | IOException e) {
logger.error("Failed to connect to config node.");
authorizerResp.setStatus(
RpcUtils.getStatus(
TSStatusCode.EXECUTE_STATEMENT_ERROR, "Failed to connect to
config node."));
+ } catch (AuthException e) {
+ future.setException(e);
}
return future;
}
@@ -199,7 +210,7 @@ public class ClusterAuthorityFetcher implements
IAuthorityFetcher {
TLoginReq req = new TLoginReq(username, password);
TPermissionInfoResp status = null;
try (ConfigNodeClient configNodeClient =
-
configNodeClientManager.borrowClient(ConfigNodeInfo.partitionRegionId)) {
+
CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.partitionRegionId)) {
// Send request to some API server
status = configNodeClient.login(req);
} catch (TException | IOException e) {
@@ -226,7 +237,7 @@ public class ClusterAuthorityFetcher implements
IAuthorityFetcher {
TCheckUserPrivilegesReq req = new TCheckUserPrivilegesReq(username,
allPath, permission);
TPermissionInfoResp permissionInfoResp;
try (ConfigNodeClient configNodeClient =
-
configNodeClientManager.borrowClient(ConfigNodeInfo.partitionRegionId)) {
+
CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.partitionRegionId)) {
// Send request to some API server
permissionInfoResp = configNodeClient.checkUserPrivileges(req);
} catch (TException | IOException e) {
@@ -331,6 +342,18 @@ public class ClusterAuthorityFetcher implements
IAuthorityFetcher {
return pathPrivilege;
}
+ private TAuthorizerReq statementToAuthorizerReq(AuthorStatement
authorStatement)
+ throws AuthException {
+ return new TAuthorizerReq(
+ authorStatement.getAuthorType().ordinal(),
+ authorStatement.getUserName() == null ? "" :
authorStatement.getUserName(),
+ authorStatement.getRoleName() == null ? "" :
authorStatement.getRoleName(),
+ authorStatement.getPassWord() == null ? "" :
authorStatement.getPassWord(),
+ authorStatement.getNewPassword() == null ? "" :
authorStatement.getNewPassword(),
+ AuthUtils.strToPermissions(authorStatement.getPrivilegeList()),
+ authorStatement.getNodeName() == null ? "" :
authorStatement.getNodeName().getFullPath());
+ }
+
public Cache<String, User> getUserCache() {
return userCache;
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/auth/IAuthorityFetcher.java
b/server/src/main/java/org/apache/iotdb/db/auth/IAuthorityFetcher.java
index fc618d1e1a..bae7751fbd 100644
--- a/server/src/main/java/org/apache/iotdb/db/auth/IAuthorityFetcher.java
+++ b/server/src/main/java/org/apache/iotdb/db/auth/IAuthorityFetcher.java
@@ -20,9 +20,8 @@
package org.apache.iotdb.db.auth;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
-import org.apache.iotdb.confignode.rpc.thrift.TAuthorizerReq;
-import org.apache.iotdb.db.client.ConfigNodeClient;
import org.apache.iotdb.db.mpp.plan.execution.config.ConfigTaskResult;
+import org.apache.iotdb.db.mpp.plan.statement.sys.AuthorStatement;
import com.google.common.util.concurrent.SettableFuture;
@@ -34,9 +33,7 @@ public interface IAuthorityFetcher {
TSStatus checkUserPrivileges(String username, List<String> allPath, int
permission);
- SettableFuture<ConfigTaskResult> operatePermission(
- TAuthorizerReq authorizerReq, ConfigNodeClient configNodeClient);
+ SettableFuture<ConfigTaskResult> operatePermission(AuthorStatement
authorStatement);
- SettableFuture<ConfigTaskResult> queryPermission(
- TAuthorizerReq authorizerReq, ConfigNodeClient configNodeClient);
+ SettableFuture<ConfigTaskResult> queryPermission(AuthorStatement
authorStatement);
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/auth/StandaloneAuthorityFetcher.java
b/server/src/main/java/org/apache/iotdb/db/auth/StandaloneAuthorityFetcher.java
index d12d5c8329..663fd9c3c2 100644
---
a/server/src/main/java/org/apache/iotdb/db/auth/StandaloneAuthorityFetcher.java
+++
b/server/src/main/java/org/apache/iotdb/db/auth/StandaloneAuthorityFetcher.java
@@ -22,10 +22,9 @@ package org.apache.iotdb.db.auth;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.auth.AuthException;
import org.apache.iotdb.commons.utils.AuthUtils;
-import org.apache.iotdb.confignode.rpc.thrift.TAuthorizerReq;
-import org.apache.iotdb.db.client.ConfigNodeClient;
import org.apache.iotdb.db.localconfignode.LocalConfigNode;
import org.apache.iotdb.db.mpp.plan.execution.config.ConfigTaskResult;
+import org.apache.iotdb.db.mpp.plan.statement.sys.AuthorStatement;
import org.apache.iotdb.rpc.RpcUtils;
import org.apache.iotdb.rpc.TSStatusCode;
@@ -33,7 +32,6 @@ import com.google.common.util.concurrent.SettableFuture;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -102,12 +100,11 @@ public class StandaloneAuthorityFetcher implements
IAuthorityFetcher {
}
@Override
- public SettableFuture<ConfigTaskResult> operatePermission(
- TAuthorizerReq authorizerReq, ConfigNodeClient configNodeClient) {
+ public SettableFuture<ConfigTaskResult> operatePermission(AuthorStatement
authorStatement) {
SettableFuture<ConfigTaskResult> future = SettableFuture.create();
boolean status = true;
try {
- LocalConfigNode.getInstance().operatorPermission(authorizerReq);
+ LocalConfigNode.getInstance().operatorPermission(authorStatement);
} catch (AuthException e) {
future.setException(e);
status = false;
@@ -119,15 +116,16 @@ public class StandaloneAuthorityFetcher implements
IAuthorityFetcher {
}
@Override
- public SettableFuture<ConfigTaskResult> queryPermission(
- TAuthorizerReq authorizerReq, ConfigNodeClient configNodeClient) {
+ public SettableFuture<ConfigTaskResult> queryPermission(AuthorStatement
authorStatement) {
SettableFuture<ConfigTaskResult> future = SettableFuture.create();
- Map<String, List<String>> authorizerResp = new HashMap<>();
+ Map<String, List<String>> authorizerResp;
try {
- authorizerResp =
LocalConfigNode.getInstance().queryPermission(authorizerReq);
+ authorizerResp =
LocalConfigNode.getInstance().queryPermission(authorStatement);
+ // build TSBlock
+ AuthorizerManager.getInstance().buildTSBlock(authorizerResp, future);
} catch (AuthException e) {
future.setException(e);
}
- return AuthorizerManager.getInstance().buildTSBlock(authorizerResp);
+ return future;
}
}
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 3c1ba07911..daa08cae18 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
@@ -47,7 +47,6 @@ import
org.apache.iotdb.commons.partition.DataPartitionQueryParam;
import org.apache.iotdb.commons.partition.executor.SeriesPartitionExecutor;
import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.commons.utils.AuthUtils;
-import org.apache.iotdb.confignode.rpc.thrift.TAuthorizerReq;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.engine.StorageEngineV2;
@@ -70,6 +69,7 @@ import org.apache.iotdb.db.metadata.template.TemplateManager;
import org.apache.iotdb.db.metadata.utils.MetaUtils;
import org.apache.iotdb.db.mpp.common.schematree.PathPatternTree;
import org.apache.iotdb.db.mpp.plan.constant.DataNodeEndPoints;
+import org.apache.iotdb.db.mpp.plan.statement.sys.AuthorStatement;
import org.apache.iotdb.db.qp.logical.sys.AuthorOperator;
import org.apache.iotdb.db.qp.physical.sys.ActivateTemplatePlan;
import org.apache.iotdb.db.qp.physical.sys.AppendTemplatePlan;
@@ -1030,15 +1030,16 @@ public class LocalConfigNode {
// endregion
// author
- public void operatorPermission(TAuthorizerReq authorizerReq) throws
AuthException {
+ public void operatorPermission(AuthorStatement authorStatement) throws
AuthException {
AuthorOperator.AuthorType authorType =
- AuthorOperator.AuthorType.values()[authorizerReq.authorType];
- String userName = authorizerReq.getUserName();
- String roleName = authorizerReq.getRoleName();
- String password = authorizerReq.getPassword();
- String newPassword = authorizerReq.getNewPassword();
- Set<Integer> permissions = authorizerReq.getPermissions();
- String nodeName = authorizerReq.getNodeName();
+
AuthorOperator.AuthorType.values()[authorStatement.getAuthorType().ordinal()];
+ String userName = authorStatement.getUserName();
+ String roleName = authorStatement.getRoleName();
+ String password = authorStatement.getPassWord();
+ String newPassword = authorStatement.getNewPassword();
+ Set<Integer> permissions =
AuthUtils.strToPermissions(authorStatement.getPrivilegeList());
+ PartialPath partialPath = authorStatement.getNodeName();
+ String nodeName = partialPath == null ? null : partialPath.getFullPath();
switch (authorType) {
case UPDATE_USER:
iAuthorizer.updateUserPassword(userName, newPassword);
@@ -1086,23 +1087,23 @@ public class LocalConfigNode {
}
}
- public Map<String, List<String>> queryPermission(TAuthorizerReq
authorizerReq)
+ public Map<String, List<String>> queryPermission(AuthorStatement
authorStatement)
throws AuthException {
AuthorOperator.AuthorType authorType =
- AuthorOperator.AuthorType.values()[authorizerReq.authorType];
+
AuthorOperator.AuthorType.values()[authorStatement.getAuthorType().ordinal()];
switch (authorType) {
case LIST_USER:
return executeListUser();
case LIST_ROLE:
return executeListRole();
case LIST_USER_PRIVILEGE:
- return executeListUserPrivileges(authorizerReq);
+ return executeListUserPrivileges(authorStatement);
case LIST_ROLE_PRIVILEGE:
- return executeListRolePrivileges(authorizerReq);
+ return executeListRolePrivileges(authorStatement);
case LIST_USER_ROLES:
- return executeListUserRoles(authorizerReq);
+ return executeListUserRoles(authorStatement);
case LIST_ROLE_USERS:
- return executeListRoleUsers(authorizerReq);
+ return executeListRoleUsers(authorStatement);
default:
throw new AuthException("Unsupported operation " + authorType);
}
@@ -1122,14 +1123,14 @@ public class LocalConfigNode {
return permissionInfo;
}
- public Map<String, List<String>> executeListRoleUsers(TAuthorizerReq
authorizerReq)
+ public Map<String, List<String>> executeListRoleUsers(AuthorStatement
authorStatement)
throws AuthException {
Map<String, List<String>> permissionInfo = new HashMap<>();
Role role;
try {
- role = iAuthorizer.getRole(authorizerReq.getRoleName());
+ role = iAuthorizer.getRole(authorStatement.getRoleName());
if (role == null) {
- throw new AuthException("No such role : " +
authorizerReq.getRoleName());
+ throw new AuthException("No such role : " +
authorStatement.getRoleName());
}
} catch (AuthException e) {
throw new AuthException(e);
@@ -1138,7 +1139,7 @@ public class LocalConfigNode {
List<String> userList = iAuthorizer.listAllUsers();
for (String userN : userList) {
User userObj = iAuthorizer.getUser(userN);
- if (userObj != null && userObj.hasRole(authorizerReq.getRoleName())) {
+ if (userObj != null && userObj.hasRole(authorStatement.getRoleName())) {
roleUsersList.add(userN);
}
}
@@ -1146,14 +1147,14 @@ public class LocalConfigNode {
return permissionInfo;
}
- public Map<String, List<String>> executeListUserRoles(TAuthorizerReq
authorizerReq)
+ public Map<String, List<String>> executeListUserRoles(AuthorStatement
authorStatement)
throws AuthException {
Map<String, List<String>> permissionInfo = new HashMap<>();
User user;
try {
- user = iAuthorizer.getUser(authorizerReq.getUserName());
+ user = iAuthorizer.getUser(authorStatement.getUserName());
if (user == null) {
- throw new AuthException("No such user : " +
authorizerReq.getUserName());
+ throw new AuthException("No such user : " +
authorStatement.getUserName());
}
} catch (AuthException e) {
throw new AuthException(e);
@@ -1167,22 +1168,23 @@ public class LocalConfigNode {
return permissionInfo;
}
- public Map<String, List<String>> executeListRolePrivileges(TAuthorizerReq
authorizerReq)
+ public Map<String, List<String>> executeListRolePrivileges(AuthorStatement
authorStatement)
throws AuthException {
Map<String, List<String>> permissionInfo = new HashMap<>();
Role role;
try {
- role = iAuthorizer.getRole(authorizerReq.getRoleName());
+ role = iAuthorizer.getRole(authorStatement.getRoleName());
if (role == null) {
- throw new AuthException("No such role : " +
authorizerReq.getRoleName());
+ throw new AuthException("No such role : " +
authorStatement.getRoleName());
}
} catch (AuthException e) {
throw new AuthException(e);
}
List<String> rolePrivilegesList = new ArrayList<>();
for (PathPrivilege pathPrivilege : role.getPrivilegeList()) {
- if (authorizerReq.getNodeName().equals("")
- || AuthUtils.pathBelongsTo(authorizerReq.getNodeName(),
pathPrivilege.getPath())) {
+ if (authorStatement.getNodeName().getFullPath().equals("")
+ || AuthUtils.pathBelongsTo(
+ authorStatement.getNodeName().getFullPath(),
pathPrivilege.getPath())) {
rolePrivilegesList.add(pathPrivilege.toString());
}
}
@@ -1191,29 +1193,30 @@ public class LocalConfigNode {
return permissionInfo;
}
- public Map<String, List<String>> executeListUserPrivileges(TAuthorizerReq
authorizerReq)
+ public Map<String, List<String>> executeListUserPrivileges(AuthorStatement
authorStatement)
throws AuthException {
Map<String, List<String>> permissionInfo = new HashMap<>();
User user;
try {
- user = iAuthorizer.getUser(authorizerReq.getUserName());
+ user = iAuthorizer.getUser(authorStatement.getUserName());
if (user == null) {
- throw new AuthException("No such user : " +
authorizerReq.getUserName());
+ throw new AuthException("No such user : " +
authorStatement.getUserName());
}
} catch (AuthException e) {
throw new AuthException(e);
}
List<String> userPrivilegesList = new ArrayList<>();
- if (IoTDBConstant.PATH_ROOT.equals(authorizerReq.getUserName())) {
+ if (IoTDBConstant.PATH_ROOT.equals(authorStatement.getUserName())) {
for (PrivilegeType privilegeType : PrivilegeType.values()) {
userPrivilegesList.add(privilegeType.toString());
}
} else {
List<String> rolePrivileges = new ArrayList<>();
for (PathPrivilege pathPrivilege : user.getPrivilegeList()) {
- if (authorizerReq.getNodeName().equals("")
- || AuthUtils.pathBelongsTo(authorizerReq.getNodeName(),
pathPrivilege.getPath())) {
+ if (authorStatement.getNodeName().getFullPath().equals("")
+ || AuthUtils.pathBelongsTo(
+ authorStatement.getNodeName().getFullPath(),
pathPrivilege.getPath())) {
rolePrivileges.add("");
userPrivilegesList.add(pathPrivilege.toString());
}
@@ -1224,8 +1227,9 @@ public class LocalConfigNode {
continue;
}
for (PathPrivilege pathPrivilege : role.getPrivilegeList()) {
- if (authorizerReq.getNodeName().equals("")
- || AuthUtils.pathBelongsTo(authorizerReq.getNodeName(),
pathPrivilege.getPath())) {
+ if (authorStatement.getNodeName().getFullPath().equals("")
+ || AuthUtils.pathBelongsTo(
+ authorStatement.getNodeName().getFullPath(),
pathPrivilege.getPath())) {
rolePrivileges.add(roleN);
userPrivilegesList.add(pathPrivilege.toString());
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/Coordinator.java
b/server/src/main/java/org/apache/iotdb/db/mpp/plan/Coordinator.java
index 9b79b1a00e..872d8c3cfa 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/Coordinator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/Coordinator.java
@@ -22,8 +22,6 @@ import org.apache.iotdb.common.rpc.thrift.TEndPoint;
import org.apache.iotdb.commons.client.IClientManager;
import org.apache.iotdb.commons.client.sync.SyncDataNodeInternalServiceClient;
import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory;
-import org.apache.iotdb.commons.consensus.PartitionRegionId;
-import org.apache.iotdb.db.client.ConfigNodeClient;
import org.apache.iotdb.db.client.DataNodeClientPoolFactory;
import org.apache.iotdb.db.mpp.common.MPPQueryContext;
import org.apache.iotdb.db.mpp.common.QueryId;
@@ -68,10 +66,6 @@ public class Coordinator {
.createClientManager(
new
DataNodeClientPoolFactory.SyncDataNodeInternalServiceClientPoolFactory());
- private static final IClientManager<PartitionRegionId, ConfigNodeClient>
- CONFIG_NODE_CLIENT_MANAGER =
- new IClientManager.Factory<PartitionRegionId, ConfigNodeClient>()
- .createClientManager(new
DataNodeClientPoolFactory.ConfigNodeClientPoolFactory());
private final ExecutorService executor;
private final ExecutorService writeOperationExecutor;
private final ScheduledExecutorService scheduledExecutor;
@@ -96,7 +90,7 @@ public class Coordinator {
ISchemaFetcher schemaFetcher) {
if (statement instanceof IConfigStatement) {
queryContext.setQueryType(((IConfigStatement) statement).getQueryType());
- return new ConfigExecution(queryContext, statement, executor,
CONFIG_NODE_CLIENT_MANAGER);
+ return new ConfigExecution(queryContext, statement, executor);
}
return new QueryExecution(
statement,
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/AuthorizerTask.java
index 375ffe0600..f289431746 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/AuthorizerTask.java
@@ -19,33 +19,15 @@
package org.apache.iotdb.db.mpp.plan.execution.config;
-import org.apache.iotdb.commons.auth.AuthException;
-import org.apache.iotdb.commons.client.IClientManager;
-import org.apache.iotdb.commons.consensus.PartitionRegionId;
-import org.apache.iotdb.confignode.rpc.thrift.TAuthorizerReq;
import org.apache.iotdb.db.auth.AuthorizerManager;
-import org.apache.iotdb.db.client.ConfigNodeClient;
-import org.apache.iotdb.db.client.ConfigNodeInfo;
-import org.apache.iotdb.db.conf.IoTDBConfig;
-import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.mpp.plan.analyze.QueryType;
+import
org.apache.iotdb.db.mpp.plan.execution.config.executor.IConfigTaskExecutor;
import org.apache.iotdb.db.mpp.plan.statement.sys.AuthorStatement;
-import org.apache.iotdb.db.qp.physical.sys.AuthorPlan;
import com.google.common.util.concurrent.ListenableFuture;
-import com.google.common.util.concurrent.SettableFuture;
-import org.apache.thrift.TException;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.IOException;
public class AuthorizerTask implements IConfigTask {
- private static final Logger LOGGER =
LoggerFactory.getLogger(AuthorizerTask.class);
-
- private static IoTDBConfig config =
IoTDBDescriptor.getInstance().getConfig();
-
private AuthorStatement authorStatement;
private AuthorizerManager authorizerManager =
AuthorizerManager.getInstance();
@@ -54,48 +36,13 @@ public class AuthorizerTask implements IConfigTask {
}
@Override
- public ListenableFuture<ConfigTaskResult> execute(
- IClientManager<PartitionRegionId, ConfigNodeClient> clientManager) {
- SettableFuture<ConfigTaskResult> future = SettableFuture.create();
- try {
- // Construct request using statement
- TAuthorizerReq req =
- new TAuthorizerReq(
- authorStatement.getAuthorType().ordinal(),
- authorStatement.getUserName() == null ? "" :
authorStatement.getUserName(),
- authorStatement.getRoleName() == null ? "" :
authorStatement.getRoleName(),
- authorStatement.getPassWord() == null ? "" :
authorStatement.getPassWord(),
- authorStatement.getNewPassword() == null ? "" :
authorStatement.getNewPassword(),
- AuthorPlan.strToPermissions(authorStatement.getPrivilegeList()),
- authorStatement.getNodeName() == null
- ? ""
- : authorStatement.getNodeName().getFullPath());
- // Send request to some API server
- if (config.isClusterMode()) {
- try (ConfigNodeClient configNodeClient =
- clientManager.borrowClient(ConfigNodeInfo.partitionRegionId); ) {
- if (authorStatement.getQueryType() == QueryType.WRITE) {
- future = authorizerManager.operatePermission(req,
configNodeClient);
- } else {
- future = authorizerManager.queryPermission(req, configNodeClient);
- }
- }
- } else {
- if (authorStatement.getQueryType() == QueryType.WRITE) {
- future = authorizerManager.operatePermission(req, null);
- } else {
- future = authorizerManager.queryPermission(req, null);
- }
- }
-
- } catch (IOException | TException e) {
- LOGGER.error("can't connect to all config nodes", e);
- future.setException(e);
- } catch (AuthException e) {
- future.setException(e);
- }
+ public ListenableFuture<ConfigTaskResult> execute(IConfigTaskExecutor
configTaskExecutor) {
// If the action is executed successfully, return the Future.
// If your operation is async, you can return the corresponding future
directly.
- return future;
+ if (authorStatement.getQueryType() == QueryType.WRITE) {
+ return authorizerManager.operatePermission(authorStatement);
+ } else {
+ return authorizerManager.queryPermission(authorStatement);
+ }
}
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ConfigExecution.java
b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ConfigExecution.java
index b014a11926..15ec6efa3d 100644
---
a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ConfigExecution.java
+++
b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ConfigExecution.java
@@ -19,16 +19,18 @@
package org.apache.iotdb.db.mpp.plan.execution.config;
-import org.apache.iotdb.commons.client.IClientManager;
-import org.apache.iotdb.commons.consensus.PartitionRegionId;
import org.apache.iotdb.commons.utils.TestOnly;
-import org.apache.iotdb.db.client.ConfigNodeClient;
+import org.apache.iotdb.db.conf.IoTDBConfig;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.mpp.common.MPPQueryContext;
import org.apache.iotdb.db.mpp.common.header.DatasetHeader;
import org.apache.iotdb.db.mpp.execution.QueryStateMachine;
import org.apache.iotdb.db.mpp.plan.analyze.QueryType;
import org.apache.iotdb.db.mpp.plan.execution.ExecutionResult;
import org.apache.iotdb.db.mpp.plan.execution.IQueryExecution;
+import
org.apache.iotdb.db.mpp.plan.execution.config.executor.ClusterConfigTaskExecutor;
+import
org.apache.iotdb.db.mpp.plan.execution.config.executor.IConfigTaskExecutor;
+import
org.apache.iotdb.db.mpp.plan.execution.config.executor.StandsloneConfigTaskExecutor;
import org.apache.iotdb.db.mpp.plan.statement.Statement;
import org.apache.iotdb.rpc.RpcUtils;
import org.apache.iotdb.rpc.TSStatusCode;
@@ -46,6 +48,8 @@ import java.util.concurrent.ExecutorService;
public class ConfigExecution implements IQueryExecution {
+ private static IoTDBConfig config =
IoTDBDescriptor.getInstance().getConfig();
+
private final MPPQueryContext context;
private final Statement statement;
private final ExecutorService executor;
@@ -56,14 +60,9 @@ public class ConfigExecution implements IQueryExecution {
private DatasetHeader datasetHeader;
private boolean resultSetConsumed;
private final IConfigTask task;
+ private IConfigTaskExecutor configTaskExecutor;
- private IClientManager<PartitionRegionId, ConfigNodeClient> clientManager;
-
- public ConfigExecution(
- MPPQueryContext context,
- Statement statement,
- ExecutorService executor,
- IClientManager<PartitionRegionId, ConfigNodeClient> clientManager) {
+ public ConfigExecution(MPPQueryContext context, Statement statement,
ExecutorService executor) {
this.context = context;
this.statement = statement;
this.executor = executor;
@@ -71,7 +70,11 @@ public class ConfigExecution implements IQueryExecution {
this.taskFuture = SettableFuture.create();
this.task = statement.accept(new ConfigTaskVisitor(), new
ConfigTaskVisitor.TaskContext());
this.resultSetConsumed = false;
- this.clientManager = clientManager;
+ if (config.isClusterMode()) {
+ configTaskExecutor = ClusterConfigTaskExecutor.getInstance();
+ } else {
+ configTaskExecutor = StandsloneConfigTaskExecutor.getInstance();
+ }
}
@TestOnly
@@ -88,7 +91,7 @@ public class ConfigExecution implements IQueryExecution {
@Override
public void start() {
try {
- ListenableFuture<ConfigTaskResult> future = task.execute(clientManager);
+ ListenableFuture<ConfigTaskResult> future =
task.execute(configTaskExecutor);
Futures.addCallback(
future,
new FutureCallback<ConfigTaskResult>() {
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/CountStorageGroupTask.java
index 5d8fa105a7..0aa26a7336 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/CountStorageGroupTask.java
@@ -19,18 +19,10 @@
package org.apache.iotdb.db.mpp.plan.execution.config;
-import org.apache.iotdb.commons.client.IClientManager;
import org.apache.iotdb.commons.conf.IoTDBConstant;
-import org.apache.iotdb.commons.consensus.PartitionRegionId;
-import org.apache.iotdb.commons.exception.MetadataException;
-import org.apache.iotdb.confignode.rpc.thrift.TCountStorageGroupResp;
-import org.apache.iotdb.db.client.ConfigNodeClient;
-import org.apache.iotdb.db.client.ConfigNodeInfo;
-import org.apache.iotdb.db.conf.IoTDBConfig;
-import org.apache.iotdb.db.conf.IoTDBDescriptor;
-import org.apache.iotdb.db.localconfignode.LocalConfigNode;
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.executor.IConfigTaskExecutor;
import
org.apache.iotdb.db.mpp.plan.statement.metadata.CountStorageGroupStatement;
import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
@@ -38,19 +30,10 @@ import
org.apache.iotdb.tsfile.read.common.block.TsBlockBuilder;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.SettableFuture;
-import org.apache.thrift.TException;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import java.io.IOException;
-import java.util.Arrays;
import java.util.Collections;
-import java.util.List;
public class CountStorageGroupTask implements IConfigTask {
- private static final Logger LOGGER =
LoggerFactory.getLogger(CountStorageGroupTask.class);
-
- private static final IoTDBConfig config =
IoTDBDescriptor.getInstance().getConfig();
private CountStorageGroupStatement countStorageGroupStatement;
@@ -59,33 +42,12 @@ public class CountStorageGroupTask implements IConfigTask {
}
@Override
- public ListenableFuture<ConfigTaskResult> execute(
- IClientManager<PartitionRegionId, ConfigNodeClient> clientManager)
+ public ListenableFuture<ConfigTaskResult> execute(IConfigTaskExecutor
configTaskExecutor)
throws InterruptedException {
- SettableFuture<ConfigTaskResult> future = SettableFuture.create();
- int storageGroupNum = 0;
- if (config.isClusterMode()) {
- List<String> storageGroupPathPattern =
-
Arrays.asList(countStorageGroupStatement.getPartialPath().getNodes());
- try (ConfigNodeClient client =
clientManager.borrowClient(ConfigNodeInfo.partitionRegionId)) {
- TCountStorageGroupResp resp =
client.countMatchedStorageGroups(storageGroupPathPattern);
- storageGroupNum = resp.getCount();
- } catch (TException | IOException e) {
- LOGGER.error("Failed to connect to config node.");
- future.setException(e);
- }
- } else {
- try {
- storageGroupNum =
- LocalConfigNode.getInstance()
- .getStorageGroupNum(
- countStorageGroupStatement.getPartialPath(),
- countStorageGroupStatement.isPrefixPath());
- } catch (MetadataException e) {
- future.setException(e);
- }
- }
- // build TSBlock
+ return configTaskExecutor.countStorageGroup(countStorageGroupStatement);
+ }
+
+ public static void buildTSBlock(int storageGroupNum,
SettableFuture<ConfigTaskResult> future) {
TsBlockBuilder builder = new
TsBlockBuilder(Collections.singletonList(TSDataType.INT32));
builder.getTimeColumnBuilder().writeLong(0L);
builder.getColumnBuilder(0).writeInt(storageGroupNum);
@@ -95,6 +57,5 @@ public class CountStorageGroupTask implements IConfigTask {
DatasetHeader datasetHeader =
new DatasetHeader(Collections.singletonList(storageGroupColumnHeader),
true);
future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS,
builder.build(), datasetHeader));
- return future;
}
}
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/CreateFunctionTask.java
index 56b95836bf..a0c9cfc96f 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/CreateFunctionTask.java
@@ -19,36 +19,17 @@
package org.apache.iotdb.db.mpp.plan.execution.config;
-import org.apache.iotdb.common.rpc.thrift.TSStatus;
-import org.apache.iotdb.commons.client.IClientManager;
-import org.apache.iotdb.commons.consensus.PartitionRegionId;
-import org.apache.iotdb.commons.udf.service.UDFExecutableManager;
-import org.apache.iotdb.commons.udf.service.UDFRegistrationService;
-import org.apache.iotdb.confignode.rpc.thrift.TCreateFunctionReq;
-import org.apache.iotdb.db.client.ConfigNodeClient;
-import org.apache.iotdb.db.client.ConfigNodeInfo;
-import org.apache.iotdb.db.conf.IoTDBConfig;
-import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import
org.apache.iotdb.db.mpp.plan.execution.config.executor.IConfigTaskExecutor;
import org.apache.iotdb.db.mpp.plan.statement.metadata.CreateFunctionStatement;
-import org.apache.iotdb.rpc.StatementExecutionException;
-import org.apache.iotdb.rpc.TSStatusCode;
import com.google.common.util.concurrent.ListenableFuture;
-import com.google.common.util.concurrent.SettableFuture;
-import org.apache.thrift.TException;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import java.io.IOException;
import java.net.URI;
import java.util.List;
import java.util.stream.Collectors;
public class CreateFunctionTask implements IConfigTask {
- private static final Logger LOGGER =
LoggerFactory.getLogger(CreateFunctionTask.class);
- private static final IoTDBConfig CONFIG =
IoTDBDescriptor.getInstance().getConfig();
-
private final String udfName;
private final String className;
private final List<String> uris;
@@ -61,57 +42,8 @@ public class CreateFunctionTask implements IConfigTask {
}
@Override
- public ListenableFuture<ConfigTaskResult> execute(
- IClientManager<PartitionRegionId, ConfigNodeClient> clientManager)
+ public ListenableFuture<ConfigTaskResult> execute(IConfigTaskExecutor
configTaskExecutor)
throws InterruptedException {
- SettableFuture<ConfigTaskResult> future = SettableFuture.create();
- if (CONFIG.isClusterMode()) {
- executeCluster(clientManager, future);
- } else {
- executeStandalone(future);
- }
- return future;
- }
-
- private void executeCluster(
- IClientManager<PartitionRegionId, ConfigNodeClient> clientManager,
- SettableFuture<ConfigTaskResult> future) {
- try (ConfigNodeClient client =
clientManager.borrowClient(ConfigNodeInfo.partitionRegionId)) {
- final TSStatus executionStatus =
- client.createFunction(new TCreateFunctionReq(udfName, className,
uris));
-
- if (TSStatusCode.SUCCESS_STATUS.getStatusCode() !=
executionStatus.getCode()) {
- LOGGER.error(
- "[{}] Failed to create function {}({}) in config node, URI: {}.",
- executionStatus,
- udfName,
- className,
- uris);
- future.setException(new StatementExecutionException(executionStatus));
- } else {
- future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
- }
- } catch (TException | IOException e) {
- LOGGER.error("Failed to connect to config node.");
- future.setException(e);
- }
- }
-
- private void executeStandalone(SettableFuture<ConfigTaskResult> future) {
- try {
- UDFRegistrationService.getInstance()
- .register(udfName, className, uris,
UDFExecutableManager.getInstance(), true);
- future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
- } catch (Exception e) {
- final String message =
- String.format(
- "Failed to create function %s(%s), URI: %s, because %s.",
- udfName, className, uris, e.getMessage());
- LOGGER.error(message, e);
- future.setException(
- new StatementExecutionException(
- new
TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode())
- .setMessage(message)));
- }
+ return configTaskExecutor.createFunction(udfName, className, uris);
}
}
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/DeleteStorageGroupTask.java
index dceda3cb95..3da9564ff9 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/DeleteStorageGroupTask.java
@@ -19,36 +19,12 @@
package org.apache.iotdb.db.mpp.plan.execution.config;
-import org.apache.iotdb.common.rpc.thrift.TSStatus;
-import org.apache.iotdb.commons.client.IClientManager;
-import org.apache.iotdb.commons.consensus.PartitionRegionId;
-import org.apache.iotdb.commons.exception.IllegalPathException;
-import org.apache.iotdb.commons.exception.MetadataException;
-import org.apache.iotdb.commons.path.PartialPath;
-import org.apache.iotdb.confignode.rpc.thrift.TDeleteStorageGroupsReq;
-import org.apache.iotdb.db.client.ConfigNodeClient;
-import org.apache.iotdb.db.client.ConfigNodeInfo;
-import org.apache.iotdb.db.conf.IoTDBConfig;
-import org.apache.iotdb.db.conf.IoTDBDescriptor;
-import org.apache.iotdb.db.localconfignode.LocalConfigNode;
+import
org.apache.iotdb.db.mpp.plan.execution.config.executor.IConfigTaskExecutor;
import
org.apache.iotdb.db.mpp.plan.statement.metadata.DeleteStorageGroupStatement;
-import org.apache.iotdb.rpc.StatementExecutionException;
-import org.apache.iotdb.rpc.TSStatusCode;
import com.google.common.util.concurrent.ListenableFuture;
-import com.google.common.util.concurrent.SettableFuture;
-import org.apache.thrift.TException;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.IOException;
-import java.util.List;
-import java.util.stream.Collectors;
public class DeleteStorageGroupTask implements IConfigTask {
- private static final Logger LOGGER =
LoggerFactory.getLogger(DeleteStorageGroupTask.class);
-
- private static IoTDBConfig config =
IoTDBDescriptor.getInstance().getConfig();
private final DeleteStorageGroupStatement deleteStorageGroupStatement;
@@ -57,48 +33,9 @@ public class DeleteStorageGroupTask implements IConfigTask {
}
@Override
- public ListenableFuture<ConfigTaskResult> execute(
- IClientManager<PartitionRegionId, ConfigNodeClient> clientManager) {
- SettableFuture<ConfigTaskResult> future = SettableFuture.create();
- if (config.isClusterMode()) {
- TDeleteStorageGroupsReq req =
- new
TDeleteStorageGroupsReq(deleteStorageGroupStatement.getPrefixPath());
- try (ConfigNodeClient client =
clientManager.borrowClient(ConfigNodeInfo.partitionRegionId)) {
- TSStatus tsStatus = client.deleteStorageGroups(req);
- if (TSStatusCode.SUCCESS_STATUS.getStatusCode() != tsStatus.getCode())
{
- LOGGER.error(
- "Failed to execute delete storage group {} in config node,
status is {}.",
- deleteStorageGroupStatement.getPrefixPath(),
- tsStatus);
- future.setException(new StatementExecutionException(tsStatus));
- } else {
- future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
- }
- } catch (TException | IOException e) {
- LOGGER.error("Failed to connect to config node.");
- future.setException(e);
- }
- } else {
- try {
- List<PartialPath> deletePathList =
- deleteStorageGroupStatement.getPrefixPath().stream()
- .map(
- path -> {
- try {
- return new PartialPath(path);
- } catch (IllegalPathException e) {
- return null;
- }
- })
- .collect(Collectors.toList());
- LocalConfigNode.getInstance().deleteStorageGroups(deletePathList);
- } catch (MetadataException e) {
- future.setException(e);
- }
- future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
- }
+ public ListenableFuture<ConfigTaskResult> execute(IConfigTaskExecutor
configTaskExecutor) {
// If the action is executed successfully, return the Future.
// If your operation is async, you can return the corresponding future
directly.
- return future;
+ return configTaskExecutor.deleteStorageGroup(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/DropFunctionTask.java
index 6f67983284..458f73c550 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/DropFunctionTask.java
@@ -19,32 +19,13 @@
package org.apache.iotdb.db.mpp.plan.execution.config;
-import org.apache.iotdb.common.rpc.thrift.TSStatus;
-import org.apache.iotdb.commons.client.IClientManager;
-import org.apache.iotdb.commons.consensus.PartitionRegionId;
-import org.apache.iotdb.commons.udf.service.UDFRegistrationService;
-import org.apache.iotdb.confignode.rpc.thrift.TDropFunctionReq;
-import org.apache.iotdb.db.client.ConfigNodeClient;
-import org.apache.iotdb.db.client.ConfigNodeInfo;
-import org.apache.iotdb.db.conf.IoTDBConfig;
-import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import
org.apache.iotdb.db.mpp.plan.execution.config.executor.IConfigTaskExecutor;
import org.apache.iotdb.db.mpp.plan.statement.metadata.DropFunctionStatement;
-import org.apache.iotdb.rpc.StatementExecutionException;
-import org.apache.iotdb.rpc.TSStatusCode;
import com.google.common.util.concurrent.ListenableFuture;
-import com.google.common.util.concurrent.SettableFuture;
-import org.apache.thrift.TException;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.IOException;
public class DropFunctionTask implements IConfigTask {
- private static final Logger LOGGER =
LoggerFactory.getLogger(DropFunctionTask.class);
- private static final IoTDBConfig CONFIG =
IoTDBDescriptor.getInstance().getConfig();
-
private final String udfName;
public DropFunctionTask(DropFunctionStatement dropFunctionStatement) {
@@ -52,48 +33,8 @@ public class DropFunctionTask implements IConfigTask {
}
@Override
- public ListenableFuture<ConfigTaskResult> execute(
- IClientManager<PartitionRegionId, ConfigNodeClient> clientManager)
+ public ListenableFuture<ConfigTaskResult> execute(IConfigTaskExecutor
configTaskExecutor)
throws InterruptedException {
- SettableFuture<ConfigTaskResult> future = SettableFuture.create();
- if (CONFIG.isClusterMode()) {
- executeCluster(clientManager, future);
- } else {
- executeStandalone(future);
- }
- return future;
- }
-
- private void executeCluster(
- IClientManager<PartitionRegionId, ConfigNodeClient> clientManager,
- SettableFuture<ConfigTaskResult> future) {
- try (ConfigNodeClient client =
clientManager.borrowClient(ConfigNodeInfo.partitionRegionId)) {
- final TSStatus executionStatus = client.dropFunction(new
TDropFunctionReq(udfName));
-
- if (TSStatusCode.SUCCESS_STATUS.getStatusCode() !=
executionStatus.getCode()) {
- LOGGER.error("[{}] Failed to drop function {} in config node.",
executionStatus, udfName);
- future.setException(new StatementExecutionException(executionStatus));
- } else {
- future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
- }
- } catch (TException | IOException e) {
- LOGGER.error("Failed to connect to config node.");
- future.setException(e);
- }
- }
-
- private void executeStandalone(SettableFuture<ConfigTaskResult> future) {
- try {
- UDFRegistrationService.getInstance().deregister(udfName);
- future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
- } catch (Exception e) {
- final String message =
- String.format("Failed to drop function %s, because %s.", udfName,
e.getMessage());
- LOGGER.error(message, e);
- future.setException(
- new StatementExecutionException(
- new
TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode())
- .setMessage(message)));
- }
+ return configTaskExecutor.dropFunction(udfName);
}
}
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/FlushTask.java
index 231a967ce0..dcba6618fc 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/FlushTask.java
@@ -20,33 +20,19 @@
package org.apache.iotdb.db.mpp.plan.execution.config;
import org.apache.iotdb.common.rpc.thrift.TFlushReq;
-import org.apache.iotdb.common.rpc.thrift.TSStatus;
-import org.apache.iotdb.commons.client.IClientManager;
-import org.apache.iotdb.commons.consensus.PartitionRegionId;
import org.apache.iotdb.commons.path.PartialPath;
-import org.apache.iotdb.db.client.ConfigNodeClient;
-import org.apache.iotdb.db.client.ConfigNodeInfo;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
-import org.apache.iotdb.db.localconfignode.LocalConfigNode;
+import
org.apache.iotdb.db.mpp.plan.execution.config.executor.IConfigTaskExecutor;
import org.apache.iotdb.db.mpp.plan.statement.sys.FlushStatement;
-import org.apache.iotdb.rpc.StatementExecutionException;
-import org.apache.iotdb.rpc.TSStatusCode;
import com.google.common.util.concurrent.ListenableFuture;
-import com.google.common.util.concurrent.SettableFuture;
-import org.apache.thrift.TException;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
public class FlushTask implements IConfigTask {
- private static final Logger logger =
LoggerFactory.getLogger(FlushTask.class);
-
private FlushStatement flushStatement;
public FlushTask(FlushStatement flushStatement) {
@@ -54,11 +40,8 @@ public class FlushTask implements IConfigTask {
}
@Override
- public ListenableFuture<ConfigTaskResult> execute(
- IClientManager<PartitionRegionId, ConfigNodeClient> clientManager)
+ public ListenableFuture<ConfigTaskResult> execute(IConfigTaskExecutor
configTaskExecutor)
throws InterruptedException {
- SettableFuture<ConfigTaskResult> future = SettableFuture.create();
- TSStatus tsStatus = new TSStatus();
TFlushReq tFlushReq = new TFlushReq();
List<String> storageGroups = new ArrayList<>();
if (flushStatement.getStorageGroups() != null) {
@@ -76,26 +59,8 @@ public class FlushTask implements IConfigTask {
} else {
tFlushReq.setDataNodeId(-1);
}
- if (config.isClusterMode()) {
- try (ConfigNodeClient client =
clientManager.borrowClient(ConfigNodeInfo.partitionRegionId)) {
- // Send request to some API server
- tsStatus = client.flush(tFlushReq);
- // Get response or throw exception
- } catch (IOException | TException e) {
- logger.error("Failed to connect to config node.");
- future.setException(e);
- }
- } else {
- LocalConfigNode localConfigNode = LocalConfigNode.getInstance();
- tsStatus = localConfigNode.executeFlushOperation(tFlushReq);
- }
- if (tsStatus.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
- future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
- } else {
- future.setException(new StatementExecutionException(tsStatus));
- }
// If the action is executed successfully, return the Future.
// If your operation is async, you can return the corresponding future
directly.
- return future;
+ return configTaskExecutor.flush(tFlushReq);
}
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/IConfigTask.java
b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/IConfigTask.java
index f7fbd4164e..19a37912dd 100644
---
a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/IConfigTask.java
+++
b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/IConfigTask.java
@@ -19,14 +19,11 @@
package org.apache.iotdb.db.mpp.plan.execution.config;
-import org.apache.iotdb.commons.client.IClientManager;
-import org.apache.iotdb.commons.consensus.PartitionRegionId;
-import org.apache.iotdb.db.client.ConfigNodeClient;
+import
org.apache.iotdb.db.mpp.plan.execution.config.executor.IConfigTaskExecutor;
import com.google.common.util.concurrent.ListenableFuture;
public interface IConfigTask {
- ListenableFuture<ConfigTaskResult> execute(
- IClientManager<PartitionRegionId, ConfigNodeClient> clientManager)
+ ListenableFuture<ConfigTaskResult> execute(IConfigTaskExecutor
configTaskExecutor)
throws InterruptedException;
}
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/SetStorageGroupTask.java
index ef2c251366..19b68e16c2 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/SetStorageGroupTask.java
@@ -19,32 +19,13 @@
package org.apache.iotdb.db.mpp.plan.execution.config;
-import org.apache.iotdb.common.rpc.thrift.TSStatus;
-import org.apache.iotdb.commons.client.IClientManager;
-import org.apache.iotdb.commons.consensus.PartitionRegionId;
-import org.apache.iotdb.confignode.rpc.thrift.TSetStorageGroupReq;
import org.apache.iotdb.confignode.rpc.thrift.TStorageGroupSchema;
-import org.apache.iotdb.db.client.ConfigNodeClient;
-import org.apache.iotdb.db.client.ConfigNodeInfo;
-import org.apache.iotdb.db.conf.IoTDBConfig;
-import org.apache.iotdb.db.conf.IoTDBDescriptor;
-import org.apache.iotdb.db.localconfignode.LocalConfigNode;
+import
org.apache.iotdb.db.mpp.plan.execution.config.executor.IConfigTaskExecutor;
import
org.apache.iotdb.db.mpp.plan.statement.metadata.SetStorageGroupStatement;
-import org.apache.iotdb.rpc.StatementExecutionException;
-import org.apache.iotdb.rpc.TSStatusCode;
import com.google.common.util.concurrent.ListenableFuture;
-import com.google.common.util.concurrent.SettableFuture;
-import org.apache.thrift.TException;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.IOException;
public class SetStorageGroupTask implements IConfigTask {
- private static final Logger LOGGER =
LoggerFactory.getLogger(SetStorageGroupTask.class);
-
- private static IoTDBConfig config =
IoTDBDescriptor.getInstance().getConfig();
private final SetStorageGroupStatement setStorageGroupStatement;
@@ -53,54 +34,15 @@ public class SetStorageGroupTask implements IConfigTask {
}
@Override
- public ListenableFuture<ConfigTaskResult> execute(
- IClientManager<PartitionRegionId, ConfigNodeClient> clientManager) {
- SettableFuture<ConfigTaskResult> future = SettableFuture.create();
- // TODO:(this judgement needs to be integrated in a high level framework)
- if (config.isClusterMode()) {
- // Construct request using statement
- TStorageGroupSchema storageGroupSchema = constructStorageGroupSchema();
- TSetStorageGroupReq req = new TSetStorageGroupReq(storageGroupSchema);
- try (ConfigNodeClient configNodeClient =
- clientManager.borrowClient(ConfigNodeInfo.partitionRegionId)) {
- // Send request to some API server
- TSStatus tsStatus = configNodeClient.setStorageGroup(req);
- // Get response or throw exception
- if (TSStatusCode.SUCCESS_STATUS.getStatusCode() != tsStatus.getCode())
{
- LOGGER.error(
- "Failed to execute set storage group {} in config node, status
is {}.",
- setStorageGroupStatement.getStorageGroupPath(),
- tsStatus);
- future.setException(new StatementExecutionException(tsStatus));
- } else {
- future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
- }
- } catch (TException | IOException e) {
- LOGGER.error("Failed to connect to config node.");
- future.setException(e);
- }
- } else {
- try {
- LocalConfigNode localConfigNode = LocalConfigNode.getInstance();
-
localConfigNode.setStorageGroup(setStorageGroupStatement.getStorageGroupPath());
- if (setStorageGroupStatement.getTTL() != null) {
- localConfigNode.setTTL(
- setStorageGroupStatement.getStorageGroupPath(),
setStorageGroupStatement.getTTL());
- }
- // schemaReplicationFactor, dataReplicationFactor,
timePartitionInterval are ignored
- future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
- } catch (Exception e) {
- LOGGER.error("Failed to set storage group, caused by ", e);
- future.setException(e);
- }
- }
+ public ListenableFuture<ConfigTaskResult> execute(IConfigTaskExecutor
configTaskExecutor) {
// If the action is executed successfully, return the Future.
// If your operation is async, you can return the corresponding future
directly.
- return future;
+ return configTaskExecutor.setStorageGroup(setStorageGroupStatement);
}
/** construct set storage group schema according to statement */
- private TStorageGroupSchema constructStorageGroupSchema() {
+ public static TStorageGroupSchema constructStorageGroupSchema(
+ SetStorageGroupStatement setStorageGroupStatement) {
TStorageGroupSchema storageGroupSchema = new TStorageGroupSchema();
storageGroupSchema.setName(setStorageGroupStatement.getStorageGroupPath().getFullPath());
if (setStorageGroupStatement.getTTL() != null) {
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/SetTTLTask.java
index 241634f074..3f968b8736 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/SetTTLTask.java
@@ -19,32 +19,12 @@
package org.apache.iotdb.db.mpp.plan.execution.config;
-import org.apache.iotdb.common.rpc.thrift.TSStatus;
-import org.apache.iotdb.commons.client.IClientManager;
-import org.apache.iotdb.commons.consensus.PartitionRegionId;
-import org.apache.iotdb.commons.exception.MetadataException;
-import org.apache.iotdb.confignode.rpc.thrift.TSetTTLReq;
-import org.apache.iotdb.db.client.ConfigNodeClient;
-import org.apache.iotdb.db.client.ConfigNodeInfo;
-import org.apache.iotdb.db.conf.IoTDBConfig;
-import org.apache.iotdb.db.conf.IoTDBDescriptor;
-import org.apache.iotdb.db.localconfignode.LocalConfigNode;
+import
org.apache.iotdb.db.mpp.plan.execution.config.executor.IConfigTaskExecutor;
import org.apache.iotdb.db.mpp.plan.statement.metadata.SetTTLStatement;
-import org.apache.iotdb.rpc.StatementExecutionException;
-import org.apache.iotdb.rpc.TSStatusCode;
import com.google.common.util.concurrent.ListenableFuture;
-import com.google.common.util.concurrent.SettableFuture;
-import org.apache.thrift.TException;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.IOException;
public class SetTTLTask implements IConfigTask {
- private static final Logger LOGGER =
LoggerFactory.getLogger(SetTTLTask.class);
-
- private static IoTDBConfig config =
IoTDBDescriptor.getInstance().getConfig();
protected final SetTTLStatement statement;
protected String taskName;
@@ -55,40 +35,8 @@ public class SetTTLTask implements IConfigTask {
}
@Override
- public ListenableFuture<ConfigTaskResult> execute(
- IClientManager<PartitionRegionId, ConfigNodeClient> clientManager)
+ public ListenableFuture<ConfigTaskResult> execute(IConfigTaskExecutor
configTaskExecutor)
throws InterruptedException {
- SettableFuture<ConfigTaskResult> future = SettableFuture.create();
- if (config.isClusterMode()) {
- TSetTTLReq setTTLReq =
- new TSetTTLReq(statement.getStorageGroupPath().getFullPath(),
statement.getTTL());
- try (ConfigNodeClient configNodeClient =
- clientManager.borrowClient(ConfigNodeInfo.partitionRegionId)) {
- // Send request to some API server
- TSStatus tsStatus = configNodeClient.setTTL(setTTLReq);
- // Get response or throw exception
- if (TSStatusCode.SUCCESS_STATUS.getStatusCode() != tsStatus.getCode())
{
- LOGGER.error(
- "Failed to execute {} {} in config node, status is {}.",
- taskName,
- statement.getStorageGroupPath(),
- tsStatus);
- future.setException(new StatementExecutionException(tsStatus));
- } else {
- future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
- }
- } catch (TException | IOException e) {
- LOGGER.error("Failed to connect to config node.");
- future.setException(e);
- }
- } else {
- try {
- LocalConfigNode.getInstance().setTTL(statement.getStorageGroupPath(),
statement.getTTL());
- } catch (MetadataException | IOException e) {
- future.setException(e);
- }
- future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
- }
- return future;
+ return configTaskExecutor.setTTL(statement, taskName);
}
}
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/ShowClusterTask.java
index 1be47b1c9d..fb475edfd0 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/ShowClusterTask.java
@@ -19,15 +19,10 @@
package org.apache.iotdb.db.mpp.plan.execution.config;
-import org.apache.iotdb.commons.client.IClientManager;
-import org.apache.iotdb.commons.consensus.PartitionRegionId;
import org.apache.iotdb.confignode.rpc.thrift.TClusterNodeInfos;
-import org.apache.iotdb.db.client.ConfigNodeClient;
-import org.apache.iotdb.db.client.ConfigNodeInfo;
-import org.apache.iotdb.db.conf.IoTDBConfig;
-import org.apache.iotdb.db.conf.IoTDBDescriptor;
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.executor.IConfigTaskExecutor;
import org.apache.iotdb.db.mpp.plan.statement.metadata.ShowClusterStatement;
import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.iotdb.tsfile.read.common.block.TsBlockBuilder;
@@ -35,11 +30,7 @@ import org.apache.iotdb.tsfile.utils.Binary;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.SettableFuture;
-import org.apache.thrift.TException;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import java.io.IOException;
import java.util.concurrent.atomic.AtomicInteger;
import static org.apache.iotdb.commons.conf.IoTDBConstant.NODE_STATUS_RUNNING;
@@ -48,29 +39,32 @@ import static
org.apache.iotdb.commons.conf.IoTDBConstant.NODE_TYPE_DATA_NODE;
public class ShowClusterTask implements IConfigTask {
- private static final Logger LOGGER =
LoggerFactory.getLogger(ShowClusterTask.class);
-
- private static final IoTDBConfig config =
IoTDBDescriptor.getInstance().getConfig();
-
public ShowClusterTask(ShowClusterStatement showClusterStatement) {}
@Override
- public ListenableFuture<ConfigTaskResult> execute(
- IClientManager<PartitionRegionId, ConfigNodeClient> clientManager)
+ public ListenableFuture<ConfigTaskResult> execute(IConfigTaskExecutor
configTaskExecutor)
throws InterruptedException {
- SettableFuture<ConfigTaskResult> future = SettableFuture.create();
- TClusterNodeInfos clusterNodeInfos = new TClusterNodeInfos();
+ return configTaskExecutor.showCluster();
+ }
- if (config.isClusterMode()) {
- try (ConfigNodeClient client =
clientManager.borrowClient(ConfigNodeInfo.partitionRegionId)) {
- clusterNodeInfos = client.getAllClusterNodeInfos();
- } catch (TException | IOException e) {
- LOGGER.error("Failed to connect to config node.");
- future.setException(e);
- }
- }
+ private static void buildTsBlock(
+ TsBlockBuilder builder,
+ int nodeId,
+ String nodeType,
+ String nodeStatus,
+ String hostAddress,
+ int port) {
+ builder.getTimeColumnBuilder().writeLong(0L);
+ builder.getColumnBuilder(0).writeInt(nodeId);
+ builder.getColumnBuilder(1).writeBinary(new Binary(nodeType));
+ builder.getColumnBuilder(2).writeBinary(new Binary(nodeStatus));
+ builder.getColumnBuilder(3).writeBinary(new Binary(hostAddress));
+ builder.getColumnBuilder(4).writeInt(port);
+ builder.declarePosition();
+ }
- // build TSBlock
+ public static void buildTSBlock(
+ TClusterNodeInfos clusterNodeInfos, SettableFuture<ConfigTaskResult>
future) {
TsBlockBuilder builder =
new
TsBlockBuilder(HeaderConstant.showClusterHeader.getRespDataTypes());
@@ -101,22 +95,5 @@ public class ShowClusterTask implements IConfigTask {
DatasetHeader datasetHeader = HeaderConstant.showClusterHeader;
future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS,
builder.build(), datasetHeader));
- return future;
- }
-
- private void buildTsBlock(
- TsBlockBuilder builder,
- int nodeId,
- String nodeType,
- String nodeStatus,
- String hostAddress,
- int port) {
- builder.getTimeColumnBuilder().writeLong(0L);
- builder.getColumnBuilder(0).writeInt(nodeId);
- builder.getColumnBuilder(1).writeBinary(new Binary(nodeType));
- builder.getColumnBuilder(2).writeBinary(new Binary(nodeStatus));
- builder.getColumnBuilder(3).writeBinary(new Binary(hostAddress));
- builder.getColumnBuilder(4).writeInt(port);
- builder.declarePosition();
}
}
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/ShowFunctionsTask.java
index c51b1d59c3..886a58fee4 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/ShowFunctionsTask.java
@@ -19,14 +19,12 @@
package org.apache.iotdb.db.mpp.plan.execution.config;
-import org.apache.iotdb.commons.client.IClientManager;
-import org.apache.iotdb.commons.consensus.PartitionRegionId;
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.client.ConfigNodeClient;
import org.apache.iotdb.db.mpp.common.header.HeaderConstant;
+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;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
@@ -57,8 +55,7 @@ public class ShowFunctionsTask implements IConfigTask {
private static final Logger LOGGER =
LoggerFactory.getLogger(ShowFunctionsTask.class);
@Override
- public ListenableFuture<ConfigTaskResult> execute(
- IClientManager<PartitionRegionId, ConfigNodeClient> clientManager)
+ public ListenableFuture<ConfigTaskResult> execute(IConfigTaskExecutor
configTaskExecutor)
throws InterruptedException {
final SettableFuture<ConfigTaskResult> future = SettableFuture.create();
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/ShowStorageGroupTask.java
index 79894f4824..dc31a5628a 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/ShowStorageGroupTask.java
@@ -19,20 +19,10 @@
package org.apache.iotdb.db.mpp.plan.execution.config;
-import org.apache.iotdb.commons.client.IClientManager;
-import org.apache.iotdb.commons.consensus.PartitionRegionId;
-import org.apache.iotdb.commons.exception.MetadataException;
-import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.confignode.rpc.thrift.TStorageGroupSchema;
-import org.apache.iotdb.confignode.rpc.thrift.TStorageGroupSchemaResp;
-import org.apache.iotdb.db.client.ConfigNodeClient;
-import org.apache.iotdb.db.client.ConfigNodeInfo;
-import org.apache.iotdb.db.conf.IoTDBConfig;
-import org.apache.iotdb.db.conf.IoTDBDescriptor;
-import org.apache.iotdb.db.localconfignode.LocalConfigNode;
-import org.apache.iotdb.db.metadata.mnode.IStorageGroupMNode;
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.executor.IConfigTaskExecutor;
import
org.apache.iotdb.db.mpp.plan.statement.metadata.ShowStorageGroupStatement;
import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.iotdb.tsfile.read.common.block.TsBlockBuilder;
@@ -40,20 +30,10 @@ import org.apache.iotdb.tsfile.utils.Binary;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.SettableFuture;
-import org.apache.thrift.TException;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import java.io.IOException;
-import java.util.Arrays;
-import java.util.HashMap;
-import java.util.List;
import java.util.Map;
public class ShowStorageGroupTask implements IConfigTask {
- private static final Logger LOGGER =
LoggerFactory.getLogger(ShowStorageGroupTask.class);
-
- private static final IoTDBConfig config =
IoTDBDescriptor.getInstance().getConfig();
private ShowStorageGroupStatement showStorageGroupStatement;
@@ -62,41 +42,14 @@ public class ShowStorageGroupTask implements IConfigTask {
}
@Override
- public ListenableFuture<ConfigTaskResult> execute(
- IClientManager<PartitionRegionId, ConfigNodeClient> clientManager)
+ public ListenableFuture<ConfigTaskResult> execute(IConfigTaskExecutor
configTaskExecutor)
throws InterruptedException {
- SettableFuture<ConfigTaskResult> future = SettableFuture.create();
- Map<String, TStorageGroupSchema> storageGroupSchemaMap = new HashMap<>();
- if (config.isClusterMode()) {
- List<String> storageGroupPathPattern =
- Arrays.asList(showStorageGroupStatement.getPathPattern().getNodes());
- try (ConfigNodeClient client =
clientManager.borrowClient(ConfigNodeInfo.partitionRegionId)) {
- TStorageGroupSchemaResp resp =
- client.getMatchedStorageGroupSchemas(storageGroupPathPattern);
- storageGroupSchemaMap = resp.getStorageGroupSchemaMap();
- } catch (TException | IOException e) {
- LOGGER.error("Failed to connect to config node.");
- future.setException(e);
- }
- } else {
- try {
- LocalConfigNode localConfigNode = LocalConfigNode.getInstance();
- List<PartialPath> partialPaths =
- localConfigNode.getMatchedStorageGroups(
- showStorageGroupStatement.getPathPattern(),
- showStorageGroupStatement.isPrefixPath());
- for (PartialPath storageGroupPath : partialPaths) {
- IStorageGroupMNode storageGroupMNode =
- localConfigNode.getStorageGroupNodeByPath(storageGroupPath);
- String storageGroup = storageGroupMNode.getFullPath();
- TStorageGroupSchema storageGroupSchema =
storageGroupMNode.getStorageGroupSchema();
- storageGroupSchemaMap.put(storageGroup, storageGroupSchema);
- }
- } catch (MetadataException e) {
- future.setException(e);
- }
- }
- // build TSBlock
+ return configTaskExecutor.showStorageGroup(showStorageGroupStatement);
+ }
+
+ public static void buildTSBlock(
+ Map<String, TStorageGroupSchema> storageGroupSchemaMap,
+ SettableFuture<ConfigTaskResult> future) {
TsBlockBuilder builder =
new
TsBlockBuilder(HeaderConstant.showStorageGroupHeader.getRespDataTypes());
for (Map.Entry<String, TStorageGroupSchema> entry :
storageGroupSchemaMap.entrySet()) {
@@ -116,6 +69,5 @@ public class ShowStorageGroupTask implements IConfigTask {
}
DatasetHeader datasetHeader = HeaderConstant.showStorageGroupHeader;
future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS,
builder.build(), datasetHeader));
- return future;
}
}
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/ShowTTLTask.java
index 02f348d970..f9aa46c423 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/ShowTTLTask.java
@@ -19,19 +19,9 @@
package org.apache.iotdb.db.mpp.plan.execution.config;
-import org.apache.iotdb.commons.client.IClientManager;
-import org.apache.iotdb.commons.consensus.PartitionRegionId;
-import org.apache.iotdb.commons.exception.MetadataException;
-import org.apache.iotdb.commons.path.PartialPath;
-import org.apache.iotdb.confignode.rpc.thrift.TStorageGroupSchema;
-import org.apache.iotdb.confignode.rpc.thrift.TStorageGroupSchemaResp;
-import org.apache.iotdb.db.client.ConfigNodeClient;
-import org.apache.iotdb.db.client.ConfigNodeInfo;
-import org.apache.iotdb.db.conf.IoTDBConfig;
-import org.apache.iotdb.db.conf.IoTDBDescriptor;
-import org.apache.iotdb.db.localconfignode.LocalConfigNode;
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.executor.IConfigTaskExecutor;
import org.apache.iotdb.db.mpp.plan.statement.metadata.ShowTTLStatement;
import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.iotdb.tsfile.read.common.block.TsBlockBuilder;
@@ -39,20 +29,10 @@ import org.apache.iotdb.tsfile.utils.Binary;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.SettableFuture;
-import org.apache.thrift.TException;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import java.io.IOException;
-import java.util.Arrays;
-import java.util.HashMap;
-import java.util.List;
import java.util.Map;
public class ShowTTLTask implements IConfigTask {
- private static final Logger LOGGER =
LoggerFactory.getLogger(ShowTTLTask.class);
-
- private static final IoTDBConfig config =
IoTDBDescriptor.getInstance().getConfig();
private ShowTTLStatement showTTLStatement;
@@ -61,63 +41,13 @@ public class ShowTTLTask implements IConfigTask {
}
@Override
- public ListenableFuture<ConfigTaskResult> execute(
- IClientManager<PartitionRegionId, ConfigNodeClient> clientManager)
+ public ListenableFuture<ConfigTaskResult> execute(IConfigTaskExecutor
configTaskExecutor)
throws InterruptedException {
- SettableFuture<ConfigTaskResult> future = SettableFuture.create();
- List<PartialPath> storageGroupPaths = showTTLStatement.getPaths();
- Map<String, Long> storageGroupToTTL = new HashMap<>();
- if (config.isClusterMode()) {
- try (ConfigNodeClient client =
clientManager.borrowClient(ConfigNodeInfo.partitionRegionId)) {
- if (showTTLStatement.isAll()) {
- List<String> allStorageGroupPathPattern = Arrays.asList("root",
"**");
- TStorageGroupSchemaResp resp =
- client.getMatchedStorageGroupSchemas(allStorageGroupPathPattern);
- for (Map.Entry<String, TStorageGroupSchema> entry :
- resp.getStorageGroupSchemaMap().entrySet()) {
- storageGroupToTTL.put(entry.getKey(), entry.getValue().getTTL());
- }
- } else {
- for (PartialPath storageGroupPath : storageGroupPaths) {
- List<String> storageGroupPathPattern =
Arrays.asList(storageGroupPath.getNodes());
- TStorageGroupSchemaResp resp =
- client.getMatchedStorageGroupSchemas(storageGroupPathPattern);
- for (Map.Entry<String, TStorageGroupSchema> entry :
- resp.getStorageGroupSchemaMap().entrySet()) {
- if (!storageGroupToTTL.containsKey(entry.getKey())) {
- storageGroupToTTL.put(entry.getKey(),
entry.getValue().getTTL());
- }
- }
- }
- }
- } catch (TException | IOException e) {
- LOGGER.error("Failed to connect to config node.");
- future.setException(e);
- }
- } else {
- try {
- Map<PartialPath, Long> allStorageGroupToTTL =
- LocalConfigNode.getInstance().getStorageGroupsTTL();
- for (PartialPath storageGroupPath : storageGroupPaths) {
- if (showTTLStatement.isAll()) {
- storageGroupToTTL.put(
- storageGroupPath.getFullPath(),
allStorageGroupToTTL.get(storageGroupPath));
- } else {
- List<PartialPath> matchedStorageGroupPaths =
- LocalConfigNode.getInstance()
- .getMatchedStorageGroups(storageGroupPath,
showTTLStatement.isPrefixPath());
- for (PartialPath matchedStorageGroupPath :
matchedStorageGroupPaths) {
- storageGroupToTTL.put(
- matchedStorageGroupPath.getFullPath(),
- allStorageGroupToTTL.get(matchedStorageGroupPath));
- }
- }
- }
- } catch (MetadataException e) {
- future.setException(e);
- }
- }
- // build TSBlock
+ return configTaskExecutor.showTTL(showTTLStatement);
+ }
+
+ public static void buildTSBlock(
+ Map<String, Long> storageGroupToTTL, SettableFuture<ConfigTaskResult>
future) {
TsBlockBuilder builder = new
TsBlockBuilder(HeaderConstant.showTTLHeader.getRespDataTypes());
for (Map.Entry<String, Long> entry : storageGroupToTTL.entrySet()) {
builder.getTimeColumnBuilder().writeLong(0);
@@ -131,6 +61,5 @@ public class ShowTTLTask implements IConfigTask {
}
DatasetHeader datasetHeader = HeaderConstant.showTTLHeader;
future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS,
builder.build(), datasetHeader));
- return future;
}
}
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
new file mode 100644
index 0000000000..0975776ded
--- /dev/null
+++
b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/executor/ClusterConfigTaskExecutor.java
@@ -0,0 +1,328 @@
+/*
+ * 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.execution.config.executor;
+
+import org.apache.iotdb.common.rpc.thrift.TFlushReq;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.client.IClientManager;
+import org.apache.iotdb.commons.consensus.PartitionRegionId;
+import org.apache.iotdb.commons.path.PartialPath;
+import org.apache.iotdb.confignode.rpc.thrift.TClusterNodeInfos;
+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.TSetStorageGroupReq;
+import org.apache.iotdb.confignode.rpc.thrift.TSetTTLReq;
+import org.apache.iotdb.confignode.rpc.thrift.TStorageGroupSchema;
+import org.apache.iotdb.confignode.rpc.thrift.TStorageGroupSchemaResp;
+import org.apache.iotdb.db.client.ConfigNodeClient;
+import org.apache.iotdb.db.client.ConfigNodeInfo;
+import org.apache.iotdb.db.client.DataNodeClientPoolFactory;
+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.ShowStorageGroupTask;
+import org.apache.iotdb.db.mpp.plan.execution.config.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;
+import org.apache.iotdb.db.mpp.plan.statement.metadata.SetTTLStatement;
+import
org.apache.iotdb.db.mpp.plan.statement.metadata.ShowStorageGroupStatement;
+import org.apache.iotdb.db.mpp.plan.statement.metadata.ShowTTLStatement;
+import org.apache.iotdb.rpc.StatementExecutionException;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import com.google.common.util.concurrent.SettableFuture;
+import org.apache.thrift.TException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class ClusterConfigTaskExecutor implements IConfigTaskExecutor {
+
+ private static final Logger LOGGER =
LoggerFactory.getLogger(ClusterConfigTaskExecutor.class);
+
+ private static final IClientManager<PartitionRegionId, ConfigNodeClient>
+ CONFIG_NODE_CLIENT_MANAGER =
+ new IClientManager.Factory<PartitionRegionId, ConfigNodeClient>()
+ .createClientManager(new
DataNodeClientPoolFactory.ConfigNodeClientPoolFactory());
+
+ private static final class ClusterConfigTaskExecutorHolder {
+ private static final ClusterConfigTaskExecutor INSTANCE = new
ClusterConfigTaskExecutor();
+
+ private ClusterConfigTaskExecutorHolder() {}
+ }
+
+ public static ClusterConfigTaskExecutor getInstance() {
+ return ClusterConfigTaskExecutor.ClusterConfigTaskExecutorHolder.INSTANCE;
+ }
+
+ @Override
+ public SettableFuture<ConfigTaskResult> setStorageGroup(
+ SetStorageGroupStatement setStorageGroupStatement) {
+ SettableFuture<ConfigTaskResult> future = SettableFuture.create();
+ // Construct request using statement
+ TStorageGroupSchema storageGroupSchema =
+
SetStorageGroupTask.constructStorageGroupSchema(setStorageGroupStatement);
+ TSetStorageGroupReq req = new TSetStorageGroupReq(storageGroupSchema);
+ try (ConfigNodeClient configNodeClient =
+
CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.partitionRegionId)) {
+ // Send request to some API server
+ TSStatus tsStatus = configNodeClient.setStorageGroup(req);
+ // Get response or throw exception
+ if (TSStatusCode.SUCCESS_STATUS.getStatusCode() != tsStatus.getCode()) {
+ LOGGER.error(
+ "Failed to execute set storage group {} in config node, status is
{}.",
+ setStorageGroupStatement.getStorageGroupPath(),
+ tsStatus);
+ future.setException(new StatementExecutionException(tsStatus));
+ } else {
+ future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
+ }
+ } catch (TException | IOException e) {
+ LOGGER.error("Failed to connect to config node.");
+ future.setException(e);
+ }
+ return future;
+ }
+
+ @Override
+ public SettableFuture<ConfigTaskResult> showStorageGroup(
+ ShowStorageGroupStatement showStorageGroupStatement) {
+ SettableFuture<ConfigTaskResult> future = SettableFuture.create();
+ Map<String, TStorageGroupSchema> storageGroupSchemaMap;
+ // Construct request using statement
+ List<String> storageGroupPathPattern =
+ Arrays.asList(showStorageGroupStatement.getPathPattern().getNodes());
+ try (ConfigNodeClient client =
+
CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.partitionRegionId)) {
+ // Send request to some API server
+ TStorageGroupSchemaResp resp =
client.getMatchedStorageGroupSchemas(storageGroupPathPattern);
+ storageGroupSchemaMap = resp.getStorageGroupSchemaMap();
+ // build TSBlock
+ ShowStorageGroupTask.buildTSBlock(storageGroupSchemaMap, future);
+ } catch (TException | IOException e) {
+ LOGGER.error("Failed to connect to config node.");
+ future.setException(e);
+ }
+ return future;
+ }
+
+ @Override
+ public SettableFuture<ConfigTaskResult> countStorageGroup(
+ CountStorageGroupStatement countStorageGroupStatement) {
+ SettableFuture<ConfigTaskResult> future = SettableFuture.create();
+ int storageGroupNum;
+ List<String> storageGroupPathPattern =
+ Arrays.asList(countStorageGroupStatement.getPartialPath().getNodes());
+ try (ConfigNodeClient client =
+
CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.partitionRegionId)) {
+ TCountStorageGroupResp resp =
client.countMatchedStorageGroups(storageGroupPathPattern);
+ storageGroupNum = resp.getCount();
+ // build TSBlock
+ CountStorageGroupTask.buildTSBlock(storageGroupNum, future);
+ } catch (TException | IOException e) {
+ LOGGER.error("Failed to connect to config node.");
+ future.setException(e);
+ }
+ return future;
+ }
+
+ @Override
+ public SettableFuture<ConfigTaskResult> createFunction(
+ String udfName, String className, List<String> uris) {
+ SettableFuture<ConfigTaskResult> future = SettableFuture.create();
+ try (ConfigNodeClient client =
+
CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.partitionRegionId)) {
+ final TSStatus executionStatus =
+ client.createFunction(new TCreateFunctionReq(udfName, className,
uris));
+
+ if (TSStatusCode.SUCCESS_STATUS.getStatusCode() !=
executionStatus.getCode()) {
+ LOGGER.error(
+ "[{}] Failed to create function {}({}) in config node, URI: {}.",
+ executionStatus,
+ udfName,
+ className,
+ uris);
+ future.setException(new StatementExecutionException(executionStatus));
+ } else {
+ future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
+ }
+ } catch (TException | IOException e) {
+ LOGGER.error("Failed to connect to config node.");
+ future.setException(e);
+ }
+ return future;
+ }
+
+ @Override
+ public SettableFuture<ConfigTaskResult> deleteStorageGroup(
+ DeleteStorageGroupStatement deleteStorageGroupStatement) {
+ SettableFuture<ConfigTaskResult> future = SettableFuture.create();
+ TDeleteStorageGroupsReq req =
+ new
TDeleteStorageGroupsReq(deleteStorageGroupStatement.getPrefixPath());
+ try (ConfigNodeClient client =
+
CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.partitionRegionId)) {
+ TSStatus tsStatus = client.deleteStorageGroups(req);
+ if (TSStatusCode.SUCCESS_STATUS.getStatusCode() != tsStatus.getCode()) {
+ LOGGER.error(
+ "Failed to execute delete storage group {} in config node, status
is {}.",
+ deleteStorageGroupStatement.getPrefixPath(),
+ tsStatus);
+ future.setException(new StatementExecutionException(tsStatus));
+ } else {
+ future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
+ }
+ } catch (TException | IOException e) {
+ LOGGER.error("Failed to connect to config node.");
+ future.setException(e);
+ }
+ return future;
+ }
+
+ @Override
+ public SettableFuture<ConfigTaskResult> dropFunction(String udfName) {
+ SettableFuture<ConfigTaskResult> future = SettableFuture.create();
+ try (ConfigNodeClient client =
+
CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.partitionRegionId)) {
+ final TSStatus executionStatus = client.dropFunction(new
TDropFunctionReq(udfName));
+
+ if (TSStatusCode.SUCCESS_STATUS.getStatusCode() !=
executionStatus.getCode()) {
+ LOGGER.error("[{}] Failed to drop function {} in config node.",
executionStatus, udfName);
+ future.setException(new StatementExecutionException(executionStatus));
+ } else {
+ future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
+ }
+ } catch (TException | IOException e) {
+ LOGGER.error("Failed to connect to config node.");
+ future.setException(e);
+ }
+ return future;
+ }
+
+ @Override
+ public SettableFuture<ConfigTaskResult> setTTL(SetTTLStatement
setTTLStatement, String taskName) {
+ SettableFuture<ConfigTaskResult> future = SettableFuture.create();
+ TSetTTLReq setTTLReq =
+ new TSetTTLReq(
+ setTTLStatement.getStorageGroupPath().getFullPath(),
setTTLStatement.getTTL());
+ try (ConfigNodeClient configNodeClient =
+
CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.partitionRegionId)) {
+ // Send request to some API server
+ TSStatus tsStatus = configNodeClient.setTTL(setTTLReq);
+ // Get response or throw exception
+ if (TSStatusCode.SUCCESS_STATUS.getStatusCode() != tsStatus.getCode()) {
+ LOGGER.error(
+ "Failed to execute {} {} in config node, status is {}.",
+ taskName,
+ setTTLStatement.getStorageGroupPath(),
+ tsStatus);
+ future.setException(new StatementExecutionException(tsStatus));
+ } else {
+ future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
+ }
+ } catch (TException | IOException e) {
+ LOGGER.error("Failed to connect to config node.");
+ future.setException(e);
+ }
+ return future;
+ }
+
+ @Override
+ public SettableFuture<ConfigTaskResult> flush(TFlushReq tFlushReq) {
+ SettableFuture<ConfigTaskResult> future = SettableFuture.create();
+
+ try (ConfigNodeClient client =
+
CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.partitionRegionId)) {
+ // Send request to some API server
+ TSStatus tsStatus = client.flush(tFlushReq);
+ // 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) {
+ LOGGER.error("Failed to connect to config node.");
+ future.setException(e);
+ }
+ return future;
+ }
+
+ @Override
+ public SettableFuture<ConfigTaskResult> showCluster() {
+ SettableFuture<ConfigTaskResult> future = SettableFuture.create();
+ TClusterNodeInfos clusterNodeInfos = new TClusterNodeInfos();
+ try (ConfigNodeClient client =
+
CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.partitionRegionId)) {
+ clusterNodeInfos = client.getAllClusterNodeInfos();
+ } catch (TException | IOException e) {
+ LOGGER.error("Failed to connect to config node.");
+ future.setException(e);
+ }
+ // build TSBlock
+ ShowClusterTask.buildTSBlock(clusterNodeInfos, future);
+ return future;
+ }
+
+ @Override
+ public SettableFuture<ConfigTaskResult> showTTL(ShowTTLStatement
showTTLStatement) {
+ SettableFuture<ConfigTaskResult> future = SettableFuture.create();
+ List<PartialPath> storageGroupPaths = showTTLStatement.getPaths();
+ Map<String, Long> storageGroupToTTL = new HashMap<>();
+ try (ConfigNodeClient client =
+
CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.partitionRegionId)) {
+ if (showTTLStatement.isAll()) {
+ List<String> allStorageGroupPathPattern = Arrays.asList("root", "**");
+ TStorageGroupSchemaResp resp =
+ client.getMatchedStorageGroupSchemas(allStorageGroupPathPattern);
+ for (Map.Entry<String, TStorageGroupSchema> entry :
+ resp.getStorageGroupSchemaMap().entrySet()) {
+ storageGroupToTTL.put(entry.getKey(), entry.getValue().getTTL());
+ }
+ } else {
+ for (PartialPath storageGroupPath : storageGroupPaths) {
+ List<String> storageGroupPathPattern =
Arrays.asList(storageGroupPath.getNodes());
+ TStorageGroupSchemaResp resp =
+ client.getMatchedStorageGroupSchemas(storageGroupPathPattern);
+ for (Map.Entry<String, TStorageGroupSchema> entry :
+ resp.getStorageGroupSchemaMap().entrySet()) {
+ if (!storageGroupToTTL.containsKey(entry.getKey())) {
+ storageGroupToTTL.put(entry.getKey(), entry.getValue().getTTL());
+ }
+ }
+ }
+ }
+ } catch (TException | IOException e) {
+ LOGGER.error("Failed to connect to config node.");
+ future.setException(e);
+ }
+ // build TSBlock
+ ShowTTLTask.buildTSBlock(storageGroupToTTL, future);
+ return future;
+ }
+}
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
new file mode 100644
index 0000000000..f49390553b
--- /dev/null
+++
b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/executor/IConfigTaskExecutor.java
@@ -0,0 +1,61 @@
+/*
+ * 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.execution.config.executor;
+
+import org.apache.iotdb.common.rpc.thrift.TFlushReq;
+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;
+import
org.apache.iotdb.db.mpp.plan.statement.metadata.SetStorageGroupStatement;
+import org.apache.iotdb.db.mpp.plan.statement.metadata.SetTTLStatement;
+import
org.apache.iotdb.db.mpp.plan.statement.metadata.ShowStorageGroupStatement;
+import org.apache.iotdb.db.mpp.plan.statement.metadata.ShowTTLStatement;
+
+import com.google.common.util.concurrent.SettableFuture;
+
+import java.util.List;
+
+public interface IConfigTaskExecutor {
+
+ SettableFuture<ConfigTaskResult> setStorageGroup(
+ SetStorageGroupStatement setStorageGroupStatement);
+
+ SettableFuture<ConfigTaskResult> showStorageGroup(
+ ShowStorageGroupStatement showStorageGroupStatement);
+
+ SettableFuture<ConfigTaskResult> countStorageGroup(
+ CountStorageGroupStatement countStorageGroupStatement);
+
+ SettableFuture<ConfigTaskResult> createFunction(
+ String udfName, String className, List<String> uris);
+
+ SettableFuture<ConfigTaskResult> deleteStorageGroup(
+ DeleteStorageGroupStatement deleteStorageGroupStatement);
+
+ SettableFuture<ConfigTaskResult> dropFunction(String udfName);
+
+ SettableFuture<ConfigTaskResult> setTTL(SetTTLStatement setTTLStatement,
String taskName);
+
+ SettableFuture<ConfigTaskResult> flush(TFlushReq tFlushReq);
+
+ SettableFuture<ConfigTaskResult> showCluster();
+
+ SettableFuture<ConfigTaskResult> showTTL(ShowTTLStatement showTTLStatement);
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/executor/StandsloneConfigTaskExecutor.java
b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/executor/StandsloneConfigTaskExecutor.java
new file mode 100644
index 0000000000..4a2ff08050
--- /dev/null
+++
b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/executor/StandsloneConfigTaskExecutor.java
@@ -0,0 +1,265 @@
+/*
+ * 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.execution.config.executor;
+
+import org.apache.iotdb.common.rpc.thrift.TFlushReq;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.exception.IllegalPathException;
+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.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.statement.metadata.CountStorageGroupStatement;
+import
org.apache.iotdb.db.mpp.plan.statement.metadata.DeleteStorageGroupStatement;
+import
org.apache.iotdb.db.mpp.plan.statement.metadata.SetStorageGroupStatement;
+import org.apache.iotdb.db.mpp.plan.statement.metadata.SetTTLStatement;
+import
org.apache.iotdb.db.mpp.plan.statement.metadata.ShowStorageGroupStatement;
+import org.apache.iotdb.db.mpp.plan.statement.metadata.ShowTTLStatement;
+import org.apache.iotdb.rpc.RpcUtils;
+import org.apache.iotdb.rpc.StatementExecutionException;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import com.google.common.util.concurrent.SettableFuture;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+public class StandsloneConfigTaskExecutor implements IConfigTaskExecutor {
+
+ private static final Logger LOGGER =
LoggerFactory.getLogger(StandsloneConfigTaskExecutor.class);
+
+ private static final class StandsloneConfigTaskExecutorHolder {
+ private static final StandsloneConfigTaskExecutor INSTANCE = new
StandsloneConfigTaskExecutor();
+
+ private StandsloneConfigTaskExecutorHolder() {}
+ }
+
+ public static StandsloneConfigTaskExecutor getInstance() {
+ return
StandsloneConfigTaskExecutor.StandsloneConfigTaskExecutorHolder.INSTANCE;
+ }
+
+ @Override
+ public SettableFuture<ConfigTaskResult> setStorageGroup(
+ SetStorageGroupStatement setStorageGroupStatement) {
+ SettableFuture<ConfigTaskResult> future = SettableFuture.create();
+ try {
+ LocalConfigNode localConfigNode = LocalConfigNode.getInstance();
+
localConfigNode.setStorageGroup(setStorageGroupStatement.getStorageGroupPath());
+ if (setStorageGroupStatement.getTTL() != null) {
+ localConfigNode.setTTL(
+ setStorageGroupStatement.getStorageGroupPath(),
setStorageGroupStatement.getTTL());
+ }
+ // schemaReplicationFactor, dataReplicationFactor, timePartitionInterval
are ignored
+ future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
+ } catch (Exception e) {
+ LOGGER.error("Failed to set storage group, caused by ", e);
+ future.setException(e);
+ }
+ return future;
+ }
+
+ @Override
+ public SettableFuture<ConfigTaskResult> showStorageGroup(
+ ShowStorageGroupStatement showStorageGroupStatement) {
+ SettableFuture<ConfigTaskResult> future = SettableFuture.create();
+ Map<String, TStorageGroupSchema> storageGroupSchemaMap = new HashMap<>();
+ try {
+ LocalConfigNode localConfigNode = LocalConfigNode.getInstance();
+ List<PartialPath> partialPaths =
+ localConfigNode.getMatchedStorageGroups(
+ showStorageGroupStatement.getPathPattern(),
showStorageGroupStatement.isPrefixPath());
+ for (PartialPath storageGroupPath : partialPaths) {
+ IStorageGroupMNode storageGroupMNode =
+ localConfigNode.getStorageGroupNodeByPath(storageGroupPath);
+ String storageGroup = storageGroupMNode.getFullPath();
+ TStorageGroupSchema storageGroupSchema =
storageGroupMNode.getStorageGroupSchema();
+ storageGroupSchemaMap.put(storageGroup, storageGroupSchema);
+ // build TSBlock
+ ShowStorageGroupTask.buildTSBlock(storageGroupSchemaMap, future);
+ }
+ } catch (MetadataException e) {
+ future.setException(e);
+ }
+ return future;
+ }
+
+ @Override
+ public SettableFuture<ConfigTaskResult> countStorageGroup(
+ CountStorageGroupStatement countStorageGroupStatement) {
+ SettableFuture<ConfigTaskResult> future = SettableFuture.create();
+ int storageGroupNum;
+ try {
+ storageGroupNum =
+ LocalConfigNode.getInstance()
+ .getStorageGroupNum(
+ countStorageGroupStatement.getPartialPath(),
+ countStorageGroupStatement.isPrefixPath());
+ // build TSBlock
+ CountStorageGroupTask.buildTSBlock(storageGroupNum, future);
+ } catch (MetadataException e) {
+ future.setException(e);
+ }
+ return future;
+ }
+
+ @Override
+ public SettableFuture<ConfigTaskResult> createFunction(
+ String udfName, String className, List<String> uris) {
+ SettableFuture<ConfigTaskResult> future = SettableFuture.create();
+ try {
+ UDFRegistrationService.getInstance()
+ .register(udfName, className, uris,
UDFExecutableManager.getInstance(), true);
+ future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
+ } catch (Exception e) {
+ final String message =
+ String.format(
+ "Failed to create function %s(%s), URI: %s, because %s.",
+ udfName, className, uris, e.getMessage());
+ LOGGER.error(message, e);
+ future.setException(
+ new StatementExecutionException(
+ new
TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode())
+ .setMessage(message)));
+ }
+ return future;
+ }
+
+ @Override
+ public SettableFuture<ConfigTaskResult> deleteStorageGroup(
+ DeleteStorageGroupStatement deleteStorageGroupStatement) {
+ SettableFuture<ConfigTaskResult> future = SettableFuture.create();
+ try {
+ List<PartialPath> deletePathList =
+ deleteStorageGroupStatement.getPrefixPath().stream()
+ .map(
+ path -> {
+ try {
+ return new PartialPath(path);
+ } catch (IllegalPathException e) {
+ return null;
+ }
+ })
+ .collect(Collectors.toList());
+ LocalConfigNode.getInstance().deleteStorageGroups(deletePathList);
+ } catch (MetadataException e) {
+ future.setException(e);
+ }
+ future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
+ return future;
+ }
+
+ @Override
+ public SettableFuture<ConfigTaskResult> dropFunction(String udfName) {
+ SettableFuture<ConfigTaskResult> future = SettableFuture.create();
+ try {
+ UDFRegistrationService.getInstance().deregister(udfName);
+ future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
+ } catch (Exception e) {
+ final String message =
+ String.format("Failed to drop function %s, because %s.", udfName,
e.getMessage());
+ LOGGER.error(message, e);
+ future.setException(
+ new StatementExecutionException(
+ new
TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode())
+ .setMessage(message)));
+ }
+ return future;
+ }
+
+ @Override
+ public SettableFuture<ConfigTaskResult> setTTL(SetTTLStatement
setTTLStatement, String taskName) {
+ SettableFuture<ConfigTaskResult> future = SettableFuture.create();
+ try {
+ LocalConfigNode.getInstance()
+ .setTTL(setTTLStatement.getStorageGroupPath(),
setTTLStatement.getTTL());
+ } catch (MetadataException | IOException e) {
+ future.setException(e);
+ }
+ future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
+ return future;
+ }
+
+ @Override
+ public SettableFuture<ConfigTaskResult> flush(TFlushReq tFlushReq) {
+ SettableFuture<ConfigTaskResult> future = SettableFuture.create();
+ LocalConfigNode localConfigNode = LocalConfigNode.getInstance();
+ TSStatus tsStatus = localConfigNode.executeFlushOperation(tFlushReq);
+ 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> showCluster() {
+ SettableFuture<ConfigTaskResult> future = SettableFuture.create();
+ future.setException(
+ new StatementExecutionException(
+ RpcUtils.getStatus(
+ TSStatusCode.EXECUTE_STATEMENT_ERROR,
+ "Executing this command in standalone mode is not
supported")));
+ return future;
+ }
+
+ @Override
+ public SettableFuture<ConfigTaskResult> showTTL(ShowTTLStatement
showTTLStatement) {
+ SettableFuture<ConfigTaskResult> future = SettableFuture.create();
+ List<PartialPath> storageGroupPaths = showTTLStatement.getPaths();
+ Map<String, Long> storageGroupToTTL = new HashMap<>();
+ try {
+ Map<PartialPath, Long> allStorageGroupToTTL =
+ LocalConfigNode.getInstance().getStorageGroupsTTL();
+ for (PartialPath storageGroupPath : storageGroupPaths) {
+ if (showTTLStatement.isAll()) {
+ storageGroupToTTL.put(
+ storageGroupPath.getFullPath(),
allStorageGroupToTTL.get(storageGroupPath));
+ } else {
+ List<PartialPath> matchedStorageGroupPaths =
+ LocalConfigNode.getInstance()
+ .getMatchedStorageGroups(storageGroupPath,
showTTLStatement.isPrefixPath());
+ for (PartialPath matchedStorageGroupPath : matchedStorageGroupPaths)
{
+ storageGroupToTTL.put(
+ matchedStorageGroupPath.getFullPath(),
+ allStorageGroupToTTL.get(matchedStorageGroupPath));
+ }
+ }
+ }
+ } catch (MetadataException e) {
+ future.setException(e);
+ }
+ // build TSBlock
+ ShowTTLTask.buildTSBlock(storageGroupToTTL, future);
+ return future;
+ }
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/AuthorPlan.java
b/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/AuthorPlan.java
index 063d172ae6..cb39680963 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/AuthorPlan.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/AuthorPlan.java
@@ -19,9 +19,9 @@
package org.apache.iotdb.db.qp.physical.sys;
import org.apache.iotdb.commons.auth.AuthException;
-import org.apache.iotdb.commons.auth.entity.PrivilegeType;
import org.apache.iotdb.commons.exception.IllegalPathException;
import org.apache.iotdb.commons.path.PartialPath;
+import org.apache.iotdb.commons.utils.AuthUtils;
import org.apache.iotdb.db.qp.logical.Operator;
import org.apache.iotdb.db.qp.logical.Operator.OperatorType;
import org.apache.iotdb.db.qp.logical.sys.AuthorOperator;
@@ -74,7 +74,7 @@ public class AuthorPlan extends PhysicalPlan {
this.roleName = roleName;
this.password = password;
this.newPassword = newPassword;
- this.permissions = strToPermissions(authorizationList);
+ this.permissions = AuthUtils.strToPermissions(authorizationList);
this.nodeName = nodeName;
switch (authorType) {
case DROP_ROLE:
@@ -222,28 +222,6 @@ public class AuthorPlan extends PhysicalPlan {
return userName;
}
- public static Set<Integer> strToPermissions(String[] authorizationList)
throws AuthException {
- Set<Integer> result = new HashSet<>();
- if (authorizationList == null) {
- return result;
- }
- for (String s : authorizationList) {
- PrivilegeType[] types = PrivilegeType.values();
- boolean legal = false;
- for (PrivilegeType privilegeType : types) {
- if (s.equalsIgnoreCase(privilegeType.name())) {
- result.add(privilegeType.ordinal());
- legal = true;
- break;
- }
- }
- if (!legal) {
- throw new AuthException("No such privilege " + s);
- }
- }
- return result;
- }
-
@Override
public String toString() {
return "userName: "
diff --git
a/server/src/test/java/org/apache/iotdb/db/mpp/execution/ConfigExecutionTest.java
b/server/src/test/java/org/apache/iotdb/db/mpp/execution/ConfigExecutionTest.java
index 0741a6b4fa..f43e702772 100644
---
a/server/src/test/java/org/apache/iotdb/db/mpp/execution/ConfigExecutionTest.java
+++
b/server/src/test/java/org/apache/iotdb/db/mpp/execution/ConfigExecutionTest.java
@@ -19,10 +19,7 @@
package org.apache.iotdb.db.mpp.execution;
-import org.apache.iotdb.commons.client.IClientManager;
import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory;
-import org.apache.iotdb.commons.consensus.PartitionRegionId;
-import org.apache.iotdb.db.client.ConfigNodeClient;
import org.apache.iotdb.db.mpp.common.MPPQueryContext;
import org.apache.iotdb.db.mpp.common.QueryId;
import org.apache.iotdb.db.mpp.common.header.ColumnHeader;
@@ -32,6 +29,7 @@ import org.apache.iotdb.db.mpp.plan.execution.ExecutionResult;
import org.apache.iotdb.db.mpp.plan.execution.config.ConfigExecution;
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.rpc.TSStatusCode;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.read.common.block.TsBlock;
@@ -114,8 +112,7 @@ public class ConfigExecutionTest {
}
@Override
- public ListenableFuture<ConfigTaskResult> execute(
- IClientManager<PartitionRegionId, ConfigNodeClient> clientManager)
+ public ListenableFuture<ConfigTaskResult> execute(IConfigTaskExecutor
configTaskFetcher)
throws InterruptedException {
return result;
}
