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 2023/05/14 12:51:55 UTC
[iotdb] 02/02: fix de ser bugs
This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch pipe-management-fix
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 908a84f30280f2779620dafccf2eae71d7fe31a0
Author: Steve Yurong Su <ro...@apache.org>
AuthorDate: Sun May 14 20:51:26 2023 +0800
fix de ser bugs
---
.../runtime/PipeHandleLeaderChangeProcedure.java | 4 +-
.../pipe/task/AbstractOperatePipeProcedureV2.java | 2 +-
.../impl/pipe/task/CreatePipeProcedureV2.java | 16 ++---
.../procedure/store/ProcedureFactory.java | 10 +++
.../iotdb/commons/pipe/task/meta/PipeMeta.java | 9 ++-
.../commons/pipe/task/meta/PipeRuntimeMeta.java | 12 ++++
.../commons/pipe/task/meta/PipeStaticMeta.java | 55 +++++++++++-----
.../iotdb/commons/pipe/task/meta/PipeTaskMeta.java | 12 ++++
.../commons/pipe/task/meta/PipeMetaDeSerTest.java | 77 ++++++++++++++++++++++
.../iotdb/db/mpp/plan/parser/ASTVisitor.java | 12 +++-
.../impl/DataNodeInternalRPCServiceImpl.java | 6 +-
11 files changed, 176 insertions(+), 39 deletions(-)
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/runtime/PipeHandleLeaderChangeProcedure.java b/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/runtime/PipeHandleLeaderChangeProcedure.java
index adc787f58f9..42de84cc9e5 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/runtime/PipeHandleLeaderChangeProcedure.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/runtime/PipeHandleLeaderChangeProcedure.java
@@ -153,7 +153,7 @@ public class PipeHandleLeaderChangeProcedure extends AbstractOperatePipeProcedur
public void serialize(DataOutputStream stream) throws IOException {
stream.writeShort(ProcedureType.PIPE_HANDLE_LEADER_CHANGE_PROCEDURE.getTypeCode());
super.serialize(stream);
- stream.writeInt(dataRegionGroupToOldAndNewLeaderPairMap.size());
+ ReadWriteIOUtils.write(dataRegionGroupToOldAndNewLeaderPairMap.size(), stream);
for (Map.Entry<TConsensusGroupId, Pair<Integer, Integer>> entry :
dataRegionGroupToOldAndNewLeaderPairMap.entrySet()) {
ReadWriteIOUtils.write(entry.getKey().getId(), stream);
@@ -165,7 +165,7 @@ public class PipeHandleLeaderChangeProcedure extends AbstractOperatePipeProcedur
@Override
public void deserialize(ByteBuffer byteBuffer) {
super.deserialize(byteBuffer);
- final int size = byteBuffer.getInt();
+ final int size = ReadWriteIOUtils.readInt(byteBuffer);
for (int i = 0; i < size; ++i) {
final int dataRegionGroupId = ReadWriteIOUtils.readInt(byteBuffer);
final int oldDataRegionLeaderId = ReadWriteIOUtils.readInt(byteBuffer);
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/AbstractOperatePipeProcedureV2.java b/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/AbstractOperatePipeProcedureV2.java
index 15b36ecc8dd..402a9d43bf3 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/AbstractOperatePipeProcedureV2.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/AbstractOperatePipeProcedureV2.java
@@ -208,7 +208,7 @@ public abstract class AbstractOperatePipeProcedureV2
@Override
public void serialize(DataOutputStream stream) throws IOException {
super.serialize(stream);
- stream.writeBoolean(isRollbackFromOperateOnDataNodesSuccessful);
+ ReadWriteIOUtils.write(isRollbackFromOperateOnDataNodesSuccessful, stream);
}
@Override
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/CreatePipeProcedureV2.java b/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/CreatePipeProcedureV2.java
index f3bc82dd725..ec45d1c26b4 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/CreatePipeProcedureV2.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/CreatePipeProcedureV2.java
@@ -180,26 +180,26 @@ public class CreatePipeProcedureV2 extends AbstractOperatePipeProcedureV2 {
stream.writeShort(ProcedureType.CREATE_PIPE_PROCEDURE_V2.getTypeCode());
super.serialize(stream);
ReadWriteIOUtils.write(createPipeRequest.getPipeName(), stream);
- stream.writeInt(createPipeRequest.getCollectorAttributesSize());
+ ReadWriteIOUtils.write(createPipeRequest.getCollectorAttributesSize(), stream);
for (Map.Entry<String, String> entry : createPipeRequest.getCollectorAttributes().entrySet()) {
ReadWriteIOUtils.write(entry.getKey(), stream);
ReadWriteIOUtils.write(entry.getValue(), stream);
}
- stream.writeInt(createPipeRequest.getProcessorAttributesSize());
+ ReadWriteIOUtils.write(createPipeRequest.getProcessorAttributesSize(), stream);
for (Map.Entry<String, String> entry : createPipeRequest.getProcessorAttributes().entrySet()) {
ReadWriteIOUtils.write(entry.getKey(), stream);
ReadWriteIOUtils.write(entry.getValue(), stream);
}
- stream.writeInt(createPipeRequest.getConnectorAttributesSize());
+ ReadWriteIOUtils.write(createPipeRequest.getConnectorAttributesSize(), stream);
for (Map.Entry<String, String> entry : createPipeRequest.getConnectorAttributes().entrySet()) {
ReadWriteIOUtils.write(entry.getKey(), stream);
ReadWriteIOUtils.write(entry.getValue(), stream);
}
if (pipeStaticMeta != null) {
- stream.writeBoolean(true);
+ ReadWriteIOUtils.write(true, stream);
pipeStaticMeta.serialize(stream);
} else {
- stream.writeBoolean(false);
+ ReadWriteIOUtils.write(false, stream);
}
}
@@ -212,19 +212,19 @@ public class CreatePipeProcedureV2 extends AbstractOperatePipeProcedureV2 {
.setCollectorAttributes(new HashMap<>())
.setProcessorAttributes(new HashMap<>())
.setConnectorAttributes(new HashMap<>());
- int size = byteBuffer.getInt();
+ int size = ReadWriteIOUtils.readInt(byteBuffer);
for (int i = 0; i < size; ++i) {
createPipeRequest
.getCollectorAttributes()
.put(ReadWriteIOUtils.readString(byteBuffer), ReadWriteIOUtils.readString(byteBuffer));
}
- size = byteBuffer.getInt();
+ size = ReadWriteIOUtils.readInt(byteBuffer);
for (int i = 0; i < size; ++i) {
createPipeRequest
.getProcessorAttributes()
.put(ReadWriteIOUtils.readString(byteBuffer), ReadWriteIOUtils.readString(byteBuffer));
}
- size = byteBuffer.getInt();
+ size = ReadWriteIOUtils.readInt(byteBuffer);
for (int i = 0; i < size; ++i) {
createPipeRequest
.getConnectorAttributes()
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/procedure/store/ProcedureFactory.java b/confignode/src/main/java/org/apache/iotdb/confignode/procedure/store/ProcedureFactory.java
index 724a86f19cb..a24475fc8c9 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/procedure/store/ProcedureFactory.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/procedure/store/ProcedureFactory.java
@@ -200,6 +200,16 @@ public class ProcedureFactory implements IProcedureFactory {
return ProcedureType.CREATE_MODEL_PROCEDURE;
} else if (procedure instanceof DropModelProcedure) {
return ProcedureType.DROP_MODEL_PROCEDURE;
+ } else if (procedure instanceof CreatePipeProcedureV2) {
+ return ProcedureType.CREATE_PIPE_PROCEDURE_V2;
+ } else if (procedure instanceof StartPipeProcedureV2) {
+ return ProcedureType.START_PIPE_PROCEDURE_V2;
+ } else if (procedure instanceof StopPipeProcedureV2) {
+ return ProcedureType.STOP_PIPE_PROCEDURE_V2;
+ } else if (procedure instanceof DropPipeProcedureV2) {
+ return ProcedureType.DROP_PIPE_PROCEDURE_V2;
+ } else if (procedure instanceof PipeHandleLeaderChangeProcedure) {
+ return ProcedureType.PIPE_HANDLE_LEADER_CHANGE_PROCEDURE;
}
return null;
}
diff --git a/node-commons/src/main/java/org/apache/iotdb/commons/pipe/task/meta/PipeMeta.java b/node-commons/src/main/java/org/apache/iotdb/commons/pipe/task/meta/PipeMeta.java
index 1fa610cc35c..8b15b0ffa9c 100644
--- a/node-commons/src/main/java/org/apache/iotdb/commons/pipe/task/meta/PipeMeta.java
+++ b/node-commons/src/main/java/org/apache/iotdb/commons/pipe/task/meta/PipeMeta.java
@@ -20,7 +20,6 @@
package org.apache.iotdb.commons.pipe.task.meta;
import org.apache.iotdb.tsfile.utils.PublicBAOS;
-import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
import java.io.DataOutputStream;
import java.io.FileInputStream;
@@ -55,13 +54,13 @@ public class PipeMeta {
}
public void serialize(DataOutputStream outputStream) throws IOException {
- ReadWriteIOUtils.write(staticMeta.serialize(), outputStream);
- ReadWriteIOUtils.write(runtimeMeta.serialize(), outputStream);
+ staticMeta.serialize(outputStream);
+ runtimeMeta.serialize(outputStream);
}
public void serialize(FileOutputStream outputStream) throws IOException {
- ReadWriteIOUtils.write(staticMeta.serialize(), outputStream);
- ReadWriteIOUtils.write(runtimeMeta.serialize(), outputStream);
+ staticMeta.serialize(outputStream);
+ runtimeMeta.serialize(outputStream);
}
public static PipeMeta deserialize(FileInputStream fileInputStream) throws IOException {
diff --git a/node-commons/src/main/java/org/apache/iotdb/commons/pipe/task/meta/PipeRuntimeMeta.java b/node-commons/src/main/java/org/apache/iotdb/commons/pipe/task/meta/PipeRuntimeMeta.java
index 683177573ad..6a01db5de5f 100644
--- a/node-commons/src/main/java/org/apache/iotdb/commons/pipe/task/meta/PipeRuntimeMeta.java
+++ b/node-commons/src/main/java/org/apache/iotdb/commons/pipe/task/meta/PipeRuntimeMeta.java
@@ -25,6 +25,7 @@ import org.apache.iotdb.tsfile.utils.PublicBAOS;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
import java.io.DataOutputStream;
+import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.nio.ByteBuffer;
@@ -74,6 +75,17 @@ public class PipeRuntimeMeta {
}
}
+ public void serialize(FileOutputStream outputStream) throws IOException {
+ ReadWriteIOUtils.write(status.get().getType(), outputStream);
+
+ ReadWriteIOUtils.write(consensusGroupIdToTaskMetaMap.size(), outputStream);
+ for (Map.Entry<TConsensusGroupId, PipeTaskMeta> entry :
+ consensusGroupIdToTaskMetaMap.entrySet()) {
+ ReadWriteIOUtils.write(entry.getKey().getId(), outputStream);
+ entry.getValue().serialize(outputStream);
+ }
+ }
+
public static PipeRuntimeMeta deserialize(InputStream inputStream) throws IOException {
return deserialize(
ByteBuffer.wrap(ReadWriteIOUtils.readBytesWithSelfDescriptionLength(inputStream)));
diff --git a/node-commons/src/main/java/org/apache/iotdb/commons/pipe/task/meta/PipeStaticMeta.java b/node-commons/src/main/java/org/apache/iotdb/commons/pipe/task/meta/PipeStaticMeta.java
index f487de445cb..b701286d265 100644
--- a/node-commons/src/main/java/org/apache/iotdb/commons/pipe/task/meta/PipeStaticMeta.java
+++ b/node-commons/src/main/java/org/apache/iotdb/commons/pipe/task/meta/PipeStaticMeta.java
@@ -23,6 +23,7 @@ import org.apache.iotdb.tsfile.utils.PublicBAOS;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
import java.io.DataOutputStream;
+import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.nio.ByteBuffer;
@@ -85,17 +86,38 @@ public class PipeStaticMeta {
ReadWriteIOUtils.write(pipeName, outputStream);
ReadWriteIOUtils.write(creationTime, outputStream);
- outputStream.writeInt(collectorParameters.getAttribute().size());
+ ReadWriteIOUtils.write(collectorParameters.getAttribute().size(), outputStream);
for (Map.Entry<String, String> entry : collectorParameters.getAttribute().entrySet()) {
ReadWriteIOUtils.write(entry.getKey(), outputStream);
ReadWriteIOUtils.write(entry.getValue(), outputStream);
}
- outputStream.writeInt(processorParameters.getAttribute().size());
+ ReadWriteIOUtils.write(processorParameters.getAttribute().size(), outputStream);
for (Map.Entry<String, String> entry : processorParameters.getAttribute().entrySet()) {
ReadWriteIOUtils.write(entry.getKey(), outputStream);
ReadWriteIOUtils.write(entry.getValue(), outputStream);
}
- outputStream.writeInt(connectorParameters.getAttribute().size());
+ ReadWriteIOUtils.write(connectorParameters.getAttribute().size(), outputStream);
+ for (Map.Entry<String, String> entry : connectorParameters.getAttribute().entrySet()) {
+ ReadWriteIOUtils.write(entry.getKey(), outputStream);
+ ReadWriteIOUtils.write(entry.getValue(), outputStream);
+ }
+ }
+
+ public void serialize(FileOutputStream outputStream) throws IOException {
+ ReadWriteIOUtils.write(pipeName, outputStream);
+ ReadWriteIOUtils.write(creationTime, outputStream);
+
+ ReadWriteIOUtils.write(collectorParameters.getAttribute().size(), outputStream);
+ for (Map.Entry<String, String> entry : collectorParameters.getAttribute().entrySet()) {
+ ReadWriteIOUtils.write(entry.getKey(), outputStream);
+ ReadWriteIOUtils.write(entry.getValue(), outputStream);
+ }
+ ReadWriteIOUtils.write(processorParameters.getAttribute().size(), outputStream);
+ for (Map.Entry<String, String> entry : processorParameters.getAttribute().entrySet()) {
+ ReadWriteIOUtils.write(entry.getKey(), outputStream);
+ ReadWriteIOUtils.write(entry.getValue(), outputStream);
+ }
+ ReadWriteIOUtils.write(connectorParameters.getAttribute().size(), outputStream);
for (Map.Entry<String, String> entry : connectorParameters.getAttribute().entrySet()) {
ReadWriteIOUtils.write(entry.getKey(), outputStream);
ReadWriteIOUtils.write(entry.getValue(), outputStream);
@@ -117,26 +139,23 @@ public class PipeStaticMeta {
pipeStaticMeta.processorParameters = new PipeParameters(new HashMap<>());
pipeStaticMeta.connectorParameters = new PipeParameters(new HashMap<>());
- int size = byteBuffer.getInt();
+ int size = ReadWriteIOUtils.readInt(byteBuffer);
for (int i = 0; i < size; ++i) {
- pipeStaticMeta
- .collectorParameters
- .getAttribute()
- .put(ReadWriteIOUtils.readString(byteBuffer), ReadWriteIOUtils.readString(byteBuffer));
+ final String key = ReadWriteIOUtils.readString(byteBuffer);
+ final String value = ReadWriteIOUtils.readString(byteBuffer);
+ pipeStaticMeta.collectorParameters.getAttribute().put(key, value);
}
- size = byteBuffer.getInt();
+ size = ReadWriteIOUtils.readInt(byteBuffer);
for (int i = 0; i < size; ++i) {
- pipeStaticMeta
- .processorParameters
- .getAttribute()
- .put(ReadWriteIOUtils.readString(byteBuffer), ReadWriteIOUtils.readString(byteBuffer));
+ final String key = ReadWriteIOUtils.readString(byteBuffer);
+ final String value = ReadWriteIOUtils.readString(byteBuffer);
+ pipeStaticMeta.processorParameters.getAttribute().put(key, value);
}
- size = byteBuffer.getInt();
+ size = ReadWriteIOUtils.readInt(byteBuffer);
for (int i = 0; i < size; ++i) {
- pipeStaticMeta
- .connectorParameters
- .getAttribute()
- .put(ReadWriteIOUtils.readString(byteBuffer), ReadWriteIOUtils.readString(byteBuffer));
+ final String key = ReadWriteIOUtils.readString(byteBuffer);
+ final String value = ReadWriteIOUtils.readString(byteBuffer);
+ pipeStaticMeta.connectorParameters.getAttribute().put(key, value);
}
return pipeStaticMeta;
diff --git a/node-commons/src/main/java/org/apache/iotdb/commons/pipe/task/meta/PipeTaskMeta.java b/node-commons/src/main/java/org/apache/iotdb/commons/pipe/task/meta/PipeTaskMeta.java
index 1131047c104..3e0e4b34d93 100644
--- a/node-commons/src/main/java/org/apache/iotdb/commons/pipe/task/meta/PipeTaskMeta.java
+++ b/node-commons/src/main/java/org/apache/iotdb/commons/pipe/task/meta/PipeTaskMeta.java
@@ -25,6 +25,7 @@ import org.apache.iotdb.pipe.api.exception.PipeRuntimeNonCriticalException;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
import java.io.DataOutputStream;
+import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.nio.ByteBuffer;
@@ -93,6 +94,17 @@ public class PipeTaskMeta {
}
}
+ public void serialize(FileOutputStream outputStream) throws IOException {
+ ReadWriteIOUtils.write(progressIndex.get(), outputStream);
+ ReadWriteIOUtils.write(regionLeader.get(), outputStream);
+ ReadWriteIOUtils.write(exceptionMessages.size(), outputStream);
+ for (final PipeRuntimeException exceptionMessage : exceptionMessages) {
+ ReadWriteIOUtils.write(
+ exceptionMessage instanceof PipeRuntimeCriticalException, outputStream);
+ ReadWriteIOUtils.write(exceptionMessage.getMessage(), outputStream);
+ }
+ }
+
public static PipeTaskMeta deserialize(ByteBuffer byteBuffer) {
final PipeTaskMeta PipeTaskMeta = new PipeTaskMeta();
PipeTaskMeta.progressIndex.set(ReadWriteIOUtils.readLong(byteBuffer));
diff --git a/node-commons/src/test/java/org/apache/iotdb/commons/pipe/task/meta/PipeMetaDeSerTest.java b/node-commons/src/test/java/org/apache/iotdb/commons/pipe/task/meta/PipeMetaDeSerTest.java
new file mode 100644
index 00000000000..08e7f86007c
--- /dev/null
+++ b/node-commons/src/test/java/org/apache/iotdb/commons/pipe/task/meta/PipeMetaDeSerTest.java
@@ -0,0 +1,77 @@
+/*
+ * 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.commons.pipe.task.meta;
+
+import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
+import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.HashMap;
+
+public class PipeMetaDeSerTest {
+
+ @Test
+ public void test() throws IOException {
+ PipeStaticMeta pipeStaticMeta =
+ new PipeStaticMeta(
+ "pipeName",
+ 123L,
+ new HashMap() {
+ {
+ put("collector-key", "collector-value");
+ }
+ },
+ new HashMap() {
+ {
+ put("processor-key-1", "processor-value-1");
+ put("processor-key-2", "processor-value-2");
+ }
+ },
+ new HashMap() {});
+ ByteBuffer staticByteBuffer = pipeStaticMeta.serialize();
+ PipeStaticMeta pipeStaticMeta1 = PipeStaticMeta.deserialize(staticByteBuffer);
+ Assert.assertEquals(pipeStaticMeta, pipeStaticMeta1);
+
+ PipeRuntimeMeta pipeRuntimeMeta =
+ new PipeRuntimeMeta(
+ new HashMap() {
+ {
+ put(
+ new TConsensusGroupId(TConsensusGroupType.DataRegion, 456),
+ new PipeTaskMeta(789, 987));
+ put(
+ new TConsensusGroupId(TConsensusGroupType.DataRegion, 123),
+ new PipeTaskMeta(456, 789));
+ }
+ });
+ ByteBuffer runtimeByteBuffer = pipeRuntimeMeta.serialize();
+ PipeRuntimeMeta pipeRuntimeMeta1 = PipeRuntimeMeta.deserialize(runtimeByteBuffer);
+ Assert.assertEquals(pipeRuntimeMeta, pipeRuntimeMeta1);
+
+ PipeMeta pipeMeta = new PipeMeta(pipeStaticMeta, pipeRuntimeMeta);
+ ByteBuffer byteBuffer = pipeMeta.serialize();
+ PipeMeta pipeMeta1 = PipeMeta.deserialize(byteBuffer);
+ Assert.assertEquals(pipeMeta, pipeMeta1);
+ }
+}
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 bea8913d268..82b7b494f91 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
@@ -3384,7 +3384,9 @@ public class ASTVisitor extends IoTDBSqlParserBaseVisitor<Statement> {
Map<String, String> collectorMap = new HashMap<>();
for (IoTDBSqlParser.CollectorAttributeClauseContext singleCtx :
ctx.collectorAttributeClause()) {
- collectorMap.put(singleCtx.collectorKey.getText(), singleCtx.collectorValue.getText());
+ collectorMap.put(
+ parseStringLiteral(singleCtx.collectorKey.getText()),
+ parseStringLiteral(singleCtx.collectorValue.getText()));
}
return collectorMap;
}
@@ -3394,7 +3396,9 @@ public class ASTVisitor extends IoTDBSqlParserBaseVisitor<Statement> {
Map<String, String> processorMap = new HashMap<>();
for (IoTDBSqlParser.ProcessorAttributeClauseContext singleCtx :
ctx.processorAttributeClause()) {
- processorMap.put(singleCtx.processorKey.getText(), singleCtx.processorValue.getText());
+ processorMap.put(
+ parseStringLiteral(singleCtx.processorKey.getText()),
+ parseStringLiteral(singleCtx.processorValue.getText()));
}
return processorMap;
}
@@ -3404,7 +3408,9 @@ public class ASTVisitor extends IoTDBSqlParserBaseVisitor<Statement> {
Map<String, String> connectorMap = new HashMap<>();
for (IoTDBSqlParser.ConnectorAttributeClauseContext singleCtx :
ctx.connectorAttributeClause()) {
- connectorMap.put(singleCtx.connectorKey.getText(), singleCtx.connectorValue.getText());
+ connectorMap.put(
+ parseStringLiteral(singleCtx.connectorKey.getText()),
+ parseStringLiteral(singleCtx.connectorValue.getText()));
}
return connectorMap;
}
diff --git a/server/src/main/java/org/apache/iotdb/db/service/thrift/impl/DataNodeInternalRPCServiceImpl.java b/server/src/main/java/org/apache/iotdb/db/service/thrift/impl/DataNodeInternalRPCServiceImpl.java
index c8931a521f8..6e0f82681a6 100644
--- a/server/src/main/java/org/apache/iotdb/db/service/thrift/impl/DataNodeInternalRPCServiceImpl.java
+++ b/server/src/main/java/org/apache/iotdb/db/service/thrift/impl/DataNodeInternalRPCServiceImpl.java
@@ -804,8 +804,10 @@ public class DataNodeInternalRPCServiceImpl implements IDataNodeRPCService.Iface
@Override
public TSStatus pushPipeMeta(TPushPipeMetaReq req) {
- List<PipeMeta> pipeMetas = new ArrayList<>();
- req.getPipeMetas().forEach(byteBuffer -> pipeMetas.add(PipeMeta.deserialize(byteBuffer)));
+ final List<PipeMeta> pipeMetas = new ArrayList<>();
+ for (ByteBuffer byteBuffer : req.getPipeMetas()) {
+ pipeMetas.add(PipeMeta.deserialize(byteBuffer));
+ }
PipeAgent.task().handlePipeMetaChanges(pipeMetas);
return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
}