TD-12034 Organize planner module interface.

include/libs/planner/planner.h

@@ -22,6 +22,7 @@ extern "C" {
 
 #define QUERY_TYPE_MERGE       1
 #define QUERY_TYPE_PARTIAL     2
+#define QUERY_TYPE_SCAN        3
 
 enum OPERATOR_TYPE_E {
   OP_TableScan         = 1,
@@ -54,90 +55,37 @@ enum OPERATOR_TYPE_E {
 
 struct SEpSet;
 struct SQueryPlanNode;
-struct SQueryPhyPlanNode;
+struct SPhyNode;
 struct SQueryStmtInfo;
 
-typedef struct SSubquery {
+typedef struct SSubplan {
-  int64_t   queryId;            // the subquery id created by qnode
+  int32_t   type;               // QUERY_TYPE_MERGE|QUERY_TYPE_PARTIAL|QUERY_TYPE_SCAN
-  int32_t   type;               // QUERY_TYPE_MERGE|QUERY_TYPE_PARTIAL
+  SArray   *pDatasource;          // the datasource subplan,from which to fetch the result
-  int32_t   level;              // the execution level of current subquery, starting from 0.
+  struct SPhyNode *pNode;  // physical plan of current subplan
-  SArray   *pUpstream;          // the upstream,from which to fetch the result
+} SSubplan;
-  struct SQueryPhyPlanNode *pNode;  // physical plan of current subquery
-} SSubquery;
-
-typedef struct SQueryJob {
-  SArray  **pSubqueries;
-  int32_t   numOfLevels;
-  int32_t   currentLevel;
-} SQueryJob;
 
+typedef struct SQueryDag {
+  SArray  **pSubplans;
+} SQueryDag;
 
 /**
- * Optimize the query execution plan, currently not implement yet.
+ * Create the physical plan for the query, according to the AST.
- * @param pQueryNode
- * @return
  */
-int32_t qOptimizeQueryPlan(struct SQueryPlanNode* pQueryNode);
+int32_t qCreateQueryDag(const struct SQueryStmtInfo* pQueryInfo, struct SEpSet* pQnode, struct SQueryDag** pDag);
+
+int32_t qExplainQuery(const struct SQueryStmtInfo* pQueryInfo, struct SEpSet* pQnode, char** str);
 
 /**
- * Create the query plan according to the bound AST, which is in the form of pQueryInfo
+ * Convert to subplan to string for the scheduler to send to the executor
- * @param pQueryInfo
- * @param pQueryNode
- * @return
  */
-int32_t qCreateQueryPlan(const struct SQueryStmtInfo* pQueryInfo, struct SQueryPlanNode** pQueryNode);
+int32_t qSubPlanToString(struct SSubplan *pPhyNode, char** str);
-
-/**
- * Convert the query plan to string, in order to display it in the shell.
- * @param pQueryNode
- * @return
- */
-int32_t qQueryPlanToString(struct SQueryPlanNode* pQueryNode, char** str);
-
-/**
- * Restore the SQL statement according to the logic query plan.
- * @param pQueryNode
- * @param sql
- * @return
- */
-int32_t qQueryPlanToSql(struct SQueryPlanNode* pQueryNode, char** sql);
-
-/**
- * Create the physical plan for the query, according to the logic plan.
- * @param pQueryNode
- * @param pPhyNode
- * @return
- */
-int32_t qCreatePhysicalPlan(struct SQueryPlanNode* pQueryNode, struct SEpSet* pQnode, struct SQueryPhyPlanNode *pPhyNode);
-
-/**
- * Convert to physical plan to string to enable to print it out in the shell.
- * @param pPhyNode
- * @param str
- * @return
- */
-int32_t qPhyPlanToString(struct SQueryPhyPlanNode *pPhyNode, char** str);
-
-/**
- * Destroy the query plan object.
- * @return
- */
-void* qDestroyQueryPlan(struct SQueryPlanNode* pQueryNode);
 
 /**
  * Destroy the physical plan.
  * @param pQueryPhyNode
  * @return
  */
-void* qDestroyQueryPhyPlan(struct SQueryPhyPlanNode* pQueryPhyNode);
+void* qDestroyQueryDag(struct SQueryDag* pDag);
-
-/**
- * Create the query job from the physical execution plan
- * @param pPhyNode
- * @param pJob
- * @return
- */
-int32_t qCreateQueryJob(const struct SQueryPhyPlanNode* pPhyNode, struct SQueryJob** pJob);
 
 #ifdef __cplusplus
 }

include/libs/scheduler/scheduler.h

@@ -20,7 +20,42 @@
 extern "C" {
 #endif
 
-struct SQueryJob;
+typedef struct SQueryProfileSummary {
+  int64_t startTs;      // Object created and added into the message queue
+  int64_t endTs;        // the timestamp when the task is completed
+  int64_t cputime;      // total cpu cost, not execute elapsed time
+
+  int64_t loadRemoteDataDuration;       // remote io time
+  int64_t loadNativeDataDuration;       // native disk io time
+
+  uint64_t loadNativeData; // blocks + SMA + header files
+  uint64_t loadRemoteData; // remote data acquired by exchange operator.
+
+  uint64_t waitDuration; // the time to waiting to be scheduled in queue does matter, so we need to record it
+  int64_t  addQTs;       // the time to be added into the message queue, used to calculate the waiting duration in queue.
+
+  uint64_t totalRows;
+  uint64_t loadRows;
+  uint32_t totalBlocks;
+  uint32_t loadBlocks;
+  uint32_t loadBlockAgg;
+  uint32_t skipBlocks;
+  uint64_t resultSize;   // generated result size in Kb.
+} SQueryProfileSummary;
+
+typedef struct SQueryTask {
+  uint64_t            queryId; // query id
+  uint64_t            taskId;  // task id
+  char     *pSubplan;   // operator tree
+  uint64_t            status;  // task status
+  SQueryProfileSummary summary; // task execution summary
+  void               *pOutputHandle; // result buffer handle, to temporarily keep the output result for next stage
+} SQueryTask;
+
+typedef struct SQueryJob {
+  SArray  **pSubtasks;
+  // todo
+} SQueryJob;
 
 /**
  * Process the query job, generated according to the query physical plan.

source/libs/parser/test/plannerTest.cpp

@@ -30,6 +30,7 @@
 #include "tdef.h"
 #include "tvariant.h"
 #include "planner.h"
+#include "../../planner/inc/plannerInt.h"
 
 namespace {
 void setSchema(SSchema* p, int32_t type, int32_t bytes, const char* name, int32_t colId) {
@@ -92,10 +93,10 @@ void generateLogicplan(const char* sql) {
   ASSERT_EQ(ret, 0);
 
   struct SQueryPlanNode* n = nullptr;
-  code = qCreateQueryPlan(pQueryInfo, &n);
+  code = createQueryPlan(pQueryInfo, &n);
 
   char* str = NULL;
-  qQueryPlanToString(n, &str);
+  queryPlanToString(n, &str);
 
   printf("--------SQL:%s\n", sql);
   printf("%s\n", str);
@@ -155,10 +156,10 @@ TEST(testCase, planner_test) {
   ASSERT_EQ(pQueryInfo->fieldsInfo.numOfOutput, 2);
 
   struct SQueryPlanNode* n = nullptr;
-  code = qCreateQueryPlan(pQueryInfo, &n);
+  code = createQueryPlan(pQueryInfo, &n);
 
   char* str = NULL;
-  qQueryPlanToString(n, &str);
+  queryPlanToString(n, &str);
   printf("%s\n", str);
 
   destroyQueryInfo(pQueryInfo);

source/libs/planner/inc/plannerInt.h

@@ -25,6 +25,19 @@ extern "C" {
 #include "planner.h"
 #include "taosmsg.h"
 
+enum LOGIC_PLAN_E {
+  LP_SCAN     = 1,
+  LP_SESSION  = 2,
+  LP_STATE    = 3,
+  LP_INTERVAL = 4,
+  LP_FILL     = 5,
+  LP_AGG      = 6,
+  LP_JOIN     = 7,
+  LP_PROJECT  = 8,
+  LP_DISTINCT = 9,
+  LP_ORDER    = 10
+};
+
 typedef struct SQueryNodeBasicInfo {
   int32_t   type;          // operator type
   char     *name;          // operator name
@@ -57,68 +70,94 @@ typedef struct SQueryPlanNode {
   struct SQueryPlanNode  *nextNode;
 } SQueryPlanNode;
 
+typedef SSchema SSlotSchema;
+
 typedef struct SDataBlockSchema {
   int32_t             index;
-  SSchema            *pSchema;      // the schema of the SSDatablock
+  SSlotSchema        *pSchema;
   int32_t             numOfCols;    // number of columns
 } SDataBlockSchema;
 
-typedef struct SQueryPhyPlanNode {
+typedef struct SPhyNode {
   SQueryNodeBasicInfo info;
-  SArray             *pTarget;      // target list to be computed at this node
+  SArray             *pTargets;      // target list to be computed or scanned at this node
-  SArray             *qual;         // implicitly-ANDed qual conditions
+  SArray             *pConditions;         // implicitly-ANDed qual conditions
   SDataBlockSchema    targetSchema;
   // children plan to generated result for current node to process
   // in case of join, multiple plan nodes exist.
   SArray             *pChildren;
-} SQueryPhyPlanNode;
+} SPhyNode;
 
-typedef struct SQueryScanPhyNode {
+typedef struct SScanPhyNode {
-  SQueryPhyPlanNode node;
+  SPhyNode node;
-  uint64_t          uid;
+  uint64_t     uid;  // unique id of the table
-} SQueryScanPhyNode;
+} SScanPhyNode;
 
-typedef struct SQueryProjectPhyNode {
+typedef SScanPhyNode STagScanPhyNode;
-  SQueryPhyPlanNode node;
-} SQueryProjectPhyNode;
 
-typedef struct SQueryAggPhyNode {
+typedef SScanPhyNode SSystemTableScanPhyNode;
-  SQueryPhyPlanNode node;
-  SArray           *pGroup;
-  // SInterval
-} SQueryAggPhyNode;
 
-typedef struct SQueryProfileSummary {
+typedef struct SMultiTableScanPhyNode {
-  int64_t startTs;      // Object created and added into the message queue
+  SScanPhyNode scan;
-  int64_t endTs;        // the timestamp when the task is completed
+  SArray      *pTagsConditions; // implicitly-ANDed tag qual conditions
-  int64_t cputime;      // total cpu cost, not execute elapsed time
+} SMultiTableScanPhyNode;
 
-  int64_t loadRemoteDataDuration;       // remote io time
+typedef SMultiTableScanPhyNode SMultiTableSeqScanPhyNode;
-  int64_t loadNativeDataDuration;       // native disk io time
 
-  uint64_t loadNativeData; // blocks + SMA + header files
+typedef struct SProjectPhyNode {
-  uint64_t loadRemoteData; // remote data acquired by exchange operator.
+  SPhyNode node;
+} SProjectPhyNode;
 
-  uint64_t waitDuration; // the time to waiting to be scheduled in queue does matter, so we need to record it
+/**
-  int64_t  addQTs;       // the time to be added into the message queue, used to calculate the waiting duration in queue.
+ * Optimize the query execution plan, currently not implement yet.
+ * @param pQueryNode
+ * @return
+ */
+int32_t optimizeQueryPlan(struct SQueryPlanNode* pQueryNode);
 
-  uint64_t totalRows;
+/**
-  uint64_t loadRows;
+ * Create the query plan according to the bound AST, which is in the form of pQueryInfo
-  uint32_t totalBlocks;
+ * @param pQueryInfo
-  uint32_t loadBlocks;
+ * @param pQueryNode
-  uint32_t loadBlockAgg;
+ * @return
-  uint32_t skipBlocks;
+ */
-  uint64_t resultSize;   // generated result size in Kb.
+int32_t createQueryPlan(const struct SQueryStmtInfo* pQueryInfo, struct SQueryPlanNode** pQueryNode);
-} SQueryProfileSummary;
 
-typedef struct SQueryTask {
+/**
-  uint64_t            queryId; // query id
+ * Convert the query plan to string, in order to display it in the shell.
-  uint64_t            taskId;  // task id
+ * @param pQueryNode
-  SQueryPhyPlanNode *pNode;   // operator tree
+ * @return
-  uint64_t            status;  // task status
+ */
-  SQueryProfileSummary summary; // task execution summary
+int32_t queryPlanToString(struct SQueryPlanNode* pQueryNode, char** str);
-  void               *pOutputHandle; // result buffer handle, to temporarily keep the output result for next stage
+
-} SQueryTask;
+/**
+ * Restore the SQL statement according to the logic query plan.
+ * @param pQueryNode
+ * @param sql
+ * @return
+ */
+int32_t queryPlanToSql(struct SQueryPlanNode* pQueryNode, char** sql);
+
+/**
+ * Convert to physical plan to string to enable to print it out in the shell.
+ * @param pPhyNode
+ * @param str
+ * @return
+ */
+int32_t phyPlanToString(struct SPhyNode *pPhyNode, char** str);
+
+/**
+ * Destroy the query plan object.
+ * @return
+ */
+void* destroyQueryPlan(struct SQueryPlanNode* pQueryNode);
+
+/**
+ * Destroy the physical plan.
+ * @param pQueryPhyNode
+ * @return
+ */
+void* destroyQueryPhyPlan(struct SPhyNode* pQueryPhyNode);
 
 #ifdef __cplusplus
 }

source/libs/planner/src/physicalPlan.c

@@ -0,0 +1,36 @@
+/*
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#include "plannerInt.h"
+
+SPhyNode* createScanNode(SQueryPlanNode* pPlanNode) {
+  return NULL;
+}
+
+SPhyNode* createPhyNode(SQueryPlanNode* node) {
+  switch (node->info.type) {
+    case LP_SCAN:
+      return createScanNode(node);
+  }
+  return NULL;
+}
+
+SPhyNode* createSubplan(SQueryPlanNode* pSubquery) {
+  return NULL;
+}
+
+int32_t createDag(struct SQueryPlanNode* pQueryNode, struct SEpSet* pQnode, struct SQueryDag** pDag) {
+  return 0;
+}

source/libs/planner/src/planner.c

@@ -48,11 +48,11 @@ static SArray* createQueryPlanImpl(SQueryStmtInfo* pQueryInfo);
 static void doDestroyQueryNode(SQueryPlanNode* pQueryNode);
 
 int32_t printExprInfo(const char* buf, const SQueryPlanNode* pQueryNode, int32_t len);
-int32_t qOptimizeQueryPlan(struct SQueryPlanNode* pQueryNode) {
+int32_t optimizeQueryPlan(struct SQueryPlanNode* pQueryNode) {
   return 0;
 }
 
-int32_t qCreateQueryPlan(const struct SQueryStmtInfo* pQueryInfo, struct SQueryPlanNode** pQueryNode) {
+int32_t createQueryPlan(const struct SQueryStmtInfo* pQueryInfo, struct SQueryPlanNode** pQueryNode) {
   SArray* upstream = createQueryPlanImpl((struct SQueryStmtInfo*) pQueryInfo);
   assert(taosArrayGetSize(upstream) == 1);
 
@@ -62,20 +62,20 @@ int32_t qCreateQueryPlan(const struct SQueryStmtInfo* pQueryInfo, struct SQueryP
   return TSDB_CODE_SUCCESS;
 }
 
-int32_t qQueryPlanToSql(struct SQueryPlanNode* pQueryNode, char** sql) {
+int32_t queryPlanToSql(struct SQueryPlanNode* pQueryNode, char** sql) {
   return 0;
 }
 
-int32_t qCreatePhysicalPlan(struct SQueryPlanNode* pQueryNode, struct SEpSet* pQnode, struct SQueryPhyPlanNode *pPhyNode) {
+int32_t qCreatePhysicalPlan(struct SQueryPlanNode* pQueryNode, struct SEpSet* pQnode, struct SQueryDag** pDag) {
 
   return 0;
 }
 
-int32_t qPhyPlanToString(struct SQueryPhyPlanNode *pPhyNode, char** str) {
+int32_t phyPlanToString(struct SPhyNode *pPhyNode, char** str) {
   return 0;
 }
 
-void* qDestroyQueryPlan(SQueryPlanNode* pQueryNode) {
+void* destroyQueryPlan(SQueryPlanNode* pQueryNode) {
   if (pQueryNode == NULL) {
     return NULL;
   }
@@ -84,14 +84,10 @@ void* qDestroyQueryPlan(SQueryPlanNode* pQueryNode) {
   return NULL;
 }
 
-void* qDestroyQueryPhyPlan(struct SQueryPhyPlanNode* pQueryPhyNode) {
+void* destroyQueryPhyPlan(struct SPhyNode* pQueryPhyNode) {
   return NULL;
 }
 
-int32_t qCreateQueryJob(const struct SQueryPhyPlanNode* pPhyNode, struct SQueryJob** pJob) {
-  return 0;
-}
-
 //======================================================================================================================
 
 static SQueryPlanNode* createQueryNode(int32_t type, const char* name, SQueryPlanNode** prev, int32_t numOfPrev,
@@ -620,7 +616,7 @@ int32_t queryPlanToStringImpl(char* buf, SQueryPlanNode* pQueryNode, int32_t lev
   return len;
 }
 
-int32_t qQueryPlanToString(struct SQueryPlanNode* pQueryNode, char** str) {
+int32_t queryPlanToString(struct SQueryPlanNode* pQueryNode, char** str) {
   assert(pQueryNode);
 
   *str = calloc(1, 4096);