You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ta...@apache.org on 2022/06/07 09:58:06 UTC
[iotdb] branch jira3359 created (now 66706daee4)
This is an automated email from the ASF dual-hosted git repository.
tanxinyu pushed a change to branch jira3359
in repository https://gitbox.apache.org/repos/asf/iotdb.git
at 66706daee4 init serializeToByteBuffer interface
This branch includes the following new commits:
new 66706daee4 init serializeToByteBuffer interface
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[iotdb] 01/01: init serializeToByteBuffer interface
Posted by ta...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
tanxinyu pushed a commit to branch jira3359
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 66706daee42c5fa03493cbb70cc999e9fad6b17c
Author: LebronAl <TX...@gmail.com>
AuthorDate: Tue Jun 7 17:57:44 2022 +0800
init serializeToByteBuffer interface
---
.../consensus/request/ConfigRequest.java | 16 +++
.../consensus/request/auth/AuthorReq.java | 21 +++
.../request/read/CountStorageGroupReq.java | 12 ++
.../consensus/request/read/GetDataNodeInfoReq.java | 8 ++
.../request/read/GetDataPartitionReq.java | 28 ++++
.../request/read/GetNodePathsPartitionReq.java | 7 +
.../request/read/GetSchemaPartitionReq.java | 18 +++
.../request/write/ApplyConfigNodeReq.java | 8 ++
.../request/write/CreateDataPartitionReq.java | 28 ++++
.../consensus/request/write/CreateFunctionReq.java | 15 ++
.../consensus/request/write/CreateRegionsReq.java | 18 +++
.../request/write/CreateSchemaPartitionReq.java | 21 +++
.../request/write/DeleteProcedureReq.java | 7 +
.../consensus/request/write/DeleteRegionsReq.java | 11 ++
.../request/write/DeleteStorageGroupReq.java | 8 ++
.../consensus/request/write/DropFunctionReq.java | 7 +
.../request/write/PreDeleteStorageGroupReq.java | 8 ++
.../request/write/RegisterDataNodeReq.java | 8 ++
.../request/write/SetDataReplicationFactorReq.java | 9 ++
.../write/SetSchemaReplicationFactorReq.java | 9 ++
.../request/write/SetStorageGroupReq.java | 7 +
.../consensus/request/write/SetTTLReq.java | 9 ++
.../request/write/SetTimePartitionIntervalReq.java | 9 ++
.../request/write/UpdateProcedureReq.java | 9 ++
.../statemachine/PartitionRegionStateMachine.java | 4 +-
.../iotdb/confignode/procedure/Procedure.java | 55 +++++++
.../org/apache/iotdb/consensus/common/Peer.java | 7 +
.../common/request/ByteBufferConsensusRequest.java | 3 +-
.../common/request/IConsensusRequest.java | 2 +
.../common/request/IndexedConsensusRequest.java | 5 +
.../multileader/logdispatcher/LogDispatcher.java | 13 +-
.../multileader/wal/GetConsensusReqReaderPlan.java | 5 +
.../iotdb/consensus/ratis/RequestMessage.java | 18 +--
.../multileader/MultiLeaderConsensusTest.java | 16 ++-
.../apache/iotdb/consensus/ratis/TestUtils.java | 16 ++-
.../standalone/StandAloneConsensusTest.java | 9 +-
.../org/apache/iotdb/commons/path/PartialPath.java | 8 ++
.../org/apache/iotdb/commons/path/PathType.java | 10 +-
.../commons/utils/BasicStructureSerDeUtil.java | 60 ++++++++
.../commons/utils/ThriftCommonsSerDeUtils.java | 69 +++++++++
.../commons/utils/ThriftConfigNodeSerDeUtils.java | 10 ++
.../commons/utils/ThriftCommonsSerDeUtilsTest.java | 18 ++-
.../consensus/statemachine/BaseStateMachine.java | 3 +-
.../statemachine/DataRegionStateMachine.java | 3 +-
.../iotdb/db/mpp/common/FragmentInstanceId.java | 7 +
.../apache/iotdb/db/mpp/common/PlanFragmentId.java | 9 +-
.../org/apache/iotdb/db/mpp/common/QueryId.java | 6 +
.../db/mpp/common/schematree/PathPatternNode.java | 13 ++
.../db/mpp/common/schematree/PathPatternTree.java | 6 +
.../iotdb/db/mpp/plan/analyze/TypeProvider.java | 10 ++
.../iotdb/db/mpp/plan/expression/Expression.java | 14 ++
.../plan/expression/binary/BinaryExpression.java | 7 +
.../mpp/plan/expression/leaf/ConstantOperand.java | 7 +
.../plan/expression/leaf/TimeSeriesOperand.java | 6 +
.../mpp/plan/expression/leaf/TimestampOperand.java | 6 +
.../plan/expression/multi/FunctionExpression.java | 11 ++
.../db/mpp/plan/expression/unary/InExpression.java | 12 ++
.../mpp/plan/expression/unary/LikeExpression.java | 8 ++
.../plan/expression/unary/RegularExpression.java | 8 ++
.../mpp/plan/expression/unary/UnaryExpression.java | 6 +
.../db/mpp/plan/planner/plan/FragmentInstance.java | 33 +++++
.../db/mpp/plan/planner/plan/PlanFragment.java | 13 ++
.../plan/planner/plan/node/DeleteRegionNode.java | 27 ++++
.../db/mpp/plan/planner/plan/node/PlanNode.java | 18 +++
.../db/mpp/plan/planner/plan/node/PlanNodeId.java | 8 +-
.../mpp/plan/planner/plan/node/PlanNodeType.java | 8 +-
.../node/metedata/read/CountSchemaMergeNode.java | 7 +
.../plan/node/metedata/read/DevicesCountNode.java | 9 ++
.../node/metedata/read/DevicesSchemaScanNode.java | 12 ++
.../metedata/read/LevelTimeSeriesCountNode.java | 10 ++
.../read/NodeManagementMemoryMergeNode.java | 12 ++
.../node/metedata/read/NodePathsConvertNode.java | 7 +
.../node/metedata/read/NodePathsCountNode.java | 7 +
.../metedata/read/NodePathsSchemaScanNode.java | 9 ++
.../node/metedata/read/SchemaFetchMergeNode.java | 7 +
.../node/metedata/read/SchemaFetchScanNode.java | 9 ++
.../node/metedata/read/SchemaQueryMergeNode.java | 7 +
.../metedata/read/SchemaQueryOrderByHeatNode.java | 7 +
.../node/metedata/read/TimeSeriesCountNode.java | 9 ++
.../metedata/read/TimeSeriesSchemaScanNode.java | 15 ++
.../node/metedata/write/AlterTimeSeriesNode.java | 49 +++++++
.../write/CreateAlignedTimeSeriesNode.java | 67 +++++++++
.../metedata/write/CreateMultiTimeSeriesNode.java | 13 ++
.../node/metedata/write/CreateTimeSeriesNode.java | 52 +++++++
.../node/metedata/write/DeleteTimeSeriesNode.java | 11 ++
.../metedata/write/InvalidateSchemaCacheNode.java | 16 +++
.../plan/node/metedata/write/MeasurementGroup.java | 61 ++++++++
.../planner/plan/node/process/AggregationNode.java | 18 +++
.../planner/plan/node/process/DeviceMergeNode.java | 13 ++
.../planner/plan/node/process/DeviceViewNode.java | 25 ++++
.../planner/plan/node/process/ExchangeNode.java | 15 ++
.../plan/planner/plan/node/process/FillNode.java | 8 ++
.../plan/planner/plan/node/process/FilterNode.java | 14 ++
.../planner/plan/node/process/FilterNullNode.java | 8 ++
.../plan/node/process/GroupByLevelNode.java | 18 +++
.../plan/node/process/LastQueryMergeNode.java | 13 ++
.../plan/planner/plan/node/process/LimitNode.java | 8 ++
.../plan/planner/plan/node/process/OffsetNode.java | 8 ++
.../planner/plan/node/process/ProjectNode.java | 11 ++
.../node/process/SlidingWindowAggregationNode.java | 18 +++
.../plan/planner/plan/node/process/SortNode.java | 8 ++
.../planner/plan/node/process/TimeJoinNode.java | 8 ++
.../planner/plan/node/process/TransformNode.java | 13 ++
.../planner/plan/node/sink/FragmentSinkNode.java | 11 ++
.../plan/node/source/AlignedLastQueryScanNode.java | 8 ++
.../source/AlignedSeriesAggregationScanNode.java | 25 ++++
.../plan/node/source/AlignedSeriesScanNode.java | 23 +++
.../plan/node/source/LastQueryScanNode.java | 8 ++
.../node/source/SeriesAggregationScanNode.java | 25 ++++
.../planner/plan/node/source/SeriesScanNode.java | 23 +++
.../planner/plan/node/write/DeleteDataNode.java | 13 ++
.../plan/node/write/InsertMultiTabletsNode.java | 21 ++-
.../plan/planner/plan/node/write/InsertNode.java | 25 ++++
.../planner/plan/node/write/InsertRowNode.java | 94 +++++++++++-
.../planner/plan/node/write/InsertRowsNode.java | 21 ++-
.../plan/node/write/InsertRowsOfOneDeviceNode.java | 24 +++-
.../planner/plan/node/write/InsertTabletNode.java | 158 +++++++++++++++++++--
.../plan/parameter/AggregationDescriptor.java | 11 ++
.../planner/plan/parameter/AggregationStep.java | 7 +
.../planner/plan/parameter/FillDescriptor.java | 9 ++
.../plan/parameter/FilterNullParameter.java | 10 ++
.../plan/parameter/GroupByTimeParameter.java | 12 ++
.../scheduler/FragmentInstanceDispatcherImpl.java | 4 +-
.../scheduler/SimpleFragInstanceDispatcher.java | 5 +-
.../mpp/plan/statement/literal/BooleanLiteral.java | 8 ++
.../mpp/plan/statement/literal/DoubleLiteral.java | 8 ++
.../db/mpp/plan/statement/literal/Literal.java | 4 +
.../db/mpp/plan/statement/literal/LongLiteral.java | 8 ++
.../db/mpp/plan/statement/literal/NullLiteral.java | 7 +
.../mpp/plan/statement/literal/StringLiteral.java | 8 ++
.../apache/iotdb/db/qp/physical/PhysicalPlan.java | 13 ++
.../mpp/plan/plan/FragmentInstanceSerdeTest.java | 8 +-
.../org/apache/iotdb/tsfile/read/common/Path.java | 15 +-
.../org/apache/iotdb/tsfile/utils/PublicBAOS.java | 1 +
134 files changed, 2009 insertions(+), 88 deletions(-)
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 83864e30dc..654c7fac56 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
@@ -45,10 +45,12 @@ import org.apache.iotdb.confignode.consensus.request.write.SetTTLReq;
import org.apache.iotdb.confignode.consensus.request.write.SetTimePartitionIntervalReq;
import org.apache.iotdb.confignode.consensus.request.write.UpdateProcedureReq;
import org.apache.iotdb.consensus.common.request.IConsensusRequest;
+import org.apache.iotdb.tsfile.utils.PublicBAOS;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.BufferOverflowException;
import java.nio.ByteBuffer;
@@ -90,8 +92,22 @@ public abstract class ConfigRequest implements IConsensusRequest {
}
}
+ @Override
+ public ByteBuffer serializeToByteBuffer() {
+ try (PublicBAOS byteArrayOutputStream = new PublicBAOS();
+ DataOutputStream outputStream = new DataOutputStream(byteArrayOutputStream)) {
+ serializeImpl(outputStream);
+ return ByteBuffer.wrap(byteArrayOutputStream.getBuf(), 0, byteArrayOutputStream.size());
+ } catch (IOException e) {
+ LOGGER.error("Unexpected error occurs when serializing this ConfigRequest.", e);
+ throw new RuntimeException(e);
+ }
+ }
+
protected abstract void serializeImpl(ByteBuffer buffer);
+ protected abstract void serializeImpl(DataOutputStream stream) throws IOException;
+
protected abstract void deserializeImpl(ByteBuffer buffer) throws IOException;
public static class Factory {
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/auth/AuthorReq.java b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/auth/AuthorReq.java
index d8f905ea99..f7235c59a9 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/auth/AuthorReq.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/auth/AuthorReq.java
@@ -23,6 +23,8 @@ import org.apache.iotdb.commons.utils.BasicStructureSerDeUtil;
import org.apache.iotdb.confignode.consensus.request.ConfigRequest;
import org.apache.iotdb.confignode.consensus.request.ConfigRequestType;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.HashSet;
import java.util.Objects;
@@ -149,6 +151,25 @@ public class AuthorReq extends ConfigRequest {
BasicStructureSerDeUtil.write(nodeName, buffer);
}
+ @Override
+ protected void serializeImpl(DataOutputStream stream) throws IOException {
+ BasicStructureSerDeUtil.write(getPlanTypeOrdinal(authorType), stream);
+ BasicStructureSerDeUtil.write(userName, stream);
+ BasicStructureSerDeUtil.write(roleName, stream);
+ BasicStructureSerDeUtil.write(password, stream);
+ BasicStructureSerDeUtil.write(newPassword, stream);
+ if (permissions == null) {
+ stream.write((byte) 0);
+ } else {
+ stream.write((byte) 1);
+ stream.write(permissions.size());
+ for (int permission : permissions) {
+ stream.write(permission);
+ }
+ }
+ BasicStructureSerDeUtil.write(nodeName, stream);
+ }
+
@Override
protected void deserializeImpl(ByteBuffer buffer) {
userName = BasicStructureSerDeUtil.readString(buffer);
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/read/CountStorageGroupReq.java b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/read/CountStorageGroupReq.java
index 5b788d6519..c3f1e3d6ca 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/read/CountStorageGroupReq.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/read/CountStorageGroupReq.java
@@ -22,6 +22,8 @@ import org.apache.iotdb.commons.utils.BasicStructureSerDeUtil;
import org.apache.iotdb.confignode.consensus.request.ConfigRequest;
import org.apache.iotdb.confignode.consensus.request.ConfigRequestType;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Arrays;
import java.util.List;
@@ -62,6 +64,16 @@ public class CountStorageGroupReq extends ConfigRequest {
}
}
+ @Override
+ protected void serializeImpl(DataOutputStream stream) throws IOException {
+ stream.write(getType().ordinal());
+
+ stream.write(storageGroupPattern.length);
+ for (String node : storageGroupPattern) {
+ BasicStructureSerDeUtil.write(node, stream);
+ }
+ }
+
@Override
protected void deserializeImpl(ByteBuffer buffer) {
int length = buffer.getInt();
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/read/GetDataNodeInfoReq.java b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/read/GetDataNodeInfoReq.java
index 8436814bff..5971aa52c3 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/read/GetDataNodeInfoReq.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/read/GetDataNodeInfoReq.java
@@ -21,6 +21,8 @@ package org.apache.iotdb.confignode.consensus.request.read;
import org.apache.iotdb.confignode.consensus.request.ConfigRequest;
import org.apache.iotdb.confignode.consensus.request.ConfigRequestType;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Objects;
@@ -48,6 +50,12 @@ public class GetDataNodeInfoReq extends ConfigRequest {
buffer.putInt(dataNodeID);
}
+ @Override
+ protected void serializeImpl(DataOutputStream stream) throws IOException {
+ stream.write(ConfigRequestType.GetDataNodeInfo.ordinal());
+ stream.write(dataNodeID);
+ }
+
@Override
protected void deserializeImpl(ByteBuffer buffer) {
this.dataNodeID = buffer.getInt();
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/read/GetDataPartitionReq.java b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/read/GetDataPartitionReq.java
index 9547e32c0e..ce1f5bb946 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/read/GetDataPartitionReq.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/read/GetDataPartitionReq.java
@@ -27,11 +27,14 @@ import org.apache.iotdb.confignode.consensus.request.ConfigRequest;
import org.apache.iotdb.confignode.consensus.request.ConfigRequestType;
import org.apache.iotdb.confignode.rpc.thrift.TDataPartitionReq;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Map.Entry;
import java.util.Objects;
/** Get or create DataPartition by the specific partitionSlotsMap. */
@@ -109,6 +112,31 @@ public class GetDataPartitionReq extends ConfigRequest {
}));
}
+ @Override
+ protected void serializeImpl(DataOutputStream stream) throws IOException {
+ stream.write(getType().ordinal());
+
+ stream.write(partitionSlotsMap.size());
+ for (Entry<String, Map<TSeriesPartitionSlot, List<TTimePartitionSlot>>> entry :
+ partitionSlotsMap.entrySet()) {
+ String storageGroup = entry.getKey();
+ Map<TSeriesPartitionSlot, List<TTimePartitionSlot>> seriesPartitionTimePartitionSlots =
+ entry.getValue();
+ BasicStructureSerDeUtil.write(storageGroup, stream);
+ stream.write(seriesPartitionTimePartitionSlots.size());
+ for (Entry<TSeriesPartitionSlot, List<TTimePartitionSlot>> e :
+ seriesPartitionTimePartitionSlots.entrySet()) {
+ TSeriesPartitionSlot seriesPartitionSlot = e.getKey();
+ List<TTimePartitionSlot> timePartitionSlots = e.getValue();
+ ThriftCommonsSerDeUtils.serializeTSeriesPartitionSlot(seriesPartitionSlot, stream);
+ stream.write(timePartitionSlots.size());
+ timePartitionSlots.forEach(
+ timePartitionSlot ->
+ ThriftCommonsSerDeUtils.serializeTTimePartitionSlot(timePartitionSlot, stream));
+ }
+ }
+ }
+
@Override
protected void deserializeImpl(ByteBuffer buffer) {
partitionSlotsMap = new HashMap<>();
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/read/GetNodePathsPartitionReq.java b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/read/GetNodePathsPartitionReq.java
index cb7abd836e..3ea77ad6b9 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/read/GetNodePathsPartitionReq.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/read/GetNodePathsPartitionReq.java
@@ -24,6 +24,7 @@ import org.apache.iotdb.confignode.consensus.request.ConfigRequest;
import org.apache.iotdb.confignode.consensus.request.ConfigRequestType;
import org.apache.iotdb.db.metadata.path.PathDeserializeUtil;
+import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Objects;
@@ -58,6 +59,12 @@ public class GetNodePathsPartitionReq extends ConfigRequest {
buffer.putInt(level);
}
+ @Override
+ protected void serializeImpl(DataOutputStream stream) throws IOException {
+ partialPath.serialize(stream);
+ stream.write(level);
+ }
+
@Override
protected void deserializeImpl(ByteBuffer buffer) throws IOException {
partialPath = (PartialPath) PathDeserializeUtil.deserialize(buffer);
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/read/GetSchemaPartitionReq.java b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/read/GetSchemaPartitionReq.java
index 76b957934f..36f03228b9 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/read/GetSchemaPartitionReq.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/read/GetSchemaPartitionReq.java
@@ -24,12 +24,14 @@ import org.apache.iotdb.commons.utils.ThriftCommonsSerDeUtils;
import org.apache.iotdb.confignode.consensus.request.ConfigRequest;
import org.apache.iotdb.confignode.consensus.request.ConfigRequestType;
+import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Map.Entry;
import java.util.Objects;
/** Get or create SchemaPartition by the specific partitionSlotsMap. */
@@ -72,6 +74,22 @@ public class GetSchemaPartitionReq extends ConfigRequest {
});
}
+ @Override
+ protected void serializeImpl(DataOutputStream stream) throws IOException {
+ stream.write(getType().ordinal());
+
+ stream.write(partitionSlotsMap.size());
+ for (Entry<String, List<TSeriesPartitionSlot>> entry : partitionSlotsMap.entrySet()) {
+ String storageGroup = entry.getKey();
+ List<TSeriesPartitionSlot> seriesPartitionSlots = entry.getValue();
+ BasicStructureSerDeUtil.write(storageGroup, stream);
+ stream.write(seriesPartitionSlots.size());
+ seriesPartitionSlots.forEach(
+ seriesPartitionSlot ->
+ ThriftCommonsSerDeUtils.serializeTSeriesPartitionSlot(seriesPartitionSlot, stream));
+ }
+ }
+
@Override
protected void deserializeImpl(ByteBuffer buffer) throws IOException {
partitionSlotsMap = new HashMap<>();
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/ApplyConfigNodeReq.java b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/ApplyConfigNodeReq.java
index 3eeceae8b7..c4c6a5301b 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/ApplyConfigNodeReq.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/ApplyConfigNodeReq.java
@@ -23,6 +23,7 @@ import org.apache.iotdb.commons.utils.ThriftConfigNodeSerDeUtils;
import org.apache.iotdb.confignode.consensus.request.ConfigRequest;
import org.apache.iotdb.confignode.consensus.request.ConfigRequestType;
+import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Objects;
@@ -51,6 +52,13 @@ public class ApplyConfigNodeReq extends ConfigRequest {
ThriftConfigNodeSerDeUtils.serializeTConfigNodeLocation(configNodeLocation, buffer);
}
+ @Override
+ protected void serializeImpl(DataOutputStream stream) throws IOException {
+ stream.write(ConfigRequestType.ApplyConfigNode.ordinal());
+
+ ThriftConfigNodeSerDeUtils.serializeTConfigNodeLocation(configNodeLocation, stream);
+ }
+
@Override
protected void deserializeImpl(ByteBuffer buffer) throws IOException {
configNodeLocation = ThriftConfigNodeSerDeUtils.deserializeTConfigNodeLocation(buffer);
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/CreateDataPartitionReq.java b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/CreateDataPartitionReq.java
index 7169656c6f..0d42dad895 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/CreateDataPartitionReq.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/CreateDataPartitionReq.java
@@ -26,6 +26,7 @@ import org.apache.iotdb.commons.utils.ThriftCommonsSerDeUtils;
import org.apache.iotdb.confignode.consensus.request.ConfigRequest;
import org.apache.iotdb.confignode.consensus.request.ConfigRequestType;
+import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
@@ -82,6 +83,33 @@ public class CreateDataPartitionReq extends ConfigRequest {
}
}
+ @Override
+ protected void serializeImpl(DataOutputStream stream) throws IOException {
+ stream.write(ConfigRequestType.CreateDataPartition.ordinal());
+
+ stream.write(assignedDataPartition.size());
+ for (Map.Entry<
+ String, Map<TSeriesPartitionSlot, Map<TTimePartitionSlot, List<TRegionReplicaSet>>>>
+ seriesPartitionTimePartitionEntry : assignedDataPartition.entrySet()) {
+ BasicStructureSerDeUtil.write(seriesPartitionTimePartitionEntry.getKey(), stream);
+ stream.write(seriesPartitionTimePartitionEntry.getValue().size());
+ for (Map.Entry<TSeriesPartitionSlot, Map<TTimePartitionSlot, List<TRegionReplicaSet>>>
+ timePartitionEntry : seriesPartitionTimePartitionEntry.getValue().entrySet()) {
+ ThriftCommonsSerDeUtils.serializeTSeriesPartitionSlot(timePartitionEntry.getKey(), stream);
+ stream.write(timePartitionEntry.getValue().size());
+ for (Map.Entry<TTimePartitionSlot, List<TRegionReplicaSet>> regionReplicaSetEntry :
+ timePartitionEntry.getValue().entrySet()) {
+ ThriftCommonsSerDeUtils.serializeTTimePartitionSlot(
+ regionReplicaSetEntry.getKey(), stream);
+ stream.write(regionReplicaSetEntry.getValue().size());
+ for (TRegionReplicaSet regionReplicaSet : regionReplicaSetEntry.getValue()) {
+ ThriftCommonsSerDeUtils.serializeTRegionReplicaSet(regionReplicaSet, stream);
+ }
+ }
+ }
+ }
+ }
+
@Override
protected void deserializeImpl(ByteBuffer buffer) throws IOException {
assignedDataPartition = new HashMap<>();
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/CreateFunctionReq.java b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/CreateFunctionReq.java
index 4cc034a335..44e94456ad 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/CreateFunctionReq.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/CreateFunctionReq.java
@@ -23,6 +23,7 @@ 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.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
@@ -71,6 +72,20 @@ public class CreateFunctionReq extends ConfigRequest {
}
}
+ @Override
+ protected void serializeImpl(DataOutputStream stream) throws IOException {
+ stream.write(getType().ordinal());
+
+ ReadWriteIOUtils.write(functionName, stream);
+ ReadWriteIOUtils.write(className, stream);
+
+ final int size = uris.size();
+ ReadWriteIOUtils.write(size, stream);
+ for (String uri : uris) {
+ ReadWriteIOUtils.write(uri, stream);
+ }
+ }
+
@Override
protected void deserializeImpl(ByteBuffer buffer) throws IOException {
functionName = ReadWriteIOUtils.readString(buffer);
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/CreateRegionsReq.java b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/CreateRegionsReq.java
index 5b9a900f31..b67fb553f4 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/CreateRegionsReq.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/CreateRegionsReq.java
@@ -24,11 +24,13 @@ import org.apache.iotdb.commons.utils.ThriftCommonsSerDeUtils;
import org.apache.iotdb.confignode.consensus.request.ConfigRequest;
import org.apache.iotdb.confignode.consensus.request.ConfigRequestType;
+import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
+import java.util.Map.Entry;
import java.util.Objects;
import java.util.TreeMap;
@@ -67,6 +69,22 @@ public class CreateRegionsReq extends ConfigRequest {
});
}
+ @Override
+ protected void serializeImpl(DataOutputStream stream) throws IOException {
+ stream.write(ConfigRequestType.CreateRegions.ordinal());
+
+ stream.write(regionMap.size());
+ for (Entry<String, List<TRegionReplicaSet>> entry : regionMap.entrySet()) {
+ String storageGroup = entry.getKey();
+ List<TRegionReplicaSet> regionReplicaSets = entry.getValue();
+ BasicStructureSerDeUtil.write(storageGroup, stream);
+ stream.write(regionReplicaSets.size());
+ regionReplicaSets.forEach(
+ regionReplicaSet ->
+ ThriftCommonsSerDeUtils.serializeTRegionReplicaSet(regionReplicaSet, stream));
+ }
+ }
+
@Override
protected void deserializeImpl(ByteBuffer buffer) throws IOException {
int storageGroupNum = buffer.getInt();
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/CreateSchemaPartitionReq.java b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/CreateSchemaPartitionReq.java
index 859ac98efd..93401b1eb1 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/CreateSchemaPartitionReq.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/CreateSchemaPartitionReq.java
@@ -25,10 +25,12 @@ import org.apache.iotdb.commons.utils.ThriftCommonsSerDeUtils;
import org.apache.iotdb.confignode.consensus.request.ConfigRequest;
import org.apache.iotdb.confignode.consensus.request.ConfigRequestType;
+import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.Map;
+import java.util.Map.Entry;
import java.util.Objects;
/** Create SchemaPartition by assignedSchemaPartition */
@@ -66,6 +68,25 @@ public class CreateSchemaPartitionReq extends ConfigRequest {
});
}
+ @Override
+ protected void serializeImpl(DataOutputStream stream) throws IOException {
+ stream.write(ConfigRequestType.CreateSchemaPartition.ordinal());
+
+ stream.write(assignedSchemaPartition.size());
+ for (Entry<String, Map<TSeriesPartitionSlot, TRegionReplicaSet>> entry :
+ assignedSchemaPartition.entrySet()) {
+ String storageGroup = entry.getKey();
+ Map<TSeriesPartitionSlot, TRegionReplicaSet> partitionSlots = entry.getValue();
+ BasicStructureSerDeUtil.write(storageGroup, stream);
+ stream.write(partitionSlots.size());
+ partitionSlots.forEach(
+ (seriesPartitionSlot, regionReplicaSet) -> {
+ ThriftCommonsSerDeUtils.serializeTSeriesPartitionSlot(seriesPartitionSlot, stream);
+ ThriftCommonsSerDeUtils.serializeTRegionReplicaSet(regionReplicaSet, stream);
+ });
+ }
+ }
+
@Override
protected void deserializeImpl(ByteBuffer buffer) throws IOException {
assignedSchemaPartition = new HashMap<>();
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/DeleteProcedureReq.java b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/DeleteProcedureReq.java
index 16b6b9fca8..2ee701ee5d 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/DeleteProcedureReq.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/DeleteProcedureReq.java
@@ -22,6 +22,7 @@ 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 java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Objects;
@@ -48,6 +49,12 @@ public class DeleteProcedureReq extends ConfigRequest {
buffer.putLong(procId);
}
+ @Override
+ protected void serializeImpl(DataOutputStream stream) throws IOException {
+ stream.write(ConfigRequestType.DeleteProcedure.ordinal());
+ stream.writeLong(procId);
+ }
+
@Override
protected void deserializeImpl(ByteBuffer buffer) throws IOException {
this.procId = buffer.getLong();
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/DeleteRegionsReq.java b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/DeleteRegionsReq.java
index 31436bd53a..736cd87e6a 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/DeleteRegionsReq.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/DeleteRegionsReq.java
@@ -23,6 +23,7 @@ import org.apache.iotdb.commons.utils.ThriftCommonsSerDeUtils;
import org.apache.iotdb.confignode.consensus.request.ConfigRequest;
import org.apache.iotdb.confignode.consensus.request.ConfigRequestType;
+import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
@@ -56,6 +57,16 @@ public class DeleteRegionsReq extends ConfigRequest {
}
}
+ @Override
+ protected void serializeImpl(DataOutputStream stream) throws IOException {
+ stream.write(ConfigRequestType.DeleteRegions.ordinal());
+
+ stream.write(consensusGroupIds.size());
+ for (TConsensusGroupId consensusGroupId : consensusGroupIds) {
+ ThriftCommonsSerDeUtils.serializeTConsensusGroupId(consensusGroupId, stream);
+ }
+ }
+
@Override
protected void deserializeImpl(ByteBuffer buffer) throws IOException {
int length = buffer.getInt();
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/DeleteStorageGroupReq.java b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/DeleteStorageGroupReq.java
index 4855a2dff2..544c313d56 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/DeleteStorageGroupReq.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/DeleteStorageGroupReq.java
@@ -23,6 +23,8 @@ import org.apache.iotdb.confignode.consensus.request.ConfigRequest;
import org.apache.iotdb.confignode.consensus.request.ConfigRequestType;
import org.apache.iotdb.confignode.rpc.thrift.TStorageGroupSchema;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Objects;
@@ -49,6 +51,12 @@ public class DeleteStorageGroupReq extends ConfigRequest {
ThriftConfigNodeSerDeUtils.serializeTStorageGroupSchema(storageGroup, buffer);
}
+ @Override
+ protected void serializeImpl(DataOutputStream stream) throws IOException {
+ stream.write(ConfigRequestType.DeleteStorageGroup.ordinal());
+ ThriftConfigNodeSerDeUtils.serializeTStorageGroupSchema(storageGroup, stream);
+ }
+
@Override
protected void deserializeImpl(ByteBuffer buffer) {
storageGroup = ThriftConfigNodeSerDeUtils.deserializeTStorageGroupSchema(buffer);
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
index 028660dbe4..1cd597e2ac 100644
--- 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
@@ -23,6 +23,7 @@ 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.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
@@ -49,6 +50,12 @@ public class DropFunctionReq extends ConfigRequest {
ReadWriteIOUtils.write(functionName, buffer);
}
+ @Override
+ protected void serializeImpl(DataOutputStream stream) throws IOException {
+ stream.write(getType().ordinal());
+ ReadWriteIOUtils.write(functionName, stream);
+ }
+
@Override
protected void deserializeImpl(ByteBuffer buffer) throws IOException {
functionName = ReadWriteIOUtils.readString(buffer);
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/PreDeleteStorageGroupReq.java b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/PreDeleteStorageGroupReq.java
index d0aab342a8..ca69297859 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/PreDeleteStorageGroupReq.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/PreDeleteStorageGroupReq.java
@@ -23,6 +23,7 @@ import org.apache.iotdb.commons.utils.BasicStructureSerDeUtil;
import org.apache.iotdb.confignode.consensus.request.ConfigRequest;
import org.apache.iotdb.confignode.consensus.request.ConfigRequestType;
+import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
@@ -63,6 +64,13 @@ public class PreDeleteStorageGroupReq extends ConfigRequest {
buffer.put(preDeleteType.getType());
}
+ @Override
+ protected void serializeImpl(DataOutputStream stream) throws IOException {
+ stream.write(ConfigRequestType.PreDeleteStorageGroup.ordinal());
+ BasicStructureSerDeUtil.write(storageGroup, stream);
+ stream.write(preDeleteType.getType());
+ }
+
@Override
protected void deserializeImpl(ByteBuffer buffer) throws IOException {
this.storageGroup = BasicStructureSerDeUtil.readString(buffer);
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/RegisterDataNodeReq.java b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/RegisterDataNodeReq.java
index 8800891207..9327021320 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/RegisterDataNodeReq.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/RegisterDataNodeReq.java
@@ -23,6 +23,8 @@ import org.apache.iotdb.commons.utils.ThriftCommonsSerDeUtils;
import org.apache.iotdb.confignode.consensus.request.ConfigRequest;
import org.apache.iotdb.confignode.consensus.request.ConfigRequestType;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Objects;
@@ -49,6 +51,12 @@ public class RegisterDataNodeReq extends ConfigRequest {
ThriftCommonsSerDeUtils.serializeTDataNodeInfo(info, buffer);
}
+ @Override
+ protected void serializeImpl(DataOutputStream stream) throws IOException {
+ stream.write(ConfigRequestType.RegisterDataNode.ordinal());
+ ThriftCommonsSerDeUtils.serializeTDataNodeInfo(info, stream);
+ }
+
@Override
protected void deserializeImpl(ByteBuffer buffer) {
info = ThriftCommonsSerDeUtils.deserializeTDataNodeInfo(buffer);
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/SetDataReplicationFactorReq.java b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/SetDataReplicationFactorReq.java
index 971d925d2c..b59abcd0a0 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/SetDataReplicationFactorReq.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/SetDataReplicationFactorReq.java
@@ -22,6 +22,7 @@ import org.apache.iotdb.commons.utils.BasicStructureSerDeUtil;
import org.apache.iotdb.confignode.consensus.request.ConfigRequest;
import org.apache.iotdb.confignode.consensus.request.ConfigRequestType;
+import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Objects;
@@ -58,6 +59,14 @@ public class SetDataReplicationFactorReq extends ConfigRequest {
buffer.putInt(dataReplicationFactor);
}
+ @Override
+ protected void serializeImpl(DataOutputStream stream) throws IOException {
+ stream.write(getType().ordinal());
+
+ BasicStructureSerDeUtil.write(storageGroup, stream);
+ stream.write(dataReplicationFactor);
+ }
+
@Override
protected void deserializeImpl(ByteBuffer buffer) throws IOException {
storageGroup = BasicStructureSerDeUtil.readString(buffer);
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/SetSchemaReplicationFactorReq.java b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/SetSchemaReplicationFactorReq.java
index 95166663d1..501ffb77b7 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/SetSchemaReplicationFactorReq.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/SetSchemaReplicationFactorReq.java
@@ -22,6 +22,7 @@ import org.apache.iotdb.commons.utils.BasicStructureSerDeUtil;
import org.apache.iotdb.confignode.consensus.request.ConfigRequest;
import org.apache.iotdb.confignode.consensus.request.ConfigRequestType;
+import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Objects;
@@ -58,6 +59,14 @@ public class SetSchemaReplicationFactorReq extends ConfigRequest {
buffer.putInt(schemaReplicationFactor);
}
+ @Override
+ protected void serializeImpl(DataOutputStream stream) throws IOException {
+ stream.write(getType().ordinal());
+
+ BasicStructureSerDeUtil.write(storageGroup, stream);
+ stream.write(schemaReplicationFactor);
+ }
+
@Override
protected void deserializeImpl(ByteBuffer buffer) throws IOException {
storageGroup = BasicStructureSerDeUtil.readString(buffer);
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/SetStorageGroupReq.java b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/SetStorageGroupReq.java
index c546f07676..0982f83679 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/SetStorageGroupReq.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/SetStorageGroupReq.java
@@ -23,6 +23,7 @@ import org.apache.iotdb.confignode.consensus.request.ConfigRequest;
import org.apache.iotdb.confignode.consensus.request.ConfigRequestType;
import org.apache.iotdb.confignode.rpc.thrift.TStorageGroupSchema;
+import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Objects;
@@ -51,6 +52,12 @@ public class SetStorageGroupReq extends ConfigRequest {
ThriftConfigNodeSerDeUtils.serializeTStorageGroupSchema(schema, buffer);
}
+ @Override
+ protected void serializeImpl(DataOutputStream stream) throws IOException {
+ stream.write(ConfigRequestType.SetStorageGroup.ordinal());
+ ThriftConfigNodeSerDeUtils.serializeTStorageGroupSchema(schema, stream);
+ }
+
@Override
protected void deserializeImpl(ByteBuffer buffer) throws IOException {
schema = ThriftConfigNodeSerDeUtils.deserializeTStorageGroupSchema(buffer);
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/SetTTLReq.java b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/SetTTLReq.java
index 8e11cc58b1..5f9aaed92f 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/SetTTLReq.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/SetTTLReq.java
@@ -22,6 +22,7 @@ import org.apache.iotdb.commons.utils.BasicStructureSerDeUtil;
import org.apache.iotdb.confignode.consensus.request.ConfigRequest;
import org.apache.iotdb.confignode.consensus.request.ConfigRequestType;
+import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Objects;
@@ -58,6 +59,14 @@ public class SetTTLReq extends ConfigRequest {
buffer.putLong(TTL);
}
+ @Override
+ protected void serializeImpl(DataOutputStream stream) throws IOException {
+ stream.write(getType().ordinal());
+
+ BasicStructureSerDeUtil.write(storageGroup, stream);
+ stream.writeLong(TTL);
+ }
+
@Override
protected void deserializeImpl(ByteBuffer buffer) throws IOException {
storageGroup = BasicStructureSerDeUtil.readString(buffer);
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/SetTimePartitionIntervalReq.java b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/SetTimePartitionIntervalReq.java
index 97c99583f8..6b3dcfee95 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/SetTimePartitionIntervalReq.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/SetTimePartitionIntervalReq.java
@@ -22,6 +22,7 @@ import org.apache.iotdb.commons.utils.BasicStructureSerDeUtil;
import org.apache.iotdb.confignode.consensus.request.ConfigRequest;
import org.apache.iotdb.confignode.consensus.request.ConfigRequestType;
+import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Objects;
@@ -58,6 +59,14 @@ public class SetTimePartitionIntervalReq extends ConfigRequest {
buffer.putLong(timePartitionInterval);
}
+ @Override
+ protected void serializeImpl(DataOutputStream stream) throws IOException {
+ stream.write(getType().ordinal());
+
+ BasicStructureSerDeUtil.write(storageGroup, stream);
+ stream.writeLong(timePartitionInterval);
+ }
+
@Override
protected void deserializeImpl(ByteBuffer buffer) throws IOException {
storageGroup = BasicStructureSerDeUtil.readString(buffer);
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/UpdateProcedureReq.java b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/UpdateProcedureReq.java
index 2e99380bc2..025edf436b 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/UpdateProcedureReq.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/UpdateProcedureReq.java
@@ -24,6 +24,7 @@ import org.apache.iotdb.confignode.consensus.request.ConfigRequestType;
import org.apache.iotdb.confignode.procedure.Procedure;
import org.apache.iotdb.confignode.procedure.store.ProcedureFactory;
+import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Objects;
@@ -52,6 +53,14 @@ public class UpdateProcedureReq extends ConfigRequest {
}
}
+ @Override
+ protected void serializeImpl(DataOutputStream stream) throws IOException {
+ stream.write(ConfigRequestType.UpdateProcedure.ordinal());
+ if (procedure != null) {
+ procedure.serialize(stream);
+ }
+ }
+
@Override
protected void deserializeImpl(ByteBuffer buffer) throws IOException {
Procedure procedure = ProcedureFactory.getInstance().create(buffer);
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/statemachine/PartitionRegionStateMachine.java b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/statemachine/PartitionRegionStateMachine.java
index 26a077d376..5558610c99 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/statemachine/PartitionRegionStateMachine.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/statemachine/PartitionRegionStateMachine.java
@@ -69,7 +69,7 @@ public class PartitionRegionStateMachine implements IStateMachine, IStateMachine
ConfigRequest plan;
if (request instanceof ByteBufferConsensusRequest) {
try {
- plan = ConfigRequest.Factory.create(((ByteBufferConsensusRequest) request).getContent());
+ plan = ConfigRequest.Factory.create(request.serializeToByteBuffer());
} catch (IOException e) {
LOGGER.error("Deserialization error for write plan : {}", request, e);
return new TSStatus(TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode());
@@ -100,7 +100,7 @@ public class PartitionRegionStateMachine implements IStateMachine, IStateMachine
ConfigRequest plan;
if (request instanceof ByteBufferConsensusRequest) {
try {
- plan = ConfigRequest.Factory.create(((ByteBufferConsensusRequest) request).getContent());
+ plan = ConfigRequest.Factory.create(request.serializeToByteBuffer());
} catch (IOException e) {
LOGGER.error("Deserialization error for write plan : {}", request);
return null;
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/procedure/Procedure.java b/confignode/src/main/java/org/apache/iotdb/confignode/procedure/Procedure.java
index ec2a8d024a..f535ff9b7a 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/procedure/Procedure.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/procedure/Procedure.java
@@ -31,6 +31,7 @@ import org.apache.iotdb.confignode.procedure.store.IProcedureStore;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.io.DataOutputStream;
import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.nio.ByteBuffer;
@@ -179,6 +180,60 @@ public abstract class Procedure<Env> implements Comparable<Procedure<Env>> {
byteBuffer.put(this.hasLock() ? (byte) 1 : (byte) 0);
}
+ public void serialize(DataOutputStream stream) throws IOException {
+ // procid
+ stream.writeLong(this.procId);
+ // state
+ stream.write(this.state.ordinal());
+ // submit time
+ stream.writeLong(this.submittedTime);
+ // last updated
+ stream.writeLong(this.lastUpdate);
+ // parent id
+ stream.writeLong(this.parentProcId);
+ // time out
+ stream.writeLong(this.timeout);
+ // stack indexes
+ if (stackIndexes != null) {
+ stream.write(stackIndexes.length);
+ for (int index : stackIndexes) {
+ stream.write(index);
+ }
+ } else {
+ stream.write(-1);
+ }
+
+ // exceptions
+ if (hasException()) {
+ stream.write((byte) 1);
+ String exceptionClassName = exception.getClass().getName();
+ byte[] exceptionClassNameBytes = exceptionClassName.getBytes(StandardCharsets.UTF_8);
+ stream.write(exceptionClassNameBytes.length);
+ stream.write(exceptionClassNameBytes);
+ String message = this.exception.getMessage();
+ if (message != null) {
+ byte[] messageBytes = message.getBytes(StandardCharsets.UTF_8);
+ stream.write(messageBytes.length);
+ stream.write(messageBytes);
+ } else {
+ stream.write(-1);
+ }
+ } else {
+ stream.write((byte) 0);
+ }
+
+ // result
+ if (result != null) {
+ stream.write(result.length);
+ stream.write(result);
+ } else {
+ stream.write(-1);
+ }
+
+ // has lock
+ stream.write(this.hasLock() ? (byte) 1 : (byte) 0);
+ }
+
public void deserialize(ByteBuffer byteBuffer) {
// procid
this.setProcId(byteBuffer.getLong());
diff --git a/consensus/src/main/java/org/apache/iotdb/consensus/common/Peer.java b/consensus/src/main/java/org/apache/iotdb/consensus/common/Peer.java
index df97102dd2..539226b6a9 100644
--- a/consensus/src/main/java/org/apache/iotdb/consensus/common/Peer.java
+++ b/consensus/src/main/java/org/apache/iotdb/consensus/common/Peer.java
@@ -23,6 +23,7 @@ import org.apache.iotdb.common.rpc.thrift.TEndPoint;
import org.apache.iotdb.commons.consensus.ConsensusGroupId;
import org.apache.iotdb.commons.utils.ThriftCommonsSerDeUtils;
+import java.io.DataOutputStream;
import java.nio.ByteBuffer;
import java.util.Objects;
@@ -51,6 +52,12 @@ public class Peer {
ThriftCommonsSerDeUtils.serializeTEndPoint(endpoint, buffer);
}
+ public void serialize(DataOutputStream stream) {
+ ThriftCommonsSerDeUtils.serializeTConsensusGroupId(
+ groupId.convertToTConsensusGroupId(), stream);
+ ThriftCommonsSerDeUtils.serializeTEndPoint(endpoint, stream);
+ }
+
public static Peer deserialize(ByteBuffer buffer) {
return new Peer(
ConsensusGroupId.Factory.createFromTConsensusGroupId(
diff --git a/consensus/src/main/java/org/apache/iotdb/consensus/common/request/ByteBufferConsensusRequest.java b/consensus/src/main/java/org/apache/iotdb/consensus/common/request/ByteBufferConsensusRequest.java
index 19ec78908a..fcfaa1a5dd 100644
--- a/consensus/src/main/java/org/apache/iotdb/consensus/common/request/ByteBufferConsensusRequest.java
+++ b/consensus/src/main/java/org/apache/iotdb/consensus/common/request/ByteBufferConsensusRequest.java
@@ -38,7 +38,8 @@ public class ByteBufferConsensusRequest implements IConsensusRequest {
@Override
public void serializeRequest(ByteBuffer buffer) {}
- public ByteBuffer getContent() {
+ @Override
+ public ByteBuffer serializeToByteBuffer() {
return byteBuffer;
}
}
diff --git a/consensus/src/main/java/org/apache/iotdb/consensus/common/request/IConsensusRequest.java b/consensus/src/main/java/org/apache/iotdb/consensus/common/request/IConsensusRequest.java
index 0a5aacfdf2..cf393b1c6a 100644
--- a/consensus/src/main/java/org/apache/iotdb/consensus/common/request/IConsensusRequest.java
+++ b/consensus/src/main/java/org/apache/iotdb/consensus/common/request/IConsensusRequest.java
@@ -24,4 +24,6 @@ import java.nio.ByteBuffer;
public interface IConsensusRequest {
void serializeRequest(ByteBuffer buffer);
+
+ ByteBuffer serializeToByteBuffer();
}
diff --git a/consensus/src/main/java/org/apache/iotdb/consensus/common/request/IndexedConsensusRequest.java b/consensus/src/main/java/org/apache/iotdb/consensus/common/request/IndexedConsensusRequest.java
index 24ba1ae13f..fff494b9cb 100644
--- a/consensus/src/main/java/org/apache/iotdb/consensus/common/request/IndexedConsensusRequest.java
+++ b/consensus/src/main/java/org/apache/iotdb/consensus/common/request/IndexedConsensusRequest.java
@@ -50,6 +50,11 @@ public class IndexedConsensusRequest implements IConsensusRequest {
request.serializeRequest(buffer);
}
+ @Override
+ public ByteBuffer serializeToByteBuffer() {
+ return request.serializeToByteBuffer();
+ }
+
public IConsensusRequest getRequest() {
return request;
}
diff --git a/consensus/src/main/java/org/apache/iotdb/consensus/multileader/logdispatcher/LogDispatcher.java b/consensus/src/main/java/org/apache/iotdb/consensus/multileader/logdispatcher/LogDispatcher.java
index 398d50fc93..a6cbacd925 100644
--- a/consensus/src/main/java/org/apache/iotdb/consensus/multileader/logdispatcher/LogDispatcher.java
+++ b/consensus/src/main/java/org/apache/iotdb/consensus/multileader/logdispatcher/LogDispatcher.java
@@ -42,7 +42,6 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
-import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.LinkedList;
@@ -282,13 +281,9 @@ public class LogDispatcher {
// TODO iterator
IConsensusRequest data = reader.getReq(currentIndex++);
if (data != null) {
- // TODO fix byteBuffer overflow
- ByteBuffer buffer = ByteBuffer.allocate(DEFAULT_BUFFER_SIZE);
- data.serializeRequest(buffer);
- buffer.flip();
// since WAL can no longer recover FragmentInstance, but only PlanNode, we need to give
// special flags to use different deserialization methods in the dataRegion stateMachine
- logBatches.add(new TLogBatch(TLogType.InsertNode, buffer));
+ logBatches.add(new TLogBatch(TLogType.InsertNode, data.serializeToByteBuffer()));
}
}
return currentIndex - 1;
@@ -296,11 +291,7 @@ public class LogDispatcher {
private void constructBatchIndexedFromConsensusRequest(
IndexedConsensusRequest request, List<TLogBatch> logBatches) {
- // TODO fix byteBuffer overflow
- ByteBuffer buffer = ByteBuffer.allocate(DEFAULT_BUFFER_SIZE);
- request.serializeRequest(buffer);
- buffer.flip();
- logBatches.add(new TLogBatch(TLogType.FragmentInstance, buffer));
+ logBatches.add(new TLogBatch(TLogType.FragmentInstance, request.serializeToByteBuffer()));
}
}
}
diff --git a/consensus/src/main/java/org/apache/iotdb/consensus/multileader/wal/GetConsensusReqReaderPlan.java b/consensus/src/main/java/org/apache/iotdb/consensus/multileader/wal/GetConsensusReqReaderPlan.java
index a9e9f8f34e..e22f9928f0 100644
--- a/consensus/src/main/java/org/apache/iotdb/consensus/multileader/wal/GetConsensusReqReaderPlan.java
+++ b/consensus/src/main/java/org/apache/iotdb/consensus/multileader/wal/GetConsensusReqReaderPlan.java
@@ -27,4 +27,9 @@ public class GetConsensusReqReaderPlan implements IConsensusRequest {
@Override
public void serializeRequest(ByteBuffer buffer) {}
+
+ @Override
+ public ByteBuffer serializeToByteBuffer() {
+ return null;
+ }
}
diff --git a/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RequestMessage.java b/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RequestMessage.java
index 004aae81bb..2067a54e42 100644
--- a/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RequestMessage.java
+++ b/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RequestMessage.java
@@ -18,19 +18,16 @@
*/
package org.apache.iotdb.consensus.ratis;
-import org.apache.iotdb.consensus.common.request.ByteBufferConsensusRequest;
import org.apache.iotdb.consensus.common.request.IConsensusRequest;
import org.apache.ratis.protocol.Message;
import org.apache.ratis.thirdparty.com.google.protobuf.ByteString;
-
-import java.nio.ByteBuffer;
+import org.apache.ratis.thirdparty.com.google.protobuf.UnsafeByteOperations;
public class RequestMessage implements Message {
private final IConsensusRequest actualRequest;
private volatile ByteString serializedContent;
- private static final int DEFAULT_BUFFER_SIZE = 1024 * 30;
public RequestMessage(IConsensusRequest request) {
this.actualRequest = request;
@@ -46,17 +43,8 @@ public class RequestMessage implements Message {
if (serializedContent == null) {
synchronized (this) {
if (serializedContent == null) {
- if (actualRequest instanceof ByteBufferConsensusRequest) {
- ByteBufferConsensusRequest req = (ByteBufferConsensusRequest) actualRequest;
- serializedContent = ByteString.copyFrom(req.getContent());
- req.getContent().flip();
- } else {
- // TODO Pooling
- ByteBuffer byteBuffer = ByteBuffer.allocate(DEFAULT_BUFFER_SIZE);
- actualRequest.serializeRequest(byteBuffer);
- byteBuffer.flip();
- serializedContent = ByteString.copyFrom(byteBuffer);
- }
+ serializedContent =
+ UnsafeByteOperations.unsafeWrap(actualRequest.serializeToByteBuffer());
}
}
}
diff --git a/consensus/src/test/java/org/apache/iotdb/consensus/multileader/MultiLeaderConsensusTest.java b/consensus/src/test/java/org/apache/iotdb/consensus/multileader/MultiLeaderConsensusTest.java
index 045c3c44d8..0b77a33079 100644
--- a/consensus/src/test/java/org/apache/iotdb/consensus/multileader/MultiLeaderConsensusTest.java
+++ b/consensus/src/test/java/org/apache/iotdb/consensus/multileader/MultiLeaderConsensusTest.java
@@ -36,6 +36,7 @@ import org.apache.iotdb.consensus.multileader.logdispatcher.IndexController;
import org.apache.iotdb.consensus.multileader.thrift.TLogType;
import org.apache.iotdb.consensus.multileader.wal.ConsensusReqReader;
import org.apache.iotdb.consensus.multileader.wal.GetConsensusReqReaderPlan;
+import org.apache.iotdb.tsfile.utils.PublicBAOS;
import org.apache.ratis.util.FileUtils;
import org.junit.After;
@@ -45,6 +46,7 @@ import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.io.DataOutputStream;
import java.io.File;
import java.io.IOException;
import java.nio.ByteBuffer;
@@ -314,6 +316,18 @@ public class MultiLeaderConsensusTest {
peer.serialize(buffer);
}
+ @Override
+ public ByteBuffer serializeToByteBuffer() {
+ try (PublicBAOS publicBAOS = new PublicBAOS();
+ DataOutputStream outputStream = new DataOutputStream(publicBAOS)) {
+ outputStream.writeInt(num);
+ peer.serialize(outputStream);
+ return ByteBuffer.wrap(publicBAOS.getBuf(), 0, publicBAOS.size());
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
@Override
public boolean equals(Object o) {
if (this == o) {
@@ -361,7 +375,7 @@ public class MultiLeaderConsensusTest {
public synchronized TSStatus write(IConsensusRequest request) {
IConsensusRequest innerRequest = ((IndexedConsensusRequest) request).getRequest();
if (innerRequest instanceof ByteBufferConsensusRequest) {
- ByteBuffer buffer = ((ByteBufferConsensusRequest) innerRequest).getContent();
+ ByteBuffer buffer = innerRequest.serializeToByteBuffer();
requestSet.add(
new IndexedConsensusRequest(
((IndexedConsensusRequest) request).getSearchIndex(),
diff --git a/consensus/src/test/java/org/apache/iotdb/consensus/ratis/TestUtils.java b/consensus/src/test/java/org/apache/iotdb/consensus/ratis/TestUtils.java
index 3aad1fd504..0c8cd19505 100644
--- a/consensus/src/test/java/org/apache/iotdb/consensus/ratis/TestUtils.java
+++ b/consensus/src/test/java/org/apache/iotdb/consensus/ratis/TestUtils.java
@@ -53,7 +53,7 @@ public class TestUtils {
}
}
- public static class TestRequest {
+ public static class TestRequest implements IConsensusRequest {
private final int cmd;
public TestRequest(ByteBuffer buffer) {
@@ -63,6 +63,18 @@ public class TestUtils {
public boolean isIncr() {
return cmd == 1;
}
+
+ @Override
+ public void serializeRequest(ByteBuffer buffer) {
+ buffer.putInt(cmd);
+ }
+
+ @Override
+ public ByteBuffer serializeToByteBuffer() {
+ ByteBuffer buffer = ByteBuffer.allocate(4).putInt(cmd);
+ buffer.flip();
+ return buffer;
+ }
}
public static class IntegerCounter implements IStateMachine, IStateMachine.EventApi {
@@ -83,7 +95,7 @@ public class TestUtils {
public TSStatus write(IConsensusRequest request) {
TestRequest testRequest;
if (request instanceof ByteBufferConsensusRequest) {
- testRequest = new TestRequest(((ByteBufferConsensusRequest) request).getContent());
+ testRequest = new TestRequest(request.serializeToByteBuffer());
} else {
testRequest = (TestRequest) request;
}
diff --git a/consensus/src/test/java/org/apache/iotdb/consensus/standalone/StandAloneConsensusTest.java b/consensus/src/test/java/org/apache/iotdb/consensus/standalone/StandAloneConsensusTest.java
index 2973fe7f05..056e449e72 100644
--- a/consensus/src/test/java/org/apache/iotdb/consensus/standalone/StandAloneConsensusTest.java
+++ b/consensus/src/test/java/org/apache/iotdb/consensus/standalone/StandAloneConsensusTest.java
@@ -78,6 +78,13 @@ public class StandAloneConsensusTest {
public void serializeRequest(ByteBuffer buffer) {
buffer.putInt(num);
}
+
+ @Override
+ public ByteBuffer serializeToByteBuffer() {
+ ByteBuffer buffer = ByteBuffer.allocate(4).putInt(num);
+ buffer.flip();
+ return buffer;
+ }
}
private static class TestStateMachine implements IStateMachine, IStateMachine.EventApi {
@@ -97,7 +104,7 @@ public class StandAloneConsensusTest {
@Override
public TSStatus write(IConsensusRequest request) {
if (request instanceof ByteBufferConsensusRequest) {
- return new TSStatus(((ByteBufferConsensusRequest) request).getContent().getInt());
+ return new TSStatus(request.serializeToByteBuffer().getInt());
} else if (request instanceof TestEntry) {
return new TSStatus(
direction ? ((TestEntry) request).num + 1 : ((TestEntry) request).num - 1);
diff --git a/node-commons/src/main/java/org/apache/iotdb/commons/path/PartialPath.java b/node-commons/src/main/java/org/apache/iotdb/commons/path/PartialPath.java
index 510a02dde3..d90cf0cb8e 100644
--- a/node-commons/src/main/java/org/apache/iotdb/commons/path/PartialPath.java
+++ b/node-commons/src/main/java/org/apache/iotdb/commons/path/PartialPath.java
@@ -32,6 +32,8 @@ import org.apache.commons.lang3.Validate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
@@ -522,6 +524,12 @@ public class PartialPath extends Path implements Comparable<Path>, Cloneable {
serializeWithoutType(byteBuffer);
}
+ @Override
+ public void serialize(DataOutputStream stream) throws IOException {
+ PathType.Partial.serialize(stream);
+ serializeWithoutType(stream);
+ }
+
@Override
protected void serializeWithoutType(ByteBuffer byteBuffer) {
super.serializeWithoutType(byteBuffer);
diff --git a/node-commons/src/main/java/org/apache/iotdb/commons/path/PathType.java b/node-commons/src/main/java/org/apache/iotdb/commons/path/PathType.java
index cc082a8682..c27814afa4 100644
--- a/node-commons/src/main/java/org/apache/iotdb/commons/path/PathType.java
+++ b/node-commons/src/main/java/org/apache/iotdb/commons/path/PathType.java
@@ -18,6 +18,10 @@
*/
package org.apache.iotdb.commons.path;
+import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
public enum PathType {
@@ -33,6 +37,10 @@ public enum PathType {
}
public void serialize(ByteBuffer buffer) {
- buffer.put(pathType);
+ ReadWriteIOUtils.write(pathType, buffer);
+ }
+
+ public void serialize(DataOutputStream stream) throws IOException {
+ ReadWriteIOUtils.write(pathType, stream);
}
}
diff --git a/node-commons/src/main/java/org/apache/iotdb/commons/utils/BasicStructureSerDeUtil.java b/node-commons/src/main/java/org/apache/iotdb/commons/utils/BasicStructureSerDeUtil.java
index 215829d396..16d32eab88 100644
--- a/node-commons/src/main/java/org/apache/iotdb/commons/utils/BasicStructureSerDeUtil.java
+++ b/node-commons/src/main/java/org/apache/iotdb/commons/utils/BasicStructureSerDeUtil.java
@@ -19,6 +19,8 @@
package org.apache.iotdb.commons.utils;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.HashMap;
@@ -65,6 +67,23 @@ public class BasicStructureSerDeUtil {
return len;
}
+ /**
+ * write string to byteBuffer.
+ *
+ * @return the length of string represented by byte[].
+ */
+ public static int write(String s, DataOutputStream stream) throws IOException {
+ if (s == null) {
+ return write(-1, stream);
+ }
+ int len = 0;
+ byte[] bytes = s.getBytes();
+ len += write(bytes.length, stream);
+ stream.write(bytes);
+ len += bytes.length;
+ return len;
+ }
+
/**
* write a int n to byteBuffer.
*
@@ -75,6 +94,16 @@ public class BasicStructureSerDeUtil {
return INT_LEN;
}
+ /**
+ * write a int n to dataOutputStream.
+ *
+ * @return The number of bytes used to represent n.
+ */
+ public static int write(int n, DataOutputStream stream) throws IOException {
+ stream.write(n);
+ return INT_LEN;
+ }
+
/**
* write a map to buffer
*
@@ -106,6 +135,37 @@ public class BasicStructureSerDeUtil {
return length;
}
+ /**
+ * write a map to dataOutputStream
+ *
+ * @param map Map<String, String>
+ * @param stream DataOutputStream
+ * @return length
+ */
+ public static int write(Map<String, String> map, DataOutputStream stream) throws IOException {
+ if (map == null) {
+ return write(-1, stream);
+ }
+
+ int length = 0;
+ byte[] bytes;
+ stream.write(map.size());
+ length += 4;
+ for (Map.Entry<String, String> entry : map.entrySet()) {
+ bytes = entry.getKey().getBytes();
+ stream.write(bytes.length);
+ length += 4;
+ stream.write(bytes);
+ length += bytes.length;
+ bytes = entry.getValue().getBytes();
+ stream.write(bytes.length);
+ length += 4;
+ stream.write(bytes);
+ length += bytes.length;
+ }
+ return length;
+ }
+
/**
* read map from buffer
*
diff --git a/node-commons/src/main/java/org/apache/iotdb/commons/utils/ThriftCommonsSerDeUtils.java b/node-commons/src/main/java/org/apache/iotdb/commons/utils/ThriftCommonsSerDeUtils.java
index e689d09c95..fbe36ade09 100644
--- a/node-commons/src/main/java/org/apache/iotdb/commons/utils/ThriftCommonsSerDeUtils.java
+++ b/node-commons/src/main/java/org/apache/iotdb/commons/utils/ThriftCommonsSerDeUtils.java
@@ -30,9 +30,11 @@ import org.apache.iotdb.commons.exception.runtime.ThriftSerDeException;
import org.apache.thrift.TException;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.transport.TByteBuffer;
+import org.apache.thrift.transport.TIOStreamTransport;
import org.apache.thrift.transport.TTransport;
import org.apache.thrift.transport.TTransportException;
+import java.io.DataOutputStream;
import java.nio.ByteBuffer;
/** Utils for serialize and deserialize all the data struct defined by thrift-commons */
@@ -48,6 +50,12 @@ public class ThriftCommonsSerDeUtils {
return new TBinaryProtocol(transport);
}
+ private static TBinaryProtocol generateWriteProtocol(DataOutputStream stream)
+ throws TTransportException {
+ TTransport transport = new TIOStreamTransport(stream);
+ return new TBinaryProtocol(transport);
+ }
+
private static TBinaryProtocol generateReadProtocol(ByteBuffer buffer)
throws TTransportException {
TTransport transport = new TByteBuffer(buffer);
@@ -62,6 +70,14 @@ public class ThriftCommonsSerDeUtils {
}
}
+ public static void serializeTEndPoint(TEndPoint endPoint, DataOutputStream stream) {
+ try {
+ endPoint.write(generateWriteProtocol(stream));
+ } catch (TException e) {
+ throw new ThriftSerDeException("Write TEndPoint failed: ", e);
+ }
+ }
+
public static TEndPoint deserializeTEndPoint(ByteBuffer buffer) {
TEndPoint endPoint = new TEndPoint();
try {
@@ -81,6 +97,15 @@ public class ThriftCommonsSerDeUtils {
}
}
+ public static void serializeTDataNodeLocation(
+ TDataNodeLocation dataNodeLocation, DataOutputStream stream) {
+ try {
+ dataNodeLocation.write(generateWriteProtocol(stream));
+ } catch (TException e) {
+ throw new ThriftSerDeException("Write TDataNodeLocation failed: ", e);
+ }
+ }
+
public static TDataNodeLocation deserializeTDataNodeLocation(ByteBuffer buffer) {
TDataNodeLocation dataNodeLocation = new TDataNodeLocation();
try {
@@ -99,6 +124,14 @@ public class ThriftCommonsSerDeUtils {
}
}
+ public static void serializeTDataNodeInfo(TDataNodeInfo dataNodeInfo, DataOutputStream stream) {
+ try {
+ dataNodeInfo.write(generateWriteProtocol(stream));
+ } catch (TException e) {
+ throw new ThriftSerDeException("Write TDataNodeInfo failed: ", e);
+ }
+ }
+
public static TDataNodeInfo deserializeTDataNodeInfo(ByteBuffer buffer) {
TDataNodeInfo dataNodeInfo = new TDataNodeInfo();
try {
@@ -118,6 +151,15 @@ public class ThriftCommonsSerDeUtils {
}
}
+ public static void serializeTSeriesPartitionSlot(
+ TSeriesPartitionSlot seriesPartitionSlot, DataOutputStream stream) {
+ try {
+ seriesPartitionSlot.write(generateWriteProtocol(stream));
+ } catch (TException e) {
+ throw new ThriftSerDeException("Write TSeriesPartitionSlot failed: ", e);
+ }
+ }
+
public static TSeriesPartitionSlot deserializeTSeriesPartitionSlot(ByteBuffer buffer) {
TSeriesPartitionSlot seriesPartitionSlot = new TSeriesPartitionSlot();
try {
@@ -137,6 +179,15 @@ public class ThriftCommonsSerDeUtils {
}
}
+ public static void serializeTTimePartitionSlot(
+ TTimePartitionSlot timePartitionSlot, DataOutputStream stream) {
+ try {
+ timePartitionSlot.write(generateWriteProtocol(stream));
+ } catch (TException e) {
+ throw new ThriftSerDeException("Write TTimePartitionSlot failed: ", e);
+ }
+ }
+
public static TTimePartitionSlot deserializeTTimePartitionSlot(ByteBuffer buffer) {
TTimePartitionSlot timePartitionSlot = new TTimePartitionSlot();
try {
@@ -156,6 +207,15 @@ public class ThriftCommonsSerDeUtils {
}
}
+ public static void serializeTConsensusGroupId(
+ TConsensusGroupId consensusGroupId, DataOutputStream stream) {
+ try {
+ consensusGroupId.write(generateWriteProtocol(stream));
+ } catch (TException e) {
+ throw new ThriftSerDeException("Write TConsensusGroupId failed: ", e);
+ }
+ }
+
public static TConsensusGroupId deserializeTConsensusGroupId(ByteBuffer buffer) {
TConsensusGroupId consensusGroupId = new TConsensusGroupId();
try {
@@ -175,6 +235,15 @@ public class ThriftCommonsSerDeUtils {
}
}
+ public static void serializeTRegionReplicaSet(
+ TRegionReplicaSet regionReplicaSet, DataOutputStream stream) {
+ try {
+ regionReplicaSet.write(generateWriteProtocol(stream));
+ } catch (TException e) {
+ throw new ThriftSerDeException("Write TRegionReplicaSet failed: ", e);
+ }
+ }
+
public static TRegionReplicaSet deserializeTRegionReplicaSet(ByteBuffer buffer) {
TRegionReplicaSet regionReplicaSet = new TRegionReplicaSet();
try {
diff --git a/node-commons/src/main/java/org/apache/iotdb/commons/utils/ThriftConfigNodeSerDeUtils.java b/node-commons/src/main/java/org/apache/iotdb/commons/utils/ThriftConfigNodeSerDeUtils.java
index 25063bb7ce..e71fbbe32c 100644
--- a/node-commons/src/main/java/org/apache/iotdb/commons/utils/ThriftConfigNodeSerDeUtils.java
+++ b/node-commons/src/main/java/org/apache/iotdb/commons/utils/ThriftConfigNodeSerDeUtils.java
@@ -29,6 +29,7 @@ import org.apache.thrift.transport.TIOStreamTransport;
import org.apache.thrift.transport.TTransport;
import org.apache.thrift.transport.TTransportException;
+import java.io.DataOutputStream;
import java.io.InputStream;
import java.io.OutputStream;
import java.nio.ByteBuffer;
@@ -111,6 +112,15 @@ public class ThriftConfigNodeSerDeUtils {
}
}
+ public static void serializeTConfigNodeLocation(
+ TConfigNodeLocation configNodeLocation, DataOutputStream stream) {
+ try {
+ configNodeLocation.write(generateWriteProtocol(stream));
+ } catch (TException e) {
+ throw new ThriftSerDeException("Write TConfigNodeLocation failed: ", e);
+ }
+ }
+
public static TConfigNodeLocation deserializeTConfigNodeLocation(ByteBuffer buffer) {
TConfigNodeLocation configNodeLocation = new TConfigNodeLocation();
try {
diff --git a/node-commons/src/test/java/org/apache/iotdb/commons/utils/ThriftCommonsSerDeUtilsTest.java b/node-commons/src/test/java/org/apache/iotdb/commons/utils/ThriftCommonsSerDeUtilsTest.java
index 94a2a81427..eb4ce9951c 100644
--- a/node-commons/src/test/java/org/apache/iotdb/commons/utils/ThriftCommonsSerDeUtilsTest.java
+++ b/node-commons/src/test/java/org/apache/iotdb/commons/utils/ThriftCommonsSerDeUtilsTest.java
@@ -25,11 +25,14 @@ import org.apache.iotdb.common.rpc.thrift.TEndPoint;
import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
import org.apache.iotdb.common.rpc.thrift.TSeriesPartitionSlot;
import org.apache.iotdb.common.rpc.thrift.TTimePartitionSlot;
+import org.apache.iotdb.tsfile.utils.PublicBAOS;
import org.junit.After;
import org.junit.Assert;
import org.junit.Test;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
@@ -99,7 +102,7 @@ public class ThriftCommonsSerDeUtilsTest {
}
@Test
- public void readWriteTRegionReplicaSetTest() {
+ public void readWriteTRegionReplicaSetTest() throws IOException {
TRegionReplicaSet regionReplicaSet0 = new TRegionReplicaSet();
regionReplicaSet0.setRegionId(new TConsensusGroupId(TConsensusGroupType.SchemaRegion, 0));
regionReplicaSet0.setDataNodeLocations(new ArrayList<>());
@@ -113,10 +116,13 @@ public class ThriftCommonsSerDeUtilsTest {
dataNodeLocation.setSchemaRegionConsensusEndPoint(new TEndPoint("0.0.0.0", 50010 + i));
regionReplicaSet0.getDataNodeLocations().add(dataNodeLocation);
}
- ThriftCommonsSerDeUtils.serializeTRegionReplicaSet(regionReplicaSet0, buffer);
- buffer.flip();
- TRegionReplicaSet regionReplicaSet1 =
- ThriftCommonsSerDeUtils.deserializeTRegionReplicaSet(buffer);
- Assert.assertEquals(regionReplicaSet0, regionReplicaSet1);
+ try (PublicBAOS byteArrayOutputStream = new PublicBAOS();
+ DataOutputStream outputStream = new DataOutputStream(byteArrayOutputStream)) {
+ ThriftCommonsSerDeUtils.serializeTRegionReplicaSet(regionReplicaSet0, outputStream);
+ TRegionReplicaSet regionReplicaSet1 =
+ ThriftCommonsSerDeUtils.deserializeTRegionReplicaSet(
+ ByteBuffer.wrap(byteArrayOutputStream.getBuf(), 0, byteArrayOutputStream.size()));
+ Assert.assertEquals(regionReplicaSet0, regionReplicaSet1);
+ }
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/consensus/statemachine/BaseStateMachine.java b/server/src/main/java/org/apache/iotdb/db/consensus/statemachine/BaseStateMachine.java
index 0387022154..e42e393788 100644
--- a/server/src/main/java/org/apache/iotdb/db/consensus/statemachine/BaseStateMachine.java
+++ b/server/src/main/java/org/apache/iotdb/db/consensus/statemachine/BaseStateMachine.java
@@ -34,8 +34,7 @@ public abstract class BaseStateMachine implements IStateMachine, IStateMachine.E
protected FragmentInstance getFragmentInstance(IConsensusRequest request) {
FragmentInstance instance;
if (request instanceof ByteBufferConsensusRequest) {
- instance =
- FragmentInstance.deserializeFrom(((ByteBufferConsensusRequest) request).getContent());
+ instance = FragmentInstance.deserializeFrom(request.serializeToByteBuffer());
} else if (request instanceof FragmentInstance) {
instance = (FragmentInstance) request;
} else {
diff --git a/server/src/main/java/org/apache/iotdb/db/consensus/statemachine/DataRegionStateMachine.java b/server/src/main/java/org/apache/iotdb/db/consensus/statemachine/DataRegionStateMachine.java
index 8387d98aeb..1cf07f56d4 100644
--- a/server/src/main/java/org/apache/iotdb/db/consensus/statemachine/DataRegionStateMachine.java
+++ b/server/src/main/java/org/apache/iotdb/db/consensus/statemachine/DataRegionStateMachine.java
@@ -22,7 +22,6 @@ package org.apache.iotdb.db.consensus.statemachine;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.consensus.DataRegionId;
import org.apache.iotdb.consensus.common.DataSet;
-import org.apache.iotdb.consensus.common.request.ByteBufferConsensusRequest;
import org.apache.iotdb.consensus.common.request.IConsensusRequest;
import org.apache.iotdb.consensus.common.request.IndexedConsensusRequest;
import org.apache.iotdb.consensus.multileader.thrift.TLogType;
@@ -108,7 +107,7 @@ public class DataRegionStateMachine extends BaseStateMachine {
if (indexedConsensusRequest.getType() == TLogType.InsertNode) {
planNode =
PlanNodeType.deserialize(
- ((ByteBufferConsensusRequest) indexedConsensusRequest.getRequest()).getContent());
+ indexedConsensusRequest.getRequest().serializeToByteBuffer());
} else {
planNode =
getFragmentInstance(indexedConsensusRequest.getRequest()).getFragment().getRoot();
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/common/FragmentInstanceId.java b/server/src/main/java/org/apache/iotdb/db/mpp/common/FragmentInstanceId.java
index 8c0336adbd..58386a071f 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/common/FragmentInstanceId.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/common/FragmentInstanceId.java
@@ -21,6 +21,8 @@ package org.apache.iotdb.db.mpp.common;
import org.apache.iotdb.mpp.rpc.thrift.TFragmentInstanceId;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Objects;
@@ -70,6 +72,11 @@ public class FragmentInstanceId {
ReadWriteIOUtils.write(instanceId, byteBuffer);
}
+ public void serialize(DataOutputStream stream) throws IOException {
+ fragmentId.serialize(stream);
+ ReadWriteIOUtils.write(instanceId, stream);
+ }
+
public TFragmentInstanceId toThrift() {
return new TFragmentInstanceId(queryId.getId(), fragmentId.getId(), instanceId);
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/common/PlanFragmentId.java b/server/src/main/java/org/apache/iotdb/db/mpp/common/PlanFragmentId.java
index 9a7983e401..7f28284a10 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/common/PlanFragmentId.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/common/PlanFragmentId.java
@@ -20,6 +20,8 @@ package org.apache.iotdb.db.mpp.common;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.List;
import java.util.Objects;
@@ -72,7 +74,12 @@ public class PlanFragmentId {
public void serialize(ByteBuffer byteBuffer) {
queryId.serialize(byteBuffer);
- byteBuffer.putInt(id);
+ ReadWriteIOUtils.write(id, byteBuffer);
+ }
+
+ public void serialize(DataOutputStream stream) throws IOException {
+ queryId.serialize(stream);
+ ReadWriteIOUtils.write(id, stream);
}
public static PlanFragmentId deserialize(ByteBuffer byteBuffer) {
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/common/QueryId.java b/server/src/main/java/org/apache/iotdb/db/mpp/common/QueryId.java
index 3524a52da0..d927fe3040 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/common/QueryId.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/common/QueryId.java
@@ -21,6 +21,8 @@ package org.apache.iotdb.db.mpp.common;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeId;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Arrays;
import java.util.List;
@@ -132,6 +134,10 @@ public class QueryId {
ReadWriteIOUtils.write(this.id, byteBuffer);
}
+ public void serialize(DataOutputStream stream) throws IOException {
+ ReadWriteIOUtils.write(this.id, stream);
+ }
+
public static QueryId deserialize(ByteBuffer byteBuffer) {
return new QueryId(ReadWriteIOUtils.readString(byteBuffer));
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/common/schematree/PathPatternNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/common/schematree/PathPatternNode.java
index baf0a5a18e..42db1e1995 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/common/schematree/PathPatternNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/common/schematree/PathPatternNode.java
@@ -23,6 +23,7 @@ import org.apache.iotdb.commons.utils.TestOnly;
import org.apache.iotdb.tsfile.utils.PublicBAOS;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.HashMap;
@@ -108,9 +109,21 @@ public class PathPatternNode {
serializeChildren(outputStream);
}
+ public void serialize(DataOutputStream outputStream) throws IOException {
+ ReadWriteIOUtils.write(name, outputStream);
+ ReadWriteIOUtils.write(children.size(), outputStream);
+ serializeChildren(outputStream);
+ }
+
void serializeChildren(PublicBAOS outputStream) throws IOException {
for (PathPatternNode childNode : children.values()) {
childNode.serialize(outputStream);
}
}
+
+ void serializeChildren(DataOutputStream outputStream) throws IOException {
+ for (PathPatternNode childNode : children.values()) {
+ childNode.serialize(outputStream);
+ }
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/common/schematree/PathPatternTree.java b/server/src/main/java/org/apache/iotdb/db/mpp/common/schematree/PathPatternTree.java
index 1700a4faa7..84047dd825 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/common/schematree/PathPatternTree.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/common/schematree/PathPatternTree.java
@@ -27,6 +27,7 @@ import org.apache.iotdb.tsfile.common.constant.TsFileConstant;
import org.apache.iotdb.tsfile.utils.PublicBAOS;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayDeque;
@@ -198,6 +199,11 @@ public class PathPatternTree {
root.serialize(outputStream);
}
+ public void serialize(DataOutputStream stream) throws IOException {
+ constructTree();
+ root.serialize(stream);
+ }
+
public void serialize(ByteBuffer buffer) {
constructTree();
root.serialize(buffer);
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/TypeProvider.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/TypeProvider.java
index 6c70c44e2d..b196d04a2b 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/TypeProvider.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/TypeProvider.java
@@ -23,6 +23,8 @@ import org.apache.iotdb.db.exception.sql.StatementAnalyzeException;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.Map;
@@ -67,6 +69,14 @@ public class TypeProvider {
}
}
+ public void serialize(DataOutputStream stream) throws IOException {
+ ReadWriteIOUtils.write(typeMap.size(), stream);
+ for (Map.Entry<String, TSDataType> entry : typeMap.entrySet()) {
+ ReadWriteIOUtils.write(entry.getKey(), stream);
+ ReadWriteIOUtils.write(entry.getValue().ordinal(), stream);
+ }
+ }
+
public static TypeProvider deserialize(ByteBuffer byteBuffer) {
int mapSize = ReadWriteIOUtils.readInt(byteBuffer);
Map<String, TSDataType> typeMap = new HashMap<>();
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/expression/Expression.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/expression/Expression.java
index bf0b1efcf7..534e52ec41 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/expression/Expression.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/expression/Expression.java
@@ -56,6 +56,7 @@ import org.apache.iotdb.db.qp.physical.crud.UDTFPlan;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.time.ZoneId;
@@ -296,6 +297,17 @@ public abstract class Expression {
}
}
+ public static void serialize(Expression expression, DataOutputStream stream) throws IOException {
+ ReadWriteIOUtils.write(expression.getExpressionType().getExpressionTypeInShortEnum(), stream);
+
+ expression.serialize(stream);
+
+ ReadWriteIOUtils.write(expression.inputColumnIndex != null, stream);
+ if (expression.inputColumnIndex != null) {
+ ReadWriteIOUtils.write(expression.inputColumnIndex, stream);
+ }
+ }
+
public static Expression deserialize(ByteBuffer byteBuffer) {
short type = ReadWriteIOUtils.readShort(byteBuffer);
@@ -391,4 +403,6 @@ public abstract class Expression {
public abstract ExpressionType getExpressionType();
protected abstract void serialize(ByteBuffer byteBuffer);
+
+ protected abstract void serialize(DataOutputStream stream) throws IOException;
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/expression/binary/BinaryExpression.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/expression/binary/BinaryExpression.java
index 218b67af82..8f6a5e1528 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/expression/binary/BinaryExpression.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/expression/binary/BinaryExpression.java
@@ -38,6 +38,7 @@ import org.apache.iotdb.db.mpp.transformation.dag.udf.UDTFExecutor;
import org.apache.iotdb.db.qp.physical.crud.UDTFPlan;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.time.ZoneId;
@@ -344,4 +345,10 @@ public abstract class BinaryExpression extends Expression {
Expression.serialize(leftExpression, byteBuffer);
Expression.serialize(rightExpression, byteBuffer);
}
+
+ @Override
+ protected void serialize(DataOutputStream stream) throws IOException {
+ Expression.serialize(leftExpression, stream);
+ Expression.serialize(rightExpression, stream);
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/expression/leaf/ConstantOperand.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/expression/leaf/ConstantOperand.java
index db142fc023..c8da18f69e 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/expression/leaf/ConstantOperand.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/expression/leaf/ConstantOperand.java
@@ -36,6 +36,7 @@ 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;
import java.util.List;
@@ -168,4 +169,10 @@ public class ConstantOperand extends LeafOperand {
dataType.serializeTo(byteBuffer);
ReadWriteIOUtils.write(valueString, byteBuffer);
}
+
+ @Override
+ protected void serialize(DataOutputStream stream) throws IOException {
+ dataType.serializeTo(stream);
+ ReadWriteIOUtils.write(valueString, stream);
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/expression/leaf/TimeSeriesOperand.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/expression/leaf/TimeSeriesOperand.java
index 9326672048..28963ad7f5 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/expression/leaf/TimeSeriesOperand.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/expression/leaf/TimeSeriesOperand.java
@@ -37,6 +37,7 @@ import org.apache.iotdb.db.mpp.transformation.dag.udf.UDTFContext;
import org.apache.iotdb.db.qp.physical.crud.UDTFPlan;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.List;
@@ -183,4 +184,9 @@ public class TimeSeriesOperand extends LeafOperand {
protected void serialize(ByteBuffer byteBuffer) {
path.serialize(byteBuffer);
}
+
+ @Override
+ protected void serialize(DataOutputStream stream) throws IOException {
+ path.serialize(stream);
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/expression/leaf/TimestampOperand.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/expression/leaf/TimestampOperand.java
index fec495d53e..7adf1a541c 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/expression/leaf/TimestampOperand.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/expression/leaf/TimestampOperand.java
@@ -36,6 +36,7 @@ import org.apache.iotdb.db.mpp.transformation.dag.udf.UDTFContext;
import org.apache.iotdb.db.qp.physical.crud.UDTFPlan;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.List;
@@ -171,4 +172,9 @@ public class TimestampOperand extends LeafOperand {
protected void serialize(ByteBuffer byteBuffer) {
// do nothing
}
+
+ @Override
+ protected void serialize(DataOutputStream stream) throws IOException {
+ // do nothing
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/expression/multi/FunctionExpression.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/expression/multi/FunctionExpression.java
index 5841a07c00..136e91c182 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/expression/multi/FunctionExpression.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/expression/multi/FunctionExpression.java
@@ -51,6 +51,7 @@ import org.apache.iotdb.db.utils.TypeInferenceUtils;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.time.ZoneId;
@@ -610,4 +611,14 @@ public class FunctionExpression extends Expression {
Expression.serialize(expression, byteBuffer);
}
}
+
+ @Override
+ protected void serialize(DataOutputStream stream) throws IOException {
+ ReadWriteIOUtils.write(functionName, stream);
+ ReadWriteIOUtils.write(functionAttributes, stream);
+ ReadWriteIOUtils.write(expressions.size(), stream);
+ for (Expression expression : expressions) {
+ Expression.serialize(expression, stream);
+ }
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/expression/unary/InExpression.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/expression/unary/InExpression.java
index 28e5c3307d..2a23e2237f 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/expression/unary/InExpression.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/expression/unary/InExpression.java
@@ -28,6 +28,8 @@ import org.apache.iotdb.db.mpp.transformation.dag.transformer.unary.InTransforme
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Iterator;
import java.util.LinkedHashSet;
@@ -107,4 +109,14 @@ public class InExpression extends UnaryExpression {
ReadWriteIOUtils.write(value, byteBuffer);
}
}
+
+ @Override
+ protected void serialize(DataOutputStream stream) throws IOException {
+ super.serialize(stream);
+ ReadWriteIOUtils.write(isNotIn, stream);
+ ReadWriteIOUtils.write(values.size(), stream);
+ for (String value : values) {
+ ReadWriteIOUtils.write(value, stream);
+ }
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/expression/unary/LikeExpression.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/expression/unary/LikeExpression.java
index 1c9164240b..53b062a092 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/expression/unary/LikeExpression.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/expression/unary/LikeExpression.java
@@ -29,6 +29,8 @@ import org.apache.iotdb.db.mpp.transformation.dag.transformer.unary.RegularTrans
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.regex.Pattern;
@@ -154,4 +156,10 @@ public class LikeExpression extends UnaryExpression {
super.serialize(byteBuffer);
ReadWriteIOUtils.write(patternString, byteBuffer);
}
+
+ @Override
+ protected void serialize(DataOutputStream stream) throws IOException {
+ super.serialize(stream);
+ ReadWriteIOUtils.write(patternString, stream);
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/expression/unary/RegularExpression.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/expression/unary/RegularExpression.java
index 8d8b1e1186..c0fb3cf6b0 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/expression/unary/RegularExpression.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/expression/unary/RegularExpression.java
@@ -31,6 +31,8 @@ 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;
import java.util.regex.Pattern;
@@ -101,4 +103,10 @@ public class RegularExpression extends UnaryExpression {
super.serialize(byteBuffer);
ReadWriteIOUtils.write(patternString, byteBuffer);
}
+
+ @Override
+ protected void serialize(DataOutputStream stream) throws IOException {
+ super.serialize(stream);
+ ReadWriteIOUtils.write(patternString, stream);
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/expression/unary/UnaryExpression.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/expression/unary/UnaryExpression.java
index 8370eba56a..471bb1c22b 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/expression/unary/UnaryExpression.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/expression/unary/UnaryExpression.java
@@ -38,6 +38,7 @@ import org.apache.iotdb.db.qp.physical.crud.UDTFPlan;
import org.apache.iotdb.db.qp.utils.WildcardsRemover;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.time.ZoneId;
@@ -217,4 +218,9 @@ public abstract class UnaryExpression extends Expression {
protected void serialize(ByteBuffer byteBuffer) {
Expression.serialize(expression, byteBuffer);
}
+
+ @Override
+ protected void serialize(DataOutputStream stream) throws IOException {
+ Expression.serialize(expression, stream);
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/FragmentInstance.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/FragmentInstance.java
index d57a6b5b47..d4ac9cf831 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/FragmentInstance.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/FragmentInstance.java
@@ -29,13 +29,21 @@ import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeUtil;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.sink.FragmentSinkNode;
import org.apache.iotdb.tsfile.read.filter.basic.Filter;
import org.apache.iotdb.tsfile.read.filter.factory.FilterFactory;
+import org.apache.iotdb.tsfile.utils.PublicBAOS;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Objects;
public class FragmentInstance implements IConsensusRequest {
+ private final Logger logger = LoggerFactory.getLogger(FragmentInstance.class);
+
private final FragmentInstanceId id;
private final QueryType type;
// The reference of PlanFragment which this instance is generated from
@@ -162,6 +170,31 @@ public class FragmentInstance implements IConsensusRequest {
}
}
+ public ByteBuffer serializeToByteBuffer() {
+ try (PublicBAOS publicBAOS = new PublicBAOS();
+ DataOutputStream outputStream = new DataOutputStream(publicBAOS)) {
+ id.serialize(outputStream);
+ fragment.serialize(outputStream);
+ ReadWriteIOUtils.write(timeFilter != null, outputStream);
+ if (timeFilter != null) {
+ timeFilter.serialize(outputStream);
+ }
+ ReadWriteIOUtils.write(type.ordinal(), outputStream);
+ ReadWriteIOUtils.write(regionReplicaSet != null, outputStream);
+ if (regionReplicaSet != null) {
+ ThriftCommonsSerDeUtils.serializeTRegionReplicaSet(regionReplicaSet, outputStream);
+ }
+ ReadWriteIOUtils.write(hostDataNode != null, outputStream);
+ if (hostDataNode != null) {
+ ThriftCommonsSerDeUtils.serializeTDataNodeLocation(hostDataNode, outputStream);
+ }
+ return ByteBuffer.wrap(publicBAOS.getBuf(), 0, publicBAOS.size());
+ } catch (IOException e) {
+ logger.error("Unexpected error occurs when serializing this FragmentInstance.", e);
+ throw new RuntimeException(e);
+ }
+ }
+
@Override
public boolean equals(Object o) {
if (this == o) return true;
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/PlanFragment.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/PlanFragment.java
index a500f65d18..b9c835dca0 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/PlanFragment.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/PlanFragment.java
@@ -27,6 +27,8 @@ import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeId;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeType;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Objects;
@@ -117,6 +119,17 @@ public class PlanFragment {
}
}
+ public void serialize(DataOutputStream stream) throws IOException {
+ id.serialize(stream);
+ root.serialize(stream);
+ if (typeProvider == null) {
+ ReadWriteIOUtils.write((byte) 0, stream);
+ } else {
+ ReadWriteIOUtils.write((byte) 1, stream);
+ typeProvider.serialize(stream);
+ }
+ }
+
public static PlanFragment deserialize(ByteBuffer byteBuffer) {
PlanFragment planFragment =
new PlanFragment(PlanFragmentId.deserialize(byteBuffer), deserializeHelper(byteBuffer));
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/DeleteRegionNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/DeleteRegionNode.java
index 3438b50783..a37c855656 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/DeleteRegionNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/DeleteRegionNode.java
@@ -23,13 +23,21 @@ import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
import org.apache.iotdb.commons.consensus.ConsensusGroupId;
import org.apache.iotdb.consensus.common.request.IConsensusRequest;
import org.apache.iotdb.db.mpp.plan.analyze.Analysis;
+import org.apache.iotdb.tsfile.utils.PublicBAOS;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.List;
public class DeleteRegionNode extends WritePlanNode implements IConsensusRequest {
+ private final Logger logger = LoggerFactory.getLogger(DeleteRegionNode.class);
+
protected ConsensusGroupId consensusGroupId;
public DeleteRegionNode(PlanNodeId id) {
@@ -102,8 +110,27 @@ public class DeleteRegionNode extends WritePlanNode implements IConsensusRequest
ReadWriteIOUtils.write(consensusGroupId.getId(), byteBuffer);
}
+ @Override
+ protected void serializeAttributes(DataOutputStream stream) throws IOException {
+ PlanNodeType.DELETE_REGION.serialize(stream);
+ ReadWriteIOUtils.write(consensusGroupId.getType().getValue(), stream);
+ ReadWriteIOUtils.write(consensusGroupId.getId(), stream);
+ }
+
@Override
public void serializeRequest(ByteBuffer buffer) {
super.serialize(buffer);
}
+
+ @Override
+ public ByteBuffer serializeToByteBuffer() {
+ try (PublicBAOS publicBAOS = new PublicBAOS();
+ DataOutputStream outputStream = new DataOutputStream(publicBAOS)) {
+ super.serialize(outputStream);
+ return ByteBuffer.wrap(publicBAOS.getBuf(), 0, publicBAOS.size());
+ } catch (IOException e) {
+ logger.error("Unexpected error occurs when serializing this DeleteRegionNode.", e);
+ throw new RuntimeException(e);
+ }
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/PlanNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/PlanNode.java
index d5647a316d..07f8746ee9 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/PlanNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/PlanNode.java
@@ -22,6 +22,8 @@ import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
import org.apache.commons.lang.Validate;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.List;
import java.util.Objects;
@@ -97,8 +99,24 @@ public abstract class PlanNode {
}
}
+ public void serialize(DataOutputStream stream) throws IOException {
+ serializeAttributes(stream);
+ id.serialize(stream);
+ List<PlanNode> planNodes = getChildren();
+ if (planNodes == null) {
+ ReadWriteIOUtils.write(0, stream);
+ } else {
+ ReadWriteIOUtils.write(planNodes.size(), stream);
+ for (PlanNode planNode : planNodes) {
+ planNode.serialize(stream);
+ }
+ }
+ }
+
protected abstract void serializeAttributes(ByteBuffer byteBuffer);
+ protected abstract void serializeAttributes(DataOutputStream stream) throws IOException;
+
@Override
public boolean equals(Object o) {
if (this == o) {
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/PlanNodeId.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/PlanNodeId.java
index 125b0abf0c..554a915945 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/PlanNodeId.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/PlanNodeId.java
@@ -20,10 +20,12 @@ package org.apache.iotdb.db.mpp.plan.planner.plan.node;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
public class PlanNodeId {
- private String id;
+ private final String id;
public PlanNodeId(String id) {
this.id = id;
@@ -58,4 +60,8 @@ public class PlanNodeId {
public void serialize(ByteBuffer byteBuffer) {
ReadWriteIOUtils.write(id, byteBuffer);
}
+
+ public void serialize(DataOutputStream stream) throws IOException {
+ ReadWriteIOUtils.write(id, stream);
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/PlanNodeType.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/PlanNodeType.java
index 3b8dbef2c4..dfd2c5896a 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/PlanNodeType.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/PlanNodeType.java
@@ -68,8 +68,10 @@ import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.InsertRowNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.InsertRowsNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.InsertRowsOfOneDeviceNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.InsertTabletNode;
+import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
import java.io.DataInputStream;
+import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
@@ -132,7 +134,11 @@ public enum PlanNodeType {
}
public void serialize(ByteBuffer buffer) {
- buffer.putShort(nodeType);
+ ReadWriteIOUtils.write(nodeType, buffer);
+ }
+
+ public void serialize(DataOutputStream stream) throws IOException {
+ ReadWriteIOUtils.write(nodeType, stream);
}
public static PlanNode deserialize(DataInputStream stream)
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/read/CountSchemaMergeNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/read/CountSchemaMergeNode.java
index 369095b939..1a3f345b9b 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/read/CountSchemaMergeNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/read/CountSchemaMergeNode.java
@@ -24,6 +24,8 @@ import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeId;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeType;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanVisitor;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
public class CountSchemaMergeNode extends AbstractSchemaMergeNode {
@@ -42,6 +44,11 @@ public class CountSchemaMergeNode extends AbstractSchemaMergeNode {
PlanNodeType.COUNT_MERGE.serialize(byteBuffer);
}
+ @Override
+ protected void serializeAttributes(DataOutputStream stream) throws IOException {
+ PlanNodeType.COUNT_MERGE.serialize(stream);
+ }
+
public static PlanNode deserialize(ByteBuffer byteBuffer) {
PlanNodeId planNodeId = PlanNodeId.deserialize(byteBuffer);
return new CountSchemaMergeNode(planNodeId);
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/read/DevicesCountNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/read/DevicesCountNode.java
index 93cbba8e93..f0161a28da 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/read/DevicesCountNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/read/DevicesCountNode.java
@@ -27,6 +27,8 @@ import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeId;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeType;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.List;
@@ -53,6 +55,13 @@ public class DevicesCountNode extends SchemaQueryScanNode {
ReadWriteIOUtils.write(isPrefixPath, byteBuffer);
}
+ @Override
+ protected void serializeAttributes(DataOutputStream stream) throws IOException {
+ PlanNodeType.DEVICES_COUNT.serialize(stream);
+ ReadWriteIOUtils.write(path.getFullPath(), stream);
+ ReadWriteIOUtils.write(isPrefixPath, stream);
+ }
+
public static PlanNode deserialize(ByteBuffer buffer) {
String fullPath = ReadWriteIOUtils.readString(buffer);
PartialPath path;
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/read/DevicesSchemaScanNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/read/DevicesSchemaScanNode.java
index 175d142997..9961d5c668 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/read/DevicesSchemaScanNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/read/DevicesSchemaScanNode.java
@@ -26,6 +26,8 @@ import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeId;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeType;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.List;
import java.util.Objects;
@@ -72,6 +74,16 @@ public class DevicesSchemaScanNode extends SchemaQueryScanNode {
ReadWriteIOUtils.write(hasSgCol, byteBuffer);
}
+ @Override
+ protected void serializeAttributes(DataOutputStream stream) throws IOException {
+ PlanNodeType.DEVICES_SCHEMA_SCAN.serialize(stream);
+ ReadWriteIOUtils.write(path.getFullPath(), stream);
+ ReadWriteIOUtils.write(limit, stream);
+ ReadWriteIOUtils.write(offset, stream);
+ ReadWriteIOUtils.write(isPrefixPath, stream);
+ ReadWriteIOUtils.write(hasSgCol, stream);
+ }
+
public static DevicesSchemaScanNode deserialize(ByteBuffer byteBuffer) {
String fullPath = ReadWriteIOUtils.readString(byteBuffer);
PartialPath path;
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/read/LevelTimeSeriesCountNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/read/LevelTimeSeriesCountNode.java
index 2854f0754f..5b5b9856d2 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/read/LevelTimeSeriesCountNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/read/LevelTimeSeriesCountNode.java
@@ -27,6 +27,8 @@ import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeId;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeType;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.List;
import java.util.Objects;
@@ -62,6 +64,14 @@ public class LevelTimeSeriesCountNode extends SchemaQueryScanNode {
ReadWriteIOUtils.write(level, byteBuffer);
}
+ @Override
+ protected void serializeAttributes(DataOutputStream stream) throws IOException {
+ PlanNodeType.LEVEL_TIME_SERIES_COUNT.serialize(stream);
+ ReadWriteIOUtils.write(path.getFullPath(), stream);
+ ReadWriteIOUtils.write(isPrefixPath, stream);
+ ReadWriteIOUtils.write(level, stream);
+ }
+
public static PlanNode deserialize(ByteBuffer buffer) {
String fullPath = ReadWriteIOUtils.readString(buffer);
PartialPath path;
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/read/NodeManagementMemoryMergeNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/read/NodeManagementMemoryMergeNode.java
index f697d1c84d..a83518d6f1 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/read/NodeManagementMemoryMergeNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/read/NodeManagementMemoryMergeNode.java
@@ -28,6 +28,8 @@ import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
import com.google.common.collect.ImmutableList;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.HashSet;
import java.util.List;
@@ -89,6 +91,16 @@ public class NodeManagementMemoryMergeNode extends ProcessNode {
data.forEach(node -> ReadWriteIOUtils.write(node, byteBuffer));
}
+ @Override
+ protected void serializeAttributes(DataOutputStream stream) throws IOException {
+ PlanNodeType.NODE_MANAGEMENT_MEMORY_MERGE.serialize(stream);
+ int size = data.size();
+ ReadWriteIOUtils.write(size, stream);
+ for (String node : data) {
+ ReadWriteIOUtils.write(node, stream);
+ }
+ }
+
public static NodeManagementMemoryMergeNode deserialize(ByteBuffer byteBuffer) {
Set<String> data = new HashSet<>();
int size = byteBuffer.getInt();
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/read/NodePathsConvertNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/read/NodePathsConvertNode.java
index fd530bff84..e73f530199 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/read/NodePathsConvertNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/read/NodePathsConvertNode.java
@@ -27,6 +27,8 @@ import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.ProcessNode;
import com.google.common.collect.ImmutableList;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.List;
@@ -77,6 +79,11 @@ public class NodePathsConvertNode extends ProcessNode {
PlanNodeType.NODE_PATHS_CONVERT.serialize(byteBuffer);
}
+ @Override
+ protected void serializeAttributes(DataOutputStream stream) throws IOException {
+ PlanNodeType.NODE_PATHS_CONVERT.serialize(stream);
+ }
+
public static NodePathsConvertNode deserialize(ByteBuffer byteBuffer) {
PlanNodeId planNodeId = PlanNodeId.deserialize(byteBuffer);
return new NodePathsConvertNode(planNodeId);
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/read/NodePathsCountNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/read/NodePathsCountNode.java
index 238039ded7..ac4b61d8d6 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/read/NodePathsCountNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/read/NodePathsCountNode.java
@@ -27,6 +27,8 @@ import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.ProcessNode;
import com.google.common.collect.ImmutableList;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.List;
@@ -77,6 +79,11 @@ public class NodePathsCountNode extends ProcessNode {
PlanNodeType.NODE_PATHS_COUNT.serialize(byteBuffer);
}
+ @Override
+ protected void serializeAttributes(DataOutputStream stream) throws IOException {
+ PlanNodeType.NODE_PATHS_COUNT.serialize(stream);
+ }
+
public static NodePathsCountNode deserialize(ByteBuffer byteBuffer) {
PlanNodeId planNodeId = PlanNodeId.deserialize(byteBuffer);
return new NodePathsCountNode(planNodeId);
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/read/NodePathsSchemaScanNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/read/NodePathsSchemaScanNode.java
index fe6c8318d3..01a2002130 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/read/NodePathsSchemaScanNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/read/NodePathsSchemaScanNode.java
@@ -26,6 +26,8 @@ import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeId;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeType;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.List;
import java.util.Objects;
@@ -66,6 +68,13 @@ public class NodePathsSchemaScanNode extends SchemaQueryScanNode {
byteBuffer.putInt(level);
}
+ @Override
+ protected void serializeAttributes(DataOutputStream stream) throws IOException {
+ PlanNodeType.NODE_PATHS_SCAN.serialize(stream);
+ prefixPath.serialize(stream);
+ stream.write(level);
+ }
+
public static PlanNode deserialize(ByteBuffer buffer) {
PartialPath path = (PartialPath) PathDeserializeUtil.deserialize(buffer);
int level = buffer.getInt();
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/read/SchemaFetchMergeNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/read/SchemaFetchMergeNode.java
index 57b59e95f2..2552c4540f 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/read/SchemaFetchMergeNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/read/SchemaFetchMergeNode.java
@@ -24,6 +24,8 @@ import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeId;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeType;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanVisitor;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
/** This class defines the scanned result merge task of schema fetcher. */
@@ -43,6 +45,11 @@ public class SchemaFetchMergeNode extends AbstractSchemaMergeNode {
PlanNodeType.SCHEMA_FETCH_MERGE.serialize(byteBuffer);
}
+ @Override
+ protected void serializeAttributes(DataOutputStream stream) throws IOException {
+ PlanNodeType.SCHEMA_FETCH_MERGE.serialize(stream);
+ }
+
public static PlanNode deserialize(ByteBuffer byteBuffer) {
PlanNodeId planNodeId = PlanNodeId.deserialize(byteBuffer);
return new SchemaFetchMergeNode(planNodeId);
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/read/SchemaFetchScanNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/read/SchemaFetchScanNode.java
index 7f0f0e9310..031fb7b2fb 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/read/SchemaFetchScanNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/read/SchemaFetchScanNode.java
@@ -31,6 +31,8 @@ import org.apache.iotdb.db.mpp.plan.planner.plan.node.source.SourceNode;
import com.google.common.collect.ImmutableList;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Collections;
import java.util.List;
@@ -88,6 +90,13 @@ public class SchemaFetchScanNode extends SourceNode {
patternTree.serialize(byteBuffer);
}
+ @Override
+ protected void serializeAttributes(DataOutputStream stream) throws IOException {
+ PlanNodeType.SCHEMA_FETCH_SCAN.serialize(stream);
+ storageGroup.serialize(stream);
+ patternTree.serialize(stream);
+ }
+
public static SchemaFetchScanNode deserialize(ByteBuffer byteBuffer) {
PartialPath storageGroup = (PartialPath) PathDeserializeUtil.deserialize(byteBuffer);
PathPatternTree patternTree = PathPatternTree.deserialize(byteBuffer);
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/read/SchemaQueryMergeNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/read/SchemaQueryMergeNode.java
index 53422d2e78..b12b67702e 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/read/SchemaQueryMergeNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/read/SchemaQueryMergeNode.java
@@ -23,6 +23,8 @@ import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeId;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeType;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanVisitor;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
public class SchemaQueryMergeNode extends AbstractSchemaMergeNode {
@@ -52,6 +54,11 @@ public class SchemaQueryMergeNode extends AbstractSchemaMergeNode {
PlanNodeType.SCHEMA_QUERY_MERGE.serialize(byteBuffer);
}
+ @Override
+ protected void serializeAttributes(DataOutputStream stream) throws IOException {
+ PlanNodeType.SCHEMA_QUERY_MERGE.serialize(stream);
+ }
+
public static SchemaQueryMergeNode deserialize(ByteBuffer byteBuffer) {
PlanNodeId id = PlanNodeId.deserialize(byteBuffer);
return new SchemaQueryMergeNode(id);
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/read/SchemaQueryOrderByHeatNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/read/SchemaQueryOrderByHeatNode.java
index 93df4f4bd6..e72a2e3dd7 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/read/SchemaQueryOrderByHeatNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/read/SchemaQueryOrderByHeatNode.java
@@ -25,6 +25,8 @@ import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeType;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanVisitor;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.MultiChildNode;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.List;
@@ -75,6 +77,11 @@ public class SchemaQueryOrderByHeatNode extends MultiChildNode {
PlanNodeType.SCHEMA_QUERY_ORDER_BY_HEAT.serialize(byteBuffer);
}
+ @Override
+ protected void serializeAttributes(DataOutputStream stream) throws IOException {
+ PlanNodeType.SCHEMA_QUERY_ORDER_BY_HEAT.serialize(stream);
+ }
+
public static SchemaQueryOrderByHeatNode deserialize(ByteBuffer byteBuffer) {
PlanNodeId planNodeId = PlanNodeId.deserialize(byteBuffer);
return new SchemaQueryOrderByHeatNode(planNodeId);
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/read/TimeSeriesCountNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/read/TimeSeriesCountNode.java
index dfe9146834..b98f3447e5 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/read/TimeSeriesCountNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/read/TimeSeriesCountNode.java
@@ -27,6 +27,8 @@ import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeId;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeType;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.List;
@@ -53,6 +55,13 @@ public class TimeSeriesCountNode extends SchemaQueryScanNode {
ReadWriteIOUtils.write(isPrefixPath, byteBuffer);
}
+ @Override
+ protected void serializeAttributes(DataOutputStream stream) throws IOException {
+ PlanNodeType.TIME_SERIES_COUNT.serialize(stream);
+ ReadWriteIOUtils.write(path.getFullPath(), stream);
+ ReadWriteIOUtils.write(isPrefixPath, stream);
+ }
+
public static PlanNode deserialize(ByteBuffer buffer) {
String fullPath = ReadWriteIOUtils.readString(buffer);
PartialPath path;
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/read/TimeSeriesSchemaScanNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/read/TimeSeriesSchemaScanNode.java
index 5f7ef53576..d319fbf6c3 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/read/TimeSeriesSchemaScanNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/read/TimeSeriesSchemaScanNode.java
@@ -26,6 +26,8 @@ import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeId;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeType;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.List;
import java.util.Objects;
@@ -69,6 +71,19 @@ public class TimeSeriesSchemaScanNode extends SchemaQueryScanNode {
ReadWriteIOUtils.write(isPrefixPath, byteBuffer);
}
+ @Override
+ protected void serializeAttributes(DataOutputStream stream) throws IOException {
+ PlanNodeType.TIME_SERIES_SCHEMA_SCAN.serialize(stream);
+ ReadWriteIOUtils.write(path.getFullPath(), stream);
+ ReadWriteIOUtils.write(key, stream);
+ ReadWriteIOUtils.write(value, stream);
+ ReadWriteIOUtils.write(limit, stream);
+ ReadWriteIOUtils.write(offset, stream);
+ ReadWriteIOUtils.write(orderByHeat, stream);
+ ReadWriteIOUtils.write(isContains, stream);
+ ReadWriteIOUtils.write(isPrefixPath, stream);
+ }
+
public static TimeSeriesSchemaScanNode deserialize(ByteBuffer byteBuffer) {
String fullPath = ReadWriteIOUtils.readString(byteBuffer);
PartialPath path;
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/write/AlterTimeSeriesNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/write/AlterTimeSeriesNode.java
index ed32f6b9cf..97ba1cec81 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/write/AlterTimeSeriesNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/write/AlterTimeSeriesNode.java
@@ -34,6 +34,8 @@ import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
import com.google.common.collect.ImmutableList;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.List;
@@ -253,6 +255,53 @@ public class AlterTimeSeriesNode extends WritePlanNode {
}
}
+ @Override
+ protected void serializeAttributes(DataOutputStream stream) throws IOException {
+ PlanNodeType.ALTER_TIME_SERIES.serialize(stream);
+ byte[] bytes = path.getFullPath().getBytes();
+ stream.write(bytes.length);
+ stream.write(bytes);
+ stream.write((byte) alterType.ordinal());
+
+ // alias
+ if (alias != null) {
+ stream.write((byte) 1);
+ ReadWriteIOUtils.write(alias, stream);
+ } else {
+ stream.write((byte) 0);
+ }
+
+ // alterMap
+ if (alterMap == null) {
+ stream.write((byte) -1);
+ } else if (alterMap.isEmpty()) {
+ stream.write((byte) 0);
+ } else {
+ stream.write((byte) 1);
+ ReadWriteIOUtils.write(alterMap, stream);
+ }
+
+ // tagsMap
+ if (tagsMap == null) {
+ stream.write((byte) -1);
+ } else if (tagsMap.isEmpty()) {
+ stream.write((byte) 0);
+ } else {
+ stream.write((byte) 1);
+ ReadWriteIOUtils.write(tagsMap, stream);
+ }
+
+ // attributesMap
+ if (attributesMap == null) {
+ stream.write((byte) -1);
+ } else if (attributesMap.isEmpty()) {
+ stream.write((byte) 0);
+ } else {
+ stream.write((byte) 1);
+ ReadWriteIOUtils.write(attributesMap, stream);
+ }
+ }
+
@Override
public boolean equals(Object o) {
if (this == o) {
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/write/CreateAlignedTimeSeriesNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/write/CreateAlignedTimeSeriesNode.java
index 28823935ad..0f68469de0 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/write/CreateAlignedTimeSeriesNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/write/CreateAlignedTimeSeriesNode.java
@@ -36,6 +36,8 @@ import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
import com.google.common.collect.ImmutableList;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
@@ -336,6 +338,71 @@ public class CreateAlignedTimeSeriesNode extends WritePlanNode {
}
}
+ @Override
+ protected void serializeAttributes(DataOutputStream stream) throws IOException {
+ PlanNodeType.CREATE_ALIGNED_TIME_SERIES.serialize(stream);
+ byte[] bytes = devicePath.getFullPath().getBytes();
+ stream.write(bytes.length);
+ stream.write(bytes);
+
+ // measurements
+ stream.write(measurements.size());
+ for (String measurement : measurements) {
+ ReadWriteIOUtils.write(measurement, stream);
+ }
+
+ // dataTypes
+ for (TSDataType dataType : dataTypes) {
+ stream.write((byte) dataType.ordinal());
+ }
+
+ // encodings
+ for (TSEncoding encoding : encodings) {
+ stream.write((byte) encoding.ordinal());
+ }
+
+ // compressors
+ for (CompressionType compressor : compressors) {
+ stream.write((byte) compressor.ordinal());
+ }
+
+ // alias
+ if (aliasList == null) {
+ stream.write((byte) -1);
+ } else if (aliasList.isEmpty()) {
+ stream.write((byte) 0);
+ } else {
+ stream.write((byte) 1);
+ for (String alias : aliasList) {
+ ReadWriteIOUtils.write(alias, stream);
+ }
+ }
+
+ // tags
+ if (tagsList == null) {
+ stream.write((byte) -1);
+ } else if (tagsList.isEmpty()) {
+ stream.write((byte) 0);
+ } else {
+ stream.write((byte) 1);
+ for (Map<String, String> tags : tagsList) {
+ ReadWriteIOUtils.write(tags, stream);
+ }
+ }
+
+ // attributes
+ if (attributesList == null) {
+ stream.write((byte) -1);
+ } else if (attributesList.isEmpty()) {
+ stream.write((byte) 0);
+ } else {
+ stream.write((byte) 1);
+ for (Map<String, String> attributes : attributesList) {
+ ReadWriteIOUtils.write(attributes, stream);
+ }
+ }
+ }
+
public int hashCode() {
return Objects.hash(
this.getPlanNodeId(),
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/write/CreateMultiTimeSeriesNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/write/CreateMultiTimeSeriesNode.java
index 3763a92df2..de12ab542e 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/write/CreateMultiTimeSeriesNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/write/CreateMultiTimeSeriesNode.java
@@ -33,6 +33,8 @@ import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.HashMap;
@@ -178,6 +180,17 @@ public class CreateMultiTimeSeriesNode extends WritePlanNode {
}
}
+ @Override
+ protected void serializeAttributes(DataOutputStream stream) throws IOException {
+ PlanNodeType.CREATE_MULTI_TIME_SERIES.serialize(stream);
+
+ stream.write(measurementGroupMap.size());
+ for (Map.Entry<PartialPath, MeasurementGroup> entry : measurementGroupMap.entrySet()) {
+ entry.getKey().serialize(stream);
+ entry.getValue().serialize(stream);
+ }
+ }
+
public int hashCode() {
return Objects.hash(this.getPlanNodeId(), measurementGroupMap);
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/write/CreateTimeSeriesNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/write/CreateTimeSeriesNode.java
index 5da2181aa4..46c124d8c1 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/write/CreateTimeSeriesNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/write/CreateTimeSeriesNode.java
@@ -36,6 +36,8 @@ import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
import com.google.common.collect.ImmutableList;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.HashMap;
@@ -273,6 +275,56 @@ public class CreateTimeSeriesNode extends WritePlanNode {
}
}
+ @Override
+ protected void serializeAttributes(DataOutputStream stream) throws IOException {
+ PlanNodeType.CREATE_TIME_SERIES.serialize(stream);
+
+ byte[] bytes = path.getFullPath().getBytes();
+ stream.write(bytes.length);
+ stream.write(bytes);
+ stream.write((byte) dataType.ordinal());
+ stream.write((byte) encoding.ordinal());
+ stream.write((byte) compressor.ordinal());
+
+ // alias
+ if (alias != null) {
+ stream.write((byte) 1);
+ ReadWriteIOUtils.write(alias, stream);
+ } else {
+ stream.write((byte) 0);
+ }
+
+ // props
+ if (props == null) {
+ stream.write((byte) -1);
+ } else if (props.isEmpty()) {
+ stream.write((byte) 0);
+ } else {
+ stream.write((byte) 1);
+ ReadWriteIOUtils.write(props, stream);
+ }
+
+ // tags
+ if (tags == null) {
+ stream.write((byte) -1);
+ } else if (tags.isEmpty()) {
+ stream.write((byte) 0);
+ } else {
+ stream.write((byte) 1);
+ ReadWriteIOUtils.write(tags, stream);
+ }
+
+ // attributes
+ if (attributes == null) {
+ stream.write((byte) -1);
+ } else if (attributes.isEmpty()) {
+ stream.write((byte) 0);
+ } else {
+ stream.write((byte) 1);
+ ReadWriteIOUtils.write(attributes, stream);
+ }
+ }
+
@Override
public <R, C> R accept(PlanVisitor<R, C> visitor, C schemaRegion) {
return visitor.visitCreateTimeSeries(this, schemaRegion);
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/write/DeleteTimeSeriesNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/write/DeleteTimeSeriesNode.java
index 454db8da22..a1bf8ac56a 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/write/DeleteTimeSeriesNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/write/DeleteTimeSeriesNode.java
@@ -29,6 +29,8 @@ import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeType;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanVisitor;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
@@ -89,6 +91,15 @@ public class DeleteTimeSeriesNode extends PlanNode implements IPartitionRelatedN
}
}
+ @Override
+ protected void serializeAttributes(DataOutputStream stream) throws IOException {
+ PlanNodeType.DELETE_TIMESERIES.serialize(stream);
+ ReadWriteIOUtils.write(pathList.size(), stream);
+ for (PartialPath path : pathList) {
+ path.serialize(stream);
+ }
+ }
+
public static DeleteTimeSeriesNode deserialize(ByteBuffer byteBuffer) {
int size = ReadWriteIOUtils.readInt(byteBuffer);
List<PartialPath> pathList = new ArrayList<>(size);
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/write/InvalidateSchemaCacheNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/write/InvalidateSchemaCacheNode.java
index 8af09b8a3d..c0a9f90102 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/write/InvalidateSchemaCacheNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/write/InvalidateSchemaCacheNode.java
@@ -27,6 +27,8 @@ import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeId;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeType;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
@@ -96,6 +98,20 @@ public class InvalidateSchemaCacheNode extends PlanNode {
}
}
+ @Override
+ protected void serializeAttributes(DataOutputStream stream) throws IOException {
+ PlanNodeType.INVALIDATE_SCHEMA_CACHE.serialize(stream);
+ queryId.serialize(stream);
+ ReadWriteIOUtils.write(pathList.size(), stream);
+ for (PartialPath path : pathList) {
+ path.serialize(stream);
+ }
+ ReadWriteIOUtils.write(storageGroups.size(), stream);
+ for (String storageGroup : storageGroups) {
+ ReadWriteIOUtils.write(storageGroup, stream);
+ }
+ }
+
public static InvalidateSchemaCacheNode deserialize(ByteBuffer byteBuffer) {
QueryId queryId = QueryId.deserialize(byteBuffer);
int size = ReadWriteIOUtils.readInt(byteBuffer);
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/write/MeasurementGroup.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/write/MeasurementGroup.java
index ca94bdacef..5c2c22a43d 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/write/MeasurementGroup.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/write/MeasurementGroup.java
@@ -24,6 +24,8 @@ import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
@@ -175,6 +177,65 @@ public class MeasurementGroup {
}
}
+ public void serialize(DataOutputStream stream) throws IOException {
+ // measurements
+ stream.write(measurements.size());
+ for (String measurement : measurements) {
+ ReadWriteIOUtils.write(measurement, stream);
+ }
+
+ // dataTypes
+ for (TSDataType dataType : dataTypes) {
+ stream.write((byte) dataType.ordinal());
+ }
+
+ // encodings
+ for (TSEncoding encoding : encodings) {
+ stream.write((byte) encoding.ordinal());
+ }
+
+ // compressors
+ for (CompressionType compressor : compressors) {
+ stream.write((byte) compressor.ordinal());
+ }
+
+ // alias
+ if (aliasList == null) {
+ stream.write((byte) -1);
+ } else if (aliasList.isEmpty()) {
+ stream.write((byte) 0);
+ } else {
+ stream.write((byte) 1);
+ for (String alias : aliasList) {
+ ReadWriteIOUtils.write(alias, stream);
+ }
+ }
+
+ // tags
+ if (tagsList == null) {
+ stream.write((byte) -1);
+ } else if (tagsList.isEmpty()) {
+ stream.write((byte) 0);
+ } else {
+ stream.write((byte) 1);
+ for (Map<String, String> tags : tagsList) {
+ ReadWriteIOUtils.write(tags, stream);
+ }
+ }
+
+ // attributes
+ if (attributesList == null) {
+ stream.write((byte) -1);
+ } else if (attributesList.isEmpty()) {
+ stream.write((byte) 0);
+ } else {
+ stream.write((byte) 1);
+ for (Map<String, String> attributes : attributesList) {
+ ReadWriteIOUtils.write(attributes, stream);
+ }
+ }
+ }
+
public void deserialize(ByteBuffer byteBuffer) {
measurements = new ArrayList<>();
int size = byteBuffer.getInt();
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/AggregationNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/AggregationNode.java
index f2286dc21c..f3ba16a795 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/AggregationNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/AggregationNode.java
@@ -29,6 +29,8 @@ import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
import javax.annotation.Nullable;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.HashMap;
@@ -143,6 +145,22 @@ public class AggregationNode extends MultiChildNode {
ReadWriteIOUtils.write(scanOrder.ordinal(), byteBuffer);
}
+ @Override
+ protected void serializeAttributes(DataOutputStream stream) throws IOException {
+ PlanNodeType.AGGREGATE.serialize(stream);
+ ReadWriteIOUtils.write(aggregationDescriptorList.size(), stream);
+ for (AggregationDescriptor aggregationDescriptor : aggregationDescriptorList) {
+ aggregationDescriptor.serialize(stream);
+ }
+ if (groupByTimeParameter == null) {
+ ReadWriteIOUtils.write((byte) 0, stream);
+ } else {
+ ReadWriteIOUtils.write((byte) 1, stream);
+ groupByTimeParameter.serialize(stream);
+ }
+ ReadWriteIOUtils.write(scanOrder.ordinal(), stream);
+ }
+
public static AggregationNode deserialize(ByteBuffer byteBuffer) {
int descriptorSize = ReadWriteIOUtils.readInt(byteBuffer);
List<AggregationDescriptor> aggregationDescriptorList = new ArrayList<>();
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/DeviceMergeNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/DeviceMergeNode.java
index f10b69a474..cdcfedab24 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/DeviceMergeNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/DeviceMergeNode.java
@@ -26,6 +26,8 @@ import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanVisitor;
import org.apache.iotdb.db.mpp.plan.statement.component.OrderBy;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
@@ -109,6 +111,17 @@ public class DeviceMergeNode extends MultiChildNode {
}
}
+ @Override
+ protected void serializeAttributes(DataOutputStream stream) throws IOException {
+ PlanNodeType.DEVICE_MERGE.serialize(stream);
+ ReadWriteIOUtils.write(mergeOrders.get(0).ordinal(), stream);
+ ReadWriteIOUtils.write(mergeOrders.get(1).ordinal(), stream);
+ ReadWriteIOUtils.write(devices.size(), stream);
+ for (String deviceName : devices) {
+ ReadWriteIOUtils.write(deviceName, stream);
+ }
+ }
+
public static DeviceMergeNode deserialize(ByteBuffer byteBuffer) {
List<OrderBy> mergeOrders = new ArrayList<>();
mergeOrders.add(OrderBy.values()[ReadWriteIOUtils.readInt(byteBuffer)]);
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/DeviceViewNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/DeviceViewNode.java
index a70252c55d..3a54be0b34 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/DeviceViewNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/DeviceViewNode.java
@@ -25,6 +25,8 @@ import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanVisitor;
import org.apache.iotdb.db.mpp.plan.statement.component.OrderBy;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.HashMap;
@@ -151,6 +153,29 @@ public class DeviceViewNode extends MultiChildNode {
}
}
+ @Override
+ protected void serializeAttributes(DataOutputStream stream) throws IOException {
+ PlanNodeType.DEVICE_VIEW.serialize(stream);
+ ReadWriteIOUtils.write(mergeOrders.get(0).ordinal(), stream);
+ ReadWriteIOUtils.write(mergeOrders.get(1).ordinal(), stream);
+ ReadWriteIOUtils.write(outputColumnNames.size(), stream);
+ for (String column : outputColumnNames) {
+ ReadWriteIOUtils.write(column, stream);
+ }
+ ReadWriteIOUtils.write(devices.size(), stream);
+ for (String deviceName : devices) {
+ ReadWriteIOUtils.write(deviceName, stream);
+ }
+ ReadWriteIOUtils.write(deviceToMeasurementIndexesMap.size(), stream);
+ for (Map.Entry<String, List<Integer>> entry : deviceToMeasurementIndexesMap.entrySet()) {
+ ReadWriteIOUtils.write(entry.getKey(), stream);
+ ReadWriteIOUtils.write(entry.getValue().size(), stream);
+ for (Integer index : entry.getValue()) {
+ ReadWriteIOUtils.write(index, stream);
+ }
+ }
+ }
+
public static DeviceViewNode deserialize(ByteBuffer byteBuffer) {
List<OrderBy> mergeOrders = new ArrayList<>();
mergeOrders.add(OrderBy.values()[ReadWriteIOUtils.readInt(byteBuffer)]);
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/ExchangeNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/ExchangeNode.java
index fe110473c9..780a0e629f 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/ExchangeNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/ExchangeNode.java
@@ -30,6 +30,8 @@ import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
import com.google.common.collect.ImmutableList;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
@@ -135,6 +137,19 @@ public class ExchangeNode extends PlanNode {
}
}
+ @Override
+ protected void serializeAttributes(DataOutputStream stream) throws IOException {
+ PlanNodeType.EXCHANGE.serialize(stream);
+ ReadWriteIOUtils.write(upstreamEndpoint.getIp(), stream);
+ ReadWriteIOUtils.write(upstreamEndpoint.getPort(), stream);
+ upstreamInstanceId.serialize(stream);
+ upstreamPlanNodeId.serialize(stream);
+ ReadWriteIOUtils.write(outputColumnNames.size(), stream);
+ for (String outputColumnName : outputColumnNames) {
+ ReadWriteIOUtils.write(outputColumnName, stream);
+ }
+ }
+
public PlanNode getChild() {
return child;
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/FillNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/FillNode.java
index eaf55ad0d6..c9111d4f25 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/FillNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/FillNode.java
@@ -26,6 +26,8 @@ import org.apache.iotdb.db.mpp.plan.planner.plan.parameter.FillDescriptor;
import com.google.common.collect.ImmutableList;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.List;
import java.util.Objects;
@@ -92,6 +94,12 @@ public class FillNode extends ProcessNode {
fillDescriptor.serialize(byteBuffer);
}
+ @Override
+ protected void serializeAttributes(DataOutputStream stream) throws IOException {
+ PlanNodeType.FILL.serialize(stream);
+ fillDescriptor.serialize(stream);
+ }
+
public static FillNode deserialize(ByteBuffer byteBuffer) {
FillDescriptor fillDescriptor = FillDescriptor.deserialize(byteBuffer);
PlanNodeId planNodeId = PlanNodeId.deserialize(byteBuffer);
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/FilterNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/FilterNode.java
index 3b9a1ec475..9e1fa43eb3 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/FilterNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/FilterNode.java
@@ -25,6 +25,8 @@ import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeType;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanVisitor;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.time.ZoneId;
import java.util.Objects;
@@ -76,6 +78,18 @@ public class FilterNode extends TransformNode {
ReadWriteIOUtils.write(zoneId.getId(), byteBuffer);
}
+ @Override
+ protected void serializeAttributes(DataOutputStream stream) throws IOException {
+ PlanNodeType.FILTER.serialize(stream);
+ ReadWriteIOUtils.write(outputExpressions.length, stream);
+ for (Expression expression : outputExpressions) {
+ Expression.serialize(expression, stream);
+ }
+ Expression.serialize(predicate, stream);
+ ReadWriteIOUtils.write(keepNull, stream);
+ ReadWriteIOUtils.write(zoneId.getId(), stream);
+ }
+
public static FilterNode deserialize(ByteBuffer byteBuffer) {
int outputExpressionsLength = ReadWriteIOUtils.readInt(byteBuffer);
Expression[] outputExpressions = new Expression[outputExpressionsLength];
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/FilterNullNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/FilterNullNode.java
index 7c76a3cfc9..535ed4fa95 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/FilterNullNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/FilterNullNode.java
@@ -28,6 +28,8 @@ import org.apache.iotdb.db.mpp.plan.statement.component.FilterNullPolicy;
import com.google.common.collect.ImmutableList;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.List;
import java.util.Objects;
@@ -100,6 +102,12 @@ public class FilterNullNode extends ProcessNode {
filterNullParameter.serialize(byteBuffer);
}
+ @Override
+ protected void serializeAttributes(DataOutputStream stream) throws IOException {
+ PlanNodeType.FILTER_NULL.serialize(stream);
+ filterNullParameter.serialize(stream);
+ }
+
public static FilterNullNode deserialize(ByteBuffer byteBuffer) {
FilterNullParameter filterNullParameter = FilterNullParameter.deserialize(byteBuffer);
PlanNodeId planNodeId = PlanNodeId.deserialize(byteBuffer);
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/GroupByLevelNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/GroupByLevelNode.java
index 34c5bc7708..2919877a55 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/GroupByLevelNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/GroupByLevelNode.java
@@ -30,6 +30,8 @@ import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
import javax.annotation.Nullable;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
@@ -143,6 +145,22 @@ public class GroupByLevelNode extends MultiChildNode {
ReadWriteIOUtils.write(scanOrder.ordinal(), byteBuffer);
}
+ @Override
+ protected void serializeAttributes(DataOutputStream stream) throws IOException {
+ PlanNodeType.GROUP_BY_LEVEL.serialize(stream);
+ ReadWriteIOUtils.write(groupByLevelDescriptors.size(), stream);
+ for (GroupByLevelDescriptor groupByLevelDescriptor : groupByLevelDescriptors) {
+ groupByLevelDescriptor.serialize(stream);
+ }
+ if (groupByTimeParameter == null) {
+ ReadWriteIOUtils.write((byte) 0, stream);
+ } else {
+ ReadWriteIOUtils.write((byte) 1, stream);
+ groupByTimeParameter.serialize(stream);
+ }
+ ReadWriteIOUtils.write(scanOrder.ordinal(), stream);
+ }
+
public static GroupByLevelNode deserialize(ByteBuffer byteBuffer) {
int descriptorSize = ReadWriteIOUtils.readInt(byteBuffer);
List<GroupByLevelDescriptor> groupByLevelDescriptors = new ArrayList<>();
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/LastQueryMergeNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/LastQueryMergeNode.java
index bb61f16b51..ae51fdf01e 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/LastQueryMergeNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/LastQueryMergeNode.java
@@ -28,6 +28,8 @@ import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
import javax.annotation.Nullable;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
@@ -114,6 +116,17 @@ public class LastQueryMergeNode extends MultiChildNode {
}
}
+ @Override
+ protected void serializeAttributes(DataOutputStream stream) throws IOException {
+ PlanNodeType.LAST_QUERY_MERGE.serialize(stream);
+ if (timeFilter == null) {
+ ReadWriteIOUtils.write((byte) 0, stream);
+ } else {
+ ReadWriteIOUtils.write((byte) 1, stream);
+ timeFilter.serialize(stream);
+ }
+ }
+
public static LastQueryMergeNode deserialize(ByteBuffer byteBuffer) {
Filter timeFilter = null;
if (ReadWriteIOUtils.readByte(byteBuffer) == 1) {
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/LimitNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/LimitNode.java
index 0df2d5097d..9d3badac0e 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/LimitNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/LimitNode.java
@@ -26,6 +26,8 @@ import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
import com.google.common.collect.ImmutableList;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.List;
import java.util.Objects;
@@ -95,6 +97,12 @@ public class LimitNode extends ProcessNode {
ReadWriteIOUtils.write(limit, byteBuffer);
}
+ @Override
+ protected void serializeAttributes(DataOutputStream stream) throws IOException {
+ PlanNodeType.LIMIT.serialize(stream);
+ ReadWriteIOUtils.write(limit, stream);
+ }
+
public static LimitNode deserialize(ByteBuffer byteBuffer) {
int limit = ReadWriteIOUtils.readInt(byteBuffer);
PlanNodeId planNodeId = PlanNodeId.deserialize(byteBuffer);
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/OffsetNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/OffsetNode.java
index c30574fde0..0407f6d946 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/OffsetNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/OffsetNode.java
@@ -26,6 +26,8 @@ import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
import com.google.common.collect.ImmutableList;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.List;
import java.util.Objects;
@@ -86,6 +88,12 @@ public class OffsetNode extends ProcessNode {
ReadWriteIOUtils.write(offset, byteBuffer);
}
+ @Override
+ protected void serializeAttributes(DataOutputStream stream) throws IOException {
+ PlanNodeType.OFFSET.serialize(stream);
+ ReadWriteIOUtils.write(offset, stream);
+ }
+
public static OffsetNode deserialize(ByteBuffer byteBuffer) {
int offset = ReadWriteIOUtils.readInt(byteBuffer);
PlanNodeId planNodeId = PlanNodeId.deserialize(byteBuffer);
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/ProjectNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/ProjectNode.java
index 29af95d572..3a9b5a7268 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/ProjectNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/ProjectNode.java
@@ -27,6 +27,8 @@ import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
import com.google.common.collect.ImmutableList;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
@@ -88,6 +90,15 @@ public class ProjectNode extends ProcessNode {
}
}
+ @Override
+ protected void serializeAttributes(DataOutputStream stream) throws IOException {
+ PlanNodeType.PROJECT.serialize(stream);
+ ReadWriteIOUtils.write(outputColumnNames.size(), stream);
+ for (String outputColumnName : outputColumnNames) {
+ ReadWriteIOUtils.write(outputColumnName, stream);
+ }
+ }
+
public static ProjectNode deserialize(ByteBuffer byteBuffer) {
int outputColumnNamesSize = ReadWriteIOUtils.readInt(byteBuffer);
List<String> outputColumnNames = new ArrayList<>();
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/SlidingWindowAggregationNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/SlidingWindowAggregationNode.java
index 9362e40227..0e860b335b 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/SlidingWindowAggregationNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/SlidingWindowAggregationNode.java
@@ -30,6 +30,8 @@ import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
import com.google.common.collect.ImmutableList;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
@@ -140,6 +142,22 @@ public class SlidingWindowAggregationNode extends ProcessNode {
ReadWriteIOUtils.write(scanOrder.ordinal(), byteBuffer);
}
+ @Override
+ protected void serializeAttributes(DataOutputStream stream) throws IOException {
+ PlanNodeType.SLIDING_WINDOW_AGGREGATION.serialize(stream);
+ ReadWriteIOUtils.write(aggregationDescriptorList.size(), stream);
+ for (AggregationDescriptor aggregationDescriptor : aggregationDescriptorList) {
+ aggregationDescriptor.serialize(stream);
+ }
+ if (groupByTimeParameter == null) {
+ ReadWriteIOUtils.write((byte) 0, stream);
+ } else {
+ ReadWriteIOUtils.write((byte) 1, stream);
+ groupByTimeParameter.serialize(stream);
+ }
+ ReadWriteIOUtils.write(scanOrder.ordinal(), stream);
+ }
+
public static SlidingWindowAggregationNode deserialize(ByteBuffer byteBuffer) {
int descriptorSize = ReadWriteIOUtils.readInt(byteBuffer);
List<AggregationDescriptor> aggregationDescriptorList = new ArrayList<>();
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/SortNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/SortNode.java
index b61f5ebfd4..509d7b9398 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/SortNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/SortNode.java
@@ -27,6 +27,8 @@ import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
import com.google.common.collect.ImmutableList;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.List;
import java.util.Objects;
@@ -91,6 +93,12 @@ public class SortNode extends ProcessNode {
ReadWriteIOUtils.write(sortOrder.ordinal(), byteBuffer);
}
+ @Override
+ protected void serializeAttributes(DataOutputStream stream) throws IOException {
+ PlanNodeType.SORT.serialize(stream);
+ ReadWriteIOUtils.write(sortOrder.ordinal(), stream);
+ }
+
public static SortNode deserialize(ByteBuffer byteBuffer) {
OrderBy orderBy = OrderBy.values()[ReadWriteIOUtils.readInt(byteBuffer)];
PlanNodeId planNodeId = PlanNodeId.deserialize(byteBuffer);
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/TimeJoinNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/TimeJoinNode.java
index f081270e2b..1e9fe6424b 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/TimeJoinNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/TimeJoinNode.java
@@ -25,6 +25,8 @@ import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanVisitor;
import org.apache.iotdb.db.mpp.plan.statement.component.OrderBy;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
@@ -95,6 +97,12 @@ public class TimeJoinNode extends MultiChildNode {
ReadWriteIOUtils.write(mergeOrder.ordinal(), byteBuffer);
}
+ @Override
+ protected void serializeAttributes(DataOutputStream stream) throws IOException {
+ PlanNodeType.TIME_JOIN.serialize(stream);
+ ReadWriteIOUtils.write(mergeOrder.ordinal(), stream);
+ }
+
public static TimeJoinNode deserialize(ByteBuffer byteBuffer) {
OrderBy mergeOrder = OrderBy.values()[ReadWriteIOUtils.readInt(byteBuffer)];
PlanNodeId planNodeId = PlanNodeId.deserialize(byteBuffer);
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/TransformNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/TransformNode.java
index 7f9870fc18..f01ba32dfa 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/TransformNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/TransformNode.java
@@ -28,6 +28,8 @@ import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
import com.google.common.collect.ImmutableList;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.time.ZoneId;
import java.util.ArrayList;
@@ -113,6 +115,17 @@ public class TransformNode extends ProcessNode {
ReadWriteIOUtils.write(zoneId.getId(), byteBuffer);
}
+ @Override
+ protected void serializeAttributes(DataOutputStream stream) throws IOException {
+ PlanNodeType.TRANSFORM.serialize(stream);
+ ReadWriteIOUtils.write(outputExpressions.length, stream);
+ for (Expression expression : outputExpressions) {
+ Expression.serialize(expression, stream);
+ }
+ ReadWriteIOUtils.write(keepNull, stream);
+ ReadWriteIOUtils.write(zoneId.getId(), stream);
+ }
+
public static TransformNode deserialize(ByteBuffer byteBuffer) {
int outputExpressionsLength = ReadWriteIOUtils.readInt(byteBuffer);
Expression[] outputExpressions = new Expression[outputExpressionsLength];
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/sink/FragmentSinkNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/sink/FragmentSinkNode.java
index dd8ce71ddb..d53f8b4e97 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/sink/FragmentSinkNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/sink/FragmentSinkNode.java
@@ -29,6 +29,8 @@ import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
import com.google.common.collect.ImmutableList;
import org.apache.commons.lang.Validate;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.List;
import java.util.Objects;
@@ -112,6 +114,15 @@ public class FragmentSinkNode extends SinkNode {
downStreamPlanNodeId.serialize(byteBuffer);
}
+ @Override
+ protected void serializeAttributes(DataOutputStream stream) throws IOException {
+ PlanNodeType.FRAGMENT_SINK.serialize(stream);
+ ReadWriteIOUtils.write(downStreamEndpoint.getIp(), stream);
+ ReadWriteIOUtils.write(downStreamEndpoint.getPort(), stream);
+ downStreamInstanceId.serialize(stream);
+ downStreamPlanNodeId.serialize(stream);
+ }
+
@Override
public void send() {}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/source/AlignedLastQueryScanNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/source/AlignedLastQueryScanNode.java
index 514d2485ed..abb80f86b4 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/source/AlignedLastQueryScanNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/source/AlignedLastQueryScanNode.java
@@ -30,6 +30,8 @@ import org.apache.iotdb.tsfile.read.filter.basic.Filter;
import com.google.common.collect.ImmutableList;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.List;
import java.util.Objects;
@@ -129,6 +131,12 @@ public class AlignedLastQueryScanNode extends SeriesSourceNode {
seriesPath.serialize(byteBuffer);
}
+ @Override
+ protected void serializeAttributes(DataOutputStream stream) throws IOException {
+ PlanNodeType.ALIGNED_LAST_QUERY_SCAN.serialize(stream);
+ seriesPath.serialize(stream);
+ }
+
public static AlignedLastQueryScanNode deserialize(ByteBuffer byteBuffer) {
AlignedPath partialPath = (AlignedPath) PathDeserializeUtil.deserialize(byteBuffer);
PlanNodeId planNodeId = PlanNodeId.deserialize(byteBuffer);
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/source/AlignedSeriesAggregationScanNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/source/AlignedSeriesAggregationScanNode.java
index 050597073b..c3f1bc2fc7 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/source/AlignedSeriesAggregationScanNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/source/AlignedSeriesAggregationScanNode.java
@@ -39,6 +39,8 @@ import com.google.common.collect.ImmutableList;
import javax.annotation.Nullable;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
@@ -189,6 +191,29 @@ public class AlignedSeriesAggregationScanNode extends SeriesAggregationSourceNod
}
}
+ @Override
+ protected void serializeAttributes(DataOutputStream stream) throws IOException {
+ PlanNodeType.ALIGNED_SERIES_AGGREGATE_SCAN.serialize(stream);
+ alignedPath.serialize(stream);
+ ReadWriteIOUtils.write(aggregationDescriptorList.size(), stream);
+ for (AggregationDescriptor aggregationDescriptor : aggregationDescriptorList) {
+ aggregationDescriptor.serialize(stream);
+ }
+ ReadWriteIOUtils.write(scanOrder.ordinal(), stream);
+ if (timeFilter == null) {
+ ReadWriteIOUtils.write((byte) 0, stream);
+ } else {
+ ReadWriteIOUtils.write((byte) 1, stream);
+ timeFilter.serialize(stream);
+ }
+ if (groupByTimeParameter == null) {
+ ReadWriteIOUtils.write((byte) 0, stream);
+ } else {
+ ReadWriteIOUtils.write((byte) 1, stream);
+ groupByTimeParameter.serialize(stream);
+ }
+ }
+
public static AlignedSeriesAggregationScanNode deserialize(ByteBuffer byteBuffer) {
AlignedPath alignedPath = (AlignedPath) PathDeserializeUtil.deserialize(byteBuffer);
int aggregateDescriptorSize = ReadWriteIOUtils.readInt(byteBuffer);
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/source/AlignedSeriesScanNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/source/AlignedSeriesScanNode.java
index ad8e6a488a..1af4949d50 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/source/AlignedSeriesScanNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/source/AlignedSeriesScanNode.java
@@ -37,6 +37,8 @@ import com.google.common.collect.ImmutableList;
import javax.annotation.Nullable;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
@@ -204,6 +206,27 @@ public class AlignedSeriesScanNode extends SeriesSourceNode {
ReadWriteIOUtils.write(offset, byteBuffer);
}
+ @Override
+ protected void serializeAttributes(DataOutputStream stream) throws IOException {
+ PlanNodeType.ALIGNED_SERIES_SCAN.serialize(stream);
+ alignedPath.serialize(stream);
+ ReadWriteIOUtils.write(scanOrder.ordinal(), stream);
+ if (timeFilter == null) {
+ ReadWriteIOUtils.write((byte) 0, stream);
+ } else {
+ ReadWriteIOUtils.write((byte) 1, stream);
+ timeFilter.serialize(stream);
+ }
+ if (valueFilter == null) {
+ ReadWriteIOUtils.write((byte) 0, stream);
+ } else {
+ ReadWriteIOUtils.write((byte) 1, stream);
+ valueFilter.serialize(stream);
+ }
+ ReadWriteIOUtils.write(limit, stream);
+ ReadWriteIOUtils.write(offset, stream);
+ }
+
public static AlignedSeriesScanNode deserialize(ByteBuffer byteBuffer) {
AlignedPath alignedPath = (AlignedPath) PathDeserializeUtil.deserialize(byteBuffer);
OrderBy scanOrder = OrderBy.values()[ReadWriteIOUtils.readInt(byteBuffer)];
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/source/LastQueryScanNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/source/LastQueryScanNode.java
index e1db6e1d6a..8edacb9f05 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/source/LastQueryScanNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/source/LastQueryScanNode.java
@@ -30,6 +30,8 @@ import org.apache.iotdb.tsfile.read.filter.basic.Filter;
import com.google.common.collect.ImmutableList;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.List;
import java.util.Objects;
@@ -135,6 +137,12 @@ public class LastQueryScanNode extends SeriesSourceNode {
seriesPath.serialize(byteBuffer);
}
+ @Override
+ protected void serializeAttributes(DataOutputStream stream) throws IOException {
+ PlanNodeType.LAST_QUERY_SCAN.serialize(stream);
+ seriesPath.serialize(stream);
+ }
+
public static LastQueryScanNode deserialize(ByteBuffer byteBuffer) {
MeasurementPath partialPath = (MeasurementPath) PathDeserializeUtil.deserialize(byteBuffer);
PlanNodeId planNodeId = PlanNodeId.deserialize(byteBuffer);
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/source/SeriesAggregationScanNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/source/SeriesAggregationScanNode.java
index 77b0eff6d8..cb1bfdbb56 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/source/SeriesAggregationScanNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/source/SeriesAggregationScanNode.java
@@ -38,6 +38,8 @@ import com.google.common.collect.ImmutableList;
import javax.annotation.Nullable;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
@@ -200,6 +202,29 @@ public class SeriesAggregationScanNode extends SeriesAggregationSourceNode {
}
}
+ @Override
+ protected void serializeAttributes(DataOutputStream stream) throws IOException {
+ PlanNodeType.SERIES_AGGREGATE_SCAN.serialize(stream);
+ seriesPath.serialize(stream);
+ ReadWriteIOUtils.write(aggregationDescriptorList.size(), stream);
+ for (AggregationDescriptor aggregationDescriptor : aggregationDescriptorList) {
+ aggregationDescriptor.serialize(stream);
+ }
+ ReadWriteIOUtils.write(scanOrder.ordinal(), stream);
+ if (timeFilter == null) {
+ ReadWriteIOUtils.write((byte) 0, stream);
+ } else {
+ ReadWriteIOUtils.write((byte) 1, stream);
+ timeFilter.serialize(stream);
+ }
+ if (groupByTimeParameter == null) {
+ ReadWriteIOUtils.write((byte) 0, stream);
+ } else {
+ ReadWriteIOUtils.write((byte) 1, stream);
+ groupByTimeParameter.serialize(stream);
+ }
+ }
+
public static SeriesAggregationScanNode deserialize(ByteBuffer byteBuffer) {
MeasurementPath partialPath = (MeasurementPath) PathDeserializeUtil.deserialize(byteBuffer);
int aggregateDescriptorSize = ReadWriteIOUtils.readInt(byteBuffer);
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/source/SeriesScanNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/source/SeriesScanNode.java
index 96b557029f..c919512066 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/source/SeriesScanNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/source/SeriesScanNode.java
@@ -35,6 +35,8 @@ import com.google.common.collect.ImmutableList;
import javax.annotation.Nullable;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.List;
import java.util.Objects;
@@ -215,6 +217,27 @@ public class SeriesScanNode extends SeriesSourceNode {
ReadWriteIOUtils.write(offset, byteBuffer);
}
+ @Override
+ protected void serializeAttributes(DataOutputStream stream) throws IOException {
+ PlanNodeType.SERIES_SCAN.serialize(stream);
+ seriesPath.serialize(stream);
+ ReadWriteIOUtils.write(scanOrder.ordinal(), stream);
+ if (timeFilter == null) {
+ ReadWriteIOUtils.write((byte) 0, stream);
+ } else {
+ ReadWriteIOUtils.write((byte) 1, stream);
+ timeFilter.serialize(stream);
+ }
+ if (valueFilter == null) {
+ ReadWriteIOUtils.write((byte) 0, stream);
+ } else {
+ ReadWriteIOUtils.write((byte) 1, stream);
+ valueFilter.serialize(stream);
+ }
+ ReadWriteIOUtils.write(limit, stream);
+ ReadWriteIOUtils.write(offset, stream);
+ }
+
public static SeriesScanNode deserialize(ByteBuffer byteBuffer) {
MeasurementPath partialPath = (MeasurementPath) PathDeserializeUtil.deserialize(byteBuffer);
OrderBy scanOrder = OrderBy.values()[ReadWriteIOUtils.readInt(byteBuffer)];
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/write/DeleteDataNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/write/DeleteDataNode.java
index 16d6746ff9..bc5efc0d10 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/write/DeleteDataNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/write/DeleteDataNode.java
@@ -33,6 +33,8 @@ import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanVisitor;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.WritePlanNode;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collections;
@@ -118,6 +120,17 @@ public class DeleteDataNode extends WritePlanNode {
ReadWriteIOUtils.write(deleteEndTime, byteBuffer);
}
+ @Override
+ protected void serializeAttributes(DataOutputStream stream) throws IOException {
+ PlanNodeType.DELETE_DATA.serialize(stream);
+ ReadWriteIOUtils.write(pathList.size(), stream);
+ for (PartialPath path : pathList) {
+ path.serialize(stream);
+ }
+ ReadWriteIOUtils.write(deleteStartTime, stream);
+ ReadWriteIOUtils.write(deleteEndTime, stream);
+ }
+
public static DeleteDataNode deserialize(ByteBuffer byteBuffer) {
int size = ReadWriteIOUtils.readInt(byteBuffer);
List<PartialPath> pathList = new ArrayList<>(size);
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/write/InsertMultiTabletsNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/write/InsertMultiTabletsNode.java
index aff9170d6e..4b9f269302 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/write/InsertMultiTabletsNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/write/InsertMultiTabletsNode.java
@@ -31,7 +31,10 @@ import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanVisitor;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.WritePlanNode;
import org.apache.iotdb.tsfile.exception.NotImplementedException;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.HashMap;
@@ -267,13 +270,27 @@ public class InsertMultiTabletsNode extends InsertNode implements BatchInsertNod
protected void serializeAttributes(ByteBuffer byteBuffer) {
PlanNodeType.INSERT_MULTI_TABLET.serialize(byteBuffer);
- byteBuffer.putInt(insertTabletNodeList.size());
+ ReadWriteIOUtils.write(insertTabletNodeList.size(), byteBuffer);
for (InsertTabletNode node : insertTabletNodeList) {
node.subSerialize(byteBuffer);
}
for (Integer index : parentInsertTabletNodeIndexList) {
- byteBuffer.putInt(index);
+ ReadWriteIOUtils.write(index, byteBuffer);
+ }
+ }
+
+ @Override
+ protected void serializeAttributes(DataOutputStream stream) throws IOException {
+ PlanNodeType.INSERT_MULTI_TABLET.serialize(stream);
+
+ ReadWriteIOUtils.write(insertTabletNodeList.size(), stream);
+
+ for (InsertTabletNode node : insertTabletNodeList) {
+ node.subSerialize(stream);
+ }
+ for (Integer index : parentInsertTabletNodeIndexList) {
+ ReadWriteIOUtils.write(index, stream);
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/write/InsertNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/write/InsertNode.java
index 4050db50fb..471a6df7e4 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/write/InsertNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/write/InsertNode.java
@@ -31,9 +31,14 @@ import org.apache.iotdb.db.wal.buffer.IWALByteBufferView;
import org.apache.iotdb.db.wal.utils.WALWriteUtils;
import org.apache.iotdb.tsfile.exception.NotImplementedException;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.utils.PublicBAOS;
import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import java.io.DataInputStream;
+import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Arrays;
@@ -44,6 +49,8 @@ import java.util.Objects;
import java.util.stream.Collectors;
public abstract class InsertNode extends WritePlanNode implements IConsensusRequest {
+
+ private final Logger logger = LoggerFactory.getLogger(InsertNode.class);
/** this insert node doesn't need to participate in multi-leader consensus */
public static final long NO_CONSENSUS_INDEX = -1;
/** no multi-leader consensus, all insert nodes can be safely deleted */
@@ -181,11 +188,29 @@ public abstract class InsertNode extends WritePlanNode implements IConsensusRequ
getPlanNodeId().serialize(buffer);
}
+ @Override
+ public ByteBuffer serializeToByteBuffer() {
+ try (PublicBAOS byteArrayOutputStream = new PublicBAOS();
+ DataOutputStream outputStream = new DataOutputStream(byteArrayOutputStream)) {
+ serializeAttributes(outputStream);
+ getPlanNodeId().serialize(outputStream);
+ return ByteBuffer.wrap(byteArrayOutputStream.getBuf(), 0, byteArrayOutputStream.size());
+ } catch (IOException e) {
+ logger.error("Unexpected error occurs when serializing this InsertNode.", e);
+ throw new RuntimeException(e);
+ }
+ }
+
@Override
protected void serializeAttributes(ByteBuffer byteBuffer) {
throw new NotImplementedException("serializeAttributes of InsertNode is not implemented");
}
+ @Override
+ protected void serializeAttributes(DataOutputStream stream) throws IOException {
+ throw new NotImplementedException("serializeAttributes of InsertNode is not implemented");
+ }
+
// region Serialization methods for WAL
/** Serialized size of measurement schemas, ignoring failed time series */
protected int serializeMeasurementSchemasSize() {
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/write/InsertRowNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/write/InsertRowNode.java
index 8e40d44869..d5b75c05f0 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/write/InsertRowNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/write/InsertRowNode.java
@@ -52,6 +52,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.DataInputStream;
+import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Arrays;
@@ -260,24 +261,45 @@ public class InsertRowNode extends InsertNode implements WALEntryValue {
subSerialize(byteBuffer);
}
+ @Override
+ protected void serializeAttributes(DataOutputStream stream) throws IOException {
+ PlanNodeType.INSERT_ROW.serialize(stream);
+ subSerialize(stream);
+ }
+
void subSerialize(ByteBuffer buffer) {
- buffer.putLong(time);
+ ReadWriteIOUtils.write(time, buffer);
ReadWriteIOUtils.write(devicePath.getFullPath(), buffer);
serializeMeasurementsAndValues(buffer);
}
+ void subSerialize(DataOutputStream stream) throws IOException {
+ ReadWriteIOUtils.write(time, stream);
+ ReadWriteIOUtils.write(devicePath.getFullPath(), stream);
+ serializeMeasurementsAndValues(stream);
+ }
+
/** Serialize measurements and values, ignoring failed time series */
void serializeMeasurementsAndValues(ByteBuffer buffer) {
- buffer.putInt(measurements.length - getFailedMeasurementNumber());
+ ReadWriteIOUtils.write(measurements.length - getFailedMeasurementNumber(), buffer);
serializeMeasurementsOrSchemas(buffer);
putDataTypesAndValues(buffer);
- buffer.put((byte) (isNeedInferType ? 1 : 0));
- buffer.put((byte) (isAligned ? 1 : 0));
+ ReadWriteIOUtils.write((byte) (isNeedInferType ? 1 : 0), buffer);
+ ReadWriteIOUtils.write((byte) (isAligned ? 1 : 0), buffer);
+ }
+
+ /** Serialize measurements and values, ignoring failed time series */
+ void serializeMeasurementsAndValues(DataOutputStream stream) throws IOException {
+ ReadWriteIOUtils.write(measurements.length - getFailedMeasurementNumber(), stream);
+ serializeMeasurementsOrSchemas(stream);
+ putDataTypesAndValues(stream);
+ ReadWriteIOUtils.write((byte) (isNeedInferType ? 1 : 0), stream);
+ ReadWriteIOUtils.write((byte) (isAligned ? 1 : 0), stream);
}
/** Serialize measurements or measurement schemas, ignoring failed time series */
private void serializeMeasurementsOrSchemas(ByteBuffer buffer) {
- buffer.put((byte) (measurementSchemas != null ? 1 : 0));
+ ReadWriteIOUtils.write((byte) (measurementSchemas != null ? 1 : 0), buffer);
for (int i = 0; i < measurements.length; i++) {
// ignore failed partial insert
if (measurements[i] == null) {
@@ -292,6 +314,23 @@ public class InsertRowNode extends InsertNode implements WALEntryValue {
}
}
+ /** Serialize measurements or measurement schemas, ignoring failed time series */
+ private void serializeMeasurementsOrSchemas(DataOutputStream stream) throws IOException {
+ ReadWriteIOUtils.write((byte) (measurementSchemas != null ? 1 : 0), stream);
+ for (int i = 0; i < measurements.length; i++) {
+ // ignore failed partial insert
+ if (measurements[i] == null) {
+ continue;
+ }
+ // serialize measurement schemas when exist
+ if (measurementSchemas != null) {
+ measurementSchemas[i].serializeTo(stream);
+ } else {
+ ReadWriteIOUtils.write(measurements[i], stream);
+ }
+ }
+ }
+
/** Serialize data types and values, ignoring failed time series */
private void putDataTypesAndValues(ByteBuffer buffer) {
for (int i = 0; i < values.length; i++) {
@@ -337,6 +376,51 @@ public class InsertRowNode extends InsertNode implements WALEntryValue {
}
}
+ /** Serialize data types and values, ignoring failed time series */
+ private void putDataTypesAndValues(DataOutputStream stream) throws IOException {
+ for (int i = 0; i < values.length; i++) {
+ // ignore failed partial insert
+ if (measurements[i] == null) {
+ continue;
+ }
+ // serialize null value
+ if (values[i] == null) {
+ ReadWriteIOUtils.write(TYPE_NULL, stream);
+ continue;
+ }
+ // types are not determined, the situation mainly occurs when the plan uses string values
+ // and is forwarded to other nodes
+ if (isNeedInferType) {
+ ReadWriteIOUtils.write(TYPE_RAW_STRING, stream);
+ ReadWriteIOUtils.write(values[i].toString(), stream);
+ } else {
+ ReadWriteIOUtils.write(dataTypes[i], stream);
+ switch (dataTypes[i]) {
+ case BOOLEAN:
+ ReadWriteIOUtils.write((Boolean) values[i], stream);
+ break;
+ case INT32:
+ ReadWriteIOUtils.write((Integer) values[i], stream);
+ break;
+ case INT64:
+ ReadWriteIOUtils.write((Long) values[i], stream);
+ break;
+ case FLOAT:
+ ReadWriteIOUtils.write((Float) values[i], stream);
+ break;
+ case DOUBLE:
+ ReadWriteIOUtils.write((Double) values[i], stream);
+ break;
+ case TEXT:
+ ReadWriteIOUtils.write((Binary) values[i], stream);
+ break;
+ default:
+ throw new RuntimeException("Unsupported data type:" + dataTypes[i]);
+ }
+ }
+ }
+ }
+
public static InsertRowNode deserialize(ByteBuffer byteBuffer) {
// TODO: (xingtanzjr) remove placeholder
InsertRowNode insertNode = new InsertRowNode(new PlanNodeId(""));
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/write/InsertRowsNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/write/InsertRowsNode.java
index b130ccc09d..9b3d405d8e 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/write/InsertRowsNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/write/InsertRowsNode.java
@@ -32,7 +32,10 @@ import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanVisitor;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.WritePlanNode;
import org.apache.iotdb.tsfile.exception.NotImplementedException;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.HashMap;
@@ -228,13 +231,27 @@ public class InsertRowsNode extends InsertNode implements BatchInsertNode {
protected void serializeAttributes(ByteBuffer byteBuffer) {
PlanNodeType.INSERT_ROWS.serialize(byteBuffer);
- byteBuffer.putInt(insertRowNodeList.size());
+ ReadWriteIOUtils.write(insertRowNodeList.size(), byteBuffer);
for (InsertRowNode node : insertRowNodeList) {
node.subSerialize(byteBuffer);
}
for (Integer index : insertRowNodeIndexList) {
- byteBuffer.putInt(index);
+ ReadWriteIOUtils.write(index, byteBuffer);
+ }
+ }
+
+ @Override
+ protected void serializeAttributes(DataOutputStream stream) throws IOException {
+ PlanNodeType.INSERT_ROWS.serialize(stream);
+
+ ReadWriteIOUtils.write(insertRowNodeList.size(), stream);
+
+ for (InsertRowNode node : insertRowNodeList) {
+ node.subSerialize(stream);
+ }
+ for (Integer index : insertRowNodeIndexList) {
+ ReadWriteIOUtils.write(index, stream);
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/write/InsertRowsOfOneDeviceNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/write/InsertRowsOfOneDeviceNode.java
index df87d6ac9d..58c9aeffd9 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/write/InsertRowsOfOneDeviceNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/write/InsertRowsOfOneDeviceNode.java
@@ -35,6 +35,8 @@ import org.apache.iotdb.tsfile.exception.NotImplementedException;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
@@ -247,14 +249,30 @@ public class InsertRowsOfOneDeviceNode extends InsertNode implements BatchInsert
PlanNodeType.INSERT_ROWS_OF_ONE_DEVICE.serialize(byteBuffer);
ReadWriteIOUtils.write(devicePath.getFullPath(), byteBuffer);
- byteBuffer.putInt(insertRowNodeList.size());
+ ReadWriteIOUtils.write(insertRowNodeList.size(), byteBuffer);
for (InsertRowNode node : insertRowNodeList) {
- byteBuffer.putLong(node.getTime());
+ ReadWriteIOUtils.write(node.getTime(), byteBuffer);
node.serializeMeasurementsAndValues(byteBuffer);
}
for (Integer index : insertRowNodeIndexList) {
- byteBuffer.putInt(index);
+ ReadWriteIOUtils.write(index, byteBuffer);
+ }
+ }
+
+ @Override
+ protected void serializeAttributes(DataOutputStream stream) throws IOException {
+ PlanNodeType.INSERT_ROWS_OF_ONE_DEVICE.serialize(stream);
+ ReadWriteIOUtils.write(devicePath.getFullPath(), stream);
+
+ ReadWriteIOUtils.write(insertRowNodeList.size(), stream);
+
+ for (InsertRowNode node : insertRowNodeList) {
+ ReadWriteIOUtils.write(node.getTime(), stream);
+ node.serializeMeasurementsAndValues(stream);
+ }
+ for (Integer index : insertRowNodeIndexList) {
+ ReadWriteIOUtils.write(index, stream);
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/write/InsertTabletNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/write/InsertTabletNode.java
index 21422d124d..db9baaf74c 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/write/InsertTabletNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/write/InsertTabletNode.java
@@ -48,6 +48,7 @@ import org.apache.iotdb.tsfile.utils.TsPrimitiveType;
import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
import java.io.DataInputStream;
+import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
@@ -356,6 +357,12 @@ public class InsertTabletNode extends InsertNode implements WALEntryValue {
subSerialize(byteBuffer);
}
+ @Override
+ protected void serializeAttributes(DataOutputStream stream) throws IOException {
+ PlanNodeType.INSERT_TABLET.serialize(stream);
+ subSerialize(stream);
+ }
+
void subSerialize(ByteBuffer buffer) {
ReadWriteIOUtils.write(devicePath.getFullPath(), buffer);
writeMeasurementsOrSchemas(buffer);
@@ -363,14 +370,24 @@ public class InsertTabletNode extends InsertNode implements WALEntryValue {
writeTimes(buffer);
writeBitMaps(buffer);
writeValues(buffer);
- buffer.put((byte) (isAligned ? 1 : 0));
+ ReadWriteIOUtils.write((byte) (isAligned ? 1 : 0), buffer);
+ }
+
+ void subSerialize(DataOutputStream stream) throws IOException {
+ ReadWriteIOUtils.write(devicePath.getFullPath(), stream);
+ writeMeasurementsOrSchemas(stream);
+ writeDataTypes(stream);
+ writeTimes(stream);
+ writeBitMaps(stream);
+ writeValues(stream);
+ ReadWriteIOUtils.write((byte) (isAligned ? 1 : 0), stream);
}
/** Serialize measurements or measurement schemas, ignoring failed time series */
private void writeMeasurementsOrSchemas(ByteBuffer buffer) {
- buffer.putInt(measurements.length - getFailedMeasurementNumber());
+ ReadWriteIOUtils.write(measurements.length - getFailedMeasurementNumber(), buffer);
+ ReadWriteIOUtils.write((byte) (measurementSchemas != null ? 1 : 0), buffer);
- buffer.put((byte) (measurementSchemas != null ? 1 : 0));
for (int i = 0; i < measurements.length; i++) {
// ignore failed partial insert
if (measurements[i] == null) {
@@ -385,6 +402,25 @@ public class InsertTabletNode extends InsertNode implements WALEntryValue {
}
}
+ /** Serialize measurements or measurement schemas, ignoring failed time series */
+ private void writeMeasurementsOrSchemas(DataOutputStream stream) throws IOException {
+ ReadWriteIOUtils.write(measurements.length - getFailedMeasurementNumber(), stream);
+ ReadWriteIOUtils.write((byte) (measurementSchemas != null ? 1 : 0), stream);
+
+ for (int i = 0; i < measurements.length; i++) {
+ // ignore failed partial insert
+ if (measurements[i] == null) {
+ continue;
+ }
+ // serialize measurement schemas when exist
+ if (measurementSchemas != null) {
+ measurementSchemas[i].serializeTo(stream);
+ } else {
+ ReadWriteIOUtils.write(measurements[i], stream);
+ }
+ }
+ }
+
/** Serialize data types, ignoring failed time series */
private void writeDataTypes(ByteBuffer buffer) {
for (int i = 0; i < dataTypes.length; i++) {
@@ -396,16 +432,34 @@ public class InsertTabletNode extends InsertNode implements WALEntryValue {
}
}
+ /** Serialize data types, ignoring failed time series */
+ private void writeDataTypes(DataOutputStream stream) throws IOException {
+ for (int i = 0; i < dataTypes.length; i++) {
+ // ignore failed partial insert
+ if (measurements[i] == null) {
+ continue;
+ }
+ dataTypes[i].serializeTo(stream);
+ }
+ }
+
private void writeTimes(ByteBuffer buffer) {
- buffer.putInt(rowCount);
+ ReadWriteIOUtils.write(rowCount, buffer);
for (long time : times) {
- buffer.putLong(time);
+ ReadWriteIOUtils.write(time, buffer);
+ }
+ }
+
+ private void writeTimes(DataOutputStream stream) throws IOException {
+ ReadWriteIOUtils.write(rowCount, stream);
+ for (long time : times) {
+ ReadWriteIOUtils.write(time, stream);
}
}
/** Serialize bitmaps, ignoring failed time series */
private void writeBitMaps(ByteBuffer buffer) {
- buffer.put(BytesUtils.boolToByte(bitMaps != null));
+ ReadWriteIOUtils.write(BytesUtils.boolToByte(bitMaps != null), buffer);
if (bitMaps != null) {
for (int i = 0; i < bitMaps.length; i++) {
// ignore failed partial insert
@@ -414,15 +468,35 @@ public class InsertTabletNode extends InsertNode implements WALEntryValue {
}
if (bitMaps[i] == null) {
- buffer.put(BytesUtils.boolToByte(false));
+ ReadWriteIOUtils.write(BytesUtils.boolToByte(false), buffer);
} else {
- buffer.put(BytesUtils.boolToByte(true));
+ ReadWriteIOUtils.write(BytesUtils.boolToByte(true), buffer);
buffer.put(bitMaps[i].getByteArray());
}
}
}
}
+ /** Serialize bitmaps, ignoring failed time series */
+ private void writeBitMaps(DataOutputStream stream) throws IOException {
+ ReadWriteIOUtils.write(BytesUtils.boolToByte(bitMaps != null), stream);
+ if (bitMaps != null) {
+ for (int i = 0; i < bitMaps.length; i++) {
+ // ignore failed partial insert
+ if (measurements[i] == null) {
+ continue;
+ }
+
+ if (bitMaps[i] == null) {
+ ReadWriteIOUtils.write(BytesUtils.boolToByte(false), stream);
+ } else {
+ ReadWriteIOUtils.write(BytesUtils.boolToByte(true), stream);
+ stream.write(bitMaps[i].getByteArray());
+ }
+ }
+ }
+ }
+
/** Serialize values, ignoring failed time series */
private void writeValues(ByteBuffer buffer) {
for (int i = 0; i < columns.length; i++) {
@@ -434,43 +508,97 @@ public class InsertTabletNode extends InsertNode implements WALEntryValue {
}
}
+ /** Serialize values, ignoring failed time series */
+ private void writeValues(DataOutputStream stream) throws IOException {
+ for (int i = 0; i < columns.length; i++) {
+ // ignore failed partial insert
+ if (measurements[i] == null) {
+ continue;
+ }
+ serializeColumn(dataTypes[i], columns[i], stream);
+ }
+ }
+
private void serializeColumn(TSDataType dataType, Object column, ByteBuffer buffer) {
switch (dataType) {
case INT32:
int[] intValues = (int[]) column;
for (int j = 0; j < rowCount; j++) {
- buffer.putInt(intValues[j]);
+ ReadWriteIOUtils.write(intValues[j], buffer);
}
break;
case INT64:
long[] longValues = (long[]) column;
for (int j = 0; j < rowCount; j++) {
- buffer.putLong(longValues[j]);
+ ReadWriteIOUtils.write(longValues[j], buffer);
}
break;
case FLOAT:
float[] floatValues = (float[]) column;
for (int j = 0; j < rowCount; j++) {
- buffer.putFloat(floatValues[j]);
+ ReadWriteIOUtils.write(floatValues[j], buffer);
}
break;
case DOUBLE:
double[] doubleValues = (double[]) column;
for (int j = 0; j < rowCount; j++) {
- buffer.putDouble(doubleValues[j]);
+ ReadWriteIOUtils.write(doubleValues[j], buffer);
}
break;
case BOOLEAN:
boolean[] boolValues = (boolean[]) column;
for (int j = 0; j < rowCount; j++) {
- buffer.put(BytesUtils.boolToByte(boolValues[j]));
+ ReadWriteIOUtils.write(BytesUtils.boolToByte(boolValues[j]), buffer);
}
break;
case TEXT:
Binary[] binaryValues = (Binary[]) column;
for (int j = 0; j < rowCount; j++) {
- buffer.putInt(binaryValues[j].getLength());
- buffer.put(binaryValues[j].getValues());
+ ReadWriteIOUtils.write(binaryValues[j], buffer);
+ }
+ break;
+ default:
+ throw new UnSupportedDataTypeException(String.format(DATATYPE_UNSUPPORTED, dataType));
+ }
+ }
+
+ private void serializeColumn(TSDataType dataType, Object column, DataOutputStream stream)
+ throws IOException {
+ switch (dataType) {
+ case INT32:
+ int[] intValues = (int[]) column;
+ for (int j = 0; j < rowCount; j++) {
+ ReadWriteIOUtils.write(intValues[j], stream);
+ }
+ break;
+ case INT64:
+ long[] longValues = (long[]) column;
+ for (int j = 0; j < rowCount; j++) {
+ ReadWriteIOUtils.write(longValues[j], stream);
+ }
+ break;
+ case FLOAT:
+ float[] floatValues = (float[]) column;
+ for (int j = 0; j < rowCount; j++) {
+ ReadWriteIOUtils.write(floatValues[j], stream);
+ }
+ break;
+ case DOUBLE:
+ double[] doubleValues = (double[]) column;
+ for (int j = 0; j < rowCount; j++) {
+ ReadWriteIOUtils.write(doubleValues[j], stream);
+ }
+ break;
+ case BOOLEAN:
+ boolean[] boolValues = (boolean[]) column;
+ for (int j = 0; j < rowCount; j++) {
+ ReadWriteIOUtils.write(BytesUtils.boolToByte(boolValues[j]), stream);
+ }
+ break;
+ case TEXT:
+ Binary[] binaryValues = (Binary[]) column;
+ for (int j = 0; j < rowCount; j++) {
+ ReadWriteIOUtils.write(binaryValues[j], stream);
}
break;
default:
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/parameter/AggregationDescriptor.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/parameter/AggregationDescriptor.java
index f97add0369..d38188562f 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/parameter/AggregationDescriptor.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/parameter/AggregationDescriptor.java
@@ -23,6 +23,8 @@ import org.apache.iotdb.db.mpp.plan.expression.Expression;
import org.apache.iotdb.db.query.aggregation.AggregationType;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collections;
@@ -190,6 +192,15 @@ public class AggregationDescriptor {
}
}
+ public void serialize(DataOutputStream stream) throws IOException {
+ ReadWriteIOUtils.write(aggregationType.ordinal(), stream);
+ step.serialize(stream);
+ ReadWriteIOUtils.write(inputExpressions.size(), stream);
+ for (Expression expression : inputExpressions) {
+ Expression.serialize(expression, stream);
+ }
+ }
+
public static AggregationDescriptor deserialize(ByteBuffer byteBuffer) {
AggregationType aggregationType =
AggregationType.values()[ReadWriteIOUtils.readInt(byteBuffer)];
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/parameter/AggregationStep.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/parameter/AggregationStep.java
index 61883cb413..9e7c3e701a 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/parameter/AggregationStep.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/parameter/AggregationStep.java
@@ -21,6 +21,8 @@ package org.apache.iotdb.db.mpp.plan.planner.plan.parameter;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
/**
@@ -82,6 +84,11 @@ public enum AggregationStep {
ReadWriteIOUtils.write(isOutputPartial(), byteBuffer);
}
+ public void serialize(DataOutputStream stream) throws IOException {
+ ReadWriteIOUtils.write(isInputRaw(), stream);
+ ReadWriteIOUtils.write(isOutputPartial(), stream);
+ }
+
public static AggregationStep deserialize(ByteBuffer byteBuffer) {
boolean isInputRaw = ReadWriteIOUtils.readBool(byteBuffer);
boolean isOutputPartial = ReadWriteIOUtils.readBool(byteBuffer);
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/parameter/FillDescriptor.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/parameter/FillDescriptor.java
index d695ab5c35..72caf427e3 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/parameter/FillDescriptor.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/parameter/FillDescriptor.java
@@ -23,6 +23,8 @@ import org.apache.iotdb.db.mpp.plan.statement.component.FillPolicy;
import org.apache.iotdb.db.mpp.plan.statement.literal.Literal;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Objects;
@@ -50,6 +52,13 @@ public class FillDescriptor {
}
}
+ public void serialize(DataOutputStream stream) throws IOException {
+ ReadWriteIOUtils.write(fillPolicy.ordinal(), stream);
+ if (fillPolicy == FillPolicy.VALUE) {
+ fillValue.serialize(stream);
+ }
+ }
+
public static FillDescriptor deserialize(ByteBuffer byteBuffer) {
FillPolicy fillPolicy = FillPolicy.values()[ReadWriteIOUtils.readInt(byteBuffer)];
if (fillPolicy == FillPolicy.VALUE) {
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/parameter/FilterNullParameter.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/parameter/FilterNullParameter.java
index c357fb2966..803dff7c00 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/parameter/FilterNullParameter.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/parameter/FilterNullParameter.java
@@ -23,6 +23,8 @@ import org.apache.iotdb.db.mpp.plan.expression.Expression;
import org.apache.iotdb.db.mpp.plan.statement.component.FilterNullPolicy;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
@@ -68,6 +70,14 @@ public class FilterNullParameter {
}
}
+ public void serialize(DataOutputStream stream) throws IOException {
+ ReadWriteIOUtils.write(filterNullPolicy.ordinal(), stream);
+ ReadWriteIOUtils.write(filterNullColumns.size(), stream);
+ for (Expression filterNullColumn : filterNullColumns) {
+ Expression.serialize(filterNullColumn, stream);
+ }
+ }
+
public static FilterNullParameter deserialize(ByteBuffer byteBuffer) {
FilterNullPolicy filterNullPolicy =
FilterNullPolicy.values()[ReadWriteIOUtils.readInt(byteBuffer)];
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/parameter/GroupByTimeParameter.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/parameter/GroupByTimeParameter.java
index 1210bc55ce..6508ecd1e4 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/parameter/GroupByTimeParameter.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/parameter/GroupByTimeParameter.java
@@ -22,6 +22,8 @@ package org.apache.iotdb.db.mpp.plan.planner.plan.parameter;
import org.apache.iotdb.db.mpp.plan.statement.component.GroupByTimeComponent;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Objects;
@@ -153,6 +155,16 @@ public class GroupByTimeParameter {
ReadWriteIOUtils.write(leftCRightO, buffer);
}
+ public void serialize(DataOutputStream stream) throws IOException {
+ ReadWriteIOUtils.write(startTime, stream);
+ ReadWriteIOUtils.write(endTime, stream);
+ ReadWriteIOUtils.write(interval, stream);
+ ReadWriteIOUtils.write(slidingStep, stream);
+ ReadWriteIOUtils.write(isIntervalByMonth, stream);
+ ReadWriteIOUtils.write(isSlidingStepByMonth, stream);
+ ReadWriteIOUtils.write(leftCRightO, stream);
+ }
+
public static GroupByTimeParameter deserialize(ByteBuffer buffer) {
GroupByTimeParameter groupByTimeParameter = new GroupByTimeParameter();
groupByTimeParameter.setStartTime(ReadWriteIOUtils.readLong(buffer));
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/FragmentInstanceDispatcherImpl.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/FragmentInstanceDispatcherImpl.java
index a6da556732..cc5d63323b 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/FragmentInstanceDispatcherImpl.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/FragmentInstanceDispatcherImpl.java
@@ -150,9 +150,7 @@ public class FragmentInstanceDispatcherImpl implements IFragInstanceDispatcher {
throws FragmentInstanceDispatchException {
try (SyncDataNodeInternalServiceClient client =
internalServiceClientManager.borrowClient(endPoint)) {
- ByteBuffer buffer = ByteBuffer.allocate(1024 * 1024);
- instance.serializeRequest(buffer);
- buffer.flip();
+ ByteBuffer buffer = instance.serializeToByteBuffer();
TConsensusGroupId groupId = instance.getRegionReplicaSet().getRegionId();
TSendFragmentInstanceReq req =
new TSendFragmentInstanceReq(
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/SimpleFragInstanceDispatcher.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/SimpleFragInstanceDispatcher.java
index ba53e4fe2d..62ef784c21 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/SimpleFragInstanceDispatcher.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/SimpleFragInstanceDispatcher.java
@@ -62,10 +62,7 @@ public class SimpleFragInstanceDispatcher implements IFragInstanceDispatcher {
// TODO: (jackie tien) change the port
try (SyncDataNodeInternalServiceClient client =
internalServiceClientManager.borrowClient(endPoint)) {
- // TODO: (xingtanzjr) consider how to handle the buffer here
- ByteBuffer buffer = ByteBuffer.allocate(1024 * 1024);
- instance.serializeRequest(buffer);
- buffer.flip();
+ ByteBuffer buffer = instance.serializeToByteBuffer();
TConsensusGroupId groupId = instance.getRegionReplicaSet().getRegionId();
TSendFragmentInstanceReq req =
new TSendFragmentInstanceReq(
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/literal/BooleanLiteral.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/literal/BooleanLiteral.java
index 6c419ce5c9..334ec1af18 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/literal/BooleanLiteral.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/literal/BooleanLiteral.java
@@ -22,6 +22,8 @@ package org.apache.iotdb.db.mpp.plan.statement.literal;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Objects;
@@ -45,6 +47,12 @@ public class BooleanLiteral extends Literal {
ReadWriteIOUtils.write(value, byteBuffer);
}
+ @Override
+ public void serialize(DataOutputStream stream) throws IOException {
+ ReadWriteIOUtils.write(LiteralType.BOOLEAN.ordinal(), stream);
+ ReadWriteIOUtils.write(value, stream);
+ }
+
@Override
public boolean isDataTypeConsistency(TSDataType dataType) {
return dataType == TSDataType.BOOLEAN;
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/literal/DoubleLiteral.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/literal/DoubleLiteral.java
index d24aed13f9..4a613efb41 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/literal/DoubleLiteral.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/literal/DoubleLiteral.java
@@ -22,6 +22,8 @@ package org.apache.iotdb.db.mpp.plan.statement.literal;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Objects;
@@ -42,6 +44,12 @@ public class DoubleLiteral extends Literal {
ReadWriteIOUtils.write(value, byteBuffer);
}
+ @Override
+ public void serialize(DataOutputStream stream) throws IOException {
+ ReadWriteIOUtils.write(LiteralType.DOUBLE.ordinal(), stream);
+ ReadWriteIOUtils.write(value, stream);
+ }
+
@Override
public boolean isDataTypeConsistency(TSDataType dataType) {
return dataType == TSDataType.FLOAT || dataType == TSDataType.DOUBLE;
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/literal/Literal.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/literal/Literal.java
index cd089aa9d7..ec9c348f4c 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/literal/Literal.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/literal/Literal.java
@@ -24,6 +24,8 @@ import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.utils.Binary;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
public abstract class Literal extends StatementNode {
@@ -56,6 +58,8 @@ public abstract class Literal extends StatementNode {
public abstract void serialize(ByteBuffer byteBuffer);
+ public abstract void serialize(DataOutputStream stream) throws IOException;
+
public abstract boolean isDataTypeConsistency(TSDataType dataType);
public boolean getBoolean() {
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/literal/LongLiteral.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/literal/LongLiteral.java
index 5b10e106b8..d8acd1016b 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/literal/LongLiteral.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/literal/LongLiteral.java
@@ -22,6 +22,8 @@ package org.apache.iotdb.db.mpp.plan.statement.literal;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Objects;
@@ -46,6 +48,12 @@ public class LongLiteral extends Literal {
ReadWriteIOUtils.write(value, byteBuffer);
}
+ @Override
+ public void serialize(DataOutputStream stream) throws IOException {
+ ReadWriteIOUtils.write(LiteralType.LONG.ordinal(), stream);
+ ReadWriteIOUtils.write(value, stream);
+ }
+
@Override
public boolean isDataTypeConsistency(TSDataType dataType) {
return dataType == TSDataType.INT32
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/literal/NullLiteral.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/literal/NullLiteral.java
index 655c848288..4615821631 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/literal/NullLiteral.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/literal/NullLiteral.java
@@ -22,6 +22,8 @@ package org.apache.iotdb.db.mpp.plan.statement.literal;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
public class NullLiteral extends Literal {
@@ -31,6 +33,11 @@ public class NullLiteral extends Literal {
ReadWriteIOUtils.write(LiteralType.NULL.ordinal(), byteBuffer);
}
+ @Override
+ public void serialize(DataOutputStream stream) throws IOException {
+ ReadWriteIOUtils.write(LiteralType.NULL.ordinal(), stream);
+ }
+
@Override
public boolean isDataTypeConsistency(TSDataType dataType) {
return false;
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/literal/StringLiteral.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/literal/StringLiteral.java
index adedd93603..6bfb05e561 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/literal/StringLiteral.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/literal/StringLiteral.java
@@ -23,6 +23,8 @@ import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.utils.Binary;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Objects;
@@ -43,6 +45,12 @@ public class StringLiteral extends Literal {
ReadWriteIOUtils.write(value, byteBuffer);
}
+ @Override
+ public void serialize(DataOutputStream stream) throws IOException {
+ ReadWriteIOUtils.write(LiteralType.STRING.ordinal(), stream);
+ ReadWriteIOUtils.write(value, stream);
+ }
+
@Override
public boolean isDataTypeConsistency(TSDataType dataType) {
return dataType == TSDataType.TEXT;
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/physical/PhysicalPlan.java b/server/src/main/java/org/apache/iotdb/db/qp/physical/PhysicalPlan.java
index a29732e5fd..a98ce89eae 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/physical/PhysicalPlan.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/physical/PhysicalPlan.java
@@ -74,6 +74,7 @@ import org.apache.iotdb.db.qp.physical.sys.StopPipeServerPlan;
import org.apache.iotdb.db.qp.physical.sys.StopTriggerPlan;
import org.apache.iotdb.db.qp.physical.sys.StorageGroupMNodePlan;
import org.apache.iotdb.db.qp.physical.sys.UnsetTemplatePlan;
+import org.apache.iotdb.tsfile.utils.PublicBAOS;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
import org.slf4j.Logger;
@@ -181,6 +182,18 @@ public abstract class PhysicalPlan implements IConsensusRequest {
serialize(buffer);
}
+ @Override
+ public ByteBuffer serializeToByteBuffer() {
+ try (PublicBAOS byteArrayOutputStream = new PublicBAOS();
+ DataOutputStream outputStream = new DataOutputStream(byteArrayOutputStream)) {
+ serialize(outputStream);
+ return ByteBuffer.wrap(byteArrayOutputStream.getBuf(), 0, byteArrayOutputStream.size());
+ } catch (IOException e) {
+ logger.error("Unexpected error occurs when serializing this physical plan.", e);
+ throw new RuntimeException(e);
+ }
+ }
+
public void deserialize(DataInputStream stream) throws IOException, IllegalPathException {
throw new UnsupportedOperationException(SERIALIZATION_UNIMPLEMENTED);
}
diff --git a/server/src/test/java/org/apache/iotdb/db/mpp/plan/plan/FragmentInstanceSerdeTest.java b/server/src/test/java/org/apache/iotdb/db/mpp/plan/plan/FragmentInstanceSerdeTest.java
index fdb2c1924a..3510212b30 100644
--- a/server/src/test/java/org/apache/iotdb/db/mpp/plan/plan/FragmentInstanceSerdeTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/mpp/plan/plan/FragmentInstanceSerdeTest.java
@@ -73,9 +73,7 @@ public class FragmentInstanceSerdeTest {
ImmutableList.of(dataNodeLocation));
fragmentInstance.setDataRegionAndHost(regionReplicaSet);
- ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
- fragmentInstance.serializeRequest(byteBuffer);
- byteBuffer.flip();
+ ByteBuffer byteBuffer = fragmentInstance.serializeToByteBuffer();
FragmentInstance deserializeFragmentInstance = FragmentInstance.deserializeFrom(byteBuffer);
assertEquals(deserializeFragmentInstance, fragmentInstance);
}
@@ -103,9 +101,7 @@ public class FragmentInstanceSerdeTest {
ImmutableList.of(dataNodeLocation));
fragmentInstance.setDataRegionAndHost(regionReplicaSet);
- ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
- fragmentInstance.serializeRequest(byteBuffer);
- byteBuffer.flip();
+ ByteBuffer byteBuffer = fragmentInstance.serializeToByteBuffer();
FragmentInstance deserializeFragmentInstance = FragmentInstance.deserializeFrom(byteBuffer);
assertEquals(deserializeFragmentInstance, fragmentInstance);
}
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/Path.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/Path.java
index df6b3920fe..abcb013904 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/Path.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/Path.java
@@ -26,6 +26,8 @@ import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.Validate;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.io.Serializable;
import java.nio.ByteBuffer;
@@ -175,16 +177,27 @@ public class Path implements Serializable, Comparable<Path> {
}
public void serialize(ByteBuffer byteBuffer) {
- byteBuffer.put((byte) 3); // org.apache.iotdb.db.metadata.path#PathType
+ ReadWriteIOUtils.write((byte) 3, byteBuffer); // org.apache.iotdb.db.metadata.path#PathType
serializeWithoutType(byteBuffer);
}
+ public void serialize(DataOutputStream stream) throws IOException {
+ ReadWriteIOUtils.write((byte) 3, stream); // org.apache.iotdb.db.metadata.path#PathType
+ serializeWithoutType(stream);
+ }
+
protected void serializeWithoutType(ByteBuffer byteBuffer) {
ReadWriteIOUtils.write(measurement, byteBuffer);
ReadWriteIOUtils.write(device, byteBuffer);
ReadWriteIOUtils.write(fullPath, byteBuffer);
}
+ protected void serializeWithoutType(DataOutputStream stream) throws IOException {
+ ReadWriteIOUtils.write(measurement, stream);
+ ReadWriteIOUtils.write(device, stream);
+ ReadWriteIOUtils.write(fullPath, stream);
+ }
+
public static Path deserialize(ByteBuffer byteBuffer) {
Path path = new Path();
path.measurement = ReadWriteIOUtils.readString(byteBuffer);
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/utils/PublicBAOS.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/utils/PublicBAOS.java
index 90c2d0f22c..18a459a6ee 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/utils/PublicBAOS.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/utils/PublicBAOS.java
@@ -16,6 +16,7 @@
* specific language governing permissions and limitations
* under the License.
*/
+
package org.apache.iotdb.tsfile.utils;
import java.io.ByteArrayOutputStream;