Fix error of non-mappable udtf query in align by device while existing any devices' data cross region

Author: JackieTien97

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

@@ -193,7 +193,8 @@ aggregation results last_value(temperature) and last_value(status), whereas buck
 
   // indicates whether DeviceView need special process when rewriteSource in DistributionPlan,
   // you can see SourceRewriter#visitDeviceView to get more information
-  // deviceViewSpecialProcess equals true when all Aggregation Functions and DIFF
+  // deviceViewSpecialProcess equals true when all Aggregation Functions and non-mappable UDTF and
+  // DIFF
   private boolean deviceViewSpecialProcess;
 
   /////////////////////////////////////////////////////////////////////////////////////////////////

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

@@ -35,6 +35,7 @@
 import org.apache.iotdb.db.queryengine.plan.expression.Expression;
 import org.apache.iotdb.db.queryengine.plan.expression.leaf.TimeSeriesOperand;
 import org.apache.iotdb.db.queryengine.plan.expression.multi.FunctionExpression;
+import org.apache.iotdb.db.queryengine.plan.expression.multi.FunctionType;
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.BaseSourceRewriter;
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNode;
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeId;
@@ -217,7 +218,7 @@ public List<PlanNode> visitDeviceView(DeviceViewNode node, DistributionPlanConte
                   analysis.getPartitionInfo(outputDevice, context.getPartitionTimeFilter()));
       if (regionReplicaSets.size() > 1 && !existDeviceCrossRegion) {
         existDeviceCrossRegion = true;
-        if (analysis.isDeviceViewSpecialProcess() && aggregationCannotUseMergeSort()) {
+        if (analysis.isDeviceViewSpecialProcess() && cannotUseAggMergeSort()) {
           return processSpecialDeviceView(node, context);
         }
       }
@@ -387,18 +388,22 @@ private static List<Expression> getWithDeviceExpressions(
   }
 
   /**
-   * aggregation align by device, and aggregation is `count_if` or `diff`, or aggregation used with
-   * group by parameter (session, variation, count), use the old aggregation logic
+   * 1. aggregation align by device, and aggregation is `count_if` or `diff`, or aggregation used
+   * with 2. group by parameter (session, variation, count), use the old aggregation logic 3.
+   * non-mappable UDTF, we just need to check UDTF in this method, because caller has already
+   * checked analysis.isDeviceViewSpecialProcess()
    */
-  private boolean aggregationCannotUseMergeSort() {
+  private boolean cannotUseAggMergeSort() {
     if (analysis.hasGroupByParameter()) {
       return true;
     }
 
     for (Expression expression : analysis.getDeviceViewOutputExpressions()) {
       if (expression instanceof FunctionExpression) {
         String functionName = ((FunctionExpression) expression).getFunctionName();
-        if (COUNT_IF.equalsIgnoreCase(functionName) || DIFF.equalsIgnoreCase(functionName)) {
+        if (((FunctionExpression) expression).getFunctionType() == FunctionType.UDTF
+            || COUNT_IF.equalsIgnoreCase(functionName)
+            || DIFF.equalsIgnoreCase(functionName)) {
           return true;
         }
       }