tq.c 60.2 KB
Newer Older
H
refact  
Hongze Cheng 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13
/*
 * 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/>.
S
Shengliang Guan 已提交
14 15
 */

H
Hongze Cheng 已提交
16
#include "tq.h"
S
Shengliang Guan 已提交
17

dengyihao's avatar
dengyihao 已提交
18 19 20
// 0: not init
// 1: already inited
// 2: wait to be inited or cleaup
21
static int32_t tqInitialize(STQ* pTq);
dengyihao's avatar
dengyihao 已提交
22

wmmhello's avatar
wmmhello 已提交
23
static FORCE_INLINE bool tqIsHandleExec(STqHandle* pHandle) { return TMQ_HANDLE_STATUS_EXEC == pHandle->status; }
dengyihao's avatar
dengyihao 已提交
24 25
static FORCE_INLINE void tqSetHandleExec(STqHandle* pHandle) { pHandle->status = TMQ_HANDLE_STATUS_EXEC; }
static FORCE_INLINE void tqSetHandleIdle(STqHandle* pHandle) { pHandle->status = TMQ_HANDLE_STATUS_IDLE; }
wmmhello's avatar
wmmhello 已提交
26

L
Liu Jicong 已提交
27
int32_t tqInit() {
L
Liu Jicong 已提交
28 29 30 31 32 33
  int8_t old;
  while (1) {
    old = atomic_val_compare_exchange_8(&tqMgmt.inited, 0, 2);
    if (old != 2) break;
  }

34 35 36 37 38 39
  if (old == 0) {
    tqMgmt.timer = taosTmrInit(10000, 100, 10000, "TQ");
    if (tqMgmt.timer == NULL) {
      atomic_store_8(&tqMgmt.inited, 0);
      return -1;
    }
40 41 42
    if (streamInit() < 0) {
      return -1;
    }
L
Liu Jicong 已提交
43
    atomic_store_8(&tqMgmt.inited, 1);
44
  }
45

L
Liu Jicong 已提交
46 47
  return 0;
}
L
Liu Jicong 已提交
48

49
void tqCleanUp() {
L
Liu Jicong 已提交
50 51 52 53 54 55 56 57
  int8_t old;
  while (1) {
    old = atomic_val_compare_exchange_8(&tqMgmt.inited, 1, 2);
    if (old != 2) break;
  }

  if (old == 1) {
    taosTmrCleanUp(tqMgmt.timer);
L
Liu Jicong 已提交
58
    streamCleanUp();
L
Liu Jicong 已提交
59 60
    atomic_store_8(&tqMgmt.inited, 0);
  }
61
}
L
Liu Jicong 已提交
62

63
void tqDestroyTqHandle(void* data) {
64 65
  STqHandle* pData = (STqHandle*)data;
  qDestroyTask(pData->execHandle.task);
wmmhello's avatar
wmmhello 已提交
66

67
  if (pData->execHandle.subType == TOPIC_SUB_TYPE__COLUMN) {
L
Liu Jicong 已提交
68
    taosMemoryFreeClear(pData->execHandle.execCol.qmsg);
69
  } else if (pData->execHandle.subType == TOPIC_SUB_TYPE__DB) {
70
    tqReaderClose(pData->execHandle.pTqReader);
71 72
    walCloseReader(pData->pWalReader);
    taosHashCleanup(pData->execHandle.execDb.pFilterOutTbUid);
L
Liu Jicong 已提交
73
  } else if (pData->execHandle.subType == TOPIC_SUB_TYPE__TABLE) {
74
    walCloseReader(pData->pWalReader);
75
    tqReaderClose(pData->execHandle.pTqReader);
76 77
    taosMemoryFreeClear(pData->execHandle.execTb.qmsg);
    nodesDestroyNode(pData->execHandle.execTb.node);
78
  }
dengyihao's avatar
dengyihao 已提交
79
  if (pData->msg != NULL) {
80 81 82
    rpcFreeCont(pData->msg->pCont);
    taosMemoryFree(pData->msg);
    pData->msg = NULL;
D
dapan1121 已提交
83
  }
L
Liu Jicong 已提交
84 85
}

86
static bool tqOffsetEqual(const STqOffset* pLeft, const STqOffset* pRight) {
87
  return pLeft->val.type == TMQ_OFFSET__LOG && pRight->val.type == TMQ_OFFSET__LOG &&
88
         pLeft->val.version == pRight->val.version;
89 90
}

L
Liu Jicong 已提交
91
STQ* tqOpen(const char* path, SVnode* pVnode) {
92
  STQ* pTq = taosMemoryCalloc(1, sizeof(STQ));
L
Liu Jicong 已提交
93
  if (pTq == NULL) {
S
Shengliang Guan 已提交
94
    terrno = TSDB_CODE_OUT_OF_MEMORY;
L
Liu Jicong 已提交
95 96
    return NULL;
  }
97

98
  pTq->path = taosStrdup(path);
L
Liu Jicong 已提交
99
  pTq->pVnode = pVnode;
L
Liu Jicong 已提交
100
  pTq->walLogLastVer = pVnode->pWal->vers.lastVer;
101

102
  pTq->pHandle = taosHashInit(64, MurmurHash3_32, true, HASH_ENTRY_LOCK);
103
  taosHashSetFreeFp(pTq->pHandle, tqDestroyTqHandle);
104

105
  taosInitRWLatch(&pTq->lock);
106
  pTq->pPushMgr = taosHashInit(64, MurmurHash3_32, false, HASH_NO_LOCK);
L
Liu Jicong 已提交
107

108
  pTq->pCheckInfo = taosHashInit(64, MurmurHash3_32, true, HASH_ENTRY_LOCK);
L
Liu Jicong 已提交
109
  taosHashSetFreeFp(pTq->pCheckInfo, (FDelete)tDeleteSTqCheckInfo);
L
Liu Jicong 已提交
110

111 112 113 114 115 116 117
  int32_t code = tqInitialize(pTq);
  if (code != TSDB_CODE_SUCCESS) {
    tqClose(pTq);
    return NULL;
  } else {
    return pTq;
  }
118 119 120
}

int32_t tqInitialize(STQ* pTq) {
L
Liu Jicong 已提交
121
  if (tqMetaOpen(pTq) < 0) {
122
    return -1;
123 124
  }

L
Liu Jicong 已提交
125 126
  pTq->pOffsetStore = tqOffsetOpen(pTq);
  if (pTq->pOffsetStore == NULL) {
127
    return -1;
128 129
  }

130
  pTq->pStreamMeta = streamMetaOpen(pTq->path, pTq, (FTaskExpand*)tqExpandTask, pTq->pVnode->config.vgId);
L
Liu Jicong 已提交
131
  if (pTq->pStreamMeta == NULL) {
132
    return -1;
L
Liu Jicong 已提交
133 134
  }

135 136
  // the version is kept in task's meta data
  // todo check if this version is required or not
137 138
  if (streamLoadTasks(pTq->pStreamMeta, walGetCommittedVer(pTq->pVnode->pWal)) < 0) {
    return -1;
L
Liu Jicong 已提交
139 140
  }

141
  return 0;
L
Liu Jicong 已提交
142
}
L
Liu Jicong 已提交
143

L
Liu Jicong 已提交
144
void tqClose(STQ* pTq) {
145 146
  if (pTq == NULL) {
    return;
H
Hongze Cheng 已提交
147
  }
148

wmmhello's avatar
wmmhello 已提交
149 150 151 152 153 154 155 156 157 158 159 160 161 162
  void* pIter = taosHashIterate(pTq->pPushMgr, NULL);
  while (pIter) {
    STqHandle* pHandle = *(STqHandle**)pIter;
    int32_t    vgId = TD_VID(pTq->pVnode);

    if(pHandle->msg != NULL) {
      tqPushEmptyDataRsp(pHandle, vgId);
      rpcFreeCont(pHandle->msg->pCont);
      taosMemoryFree(pHandle->msg);
      pHandle->msg = NULL;
    }
    pIter = taosHashIterate(pTq->pPushMgr, pIter);
  }

163 164 165 166 167 168 169 170
  tqOffsetClose(pTq->pOffsetStore);
  taosHashCleanup(pTq->pHandle);
  taosHashCleanup(pTq->pPushMgr);
  taosHashCleanup(pTq->pCheckInfo);
  taosMemoryFree(pTq->path);
  tqMetaClose(pTq);
  streamMetaClose(pTq->pStreamMeta);
  taosMemoryFree(pTq);
L
Liu Jicong 已提交
171
}
L
Liu Jicong 已提交
172

173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195
static bool hasStreamTaskInTimer(SStreamMeta* pMeta) {
  bool inTimer = false;

  taosWLockLatch(&pMeta->lock);

  void* pIter = NULL;
  while(1) {
    pIter = taosHashIterate(pMeta->pTasks, pIter);
    if (pIter == NULL) {
      break;
    }

    SStreamTask* pTask = *(SStreamTask**)pIter;
    if (pTask->status.timerActive == 1) {
      inTimer = true;
    }
  }

  taosWUnLockLatch(&pMeta->lock);

  return inTimer;
}

H
Haojun Liao 已提交
196 197 198 199 200 201 202 203 204 205 206 207
void tqNotifyClose(STQ* pTq) {
  if (pTq != NULL) {
    taosWLockLatch(&pTq->pStreamMeta->lock);

    void* pIter = NULL;
    while (1) {
      pIter = taosHashIterate(pTq->pStreamMeta->pTasks, pIter);
      if (pIter == NULL) {
        break;
      }

      SStreamTask* pTask = *(SStreamTask**)pIter;
H
Haojun Liao 已提交
208
      tqDebug("vgId:%d s-task:%s set closing flag", pTq->pStreamMeta->vgId, pTask->id.idStr);
209 210 211
      pTask->status.taskStatus = TASK_STATUS__STOP;

      int64_t st = taosGetTimestampMs();
H
Haojun Liao 已提交
212
      qKillTask(pTask->exec.pExecutor, TSDB_CODE_SUCCESS);
H
Haojun Liao 已提交
213

214
      int64_t el = taosGetTimestampMs() - st;
H
Haojun Liao 已提交
215
      tqDebug("vgId:%d s-task:%s is closed in %" PRId64 " ms", pTq->pStreamMeta->vgId, pTask->id.idStr, el);
H
Haojun Liao 已提交
216 217 218
    }

    taosWUnLockLatch(&pTq->pStreamMeta->lock);
H
Haojun Liao 已提交
219 220 221 222 223

    tqDebug("vgId:%d start to check all tasks", pTq->pStreamMeta->vgId);

    int64_t st = taosGetTimestampMs();

224 225 226
    while(hasStreamTaskInTimer(pTq->pStreamMeta)) {
      tqDebug("vgId:%d some tasks in timer, wait for 100ms and recheck", pTq->pStreamMeta->vgId);
      taosMsleep(100);
H
Haojun Liao 已提交
227 228 229 230
    }

    int64_t el = taosGetTimestampMs() - st;
    tqDebug("vgId:%d all stream tasks are not in timer, continue close, elapsed time:%"PRId64" ms", pTq->pStreamMeta->vgId, el);
H
Haojun Liao 已提交
231 232 233
  }
}

234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281
//static int32_t doSendDataRsp(const SRpcHandleInfo* pRpcHandleInfo, const SMqDataRsp* pRsp, int32_t epoch,
//                             int64_t consumerId, int32_t type) {
//  int32_t len = 0;
//  int32_t code = 0;
//
//  if (type == TMQ_MSG_TYPE__POLL_DATA_RSP) {
//    tEncodeSize(tEncodeMqDataRsp, pRsp, len, code);
//  } else if (type == TMQ_MSG_TYPE__POLL_DATA_META_RSP) {
//    tEncodeSize(tEncodeSTaosxRsp, (STaosxRsp*)pRsp, len, code);
//  }
//
//  if (code < 0) {
//    return -1;
//  }
//
//  int32_t tlen = sizeof(SMqRspHead) + len;
//  void*   buf = rpcMallocCont(tlen);
//  if (buf == NULL) {
//    return -1;
//  }
//
//  ((SMqRspHead*)buf)->mqMsgType = type;
//  ((SMqRspHead*)buf)->epoch = epoch;
//  ((SMqRspHead*)buf)->consumerId = consumerId;
//
//  void* abuf = POINTER_SHIFT(buf, sizeof(SMqRspHead));
//
//  SEncoder encoder = {0};
//  tEncoderInit(&encoder, abuf, len);
//
//  if (type == TMQ_MSG_TYPE__POLL_DATA_RSP) {
//    tEncodeMqDataRsp(&encoder, pRsp);
//  } else if (type == TMQ_MSG_TYPE__POLL_DATA_META_RSP) {
//    tEncodeSTaosxRsp(&encoder, (STaosxRsp*)pRsp);
//  }
//
//  tEncoderClear(&encoder);
//
//  SRpcMsg rsp = {
//      .info = *pRpcHandleInfo,
//      .pCont = buf,
//      .contLen = tlen,
//      .code = 0,
//  };
//
//  tmsgSendRsp(&rsp);
//  return 0;
//}
L
Liu Jicong 已提交
282

283 284 285 286 287 288 289
int32_t tqPushEmptyDataRsp(STqHandle* pHandle, int32_t vgId) {
  SMqPollReq req = {0};
  if (tDeserializeSMqPollReq(pHandle->msg->pCont, pHandle->msg->contLen, &req) < 0) {
    tqError("tDeserializeSMqPollReq %d failed", pHandle->msg->contLen);
    terrno = TSDB_CODE_INVALID_MSG;
    return -1;
  }
D
dapan1121 已提交
290

291
  SMqDataRsp dataRsp = {0};
292
  tqInitDataRsp(&dataRsp, req.reqOffset);
293
  dataRsp.blockNum = 0;
wmmhello's avatar
wmmhello 已提交
294 295 296 297
  char buf[TSDB_OFFSET_LEN] = {0};
  tFormatOffset(buf, TSDB_OFFSET_LEN, &dataRsp.reqOffset);
  tqInfo("tqPushEmptyDataRsp to consumer:0x%"PRIx64 " vgId:%d, offset:%s, reqId:0x%" PRIx64, req.consumerId, vgId, buf, req.reqId);

H
Haojun Liao 已提交
298
  tqSendDataRsp(pHandle, pHandle->msg, &req, &dataRsp, TMQ_MSG_TYPE__POLL_DATA_RSP, vgId);
299
  tDeleteMqDataRsp(&dataRsp);
L
Liu Jicong 已提交
300 301 302
  return 0;
}

303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322
//int32_t tqPushDataRsp(STqHandle* pHandle, int32_t vgId) {
//  SMqDataRsp dataRsp = {0};
//  dataRsp.head.consumerId = pHandle->consumerId;
//  dataRsp.head.epoch = pHandle->epoch;
//  dataRsp.head.mqMsgType = TMQ_MSG_TYPE__POLL_RSP;
//
//  int64_t sver = 0, ever = 0;
//  walReaderValidVersionRange(pHandle->execHandle.pTqReader->pWalReader, &sver, &ever);
//  tqDoSendDataRsp(&pHandle->msg->info, &dataRsp, pHandle->epoch, pHandle->consumerId, TMQ_MSG_TYPE__POLL_RSP, sver,
//                  ever);
//
//  char buf1[TSDB_OFFSET_LEN] = {0};
//  char buf2[TSDB_OFFSET_LEN] = {0};
//  tFormatOffset(buf1, tListLen(buf1), &dataRsp.reqOffset);
//  tFormatOffset(buf2, tListLen(buf2), &dataRsp.rspOffset);
//  tqDebug("vgId:%d, from consumer:0x%" PRIx64 " (epoch %d) push rsp, block num: %d, req:%s, rsp:%s", vgId,
//          dataRsp.head.consumerId, dataRsp.head.epoch, dataRsp.blockNum, buf1, buf2);
//  return 0;
//}

323 324 325 326 327 328
int32_t tqSendDataRsp(STqHandle* pHandle, const SRpcMsg* pMsg, const SMqPollReq* pReq, const SMqDataRsp* pRsp,
                      int32_t type, int32_t vgId) {
  int64_t sver = 0, ever = 0;
  walReaderValidVersionRange(pHandle->execHandle.pTqReader->pWalReader, &sver, &ever);

  tqDoSendDataRsp(&pMsg->info, pRsp, pReq->epoch, pReq->consumerId, type, sver, ever);
329

330 331 332 333
  char buf1[TSDB_OFFSET_LEN] = {0};
  char buf2[TSDB_OFFSET_LEN] = {0};
  tFormatOffset(buf1, TSDB_OFFSET_LEN, &pRsp->reqOffset);
  tFormatOffset(buf2, TSDB_OFFSET_LEN, &pRsp->rspOffset);
334

335
  tqDebug("tmq poll vgId:%d consumer:0x%" PRIx64 " (epoch %d) send rsp, block num:%d, req:%s, rsp:%s, reqId:0x%" PRIx64, vgId,
dengyihao's avatar
dengyihao 已提交
336
          pReq->consumerId, pReq->epoch, pRsp->blockNum, buf1, buf2, pReq->reqId);
337 338 339 340

  return 0;
}

341
int32_t tqProcessOffsetCommitReq(STQ* pTq, int64_t sversion, char* msg, int32_t msgLen) {
342 343
  SMqVgOffset vgOffset = {0};
  int32_t     vgId = TD_VID(pTq->pVnode);
344

X
Xiaoyu Wang 已提交
345 346
  SDecoder decoder;
  tDecoderInit(&decoder, (uint8_t*)msg, msgLen);
347
  if (tDecodeMqVgOffset(&decoder, &vgOffset) < 0) {
348 349
    return -1;
  }
350

351 352
  tDecoderClear(&decoder);

353 354 355
  STqOffset* pOffset = &vgOffset.offset;

  if (pOffset->val.type == TMQ_OFFSET__SNAPSHOT_DATA || pOffset->val.type == TMQ_OFFSET__SNAPSHOT_META) {
356
    tqDebug("receive offset commit msg to %s on vgId:%d, offset(type:snapshot) uid:%" PRId64 ", ts:%" PRId64,
357 358
            pOffset->subKey, vgId, pOffset->val.uid, pOffset->val.ts);
  } else if (pOffset->val.type == TMQ_OFFSET__LOG) {
359
    tqDebug("receive offset commit msg to %s on vgId:%d, offset(type:log) version:%" PRId64, pOffset->subKey, vgId,
360
            pOffset->val.version);
361
  } else {
362
    tqError("invalid commit offset type:%d", pOffset->val.type);
363
    return -1;
364
  }
365

366
  STqOffset* pSavedOffset = tqOffsetRead(pTq->pOffsetStore, pOffset->subKey);
367 368
  if (pSavedOffset != NULL && tqOffsetEqual(pOffset, pSavedOffset)) {
    tqInfo("not update the offset, vgId:%d sub:%s since committed:%" PRId64 " less than/equal to existed:%" PRId64,
369
            vgId, pOffset->subKey, pOffset->val.version, pSavedOffset->val.version);
370
    return 0;  // no need to update the offset value
371 372
  }

373
  // save the new offset value
374
  if (tqOffsetWrite(pTq->pOffsetStore, pOffset) < 0) {
375
    return -1;
376
  }
377

378 379 380
  return 0;
}

381 382
int32_t tqProcessSeekReq(STQ* pTq, SRpcMsg* pMsg) {
  SMqSeekReq  req = {0};
383
  int32_t     vgId = TD_VID(pTq->pVnode);
384 385
  SRpcMsg     rsp = {.info = pMsg->info};
  int         code = 0;
386

387 388 389
  if (tDeserializeSMqSeekReq(pMsg->pCont, pMsg->contLen, &req) < 0) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    goto end;
390 391
  }

392
  tqDebug("tmq seek: consumer:0x%" PRIx64 " vgId:%d, subkey %s", req.consumerId, vgId, req.subKey);
393
  STqHandle* pHandle = taosHashGet(pTq->pHandle, req.subKey, strlen(req.subKey));
394
  if (pHandle == NULL) {
395 396 397
    tqWarn("tmq seek: consumer:0x%" PRIx64 " vgId:%d subkey %s not found", req.consumerId, vgId, req.subKey);
    code = 0;
    goto end;
398 399
  }

400 401
  // 2. check consumer-vg assignment status
  taosRLockLatch(&pTq->lock);
402 403 404
  if (pHandle->consumerId != req.consumerId) {
    tqError("ERROR tmq seek: consumer:0x%" PRIx64 " vgId:%d, subkey %s, mismatch for saved handle consumer:0x%" PRIx64,
            req.consumerId, vgId, req.subKey, pHandle->consumerId);
405
    taosRUnLockLatch(&pTq->lock);
406 407
    code = TSDB_CODE_TMQ_CONSUMER_MISMATCH;
    goto end;
408 409
  }

410 411 412 413
  //if consumer register to push manager, push empty to consumer to change vg status from TMQ_VG_STATUS__WAIT to TMQ_VG_STATUS__IDLE,
  //otherwise poll data failed after seek.
  tqUnregisterPushHandle(pTq, pHandle);
  taosRUnLockLatch(&pTq->lock);
H
Haojun Liao 已提交
414

415 416 417
end:
  rsp.code = code;
  tmsgSendRsp(&rsp);
418
  return 0;
419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498

//  SMqVgOffset vgOffset = {0};
//  int32_t     vgId = TD_VID(pTq->pVnode);
//
//  SDecoder decoder;
//  tDecoderInit(&decoder, (uint8_t*)msg, msgLen);
//  if (tDecodeMqVgOffset(&decoder, &vgOffset) < 0) {
//    tqError("vgId:%d failed to decode seek msg", vgId);
//    return -1;
//  }
//
//  tDecoderClear(&decoder);
//
//  tqDebug("topic:%s, vgId:%d process offset seek by consumer:0x%" PRIx64 ", req offset:%" PRId64,
//          vgOffset.offset.subKey, vgId, vgOffset.consumerId, vgOffset.offset.val.version);
//
//  STqOffset* pOffset = &vgOffset.offset;
//  if (pOffset->val.type != TMQ_OFFSET__LOG) {
//    tqError("vgId:%d, subKey:%s invalid seek offset type:%d", vgId, pOffset->subKey, pOffset->val.type);
//    return -1;
//  }
//
//  STqHandle* pHandle = taosHashGet(pTq->pHandle, pOffset->subKey, strlen(pOffset->subKey));
//  if (pHandle == NULL) {
//    tqError("tmq seek: consumer:0x%" PRIx64 " vgId:%d subkey %s not found", vgOffset.consumerId, vgId, pOffset->subKey);
//    terrno = TSDB_CODE_INVALID_MSG;
//    return -1;
//  }
//
//  // 2. check consumer-vg assignment status
//  taosRLockLatch(&pTq->lock);
//  if (pHandle->consumerId != vgOffset.consumerId) {
//    tqDebug("ERROR tmq seek: consumer:0x%" PRIx64 " vgId:%d, subkey %s, mismatch for saved handle consumer:0x%" PRIx64,
//            vgOffset.consumerId, vgId, pOffset->subKey, pHandle->consumerId);
//    terrno = TSDB_CODE_TMQ_CONSUMER_MISMATCH;
//    taosRUnLockLatch(&pTq->lock);
//    return -1;
//  }
//  taosRUnLockLatch(&pTq->lock);
//
//  // 3. check the offset info
//  STqOffset* pSavedOffset = tqOffsetRead(pTq->pOffsetStore, pOffset->subKey);
//  if (pSavedOffset != NULL) {
//    if (pSavedOffset->val.type != TMQ_OFFSET__LOG) {
//      tqError("invalid saved offset type, vgId:%d sub:%s", vgId, pOffset->subKey);
//      return 0;  // no need to update the offset value
//    }
//
//    if (pSavedOffset->val.version == pOffset->val.version) {
//      tqDebug("vgId:%d subKey:%s no need to seek to %" PRId64 " prev offset:%" PRId64, vgId, pOffset->subKey,
//              pOffset->val.version, pSavedOffset->val.version);
//      return 0;
//    }
//  }
//
//  int64_t sver = 0, ever = 0;
//  walReaderValidVersionRange(pHandle->execHandle.pTqReader->pWalReader, &sver, &ever);
//  if (pOffset->val.version < sver) {
//    pOffset->val.version = sver;
//  } else if (pOffset->val.version > ever) {
//    pOffset->val.version = ever;
//  }
//
//  // save the new offset value
//  if (pSavedOffset != NULL) {
//    tqDebug("vgId:%d sub:%s seek to:%" PRId64 " prev offset:%" PRId64, vgId, pOffset->subKey, pOffset->val.version,
//            pSavedOffset->val.version);
//  } else {
//    tqDebug("vgId:%d sub:%s seek to:%" PRId64 " not saved yet", vgId, pOffset->subKey, pOffset->val.version);
//  }
//
//  if (tqOffsetWrite(pTq->pOffsetStore, pOffset) < 0) {
//    tqError("failed to save offset, vgId:%d sub:%s seek to %" PRId64, vgId, pOffset->subKey, pOffset->val.version);
//    return -1;
//  }
//
//  tqDebug("topic:%s, vgId:%d consumer:0x%" PRIx64 " offset is update to:%" PRId64, vgOffset.offset.subKey, vgId,
//          vgOffset.consumerId, vgOffset.offset.val.version);
//
//  return 0;
499 500
}

L
Liu Jicong 已提交
501
int32_t tqCheckColModifiable(STQ* pTq, int64_t tbUid, int32_t colId) {
L
Liu Jicong 已提交
502
  void* pIter = NULL;
503

L
Liu Jicong 已提交
504
  while (1) {
505
    pIter = taosHashIterate(pTq->pCheckInfo, pIter);
506 507 508 509
    if (pIter == NULL) {
      break;
    }

510
    STqCheckInfo* pCheck = (STqCheckInfo*)pIter;
511

L
Liu Jicong 已提交
512 513
    if (pCheck->ntbUid == tbUid) {
      int32_t sz = taosArrayGetSize(pCheck->colIdList);
L
Liu Jicong 已提交
514
      for (int32_t i = 0; i < sz; i++) {
L
Liu Jicong 已提交
515 516
        int16_t forbidColId = *(int16_t*)taosArrayGet(pCheck->colIdList, i);
        if (forbidColId == colId) {
517
          taosHashCancelIterate(pTq->pCheckInfo, pIter);
L
Liu Jicong 已提交
518 519 520 521 522
          return -1;
        }
      }
    }
  }
523

L
Liu Jicong 已提交
524 525 526
  return 0;
}

527 528 529 530 531 532 533 534
int32_t tqProcessPollPush(STQ* pTq, SRpcMsg* pMsg) {
  int32_t vgId = TD_VID(pTq->pVnode);
  taosWLockLatch(&pTq->lock);
  if (taosHashGetSize(pTq->pPushMgr) > 0) {
    void* pIter = taosHashIterate(pTq->pPushMgr, NULL);

    while (pIter) {
      STqHandle* pHandle = *(STqHandle**)pIter;
wmmhello's avatar
wmmhello 已提交
535
      tqInfo("vgId:%d start set submit for pHandle:%p, consumer:0x%" PRIx64, vgId, pHandle, pHandle->consumerId);
536 537 538

      if (ASSERT(pHandle->msg != NULL)) {
        tqError("pHandle->msg should not be null");
wmmhello's avatar
wmmhello 已提交
539
        taosHashCancelIterate(pTq->pPushMgr, pIter);
540 541 542 543 544 545 546 547 548 549 550 551 552
        break;
      }else{
        SRpcMsg msg = {.msgType = TDMT_VND_TMQ_CONSUME, .pCont = pHandle->msg->pCont, .contLen = pHandle->msg->contLen, .info = pHandle->msg->info};
        tmsgPutToQueue(&pTq->pVnode->msgCb, QUERY_QUEUE, &msg);
        taosMemoryFree(pHandle->msg);
        pHandle->msg = NULL;
      }

      pIter = taosHashIterate(pTq->pPushMgr, pIter);
    }

    taosHashClear(pTq->pPushMgr);
  }
wmmhello's avatar
wmmhello 已提交
553
  taosWUnLockLatch(&pTq->lock);
554 555 556
  return 0;
}

D
dapan1121 已提交
557
int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
X
Xiaoyu Wang 已提交
558
  SMqPollReq req = {0};
dengyihao's avatar
dengyihao 已提交
559
  int        code = 0;
D
dapan1121 已提交
560 561 562 563 564 565 566 567 568 569
  if (tDeserializeSMqPollReq(pMsg->pCont, pMsg->contLen, &req) < 0) {
    tqError("tDeserializeSMqPollReq %d failed", pMsg->contLen);
    terrno = TSDB_CODE_INVALID_MSG;
    return -1;
  }

  int64_t      consumerId = req.consumerId;
  int32_t      reqEpoch = req.epoch;
  STqOffsetVal reqOffset = req.reqOffset;
  int32_t      vgId = TD_VID(pTq->pVnode);
wmmhello's avatar
wmmhello 已提交
570
  STqHandle*   pHandle = NULL;
D
dapan1121 已提交
571

wmmhello's avatar
wmmhello 已提交
572 573 574 575 576
  while (1) {
    taosWLockLatch(&pTq->lock);
    // 1. find handle
    pHandle = taosHashGet(pTq->pHandle, req.subKey, strlen(req.subKey));
    if (pHandle == NULL) {
wmmhello's avatar
wmmhello 已提交
577 578 579 580 581 582 583 584 585 586 587 588
      do{
        if (tqMetaGetHandle(pTq, req.subKey) == 0){
          pHandle = taosHashGet(pTq->pHandle, req.subKey, strlen(req.subKey));
          if(pHandle != NULL){
            break;
          }
        }
        tqError("tmq poll: consumer:0x%" PRIx64 " vgId:%d subkey %s not found", consumerId, vgId, req.subKey);
        terrno = TSDB_CODE_INVALID_MSG;
        taosWUnLockLatch(&pTq->lock);
        return -1;
      }while(0);
wmmhello's avatar
wmmhello 已提交
589
    }
D
dapan1121 已提交
590

591 592
    // 2. check re-balance status
    if (pHandle->consumerId != consumerId) {
dengyihao's avatar
dengyihao 已提交
593 594
      tqError("ERROR tmq poll: consumer:0x%" PRIx64
              " vgId:%d, subkey %s, mismatch for saved handle consumer:0x%" PRIx64,
595
              consumerId, TD_VID(pTq->pVnode), req.subKey, pHandle->consumerId);
596
      terrno = TSDB_CODE_TMQ_CONSUMER_MISMATCH;
597 598 599
      taosWUnLockLatch(&pTq->lock);
      return -1;
    }
wmmhello's avatar
wmmhello 已提交
600

wmmhello's avatar
wmmhello 已提交
601
    bool exec = tqIsHandleExec(pHandle);
dengyihao's avatar
dengyihao 已提交
602
    if (!exec) {
wmmhello's avatar
wmmhello 已提交
603
      tqSetHandleExec(pHandle);
dengyihao's avatar
dengyihao 已提交
604
      //      qSetTaskCode(pHandle->execHandle.task, TDB_CODE_SUCCESS);
605
      tqDebug("tmq poll: consumer:0x%" PRIx64 " vgId:%d, topic:%s, set handle exec, pHandle:%p", consumerId, vgId,
dengyihao's avatar
dengyihao 已提交
606
              req.subKey, pHandle);
wmmhello's avatar
wmmhello 已提交
607 608 609
      taosWUnLockLatch(&pTq->lock);
      break;
    }
610
    taosWUnLockLatch(&pTq->lock);
611

dengyihao's avatar
dengyihao 已提交
612 613 614
    tqDebug("tmq poll: consumer:0x%" PRIx64
            "vgId:%d, topic:%s, subscription is executing, wait for 10ms and retry, pHandle:%p",
            consumerId, vgId, req.subKey, pHandle);
wmmhello's avatar
wmmhello 已提交
615
    taosMsleep(10);
D
dapan1121 已提交
616 617 618
  }

  // 3. update the epoch value
619
  if (pHandle->epoch < reqEpoch) {
dengyihao's avatar
dengyihao 已提交
620
    tqDebug("tmq poll: consumer:0x%" PRIx64 " epoch update from %d to %d by poll req", consumerId, pHandle->epoch,
X
Xiaoyu Wang 已提交
621
            reqEpoch);
D
dapan1121 已提交
622 623 624
    pHandle->epoch = reqEpoch;
  }

wmmhello's avatar
wmmhello 已提交
625
  char buf[TSDB_OFFSET_LEN] = {0};
626
  tFormatOffset(buf, TSDB_OFFSET_LEN, &reqOffset);
D
dapan1121 已提交
627 628 629
  tqDebug("tmq poll: consumer:0x%" PRIx64 " (epoch %d), subkey %s, recv poll req vgId:%d, req:%s, reqId:0x%" PRIx64,
          consumerId, req.epoch, pHandle->subKey, vgId, buf, req.reqId);

wmmhello's avatar
wmmhello 已提交
630
  code = tqExtractDataForMq(pTq, pHandle, &req, pMsg);
631
  tqSetHandleIdle(pHandle);
632

633
  tqDebug("tmq poll: consumer:0x%" PRIx64 " vgId:%d, topic:%s, set handle idle, pHandle:%p", consumerId, vgId,
dengyihao's avatar
dengyihao 已提交
634
          req.subKey, pHandle);
635
  return code;
D
dapan1121 已提交
636 637
}

638 639 640 641 642 643 644 645 646 647 648 649 650 651 652 653 654 655 656 657 658 659 660 661 662 663 664
int32_t tqProcessVgCommittedInfoReq(STQ* pTq, SRpcMsg* pMsg) {
  void* data = POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead));
  int32_t len = pMsg->contLen - sizeof(SMsgHead);

  SMqVgOffset vgOffset = {0};

  SDecoder decoder;
  tDecoderInit(&decoder, (uint8_t*)data, len);
  if (tDecodeMqVgOffset(&decoder, &vgOffset) < 0) {
    return TSDB_CODE_OUT_OF_MEMORY;
  }

  tDecoderClear(&decoder);

  STqOffset* pOffset = &vgOffset.offset;
  STqOffset* pSavedOffset = tqOffsetRead(pTq->pOffsetStore, pOffset->subKey);
  if (pSavedOffset == NULL) {
    return TSDB_CODE_TMQ_NO_COMMITTED;
  }
  vgOffset.offset = *pSavedOffset;

  int32_t code = 0;
  tEncodeSize(tEncodeMqVgOffset, &vgOffset, len, code);
  if (code < 0) {
    return TSDB_CODE_INVALID_PARA;
  }

665
  void* buf = rpcMallocCont(len);
666 667 668 669 670 671 672 673 674 675 676 677 678 679 680
  if (buf == NULL) {
    return TSDB_CODE_OUT_OF_MEMORY;
  }
  SEncoder encoder;
  tEncoderInit(&encoder, buf, len);
  tEncodeMqVgOffset(&encoder, &vgOffset);
  tEncoderClear(&encoder);

  SRpcMsg rsp = {.info = pMsg->info, .pCont = buf, .contLen = len, .code = 0};

  tmsgSendRsp(&rsp);

  return 0;
}

681 682 683 684 685 686 687 688 689 690 691 692 693 694 695 696 697 698 699 700 701 702 703 704 705 706 707 708 709 710 711 712 713 714 715
int32_t tqProcessVgWalInfoReq(STQ* pTq, SRpcMsg* pMsg) {
  SMqPollReq req = {0};
  if (tDeserializeSMqPollReq(pMsg->pCont, pMsg->contLen, &req) < 0) {
    tqError("tDeserializeSMqPollReq %d failed", pMsg->contLen);
    terrno = TSDB_CODE_INVALID_MSG;
    return -1;
  }

  int64_t      consumerId = req.consumerId;
  STqOffsetVal reqOffset = req.reqOffset;
  int32_t      vgId = TD_VID(pTq->pVnode);

  // 1. find handle
  STqHandle* pHandle = taosHashGet(pTq->pHandle, req.subKey, strlen(req.subKey));
  if (pHandle == NULL) {
    tqError("consumer:0x%" PRIx64 " vgId:%d subkey:%s not found", consumerId, vgId, req.subKey);
    terrno = TSDB_CODE_INVALID_MSG;
    return -1;
  }

  // 2. check re-balance status
  taosRLockLatch(&pTq->lock);
  if (pHandle->consumerId != consumerId) {
    tqDebug("ERROR consumer:0x%" PRIx64 " vgId:%d, subkey %s, mismatch for saved handle consumer:0x%" PRIx64,
            consumerId, vgId, req.subKey, pHandle->consumerId);
    terrno = TSDB_CODE_TMQ_CONSUMER_MISMATCH;
    taosRUnLockLatch(&pTq->lock);
    return -1;
  }
  taosRUnLockLatch(&pTq->lock);

  int64_t sver = 0, ever = 0;
  walReaderValidVersionRange(pHandle->execHandle.pTqReader->pWalReader, &sver, &ever);

  SMqDataRsp dataRsp = {0};
716
  tqInitDataRsp(&dataRsp, req.reqOffset);
717

wmmhello's avatar
wmmhello 已提交
718 719 720 721 722 723
  if (req.useSnapshot == true) {
    tqError("consumer:0x%" PRIx64 " vgId:%d subkey:%s snapshot not support wal info", consumerId, vgId, req.subKey);
    terrno = TSDB_CODE_INVALID_PARA;
    tDeleteMqDataRsp(&dataRsp);
    return -1;
  }
724

wmmhello's avatar
wmmhello 已提交
725
  dataRsp.rspOffset.type = TMQ_OFFSET__LOG;
726

wmmhello's avatar
wmmhello 已提交
727 728
  if (reqOffset.type == TMQ_OFFSET__LOG) {
    dataRsp.rspOffset.version = reqOffset.version;
729 730 731 732 733 734 735 736 737
  } else if(reqOffset.type < 0){
    STqOffset* pOffset = tqOffsetRead(pTq->pOffsetStore, req.subKey);
    if (pOffset != NULL) {
      if (pOffset->val.type != TMQ_OFFSET__LOG) {
        tqError("consumer:0x%" PRIx64 " vgId:%d subkey:%s, no valid wal info", consumerId, vgId, req.subKey);
        terrno = TSDB_CODE_INVALID_PARA;
        tDeleteMqDataRsp(&dataRsp);
        return -1;
      }
738

739 740 741 742 743 744 745
      dataRsp.rspOffset.version = pOffset->val.version;
      tqInfo("consumer:0x%" PRIx64 " vgId:%d subkey:%s get assignment from store:%"PRId64, consumerId, vgId, req.subKey, dataRsp.rspOffset.version);
    }else{
      if (reqOffset.type == TMQ_OFFSET__RESET_EARLIEST) {
        dataRsp.rspOffset.version = sver;  // not consume yet, set the earliest position
      } else if (reqOffset.type == TMQ_OFFSET__RESET_LATEST) {
        dataRsp.rspOffset.version = ever;
746
      }
747
      tqInfo("consumer:0x%" PRIx64 " vgId:%d subkey:%s get assignment from init:%"PRId64, consumerId, vgId, req.subKey, dataRsp.rspOffset.version);
748
    }
wmmhello's avatar
wmmhello 已提交
749 750 751 752 753 754
  } else {
    tqError("consumer:0x%" PRIx64 " vgId:%d subkey:%s invalid offset type:%d", consumerId, vgId, req.subKey,
            reqOffset.type);
    terrno = TSDB_CODE_INVALID_PARA;
    tDeleteMqDataRsp(&dataRsp);
    return -1;
755 756 757
  }

  tqDoSendDataRsp(&pMsg->info, &dataRsp, req.epoch, req.consumerId, TMQ_MSG_TYPE__WALINFO_RSP, sver, ever);
wmmhello's avatar
wmmhello 已提交
758
  tDeleteMqDataRsp(&dataRsp);
759 760 761
  return 0;
}

762
int32_t tqProcessDeleteSubReq(STQ* pTq, int64_t sversion, char* msg, int32_t msgLen) {
L
Liu Jicong 已提交
763
  SMqVDeleteReq* pReq = (SMqVDeleteReq*)msg;
dengyihao's avatar
dengyihao 已提交
764
  int32_t        vgId = TD_VID(pTq->pVnode);
L
Liu Jicong 已提交
765

766
  tqInfo("vgId:%d, tq process delete sub req %s", vgId, pReq->subKey);
wmmhello's avatar
wmmhello 已提交
767
  int32_t code = 0;
L
Liu Jicong 已提交
768

wmmhello's avatar
wmmhello 已提交
769
  taosWLockLatch(&pTq->lock);
L
Liu Jicong 已提交
770 771
  STqHandle* pHandle = taosHashGet(pTq->pHandle, pReq->subKey, strlen(pReq->subKey));
  if (pHandle) {
wmmhello's avatar
wmmhello 已提交
772
    while (tqIsHandleExec(pHandle)) {
dengyihao's avatar
dengyihao 已提交
773 774
      tqDebug("vgId:%d, topic:%s, subscription is executing, wait for 10ms and retry, pHandle:%p", vgId,
              pHandle->subKey, pHandle);
wmmhello's avatar
wmmhello 已提交
775
      taosMsleep(10);
776
    }
777

L
Liu Jicong 已提交
778 779 780
    if (pHandle->pRef) {
      walCloseRef(pTq->pVnode->pWal, pHandle->pRef->refId);
    }
781

L
Liu Jicong 已提交
782 783 784 785
    code = taosHashRemove(pTq->pHandle, pReq->subKey, strlen(pReq->subKey));
    if (code != 0) {
      tqError("cannot process tq delete req %s, since no such handle", pReq->subKey);
    }
L
Liu Jicong 已提交
786
  }
787

L
Liu Jicong 已提交
788 789
  code = tqOffsetDelete(pTq->pOffsetStore, pReq->subKey);
  if (code != 0) {
790
    tqError("cannot process tq delete req %s, since no such offset in cache", pReq->subKey);
L
Liu Jicong 已提交
791
  }
L
Liu Jicong 已提交
792

L
Liu Jicong 已提交
793
  if (tqMetaDeleteHandle(pTq, pReq->subKey) < 0) {
L
Liu Jicong 已提交
794
    tqError("cannot process tq delete req %s, since no such offset in tdb", pReq->subKey);
795
  }
wmmhello's avatar
wmmhello 已提交
796 797
  taosWUnLockLatch(&pTq->lock);

L
Liu Jicong 已提交
798
  return 0;
L
Liu Jicong 已提交
799 800
}

801
int32_t tqProcessAddCheckInfoReq(STQ* pTq, int64_t sversion, char* msg, int32_t msgLen) {
802 803
  STqCheckInfo info = {0};
  SDecoder     decoder;
X
Xiaoyu Wang 已提交
804
  tDecoderInit(&decoder, (uint8_t*)msg, msgLen);
805
  if (tDecodeSTqCheckInfo(&decoder, &info) < 0) {
L
Liu Jicong 已提交
806 807 808 809
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return -1;
  }
  tDecoderClear(&decoder);
810 811 812 813 814
  if (taosHashPut(pTq->pCheckInfo, info.topic, strlen(info.topic), &info, sizeof(STqCheckInfo)) < 0) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return -1;
  }
  if (tqMetaSaveCheckInfo(pTq, info.topic, msg, msgLen) < 0) {
L
Liu Jicong 已提交
815 816 817 818 819 820
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return -1;
  }
  return 0;
}

821
int32_t tqProcessDelCheckInfoReq(STQ* pTq, int64_t sversion, char* msg, int32_t msgLen) {
822 823 824 825 826 827 828 829 830 831 832
  if (taosHashRemove(pTq->pCheckInfo, msg, strlen(msg)) < 0) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return -1;
  }
  if (tqMetaDeleteCheckInfo(pTq, msg) < 0) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return -1;
  }
  return 0;
}

833
int32_t tqProcessSubscribeReq(STQ* pTq, int64_t sversion, char* msg, int32_t msgLen) {
dengyihao's avatar
dengyihao 已提交
834
  int         ret = 0;
L
Liu Jicong 已提交
835
  SMqRebVgReq req = {0};
dengyihao's avatar
dengyihao 已提交
836
  SDecoder    dc = {0};
837 838 839 840 841 842 843 844 845

  tDecoderInit(&dc, msg, msgLen);

  // decode req
  if (tDecodeSMqRebVgReq(&dc, &req) < 0) {
    terrno = TSDB_CODE_INVALID_MSG;
    tDecoderClear(&dc);
    return -1;
  }
L
Liu Jicong 已提交
846

847
  tqInfo("vgId:%d, tq process sub req:%s, Id:0x%" PRIx64 " -> Id:0x%" PRIx64, pTq->pVnode->config.vgId, req.subKey,
D
dapan1121 已提交
848
          req.oldConsumerId, req.newConsumerId);
L
Liu Jicong 已提交
849

850 851 852 853 854 855 856 857
  STqHandle* pHandle = NULL;
  while(1){
    pHandle = taosHashGet(pTq->pHandle, req.subKey, strlen(req.subKey));
    if (pHandle || tqMetaGetHandle(pTq, req.subKey) < 0){
      break;
    }
  }

L
Liu Jicong 已提交
858
  if (pHandle == NULL) {
L
Liu Jicong 已提交
859
    if (req.oldConsumerId != -1) {
860
      tqError("vgId:%d, build new consumer handle %s for consumer:0x%" PRIx64 ", but old consumerId:0x%" PRIx64,
861
              req.vgId, req.subKey, req.newConsumerId, req.oldConsumerId);
L
Liu Jicong 已提交
862
    }
L
Liu Jicong 已提交
863
    if (req.newConsumerId == -1) {
864
      tqError("vgId:%d, tq invalid re-balance request, new consumerId %" PRId64 "", req.vgId, req.newConsumerId);
865
      goto end;
L
Liu Jicong 已提交
866
    }
867 868 869 870
    STqHandle handle = {0};
    ret = tqCreateHandle(pTq, &req, &handle);
    if(ret < 0){
      tqDestroyTqHandle(&handle);
871
      goto end;
872
    }
873
    ret = tqMetaSaveHandle(pTq, req.subKey, &handle);
L
Liu Jicong 已提交
874
  } else {
875
    taosWLockLatch(&pTq->lock);
wmmhello's avatar
wmmhello 已提交
876

D
dapan1121 已提交
877
    if (pHandle->consumerId == req.newConsumerId) {  // do nothing
wmmhello's avatar
wmmhello 已提交
878
      tqInfo("vgId:%d no switch consumer:0x%" PRIx64 " remains, because redo wal log", req.vgId, req.newConsumerId);
879
    } else {
wmmhello's avatar
wmmhello 已提交
880
      tqInfo("vgId:%d switch consumer from Id:0x%" PRIx64 " to Id:0x%" PRIx64, req.vgId, pHandle->consumerId, req.newConsumerId);
881
      atomic_store_64(&pHandle->consumerId, req.newConsumerId);
wmmhello's avatar
wmmhello 已提交
882 883 884 885 886 887 888 889 890 891 892 893 894 895 896 897
      //    atomic_add_fetch_32(&pHandle->epoch, 1);

      // kill executing task
      //    if(tqIsHandleExec(pHandle)) {
      //      qTaskInfo_t pTaskInfo = pHandle->execHandle.task;
      //      if (pTaskInfo != NULL) {
      //        qKillTask(pTaskInfo, TSDB_CODE_SUCCESS);
      //      }

      //      if (pHandle->execHandle.subType == TOPIC_SUB_TYPE__COLUMN) {
      //        qStreamCloseTsdbReader(pTaskInfo);
      //      }
      //    }
      // remove if it has been register in the push manager, and return one empty block to consumer
      tqUnregisterPushHandle(pTq, pHandle);
      ret = tqMetaSaveHandle(pTq, req.subKey, pHandle);
898
    }
899
    taosWUnLockLatch(&pTq->lock);
L
Liu Jicong 已提交
900
  }
L
Liu Jicong 已提交
901

902
end:
903
  tDecoderClear(&dc);
904
  return ret;
L
Liu Jicong 已提交
905
}
906

dengyihao's avatar
dengyihao 已提交
907
void freePtr(void* ptr) { taosMemoryFree(*(void**)ptr); }
L
liuyao 已提交
908

909
int32_t tqExpandTask(STQ* pTq, SStreamTask* pTask, int64_t ver) {
D
dapan1121 已提交
910
  int32_t vgId = TD_VID(pTq->pVnode);
911

912
  pTask->id.idStr = createStreamTaskIdStr(pTask->id.streamId, pTask->id.taskId);
L
Liu Jicong 已提交
913
  pTask->refCnt = 1;
914
  pTask->status.schedStatus = TASK_SCHED_STATUS__INACTIVE;
dengyihao's avatar
dengyihao 已提交
915
  pTask->inputQueue = streamQueueOpen(512 << 10);
916
  pTask->outputInfo.queue = streamQueueOpen(512 << 10);
L
Liu Jicong 已提交
917

918
  if (pTask->inputQueue == NULL || pTask->outputInfo.queue == NULL) {
H
Haojun Liao 已提交
919
    tqError("s-task:%s failed to prepare the input/output queue, initialize task failed", pTask->id.idStr);
L
Liu Jicong 已提交
920
    return -1;
L
Liu Jicong 已提交
921 922
  }

H
Haojun Liao 已提交
923
  pTask->tsInfo.init = taosGetTimestampMs();
L
Liu Jicong 已提交
924
  pTask->inputStatus = TASK_INPUT_STATUS__NORMAL;
925
  pTask->outputInfo.status = TASK_OUTPUT_STATUS__NORMAL;
926
  pTask->pMsgCb = &pTq->pVnode->msgCb;
927
  pTask->pMeta = pTq->pStreamMeta;
928

929 930
  streamTaskOpenAllUpstreamInput(pTask);

931
  // backup the initial status, and set it to be TASK_STATUS__INIT
932
  pTask->chkInfo.version = ver;
933
  pTask->chkInfo.currentVer = ver;
934

935 936
  pTask->dataRange.range.maxVer = ver;
  pTask->dataRange.range.minVer = ver;
937

938
  if (pTask->info.taskLevel == TASK_LEVEL__SOURCE) {
L
liuyao 已提交
939
    SStreamTask* pSateTask = pTask;
L
liuyao 已提交
940
    SStreamTask task = {0};
L
liuyao 已提交
941
    if (pTask->info.fillHistory) {
L
liuyao 已提交
942 943 944
      task.id = pTask->streamTaskId;
      task.pMeta = pTask->pMeta;
      pSateTask = &task;
L
liuyao 已提交
945
    }
946

L
liuyao 已提交
947
    pTask->pState = streamStateOpen(pTq->pStreamMeta->path, pSateTask, false, -1, -1);
948 949 950 951
    if (pTask->pState == NULL) {
      return -1;
    }

L
liuyao 已提交
952 953 954 955 956
    SReadHandle handle = {.vnode = pTq->pVnode,
                          .initTqReader = 1,
                          .pStateBackend = pTask->pState,
                          .fillHistory = pTask->info.fillHistory,
                          .winRange = pTask->dataRange.window};
957
    initStorageAPI(&handle.api);
958

959
    pTask->exec.pExecutor = qCreateStreamExecTaskInfo(pTask->exec.qmsg, &handle, vgId, pTask->id.taskId);
960
    if (pTask->exec.pExecutor == NULL) {
L
Liu Jicong 已提交
961 962
      return -1;
    }
963

964
    qSetTaskId(pTask->exec.pExecutor, pTask->id.taskId, pTask->id.streamId);
965
  } else if (pTask->info.taskLevel == TASK_LEVEL__AGG) {
L
liuyao 已提交
966
    SStreamTask* pSateTask = pTask;
L
liuyao 已提交
967
    SStreamTask task = {0};
L
liuyao 已提交
968
    if (pTask->info.fillHistory) {
L
liuyao 已提交
969 970 971
      task.id = pTask->streamTaskId;
      task.pMeta = pTask->pMeta;
      pSateTask = &task;
L
liuyao 已提交
972
    }
L
liuyao 已提交
973
    pTask->pState = streamStateOpen(pTq->pStreamMeta->path, pSateTask, false, -1, -1);
974 975 976
    if (pTask->pState == NULL) {
      return -1;
    }
977

978
    int32_t     numOfVgroups = (int32_t)taosArrayGetSize(pTask->pUpstreamEpInfoList);
L
liuyao 已提交
979 980 981 982 983
    SReadHandle handle = {.vnode = NULL,
                          .numOfVgroups = numOfVgroups,
                          .pStateBackend = pTask->pState,
                          .fillHistory = pTask->info.fillHistory,
                          .winRange = pTask->dataRange.window};
984
    initStorageAPI(&handle.api);
985

986
    pTask->exec.pExecutor = qCreateStreamExecTaskInfo(pTask->exec.qmsg, &handle, vgId, pTask->id.taskId);
987
    if (pTask->exec.pExecutor == NULL) {
L
Liu Jicong 已提交
988 989
      return -1;
    }
990
    qSetTaskId(pTask->exec.pExecutor, pTask->id.taskId, pTask->id.streamId);
L
Liu Jicong 已提交
991
  }
L
Liu Jicong 已提交
992 993

  // sink
994
  if (pTask->outputInfo.type == TASK_OUTPUT__SMA) {
L
Liu Jicong 已提交
995
    pTask->smaSink.vnode = pTq->pVnode;
L
Liu Jicong 已提交
996
    pTask->smaSink.smaSink = smaHandleRes;
997
  } else if (pTask->outputInfo.type == TASK_OUTPUT__TABLE) {
L
Liu Jicong 已提交
998
    pTask->tbSink.vnode = pTq->pVnode;
H
Haojun Liao 已提交
999
    pTask->tbSink.tbSinkFunc = tqSinkToTablePipeline;
L
Liu Jicong 已提交
1000

X
Xiaoyu Wang 已提交
1001
    int32_t   ver1 = 1;
5
54liuyao 已提交
1002
    SMetaInfo info = {0};
dengyihao's avatar
dengyihao 已提交
1003
    int32_t   code = metaGetInfo(pTq->pVnode->pMeta, pTask->tbSink.stbUid, &info, NULL);
5
54liuyao 已提交
1004
    if (code == TSDB_CODE_SUCCESS) {
D
dapan1121 已提交
1005
      ver1 = info.skmVer;
5
54liuyao 已提交
1006
    }
L
Liu Jicong 已提交
1007

1008 1009
    SSchemaWrapper* pschemaWrapper = pTask->tbSink.pSchemaWrapper;
    pTask->tbSink.pTSchema = tBuildTSchema(pschemaWrapper->pSchema, pschemaWrapper->nCols, ver1);
1010
    if (pTask->tbSink.pTSchema == NULL) {
D
dapan1121 已提交
1011 1012
      return -1;
    }
L
liuyao 已提交
1013 1014
    pTask->tbSink.pTblInfo = tSimpleHashInit(10240, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT));
    tSimpleHashSetFreeFp(pTask->tbSink.pTblInfo, freePtr);
L
Liu Jicong 已提交
1015
  }
1016

1017
  if (pTask->info.taskLevel == TASK_LEVEL__SOURCE) {
1018
    SWalFilterCond cond = {.deleteMsg = 1};  // delete msg also extract from wal files
1019
    pTask->exec.pWalReader = walOpenReader(pTq->pVnode->pWal, &cond);
1020 1021
  }

1022 1023 1024 1025 1026 1027
  // reset the task status from unfinished transaction
  if (pTask->status.taskStatus == TASK_STATUS__PAUSE) {
    tqWarn("s-task:%s reset task status to be normal, kept in meta status: Paused", pTask->id.idStr);
    pTask->status.taskStatus = TASK_STATUS__NORMAL;
  }

H
Haojun Liao 已提交
1028
  taosThreadMutexInit(&pTask->lock, NULL);
1029
  streamSetupScheduleTrigger(pTask);
1030

1031
  tqInfo("vgId:%d expand stream task, s-task:%s, checkpoint ver:%" PRId64
1032
         " child id:%d, level:%d, fill-history:%d, trigger:%" PRId64 " ms, disable pause",
1033
         vgId, pTask->id.idStr, pTask->chkInfo.version, pTask->info.selfChildId, pTask->info.taskLevel,
1034
         pTask->info.fillHistory, pTask->triggerParam);
1035 1036 1037

  // next valid version will add one
  pTask->chkInfo.version += 1;
L
Liu Jicong 已提交
1038
  return 0;
L
Liu Jicong 已提交
1039
}
L
Liu Jicong 已提交
1040

1041
int32_t tqProcessStreamTaskCheckReq(STQ* pTq, SRpcMsg* pMsg) {
1042 1043 1044 1045
  char*   msgStr = pMsg->pCont;
  char*   msgBody = POINTER_SHIFT(msgStr, sizeof(SMsgHead));
  int32_t msgLen = pMsg->contLen - sizeof(SMsgHead);

1046 1047
  SStreamTaskCheckReq req;
  SDecoder            decoder;
1048

X
Xiaoyu Wang 已提交
1049
  tDecoderInit(&decoder, (uint8_t*)msgBody, msgLen);
1050
  tDecodeStreamTaskCheckReq(&decoder, &req);
1051
  tDecoderClear(&decoder);
1052

1053 1054
  int32_t taskId = req.downstreamTaskId;

1055 1056 1057 1058 1059 1060 1061 1062 1063
  SStreamTaskCheckRsp rsp = {
      .reqId = req.reqId,
      .streamId = req.streamId,
      .childId = req.childId,
      .downstreamNodeId = req.downstreamNodeId,
      .downstreamTaskId = req.downstreamTaskId,
      .upstreamNodeId = req.upstreamNodeId,
      .upstreamTaskId = req.upstreamTaskId,
  };
1064

1065
  SStreamTask* pTask = streamMetaAcquireTask(pTq->pStreamMeta, req.streamId, taskId);
1066
  if (pTask != NULL) {
1067
    rsp.status = streamTaskCheckStatus(pTask);
1068 1069
    streamMetaReleaseTask(pTq->pStreamMeta, pTask);

1070 1071 1072
    const char* pStatus = streamGetTaskStatusStr(pTask->status.taskStatus);
    tqDebug("s-task:%s status:%s, recv task check req(reqId:0x%" PRIx64 ") task:0x%x (vgId:%d), ready:%d",
            pTask->id.idStr, pStatus, rsp.reqId, rsp.upstreamTaskId, rsp.upstreamNodeId, rsp.status);
1073 1074
  } else {
    rsp.status = 0;
H
Haojun Liao 已提交
1075 1076 1077
    tqDebug("tq recv task check(taskId:0x%" PRIx64 "-0x%x not built yet) req(reqId:0x%" PRIx64
            ") from task:0x%x (vgId:%d), rsp status %d",
            req.streamId, taskId, rsp.reqId, rsp.upstreamTaskId, rsp.upstreamNodeId, rsp.status);
1078 1079
  }

1080
  return streamSendCheckRsp(pTq->pStreamMeta, &req, &rsp, &pMsg->info, taskId);
1081 1082
}

1083 1084 1085 1086
int32_t tqProcessStreamTaskCheckRsp(STQ* pTq, int64_t sversion, SRpcMsg* pMsg) {
  char* pReq = POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead));
  int32_t len = pMsg->contLen - sizeof(SMsgHead);

1087 1088 1089 1090
  int32_t             code;
  SStreamTaskCheckRsp rsp;

  SDecoder decoder;
1091
  tDecoderInit(&decoder, (uint8_t*)pReq, len);
1092
  code = tDecodeStreamTaskCheckRsp(&decoder, &rsp);
1093

1094 1095 1096 1097 1098
  if (code < 0) {
    tDecoderClear(&decoder);
    return -1;
  }

1099
  tDecoderClear(&decoder);
1100 1101
  tqDebug("tq task:0x%x (vgId:%d) recv check rsp(reqId:0x%" PRIx64 ") from 0x%x (vgId:%d) status %d",
          rsp.upstreamTaskId, rsp.upstreamNodeId, rsp.reqId, rsp.downstreamTaskId, rsp.downstreamNodeId, rsp.status);
1102

1103
  SStreamTask* pTask = streamMetaAcquireTask(pTq->pStreamMeta, rsp.streamId, rsp.upstreamTaskId);
1104
  if (pTask == NULL) {
1105
    tqError("tq failed to locate the stream task:0x%x (vgId:%d), it may have been destroyed", rsp.upstreamTaskId,
1106
            pTq->pStreamMeta->vgId);
1107
    terrno = TSDB_CODE_STREAM_TASK_NOT_EXIST;
1108 1109 1110
    return -1;
  }

1111
  code = streamProcessCheckRsp(pTask, &rsp);
L
Liu Jicong 已提交
1112 1113
  streamMetaReleaseTask(pTq->pStreamMeta, pTask);
  return code;
1114 1115
}

1116
int32_t tqProcessTaskDeployReq(STQ* pTq, int64_t sversion, char* msg, int32_t msgLen) {
1117 1118 1119
  int32_t code = 0;
  int32_t vgId = TD_VID(pTq->pVnode);

5
54liuyao 已提交
1120 1121 1122
  if (tsDisableStream) {
    return 0;
  }
1123 1124 1125 1126

  // 1.deserialize msg and build task
  SStreamTask* pTask = taosMemoryCalloc(1, sizeof(SStreamTask));
  if (pTask == NULL) {
1127
    terrno = TSDB_CODE_OUT_OF_MEMORY;
dengyihao's avatar
dengyihao 已提交
1128 1129
    tqError("vgId:%d failed to create stream task due to out of memory, alloc size:%d", vgId,
            (int32_t)sizeof(SStreamTask));
1130 1131
    return -1;
  }
1132

1133 1134
  SDecoder decoder;
  tDecoderInit(&decoder, (uint8_t*)msg, msgLen);
1135
  code = tDecodeStreamTask(&decoder, pTask);
1136 1137 1138 1139 1140
  if (code < 0) {
    tDecoderClear(&decoder);
    taosMemoryFree(pTask);
    return -1;
  }
1141

1142 1143
  tDecoderClear(&decoder);

1144 1145
  SStreamMeta* pStreamMeta = pTq->pStreamMeta;

1146
  // 2.save task, use the newest commit version as the initial start version of stream task.
1147 1148
  int32_t taskId = pTask->id.taskId;
  bool    added = false;
1149

1150 1151
  taosWLockLatch(&pStreamMeta->lock);
  code = streamMetaRegisterTask(pStreamMeta, sversion, pTask, &added);
1152
  int32_t numOfTasks = streamMetaGetNumOfTasks(pStreamMeta);
1153
  taosWUnLockLatch(&pStreamMeta->lock);
1154

1155
  if (code < 0) {
1156
    tqError("vgId:%d failed to add s-task:0x%x, total:%d", vgId, pTask->id.taskId, numOfTasks);
H
Haojun Liao 已提交
1157
    tFreeStreamTask(pTask);
1158 1159 1160
    return -1;
  }

1161
  // not added into meta store
1162 1163
  if (added) {
    tqDebug("vgId:%d s-task:0x%x is deployed and add into meta, numOfTasks:%d", vgId, taskId, numOfTasks);
1164
    SStreamTask* p = streamMetaAcquireTask(pStreamMeta, pTask->id.streamId, taskId);
1165 1166 1167 1168 1169
    if (p != NULL) {  // reset the downstreamReady flag.
      streamTaskCheckDownstreamTasks(p);
    }
    streamMetaReleaseTask(pStreamMeta, p);
  } else {
1170 1171
    tqWarn("vgId:%d failed to add s-task:0x%x, already exists in meta store", vgId, taskId);
    tFreeStreamTask(pTask);
1172 1173
  }

1174 1175 1176
  return 0;
}

1177
int32_t tqProcessTaskScanHistory(STQ* pTq, SRpcMsg* pMsg) {
1178
  SStreamScanHistoryReq* pReq = (SStreamScanHistoryReq*)pMsg->pCont;
1179
  SStreamMeta*           pMeta = pTq->pStreamMeta;
1180

1181
  int32_t      code = TSDB_CODE_SUCCESS;
1182
  SStreamTask* pTask = streamMetaAcquireTask(pMeta, pReq->streamId, pReq->taskId);
1183
  if (pTask == NULL) {
1184 1185
    tqError("vgId:%d failed to acquire stream task:0x%x during stream recover, task may have been destroyed",
            pMeta->vgId, pReq->taskId);
1186 1187 1188
    return -1;
  }

1189
  // do recovery step1
1190
  const char* id = pTask->id.idStr;
1191
  const char* pStatus = streamGetTaskStatusStr(pTask->status.taskStatus);
1192
  tqDebug("s-task:%s start scan-history stage(step 1), status:%s", id, pStatus);
1193

1194
  if (pTask->tsInfo.step1Start == 0) {
1195
    ASSERT(pTask->status.pauseAllowed == false);
1196
    pTask->tsInfo.step1Start = taosGetTimestampMs();
1197 1198 1199 1200 1201
    if (pTask->info.fillHistory == 1) {
      streamTaskEnablePause(pTask);
    }
  } else {
    tqDebug("s-task:%s resume from paused, start ts:%"PRId64, pTask->id.idStr, pTask->tsInfo.step1Start);
1202
  }
1203

1204
  // we have to continue retrying to successfully execute the scan history task.
1205 1206 1207 1208 1209 1210 1211 1212
  int8_t schedStatus = atomic_val_compare_exchange_8(&pTask->status.schedStatus, TASK_SCHED_STATUS__INACTIVE,
                                                     TASK_SCHED_STATUS__WAITING);
  if (schedStatus != TASK_SCHED_STATUS__INACTIVE) {
    tqError(
        "s-task:%s failed to start scan-history in first stream time window since already started, unexpected "
        "sched-status:%d",
        id, schedStatus);
    return 0;
1213
  }
1214

1215
  if (pTask->info.fillHistory == 1) {
1216
    ASSERT(pTask->status.pauseAllowed == true);
1217 1218
  }

1219 1220
  streamSourceScanHistoryData(pTask);
  if (pTask->status.taskStatus == TASK_STATUS__PAUSE) {
1221
    double el = (taosGetTimestampMs() - pTask->tsInfo.step1Start) / 1000.0;
1222 1223
    tqDebug("s-task:%s is paused in the step1, elapsed time:%.2fs, sched-status:%d", pTask->id.idStr, el,
            TASK_SCHED_STATUS__INACTIVE);
L
liuyao 已提交
1224
    atomic_store_8(&pTask->status.schedStatus, TASK_SCHED_STATUS__INACTIVE);
L
Liu Jicong 已提交
1225 1226 1227
    return 0;
  }

1228 1229
  // the following procedure should be executed, no matter status is stop/pause or not
  double el = (taosGetTimestampMs() - pTask->tsInfo.step1Start) / 1000.0;
H
Haojun Liao 已提交
1230
  tqDebug("s-task:%s scan-history stage(step 1) ended, elapsed time:%.2fs", id, el);
H
Haojun Liao 已提交
1231

H
Haojun Liao 已提交
1232
  if (pTask->info.fillHistory) {
L
liuyao 已提交
1233
    SVersionRange* pRange = NULL;
1234
    SStreamTask*   pStreamTask = NULL;
1235
    bool           done = false;
1236

1237
    // 1. get the related stream task
1238
    pStreamTask = streamMetaAcquireTask(pMeta, pTask->streamTaskId.streamId, pTask->streamTaskId.taskId);
1239 1240 1241 1242
    if (pStreamTask == NULL) {
      // todo delete this task, if the related stream task is dropped
      qError("failed to find s-task:0x%x, it may have been destroyed, drop fill-history task:%s",
             pTask->streamTaskId.taskId, pTask->id.idStr);
1243

1244
      tqDebug("s-task:%s fill-history task set status to be dropping", id);
1245

H
Haojun Liao 已提交
1246
      streamMetaUnregisterTask(pMeta, pTask->id.streamId, pTask->id.taskId);
1247 1248 1249
      streamMetaReleaseTask(pMeta, pTask);
      return -1;
    }
L
Liu Jicong 已提交
1250

1251
    ASSERT(pStreamTask->info.taskLevel == TASK_LEVEL__SOURCE);
L
Liu Jicong 已提交
1252

1253 1254
    // 2. it cannot be paused, when the stream task in TASK_STATUS__SCAN_HISTORY status. Let's wait for the
    // stream task get ready for scan history data
1255 1256 1257 1258 1259 1260
    while (pStreamTask->status.taskStatus == TASK_STATUS__SCAN_HISTORY) {
      tqDebug(
          "s-task:%s level:%d related stream task:%s(status:%s) not ready for halt, wait for it and recheck in 100ms",
          id, pTask->info.taskLevel, pStreamTask->id.idStr, streamGetTaskStatusStr(pStreamTask->status.taskStatus));
      taosMsleep(100);
    }
1261

1262 1263
    // now we can stop the stream task execution
    streamTaskHalt(pStreamTask);
1264

1265 1266
    tqDebug("s-task:%s level:%d sched-status:%d is halt by fill-history task:%s", pStreamTask->id.idStr,
            pStreamTask->info.taskLevel, pStreamTask->status.schedStatus, id);
H
Haojun Liao 已提交
1267

1268 1269 1270 1271
    // if it's an source task, extract the last version in wal.
    pRange = &pTask->dataRange.range;
    int64_t latestVer = walReaderGetCurrentVer(pStreamTask->exec.pWalReader);
    done = streamHistoryTaskSetVerRangeStep2(pTask, latestVer);
1272

1273
    if (done) {
H
Haojun Liao 已提交
1274
      pTask->tsInfo.step2Start = taosGetTimestampMs();
1275 1276
      qDebug("s-task:%s scan-history from WAL stage(step 2) ended, elapsed time:%.2fs", id, 0.0);
      appendTranstateIntoInputQ(pTask);
1277
      streamTryExec(pTask);  // exec directly
1278
    } else {
1279 1280 1281 1282 1283 1284
      STimeWindow* pWindow = &pTask->dataRange.window;
      tqDebug("s-task:%s level:%d verRange:%" PRId64 " - %" PRId64 " window:%" PRId64 "-%" PRId64
              ", do secondary scan-history from WAL after halt the related stream task:%s",
              id, pTask->info.taskLevel, pRange->minVer, pRange->maxVer, pWindow->skey, pWindow->ekey,
              pStreamTask->id.idStr);
      ASSERT(pTask->status.schedStatus == TASK_SCHED_STATUS__WAITING);
1285

1286 1287
      pTask->tsInfo.step2Start = taosGetTimestampMs();
      streamSetParamForStreamScannerStep2(pTask, pRange, pWindow);
H
Haojun Liao 已提交
1288

1289
      int64_t dstVer = pTask->dataRange.range.minVer - 1;
H
Haojun Liao 已提交
1290

1291 1292 1293 1294
      pTask->chkInfo.currentVer = dstVer;
      walReaderSetSkipToVersion(pTask->exec.pWalReader, dstVer);
      tqDebug("s-task:%s wal reader start scan WAL verRange:%" PRId64 "-%" PRId64 ", set sched-status:%d", id, dstVer,
              pTask->dataRange.range.maxVer, TASK_SCHED_STATUS__INACTIVE);
H
Haojun Liao 已提交
1295

1296 1297
      atomic_store_8(&pTask->status.schedStatus, TASK_SCHED_STATUS__INACTIVE);

1298 1299 1300 1301 1302
      // set the fill-history task to be normal
      if (pTask->info.fillHistory == 1) {
        streamSetStatusNormal(pTask);
      }

1303 1304
      tqStartStreamTasks(pTq);
    }
1305 1306 1307

    streamMetaReleaseTask(pMeta, pTask);
    streamMetaReleaseTask(pMeta, pStreamTask);
1308 1309 1310 1311
  } else {
    // todo update the chkInfo version for current task.
    // this task has an associated history stream task, so we need to scan wal from the end version of
    // history scan. The current version of chkInfo.current is not updated during the history scan
1312 1313
    STimeWindow* pWindow = &pTask->dataRange.window;

1314
    if (pTask->historyTaskId.taskId == 0) {
1315
      *pWindow = (STimeWindow){INT64_MIN, INT64_MAX};
1316
      tqDebug(
1317
          "s-task:%s scan-history in stream time window completed, no related fill-history task, reset the time "
1318 1319
          "window:%" PRId64 " - %" PRId64,
          id, pWindow->skey, pWindow->ekey);
1320
      qStreamInfoResetTimewindowFilter(pTask->exec.pExecutor);
1321
    } else {
H
Haojun Liao 已提交
1322 1323
      // when related fill-history task exists, update the fill-history time window only when the
      // state transfer is completed.
1324
      tqDebug(
1325
          "s-task:%s scan-history in stream time window completed, now start to handle data from WAL, start "
1326
          "ver:%" PRId64 ", window:%" PRId64 " - %" PRId64,
1327
          id, pTask->chkInfo.currentVer, pWindow->skey, pWindow->ekey);
1328
    }
1329

1330
    // notify the downstream agg tasks that upstream tasks are ready to processing the WAL data, update the
1331 1332
    code = streamTaskScanHistoryDataComplete(pTask);
    streamMetaReleaseTask(pMeta, pTask);
1333

1334 1335
    // when all source task complete to scan history data in stream time window, they are allowed to handle stream data
    // at the same time.
1336 1337
    return code;
  }
1338 1339 1340 1341

  return 0;
}

1342
int32_t tqProcessTaskScanHistoryFinishReq(STQ* pTq, SRpcMsg* pMsg) {
L
Liu Jicong 已提交
1343 1344
  char*   msg = POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead));
  int32_t msgLen = pMsg->contLen - sizeof(SMsgHead);
1345 1346

  // deserialize
1347
  SStreamScanHistoryFinishReq req = {0};
1348 1349

  SDecoder decoder;
X
Xiaoyu Wang 已提交
1350
  tDecoderInit(&decoder, (uint8_t*)msg, msgLen);
1351
  tDecodeStreamScanHistoryFinishReq(&decoder, &req);
1352 1353
  tDecoderClear(&decoder);

1354
  SStreamTask* pTask = streamMetaAcquireTask(pTq->pStreamMeta, req.streamId, req.downstreamTaskId);
1355
  if (pTask == NULL) {
1356 1357
    tqError("vgId:%d process scan history finish msg, failed to find task:0x%x, it may be destroyed",
            pTq->pStreamMeta->vgId, req.downstreamTaskId);
1358 1359 1360
    return -1;
  }

1361 1362 1363
  tqDebug("s-task:%s receive scan-history finish msg from task:0x%x", pTask->id.idStr, req.upstreamTaskId);

  int32_t code = streamProcessScanHistoryFinishReq(pTask, &req, &pMsg->info);
L
Liu Jicong 已提交
1364
  streamMetaReleaseTask(pTq->pStreamMeta, pTask);
1365
  return code;
L
Liu Jicong 已提交
1366
}
L
Liu Jicong 已提交
1367

1368 1369 1370 1371 1372 1373 1374 1375 1376 1377 1378 1379
int32_t tqProcessTaskScanHistoryFinishRsp(STQ* pTq, SRpcMsg* pMsg) {
  char*   msg = POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead));
  int32_t msgLen = pMsg->contLen - sizeof(SMsgHead);

  // deserialize
  SStreamCompleteHistoryMsg req = {0};

  SDecoder decoder;
  tDecoderInit(&decoder, (uint8_t*)msg, msgLen);
  tDecodeCompleteHistoryDataMsg(&decoder, &req);
  tDecoderClear(&decoder);

1380
  SStreamTask* pTask = streamMetaAcquireTask(pTq->pStreamMeta, req.streamId, req.upstreamTaskId);
1381 1382 1383 1384 1385 1386 1387 1388
  if (pTask == NULL) {
    tqError("vgId:%d process scan history finish rsp, failed to find task:0x%x, it may be destroyed",
            pTq->pStreamMeta->vgId, req.upstreamTaskId);
    return -1;
  }

  int32_t remain = atomic_sub_fetch_32(&pTask->notReadyTasks, 1);
  if (remain > 0) {
1389 1390
    tqDebug("s-task:%s scan-history finish rsp received from downstream task:0x%x, remain:%d not send finish rsp",
            pTask->id.idStr, req.downstreamId, remain);
1391
  } else {
1392 1393
    tqDebug(
        "s-task:%s scan-history finish rsp received from downstream task:0x%x, all downstream tasks rsp scan-history "
H
Haojun Liao 已提交
1394
        "completed msg", pTask->id.idStr, req.downstreamId);
1395 1396 1397 1398
    streamProcessScanHistoryFinishRsp(pTask);
  }

  streamMetaReleaseTask(pTq->pStreamMeta, pTask);
L
Liu Jicong 已提交
1399 1400 1401
  return 0;
}

1402 1403 1404 1405
int32_t extractDelDataBlock(const void* pData, int32_t len, int64_t ver, SStreamRefDataBlock** pRefBlock) {
  SDecoder*   pCoder = &(SDecoder){0};
  SDeleteRes* pRes = &(SDeleteRes){0};

H
Haojun Liao 已提交
1406
  (*pRefBlock) = NULL;
H
Haojun Liao 已提交
1407

1408 1409 1410 1411 1412 1413 1414 1415 1416 1417 1418 1419 1420 1421 1422 1423 1424 1425 1426 1427 1428 1429 1430 1431 1432 1433 1434 1435 1436 1437 1438 1439 1440 1441 1442 1443 1444 1445
  pRes->uidList = taosArrayInit(0, sizeof(tb_uid_t));
  if (pRes->uidList == NULL) {
    return TSDB_CODE_OUT_OF_MEMORY;
  }

  tDecoderInit(pCoder, (uint8_t*)pData, len);
  tDecodeDeleteRes(pCoder, pRes);
  tDecoderClear(pCoder);

  int32_t numOfTables = taosArrayGetSize(pRes->uidList);
  if (numOfTables == 0 || pRes->affectedRows == 0) {
    taosArrayDestroy(pRes->uidList);
    return TSDB_CODE_SUCCESS;
  }

  SSDataBlock* pDelBlock = createSpecialDataBlock(STREAM_DELETE_DATA);
  blockDataEnsureCapacity(pDelBlock, numOfTables);
  pDelBlock->info.rows = numOfTables;
  pDelBlock->info.version = ver;

  for (int32_t i = 0; i < numOfTables; i++) {
    // start key column
    SColumnInfoData* pStartCol = taosArrayGet(pDelBlock->pDataBlock, START_TS_COLUMN_INDEX);
    colDataSetVal(pStartCol, i, (const char*)&pRes->skey, false);  // end key column
    SColumnInfoData* pEndCol = taosArrayGet(pDelBlock->pDataBlock, END_TS_COLUMN_INDEX);
    colDataSetVal(pEndCol, i, (const char*)&pRes->ekey, false);
    // uid column
    SColumnInfoData* pUidCol = taosArrayGet(pDelBlock->pDataBlock, UID_COLUMN_INDEX);
    int64_t*         pUid = taosArrayGet(pRes->uidList, i);
    colDataSetVal(pUidCol, i, (const char*)pUid, false);

    colDataSetNULL(taosArrayGet(pDelBlock->pDataBlock, GROUPID_COLUMN_INDEX), i);
    colDataSetNULL(taosArrayGet(pDelBlock->pDataBlock, CALCULATE_START_TS_COLUMN_INDEX), i);
    colDataSetNULL(taosArrayGet(pDelBlock->pDataBlock, CALCULATE_END_TS_COLUMN_INDEX), i);
  }

  taosArrayDestroy(pRes->uidList);
  *pRefBlock = taosAllocateQitem(sizeof(SStreamRefDataBlock), DEF_QITEM, 0);
H
Haojun Liao 已提交
1446
  if ((*pRefBlock) == NULL) {
1447 1448 1449 1450 1451 1452 1453 1454
    return TSDB_CODE_OUT_OF_MEMORY;
  }

  (*pRefBlock)->type = STREAM_INPUT__REF_DATA_BLOCK;
  (*pRefBlock)->pBlock = pDelBlock;
  return TSDB_CODE_SUCCESS;
}

L
Liu Jicong 已提交
1455 1456
int32_t tqProcessTaskRunReq(STQ* pTq, SRpcMsg* pMsg) {
  SStreamTaskRunReq* pReq = pMsg->pCont;
1457 1458 1459 1460

  int32_t taskId = pReq->taskId;
  int32_t vgId = TD_VID(pTq->pVnode);

1461 1462 1463 1464 1465 1466
  if (taskId == STREAM_TASK_STATUS_CHECK_ID) {
    tqStreamTasksStatusCheck(pTq);
    return 0;
  }

  if (taskId == EXTRACT_DATA_FROM_WAL_ID) {  // all tasks are extracted submit data from the wal
1467
    tqStreamTasksScanWal(pTq);
L
Liu Jicong 已提交
1468
    return 0;
1469
  }
1470

1471
  SStreamTask* pTask = streamMetaAcquireTask(pTq->pStreamMeta, pReq->streamId, taskId);
1472
  if (pTask != NULL) {
1473
    // even in halt status, the data in inputQ must be processed
H
Haojun Liao 已提交
1474
    int8_t st = pTask->status.taskStatus;
1475
    if (st == TASK_STATUS__NORMAL || st == TASK_STATUS__SCAN_HISTORY) {
1476
      tqDebug("vgId:%d s-task:%s start to process block from inputQ, last chk point:%" PRId64, vgId, pTask->id.idStr,
dengyihao's avatar
dengyihao 已提交
1477
              pTask->chkInfo.version);
1478
      streamProcessRunReq(pTask);
1479
    } else {
1480
      atomic_store_8(&pTask->status.schedStatus, TASK_SCHED_STATUS__INACTIVE);
1481 1482
      tqDebug("vgId:%d s-task:%s ignore run req since not in ready state, status:%s, sched-status:%d", vgId,
              pTask->id.idStr, streamGetTaskStatusStr(pTask->status.taskStatus), pTask->status.schedStatus);
1483
    }
1484

L
Liu Jicong 已提交
1485
    streamMetaReleaseTask(pTq->pStreamMeta, pTask);
1486
    tqStartStreamTasks(pTq);
L
Liu Jicong 已提交
1487
    return 0;
1488 1489 1490
  } else { // NOTE: pTask->status.schedStatus is not updated since it is not be handled by the run exec.
    // todo add one function to handle this
    tqError("vgId:%d failed to found s-task, taskId:0x%x may have been dropped", vgId, taskId);
1491
    return -1;
L
Liu Jicong 已提交
1492
  }
L
Liu Jicong 已提交
1493 1494
}

L
Liu Jicong 已提交
1495
int32_t tqProcessTaskDispatchReq(STQ* pTq, SRpcMsg* pMsg, bool exec) {
dengyihao's avatar
dengyihao 已提交
1496 1497 1498
  char*   msgStr = pMsg->pCont;
  char*   msgBody = POINTER_SHIFT(msgStr, sizeof(SMsgHead));
  int32_t msgLen = pMsg->contLen - sizeof(SMsgHead);
1499 1500 1501 1502

  SStreamDispatchReq req = {0};

  SDecoder decoder;
L
Liu Jicong 已提交
1503
  tDecoderInit(&decoder, (uint8_t*)msgBody, msgLen);
1504
  tDecodeStreamDispatchReq(&decoder, &req);
L
Liu Jicong 已提交
1505

1506
  SStreamTask* pTask = streamMetaAcquireTask(pTq->pStreamMeta, req.streamId, req.taskId);
L
Liu Jicong 已提交
1507
  if (pTask) {
1508
    SRpcMsg rsp = {.info = pMsg->info, .code = 0};
1509
    streamProcessDispatchMsg(pTask, &req, &rsp, exec);
L
Liu Jicong 已提交
1510
    streamMetaReleaseTask(pTq->pStreamMeta, pTask);
L
Liu Jicong 已提交
1511
    return 0;
1512
  } else {
L
liuyao 已提交
1513
    tDeleteStreamDispatchReq(&req);
1514
    return -1;
L
Liu Jicong 已提交
1515
  }
L
Liu Jicong 已提交
1516 1517
}

L
Liu Jicong 已提交
1518 1519
int32_t tqProcessTaskDispatchRsp(STQ* pTq, SRpcMsg* pMsg) {
  SStreamDispatchRsp* pRsp = POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead));
1520

1521 1522 1523 1524 1525
  int32_t      vgId = pTq->pStreamMeta->vgId;
  int32_t      taskId = htonl(pRsp->upstreamTaskId);
  int64_t      streamId = htobe64(pRsp->streamId);
  SStreamTask* pTask = streamMetaAcquireTask(pTq->pStreamMeta, streamId, taskId);

L
Liu Jicong 已提交
1526
  if (pTask) {
1527
    streamProcessDispatchRsp(pTask, pRsp, pMsg->code);
L
Liu Jicong 已提交
1528
    streamMetaReleaseTask(pTq->pStreamMeta, pTask);
1529
    return TSDB_CODE_SUCCESS;
1530
  } else {
1531
    tqDebug("vgId:%d failed to handle the dispatch rsp, since find task:0x%x failed", vgId, taskId);
1532
    return TSDB_CODE_INVALID_MSG;
L
Liu Jicong 已提交
1533
  }
L
Liu Jicong 已提交
1534
}
L
Liu Jicong 已提交
1535

1536
int32_t tqProcessTaskDropReq(STQ* pTq, int64_t sversion, char* msg, int32_t msgLen) {
L
Liu Jicong 已提交
1537
  SVDropStreamTaskReq* pReq = (SVDropStreamTaskReq*)msg;
1538
  tqDebug("vgId:%d receive msg to drop stream task:0x%x", TD_VID(pTq->pVnode), pReq->taskId);
1539
  SStreamTask* pTask = streamMetaAcquireTask(pTq->pStreamMeta, pReq->streamId, pReq->taskId);
1540 1541 1542 1543
  if (pTask == NULL) {
    tqError("vgId:%d failed to acquire s-task:0x%x when dropping it", pTq->pStreamMeta->vgId, pReq->taskId);
    return 0;
  }
1544

H
Haojun Liao 已提交
1545
  streamMetaUnregisterTask(pTq->pStreamMeta, pReq->streamId, pReq->taskId);
1546
  streamMetaReleaseTask(pTq->pStreamMeta, pTask);
L
Liu Jicong 已提交
1547
  return 0;
L
Liu Jicong 已提交
1548
}
L
Liu Jicong 已提交
1549

5
54liuyao 已提交
1550 1551
int32_t tqProcessTaskPauseReq(STQ* pTq, int64_t sversion, char* msg, int32_t msgLen) {
  SVPauseStreamTaskReq* pReq = (SVPauseStreamTaskReq*)msg;
1552 1553

  SStreamMeta* pMeta = pTq->pStreamMeta;
1554
  SStreamTask* pTask = streamMetaAcquireTask(pMeta, pReq->streamId, pReq->taskId);
1555
  if (pTask == NULL) {
1556
    tqError("vgId:%d process pause req, failed to acquire task:0x%x, it may have been dropped already", pMeta->vgId,
1557
            pReq->taskId);
1558 1559
    // since task is in [STOP|DROPPING] state, it is safe to assume the pause is active
    return TSDB_CODE_SUCCESS;
1560 1561 1562
  }

  tqDebug("s-task:%s receive pause msg from mnode", pTask->id.idStr);
1563
  streamTaskPause(pTask);
1564 1565 1566

  SStreamTask* pHistoryTask = NULL;
  if (pTask->historyTaskId.taskId != 0) {
1567
    pHistoryTask = streamMetaAcquireTask(pMeta, pTask->historyTaskId.streamId, pTask->historyTaskId.taskId);
1568
    if (pHistoryTask == NULL) {
1569
      tqError("vgId:%d process pause req, failed to acquire fill-history task:0x%x, it may have been dropped already",
1570
              pMeta->vgId, pTask->historyTaskId.taskId);
1571
      streamMetaReleaseTask(pMeta, pTask);
1572 1573 1574

      // since task is in [STOP|DROPPING] state, it is safe to assume the pause is active
      return TSDB_CODE_SUCCESS;
1575 1576
    }

1577
    tqDebug("s-task:%s fill-history task handle paused along with related stream task", pHistoryTask->id.idStr);
1578

1579
    streamTaskPause(pHistoryTask);
1580
    streamMetaReleaseTask(pMeta, pHistoryTask);
L
liuyao 已提交
1581
  }
1582

1583
  streamMetaReleaseTask(pMeta, pTask);
1584
  return TSDB_CODE_SUCCESS;
L
liuyao 已提交
1585 1586 1587 1588
}

int32_t tqProcessTaskResumeImpl(STQ* pTq, SStreamTask* pTask, int64_t sversion, int8_t igUntreated) {
  int32_t vgId = pTq->pStreamMeta->vgId;
1589 1590 1591
  if (pTask == NULL) {
    return -1;
  }
L
liuyao 已提交
1592

1593 1594
  // todo: handle the case: resume from halt to pause/ from halt to normal/ from pause to normal
  streamTaskResume(pTask);
1595

1596 1597 1598
  int32_t level = pTask->info.taskLevel;
  int8_t  status = pTask->status.taskStatus;
  if (status == TASK_STATUS__NORMAL || status == TASK_STATUS__SCAN_HISTORY) {
1599
    // no lock needs to secure the access of the version
1600
    if (igUntreated && level == TASK_LEVEL__SOURCE && !pTask->info.fillHistory) {
1601 1602 1603 1604 1605 1606 1607 1608 1609 1610
      // discard all the data  when the stream task is suspended.
      walReaderSetSkipToVersion(pTask->exec.pWalReader, sversion);
      tqDebug("vgId:%d s-task:%s resume to exec, prev paused version:%" PRId64 ", start from vnode ver:%" PRId64
              ", schedStatus:%d",
              vgId, pTask->id.idStr, pTask->chkInfo.currentVer, sversion, pTask->status.schedStatus);
    } else {  // from the previous paused version and go on
      tqDebug("vgId:%d s-task:%s resume to exec, from paused ver:%" PRId64 ", vnode ver:%" PRId64 ", schedStatus:%d",
              vgId, pTask->id.idStr, pTask->chkInfo.currentVer, sversion, pTask->status.schedStatus);
    }

1611
    if (level == TASK_LEVEL__SOURCE && pTask->info.fillHistory && pTask->status.taskStatus == TASK_STATUS__SCAN_HISTORY) {
1612
      streamStartScanHistoryAsync(pTask, igUntreated);
1613
    } else if (level == TASK_LEVEL__SOURCE && (taosQueueItemSize(pTask->inputQueue->queue) == 0)) {
1614 1615 1616
      tqStartStreamTasks(pTq);
    } else {
      streamSchedExec(pTask);
L
liuyao 已提交
1617
    }
L
liuyao 已提交
1618
  }
1619 1620

  streamMetaReleaseTask(pTq->pStreamMeta, pTask);
5
54liuyao 已提交
1621 1622 1623 1624 1625
  return 0;
}

int32_t tqProcessTaskResumeReq(STQ* pTq, int64_t sversion, char* msg, int32_t msgLen) {
  SVResumeStreamTaskReq* pReq = (SVResumeStreamTaskReq*)msg;
1626
  SStreamTask* pTask = streamMetaAcquireTask(pTq->pStreamMeta, pReq->streamId, pReq->taskId);
L
liuyao 已提交
1627 1628 1629
  int32_t code = tqProcessTaskResumeImpl(pTq, pTask, sversion, pReq->igUntreated);
  if (code != 0) {
    return code;
L
liuyao 已提交
1630
  }
1631

1632
  SStreamTask* pHistoryTask = streamMetaAcquireTask(pTq->pStreamMeta, pTask->historyTaskId.streamId, pTask->historyTaskId.taskId);
L
liuyao 已提交
1633 1634 1635
  if (pHistoryTask) {
    code = tqProcessTaskResumeImpl(pTq, pHistoryTask, sversion, pReq->igUntreated);
  }
1636

L
liuyao 已提交
1637
  return code;
5
54liuyao 已提交
1638 1639
}

L
Liu Jicong 已提交
1640
int32_t tqProcessTaskRetrieveReq(STQ* pTq, SRpcMsg* pMsg) {
1641 1642 1643 1644 1645
  char*    msgStr = pMsg->pCont;
  char*    msgBody = POINTER_SHIFT(msgStr, sizeof(SMsgHead));
  int32_t  msgLen = pMsg->contLen - sizeof(SMsgHead);
  SDecoder decoder;

L
Liu Jicong 已提交
1646
  SStreamRetrieveReq req;
1647
  tDecoderInit(&decoder, (uint8_t*)msgBody, msgLen);
L
Liu Jicong 已提交
1648
  tDecodeStreamRetrieveReq(&decoder, &req);
L
Liu Jicong 已提交
1649
  tDecoderClear(&decoder);
1650

1651
  SStreamTask* pTask = streamMetaAcquireTask(pTq->pStreamMeta, req.streamId, req.dstTaskId);
1652

L
Liu Jicong 已提交
1653
  if (pTask) {
1654
    SRpcMsg rsp = {.info = pMsg->info, .code = 0};
L
Liu Jicong 已提交
1655
    streamProcessRetrieveReq(pTask, &req, &rsp);
1656

L
Liu Jicong 已提交
1657
    streamMetaReleaseTask(pTq->pStreamMeta, pTask);
L
Liu Jicong 已提交
1658
    tDeleteStreamRetrieveReq(&req);
L
Liu Jicong 已提交
1659
    return 0;
L
Liu Jicong 已提交
1660
  } else {
L
liuyao 已提交
1661
    tDeleteStreamRetrieveReq(&req);
L
Liu Jicong 已提交
1662
    return -1;
L
Liu Jicong 已提交
1663 1664 1665 1666 1667 1668 1669
  }
}

int32_t tqProcessTaskRetrieveRsp(STQ* pTq, SRpcMsg* pMsg) {
  //
  return 0;
}
L
Liu Jicong 已提交
1670

1671 1672
int32_t vnodeEnqueueStreamMsg(SVnode* pVnode, SRpcMsg* pMsg) {
  STQ*      pTq = pVnode->pTq;
1673 1674
  int32_t vgId = pVnode->config.vgId;

1675 1676 1677 1678
  SMsgHead* msgStr = pMsg->pCont;
  char*     msgBody = POINTER_SHIFT(msgStr, sizeof(SMsgHead));
  int32_t   msgLen = pMsg->contLen - sizeof(SMsgHead);
  int32_t   code = 0;
L
Liu Jicong 已提交
1679 1680 1681

  SStreamDispatchReq req;
  SDecoder           decoder;
1682
  tDecoderInit(&decoder, (uint8_t*)msgBody, msgLen);
L
Liu Jicong 已提交
1683 1684
  if (tDecodeStreamDispatchReq(&decoder, &req) < 0) {
    code = TSDB_CODE_MSG_DECODE_ERROR;
L
Liu Jicong 已提交
1685
    tDecoderClear(&decoder);
L
Liu Jicong 已提交
1686 1687
    goto FAIL;
  }
L
Liu Jicong 已提交
1688
  tDecoderClear(&decoder);
L
Liu Jicong 已提交
1689

L
Liu Jicong 已提交
1690
  int32_t taskId = req.taskId;
1691 1692 1693
  tqDebug("vgId:%d receive dispatch msg to s-task:0x%"PRIx64"-0x%x", vgId, req.streamId, taskId);

  SStreamTask* pTask = streamMetaAcquireTask(pTq->pStreamMeta, req.streamId, taskId);
1694
  if (pTask != NULL) {
1695
    SRpcMsg rsp = {.info = pMsg->info, .code = 0};
1696
    streamProcessDispatchMsg(pTask, &req, &rsp, false);
L
Liu Jicong 已提交
1697
    streamMetaReleaseTask(pTq->pStreamMeta, pTask);
L
Liu Jicong 已提交
1698 1699
    rpcFreeCont(pMsg->pCont);
    taosFreeQitem(pMsg);
1700
    return 0;
5
54liuyao 已提交
1701
  } else {
1702

5
54liuyao 已提交
1703
    tDeleteStreamDispatchReq(&req);
L
Liu Jicong 已提交
1704
  }
L
Liu Jicong 已提交
1705

1706 1707
  code = TSDB_CODE_STREAM_TASK_NOT_EXIST;

L
Liu Jicong 已提交
1708
FAIL:
1709
  if (pMsg->info.handle == NULL) {
1710
    tqError("s-task:0x%x vgId:%d msg handle is null, abort enqueue dispatch msg", vgId, taskId);
1711 1712
    return -1;
  }
1713 1714 1715

  SMsgHead* pRspHead = rpcMallocCont(sizeof(SMsgHead) + sizeof(SStreamDispatchRsp));
  if (pRspHead == NULL) {
1716
    SRpcMsg rsp = {.code = TSDB_CODE_OUT_OF_MEMORY, .info = pMsg->info};
1717
    tqError("s-task:0x%x send dispatch error rsp, code:%s", taskId, tstrerror(code));
1718 1719 1720 1721 1722 1723 1724 1725 1726 1727 1728 1729 1730 1731 1732
    tmsgSendRsp(&rsp);
    rpcFreeCont(pMsg->pCont);
    taosFreeQitem(pMsg);
    return -1;
  }

  pRspHead->vgId = htonl(req.upstreamNodeId);
  SStreamDispatchRsp* pRsp = POINTER_SHIFT(pRspHead, sizeof(SMsgHead));
  pRsp->streamId = htobe64(req.streamId);
  pRsp->upstreamTaskId = htonl(req.upstreamTaskId);
  pRsp->upstreamNodeId = htonl(req.upstreamNodeId);
  pRsp->downstreamNodeId = htonl(pVnode->config.vgId);
  pRsp->downstreamTaskId = htonl(req.taskId);
  pRsp->inputStatus = TASK_OUTPUT_STATUS__NORMAL;

1733 1734 1735 1736
  int32_t len = sizeof(SMsgHead) + sizeof(SStreamDispatchRsp);
  SRpcMsg rsp = { .code = code, .info = pMsg->info, .contLen = len, .pCont = pRspHead};
  tqError("s-task:0x%x send dispatch error rsp, code:%s", taskId, tstrerror(code));

L
Liu Jicong 已提交
1737
  tmsgSendRsp(&rsp);
L
Liu Jicong 已提交
1738 1739
  rpcFreeCont(pMsg->pCont);
  taosFreeQitem(pMsg);
1740
  return -1;
L
Liu Jicong 已提交
1741
}
L
Liu Jicong 已提交
1742

1743
int32_t tqCheckLogInWal(STQ* pTq, int64_t sversion) { return sversion <= pTq->walLogLastVer; }
1744