Mark wether a PlanFragement is contains TableScanNode or not

There may be a little bit performance issue while response for api of cluster stats, if cluster has huge amounts of tasks.
Most of time was spent on distinguish the type of fragements temporarily.

core/trino-main/src/main/java/io/trino/execution/QueryStateMachine.java

@@ -54,7 +54,6 @@
 import io.trino.spi.type.Type;
 import io.trino.sql.analyzer.Output;
 import io.trino.sql.planner.PlanFragment;
-import io.trino.sql.planner.plan.TableScanNode;
 import io.trino.tracing.TrinoAttributes;
 import io.trino.transaction.TransactionId;
 import io.trino.transaction.TransactionInfo;
@@ -669,7 +668,7 @@ private QueryStats getQueryStats(Optional<StageInfo> rootStage, List<StageInfo>
             failedInternalNetworkInputPositions += stageStats.getFailedInternalNetworkInputPositions();
 
             PlanFragment plan = stageInfo.getPlan();
-            if (plan != null && plan.getPartitionedSourceNodes().stream().anyMatch(TableScanNode.class::isInstance)) {
+            if (plan != null && plan.containsTableScanNode()) {
                 rawInputDataSize += stageStats.getRawInputDataSize().toBytes();
                 failedRawInputDataSize += stageStats.getFailedRawInputDataSize().toBytes();
                 rawInputPositions += stageStats.getRawInputPositions();

core/trino-main/src/main/java/io/trino/execution/StageStateMachine.java

@@ -33,7 +33,6 @@
 import io.trino.spi.eventlistener.StageGcStatistics;
 import io.trino.sql.planner.PlanFragment;
 import io.trino.sql.planner.plan.PlanNodeId;
-import io.trino.sql.planner.plan.TableScanNode;
 import io.trino.tracing.TrinoAttributes;
 import io.trino.util.Failures;
 import it.unimi.dsi.fastutil.longs.Long2ObjectOpenHashMap;
@@ -338,7 +337,7 @@ public BasicStageStats getBasicStageStats(Supplier<Iterable<TaskInfo>> taskInfos
             internalNetworkInputDataSize += taskStats.getInternalNetworkInputDataSize().toBytes();
             internalNetworkInputPositions += taskStats.getInternalNetworkInputPositions();
 
-            if (fragment.getPartitionedSourceNodes().stream().anyMatch(TableScanNode.class::isInstance)) {
+            if (fragment.containsTableScanNode()) {
                 rawInputDataSize += taskStats.getRawInputDataSize().toBytes();
                 rawInputPositions += taskStats.getRawInputPositions();
             }

core/trino-main/src/main/java/io/trino/sql/planner/PlanFragment.java

@@ -26,6 +26,7 @@
 import io.trino.sql.planner.plan.PlanNode;
 import io.trino.sql.planner.plan.PlanNodeId;
 import io.trino.sql.planner.plan.RemoteSourceNode;
+import io.trino.sql.planner.plan.TableScanNode;
 
 import java.util.List;
 import java.util.Map;
@@ -55,6 +56,7 @@ public class PlanFragment
     private final List<CatalogProperties> activeCatalogs;
     private final List<LanguageScalarFunctionData> languageFunctions;
     private final Optional<String> jsonRepresentation;
+    private final boolean containsTableScanNode;
 
     // Only for creating instances without the JSON representation embedded
     private PlanFragment(
@@ -88,6 +90,7 @@ private PlanFragment(
         this.activeCatalogs = requireNonNull(activeCatalogs, "activeCatalogs is null");
         this.languageFunctions = requireNonNull(languageFunctions, "languageFunctions is null");
         this.jsonRepresentation = Optional.empty();
+        this.containsTableScanNode = partitionedSourceNodes.stream().anyMatch(TableScanNode.class::isInstance);
     }
 
     @JsonCreator
@@ -135,6 +138,7 @@ public PlanFragment(
         this.remoteSourceNodes = remoteSourceNodes.build();
 
         this.outputPartitioningScheme = requireNonNull(outputPartitioningScheme, "partitioningScheme is null");
+        this.containsTableScanNode = partitionedSourceNodes.stream().anyMatch(TableScanNode.class::isInstance);
     }
 
     @JsonProperty
@@ -372,4 +376,9 @@ public PlanFragment withActiveCatalogs(List<CatalogProperties> activeCatalogs)
                 this.languageFunctions,
                 this.jsonRepresentation);
     }
+
+    public boolean containsTableScanNode()
+    {
+        return containsTableScanNode;
+    }
 }