fix: CteMaterializer single instance

Author: shizy818

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/QueryExecution.java

@@ -47,7 +47,7 @@
 import org.apache.iotdb.db.queryengine.plan.planner.plan.FragmentInstance;
 import org.apache.iotdb.db.queryengine.plan.planner.plan.LogicalQueryPlan;
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeUtil;
-import org.apache.iotdb.db.queryengine.plan.relational.planner.ir.CteMaterializer;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.CteMaterializer;
 import org.apache.iotdb.db.queryengine.plan.scheduler.IScheduler;
 import org.apache.iotdb.db.utils.SetThreadName;
 import org.apache.iotdb.mpp.rpc.thrift.TFragmentInstanceId;
@@ -351,7 +351,7 @@ private void releaseResource() {
       cleanUpResultHandle();
     }
     if (getSQLDialect().equals(IClientSession.SqlDialect.TABLE) && !context.isSubquery()) {
-      CteMaterializer.cleanUpCTE(context);
+      CteMaterializer.getInstance().cleanUpCTE(context);
     }
   }
 
@@ -397,7 +397,7 @@ private void releaseResource(Throwable t) {
       cleanUpResultHandle();
     }
     if (getSQLDialect().equals(IClientSession.SqlDialect.TABLE) && !context.isSubquery()) {
-      CteMaterializer.cleanUpCTE(context);
+      CteMaterializer.getInstance().cleanUpCTE(context);
     }
   }
 

…lational/planner/ir/CteMaterializer.java …/relational/planner/CteMaterializer.javaiotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/ir/CteMaterializer.java renamed to iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/CteMaterializer.java iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/ir/CteMaterializer.java renamed to iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/CteMaterializer.java

@@ -19,7 +19,7 @@
  *
  */
 
-package org.apache.iotdb.db.queryengine.plan.relational.planner.ir;
+package org.apache.iotdb.db.queryengine.plan.relational.planner;
 
 import org.apache.iotdb.common.rpc.thrift.TEndPoint;
 import org.apache.iotdb.commons.client.IClientManager;
@@ -70,13 +70,15 @@
 public class CteMaterializer {
   private static final Logger LOGGER = LoggerFactory.getLogger(CteMaterializer.class);
   private static String CTE_MATERIALIZATION_FAILURE_WARNING =
-      "***** CTE MATERIALIZATION failed! INLINE mode is adopted in main query *****";
+      "***** Failed to materialize CTE! INLINE mode is adopted in the main query *****";
 
   private static final Coordinator coordinator = Coordinator.getInstance();
 
-  private CteMaterializer() {}
+  public void materializeCTE(Analysis analysis, MPPQueryContext context) {
+    if (coordinator == null) {
+      return;
+    }
 
-  public static void materializeCTE(Analysis analysis, MPPQueryContext context) {
     analysis
         .getNamedQueries()
         .forEach(
@@ -104,7 +106,7 @@ public static void materializeCTE(Analysis analysis, MPPQueryContext context) {
             });
   }
 
-  public static void cleanUpCTE(MPPQueryContext context) {
+  public void cleanUpCTE(MPPQueryContext context) {
     Map<NodeRef<Table>, CteDataStore> cteDataStores = context.getCteDataStores();
     cteDataStores
         .values()
@@ -116,8 +118,7 @@ public static void cleanUpCTE(MPPQueryContext context) {
     cteDataStores.clear();
   }
 
-  private static CteDataStore fetchCteQueryResult(
-      Table table, Query query, MPPQueryContext context) {
+  private CteDataStore fetchCteQueryResult(Table table, Query query, MPPQueryContext context) {
     final long queryId = SessionManager.getInstance().requestQueryId();
     Throwable t = null;
     try {
@@ -184,7 +185,7 @@ private static CteDataStore fetchCteQueryResult(
     return null;
   }
 
-  private static TableSchema getTableSchema(DatasetHeader datasetHeader, String cteName) {
+  private TableSchema getTableSchema(DatasetHeader datasetHeader, String cteName) {
     final List<String> columnNames = datasetHeader.getRespColumns();
     final List<TSDataType> columnDataTypes = datasetHeader.getRespDataTypes();
     if (columnNames.size() != columnDataTypes.size()) {
@@ -213,7 +214,7 @@ private static TableSchema getTableSchema(DatasetHeader datasetHeader, String ct
     return new TableSchema(cteName, columnSchemaList);
   }
 
-  private static List<String> getCteExplainAnalyzeLines(
+  private List<String> getCteExplainAnalyzeLines(
       FragmentInstanceStatisticsDrawer fragmentInstanceStatisticsDrawer,
       List<FragmentInstance> instances,
       boolean verbose)
@@ -231,7 +232,7 @@ private static List<String> getCteExplainAnalyzeLines(
     return statisticLines.stream().map(StatisticLine::getValue).collect(Collectors.toList());
   }
 
-  private static void handleCteExplainAnalyzeResults(
+  private void handleCteExplainAnalyzeResults(
       MPPQueryContext context, long queryId, Table table, String warnMessage) {
     QueryExecution execution = (QueryExecution) coordinator.getQueryExecution(queryId);
     DistributedQueryPlan distributedQueryPlan = execution.getDistributedPlan();
@@ -265,7 +266,7 @@ private static void handleCteExplainAnalyzeResults(
     }
   }
 
-  private static void handleCteExplainResults(MPPQueryContext context, long queryId, Table table) {
+  private void handleCteExplainResults(MPPQueryContext context, long queryId, Table table) {
     QueryExecution execution = (QueryExecution) coordinator.getQueryExecution(queryId);
     DistributedQueryPlan distributedQueryPlan = execution.getDistributedPlan();
     if (distributedQueryPlan == null) {
@@ -276,4 +277,16 @@ private static void handleCteExplainResults(MPPQueryContext context, long queryI
     List<String> lines = distributedQueryPlan.getPlanText();
     context.addCteExplainResult(table, new Pair<>(-1, lines));
   }
+
+  private static class CteMaterializerHolder {
+    private static final CteMaterializer INSTANCE = new CteMaterializer();
+
+    private CteMaterializerHolder() {
+      // Empty constructor
+    }
+  }
+
+  public static CteMaterializer getInstance() {
+    return CteMaterializerHolder.INSTANCE;
+  }
 }

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

@@ -26,6 +26,7 @@
 import org.apache.iotdb.commons.schema.table.TreeViewSchema;
 import org.apache.iotdb.commons.schema.table.TsTable;
 import org.apache.iotdb.commons.utils.TestOnly;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
 import org.apache.iotdb.db.exception.sql.SemanticException;
 import org.apache.iotdb.db.queryengine.common.MPPQueryContext;
 import org.apache.iotdb.db.queryengine.common.QueryId;
@@ -50,7 +51,6 @@
 import org.apache.iotdb.db.queryengine.plan.relational.metadata.QualifiedObjectName;
 import org.apache.iotdb.db.queryengine.plan.relational.metadata.TableMetadataImpl;
 import org.apache.iotdb.db.queryengine.plan.relational.metadata.TableSchema;
-import org.apache.iotdb.db.queryengine.plan.relational.planner.ir.CteMaterializer;
 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.IntoNode;
@@ -374,7 +374,10 @@ private RelationPlan createRelationPlan(Analysis analysis, PipeEnriched pipeEnri
 
   private RelationPlan createRelationPlan(Analysis analysis, Query query) {
     // materialize cte if needed
-    CteMaterializer.materializeCTE(analysis, queryContext);
+    final int dataNodeId = IoTDBDescriptor.getInstance().getConfig().getDataNodeId();
+    if (dataNodeId != -1) {
+      CteMaterializer.getInstance().materializeCTE(analysis, queryContext);
+    }
     return getRelationPlanner(analysis).process(query, null);
   }