Merge pull request #9984 from taosdata/feature/qnode

Feature/qnode

include/libs/scheduler/scheduler.h

@@ -72,7 +72,7 @@ int32_t schedulerInit(SSchedulerCfg *cfg);
  * @param nodeList  Qnode/Vnode address list, element is SQueryNodeAddr
  * @return
  */
-int32_t scheduleExecJob(void *transport, SArray *nodeList, SQueryDag* pDag, struct SSchJob** pJob, SQueryResult *pRes);
+int32_t schedulerExecJob(void *transport, SArray *nodeList, SQueryDag* pDag, struct SSchJob** pJob, SQueryResult *pRes);
 
 /**
  * Process the query job, generated according to the query physical plan.
@@ -80,7 +80,7 @@ int32_t scheduleExecJob(void *transport, SArray *nodeList, SQueryDag* pDag, stru
  * @param nodeList  Qnode/Vnode address list, element is SQueryNodeAddr
  * @return
  */
-int32_t scheduleAsyncExecJob(void *transport, SArray *nodeList, SQueryDag* pDag, struct SSchJob** pJob);
+int32_t schedulerAsyncExecJob(void *transport, SArray *nodeList, SQueryDag* pDag, struct SSchJob** pJob);
 
 /**
  * Fetch query result from the remote query executor
@@ -88,7 +88,7 @@ int32_t scheduleAsyncExecJob(void *transport, SArray *nodeList, SQueryDag* pDag,
  * @param data
  * @return
  */
-int32_t scheduleFetchRows(struct SSchJob *pJob, void **data);
+int32_t schedulerFetchRows(struct SSchJob *pJob, void **data);
 
 
 /**
@@ -102,7 +102,7 @@ int32_t scheduleFetchRows(struct SSchJob *pJob, void **data);
  * Free the query job
  * @param pJob
  */
-void scheduleFreeJob(void *pJob);
+void schedulerFreeJob(void *pJob);
 
 void schedulerDestroy(void);
 

include/util/taoserror.h

@@ -354,7 +354,7 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_QRY_SCH_NOT_EXIST             TAOS_DEF_ERROR_CODE(0, 0x0710)  //"Scheduler not exist")
 #define TSDB_CODE_QRY_TASK_NOT_EXIST            TAOS_DEF_ERROR_CODE(0, 0x0711)  //"Task not exist")
 #define TSDB_CODE_QRY_TASK_ALREADY_EXIST        TAOS_DEF_ERROR_CODE(0, 0x0712)  //"Task already exist")
-#define TSDB_CODE_QRY_RES_CACHE_NOT_EXIST       TAOS_DEF_ERROR_CODE(0, 0x0713)  //"Task result cache not exist")
+#define TSDB_CODE_QRY_TASK_CTX_NOT_EXIST        TAOS_DEF_ERROR_CODE(0, 0x0713)  //"Task context not exist")
 #define TSDB_CODE_QRY_TASK_CANCELLED            TAOS_DEF_ERROR_CODE(0, 0x0714)  //"Task cancelled")
 #define TSDB_CODE_QRY_TASK_DROPPED              TAOS_DEF_ERROR_CODE(0, 0x0715)  //"Task dropped")
 #define TSDB_CODE_QRY_TASK_CANCELLING           TAOS_DEF_ERROR_CODE(0, 0x0716)  //"Task cancelling")

source/client/src/clientImpl.c

@@ -237,12 +237,12 @@ int32_t scheduleQuery(SRequestObj* pRequest, SQueryDag* pDag, SArray* pNodeList)
   if (TSDB_SQL_INSERT == pRequest->type || TSDB_SQL_CREATE_TABLE == pRequest->type) {
     SQueryResult res = {.code = 0, .numOfRows = 0, .msgSize = ERROR_MSG_BUF_DEFAULT_SIZE, .msg = pRequest->msgBuf};
 
-    int32_t code = scheduleExecJob(pRequest->pTscObj->pAppInfo->pTransporter, NULL, pDag, &pRequest->body.pQueryJob, &res);
+    int32_t code = schedulerExecJob(pRequest->pTscObj->pAppInfo->pTransporter, NULL, pDag, &pRequest->body.pQueryJob, &res);
     if (code != TSDB_CODE_SUCCESS) {
       // handle error and retry
     } else {
       if (pRequest->body.pQueryJob != NULL) {
-        scheduleFreeJob(pRequest->body.pQueryJob);
+        schedulerFreeJob(pRequest->body.pQueryJob);
       }
     }
 
@@ -251,7 +251,7 @@ int32_t scheduleQuery(SRequestObj* pRequest, SQueryDag* pDag, SArray* pNodeList)
     return pRequest->code;
   }
 
-  return scheduleAsyncExecJob(pRequest->pTscObj->pAppInfo->pTransporter, pNodeList, pDag, &pRequest->body.pQueryJob);
+  return schedulerAsyncExecJob(pRequest->pTscObj->pAppInfo->pTransporter, pNodeList, pDag, &pRequest->body.pQueryJob);
 }
 
 typedef struct tmq_t tmq_t;
@@ -711,7 +711,7 @@ void* doFetchRow(SRequestObj* pRequest) {
       }
 
       SReqResultInfo* pResInfo = &pRequest->body.resInfo;
-      int32_t code = scheduleFetchRows(pRequest->body.pQueryJob, (void **)&pResInfo->pData);
+      int32_t code = schedulerFetchRows(pRequest->body.pQueryJob, (void **)&pResInfo->pData);
       if (code != TSDB_CODE_SUCCESS) {
         pRequest->code = code;
         return NULL;

source/dnode/mgmt/impl/src/dndTransport.c

@@ -116,6 +116,7 @@ static void dndInitMsgFp(STransMgmt *pMgmt) {
   // Requests handled by VNODE
   pMgmt->msgFp[TMSG_INDEX(TDMT_VND_SUBMIT)] = dndProcessVnodeWriteMsg;
   pMgmt->msgFp[TMSG_INDEX(TDMT_VND_QUERY)] = dndProcessVnodeQueryMsg;
+  pMgmt->msgFp[TMSG_INDEX(TDMT_VND_QUERY_CONTINUE)] = dndProcessVnodeQueryMsg;  
   pMgmt->msgFp[TMSG_INDEX(TDMT_VND_FETCH)] = dndProcessVnodeFetchMsg;
   pMgmt->msgFp[TMSG_INDEX(TDMT_VND_ALTER_TABLE)] = dndProcessVnodeWriteMsg;
   pMgmt->msgFp[TMSG_INDEX(TDMT_VND_UPDATE_TAG_VAL)] = dndProcessVnodeWriteMsg;

source/dnode/mgmt/impl/src/dndVnodes.c

@@ -892,7 +892,7 @@ int32_t dndPutReqToVQueryQ(SDnode *pDnode, SRpcMsg *pMsg) {
   SVnodeObj *pVnode = dndAcquireVnode(pDnode, pHead->vgId);
   if (pVnode == NULL) return -1;
 
-  int32_t code = dndWriteRpcMsgToVnodeQueue(pVnode->pFetchQ, pMsg, false);
+  int32_t code = dndWriteRpcMsgToVnodeQueue(pVnode->pQueryQ, pMsg, false);
   dndReleaseVnode(pDnode, pVnode);
   return code;
 }

source/dnode/vnode/src/vnd/vnodeQuery.c

@@ -25,7 +25,7 @@ int vnodeQueryOpen(SVnode *pVnode) {
 }
 
 int vnodeProcessQueryReq(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
-  vTrace("query message is processing");
+  vTrace("message in query queue is processing");
 
   switch (pMsg->msgType) {
     case TDMT_VND_QUERY:
@@ -39,7 +39,7 @@ int vnodeProcessQueryReq(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
 }
 
 int vnodeProcessFetchReq(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
-  vTrace("fetch message is processed");
+  vTrace("message in fetch queue is processing");
   switch (pMsg->msgType) {
     case TDMT_VND_FETCH:
       return qWorkerProcessFetchMsg(pVnode, pVnode->pQuery, pMsg);

source/libs/qworker/inc/qworkerInt.h

@@ -22,17 +22,17 @@ extern "C" {
 
 #include "tlockfree.h"
 
-#define QWORKER_DEFAULT_SCHEDULER_NUMBER 10000
-#define QWORKER_DEFAULT_TASK_NUMBER 10000
-#define QWORKER_DEFAULT_SCH_TASK_NUMBER 10000
-
+#define QW_DEFAULT_SCHEDULER_NUMBER 10000
+#define QW_DEFAULT_TASK_NUMBER 10000
+#define QW_DEFAULT_SCH_TASK_NUMBER 10000
+#define QW_DEFAULT_SHORT_RUN_TIMES 2
 enum {
   QW_PHASE_PRE_QUERY = 1,
   QW_PHASE_POST_QUERY,
-  QW_PHASE_PRE_CQUERY,
-  QW_PHASE_POST_CQUERY,
   QW_PHASE_PRE_FETCH,
   QW_PHASE_POST_FETCH,
+  QW_PHASE_PRE_CQUERY,
+  QW_PHASE_POST_CQUERY,
 };
 
 enum {
@@ -83,11 +83,9 @@ typedef struct SQWMsg {
 } SQWMsg;
 
 typedef struct SQWPhaseInput {
-  int8_t         status;
+  int8_t         taskStatus;
   int8_t         taskType;
   int32_t        code;
-  qTaskInfo_t    taskHandle;
-  DataSinkHandle sinkHandle;
 } SQWPhaseInput;
 
 typedef struct SQWPhaseOutput {
@@ -111,6 +109,7 @@ typedef struct SQWTaskCtx {
   void           *cancelConnection;
   
   bool            emptyRes;
+  bool            multiExec;
   int8_t          queryContinue;
   int8_t          queryInQueue;
   int32_t         rspCode; 
@@ -133,7 +132,7 @@ typedef struct SQWorkerMgmt {
   int8_t           nodeType;
   int32_t          nodeId;
   SRWLatch         schLock;
-  SRWLatch         ctxLock;
+  //SRWLatch         ctxLock;
   SHashObj        *schHash;       //key: schedulerId,    value: SQWSchStatus
   SHashObj        *ctxHash;       //key: queryId+taskId, value: SQWTaskCtx
   void            *nodeObj;
@@ -144,6 +143,8 @@ typedef struct SQWorkerMgmt {
 #define QW_IDS() sId, qId, tId
 #define QW_FPARAMS() mgmt, QW_IDS()
 
+#define QW_GET_EVENT_VALUE(ctx, event) atomic_load_8(&(ctx)->events[event])
+
 #define QW_IS_EVENT_RECEIVED(ctx, event) (atomic_load_8(&(ctx)->events[event]) == QW_EVENT_RECEIVED)
 #define QW_IS_EVENT_PROCESSED(ctx, event) (atomic_load_8(&(ctx)->events[event]) == QW_EVENT_PROCESSED)
 #define QW_SET_EVENT_RECEIVED(ctx, event) atomic_store_8(&(ctx)->events[event], QW_EVENT_RECEIVED)
@@ -151,9 +152,10 @@ typedef struct SQWorkerMgmt {
 
 #define QW_GET_PHASE(ctx) atomic_load_8(&(ctx)->phase)
 
-#define QW_SET_RSP_CODE(ctx, code) atomic_val_compare_exchange_32(&(ctx)->rspCode, 0, code)
+#define QW_SET_RSP_CODE(ctx, code) atomic_store_32(&(ctx)->rspCode, code)
+#define QW_UPDATE_RSP_CODE(ctx, code) atomic_val_compare_exchange_32(&(ctx)->rspCode, 0, code)
 
-#define QW_IN_EXECUTOR(ctx) (QW_GET_PHASE(ctx) == QW_PHASE_PRE_QUERY || QW_GET_PHASE(ctx) == QW_PHASE_PRE_CQUERY || QW_GET_PHASE(ctx) == QW_PHASE_PRE_FETCH)
+#define QW_IS_QUERY_RUNNING(ctx) (QW_GET_PHASE(ctx) == QW_PHASE_PRE_QUERY || QW_GET_PHASE(ctx) == QW_PHASE_PRE_CQUERY)
 
 #define QW_TASK_NOT_EXIST(code) (TSDB_CODE_QRY_SCH_NOT_EXIST == (code) || TSDB_CODE_QRY_TASK_NOT_EXIST == (code))
 #define QW_TASK_ALREADY_EXIST(code) (TSDB_CODE_QRY_TASK_ALREADY_EXIST == (code))

source/libs/qworker/inc/qworkerMsg.h

@@ -30,6 +30,7 @@ int32_t qwProcessFetch(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_t
 int32_t qwProcessDrop(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_t tId, SQWMsg *qwMsg);
 
 int32_t qwBuildAndSendDropRsp(void *connection, int32_t code);
+int32_t qwBuildAndSendCancelRsp(SRpcMsg *pMsg, int32_t code);
 int32_t qwBuildAndSendFetchRsp(void *connection, SRetrieveTableRsp *pRsp, int32_t dataLength, int32_t code);
 void qwBuildFetchRsp(void *msg, SOutputData *input, int32_t len);
 int32_t qwBuildAndSendCQueryMsg(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_t tId, void *connection);

source/libs/qworker/src/qworkerMsg.c

@@ -50,6 +50,7 @@ int32_t qwBuildAndSendQueryRsp(void *connection, int32_t code) {
   pRsp->code = code;
 
   SRpcMsg rpcRsp = {
+    .msgType = TDMT_VND_QUERY_RSP,
     .handle  = pMsg->handle,
     .ahandle = pMsg->ahandle,
     .pCont   = pRsp,
@@ -68,6 +69,7 @@ int32_t qwBuildAndSendReadyRsp(void *connection, int32_t code) {
   pRsp->code = code;
 
   SRpcMsg rpcRsp = {
+    .msgType = TDMT_VND_RES_READY_RSP,
     .handle  = pMsg->handle,
     .ahandle = pMsg->ahandle,
     .pCont   = pRsp,
@@ -98,7 +100,7 @@ int32_t qwBuildAndSendStatusRsp(SRpcMsg *pMsg, SSchedulerStatusRsp *sStatus) {
   }
 
   SRpcMsg rpcRsp = {
-    .msgType = pMsg->msgType + 1,
+    .msgType = TDMT_VND_TASKS_STATUS_RSP,
     .handle  = pMsg->handle,
     .ahandle = pMsg->ahandle,
     .pCont   = pRsp,
@@ -121,6 +123,7 @@ int32_t qwBuildAndSendFetchRsp(void *connection, SRetrieveTableRsp *pRsp, int32_
   }
 
   SRpcMsg rpcRsp = {
+    .msgType = TDMT_VND_FETCH_RSP,
     .handle  = pMsg->handle,
     .ahandle = pMsg->ahandle,
     .pCont   = pRsp,
@@ -138,6 +141,7 @@ int32_t qwBuildAndSendCancelRsp(SRpcMsg *pMsg, int32_t code) {
   pRsp->code = code;
 
   SRpcMsg rpcRsp = {
+    .msgType = TDMT_VND_CANCEL_TASK_RSP,
     .handle  = pMsg->handle,
     .ahandle = pMsg->ahandle,
     .pCont   = pRsp,
@@ -155,6 +159,7 @@ int32_t qwBuildAndSendDropRsp(void *connection, int32_t code) {
   pRsp->code = code;
 
   SRpcMsg rpcRsp = {
+    .msgType = TDMT_VND_DROP_TASK_RSP,
     .handle  = pMsg->handle,
     .ahandle = pMsg->ahandle,
     .pCont   = pRsp,
@@ -273,7 +278,7 @@ int32_t qWorkerProcessQueryMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
   SQWorkerMgmt *mgmt = (SQWorkerMgmt *)qWorkerMgmt;
   
   if (NULL == msg || pMsg->contLen <= sizeof(*msg)) {
-    QW_ELOG("invalid query msg, contLen:%d", pMsg->contLen);
+    QW_ELOG("invalid query msg, msg:%p, msgLen:%d", msg, pMsg->contLen);
     QW_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT);
   }
 
@@ -306,15 +311,11 @@ int32_t qWorkerProcessCQueryMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
   SQWTaskCtx *handles = NULL;
   SQWorkerMgmt *mgmt = (SQWorkerMgmt *)qWorkerMgmt;
 
-  if (NULL == msg || pMsg->contLen <= sizeof(*msg)) {
-    QW_ELOG("invalid cquery msg, contLen:%d", pMsg->contLen);
+  if (NULL == msg || pMsg->contLen < sizeof(*msg)) {
+    QW_ELOG("invalid cquery msg, msg:%p, msgLen:%d", msg, pMsg->contLen);
     QW_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT);
   }
 
-  msg->sId = be64toh(msg->sId);
-  msg->queryId = be64toh(msg->queryId);
-  msg->taskId = be64toh(msg->taskId);
-  
   uint64_t sId = msg->sId;
   uint64_t qId = msg->queryId;
   uint64_t tId = msg->taskId;
@@ -335,14 +336,13 @@ int32_t qWorkerProcessReadyMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg){
     return TSDB_CODE_QRY_INVALID_INPUT;
   }
 
+  SQWorkerMgmt *mgmt = (SQWorkerMgmt *)qWorkerMgmt;
   SResReadyReq *msg = pMsg->pCont;
   if (NULL == msg || pMsg->contLen < sizeof(*msg)) {
-    qError("invalid task status msg");  
+    QW_ELOG("invalid task ready msg, msg:%p, msgLen:%d", msg, pMsg->contLen);
     QW_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT);
   }  
 
-  SQWorkerMgmt *mgmt = (SQWorkerMgmt *)qWorkerMgmt;
-
   msg->sId = be64toh(msg->sId);
   msg->queryId = be64toh(msg->queryId);
   msg->taskId = be64toh(msg->taskId);
@@ -398,6 +398,7 @@ int32_t qWorkerProcessFetchMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
   SQWorkerMgmt *mgmt = (SQWorkerMgmt *)qWorkerMgmt;
   
   if (NULL == msg || pMsg->contLen < sizeof(*msg)) {
+    QW_ELOG("invalid fetch msg, msg:%p, msgLen:%d", msg, pMsg->contLen);  
     QW_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT);
   }  
 

source/libs/scheduler/src/scheduler.c

@@ -275,10 +275,13 @@ int32_t schBuildTaskRalation(SSchJob *pJob, SHashObj *planToTask) {
 }
 
 
-int32_t schRecordTaskSucceedNode(SSchTask *pTask) {
-  SQueryNodeAddr *addr = taosArrayGet(pTask->candidateAddrs, atomic_load_8(&pTask->candidateIdx));
-
-  assert(NULL != addr);
+int32_t schRecordTaskSucceedNode(SSchJob *pJob, SSchTask *pTask) {
+  int32_t idx = atomic_load_8(&pTask->candidateIdx);
+  SQueryNodeAddr *addr = taosArrayGet(pTask->candidateAddrs, idx);
+  if (NULL == addr) {
+    SCH_TASK_ELOG("taosArrayGet candidate addr failed, idx:%d, size:%d", idx, (int32_t)taosArrayGetSize(pTask->candidateAddrs));
+    SCH_ERR_RET(TSDB_CODE_SCH_INTERNAL_ERROR);
+  }
 
   pTask->succeedAddr = *addr;
 
@@ -578,9 +581,10 @@ int32_t schProcessOnJobFailureImpl(SSchJob *pJob, int32_t status, int32_t errCod
     tsem_post(&pJob->rspSem);
   }
 
-  SCH_ERR_RET(atomic_load_32(&pJob->errCode));
+  int32_t code = atomic_load_32(&pJob->errCode);
+  SCH_ERR_RET(code);
 
-  assert(0);
+  SCH_JOB_ELOG("job errCode is invalid, errCode:%d", code);
 }
 
 
@@ -721,7 +725,7 @@ int32_t schProcessOnTaskSuccess(SSchJob *pJob, SSchTask *pTask) {
 
   SCH_SET_TASK_STATUS(pTask, JOB_TASK_STATUS_PARTIAL_SUCCEED);
 
-  SCH_ERR_JRET(schRecordTaskSucceedNode(pTask));
+  SCH_ERR_JRET(schRecordTaskSucceedNode(pJob, pTask));
   
   int32_t parentNum = pTask->parents ? (int32_t)taosArrayGetSize(pTask->parents) : 0;
   if (parentNum == 0) {
@@ -738,7 +742,7 @@ int32_t schProcessOnTaskSuccess(SSchJob *pJob, SSchTask *pTask) {
         
         return TSDB_CODE_SUCCESS;
       } else if (taskDone > pTask->level->taskNum) {
-        assert(0);
+        SCH_TASK_ELOG("taskDone number invalid, done:%d, total:%d", taskDone, pTask->level->taskNum);
       }
 
       if (pTask->level->taskFailed > 0) {
@@ -871,18 +875,21 @@ int32_t schHandleResponseMsg(SSchJob *pJob, SSchTask *pTask, int32_t msgType, ch
         }
 
         atomic_store_ptr(&pJob->res, rsp);
-        atomic_store_32(&pJob->resNumOfRows, rsp->numOfRows);
+        atomic_add_fetch_32(&pJob->resNumOfRows, htonl(rsp->numOfRows));
 
         if (rsp->completed) {
           SCH_SET_TASK_STATUS(pTask, JOB_TASK_STATUS_SUCCEED);
         }
 
+        SCH_TASK_DLOG("got fetch rsp, rows:%d, complete:%d", htonl(rsp->numOfRows), rsp->completed);
+
         SCH_ERR_JRET(schProcessOnDataFetched(pJob));
         break;
       }
-    case TDMT_VND_DROP_TASK: {
+    case TDMT_VND_DROP_TASK_RSP: {
         // SHOULD NEVER REACH HERE
-        assert(0);
+        SCH_TASK_ELOG("invalid status to handle drop task rsp, ref:%d", atomic_load_32(&pJob->ref));
+        SCH_ERR_JRET(TSDB_CODE_SCH_INTERNAL_ERROR);
         break;
       }
     default:
@@ -1030,6 +1037,8 @@ int32_t schAsyncSendMsg(void *transport, SEpSet* epSet, uint64_t qId, uint64_t t
     qError("QID:%"PRIx64 ",TID:%"PRIx64 " asyncSendMsgToServer failed, code:%x", qId, tId, code);
     SCH_ERR_JRET(code);
   }
+
+  qDebug("QID:%"PRIx64 ",TID:%"PRIx64 " req msg sent, type:%d, %s", qId, tId, msgType, TMSG_INFO(msgType));
   
   return TSDB_CODE_SUCCESS;
 
@@ -1289,6 +1298,8 @@ void schDropJobAllTasks(SSchJob *pJob) {
 }
 
 int32_t schExecJobImpl(void *transport, SArray *nodeList, SQueryDag* pDag, struct SSchJob** job, bool syncSchedule) {
+  qDebug("QID:%"PRIx64" job started", pDag->queryId);
+  
   if (nodeList && taosArrayGetSize(nodeList) <= 0) {
     qInfo("QID:%"PRIx64" input nodeList is empty", pDag->queryId);
   }
@@ -1356,7 +1367,7 @@ _return:
 
   *(SSchJob **)job = NULL;
   
-  scheduleFreeJob(pJob);
+  schedulerFreeJob(pJob);
   
   SCH_RET(code);
 }
@@ -1401,7 +1412,7 @@ int32_t schedulerInit(SSchedulerCfg *cfg) {
   return TSDB_CODE_SUCCESS;
 }
 
-int32_t scheduleExecJob(void *transport, SArray *nodeList, SQueryDag* pDag, struct SSchJob** pJob, SQueryResult *pRes) {
+int32_t schedulerExecJob(void *transport, SArray *nodeList, SQueryDag* pDag, struct SSchJob** pJob, SQueryResult *pRes) {
   if (NULL == transport || NULL == pDag || NULL == pDag->pSubplans || NULL == pJob || NULL == pRes) {
     SCH_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT);
   }
@@ -1418,7 +1429,7 @@ int32_t scheduleExecJob(void *transport, SArray *nodeList, SQueryDag* pDag, stru
   return TSDB_CODE_SUCCESS;
 }
 
-int32_t scheduleAsyncExecJob(void *transport, SArray *nodeList, SQueryDag* pDag, struct SSchJob** pJob) {
+int32_t schedulerAsyncExecJob(void *transport, SArray *nodeList, SQueryDag* pDag, struct SSchJob** pJob) {
   if (NULL == transport || NULL == pDag || NULL == pDag->pSubplans || NULL == pJob) {
     SCH_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT);
   }
@@ -1551,7 +1562,7 @@ _return:
 }
 
 
-int32_t scheduleFetchRows(SSchJob *pJob, void** pData) {
+int32_t schedulerFetchRows(SSchJob *pJob, void** pData) {
   if (NULL == pJob || NULL == pData) {
     SCH_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT);
   }
@@ -1616,11 +1627,12 @@ _return:
     }
 
     *pData = rsp;
+    SCH_JOB_DLOG("empty res and set query complete, code:%x", code);
   }
 
   atomic_val_compare_exchange_8(&pJob->userFetch, 1, 0);
 
-  SCH_JOB_DLOG("fetch done, code:%s", tstrerror(code));
+  SCH_JOB_DLOG("fetch done, totalRows:%d, code:%s", pJob->resNumOfRows, tstrerror(code));
   atomic_sub_fetch_32(&pJob->ref, 1);
 
   SCH_RET(code);
@@ -1638,7 +1650,7 @@ int32_t scheduleCancelJob(void *job) {
   SCH_RET(code);
 }
 
-void scheduleFreeJob(void *job) {
+void schedulerFreeJob(void *job) {
   if (NULL == job) {
     return;
   }
@@ -1667,7 +1679,8 @@ void scheduleFreeJob(void *job) {
 
         usleep(1);
       } else {
-        assert(0);
+        SCH_JOB_ELOG("invalid job ref number, ref:%d", ref);
+        break;
       }
     }
 

source/libs/scheduler/test/schedulerTests.cpp

@@ -334,7 +334,7 @@ void schtFreeQueryJob(int32_t freeThread) {
   SSchJob *job = atomic_load_ptr(&pQueryJob);
   
   if (job && atomic_val_compare_exchange_ptr(&pQueryJob, job, NULL)) {
-    scheduleFreeJob(job);
+    schedulerFreeJob(job);
     if (freeThread) {
       if (++freeNum % schtTestPrintNum == 0) {
         printf("FreeNum:%d\n", freeNum);
@@ -378,7 +378,7 @@ void* schtRunJobThread(void *aa) {
     qnodeAddr.port = 6031;
     taosArrayPush(qnodeList, &qnodeAddr);
 
-    code = scheduleAsyncExecJob(mockPointer, qnodeList, &dag, &job);
+    code = schedulerAsyncExecJob(mockPointer, qnodeList, &dag, &job);
     assert(code == 0);
 
     execTasks = taosHashInit(5, taosGetDefaultHashFunction(TSDB_DATA_TYPE_UBIGINT), false, HASH_ENTRY_LOCK);
@@ -472,7 +472,7 @@ void* schtRunJobThread(void *aa) {
     atomic_store_32(&schtStartFetch, 1);
 
     void *data = NULL;  
-    code = scheduleFetchRows(pQueryJob, &data);
+    code = schedulerFetchRows(pQueryJob, &data);
     assert(code == 0 || code);
 
     if (0 == code) {
@@ -482,7 +482,7 @@ void* schtRunJobThread(void *aa) {
     }
 
     data = NULL;
-    code = scheduleFetchRows(pQueryJob, &data);
+    code = schedulerFetchRows(pQueryJob, &data);
     assert(code == 0 || code);
     
     schtFreeQueryJob(0);
@@ -539,7 +539,7 @@ TEST(queryTest, normalCase) {
   schtSetExecNode();
   schtSetAsyncSendMsgToServer();
   
-  code = scheduleAsyncExecJob(mockPointer, qnodeList, &dag, &pJob);
+  code = schedulerAsyncExecJob(mockPointer, qnodeList, &dag, &pJob);
   ASSERT_EQ(code, 0);
 
   SSchJob *job = (SSchJob *)pJob;
@@ -594,7 +594,7 @@ TEST(queryTest, normalCase) {
   pthread_create(&(thread1), &thattr, schtCreateFetchRspThread, job);
 
   void *data = NULL;  
-  code = scheduleFetchRows(job, &data);
+  code = schedulerFetchRows(job, &data);
   ASSERT_EQ(code, 0);
 
   SRetrieveTableRsp *pRsp = (SRetrieveTableRsp *)data;
@@ -603,11 +603,11 @@ TEST(queryTest, normalCase) {
   tfree(data);
 
   data = NULL;
-  code = scheduleFetchRows(job, &data);
+  code = schedulerFetchRows(job, &data);
   ASSERT_EQ(code, 0);
   ASSERT_TRUE(data);
 
-  scheduleFreeJob(pJob);
+  schedulerFreeJob(pJob);
 
   schtFreeQueryDag(&dag);
 
@@ -649,11 +649,11 @@ TEST(insertTest, normalCase) {
   pthread_create(&(thread1), &thattr, schtSendRsp, &pInsertJob);
 
   SQueryResult res = {0};
-  code = scheduleExecJob(mockPointer, qnodeList, &dag, &pInsertJob, &res);
+  code = schedulerExecJob(mockPointer, qnodeList, &dag, &pInsertJob, &res);
   ASSERT_EQ(code, 0);
   ASSERT_EQ(res.numOfRows, 20);
 
-  scheduleFreeJob(pInsertJob);
+  schedulerFreeJob(pInsertJob);
 
   schedulerDestroy();  
 }

source/util/src/terror.c

@@ -353,7 +353,7 @@ TAOS_DEFINE_ERROR(TSDB_CODE_QRY_INVALID_INPUT,            "invalid input")
 TAOS_DEFINE_ERROR(TSDB_CODE_QRY_SCH_NOT_EXIST,            "Scheduler not exist")
 TAOS_DEFINE_ERROR(TSDB_CODE_QRY_TASK_NOT_EXIST,           "Task not exist")
 TAOS_DEFINE_ERROR(TSDB_CODE_QRY_TASK_ALREADY_EXIST,       "Task already exist")
-TAOS_DEFINE_ERROR(TSDB_CODE_QRY_RES_CACHE_NOT_EXIST,      "Task result cache not exist")
+TAOS_DEFINE_ERROR(TSDB_CODE_QRY_TASK_CTX_NOT_EXIST,       "Task context not exist")
 TAOS_DEFINE_ERROR(TSDB_CODE_QRY_TASK_CANCELLED,           "Task cancelled")
 TAOS_DEFINE_ERROR(TSDB_CODE_QRY_TASK_DROPPED,             "Task dropped")
 TAOS_DEFINE_ERROR(TSDB_CODE_QRY_TASK_CANCELLING,          "Task cancelling")