Fix wrong push limit down to AggTableScanNode (apache#16696)

Author: JackieTien97

integration-test/src/test/java/org/apache/iotdb/relational/it/query/recent/IoTDBTableAggregationIT.java

@@ -5491,4 +5491,17 @@ public void emptyTimeRangeQueryTest() {
         retArray,
         DATABASE_NAME);
   }
+
+  @Test
+  public void orderByLimitTest() {
+    String[] expectedHeader =
+        new String[] {"province", "city", "region", "device_id", "_col4", "_col5"};
+    String[] retArray = new String[] {"beijing,beijing,chaoyang,d09,2024-09-24T06:00:00.000Z,2,"};
+
+    tableResultSetEqualTest(
+        "select province, city, region, device_id, date_bin(1h, time), count(s1) from table1 where s1 >= 40 group by 1,2,3,4,5 order by province, city, region, device_id, date_bin(1h, time) limit 1",
+        expectedHeader,
+        retArray,
+        DATABASE_NAME);
+  }
 }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanGenerator.java

@@ -423,7 +423,7 @@ public List<PlanNode> visitTopK(TopKNode node, PlanContext context) {
   private boolean canTopKEliminated(OrderingScheme orderingScheme, long k, PlanNode child) {
     // if DeviceTableScanNode has limit <= K and with same order, we can directly return
     // DeviceTableScanNode
-    if (child instanceof DeviceTableScanNode) {
+    if (child instanceof DeviceTableScanNode && !(child instanceof AggregationTableScanNode)) {
       DeviceTableScanNode tableScanNode = (DeviceTableScanNode) child;
       if (canSortEliminated(orderingScheme, nodeOrderingMap.get(child.getPlanNodeId()))) {
         if (tableScanNode.getPushDownLimit() <= 0) {

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/iterative/rule/PushDownOffsetIntoTableScan.java

@@ -20,6 +20,7 @@
 package org.apache.iotdb.db.queryengine.plan.relational.planner.iterative.rule;
 
 import org.apache.iotdb.db.queryengine.plan.relational.planner.iterative.Rule;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.AggregationTableScanNode;
 import org.apache.iotdb.db.queryengine.plan.relational.planner.node.DeviceTableScanNode;
 import org.apache.iotdb.db.queryengine.plan.relational.planner.node.OffsetNode;
 import org.apache.iotdb.db.queryengine.plan.relational.planner.node.TableScanNode;
@@ -53,7 +54,8 @@ public Pattern<OffsetNode> getPattern() {
   @Override
   public Result apply(OffsetNode parent, Captures captures, Context context) {
     TableScanNode tableScanNode = captures.get(CHILD);
-    if (tableScanNode instanceof DeviceTableScanNode
+    if ((tableScanNode instanceof DeviceTableScanNode
+            && !(tableScanNode instanceof AggregationTableScanNode))
         && !((DeviceTableScanNode) tableScanNode).isPushLimitToEachDevice()) {
       tableScanNode.setPushDownOffset(parent.getCount());
       // consider case that there is no limit

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

@@ -623,4 +623,14 @@ public void setDeviceCountMap(Map<DeviceEntry, Integer> deviceCountMap) {
   public Map<DeviceEntry, Integer> getDeviceCountMap() {
     return deviceCountMap;
   }
+
+  @Override
+  public void setPushDownLimit(long pushDownLimit) {
+    throw new IllegalStateException("Should never push down limit to AggregationTableScanNode.");
+  }
+
+  @Override
+  public void setPushDownOffset(long pushDownOffset) {
+    throw new IllegalStateException("Should never push down offset to AggregationTableScanNode.");
+  }
 }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/PushLimitOffsetIntoTableScan.java

@@ -50,7 +50,7 @@
 import static org.apache.iotdb.db.queryengine.plan.relational.planner.optimizations.PushPredicateIntoTableScan.containsDiffFunction;
 
 /**
- * <b>Optimization phase:</b> Distributed plan planning.
+ * <b>Optimization phase:</b> Logical plan planning.
  *
  * <p>The LIMIT OFFSET condition can be pushed down to the DeviceTableScanNode, when the following
  * conditions are met: