taosdata/TDengine | Build 3621 | source/libs/executor/src/executor.c | Coveralls

UNCOV

static void cleanupRefPool() {

×

UNCOV

33

  int32_t ref = atomic_val_compare_exchange_32(&exchangeObjRefPool, exchangeObjRefPool, 0);

×

UNCOV

34

  taosCloseRef(ref);

×

UNCOV

35

×

UNCOV

37

static void initRefPool() {

×

UNCOV

38

  exchangeObjRefPool = taosOpenRef(1024, doDestroyExchangeOperatorInfo);

×

UNCOV

39

  (void)atexit(cleanupRefPool);

×

UNCOV

40

×

UNCOV

42

static int32_t doSetSMABlock(SOperatorInfo* pOperator, void* input, size_t numOfBlocks, int32_t type, char* id) {

×

UNCOV

43

  int32_t code = TSDB_CODE_SUCCESS;

×

UNCOV

44

  int32_t lino = 0;

×

UNCOV

45

  if (pOperator->operatorType != QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) {

×

UNCOV

46

    if (pOperator->numOfDownstream == 0) {

×

47

      qError("failed to find stream scan operator to set the input data block, %s" PRIx64, id);

×

48

      return TSDB_CODE_APP_ERROR;

×

UNCOV

51

    if (pOperator->numOfDownstream > 1) {  // not handle this in join query

×

52

      qError("join not supported for stream block scan, %s" PRIx64, id);

×

53

      return TSDB_CODE_APP_ERROR;

×

UNCOV

55

    pOperator->status = OP_NOT_OPENED;

×

UNCOV

56

    return doSetSMABlock(pOperator->pDownstream[0], input, numOfBlocks, type, id);

×

UNCOV

58

    pOperator->status = OP_NOT_OPENED;

×

UNCOV

60

    SStreamScanInfo* pInfo = pOperator->info;

×

UNCOV

62

    if (type == STREAM_INPUT__MERGED_SUBMIT) {

×

UNCOV

63

      for (int32_t i = 0; i < numOfBlocks; i++) {

×

UNCOV

64

        SPackedData* pReq = POINTER_SHIFT(input, i * sizeof(SPackedData));

×

UNCOV

65

        void*        tmp = taosArrayPush(pInfo->pBlockLists, pReq);

×

UNCOV

66

        QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

×

UNCOV

68

      pInfo->blockType = STREAM_INPUT__DATA_SUBMIT;

×

UNCOV

69

    } else if (type == STREAM_INPUT__DATA_SUBMIT) {

×

70

      void* tmp = taosArrayPush(pInfo->pBlockLists, &input);

×

71

      QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

×

72

      pInfo->blockType = STREAM_INPUT__DATA_SUBMIT;

×

UNCOV

73

    } else if (type == STREAM_INPUT__DATA_BLOCK) {

×

UNCOV

74

      for (int32_t i = 0; i < numOfBlocks; ++i) {

×

UNCOV

75

        SSDataBlock* pDataBlock = &((SSDataBlock*)input)[i];

×

UNCOV

76

        SPackedData  tmp = {.pDataBlock = pDataBlock};

×

UNCOV

77

        void*        tmpItem = taosArrayPush(pInfo->pBlockLists, &tmp);

×

UNCOV

78

        QUERY_CHECK_NULL(tmpItem, code, lino, _end, terrno);

×

UNCOV

80

      pInfo->blockType = STREAM_INPUT__DATA_BLOCK;

×

UNCOV

81

    } else if (type == STREAM_INPUT__CHECKPOINT) {

×

UNCOV

82

      SPackedData tmp = {.pDataBlock = input};

×

UNCOV

83

      void*       tmpItem = taosArrayPush(pInfo->pBlockLists, &tmp);

×

UNCOV

84

      QUERY_CHECK_NULL(tmpItem, code, lino, _end, terrno);

×

UNCOV

85

      pInfo->blockType = STREAM_INPUT__CHECKPOINT;

×

UNCOV

86

    } else if (type == STREAM_INPUT__REF_DATA_BLOCK) {

×

UNCOV

87

      for (int32_t i = 0; i < numOfBlocks; ++i) {

×

UNCOV

88

        SPackedData* pReq = POINTER_SHIFT(input, i * sizeof(SPackedData));

×

UNCOV

89

        void*        tmp = taosArrayPush(pInfo->pBlockLists, pReq);

×

UNCOV

90

        QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

×

UNCOV

92

      pInfo->blockType = STREAM_INPUT__DATA_BLOCK;

×

UNCOV

95

    return TSDB_CODE_SUCCESS;

×

98

_end:

×

99

  if (code != TSDB_CODE_SUCCESS) {

×

100

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

×

102

  return code;

×

UNCOV

105

static int32_t doSetStreamOpOpen(SOperatorInfo* pOperator, char* id) {

×

UNCOV

106

  if (pOperator->operatorType != QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) {

×

UNCOV

107

    if (pOperator->numOfDownstream == 0) {

×

108

      qError("failed to find stream scan operator to set the input data block, %s" PRIx64, id);

×

109

      return TSDB_CODE_APP_ERROR;

×

UNCOV

112

    if (pOperator->numOfDownstream > 1) {  // not handle this in join query

×

113

      qError("join not supported for stream block scan, %s" PRIx64, id);

×

114

      return TSDB_CODE_APP_ERROR;

×

UNCOV

117

    pOperator->status = OP_NOT_OPENED;

×

UNCOV

118

    return doSetStreamOpOpen(pOperator->pDownstream[0], id);

×

UNCOV

120

  return 0;

×

123

static void clearStreamBlock(SOperatorInfo* pOperator) {

×

124

  if (pOperator->operatorType != QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) {

×

125

    if (pOperator->numOfDownstream == 1) {

×

126

      return clearStreamBlock(pOperator->pDownstream[0]);

×

129

    SStreamScanInfo* pInfo = pOperator->info;

×

130

    doClearBufferedBlocks(pInfo);

×

134

void qResetTaskInfoCode(qTaskInfo_t tinfo) {

×

135

  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;

×

136

  pTaskInfo->code = 0;

×

137

  clearStreamBlock(pTaskInfo->pRoot);

×

138

×

UNCOV

140

static int32_t doSetStreamBlock(SOperatorInfo* pOperator, void* input, size_t numOfBlocks, int32_t type,

×

UNCOV

142

  int32_t code = TSDB_CODE_SUCCESS;

×

UNCOV

143

  int32_t lino = 0;

×

UNCOV

144

  if (pOperator->operatorType != QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) {

×

UNCOV

145

    if (pOperator->numOfDownstream == 0) {

×

146

      qError("failed to find stream scan operator to set the input data block, %s" PRIx64, id);

×

147

      return TSDB_CODE_APP_ERROR;

×

UNCOV

150

    if (pOperator->numOfDownstream > 1) {  // not handle this in join query

×

151

      qError("join not supported for stream block scan, %s" PRIx64, id);

×

152

      return TSDB_CODE_APP_ERROR;

×

UNCOV

154

    pOperator->status = OP_NOT_OPENED;

×

UNCOV

155

    return doSetStreamBlock(pOperator->pDownstream[0], input, numOfBlocks, type, id);

×

UNCOV

157

    pOperator->status = OP_NOT_OPENED;

×

UNCOV

158

    SStreamScanInfo* pInfo = pOperator->info;

×

UNCOV

160

    qDebug("s-task:%s in this batch, %d blocks need to be processed", id, (int32_t)numOfBlocks);

×

UNCOV

161

    QUERY_CHECK_CONDITION((pInfo->validBlockIndex == 0 && taosArrayGetSize(pInfo->pBlockLists) == 0), code, lino, _end,

×

UNCOV

164

    if (type == STREAM_INPUT__MERGED_SUBMIT) {

×

UNCOV

165

      for (int32_t i = 0; i < numOfBlocks; i++) {

×

UNCOV

166

        SPackedData* pReq = POINTER_SHIFT(input, i * sizeof(SPackedData));

×

UNCOV

167

        void*        tmp = taosArrayPush(pInfo->pBlockLists, pReq);

×

UNCOV

168

        QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

×

UNCOV

171

      pInfo->blockType = STREAM_INPUT__DATA_SUBMIT;

×

UNCOV

172

    } else if (type == STREAM_INPUT__DATA_SUBMIT) {

×

UNCOV

173

      void* tmp = taosArrayPush(pInfo->pBlockLists, input);

×

UNCOV

174

      QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

×

UNCOV

176

      pInfo->blockType = STREAM_INPUT__DATA_SUBMIT;

×

UNCOV

177

    } else if (type == STREAM_INPUT__DATA_BLOCK) {

×

UNCOV

178

      for (int32_t i = 0; i < numOfBlocks; ++i) {

×

UNCOV

179

        SSDataBlock* pDataBlock = &((SSDataBlock*)input)[i];

×

UNCOV

180

        SPackedData  tmp = {.pDataBlock = pDataBlock};

×

UNCOV

181

        void*        tmpItem = taosArrayPush(pInfo->pBlockLists, &tmp);

×

UNCOV

182

        QUERY_CHECK_NULL(tmpItem, code, lino, _end, terrno);

×

UNCOV

185

      pInfo->blockType = STREAM_INPUT__DATA_BLOCK;

×

UNCOV

186

    } else if (type == STREAM_INPUT__CHECKPOINT_TRIGGER) {

×

UNCOV

187

      SPackedData tmp = {.pDataBlock = input};

×

UNCOV

188

      void*       tmpItem = taosArrayPush(pInfo->pBlockLists, &tmp);

×

UNCOV

189

      QUERY_CHECK_NULL(tmpItem, code, lino, _end, terrno);

×

UNCOV

191

      pInfo->blockType = STREAM_INPUT__CHECKPOINT;

×

UNCOV

193

      code = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;

×

UNCOV

194

      QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

197

    return TSDB_CODE_SUCCESS;

×

200

_end:

×

201

  if (code != TSDB_CODE_SUCCESS) {

×

202

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

×

204

  return code;

×

UNCOV

207

int32_t doSetTaskId(SOperatorInfo* pOperator, SStorageAPI* pAPI) {

×

UNCOV

208

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

×

UNCOV

209

  if (pOperator->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) {

×

UNCOV

210

    SStreamScanInfo* pStreamScanInfo = pOperator->info;

×

UNCOV

211

    if (pStreamScanInfo->pTableScanOp != NULL) {

×

UNCOV

212

      STableScanInfo* pScanInfo = pStreamScanInfo->pTableScanOp->info;

×

UNCOV

213

      if (pScanInfo->base.dataReader != NULL) {

×

UNCOV

214

        int32_t code = pAPI->tsdReader.tsdSetReaderTaskId(pScanInfo->base.dataReader, pTaskInfo->id.str);

×

UNCOV

215

        if (code) {

×

216

          qError("failed to set reader id for executor, code:%s", tstrerror(code));

×

217

          return code;

×

UNCOV

222

    return doSetTaskId(pOperator->pDownstream[0], pAPI);

×

UNCOV

225

  return 0;

×

UNCOV

228

int32_t qSetTaskId(qTaskInfo_t tinfo, uint64_t taskId, uint64_t queryId) {

×

UNCOV

229

  SExecTaskInfo* pTaskInfo = tinfo;

×

UNCOV

230

  pTaskInfo->id.queryId = queryId;

×

UNCOV

231

  buildTaskId(taskId, queryId, pTaskInfo->id.str);

×

UNCOV

234

  return doSetTaskId(pTaskInfo->pRoot, &pTaskInfo->storageAPI);

×

UNCOV

237

int32_t qSetStreamOpOpen(qTaskInfo_t tinfo) {

×

UNCOV

238

  if (tinfo == NULL) {

×

239

    return TSDB_CODE_APP_ERROR;

×

UNCOV

242

  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;

×

UNCOV

243

  int32_t        code = doSetStreamOpOpen(pTaskInfo->pRoot, GET_TASKID(pTaskInfo));

×

UNCOV

244

  if (code != TSDB_CODE_SUCCESS) {

×

245

    qError("%s failed to set the stream block data", GET_TASKID(pTaskInfo));

×

UNCOV

247

    qDebug("%s set the stream block successfully", GET_TASKID(pTaskInfo));

×

UNCOV

250

  return code;

×

UNCOV

253

int32_t qSetStreamNotifyInfo(qTaskInfo_t tinfo, int32_t eventTypes, const SSchemaWrapper* pSchemaWrapper,

×

UNCOV

255

  int32_t code = TSDB_CODE_SUCCESS;

×

UNCOV

256

  SStreamTaskInfo *pStreamInfo = NULL;

×

UNCOV

258

  if (tinfo == 0 || eventTypes == 0 || pSchemaWrapper == NULL || stbFullName == NULL) {

×

UNCOV

259

    goto _end;

×

262

  pStreamInfo = &((SExecTaskInfo*)tinfo)->streamInfo;

×

263

  pStreamInfo->eventTypes = eventTypes;

×

264

  pStreamInfo->notifyResultSchema = tCloneSSchemaWrapper(pSchemaWrapper);

×

265

  if (pStreamInfo->notifyResultSchema == NULL) {

×

266

    code = terrno;

×

268

  pStreamInfo->stbFullName = taosStrdup(stbFullName);

×

269

  pStreamInfo->newSubTableRule = newSubTableRule;

×

270

  pStreamInfo->pNotifyEventStat = pNotifyEventStat;

×

UNCOV

272

_end:

×

UNCOV

273

  return code;

×

UNCOV

276

int32_t qSetMultiStreamInput(qTaskInfo_t tinfo, const void* pBlocks, size_t numOfBlocks, int32_t type) {

×

UNCOV

277

  if (tinfo == NULL) {

×

278

    return TSDB_CODE_APP_ERROR;

×

UNCOV

281

  if (pBlocks == NULL || numOfBlocks == 0) {

×

282

    return TSDB_CODE_SUCCESS;

×

UNCOV

285

  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;

×

UNCOV

287

  int32_t code = doSetStreamBlock(pTaskInfo->pRoot, (void*)pBlocks, numOfBlocks, type, GET_TASKID(pTaskInfo));

×

UNCOV

288

  if (code != TSDB_CODE_SUCCESS) {

×

289

    qError("%s failed to set the stream block data", GET_TASKID(pTaskInfo));

×

UNCOV

291

    qDebug("%s set the stream block successfully", GET_TASKID(pTaskInfo));

×

UNCOV

294

  return code;

×

UNCOV

297

int32_t qSetSMAInput(qTaskInfo_t tinfo, const void* pBlocks, size_t numOfBlocks, int32_t type) {

×

UNCOV

298

  if (tinfo == NULL) {

×

299

    return TSDB_CODE_APP_ERROR;

×

UNCOV

302

  if (pBlocks == NULL || numOfBlocks == 0) {

×

303

    return TSDB_CODE_SUCCESS;

×

UNCOV

306

  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;

×

UNCOV

308

  int32_t code = doSetSMABlock(pTaskInfo->pRoot, (void*)pBlocks, numOfBlocks, type, GET_TASKID(pTaskInfo));

×

UNCOV

309

  if (code != TSDB_CODE_SUCCESS) {

×

310

    qError("%s failed to set the sma block data", GET_TASKID(pTaskInfo));

×

UNCOV

312

    qDebug("%s set the sma block successfully", GET_TASKID(pTaskInfo));

×

UNCOV

315

  return code;

×

UNCOV

318

qTaskInfo_t qCreateQueueExecTaskInfo(void* msg, SReadHandle* pReaderHandle, int32_t vgId, int32_t* numOfCols,

×

UNCOV

320

  if (msg == NULL) {  // create raw scan

×

UNCOV

321

    SExecTaskInfo* pTaskInfo = NULL;

×

UNCOV

323

    int32_t code = doCreateTask(0, id, vgId, OPTR_EXEC_MODEL_QUEUE, &pReaderHandle->api, &pTaskInfo);

×

UNCOV

324

    if (NULL == pTaskInfo || code != 0) {

×

325

      return NULL;

×

UNCOV

328

    code = createRawScanOperatorInfo(pReaderHandle, pTaskInfo, &pTaskInfo->pRoot);

×

UNCOV

329

    if (NULL == pTaskInfo->pRoot || code != 0) {

×

330

      taosMemoryFree(pTaskInfo);

×

331

      return NULL;

×

UNCOV

334

    pTaskInfo->storageAPI = pReaderHandle->api;

×

UNCOV

335

    qDebug("create raw scan task info completed, vgId:%d, %s", vgId, GET_TASKID(pTaskInfo));

×

UNCOV

336

    return pTaskInfo;

×

UNCOV

339

  SSubplan* pPlan = NULL;

×

UNCOV

340

  int32_t   code = qStringToSubplan(msg, &pPlan);

×

UNCOV

341

  if (code != TSDB_CODE_SUCCESS) {

×

342

    terrno = code;

×

343

    return NULL;

×

UNCOV

346

  qTaskInfo_t pTaskInfo = NULL;

×

UNCOV

347

  code = qCreateExecTask(pReaderHandle, vgId, 0, pPlan, &pTaskInfo, NULL, 0, NULL, OPTR_EXEC_MODEL_QUEUE);

×

UNCOV

348

  if (code != TSDB_CODE_SUCCESS) {

×

349

    qDestroyTask(pTaskInfo);

×

350

    terrno = code;

×

351

    return NULL;

×

UNCOV

355

  SDataBlockDescNode* pDescNode = pPlan->pNode->pOutputDataBlockDesc;

×

UNCOV

356

  *numOfCols = 0;

×

UNCOV

359

  FOREACH(pNode, pDescNode->pSlots) {

×

UNCOV

360

    SSlotDescNode* pSlotDesc = (SSlotDescNode*)pNode;

×

UNCOV

361

    if (pSlotDesc->output) {

×

UNCOV

362

      ++(*numOfCols);

×

UNCOV

366

  return pTaskInfo;

×

UNCOV

369

int32_t qCreateStreamExecTaskInfo(qTaskInfo_t* pTaskInfo, void* msg, SReadHandle* readers, int32_t vgId, int32_t taskId) {

×

UNCOV

370

  if (msg == NULL) {

×

371

    return TSDB_CODE_INVALID_PARA;

×

UNCOV

374

  *pTaskInfo = NULL;

×

UNCOV

376

  SSubplan* pPlan = NULL;

×

UNCOV

377

  int32_t   code = qStringToSubplan(msg, &pPlan);

×

UNCOV

378

  if (code != TSDB_CODE_SUCCESS) {

×

379

    return code;

×

UNCOV

382

  code = qCreateExecTask(readers, vgId, taskId, pPlan, pTaskInfo, NULL, 0, NULL, OPTR_EXEC_MODEL_STREAM);

×

UNCOV

383

  if (code != TSDB_CODE_SUCCESS) {

×

384

    qDestroyTask(*pTaskInfo);

×

385

    return code;

×

UNCOV

388

  code = qStreamInfoResetTimewindowFilter(*pTaskInfo);

×

UNCOV

389

  if (code != TSDB_CODE_SUCCESS) {

×

390

    qDestroyTask(*pTaskInfo);

×

UNCOV

393

  return code;

×

UNCOV

396

static int32_t filterUnqualifiedTables(const SStreamScanInfo* pScanInfo, const SArray* tableIdList, const char* idstr,

×

UNCOV

398

  int32_t code = TSDB_CODE_SUCCESS;

×

UNCOV

399

  int32_t lino = 0;

×

UNCOV

400

  SArray* qa = taosArrayInit(4, sizeof(tb_uid_t));

×

UNCOV

401

  QUERY_CHECK_NULL(qa, code, lino, _error, terrno);

×

UNCOV

402

  int32_t numOfUids = taosArrayGetSize(tableIdList);

×

UNCOV

403

  if (numOfUids == 0) {

×

UNCOV

404

    (*ppArrayRes) = qa;

×

UNCOV

405

    goto _error;

×

UNCOV

408

  STableScanInfo* pTableScanInfo = pScanInfo->pTableScanOp->info;

×

UNCOV

410

  uint64_t suid = 0;

×

UNCOV

411

  uint64_t uid = 0;

×

UNCOV

412

  int32_t  type = 0;

×

UNCOV

413

  tableListGetSourceTableInfo(pTableScanInfo->base.pTableListInfo, &suid, &uid, &type);

×

UNCOV

416

  SMetaReader mr = {0};

×

UNCOV

417

  pAPI->metaReaderFn.initReader(&mr, pScanInfo->readHandle.vnode, META_READER_LOCK, &pAPI->metaFn);

×

UNCOV

418

  for (int32_t i = 0; i < numOfUids; ++i) {

×

UNCOV

419

    uint64_t* id = (uint64_t*)taosArrayGet(tableIdList, i);

×

UNCOV

420

    QUERY_CHECK_NULL(id, code, lino, _end, terrno);

×

UNCOV

422

    int32_t code = pAPI->metaReaderFn.getTableEntryByUid(&mr, *id);

×

UNCOV

423

    if (code != TSDB_CODE_SUCCESS) {

×

424

      qError("failed to get table meta, uid:%" PRIu64 " code:%s, %s", *id, tstrerror(terrno), idstr);

×

425

      continue;

×

UNCOV

428

    tDecoderClear(&mr.coder);

×

UNCOV

430

    if (mr.me.type == TSDB_SUPER_TABLE) {

×

431

      continue;

×

UNCOV

433

      if (type == TSDB_SUPER_TABLE) {

×

UNCOV

435

        if (mr.me.type != TSDB_CHILD_TABLE || mr.me.ctbEntry.suid != suid) {

×

UNCOV

436

          continue;

×

UNCOV

441

        if (mr.me.uid != uid) {

×

UNCOV

442

          continue;

×

UNCOV

447

    if (pScanInfo->pTagCond != NULL) {

×

UNCOV

448

      bool          qualified = false;

×

UNCOV

449

      STableKeyInfo info = {.groupId = 0, .uid = mr.me.uid};

×

UNCOV

450

      code = isQualifiedTable(&info, pScanInfo->pTagCond, pScanInfo->readHandle.vnode, &qualified, pAPI);

×

UNCOV

451

      if (code != TSDB_CODE_SUCCESS) {

×

452

        qError("failed to filter new table, uid:0x%" PRIx64 ", %s", info.uid, idstr);

×

UNCOV

453

        continue;

×

UNCOV

456

      if (!qualified) {

×

UNCOV

457

        continue;

×

UNCOV

462

    void* tmp = taosArrayPush(qa, id);

×

UNCOV

463

    QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

×

UNCOV

466

_end:

×

UNCOV

468

  pAPI->metaReaderFn.clearReader(&mr);

×

UNCOV

469

  (*ppArrayRes) = qa;

×

UNCOV

471

_error:

×

UNCOV

472

  if (code != TSDB_CODE_SUCCESS) {

×

473

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

×

UNCOV

475

  return code;

×

UNCOV

478

int32_t qUpdateTableListForStreamScanner(qTaskInfo_t tinfo, const SArray* tableIdList, bool isAdd) {

×

UNCOV

479

  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;

×

UNCOV

480

  const char*    id = GET_TASKID(pTaskInfo);

×

UNCOV

481

  int32_t        code = 0;

×

UNCOV

483

  if (isAdd) {

×

UNCOV

484

    qDebug("try to add %d tables id into query list, %s", (int32_t)taosArrayGetSize(tableIdList), id);

×

UNCOV

488

  SOperatorInfo* pInfo = NULL;

×

UNCOV

489

  code = extractOperatorInTree(pTaskInfo->pRoot, QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN, id, &pInfo);

×

UNCOV

490

  if (code != 0 || pInfo == NULL) {

×

UNCOV

491

    return code;

×

UNCOV

494

  SStreamScanInfo* pScanInfo = pInfo->info;

×

UNCOV

495

  if (pInfo->pTaskInfo->execModel == OPTR_EXEC_MODEL_QUEUE) {  // clear meta cache for subscription if tag is changed

×

UNCOV

496

    for (int32_t i = 0; i < taosArrayGetSize(tableIdList); ++i) {

×

UNCOV

497

      int64_t*        uid = (int64_t*)taosArrayGet(tableIdList, i);

×

UNCOV

498

      STableScanInfo* pTableScanInfo = pScanInfo->pTableScanOp->info;

×

UNCOV

499

      taosLRUCacheErase(pTableScanInfo->base.metaCache.pTableMetaEntryCache, uid, LONG_BYTES);

×

UNCOV

503

  if (isAdd) {  // add new table id

×

UNCOV

504

    SArray* qa = NULL;

×

UNCOV

505

    code = filterUnqualifiedTables(pScanInfo, tableIdList, id, &pTaskInfo->storageAPI, &qa);

×

UNCOV

506

    if (code != TSDB_CODE_SUCCESS) {

×

507

      taosArrayDestroy(qa);

×

508

      return code;

×

UNCOV

510

    int32_t numOfQualifiedTables = taosArrayGetSize(qa);

×

UNCOV

511

    qDebug("%d qualified child tables added into stream scanner, %s", numOfQualifiedTables, id);

×

UNCOV

512

    pTaskInfo->storageAPI.tqReaderFn.tqReaderAddTables(pScanInfo->tqReader, qa);

×

UNCOV

514

    bool   assignUid = false;

×

UNCOV

515

    size_t bufLen = (pScanInfo->pGroupTags != NULL) ? getTableTagsBufLen(pScanInfo->pGroupTags) : 0;

×

UNCOV

516

    char*  keyBuf = NULL;

×

UNCOV

517

    if (bufLen > 0) {

×

UNCOV

518

      assignUid = groupbyTbname(pScanInfo->pGroupTags);

×

UNCOV

519

      keyBuf = taosMemoryMalloc(bufLen);

×

UNCOV

520

      if (keyBuf == NULL) {

×

521

        taosArrayDestroy(qa);

×

522

        return terrno;

×

UNCOV

526

    STableListInfo* pTableListInfo = ((STableScanInfo*)pScanInfo->pTableScanOp->info)->base.pTableListInfo;

×

UNCOV

527

    taosWLockLatch(&pTaskInfo->lock);

×

UNCOV

529

    for (int32_t i = 0; i < numOfQualifiedTables; ++i) {

×

UNCOV

530

      uint64_t* uid = taosArrayGet(qa, i);

×

UNCOV

531

      if (!uid) {

×

532

        taosMemoryFree(keyBuf);

×

533

        taosArrayDestroy(qa);

×

534

        taosWUnLockLatch(&pTaskInfo->lock);

×

535

        return terrno;

×

UNCOV

537

      STableKeyInfo keyInfo = {.uid = *uid, .groupId = 0};

×

UNCOV

539

      if (bufLen > 0) {

×

UNCOV

540

        if (assignUid) {

×

UNCOV

541

          keyInfo.groupId = keyInfo.uid;

×

UNCOV

543

          code = getGroupIdFromTagsVal(pScanInfo->readHandle.vnode, keyInfo.uid, pScanInfo->pGroupTags, keyBuf,

×

UNCOV

545

          if (code != TSDB_CODE_SUCCESS) {

×

546

            taosMemoryFree(keyBuf);

×

547

            taosArrayDestroy(qa);

×

548

            taosWUnLockLatch(&pTaskInfo->lock);

×

549

            return code;

×

UNCOV

554

      code = tableListAddTableInfo(pTableListInfo, keyInfo.uid, keyInfo.groupId);

×

UNCOV

555

      if (code != TSDB_CODE_SUCCESS) {

×

556

        taosMemoryFree(keyBuf);

×

557

        taosArrayDestroy(qa);

×

558

        taosWUnLockLatch(&pTaskInfo->lock);

×

559

        return code;

×

UNCOV

563

    taosWUnLockLatch(&pTaskInfo->lock);

×

UNCOV

564

    if (keyBuf != NULL) {

×

UNCOV

565

      taosMemoryFree(keyBuf);

×

UNCOV

568

    taosArrayDestroy(qa);

×

UNCOV

570

    qDebug("%d remove child tables from the stream scanner, %s", (int32_t)taosArrayGetSize(tableIdList), id);

×

UNCOV

571

    taosWLockLatch(&pTaskInfo->lock);

×

UNCOV

572

    pTaskInfo->storageAPI.tqReaderFn.tqReaderRemoveTables(pScanInfo->tqReader, tableIdList);

×

UNCOV

573

    taosWUnLockLatch(&pTaskInfo->lock);

×

UNCOV

576

  return code;

×

UNCOV

579

int32_t qGetQueryTableSchemaVersion(qTaskInfo_t tinfo, char* dbName, int32_t dbNameBuffLen, char* tableName,

×

UNCOV

582

  *tbGet = false;

×

UNCOV

584

  if (tinfo == NULL || dbName == NULL || tableName == NULL) {

×

585

    return TSDB_CODE_INVALID_PARA;

×

UNCOV

587

  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;

×

UNCOV

589

  if (taosArrayGetSize(pTaskInfo->schemaInfos) <= idx) {

×

UNCOV

590

    return TSDB_CODE_SUCCESS;

×

UNCOV

593

  SSchemaInfo* pSchemaInfo = taosArrayGet(pTaskInfo->schemaInfos, idx);

×

UNCOV

594

  if (!pSchemaInfo) {

×

595

    qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));

×

596

    return terrno;

×

UNCOV

599

  *sversion = pSchemaInfo->sw->version;

×

UNCOV

600

  *tversion = pSchemaInfo->tversion;

×

UNCOV

601

  if (pSchemaInfo->dbname) {

×

UNCOV

602

    tstrncpy(dbName, pSchemaInfo->dbname, dbNameBuffLen);

×

604

    dbName[0] = 0;

×

UNCOV

606

  if (pSchemaInfo->tablename) {

×

UNCOV

607

    tstrncpy(tableName, pSchemaInfo->tablename, tbaleNameBuffLen);

×

609

    tableName[0] = 0;

×

UNCOV

612

  *tbGet = true;

×

UNCOV

614

  return TSDB_CODE_SUCCESS;

×

UNCOV

617

bool qIsDynamicExecTask(qTaskInfo_t tinfo) { return ((SExecTaskInfo*)tinfo)->dynamicTask; }

×

UNCOV

619

void destroyOperatorParam(SOperatorParam* pParam) {

×

UNCOV

620

  if (NULL == pParam) {

×

UNCOV

621

    return;

×

UNCOV

627

void qUpdateOperatorParam(qTaskInfo_t tinfo, void* pParam) {

×

UNCOV

628

  destroyOperatorParam(((SExecTaskInfo*)tinfo)->pOpParam);

×

UNCOV

629

  ((SExecTaskInfo*)tinfo)->pOpParam = pParam;

×

UNCOV

630

  ((SExecTaskInfo*)tinfo)->paramSet = false;

×

UNCOV

631

×

UNCOV

633

int32_t qExecutorInit(void) {

×

NEW

634

  (void)taosThreadOnce(&initPoolOnce, initRefPool);

×

UNCOV

635

  return TSDB_CODE_SUCCESS;

×

UNCOV

638

int32_t qCreateExecTask(SReadHandle* readHandle, int32_t vgId, uint64_t taskId, SSubplan* pSubplan,

×

UNCOV

641

  SExecTaskInfo** pTask = (SExecTaskInfo**)pTaskInfo;

×

UNCOV

642

  (void)taosThreadOnce(&initPoolOnce, initRefPool);

×

UNCOV

644

  qDebug("start to create task, TID:0x%" PRIx64 " QID:0x%" PRIx64 ", vgId:%d", taskId, pSubplan->id.queryId, vgId);

×

UNCOV

646

  int32_t code = createExecTaskInfo(pSubplan, pTask, readHandle, taskId, vgId, sql, model);

×

UNCOV

647

  if (code != TSDB_CODE_SUCCESS || NULL == *pTask) {

×

648

    qError("failed to createExecTaskInfo, code: %s", tstrerror(code));

×

UNCOV

649

    goto _error;

×

UNCOV

652

  if (handle) {

×

UNCOV

653

    SDataSinkMgtCfg cfg = {.maxDataBlockNum = 500, .maxDataBlockNumPerQuery = 50, .compress = compressResult};

×

UNCOV

654

    void*           pSinkManager = NULL;

×

UNCOV

655

    code = dsDataSinkMgtInit(&cfg, &(*pTask)->storageAPI, &pSinkManager);

×

UNCOV

656

    if (code != TSDB_CODE_SUCCESS) {

×

657

      qError("failed to dsDataSinkMgtInit, code:%s, %s", tstrerror(code), (*pTask)->id.str);

×

658

      goto _error;

×

UNCOV

661

    void* pSinkParam = NULL;

×

UNCOV

662

    code = createDataSinkParam(pSubplan->pDataSink, &pSinkParam, (*pTask), readHandle);

×

UNCOV

663

    if (code != TSDB_CODE_SUCCESS) {

×

664

      qError("failed to createDataSinkParam, vgId:%d, code:%s, %s", vgId, tstrerror(code), (*pTask)->id.str);

×

665

      taosMemoryFree(pSinkManager);

×

666

      goto _error;

×

UNCOV

669

    SDataSinkNode* pSink = NULL;

×

UNCOV

670

    if (readHandle->localExec) {

×

UNCOV

671

      code = nodesCloneNode((SNode *)pSubplan->pDataSink, (SNode **)&pSink);

×

UNCOV

672

      if (code != TSDB_CODE_SUCCESS) {

×

UNCOV

673

        qError("failed to nodesCloneNode, srcType:%d, code:%s, %s", nodeType(pSubplan->pDataSink), tstrerror(code), (*pTask)->id.str);

×

UNCOV

674

        taosMemoryFree(pSinkManager);

×

675

        goto _error;

×

UNCOV

680

    code = dsCreateDataSinker(pSinkManager, readHandle->localExec ? &pSink : &pSubplan->pDataSink, handle, pSinkParam, (*pTask)->id.str);

×

UNCOV

681

    if (code) {

×

682

      qError("s-task:%s failed to create data sinker, code:%s", (*pTask)->id.str, tstrerror(code));

×

UNCOV

686

  qDebug("subplan task create completed, TID:0x%" PRIx64 " QID:0x%" PRIx64 " code:%s", taskId, pSubplan->id.queryId,

×

UNCOV

689

_error:

×

UNCOV

691

  return code;

×

694

static void freeBlock(void* param) {

×

695

  SSDataBlock* pBlock = *(SSDataBlock**)param;

×

696

  blockDataDestroy(pBlock);

×

697

×

UNCOV

699

int32_t qExecTaskOpt(qTaskInfo_t tinfo, SArray* pResList, uint64_t* useconds, bool* hasMore, SLocalFetch* pLocal) {

×

UNCOV

700

  int32_t        code = TSDB_CODE_SUCCESS;

×

UNCOV

701

  int32_t        lino = 0;

×

UNCOV

702

  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;

×

UNCOV

703

  int64_t        threadId = taosGetSelfPthreadId();

×

UNCOV

705

  if (pLocal) {

×

UNCOV

706

    memcpy(&pTaskInfo->localFetch, pLocal, sizeof(*pLocal));

×

UNCOV

709

  taosArrayClear(pResList);

×

UNCOV

711

  int64_t curOwner = 0;

×

UNCOV

712

  if ((curOwner = atomic_val_compare_exchange_64(&pTaskInfo->owner, 0, threadId)) != 0) {

×

713

    qError("%s-%p execTask is now executed by thread:%p", GET_TASKID(pTaskInfo), pTaskInfo, (void*)curOwner);

×

714

    pTaskInfo->code = TSDB_CODE_QRY_IN_EXEC;

×

715

    return pTaskInfo->code;

×

UNCOV

718

  if (pTaskInfo->cost.start == 0) {

×

UNCOV

719

    pTaskInfo->cost.start = taosGetTimestampUs();

×

UNCOV

722

  if (isTaskKilled(pTaskInfo)) {

×

UNCOV

723

    atomic_store_64(&pTaskInfo->owner, 0);

×

UNCOV

724

    qDebug("%s already killed, abort", GET_TASKID(pTaskInfo));

×

UNCOV

725

    return pTaskInfo->code;

×

UNCOV

729

  int32_t ret = setjmp(pTaskInfo->env);

×

UNCOV

730

  if (ret != TSDB_CODE_SUCCESS) {

×

UNCOV

731

    pTaskInfo->code = ret;

×

UNCOV

732

    (void)cleanUpUdfs();

×

UNCOV

734

    qDebug("%s task abort due to error/cancel occurs, code:%s", GET_TASKID(pTaskInfo), tstrerror(pTaskInfo->code));

×

UNCOV

735

    atomic_store_64(&pTaskInfo->owner, 0);

×

UNCOV

737

    return pTaskInfo->code;

×

UNCOV

740

  qDebug("%s execTask is launched", GET_TASKID(pTaskInfo));

×

UNCOV

742

  int32_t      current = 0;

×

UNCOV

743

  SSDataBlock* pRes = NULL;

×

UNCOV

744

  int64_t      st = taosGetTimestampUs();

×

UNCOV

746

  if (pTaskInfo->pOpParam && !pTaskInfo->paramSet) {

×

UNCOV

747

    pTaskInfo->paramSet = true;

×

UNCOV

748

    code = pTaskInfo->pRoot->fpSet.getNextExtFn(pTaskInfo->pRoot, pTaskInfo->pOpParam, &pRes);

×

UNCOV

750

    code = pTaskInfo->pRoot->fpSet.getNextFn(pTaskInfo->pRoot, &pRes);

×

UNCOV

753

  QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

754

  code = blockDataCheck(pRes);

×

UNCOV

755

  QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

757

  if (pRes == NULL) {

×

UNCOV

758

    st = taosGetTimestampUs();

×

UNCOV

761

  int32_t rowsThreshold = pTaskInfo->pSubplan->rowsThreshold;

×

UNCOV

762

  if (!pTaskInfo->pSubplan->dynamicRowThreshold || 4096 <= pTaskInfo->pSubplan->rowsThreshold) {

×

UNCOV

763

    rowsThreshold = 4096;

×

UNCOV

766

  int32_t blockIndex = 0;

×

UNCOV

767

  while (pRes != NULL) {

×

UNCOV

768

    SSDataBlock* p = NULL;

×

UNCOV

769

    if (blockIndex >= taosArrayGetSize(pTaskInfo->pResultBlockList)) {

×

UNCOV

770

      SSDataBlock* p1 = NULL;

×

UNCOV

771

      code = createOneDataBlock(pRes, true, &p1);

×

UNCOV

772

      QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

774

      void* tmp = taosArrayPush(pTaskInfo->pResultBlockList, &p1);

×

UNCOV

775

      QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

×

UNCOV

776

      p = p1;

×

UNCOV

778

      void* tmp = taosArrayGet(pTaskInfo->pResultBlockList, blockIndex);

×

UNCOV

779

      QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

×

UNCOV

781

      p = *(SSDataBlock**)tmp;

×

UNCOV

782

      code = copyDataBlock(p, pRes);

×

UNCOV

783

      QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

786

    blockIndex += 1;

×

UNCOV

788

    current += p->info.rows;

×

UNCOV

789

    QUERY_CHECK_CONDITION((p->info.rows > 0 || p->info.type == STREAM_CHECKPOINT), code, lino, _end,

×

UNCOV

791

    void* tmp = taosArrayPush(pResList, &p);

×

UNCOV

792

    QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

×

UNCOV

794

    if (current >= rowsThreshold) {

×

UNCOV

795

      break;

×

UNCOV

798

    code = pTaskInfo->pRoot->fpSet.getNextFn(pTaskInfo->pRoot, &pRes);

×

UNCOV

799

    QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

800

    code = blockDataCheck(pRes);

×

UNCOV

801

    QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

804

  if (pTaskInfo->pSubplan->dynamicRowThreshold) {

×

UNCOV

805

    pTaskInfo->pSubplan->rowsThreshold -= current;

×

UNCOV

808

  *hasMore = (pRes != NULL);

×

UNCOV

809

  uint64_t el = (taosGetTimestampUs() - st);

×

UNCOV

811

  pTaskInfo->cost.elapsedTime += el;

×

UNCOV

812

  if (NULL == pRes) {

×

UNCOV

813

    *useconds = pTaskInfo->cost.elapsedTime;

×

UNCOV

816

_end:

×

UNCOV

817

  (void)cleanUpUdfs();

×

UNCOV

819

  uint64_t total = pTaskInfo->pRoot->resultInfo.totalRows;

×

UNCOV

820

  qDebug("%s task suspended, %d rows in %d blocks returned, total:%" PRId64 " rows, in sinkNode:%d, elapsed:%.2f ms",

×

UNCOV

823

  atomic_store_64(&pTaskInfo->owner, 0);

×

UNCOV

824

  if (code) {

×

825

    pTaskInfo->code = code;

×

826

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

×

UNCOV

829

  return pTaskInfo->code;

×

UNCOV

832

void qCleanExecTaskBlockBuf(qTaskInfo_t tinfo) {

×

UNCOV

833

  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;

×

UNCOV

834

  SArray*        pList = pTaskInfo->pResultBlockList;

×

UNCOV

835

  size_t         num = taosArrayGetSize(pList);

×

UNCOV

836

  for (int32_t i = 0; i < num; ++i) {

×

UNCOV

837

    SSDataBlock** p = taosArrayGet(pTaskInfo->pResultBlockList, i);

×

UNCOV

838

    if (p) {

×

UNCOV

839

      blockDataDestroy(*p);

×

UNCOV

843

  taosArrayClear(pTaskInfo->pResultBlockList);

×

UNCOV

844

×

UNCOV

846

int32_t qExecTask(qTaskInfo_t tinfo, SSDataBlock** pRes, uint64_t* useconds) {

×

UNCOV

847

  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;

×

UNCOV

848

  int64_t        threadId = taosGetSelfPthreadId();

×

UNCOV

849

  int64_t        curOwner = 0;

×

UNCOV

851

  *pRes = NULL;

×

UNCOV

854

  taosRLockLatch(&pTaskInfo->lock);

×

UNCOV

855

  bool isKilled = isTaskKilled(pTaskInfo);

×

UNCOV

856

  if (isKilled) {

×

857

    clearStreamBlock(pTaskInfo->pRoot);

×

858

    qDebug("%s already killed, abort", GET_TASKID(pTaskInfo));

×

860

    taosRUnLockLatch(&pTaskInfo->lock);

×

861

    return pTaskInfo->code;

×

UNCOV

864

  if (pTaskInfo->owner != 0) {

×

865

    qError("%s-%p execTask is now executed by thread:%p", GET_TASKID(pTaskInfo), pTaskInfo, (void*)curOwner);

×

866

    pTaskInfo->code = TSDB_CODE_QRY_IN_EXEC;

×

868

    taosRUnLockLatch(&pTaskInfo->lock);

×

869

    return pTaskInfo->code;

×

UNCOV

872

  pTaskInfo->owner = threadId;

×

UNCOV

873

  taosRUnLockLatch(&pTaskInfo->lock);

×

UNCOV

875

  if (pTaskInfo->cost.start == 0) {

×

UNCOV

876

    pTaskInfo->cost.start = taosGetTimestampUs();

×

UNCOV

880

  int32_t ret = setjmp(pTaskInfo->env);

×

UNCOV

881

  if (ret != TSDB_CODE_SUCCESS) {

×

UNCOV

882

    pTaskInfo->code = ret;

×

UNCOV

883

    (void)cleanUpUdfs();

×

UNCOV

884

    qDebug("%s task abort due to error/cancel occurs, code:%s", GET_TASKID(pTaskInfo), tstrerror(pTaskInfo->code));

×

UNCOV

885

    atomic_store_64(&pTaskInfo->owner, 0);

×

UNCOV

886

    return pTaskInfo->code;

×

UNCOV

889

  qDebug("%s execTask is launched", GET_TASKID(pTaskInfo));

×

UNCOV

891

  int64_t st = taosGetTimestampUs();

×

UNCOV

893

  int32_t code = pTaskInfo->pRoot->fpSet.getNextFn(pTaskInfo->pRoot, pRes);

×

UNCOV

894

  if (code) {

×

895

    pTaskInfo->code = code;

×

896

    qError("%s failed at line %d, code:%s %s", __func__, __LINE__, tstrerror(code), GET_TASKID(pTaskInfo));

×

UNCOV

899

  code = blockDataCheck(*pRes);

×

UNCOV

900

  if (code) {

×

901

    pTaskInfo->code = code;

×

902

    qError("%s failed at line %d, code:%s %s", __func__, __LINE__, tstrerror(code), GET_TASKID(pTaskInfo));

×

UNCOV

905

  uint64_t el = (taosGetTimestampUs() - st);

×

UNCOV

907

  pTaskInfo->cost.elapsedTime += el;

×

UNCOV

908

  if (NULL == *pRes) {

×

UNCOV

909

    *useconds = pTaskInfo->cost.elapsedTime;

×

UNCOV

912

  (void) cleanUpUdfs();

×

UNCOV

914

  int32_t  current = (*pRes != NULL) ? (*pRes)->info.rows : 0;

×

UNCOV

915

  uint64_t total = pTaskInfo->pRoot->resultInfo.totalRows;

×

UNCOV

917

  qDebug("%s task suspended, %d rows returned, total:%" PRId64 " rows, in sinkNode:%d, elapsed:%.2f ms",

×

UNCOV

920

  atomic_store_64(&pTaskInfo->owner, 0);

×

UNCOV

921

  return pTaskInfo->code;

×

UNCOV

924

int32_t qAppendTaskStopInfo(SExecTaskInfo* pTaskInfo, SExchangeOpStopInfo* pInfo) {

×

UNCOV

925

  taosWLockLatch(&pTaskInfo->stopInfo.lock);

×

UNCOV

926

  void* tmp = taosArrayPush(pTaskInfo->stopInfo.pStopInfo, pInfo);

×

UNCOV

927

  taosWUnLockLatch(&pTaskInfo->stopInfo.lock);

×

UNCOV

929

  if (!tmp) {

×

UNCOV

930

    qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));

×

UNCOV

931

    return terrno;

×

UNCOV

933

  return TSDB_CODE_SUCCESS;

×

936

int32_t stopInfoComp(void const* lp, void const* rp) {

×

937

  SExchangeOpStopInfo* key = (SExchangeOpStopInfo*)lp;

×

938

  SExchangeOpStopInfo* pInfo = (SExchangeOpStopInfo*)rp;

×

940

  if (key->refId < pInfo->refId) {

×

941

    return -1;

×

942

  } else if (key->refId > pInfo->refId) {

×

943

    return 1;

×

946

  return 0;

×

949

void qRemoveTaskStopInfo(SExecTaskInfo* pTaskInfo, SExchangeOpStopInfo* pInfo) {

×

950

  taosWLockLatch(&pTaskInfo->stopInfo.lock);

×

951

  int32_t idx = taosArraySearchIdx(pTaskInfo->stopInfo.pStopInfo, pInfo, stopInfoComp, TD_EQ);

×

952

  if (idx >= 0) {

×

953

    taosArrayRemove(pTaskInfo->stopInfo.pStopInfo, idx);

×

955

  taosWUnLockLatch(&pTaskInfo->stopInfo.lock);

×

956

×

UNCOV

958

void qStopTaskOperators(SExecTaskInfo* pTaskInfo) {

×

UNCOV

959

  taosWLockLatch(&pTaskInfo->stopInfo.lock);

×

UNCOV

961

  int32_t num = taosArrayGetSize(pTaskInfo->stopInfo.pStopInfo);

×

UNCOV

962

  for (int32_t i = 0; i < num; ++i) {

×

UNCOV

963

    SExchangeOpStopInfo* pStop = taosArrayGet(pTaskInfo->stopInfo.pStopInfo, i);

×

UNCOV

964

    if (!pStop) {

×

965

      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));

×

966

      continue;

×

UNCOV

968

    SExchangeInfo* pExchangeInfo = taosAcquireRef(exchangeObjRefPool, pStop->refId);

×

UNCOV

969

    if (pExchangeInfo) {

×

UNCOV

970

      int32_t code = tsem_post(&pExchangeInfo->ready);

×

UNCOV

971

      if (code != TSDB_CODE_SUCCESS) {

×

972

        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));

×

UNCOV

974

      code = taosReleaseRef(exchangeObjRefPool, pStop->refId);

×

UNCOV

975

      if (code != TSDB_CODE_SUCCESS) {

×

976

        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));

×

UNCOV

981

  taosWUnLockLatch(&pTaskInfo->stopInfo.lock);

×

UNCOV

982

×

UNCOV

984

int32_t qAsyncKillTask(qTaskInfo_t qinfo, int32_t rspCode) {

×

UNCOV

985

  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)qinfo;

×

UNCOV

986

  if (pTaskInfo == NULL) {

×

987

    return TSDB_CODE_QRY_INVALID_QHANDLE;

×

UNCOV

990

  qDebug("%s execTask async killed", GET_TASKID(pTaskInfo));

×

UNCOV

992

  setTaskKilled(pTaskInfo, rspCode);

×

UNCOV

993

  qStopTaskOperators(pTaskInfo);

×

UNCOV

995

  return TSDB_CODE_SUCCESS;

×

NEW

998

int32_t qKillTask(qTaskInfo_t tinfo, int32_t rspCode, int64_t waitDuration) {

×

NEW

999

  int64_t        st = taosGetTimestampMs();

×

UNCOV

1000

  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;

×

UNCOV

1001

  if (pTaskInfo == NULL) {

×

UNCOV

1002

    return TSDB_CODE_QRY_INVALID_QHANDLE;

×

NEW

1005

  if (waitDuration > 0) {

×

NEW

1006

    qDebug("%s sync killed execTask, and waiting for %.2fs", GET_TASKID(pTaskInfo), waitDuration/1000.0);

×

NEW

1008

    qDebug("%s async killed execTask", GET_TASKID(pTaskInfo));

×

UNCOV

1011

  setTaskKilled(pTaskInfo, TSDB_CODE_TSC_QUERY_KILLED);

×

NEW

1013

  if (waitDuration > 0) {

×

NEW

1015

      taosWLockLatch(&pTaskInfo->lock);

×

NEW

1016

      if (qTaskIsExecuting(pTaskInfo)) {  // let's wait for 100 ms and try again

×

NEW

1017

        taosWUnLockLatch(&pTaskInfo->lock);

×

NEW

1019

        taosMsleep(200);

×

NEW

1021

        int64_t d = taosGetTimestampMs() - st;

×

NEW

1022

        if (d >= waitDuration && waitDuration >= 0) {

×

NEW

1023

          qWarn("%s waiting more than %.2fs, not wait anymore", GET_TASKID(pTaskInfo), waitDuration / 1000.0);

×

NEW

1024

          return TSDB_CODE_SUCCESS;

×

NEW

1027

        pTaskInfo->code = rspCode;

×

NEW

1028

        taosWUnLockLatch(&pTaskInfo->lock);

×

NEW

1029

        return TSDB_CODE_SUCCESS;

×

NEW

1034

  return TSDB_CODE_SUCCESS;

×

UNCOV

1037

bool qTaskIsExecuting(qTaskInfo_t qinfo) {

×

UNCOV

1038

  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)qinfo;

×

UNCOV

1039

  if (NULL == pTaskInfo) {

×

1040

    return false;

×

UNCOV

1043

  return 0 != atomic_load_64(&pTaskInfo->owner);

×

UNCOV

1046

static void printTaskExecCostInLog(SExecTaskInfo* pTaskInfo) {

×

UNCOV

1047

  STaskCostInfo* pSummary = &pTaskInfo->cost;

×

UNCOV

1048

  int64_t        idleTime = pSummary->start - pSummary->created;

×

UNCOV

1050

  SFileBlockLoadRecorder* pRecorder = pSummary->pRecoder;

×

UNCOV

1051

  if (pSummary->pRecoder != NULL) {

×

UNCOV

1052

    qDebug(

×

UNCOV

1060

    qDebug("%s :cost summary: idle in queue:%.2f ms, elapsed time:%.2f ms", GET_TASKID(pTaskInfo), idleTime / 1000.0,

×

UNCOV

1063

×

UNCOV

1065

void qDestroyTask(qTaskInfo_t qTaskHandle) {

×

UNCOV

1066

  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)qTaskHandle;

×

UNCOV

1067

  if (pTaskInfo == NULL) {

×

UNCOV

1068

    return;

×

UNCOV

1071

  if (pTaskInfo->pRoot != NULL) {

×

UNCOV

1072

    qDebug("%s execTask completed, numOfRows:%" PRId64, GET_TASKID(pTaskInfo), pTaskInfo->pRoot->resultInfo.totalRows);

×

1074

    qDebug("%s execTask completed", GET_TASKID(pTaskInfo));

×

UNCOV

1077

  printTaskExecCostInLog(pTaskInfo);  // print the query cost summary

×

UNCOV

1078

  doDestroyTask(pTaskInfo);

×

UNCOV

1081

int32_t qGetExplainExecInfo(qTaskInfo_t tinfo, SArray* pExecInfoList) {

×

UNCOV

1082

  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;

×

UNCOV

1083

  return getOperatorExplainExecInfo(pTaskInfo->pRoot, pExecInfoList);

×

UNCOV

1086

void qExtractStreamScanner(qTaskInfo_t tinfo, void** scanner) {

×

UNCOV

1087

  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;

×

UNCOV

1088

  SOperatorInfo* pOperator = pTaskInfo->pRoot;

×

UNCOV

1090

  while (1) {

×

UNCOV

1091

    uint16_t type = pOperator->operatorType;

×

UNCOV

1092

    if (type == QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) {

×

UNCOV

1093

      *scanner = pOperator->info;

×

UNCOV

1094

      break;

×

UNCOV

1096

      pOperator = pOperator->pDownstream[0];

×

UNCOV

1099

×

UNCOV

1101

int32_t qStreamSourceScanParamForHistoryScanStep1(qTaskInfo_t tinfo, SVersionRange* pVerRange, STimeWindow* pWindow) {

×

UNCOV

1102

  int32_t        code = TSDB_CODE_SUCCESS;

×

UNCOV

1103

  int32_t        lino = 0;

×

UNCOV

1104

  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;

×

UNCOV

1105

  QUERY_CHECK_CONDITION((pTaskInfo->execModel == OPTR_EXEC_MODEL_STREAM), code, lino, _end,

×

UNCOV

1108

  SStreamTaskInfo* pStreamInfo = &pTaskInfo->streamInfo;

×

UNCOV

1110

  pStreamInfo->fillHistoryVer = *pVerRange;

×

UNCOV

1111

  pStreamInfo->fillHistoryWindow = *pWindow;

×

UNCOV

1112

  pStreamInfo->recoverStep = STREAM_RECOVER_STEP__PREPARE1;

×

UNCOV

1114

  qDebug("%s step 1. set param for stream scanner for scan-history data, verRange:%" PRId64 " - %" PRId64

×

UNCOV

1118

_end:

×

UNCOV

1119

  if (code != TSDB_CODE_SUCCESS) {

×

1120

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

×

UNCOV

1122

  return code;

×

UNCOV

1125

int32_t qStreamSourceScanParamForHistoryScanStep2(qTaskInfo_t tinfo, SVersionRange* pVerRange, STimeWindow* pWindow) {

×

UNCOV

1126

  int32_t        code = TSDB_CODE_SUCCESS;

×

UNCOV

1127

  int32_t        lino = 0;

×

UNCOV

1128

  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;

×

UNCOV

1129

  QUERY_CHECK_CONDITION((pTaskInfo->execModel == OPTR_EXEC_MODEL_STREAM), code, lino, _end,

×

UNCOV

1132

  SStreamTaskInfo* pStreamInfo = &pTaskInfo->streamInfo;

×

UNCOV

1134

  pStreamInfo->fillHistoryVer = *pVerRange;

×

UNCOV

1135

  pStreamInfo->fillHistoryWindow = *pWindow;

×

UNCOV

1136

  pStreamInfo->recoverStep = STREAM_RECOVER_STEP__PREPARE2;

×

UNCOV

1138

  qDebug("%s step 2. set param for stream scanner scan wal, verRange:%" PRId64 "-%" PRId64 ", window:%" PRId64

×

UNCOV

1142

_end:

×

UNCOV

1143

  if (code != TSDB_CODE_SUCCESS) {

×

1144

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

×

UNCOV

1146

  return code;

×

1149

int32_t qStreamRecoverFinish(qTaskInfo_t tinfo) {

×

1150

  int32_t        code = TSDB_CODE_SUCCESS;

×

1151

  int32_t        lino = 0;

×

1152

  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;

×

1153

  QUERY_CHECK_CONDITION((pTaskInfo->execModel == OPTR_EXEC_MODEL_STREAM), code, lino, _end,

×

1155

  pTaskInfo->streamInfo.recoverStep = STREAM_RECOVER_STEP__NONE;

×

1157

_end:

×

1158

  if (code != TSDB_CODE_SUCCESS) {

×

1159

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

×

1161

  return code;

×

UNCOV

1164

static int32_t getOpratorIntervalInfo(SOperatorInfo* pOperator, int64_t* pWaterMark, SInterval* pInterval, STimeWindow* pLastWindow) {

×

UNCOV

1165

  if (pOperator->operatorType != QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) {

×

UNCOV

1166

    return getOpratorIntervalInfo(pOperator->pDownstream[0], pWaterMark, pInterval, pLastWindow);

×

UNCOV

1168

  SStreamScanInfo* pScanOp = (SStreamScanInfo*) pOperator->info;

×

UNCOV

1169

  *pWaterMark = pScanOp->twAggSup.waterMark;

×

UNCOV

1170

  *pInterval = pScanOp->interval;

×

UNCOV

1171

  *pLastWindow = pScanOp->lastScanRange;

×

UNCOV

1172

  return TSDB_CODE_SUCCESS;

×

UNCOV

1175

int32_t qGetStreamIntervalExecInfo(qTaskInfo_t tinfo, int64_t* pWaterMark, SInterval* pInterval, STimeWindow* pLastWindow) {

×

UNCOV

1176

  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;

×

UNCOV

1177

  SOperatorInfo* pOperator = pTaskInfo->pRoot;

×

UNCOV

1178

  return getOpratorIntervalInfo(pOperator, pWaterMark, pInterval, pLastWindow);

×

UNCOV

1181

int32_t qSetStreamOperatorOptionForScanHistory(qTaskInfo_t tinfo) {

×

UNCOV

1182

  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;

×

UNCOV

1183

  SOperatorInfo* pOperator = pTaskInfo->pRoot;

×

UNCOV

1185

  while (1) {

×

UNCOV

1186

    int32_t type = pOperator->operatorType;

×

UNCOV

1187

    if (type == QUERY_NODE_PHYSICAL_PLAN_STREAM_INTERVAL || type == QUERY_NODE_PHYSICAL_PLAN_STREAM_SEMI_INTERVAL ||

×

UNCOV

1188

        type == QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_INTERVAL ||

×

UNCOV

1189

        type == QUERY_NODE_PHYSICAL_PLAN_STREAM_MID_INTERVAL) {

×

UNCOV

1190

      SStreamIntervalOperatorInfo* pInfo = pOperator->info;

×

UNCOV

1191

      STimeWindowAggSupp*          pSup = &pInfo->twAggSup;

×

UNCOV

1193

      qInfo("save stream param for interval: %d,  %" PRId64, pSup->calTrigger, pSup->deleteMark);

×

UNCOV

1195

      pSup->calTriggerSaved = pSup->calTrigger;

×

UNCOV

1196

      pSup->deleteMarkSaved = pSup->deleteMark;

×

UNCOV

1197

      pSup->calTrigger = STREAM_TRIGGER_AT_ONCE;

×

UNCOV

1198

      pSup->deleteMark = INT64_MAX;

×

UNCOV

1199

      pInfo->ignoreExpiredDataSaved = pInfo->ignoreExpiredData;

×

UNCOV

1200

      pInfo->ignoreExpiredData = false;

×

UNCOV

1201

    } else if (type == QUERY_NODE_PHYSICAL_PLAN_STREAM_SESSION ||

×

UNCOV

1202

               type == QUERY_NODE_PHYSICAL_PLAN_STREAM_SEMI_SESSION ||

×

UNCOV

1203

               type == QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_SESSION) {

×

UNCOV

1204

      SStreamSessionAggOperatorInfo* pInfo = pOperator->info;

×

UNCOV

1205

      STimeWindowAggSupp*            pSup = &pInfo->twAggSup;

×

UNCOV

1207

      qInfo("save stream param for session: %d,  %" PRId64, pSup->calTrigger, pSup->deleteMark);

×

UNCOV

1209

      pSup->calTriggerSaved = pSup->calTrigger;

×

UNCOV

1210

      pSup->deleteMarkSaved = pSup->deleteMark;

×

UNCOV

1211

      pSup->calTrigger = STREAM_TRIGGER_AT_ONCE;

×

UNCOV

1212

      pSup->deleteMark = INT64_MAX;

×

UNCOV

1213

      pInfo->ignoreExpiredDataSaved = pInfo->ignoreExpiredData;

×

UNCOV

1214

      pInfo->ignoreExpiredData = false;

×

UNCOV

1215

    } else if (type == QUERY_NODE_PHYSICAL_PLAN_STREAM_STATE) {

×

UNCOV

1216

      SStreamStateAggOperatorInfo* pInfo = pOperator->info;

×

UNCOV

1217

      STimeWindowAggSupp*          pSup = &pInfo->twAggSup;

×

UNCOV

1219

      qInfo("save stream param for state: %d,  %" PRId64, pSup->calTrigger, pSup->deleteMark);

×

UNCOV

1221

      pSup->calTriggerSaved = pSup->calTrigger;

×

UNCOV

1222

      pSup->deleteMarkSaved = pSup->deleteMark;

×

UNCOV

1223

      pSup->calTrigger = STREAM_TRIGGER_AT_ONCE;

×

UNCOV

1224

      pSup->deleteMark = INT64_MAX;

×

UNCOV

1225

      pInfo->ignoreExpiredDataSaved = pInfo->ignoreExpiredData;

×

UNCOV

1226

      pInfo->ignoreExpiredData = false;

×

UNCOV

1227

    } else if (type == QUERY_NODE_PHYSICAL_PLAN_STREAM_EVENT) {

×

UNCOV

1228

      SStreamEventAggOperatorInfo* pInfo = pOperator->info;

×

UNCOV

1229

      STimeWindowAggSupp*          pSup = &pInfo->twAggSup;

×

UNCOV

1231

      qInfo("save stream param for state: %d,  %" PRId64, pSup->calTrigger, pSup->deleteMark);

×

UNCOV

1233

      pSup->calTriggerSaved = pSup->calTrigger;

×

UNCOV

1234

      pSup->deleteMarkSaved = pSup->deleteMark;

×

UNCOV

1235

      pSup->calTrigger = STREAM_TRIGGER_AT_ONCE;

×

UNCOV

1236

      pSup->deleteMark = INT64_MAX;

×

UNCOV

1237

      pInfo->ignoreExpiredDataSaved = pInfo->ignoreExpiredData;

×

UNCOV

1238

      pInfo->ignoreExpiredData = false;

×

UNCOV

1239

    } else if (type == QUERY_NODE_PHYSICAL_PLAN_STREAM_COUNT) {

×

UNCOV

1240

      SStreamCountAggOperatorInfo* pInfo = pOperator->info;

×

UNCOV

1241

      STimeWindowAggSupp*          pSup = &pInfo->twAggSup;

×

UNCOV

1243

      qInfo("save stream param for state: %d,  %" PRId64, pSup->calTrigger, pSup->deleteMark);

×

UNCOV

1245

      pSup->calTriggerSaved = pSup->calTrigger;

×

UNCOV

1246

      pSup->deleteMarkSaved = pSup->deleteMark;

×

UNCOV

1247

      pSup->calTrigger = STREAM_TRIGGER_AT_ONCE;

×

UNCOV

1248

      pSup->deleteMark = INT64_MAX;

×

UNCOV

1249

      pInfo->ignoreExpiredDataSaved = pInfo->ignoreExpiredData;

×

UNCOV

1250

      pInfo->ignoreExpiredData = false;

×

UNCOV

1251

      qInfo("save stream task:%s, param for state: %d", GET_TASKID(pTaskInfo), pInfo->ignoreExpiredData);

×

UNCOV

1252

    } else if (type == QUERY_NODE_PHYSICAL_PLAN_STREAM_INTERP_FUNC) {

×

UNCOV

1253

      SStreamTimeSliceOperatorInfo* pInfo = pOperator->info;

×

UNCOV

1254

      STimeWindowAggSupp*           pSup = &pInfo->twAggSup;

×

UNCOV

1256

      qInfo("save stream param for state: %d,  %" PRId64, pSup->calTrigger, pSup->deleteMark);

×

UNCOV

1258

      pSup->calTriggerSaved = pSup->calTrigger;

×

UNCOV

1259

      pSup->deleteMarkSaved = pSup->deleteMark;

×

UNCOV

1260

      pSup->calTrigger = STREAM_TRIGGER_AT_ONCE;

×

UNCOV

1261

      pSup->deleteMark = INT64_MAX;

×

UNCOV

1262

      pInfo->ignoreExpiredDataSaved = pInfo->ignoreExpiredData;

×

UNCOV

1263

      pInfo->ignoreExpiredData = false;

×

UNCOV

1264

      qInfo("save stream task:%s, param for state: %d", GET_TASKID(pTaskInfo), pInfo->ignoreExpiredData);

×

UNCOV

1268

    if (pOperator->numOfDownstream != 1 || pOperator->pDownstream[0] == NULL) {

×

UNCOV

1269

      if (pOperator->numOfDownstream > 1) {

×

1270

        qError("unexpected stream, multiple downstream");

×

1271

        return -1;

×

UNCOV

1273

      return 0;

×

UNCOV

1275

      pOperator = pOperator->pDownstream[0];

×

UNCOV

1282

bool qStreamScanhistoryFinished(qTaskInfo_t tinfo) {

×

UNCOV

1283

  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;

×

UNCOV

1284

  return pTaskInfo->streamInfo.recoverScanFinished;

×

UNCOV

1287

int32_t qStreamInfoResetTimewindowFilter(qTaskInfo_t tinfo) {

×

UNCOV

1288

  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;

×

UNCOV

1289

  STimeWindow*   pWindow = &pTaskInfo->streamInfo.fillHistoryWindow;

×

UNCOV

1291

  qDebug("%s remove timeWindow filter:%" PRId64 "-%" PRId64 ", set new window:%" PRId64 "-%" PRId64,

×

UNCOV

1294

  pWindow->skey = INT64_MIN;

×

UNCOV

1295

  pWindow->ekey = INT64_MAX;

×

UNCOV

1296

  return 0;

×

UNCOV

1299

void* qExtractReaderFromStreamScanner(void* scanner) {

×

UNCOV

1300

  SStreamScanInfo* pInfo = scanner;

×

UNCOV

1301

  return (void*)pInfo->tqReader;

×

UNCOV

1304

const SSchemaWrapper* qExtractSchemaFromTask(qTaskInfo_t tinfo) {

×

UNCOV

1305

  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;

×

UNCOV

1306

  return pTaskInfo->streamInfo.schema;

×

UNCOV

1309

const char* qExtractTbnameFromTask(qTaskInfo_t tinfo) {

×

UNCOV

1310

  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;

×

UNCOV

1311

  return pTaskInfo->streamInfo.tbName;

×

UNCOV

1314

SMqBatchMetaRsp* qStreamExtractMetaMsg(qTaskInfo_t tinfo) {

×

UNCOV

1315

  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;

×

UNCOV

1316

  return &pTaskInfo->streamInfo.btMetaRsp;

×

UNCOV

1319

int32_t qStreamExtractOffset(qTaskInfo_t tinfo, STqOffsetVal* pOffset) {

×

UNCOV

1320

  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;

×

UNCOV

1321

  tOffsetCopy(pOffset, &pTaskInfo->streamInfo.currentOffset);

×

UNCOV

1322

  return 0;

×

UNCOV

1330

int32_t initQueryTableDataCondForTmq(SQueryTableDataCond* pCond, SSnapContext* sContext, SMetaTableInfo* pMtInfo) {

×

UNCOV

1331

  memset(pCond, 0, sizeof(SQueryTableDataCond));

×

UNCOV

1332

  pCond->order = TSDB_ORDER_ASC;

×

UNCOV

1333

  pCond->numOfCols = pMtInfo->schema->nCols;

×

UNCOV

1334

  pCond->colList = taosMemoryCalloc(pCond->numOfCols, sizeof(SColumnInfo));

×

UNCOV

1335

  pCond->pSlotList = taosMemoryMalloc(sizeof(int32_t) * pCond->numOfCols);

×

UNCOV

1336

  if (pCond->colList == NULL || pCond->pSlotList == NULL) {

×

1337

    taosMemoryFreeClear(pCond->colList);

×

1338

    taosMemoryFreeClear(pCond->pSlotList);

×

1339

    return terrno;

×

UNCOV

1342

  pCond->twindows = TSWINDOW_INITIALIZER;

×

UNCOV

1343

  pCond->suid = pMtInfo->suid;

×

UNCOV

1344

  pCond->type = TIMEWINDOW_RANGE_CONTAINED;

×

UNCOV

1345

  pCond->startVersion = -1;

×

UNCOV

1346

  pCond->endVersion = sContext->snapVersion;

×

UNCOV

1348

  for (int32_t i = 0; i < pCond->numOfCols; ++i) {

×

UNCOV

1349

    SColumnInfo* pColInfo = &pCond->colList[i];

×

UNCOV

1350

    pColInfo->type = pMtInfo->schema->pSchema[i].type;

×

UNCOV

1351

    pColInfo->bytes = pMtInfo->schema->pSchema[i].bytes;

×

UNCOV

1352

    pColInfo->colId = pMtInfo->schema->pSchema[i].colId;

×

UNCOV

1353

    pColInfo->pk = pMtInfo->schema->pSchema[i].flags & COL_IS_KEY;

×

UNCOV

1355

    pCond->pSlotList[i] = i;

×

UNCOV

1358

  return TSDB_CODE_SUCCESS;

×

UNCOV

1361

void qStreamSetOpen(qTaskInfo_t tinfo) {

×

UNCOV

1362

  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;

×

UNCOV

1363

  SOperatorInfo* pOperator = pTaskInfo->pRoot;

×

UNCOV

1364

  pOperator->status = OP_NOT_OPENED;

×

UNCOV

1365

×

UNCOV

1367

void qStreamSetSourceExcluded(qTaskInfo_t tinfo, int8_t sourceExcluded) {

×

UNCOV

1368

  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;

×

UNCOV

1369

  pTaskInfo->streamInfo.sourceExcluded = sourceExcluded;

×

UNCOV

1370

×

UNCOV

1372

int32_t qStreamPrepareScan(qTaskInfo_t tinfo, STqOffsetVal* pOffset, int8_t subType) {

×

UNCOV

1373

  int32_t        code = TSDB_CODE_SUCCESS;

×

UNCOV

1374

  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;

×

UNCOV

1375

  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;

×

UNCOV

1377

  SOperatorInfo* pOperator = pTaskInfo->pRoot;

×

UNCOV

1378

  const char*    id = GET_TASKID(pTaskInfo);

×

UNCOV

1380

  if (subType == TOPIC_SUB_TYPE__COLUMN && pOffset->type == TMQ_OFFSET__LOG) {

×

UNCOV

1381

    code = extractOperatorInTree(pOperator, QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN, id, &pOperator);

×

UNCOV

1382

    if (pOperator == NULL || code != 0) {

×

1383

      return code;

×

UNCOV

1386

    SStreamScanInfo* pInfo = pOperator->info;

×

UNCOV

1387

    SStoreTqReader*  pReaderAPI = &pTaskInfo->storageAPI.tqReaderFn;

×

UNCOV

1388

    SWalReader*      pWalReader = pReaderAPI->tqReaderGetWalReader(pInfo->tqReader);

×

UNCOV

1389

    walReaderVerifyOffset(pWalReader, pOffset);

×

UNCOV

1392

  if (tOffsetEqual(pOffset, &pTaskInfo->streamInfo.currentOffset)) {

×

UNCOV

1393

    return 0;

×

UNCOV

1396

  if (subType == TOPIC_SUB_TYPE__COLUMN) {

×

UNCOV

1397

    code = extractOperatorInTree(pOperator, QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN, id, &pOperator);

×

UNCOV

1398

    if (pOperator == NULL || code != 0) {

×

1399

      return code;

×

UNCOV

1402

    SStreamScanInfo* pInfo = pOperator->info;

×

UNCOV

1403

    STableScanInfo*  pScanInfo = pInfo->pTableScanOp->info;

×

UNCOV

1404

    STableScanBase*  pScanBaseInfo = &pScanInfo->base;

×

UNCOV

1405

    STableListInfo*  pTableListInfo = pScanBaseInfo->pTableListInfo;

×

UNCOV

1407

    if (pOffset->type == TMQ_OFFSET__LOG) {

×

UNCOV

1408

      pTaskInfo->storageAPI.tsdReader.tsdReaderClose(pScanBaseInfo->dataReader);

×

UNCOV

1409

      pScanBaseInfo->dataReader = NULL;

×

UNCOV

1411

      SStoreTqReader* pReaderAPI = &pTaskInfo->storageAPI.tqReaderFn;

×

UNCOV

1412

      SWalReader*     pWalReader = pReaderAPI->tqReaderGetWalReader(pInfo->tqReader);

×

UNCOV

1413

      walReaderVerifyOffset(pWalReader, pOffset);

×

UNCOV

1414

      code = pReaderAPI->tqReaderSeek(pInfo->tqReader, pOffset->version, id);

×

UNCOV

1415

      if (code < 0) {

×

UNCOV

1416

        qError("tqReaderSeek failed ver:%" PRId64 ", %s", pOffset->version, id);

×

UNCOV

1417

        return code;

×

UNCOV

1419

    } else if (pOffset->type == TMQ_OFFSET__SNAPSHOT_DATA) {

×

UNCOV

1422

      int64_t uid = pOffset->uid;

×

UNCOV

1423

      int64_t ts = pOffset->ts;

×

UNCOV

1424

      int32_t index = 0;

×

UNCOV

1427

      taosRLockLatch(&pTaskInfo->lock);

×

UNCOV

1428

      int32_t numOfTables = 0;

×

UNCOV

1429

      code = tableListGetSize(pTableListInfo, &numOfTables);

×

UNCOV

1430

      if (code != TSDB_CODE_SUCCESS) {

×

1431

        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));

×

1432

        taosRUnLockLatch(&pTaskInfo->lock);

×

UNCOV

1433

        return code;

×

UNCOV

1436

      if (uid == 0) {

×

UNCOV

1437

        if (numOfTables != 0) {

×

UNCOV

1438

          STableKeyInfo* tmp = tableListGetInfo(pTableListInfo, 0);

×

UNCOV

1439

          if (!tmp) {

×

1440

            qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));

×

1441

            taosRUnLockLatch(&pTaskInfo->lock);

×

1442

            return terrno;

×

UNCOV

1444

          if (tmp) uid = tmp->uid;

×

UNCOV

1445

          ts = INT64_MIN;

×

UNCOV

1446

          pScanInfo->currentTable = 0;

×

UNCOV

1448

          taosRUnLockLatch(&pTaskInfo->lock);

×

UNCOV

1449

          qError("no table in table list, %s", id);

×

UNCOV

1450

          return TSDB_CODE_TMQ_NO_TABLE_QUALIFIED;

×

UNCOV

1453

      pTaskInfo->storageAPI.tqReaderFn.tqSetTablePrimaryKey(pInfo->tqReader, uid);

×

UNCOV

1455

      qDebug("switch to table uid:%" PRId64 " ts:%" PRId64 "% " PRId64 " rows returned", uid, ts,

×

UNCOV

1457

      pInfo->pTableScanOp->resultInfo.totalRows = 0;

×

UNCOV

1462

      index = tableListFind(pTableListInfo, uid, 0);

×

UNCOV

1463

      taosRUnLockLatch(&pTaskInfo->lock);

×

UNCOV

1465

      if (index >= 0) {

×

UNCOV

1466

        pScanInfo->currentTable = index;

×

1468

        qError("vgId:%d uid:%" PRIu64 " not found in table list, total:%d, index:%d %s", pTaskInfo->id.vgId, uid,

×

1470

        return TSDB_CODE_TMQ_NO_TABLE_QUALIFIED;

×

UNCOV

1473

      STableKeyInfo keyInfo = {.uid = uid};

×

UNCOV

1474

      int64_t       oldSkey = pScanBaseInfo->cond.twindows.skey;

×

UNCOV

1477

      if (pTaskInfo->storageAPI.tqReaderFn.tqGetTablePrimaryKey(pInfo->tqReader)) {

×

1478

        pScanBaseInfo->cond.twindows.skey = ts;

×

UNCOV

1480

        pScanBaseInfo->cond.twindows.skey = ts + 1;

×

UNCOV

1482

      pScanInfo->scanTimes = 0;

×

UNCOV

1484

      if (pScanBaseInfo->dataReader == NULL) {

×

UNCOV

1485

        code = pTaskInfo->storageAPI.tsdReader.tsdReaderOpen(pScanBaseInfo->readHandle.vnode, &pScanBaseInfo->cond,

×

UNCOV

1487

                                                             (void**)&pScanBaseInfo->dataReader, id, NULL);

×

UNCOV

1488

        if (code != TSDB_CODE_SUCCESS) {

×

1489

          qError("prepare read tsdb snapshot failed, uid:%" PRId64 ", code:%s %s", pOffset->uid, tstrerror(code), id);

×

1490

          return code;

×

UNCOV

1493

        qDebug("tsdb reader created with offset(snapshot) uid:%" PRId64 " ts:%" PRId64 " table index:%d, total:%d, %s",

×

UNCOV

1496

        code = pTaskInfo->storageAPI.tsdReader.tsdSetQueryTableList(pScanBaseInfo->dataReader, &keyInfo, 1);

×

UNCOV

1497

        if (code != TSDB_CODE_SUCCESS) {

×

1498

          qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));

×

1499

          return code;

×

UNCOV

1502

        code = pTaskInfo->storageAPI.tsdReader.tsdReaderResetStatus(pScanBaseInfo->dataReader, &pScanBaseInfo->cond);

×

UNCOV

1503

        if (code != TSDB_CODE_SUCCESS) {

×

1504

          qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));

×

1505

          return code;

×

UNCOV

1507

        qDebug("tsdb reader offset seek snapshot to uid:%" PRId64 " ts %" PRId64 "  table index:%d numOfTable:%d, %s",

×

UNCOV

1512

      pScanBaseInfo->cond.twindows.skey = oldSkey;

×

1514

      qError("invalid pOffset->type:%d, %s", pOffset->type, id);

×

1515

      return TSDB_CODE_PAR_INTERNAL_ERROR;

×

UNCOV

1519

    if (pOffset->type == TMQ_OFFSET__SNAPSHOT_DATA) {

×

UNCOV

1520

      SStreamRawScanInfo* pInfo = pOperator->info;

×

UNCOV

1521

      SSnapContext*       sContext = pInfo->sContext;

×

UNCOV

1522

      SOperatorInfo*      p = NULL;

×

UNCOV

1524

      code = extractOperatorInTree(pOperator, QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN, id, &p);

×

UNCOV

1525

      if (code != 0) {

×

1526

        return code;

×

UNCOV

1529

      STableListInfo* pTableListInfo = ((SStreamRawScanInfo*)(p->info))->pTableListInfo;

×

UNCOV

1531

      if (pAPI->snapshotFn.setForSnapShot(sContext, pOffset->uid) != 0) {

×

1532

        qError("setDataForSnapShot error. uid:%" PRId64 " , %s", pOffset->uid, id);

×

1533

        return TSDB_CODE_PAR_INTERNAL_ERROR;

×

UNCOV

1536

      SMetaTableInfo mtInfo = {0};

×

UNCOV

1537

      code = pTaskInfo->storageAPI.snapshotFn.getMetaTableInfoFromSnapshot(sContext, &mtInfo);

×

UNCOV

1538

      if (code != 0) {

×

1539

        return code;

×

UNCOV

1541

      pTaskInfo->storageAPI.tsdReader.tsdReaderClose(pInfo->dataReader);

×

UNCOV

1542

      pInfo->dataReader = NULL;

×

UNCOV

1544

      cleanupQueryTableDataCond(&pTaskInfo->streamInfo.tableCond);

×

UNCOV

1545

      tableListClear(pTableListInfo);

×

UNCOV

1547

      if (mtInfo.uid == 0) {

×

UNCOV

1548

        tDeleteSchemaWrapper(mtInfo.schema);

×

UNCOV

1549

        goto end;  // no data

×

UNCOV

1552

      pAPI->snapshotFn.taosXSetTablePrimaryKey(sContext, mtInfo.uid);

×

UNCOV

1553

      code = initQueryTableDataCondForTmq(&pTaskInfo->streamInfo.tableCond, sContext, &mtInfo);

×

UNCOV

1554

      if (code != TSDB_CODE_SUCCESS) {

×

1555

        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));

×

1556

        tDeleteSchemaWrapper(mtInfo.schema);

×

1557

        return code;

×

UNCOV

1559

      if (pAPI->snapshotFn.taosXGetTablePrimaryKey(sContext)) {

×

1560

        pTaskInfo->streamInfo.tableCond.twindows.skey = pOffset->ts;

×

UNCOV

1562

        pTaskInfo->streamInfo.tableCond.twindows.skey = pOffset->ts + 1;

×

UNCOV

1565

      code = tableListAddTableInfo(pTableListInfo, mtInfo.uid, 0);

×

UNCOV

1566

      if (code != TSDB_CODE_SUCCESS) {

×

1567

        tDeleteSchemaWrapper(mtInfo.schema);

×

1568

        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));

×

1569

        return code;

×

UNCOV

1572

      STableKeyInfo* pList = tableListGetInfo(pTableListInfo, 0);

×

UNCOV

1573

      if (!pList) {

×

1574

        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));

×

1575

        tDeleteSchemaWrapper(mtInfo.schema);

×

1576

        return code;

×

UNCOV

1578

      int32_t size = 0;

×

UNCOV

1579

      code = tableListGetSize(pTableListInfo, &size);

×

UNCOV

1580

      if (code != TSDB_CODE_SUCCESS) {

×

1581

        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));

×

1582

        tDeleteSchemaWrapper(mtInfo.schema);

×

1583

        return code;

×

UNCOV

1586

      code = pTaskInfo->storageAPI.tsdReader.tsdReaderOpen(pInfo->vnode, &pTaskInfo->streamInfo.tableCond, pList, size,

×

UNCOV

1587

                                                           NULL, (void**)&pInfo->dataReader, NULL, NULL);

×

UNCOV

1588

      if (code != TSDB_CODE_SUCCESS) {

×

1589

        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));

×

1590

        tDeleteSchemaWrapper(mtInfo.schema);

×

1591

        return code;

×

UNCOV

1594

      cleanupQueryTableDataCond(&pTaskInfo->streamInfo.tableCond);

×

UNCOV

1595

      tstrncpy(pTaskInfo->streamInfo.tbName, mtInfo.tbName, TSDB_TABLE_NAME_LEN);

×

UNCOV

1597

      tDeleteSchemaWrapper(pTaskInfo->streamInfo.schema);

×

UNCOV

1598

      pTaskInfo->streamInfo.schema = mtInfo.schema;

×

UNCOV

1600

      qDebug("tmqsnap qStreamPrepareScan snapshot data uid:%" PRId64 " ts %" PRId64 " %s", mtInfo.uid, pOffset->ts, id);

×

UNCOV

1601

    } else if (pOffset->type == TMQ_OFFSET__SNAPSHOT_META) {

×

UNCOV

1602

      SStreamRawScanInfo* pInfo = pOperator->info;

×

UNCOV

1603

      SSnapContext*       sContext = pInfo->sContext;

×

UNCOV

1604

      code = pTaskInfo->storageAPI.snapshotFn.setForSnapShot(sContext, pOffset->uid);

×

UNCOV

1605

      if (code != 0) {

×

1606

        qError("setForSnapShot error. uid:%" PRIu64 " ,version:%" PRId64, pOffset->uid, pOffset->version);

×

1607

        return code;

×

UNCOV

1609

      qDebug("tmqsnap qStreamPrepareScan snapshot meta uid:%" PRId64 " ts %" PRId64 " %s", pOffset->uid, pOffset->ts,

×

UNCOV

1611

    } else if (pOffset->type == TMQ_OFFSET__LOG) {

×

UNCOV

1612

      SStreamRawScanInfo* pInfo = pOperator->info;

×

UNCOV

1613

      pTaskInfo->storageAPI.tsdReader.tsdReaderClose(pInfo->dataReader);

×

UNCOV

1614

      pInfo->dataReader = NULL;

×

UNCOV

1615

      qDebug("tmqsnap qStreamPrepareScan snapshot log, %s", id);

×

UNCOV

1619

end:

×

UNCOV

1620

  tOffsetCopy(&pTaskInfo->streamInfo.currentOffset, pOffset);

×

UNCOV

1621

  return 0;

×

UNCOV

1624

void qProcessRspMsg(void* parent, SRpcMsg* pMsg, SEpSet* pEpSet) {

×

UNCOV

1625

  SMsgSendInfo* pSendInfo = (SMsgSendInfo*)pMsg->info.ahandle;

×

UNCOV

1626

  if (pMsg->info.ahandle == NULL) {

×

1627

    qError("pMsg->info.ahandle is NULL");

×

1628

    return;

×

UNCOV

1631

  SDataBuf buf = {.len = pMsg->contLen, .pData = NULL};

×

UNCOV

1633

  if (pMsg->contLen > 0) {

×

UNCOV

1634

    buf.pData = taosMemoryCalloc(1, pMsg->contLen);

×

UNCOV

1635

    if (buf.pData == NULL) {

×

1636

      pMsg->code = terrno;

×

UNCOV

1638

      memcpy(buf.pData, pMsg->pCont, pMsg->contLen);

×

UNCOV

1642

  (void)pSendInfo->fp(pSendInfo->param, &buf, pMsg->code);

×

UNCOV

1643

  rpcFreeCont(pMsg->pCont);

×

UNCOV

1644

  destroySendMsgInfo(pSendInfo);

×

1647

SArray* qGetQueriedTableListInfo(qTaskInfo_t tinfo) {

×

1648

  int32_t        code = TSDB_CODE_SUCCESS;

×

1649

  int32_t        lino = 0;

×

1650

  SExecTaskInfo* pTaskInfo = tinfo;

×

1651

  SArray*        plist = NULL;

×

1653

  code = getTableListInfo(pTaskInfo, &plist);

×

1654

  if (code || plist == NULL) {

×

1655

    return NULL;

×

1659

  STableListInfo* pTableListInfo = taosArrayGetP(plist, 0);

×

1661

  SArray* pUidList = taosArrayInit(10, sizeof(uint64_t));

×

1662

  QUERY_CHECK_NULL(pUidList, code, lino, _end, terrno);

×

1664

  int32_t numOfTables = 0;

×

1665

  code = tableListGetSize(pTableListInfo, &numOfTables);

×

1666

  QUERY_CHECK_CODE(code, lino, _end);

×

1668

  for (int32_t i = 0; i < numOfTables; ++i) {

×

1669

    STableKeyInfo* pKeyInfo = tableListGetInfo(pTableListInfo, i);

×

1670

    QUERY_CHECK_NULL(pKeyInfo, code, lino, _end, terrno);

×

1671

    void* tmp = taosArrayPush(pUidList, &pKeyInfo->uid);

×

1672

    QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

×

1675

  taosArrayDestroy(plist);

×

1677

_end:

×

1678

  if (code != TSDB_CODE_SUCCESS) {

×

1679

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

×

1680

    T_LONG_JMP(pTaskInfo->env, code);

×

1682

  return pUidList;

×

UNCOV

1685

static int32_t extractTableList(SArray* pList, const SOperatorInfo* pOperator) {

×

UNCOV

1686

  int32_t        code = TSDB_CODE_SUCCESS;

×

UNCOV

1687

  int32_t        lino = 0;

×

UNCOV

1688

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

×

UNCOV

1690

  if (pOperator->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) {

×

1691

    SStreamScanInfo* pScanInfo = pOperator->info;

×

1692

    STableScanInfo*  pTableScanInfo = pScanInfo->pTableScanOp->info;

×

1694

    void* tmp = taosArrayPush(pList, &pTableScanInfo->base.pTableListInfo);

×

1695

    QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

×

UNCOV

1696

  } else if (pOperator->operatorType == QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN) {

×

UNCOV

1697

    STableScanInfo* pScanInfo = pOperator->info;

×

UNCOV

1699

    void* tmp = taosArrayPush(pList, &pScanInfo->base.pTableListInfo);

×

UNCOV

1700

    QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

×

UNCOV

1702

    if (pOperator->pDownstream != NULL && pOperator->pDownstream[0] != NULL) {

×

UNCOV

1703

      code = extractTableList(pList, pOperator->pDownstream[0]);

×

1707

_end:

×

UNCOV

1708

  if (code != TSDB_CODE_SUCCESS) {

×

1709

    qError("%s %s failed at line %d since %s", pTaskInfo->id.str, __func__, lino, tstrerror(code));

×

UNCOV

1711

  return code;

×

UNCOV

1714

int32_t getTableListInfo(const SExecTaskInfo* pTaskInfo, SArray** pList) {

×

UNCOV

1715

  if (pList == NULL) {

×

1716

    return TSDB_CODE_INVALID_PARA;

×

UNCOV

1719

  *pList = NULL;

×

UNCOV

1720

  SArray* pArray = taosArrayInit(0, POINTER_BYTES);

×

UNCOV

1721

  if (pArray == NULL) {

×

1722

    return terrno;

×

UNCOV

1725

  int32_t code = extractTableList(pArray, pTaskInfo->pRoot);

×

UNCOV

1726

  if (code == 0) {

×

UNCOV

1727

    *pList = pArray;

×

1729

    taosArrayDestroy(pArray);

×

UNCOV

1731

  return code;

×

UNCOV

1734

int32_t qStreamOperatorReleaseState(qTaskInfo_t tInfo) {

×

UNCOV

1735

  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tInfo;

×

UNCOV

1736

  if (pTaskInfo->pRoot->fpSet.releaseStreamStateFn != NULL) {

×

UNCOV

1737

    pTaskInfo->pRoot->fpSet.releaseStreamStateFn(pTaskInfo->pRoot);

×

UNCOV

1739

  return 0;

×

UNCOV

1742

int32_t qStreamOperatorReloadState(qTaskInfo_t tInfo) {

×

UNCOV

1743

  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tInfo;

×

UNCOV

1744

  if (pTaskInfo->pRoot->fpSet.reloadStreamStateFn != NULL) {

×

UNCOV

1745

    pTaskInfo->pRoot->fpSet.reloadStreamStateFn(pTaskInfo->pRoot);

×

UNCOV

1747

  return 0;

×

UNCOV

1750

void qResetTaskCode(qTaskInfo_t tinfo) {

×

UNCOV

1751

  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;

×

UNCOV

1753

  int32_t code = pTaskInfo->code;

×

UNCOV

1754

  pTaskInfo->code = 0;

×

UNCOV

1755

  qDebug("0x%" PRIx64 " reset task code to be success, prev:%s", pTaskInfo->id.taskId, tstrerror(code));

×

UNCOV

1756

×

taosdata / TDengine / #3621

Source File
Press 'n' to go to next uncovered line, 'b' for previous

taosdata / TDengine / #3621

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous