Merge LastQueryScanNode of same device

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/last/LastQueryOperator.java

@@ -116,6 +116,7 @@ public TsBlock next() throws Exception {
           return null;
         } else if (!tsBlock.isEmpty()) {
           LastQueryUtil.appendLastValue(tsBlockBuilder, tsBlock);
+          continue;
         }
       } else {
         children.get(currentIndex).close();

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/last/LastQuerySortOperator.java

@@ -153,7 +153,8 @@ private TsBlock buildResult() throws Exception {
 
     while (keepGoing(start, maxRuntime, endIndex)) {
 
-      if (prepareData()) {
+      prepareData();
+      if (previousTsBlock == null) {
         return null;
       }
 
@@ -179,21 +180,20 @@ private boolean keepGoing(long start, long maxRuntime, int endIndex) {
         && !tsBlockBuilder.isFull();
   }
 
-  private boolean prepareData() throws Exception {
+  private void prepareData() throws Exception {
     if (previousTsBlock == null || previousTsBlock.getPositionCount() <= previousTsBlockIndex) {
       if (children.get(currentIndex).hasNextWithTimer()) {
         previousTsBlock = children.get(currentIndex).nextWithTimer();
         previousTsBlockIndex = 0;
-        if (previousTsBlock == null) {
-          return true;
+        if (previousTsBlock == null || children.get(currentIndex).hasNextWithTimer()) {
+          return;
         }
       } else {
         children.get(currentIndex).close();
         children.set(currentIndex, null);
       }
       currentIndex++;
     }
-    return false;
   }
 
   @Override

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

@@ -81,10 +81,9 @@
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.join.FullOuterTimeJoinNode;
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.last.LastQueryNode;
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.last.LastQueryTransformNode;
-import org.apache.iotdb.db.queryengine.plan.planner.plan.node.source.AlignedLastQueryScanNode;
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.source.AlignedSeriesScanNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.source.DeviceLastQueryScanNode;
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.source.DeviceRegionScanNode;
-import org.apache.iotdb.db.queryengine.plan.planner.plan.node.source.LastQueryScanNode;
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.source.SeriesScanNode;
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.source.SeriesSourceNode;
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.source.ShowQueriesNode;
@@ -110,6 +109,7 @@
 import org.apache.commons.lang3.Validate;
 import org.apache.tsfile.enums.TSDataType;
 import org.apache.tsfile.file.metadata.IDeviceID;
+import org.apache.tsfile.write.schema.IMeasurementSchema;
 
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -234,7 +234,6 @@ public LogicalPlanBuilder planRawDataSource(
   }
 
   public LogicalPlanBuilder planLast(Analysis analysis, Ordering timeseriesOrdering) {
-    Set<IDeviceID> deviceAlignedSet = new HashSet<>();
     Set<IDeviceID> deviceExistViewSet = new HashSet<>();
     // <Device, <Measurement, Expression>>
     Map<IDeviceID, Map<String, Expression>> outputPathToSourceExpressionMap = new LinkedHashMap<>();
@@ -254,9 +253,6 @@ public LogicalPlanBuilder planLast(Analysis analysis, Ordering timeseriesOrderin
                       ? new TreeMap<>(timeseriesOrdering.getStringComparator())
                       : new LinkedHashMap<>())
           .put(outputPath.getMeasurement(), sourceExpression);
-      if (outputPath.isUnderAlignedEntity()) {
-        deviceAlignedSet.add(outputDevice);
-      }
       if (sourceExpression.isViewExpression()) {
         deviceExistViewSet.add(outputDevice);
       }
@@ -278,46 +274,34 @@ public LogicalPlanBuilder planLast(Analysis analysis, Ordering timeseriesOrderin
                   ? sourceExpression.getViewPath().getFullPath()
                   : null;
 
-          if (selectedPath.isUnderAlignedEntity()) { // aligned series
-            sourceNodeList.add(
-                reserveMemoryForSeriesSourceNode(
-                    new AlignedLastQueryScanNode(
-                        context.getQueryId().genPlanNodeId(),
-                        new AlignedPath(selectedPath),
-                        outputViewPath)));
-          } else { // non-aligned series
-            sourceNodeList.add(
-                reserveMemoryForSeriesSourceNode(
-                    new LastQueryScanNode(
-                        context.getQueryId().genPlanNodeId(), selectedPath, outputViewPath)));
-          }
-        }
-      } else {
-        if (deviceAlignedSet.contains(outputDevice)) {
-          // aligned series
-          List<MeasurementPath> measurementPaths =
-              measurementToExpressionsOfDevice.values().stream()
-                  .map(expression -> (MeasurementPath) ((TimeSeriesOperand) expression).getPath())
-                  .collect(Collectors.toList());
-          AlignedPath alignedPath = new AlignedPath(measurementPaths.get(0).getDevicePath());
-          for (MeasurementPath measurementPath : measurementPaths) {
-            alignedPath.addMeasurement(measurementPath);
-          }
           sourceNodeList.add(
               reserveMemoryForSeriesSourceNode(
-                  new AlignedLastQueryScanNode(
-                      context.getQueryId().genPlanNodeId(), alignedPath, null)));
-        } else {
-          // non-aligned series
-          for (Expression sourceExpression : measurementToExpressionsOfDevice.values()) {
-            MeasurementPath selectedPath =
-                (MeasurementPath) ((TimeSeriesOperand) sourceExpression).getPath();
-            sourceNodeList.add(
-                reserveMemoryForSeriesSourceNode(
-                    new LastQueryScanNode(
-                        context.getQueryId().genPlanNodeId(), selectedPath, null)));
-          }
+                  new DeviceLastQueryScanNode(
+                      context.getQueryId().genPlanNodeId(),
+                      selectedPath.getDevicePath(),
+                      selectedPath.isUnderAlignedEntity(),
+                      Collections.singletonList(selectedPath.getMeasurementSchema()),
+                      outputViewPath)));
         }
+      } else {
+        boolean aligned = false;
+        List<IMeasurementSchema> measurementSchemas =
+            new ArrayList<>(measurementToExpressionsOfDevice.size());
+        PartialPath devicePath = null;
+        for (Expression sourceExpression : measurementToExpressionsOfDevice.values()) {
+          MeasurementPath selectedPath =
+              (MeasurementPath) ((TimeSeriesOperand) sourceExpression).getPath();
+          aligned = selectedPath.isUnderAlignedEntity();
+          devicePath = devicePath == null ? selectedPath.getDevicePath() : devicePath;
+          measurementSchemas.add(selectedPath.getMeasurementSchema());
+        }
+        sourceNodeList.add(
+            new DeviceLastQueryScanNode(
+                context.getQueryId().genPlanNodeId(),
+                devicePath,
+                aligned,
+                measurementSchemas,
+                null));
       }
     }
 
@@ -328,10 +312,8 @@ public LogicalPlanBuilder planLast(Analysis analysis, Ordering timeseriesOrderin
           Comparator.comparing(
               child -> {
                 String sortKey = "";
-                if (child instanceof LastQueryScanNode) {
-                  sortKey = ((LastQueryScanNode) child).getOutputSymbolForSort();
-                } else if (child instanceof AlignedLastQueryScanNode) {
-                  sortKey = ((AlignedLastQueryScanNode) child).getOutputSymbolForSort();
+                if (child instanceof DeviceLastQueryScanNode) {
+                  sortKey = ((DeviceLastQueryScanNode) child).getOutputSymbolForSort();
                 } else if (child instanceof LastQueryTransformNode) {
                   sortKey = ((LastQueryTransformNode) child).getOutputSymbolForSort();
                 }