[To dev/1.3]Support order by time for Agg align by device query with cross region data

Author: Wei-hao-Li

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

@@ -1174,8 +1174,8 @@ public Operator visitAggregationMergeSort(
 
     List<SortItem> sortItemList = node.getMergeOrderParameter().getSortItemList();
     if (!sortItemList.get(0).getSortKey().equalsIgnoreCase("Device")) {
-      throw new IllegalArgumentException(
-          "Only order by device align by device support AggregationMergeSortNode.");
+      throw new IllegalStateException(
+          "AggregationMergeSortNode without order by device should not appear here");
     }
 
     boolean timeAscending = true;

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

@@ -82,6 +82,7 @@
 import org.apache.iotdb.db.queryengine.plan.statement.component.SortItem;
 import org.apache.iotdb.db.utils.constant.SqlConstant;
 
+import com.google.common.collect.ImmutableList;
 import org.apache.tsfile.enums.TSDataType;
 
 import java.util.ArrayList;
@@ -330,12 +331,23 @@ public List<PlanNode> visitDeviceView(DeviceViewNode node, DistributionPlanConte
         }
       }
 
+      OrderByParameter orderByParameter;
+      List<SortItem> sortItemList = node.getMergeOrderParameter().getSortItemList();
+      if (!sortItemList.get(0).getSortKey().equalsIgnoreCase("Device")) {
+        // When reach here, it means DeviceView is order by time with only one device, it is no
+        // problem to transform order by time to order by device.
+        // SortItems here will only be Time and Device, see planDeviceView().
+        orderByParameter =
+            new OrderByParameter(ImmutableList.of(sortItemList.get(1), sortItemList.get(0)));
+      } else {
+        orderByParameter = node.getMergeOrderParameter();
+      }
       boolean hasGroupBy =
           analysis.getGroupByTimeParameter() != null || analysis.hasGroupByParameter();
       AggregationMergeSortNode mergeSortNode =
           new AggregationMergeSortNode(
               context.queryContext.getQueryId().genPlanNodeId(),
-              node.getMergeOrderParameter(),
+              orderByParameter,
               node.getOutputColumnNames(),
               deviceViewOutputExpressions,
               hasGroupBy);

iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/AggregationAlignByDeviceTest.java

@@ -39,10 +39,13 @@
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.sink.IdentitySinkNode;
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.sink.ShuffleSinkNode;
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.source.SeriesAggregationScanNode;
+import org.apache.iotdb.db.queryengine.plan.statement.component.SortItem;
 
 import com.google.common.collect.ImmutableList;
 import org.junit.Test;
 
+import java.util.List;
+
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
 
@@ -465,5 +468,21 @@ public void crossRegionTest() {
               "last_value(root.sg.d1.s2)",
               "max_time(root.sg.d1.s2)"));
     }
+
+    // order by time
+    sql = "select last_value(s1),last_value(s2)from root.sg.d1 order by time align by device";
+    analysis = Util.analyze(sql, context);
+    logicalPlanNode = Util.genLogicalPlan(analysis, context);
+    planner = new DistributionPlanner(analysis, new LogicalQueryPlan(context, logicalPlanNode));
+    plan = planner.planFragments();
+    assertEquals(2, plan.getInstances().size());
+
+    firstFiRoot = plan.getInstances().get(0).getFragment().getPlanNodeTree().getChildren().get(0);
+    assertTrue(firstFiRoot instanceof AggregationMergeSortNode);
+    assertTrue(firstFiRoot.getChildren().get(0) instanceof DeviceViewNode);
+    List<SortItem> sortItemList =
+        ((AggregationMergeSortNode) firstFiRoot).getMergeOrderParameter().getSortItemList();
+    assertEquals(sortItemList.get(0).getSortKey().toLowerCase(), "device");
+    assertEquals(sortItemList.get(1).getSortKey().toLowerCase(), "time");
   }
 }