Merge pull request #19845 from taosdata/enh/3.0_planner_optimize

feat: add explain test case

include/libs/nodes/plannodes.h

@@ -562,6 +562,7 @@ typedef struct SQueryInserterNode {
   char          tableName[TSDB_TABLE_NAME_LEN];
   int32_t       vgId;
   SEpSet        epSet;
+  bool          explain;
 } SQueryInserterNode;
 
 typedef struct SDataDeleterNode {

source/libs/executor/src/dataInserter.c

@@ -46,6 +46,7 @@ typedef struct SDataInserterHandle {
   uint64_t            cachedSize;
   TdThreadMutex       mutex;
   tsem_t              ready;
+  bool                explain;
 } SDataInserterHandle;
 
 typedef struct SSubmitRspParam {
@@ -333,26 +334,28 @@ int32_t dataBlocksToSubmitReq(SDataInserterHandle* pInserter, void** pMsg, int32
 
 static int32_t putDataBlock(SDataSinkHandle* pHandle, const SInputData* pInput, bool* pContinue) {
   SDataInserterHandle* pInserter = (SDataInserterHandle*)pHandle;
-  taosArrayPush(pInserter->pDataBlocks, &pInput->pData);
-  void*   pMsg = NULL;
-  int32_t msgLen = 0;
-  int32_t code = dataBlocksToSubmitReq(pInserter, &pMsg, &msgLen);
-  if (code) {
-    return code;
-  }
+  if (!pInserter->explain) {
+    taosArrayPush(pInserter->pDataBlocks, &pInput->pData);
+    void*   pMsg = NULL;
+    int32_t msgLen = 0;
+    int32_t code = dataBlocksToSubmitReq(pInserter, &pMsg, &msgLen);
+    if (code) {
+      return code;
+    }
 
-  taosArrayClear(pInserter->pDataBlocks);
+    taosArrayClear(pInserter->pDataBlocks);
 
-  code = sendSubmitRequest(pInserter, pMsg, msgLen, pInserter->pParam->readHandle->pMsgCb->clientRpc,
-                           &pInserter->pNode->epSet);
-  if (code) {
-    return code;
-  }
+    code = sendSubmitRequest(pInserter, pMsg, msgLen, pInserter->pParam->readHandle->pMsgCb->clientRpc,
+                             &pInserter->pNode->epSet);
+    if (code) {
+      return code;
+    }
 
-  tsem_wait(&pInserter->ready);
+    tsem_wait(&pInserter->ready);
 
-  if (pInserter->submitRes.code) {
-    return pInserter->submitRes.code;
+    if (pInserter->submitRes.code) {
+      return pInserter->submitRes.code;
+    }
   }
 
   *pContinue = true;
@@ -412,6 +415,7 @@ int32_t createDataInserter(SDataSinkManager* pManager, const SDataSinkNode* pDat
   inserter->pParam = pParam;
   inserter->status = DS_BUF_EMPTY;
   inserter->queryEnd = false;
+  inserter->explain = pInserterNode->explain;
 
   int64_t suid = 0;
   int32_t code =

source/libs/nodes/src/nodesCodeFuncs.c

@@ -176,7 +176,7 @@ const char* nodesNodeName(ENodeType type) {
     case QUERY_NODE_SHOW_DB_ALIVE_STMT:
       return "ShowDbAliveStmt";
     case QUERY_NODE_SHOW_CLUSTER_ALIVE_STMT:
-      return "ShowClusterAliveStmt";      
+      return "ShowClusterAliveStmt";
     case QUERY_NODE_REDISTRIBUTE_VGROUP_STMT:
       return "RedistributeVgroupStmt";
     case QUERY_NODE_SPLIT_VGROUP_STMT:
@@ -2595,6 +2595,7 @@ static const char* jkQueryInsertPhysiPlanTableType = "TableType";
 static const char* jkQueryInsertPhysiPlanTableFName = "TableFName";
 static const char* jkQueryInsertPhysiPlanVgId = "VgId";
 static const char* jkQueryInsertPhysiPlanEpSet = "EpSet";
+static const char* jkQueryInsertPhysiPlanExplain = "Explain";
 
 static int32_t physiQueryInsertNodeToJson(const void* pObj, SJson* pJson) {
   const SQueryInserterNode* pNode = (const SQueryInserterNode*)pObj;
@@ -2621,6 +2622,9 @@ static int32_t physiQueryInsertNodeToJson(const void* pObj, SJson* pJson) {
   if (TSDB_CODE_SUCCESS == code) {
     code = tjsonAddObject(pJson, jkQueryInsertPhysiPlanEpSet, epSetToJson, &pNode->epSet);
   }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = tjsonAddBoolToObject(pJson, jkQueryInsertPhysiPlanExplain, pNode->explain);
+  }
 
   return code;
 }
@@ -2650,6 +2654,9 @@ static int32_t jsonToPhysiQueryInsertNode(const SJson* pJson, void* pObj) {
   if (TSDB_CODE_SUCCESS == code) {
     code = tjsonToObject(pJson, jkQueryInsertPhysiPlanEpSet, jsonToEpSet, &pNode->epSet);
   }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = tjsonGetBoolValue(pJson, jkQueryInsertPhysiPlanExplain, &pNode->explain);
+  }
 
   return code;
 }

source/libs/nodes/src/nodesMsgFuncs.c

@@ -3244,7 +3244,8 @@ enum {
   PHY_QUERY_INSERT_CODE_TABLE_TYPE,
   PHY_QUERY_INSERT_CODE_TABLE_NAME,
   PHY_QUERY_INSERT_CODE_VG_ID,
-  PHY_QUERY_INSERT_CODE_EP_SET
+  PHY_QUERY_INSERT_CODE_EP_SET,
+  PHY_QUERY_INSERT_CODE_EXPLAIN
 };
 
 static int32_t physiQueryInsertNodeToMsg(const void* pObj, STlvEncoder* pEncoder) {
@@ -3272,6 +3273,9 @@ static int32_t physiQueryInsertNodeToMsg(const void* pObj, STlvEncoder* pEncoder
   if (TSDB_CODE_SUCCESS == code) {
     code = tlvEncodeObj(pEncoder, PHY_QUERY_INSERT_CODE_EP_SET, epSetToMsg, &pNode->epSet);
   }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = tlvEncodeBool(pEncoder, PHY_QUERY_INSERT_CODE_EXPLAIN, pNode->explain);
+  }
 
   return code;
 }
@@ -3307,6 +3311,9 @@ static int32_t msgToPhysiQueryInsertNode(STlvDecoder* pDecoder, void* pObj) {
       case PHY_QUERY_INSERT_CODE_EP_SET:
         code = tlvDecodeObjFromTlv(pTlv, msgToEpSet, &pNode->epSet);
         break;
+      case PHY_QUERY_INSERT_CODE_EXPLAIN:
+        code = tlvDecodeBool(pTlv, &pNode->explain);
+        break;
       default:
         break;
     }

source/libs/planner/src/planPhysiCreater.c

@@ -1718,6 +1718,7 @@ static int32_t createQueryInserter(SPhysiPlanContext* pCxt, SVnodeModifyLogicNod
   strcpy(pInserter->tableName, pModify->tableName);
   pInserter->vgId = pModify->pVgroupList->vgroups[0].vgId;
   pInserter->epSet = pModify->pVgroupList->vgroups[0].epSet;
+  pInserter->explain = (QUERY_NODE_EXPLAIN_STMT == nodeType(pCxt->pPlanCxt->pAstRoot) ? true : false);
   vgroupInfoToNodeAddr(pModify->pVgroupList->vgroups, &pSubplan->execNode);
 
   int32_t code = setListSlotId(pCxt, pSubplan->pNode->pOutputDataBlockDesc->dataBlockId, -1, pModify->pInsertCols,

source/libs/scheduler/src/schJob.c

@@ -537,7 +537,9 @@ int32_t schProcessOnExplainDone(SSchJob *pJob, SSchTask *pTask, SRetrieveTableRs
 
   SCH_SET_TASK_STATUS(pTask, JOB_TASK_STATUS_SUCC);
 
-  schProcessOnDataFetched(pJob);
+  if (!SCH_IS_INSERT_JOB(pJob)) {
+    schProcessOnDataFetched(pJob);
+  }
 
   return TSDB_CODE_SUCCESS;
 }
@@ -682,7 +684,7 @@ void schFreeJobImpl(void *job) {
 int32_t schJobFetchRows(SSchJob *pJob) {
   int32_t code = 0;
 
-  if (!(pJob->attr.explainMode == EXPLAIN_MODE_STATIC)) {
+  if (!(pJob->attr.explainMode == EXPLAIN_MODE_STATIC) && !(SCH_IS_EXPLAIN_JOB(pJob) && SCH_IS_INSERT_JOB(pJob))) {
     SCH_ERR_RET(schLaunchFetchTask(pJob));
 
     if (schChkCurrentOp(pJob, SCH_OP_FETCH, true)) {

source/libs/scheduler/src/schRemote.c

@@ -341,6 +341,14 @@ int32_t schHandleResponseMsg(SSchJob *pJob, SSchTask *pTask, int32_t execId, SDa
         SCH_ERR_JRET(TSDB_CODE_QRY_INVALID_INPUT);
       }
 
+      if (taosArrayGetSize(pTask->parents) == 0 && SCH_IS_EXPLAIN_JOB(pJob) && SCH_IS_INSERT_JOB(pJob)) {
+        SRetrieveTableRsp *pRsp = NULL;
+        SCH_ERR_JRET(qExecExplainEnd(pJob->explainCtx, &pRsp));
+        if (pRsp) {
+          SCH_ERR_JRET(schProcessOnExplainDone(pJob, pTask, pRsp));
+        }
+      }
+
       SQueryTableRsp rsp = {0};
       if (tDeserializeSQueryTableRsp(msg, msgSize, &rsp) < 0) {
         SCH_TASK_ELOG("tDeserializeSQueryTableRsp failed, msgSize:%d", msgSize);

tests/script/tsim/query/explain.sim

@@ -29,43 +29,62 @@ sql insert into tb4 values (now, 4, "Bitmap Heap Scan on tenk1 t1  (cost=5.07..2
 #sql create table tb4 using st2 tags(4);
 #sql insert into tb4 values (now, 4, "Bitmap Heap Scan on tenk1 t1  (cost=5.07..229.20 rows=101 width=244) (actual time=0.080..0.526 rows=100 loops=1)");
 
+# for explain insert into select
+sql create table t1 (ts timestamp, f1 int, f2 binary(200), t1 int);
 
 print ======== step2
 sql explain select * from st1 where -2;
+sql explain insert into t1 select * from st1 where -2;
 sql explain select ts from tb1;
+sql explain insert into t1(ts) select ts from tb1;
 sql explain select * from st1;
+sql explain insert into t1 select * from st1;
 sql explain select * from st1 order by ts;
+sql explain insert into t1 select * from st1 order by ts;
 sql explain select * from information_schema.ins_stables;
 sql explain select count(*),sum(f1) from tb1;
 sql explain select count(*),sum(f1) from st1;
 sql explain select count(*),sum(f1) from st1 group by f1;
 #sql explain select count(f1) from tb1 interval(10s, 2s) sliding(3s) fill(prev);
+sql explain insert into t1(ts, t1) select _wstart, count(*) from st1 interval(10s);
 
 print ======== step3
 sql explain verbose true select * from st1 where -2;
+sql explain verbose true insert into t1 select * from st1 where -2;
 sql explain verbose true select ts from tb1 where f1 > 0;
+sql explain verbose true insert into t1(ts) select ts from tb1 where f1 > 0;
 sql explain verbose true select * from st1 where f1 > 0 and ts > '2020-10-31 00:00:00' and ts < '2021-10-31 00:00:00';
+sql explain verbose true insert into t1 select * from st1 where f1 > 0 and ts > '2020-10-31 00:00:00' and ts < '2021-10-31 00:00:00';
 sql explain verbose true select count(*) from st1 partition by tbname slimit 1 soffset 2 limit 2 offset 1;
 sql explain verbose true select * from information_schema.ins_stables where db_name='db2';
 sql explain verbose true select st1.f1 from st1 join st2 on st1.ts=st2.ts and st1.f1 > 0;
+sql explain verbose true insert into t1(ts) select st1.f1 from st1 join st2 on st1.ts=st2.ts and st1.f1 > 0;
+sql explain verbose true insert into t1(ts, t1) select _wstart, count(*) from st1 interval(10s);
 
 print ======== step4
 sql explain analyze select ts from st1 where -2;
+sql explain analyze insert into t1(ts) select ts from st1 where -2;
 sql explain analyze select ts from tb1;
+sql explain analyze insert into t1(ts) select ts from tb1;
 sql explain analyze select ts from st1;
-sql explain analyze select ts from st1;
+sql explain analyze insert into t1(ts) select ts from st1;
 sql explain analyze select ts from st1 order by ts;
+sql explain analyze insert into t1(ts) select ts from st1 order by ts;
 sql explain analyze select * from information_schema.ins_stables;
 sql explain analyze select count(*),sum(f1) from tb1;
 sql explain analyze select count(*),sum(f1) from st1;
 sql explain analyze select count(*),sum(f1) from st1 group by f1;
+sql explain analyze insert into t1(ts, t1) select _wstart, count(*) from st1 interval(10s);
 
 print ======== step5
 sql explain analyze verbose true select ts from st1 where -2;
+sql explain analyze verbose true insert into t1(ts) select ts from st1 where -2;
 sql explain analyze verbose true select ts from tb1;
+sql explain analyze verbose true insert into t1(ts) select ts from tb1;
 sql explain analyze verbose true select ts from st1;
-sql explain analyze verbose true select ts from st1;
+sql explain analyze verbose true insert into t1(ts) select ts from st1;
 sql explain analyze verbose true select ts from st1 order by ts;
+sql explain analyze verbose true insert into t1(ts) select ts from st1 order by ts;
 sql explain analyze verbose true select * from information_schema.ins_stables;
 sql explain analyze verbose true select count(*),sum(f1) from tb1;
 sql explain analyze verbose true select count(*),sum(f1) from st1;