[Enhancement] check tuple id is in Descriptor table

fe/fe-core/src/main/java/com/starrocks/planner/PlanFragment.java

@@ -38,6 +38,7 @@
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
 import com.google.common.collect.Sets;
+import com.starrocks.analysis.DescriptorTable;
 import com.starrocks.analysis.Expr;
 import com.starrocks.common.Config;
 import com.starrocks.common.FeConstants;
@@ -63,7 +64,6 @@
 
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
-import java.util.Collections;
 import java.util.Comparator;
 import java.util.Deque;
 import java.util.LinkedList;
@@ -72,7 +72,6 @@
 import java.util.Queue;
 import java.util.Set;
 import java.util.function.Consumer;
-import java.util.function.Function;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
@@ -418,6 +417,25 @@ public void setOutputExprs(List<Expr> outputExprs) {
         this.outputExprs = Expr.cloneList(outputExprs, null);
     }
 
+    public boolean checkFragmentTupleId(DescriptorTable descriptorTable) {
+        Queue<PlanNode> queue = Lists.newLinkedList();
+        queue.add(planRoot);
+        while (!queue.isEmpty()) {
+            PlanNode node = queue.poll();
+            boolean isValid =
+                    node.getTupleIds().stream().allMatch(tupleId -> descriptorTable.getTupleDesc(tupleId) != null);
+            if (!isValid) {
+                // query dump should be able to work
+                if (ConnectContext.get() != null && (ConnectContext.get().isHTTPQueryDump())) {
+                    return true;
+                }
+                return false;
+            }
+            queue.addAll(node.getChildren());
+        }
+        return true;
+    }
+
     /**
      * Finalize plan tree and create stream sink, if needed.
      */

fe/fe-core/src/main/java/com/starrocks/sql/plan/PlanFragmentBuilder.java

@@ -220,6 +220,7 @@
 import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
+import java.util.Queue;
 import java.util.Set;
 import java.util.TreeSet;
 import java.util.stream.Collectors;
@@ -411,9 +412,25 @@ private static ExecPlan finalizeFragments(ExecPlan execPlan, TResultSinkType res
             }
         }
 
+        if (!checkTupleId(execPlan.getDescTbl(), execPlan.getFragments().get(0))) {
+            throw new StarRocksPlannerException("TupleId check failed", INTERNAL_ERROR);
+        }
+
         return execPlan;
     }
 
+    private static boolean checkTupleId(DescriptorTable descriptorTable, PlanFragment root) {
+        Queue<PlanFragment> queue = Lists.newLinkedList();
+        queue.add(root);
+        while (!queue.isEmpty()) {
+            PlanFragment planFragment = queue.poll();
+            if (!planFragment.checkFragmentTupleId(descriptorTable)) {
+                return false;
+            }
+        }
+        return true;
+    }
+
     private static class PhysicalPlanTranslator extends OptExpressionVisitor<PlanFragment, ExecPlan> {
         private final ColumnRefFactory columnRefFactory;
         private final IdGenerator<RuntimeFilterId> runtimeFilterIdIdGenerator = RuntimeFilterId.createGenerator();