add distributed serialize

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/TemplatedInfo.java

@@ -104,6 +104,7 @@ public TemplatedInfo(
       Map<String, IMeasurementSchema> schemaMap,
       Map<String, List<InputLocation>> layoutMap,
       Expression pushDownPredicate,
+      List<AggregationDescriptor> aggregationDescriptorList,
       GroupByTimeParameter groupByTimeParameter,
       boolean outputEndTime) {
     this.measurementList = measurementList;
@@ -123,6 +124,7 @@ public TemplatedInfo(
     }
     this.pushDownPredicate = pushDownPredicate;
 
+    this.aggregationDescriptorList = aggregationDescriptorList;
     this.groupByTimeParameter = groupByTimeParameter;
     this.outputEndTime = outputEndTime;
   }
@@ -259,6 +261,13 @@ public void serialize(ByteBuffer byteBuffer) {
     } else {
       ReadWriteIOUtils.write((byte) 0, byteBuffer);
     }
+
+    if (aggregationDescriptorList != null) {
+      ReadWriteIOUtils.write(aggregationDescriptorList.size(), byteBuffer);
+      aggregationDescriptorList.forEach(d -> d.serialize(byteBuffer));
+    } else {
+      ReadWriteIOUtils.write((byte) 0, byteBuffer);
+    }
   }
 
   public void serialize(DataOutputStream stream) throws IOException {
@@ -302,6 +311,15 @@ public void serialize(DataOutputStream stream) throws IOException {
     } else {
       ReadWriteIOUtils.write((byte) 0, stream);
     }
+
+    if (aggregationDescriptorList != null) {
+      ReadWriteIOUtils.write(aggregationDescriptorList.size(), stream);
+      for (AggregationDescriptor descriptor : aggregationDescriptorList) {
+        descriptor.serialize(stream);
+      }
+    } else {
+      ReadWriteIOUtils.write((byte) 0, stream);
+    }
   }
 
   public static TemplatedInfo deserialize(ByteBuffer byteBuffer) {
@@ -367,6 +385,15 @@ public static TemplatedInfo deserialize(ByteBuffer byteBuffer) {
       pushDownPredicate = Expression.deserialize(byteBuffer);
     }
 
+    List<AggregationDescriptor> aggregationDescriptorList = null;
+    listSize = ReadWriteIOUtils.readInt(byteBuffer);
+    if (listSize > 0) {
+      aggregationDescriptorList = new ArrayList<>(listSize);
+      while (listSize-- > 0) {
+        aggregationDescriptorList.add(AggregationDescriptor.deserialize(byteBuffer));
+      }
+    }
+
     // TODO add groupByTimeParameter, outputEndTime serialization and deserialization
 
     return new TemplatedInfo(
@@ -384,6 +411,7 @@ public static TemplatedInfo deserialize(ByteBuffer byteBuffer) {
         currentSchemaMap,
         layoutMap,
         pushDownPredicate,
+        aggregationDescriptorList,
         null,
         false);
   }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/TemplatedLogicalPlan.java

@@ -155,6 +155,7 @@ private void initAggQueryCommonVariables() {
                 analysis.getDeviceTemplate().getSchemaMap(),
                 filterLayoutMap,
                 null,
+                null,
                 analysis.getGroupByTimeParameter(),
                 queryStatement.isOutputEndTime()));
   }
@@ -212,6 +213,7 @@ private void initNonAggQueryCommonVariables() {
                 analysis.getDeviceTemplate().getSchemaMap(),
                 filterLayoutMap,
                 null,
+                null,
                 analysis.getGroupByTimeParameter(),
                 queryStatement.isOutputEndTime()));
   }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanNodeType.java

@@ -462,6 +462,8 @@ public static PlanNode deserializeWithTemplate(ByteBuffer buffer, TypeProvider t
         return FilterNode.deserializeUseTemplate(buffer, typeProvider);
       case 33:
         return AlignedSeriesScanNode.deserializeUseTemplate(buffer, typeProvider);
+      case 34:
+        return AlignedSeriesAggregationScanNode.deserializeUseTemplate(buffer, typeProvider);
       case 65:
         return SingleDeviceViewNode.deserializeUseTemplate(buffer, typeProvider);
       case 32:

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/source/AlignedSeriesAggregationScanNode.java

@@ -23,6 +23,7 @@
 import org.apache.iotdb.commons.path.AlignedPath;
 import org.apache.iotdb.commons.path.PartialPath;
 import org.apache.iotdb.commons.path.PathDeserializeUtil;
+import org.apache.iotdb.db.queryengine.plan.analyze.TypeProvider;
 import org.apache.iotdb.db.queryengine.plan.expression.Expression;
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNode;
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeId;
@@ -255,6 +256,41 @@ public static AlignedSeriesAggregationScanNode deserialize(ByteBuffer byteBuffer
         null);
   }
 
+  @Override
+  public void serializeUseTemplate(DataOutputStream stream, TypeProvider typeProvider)
+      throws IOException {
+    PlanNodeType.ALIGNED_SERIES_AGGREGATE_SCAN.serialize(stream);
+    id.serialize(stream);
+    ReadWriteIOUtils.write(alignedPath.getNodes().length, stream);
+    for (String node : alignedPath.getNodes()) {
+      ReadWriteIOUtils.write(node, stream);
+    }
+  }
+
+  public static AlignedSeriesAggregationScanNode deserializeUseTemplate(
+      ByteBuffer byteBuffer, TypeProvider typeProvider) {
+    PlanNodeId planNodeId = PlanNodeId.deserialize(byteBuffer);
+
+    int nodeSize = ReadWriteIOUtils.readInt(byteBuffer);
+    String[] nodes = new String[nodeSize];
+    for (int i = 0; i < nodeSize; i++) {
+      nodes[i] = ReadWriteIOUtils.readString(byteBuffer);
+    }
+    AlignedPath alignedPath = new AlignedPath(new PartialPath(nodes));
+    alignedPath.setMeasurementList(typeProvider.getTemplatedInfo().getMeasurementList());
+    alignedPath.addSchemas(typeProvider.getTemplatedInfo().getSchemaList());
+
+    return new AlignedSeriesAggregationScanNode(
+        planNodeId,
+        alignedPath,
+        typeProvider.getTemplatedInfo().aggregationDescriptorList,
+        typeProvider.getTemplatedInfo().getScanOrder(),
+        typeProvider.getTemplatedInfo().outputEndTime,
+        typeProvider.getTemplatedInfo().getPushDownPredicate(),
+        typeProvider.getTemplatedInfo().groupByTimeParameter,
+        null);
+  }
+
   @Override
   public boolean equals(Object o) {
     if (this == o) {