You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ro...@apache.org on 2022/05/30 06:41:18 UTC
[iotdb] branch master updated: [IOTDB-3228] UDF Management in MPP Cluster: DROP FUNCTION (#6064)
This is an automated email from the ASF dual-hosted git repository.
rong 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 fcb27a8449 [IOTDB-3228] UDF Management in MPP Cluster: DROP FUNCTION (#6064)
fcb27a8449 is described below
commit fcb27a844972cc978729f4cfa5161f86c0fcb412
Author: Steve Yurong Su <ro...@apache.org>
AuthorDate: Mon May 30 14:41:13 2022 +0800
[IOTDB-3228] UDF Management in MPP Cluster: DROP FUNCTION (#6064)
---
.../confignode/client/AsyncDataNodeClientPool.java | 19 ++++-
...Handler.java => FunctionManagementHandler.java} | 6 +-
.../consensus/request/ConfigRequest.java | 4 +
.../consensus/request/ConfigRequestType.java | 1 +
.../consensus/request/write/DropFunctionReq.java | 56 ++++++++++++
.../iotdb/confignode/manager/ConfigManager.java | 8 ++
.../apache/iotdb/confignode/manager/Manager.java | 2 +
.../iotdb/confignode/manager/UDFManager.java | 59 +++++++++++--
.../iotdb/confignode/persistence/UDFInfo.java | 25 +++++-
.../executor/ConfigRequestExecutor.java | 3 +
.../thrift/ConfigNodeRPCServiceProcessor.java | 6 ++
.../udf/service/UDFRegistrationService.java | 8 +-
.../apache/iotdb/db/client/ConfigNodeClient.java | 17 ++++
.../plan/execution/config/ConfigTaskVisitor.java | 7 ++
.../plan/execution/config/DropFunctionTask.java | 99 ++++++++++++++++++++++
.../iotdb/db/mpp/plan/parser/ASTVisitor.java | 8 ++
.../db/mpp/plan/statement/StatementVisitor.java | 5 ++
.../statement/metadata/DropFunctionStatement.java | 60 +++++++++++++
.../service/thrift/impl/InternalServiceImpl.java | 12 +++
.../src/main/thrift/confignode.thrift | 8 +-
thrift/src/main/thrift/mpp.thrift | 11 +++
21 files changed, 404 insertions(+), 20 deletions(-)
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/client/AsyncDataNodeClientPool.java b/confignode/src/main/java/org/apache/iotdb/confignode/client/AsyncDataNodeClientPool.java
index 17d93b8f85..a38218ba04 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/client/AsyncDataNodeClientPool.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/client/AsyncDataNodeClientPool.java
@@ -25,13 +25,14 @@ import org.apache.iotdb.common.rpc.thrift.THeartbeatReq;
import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
import org.apache.iotdb.commons.client.IClientManager;
import org.apache.iotdb.commons.client.async.AsyncDataNodeInternalServiceClient;
-import org.apache.iotdb.confignode.client.handlers.CreateFunctionHandler;
import org.apache.iotdb.confignode.client.handlers.CreateRegionHandler;
+import org.apache.iotdb.confignode.client.handlers.FunctionManagementHandler;
import org.apache.iotdb.confignode.client.handlers.HeartbeatHandler;
import org.apache.iotdb.confignode.consensus.request.write.CreateRegionsReq;
import org.apache.iotdb.mpp.rpc.thrift.TCreateDataRegionReq;
import org.apache.iotdb.mpp.rpc.thrift.TCreateFunctionRequest;
import org.apache.iotdb.mpp.rpc.thrift.TCreateSchemaRegionReq;
+import org.apache.iotdb.mpp.rpc.thrift.TDropFunctionRequest;
import org.apache.thrift.TException;
import org.slf4j.Logger;
@@ -241,7 +242,7 @@ public class AsyncDataNodeClientPool {
* @param endPoint The specific DataNode
*/
public void createFunction(
- TEndPoint endPoint, TCreateFunctionRequest request, CreateFunctionHandler handler) {
+ TEndPoint endPoint, TCreateFunctionRequest request, FunctionManagementHandler handler) {
try {
clientManager.borrowClient(endPoint).createFunction(request, handler);
} catch (Exception e) {
@@ -249,6 +250,20 @@ public class AsyncDataNodeClientPool {
}
}
+ /**
+ * Only used in UDFManager
+ *
+ * @param endPoint The specific DataNode
+ */
+ public void dropFunction(
+ TEndPoint endPoint, TDropFunctionRequest request, FunctionManagementHandler handler) {
+ try {
+ clientManager.borrowClient(endPoint).dropFunction(request, handler);
+ } catch (Exception e) {
+ LOGGER.error("Failed to asking DataNode to create function: {}", endPoint, e);
+ }
+ }
+
// TODO: Is the ClientPool must be a singleton?
private static class ClientPoolHolder {
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/client/handlers/CreateFunctionHandler.java b/confignode/src/main/java/org/apache/iotdb/confignode/client/handlers/FunctionManagementHandler.java
similarity index 89%
rename from confignode/src/main/java/org/apache/iotdb/confignode/client/handlers/CreateFunctionHandler.java
rename to confignode/src/main/java/org/apache/iotdb/confignode/client/handlers/FunctionManagementHandler.java
index e740a5c6a9..6ec4b4f06f 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/client/handlers/CreateFunctionHandler.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/client/handlers/FunctionManagementHandler.java
@@ -27,14 +27,14 @@ import org.apache.thrift.async.AsyncMethodCallback;
import java.util.List;
import java.util.concurrent.CountDownLatch;
-public class CreateFunctionHandler implements AsyncMethodCallback<TSStatus> {
+public class FunctionManagementHandler implements AsyncMethodCallback<TSStatus> {
private final CountDownLatch countDownLatch;
private final List<TSStatus> dataNodeResponseStatus;
private final String ip;
private final int port;
- public CreateFunctionHandler(
+ public FunctionManagementHandler(
CountDownLatch countDownLatch, List<TSStatus> dataNodeResponseStatus, String ip, int port) {
this.countDownLatch = countDownLatch;
this.dataNodeResponseStatus = dataNodeResponseStatus;
@@ -53,6 +53,6 @@ public class CreateFunctionHandler implements AsyncMethodCallback<TSStatus> {
countDownLatch.countDown();
dataNodeResponseStatus.add(
new TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode())
- .setMessage("[" + ip + ":" + port + "] " + exception.getMessage()));
+ .setMessage("DataNode[" + ip + ":" + port + "] " + exception.getMessage()));
}
}
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigRequest.java b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigRequest.java
index 92228a7d26..83864e30dc 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigRequest.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigRequest.java
@@ -35,6 +35,7 @@ import org.apache.iotdb.confignode.consensus.request.write.CreateSchemaPartition
import org.apache.iotdb.confignode.consensus.request.write.DeleteProcedureReq;
import org.apache.iotdb.confignode.consensus.request.write.DeleteRegionsReq;
import org.apache.iotdb.confignode.consensus.request.write.DeleteStorageGroupReq;
+import org.apache.iotdb.confignode.consensus.request.write.DropFunctionReq;
import org.apache.iotdb.confignode.consensus.request.write.PreDeleteStorageGroupReq;
import org.apache.iotdb.confignode.consensus.request.write.RegisterDataNodeReq;
import org.apache.iotdb.confignode.consensus.request.write.SetDataReplicationFactorReq;
@@ -191,6 +192,9 @@ public abstract class ConfigRequest implements IConsensusRequest {
case CreateFunction:
req = new CreateFunctionReq();
break;
+ case DropFunction:
+ req = new DropFunctionReq();
+ break;
case GetNodePathsPartition:
req = new GetNodePathsPartitionReq();
break;
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigRequestType.java b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigRequestType.java
index f0cb20ea8c..a8f122e412 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigRequestType.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigRequestType.java
@@ -60,5 +60,6 @@ public enum ConfigRequestType {
ListRoleUsers,
ApplyConfigNode,
CreateFunction,
+ DropFunction,
GetNodePathsPartition;
}
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/DropFunctionReq.java b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/DropFunctionReq.java
new file mode 100644
index 0000000000..028660dbe4
--- /dev/null
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/DropFunctionReq.java
@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.confignode.consensus.request.write;
+
+import org.apache.iotdb.confignode.consensus.request.ConfigRequest;
+import org.apache.iotdb.confignode.consensus.request.ConfigRequestType;
+import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+public class DropFunctionReq extends ConfigRequest {
+
+ private String functionName;
+
+ public DropFunctionReq() {
+ super(ConfigRequestType.DropFunction);
+ }
+
+ public DropFunctionReq(String functionName) {
+ super(ConfigRequestType.DropFunction);
+ this.functionName = functionName;
+ }
+
+ public String getFunctionName() {
+ return functionName;
+ }
+
+ @Override
+ protected void serializeImpl(ByteBuffer buffer) {
+ buffer.putInt(getType().ordinal());
+ ReadWriteIOUtils.write(functionName, buffer);
+ }
+
+ @Override
+ protected void deserializeImpl(ByteBuffer buffer) throws IOException {
+ functionName = ReadWriteIOUtils.readString(buffer);
+ }
+}
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java b/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java
index 89ce0e1562..11ec6e2abd 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java
@@ -612,6 +612,14 @@ public class ConfigManager implements Manager {
: status;
}
+ @Override
+ public TSStatus dropFunction(String udfName) {
+ TSStatus status = confirmLeader();
+ return status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()
+ ? udfManager.dropFunction(udfName)
+ : status;
+ }
+
@Override
public UDFManager getUDFManager() {
return udfManager;
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/manager/Manager.java b/confignode/src/main/java/org/apache/iotdb/confignode/manager/Manager.java
index b689f52258..ef930f3238 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/manager/Manager.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/manager/Manager.java
@@ -220,4 +220,6 @@ public interface Manager {
TSStatus applyConfigNode(ApplyConfigNodeReq applyConfigNodeReq);
TSStatus createFunction(String udfName, String className, List<String> uris);
+
+ TSStatus dropFunction(String udfName);
}
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/manager/UDFManager.java b/confignode/src/main/java/org/apache/iotdb/confignode/manager/UDFManager.java
index a3f7b03034..f1e861cad2 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/manager/UDFManager.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/manager/UDFManager.java
@@ -23,10 +23,12 @@ import org.apache.iotdb.common.rpc.thrift.TDataNodeInfo;
import org.apache.iotdb.common.rpc.thrift.TEndPoint;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.confignode.client.AsyncDataNodeClientPool;
-import org.apache.iotdb.confignode.client.handlers.CreateFunctionHandler;
+import org.apache.iotdb.confignode.client.handlers.FunctionManagementHandler;
import org.apache.iotdb.confignode.consensus.request.write.CreateFunctionReq;
+import org.apache.iotdb.confignode.consensus.request.write.DropFunctionReq;
import org.apache.iotdb.confignode.persistence.UDFInfo;
import org.apache.iotdb.mpp.rpc.thrift.TCreateFunctionRequest;
+import org.apache.iotdb.mpp.rpc.thrift.TDropFunctionRequest;
import org.apache.iotdb.rpc.TSStatusCode;
import org.slf4j.Logger;
@@ -63,14 +65,13 @@ public class UDFManager {
return configNodeStatus;
}
- return squashDataNodeResponseStatusList(
- createFunctionOnDataNodes(functionName, className, uris));
+ return squashResponseStatusList(createFunctionOnDataNodes(functionName, className, uris));
} catch (Exception e) {
final String errorMessage =
String.format(
"Failed to register UDF %s(class name: %s, uris: %s), because of exception: %s",
functionName, className, uris, e);
- LOGGER.warn(errorMessage);
+ LOGGER.warn(errorMessage, e);
return new TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode())
.setMessage(errorMessage);
}
@@ -92,7 +93,7 @@ public class UDFManager {
.createFunction(
endPoint,
request,
- new CreateFunctionHandler(
+ new FunctionManagementHandler(
countDownLatch, dataNodeResponseStatus, endPoint.getIp(), endPoint.getPort()));
}
@@ -106,9 +107,53 @@ public class UDFManager {
return dataNodeResponseStatus;
}
- private TSStatus squashDataNodeResponseStatusList(List<TSStatus> dataNodeResponseStatusList) {
+ public TSStatus dropFunction(String functionName) {
+ try {
+ final List<TSStatus> nodeResponseList = dropFunctionOnDataNodes(functionName);
+ final TSStatus configNodeStatus =
+ configManager.getConsensusManager().write(new DropFunctionReq(functionName)).getStatus();
+ nodeResponseList.add(configNodeStatus);
+ return squashResponseStatusList(nodeResponseList);
+ } catch (Exception e) {
+ final String errorMessage =
+ String.format("Failed to deregister UDF %s, because of exception: %s", functionName, e);
+ LOGGER.warn(errorMessage, e);
+ return new TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode())
+ .setMessage(errorMessage);
+ }
+ }
+
+ private List<TSStatus> dropFunctionOnDataNodes(String functionName) {
+ final List<TDataNodeInfo> onlineDataNodes =
+ configManager.getNodeManager().getOnlineDataNodes(-1);
+ final List<TSStatus> dataNodeResponseStatus =
+ Collections.synchronizedList(new ArrayList<>(onlineDataNodes.size()));
+ final CountDownLatch countDownLatch = new CountDownLatch(onlineDataNodes.size());
+ final TDropFunctionRequest request = new TDropFunctionRequest(functionName);
+
+ for (TDataNodeInfo dataNodeInfo : onlineDataNodes) {
+ final TEndPoint endPoint = dataNodeInfo.getLocation().getInternalEndPoint();
+ AsyncDataNodeClientPool.getInstance()
+ .dropFunction(
+ endPoint,
+ request,
+ new FunctionManagementHandler(
+ countDownLatch, dataNodeResponseStatus, endPoint.getIp(), endPoint.getPort()));
+ }
+
+ try {
+ countDownLatch.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ LOGGER.error("UDFManager was interrupted during dropping functions on data nodes", e);
+ }
+
+ return dataNodeResponseStatus;
+ }
+
+ private TSStatus squashResponseStatusList(List<TSStatus> responseStatusList) {
final List<TSStatus> failedStatus =
- dataNodeResponseStatusList.stream()
+ responseStatusList.stream()
.filter(status -> status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode())
.collect(Collectors.toList());
return failedStatus.isEmpty()
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/persistence/UDFInfo.java b/confignode/src/main/java/org/apache/iotdb/confignode/persistence/UDFInfo.java
index f0aa594d44..cc4188bf0c 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/persistence/UDFInfo.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/persistence/UDFInfo.java
@@ -28,6 +28,7 @@ import org.apache.iotdb.commons.udf.service.UDFRegistrationService;
import org.apache.iotdb.confignode.conf.ConfigNodeConf;
import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor;
import org.apache.iotdb.confignode.consensus.request.write.CreateFunctionReq;
+import org.apache.iotdb.confignode.consensus.request.write.DropFunctionReq;
import org.apache.iotdb.rpc.TSStatusCode;
import org.slf4j.Logger;
@@ -48,8 +49,11 @@ public class UDFInfo implements SnapshotProcessor {
private final UDFRegistrationService udfRegistrationService;
public UDFInfo() {
- udfExecutableManager = UDFExecutableManager.getInstance();
- udfRegistrationService = UDFRegistrationService.getInstance();
+ udfExecutableManager =
+ UDFExecutableManager.setupAndGetInstance(
+ CONFIG_NODE_CONF.getTemporaryLibDir(), CONFIG_NODE_CONF.getUdfLibDir());
+ udfRegistrationService =
+ UDFRegistrationService.setupAndGetInstance(CONFIG_NODE_CONF.getSystemUdfDir());
}
public synchronized void validateBeforeRegistration(
@@ -95,7 +99,7 @@ public class UDFInfo implements SnapshotProcessor {
} catch (Exception e) {
final String errorMessage =
String.format(
- "Failed to register UDF %s(class name: %s, uris: %s), because of exception: %s",
+ "[ConfigNode] Failed to register UDF %s(class name: %s, uris: %s), because of exception: %s",
functionName, className, uris, e);
LOGGER.warn(errorMessage, e);
return new TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode())
@@ -103,6 +107,21 @@ public class UDFInfo implements SnapshotProcessor {
}
}
+ public synchronized TSStatus dropFunction(DropFunctionReq req) {
+ try {
+ udfRegistrationService.deregister(req.getFunctionName());
+ return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
+ } catch (Exception e) {
+ final String errorMessage =
+ String.format(
+ "[ConfigNode] Failed to deregister UDF %s, because of exception: %s",
+ req.getFunctionName(), e);
+ LOGGER.warn(errorMessage, e);
+ return new TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode())
+ .setMessage(errorMessage);
+ }
+ }
+
@Override
public synchronized boolean processTakeSnapshot(File snapshotDir) throws IOException {
return udfExecutableManager.processTakeSnapshot(snapshotDir)
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/persistence/executor/ConfigRequestExecutor.java b/confignode/src/main/java/org/apache/iotdb/confignode/persistence/executor/ConfigRequestExecutor.java
index 5db1567af4..311ac7e575 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/persistence/executor/ConfigRequestExecutor.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/persistence/executor/ConfigRequestExecutor.java
@@ -38,6 +38,7 @@ import org.apache.iotdb.confignode.consensus.request.write.CreateSchemaPartition
import org.apache.iotdb.confignode.consensus.request.write.DeleteProcedureReq;
import org.apache.iotdb.confignode.consensus.request.write.DeleteRegionsReq;
import org.apache.iotdb.confignode.consensus.request.write.DeleteStorageGroupReq;
+import org.apache.iotdb.confignode.consensus.request.write.DropFunctionReq;
import org.apache.iotdb.confignode.consensus.request.write.PreDeleteStorageGroupReq;
import org.apache.iotdb.confignode.consensus.request.write.RegisterDataNodeReq;
import org.apache.iotdb.confignode.consensus.request.write.SetDataReplicationFactorReq;
@@ -187,6 +188,8 @@ public class ConfigRequestExecutor {
return nodeInfo.updateConfigNodeList((ApplyConfigNodeReq) req);
case CreateFunction:
return udfInfo.createFunction((CreateFunctionReq) req);
+ case DropFunction:
+ return udfInfo.dropFunction((DropFunctionReq) req);
default:
throw new UnknownPhysicalPlanTypeException(req.getType());
}
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java b/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java
index 1d46688414..045a8e3b1f 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java
@@ -64,6 +64,7 @@ import org.apache.iotdb.confignode.rpc.thrift.TDataPartitionReq;
import org.apache.iotdb.confignode.rpc.thrift.TDataPartitionResp;
import org.apache.iotdb.confignode.rpc.thrift.TDeleteStorageGroupReq;
import org.apache.iotdb.confignode.rpc.thrift.TDeleteStorageGroupsReq;
+import org.apache.iotdb.confignode.rpc.thrift.TDropFunctionReq;
import org.apache.iotdb.confignode.rpc.thrift.TLoginReq;
import org.apache.iotdb.confignode.rpc.thrift.TPermissionInfoResp;
import org.apache.iotdb.confignode.rpc.thrift.TSchemaNodeManagementReq;
@@ -388,6 +389,11 @@ public class ConfigNodeRPCServiceProcessor implements ConfigIService.Iface {
return configManager.createFunction(req.getUdfName(), req.getClassName(), req.getUris());
}
+ @Override
+ public TSStatus dropFunction(TDropFunctionReq req) throws TException {
+ return configManager.dropFunction(req.getUdfName());
+ }
+
public void handleClientExit() {}
// TODO: Interfaces for data operations
diff --git a/node-commons/src/main/java/org/apache/iotdb/commons/udf/service/UDFRegistrationService.java b/node-commons/src/main/java/org/apache/iotdb/commons/udf/service/UDFRegistrationService.java
index 46a37eb356..52e618d4aa 100644
--- a/node-commons/src/main/java/org/apache/iotdb/commons/udf/service/UDFRegistrationService.java
+++ b/node-commons/src/main/java/org/apache/iotdb/commons/udf/service/UDFRegistrationService.java
@@ -184,7 +184,7 @@ public class UDFRegistrationService implements IService, SnapshotProcessor {
String.format(
"Failed to register UDF %s(%s) because failed to fetch UDF executables(%s)",
functionName, className, uris);
- LOGGER.warn(errorMessage);
+ LOGGER.warn(errorMessage, outerException);
throw new UDFRegistrationException(errorMessage, outerException);
}
}
@@ -212,7 +212,7 @@ public class UDFRegistrationService implements IService, SnapshotProcessor {
String.format(
"Failed to register UDF %s(%s), because its instance can not be constructed successfully. Exception: %s",
functionName, className, e);
- LOGGER.warn(errorMessage);
+ LOGGER.warn(errorMessage, e);
throw new UDFRegistrationException(errorMessage);
} finally {
releaseRegistrationLock();
@@ -260,7 +260,7 @@ public class UDFRegistrationService implements IService, SnapshotProcessor {
if (information.isBuiltin()) {
String errorMessage =
String.format("Built-in function %s can not be deregistered.", functionName);
- LOGGER.error(errorMessage);
+ LOGGER.warn(errorMessage);
throw new UDFRegistrationException(errorMessage);
}
@@ -322,7 +322,7 @@ public class UDFRegistrationService implements IService, SnapshotProcessor {
String.format(
"Failed to reflect UDF %s(%s) instance, because %s",
functionName, information.getClassName(), e);
- LOGGER.warn(errorMessage);
+ LOGGER.warn(errorMessage, e);
throw new RuntimeException(errorMessage);
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/client/ConfigNodeClient.java b/server/src/main/java/org/apache/iotdb/db/client/ConfigNodeClient.java
index 2e0ce66ead..563897078d 100644
--- a/server/src/main/java/org/apache/iotdb/db/client/ConfigNodeClient.java
+++ b/server/src/main/java/org/apache/iotdb/db/client/ConfigNodeClient.java
@@ -44,6 +44,7 @@ import org.apache.iotdb.confignode.rpc.thrift.TDataPartitionReq;
import org.apache.iotdb.confignode.rpc.thrift.TDataPartitionResp;
import org.apache.iotdb.confignode.rpc.thrift.TDeleteStorageGroupReq;
import org.apache.iotdb.confignode.rpc.thrift.TDeleteStorageGroupsReq;
+import org.apache.iotdb.confignode.rpc.thrift.TDropFunctionReq;
import org.apache.iotdb.confignode.rpc.thrift.TLoginReq;
import org.apache.iotdb.confignode.rpc.thrift.TPermissionInfoResp;
import org.apache.iotdb.confignode.rpc.thrift.TSchemaNodeManagementReq;
@@ -605,6 +606,22 @@ public class ConfigNodeClient implements ConfigIService.Iface, SyncThriftClient,
throw new TException(MSG_RECONNECTION_FAIL);
}
+ @Override
+ public TSStatus dropFunction(TDropFunctionReq req) throws TException {
+ for (int i = 0; i < RETRY_NUM; i++) {
+ try {
+ TSStatus status = client.dropFunction(req);
+ if (!updateConfigNodeLeader(status)) {
+ return status;
+ }
+ } catch (TException e) {
+ configLeader = null;
+ }
+ reconnect();
+ }
+ throw new TException(MSG_RECONNECTION_FAIL);
+ }
+
public static class Factory extends BaseClientFactory<PartitionRegionId, ConfigNodeClient> {
public Factory(
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ConfigTaskVisitor.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ConfigTaskVisitor.java
index ebd9327e71..7e6038049b 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ConfigTaskVisitor.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/ConfigTaskVisitor.java
@@ -25,6 +25,7 @@ import org.apache.iotdb.db.mpp.plan.statement.StatementVisitor;
import org.apache.iotdb.db.mpp.plan.statement.metadata.CountStorageGroupStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.CreateFunctionStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.DeleteStorageGroupStatement;
+import org.apache.iotdb.db.mpp.plan.statement.metadata.DropFunctionStatement;
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;
@@ -96,5 +97,11 @@ public class ConfigTaskVisitor
return new CreateFunctionTask(createFunctionStatement);
}
+ @Override
+ public IConfigTask visitDropFunction(
+ DropFunctionStatement dropFunctionStatement, TaskContext context) {
+ return new DropFunctionTask(dropFunctionStatement);
+ }
+
public static class TaskContext {}
}
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
new file mode 100644
index 0000000000..6f67983284
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/DropFunctionTask.java
@@ -0,0 +1,99 @@
+/*
+ * 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;
+
+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.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) {
+ udfName = dropFunctionStatement.getUdfName();
+ }
+
+ @Override
+ public ListenableFuture<ConfigTaskResult> execute(
+ IClientManager<PartitionRegionId, ConfigNodeClient> clientManager)
+ 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)));
+ }
+ }
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/parser/ASTVisitor.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/parser/ASTVisitor.java
index 463ff16649..7227a43ed7 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/parser/ASTVisitor.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/parser/ASTVisitor.java
@@ -82,6 +82,7 @@ import org.apache.iotdb.db.mpp.plan.statement.metadata.CreateFunctionStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.CreateTimeSeriesStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.DeleteStorageGroupStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.DeleteTimeSeriesStatement;
+import org.apache.iotdb.db.mpp.plan.statement.metadata.DropFunctionStatement;
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.ShowChildNodesStatement;
@@ -101,6 +102,7 @@ import org.apache.iotdb.db.qp.sql.IoTDBSqlParser.CountNodesContext;
import org.apache.iotdb.db.qp.sql.IoTDBSqlParser.CountStorageGroupContext;
import org.apache.iotdb.db.qp.sql.IoTDBSqlParser.CountTimeseriesContext;
import org.apache.iotdb.db.qp.sql.IoTDBSqlParser.CreateFunctionContext;
+import org.apache.iotdb.db.qp.sql.IoTDBSqlParser.DropFunctionContext;
import org.apache.iotdb.db.qp.sql.IoTDBSqlParser.ExpressionContext;
import org.apache.iotdb.db.qp.sql.IoTDBSqlParser.UriContext;
import org.apache.iotdb.db.qp.sql.IoTDBSqlParserBaseVisitor;
@@ -624,6 +626,12 @@ public class ASTVisitor extends IoTDBSqlParserBaseVisitor<Statement> {
return uris;
}
+ // Drop Function
+ @Override
+ public Statement visitDropFunction(DropFunctionContext ctx) {
+ return new DropFunctionStatement(parseIdentifier(ctx.udfName.getText()));
+ }
+
// Show Child Paths =====================================================================
@Override
public Statement visitShowChildPaths(IoTDBSqlParser.ShowChildPathsContext ctx) {
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/StatementVisitor.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/StatementVisitor.java
index 80a1111a02..5cb8ba2db0 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/StatementVisitor.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/StatementVisitor.java
@@ -41,6 +41,7 @@ import org.apache.iotdb.db.mpp.plan.statement.metadata.CreateTimeSeriesByDeviceS
import org.apache.iotdb.db.mpp.plan.statement.metadata.CreateTimeSeriesStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.DeleteStorageGroupStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.DeleteTimeSeriesStatement;
+import org.apache.iotdb.db.mpp.plan.statement.metadata.DropFunctionStatement;
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.ShowChildNodesStatement;
@@ -134,6 +135,10 @@ public abstract class StatementVisitor<R, C> {
return visitStatement(createFunctionStatement, context);
}
+ public R visitDropFunction(DropFunctionStatement dropFunctionStatement, C context) {
+ return visitStatement(dropFunctionStatement, context);
+ }
+
/** Data Manipulation Language (DML) */
// Select Statement
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/metadata/DropFunctionStatement.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/metadata/DropFunctionStatement.java
new file mode 100644
index 0000000000..e5850d4a81
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/metadata/DropFunctionStatement.java
@@ -0,0 +1,60 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.mpp.plan.statement.metadata;
+
+import org.apache.iotdb.commons.path.PartialPath;
+import org.apache.iotdb.db.mpp.plan.analyze.QueryType;
+import org.apache.iotdb.db.mpp.plan.constant.StatementType;
+import org.apache.iotdb.db.mpp.plan.statement.IConfigStatement;
+import org.apache.iotdb.db.mpp.plan.statement.Statement;
+import org.apache.iotdb.db.mpp.plan.statement.StatementVisitor;
+
+import java.util.Collections;
+import java.util.List;
+
+public class DropFunctionStatement extends Statement implements IConfigStatement {
+
+ private final String udfName;
+
+ public DropFunctionStatement(String udfName) {
+ super();
+ statementType = StatementType.DROP_FUNCTION;
+ this.udfName = udfName;
+ }
+
+ public String getUdfName() {
+ return udfName;
+ }
+
+ @Override
+ public <R, C> R accept(StatementVisitor<R, C> visitor, C context) {
+ return visitor.visitDropFunction(this, context);
+ }
+
+ @Override
+ public QueryType getQueryType() {
+ return QueryType.WRITE;
+ }
+
+ @Override
+ public List<? extends PartialPath> getPaths() {
+ return Collections.emptyList();
+ }
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/service/thrift/impl/InternalServiceImpl.java b/server/src/main/java/org/apache/iotdb/db/service/thrift/impl/InternalServiceImpl.java
index eff3a3ffa0..ab75ed7da9 100644
--- a/server/src/main/java/org/apache/iotdb/db/service/thrift/impl/InternalServiceImpl.java
+++ b/server/src/main/java/org/apache/iotdb/db/service/thrift/impl/InternalServiceImpl.java
@@ -76,6 +76,7 @@ import org.apache.iotdb.mpp.rpc.thrift.TCancelResp;
import org.apache.iotdb.mpp.rpc.thrift.TCreateDataRegionReq;
import org.apache.iotdb.mpp.rpc.thrift.TCreateFunctionRequest;
import org.apache.iotdb.mpp.rpc.thrift.TCreateSchemaRegionReq;
+import org.apache.iotdb.mpp.rpc.thrift.TDropFunctionRequest;
import org.apache.iotdb.mpp.rpc.thrift.TFetchFragmentInstanceStateReq;
import org.apache.iotdb.mpp.rpc.thrift.TFragmentInstanceStateResp;
import org.apache.iotdb.mpp.rpc.thrift.TInvalidateCacheReq;
@@ -398,5 +399,16 @@ public class InternalServiceImpl implements InternalService.Iface {
}
}
+ @Override
+ public TSStatus dropFunction(TDropFunctionRequest request) {
+ try {
+ UDFRegistrationService.getInstance().deregister(request.getUdfName());
+ return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
+ } catch (Exception e) {
+ return new TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode())
+ .setMessage(e.getMessage());
+ }
+ }
+
public void handleClientExit() {}
}
diff --git a/thrift-confignode/src/main/thrift/confignode.thrift b/thrift-confignode/src/main/thrift/confignode.thrift
index ac3f749041..3bba5e3bc5 100644
--- a/thrift-confignode/src/main/thrift/confignode.thrift
+++ b/thrift-confignode/src/main/thrift/confignode.thrift
@@ -215,6 +215,10 @@ struct TCreateFunctionReq {
3: required list<string> uris
}
+struct TDropFunctionReq {
+ 1: required string udfName
+}
+
service ConfigIService {
/* DataNode */
@@ -278,4 +282,6 @@ service ConfigIService {
/* UDF */
common.TSStatus createFunction(TCreateFunctionReq req)
-}
\ No newline at end of file
+
+ common.TSStatus dropFunction(TDropFunctionReq req)
+}
diff --git a/thrift/src/main/thrift/mpp.thrift b/thrift/src/main/thrift/mpp.thrift
index eaecf45dc7..702050e540 100644
--- a/thrift/src/main/thrift/mpp.thrift
+++ b/thrift/src/main/thrift/mpp.thrift
@@ -141,6 +141,10 @@ struct TCreateFunctionRequest {
3: required list<string> uris
}
+struct TDropFunctionRequest {
+ 1: required string udfName
+}
+
struct TInvalidatePermissionCacheReq {
1: required string username
2: required string roleName
@@ -229,6 +233,13 @@ service InternalService {
**/
common.TSStatus createFunction(TCreateFunctionRequest req)
+ /**
+ * Config node will drop a function on a list of data nodes.
+ *
+ * @param function name
+ **/
+ common.TSStatus dropFunction(TDropFunctionRequest req)
+
/**
* Config node will invalidate permission Info cache.
*