qworker.c 39.5 KB
Newer Older
1 2
#include "qworker.h"

dengyihao's avatar
dengyihao 已提交
3
#include "dataSinkMgt.h"
4
#include "executor.h"
D
dapan1121 已提交
5
#include "planner.h"
H
Haojun Liao 已提交
6
#include "query.h"
D
dapan1121 已提交
7 8
#include "qwInt.h"
#include "qwMsg.h"
dengyihao's avatar
dengyihao 已提交
9
#include "tcommon.h"
H
Hongze Cheng 已提交
10
#include "tdatablock.h"
dengyihao's avatar
dengyihao 已提交
11
#include "tglobal.h"
H
Haojun Liao 已提交
12
#include "tmsg.h"
13
#include "tname.h"
D
dapan1121 已提交
14

D
dapan1121 已提交
15
SQWorkerMgmt gQwMgmt = {
16 17 18
    .lock = 0,
    .qwRef = -1,
    .qwNum = 0,
D
dapan1121 已提交
19
};
20

D
dapan1121 已提交
21 22 23 24
int32_t qwProcessHbLinkBroken(SQWorker *mgmt, SQWMsg *qwMsg, SSchedulerHbReq *req) {
  int32_t         code = 0;
  SSchedulerHbRsp rsp = {0};
  SQWSchStatus   *sch = NULL;
D
dapan1121 已提交
25

D
dapan1121 已提交
26
  QW_ERR_RET(qwAcquireScheduler(mgmt, req->sId, QW_READ, &sch));
D
dapan1121 已提交
27

D
dapan1121 已提交
28
  QW_LOCK(QW_WRITE, &sch->hbConnLock);
D
dapan1121 已提交
29

D
dapan1121 已提交
30
  sch->hbBrokenTs = taosGetTimestampMs();
31

D
dapan1121 已提交
32 33 34 35
  if (qwMsg->connInfo.handle == sch->hbConnInfo.handle) {
    tmsgReleaseHandle(&sch->hbConnInfo, TAOS_CONN_SERVER);
    sch->hbConnInfo.handle = NULL;
    sch->hbConnInfo.ahandle = NULL;
D
dapan1121 已提交
36

D
dapan1121 已提交
37 38 39
    QW_DLOG("release hb handle due to connection broken, handle:%p", qwMsg->connInfo.handle);
  } else {
    QW_DLOG("ignore hb connection broken, handle:%p, currentHandle:%p", qwMsg->connInfo.handle, sch->hbConnInfo.handle);
D
dapan1121 已提交
40
  }
D
dapan1121 已提交
41

D
dapan1121 已提交
42
  QW_UNLOCK(QW_WRITE, &sch->hbConnLock);
D
dapan1121 已提交
43

D
dapan1121 已提交
44
  qwReleaseScheduler(QW_READ, mgmt);
D
dapan1121 已提交
45

D
dapan1121 已提交
46
  QW_RET(TSDB_CODE_SUCCESS);
D
dapan1121 已提交
47 48
}

49 50
static void freeItem(void *param) {
  SExplainExecInfo *pInfo = param;
H
Haojun Liao 已提交
51 52 53
  taosMemoryFree(pInfo->verboseInfo);
}

D
dapan1121 已提交
54
int32_t qwHandleTaskComplete(QW_FPARAMS_DEF, SQWTaskCtx *ctx) {
D
dapan1121 已提交
55
  qTaskInfo_t taskHandle = ctx->taskHandle;
D
dapan1121 已提交
56

D
dapan1121 已提交
57 58
  ctx->queryExecDone = true;

D
dapan1121 已提交
59
  if (TASK_TYPE_TEMP == ctx->taskType && taskHandle) {
D
dapan1121 已提交
60
    if (ctx->explain) {
61
      SArray *execInfoList = taosArrayInit(4, sizeof(SExplainExecInfo));
H
Haojun Liao 已提交
62
      QW_ERR_RET(qGetExplainExecInfo(taskHandle, execInfoList));
63

D
dapan1121 已提交
64
      if (ctx->localExec) {
65 66 67 68 69 70 71 72 73 74 75
        SExplainLocalRsp localRsp = {0};
        localRsp.rsp.numOfPlans = taosArrayGetSize(execInfoList);
        SExplainExecInfo *pExec = taosMemoryCalloc(localRsp.rsp.numOfPlans, sizeof(SExplainExecInfo));
        memcpy(pExec, taosArrayGet(execInfoList, 0), localRsp.rsp.numOfPlans * sizeof(SExplainExecInfo));
        localRsp.rsp.subplanInfo = pExec;
        localRsp.qId = qId;
        localRsp.tId = tId;
        localRsp.rId = rId;
        localRsp.eId = eId;
        taosArrayPush(ctx->explainRes, &localRsp);
        taosArrayDestroy(execInfoList);
D
dapan1121 已提交
76 77 78 79 80 81 82
      } else {
        SRpcHandleInfo connInfo = ctx->ctrlConnInfo;
        connInfo.ahandle = NULL;
        int32_t code = qwBuildAndSendExplainRsp(&connInfo, execInfoList);
        taosArrayDestroyEx(execInfoList, freeItem);
        QW_ERR_RET(code);
      }
D
dapan1121 已提交
83
    }
D
dapan1121 已提交
84 85 86 87

    if (!ctx->needFetch) {
      dsGetDataLength(ctx->sinkHandle, &ctx->affectedRows, NULL);
    }
D
dapan1121 已提交
88 89 90 91 92
  }

  return TSDB_CODE_SUCCESS;
}

D
dapan1121 已提交
93 94
int32_t qwSendQueryRsp(QW_FPARAMS_DEF, int32_t msgType, SQWTaskCtx *ctx, int32_t rspCode, bool quickRsp) {
  if ((!quickRsp) || QUERY_RSP_POLICY_QUICK == tsQueryRspPolicy) {
95 96 97 98 99
    if (!ctx->localExec) {
      qwBuildAndSendQueryRsp(msgType, &ctx->ctrlConnInfo, rspCode, ctx);
      QW_TASK_DLOG("query msg rsped, handle:%p, code:%x - %s", ctx->ctrlConnInfo.handle, rspCode, tstrerror(rspCode));
    }

D
dapan1121 已提交
100 101 102 103 104 105
    ctx->queryRsped = true;
  }

  return TSDB_CODE_SUCCESS;
}

106
int32_t qwExecTask(QW_FPARAMS_DEF, SQWTaskCtx *ctx, bool *queryStop) {
dengyihao's avatar
dengyihao 已提交
107 108 109 110 111
  int32_t        code = 0;
  bool           qcontinue = true;
  uint64_t       useconds = 0;
  int32_t        i = 0;
  int32_t        execNum = 0;
D
dapan1121 已提交
112
  qTaskInfo_t    taskHandle = ctx->taskHandle;
D
dapan1121 已提交
113
  DataSinkHandle sinkHandle = ctx->sinkHandle;
H
Hongze Cheng 已提交
114
  SLocalFetch    localFetch = {(void *)mgmt, ctx->localExec, qWorkerProcessLocalFetch, ctx->explainRes};
dengyihao's avatar
dengyihao 已提交
115

D
dapan1121 已提交
116 117 118 119
  if (ctx->queryExecDone) {
    if (queryStop) {
      *queryStop = true;
    }
dengyihao's avatar
dengyihao 已提交
120

D
dapan1121 已提交
121 122 123
    return TSDB_CODE_SUCCESS;
  }

124
  SArray *pResList = taosArrayInit(4, POINTER_BYTES);
D
dapan1121 已提交
125
  while (true) {
H
Haojun Liao 已提交
126
    QW_TASK_DLOG("start to execTask, loopIdx:%d", i++);
D
dapan1121 已提交
127

D
dapan1121 已提交
128
    // if *taskHandle is NULL, it's killed right now
129 130
    bool hasMore = false;

D
dapan1121 已提交
131
    if (taskHandle) {
D
dapan1121 已提交
132
      qwDbgSimulateSleep();
133 134

      code = qExecTaskOpt(taskHandle, pResList, &useconds, &hasMore, &localFetch);
D
dapan1121 已提交
135
      if (code) {
136 137 138 139 140
        if (code != TSDB_CODE_OPS_NOT_SUPPORT) {
          QW_TASK_ELOG("qExecTask failed, code:%x - %s", code, tstrerror(code));
        } else {
          QW_TASK_DLOG("qExecTask failed, code:%x - %s", code, tstrerror(code));
        }
141
        QW_ERR_JRET(code);
D
dapan1121 已提交
142
      }
D
dapan1121 已提交
143 144
    }

D
dapan1121 已提交
145 146
    ++execNum;

147 148
    size_t numOfResBlock = taosArrayGetSize(pResList);
    for (int32_t j = 0; j < numOfResBlock; ++j) {
H
Haojun Liao 已提交
149
      SSDataBlock *pRes = taosArrayGetP(pResList, j);
H
Haojun Liao 已提交
150

H
Haojun Liao 已提交
151 152 153 154
      SInputData inputData = {.pData = pRes};
      code = dsPutDataBlock(sinkHandle, &inputData, &qcontinue);
      if (code) {
        QW_TASK_ELOG("dsPutDataBlock failed, code:%x - %s", code, tstrerror(code));
155
        QW_ERR_JRET(code);
H
Haojun Liao 已提交
156
      }
157

H
Haojun Liao 已提交
158
      QW_TASK_DLOG("data put into sink, rows:%d, continueExecTask:%d", pRes->info.rows, qcontinue);
D
dapan1121 已提交
159
    }
D
dapan1121 已提交
160

161 162 163
    if (numOfResBlock == 0 || (hasMore == false)) {
      if (numOfResBlock == 0) {
        QW_TASK_DLOG("qExecTask end with empty res, useconds:%" PRIu64, useconds);
H
Haojun Liao 已提交
164 165
      } else {
        QW_TASK_DLOG("qExecTask done, useconds:%" PRIu64, useconds);
166 167 168 169 170 171 172 173 174 175 176 177
      }

      dsEndPut(sinkHandle, useconds);
      QW_ERR_JRET(qwHandleTaskComplete(QW_FPARAMS(), ctx));

      if (queryStop) {
        *queryStop = true;
      }

      break;
    }

D
dapan1121 已提交
178
    if (!qcontinue) {
179 180 181
      if (queryStop) {
        *queryStop = true;
      }
182

D
dapan1121 已提交
183 184 185
      break;
    }

D
dapan1121 已提交
186
    if (ctx->needFetch && (!ctx->queryRsped) && execNum >= QW_DEFAULT_SHORT_RUN_TIMES) {
D
dapan1121 已提交
187 188 189
      break;
    }

D
dapan1121 已提交
190
    if (QW_EVENT_RECEIVED(ctx, QW_EVENT_FETCH)) {
D
dapan1121 已提交
191 192
      break;
    }
D
dapan1121 已提交
193

D
dapan1121 已提交
194 195 196
    if (atomic_load_32(&ctx->rspCode)) {
      break;
    }
D
dapan1121 已提交
197 198
  }

199
_return:
H
Haojun Liao 已提交
200
  taosArrayDestroy(pResList);
D
dapan1121 已提交
201
  QW_RET(code);
D
dapan1121 已提交
202
}
D
dapan1121 已提交
203

204 205 206 207 208 209 210 211 212
bool qwTaskNotInExec(SQWTaskCtx *ctx) {
  qTaskInfo_t    taskHandle = ctx->taskHandle;
  if (NULL == taskHandle || !qTaskIsExecuting(taskHandle)) {
    return true;
  }

  return false;
}

D
dapan1121 已提交
213
int32_t qwGenerateSchHbRsp(SQWorker *mgmt, SQWSchStatus *sch, SQWHbInfo *hbInfo) {
D
dapan1121 已提交
214 215
  int32_t taskNum = 0;

D
dapan1121 已提交
216
  hbInfo->connInfo = sch->hbConnInfo;
D
dapan1121 已提交
217
  hbInfo->rsp.epId = sch->hbEpId;
D
dapan1121 已提交
218

D
dapan1121 已提交
219
  QW_LOCK(QW_READ, &sch->tasksLock);
dengyihao's avatar
dengyihao 已提交
220

D
dapan1121 已提交
221
  taskNum = taosHashGetSize(sch->tasksHash);
D
dapan1121 已提交
222 223 224

  hbInfo->rsp.taskStatus = taosArrayInit(taskNum, sizeof(STaskStatus));
  if (NULL == hbInfo->rsp.taskStatus) {
D
dapan1121 已提交
225
    QW_UNLOCK(QW_READ, &sch->tasksLock);
D
dapan1121 已提交
226
    QW_ELOG("taosArrayInit taskStatus failed, num:%d", taskNum);
S
Shengliang Guan 已提交
227
    return TSDB_CODE_OUT_OF_MEMORY;
D
dapan1121 已提交
228 229
  }

230
  void       *key = NULL;
dengyihao's avatar
dengyihao 已提交
231 232
  size_t      keyLen = 0;
  int32_t     i = 0;
D
dapan1121 已提交
233
  STaskStatus status = {0};
D
dapan1121 已提交
234 235 236 237

  void *pIter = taosHashIterate(sch->tasksHash, NULL);
  while (pIter) {
    SQWTaskStatus *taskStatus = (SQWTaskStatus *)pIter;
D
dapan1121 已提交
238
    key = taosHashGetKey(pIter, &keyLen);
D
dapan1121 已提交
239

dengyihao's avatar
dengyihao 已提交
240
    // TODO GET EXECUTOR API TO GET MORE INFO
D
dapan1121 已提交
241

D
dapan1121 已提交
242
    QW_GET_QTID(key, status.queryId, status.taskId, status.execId);
D
dapan1121 已提交
243 244
    status.status = taskStatus->status;
    status.refId = taskStatus->refId;
dengyihao's avatar
dengyihao 已提交
245

D
dapan1121 已提交
246
    taosArrayPush(hbInfo->rsp.taskStatus, &status);
dengyihao's avatar
dengyihao 已提交
247

D
dapan1121 已提交
248 249
    ++i;
    pIter = taosHashIterate(sch->tasksHash, pIter);
dengyihao's avatar
dengyihao 已提交
250
  }
D
dapan1121 已提交
251 252 253 254 255 256

  QW_UNLOCK(QW_READ, &sch->tasksLock);

  return TSDB_CODE_SUCCESS;
}

D
dapan1121 已提交
257
int32_t qwGetQueryResFromSink(QW_FPARAMS_DEF, SQWTaskCtx *ctx, int32_t *dataLen, void **rspMsg, SOutputData *pOutput) {
D
dapan1121 已提交
258
  int64_t            len = 0;
D
dapan1121 已提交
259
  SRetrieveTableRsp *rsp = NULL;
dengyihao's avatar
dengyihao 已提交
260 261
  bool               queryEnd = false;
  int32_t            code = 0;
262
  SOutputData        output = {0};
D
dapan1121 已提交
263

264
  if (NULL == ctx->sinkHandle) {
D
dapan1121 已提交
265
    pOutput->queryEnd = true;
266 267 268
    return TSDB_CODE_SUCCESS;
  }

269
  *dataLen = 0;
D
dapan1121 已提交
270

271 272
  while (true) {
    dsGetDataLength(ctx->sinkHandle, &len, &queryEnd);
D
dapan1121 已提交
273

274
    if (len < 0) {
275
      QW_TASK_ELOG("invalid length from dsGetDataLength, length:%" PRId64 "", len);
276 277
      QW_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT);
    }
dengyihao's avatar
dengyihao 已提交
278

279 280 281 282 283 284 285
    if (len == 0) {
      if (queryEnd) {
        code = dsGetDataBlock(ctx->sinkHandle, &output);
        if (code) {
          QW_TASK_ELOG("dsGetDataBlock failed, code:%x - %s", code, tstrerror(code));
          QW_ERR_RET(code);
        }
H
Haojun Liao 已提交
286

dengyihao's avatar
dengyihao 已提交
287
        QW_TASK_DLOG("no more data in sink and query end, fetched blocks %d rows %" PRId64, pOutput->numOfBlocks,
288
                     pOutput->numOfRows);
289

290 291
        qwUpdateTaskStatus(QW_FPARAMS(), JOB_TASK_STATUS_SUCC);
        if (NULL == rsp) {
D
dapan1121 已提交
292
          QW_ERR_RET(qwMallocFetchRsp(!ctx->localExec, len, &rsp));
293 294 295 296 297 298 299 300 301 302 303 304 305
          *pOutput = output;
        } else {
          pOutput->queryEnd = output.queryEnd;
          pOutput->bufStatus = output.bufStatus;
          pOutput->useconds = output.useconds;
        }

        break;
      }

      pOutput->bufStatus = DS_BUF_EMPTY;

      break;
D
dapan1121 已提交
306
    }
D
dapan1121 已提交
307

308
    // Got data from sink
309
    QW_TASK_DLOG("there are data in sink, dataLength:%" PRId64 "", len);
D
dapan1121 已提交
310

311
    *dataLen += len;
D
dapan1121 已提交
312

D
dapan1121 已提交
313
    QW_ERR_RET(qwMallocFetchRsp(!ctx->localExec, *dataLen, &rsp));
D
dapan1121 已提交
314

315 316 317 318 319 320
    output.pData = rsp->data + *dataLen - len;
    code = dsGetDataBlock(ctx->sinkHandle, &output);
    if (code) {
      QW_TASK_ELOG("dsGetDataBlock failed, code:%x - %s", code, tstrerror(code));
      QW_ERR_RET(code);
    }
dengyihao's avatar
dengyihao 已提交
321

322 323 324 325 326 327 328 329 330 331
    pOutput->queryEnd = output.queryEnd;
    pOutput->precision = output.precision;
    pOutput->bufStatus = output.bufStatus;
    pOutput->useconds = output.useconds;
    pOutput->compressed = output.compressed;
    pOutput->numOfCols = output.numOfCols;
    pOutput->numOfRows += output.numOfRows;
    pOutput->numOfBlocks++;

    if (DS_BUF_EMPTY == pOutput->bufStatus && pOutput->queryEnd) {
332
      QW_TASK_DLOG("task all data fetched and done, fetched blocks %d rows %" PRId64, pOutput->numOfBlocks,
333
                   pOutput->numOfRows);
334 335 336
      qwUpdateTaskStatus(QW_FPARAMS(), JOB_TASK_STATUS_SUCC);
      break;
    }
dengyihao's avatar
dengyihao 已提交
337

338
    if (0 == ctx->level) {
dengyihao's avatar
dengyihao 已提交
339 340
      QW_TASK_DLOG("task fetched blocks %d rows %" PRId64 ", level %d", pOutput->numOfBlocks, pOutput->numOfRows,
                   ctx->level);
341 342
      break;
    }
D
dapan1121 已提交
343

344
    if (pOutput->numOfRows >= QW_MIN_RES_ROWS) {
dengyihao's avatar
dengyihao 已提交
345 346
      QW_TASK_DLOG("task fetched blocks %d rows %" PRId64 " reaches the min rows", pOutput->numOfBlocks,
                   pOutput->numOfRows);
347 348
      break;
    }
D
dapan1121 已提交
349 350
  }

351 352
  *rspMsg = rsp;

D
dapan1121 已提交
353
  return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
354 355
}

356
int32_t qwGetDeleteResFromSink(QW_FPARAMS_DEF, SQWTaskCtx *ctx, SDeleteRes *pRes) {
357 358 359 360
  int64_t     len = 0;
  bool        queryEnd = false;
  int32_t     code = 0;
  SOutputData output = {0};
D
dapan1121 已提交
361 362 363

  dsGetDataLength(ctx->sinkHandle, &len, &queryEnd);

D
dapan1121 已提交
364
  if (len <= 0 || len != sizeof(SDeleterRes)) {
D
dapan1121 已提交
365
    QW_TASK_ELOG("invalid length from dsGetDataLength, length:%" PRId64, len);
D
dapan1121 已提交
366 367 368 369 370 371 372
    QW_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT);
  }

  output.pData = taosMemoryCalloc(1, len);
  if (NULL == output.pData) {
    QW_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
  }
373

D
dapan1121 已提交
374 375 376 377 378 379 380
  code = dsGetDataBlock(ctx->sinkHandle, &output);
  if (code) {
    QW_TASK_ELOG("dsGetDataBlock failed, code:%x - %s", code, tstrerror(code));
    taosMemoryFree(output.pData);
    QW_ERR_RET(code);
  }

381 382
  SDeleterRes *pDelRes = (SDeleterRes *)output.pData;

D
dapan1121 已提交
383
  pRes->suid = pDelRes->suid;
D
dapan1121 已提交
384 385 386
  pRes->uidList = pDelRes->uidList;
  pRes->skey = pDelRes->skey;
  pRes->ekey = pDelRes->ekey;
387
  pRes->affectedRows = pDelRes->affectedRows;
388 389
  strcpy(pRes->tableFName, pDelRes->tableName);
  strcpy(pRes->tsColName, pDelRes->tsColName);
D
dapan1121 已提交
390
  taosMemoryFree(output.pData);
391

D
dapan1121 已提交
392 393 394
  return TSDB_CODE_SUCCESS;
}

D
dapan1121 已提交
395
int32_t qwHandlePrePhaseEvents(QW_FPARAMS_DEF, int8_t phase, SQWPhaseInput *input, SQWPhaseOutput *output) {
396 397
  int32_t     code = 0;
  SQWTaskCtx *ctx = NULL;
D
dapan1121 已提交
398

D
dapan1121 已提交
399
  QW_TASK_DLOG("start to handle event at phase %s", qwPhaseStr(phase));
D
dapan1121 已提交
400

D
dapan1121 已提交
401
  QW_ERR_JRET(qwAcquireTaskCtx(QW_FPARAMS(), &ctx));
dengyihao's avatar
dengyihao 已提交
402

D
dapan1121 已提交
403
  QW_LOCK(QW_WRITE, &ctx->lock);
D
dapan1121 已提交
404

D
dapan1121 已提交
405
  QW_SET_PHASE(ctx, phase);
D
dapan1121 已提交
406

dengyihao's avatar
dengyihao 已提交
407
  if (atomic_load_8((int8_t *)&ctx->queryEnd)) {
D
dapan1121 已提交
408 409 410
    QW_TASK_ELOG_E("query already end");
    QW_ERR_JRET(TSDB_CODE_QW_MSG_ERROR);
  }
D
dapan1121 已提交
411

D
dapan1121 已提交
412 413
  switch (phase) {
    case QW_PHASE_PRE_QUERY: {
D
dapan1121 已提交
414
      if (QW_EVENT_PROCESSED(ctx, QW_EVENT_DROP)) {
H
Haojun Liao 已提交
415
        QW_TASK_ELOG("task already dropped at wrong phase %s", qwPhaseStr(phase));
D
dapan1121 已提交
416
        QW_ERR_JRET(TSDB_CODE_QRY_TASK_STATUS_ERROR);
D
dapan1121 已提交
417
      }
D
dapan1121 已提交
418

D
dapan1121 已提交
419
      if (QW_EVENT_RECEIVED(ctx, QW_EVENT_DROP)) {
D
dapan1121 已提交
420
        QW_ERR_JRET(qwDropTask(QW_FPARAMS()));
421

422 423
        // qwBuildAndSendDropRsp(&ctx->ctrlConnInfo, code);
        // QW_TASK_DLOG("drop rsp send, handle:%p, code:%x - %s", ctx->ctrlConnInfo.handle, code, tstrerror(code));
D
dapan1121 已提交
424

D
dapan1121 已提交
425
        QW_ERR_JRET(ctx->rspCode);
D
dapan1121 已提交
426
      }
D
dapan1121 已提交
427

D
dapan1121 已提交
428
      QW_ERR_JRET(qwUpdateTaskStatus(QW_FPARAMS(), JOB_TASK_STATUS_EXEC));
D
dapan1121 已提交
429 430
      break;
    }
D
dapan1121 已提交
431
    case QW_PHASE_PRE_FETCH: {
D
dapan1121 已提交
432
      if (QW_EVENT_PROCESSED(ctx, QW_EVENT_DROP) || QW_EVENT_RECEIVED(ctx, QW_EVENT_DROP)) {
D
dapan1121 已提交
433
        QW_TASK_WLOG("task dropping or already dropped, phase:%s", qwPhaseStr(phase));
D
dapan1121 已提交
434
        QW_ERR_JRET(ctx->rspCode);
D
dapan1121 已提交
435
      }
D
dapan1121 已提交
436

D
dapan1121 已提交
437
      if (QW_EVENT_RECEIVED(ctx, QW_EVENT_FETCH)) {
D
dapan1121 已提交
438
        QW_TASK_WLOG("last fetch still not processed, phase:%s", qwPhaseStr(phase));
D
dapan1121 已提交
439 440 441
        QW_ERR_JRET(TSDB_CODE_QRY_DUPLICATTED_OPERATION);
      }

D
dapan1121 已提交
442 443 444 445 446
      if (ctx->rspCode) {
        QW_TASK_ELOG("task already failed cause of %s, phase:%s", tstrerror(ctx->rspCode), qwPhaseStr(phase));
        QW_ERR_JRET(ctx->rspCode);
      }

D
dapan1121 已提交
447
      if (!ctx->queryRsped) {
D
dapan1121 已提交
448
        QW_TASK_ELOG("ready msg has not been processed, phase:%s", qwPhaseStr(phase));
D
dapan1121 已提交
449
        QW_ERR_JRET(TSDB_CODE_QRY_TASK_MSG_ERROR);
D
dapan1121 已提交
450 451
      }
      break;
dengyihao's avatar
dengyihao 已提交
452
    }
D
dapan1121 已提交
453
    case QW_PHASE_PRE_CQUERY: {
D
dapan1121 已提交
454
      if (QW_EVENT_PROCESSED(ctx, QW_EVENT_DROP)) {
D
dapan1121 已提交
455
        QW_TASK_WLOG("task already dropped, phase:%s", qwPhaseStr(phase));
D
dapan1121 已提交
456
        QW_ERR_JRET(ctx->rspCode);
D
dapan1121 已提交
457
      }
D
dapan1121 已提交
458

D
dapan1121 已提交
459 460 461 462 463
      if (ctx->rspCode) {
        QW_TASK_ELOG("task already failed cause of %s, phase:%s", tstrerror(ctx->rspCode), qwPhaseStr(phase));
        QW_ERR_JRET(ctx->rspCode);
      }

D
dapan1121 已提交
464
      if (QW_EVENT_RECEIVED(ctx, QW_EVENT_DROP)) {
D
dapan1121 已提交
465
        QW_ERR_JRET(qwDropTask(QW_FPARAMS()));
H
Haojun Liao 已提交
466

467 468
        // qwBuildAndSendDropRsp(&ctx->ctrlConnInfo, code);
        // QW_TASK_DLOG("drop rsp send, handle:%p, code:%x - %s", ctx->ctrlConnInfo.handle, code, tstrerror(code));
469

D
dapan1121 已提交
470
        QW_ERR_JRET(ctx->rspCode);
D
dapan1121 已提交
471
      }
D
dapan1121 已提交
472

D
dapan1121 已提交
473
      break;
D
dapan1121 已提交
474 475 476
    }
    default:
      QW_TASK_ELOG("invalid phase %s", qwPhaseStr(phase));
S
Shengliang Guan 已提交
477
      QW_ERR_JRET(TSDB_CODE_APP_ERROR);
D
dapan1121 已提交
478 479 480
  }

  if (ctx->rspCode) {
481
    QW_TASK_ELOG("task already failed at phase %s, code:%s", qwPhaseStr(phase), tstrerror(ctx->rspCode));
D
dapan1121 已提交
482
    QW_ERR_JRET(ctx->rspCode);
D
dapan1121 已提交
483
  }
D
dapan1121 已提交
484

D
dapan1121 已提交
485
_return:
D
dapan1121 已提交
486

D
dapan1121 已提交
487
  if (ctx) {
D
dapan1121 已提交
488
    QW_UPDATE_RSP_CODE(ctx, code);
dengyihao's avatar
dengyihao 已提交
489

D
dapan1121 已提交
490
    QW_UNLOCK(QW_WRITE, &ctx->lock);
D
dapan1121 已提交
491 492
    qwReleaseTaskCtx(mgmt, ctx);
  }
D
dapan1121 已提交
493

494 495 496 497 498
  if (code != TSDB_CODE_SUCCESS) {
    QW_TASK_ELOG("end to handle event at phase %s, code:%s", qwPhaseStr(phase), tstrerror(code));
  } else {
    QW_TASK_DLOG("end to handle event at phase %s, code:%s", qwPhaseStr(phase), tstrerror(code));
  }
D
dapan1121 已提交
499 500 501 502 503

  QW_RET(code);
}

int32_t qwHandlePostPhaseEvents(QW_FPARAMS_DEF, int8_t phase, SQWPhaseInput *input, SQWPhaseOutput *output) {
504 505 506
  int32_t        code = 0;
  SQWTaskCtx    *ctx = NULL;
  SRpcHandleInfo connInfo = {0};
D
dapan1121 已提交
507

D
dapan1121 已提交
508
  QW_TASK_DLOG("start to handle event at phase %s", qwPhaseStr(phase));
dengyihao's avatar
dengyihao 已提交
509

D
dapan1121 已提交
510
  QW_ERR_JRET(qwAcquireTaskCtx(QW_FPARAMS(), &ctx));
dengyihao's avatar
dengyihao 已提交
511

D
dapan1121 已提交
512 513
  QW_LOCK(QW_WRITE, &ctx->lock);

D
dapan1121 已提交
514
  if (QW_EVENT_PROCESSED(ctx, QW_EVENT_DROP)) {
D
dapan1121 已提交
515
    QW_TASK_WLOG("task already dropped, phase:%s", qwPhaseStr(phase));
D
dapan1121 已提交
516
    QW_ERR_JRET(ctx->rspCode);
D
dapan1121 已提交
517 518
  }

D
dapan1121 已提交
519
  if (QW_EVENT_RECEIVED(ctx, QW_EVENT_DROP)) {
520
    if (QW_PHASE_POST_FETCH != phase || ((!QW_QUERY_RUNNING(ctx)) && qwTaskNotInExec(ctx))) {
521 522 523
      QW_ERR_JRET(qwDropTask(QW_FPARAMS()));
      QW_ERR_JRET(ctx->rspCode);
    }
D
dapan1121 已提交
524 525 526
  }

  if (ctx->rspCode) {
dengyihao's avatar
dengyihao 已提交
527 528
    QW_TASK_ELOG("task already failed, phase %s, error:%x - %s", qwPhaseStr(phase), ctx->rspCode,
                 tstrerror(ctx->rspCode));
D
dapan1121 已提交
529
    QW_ERR_JRET(ctx->rspCode);
dengyihao's avatar
dengyihao 已提交
530
  }
D
dapan1121 已提交
531

D
dapan1121 已提交
532
  QW_ERR_JRET(input->code);
D
dapan1121 已提交
533 534 535

_return:

D
dapan1121 已提交
536
  if (TSDB_CODE_SUCCESS == code && QW_PHASE_POST_QUERY == phase) {
D
dapan1121 已提交
537
    qwUpdateTaskStatus(QW_FPARAMS(), JOB_TASK_STATUS_PART_SUCC);
538
    ctx->queryGotData = true;
D
dapan1121 已提交
539 540
  }

541
  if (QW_PHASE_POST_QUERY == phase && ctx && !ctx->queryRsped) {
D
dapan1121 已提交
542 543 544 545
    bool   rsped = false;
    SQWMsg qwMsg = {.msgType = ctx->msgType, .connInfo = ctx->ctrlConnInfo};
    qwDbgSimulateRedirect(&qwMsg, ctx, &rsped);
    qwDbgSimulateDead(QW_FPARAMS(), ctx, &rsped);
dengyihao's avatar
dengyihao 已提交
546
    if (!rsped) {
D
dapan1121 已提交
547
      qwSendQueryRsp(QW_FPARAMS(), input->msgType + 1, ctx, code, false);
548
    }
549 550
  }

D
dapan1121 已提交
551
  if (ctx) {
D
dapan1121 已提交
552
    QW_UPDATE_RSP_CODE(ctx, code);
D
dapan1121 已提交
553

554 555 556
    if (QW_PHASE_POST_CQUERY != phase) {
      QW_SET_PHASE(ctx, phase);
    }
dengyihao's avatar
dengyihao 已提交
557

D
dapan1121 已提交
558
    QW_UNLOCK(QW_WRITE, &ctx->lock);
D
dapan1121 已提交
559
    qwReleaseTaskCtx(mgmt, ctx);
D
dapan1121 已提交
560 561
  }

D
dapan1121 已提交
562
  if (code) {
D
dapan1121 已提交
563
    qwUpdateTaskStatus(QW_FPARAMS(), JOB_TASK_STATUS_FAIL);
D
dapan1121 已提交
564 565
  }

D
dapan1121 已提交
566
  QW_TASK_DLOG("end to handle event at phase %s, code:%x - %s", qwPhaseStr(phase), code, tstrerror(code));
D
dapan1121 已提交
567

D
dapan1121 已提交
568 569 570
  QW_RET(code);
}

D
dapan1121 已提交
571 572 573 574 575 576
int32_t qwAbortPrerocessQuery(QW_FPARAMS_DEF) {
  QW_ERR_RET(qwDropTask(QW_FPARAMS()));

  QW_RET(TSDB_CODE_SUCCESS);
}

D
dapan1121 已提交
577
int32_t qwPreprocessQuery(QW_FPARAMS_DEF, SQWMsg *qwMsg) {
H
Hongze Cheng 已提交
578 579
  int32_t     code = 0;
  SQWTaskCtx *ctx = NULL;
D
dapan1121 已提交
580

D
dapan1121 已提交
581
  QW_ERR_JRET(qwRegisterQueryBrokenLinkArg(QW_FPARAMS(), &qwMsg->connInfo));
D
dapan1121 已提交
582

D
dapan1121 已提交
583 584 585
  QW_ERR_JRET(qwAddTaskCtx(QW_FPARAMS()));

  QW_ERR_JRET(qwAcquireTaskCtx(QW_FPARAMS(), &ctx));
D
dapan1121 已提交
586

D
dapan1121 已提交
587
  ctx->ctrlConnInfo = qwMsg->connInfo;
588
  ctx->sId = sId;
D
dapan1121 已提交
589
  ctx->phase = -1;
D
dapan1121 已提交
590

D
dapan1121 已提交
591
  QW_ERR_JRET(qwAddTaskStatus(QW_FPARAMS(), JOB_TASK_STATUS_INIT));
D
dapan1121 已提交
592 593 594 595 596 597 598 599 600 601 602

_return:

  if (ctx) {
    QW_UPDATE_RSP_CODE(ctx, code);
    qwReleaseTaskCtx(mgmt, ctx);
  }

  QW_RET(TSDB_CODE_SUCCESS);
}

603
int32_t qwProcessQuery(QW_FPARAMS_DEF, SQWMsg *qwMsg, char *sql) {
D
dapan1121 已提交
604 605 606 607 608 609 610 611
  int32_t        code = 0;
  bool           queryRsped = false;
  SSubplan      *plan = NULL;
  SQWPhaseInput  input = {0};
  qTaskInfo_t    pTaskInfo = NULL;
  DataSinkHandle sinkHandle = NULL;
  SQWTaskCtx    *ctx = NULL;

D
dapan1121 已提交
612
  QW_ERR_JRET(qwHandlePrePhaseEvents(QW_FPARAMS(), QW_PHASE_PRE_QUERY, &input, NULL));
D
dapan1121 已提交
613 614

  QW_ERR_JRET(qwGetTaskCtx(QW_FPARAMS(), &ctx));
dengyihao's avatar
dengyihao 已提交
615

D
dapan1121 已提交
616 617 618
  ctx->taskType = qwMsg->msgInfo.taskType;
  ctx->explain = qwMsg->msgInfo.explain;
  ctx->needFetch = qwMsg->msgInfo.needFetch;
D
dapan1121 已提交
619
  ctx->msgType = qwMsg->msgType;
D
dapan1121 已提交
620
  ctx->localExec = false;
X
Xiaoyu Wang 已提交
621

622
  // QW_TASK_DLOGL("subplan json string, len:%d, %s", qwMsg->msgLen, qwMsg->msg);
X
Xiaoyu Wang 已提交
623

624
  code = qMsgToSubplan(qwMsg->msg, qwMsg->msgLen, &plan);
D
dapan1121 已提交
625
  if (TSDB_CODE_SUCCESS != code) {
626 627
    code = TSDB_CODE_INVALID_MSG;
    QW_TASK_ELOG("task physical plan to subplan failed, code:%x - %s", code, tstrerror(code));
D
dapan1121 已提交
628
    QW_ERR_JRET(code);
D
dapan1121 已提交
629
  }
dengyihao's avatar
dengyihao 已提交
630

631
  code = qCreateExecTask(qwMsg->node, mgmt->nodeId, tId, plan, &pTaskInfo, &sinkHandle, sql, OPTR_EXEC_MODEL_BATCH);
D
dapan1121 已提交
632
  sql = NULL;
D
dapan1121 已提交
633
  if (code) {
D
dapan1121 已提交
634
    QW_TASK_ELOG("qCreateExecTask failed, code:%x - %s", code, tstrerror(code));
D
dapan1121 已提交
635
    QW_ERR_JRET(code);
D
dapan1121 已提交
636
  }
D
dapan1121 已提交
637

H
Haojun Liao 已提交
638
  if (NULL == sinkHandle || NULL == pTaskInfo) {
D
dapan1121 已提交
639
    QW_TASK_ELOG("create task result error, taskHandle:%p, sinkHandle:%p", pTaskInfo, sinkHandle);
S
Shengliang Guan 已提交
640
    QW_ERR_JRET(TSDB_CODE_APP_ERROR);
D
dapan1121 已提交
641 642
  }

D
dapan1121 已提交
643 644
  qwSendQueryRsp(QW_FPARAMS(), qwMsg->msgType + 1, ctx, code, true);

645
  ctx->level = plan->level;
D
dapan1121 已提交
646 647 648
  atomic_store_ptr(&ctx->taskHandle, pTaskInfo);
  atomic_store_ptr(&ctx->sinkHandle, sinkHandle);

D
dapan1121 已提交
649 650
  qwSaveTbVersionInfo(pTaskInfo, ctx);
  QW_ERR_JRET(qwExecTask(QW_FPARAMS(), ctx, NULL));
dengyihao's avatar
dengyihao 已提交
651

D
dapan1121 已提交
652 653
_return:

D
dapan1121 已提交
654
  taosMemoryFree(sql);
655

D
dapan1121 已提交
656
  input.code = code;
D
dapan1121 已提交
657
  input.msgType = qwMsg->msgType;
D
dapan1121 已提交
658
  code = qwHandlePostPhaseEvents(QW_FPARAMS(), QW_PHASE_POST_QUERY, &input, NULL);
659

D
dapan1121 已提交
660
  if (QUERY_RSP_POLICY_QUICK == tsQueryRspPolicy && ctx != NULL && QW_EVENT_RECEIVED(ctx, QW_EVENT_FETCH)) {
dengyihao's avatar
dengyihao 已提交
661 662
    void       *rsp = NULL;
    int32_t     dataLen = 0;
663
    SOutputData sOutput = {0};
D
dapan1121 已提交
664 665 666
    if (qwGetQueryResFromSink(QW_FPARAMS(), ctx, &dataLen, &rsp, &sOutput)) {
      return TSDB_CODE_SUCCESS;
    }
H
Haojun Liao 已提交
667

668 669
    if (rsp) {
      bool qComplete = (DS_BUF_EMPTY == sOutput.bufStatus && sOutput.queryEnd);
H
Haojun Liao 已提交
670

671 672 673 674
      qwBuildFetchRsp(rsp, &sOutput, dataLen, qComplete);
      if (qComplete) {
        atomic_store_8((int8_t *)&ctx->queryEnd, true);
      }
H
Haojun Liao 已提交
675

676 677
      qwMsg->connInfo = ctx->dataConnInfo;
      QW_SET_EVENT_PROCESSED(ctx, QW_EVENT_FETCH);
H
Haojun Liao 已提交
678

679
      qwBuildAndSendFetchRsp(ctx->msgType + 1, &qwMsg->connInfo, rsp, dataLen, code);
680
      rsp = NULL;
H
Haojun Liao 已提交
681

dengyihao's avatar
dengyihao 已提交
682 683
      QW_TASK_DLOG("fetch rsp send, handle:%p, code:%x - %s, dataLen:%d", qwMsg->connInfo.handle, code, tstrerror(code),
                   dataLen);
684 685 686
    }
  }

D
dapan1121 已提交
687
  QW_RET(TSDB_CODE_SUCCESS);
D
dapan1121 已提交
688 689
}

D
dapan1121 已提交
690
int32_t qwProcessCQuery(QW_FPARAMS_DEF, SQWMsg *qwMsg) {
691
  SQWTaskCtx   *ctx = NULL;
dengyihao's avatar
dengyihao 已提交
692
  int32_t       code = 0;
693
  SQWPhaseInput input = {0};
694
  void         *rsp = NULL;
dengyihao's avatar
dengyihao 已提交
695
  int32_t       dataLen = 0;
696
  bool          queryStop = false;
D
dapan1121 已提交
697
  bool          qComplete = false;
dengyihao's avatar
dengyihao 已提交
698

D
dapan1121 已提交
699
  do {
D
dapan1121 已提交
700
    ctx = NULL;
dengyihao's avatar
dengyihao 已提交
701

D
dapan1121 已提交
702
    QW_ERR_JRET(qwHandlePrePhaseEvents(QW_FPARAMS(), QW_PHASE_PRE_CQUERY, &input, NULL));
D
dapan1121 已提交
703

D
dapan1121 已提交
704
    QW_ERR_JRET(qwGetTaskCtx(QW_FPARAMS(), &ctx));
D
dapan1121 已提交
705

dengyihao's avatar
dengyihao 已提交
706 707
    atomic_store_8((int8_t *)&ctx->queryInQueue, 0);
    atomic_store_8((int8_t *)&ctx->queryContinue, 0);
D
dapan1121 已提交
708

709
    QW_ERR_JRET(qwExecTask(QW_FPARAMS(), ctx, &queryStop));
D
dapan1121 已提交
710

D
dapan1121 已提交
711
    if (QW_EVENT_RECEIVED(ctx, QW_EVENT_FETCH)) {
D
dapan1121 已提交
712
      SOutputData sOutput = {0};
D
dapan1121 已提交
713
      QW_ERR_JRET(qwGetQueryResFromSink(QW_FPARAMS(), ctx, &dataLen, &rsp, &sOutput));
dengyihao's avatar
dengyihao 已提交
714 715

      if ((!sOutput.queryEnd) && (DS_BUF_LOW == sOutput.bufStatus || DS_BUF_EMPTY == sOutput.bufStatus)) {
D
dapan1121 已提交
716
        QW_TASK_DLOG("task not end and buf is %s, need to continue query", qwBufStatusStr(sOutput.bufStatus));
dengyihao's avatar
dengyihao 已提交
717 718

        atomic_store_8((int8_t *)&ctx->queryContinue, 1);
719
      }
dengyihao's avatar
dengyihao 已提交
720

D
dapan1121 已提交
721
      if (rsp) {
D
dapan1121 已提交
722
        qComplete = (DS_BUF_EMPTY == sOutput.bufStatus && sOutput.queryEnd);
dengyihao's avatar
dengyihao 已提交
723

D
dapan1121 已提交
724
        qwBuildFetchRsp(rsp, &sOutput, dataLen, qComplete);
D
dapan1121 已提交
725
        if (qComplete) {
dengyihao's avatar
dengyihao 已提交
726
          atomic_store_8((int8_t *)&ctx->queryEnd, true);
D
dapan1121 已提交
727
          atomic_store_8((int8_t *)&ctx->queryContinue, 0);
D
dapan1121 已提交
728
        }
H
Haojun Liao 已提交
729

D
dapan1121 已提交
730
        qwMsg->connInfo = ctx->dataConnInfo;
dengyihao's avatar
dengyihao 已提交
731 732
        QW_SET_EVENT_PROCESSED(ctx, QW_EVENT_FETCH);

733
        qwBuildAndSendFetchRsp(ctx->msgType + 1, &qwMsg->connInfo, rsp, dataLen, code);
D
dapan1121 已提交
734
        rsp = NULL;
735

dengyihao's avatar
dengyihao 已提交
736 737
        QW_TASK_DLOG("fetch rsp send, handle:%p, code:%x - %s, dataLen:%d", qwMsg->connInfo.handle, code,
                     tstrerror(code), dataLen);
D
dapan1121 已提交
738
      } else {
dengyihao's avatar
dengyihao 已提交
739
        atomic_store_8((int8_t *)&ctx->queryContinue, 1);
740 741 742
      }
    }

dengyihao's avatar
dengyihao 已提交
743
  _return:
744

D
dapan1121 已提交
745 746 747 748
    if (NULL == ctx) {
      break;
    }

D
dapan1121 已提交
749
    if (code && QW_EVENT_RECEIVED(ctx, QW_EVENT_FETCH)) {
dengyihao's avatar
dengyihao 已提交
750
      QW_SET_EVENT_PROCESSED(ctx, QW_EVENT_FETCH);
751 752
      qwFreeFetchRsp(rsp);
      rsp = NULL;
dengyihao's avatar
dengyihao 已提交
753

D
dapan1121 已提交
754
      qwMsg->connInfo = ctx->dataConnInfo;
755
      qwBuildAndSendFetchRsp(ctx->msgType + 1, &qwMsg->connInfo, NULL, 0, code);
dengyihao's avatar
dengyihao 已提交
756 757
      QW_TASK_DLOG("fetch rsp send, handle:%p, code:%x - %s, dataLen:%d", qwMsg->connInfo.handle, code, tstrerror(code),
                   0);
758
    }
D
dapan1121 已提交
759

D
dapan1121 已提交
760
    QW_LOCK(QW_WRITE, &ctx->lock);
D
dapan1121 已提交
761
    if (atomic_load_8((int8_t*)&ctx->queryEnd) || (queryStop && (0 == atomic_load_8((int8_t *)&ctx->queryContinue))) || code) {
D
dapan1121 已提交
762
      // Note: query is not running anymore
763
      QW_SET_PHASE(ctx, QW_PHASE_POST_CQUERY);
dengyihao's avatar
dengyihao 已提交
764
      QW_UNLOCK(QW_WRITE, &ctx->lock);
D
dapan1121 已提交
765 766
      break;
    }
dengyihao's avatar
dengyihao 已提交
767
    QW_UNLOCK(QW_WRITE, &ctx->lock);
D
dapan1121 已提交
768
  } while (true);
D
dapan1121 已提交
769

D
dapan1121 已提交
770
  input.code = code;
dengyihao's avatar
dengyihao 已提交
771
  qwHandlePostPhaseEvents(QW_FPARAMS(), QW_PHASE_POST_CQUERY, &input, NULL);
D
dapan1121 已提交
772

dengyihao's avatar
dengyihao 已提交
773
  QW_RET(TSDB_CODE_SUCCESS);
D
dapan1121 已提交
774
}
D
dapan1121 已提交
775

D
dapan1121 已提交
776
int32_t qwProcessFetch(QW_FPARAMS_DEF, SQWMsg *qwMsg) {
dengyihao's avatar
dengyihao 已提交
777 778 779
  int32_t       code = 0;
  int32_t       dataLen = 0;
  bool          locked = false;
780 781
  SQWTaskCtx   *ctx = NULL;
  void         *rsp = NULL;
D
dapan1121 已提交
782
  SQWPhaseInput input = {0};
D
dapan1121 已提交
783

D
dapan1121 已提交
784
  QW_ERR_JRET(qwHandlePrePhaseEvents(QW_FPARAMS(), QW_PHASE_PRE_FETCH, &input, NULL));
785

786
  QW_ERR_JRET(qwGetTaskCtx(QW_FPARAMS(), &ctx));
dengyihao's avatar
dengyihao 已提交
787

D
dapan1121 已提交
788
  ctx->msgType = qwMsg->msgType;
D
dapan1121 已提交
789
  ctx->dataConnInfo = qwMsg->connInfo;
D
dapan1121 已提交
790

D
dapan 已提交
791
  SOutputData sOutput = {0};
D
dapan1121 已提交
792
  QW_ERR_JRET(qwGetQueryResFromSink(QW_FPARAMS(), ctx, &dataLen, &rsp, &sOutput));
D
dapan1121 已提交
793

794 795
  if (NULL == rsp) {
    QW_SET_EVENT_RECEIVED(ctx, QW_EVENT_FETCH);
D
dapan1121 已提交
796
  } else {
D
dapan1121 已提交
797
    bool qComplete = (DS_BUF_EMPTY == sOutput.bufStatus && sOutput.queryEnd);
dengyihao's avatar
dengyihao 已提交
798

D
dapan1121 已提交
799
    qwBuildFetchRsp(rsp, &sOutput, dataLen, qComplete);
D
dapan1121 已提交
800
    if (qComplete) {
dengyihao's avatar
dengyihao 已提交
801
      atomic_store_8((int8_t *)&ctx->queryEnd, true);
D
dapan1121 已提交
802
    }
D
dapan1121 已提交
803 804
  }

dengyihao's avatar
dengyihao 已提交
805
  if ((!sOutput.queryEnd) && (DS_BUF_LOW == sOutput.bufStatus || DS_BUF_EMPTY == sOutput.bufStatus)) {
D
dapan1121 已提交
806
    QW_TASK_DLOG("task not end and buf is %s, need to continue query", qwBufStatusStr(sOutput.bufStatus));
D
dapan1121 已提交
807

D
dapan1121 已提交
808 809
    QW_LOCK(QW_WRITE, &ctx->lock);
    locked = true;
810

D
dapan1121 已提交
811
    // RC WARNING
812 813 814
    if (-1 == ctx->phase || false == ctx->queryGotData) {
      QW_TASK_DLOG_E("task query unfinished");
    } else if (QW_QUERY_RUNNING(ctx)) {
dengyihao's avatar
dengyihao 已提交
815 816
      atomic_store_8((int8_t *)&ctx->queryContinue, 1);
    } else if (0 == atomic_load_8((int8_t *)&ctx->queryInQueue)) {
D
dapan1121 已提交
817
      qwUpdateTaskStatus(QW_FPARAMS(), JOB_TASK_STATUS_EXEC);
D
dapan1121 已提交
818

dengyihao's avatar
dengyihao 已提交
819 820
      atomic_store_8((int8_t *)&ctx->queryInQueue, 1);

D
dapan1121 已提交
821
      QW_ERR_JRET(qwBuildAndSendCQueryMsg(QW_FPARAMS(), &qwMsg->connInfo));
822
    }
D
dapan 已提交
823
  }
dengyihao's avatar
dengyihao 已提交
824

D
dapan1121 已提交
825
_return:
D
dapan1121 已提交
826

D
dapan1121 已提交
827 828 829 830 831
  if (locked) {
    QW_UNLOCK(QW_WRITE, &ctx->lock);
  }

  input.code = code;
D
dapan1121 已提交
832
  code = qwHandlePostPhaseEvents(QW_FPARAMS(), QW_PHASE_POST_FETCH, &input, NULL);
D
dapan1121 已提交
833

D
dapan 已提交
834 835 836
  if (code) {
    qwFreeFetchRsp(rsp);
    rsp = NULL;
D
dapan1121 已提交
837
    dataLen = 0;
D
dapan1121 已提交
838 839 840
  }

  if (code || rsp) {
D
dapan1121 已提交
841
    bool rsped = false;
D
dapan1121 已提交
842
    if (ctx) {
843
      qwDbgSimulateRedirect(qwMsg, ctx, &rsped);
D
dapan1121 已提交
844 845
      qwDbgSimulateDead(QW_FPARAMS(), ctx, &rsped);
    }
D
dapan1121 已提交
846 847
    if (!rsped) {
      qwBuildAndSendFetchRsp(qwMsg->msgType + 1, &qwMsg->connInfo, rsp, dataLen, code);
848 849
      QW_TASK_DLOG("%s send, handle:%p, code:%x - %s, dataLen:%d", TMSG_INFO(qwMsg->msgType + 1),
                   qwMsg->connInfo.handle, code, tstrerror(code), dataLen);
850 851 852
    } else {
      qwFreeFetchRsp(rsp);
      rsp = NULL;
D
dapan1121 已提交
853
    }
D
dapan1121 已提交
854 855
  }

D
dapan1121 已提交
856
  QW_RET(TSDB_CODE_SUCCESS);
D
dapan1121 已提交
857
}
D
dapan1121 已提交
858

D
dapan1121 已提交
859
int32_t qwProcessDrop(QW_FPARAMS_DEF, SQWMsg *qwMsg) {
dengyihao's avatar
dengyihao 已提交
860
  int32_t     code = 0;
D
dapan1121 已提交
861
  bool        dropped = false;
D
dapan1121 已提交
862
  SQWTaskCtx *ctx = NULL;
dengyihao's avatar
dengyihao 已提交
863
  bool        locked = false;
D
dapan1121 已提交
864

D
dapan1121 已提交
865
  QW_ERR_JRET(qwAcquireTaskCtx(QW_FPARAMS(), &ctx));
dengyihao's avatar
dengyihao 已提交
866

D
dapan1121 已提交
867 868 869 870
  QW_LOCK(QW_WRITE, &ctx->lock);

  locked = true;

D
dapan1121 已提交
871
  if (QW_EVENT_RECEIVED(ctx, QW_EVENT_DROP)) {
D
dapan1121 已提交
872
    QW_TASK_WLOG_E("task already dropping");
D
dapan1121 已提交
873 874 875
    QW_ERR_JRET(TSDB_CODE_QRY_DUPLICATTED_OPERATION);
  }

D
dapan1121 已提交
876
  if (QW_QUERY_RUNNING(ctx)) {
D
dapan1121 已提交
877
    QW_ERR_JRET(qwKillTaskHandle(ctx, TSDB_CODE_TSC_QUERY_CANCELLED));
D
dapan1121 已提交
878
    qwUpdateTaskStatus(QW_FPARAMS(), JOB_TASK_STATUS_DROP);
D
dapan1121 已提交
879
  } else {
D
dapan1121 已提交
880
    QW_ERR_JRET(qwDropTask(QW_FPARAMS()));
D
dapan1121 已提交
881
    dropped = true;
D
dapan1121 已提交
882
  }
D
dapan1121 已提交
883

D
dapan1121 已提交
884
  if (!dropped) {
D
dapan1121 已提交
885
    QW_UPDATE_RSP_CODE(ctx, TSDB_CODE_TSC_QUERY_CANCELLED);
D
dapan1121 已提交
886 887
    QW_SET_EVENT_RECEIVED(ctx, QW_EVENT_DROP);
  }
888

D
dapan1121 已提交
889
_return:
D
dapan1121 已提交
890

D
dapan1121 已提交
891
  if (code) {
D
dapan1121 已提交
892 893 894
    if (ctx) {
      QW_UPDATE_RSP_CODE(ctx, code);
    }
H
Haojun Liao 已提交
895

D
dapan1121 已提交
896
    qwUpdateTaskStatus(QW_FPARAMS(), JOB_TASK_STATUS_FAIL);
D
dapan1121 已提交
897 898
  }

D
dapan 已提交
899 900 901 902
  if (locked) {
    QW_UNLOCK(QW_WRITE, &ctx->lock);
  }

D
dapan1121 已提交
903
  if (ctx) {
D
dapan1121 已提交
904
    qwReleaseTaskCtx(mgmt, ctx);
D
dapan1121 已提交
905 906
  }

D
dapan1121 已提交
907 908 909
  QW_RET(TSDB_CODE_SUCCESS);
}

D
dapan1121 已提交
910
int32_t qwProcessHb(SQWorker *mgmt, SQWMsg *qwMsg, SSchedulerHbReq *req) {
dengyihao's avatar
dengyihao 已提交
911
  int32_t         code = 0;
D
dapan1121 已提交
912
  SSchedulerHbRsp rsp = {0};
913
  SQWSchStatus   *sch = NULL;
D
dapan1121 已提交
914

D
dapan1121 已提交
915 916 917
  if (qwMsg->code) {
    QW_RET(qwProcessHbLinkBroken(mgmt, qwMsg, req));
  }
D
dapan1121 已提交
918 919

  QW_ERR_JRET(qwAcquireAddScheduler(mgmt, req->sId, QW_READ, &sch));
D
dapan1121 已提交
920 921
  QW_ERR_JRET(qwRegisterHbBrokenLinkArg(mgmt, req->sId, &qwMsg->connInfo));

D
dapan1121 已提交
922 923
  sch->hbBrokenTs = 0;

D
dapan1121 已提交
924
  QW_LOCK(QW_WRITE, &sch->hbConnLock);
D
dapan1121 已提交
925

D
dapan1121 已提交
926
  if (sch->hbConnInfo.handle) {
927
    tmsgReleaseHandle(&sch->hbConnInfo, TAOS_CONN_SERVER);
S
Shengliang Guan 已提交
928
    sch->hbConnInfo.handle = NULL;
D
dapan1121 已提交
929
  }
D
dapan1121 已提交
930

D
dapan1121 已提交
931
  memcpy(&sch->hbConnInfo, &qwMsg->connInfo, sizeof(qwMsg->connInfo));
D
dapan1121 已提交
932
  memcpy(&sch->hbEpId, &req->epId, sizeof(req->epId));
dengyihao's avatar
dengyihao 已提交
933

D
dapan1121 已提交
934
  QW_UNLOCK(QW_WRITE, &sch->hbConnLock);
dengyihao's avatar
dengyihao 已提交
935 936 937

  QW_DLOG("hb connection updated, sId:%" PRIx64 ", nodeId:%d, fqdn:%s, port:%d, handle:%p, ahandle:%p", req->sId,
          req->epId.nodeId, req->epId.ep.fqdn, req->epId.ep.port, qwMsg->connInfo.handle, qwMsg->connInfo.ahandle);
D
dapan1121 已提交
938

D
dapan1121 已提交
939 940 941 942
  qwReleaseScheduler(QW_READ, mgmt);

_return:

D
dapan1121 已提交
943 944
  memcpy(&rsp.epId, &req->epId, sizeof(req->epId));

S
shm  
Shengliang Guan 已提交
945
  qwBuildAndSendHbRsp(&qwMsg->connInfo, &rsp, code);
D
dapan1121 已提交
946 947

  if (code) {
948
    tmsgReleaseHandle(&qwMsg->connInfo, TAOS_CONN_SERVER);
S
Shengliang Guan 已提交
949
    qwMsg->connInfo.handle = NULL;
D
dapan1121 已提交
950
  }
dengyihao's avatar
dengyihao 已提交
951

952
  /*QW_DLOG("hb rsp send, handle:%p, code:%x - %s", qwMsg->connInfo.handle, code, tstrerror(code));*/
dengyihao's avatar
dengyihao 已提交
953

D
dapan1121 已提交
954
  QW_RET(TSDB_CODE_SUCCESS);
D
dapan1121 已提交
955 956 957
}

void qwProcessHbTimerEvent(void *param, void *tmrId) {
958
  SQWHbParam *hbParam = (SQWHbParam *)param;
D
dapan1121 已提交
959 960 961 962
  if (hbParam->qwrId != atomic_load_32(&gQwMgmt.qwRef)) {
    return;
  }

963
  int64_t   refId = hbParam->refId;
D
dapan1121 已提交
964 965 966 967 968
  SQWorker *mgmt = qwAcquire(refId);
  if (NULL == mgmt) {
    QW_DLOG("qwAcquire %" PRIx64 "failed", refId);
    return;
  }
969

D
dapan1121 已提交
970
  SQWSchStatus *sch = NULL;
dengyihao's avatar
dengyihao 已提交
971
  int32_t       taskNum = 0;
972
  SQWHbInfo    *rspList = NULL;
D
dapan1121 已提交
973
  SArray       *pExpiredSch = NULL;
dengyihao's avatar
dengyihao 已提交
974
  int32_t       code = 0;
D
dapan1121 已提交
975

D
dapan1121 已提交
976 977
  qwDbgDumpMgmtInfo(mgmt);

D
dapan1121 已提交
978 979 980 981 982
  QW_LOCK(QW_READ, &mgmt->schLock);

  int32_t schNum = taosHashGetSize(mgmt->schHash);
  if (schNum <= 0) {
    QW_UNLOCK(QW_READ, &mgmt->schLock);
D
dapan1121 已提交
983
    taosTmrReset(qwProcessHbTimerEvent, QW_DEFAULT_HEARTBEAT_MSEC, param, mgmt->timer, &mgmt->hbTimer);
D
dapan1121 已提交
984
    qwRelease(refId);
D
dapan1121 已提交
985
    return;
D
dapan1121 已提交
986 987
  }

wafwerar's avatar
wafwerar 已提交
988
  rspList = taosMemoryCalloc(schNum, sizeof(SQWHbInfo));
D
dapan1121 已提交
989 990
  pExpiredSch = taosArrayInit(schNum, sizeof(uint64_t));
  if (NULL == rspList || NULL == pExpiredSch) {
D
dapan1121 已提交
991
    QW_UNLOCK(QW_READ, &mgmt->schLock);
D
dapan1121 已提交
992 993
    taosMemoryFree(rspList);
    taosArrayDestroy(pExpiredSch);
D
dapan1121 已提交
994 995
    QW_ELOG("calloc %d SQWHbInfo failed", schNum);
    taosTmrReset(qwProcessHbTimerEvent, QW_DEFAULT_HEARTBEAT_MSEC, param, mgmt->timer, &mgmt->hbTimer);
D
dapan1121 已提交
996
    qwRelease(refId);
D
dapan1121 已提交
997
    return;
D
dapan1121 已提交
998 999
  }

1000
  void   *key = NULL;
dengyihao's avatar
dengyihao 已提交
1001
  size_t  keyLen = 0;
D
dapan1121 已提交
1002
  int32_t i = 0;
D
dapan1121 已提交
1003
  int64_t currentMs = taosGetTimestampMs();
D
dapan1121 已提交
1004 1005 1006

  void *pIter = taosHashIterate(mgmt->schHash, NULL);
  while (pIter) {
H
Haojun Liao 已提交
1007 1008
    SQWSchStatus *sch1 = (SQWSchStatus *)pIter;
    if (NULL == sch1->hbConnInfo.handle) {
D
dapan1121 已提交
1009
      uint64_t *sId = taosHashGetKey(pIter, NULL);
D
dapan1121 已提交
1010
      QW_TLOG("cancel send hb to sch %" PRIx64 " cause of no connection handle", *sId);
D
dapan1121 已提交
1011

H
Haojun Liao 已提交
1012 1013
      if (sch1->hbBrokenTs > 0 && ((currentMs - sch1->hbBrokenTs) > QW_SCH_TIMEOUT_MSEC) &&
          taosHashGetSize(sch1->tasksHash) <= 0) {
D
dapan1121 已提交
1014 1015
        taosArrayPush(pExpiredSch, sId);
      }
1016

D
dapan1121 已提交
1017 1018 1019
      pIter = taosHashIterate(mgmt->schHash, pIter);
      continue;
    }
dengyihao's avatar
dengyihao 已提交
1020

D
dapan1121 已提交
1021 1022 1023 1024 1025 1026 1027 1028 1029 1030 1031 1032 1033 1034 1035
    code = qwGenerateSchHbRsp(mgmt, (SQWSchStatus *)pIter, &rspList[i]);
    if (code) {
      taosHashCancelIterate(mgmt->schHash, pIter);
      QW_ERR_JRET(code);
    }

    ++i;
    pIter = taosHashIterate(mgmt->schHash, pIter);
  }

_return:

  QW_UNLOCK(QW_READ, &mgmt->schLock);

  for (int32_t j = 0; j < i; ++j) {
S
shm  
Shengliang Guan 已提交
1036
    qwBuildAndSendHbRsp(&rspList[j].connInfo, &rspList[j].rsp, code);
1037 1038
    /*QW_DLOG("hb rsp send, handle:%p, code:%x - %s, taskNum:%d", rspList[j].connInfo.handle, code, tstrerror(code),*/
    /*(rspList[j].rsp.taskStatus ? (int32_t)taosArrayGetSize(rspList[j].rsp.taskStatus) : 0));*/
D
dapan1121 已提交
1039
    tFreeSSchedulerHbRsp(&rspList[j].rsp);
D
dapan1121 已提交
1040 1041
  }

D
dapan1121 已提交
1042
  if (taosArrayGetSize(pExpiredSch) > 0) {
D
dapan1121 已提交
1043
    qwClearExpiredSch(mgmt, pExpiredSch);
D
dapan1121 已提交
1044 1045
  }

wafwerar's avatar
wafwerar 已提交
1046
  taosMemoryFreeClear(rspList);
D
dapan1121 已提交
1047
  taosArrayDestroy(pExpiredSch);
D
dapan1121 已提交
1048

dengyihao's avatar
dengyihao 已提交
1049
  taosTmrReset(qwProcessHbTimerEvent, QW_DEFAULT_HEARTBEAT_MSEC, param, mgmt->timer, &mgmt->hbTimer);
1050
  qwRelease(refId);
D
dapan1121 已提交
1051 1052
}

1053
int32_t qwProcessDelete(QW_FPARAMS_DEF, SQWMsg *qwMsg, SDeleteRes *pRes) {
D
dapan1121 已提交
1054 1055 1056 1057 1058 1059
  int32_t        code = 0;
  SSubplan      *plan = NULL;
  qTaskInfo_t    pTaskInfo = NULL;
  DataSinkHandle sinkHandle = NULL;
  SQWTaskCtx     ctx = {0};

1060
  code = qMsgToSubplan(qwMsg->msg, qwMsg->msgLen, &plan);
D
dapan1121 已提交
1061 1062 1063 1064 1065 1066
  if (TSDB_CODE_SUCCESS != code) {
    code = TSDB_CODE_INVALID_MSG;
    QW_TASK_ELOG("task physical plan to subplan failed, code:%x - %s", code, tstrerror(code));
    QW_ERR_JRET(code);
  }

1067
  code = qCreateExecTask(qwMsg->node, mgmt->nodeId, tId, plan, &pTaskInfo, &sinkHandle, NULL, OPTR_EXEC_MODEL_BATCH);
D
dapan1121 已提交
1068 1069 1070 1071 1072 1073 1074
  if (code) {
    QW_TASK_ELOG("qCreateExecTask failed, code:%x - %s", code, tstrerror(code));
    QW_ERR_JRET(code);
  }

  if (NULL == sinkHandle || NULL == pTaskInfo) {
    QW_TASK_ELOG("create task result error, taskHandle:%p, sinkHandle:%p", pTaskInfo, sinkHandle);
S
Shengliang Guan 已提交
1075
    QW_ERR_JRET(TSDB_CODE_APP_ERROR);
D
dapan1121 已提交
1076 1077
  }

D
dapan1121 已提交
1078 1079
  ctx.taskHandle = pTaskInfo;
  ctx.sinkHandle = sinkHandle;
D
dapan1121 已提交
1080

D
dapan1121 已提交
1081
  QW_ERR_JRET(qwExecTask(QW_FPARAMS(), &ctx, NULL));
D
dapan1121 已提交
1082

1083
  QW_ERR_JRET(qwGetDeleteResFromSink(QW_FPARAMS(), &ctx, pRes));
D
dapan1121 已提交
1084 1085 1086

_return:

D
dapan1121 已提交
1087
  qwFreeTaskCtx(&ctx);
D
dapan1121 已提交
1088 1089 1090 1091

  QW_RET(TSDB_CODE_SUCCESS);
}

D
dapan1121 已提交
1092 1093
int32_t qWorkerInit(int8_t nodeType, int32_t nodeId, void **qWorkerMgmt, const SMsgCb *pMsgCb) {
  if (NULL == qWorkerMgmt || (pMsgCb && pMsgCb->mgmt == NULL)) {
D
dapan1121 已提交
1094 1095 1096
    qError("invalid param to init qworker");
    QW_RET(TSDB_CODE_QRY_INVALID_INPUT);
  }
S
Shengliang 已提交
1097

D
dapan1121 已提交
1098 1099 1100 1101
  int32_t qwNum = atomic_add_fetch_32(&gQwMgmt.qwNum, 1);
  if (1 == qwNum) {
    memset(gQwMgmt.param, 0, sizeof(gQwMgmt.param));
  }
D
dapan1121 已提交
1102 1103 1104 1105 1106 1107 1108 1109

  int32_t code = qwOpenRef();
  if (code) {
    atomic_sub_fetch_32(&gQwMgmt.qwNum, 1);
    QW_RET(code);
  }

  SQWorker *mgmt = taosMemoryCalloc(1, sizeof(SQWorker));
D
dapan1121 已提交
1110
  if (NULL == mgmt) {
D
dapan1121 已提交
1111 1112
    qError("calloc %d failed", (int32_t)sizeof(SQWorker));
    atomic_sub_fetch_32(&gQwMgmt.qwNum, 1);
S
Shengliang Guan 已提交
1113
    QW_RET(TSDB_CODE_OUT_OF_MEMORY);
D
dapan1121 已提交
1114 1115
  }

D
dapan1121 已提交
1116 1117 1118
  mgmt->cfg.maxSchedulerNum = QW_DEFAULT_SCHEDULER_NUMBER;
  mgmt->cfg.maxTaskNum = QW_DEFAULT_TASK_NUMBER;
  mgmt->cfg.maxSchTaskNum = QW_DEFAULT_SCH_TASK_NUMBER;
D
dapan1121 已提交
1119

dengyihao's avatar
dengyihao 已提交
1120 1121
  mgmt->schHash = taosHashInit(mgmt->cfg.maxSchedulerNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_UBIGINT), false,
                               HASH_ENTRY_LOCK);
D
dapan1121 已提交
1122
  if (NULL == mgmt->schHash) {
wafwerar's avatar
wafwerar 已提交
1123
    taosMemoryFreeClear(mgmt);
D
dapan1121 已提交
1124
    qError("init %d scheduler hash failed", mgmt->cfg.maxSchedulerNum);
S
Shengliang Guan 已提交
1125
    QW_ERR_JRET(TSDB_CODE_OUT_OF_MEMORY);
D
dapan1121 已提交
1126 1127
  }

dengyihao's avatar
dengyihao 已提交
1128 1129
  mgmt->ctxHash =
      taosHashInit(mgmt->cfg.maxTaskNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_ENTRY_LOCK);
D
dapan1121 已提交
1130
  if (NULL == mgmt->ctxHash) {
D
dapan1121 已提交
1131
    qError("init %d task ctx hash failed", mgmt->cfg.maxTaskNum);
S
Shengliang Guan 已提交
1132
    QW_ERR_JRET(TSDB_CODE_OUT_OF_MEMORY);
D
dapan1121 已提交
1133 1134 1135 1136 1137
  }

  mgmt->timer = taosTmrInit(0, 0, 0, "qworker");
  if (NULL == mgmt->timer) {
    qError("init timer failed, error:%s", tstrerror(terrno));
S
Shengliang Guan 已提交
1138
    QW_ERR_JRET(TSDB_CODE_OUT_OF_MEMORY);
D
dapan1121 已提交
1139 1140
  }

D
dapan1121 已提交
1141 1142
  mgmt->nodeType = nodeType;
  mgmt->nodeId = nodeId;
D
dapan1121 已提交
1143 1144 1145 1146 1147
  if (pMsgCb) {
    mgmt->msgCb = *pMsgCb;
  } else {
    memset(&mgmt->msgCb, 0, sizeof(mgmt->msgCb));
  }
D
dapan1121 已提交
1148

D
dapan1121 已提交
1149 1150 1151 1152 1153 1154
  mgmt->refId = taosAddRef(gQwMgmt.qwRef, mgmt);
  if (mgmt->refId < 0) {
    qError("taosAddRef qw failed, error:%s", tstrerror(terrno));
    QW_ERR_JRET(terrno);
  }

D
dapan1121 已提交
1155 1156 1157
  SQWHbParam *param = NULL;
  qwSetHbParam(mgmt->refId, &param);

1158
  mgmt->hbTimer = taosTmrStart(qwProcessHbTimerEvent, QW_DEFAULT_HEARTBEAT_MSEC, (void *)param, mgmt->timer);
D
dapan1121 已提交
1159 1160
  if (NULL == mgmt->hbTimer) {
    qError("start hb timer failed");
S
Shengliang Guan 已提交
1161
    QW_ERR_JRET(TSDB_CODE_OUT_OF_MEMORY);
D
dapan1121 已提交
1162
  }
1163

D
dapan1121 已提交
1164 1165
  *qWorkerMgmt = mgmt;

S
Shengliang Guan 已提交
1166
  qDebug("qworker initialized, type:%d, id:%d, handle:%p", mgmt->nodeType, mgmt->nodeId, mgmt);
D
dapan1121 已提交
1167

D
dapan1121 已提交
1168
  return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
1169 1170 1171

_return:

D
dapan1121 已提交
1172 1173 1174 1175 1176 1177 1178
  if (mgmt->refId >= 0) {
    qwRelease(mgmt->refId);
  } else {
    taosHashCleanup(mgmt->schHash);
    taosHashCleanup(mgmt->ctxHash);
    taosTmrCleanUp(mgmt->timer);
    taosMemoryFreeClear(mgmt);
D
dapan1121 已提交
1179

1180
    atomic_sub_fetch_32(&gQwMgmt.qwNum, 1);
D
dapan1121 已提交
1181
  }
1182

D
dapan1121 已提交
1183
  QW_RET(code);
D
dapan1121 已提交
1184
}
D
dapan1121 已提交
1185

D
dapan1121 已提交
1186
void qWorkerStopAllTasks(void *qWorkerMgmt) {
D
dapan1121 已提交
1187 1188 1189
  SQWorker *mgmt = (SQWorker *)qWorkerMgmt;

  QW_DLOG("start to stop all tasks, taskNum:%d", taosHashGetSize(mgmt->ctxHash));
D
dapan1121 已提交
1190

1191
  uint64_t qId, tId, sId;
dengyihao's avatar
dengyihao 已提交
1192
  int32_t  eId;
1193
  int64_t  rId = 0;
1194 1195 1196
  
  atomic_store_8(&mgmt->nodeStopped, 1);

D
dapan1121 已提交
1197 1198 1199 1200 1201 1202 1203 1204
  void    *pIter = taosHashIterate(mgmt->ctxHash, NULL);
  while (pIter) {
    SQWTaskCtx *ctx = (SQWTaskCtx *)pIter;
    void       *key = taosHashGetKey(pIter, NULL);
    QW_GET_QTID(key, qId, tId, eId);

    QW_LOCK(QW_WRITE, &ctx->lock);

1205 1206
    sId = ctx->sId;

D
dapan1121 已提交
1207
    QW_TASK_DLOG_E("start to force stop task");
dengyihao's avatar
dengyihao 已提交
1208

D
dapan1121 已提交
1209 1210 1211 1212 1213 1214 1215
    if (QW_EVENT_RECEIVED(ctx, QW_EVENT_DROP) || QW_EVENT_PROCESSED(ctx, QW_EVENT_DROP)) {
      QW_TASK_WLOG_E("task already dropping");
      QW_UNLOCK(QW_WRITE, &ctx->lock);

      pIter = taosHashIterate(mgmt->ctxHash, pIter);
      continue;
    }
dengyihao's avatar
dengyihao 已提交
1216

D
dapan1121 已提交
1217
    if (QW_QUERY_RUNNING(ctx)) {
D
dapan1121 已提交
1218
      qwKillTaskHandle(ctx, TSDB_CODE_VND_STOPPED);
1219
    } else if (QW_FETCH_RUNNING(ctx)) {
D
dapan1121 已提交
1220
      QW_UPDATE_RSP_CODE(ctx, TSDB_CODE_VND_STOPPED);
1221 1222
      QW_SET_EVENT_RECEIVED(ctx, QW_EVENT_DROP);    
    } else {
1223
      (void)qwDropTask(QW_FPARAMS());
D
dapan1121 已提交
1224 1225 1226 1227 1228 1229 1230 1231
    }

    QW_UNLOCK(QW_WRITE, &ctx->lock);

    pIter = taosHashIterate(mgmt->ctxHash, pIter);
  }
}

D
dapan1121 已提交
1232 1233 1234
void qWorkerDestroy(void **qWorkerMgmt) {
  if (NULL == qWorkerMgmt || NULL == *qWorkerMgmt) {
    return;
D
dapan1121 已提交
1235
  }
D
dapan 已提交
1236

H
Hongze Cheng 已提交
1237
  int32_t   destroyed = 0;
D
dapan1121 已提交
1238
  SQWorker *mgmt = *qWorkerMgmt;
D
dapan1121 已提交
1239
  mgmt->destroyed = &destroyed;
D
dapan1121 已提交
1240

D
dapan1121 已提交
1241 1242
  if (taosRemoveRef(gQwMgmt.qwRef, mgmt->refId)) {
    qError("remove qw from ref list failed, refId:%" PRIx64, mgmt->refId);
D
dapan1121 已提交
1243 1244 1245 1246 1247
    return;
  }

  while (0 == destroyed) {
    taosMsleep(2);
D
dapan1121 已提交
1248
  }
D
dapan1121 已提交
1249
}
D
dapan1121 已提交
1250

D
dapan1121 已提交
1251 1252 1253 1254 1255
int32_t qWorkerGetStat(SReadHandle *handle, void *qWorkerMgmt, SQWorkerStat *pStat) {
  if (NULL == handle || NULL == qWorkerMgmt || NULL == pStat) {
    QW_RET(TSDB_CODE_QRY_INVALID_INPUT);
  }

1256
  SQWorker     *mgmt = (SQWorker *)qWorkerMgmt;
D
dapan1121 已提交
1257
  SDataSinkStat sinkStat = {0};
1258

D
dapan1121 已提交
1259 1260
  dsDataSinkGetCacheSize(&sinkStat);
  pStat->cacheDataSize = sinkStat.cachedSize;
1261

D
dapan1121 已提交
1262 1263 1264 1265 1266
  pStat->queryProcessed = QW_STAT_GET(mgmt->stat.msgStat.queryProcessed);
  pStat->cqueryProcessed = QW_STAT_GET(mgmt->stat.msgStat.cqueryProcessed);
  pStat->fetchProcessed = QW_STAT_GET(mgmt->stat.msgStat.fetchProcessed);
  pStat->dropProcessed = QW_STAT_GET(mgmt->stat.msgStat.dropProcessed);
  pStat->hbProcessed = QW_STAT_GET(mgmt->stat.msgStat.hbProcessed);
D
dapan1121 已提交
1267
  pStat->deleteProcessed = QW_STAT_GET(mgmt->stat.msgStat.deleteProcessed);
D
dapan1121 已提交
1268 1269 1270 1271 1272 1273 1274

  pStat->numOfQueryInQueue = handle->pMsgCb->qsizeFp(handle->pMsgCb->mgmt, mgmt->nodeId, QUERY_QUEUE);
  pStat->numOfFetchInQueue = handle->pMsgCb->qsizeFp(handle->pMsgCb->mgmt, mgmt->nodeId, FETCH_QUEUE);
  pStat->timeInQueryQueue = qwGetTimeInQueue((SQWorker *)qWorkerMgmt, QUERY_QUEUE);
  pStat->timeInFetchQueue = qwGetTimeInQueue((SQWorker *)qWorkerMgmt, FETCH_QUEUE);

  return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
1275
}
D
dapan1121 已提交
1276

H
Hongze Cheng 已提交
1277 1278 1279
int32_t qWorkerProcessLocalQuery(void *pMgmt, uint64_t sId, uint64_t qId, uint64_t tId, int64_t rId, int32_t eId,
                                 SQWMsg *qwMsg, SArray *explainRes) {
  SQWorker      *mgmt = (SQWorker *)pMgmt;
D
dapan1121 已提交
1280 1281 1282 1283 1284 1285
  int32_t        code = 0;
  SQWTaskCtx    *ctx = NULL;
  SSubplan      *plan = (SSubplan *)qwMsg->msg;
  SQWPhaseInput  input = {0};
  qTaskInfo_t    pTaskInfo = NULL;
  DataSinkHandle sinkHandle = NULL;
H
Hongze Cheng 已提交
1286
  SReadHandle    rHandle = {0};
D
dapan1121 已提交
1287 1288

  QW_ERR_JRET(qwAddTaskCtx(QW_FPARAMS()));
D
dapan1121 已提交
1289
  QW_ERR_JRET(qwAddTaskStatus(QW_FPARAMS(), JOB_TASK_STATUS_INIT));
D
dapan1121 已提交
1290

D
dapan1121 已提交
1291
  QW_ERR_JRET(qwHandlePrePhaseEvents(QW_FPARAMS(), QW_PHASE_PRE_QUERY, &input, NULL));
D
dapan1121 已提交
1292 1293 1294 1295 1296 1297
  QW_ERR_JRET(qwAcquireTaskCtx(QW_FPARAMS(), &ctx));

  ctx->taskType = qwMsg->msgInfo.taskType;
  ctx->explain = qwMsg->msgInfo.explain;
  ctx->needFetch = qwMsg->msgInfo.needFetch;
  ctx->msgType = qwMsg->msgType;
D
dapan1121 已提交
1298 1299
  ctx->localExec = true;
  ctx->explainRes = explainRes;
D
dapan1121 已提交
1300

D
dapan1121 已提交
1301 1302
  rHandle.pMsgCb = taosMemoryCalloc(1, sizeof(SMsgCb));
  rHandle.pMsgCb->clientRpc = qwMsg->connInfo.handle;
H
Haojun Liao 已提交
1303

D
dapan1121 已提交
1304
  code = qCreateExecTask(&rHandle, mgmt->nodeId, tId, plan, &pTaskInfo, &sinkHandle, NULL, OPTR_EXEC_MODEL_BATCH);
D
dapan1121 已提交
1305 1306 1307 1308 1309 1310 1311
  if (code) {
    QW_TASK_ELOG("qCreateExecTask failed, code:%x - %s", code, tstrerror(code));
    QW_ERR_JRET(code);
  }

  if (NULL == sinkHandle || NULL == pTaskInfo) {
    QW_TASK_ELOG("create task result error, taskHandle:%p, sinkHandle:%p", pTaskInfo, sinkHandle);
S
Shengliang Guan 已提交
1312
    QW_ERR_JRET(TSDB_CODE_APP_ERROR);
D
dapan1121 已提交
1313 1314 1315 1316 1317 1318 1319 1320 1321 1322
  }

  ctx->level = plan->level;
  atomic_store_ptr(&ctx->taskHandle, pTaskInfo);
  atomic_store_ptr(&ctx->sinkHandle, sinkHandle);

  QW_ERR_JRET(qwExecTask(QW_FPARAMS(), ctx, NULL));

_return:

D
dapan1121 已提交
1323
  taosMemoryFree(rHandle.pMsgCb);
H
Haojun Liao 已提交
1324

D
dapan1121 已提交
1325 1326 1327 1328
  input.code = code;
  input.msgType = qwMsg->msgType;
  code = qwHandlePostPhaseEvents(QW_FPARAMS(), QW_PHASE_POST_QUERY, &input, NULL);

D
dapan1121 已提交
1329 1330 1331 1332 1333 1334 1335 1336
  if (ctx) {
    QW_UPDATE_RSP_CODE(ctx, code);
    qwReleaseTaskCtx(mgmt, ctx);
  }

  QW_RET(code);
}

H
Hongze Cheng 已提交
1337 1338 1339 1340 1341 1342 1343 1344
int32_t qWorkerProcessLocalFetch(void *pMgmt, uint64_t sId, uint64_t qId, uint64_t tId, int64_t rId, int32_t eId,
                                 void **pRsp, SArray *explainRes) {
  SQWorker   *mgmt = (SQWorker *)pMgmt;
  int32_t     code = 0;
  int32_t     dataLen = 0;
  SQWTaskCtx *ctx = NULL;
  void       *rsp = NULL;
  bool        queryStop = false;
D
dapan1121 已提交
1345 1346 1347 1348 1349 1350 1351 1352 1353

  SQWPhaseInput input = {0};

  QW_ERR_JRET(qwHandlePrePhaseEvents(QW_FPARAMS(), QW_PHASE_PRE_FETCH, &input, NULL));

  QW_ERR_JRET(qwGetTaskCtx(QW_FPARAMS(), &ctx));

  ctx->msgType = TDMT_SCH_MERGE_FETCH;
  ctx->explainRes = explainRes;
H
Haojun Liao 已提交
1354

D
dapan1121 已提交
1355
  SOutputData sOutput = {0};
H
Haojun Liao 已提交
1356

D
dapan1121 已提交
1357 1358 1359 1360 1361
  while (true) {
    QW_ERR_JRET(qwGetQueryResFromSink(QW_FPARAMS(), ctx, &dataLen, &rsp, &sOutput));

    if (NULL == rsp) {
      QW_ERR_JRET(qwExecTask(QW_FPARAMS(), ctx, &queryStop));
H
Haojun Liao 已提交
1362

D
dapan1121 已提交
1363 1364 1365 1366 1367 1368 1369 1370 1371 1372 1373 1374 1375 1376 1377 1378 1379 1380 1381 1382 1383 1384
      continue;
    } else {
      bool qComplete = (DS_BUF_EMPTY == sOutput.bufStatus && sOutput.queryEnd);

      qwBuildFetchRsp(rsp, &sOutput, dataLen, qComplete);
      if (qComplete) {
        atomic_store_8((int8_t *)&ctx->queryEnd, true);
      }

      break;
    }
  }

_return:

  *pRsp = rsp;

  input.code = code;
  code = qwHandlePostPhaseEvents(QW_FPARAMS(), QW_PHASE_POST_FETCH, &input, NULL);

  QW_RET(code);
}