You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by hu...@apache.org on 2023/02/17 07:00:40 UTC
[iotdb] 04/16: finish ModelManager
This is an automated email from the ASF dual-hosted git repository.
hui pushed a commit to branch lmh/modelManager
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 143fb28dfc0b110533ac3c385b8a66858a6fe67c
Author: Minghui Liu <li...@foxmail.com>
AuthorDate: Thu Jan 5 20:06:58 2023 +0800
finish ModelManager
---
.../request/read/model/ShowModelPlan.java | 69 ++++++++++---------
.../request/read/model/ShowTrailPlan.java | 68 +++++++++----------
.../request/write/model/DropModelPlan.java | 73 ++++++++++----------
.../consensus/response/ModelTableResp.java | 31 +++++++++
.../consensus/response/TrailTableResp.java | 31 +++++++++
.../iotdb/confignode/manager/ModelManager.java | 78 +++++++++++++++++++---
.../iotdb/confignode/manager/ProcedureManager.java | 9 +++
7 files changed, 246 insertions(+), 113 deletions(-)
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/read/model/ShowModelPlan.java b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/read/model/ShowModelPlan.java
index ac1d9a2e61..d5082379fd 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/read/model/ShowModelPlan.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/read/model/ShowModelPlan.java
@@ -19,7 +19,6 @@
package org.apache.iotdb.confignode.consensus.request.read.model;
-import org.apache.commons.lang3.Validate;
import org.apache.iotdb.commons.utils.ThriftCommonsSerDeUtils;
import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlan;
import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlanType;
@@ -32,45 +31,45 @@ import java.util.Objects;
public class ShowModelPlan extends ConfigPhysicalPlan {
- private TShowModelReq showModelReq;
+ private TShowModelReq showModelReq;
- public ShowModelPlan() {
- super(ConfigPhysicalPlanType.ShowModel);
- }
+ public ShowModelPlan() {
+ super(ConfigPhysicalPlanType.ShowModel);
+ }
- public ShowModelPlan(TShowModelReq showModelReq) {
- super(ConfigPhysicalPlanType.ShowModel);
- this.showModelReq = showModelReq;
- }
+ public ShowModelPlan(TShowModelReq showModelReq) {
+ super(ConfigPhysicalPlanType.ShowModel);
+ this.showModelReq = showModelReq;
+ }
- @Override
- protected void serializeImpl(DataOutputStream stream) throws IOException {
- stream.writeShort(getType().getPlanType());
- ThriftCommonsSerDeUtils.serializeTShowModelReq(showModelReq, stream);
- }
+ @Override
+ protected void serializeImpl(DataOutputStream stream) throws IOException {
+ stream.writeShort(getType().getPlanType());
+ ThriftCommonsSerDeUtils.serializeTShowModelReq(showModelReq, stream);
+ }
- @Override
- protected void deserializeImpl(ByteBuffer buffer) throws IOException {
- showModelReq = ThriftCommonsSerDeUtils.deserializeTShowModelReq(buffer);
- }
+ @Override
+ protected void deserializeImpl(ByteBuffer buffer) throws IOException {
+ showModelReq = ThriftCommonsSerDeUtils.deserializeTShowModelReq(buffer);
+ }
- @Override
- public boolean equals(Object o) {
- if (this == o) {
- return true;
- }
- if (o == null || getClass() != o.getClass()) {
- return false;
- }
- if (!super.equals(o)) {
- return false;
- }
- ShowModelPlan that = (ShowModelPlan) o;
- return Objects.equals(showModelReq, that.showModelReq);
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
}
-
- @Override
- public int hashCode() {
- return Objects.hash(super.hashCode(), showModelReq);
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ if (!super.equals(o)) {
+ return false;
}
+ ShowModelPlan that = (ShowModelPlan) o;
+ return Objects.equals(showModelReq, that.showModelReq);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(super.hashCode(), showModelReq);
+ }
}
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/read/model/ShowTrailPlan.java b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/read/model/ShowTrailPlan.java
index 448c13ab2c..aa9fb83bf2 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/read/model/ShowTrailPlan.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/read/model/ShowTrailPlan.java
@@ -31,45 +31,45 @@ import java.util.Objects;
public class ShowTrailPlan extends ConfigPhysicalPlan {
- private TShowTrailReq showTrailReq;
+ private TShowTrailReq showTrailReq;
- public ShowTrailPlan() {
- super(ConfigPhysicalPlanType.ShowTrail);
- }
+ public ShowTrailPlan() {
+ super(ConfigPhysicalPlanType.ShowTrail);
+ }
- public ShowTrailPlan(TShowTrailReq showTrailReq) {
- super(ConfigPhysicalPlanType.ShowTrail);
- this.showTrailReq = showTrailReq;
- }
+ public ShowTrailPlan(TShowTrailReq showTrailReq) {
+ super(ConfigPhysicalPlanType.ShowTrail);
+ this.showTrailReq = showTrailReq;
+ }
- @Override
- protected void serializeImpl(DataOutputStream stream) throws IOException {
- stream.writeShort(getType().getPlanType());
- ThriftCommonsSerDeUtils.serializeTShowTrailReq(showTrailReq, stream);
- }
+ @Override
+ protected void serializeImpl(DataOutputStream stream) throws IOException {
+ stream.writeShort(getType().getPlanType());
+ ThriftCommonsSerDeUtils.serializeTShowTrailReq(showTrailReq, stream);
+ }
- @Override
- protected void deserializeImpl(ByteBuffer buffer) throws IOException {
- showTrailReq = ThriftCommonsSerDeUtils.deserializeTShowTrailReq(buffer);
- }
+ @Override
+ protected void deserializeImpl(ByteBuffer buffer) throws IOException {
+ showTrailReq = ThriftCommonsSerDeUtils.deserializeTShowTrailReq(buffer);
+ }
- @Override
- public boolean equals(Object o) {
- if (this == o) {
- return true;
- }
- if (o == null || getClass() != o.getClass()) {
- return false;
- }
- if (!super.equals(o)) {
- return false;
- }
- ShowTrailPlan that = (ShowTrailPlan) o;
- return Objects.equals(showTrailReq, that.showTrailReq);
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
}
-
- @Override
- public int hashCode() {
- return Objects.hash(super.hashCode(), showTrailReq);
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ if (!super.equals(o)) {
+ return false;
}
+ ShowTrailPlan that = (ShowTrailPlan) o;
+ return Objects.equals(showTrailReq, that.showTrailReq);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(super.hashCode(), showTrailReq);
+ }
}
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/model/DropModelPlan.java b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/model/DropModelPlan.java
index 3cbec2d355..f2def81121 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/model/DropModelPlan.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/model/DropModelPlan.java
@@ -19,11 +19,12 @@
package org.apache.iotdb.confignode.consensus.request.write.model;
-import org.apache.commons.lang3.Validate;
import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlan;
import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlanType;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+import org.apache.commons.lang3.Validate;
+
import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
@@ -31,46 +32,46 @@ import java.util.Objects;
public class DropModelPlan extends ConfigPhysicalPlan {
- private String modelId;
+ private String modelId;
- public DropModelPlan() {
- super(ConfigPhysicalPlanType.DropModel);
- }
+ public DropModelPlan() {
+ super(ConfigPhysicalPlanType.DropModel);
+ }
- public DropModelPlan(String modelId) {
- super(ConfigPhysicalPlanType.DropModel);
- Validate.notNull(modelId);
- this.modelId = modelId;
- }
+ public DropModelPlan(String modelId) {
+ super(ConfigPhysicalPlanType.DropModel);
+ Validate.notNull(modelId);
+ this.modelId = modelId;
+ }
- @Override
- protected void serializeImpl(DataOutputStream stream) throws IOException {
- stream.writeShort(getType().getPlanType());
- ReadWriteIOUtils.write(modelId, stream);
- }
+ @Override
+ protected void serializeImpl(DataOutputStream stream) throws IOException {
+ stream.writeShort(getType().getPlanType());
+ ReadWriteIOUtils.write(modelId, stream);
+ }
- @Override
- protected void deserializeImpl(ByteBuffer buffer) throws IOException {
- modelId = ReadWriteIOUtils.readString(buffer);
- }
+ @Override
+ protected void deserializeImpl(ByteBuffer buffer) throws IOException {
+ modelId = ReadWriteIOUtils.readString(buffer);
+ }
- @Override
- public boolean equals(Object o) {
- if (this == o) {
- return true;
- }
- if (o == null || getClass() != o.getClass()) {
- return false;
- }
- if (!super.equals(o)) {
- return false;
- }
- DropModelPlan that = (DropModelPlan) o;
- return modelId.equals(that.modelId);
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
}
-
- @Override
- public int hashCode() {
- return Objects.hash(super.hashCode(), modelId);
+ if (o == null || getClass() != o.getClass()) {
+ return false;
}
+ if (!super.equals(o)) {
+ return false;
+ }
+ DropModelPlan that = (DropModelPlan) o;
+ return modelId.equals(that.modelId);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(super.hashCode(), modelId);
+ }
}
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/response/ModelTableResp.java b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/response/ModelTableResp.java
new file mode 100644
index 0000000000..05afc195bc
--- /dev/null
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/response/ModelTableResp.java
@@ -0,0 +1,31 @@
+/*
+ * 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.response;
+
+import org.apache.iotdb.confignode.rpc.thrift.TShowModelResp;
+import org.apache.iotdb.consensus.common.DataSet;
+
+import java.io.IOException;
+
+public class ModelTableResp implements DataSet {
+ public TShowModelResp convertToThriftResponse() throws IOException {
+ return null;
+ }
+}
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/response/TrailTableResp.java b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/response/TrailTableResp.java
new file mode 100644
index 0000000000..565297d053
--- /dev/null
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/response/TrailTableResp.java
@@ -0,0 +1,31 @@
+/*
+ * 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.response;
+
+import org.apache.iotdb.confignode.rpc.thrift.TShowTrailResp;
+import org.apache.iotdb.consensus.common.DataSet;
+
+import java.io.IOException;
+
+public class TrailTableResp implements DataSet {
+ public TShowTrailResp convertToThriftResponse() throws IOException {
+ return null;
+ }
+}
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/manager/ModelManager.java b/confignode/src/main/java/org/apache/iotdb/confignode/manager/ModelManager.java
index 81cbe8c014..347c5c0b94 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/manager/ModelManager.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/manager/ModelManager.java
@@ -20,6 +20,12 @@
package org.apache.iotdb.confignode.manager;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.model.ModelInformation;
+import org.apache.iotdb.confignode.consensus.request.read.model.ShowModelPlan;
+import org.apache.iotdb.confignode.consensus.request.read.model.ShowTrailPlan;
+import org.apache.iotdb.confignode.consensus.request.write.model.UpdateModelInfoPlan;
+import org.apache.iotdb.confignode.consensus.response.ModelTableResp;
+import org.apache.iotdb.confignode.consensus.response.TrailTableResp;
import org.apache.iotdb.confignode.persistence.ModelInfo;
import org.apache.iotdb.confignode.rpc.thrift.TCreateModelReq;
import org.apache.iotdb.confignode.rpc.thrift.TDropModelReq;
@@ -28,10 +34,16 @@ import org.apache.iotdb.confignode.rpc.thrift.TShowModelResp;
import org.apache.iotdb.confignode.rpc.thrift.TShowTrailReq;
import org.apache.iotdb.confignode.rpc.thrift.TShowTrailResp;
import org.apache.iotdb.confignode.rpc.thrift.TUpdateModelInfoReq;
+import org.apache.iotdb.consensus.common.response.ConsensusReadResponse;
+import org.apache.iotdb.consensus.common.response.ConsensusWriteResponse;
+import org.apache.iotdb.rpc.TSStatusCode;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.io.IOException;
+import java.util.Collections;
+
public class ModelManager {
private static final Logger LOGGER = LoggerFactory.getLogger(ModelManager.class);
@@ -49,22 +61,72 @@ public class ModelManager {
}
public TSStatus createModel(TCreateModelReq req) {
- return null;
+ ModelInformation modelInformation = new ModelInformation();
+ return configManager.getProcedureManager().createModel(modelInformation);
}
public TSStatus dropModel(TDropModelReq req) {
- return null;
+ return configManager.getProcedureManager().dropModel(req.getModelId());
}
- public TShowModelResp showModel(TShowModelReq req) {
- return null;
+ public TSStatus updateModelInfo(TUpdateModelInfoReq req) {
+ ConsensusWriteResponse response =
+ configManager.getConsensusManager().write(new UpdateModelInfoPlan(req));
+ if (response.getStatus() != null) {
+ return response.getStatus();
+ } else {
+ LOGGER.warn(
+ "Unexpected error happened while updating model {}: ",
+ req.getModelInfo(),
+ response.getException());
+ // consensus layer related errors
+ TSStatus res = new TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode());
+ res.setMessage(response.getErrorMessage());
+ return res;
+ }
}
- public TShowTrailResp showTrail(TShowTrailReq req) {
- return null;
+ public TShowModelResp showModel(TShowModelReq req) {
+ try {
+ ConsensusReadResponse response =
+ configManager.getConsensusManager().read(new ShowModelPlan(req));
+ if (response.getDataset() != null) {
+ return ((ModelTableResp) response.getDataset()).convertToThriftResponse();
+ } else {
+ LOGGER.warn("Unexpected error happened while showing model: ", response.getException());
+ // consensus layer related errors
+ TSStatus res = new TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode());
+ res.setMessage(response.getException().toString());
+ return new TShowModelResp(res, Collections.emptyList());
+ }
+ } catch (IOException e) {
+ LOGGER.error("Fail to get ModelTable", e);
+ return new TShowModelResp(
+ new TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode())
+ .setMessage(e.getMessage()),
+ Collections.emptyList());
+ }
}
- public TSStatus updateModelInfo(TUpdateModelInfoReq req) {
- return null;
+ public TShowTrailResp showTrail(TShowTrailReq req) {
+ try {
+ ConsensusReadResponse response =
+ configManager.getConsensusManager().read(new ShowTrailPlan(req));
+ if (response.getDataset() != null) {
+ return ((TrailTableResp) response.getDataset()).convertToThriftResponse();
+ } else {
+ LOGGER.warn("Unexpected error happened while showing trail: ", response.getException());
+ // consensus layer related errors
+ TSStatus res = new TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode());
+ res.setMessage(response.getException().toString());
+ return new TShowTrailResp(res, Collections.emptyList());
+ }
+ } catch (IOException e) {
+ LOGGER.error("Fail to get TrailTable", e);
+ return new TShowTrailResp(
+ new TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode())
+ .setMessage(e.getMessage()),
+ Collections.emptyList());
+ }
}
}
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/manager/ProcedureManager.java b/confignode/src/main/java/org/apache/iotdb/confignode/manager/ProcedureManager.java
index 17a9ea7fec..f8d88cc629 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/manager/ProcedureManager.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/manager/ProcedureManager.java
@@ -28,6 +28,7 @@ import org.apache.iotdb.commons.cluster.NodeStatus;
import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.commons.exception.IoTDBException;
import org.apache.iotdb.commons.exception.sync.PipeException;
+import org.apache.iotdb.commons.model.ModelInformation;
import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.commons.path.PathPatternTree;
import org.apache.iotdb.commons.trigger.TriggerInformation;
@@ -532,6 +533,14 @@ public class ProcedureManager {
return statusList.get(0);
}
+ public TSStatus createModel(ModelInformation modelInformation) {
+ return RpcUtils.SUCCESS_STATUS;
+ }
+
+ public TSStatus dropModel(String modelId) {
+ return RpcUtils.SUCCESS_STATUS;
+ }
+
public TSStatus createPipe(TCreatePipeReq req) {
try {
long procedureId = executor.submitProcedure(new CreatePipeProcedure(req));