Optimize the implantation of EXPLAIN ANALYZE

Author: ycycse

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

@@ -82,7 +82,6 @@
 import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.StreamingHashAggregationOperator;
 import org.apache.iotdb.db.queryengine.execution.relational.ColumnTransformerBuilder;
 import org.apache.iotdb.db.queryengine.plan.analyze.TypeProvider;
-import org.apache.iotdb.db.queryengine.plan.planner.plan.node.ExplainAnalyzeNode;
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNode;
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeId;
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanVisitor;
@@ -110,6 +109,7 @@
 import org.apache.iotdb.db.queryengine.plan.relational.planner.node.AggregationNode;
 import org.apache.iotdb.db.queryengine.plan.relational.planner.node.AggregationTableScanNode;
 import org.apache.iotdb.db.queryengine.plan.relational.planner.node.CollectNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.ExplainAnalyzeNode;
 import org.apache.iotdb.db.queryengine.plan.relational.planner.node.FilterNode;
 import org.apache.iotdb.db.queryengine.plan.relational.planner.node.GapFillNode;
 import org.apache.iotdb.db.queryengine.plan.relational.planner.node.JoinNode;

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

@@ -421,6 +421,10 @@ public PlanNode visitExplainAnalyze(ExplainAnalyzeNode node, NodeGroupContext co
     newNode.setChild(exchangeNode);
 
     context.hasExchangeNode = true;
+    context.putNodeDistribution(
+        newNode.getPlanNodeId(),
+        new NodeDistribution(
+            NodeDistributionType.DIFFERENT_FROM_ALL_CHILDREN, DataPartition.NOT_ASSIGNED));
     return newNode;
   }
 

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

@@ -21,9 +21,6 @@
 
 import org.apache.iotdb.db.queryengine.common.header.ColumnHeaderConstant;
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.SingleChildProcessNode;
-import org.apache.iotdb.db.queryengine.plan.relational.planner.Symbol;
-
-import org.apache.tsfile.utils.ReadWriteIOUtils;
 
 import java.io.DataOutputStream;
 import java.io.IOException;
@@ -60,36 +57,21 @@ public List<String> getOutputColumnNames() {
     return Collections.singletonList(ColumnHeaderConstant.EXPLAIN_ANALYZE);
   }
 
-  @Override
-  public List<Symbol> getOutputSymbols() {
-    return Collections.singletonList(Symbol.of(ColumnHeaderConstant.EXPLAIN_ANALYZE));
-  }
-
   @Override
   public PlanNode replaceChildren(List<PlanNode> newChildren) {
     return new ExplainAnalyzeNode(getPlanNodeId(), newChildren.get(0), verbose, queryId, timeout);
   }
 
+  // ExplainAnalyze should be at the same region as Coordinator all the time. Therefore, there will
+  // be no serialization and deserialization process.
   @Override
   protected void serializeAttributes(ByteBuffer byteBuffer) {
-    PlanNodeType.EXPLAIN_ANALYZE.serialize(byteBuffer);
-    ReadWriteIOUtils.write(verbose, byteBuffer);
+    throw new UnsupportedOperationException("ExplainAnalyzeNode should not be serialized");
   }
 
   @Override
   protected void serializeAttributes(DataOutputStream stream) throws IOException {
-    PlanNodeType.EXPLAIN_ANALYZE.serialize(stream);
-    ReadWriteIOUtils.write(verbose, stream);
-    ReadWriteIOUtils.write(queryId, stream);
-    ReadWriteIOUtils.write(timeout, stream);
-  }
-
-  public static ExplainAnalyzeNode deserialize(ByteBuffer byteBuffer) {
-    boolean verbose = ReadWriteIOUtils.readBool(byteBuffer);
-    long queryId = ReadWriteIOUtils.readLong(byteBuffer);
-    PlanNodeId planNodeId = PlanNodeId.deserialize(byteBuffer);
-    long timeout = ReadWriteIOUtils.readLong(byteBuffer);
-    return new ExplainAnalyzeNode(planNodeId, null, verbose, queryId, timeout);
+    throw new UnsupportedOperationException("ExplainAnalyzeNode should not be serialized");
   }
 
   public boolean isVerbose() {

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

@@ -276,6 +276,7 @@ public enum PlanNodeType {
   TABLE_AGGREGATION_NODE((short) 1015),
   TABLE_AGGREGATION_TABLE_SCAN_NODE((short) 1016),
   TABLE_GAP_FILL_NODE((short) 1017),
+  TABLE_EXPLAIN_ANALYZE_NODE((short) 1018),
 
   RELATIONAL_INSERT_TABLET((short) 2000),
   RELATIONAL_INSERT_ROW((short) 2001),
@@ -536,7 +537,7 @@ public static PlanNode deserialize(ByteBuffer buffer, short nodeType) {
       case 89:
         return AggregationMergeSortNode.deserialize(buffer);
       case 90:
-        return ExplainAnalyzeNode.deserialize(buffer);
+        throw new UnsupportedOperationException("ExplainAnalyzeNode should not be serialized");
       case 91:
         return PipeOperateSchemaQueueNode.deserialize(buffer);
       case 92:
@@ -626,6 +627,8 @@ public static PlanNode deserialize(ByteBuffer buffer, short nodeType) {
             .deserialize(buffer);
       case 1017:
         return GapFillNode.deserialize(buffer);
+      case 1018:
+        throw new UnsupportedOperationException("ExplainAnalyzeNode should not be deserialized");
       case 2000:
         return RelationalInsertTabletNode.deserialize(buffer);
       case 2001:

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

@@ -666,6 +666,12 @@ public R visitMergeSort(
     return visitMultiChildProcess(node, context);
   }
 
+  public R visitExplainAnalyze(
+      org.apache.iotdb.db.queryengine.plan.relational.planner.node.ExplainAnalyzeNode node,
+      C context) {
+    return visitSingleChildProcess(node, context);
+  }
+
   public R visitOutput(
       org.apache.iotdb.db.queryengine.plan.relational.planner.node.OutputNode node, C context) {
     return visitSingleChildProcess(node, context);

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

@@ -201,6 +201,8 @@ public class Analysis implements IAnalysis {
   // if emptyDataSource, there is no need to execute the query in BE
   private boolean emptyDataSource = false;
 
+  private boolean isQuery = false;
+
   public DataPartition getDataPartition() {
     return dataPartition;
   }
@@ -752,7 +754,11 @@ public TsBlock constructResultForMemorySource(MPPQueryContext context) {
 
   @Override
   public boolean isQuery() {
-    return getStatement().isQuery();
+    return isQuery;
+  }
+
+  public void setQuery(boolean query) {
+    isQuery = query;
   }
 
   @Override

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/Analyzer.java

@@ -90,7 +90,7 @@ public Analysis analyze(Statement statement) {
             analysis, context, session, warningCollector, CorrelationSupport.ALLOWED);
 
     analyzer.analyze(statement);
-    if (statement.isQuery()) {
+    if (analysis.isQuery()) {
       long analyzeCost = System.nanoTime() - startTime;
       QueryPlanCostMetricSet.getInstance().recordPlanCost(TABLE_TYPE, ANALYZER, analyzeCost);
       context.setAnalyzeCost(analyzeCost);

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/StatementAnalyzer.java

@@ -601,6 +601,7 @@ protected Scope visitExplainAnalyze(ExplainAnalyze node, Optional<Scope> context
 
     @Override
     protected Scope visitQuery(Query node, Optional<Scope> context) {
+      analysis.setQuery(true);
       Scope withScope = analyzeWith(node, context);
       hasFillInParentScope = node.getFill().isPresent() || hasFillInParentScope;
       Scope queryBodyScope = process(node.getQueryBody(), withScope);

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

@@ -25,7 +25,6 @@
 import org.apache.iotdb.db.queryengine.execution.warnings.WarningCollector;
 import org.apache.iotdb.db.queryengine.metric.QueryPlanCostMetricSet;
 import org.apache.iotdb.db.queryengine.plan.planner.plan.LogicalQueryPlan;
-import org.apache.iotdb.db.queryengine.plan.planner.plan.node.ExplainAnalyzeNode;
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNode;
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.WritePlanNode;
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metadata.read.CountSchemaMergeNode;
@@ -34,6 +33,7 @@
 import org.apache.iotdb.db.queryengine.plan.relational.analyzer.RelationType;
 import org.apache.iotdb.db.queryengine.plan.relational.execution.querystats.PlanOptimizersStatsCollector;
 import org.apache.iotdb.db.queryengine.plan.relational.metadata.Metadata;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.ExplainAnalyzeNode;
 import org.apache.iotdb.db.queryengine.plan.relational.planner.node.FilterNode;
 import org.apache.iotdb.db.queryengine.plan.relational.planner.node.LimitNode;
 import org.apache.iotdb.db.queryengine.plan.relational.planner.node.OffsetNode;
@@ -67,6 +67,7 @@
 import com.google.common.collect.ImmutableMap;
 import org.apache.tsfile.enums.TSDataType;
 import org.apache.tsfile.read.common.type.LongType;
+import org.apache.tsfile.read.common.type.StringType;
 import org.apache.tsfile.read.common.type.TypeFactory;
 
 import java.util.ArrayList;
@@ -127,7 +128,7 @@ public LogicalQueryPlan plan(final Analysis analysis) {
     final Statement statement = analysis.getStatement();
     PlanNode planNode = planStatement(analysis, statement);
 
-    if (statement.isQuery()) {
+    if (analysis.isQuery()) {
       long logicalPlanCostTime = System.nanoTime() - startTime;
       QueryPlanCostMetricSet.getInstance()
           .recordPlanCost(TABLE_TYPE, LOGICAL_PLANNER, logicalPlanCostTime);
@@ -448,13 +449,16 @@ private void planTraverseDevice(final AbstractTraverseDevice statement, final An
   private RelationPlan planExplainAnalyze(final ExplainAnalyze statement, final Analysis analysis) {
     RelationPlan originalQueryPlan =
         createRelationPlan(analysis, (Query) (statement.getStatement()));
+    Symbol symbol =
+        symbolAllocator.newSymbol(ColumnHeaderConstant.EXPLAIN_ANALYZE, StringType.getInstance());
     PlanNode newRoot =
         new ExplainAnalyzeNode(
             queryContext.getQueryId().genPlanNodeId(),
             originalQueryPlan.getRoot(),
             statement.isVerbose(),
             queryContext.getLocalQueryId(),
-            queryContext.getTimeOut());
+            queryContext.getTimeOut(),
+            symbol);
     return new RelationPlan(
         newRoot,
         originalQueryPlan.getScope(),

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

@@ -20,19 +20,21 @@
 package org.apache.iotdb.db.queryengine.plan.relational.planner.distribute;
 
 import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
+import org.apache.iotdb.commons.partition.DataPartition;
 import org.apache.iotdb.db.queryengine.common.MPPQueryContext;
 import org.apache.iotdb.db.queryengine.plan.planner.distribution.NodeDistribution;
-import org.apache.iotdb.db.queryengine.plan.planner.plan.node.ExplainAnalyzeNode;
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNode;
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanVisitor;
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.WritePlanNode;
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metadata.read.TableDeviceSourceNode;
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.ExchangeNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.ExplainAnalyzeNode;
 import org.apache.iotdb.db.queryengine.plan.relational.planner.node.TableScanNode;
 import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.TableDeviceFetchNode;
 import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.TableDeviceQueryCountNode;
 import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.TableDeviceQueryScanNode;
 
+import static org.apache.iotdb.db.queryengine.plan.planner.distribution.NodeDistributionType.DIFFERENT_FROM_ALL_CHILDREN;
 import static org.apache.iotdb.db.queryengine.plan.planner.distribution.NodeDistributionType.SAME_WITH_ALL_CHILDREN;
 import static org.apache.iotdb.db.queryengine.plan.planner.distribution.NodeDistributionType.SAME_WITH_SOME_CHILD;
 
@@ -59,14 +61,14 @@ public PlanNode visitPlan(PlanNode node, TableDistributedPlanGenerator.PlanConte
     PlanNode newNode = node.clone();
     if (node.getChildren().size() == 1) {
       newNode.addChild(node.getChildren().get(0).accept(this, context));
-      // Explain Analyze doesn't have region info, this may be solved by a local RegionReplicaSet
-      NodeDistribution childRegion =
-          context.nodeDistributionMap.get(node.getChildren().get(0).getPlanNodeId());
-      if (childRegion != null) {
-        context.nodeDistributionMap.put(
-            node.getPlanNodeId(),
-            new NodeDistribution(SAME_WITH_ALL_CHILDREN, childRegion.getRegion()));
-      }
+      context.nodeDistributionMap.put(
+          node.getPlanNodeId(),
+          new NodeDistribution(
+              SAME_WITH_ALL_CHILDREN,
+              context
+                  .nodeDistributionMap
+                  .get(node.getChildren().get(0).getPlanNodeId())
+                  .getRegion()));
       return newNode;
     }
 
@@ -134,6 +136,9 @@ public PlanNode visitExplainAnalyze(
     exchangeNode.setOutputSymbols(child.getOutputSymbols());
     newNode.setChild(exchangeNode);
 
+    context.nodeDistributionMap.put(
+        newNode.getPlanNodeId(),
+        new NodeDistribution(DIFFERENT_FROM_ALL_CHILDREN, DataPartition.NOT_ASSIGNED));
     context.hasExchangeNode = true;
     return newNode;
   }