(iotdb) branch dev/1.3 updated: [To dev/1.3] Fix error of non-mappable udtf query in align by device while existing any devices' data cross region

[jackietien]

This is an automated email from the ASF dual-hosted git repository.

jackietien pushed a commit to branch dev/1.3
in repository https://gitbox.apache.org/repos/asf/iotdb.git


The following commit(s) were added to refs/heads/dev/1.3 by this push:
     new 940742b99f2 [To dev/1.3] Fix error of non-mappable udtf query in align 
by device while existing any devices' data cross region
940742b99f2 is described below

commit 940742b99f22673cf8e0e154a8e1544c18c1655c
Author: Jackie Tien <jackietie...@gmail.com>
AuthorDate: Tue Aug 26 13:57:56 2025 +0800

    [To dev/1.3] Fix error of non-mappable udtf query in align by device while 
existing any devices' data cross region
    
    (cherry picked from commit ef0e45e3e19394b2aede89658fc2d7de9920e69d)
---
 .../iotdb/db/queryengine/plan/analyze/Analysis.java       |  3 ++-
 .../plan/planner/distribution/SourceRewriter.java         | 15 ++++++++++-----
 2 files changed, 12 insertions(+), 6 deletions(-)

diff --git 
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/Analysis.java
 
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/Analysis.java
index c3f31fa8a3b..01d1020934d 100644
--- 
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/Analysis.java
+++ 
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/Analysis.java
@@ -190,7 +190,8 @@ public class Analysis implements IAnalysis {
 
   // 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;
 
   
/////////////////////////////////////////////////////////////////////////////////////////////////
diff --git 
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/SourceRewriter.java
 
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/SourceRewriter.java
index 1724d2d639d..354ba952822 100644
--- 
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/SourceRewriter.java
+++ 
b/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.analyze.Analysis;
 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;
@@ -216,7 +217,7 @@ public class SourceRewriter extends 
BaseSourceRewriter<DistributionPlanContext>
                   analysis.getPartitionInfo(outputDevice, 
context.getPartitionTimeFilter()));
       if (regionReplicaSets.size() > 1 && !existDeviceCrossRegion) {
         existDeviceCrossRegion = true;
-        if (analysis.isDeviceViewSpecialProcess() && 
aggregationCannotUseMergeSort()) {
+        if (analysis.isDeviceViewSpecialProcess() && cannotUseAggMergeSort()) {
           return processSpecialDeviceView(node, context);
         }
       }
@@ -385,10 +386,12 @@ public class SourceRewriter extends 
BaseSourceRewriter<DistributionPlanContext>
   }
 
   /**
-   * 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;
     }
@@ -396,7 +399,9 @@ public class SourceRewriter extends 
BaseSourceRewriter<DistributionPlanContext>
     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;
         }
       }