taosdata/TDengine | Build 5066 | source/dnode/vnode/src/vnd/vnodeStream.c | Coveralls

64

static int64_t getSuid(SStreamTriggerReaderInfo* sStreamReaderInfo, STableKeyInfo* pList) {

8,271,911✔

65

  int64_t suid = 0;

8,271,911✔

66

  if (!sStreamReaderInfo->isVtableStream) {

8,271,911✔

67

    suid = sStreamReaderInfo->suid;

7,729,483✔

68

    goto end;

7,729,238✔

71

  if (pList == NULL) {

542,428✔

75

  taosRLockLatch(&sStreamReaderInfo->lock);

542,428✔

76

  SStreamTableMapElement* element = taosHashGet(sStreamReaderInfo->vSetTableList.uIdMap, &pList->uid, LONG_BYTES);

542,428✔

77

  if (element != 0) {

542,428✔

78

    suid = element->table->groupId;

307,164✔

79

    taosRUnLockLatch(&sStreamReaderInfo->lock);

307,164✔

80

    goto end;

307,164✔

82

  taosRUnLockLatch(&sStreamReaderInfo->lock);

235,264✔

84

end:

8,271,911✔

85

  return suid;

8,271,911✔

88

static int64_t getSessionKey(int64_t session, int64_t type) { return (session | (type << 32)); }

8,375,892✔

90

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

2,539,452✔

91

  int16_t* c1 = (int16_t*)lp;

2,539,452✔

92

  int16_t* c2 = (int16_t*)rp;

2,539,452✔

94

  if (*c1 < *c2) {

2,539,452✔

95

    return -1;

2,521,972✔

96

  } else if (*c1 > *c2) {

17,480✔

97

    return 1;

17,480✔

103

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

2,526,252✔

104

  SSchema* c1 = (SSchema*)lp;

2,526,252✔

105

  SSchema* c2 = (SSchema*)rp;

2,526,252✔

107

  if (c1->colId < c2->colId) {

2,526,252✔

108

    return -1;

2,508,772✔

109

  } else if (c1->colId > c2->colId) {

17,480✔

110

    return 1;

17,480✔

116

static int32_t addColData(SSDataBlock* pResBlock, int32_t index, void* data) {

44,847,987✔

117

  SColumnInfoData* pSrc = taosArrayGet(pResBlock->pDataBlock, index);

44,847,987✔

118

  if (pSrc == NULL) {

44,856,571✔

122

  memcpy(pSrc->pData + pResBlock->info.rows * pSrc->info.bytes, data, pSrc->info.bytes);

44,856,571✔

123

  return 0;

44,847,387✔

126

static int32_t getTableDataInfo(SStreamReaderTaskInner* pTask, bool* hasNext) {

11,139,905✔

127

  int32_t code = pTask->storageApi->tsdReader.tsdNextDataBlock(pTask->pReader, hasNext);

11,139,905✔

128

  if (code != TSDB_CODE_SUCCESS) {

11,141,305✔

132

  return code;

11,139,855✔

135

static int32_t getTableData(SStreamReaderTaskInner* pTask, SSDataBlock** ppRes) {

1,897,720✔

136

  return pTask->storageApi->tsdReader.tsdReaderRetrieveDataBlock(pTask->pReader, ppRes);

1,897,720✔

139

static int32_t buildOTableInfoRsp(const SSTriggerOrigTableInfoRsp* rsp, void** data, size_t* size) {

146,369✔

140

  int32_t code = 0;

146,369✔

141

  int32_t lino = 0;

146,369✔

142

  void*   buf = NULL;

146,369✔

143

  int32_t len = tSerializeSTriggerOrigTableInfoRsp(NULL, 0, rsp);

146,369✔

144

  STREAM_CHECK_CONDITION_GOTO(len <= 0, TSDB_CODE_INVALID_PARA);

146,369✔

145

  buf = rpcMallocCont(len);

146,369✔

146

  STREAM_CHECK_NULL_GOTO(buf, terrno);

146,369✔

147

  int32_t actLen = tSerializeSTriggerOrigTableInfoRsp(buf, len, rsp);

146,369✔

148

  STREAM_CHECK_CONDITION_GOTO(actLen != len, TSDB_CODE_INVALID_PARA);

146,369✔

149

  *data = buf;

146,369✔

150

  *size = len;

146,369✔

151

  buf = NULL;

146,369✔

152

end:

146,369✔

153

  rpcFreeCont(buf);

146,369✔

154

  return code;

146,369✔

157

static bool ignoreMetaChange(int64_t tableListVer, int64_t ver) {

229,076✔

158

  stDebug("%s tableListVer:%" PRId64 " ver:%" PRId64, __func__, tableListVer, ver);

229,076✔

159

  return tableListVer >= ver;

229,076✔

162

static bool needReLoadTableList(SStreamTriggerReaderInfo* sStreamReaderInfo, int8_t tableType, int64_t suid, int64_t uid, bool isCalc){

5,836,904✔

163

  if ((tableType == TD_CHILD_TABLE || tableType == TD_VIRTUAL_CHILD_TABLE) &&

5,836,904✔

164

      sStreamReaderInfo->tableType == TD_SUPER_TABLE &&

2,721,216✔

165

      suid == sStreamReaderInfo->suid) {

1,005,099✔

166

    taosRLockLatch(&sStreamReaderInfo->lock);

10,697✔

167

    uint64_t gid = qStreamGetGroupIdFromOrigin(sStreamReaderInfo, uid);

10,697✔

168

    taosRUnLockLatch(&sStreamReaderInfo->lock);

10,697✔

169

    if (gid == (uint64_t)-1) return true;

10,697✔

171

  return false;

5,813,535✔

174

static bool uidInTableList(SStreamTriggerReaderInfo* sStreamReaderInfo, int64_t suid, int64_t uid, uint64_t* id){

14,951,921✔

175

  int32_t  ret = false;

14,951,921✔

176

  if (sStreamReaderInfo->tableType == TD_SUPER_TABLE) {

14,951,921✔

177

    if (suid != sStreamReaderInfo->suid) goto end;

8,647,104✔

178

    if (qStreamGetTableListNum(sStreamReaderInfo) == 0) goto end;

3,940,728✔

180

  *id = qStreamGetGroupIdFromOrigin(sStreamReaderInfo, uid);

10,245,023✔

181

  if (*id == -1) goto end;

10,247,907✔

182

  ret = true;

6,261,671✔

184

end:

14,962,766✔

185

  stTrace("%s ret:%d %p %p check suid:%" PRId64 " uid:%" PRId64 " gid:%"PRIu64, __func__, ret, sStreamReaderInfo, sStreamReaderInfo->tableList.gIdMap, suid, uid, *id);

14,962,766✔

186

  return ret;

14,957,828✔

189

static bool uidInTableListOrigin(SStreamTriggerReaderInfo* sStreamReaderInfo, int64_t suid, int64_t uid, uint64_t* id) {

37,446✔

190

  return uidInTableList(sStreamReaderInfo, suid, uid, id);

37,446✔

193

static bool uidInTableListSet(SStreamTriggerReaderInfo* sStreamReaderInfo, int64_t suid, int64_t uid, uint64_t* id, bool isCalc) {

65,793,074✔

194

  bool ret = false;

65,793,074✔

195

  taosRLockLatch(&sStreamReaderInfo->lock);

65,793,074✔

196

  if (sStreamReaderInfo->isVtableStream) {

65,805,547✔

197

    int64_t tmp[2] = {suid, uid};

50,884,328✔

198

    if(tSimpleHashGet(isCalc ? sStreamReaderInfo->uidHashCalc : sStreamReaderInfo->uidHashTrigger, tmp, sizeof(tmp)) != NULL) {

50,884,328✔

199

      *id = uid;

16,211,168✔

200

      ret = true;

16,211,445✔

203

    ret = uidInTableList(sStreamReaderInfo, suid, uid, id);

14,917,676✔

206

end:

65,808,467✔

207

  taosRUnLockLatch(&sStreamReaderInfo->lock);

65,808,467✔

208

  return ret;

65,758,340✔

211

static int32_t  qTransformStreamTableList(SStreamTriggerReaderInfo* sStreamReaderInfo, void* pTableListInfo, StreamTableListInfo* tableInfo){

362,375✔

212

  SArray* pList = qStreamGetTableListArray(pTableListInfo);

362,375✔

213

  int32_t totalSize = taosArrayGetSize(pList);

362,375✔

214

  int32_t code = 0;

362,375✔

215

  void* pTask = sStreamReaderInfo->pTask;

362,375✔

216

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

973,461✔

217

    STableKeyInfo* info = taosArrayGet(pList, i);

611,086✔

218

    if (info == NULL) {

611,086✔

221

    code = cacheTag(sStreamReaderInfo->pVnode, sStreamReaderInfo->pTableMetaCacheTrigger, sStreamReaderInfo->pExprInfoTriggerTag, sStreamReaderInfo->numOfExprTriggerTag, &sStreamReaderInfo->storageApi, info->uid, 0, NULL);

611,086✔

222

    if (code != 0){

610,935✔

226

    code = cacheTag(sStreamReaderInfo->pVnode, sStreamReaderInfo->pTableMetaCacheCalc, sStreamReaderInfo->pExprInfoCalcTag, sStreamReaderInfo->numOfExprCalcTag, &sStreamReaderInfo->storageApi, info->uid, 0, NULL);

610,935✔

227

    if (code != 0){

610,935✔

231

    code = qStreamSetTableList(tableInfo, info->uid, info->groupId);

610,935✔

232

    if (code != 0){

611,086✔

236

  return 0;

362,375✔

239

static int32_t generateTablistForStreamReader(SVnode* pVnode, SStreamTriggerReaderInfo* sStreamReaderInfo) {

361,563✔

240

  int32_t                   code = 0;

361,563✔

241

  int32_t                   lino = 0;

361,563✔

242

  SNodeList* groupNew = NULL;

361,563✔

243

  void* pTableListInfo = NULL;

362,375✔

246

  STREAM_CHECK_RET_GOTO(nodesCloneList(sStreamReaderInfo->partitionCols, &groupNew));

362,375✔

248

  STREAM_CHECK_RET_GOTO(qStreamCreateTableListForReader(pVnode, sStreamReaderInfo->suid, sStreamReaderInfo->uid, sStreamReaderInfo->tableType, groupNew,

361,294✔

252

  STREAM_CHECK_RET_GOTO(qTransformStreamTableList(sStreamReaderInfo, pTableListInfo, &sStreamReaderInfo->tableList));

362,375✔

254

  void* pTask = sStreamReaderInfo->pTask;

362,375✔

255

  ST_TASK_DLOG("vgId:%d %s tablelist size:%" PRIzu, TD_VID(pVnode), __func__, taosArrayGetSize(sStreamReaderInfo->tableList.pTableList));

362,375✔

256

end:

361,903✔

257

  nodesDestroyList(groupNew);

362,375✔

258

  qStreamDestroyTableList(pTableListInfo);

362,375✔

259

  STREAM_PRINT_LOG_END(code, lino);

362,375✔

260

  return code;

362,375✔

263

static int32_t buildVTableInfoRsp(const SStreamMsgVTableInfo* rsp, void** data, size_t* size) {

85,834✔

264

  int32_t code = 0;

85,834✔

265

  int32_t lino = 0;

85,834✔

266

  void*   buf = NULL;

85,834✔

267

  int32_t len = tSerializeSStreamMsgVTableInfo(NULL, 0, rsp);

85,834✔

268

  STREAM_CHECK_CONDITION_GOTO(len <= 0, TSDB_CODE_INVALID_PARA);

85,834✔

269

  buf = rpcMallocCont(len);

85,834✔

270

  STREAM_CHECK_NULL_GOTO(buf, terrno);

85,834✔

271

  int32_t actLen = tSerializeSStreamMsgVTableInfo(buf, len, rsp);

85,834✔

272

  STREAM_CHECK_CONDITION_GOTO(actLen != len, TSDB_CODE_INVALID_PARA);

85,834✔

273

  *data = buf;

85,834✔

274

  *size = len;

85,834✔

275

  buf = NULL;

85,834✔

276

end:

85,834✔

277

  rpcFreeCont(buf);

85,834✔

278

  return code;

85,834✔

281

static int32_t buildTsRsp(const SStreamTsResponse* tsRsp, void** data, size_t* size) {

685,030✔

282

  int32_t code = 0;

685,030✔

283

  int32_t lino = 0;

685,030✔

284

  void*   buf = NULL;

685,030✔

285

  int32_t len = tSerializeSStreamTsResponse(NULL, 0, tsRsp);

685,030✔

286

  STREAM_CHECK_CONDITION_GOTO(len <= 0, TSDB_CODE_INVALID_PARA);

685,030✔

287

  buf = rpcMallocCont(len);

685,030✔

288

  STREAM_CHECK_NULL_GOTO(buf, terrno);

685,030✔

289

  int32_t actLen = tSerializeSStreamTsResponse(buf, len, tsRsp);

685,030✔

290

  STREAM_CHECK_CONDITION_GOTO(actLen != len, TSDB_CODE_INVALID_PARA);

685,301✔

291

  *data = buf;

685,301✔

292

  *size = len;

685,301✔

293

  buf = NULL;

684,210✔

294

end:

684,210✔

295

  rpcFreeCont(buf);

684,210✔

296

  return code;

684,761✔

300

static int32_t buildRsp(SSDataBlock* pBlock, void** data, size_t* size) {

9,988,821✔

301

  int32_t code = 0;

9,988,821✔

302

  int32_t lino = 0;

9,988,821✔

303

  void*   buf = NULL;

9,988,821✔

304

  STREAM_CHECK_CONDITION_GOTO(pBlock == NULL || pBlock->info.rows == 0, TSDB_CODE_SUCCESS);

9,988,821✔

305

  size_t dataEncodeSize = blockGetEncodeSize(pBlock);

2,638,841✔

306

  buf = rpcMallocCont(dataEncodeSize);

2,638,841✔

307

  STREAM_CHECK_NULL_GOTO(buf, terrno);

2,638,841✔

308

  int32_t actualLen = blockEncode(pBlock, buf, dataEncodeSize, taosArrayGetSize(pBlock->pDataBlock));

2,638,841✔

309

  STREAM_CHECK_CONDITION_GOTO(actualLen < 0, terrno);

2,638,564✔

310

  *data = buf;

2,638,564✔

311

  *size = dataEncodeSize;

2,638,564✔

312

  buf = NULL;

2,638,841✔

313

end:

9,993,231✔

314

  rpcFreeCont(buf);

9,993,231✔

315

  return code;

9,992,251✔

318

static int32_t buildArrayRsp(SArray* pBlockList, void** data, size_t* size) {

159,256✔

319

  int32_t code = 0;

159,256✔

320

  int32_t lino = 0;

159,256✔

322

  void*   buf = NULL;

159,256✔

324

  int32_t blockNum = 0;

159,256✔

325

  size_t  dataEncodeBufSize = 0;

159,256✔

326

  for(size_t i = 0; i < taosArrayGetSize(pBlockList); i++){

371,283✔

327

    SSDataBlock* pBlock = taosArrayGetP(pBlockList, i);

212,027✔

328

    if (pBlock == NULL || pBlock->info.rows == 0) continue;

212,027✔

329

    int32_t blockSize = blockGetEncodeSize(pBlock);

212,027✔

330

    dataEncodeBufSize += blockSize;

212,027✔

331

    blockNum++;

212,027✔

333

  buf = rpcMallocCont(INT_BYTES + dataEncodeBufSize);

159,256✔

334

  STREAM_CHECK_NULL_GOTO(buf, terrno);

159,256✔

336

  char* dataBuf = (char*)buf;

159,256✔

337

  *((int32_t*)(dataBuf)) = blockNum;

159,256✔

338

  dataBuf += INT_BYTES;

159,256✔

339

  for(size_t i = 0; i < taosArrayGetSize(pBlockList); i++){

371,283✔

340

    SSDataBlock* pBlock = taosArrayGetP(pBlockList, i);

212,027✔

341

    if (pBlock == NULL || pBlock->info.rows == 0) continue;

212,027✔

342

    int32_t actualLen = blockEncode(pBlock, dataBuf, dataEncodeBufSize, taosArrayGetSize(pBlock->pDataBlock));

212,027✔

343

    STREAM_CHECK_CONDITION_GOTO(actualLen < 0, terrno);

212,027✔

344

    dataBuf += actualLen;

212,027✔

346

  *data = buf;

159,256✔

347

  *size = INT_BYTES + dataEncodeBufSize;

159,256✔

348

  buf = NULL;

159,256✔

349

end:

159,256✔

350

  rpcFreeCont(buf);

159,256✔

351

  return code;

159,256✔

374

static int32_t buildWalMetaBlockNew(SSDataBlock* pBlock, int64_t id, int64_t skey, int64_t ekey, int64_t ver) {

10,867,213✔

375

  int32_t code = 0;

10,867,213✔

376

  int32_t lino = 0;

10,867,213✔

377

  int32_t index = 0;

10,867,213✔

378

  STREAM_CHECK_RET_GOTO(addColData(pBlock, index++, &id));

10,867,213✔

379

  STREAM_CHECK_RET_GOTO(addColData(pBlock, index++, &skey));

10,868,167✔

380

  STREAM_CHECK_RET_GOTO(addColData(pBlock, index++, &ekey));

10,866,939✔

381

  STREAM_CHECK_RET_GOTO(addColData(pBlock, index++, &ver));

10,867,059✔

383

end:

10,867,207✔

384

  return code;

10,867,207✔

387

static int32_t buildTableBlock(SSDataBlock* pBlock, int64_t id, int64_t ver, ETableBlockType type) {

4,072✔

388

  int32_t code = 0;

4,072✔

389

  int32_t lino = 0;

4,072✔

390

  int32_t index = 0;

4,072✔

391

  STREAM_CHECK_RET_GOTO(addColData(pBlock, index++, &id));

4,072✔

392

  STREAM_CHECK_RET_GOTO(addColData(pBlock, index++, &ver));

4,072✔

393

  STREAM_CHECK_RET_GOTO(addColData(pBlock, index++, &type));

4,072✔

395

end:

4,072✔

396

  return code;

4,072✔

407

static int32_t scanDeleteDataNew(SStreamTriggerReaderInfo* sStreamReaderInfo, SSTriggerWalNewRsp* rsp, void* data, int32_t len,

34,760✔

409

  int32_t    code = 0;

34,760✔

410

  int32_t    lino = 0;

34,760✔

411

  SDecoder   decoder = {0};

34,760✔

412

  SDeleteRes req = {0};

34,760✔

413

  void* pTask = sStreamReaderInfo->pTask;

34,760✔

415

  req.uidList = taosArrayInit(0, sizeof(tb_uid_t));

34,760✔

416

  tDecoderInit(&decoder, data, len);

34,760✔

417

  STREAM_CHECK_RET_GOTO(tDecodeDeleteRes(&decoder, &req));

34,760✔

418

  STREAM_CHECK_CONDITION_GOTO((sStreamReaderInfo->tableType == TSDB_SUPER_TABLE && !sStreamReaderInfo->isVtableStream && req.suid != sStreamReaderInfo->suid), TDB_CODE_SUCCESS);

34,760✔

420

  for (int32_t i = 0; i < taosArrayGetSize(req.uidList); i++) {

53,766✔

421

    uint64_t* uid = taosArrayGet(req.uidList, i);

30,854✔

422

    STREAM_CHECK_NULL_GOTO(uid, terrno);

30,854✔

423

    uint64_t   id = 0;

30,854✔

424

    ST_TASK_DLOG("stream reader scan delete start data:uid %" PRIu64 ", skey %" PRIu64 ", ekey %" PRIu64, *uid, req.skey, req.ekey);

30,854✔

425

    STREAM_CHECK_CONDITION_GOTO(!uidInTableListSet(sStreamReaderInfo, req.suid, *uid, &id, false), TDB_CODE_SUCCESS);

30,854✔

426

    STREAM_CHECK_RET_GOTO(blockDataEnsureCapacity(rsp->deleteBlock, ((SSDataBlock*)rsp->deleteBlock)->info.rows + 1));

22,912✔

427

    STREAM_CHECK_RET_GOTO(buildWalMetaBlockNew(rsp->deleteBlock, id, req.skey, req.ekey, ver));

22,912✔

428

    ((SSDataBlock*)rsp->deleteBlock)->info.rows++;

22,912✔

429

    rsp->totalRows++;

22,912✔

432

end:

34,760✔

433

  taosArrayDestroy(req.uidList);

34,760✔

434

  tDecoderClear(&decoder);

34,760✔

435

  return code;

34,760✔

438

static int32_t createBlockForProcessMeta(SSDataBlock** pBlock) {

3,528✔

439

  int32_t code = 0;

3,528✔

440

  int32_t lino = 0;

3,528✔

441

  SArray* schemas = NULL;

3,528✔

443

  schemas = taosArrayInit(8, sizeof(SSchema));

3,528✔

444

  STREAM_CHECK_NULL_GOTO(schemas, terrno);

3,528✔

446

  int32_t index = 0;

3,528✔

447

  STREAM_CHECK_RET_GOTO(qStreamBuildSchema(schemas, TSDB_DATA_TYPE_BIGINT, LONG_BYTES, index++))  // gid non vtable/uid vtable

3,528✔

448

  STREAM_CHECK_RET_GOTO(qStreamBuildSchema(schemas, TSDB_DATA_TYPE_BIGINT, LONG_BYTES, index++))  // ver

3,528✔

449

  STREAM_CHECK_RET_GOTO(qStreamBuildSchema(schemas, TSDB_DATA_TYPE_TINYINT, CHAR_BYTES, index++))  // type

3,528✔

451

  STREAM_CHECK_RET_GOTO(createDataBlockForStream(schemas, pBlock));

3,528✔

453

end:

3,528✔

454

  taosArrayDestroy(schemas);

3,528✔

455

  return code;

3,528✔

458

static int32_t addOneRow(void** tmp, int64_t id, int64_t ver, ETableBlockType type) {

4,072✔

459

  int32_t  code = 0;

4,072✔

460

  int32_t  lino = 0;

4,072✔

461

  if (*tmp == NULL) {

4,072✔

462

    STREAM_CHECK_RET_GOTO(createBlockForProcessMeta((SSDataBlock**)tmp));

3,528✔

464

  STREAM_CHECK_RET_GOTO(blockDataEnsureCapacity(*tmp, ((SSDataBlock*)(*tmp))->info.rows + 1));

4,072✔

465

  STREAM_CHECK_RET_GOTO(buildTableBlock(*tmp, id, ver, type));

4,072✔

466

  ((SSDataBlock*)(*tmp))->info.rows++;

4,072✔

468

end:

4,072✔

469

  return code;

4,072✔

472

static int32_t addUidListToBlock(SArray* uidListAdd, void** block, int64_t ver, int32_t* totalRows, ETableBlockType type) {

83,884✔

473

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

87,956✔

474

    uint64_t* uid = taosArrayGet(uidListAdd, i);

4,072✔

475

    if (uid == NULL) {

4,072✔

478

    int32_t code = addOneRow(block, *uid, ver, type);

4,072✔

479

    if (code != 0) {

4,072✔

482

    (*totalRows)++;

4,072✔

484

  return 0;

83,884✔

487

static int32_t qStreamGetAddTable(SStreamTriggerReaderInfo* sStreamReaderInfo, SArray* tableListAdd, SArray* uidListAdd) {

93,444✔

488

  int32_t      code = 0;

93,444✔

489

  int32_t      lino = 0;

93,444✔

490

  if (uidListAdd == NULL) {

93,444✔

491

    return 0;

85,633✔

493

  void* pTask = sStreamReaderInfo->pTask;

7,811✔

495

  taosRLockLatch(&sStreamReaderInfo->lock);

7,811✔

496

  int32_t totalSize = taosArrayGetSize(tableListAdd);

7,811✔

497

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

10,315✔

498

    STableKeyInfo* info = taosArrayGet(tableListAdd, i);

2,504✔

499

    if (info == NULL) {

2,504✔

502

    if (taosHashGet(sStreamReaderInfo->tableList.uIdMap, &info->uid, LONG_BYTES) != NULL) {

2,504✔

505

    STREAM_CHECK_NULL_GOTO(taosArrayPush(uidListAdd, &info->uid), terrno);

5,008✔

506

    ST_TASK_WLOG("%s real add table to list for uid:%" PRId64, __func__, info->uid);

2,504✔

509

end:

7,811✔

510

  taosRUnLockLatch(&sStreamReaderInfo->lock);

7,811✔

511

  return code;

7,811✔

514

static int32_t qStreamGetDelTable(SStreamTriggerReaderInfo* sStreamReaderInfo, SArray* tableListDel, SArray* uidListDel) {

55,197✔

515

  int32_t      code = 0;

55,197✔

516

  int32_t      lino = 0;

55,197✔

517

  if (uidListDel == NULL) {

55,197✔

520

  void* pTask = sStreamReaderInfo->pTask;

55,197✔

522

  taosRLockLatch(&sStreamReaderInfo->lock);

55,197✔

523

  int32_t totalSize = taosArrayGetSize(tableListDel);

55,197✔

524

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

56,231✔

525

    int64_t* uid = taosArrayGet(tableListDel, i);

1,034✔

526

    if (uid == NULL) {

1,034✔

529

    if (taosHashGet(sStreamReaderInfo->tableList.uIdMap, uid, LONG_BYTES) == NULL) {

1,034✔

532

    STREAM_CHECK_NULL_GOTO(taosArrayPush(uidListDel, uid), terrno);

1,034✔

533

    ST_TASK_WLOG("%s real del table from list for uid:%" PRId64, __func__, *uid);

1,034✔

536

end:

55,197✔

537

  taosRUnLockLatch(&sStreamReaderInfo->lock);

55,197✔

538

  return code;

55,197✔

541

static int32_t scanDropTableNew(SStreamTriggerReaderInfo* sStreamReaderInfo, SSTriggerWalNewRsp* rsp, void* data, int32_t len,

36,912✔

543

  int32_t  code = 0;

36,912✔

544

  int32_t  lino = 0;

36,912✔

545

  SDecoder decoder = {0};

36,912✔

546

  void* pTask = sStreamReaderInfo->pTask;

36,912✔

547

  SArray* uidList = NULL;

36,912✔

548

  SArray* uidListDel = NULL;

36,912✔

549

  SArray* uidListDelOutTbl = NULL;

36,912✔

550

  SVDropTbBatchReq req = {0};

36,912✔

551

  tDecoderInit(&decoder, data, len);

36,912✔

552

  STREAM_CHECK_RET_GOTO(tDecodeSVDropTbBatchReq(&decoder, &req));

36,912✔

554

  for (int32_t iReq = 0; iReq < req.nReqs; iReq++) {

73,824✔

555

    SVDropTbReq* pDropTbReq = req.pReqs + iReq;

36,912✔

556

    STREAM_CHECK_NULL_GOTO(pDropTbReq, TSDB_CODE_INVALID_PARA);

36,912✔

557

    uint64_t id = 0;

36,912✔

558

    if(!uidInTableListOrigin(sStreamReaderInfo, pDropTbReq->suid, pDropTbReq->uid, &id)) {

36,912✔

559

      continue;

35,878✔

562

    if (sStreamReaderInfo->deleteOutTbl != 0) {

1,034✔

569

    if (sStreamReaderInfo->isVtableStream) {

1,034✔

570

      if (uidList == NULL) {

1,034✔

571

        uidList = taosArrayInit(8, sizeof(tb_uid_t));

1,034✔

572

        STREAM_CHECK_NULL_GOTO(uidList, terrno);

1,034✔

574

      STREAM_CHECK_NULL_GOTO(taosArrayPush(uidList, &pDropTbReq->uid), terrno);

2,068✔

577

    ST_TASK_DLOG("stream reader scan drop uid %" PRId64 ", id %" PRIu64, pDropTbReq->uid, id);

1,034✔

579

  STREAM_CHECK_RET_GOTO(addUidListToBlock(uidListDelOutTbl, &rsp->tableBlock, ver, &rsp->totalRows, TABLE_BLOCK_DROP));

36,912✔

581

  if (sStreamReaderInfo->isVtableStream) {

36,912✔

582

    uidListDel = taosArrayInit(8, sizeof(tb_uid_t));

36,912✔

583

    STREAM_CHECK_NULL_GOTO(uidListDel, terrno);

36,912✔

584

    STREAM_CHECK_RET_GOTO(qStreamGetDelTable(sStreamReaderInfo, uidList, uidListDel));

36,912✔

585

    STREAM_CHECK_RET_GOTO(addUidListToBlock(uidListDel, &rsp->tableBlock, ver, &rsp->totalRows, TABLE_BLOCK_RETIRE));

36,912✔

588

end:

36,912✔

589

  taosArrayDestroy(uidList);

36,912✔

590

  taosArrayDestroy(uidListDel);

36,912✔

591

  taosArrayDestroy(uidListDelOutTbl);

36,912✔

592

  tDecoderClear(&decoder);

36,912✔

593

  return code;

36,912✔

596

static int32_t qStreamModifyTableList(SStreamTriggerReaderInfo* sStreamReaderInfo, SArray* tableListAdd, SArray* tableListDel) {

95,117✔

597

  int32_t      code = 0;

95,117✔

598

  int32_t      lino = 0;

95,117✔

599

  void* pTask = sStreamReaderInfo->pTask;

95,117✔

601

  taosWLockLatch(&sStreamReaderInfo->lock);

95,117✔

602

  int32_t totalSize = taosArrayGetSize(tableListDel);

95,117✔

603

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

95,117✔

611

  totalSize = taosArrayGetSize(tableListAdd);

95,117✔

612

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

122,851✔

613

    STableKeyInfo* info = taosArrayGet(tableListAdd, i);

27,734✔

614

    if (info == NULL) {

27,734✔

617

    int ret = cacheTag(sStreamReaderInfo->pVnode, sStreamReaderInfo->pTableMetaCacheTrigger, sStreamReaderInfo->pExprInfoTriggerTag, sStreamReaderInfo->numOfExprTriggerTag, &sStreamReaderInfo->storageApi, info->uid, 0, NULL);

27,734✔

618

    if (ret != 0){

27,734✔

619

      ST_TASK_WLOG("%s cacheTag trigger failed for uid:%" PRId64",code:%d", __func__, info->uid, ret);

6,047✔

620

      continue;

6,047✔

622

    ret = cacheTag(sStreamReaderInfo->pVnode, sStreamReaderInfo->pTableMetaCacheCalc, sStreamReaderInfo->pExprInfoCalcTag, sStreamReaderInfo->numOfExprCalcTag, &sStreamReaderInfo->storageApi, info->uid, 0, NULL);

21,687✔

623

    if (ret != 0){

21,687✔

627

    STREAM_CHECK_RET_GOTO(qStreamRemoveTableList(&sStreamReaderInfo->tableList, info->uid));

21,687✔

628

    STREAM_CHECK_RET_GOTO(qStreamSetTableList(&sStreamReaderInfo->tableList, info->uid, info->groupId));

21,687✔

631

end:

95,117✔

632

  taosWUnLockLatch(&sStreamReaderInfo->lock);

95,117✔

633

  return code;

95,117✔

636

static int32_t processTableList(SStreamTriggerReaderInfo* sStreamReaderInfo, SArray* uidList, SArray** tableList) {

95,117✔

637

  int32_t code = 0;

95,117✔

638

  int32_t lino = 0;

95,117✔

639

  SNodeList* groupNew = NULL;

95,117✔

641

  if (taosArrayGetSize(uidList) == 0) {

95,117✔

642

    return 0;

67,383✔

644

  STREAM_CHECK_RET_GOTO(nodesCloneList(sStreamReaderInfo->partitionCols, &groupNew));

27,734✔

645

  STREAM_CHECK_RET_GOTO(qStreamFilterTableListForReader(sStreamReaderInfo->pVnode, uidList, groupNew, sStreamReaderInfo->pTagCond,

27,734✔

649

end:

27,734✔

650

  nodesDestroyList(groupNew);

27,734✔

651

  return code;

27,734✔

654

static int32_t scanCreateTableNew(SStreamTriggerReaderInfo* sStreamReaderInfo, SSTriggerWalNewRsp* rsp, void* data, int32_t len,

75,159✔

656

  int32_t  code = 0;

75,159✔

657

  int32_t  lino = 0;

75,159✔

658

  SDecoder decoder = {0};

75,159✔

659

  SArray*  uidList = NULL;

75,159✔

660

  SArray*  tableList = NULL;

75,159✔

661

  SArray*  uidListAdd = NULL;

75,159✔

662

  void* pTask = sStreamReaderInfo->pTask;

75,159✔

664

  SVCreateTbBatchReq req = {0};

75,159✔

665

  tDecoderInit(&decoder, data, len);

75,159✔

667

  STREAM_CHECK_RET_GOTO(tDecodeSVCreateTbBatchReq(&decoder, &req));

75,159✔

669

  uidList = taosArrayInit(8, sizeof(tb_uid_t));

75,159✔

670

  STREAM_CHECK_NULL_GOTO(uidList, terrno);

74,737✔

672

  if (sStreamReaderInfo->isVtableStream) {

74,737✔

673

    uidListAdd = taosArrayInit(8, sizeof(tb_uid_t));

6,096✔

674

    STREAM_CHECK_NULL_GOTO(uidListAdd, terrno);

6,096✔

677

  SVCreateTbReq* pCreateReq = NULL;

75,159✔

678

  for (int32_t iReq = 0; iReq < req.nReqs; iReq++) {

150,318✔

679

    pCreateReq = req.pReqs + iReq;

75,159✔

680

    if (!needReLoadTableList(sStreamReaderInfo, pCreateReq->type, pCreateReq->ctb.suid, pCreateReq->uid, false)) {

75,159✔

681

      ST_TASK_DLOG("stream reader scan create table jump, %s", pCreateReq->name);

67,383✔

682

      continue;

67,383✔

684

    ST_TASK_ILOG("stream reader scan create table %s", pCreateReq->name);

7,776✔

685

    STREAM_CHECK_NULL_GOTO(taosArrayPush(uidList, &pCreateReq->uid), terrno);

15,552✔

688

  STREAM_CHECK_RET_GOTO(processTableList(sStreamReaderInfo, uidList, &tableList));

75,159✔

689

  STREAM_CHECK_RET_GOTO(qStreamGetAddTable(sStreamReaderInfo, tableList, uidListAdd));

75,159✔

690

  if (sStreamReaderInfo->isVtableStream) {

75,159✔

691

    STREAM_CHECK_RET_GOTO(addUidListToBlock(uidListAdd, &rsp->tableBlock, ver, &rsp->totalRows, TABLE_BLOCK_ADD));

6,096✔

694

  STREAM_CHECK_RET_GOTO(qStreamModifyTableList(sStreamReaderInfo, tableList, uidList));

75,159✔

695

end:

75,159✔

696

  taosArrayDestroy(uidList);

75,159✔

697

  taosArrayDestroy(uidListAdd);

75,159✔

698

  taosArrayDestroy(tableList);

75,159✔

699

  tDeleteSVCreateTbBatchReq(&req);

75,159✔

700

  tDecoderClear(&decoder);

75,159✔

701

  return code;

75,159✔

704

static int32_t processAutoCreateTableNew(SStreamTriggerReaderInfo* sStreamReaderInfo, SVCreateTbReq* pCreateReq, int64_t ver) {

5,760,593✔

705

  int32_t  code = 0;

5,760,593✔

706

  int32_t  lino = 0;

5,760,593✔

707

  void*    pTask = sStreamReaderInfo->pTask;

5,760,593✔

708

  SArray*  uidList = NULL;

5,761,599✔

709

  SArray*  tableList = NULL;

5,761,599✔

711

  ST_TASK_DLOG("%s start, name:%s uid:%"PRId64, __func__, pCreateReq->name, pCreateReq->uid);

5,761,714✔

712

  if (!needReLoadTableList(sStreamReaderInfo, pCreateReq->type, pCreateReq->ctb.suid, pCreateReq->uid, false) ||

5,764,588✔

713

      ignoreMetaChange(sStreamReaderInfo->tableList.version, ver)) {

2,297✔

714

    ST_TASK_DLOG("stream reader scan auto create table jump, %s", pCreateReq->name);

5,760,072✔

715

    goto end;

5,759,804✔

717

  uidList = taosArrayInit(8, sizeof(tb_uid_t));

1,673✔

718

  STREAM_CHECK_NULL_GOTO(uidList, terrno);

1,673✔

719

  STREAM_CHECK_NULL_GOTO(taosArrayPush(uidList, &pCreateReq->uid), terrno);

3,346✔

720

  ST_TASK_DLOG("stream reader scan auto create table %s", pCreateReq->name);

1,673✔

722

  STREAM_CHECK_RET_GOTO(processTableList(sStreamReaderInfo, uidList, &tableList));

1,673✔

723

  STREAM_CHECK_RET_GOTO(qStreamModifyTableList(sStreamReaderInfo, tableList, uidList));

1,673✔

724

end:

5,758,625✔

725

  taosArrayDestroy(uidList);

5,762,022✔

726

  taosArrayDestroy(tableList);

5,761,751✔

727

  return code;

5,761,751✔

730

static bool isColIdInList(SNodeList* colList, col_id_t cid){

534✔

731

  int32_t  code = 0;

534✔

732

  int32_t  lino = 0;

534✔

733

  SNode*  nodeItem = NULL;

534✔

734

  FOREACH(nodeItem, colList) {

1,335✔

735

    SNode*           pNode = ((STargetNode*)nodeItem)->pExpr;

1,335✔

736

    if (nodeType(pNode) == QUERY_NODE_COLUMN) {

1,335✔

737

      SColumnNode*     valueNode = (SColumnNode*)(pNode);

1,335✔

738

      if (cid == valueNode->colId) {

1,335✔

739

        return true;

534✔

747

static bool isAlteredTable(int8_t action, ETableType tbType) {

18,819✔

748

  if (action == TSDB_ALTER_TABLE_UPDATE_MULTI_TABLE_TAG_VAL && tbType == TSDB_CHILD_TABLE) {

18,819✔

749

    return true;

18,285✔

750

  } else if (action == TSDB_ALTER_TABLE_UPDATE_CHILD_TABLE_TAG_VAL && tbType == TSDB_SUPER_TABLE) {

534✔

752

  } else if ((action == TSDB_ALTER_TABLE_ALTER_COLUMN_REF || action == TSDB_ALTER_TABLE_REMOVE_COLUMN_REF) &&

534✔

753

     (tbType == TSDB_VIRTUAL_CHILD_TABLE || tbType == TSDB_VIRTUAL_NORMAL_TABLE)) {

534✔

754

    return true;

534✔

759

void getAlterColId(void* pVnode, int64_t uid, const char* colName, col_id_t* colId) {

534✔

760

  SSchemaWrapper *pSchema = metaGetTableSchema(((SVnode *)pVnode)->pMeta, uid, -1, 1, NULL, 0, false);

534✔

761

  if (pSchema == NULL) {

534✔

764

  for (int32_t i = 0; i < pSchema->nCols; i++) {

1,335✔

765

    if (strncmp(pSchema->pSchema[i].name, colName, TSDB_COL_NAME_LEN) == 0) {

1,335✔

766

      *colId = pSchema->pSchema[i].colId;

534✔

767

      break;

534✔

771

  return;

534✔

775

static int32_t scanAlterTableColumnRef(SStreamTriggerReaderInfo* sStreamReaderInfo, SSTriggerWalNewRsp* rsp,

534✔

777

  int32_t code = 0;

534✔

778

  int32_t lino = 0;

534✔

779

  void* pTask = sStreamReaderInfo->pTask;

534✔

780

  SArray* uidListAdd = NULL;

534✔

782

  uidListAdd = taosArrayInit(8, sizeof(tb_uid_t));

534✔

783

  STREAM_CHECK_NULL_GOTO(uidListAdd, terrno);

534✔

785

  uint64_t id = 0;

534✔

786

  STREAM_CHECK_CONDITION_GOTO(!uidInTableListOrigin(sStreamReaderInfo, sStreamReaderInfo->suid, uid, &id), TDB_CODE_SUCCESS);

534✔

788

  col_id_t colId = 0;

534✔

789

  getAlterColId(sStreamReaderInfo->pVnode, uid, pReq->colName, &colId);

534✔

790

  if (atomic_load_8(&sStreamReaderInfo->isVtableOnlyTs) == 0 && !isColIdInList(sStreamReaderInfo->triggerCols, colId)) {

534✔

795

  STREAM_CHECK_NULL_GOTO(taosArrayPush(uidListAdd, &uid), terrno);

534✔

796

  STREAM_CHECK_RET_GOTO(addUidListToBlock(uidListAdd, &rsp->tableBlock, ver, &rsp->totalRows, TABLE_BLOCK_ADD));

534✔

798

  ST_TASK_DLOG("stream reader scan alter table column ref %s", pReq->tbName);

534✔

800

end:

534✔

801

  taosArrayDestroy(uidListAdd);

534✔

802

  STREAM_PRINT_LOG_END_WITHID(code, lino);

534✔

803

  return code;

534✔

806

static int32_t checkAlter(SStreamTriggerReaderInfo* sStreamReaderInfo, char* tbName, int8_t action, uint64_t *uid) {

18,819✔

807

  int32_t  code = 0;

18,819✔

808

  int32_t  lino = 0;

18,819✔

809

  ETableType tbType = 0;

18,819✔

810

  uint64_t suid = 0;

18,819✔

812

  STREAM_CHECK_RET_GOTO(metaGetTableTypeSuidByName(sStreamReaderInfo->pVnode, tbName, &tbType, &suid));

18,819✔

813

  STREAM_CHECK_CONDITION_GOTO(!isAlteredTable(action, tbType), TDB_CODE_SUCCESS);

18,819✔

814

  STREAM_CHECK_CONDITION_GOTO(suid != sStreamReaderInfo->suid, TDB_CODE_SUCCESS);

18,819✔

815

  if (action == TSDB_ALTER_TABLE_UPDATE_CHILD_TABLE_TAG_VAL) {

16,859✔

819

  STREAM_CHECK_RET_GOTO(metaGetTableUidByName(sStreamReaderInfo->pVnode, tbName, uid));

16,859✔

821

end:

18,819✔

822

  return code;

18,819✔

825

static SArray* getTableListForAlterSuperTable(SStreamTriggerReaderInfo* sStreamReaderInfo, SVAlterTbReq* pReq){

18,285✔

826

  int32_t code = 0;

18,285✔

827

  int32_t lino = 0;

18,285✔

828

  void* pTask = sStreamReaderInfo->pTask;

18,285✔

829

  SArray* uidList = taosArrayInit(8, sizeof(tb_uid_t));

18,285✔

830

  STREAM_CHECK_NULL_GOTO(uidList, terrno);

18,285✔

831

  for (int32_t i = 0; i < taosArrayGetSize(pReq->tables); i++) {

36,570✔

832

    SUpdateTableTagVal *pTable = taosArrayGet(pReq->tables, i);

18,285✔

833

    uint64_t uid = 0;

18,285✔

834

    code = checkAlter(sStreamReaderInfo, pTable->tbName, pReq->action, &uid);

18,285✔

835

    if (code == TSDB_CODE_PAR_TABLE_NOT_EXIST) {

18,285✔

840

    STREAM_CHECK_RET_GOTO(code);

18,285✔

841

    STREAM_CHECK_NULL_GOTO(taosArrayPush(uidList, (const void *)&uid), terrno);

18,285✔

844

end:

18,285✔

845

  if (code != 0) {

18,285✔

850

  return uidList;

18,285✔

854

static int32_t scanAlterTableTagVal(SStreamTriggerReaderInfo* sStreamReaderInfo, SSTriggerWalNewRsp* rsp,

18,285✔

856

  int32_t code = 0;

18,285✔

857

  int32_t lino = 0;

18,285✔

858

  void* pTask = sStreamReaderInfo->pTask;

18,285✔

859

  SArray* uidListAdd = NULL;

18,285✔

860

  SArray* uidListDel = NULL;

18,285✔

861

  SArray* tableList = NULL;

18,285✔

863

  if (sStreamReaderInfo->isVtableStream) {

18,285✔

864

    uidListAdd = taosArrayInit(8, sizeof(tb_uid_t));

1,715✔

865

    STREAM_CHECK_NULL_GOTO(uidListAdd, terrno);

1,715✔

868

  uidListDel = taosArrayInit(8, sizeof(tb_uid_t));

18,285✔

869

  STREAM_CHECK_NULL_GOTO(uidListDel, terrno);

18,285✔

871

  STREAM_CHECK_RET_GOTO(processTableList(sStreamReaderInfo, uidList, &tableList));

18,285✔

872

  STREAM_CHECK_RET_GOTO(qStreamGetDelTable(sStreamReaderInfo, uidList, uidListDel));

18,285✔

874

  if (rsp->checkAlter && taosArrayGetSize(uidListDel) > 0 && rsp->totalDataRows > 0) {

18,285✔

881

  STREAM_CHECK_RET_GOTO(qStreamGetAddTable(sStreamReaderInfo, tableList, uidListAdd));

18,285✔

882

  if (sStreamReaderInfo->isVtableStream) {

18,285✔

883

    STREAM_CHECK_RET_GOTO(addUidListToBlock(uidListAdd, &rsp->tableBlock, ver, &rsp->totalRows, TABLE_BLOCK_ADD));

1,715✔

884

    STREAM_CHECK_RET_GOTO(addUidListToBlock(uidListDel, &rsp->tableBlock, ver, &rsp->totalRows, TABLE_BLOCK_RETIRE));

1,715✔

886

  STREAM_CHECK_RET_GOTO(qStreamModifyTableList(sStreamReaderInfo, tableList, uidList));

18,285✔

888

  ST_TASK_DLOG("%s stream reader scan alter table tag val", __func__);

18,285✔

890

end:

18,285✔

891

  taosArrayDestroy(uidListAdd);

18,285✔

892

  taosArrayDestroy(uidListDel);

18,285✔

893

  taosArrayDestroy(tableList);

18,285✔

894

  STREAM_PRINT_LOG_END_WITHID(code, lino);

18,285✔

895

  return code;

18,285✔

898

static int32_t scanAlterTableNew(SStreamTriggerReaderInfo* sStreamReaderInfo, SSTriggerWalNewRsp* rsp, void* data, int32_t len, int64_t ver) {

35,661✔

899

  int32_t  code = 0;

35,661✔

900

  int32_t  lino = 0;

35,661✔

901

  SDecoder decoder = {0};

35,661✔

902

  void* pTask = sStreamReaderInfo->pTask;

35,661✔

903

  SArray* uidList = NULL;

35,661✔

905

  ST_TASK_DLOG("%s start", __func__);

35,661✔

907

  SVAlterTbReq req = {0};

35,661✔

908

  tDecoderInit(&decoder, data, len);

35,661✔

910

  STREAM_CHECK_RET_GOTO(tDecodeSVAlterTbReq(&decoder, &req));

35,661✔

912

  STREAM_CHECK_CONDITION_GOTO(req.action != TSDB_ALTER_TABLE_UPDATE_MULTI_TABLE_TAG_VAL && req.action != TSDB_ALTER_TABLE_UPDATE_CHILD_TABLE_TAG_VAL &&

35,661✔

915

  uint64_t uid = 0;

24,237✔

916

  if (req.action == TSDB_ALTER_TABLE_ALTER_COLUMN_REF || req.action == TSDB_ALTER_TABLE_REMOVE_COLUMN_REF) {

24,237✔

917

    STREAM_CHECK_CONDITION_GOTO(!sStreamReaderInfo->isVtableStream, TDB_CODE_SUCCESS);

5,952✔

918

    code = checkAlter(sStreamReaderInfo, req.tbName, req.action, &uid);

534✔

919

    if (code == TSDB_CODE_PAR_TABLE_NOT_EXIST) {

534✔

924

    STREAM_CHECK_RET_GOTO(scanAlterTableColumnRef(sStreamReaderInfo, rsp, &req, uid, ver));

534✔

925

  } else if (req.action == TSDB_ALTER_TABLE_UPDATE_MULTI_TABLE_TAG_VAL) {

18,285✔

926

    uidList = getTableListForAlterSuperTable(sStreamReaderInfo, &req);

18,285✔

927

    STREAM_CHECK_NULL_GOTO(uidList, terrno);

18,285✔

928

    STREAM_CHECK_RET_GOTO(scanAlterTableTagVal(sStreamReaderInfo, rsp, uidList, ver));

18,285✔

942

  ST_TASK_DLOG("%s stream reader scan alter table", __func__);

18,819✔

944

end:

35,661✔

945

  destroyAlterTbReq(&req);

35,661✔

947

  taosArrayDestroy(uidList);

35,661✔

948

  tDecoderClear(&decoder);

35,661✔

949

  STREAM_PRINT_LOG_END_WITHID(code, lino);

35,661✔

950

  return code;

35,661✔

995

static int32_t scanSubmitTbDataForMeta(SDecoder *pCoder, SStreamTriggerReaderInfo* sStreamReaderInfo, SSHashObj* gidHash, int64_t ver) {

47,047,922✔

996

  int32_t code = 0;

47,047,922✔

997

  int32_t lino = 0;

47,047,922✔

998

  WalMetaResult walMeta = {0};

47,047,922✔

999

  SSubmitTbData submitTbData = {0};

47,051,135✔

1001

  if (tStartDecode(pCoder) < 0) {

47,052,545✔

1006

  uint8_t       version = 0;

47,055,870✔

1007

  if (tDecodeI32v(pCoder, &submitTbData.flags) < 0) {

47,054,772✔

1011

  version = (submitTbData.flags >> 8) & 0xff;

47,054,772✔

1012

  submitTbData.flags = submitTbData.flags & 0xff;

47,054,772✔

1015

  if (submitTbData.flags & SUBMIT_REQ_AUTO_CREATE_TABLE) {

47,054,772✔

1016

    submitTbData.pCreateTbReq = taosMemoryCalloc(1, sizeof(SVCreateTbReq));

4,642,993✔

1017

    STREAM_CHECK_NULL_GOTO(submitTbData.pCreateTbReq, terrno);

4,642,168✔

1018

    STREAM_CHECK_RET_GOTO(tDecodeSVCreateTbReq(pCoder, submitTbData.pCreateTbReq));

4,642,168✔

1019

    STREAM_CHECK_RET_GOTO(processAutoCreateTableNew(sStreamReaderInfo, submitTbData.pCreateTbReq, ver));

4,642,993✔

1023

  if (tDecodeI64(pCoder, &submitTbData.suid) < 0) {

47,055,578✔

1027

  if (tDecodeI64(pCoder, &submitTbData.uid) < 0) {

47,056,699✔

1032

  if (!uidInTableListSet(sStreamReaderInfo, submitTbData.suid, submitTbData.uid, &walMeta.id, false)){

47,056,699✔

1033

    goto end;

38,307,297✔

1035

  if (tDecodeI32v(pCoder, &submitTbData.sver) < 0) {

8,749,499✔

1040

  if (submitTbData.flags & SUBMIT_REQ_COLUMN_DATA_FORMAT) {

8,749,499✔

1069

    uint64_t nRow = 0;

8,749,499✔

1070

    if (tDecodeU64v(pCoder, &nRow) < 0) {

8,747,723✔

1075

    for (int32_t iRow = 0; iRow < nRow; ++iRow) {

18,861,605✔

1076

      SRow *pRow = (SRow *)(pCoder->data + pCoder->pos);

10,112,232✔

1077

      pCoder->pos += pRow->len;

10,112,780✔

1078

      if (iRow == 0){

10,112,629✔

1080

        walMeta.skey = pRow->ts;

8,747,843✔

1085

      if (iRow == nRow - 1) {

10,112,081✔

1087

        walMeta.ekey = pRow->ts;

8,747,572✔

1095

  WalMetaResult* data = (WalMetaResult*)tSimpleHashGet(gidHash, &walMeta.id, LONG_BYTES);

8,747,723✔

1096

  if (data != NULL) {

8,746,897✔

1097

    if (walMeta.skey < data->skey) data->skey = walMeta.skey;

1,108✔

1098

    if (walMeta.ekey > data->ekey) data->ekey = walMeta.ekey;

1,108✔

1100

    STREAM_CHECK_RET_GOTO(tSimpleHashPut(gidHash, &walMeta.id, LONG_BYTES, &walMeta, sizeof(WalMetaResult)));

8,745,789✔

1103

end:

47,050,824✔

1104

  tDestroySVSubmitCreateTbReq(submitTbData.pCreateTbReq, TSDB_MSG_FLG_DECODE);

47,055,285✔

1105

  taosMemoryFreeClear(submitTbData.pCreateTbReq);

47,052,818✔

1106

  tEndDecode(pCoder);

47,052,818✔

1107

  return code;

47,054,461✔

1110

static int32_t scanSubmitDataForMeta(SStreamTriggerReaderInfo* sStreamReaderInfo, SSTriggerWalNewRsp* rsp, void* data, int32_t len, int64_t ver) {

47,052,586✔

1111

  int32_t  code = 0;

47,052,586✔

1112

  int32_t  lino = 0;

47,052,586✔

1113

  SDecoder decoder = {0};

47,052,586✔

1114

  SSHashObj* gidHash = NULL;

47,058,638✔

1115

  void* pTask = sStreamReaderInfo->pTask;

47,058,638✔

1117

  tDecoderInit(&decoder, data, len);

47,059,139✔

1118

  if (tStartDecode(&decoder) < 0) {

47,054,531✔

1123

  uint64_t nSubmitTbData = 0;

47,055,618✔

1124

  if (tDecodeU64v(&decoder, &nSubmitTbData) < 0) {

47,053,660✔

1129

  gidHash = tSimpleHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT));

47,053,660✔

1130

  STREAM_CHECK_NULL_GOTO(gidHash, terrno);

47,041,034✔

1132

  for (uint64_t i = 0; i < nSubmitTbData; i++) {

94,096,936✔

1133

    STREAM_CHECK_RET_GOTO(scanSubmitTbDataForMeta(&decoder, sStreamReaderInfo, gidHash, ver));

47,042,915✔

1135

  tEndDecode(&decoder);

47,054,021✔

1137

  STREAM_CHECK_RET_GOTO(blockDataEnsureCapacity(rsp->metaBlock, ((SSDataBlock*)rsp->metaBlock)->info.rows + tSimpleHashGetSize(gidHash)));

47,050,385✔

1138

  int32_t iter = 0;

47,047,363✔

1139

  void*   px = tSimpleHashIterate(gidHash, NULL, &iter);

47,045,713✔

1140

  while (px != NULL) {

55,799,444✔

1141

    WalMetaResult* pMeta = (WalMetaResult*)px;

8,746,338✔

1142

    STREAM_CHECK_RET_GOTO(buildWalMetaBlockNew(rsp->metaBlock, pMeta->id, pMeta->skey, pMeta->ekey, ver));

8,746,338✔

1143

    ((SSDataBlock*)rsp->metaBlock)->info.rows++;

8,743,586✔

1144

    rsp->totalRows++;

8,744,688✔

1145

    ST_TASK_DLOG("stream reader scan submit data:skey %" PRId64 ", ekey %" PRId64 ", id %" PRIu64

8,744,688✔

1147

    px = tSimpleHashIterate(gidHash, px, &iter);

8,747,068✔

1149

end:

47,053,106✔

1150

  tDecoderClear(&decoder);

47,055,587✔

1151

  tSimpleHashCleanup( gidHash);

47,053,562✔

1152

  return code;

47,052,104✔

1155

static int32_t createBlockForTsdbMeta(SSDataBlock** pBlock, bool isVTable) {

591,841✔

1156

  int32_t code = 0;

591,841✔

1157

  int32_t lino = 0;

591,841✔

1158

  SArray* schemas = taosArrayInit(8, sizeof(SSchema));

591,841✔

1159

  STREAM_CHECK_NULL_GOTO(schemas, terrno);

591,841✔

1161

  int32_t index = 1;

591,841✔

1162

  STREAM_CHECK_RET_GOTO(qStreamBuildSchema(schemas, TSDB_DATA_TYPE_TIMESTAMP, LONG_BYTES, index++))  // skey

591,841✔

1163

  STREAM_CHECK_RET_GOTO(qStreamBuildSchema(schemas, TSDB_DATA_TYPE_TIMESTAMP, LONG_BYTES, index++))  // ekey

591,564✔

1164

  STREAM_CHECK_RET_GOTO(qStreamBuildSchema(schemas, TSDB_DATA_TYPE_BIGINT, LONG_BYTES, index++))  // uid

591,841✔

1165

  if (!isVTable) {

591,841✔

1166

    STREAM_CHECK_RET_GOTO(qStreamBuildSchema(schemas, TSDB_DATA_TYPE_UBIGINT, LONG_BYTES, index++))  // gid

52,713✔

1168

  STREAM_CHECK_RET_GOTO(qStreamBuildSchema(schemas, TSDB_DATA_TYPE_BIGINT, LONG_BYTES, index++))     // nrows

591,841✔

1170

  STREAM_CHECK_RET_GOTO(createDataBlockForStream(schemas, pBlock));

591,841✔

1172

end:

591,841✔

1173

  taosArrayDestroy(schemas);

591,841✔

1174

  return code;

591,841✔

1177

static int32_t createBlockForWalMetaNew(SSDataBlock** pBlock) {

374,968✔

1178

  int32_t code = 0;

374,968✔

1179

  int32_t lino = 0;

374,968✔

1180

  SArray* schemas = NULL;

374,968✔

1182

  schemas = taosArrayInit(8, sizeof(SSchema));

374,968✔

1183

  STREAM_CHECK_NULL_GOTO(schemas, terrno);

374,968✔

1185

  int32_t index = 0;

374,968✔

1186

  STREAM_CHECK_RET_GOTO(qStreamBuildSchema(schemas, TSDB_DATA_TYPE_BIGINT, LONG_BYTES, index++))  // gid non vtable/uid vtable

374,968✔

1187

  STREAM_CHECK_RET_GOTO(qStreamBuildSchema(schemas, TSDB_DATA_TYPE_BIGINT, LONG_BYTES, index++))  // skey

374,968✔

1188

  STREAM_CHECK_RET_GOTO(qStreamBuildSchema(schemas, TSDB_DATA_TYPE_BIGINT, LONG_BYTES, index++))  // ekey

374,968✔

1189

  STREAM_CHECK_RET_GOTO(qStreamBuildSchema(schemas, TSDB_DATA_TYPE_BIGINT, LONG_BYTES, index++))  // ver

374,968✔

1191

  STREAM_CHECK_RET_GOTO(createDataBlockForStream(schemas, pBlock));

374,968✔

1193

end:

374,968✔

1194

  taosArrayDestroy(schemas);

374,968✔

1195

  return code;

374,968✔

1198

static int32_t processMeta(int16_t msgType, SStreamTriggerReaderInfo* sStreamReaderInfo, void *data, int32_t len, SSTriggerWalNewRsp* rsp, int64_t ver) {

691,886✔

1199

  int32_t code = 0;

691,886✔

1200

  int32_t lino = 0;

691,886✔

1201

  void* pTask = sStreamReaderInfo->pTask;

691,886✔

1203

  ST_TASK_DLOG("%s check meta msg, stream ver:%" PRId64 ", wal ver:%" PRId64, __func__, sStreamReaderInfo->tableList.version, ver);

691,886✔

1205

  SDecoder dcoder = {0};

691,886✔

1206

  tDecoderInit(&dcoder, data, len);

691,886✔

1207

  if (msgType == TDMT_VND_DELETE && sStreamReaderInfo->deleteReCalc != 0) {

691,886✔

1208

    if (rsp->deleteBlock == NULL) {

34,760✔

1209

      STREAM_CHECK_RET_GOTO(createBlockForWalMetaNew((SSDataBlock**)&rsp->deleteBlock));

12,854✔

1212

    STREAM_CHECK_RET_GOTO(scanDeleteDataNew(sStreamReaderInfo, rsp, data, len, ver));

34,760✔

1213

  } else if (msgType == TDMT_VND_DROP_TABLE &&

657,126✔

1214

    (sStreamReaderInfo->deleteOutTbl != 0 || sStreamReaderInfo->isVtableStream)) {

56,649✔

1215

    STREAM_CHECK_RET_GOTO(scanDropTableNew(sStreamReaderInfo, rsp, data, len, ver));

36,912✔

1218

  } else if (msgType == TDMT_VND_CREATE_TABLE && !ignoreMetaChange(sStreamReaderInfo->tableList.version, ver)) {

620,214✔

1219

    STREAM_CHECK_RET_GOTO(scanCreateTableNew(sStreamReaderInfo, rsp, data, len, ver));

75,159✔

1220

  } else if (msgType == TDMT_VND_ALTER_STB && !ignoreMetaChange(sStreamReaderInfo->tableList.version, ver)) {

545,055✔

1222

  } else if (msgType == TDMT_VND_ALTER_TABLE && !ignoreMetaChange(sStreamReaderInfo->tableList.version, ver)) {

513,909✔

1223

    STREAM_CHECK_RET_GOTO(scanAlterTableNew(sStreamReaderInfo, rsp, data, len, ver));

35,661✔

1226

end:

691,886✔

1227

  tDecoderClear(&dcoder);

691,886✔

1228

  return code;

691,886✔

1230

static int32_t processWalVerMetaNew(SVnode* pVnode, SSTriggerWalNewRsp* rsp, SStreamTriggerReaderInfo* sStreamReaderInfo,

23,859,173✔

1232

  int32_t code = 0;

23,859,173✔

1233

  int32_t lino = 0;

23,859,173✔

1234

  void* pTask = sStreamReaderInfo->pTask;

23,859,173✔

1236

  SWalReader* pWalReader = walOpenReader(pVnode->pWal, 0);

23,859,173✔

1237

  STREAM_CHECK_NULL_GOTO(pWalReader, terrno);

23,857,868✔

1238

  code = walReaderSeekVer(pWalReader, rsp->ver);

23,857,868✔

1239

  if (code == TSDB_CODE_WAL_LOG_NOT_EXIST){

23,857,741✔

1240

    if (rsp->ver < walGetFirstVer(pWalReader->pWal)) {

14,858,377✔

1244

      rsp->verTime = taosGetTimestampUs();

14,858,872✔

1246

    ST_TASK_DLOG("vgId:%d %s scan wal end:%s", TD_VID(pVnode), __func__, tstrerror(code));

14,859,387✔

1247

    code = TSDB_CODE_SUCCESS;

14,859,722✔

1248

    goto end;

14,859,722✔

1250

  STREAM_CHECK_RET_GOTO(code);

8,999,364✔

1252

  STREAM_CHECK_RET_GOTO(blockDataEnsureCapacity(rsp->metaBlock, STREAM_RETURN_ROWS_NUM));

8,999,364✔

1253

  while (1) {

47,545,587✔

1254

    code = walNextValidMsg(pWalReader, true);

56,543,794✔

1255

    if (code == TSDB_CODE_WAL_LOG_NOT_EXIST){

56,548,506✔

1256

      rsp->verTime = taosGetTimestampUs();

8,996,854✔

1257

      ST_TASK_DLOG("vgId:%d %s scan wal end:%s", TD_VID(pVnode), __func__, tstrerror(code));

8,996,854✔

1258

      code = TSDB_CODE_SUCCESS;

8,996,854✔

1259

      goto end;

8,996,854✔

1261

    STREAM_CHECK_RET_GOTO(code);

47,550,557✔

1262

    rsp->ver = pWalReader->curVersion;

47,550,557✔

1263

    SWalCont* wCont = &pWalReader->pHead->head;

47,552,150✔

1264

    rsp->verTime = wCont->ingestTs;

47,548,089✔

1265

    if (wCont->ingestTs / 1000 > ctime) break;

47,551,352✔

1266

    void*   data = POINTER_SHIFT(wCont->body, sizeof(SMsgHead));

47,550,770✔

1267

    int32_t len = wCont->bodyLen - sizeof(SMsgHead);

47,551,069✔

1268

    int64_t ver = wCont->version;

47,550,477✔

1270

    ST_TASK_DLOG("vgId:%d stream reader scan wal ver:%" PRId64 "/%" PRId64 ", type:%s, deleteData:%d, deleteTb:%d",

47,551,666✔

1272

    if (wCont->msgType == TDMT_VND_SUBMIT) {

47,553,048✔

1274

      if (sStreamReaderInfo->isVtableStream && rsp->tableBlock != NULL && ((SSDataBlock*)rsp->tableBlock)->info.rows > 0) {

47,061,302✔

1275

        rsp->ver--;

2,994✔

1276

        break;

2,994✔

1278

      data = POINTER_SHIFT(wCont->body, sizeof(SSubmitReq2Msg));

47,058,031✔

1279

      len = wCont->bodyLen - sizeof(SSubmitReq2Msg);

47,057,754✔

1280

      STREAM_CHECK_RET_GOTO(scanSubmitDataForMeta(sStreamReaderInfo, rsp, data, len, ver));

47,057,754✔

1282

      STREAM_CHECK_RET_GOTO(processMeta(wCont->msgType, sStreamReaderInfo, data, len, rsp, ver));

492,688✔

1285

    if (rsp->totalRows >= STREAM_RETURN_ROWS_NUM) {

47,546,109✔

1290

end:

23,859,570✔

1291

  walCloseReader(pWalReader);

23,859,570✔

1292

  return code;

23,859,234✔

1295

int32_t cacheTag(SVnode* pVnode, SHashObj* metaCache, SExprInfo* pExprInfo, int32_t numOfExpr, SStorageAPI* api, uint64_t uid, col_id_t colId, SRWLatch* lock) {

45,184,073✔

1296

  int32_t     code = 0;

45,184,073✔

1297

  int32_t     lino = 0;

45,184,073✔

1298

  SMetaReader mr = {0};

45,184,073✔

1299

  SArray* tagCache = NULL;

45,186,089✔

1300

  char* data = NULL;

45,188,280✔

1302

  if (lock != NULL) taosWLockLatch(lock);

45,187,486✔

1303

  STREAM_CHECK_CONDITION_GOTO(numOfExpr == 0, code);

45,190,765✔

1304

  stDebug("%s start,uid:%"PRIu64, __func__, uid);

2,608,601✔

1305

  void* uidData = taosHashGet(metaCache, &uid, LONG_BYTES);

2,608,601✔

1306

  if (uidData == NULL) {

2,608,601✔

1307

    tagCache = taosArrayInit(numOfExpr, POINTER_BYTES);

2,591,406✔

1308

    STREAM_CHECK_NULL_GOTO(tagCache, terrno);

2,591,406✔

1309

    if(taosHashPut(metaCache, &uid, LONG_BYTES, &tagCache, POINTER_BYTES) != 0) {

2,591,406✔

1315

    tagCache = *(SArray**)uidData;

17,195✔

1316

    stDebug("%s found tagCache, size:%zu %d, uid:%"PRIu64, __func__, taosArrayGetSize(tagCache), numOfExpr, uid);

17,195✔

1317

    STREAM_CHECK_CONDITION_GOTO(taosArrayGetSize(tagCache) != numOfExpr, TSDB_CODE_INVALID_PARA);

17,195✔

1320

  api->metaReaderFn.initReader(&mr, pVnode, META_READER_LOCK, &api->metaFn);

2,608,601✔

1321

  code = api->metaReaderFn.getEntryGetUidCache(&mr, uid);

2,608,601✔

1322

  api->metaReaderFn.readerReleaseLock(&mr);

2,603,597✔

1323

  STREAM_CHECK_RET_GOTO(code);

2,604,697✔

1325

  for (int32_t j = 0; j < numOfExpr; ++j) {

8,976,248✔

1326

    const SExprInfo* pExpr1 = &pExprInfo[j];

6,376,927✔

1327

    int32_t functionId = pExpr1->pExpr->_function.functionId;

6,376,790✔

1328

    col_id_t cid = 0;

6,385,899✔

1330

    if (fmIsScanPseudoColumnFunc(functionId)) {

6,385,899✔

1331

      int32_t fType = pExpr1->pExpr->_function.functionType;

746,992✔

1332

      if (fType == FUNCTION_TYPE_TBNAME) {

746,992✔

1333

        data = taosMemoryCalloc(1, strlen(mr.me.name) + VARSTR_HEADER_SIZE);

746,992✔

1334

        STREAM_CHECK_NULL_GOTO(data, terrno);

746,992✔

1335

        STR_TO_VARSTR(data, mr.me.name)

746,992✔

1337

      cid = -1;

746,992✔

1339

      const char* p = NULL;

5,631,401✔

1340

      char* pData = NULL;

5,631,401✔

1341

      int8_t type = pExpr1->base.resSchema.type;

5,631,401✔

1342

      int32_t len = pExpr1->base.resSchema.bytes;

5,630,984✔

1343

      STagVal tagVal = {0};

5,636,405✔

1344

      tagVal.cid = pExpr1->base.pParam[0].pCol->colId;

5,636,604✔

1345

      cid = tagVal.cid;

5,637,592✔

1346

      if (colId != 0 && cid != colId) {

5,637,592✔

1347

        continue;

1,496✔

1349

      p = api->metaFn.extractTagVal(mr.me.ctbEntry.pTags, type, &tagVal);

5,636,096✔

1351

      if (type != TSDB_DATA_TYPE_JSON && p != NULL) {

5,626,505✔

1352

        pData = tTagValToData((const STagVal*)p, false);

5,624,240✔

1354

        pData = (char*)p;

2,265✔

1357

      if (pData != NULL && (type == TSDB_DATA_TYPE_JSON || !IS_VAR_DATA_TYPE(type))) {

5,631,509✔

1358

        if (type == TSDB_DATA_TYPE_JSON) {

2,820,934✔

1361

        data = taosMemoryCalloc(1, len);

2,820,934✔

1362

        STREAM_CHECK_NULL_GOTO(data, terrno);

2,817,597✔

1363

        (void)memcpy(data, pData, len);

2,817,597✔

1365

        data = pData;

2,810,575✔

1368

    if (uidData == NULL){

6,375,165✔

1369

      STREAM_CHECK_NULL_GOTO(taosArrayPush(tagCache, &data), terrno);

12,690,629✔

1371

      void* pre = taosArrayGetP(tagCache, j);

32,404✔

1372

      taosMemoryFree(pre);

32,404✔

1373

      taosArraySet(tagCache, j, &data);

32,404✔

1375

    data = NULL;

6,376,519✔

1378

end:

45,186,955✔

1379

  taosMemoryFree(data);

45,186,033✔

1380

  api->metaReaderFn.clearReader(&mr);

45,174,527✔

1381

  if (lock != NULL) taosWUnLockLatch(lock);

45,178,173✔

1382

  return code;

45,186,238✔

1385

int32_t fillTag(SHashObj* metaCache, SExprInfo* pExprInfo, int32_t numOfExpr,

373,792,614✔

1387

  int32_t     code = 0;

373,792,614✔

1388

  int32_t     lino = 0;

373,792,614✔

1389

  SArray* tagCache = NULL;

373,792,614✔

1390

  if (numOfExpr == 0) {

373,792,614✔

1391

    return TSDB_CODE_SUCCESS;

41,957,386✔

1394

  taosRLockLatch(lock);

331,835,228✔

1395

  void* uidData = taosHashGet(metaCache, &uid, LONG_BYTES);

331,924,131✔

1396

  if (uidData == NULL) {

331,887,932✔

1399

    tagCache = *(SArray**)uidData;

331,887,932✔

1400

    if(taosArrayGetSize(tagCache) != numOfExpr) {

331,902,140✔

1406

  for (int32_t j = 0; j < numOfExpr; ++j) {

1,138,752,204✔

1407

    const SExprInfo* pExpr1 = &pExprInfo[j];

806,884,719✔

1408

    int32_t          dstSlotId = pExpr1->base.resSchema.slotId;

806,746,494✔

1410

    SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, dstSlotId);

806,886,523✔

1411

    STREAM_CHECK_NULL_GOTO(pColInfoData, terrno);

806,621,227✔

1412

    int32_t functionId = pExpr1->pExpr->_function.functionId;

806,621,227✔

1415

    if (fmIsScanPseudoColumnFunc(functionId)) {

806,634,858✔

1416

      int32_t fType = pExpr1->pExpr->_function.functionType;

4,265,898✔

1417

      if (fType == FUNCTION_TYPE_TBNAME) {

4,265,898✔

1418

        pColInfoData->info.colId = -1;

4,265,624✔

1421

    char* data = tagCache == NULL ? NULL : taosArrayGetP(tagCache, j);

806,439,569✔

1423

    bool isNullVal = (data == NULL) || (pColInfoData->info.type == TSDB_DATA_TYPE_JSON && tTagIsJsonNull(data));

806,337,157✔

1424

    if (isNullVal) {

806,386,173✔

1427

      if (!IS_VAR_DATA_TYPE(pColInfoData->info.type)) {

806,386,173✔

1428

        for (uint32_t i = 0; i < numOfRows; i++){

2,147,483,647✔

1429

          colDataClearNull_f(pColInfoData->nullbitmap, currentRow + i);

2,147,483,647✔

1432

      code = colDataSetNItems(pColInfoData, currentRow, (const char*)data, numOfRows, numOfBlocks, false);

806,405,164✔

1433

      STREAM_CHECK_RET_GOTO(code);

806,839,550✔

1436

end:

331,867,485✔

1437

  taosRUnLockLatch(lock);

331,867,834✔

1438

  return code;

331,926,245✔

1441

static int32_t processTag(SStreamTriggerReaderInfo* info, bool isCalc,

4,140,092✔

1443

  int32_t     code = 0;

4,140,092✔

1444

  int32_t     lino = 0;

4,140,092✔

1446

  void* pTask = info->pTask;

4,140,092✔

1447

  ST_TASK_DLOG("%s start. rows:%" PRIu32 ",uid:%"PRIu64, __func__,  numOfRows, uid);

4,140,092✔

1449

  SHashObj* metaCache = isCalc ? info->pTableMetaCacheCalc : info->pTableMetaCacheTrigger;

4,140,384✔

1450

  SExprInfo*   pExprInfo = isCalc ? info->pExprInfoCalcTag : info->pExprInfoTriggerTag;

4,140,381✔

1451

  int32_t      numOfExpr = isCalc ? info->numOfExprCalcTag : info->numOfExprTriggerTag;

4,140,658✔

1453

  code = fillTag(metaCache, pExprInfo, numOfExpr, uid, pBlock, currentRow, numOfRows, numOfBlocks, &info->lock);

4,140,658✔

1454

  STREAM_CHECK_RET_GOTO(code);

4,142,847✔

1456

end:

4,142,847✔

1457

  return code;

4,142,847✔

1505

static int32_t getColId(int64_t suid, int64_t uid, int16_t i, SStreamTriggerReaderInfo* sStreamReaderInfo, SSTriggerWalNewRsp* rsp, int16_t* colId) {

24,076,083✔

1506

  int32_t code = 0;

24,076,083✔

1507

  int32_t lino = 0;

24,076,083✔

1508

  int64_t id[2] = {suid, uid};

24,076,083✔

1509

  taosRLockLatch(&sStreamReaderInfo->lock);

24,081,285✔

1510

  void *px = tSimpleHashGet(rsp->isCalc ? sStreamReaderInfo->uidHashCalc : sStreamReaderInfo->uidHashTrigger, id, sizeof(id));

24,080,731✔

1511

  STREAM_CHECK_NULL_GOTO(px, TSDB_CODE_INVALID_PARA);

24,081,834✔

1512

  SSHashObj* uInfo = *(SSHashObj **)px;

24,081,834✔

1513

  STREAM_CHECK_NULL_GOTO(uInfo, TSDB_CODE_INVALID_PARA);

24,081,834✔

1514

  int16_t*  tmp = tSimpleHashGet(uInfo, &i, sizeof(i));

24,081,834✔

1515

  if (tmp != NULL) {

24,078,003✔

1516

    *colId = *tmp;

22,105,241✔

1518

    *colId = -1;

1,972,762✔

1521

end:

24,080,757✔

1522

  taosRUnLockLatch(&sStreamReaderInfo->lock);

24,079,936✔

1523

  return code;

24,080,731✔

1526

static int32_t getSchemas(SVnode* pVnode, int64_t suid, int64_t uid, int32_t sver, SStreamTriggerReaderInfo* sStreamReaderInfo, STSchema** schema) {

11,594,776✔

1527

  int32_t code = 0;

11,594,776✔

1528

  int32_t lino = 0;

11,594,776✔

1529

  int64_t id = suid != 0 ? suid : uid;

11,594,776✔

1530

  if (sStreamReaderInfo->isVtableStream) {

11,596,149✔

1531

    STSchema** schemaTmp = taosHashGet(sStreamReaderInfo->triggerTableSchemaMapVTable, &id, LONG_BYTES);

7,744,012✔

1532

    if (schemaTmp == NULL || *schemaTmp == NULL || (*schemaTmp)->version != sver) {

7,745,120✔

1533

      *schema = metaGetTbTSchema(pVnode->pMeta, id, sver, 1);

71,033✔

1534

      STREAM_CHECK_NULL_GOTO(*schema, terrno);

71,033✔

1535

      code = taosHashPut(sStreamReaderInfo->triggerTableSchemaMapVTable, &id, LONG_BYTES, schema, POINTER_BYTES);

71,033✔

1536

      if (code != 0) {

71,033✔

1541

      *schema = *schemaTmp;

7,673,256✔

1544

    if (sStreamReaderInfo->triggerTableSchema == NULL || sStreamReaderInfo->triggerTableSchema->version != sver) {

3,851,860✔

1545

      taosMemoryFree(sStreamReaderInfo->triggerTableSchema);

110,307✔

1546

      sStreamReaderInfo->triggerTableSchema = metaGetTbTSchema(pVnode->pMeta, id, sver, 1);

110,307✔

1547

      STREAM_CHECK_NULL_GOTO(sStreamReaderInfo->triggerTableSchema, terrno);

110,307✔

1549

    *schema = sStreamReaderInfo->triggerTableSchema;

3,852,411✔

1552

end:

11,596,694✔

1553

  return code;

11,596,694✔

1556

static int32_t scanSubmitTbData(SVnode* pVnode, SDecoder *pCoder, SStreamTriggerReaderInfo* sStreamReaderInfo,

12,849,709✔

1558

  int32_t code = 0;

12,849,709✔

1559

  int32_t lino = 0;

12,849,709✔

1560

  uint64_t id = 0;

12,849,709✔

1561

  WalMetaResult walMeta = {0};

12,851,088✔

1562

  void* pTask = sStreamReaderInfo->pTask;

12,851,109✔

1563

  SSDataBlock * pBlock = (SSDataBlock*)rsp->dataBlock;

12,851,109✔

1565

  if (tStartDecode(pCoder) < 0) {

12,850,299✔

1571

  SSubmitTbData submitTbData = {0};

12,850,820✔

1572

  uint8_t       version = 0;

12,849,170✔

1573

  if (tDecodeI32v(pCoder, &submitTbData.flags) < 0) {

12,851,337✔

1578

  version = (submitTbData.flags >> 8) & 0xff;

12,851,337✔

1579

  submitTbData.flags = submitTbData.flags & 0xff;

12,851,337✔

1581

  if (submitTbData.flags & SUBMIT_REQ_AUTO_CREATE_TABLE) {

12,851,337✔

1582

    if (tStartDecode(pCoder) < 0) {

248,136✔

1587

    tEndDecode(pCoder);

248,136✔

1591

  if (tDecodeI64(pCoder, &submitTbData.suid) < 0) {

12,849,952✔

1596

  if (tDecodeI64(pCoder, &submitTbData.uid) < 0) {

12,848,853✔

1602

  ST_TASK_DLOG("%s uid:%" PRId64 ", suid:%" PRId64 ", ver:%" PRId64, __func__, submitTbData.uid, submitTbData.suid, ver);

12,848,853✔

1604

  if (rsp->uidHash != NULL) {

12,849,164✔

1605

    uint64_t* gid = tSimpleHashGet(rsp->uidHash, &submitTbData.uid, LONG_BYTES);

9,499,317✔

1606

    STREAM_CHECK_CONDITION_GOTO(gid == NULL, TDB_CODE_SUCCESS);

9,498,227✔

1607

    ST_TASK_DLOG("%s get uid gid from uidHash, uid:%" PRId64 ", suid:%" PRId64 " gid:%"PRIu64, __func__, submitTbData.uid, submitTbData.suid, *gid);

9,498,227✔

1608

    id = *gid;

9,498,227✔

1610

    STREAM_CHECK_CONDITION_GOTO(!uidInTableListSet(sStreamReaderInfo, submitTbData.suid, submitTbData.uid, &id, rsp->isCalc), TDB_CODE_SUCCESS);

3,350,958✔

1613

  walMeta.id = id;

11,594,228✔

1614

  STimeWindow window = {.skey = INT64_MIN, .ekey = INT64_MAX};

11,594,228✔

1616

  if (ranges != NULL){

11,595,601✔

1617

    void* timerange = tSimpleHashGet(ranges, &id, sizeof(id));

9,497,128✔

1618

    if (timerange == NULL) goto end;;

9,499,046✔

1619

    int64_t* pRange = (int64_t*)timerange;

9,499,046✔

1620

    window.skey = pRange[0];

9,499,046✔

1621

    window.ekey = pRange[1];

9,499,046✔

1622

    ST_TASK_DLOG("%s get time range from ranges, uid:%" PRId64 ", suid:%" PRId64 ", gid:%" PRIu64 ", skey:%" PRId64 ", ekey:%" PRId64,

9,497,667✔

1626

  if (tDecodeI32v(pCoder, &submitTbData.sver) < 0) {

11,597,522✔

1632

  STSchema*    schema = NULL;

11,597,522✔

1633

  STREAM_CHECK_RET_GOTO(getSchemas(pVnode, submitTbData.suid, submitTbData.uid, submitTbData.sver, sStreamReaderInfo, &schema));

11,594,776✔

1635

  SStreamWalDataSlice* pSlice = (SStreamWalDataSlice*)tSimpleHashGet(rsp->indexHash, &submitTbData.uid, LONG_BYTES);

11,595,327✔

1636

  int32_t blockStart = 0;

11,597,528✔

1637

  int32_t numOfRows = 0;

11,597,528✔

1638

  if (submitTbData.flags & SUBMIT_REQ_COLUMN_DATA_FORMAT) {

11,596,435✔

1721

    uint64_t nRow = 0;

11,596,435✔

1722

    if (tDecodeU64v(pCoder, &nRow) < 0) {

11,595,050✔

1726

    for (uint64_t iRow = 0; iRow < nRow; ++iRow) {

24,813,462✔

1727

      SRow *pRow = (SRow *)(pCoder->data + pCoder->pos);

13,217,539✔

1728

      pCoder->pos += pRow->len;

13,219,183✔

1730

      if (iRow == 0){

13,217,222✔

1732

        walMeta.skey = pRow->ts;

11,598,076✔

1737

      if (iRow == nRow - 1) {

13,216,674✔

1739

        walMeta.ekey = pRow->ts;

11,595,598✔

1745

      if (pRow->ts < window.skey || pRow->ts > window.ekey) {

13,218,053✔

1746

        continue;

12,241✔

1748

      STREAM_CHECK_NULL_GOTO(pSlice, TSDB_CODE_INVALID_PARA);

13,206,387✔

1749

      blockStart = pSlice->currentRowIdx;

13,206,142✔

1751

      for (int16_t i = 0; i < taosArrayGetSize(pBlock->pDataBlock); i++) {  // reader todo test null

81,941,461✔

1752

        SColumnInfoData* pColData = taosArrayGet(pBlock->pDataBlock, i);

68,739,257✔

1753

        STREAM_CHECK_NULL_GOTO(pColData, terrno);

68,737,026✔

1754

        if (pColData->info.colId <= -1) {

68,737,026✔

1755

          pColData->hasNull = true;

23,359,685✔

1756

          continue;

23,359,408✔

1758

        int16_t colId = 0;

45,386,150✔

1759

        if (sStreamReaderInfo->isVtableStream){

45,385,596✔

1760

          STREAM_CHECK_RET_GOTO(getColId(submitTbData.suid, submitTbData.uid, i, sStreamReaderInfo, rsp, &colId));

24,078,840✔

1761

          ST_TASK_TLOG("%s vtable colId:%d, i:%d, uid:%" PRId64, __func__, colId, i, submitTbData.uid);

24,080,731✔

1763

          colId = pColData->info.colId;

21,304,853✔

1766

        SColVal colVal = {0};

45,386,403✔

1767

        int32_t sourceIdx = 0;

45,386,677✔

1769

          if (sourceIdx >= schema->numOfCols) {

123,634,734✔

1770

            break;

10,348,862✔

1772

          STREAM_CHECK_RET_GOTO(tRowGet(pRow, schema, sourceIdx, &colVal));

113,284,568✔

1773

          if (colVal.cid == colId) {

113,286,658✔

1774

            break;

35,038,601✔

1776

          sourceIdx++;

78,248,057✔

1778

        if (colVal.cid == colId && COL_VAL_IS_VALUE(&colVal)) {

45,387,463✔

1779

          if (IS_VAR_DATA_TYPE(colVal.value.type) || colVal.value.type == TSDB_DATA_TYPE_DECIMAL){

32,409,766✔

1780

            STREAM_CHECK_RET_GOTO(varColSetVarData(pColData, blockStart+ numOfRows, (const char*)colVal.value.pData, colVal.value.nData, !COL_VAL_IS_VALUE(&colVal)));

49,852✔

1781

            ST_TASK_TLOG("%s vtable colId:%d, i:%d, colData:%p, data:%s, len:%d, rowIndex:%d, offset:%d, uid:%" PRId64, __func__, colId, i, pColData,

50,926✔

1784

            STREAM_CHECK_RET_GOTO(colDataSetVal(pColData, blockStart + numOfRows, (const char*)(&(colVal.value.val)), !COL_VAL_IS_VALUE(&colVal)));

32,359,914✔

1787

          colDataSetNULL(pColData, blockStart + numOfRows);

12,977,697✔

1791

      numOfRows++;

13,206,171✔

1795

  if (numOfRows > 0) {

11,596,770✔

1796

    if (!sStreamReaderInfo->isVtableStream) {

11,596,770✔

1797

      STREAM_CHECK_RET_GOTO(processTag(sStreamReaderInfo, rsp->isCalc, submitTbData.uid, pBlock, blockStart, numOfRows, 1));

3,852,443✔

1800

    SColumnInfoData* pColData = taosArrayGetLast(pBlock->pDataBlock);

11,596,758✔

1801

    STREAM_CHECK_NULL_GOTO(pColData, terrno);

11,597,548✔

1802

    STREAM_CHECK_RET_GOTO(colDataSetNItems(pColData, blockStart, (const char*)&ver, numOfRows, 1, false));

11,597,548✔

1804

    STREAM_CHECK_NULL_GOTO(pSlice, TSDB_CODE_INVALID_PARA);

11,595,907✔

1805

    ST_TASK_DLOG("%s process submit data:skey %" PRId64 ", ekey %" PRId64 ", id %" PRIu64

11,595,907✔

1808

    pSlice->currentRowIdx += numOfRows;

11,596,184✔

1809

    pBlock->info.rows += numOfRows;

11,597,274✔

1815

  if (gidHash == NULL) goto end;

11,595,901✔

1817

  WalMetaResult* data = (WalMetaResult*)tSimpleHashGet(gidHash, &walMeta.id, LONG_BYTES);

2,097,963✔

1818

  if (data != NULL) {

2,097,689✔

1822

    STREAM_CHECK_RET_GOTO(tSimpleHashPut(gidHash, &walMeta.id, LONG_BYTES, &walMeta, sizeof(WalMetaResult)));

2,097,689✔

1825

end:

12,847,373✔

1826

  if (code != 0) {                                                             \

12,849,441✔

1827

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

245✔

1829

  tEndDecode(pCoder);

12,849,441✔

1830

  return code;

12,850,229✔

1832

static int32_t scanSubmitData(SVnode* pVnode, SStreamTriggerReaderInfo* sStreamReaderInfo,

12,847,813✔

1834

  int32_t  code = 0;

12,847,813✔

1835

  int32_t  lino = 0;

12,847,813✔

1836

  SDecoder decoder = {0};

12,847,813✔

1837

  SSHashObj* gidHash = NULL;

12,848,893✔

1838

  void* pTask = sStreamReaderInfo->pTask;

12,848,893✔

1840

  tDecoderInit(&decoder, data, len);

12,849,438✔

1841

  if (tStartDecode(&decoder) < 0) {

12,849,447✔

1846

  uint64_t nSubmitTbData = 0;

12,851,624✔

1847

  if (tDecodeU64v(&decoder, &nSubmitTbData) < 0) {

12,850,823✔

1852

  if (rsp->metaBlock != NULL){

12,850,823✔

1853

    gidHash = tSimpleHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT));

3,351,503✔

1854

    STREAM_CHECK_NULL_GOTO(gidHash, terrno);

3,349,314✔

1857

  for (uint64_t i = 0; i < nSubmitTbData; i++) {

25,699,994✔

1858

    STREAM_CHECK_RET_GOTO(scanSubmitTbData(pVnode, &decoder, sStreamReaderInfo, ranges, gidHash, rsp, ver));

12,848,908✔

1861

  tEndDecode(&decoder);

12,851,086✔

1863

  if (rsp->metaBlock != NULL){

12,849,978✔

1864

    STREAM_CHECK_RET_GOTO(blockDataEnsureCapacity(rsp->metaBlock, ((SSDataBlock*)rsp->metaBlock)->info.rows + tSimpleHashGetSize(gidHash)));

3,352,865✔

1865

    int32_t iter = 0;

3,350,125✔

1866

    void*   px = tSimpleHashIterate(gidHash, NULL, &iter);

3,350,399✔

1867

    while (px != NULL) {

5,451,410✔

1868

      WalMetaResult* pMeta = (WalMetaResult*)px;

2,098,782✔

1869

      STREAM_CHECK_RET_GOTO(buildWalMetaBlockNew(rsp->metaBlock, pMeta->id, pMeta->skey, pMeta->ekey, ver));

2,098,782✔

1870

      ((SSDataBlock*)rsp->metaBlock)->info.rows++;

2,099,607✔

1871

      rsp->totalRows++;

2,099,607✔

1872

      ST_TASK_DLOG("%s process meta data:skey %" PRId64 ", ekey %" PRId64 ", id %" PRIu64

2,099,607✔

1874

      px = tSimpleHashIterate(gidHash, px, &iter);

2,099,607✔

1879

end:

12,847,799✔

1880

  tSimpleHashCleanup(gidHash);

12,851,362✔

1881

  tDecoderClear(&decoder);

12,849,438✔

1882

  return code;

12,850,838✔

1885

static int32_t scanSubmitTbDataPre(SDecoder *pCoder, SStreamTriggerReaderInfo* sStreamReaderInfo, SSHashObj* ranges,

15,363,933✔

1887

  int32_t code = 0;

15,363,933✔

1888

  int32_t lino = 0;

15,363,933✔

1889

  void* pTask = sStreamReaderInfo->pTask;

15,363,933✔

1891

  if (tStartDecode(pCoder) < 0) {

15,365,300✔

1896

  SSubmitTbData submitTbData = {0};

15,371,614✔

1897

  uint8_t       version = 0;

15,368,865✔

1898

  if (tDecodeI32v(pCoder, &submitTbData.flags) < 0) {

15,371,626✔

1902

  version = (submitTbData.flags >> 8) & 0xff;

15,371,626✔

1903

  submitTbData.flags = submitTbData.flags & 0xff;

15,371,626✔

1906

  if (submitTbData.flags & SUBMIT_REQ_AUTO_CREATE_TABLE) {

15,371,626✔

1907

    submitTbData.pCreateTbReq = taosMemoryCalloc(1, sizeof(SVCreateTbReq));

1,118,481✔

1908

    STREAM_CHECK_NULL_GOTO(submitTbData.pCreateTbReq, terrno);

1,116,566✔

1909

    STREAM_CHECK_RET_GOTO(tDecodeSVCreateTbReq(pCoder, submitTbData.pCreateTbReq));

1,116,566✔

1910

    STREAM_CHECK_RET_GOTO(processAutoCreateTableNew(sStreamReaderInfo, submitTbData.pCreateTbReq, ver));

1,118,604✔

1914

  if (tDecodeI64(pCoder, &submitTbData.suid) < 0) {

15,366,679✔

1918

  if (tDecodeI64(pCoder, uid) < 0) {

15,367,236✔

1922

  ST_TASK_DLOG("%s uid:%" PRId64 ", suid:%" PRId64, __func__, *uid, submitTbData.suid);

15,367,236✔

1923

  STREAM_CHECK_CONDITION_GOTO(!uidInTableListSet(sStreamReaderInfo, submitTbData.suid, *uid, gid, rsp->isCalc), TDB_CODE_SUCCESS);

15,366,405✔

1924

  if (rsp->uidHash != NULL) {

11,600,498✔

1925

    STREAM_CHECK_RET_GOTO(tSimpleHashPut(rsp->uidHash, uid, LONG_BYTES, gid, LONG_BYTES));

9,499,040✔

1926

    ST_TASK_DLOG("%s put uid into uidHash, uid:%" PRId64 ", suid:%" PRId64 " gid:%"PRIu64, __func__, *uid, submitTbData.suid, *gid);

9,498,486✔

1928

  STimeWindow window = {.skey = INT64_MIN, .ekey = INT64_MAX};

11,600,769✔

1930

  if (ranges != NULL){

11,599,393✔

1931

    void* timerange = tSimpleHashGet(ranges, gid, sizeof(*gid));

9,499,040✔

1932

    if (timerange == NULL) goto end;;

9,499,046✔

1933

    int64_t* pRange = (int64_t*)timerange;

9,499,046✔

1934

    window.skey = pRange[0];

9,499,046✔

1935

    window.ekey = pRange[1];

9,499,046✔

1938

  if (tDecodeI32v(pCoder, &submitTbData.sver) < 0) {

11,602,970✔

1943

  if (submitTbData.flags & SUBMIT_REQ_COLUMN_DATA_FORMAT) {

11,602,970✔

1969

    uint64_t nRow = 0;

11,602,970✔

1970

    if (tDecodeU64v(pCoder, &nRow) < 0) {

11,600,501✔

1975

    if (window.skey != INT64_MIN || window.ekey != INT64_MAX) {

11,600,501✔

1976

      for (uint64_t iRow = 0; iRow < nRow; ++iRow) {

20,305,001✔

1977

        SRow *pRow = (SRow *)(pCoder->data + pCoder->pos);

10,805,379✔

1978

        pCoder->pos += pRow->len;

10,805,933✔

1979

        if (pRow->ts < window.skey || pRow->ts > window.ekey) {

10,805,656✔

1980

          continue;

12,241✔

1982

        (*numOfRows)++;

10,793,415✔

1985

      (*numOfRows) = nRow;

2,100,879✔

1989

end:

15,370,484✔

1990

  tDestroySVSubmitCreateTbReq(submitTbData.pCreateTbReq, TSDB_MSG_FLG_DECODE);

15,369,662✔

1991

  taosMemoryFreeClear(submitTbData.pCreateTbReq);

15,365,808✔

1992

  tEndDecode(pCoder);

15,365,257✔

1993

  return code;

15,367,778✔

1996

static int32_t scanSubmitDataPre(SStreamTriggerReaderInfo* sStreamReaderInfo, void* data, int32_t len, SSHashObj* ranges, SSTriggerWalNewRsp* rsp, int64_t ver) {

15,365,026✔

1997

  int32_t  code = 0;

15,365,026✔

1998

  int32_t  lino = 0;

15,365,026✔

1999

  SDecoder decoder = {0};

15,365,026✔

2000

  void* pTask = sStreamReaderInfo->pTask;

15,366,947✔

2002

  tDecoderInit(&decoder, data, len);

15,368,868✔

2003

  if (tStartDecode(&decoder) < 0) {

15,367,230✔

2008

  uint64_t nSubmitTbData = 0;

15,368,286✔

2009

  if (tDecodeU64v(&decoder, &nSubmitTbData) < 0) {

15,368,001✔

2013

  ST_TASK_DLOG("%s nSubmitTbData:%" PRIu64 ", ver:%"PRId64 " bodyLen:%d", __func__, nSubmitTbData, ver, len);

15,368,001✔

2015

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

30,736,931✔

2016

    uint64_t gid = -1;

15,366,427✔

2017

    int64_t  uid = 0;

15,365,885✔

2018

    int32_t numOfRows = 0;

15,368,086✔

2019

    STREAM_CHECK_RET_GOTO(scanSubmitTbDataPre(&decoder, sStreamReaderInfo, ranges, &gid, &uid, &numOfRows, rsp, ver));

15,366,688✔

2020

    if (numOfRows <= 0) {

15,367,630✔

2021

      ST_TASK_DLOG("%s no valid data uid:%" PRId64 ", gid:%" PRIu64 ", numOfRows:%d, ver:%"PRId64, __func__, uid, gid, numOfRows, ver);

3,770,146✔

2022

      continue;

3,770,299✔

2024

    rsp->totalRows += numOfRows;

11,597,484✔

2025

    rsp->totalDataRows += numOfRows;

11,598,580✔

2027

    SStreamWalDataSlice* pSlice = (SStreamWalDataSlice*)tSimpleHashGet(rsp->indexHash, &uid, LONG_BYTES);

11,599,648✔

2028

    if (pSlice != NULL) {

11,601,880✔

2029

      pSlice->numRows += numOfRows;

10,782,672✔

2030

      ST_TASK_DLOG("%s again uid:%" PRId64 ", gid:%" PRIu64 ", total numOfRows:%d, hash:%p %d, ver:%"PRId64, __func__, uid, gid, pSlice->numRows, rsp->indexHash, tSimpleHashGetSize(rsp->indexHash), ver);

10,781,573✔

2031

      pSlice->gId = gid;

10,781,573✔

2033

      SStreamWalDataSlice tmp = {.gId=gid,.numRows=numOfRows,.currentRowIdx=0,.startRowIdx=0};

819,208✔

2034

      ST_TASK_DLOG("%s first uid:%" PRId64 ", gid:%" PRIu64 ", numOfRows:%d, hash:%p %d, ver:%"PRId64, __func__, uid, gid, tmp.numRows, rsp->indexHash, tSimpleHashGetSize(rsp->indexHash), ver);

818,934✔

2035

      STREAM_CHECK_RET_GOTO(tSimpleHashPut(rsp->indexHash, &uid, LONG_BYTES, &tmp, sizeof(tmp)));

818,934✔

2039

  tEndDecode(&decoder);

15,370,504✔

2041

end:

15,370,807✔

2042

  tDecoderClear(&decoder);

15,369,976✔

2043

  return code;

15,372,165✔

2046

static void buildIndexHash(SSHashObj* indexHash, void* pTask){

686,729✔

2047

  void*   pe = NULL;

686,729✔

2048

  int32_t iter = 0;

686,729✔

2049

  int32_t index = 0;

686,729✔

2050

  while ((pe = tSimpleHashIterate(indexHash, pe, &iter)) != NULL) {

1,505,666✔

2051

    SStreamWalDataSlice* pInfo = (SStreamWalDataSlice*)pe;

818,937✔

2052

    pInfo->startRowIdx = index;

818,937✔

2053

    pInfo->currentRowIdx = index;

818,937✔

2054

    index += pInfo->numRows;

818,937✔

2055

    ST_TASK_DLOG("%s uid:%" PRId64 ", gid:%" PRIu64 ", startRowIdx:%d, numRows:%d", __func__, *(int64_t*)(tSimpleHashGetKey(pe, NULL)),

1,436,254✔

2058

686,729✔

2060

static void printIndexHash(SSHashObj* indexHash, void* pTask){

686,484✔

2061

  if (qDebugFlag & DEBUG_TRACE) {

686,484✔

2062

    void*   pe = NULL;

7,458✔

2063

    int32_t iter = 0;

7,458✔

2064

    while ((pe = tSimpleHashIterate(indexHash, pe, &iter)) != NULL) {

18,514✔

2065

      SStreamWalDataSlice* pInfo = (SStreamWalDataSlice*)pe;

11,056✔

2066

      ST_TASK_TLOG("%s uid:%" PRId64 ", gid:%" PRIu64 ", startRowIdx:%d, numRows:%d", __func__, *(int64_t*)(tSimpleHashGetKey(pe, NULL)),

11,316✔

2070

686,484✔

2072

static void filterIndexHash(SSHashObj* indexHash, SColumnInfoData* pRet){

13,791✔

2073

  void*   pe = NULL;

13,791✔

2074

  int32_t iter = 0;

13,791✔

2075

  int32_t index = 0;

13,791✔

2076

  int32_t pIndex = 0;

13,791✔

2077

  int8_t* pIndicator = (int8_t*)pRet->pData;

13,791✔

2078

  while ((pe = tSimpleHashIterate(indexHash, pe, &iter)) != NULL) {

29,462✔

2079

    SStreamWalDataSlice* pInfo = (SStreamWalDataSlice*)pe;

15,671✔

2080

    pInfo->startRowIdx = index;

15,671✔

2081

    int32_t size = pInfo->numRows;

15,671✔

2082

    for (int32_t i = 0; i < pInfo->numRows; i++) {

223,795✔

2083

      if (pIndicator && !pIndicator[pIndex++]) {

208,124✔

2084

        size--;

69,530✔

2087

    pInfo->numRows = size;

15,671✔

2088

    index += pInfo->numRows;

15,671✔

2089

    stTrace("stream reader re build index hash uid:%" PRId64 ", gid:%" PRIu64 ", startRowIdx:%d, numRows:%d", *(int64_t*)(tSimpleHashGetKey(pe, NULL)),

15,671✔

2092

13,791✔

2094

static int32_t prepareIndexMetaData(SWalReader* pWalReader, SStreamTriggerReaderInfo* sStreamReaderInfo, SSTriggerWalNewRsp* resultRsp){

6,385,591✔

2095

  int32_t      code = 0;

6,385,591✔

2096

  int32_t      lino = 0;

6,385,591✔

2097

  void* pTask = sStreamReaderInfo->pTask;

6,385,591✔

2099

  code = walReaderSeekVer(pWalReader, resultRsp->ver);

6,386,019✔

2100

  if (code == TSDB_CODE_WAL_LOG_NOT_EXIST){

6,382,666✔

2101

    if (resultRsp->ver < walGetFirstVer(pWalReader->pWal)) {

5,648,058✔

2105

      resultRsp->verTime = taosGetTimestampUs();

5,647,851✔

2107

    ST_TASK_DLOG("%s scan wal end:%s",  __func__, tstrerror(code));

5,648,405✔

2108

    code = TSDB_CODE_SUCCESS;

5,648,354✔

2109

    goto end;

5,648,354✔

2111

  STREAM_CHECK_RET_GOTO(code);

734,608✔

2113

  while (1) {

6,073,419✔

2114

    code = walNextValidMsg(pWalReader, true);

6,808,027✔

2115

    if (code == TSDB_CODE_WAL_LOG_NOT_EXIST){

6,803,791✔

2116

      resultRsp->verTime = taosGetTimestampUs();

735,797✔

2117

      ST_TASK_DLOG("%s scan wal end:%s", __func__, tstrerror(code));

735,797✔

2118

      code = TSDB_CODE_SUCCESS;

735,797✔

2119

      goto end;

735,797✔

2121

    STREAM_CHECK_RET_GOTO(code);

6,065,331✔

2122

    resultRsp->ver = pWalReader->curVersion;

6,065,331✔

2123

    SWalCont* wCont = &pWalReader->pHead->head;

6,069,075✔

2124

    resultRsp->verTime = wCont->ingestTs;

6,066,995✔

2125

    void*   data = POINTER_SHIFT(wCont->body, sizeof(SMsgHead));

6,069,578✔

2126

    int32_t len = wCont->bodyLen - sizeof(SMsgHead);

6,070,406✔

2127

    int64_t ver = wCont->version;

6,069,093✔

2128

    ST_TASK_DLOG("%s scan wal ver:%" PRId64 ", type:%s, deleteData:%d, deleteTb:%d, msg len:%d", __func__,

6,069,589✔

2130

    if (wCont->msgType == TDMT_VND_SUBMIT) {

6,069,857✔

2132

      if (sStreamReaderInfo->isVtableStream && resultRsp->tableBlock != NULL && ((SSDataBlock*)resultRsp->tableBlock)->info.rows > 0) {

5,870,662✔

2136

      data = POINTER_SHIFT(wCont->body, sizeof(SSubmitReq2Msg));

5,872,854✔

2137

      len = wCont->bodyLen - sizeof(SSubmitReq2Msg);

5,872,306✔

2138

      STREAM_CHECK_RET_GOTO(scanSubmitDataPre(sStreamReaderInfo, data, len, NULL, resultRsp, ver));

5,873,128✔

2140

      STREAM_CHECK_RET_GOTO(processMeta(wCont->msgType, sStreamReaderInfo, data, len, resultRsp, ver));

199,198✔

2143

    ST_TASK_DLOG("%s scan wal next ver:%" PRId64 ", totalRows:%d", __func__, resultRsp->ver, resultRsp->totalRows);

6,069,876✔

2144

    if (resultRsp->totalRows >= STREAM_RETURN_ROWS_NUM || resultRsp->needReturn) {

6,069,854✔

2150

  STREAM_PRINT_LOG_END(code, lino);

6,384,145✔

2151

  return code;

6,384,419✔

2154

static int32_t prepareIndexData(SWalReader* pWalReader, SStreamTriggerReaderInfo* sStreamReaderInfo,

17,645,144✔

2156

  int32_t      code = 0;

17,645,144✔

2157

  int32_t      lino = 0;

17,645,144✔

2158

  void* pTask = sStreamReaderInfo->pTask;

17,645,144✔

2160

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

27,143,347✔

2161

    int64_t *ver = taosArrayGet(versions, i);

9,497,107✔

2162

    if (ver == NULL) continue;

9,497,932✔

2164

    STREAM_CHECK_RET_GOTO(walFetchHead(pWalReader, *ver));

9,497,932✔

2165

    if(pWalReader->pHead->head.msgType != TDMT_VND_SUBMIT) {

9,498,209✔

2170

    STREAM_CHECK_RET_GOTO(walFetchBody(pWalReader));

9,498,763✔

2172

    SWalCont* wCont = &pWalReader->pHead->head;

9,499,040✔

2173

    void*   pBody = POINTER_SHIFT(wCont->body, sizeof(SSubmitReq2Msg));

9,499,040✔

2174

    int32_t bodyLen = wCont->bodyLen - sizeof(SSubmitReq2Msg);

9,499,040✔

2176

    STREAM_CHECK_RET_GOTO(scanSubmitDataPre(sStreamReaderInfo, pBody, bodyLen, ranges, rsp, *ver));

9,499,040✔

2179

end:

17,645,969✔

2180

  return code;

17,645,969✔

2183

static int32_t filterData(SSTriggerWalNewRsp* resultRsp, SStreamTriggerReaderInfo* sStreamReaderInfo) {

686,755✔

2184

  int32_t      code = 0;

686,755✔

2185

  int32_t       lino = 0;

686,755✔

2186

  SColumnInfoData* pRet = NULL;

686,755✔

2188

  int64_t totalRows = ((SSDataBlock*)resultRsp->dataBlock)->info.rows;

686,755✔

2189

  STREAM_CHECK_RET_GOTO(qStreamFilter(((SSDataBlock*)resultRsp->dataBlock), sStreamReaderInfo->pFilterInfo, &pRet));

686,755✔

2191

  if (((SSDataBlock*)resultRsp->dataBlock)->info.rows < totalRows) {

686,755✔

2192

    filterIndexHash(resultRsp->indexHash, pRet);

13,791✔

2195

end:

686,484✔

2196

  colDataDestroy(pRet);

686,484✔

2197

  taosMemoryFree(pRet);

686,484✔

2198

  return code;

686,484✔

2201

static int32_t processWalVerMetaDataNew(SVnode* pVnode, SStreamTriggerReaderInfo* sStreamReaderInfo,

6,384,062✔

2203

  int32_t      code = 0;

6,384,062✔

2204

  int32_t      lino = 0;

6,384,062✔

2205

  void* pTask = sStreamReaderInfo->pTask;

6,384,062✔

2207

  SWalReader* pWalReader = walOpenReader(pVnode->pWal, 0);

6,385,177✔

2208

  STREAM_CHECK_NULL_GOTO(pWalReader, terrno);

6,382,759✔

2209

  blockDataEmpty(resultRsp->dataBlock);

6,382,759✔

2210

  blockDataEmpty(resultRsp->metaBlock);

6,381,070✔

2211

  int64_t lastVer = resultRsp->ver;

6,380,458✔

2212

  STREAM_CHECK_RET_GOTO(prepareIndexMetaData(pWalReader, sStreamReaderInfo, resultRsp));

6,380,986✔

2213

  STREAM_CHECK_CONDITION_GOTO(resultRsp->totalRows == 0, TDB_CODE_SUCCESS);

6,384,425✔

2215

  buildIndexHash(resultRsp->indexHash, pTask);

166,129✔

2216

  STREAM_CHECK_RET_GOTO(blockDataEnsureCapacity(((SSDataBlock*)resultRsp->dataBlock), resultRsp->totalRows));

166,129✔

2217

  while(lastVer < resultRsp->ver) {

3,566,862✔

2218

    STREAM_CHECK_RET_GOTO(walFetchHead(pWalReader, lastVer++));

3,401,803✔

2219

    if(pWalReader->pHead->head.msgType != TDMT_VND_SUBMIT) {

3,401,831✔

2220

      TAOS_CHECK_RETURN(walSkipFetchBody(pWalReader));

51,418✔

2221

      continue;

51,418✔

2223

    STREAM_CHECK_RET_GOTO(walFetchBody(pWalReader));

3,351,235✔

2224

    SWalCont* wCont = &pWalReader->pHead->head;

3,350,136✔

2225

    void*   pBody = POINTER_SHIFT(wCont->body, sizeof(SSubmitReq2Msg));

3,350,684✔

2226

    int32_t bodyLen = wCont->bodyLen - sizeof(SSubmitReq2Msg);

3,351,229✔

2227

    ST_TASK_DLOG("process wal ver:%" PRId64 ", type:%d, bodyLen:%d", wCont->version, wCont->msgType, bodyLen);

3,350,955✔

2228

    STREAM_CHECK_RET_GOTO(scanSubmitData(pVnode, sStreamReaderInfo, pBody, bodyLen, NULL, resultRsp, wCont->version));

3,351,232✔

2231

  int32_t metaRows = resultRsp->totalRows - ((SSDataBlock*)resultRsp->dataBlock)->info.rows;

165,884✔

2232

  STREAM_CHECK_RET_GOTO(filterData(resultRsp, sStreamReaderInfo));

165,884✔

2233

  resultRsp->totalRows = ((SSDataBlock*)resultRsp->dataBlock)->info.rows + metaRows;

165,884✔

2235

end:

6,384,970✔

2236

  ST_TASK_DLOG("vgId:%d %s end, get result totalRows:%d, process:%"PRId64"/%"PRId64, TD_VID(pVnode), __func__,

6,384,970✔

2238

  walCloseReader(pWalReader);

6,384,970✔

2239

  return code;

6,383,052✔

2242

static int32_t processWalVerDataNew(SVnode* pVnode, SStreamTriggerReaderInfo* sStreamReaderInfo,

17,645,144✔

2244

  int32_t      code = 0;

17,645,144✔

2245

  int32_t      lino = 0;

17,645,144✔

2247

  void* pTask = sStreamReaderInfo->pTask;

17,645,144✔

2248

  SWalReader* pWalReader = walOpenReader(pVnode->pWal, 0);

17,646,794✔

2249

  STREAM_CHECK_NULL_GOTO(pWalReader, terrno);

17,644,319✔

2251

  if (taosArrayGetSize(versions) > 0) {

17,644,319✔

2252

    rsp->ver = *(int64_t*)taosArrayGetLast(versions);

520,871✔

2255

  STREAM_CHECK_RET_GOTO(prepareIndexData(pWalReader, sStreamReaderInfo, versions, ranges, rsp));

17,646,246✔

2256

  STREAM_CHECK_CONDITION_GOTO(rsp->totalRows == 0, TDB_CODE_SUCCESS);

17,645,144✔

2258

  ST_TASK_TLOG("%s index hash:%p %d", __func__, rsp->indexHash, tSimpleHashGetSize(rsp->indexHash));

520,871✔

2259

  buildIndexHash(rsp->indexHash, pTask);

520,871✔

2261

  blockDataEmpty(rsp->dataBlock);

520,600✔

2262

  STREAM_CHECK_RET_GOTO(blockDataEnsureCapacity(rsp->dataBlock, rsp->totalRows));

520,871✔

2264

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

10,018,286✔

2265

    int64_t *ver = taosArrayGet(versions, i);

9,497,667✔

2266

    if (ver == NULL) continue;

9,498,209✔

2267

    ST_TASK_TLOG("vgId:%d %s scan wal process:%"PRId64"/%"PRId64, TD_VID(pVnode), __func__, *ver, walGetAppliedVer(pWalReader->pWal));

9,498,209✔

2269

    STREAM_CHECK_RET_GOTO(walFetchHead(pWalReader, *ver));

9,498,209✔

2270

    if(pWalReader->pHead->head.msgType != TDMT_VND_SUBMIT) {

9,498,486✔

2274

    STREAM_CHECK_RET_GOTO(walFetchBody(pWalReader));

9,498,492✔

2275

    SWalCont* wCont = &pWalReader->pHead->head;

9,497,938✔

2276

    void*   pBody = POINTER_SHIFT(wCont->body, sizeof(SSubmitReq2Msg));

9,498,492✔

2277

    int32_t bodyLen = wCont->bodyLen - sizeof(SSubmitReq2Msg);

9,498,218✔

2279

    STREAM_CHECK_RET_GOTO(scanSubmitData(pVnode, sStreamReaderInfo, pBody, bodyLen, ranges, rsp, wCont->version));

9,498,495✔

2282

  STREAM_CHECK_RET_GOTO(filterData(rsp, sStreamReaderInfo));

520,594✔

2283

  rsp->totalRows = ((SSDataBlock*)rsp->dataBlock)->info.rows;

520,871✔

2285

end:

17,645,698✔

2286

  ST_TASK_DLOG("vgId:%d %s end, get result totalRows:%d, process:%"PRId64"/%"PRId64, TD_VID(pVnode), __func__,

17,645,698✔

2288

  walCloseReader(pWalReader);

17,650,648✔

2289

  return code;

17,648,173✔

2292

static int32_t buildScheamFromMeta(SVnode* pVnode, int64_t uid, SArray** schemas, SStorageAPI* api) {

656,020✔

2293

  int32_t code = 0;

656,020✔

2294

  int32_t lino = 0;

656,020✔

2295

  SMetaReader metaReader = {0};

656,020✔

2296

  *schemas = taosArrayInit(8, sizeof(SSchema));

656,020✔

2297

  STREAM_CHECK_NULL_GOTO(*schemas, terrno);

656,020✔

2299

  api->metaReaderFn.initReader(&metaReader, pVnode, META_READER_LOCK, &api->metaFn);

656,020✔

2300

  STREAM_CHECK_RET_GOTO(api->metaReaderFn.getTableEntryByUid(&metaReader, uid));

656,020✔

2302

  SSchemaWrapper* sSchemaWrapper = NULL;

652,720✔

2303

  if (metaReader.me.type == TD_CHILD_TABLE) {

652,720✔

2304

    int64_t suid = metaReader.me.ctbEntry.suid;

652,720✔

2305

    tDecoderClear(&metaReader.coder);

652,720✔

2306

    STREAM_CHECK_RET_GOTO(api->metaReaderFn.getTableEntryByUid(&metaReader, suid));

652,720✔

2307

    sSchemaWrapper = &metaReader.me.stbEntry.schemaRow;

652,720✔

2314

  for (size_t j = 0; j < sSchemaWrapper->nCols; j++) {

3,863,939✔

2315

    SSchema* s = sSchemaWrapper->pSchema + j;

3,211,219✔

2316

    STREAM_CHECK_NULL_GOTO(taosArrayPush(*schemas, s), terrno);

6,422,438✔

2319

end:

656,020✔

2320

  api->metaReaderFn.clearReader(&metaReader);

656,020✔

2321

  STREAM_PRINT_LOG_END(code, lino);

656,020✔

2322

  if (code != 0)  {

656,020✔

2323

    taosArrayDestroy(*schemas);

3,300✔

2324

    *schemas = NULL;

3,300✔

2326

  return code;

656,020✔

2329

static int32_t shrinkScheams(SArray* cols, SArray* schemas) {

652,720✔

2330

  int32_t code = 0;

652,720✔

2331

  int32_t lino = 0;

652,720✔

2332

  size_t  schemaLen = taosArrayGetSize(schemas);

652,720✔

2333

  STREAM_CHECK_RET_GOTO(taosArrayEnsureCap(schemas, schemaLen + taosArrayGetSize(cols)));

652,720✔

2334

  for (size_t i = 0; i < taosArrayGetSize(cols); i++) {

2,561,811✔

2335

    col_id_t* id = taosArrayGet(cols, i);

1,909,091✔

2336

    STREAM_CHECK_NULL_GOTO(id, terrno);

1,909,091✔

2337

    for (size_t i = 0; i < schemaLen; i++) {

5,006,857✔

2338

      SSchema* s = taosArrayGet(schemas, i);

5,006,857✔

2339

      STREAM_CHECK_NULL_GOTO(s, terrno);

5,006,857✔

2340

      if (*id == s->colId) {

5,006,857✔

2341

        STREAM_CHECK_NULL_GOTO(taosArrayPush(schemas, s), terrno);

1,909,091✔

2342

        break;

1,909,091✔

2346

  taosArrayPopFrontBatch(schemas, schemaLen);

652,720✔

2348

end:

652,720✔

2349

  return code;

652,720✔

2475

static int32_t processCalaTimeRange(SStreamTriggerReaderCalcInfo* sStreamReaderCalcInfo, SResFetchReq* req,

722,579✔

2477

  int32_t code = 0;

722,579✔

2478

  int32_t lino = 0;

722,579✔

2479

  void* pTask = sStreamReaderCalcInfo->pTask;

722,579✔

2480

  STimeWindow* pWin = isExtWin ? &handle->extWinRange : &handle->winRange;

722,579✔

2481

  bool* pValid = isExtWin ? &handle->extWinRangeValid : &handle->winRangeValid;

722,579✔

2483

  if (req->pStRtFuncInfo->withExternalWindow) {

722,579✔

2484

    sStreamReaderCalcInfo->tmpRtFuncInfo.curIdx = 0;

538,559✔

2485

    sStreamReaderCalcInfo->tmpRtFuncInfo.triggerType = req->pStRtFuncInfo->triggerType;

538,559✔

2486

    sStreamReaderCalcInfo->tmpRtFuncInfo.isWindowTrigger = req->pStRtFuncInfo->isWindowTrigger;

538,559✔

2487

    sStreamReaderCalcInfo->tmpRtFuncInfo.precision = req->pStRtFuncInfo->precision;

538,559✔

2489

    SSTriggerCalcParam* pFirst = NULL;

538,559✔

2490

    SSTriggerCalcParam* pLast = NULL;

538,559✔

2491

    if (req->pStRtFuncInfo->isMultiGroupCalc) {

538,559✔

2496

      pFirst = taosArrayGet(req->pStRtFuncInfo->pStreamPesudoFuncVals, 0);

538,559✔

2497

      pLast = taosArrayGetLast(req->pStRtFuncInfo->pStreamPesudoFuncVals);

538,559✔

2498

      STREAM_CHECK_NULL_GOTO(pFirst, terrno);

538,559✔

2499

      STREAM_CHECK_NULL_GOTO(pLast, terrno);

538,559✔

2502

    if (!node->needCalc) {

538,559✔

2503

      pWin->skey = pFirst->wstart;

341,263✔

2504

      pWin->ekey = pLast->wend;

341,263✔

2505

      *pValid = true;

341,263✔

2506

      if (req->pStRtFuncInfo->triggerType == STREAM_TRIGGER_SLIDING) {

341,263✔

2507

        pWin->ekey--;

194,787✔

2510

      SSTriggerCalcParam* pTmp = taosArrayGet(sStreamReaderCalcInfo->tmpRtFuncInfo.pStreamPesudoFuncVals, 0);

197,296✔

2511

      memcpy(pTmp, pFirst, sizeof(*pTmp));

197,296✔

2513

      STREAM_CHECK_RET_GOTO(streamCalcCurrWinTimeRange(node, &sStreamReaderCalcInfo->tmpRtFuncInfo, pWin, pValid, 1));

197,296✔

2514

      if (*pValid) {

197,296✔

2515

        int64_t skey = pWin->skey;

197,296✔

2517

        memcpy(pTmp, pLast, sizeof(*pTmp));

197,296✔

2518

        STREAM_CHECK_RET_GOTO(streamCalcCurrWinTimeRange(node, &sStreamReaderCalcInfo->tmpRtFuncInfo, pWin, pValid, 2));

197,296✔

2520

        if (*pValid) {

197,296✔

2521

          pWin->skey = skey;

197,296✔

2524

      pWin->ekey--;

197,296✔

2527

    if (!node->needCalc) {

184,020✔

2528

      SSTriggerCalcParam* pCurr = taosArrayGet(req->pStRtFuncInfo->pStreamPesudoFuncVals, req->pStRtFuncInfo->curIdx);

143,654✔

2529

      pWin->skey = pCurr->wstart;

143,654✔

2530

      pWin->ekey = pCurr->wend;

143,654✔

2531

      *pValid = true;

143,654✔

2532

      if (req->pStRtFuncInfo->triggerType == STREAM_TRIGGER_SLIDING) {

143,654✔

2533

        pWin->ekey--;

88,844✔

2536

      STREAM_CHECK_RET_GOTO(streamCalcCurrWinTimeRange(node, req->pStRtFuncInfo, pWin, pValid, 3));

40,366✔

2537

      pWin->ekey--;

40,366✔

2541

  ST_TASK_DLOG("%s type:%s, withExternalWindow:%d, skey:%" PRId64 ", ekey:%" PRId64 ", validRange:%d",

722,579✔

2544

end:

5,492✔

2546

  if (code) {

722,579✔

2550

  return code;

722,579✔

2553

static int32_t createDataBlockTsUid(SSDataBlock** pBlockRet, uint32_t numOfRows) {

629,532✔

2554

  int32_t      code = 0;

629,532✔

2555

  int32_t      lino = 0;

629,532✔

2556

  SSDataBlock* pBlock = NULL;

629,532✔

2557

  STREAM_CHECK_RET_GOTO(createDataBlock(&pBlock));

629,532✔

2558

  SColumnInfoData idata = createColumnInfoData(TSDB_DATA_TYPE_TIMESTAMP, LONG_BYTES, PRIMARYKEY_TIMESTAMP_COL_ID);

629,803✔

2559

  STREAM_CHECK_RET_GOTO(blockDataAppendColInfo(pBlock, &idata));

629,381✔

2560

  idata = createColumnInfoData(TSDB_DATA_TYPE_BIGINT, LONG_BYTES, PRIMARYKEY_TIMESTAMP_COL_ID + 1);

629,803✔

2561

  STREAM_CHECK_RET_GOTO(blockDataAppendColInfo(pBlock, &idata));

629,381✔

2562

  STREAM_CHECK_RET_GOTO(blockDataEnsureCapacity(pBlock, numOfRows));

629,532✔

2564

end:

629,529✔

2565

  STREAM_PRINT_LOG_END(code, lino)

629,529✔

2566

  if (code != TSDB_CODE_SUCCESS) {

629,529✔

2570

  *pBlockRet = pBlock;

629,529✔

2571

  return code;

629,529✔

2574

static int32_t processTsOutPutAllTables(SStreamTriggerReaderInfo* sStreamReaderInfo, SStreamTsResponse* tsRsp, SSDataBlock* pResBlock, int32_t order) {

365,041✔

2575

  int32_t code = 0;

365,041✔

2576

  int32_t lino = 0;

365,041✔

2577

  void* pTask = sStreamReaderInfo->pTask;

365,041✔

2579

  tsRsp->tsInfo = taosArrayInit(pResBlock->info.rows, sizeof(STsInfo));

365,041✔

2580

  STREAM_CHECK_NULL_GOTO(tsRsp->tsInfo, terrno);

365,041✔

2581

  SColumnInfoData* pColInfoDataTs = taosArrayGet(pResBlock->pDataBlock, 0);

365,041✔

2582

  SColumnInfoData* pColInfoDataUid = taosArrayGet(pResBlock->pDataBlock, 1);

365,041✔

2583

  for (int32_t j = 0; j < pResBlock->info.rows; j++) {

1,034,613✔

2584

    if (colDataIsNull_s(pColInfoDataTs, j) || pColInfoDataTs->pData == NULL) {

1,339,698✔

2587

    STsInfo* tsInfo = taosArrayReserve(tsRsp->tsInfo, 1);

669,849✔

2588

    STREAM_CHECK_NULL_GOTO(tsInfo, terrno)

669,849✔

2589

    if (order == TSDB_ORDER_ASC) {

669,849✔

2590

      tsInfo->ts = INT64_MAX;

355,175✔

2592

      tsInfo->ts = INT64_MIN;

314,674✔

2594

    int64_t ts = *(int64_t*)colDataGetNumData(pColInfoDataTs, j);

669,849✔

2595

    if (order == TSDB_ORDER_ASC && ts < tsInfo->ts) {

669,849✔

2596

      tsInfo->ts = ts;

355,175✔

2597

    } else if (order == TSDB_ORDER_DESC && ts > tsInfo->ts) {

314,674✔

2598

      tsInfo->ts = ts;

314,674✔

2600

    tsInfo->gId = *(int64_t*)colDataGetNumData(pColInfoDataUid, j);

669,849✔

2601

    ST_TASK_DLOG("%s get ts:%" PRId64 ", gId:%" PRIu64 ", ver:%" PRId64, __func__, tsInfo->ts, tsInfo->gId, tsRsp->ver);

669,849✔

2604

end:

365,041✔

2605

  return code;

365,041✔

2608

static int32_t processTsOutPutOneGroup(SStreamTriggerReaderInfo* sStreamReaderInfo, SStreamTsResponse* tsRsp, SSDataBlock* pResBlock, int32_t order) {

129,897✔

2609

  int32_t code = 0;

129,897✔

2610

  int32_t lino = 0;

129,897✔

2611

  void* pTask = sStreamReaderInfo->pTask;

129,897✔

2613

  tsRsp->tsInfo = taosArrayInit(1, sizeof(STsInfo));

130,168✔

2614

  STREAM_CHECK_NULL_GOTO(tsRsp->tsInfo, terrno);

129,897✔

2615

  STsInfo* tsInfo = taosArrayReserve(tsRsp->tsInfo, 1);

129,897✔

2616

  STREAM_CHECK_NULL_GOTO(tsInfo, terrno)

129,897✔

2617

  if (order == TSDB_ORDER_ASC) {

129,897✔

2618

    tsInfo->ts = INT64_MAX;

99,740✔

2620

    tsInfo->ts = INT64_MIN;

30,157✔

2623

  SColumnInfoData* pColInfoDataTs = taosArrayGet(pResBlock->pDataBlock, 0);

129,897✔

2624

  SColumnInfoData* pColInfoDataUid = taosArrayGet(pResBlock->pDataBlock, 1);

130,168✔

2625

  for (int32_t j = 0; j < pResBlock->info.rows; j++) {

281,543✔

2626

    if (colDataIsNull_s(pColInfoDataTs, j) || pColInfoDataTs->pData == NULL) {

303,569✔

2629

    int64_t ts = *(int64_t*)colDataGetNumData(pColInfoDataTs, j);

151,649✔

2630

    if (order == TSDB_ORDER_ASC && ts < tsInfo->ts) {

151,104✔

2631

      tsInfo->ts = ts;

99,740✔

2632

    } else if (order == TSDB_ORDER_DESC && ts > tsInfo->ts) {

51,090✔

2633

      tsInfo->ts = ts;

30,157✔

2636

  int64_t uid = *(int64_t*)colDataGetNumData(pColInfoDataUid, 0);

129,623✔

2637

  tsInfo->gId = qStreamGetGroupIdFromSet(sStreamReaderInfo, uid);

129,623✔

2638

  ST_TASK_DLOG("%s get ts:%" PRId64 ", gId:%" PRIu64 ", ver:%" PRId64, __func__, tsInfo->ts, tsInfo->gId, tsRsp->ver);

129,897✔

2640

end:

35,470✔

2641

  return code;

130,168✔

2644

static int32_t processTsOutPutAllGroups(SStreamTriggerReaderInfo* sStreamReaderInfo, SStreamTsResponse* tsRsp, SSDataBlock* pResBlock, int32_t order) {

9,069✔

2645

  int32_t code = 0;

9,069✔

2646

  int32_t lino = 0;

9,069✔

2647

  STableKeyInfo* pList = NULL;

9,069✔

2648

  StreamTableListInfo     tableInfo = {0};

9,069✔

2650

  void* pTask = sStreamReaderInfo->pTask;

9,069✔

2651

  STREAM_CHECK_RET_GOTO(qStreamCopyTableInfo(sStreamReaderInfo, &tableInfo));

9,069✔

2653

  SSHashObj*   uidTsHash = tSimpleHashInit(pResBlock->info.rows, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT));

9,069✔

2654

  STREAM_CHECK_NULL_GOTO(uidTsHash, terrno);

9,069✔

2655

  SColumnInfoData* pColInfoDataTs = taosArrayGet(pResBlock->pDataBlock, 0);

9,069✔

2656

  SColumnInfoData* pColInfoDataUid = taosArrayGet(pResBlock->pDataBlock, 1);

9,069✔

2657

  for (int32_t j = 0; j < pResBlock->info.rows; j++) {

28,230✔

2658

    if (colDataIsNull_s(pColInfoDataTs, j) || pColInfoDataTs->pData == NULL) {

38,322✔

2661

    int64_t ts = *(int64_t*)colDataGetNumData(pColInfoDataTs, j);

19,161✔

2662

    int64_t uid = *(int64_t*)colDataGetNumData(pColInfoDataUid, j);

19,161✔

2663

    STREAM_CHECK_RET_GOTO(tSimpleHashPut(uidTsHash, &uid, LONG_BYTES, &ts, LONG_BYTES));

19,161✔

2665

  tsRsp->tsInfo = taosArrayInit(qStreamGetTableListGroupNum(sStreamReaderInfo), sizeof(STsInfo));

9,069✔

2666

  STREAM_CHECK_NULL_GOTO(tsRsp->tsInfo, terrno);

9,069✔

2667

  while (true) {

18,221✔

2668

    int32_t        pNum = 0;

27,290✔

2669

    int64_t        suid = 0;

27,290✔

2670

    STREAM_CHECK_RET_GOTO(qStreamIterTableList(&tableInfo, &pList, &pNum, &suid));

27,290✔

2671

    if(pNum == 0) break;

27,290✔

2672

    STsInfo* tsInfo = taosArrayReserve(tsRsp->tsInfo, 1);

18,862✔

2673

    STREAM_CHECK_NULL_GOTO(tsInfo, terrno)

18,862✔

2674

    if (order == TSDB_ORDER_ASC) {

18,862✔

2675

      tsInfo->ts = INT64_MAX;

8,914✔

2677

      tsInfo->ts = INT64_MIN;

9,948✔

2679

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

37,533✔

2680

      int64_t uid = pList[i].uid;

19,312✔

2681

      int64_t *ts = tSimpleHashGet(uidTsHash, &uid, LONG_BYTES);

19,312✔

2682

      STREAM_CHECK_NULL_GOTO(ts, terrno);

19,312✔

2683

      if (order == TSDB_ORDER_ASC && *ts < tsInfo->ts) {

18,671✔

2684

        tsInfo->ts = *ts;

8,914✔

2685

      } else if (order == TSDB_ORDER_DESC && *ts > tsInfo->ts) {

9,757✔

2686

        tsInfo->ts = *ts;

9,307✔

2689

    int64_t uid = pList[0].uid;

18,221✔

2690

    tsInfo->gId = qStreamGetGroupIdFromSet(sStreamReaderInfo, uid);

18,221✔

2691

    ST_TASK_DLOG("%s get ts:%" PRId64 ", gId:%" PRIu64 ", ver:%" PRId64, __func__, tsInfo->ts, tsInfo->gId, tsRsp->ver);

18,221✔

2692

    taosMemoryFreeClear(pList);

18,221✔

2695

end:

9,069✔

2696

  qStreamDestroyTableInfo(&tableInfo);

9,069✔

2697

  taosMemoryFreeClear(pList);

9,069✔

2698

  tSimpleHashCleanup(uidTsHash);

9,069✔

2699

  return code;

9,069✔

2704

static int32_t getAllTs(SVnode* pVnode, SSDataBlock*  pResBlock, SStreamReaderTaskInner* pTaskInner, STableKeyInfo* pList, int32_t pNum) {

491,907✔

2705

  int32_t code = 0;

491,907✔

2706

  int32_t lino = 0;

491,907✔

2708

  stDebug("%s getAllTs enter: pNum:%d suid:%"PRId64" order:%d skey:%"PRId64" ekey:%"PRId64" verRange:[%"PRId64",%"PRId64"]",

491,907✔

2712

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

1,816,551✔

2713

    stDebug("%s getAllTs table[%d]: uid:%"PRId64, pTaskInner->idStr, i, pList[i].uid);

1,324,644✔

2716

  STREAM_CHECK_RET_GOTO(pTaskInner->storageApi->tsdReader.tsdCreateFirstLastTsIter(pVnode, &pTaskInner->options->twindows, &(SVersionRange){.minVer = -1, .maxVer = pTaskInner->options->ver},

491,907✔

2718

  bool hasNext = true;

491,257✔

2719

  int32_t iterCount = 0;

491,408✔

2721

    STREAM_CHECK_RET_GOTO(pTaskInner->storageApi->tsdReader.tsdNextFirstLastTsBlock(pTaskInner->pReader, pResBlock, &hasNext));

491,408✔

2722

    stDebug("%s getAllTs iter[%d]: hasNext:%d pResBlock->info.rows:%"PRId64, pTaskInner->idStr, iterCount++, hasNext, pResBlock->info.rows);

491,909✔

2723

    STREAM_CHECK_CONDITION_GOTO(!hasNext, TDB_CODE_SUCCESS);

491,633✔

2726

end:

491,633✔

2727

  stDebug("%s getAllTs done: code:%d pResBlock->info.rows:%"PRId64, pTaskInner->idStr, code, pResBlock ? pResBlock->info.rows : -1);

491,633✔

2728

  pTaskInner->storageApi->tsdReader.tsdDestroyFirstLastTsIter(pTaskInner->pReader);

491,633✔

2729

  pTaskInner->pReader = NULL;

492,178✔

2730

  return code;

491,090✔

2733

static int32_t processTsVTable(SVnode* pVnode, SStreamTsResponse* tsRsp, SStreamTriggerReaderInfo* sStreamReaderInfo,

295,333✔

2735

  int32_t code = 0;

295,333✔

2736

  int32_t lino = 0;

295,333✔

2737

  STableKeyInfo* pList = NULL;

295,333✔

2738

  StreamTableListInfo     tableInfo = {0};

295,333✔

2740

  void* pTask = sStreamReaderInfo->pTask;

295,333✔

2741

  STREAM_CHECK_RET_GOTO(qStreamCopyTableInfo(sStreamReaderInfo, &tableInfo));

295,333✔

2743

  SSDataBlock*  pResBlock = NULL;

295,333✔

2744

  STREAM_CHECK_RET_GOTO(createDataBlockTsUid(&pResBlock, qStreamGetTableListNum(sStreamReaderInfo)));

295,333✔

2746

  while (true) {

157,708✔

2747

    int32_t        pNum = 0;

453,041✔

2748

    int64_t        suid = 0;

453,041✔

2749

    STREAM_CHECK_RET_GOTO(qStreamIterTableList(&tableInfo, &pList, &pNum, &suid));

453,041✔

2750

    if(pNum == 0) break;

453,041✔

2751

    pTaskInner->options->suid = suid;

157,708✔

2752

    STREAM_CHECK_RET_GOTO(getAllTs(pVnode, pResBlock, pTaskInner, pList, pNum));

157,708✔

2753

    taosMemoryFreeClear(pList);

157,708✔

2756

  STREAM_CHECK_RET_GOTO(processTsOutPutAllTables(sStreamReaderInfo, tsRsp, pResBlock, pTaskInner->options->order));

295,333✔

2758

end:

295,333✔

2759

  qStreamDestroyTableInfo(&tableInfo);

295,333✔

2760

  taosMemoryFreeClear(pList);

295,333✔

2761

  blockDataDestroy(pResBlock);

295,333✔

2762

  STREAM_PRINT_LOG_END_WITHID(code, lino);

295,333✔

2763

  return code;

295,333✔

2766

static int32_t processTsNonVTable(SVnode* pVnode, SStreamTsResponse* tsRsp, SStreamTriggerReaderInfo* sStreamReaderInfo,

320,936✔

2768

  int32_t code = 0;

320,936✔

2769

  int32_t lino = 0;

320,936✔

2770

  STableKeyInfo* pList = NULL;

320,936✔

2771

  void* pTask = sStreamReaderInfo->pTask;

320,936✔

2773

  SSDataBlock*  pResBlock = NULL;

320,936✔

2775

  int32_t        pNum = 0;

320,936✔

2776

  int64_t        suid = 0;

320,936✔

2777

  STREAM_CHECK_RET_GOTO(qStreamGetTableList(sStreamReaderInfo, 0, &pList, &pNum));

320,936✔

2778

  ST_TASK_DLOG("vgId:%d %s qStreamGetTableList returned pNum:%d", TD_VID(pVnode), __func__, pNum);

320,785✔

2779

  STREAM_CHECK_CONDITION_GOTO(pNum == 0, TSDB_CODE_SUCCESS);

320,936✔

2780

  STREAM_CHECK_RET_GOTO(createDataBlockTsUid(&pResBlock, pNum));

283,275✔

2782

  pTaskInner->options->suid = sStreamReaderInfo->suid;

283,275✔

2783

  ST_TASK_DLOG("vgId:%d %s calling getAllTs: suid:%"PRId64" order:%d skey:%"PRId64" ekey:%"PRId64" ver:%"PRId64,

283,275✔

2786

  STREAM_CHECK_RET_GOTO(getAllTs(pVnode, pResBlock, pTaskInner, pList, pNum));

283,275✔

2787

  ST_TASK_DLOG("vgId:%d %s getAllTs done: pResBlock rows:%"PRId64, TD_VID(pVnode), __func__, pResBlock->info.rows);

283,006✔

2788

  int32_t order = pTaskInner->options->order;

283,275✔

2789

  if (pResBlock->info.rows == 0 && sStreamReaderInfo->groupByTbname) {

283,275✔

2790

    tsRsp->tsInfo = taosArrayInit(pNum, sizeof(STsInfo));

43,350✔

2791

    STREAM_CHECK_NULL_GOTO(tsRsp->tsInfo, terrno);

43,350✔

2792

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

159,804✔

2793

      STsInfo* tsInfo = taosArrayReserve(tsRsp->tsInfo, 1);

116,454✔

2794

      STREAM_CHECK_NULL_GOTO(tsInfo, terrno);

116,454✔

2795

      tsInfo->gId = pList[i].uid;

116,454✔

2796

      tsInfo->ts = 0;

116,454✔

2797

      ST_TASK_DLOG("%s no data but return gId (uid):%" PRIu64 " for tbname partition", __func__, tsInfo->gId);

116,454✔

2799

    goto end;

43,350✔

2802

  STREAM_CHECK_CONDITION_GOTO(pResBlock->info.rows == 0, TDB_CODE_SUCCESS);

239,925✔

2804

  if (sStreamReaderInfo->groupByTbname) {

157,750✔

2805

    STREAM_CHECK_RET_GOTO(processTsOutPutAllTables(sStreamReaderInfo, tsRsp, pResBlock, order));

69,708✔

2806

  } else if (sStreamReaderInfo->partitionCols == NULL) {

88,042✔

2807

    STREAM_CHECK_RET_GOTO(processTsOutPutOneGroup(sStreamReaderInfo, tsRsp, pResBlock, order));

78,973✔

2809

    STREAM_CHECK_RET_GOTO(processTsOutPutAllGroups(sStreamReaderInfo, tsRsp, pResBlock, order));

9,069✔

2811

end:

320,674✔

2812

  blockDataDestroy(pResBlock);

320,936✔

2813

  taosMemoryFreeClear(pList);

320,936✔

2814

  STREAM_PRINT_LOG_END_WITHID(code, lino);

320,936✔

2815

  return code;

320,936✔

2818

static int32_t processTsOnce(SVnode* pVnode, SStreamTsResponse* tsRsp, SStreamTriggerReaderInfo* sStreamReaderInfo,

68,761✔

2820

  int32_t code = 0;

68,761✔

2821

  int32_t lino = 0;

68,761✔

2822

  STableKeyInfo* pList = NULL;

68,761✔

2823

  void* pTask = sStreamReaderInfo->pTask;

68,761✔

2825

  SSDataBlock*  pResBlock = NULL;

69,032✔

2827

  int32_t        pNum = 0;

68,761✔

2828

  STREAM_CHECK_RET_GOTO(qStreamGetTableList(sStreamReaderInfo, gid, &pList, &pNum));

68,761✔

2829

  STREAM_CHECK_CONDITION_GOTO(pNum == 0, TSDB_CODE_SUCCESS);

68,761✔

2830

  STREAM_CHECK_RET_GOTO(createDataBlockTsUid(&pResBlock, pNum));

50,924✔

2832

  pTaskInner->options->suid = sStreamReaderInfo->suid;

50,650✔

2833

  STREAM_CHECK_RET_GOTO(getAllTs(pVnode, pResBlock, pTaskInner, pList, pNum));

50,921✔

2834

  STREAM_CHECK_CONDITION_GOTO(pResBlock->info.rows == 0, TDB_CODE_SUCCESS);

50,924✔

2835

  int32_t order = pTaskInner->options->order;

50,924✔

2837

  STREAM_CHECK_RET_GOTO(processTsOutPutOneGroup(sStreamReaderInfo, tsRsp, pResBlock, order));

51,195✔

2838

end:

69,032✔

2839

  blockDataDestroy(pResBlock);

69,032✔

2840

  taosMemoryFreeClear(pList);

68,490✔

2841

  STREAM_PRINT_LOG_END_WITHID(code, lino);

68,761✔

2842

  return code;

68,761✔

2845

static int32_t processTs(SVnode* pVnode, SStreamTsResponse* tsRsp, SStreamTriggerReaderInfo* sStreamReaderInfo,

616,118✔

2847

  if (sStreamReaderInfo->isVtableStream) {

616,118✔

2848

    return processTsVTable(pVnode, tsRsp, sStreamReaderInfo, pTaskInner);

295,333✔

2851

  return processTsNonVTable(pVnode, tsRsp, sStreamReaderInfo, pTaskInner);

320,936✔

2854

static int32_t vnodeProcessStreamSetTableReq(SVnode* pVnode, SRpcMsg* pMsg, SSTriggerPullRequestUnion* req, SStreamTriggerReaderInfo* sStreamReaderInfo) {

146,369✔

2855

  int32_t code = 0;

146,369✔

2856

  int32_t lino = 0;

146,369✔

2857

  void*   buf = NULL;

146,369✔

2858

  size_t  size = 0;

146,369✔

2859

  void* pTask = sStreamReaderInfo->pTask;

146,369✔

2861

  ST_TASK_DLOG("vgId:%d %s start, trigger hash size:%d, calc hash size:%d, appver:%"PRId64, TD_VID(pVnode), __func__,

146,369✔

2864

  taosWLockLatch(&sStreamReaderInfo->lock);

146,369✔

2865

  TSWAP(sStreamReaderInfo->uidHashTrigger, req->setTableReq.uidInfoTrigger);

146,369✔

2866

  TSWAP(sStreamReaderInfo->uidHashCalc, req->setTableReq.uidInfoCalc);

146,369✔

2867

  STREAM_CHECK_NULL_GOTO(sStreamReaderInfo->uidHashTrigger, TSDB_CODE_INVALID_PARA);

146,369✔

2868

  STREAM_CHECK_NULL_GOTO(sStreamReaderInfo->uidHashCalc, TSDB_CODE_INVALID_PARA);

146,369✔

2870

  qStreamClearTableInfo(&sStreamReaderInfo->vSetTableList);

146,369✔

2871

  STREAM_CHECK_RET_GOTO(initStreamTableListInfo(&sStreamReaderInfo->vSetTableList));

146,369✔

2872

  STREAM_CHECK_RET_GOTO(qBuildVTableList(sStreamReaderInfo));

146,369✔

2873

end:

146,369✔

2874

  taosWUnLockLatch(&sStreamReaderInfo->lock);

146,369✔

2875

  STREAM_PRINT_LOG_END_WITHID(code, lino);

146,369✔

2876

  SRpcMsg rsp = {

146,369✔

2878

  tmsgSendRsp(&rsp);

146,369✔

2879

  return code;

146,369✔

2882

static int32_t vnodeProcessStreamLastTsReq(SVnode* pVnode, SRpcMsg* pMsg, SSTriggerPullRequestUnion* req, SStreamTriggerReaderInfo* sStreamReaderInfo) {

351,360✔

2883

  int32_t                 code = 0;

351,360✔

2884

  int32_t                 lino = 0;

351,360✔

2885

  SStreamReaderTaskInner* pTaskInner = NULL;

351,360✔

2886

  SStreamTsResponse       tsRsp = {0};

351,844✔

2887

  void*                   buf = NULL;

351,844✔

2888

  size_t                  size = 0;

351,844✔

2890

  void* pTask = sStreamReaderInfo->pTask;

351,844✔

2892

  ST_TASK_DLOG("vgId:%d %s start", TD_VID(pVnode), __func__);

351,844✔

2894

  BUILD_OPTION(options, 0, sStreamReaderInfo->tableList.version, TSDB_ORDER_DESC, INT64_MIN, INT64_MAX, NULL, false, NULL);

351,844✔

2895

  STREAM_CHECK_RET_GOTO(createStreamTaskForTs(&options, &pTaskInner, &sStreamReaderInfo->storageApi));

351,844✔

2897

  tsRsp.ver = sStreamReaderInfo->tableList.version + 1;

351,844✔

2899

  STREAM_CHECK_RET_GOTO(processTs(pVnode, &tsRsp, sStreamReaderInfo, pTaskInner));

351,844✔

2901

end:

351,844✔

2902

  ST_TASK_DLOG("vgId:%d %s get result size:%"PRIzu", ver:%"PRId64, TD_VID(pVnode), __func__, taosArrayGetSize(tsRsp.tsInfo), tsRsp.ver);

351,844✔

2903

  code = buildTsRsp(&tsRsp, &buf, &size);

351,844✔

2904

  STREAM_PRINT_LOG_END_WITHID(code, lino);

351,575✔

2905

  SRpcMsg rsp = {

351,844✔

2907

  tmsgSendRsp(&rsp);

351,844✔

2908

  taosArrayDestroy(tsRsp.tsInfo);

351,844✔

2909

  taosMemoryFree(pTaskInner);

351,844✔

2910

  return code;

351,844✔

2913

static int32_t vnodeProcessStreamFirstTsReq(SVnode* pVnode, SRpcMsg* pMsg, SSTriggerPullRequestUnion* req, SStreamTriggerReaderInfo* sStreamReaderInfo) {

333,186✔

2914

  int32_t                 code = 0;

333,186✔

2915

  int32_t                 lino = 0;

333,186✔

2916

  SStreamReaderTaskInner* pTaskInner = NULL;

333,186✔

2917

  SStreamTsResponse       tsRsp = {0};

333,457✔

2918

  void*                   buf = NULL;

333,457✔

2919

  size_t                  size = 0;

333,457✔

2921

  void* pTask = sStreamReaderInfo->pTask;

333,457✔

2922

  ST_TASK_DLOG("vgId:%d %s start, startTime:%"PRId64" ver:%"PRId64" gid:%"PRId64

333,186✔

2927

  int32_t        pNum = 0;

333,186✔

2929

  tsRsp.ver = pVnode->state.applied;

333,186✔

2931

  BUILD_OPTION(options, 0, req->firstTsReq.ver, TSDB_ORDER_ASC, req->firstTsReq.startTime, INT64_MAX, NULL, false, NULL);

333,186✔

2932

  STREAM_CHECK_RET_GOTO(createStreamTaskForTs(&options, &pTaskInner, &sStreamReaderInfo->storageApi));

333,186✔

2934

  if (req->firstTsReq.gid != 0) {

332,915✔

2935

    STREAM_CHECK_RET_GOTO(processTsOnce(pVnode, &tsRsp, sStreamReaderInfo, pTaskInner, req->firstTsReq.gid));

68,761✔

2937

    STREAM_CHECK_RET_GOTO(processTs(pVnode, &tsRsp, sStreamReaderInfo, pTaskInner));

264,425✔

2940

end:

333,457✔

2941

  ST_TASK_DLOG("vgId:%d %s get result size:%"PRIzu", ver:%"PRId64, TD_VID(pVnode), __func__, taosArrayGetSize(tsRsp.tsInfo), tsRsp.ver);

333,457✔

2942

  code = buildTsRsp(&tsRsp, &buf, &size);

333,457✔

2943

  STREAM_PRINT_LOG_END_WITHID(code, lino);

333,186✔

2944

  SRpcMsg rsp = {

333,186✔

2946

  tmsgSendRsp(&rsp);

332,912✔

2947

  taosArrayDestroy(tsRsp.tsInfo);

333,457✔

2948

  taosMemoryFree(pTaskInner);

333,457✔

2949

  return code;

332,915✔

2952

static int32_t vnodeProcessStreamTsdbMetaReq(SVnode* pVnode, SRpcMsg* pMsg, SSTriggerPullRequestUnion* req, SStreamTriggerReaderInfo* sStreamReaderInfo) {

591,841✔

2953

  int32_t code = 0;

591,841✔

2954

  int32_t lino = 0;

591,841✔

2955

  void*   buf = NULL;

591,841✔

2956

  size_t  size = 0;

591,841✔

2957

  STableKeyInfo* pList = NULL;

591,841✔

2959

  void* pTask = sStreamReaderInfo->pTask;

591,841✔

2960

  ST_TASK_DLOG("vgId:%d %s start, ver:%" PRId64 ",skey:%" PRId64 ",ekey:%" PRId64 ",gid:%" PRId64, TD_VID(pVnode),

591,841✔

2964

  SStreamReaderTaskInner* pTaskInner = NULL;

591,841✔

2965

  int64_t                 key = getSessionKey(req->base.sessionId, STRIGGER_PULL_TSDB_META);

591,841✔

2967

  if (req->base.type == STRIGGER_PULL_TSDB_META) {

591,841✔

2968

    int32_t        pNum = 0;

591,841✔

2969

    STREAM_CHECK_RET_GOTO(qStreamGetTableList(sStreamReaderInfo, req->tsdbMetaReq.gid, &pList, &pNum));

591,841✔

2970

    BUILD_OPTION(options, getSuid(sStreamReaderInfo, pList), req->tsdbMetaReq.ver, req->tsdbMetaReq.order, req->tsdbMetaReq.startTime, req->tsdbMetaReq.endTime,

591,841✔

2972

    STREAM_CHECK_RET_GOTO(createStreamTask(pVnode, &options, &pTaskInner, NULL, pList, pNum, &sStreamReaderInfo->storageApi));

591,841✔

2973

    STREAM_CHECK_RET_GOTO(taosHashPut(sStreamReaderInfo->streamTaskMap, &key, LONG_BYTES, &pTaskInner, sizeof(pTaskInner)));

591,841✔

2975

    STREAM_CHECK_RET_GOTO(createBlockForTsdbMeta(&pTaskInner->pResBlockDst, sStreamReaderInfo->isVtableStream));

591,841✔

2983

  blockDataCleanup(pTaskInner->pResBlockDst);

591,841✔

2984

  STREAM_CHECK_RET_GOTO(blockDataEnsureCapacity(pTaskInner->pResBlockDst, STREAM_RETURN_ROWS_NUM));

591,841✔

2985

  bool hasNext = true;

591,841✔

2986

  while (true) {

326,784✔

2987

    STREAM_CHECK_RET_GOTO(getTableDataInfo(pTaskInner, &hasNext));

918,625✔

2988

    if (!hasNext) {

918,625✔

2989

      break;

591,841✔

2991

    pTaskInner->storageApi->tsdReader.tsdReaderReleaseDataBlock(pTaskInner->pReader);

326,784✔

2992

    pTaskInner->pResBlock->info.id.groupId = qStreamGetGroupIdFromSet(sStreamReaderInfo, pTaskInner->pResBlock->info.id.uid);

326,784✔

2994

    int32_t index = 0;

326,784✔

2995

    STREAM_CHECK_RET_GOTO(addColData(pTaskInner->pResBlockDst, index++, &pTaskInner->pResBlock->info.window.skey));

326,784✔

2996

    STREAM_CHECK_RET_GOTO(addColData(pTaskInner->pResBlockDst, index++, &pTaskInner->pResBlock->info.window.ekey));

326,784✔

2997

    STREAM_CHECK_RET_GOTO(addColData(pTaskInner->pResBlockDst, index++, &pTaskInner->pResBlock->info.id.uid));

326,784✔

2998

    if (!sStreamReaderInfo->isVtableStream) {

326,784✔

2999

      STREAM_CHECK_RET_GOTO(addColData(pTaskInner->pResBlockDst, index++, &pTaskInner->pResBlock->info.id.groupId));

85,280✔

3001

    STREAM_CHECK_RET_GOTO(addColData(pTaskInner->pResBlockDst, index++, &pTaskInner->pResBlock->info.rows));

326,784✔

3003

    stDebug("vgId:%d %s get  skey:%" PRId64 ", eksy:%" PRId64 ", uid:%" PRId64 ", gId:%" PRIu64 ", rows:%" PRId64,

326,784✔

3006

            pTaskInner->pResBlockDst->info.rows++;

326,784✔

3007

    if (pTaskInner->pResBlockDst->info.rows >= STREAM_RETURN_ROWS_NUM) {

326,784✔

3012

  ST_TASK_DLOG("vgId:%d %s get result rows:%" PRId64, TD_VID(pVnode), __func__, pTaskInner->pResBlockDst->info.rows);

591,841✔

3013

  STREAM_CHECK_RET_GOTO(buildRsp(pTaskInner->pResBlockDst, &buf, &size));

591,841✔

3014

  printDataBlock(pTaskInner->pResBlockDst, __func__, "meta", ((SStreamTask *)sStreamReaderInfo->pTask)->streamId);

591,841✔

3015

  if (!hasNext) {

591,841✔

3016

    STREAM_CHECK_RET_GOTO(taosHashRemove(sStreamReaderInfo->streamTaskMap, &key, LONG_BYTES));

591,841✔

3019

end:

591,841✔

3020

  STREAM_PRINT_LOG_END_WITHID(code, lino);

591,841✔

3021

  SRpcMsg rsp = {

591,841✔

3023

  tmsgSendRsp(&rsp);

591,841✔

3024

  taosMemoryFree(pList);

591,841✔

3025

  return code;

591,841✔

3028

static int32_t vnodeProcessStreamTsdbTsDataReqNonVTable(SVnode* pVnode, SRpcMsg* pMsg, SSTriggerPullRequestUnion* req, SStreamTriggerReaderInfo* sStreamReaderInfo) {

77,838✔

3029

  int32_t                 code = 0;

77,838✔

3030

  int32_t                 lino = 0;

77,838✔

3031

  SStreamReaderTaskInner* pTaskInner = NULL;

77,838✔

3032

  void*                   buf = NULL;

77,838✔

3033

  size_t                  size = 0;

77,838✔

3034

  SSDataBlock*            pBlockRes = NULL;

77,838✔

3036

  void* pTask = sStreamReaderInfo->pTask;

77,838✔

3037

  ST_TASK_DLOG("vgId:%d %s start, ver:%"PRId64",skey:%"PRId64",ekey:%"PRId64",uid:%"PRId64",suid:%"PRId64, TD_VID(pVnode), __func__, req->tsdbTsDataReq.ver,

77,838✔

3041

  int32_t        pNum = 1;

77,838✔

3042

  STableKeyInfo  pList = {.groupId = qStreamGetGroupIdFromSet(sStreamReaderInfo, req->tsdbTsDataReq.uid), .uid = req->tsdbTsDataReq.uid};

77,838✔

3043

  STREAM_CHECK_CONDITION_GOTO(pList.groupId == -1, TSDB_CODE_INVALID_PARA);

77,838✔

3044

  BUILD_OPTION(options, getSuid(sStreamReaderInfo, &pList), req->tsdbTsDataReq.ver, TSDB_ORDER_ASC, req->tsdbTsDataReq.skey, req->tsdbTsDataReq.ekey,

77,838✔

3046

  STREAM_CHECK_RET_GOTO(createStreamTask(pVnode, &options, &pTaskInner, sStreamReaderInfo->triggerResBlock, &pList, pNum, &sStreamReaderInfo->storageApi));

77,838✔

3047

  STREAM_CHECK_RET_GOTO(createOneDataBlock(sStreamReaderInfo->triggerResBlock, false, &pTaskInner->pResBlockDst));

77,838✔

3048

  STREAM_CHECK_RET_GOTO(createOneDataBlock(sStreamReaderInfo->tsBlock, false, &pBlockRes));

77,838✔

3050

  while (1) {

77,838✔

3051

    bool hasNext = false;

155,676✔

3052

    STREAM_CHECK_RET_GOTO(getTableDataInfo(pTaskInner, &hasNext));

155,676✔

3053

    if (!hasNext) {

155,676✔

3054

      break;

77,838✔

3057

    pTaskInner->pResBlock->info.id.groupId = qStreamGetGroupIdFromSet(sStreamReaderInfo, pTaskInner->pResBlock->info.id.uid);

77,838✔

3060

    SSDataBlock* pBlock = NULL;

77,838✔

3061

    STREAM_CHECK_RET_GOTO(getTableData(pTaskInner, &pBlock));

77,838✔

3062

    if (pBlock != NULL && pBlock->info.rows > 0) {

77,838✔

3063

      STREAM_CHECK_RET_GOTO(processTag(sStreamReaderInfo, false, pBlock->info.id.uid, pBlock,

77,838✔

3067

    STREAM_CHECK_RET_GOTO(qStreamFilter(pBlock, sStreamReaderInfo->pFilterInfo, NULL));

77,838✔

3068

    STREAM_CHECK_RET_GOTO(blockDataMerge(pTaskInner->pResBlockDst, pBlock));

77,838✔

3069

    ST_TASK_DLOG("vgId:%d %s get  skey:%" PRId64 ", eksy:%" PRId64 ", uid:%" PRId64 ", gId:%" PRIu64 ", rows:%" PRId64,

77,838✔

3074

  blockDataTransform(pBlockRes, pTaskInner->pResBlockDst);

77,838✔

3076

  ST_TASK_DLOG("vgId:%d %s get result rows:%" PRId64, TD_VID(pVnode), __func__, pTaskInner->pResBlockDst->info.rows);

77,838✔

3077

  STREAM_CHECK_RET_GOTO(buildRsp(pBlockRes, &buf, &size));

77,838✔

3079

end:

77,838✔

3080

  STREAM_PRINT_LOG_END_WITHID(code, lino);

77,838✔

3081

  SRpcMsg rsp = {

77,838✔

3083

  tmsgSendRsp(&rsp);

77,838✔

3084

  blockDataDestroy(pBlockRes);

77,838✔

3086

  releaseStreamTask(&pTaskInner);

77,838✔

3087

  return code;

77,838✔

3090

static int32_t vnodeProcessStreamTsdbTsDataReqVTable(SVnode* pVnode, SRpcMsg* pMsg, SSTriggerPullRequestUnion* req, SStreamTriggerReaderInfo* sStreamReaderInfo) {

3,300✔

3091

  int32_t                 code = 0;

3,300✔

3092

  int32_t                 lino = 0;

3,300✔

3093

  SStreamReaderTaskInner* pTaskInner = NULL;

3,300✔

3094

  void*                   buf = NULL;

3,300✔

3095

  size_t                  size = 0;

3,300✔

3096

  SSDataBlock*            pBlockRes = NULL;

3,300✔

3098

  void* pTask = sStreamReaderInfo->pTask;

3,300✔

3099

  ST_TASK_DLOG("vgId:%d %s start, ver:%"PRId64",skey:%"PRId64",ekey:%"PRId64",uid:%"PRId64",suid:%"PRId64, TD_VID(pVnode), __func__, req->tsdbTsDataReq.ver,

3,300✔

3103

  int32_t        pNum = 1;

3,300✔

3104

  STableKeyInfo  pList = {.groupId = qStreamGetGroupIdFromSet(sStreamReaderInfo, req->tsdbTsDataReq.uid), .uid = req->tsdbTsDataReq.uid};

3,300✔

3105

  STREAM_CHECK_CONDITION_GOTO(pList.groupId == -1, TSDB_CODE_INVALID_PARA);

3,300✔

3106

  BUILD_OPTION(options, getSuid(sStreamReaderInfo, &pList), req->tsdbTsDataReq.ver, TSDB_ORDER_ASC, req->tsdbTsDataReq.skey, req->tsdbTsDataReq.ekey,

3,300✔

3108

  STREAM_CHECK_RET_GOTO(createStreamTask(pVnode, &options, &pTaskInner, sStreamReaderInfo->tsBlock, &pList, pNum, &sStreamReaderInfo->storageApi));

3,300✔

3109

  STREAM_CHECK_RET_GOTO(createOneDataBlock(sStreamReaderInfo->tsBlock, false, &pBlockRes));

3,300✔

3111

  while (1) {

3,300✔

3112

    bool hasNext = false;

6,600✔

3113

    STREAM_CHECK_RET_GOTO(getTableDataInfo(pTaskInner, &hasNext));

6,600✔

3114

    if (!hasNext) {

6,600✔

3115

      break;

3,300✔

3118

    SSDataBlock* pBlock = NULL;

3,300✔

3119

    STREAM_CHECK_RET_GOTO(getTableData(pTaskInner, &pBlock));

3,300✔

3120

    STREAM_CHECK_RET_GOTO(blockDataMerge(pBlockRes, pBlock));

3,300✔

3121

    ST_TASK_DLOG("vgId:%d %s get  skey:%" PRId64 ", eksy:%" PRId64 ", uid:%" PRId64 ", gId:%" PRIu64 ", rows:%" PRId64,

3,300✔

3126

  ST_TASK_DLOG("vgId:%d %s get result rows:%" PRId64, TD_VID(pVnode), __func__, pBlockRes->info.rows);

3,300✔

3127

  STREAM_CHECK_RET_GOTO(buildRsp(pBlockRes, &buf, &size));

3,300✔

3129

end:

3,300✔

3130

  STREAM_PRINT_LOG_END_WITHID(code, lino);

3,300✔

3131

  SRpcMsg rsp = {

3,300✔

3133

  tmsgSendRsp(&rsp);

3,300✔

3134

  blockDataDestroy(pBlockRes);

3,300✔

3136

  releaseStreamTask(&pTaskInner);

3,300✔

3137

  return code;

3,300✔

3140

static int32_t vnodeProcessStreamTsdbTriggerDataReq(SVnode* pVnode, SRpcMsg* pMsg, SSTriggerPullRequestUnion* req, SStreamTriggerReaderInfo* sStreamReaderInfo) {

287,024✔

3141

  int32_t code = 0;

287,024✔

3142

  int32_t lino = 0;

287,024✔

3143

  void*   buf = NULL;

287,024✔

3144

  size_t  size = 0;

287,024✔

3145

  STableKeyInfo* pList = NULL;

287,024✔

3146

  SArray*        pResList = NULL;

287,024✔

3147

  SSDataBlock*   pBlockTmp = NULL;

287,024✔

3149

  SStreamReaderTaskInner* pTaskInner = NULL;

287,024✔

3150

  void* pTask = sStreamReaderInfo->pTask;

287,024✔

3151

  ST_TASK_DLOG("vgId:%d %s start. ver:%"PRId64",order:%d,startTs:%"PRId64",gid:%"PRId64, TD_VID(pVnode), __func__, req->tsdbTriggerDataReq.ver, req->tsdbTriggerDataReq.order, req->tsdbTriggerDataReq.startTime, req->tsdbTriggerDataReq.gid);

287,024✔

3153

  int64_t                 key = getSessionKey(req->base.sessionId, STRIGGER_PULL_TSDB_TRIGGER_DATA);

287,024✔

3155

  if (req->base.type == STRIGGER_PULL_TSDB_TRIGGER_DATA) {

287,024✔

3156

    int32_t        pNum = 0;

101,634✔

3157

    STREAM_CHECK_RET_GOTO(qStreamGetTableList(sStreamReaderInfo, req->tsdbTriggerDataReq.gid, &pList, &pNum));

101,634✔

3158

    BUILD_OPTION(options, getSuid(sStreamReaderInfo, pList), req->tsdbTriggerDataReq.ver, req->tsdbTriggerDataReq.order, req->tsdbTriggerDataReq.startTime, INT64_MAX,

101,634✔

3160

    STREAM_CHECK_RET_GOTO(createStreamTask(pVnode, &options, &pTaskInner, sStreamReaderInfo->triggerResBlock, pList, pNum, &sStreamReaderInfo->storageApi));

101,634✔

3161

    STREAM_CHECK_RET_GOTO(taosHashPut(sStreamReaderInfo->streamTaskMap, &key, LONG_BYTES, &pTaskInner, sizeof(pTaskInner)));

101,634✔

3163

    void** tmp = taosHashGet(sStreamReaderInfo->streamTaskMap, &key, LONG_BYTES);

185,390✔

3164

    STREAM_CHECK_NULL_GOTO(tmp, TSDB_CODE_STREAM_NO_CONTEXT);

185,390✔

3165

    pTaskInner = *(SStreamReaderTaskInner**)tmp;

57,622✔

3166

    STREAM_CHECK_NULL_GOTO(pTaskInner, TSDB_CODE_INTERNAL_ERROR);

57,622✔

3169

  blockDataCleanup(pTaskInner->pResBlockDst);

159,256✔

3170

  bool hasNext = true;

159,256✔

3171

  int32_t totalRows = 0;

159,256✔

3173

  pResList = taosArrayInit(4, POINTER_BYTES);

159,256✔

3174

  STREAM_CHECK_NULL_GOTO(pResList, terrno);

159,256✔

3175

  while (1) {

154,405✔

3176

    STREAM_CHECK_RET_GOTO(getTableDataInfo(pTaskInner, &hasNext));

313,661✔

3177

    if (!hasNext) {

313,372✔

3178

      break;

101,634✔

3180

    pTaskInner->pResBlock->info.id.groupId = qStreamGetGroupIdFromSet(sStreamReaderInfo, pTaskInner->pResBlock->info.id.uid);

211,738✔

3183

    SSDataBlock* pBlock = NULL;

212,027✔

3184

    STREAM_CHECK_RET_GOTO(getTableData(pTaskInner, &pBlock));

212,027✔

3185

    if (pBlock != NULL && pBlock->info.rows > 0) {

211,464✔

3186

      STREAM_CHECK_RET_GOTO(

211,738✔

3189

    STREAM_CHECK_RET_GOTO(qStreamFilter(pBlock, sStreamReaderInfo->pFilterInfo, NULL));

212,027✔

3191

    ST_TASK_DLOG("vgId:%d %s get result rows:%" PRId64, TD_VID(pVnode), __func__, pBlock->info.rows);

212,027✔

3192

    STREAM_CHECK_RET_GOTO(createOneDataBlock(pBlock, true, &pBlockTmp));

212,027✔

3193

    STREAM_CHECK_NULL_GOTO(taosArrayPush(pResList, &pBlockTmp), terrno);

212,027✔

3194

    totalRows += blockDataGetNumOfRows(pBlockTmp);

212,027✔

3195

    pBlockTmp = NULL;

212,027✔

3197

    ST_TASK_DLOG("vgId:%d %s get skey:%" PRId64 ", eksy:%" PRId64 ", uid:%" PRId64 ", gId:%" PRIu64 ", rows:%" PRId64,

212,027✔

3200

    if (totalRows >= STREAM_RETURN_ROWS_NUM) {  //todo optimize send multi blocks in one group

212,027✔

3201

      break;

57,622✔

3205

  STREAM_CHECK_RET_GOTO(buildArrayRsp(pResList, &buf, &size));

159,256✔

3206

  if (!hasNext) {

159,256✔

3207

    STREAM_CHECK_RET_GOTO(taosHashRemove(sStreamReaderInfo->streamTaskMap, &key, LONG_BYTES));

101,634✔

3210

end:

286,735✔

3211

  STREAM_PRINT_LOG_END_WITHID(code, lino);

286,735✔

3212

  SRpcMsg rsp = {

287,024✔

3214

  tmsgSendRsp(&rsp);

287,024✔

3215

  taosMemoryFree(pList);

287,024✔

3216

  blockDataDestroy(pBlockTmp);

287,024✔

3217

  taosArrayDestroyP(pResList, (FDelete)blockDataDestroy);

287,024✔

3218

  return code;

287,024✔

3221

static int32_t vnodeProcessStreamTsdbCalcDataReq(SVnode* pVnode, SRpcMsg* pMsg, SSTriggerPullRequestUnion* req, SStreamTriggerReaderInfo* sStreamReaderInfo) {

7,496,782✔

3222

  int32_t code = 0;

7,496,782✔

3223

  int32_t lino = 0;

7,496,782✔

3224

  void*   buf = NULL;

7,496,782✔

3225

  size_t  size = 0;

7,497,027✔

3226

  SSDataBlock*   pBlockRes = NULL;

7,497,298✔

3227

  STableKeyInfo* pList = NULL;

7,497,298✔

3230

  void* pTask = sStreamReaderInfo->pTask;

7,497,298✔

3231

  ST_TASK_DLOG("vgId:%d %s start, skey:%"PRId64",ekey:%"PRId64",gid:%"PRId64",ver:%"PRId64, TD_VID(pVnode), __func__,

7,497,027✔

3234

  STREAM_CHECK_NULL_GOTO(sStreamReaderInfo->triggerCols, TSDB_CODE_STREAM_NOT_TABLE_SCAN_PLAN);

7,497,027✔

3236

  SStreamReaderTaskInner* pTaskInner = NULL;

7,497,053✔

3237

  int64_t                 key = getSessionKey(req->base.sessionId, STRIGGER_PULL_TSDB_CALC_DATA);

7,497,053✔

3239

  if (req->base.type == STRIGGER_PULL_TSDB_CALC_DATA) {

7,497,298✔

3240

    int32_t        pNum = 0;

7,497,298✔

3241

    STREAM_CHECK_RET_GOTO(qStreamGetTableList(sStreamReaderInfo, req->tsdbCalcDataReq.gid, &pList, &pNum));

7,497,298✔

3242

    BUILD_OPTION(options, getSuid(sStreamReaderInfo, pList), req->tsdbCalcDataReq.ver, TSDB_ORDER_ASC, req->tsdbCalcDataReq.skey, req->tsdbCalcDataReq.ekey,

7,497,053✔

3244

    STREAM_CHECK_RET_GOTO(createStreamTask(pVnode, &options, &pTaskInner, sStreamReaderInfo->triggerResBlock, pList, pNum, &sStreamReaderInfo->storageApi));

7,497,298✔

3246

    STREAM_CHECK_RET_GOTO(taosHashPut(sStreamReaderInfo->streamTaskMap, &key, LONG_BYTES, &pTaskInner, sizeof(pTaskInner)));

7,489,968✔

3247

    STREAM_CHECK_RET_GOTO(createOneDataBlock(sStreamReaderInfo->triggerResBlock, false, &pTaskInner->pResBlockDst));

7,494,623✔

3255

  blockDataCleanup(pTaskInner->pResBlockDst);

7,494,332✔

3256

  bool hasNext = true;

7,494,358✔

3257

  while (1) {

951,881✔

3258

    STREAM_CHECK_RET_GOTO(getTableDataInfo(pTaskInner, &hasNext));

8,444,769✔

3259

    if (!hasNext) {

8,440,413✔

3260

      break;

7,488,962✔

3262

    pTaskInner->pResBlock->info.id.groupId = qStreamGetGroupIdFromSet(sStreamReaderInfo, pTaskInner->pResBlock->info.id.uid);

951,451✔

3264

    SSDataBlock* pBlock = NULL;

952,106✔

3265

    STREAM_CHECK_RET_GOTO(getTableData(pTaskInner, &pBlock));

952,106✔

3266

    STREAM_CHECK_RET_GOTO(qStreamFilter(pBlock, sStreamReaderInfo->pFilterInfo, NULL));

951,385✔

3267

    STREAM_CHECK_RET_GOTO(blockDataMerge(pTaskInner->pResBlockDst, pBlock));

952,106✔

3268

    if (pTaskInner->pResBlockDst->info.rows >= STREAM_RETURN_ROWS_NUM) {

951,610✔

3273

  STREAM_CHECK_RET_GOTO(createOneDataBlock(sStreamReaderInfo->calcResBlock, false, &pBlockRes));

7,488,962✔

3274

  STREAM_CHECK_RET_GOTO(blockDataEnsureCapacity(pBlockRes, pTaskInner->pResBlockDst->info.capacity));

7,496,073✔

3275

  blockDataTransform(pBlockRes, pTaskInner->pResBlockDst);

7,492,643✔

3276

  STREAM_CHECK_RET_GOTO(buildRsp(pBlockRes, &buf, &size));

7,495,338✔

3277

  printDataBlock(pBlockRes, __func__, "tsdb_calc_data", ((SStreamTask*)pTask)->streamId);

7,493,133✔

3278

  ST_TASK_DLOG("vgId:%d %s get result rows:%" PRId64, TD_VID(pVnode), __func__, pBlockRes->info.rows);

7,489,213✔

3279

  printDataBlock(pBlockRes, __func__, "tsdb_data", ((SStreamTask*)pTask)->streamId);

7,496,563✔

3281

  if (!hasNext) {

7,497,298✔

3282

    STREAM_CHECK_RET_GOTO(taosHashRemove(sStreamReaderInfo->streamTaskMap, &key, LONG_BYTES));

7,497,298✔

3285

end:

7,496,289✔

3286

  STREAM_PRINT_LOG_END_WITHID(code, lino);

7,495,283✔

3287

  SRpcMsg rsp = {

7,496,978✔

3289

  tmsgSendRsp(&rsp);

7,497,027✔

3290

  blockDataDestroy(pBlockRes);

7,497,053✔

3291

  taosMemoryFree(pList);

7,496,563✔

3292

  return code;

7,496,312✔

3295

static int32_t vnodeProcessStreamTsdbVirtalDataReq(SVnode* pVnode, SRpcMsg* pMsg, SSTriggerPullRequestUnion* req, SStreamTriggerReaderInfo* sStreamReaderInfo) {

656,020✔

3296

  int32_t code = 0;

656,020✔

3297

  int32_t lino = 0;

656,020✔

3298

  void*   buf = NULL;

656,020✔

3299

  size_t  size = 0;

656,020✔

3300

  int32_t* slotIdList = NULL;

656,020✔

3301

  SArray* sortedCid = NULL;

656,020✔

3302

  SArray* schemas = NULL;

656,020✔

3303

  SSDataBlock*   pBlockRes = NULL;

656,020✔

3305

  void* pTask = sStreamReaderInfo->pTask;

656,020✔

3306

  ST_TASK_DLOG("vgId:%d %s start, skey:%"PRId64",ekey:%"PRId64",uid:%"PRId64",ver:%"PRId64, TD_VID(pVnode), __func__,

656,020✔

3309

  SStreamReaderTaskInner* pTaskInner = NULL;

656,020✔

3310

  int64_t key = req->tsdbDataReq.uid;

656,020✔

3312

  if (req->base.type == STRIGGER_PULL_TSDB_DATA) {

656,020✔

3314

    slotIdList = taosMemoryMalloc(taosArrayGetSize(req->tsdbDataReq.cids) * sizeof(int32_t));

656,020✔

3315

    STREAM_CHECK_NULL_GOTO(slotIdList, terrno);

656,020✔

3316

    sortedCid = taosArrayDup(req->tsdbDataReq.cids, NULL);

656,020✔

3317

    STREAM_CHECK_NULL_GOTO(sortedCid, terrno);

656,020✔

3318

    taosArraySort(sortedCid, sortCid);

656,020✔

3319

    for (int32_t i = 0; i < taosArrayGetSize(req->tsdbDataReq.cids); i++) {

2,575,011✔

3320

      int16_t* cid = taosArrayGet(req->tsdbDataReq.cids, i);

1,918,991✔

3321

      STREAM_CHECK_NULL_GOTO(cid, terrno);

1,918,991✔

3322

      for (int32_t j = 0; j < taosArrayGetSize(sortedCid); j++) {

3,796,648✔

3323

        int16_t* cidSorted = taosArrayGet(sortedCid, j);

3,796,648✔

3324

        STREAM_CHECK_NULL_GOTO(cidSorted, terrno);

3,796,648✔

3325

        if (*cid == *cidSorted) {

3,796,648✔

3326

          slotIdList[j] = i;

1,918,991✔

3327

          break;

1,918,991✔

3332

    STREAM_CHECK_RET_GOTO(buildScheamFromMeta(pVnode, req->tsdbDataReq.uid, &schemas, &sStreamReaderInfo->storageApi));

656,020✔

3333

    STREAM_CHECK_RET_GOTO(shrinkScheams(req->tsdbDataReq.cids, schemas));

652,720✔

3334

    STREAM_CHECK_RET_GOTO(createDataBlockForStream(schemas, &pBlockRes));

652,720✔

3336

    taosArraySort(schemas, sortSSchema);

652,720✔

3337

    BUILD_OPTION(options, req->tsdbDataReq.suid, req->tsdbDataReq.ver, req->tsdbDataReq.order, req->tsdbDataReq.skey,

652,720✔

3339

    STableKeyInfo       keyInfo = {.uid = req->tsdbDataReq.uid, .groupId = 0};

652,720✔

3340

    STREAM_CHECK_RET_GOTO(createStreamTask(pVnode, &options, &pTaskInner, pBlockRes, &keyInfo, 1, &sStreamReaderInfo->storageApi));

652,720✔

3341

    STREAM_CHECK_RET_GOTO(taosHashPut(sStreamReaderInfo->streamTaskMap, &key, LONG_BYTES, &pTaskInner, sizeof(pTaskInner)));

652,720✔

3342

    pTaskInner->pResBlockDst = pBlockRes;

652,720✔

3343

    pBlockRes = NULL;

652,720✔

3351

  blockDataCleanup(pTaskInner->pResBlockDst);

652,720✔

3352

  bool hasNext = true;

652,720✔

3353

  while (1) {

652,720✔

3354

    STREAM_CHECK_RET_GOTO(getTableDataInfo(pTaskInner, &hasNext));

1,305,440✔

3355

    if (!hasNext) {

1,305,440✔

3356

      break;

652,720✔

3359

    SSDataBlock* pBlock = NULL;

652,720✔

3360

    STREAM_CHECK_RET_GOTO(getTableData(pTaskInner, &pBlock));

652,720✔

3361

    STREAM_CHECK_RET_GOTO(blockDataMerge(pTaskInner->pResBlockDst, pBlock));

652,720✔

3362

    if (pTaskInner->pResBlockDst->info.rows >= STREAM_RETURN_ROWS_NUM) {

652,720✔

3366

  STREAM_CHECK_RET_GOTO(buildRsp(pTaskInner->pResBlockDst, &buf, &size));

652,720✔

3367

  ST_TASK_DLOG("vgId:%d %s get result rows:%" PRId64, TD_VID(pVnode), __func__, pTaskInner->pResBlockDst->info.rows);

652,720✔

3368

  printDataBlock(pTaskInner->pResBlockDst, __func__, "tsdb_data", ((SStreamTask*)pTask)->streamId);

652,720✔

3369

  if (!hasNext) {

652,720✔

3370

    STREAM_CHECK_RET_GOTO(taosHashRemove(sStreamReaderInfo->streamTaskMap, &key, LONG_BYTES));

652,720✔

3373

end:

656,020✔

3374

  STREAM_PRINT_LOG_END_WITHID(code, lino);

656,020✔

3375

  SRpcMsg rsp = {

656,020✔

3377

  tmsgSendRsp(&rsp);

656,020✔

3378

  taosMemFree(slotIdList);

656,020✔

3379

  taosArrayDestroy(sortedCid);

656,020✔

3380

  taosArrayDestroy(schemas);

656,020✔

3381

  blockDataDestroy(pBlockRes);

656,020✔

3382

  return code;

656,020✔

3385

static int32_t vnodeProcessStreamWalMetaNewReq(SVnode* pVnode, SRpcMsg* pMsg, SSTriggerPullRequestUnion* req, SStreamTriggerReaderInfo* sStreamReaderInfo) {

23,860,269✔

3386

  int32_t      code = 0;

23,860,269✔

3387

  int32_t      lino = 0;

23,860,269✔

3388

  void*        buf = NULL;

23,860,269✔

3389

  size_t       size = 0;

23,860,269✔

3390

  int64_t      lastVer = 0;

23,860,269✔

3391

  SSTriggerWalNewRsp resultRsp = {0};

23,860,269✔

3393

  void* pTask = sStreamReaderInfo->pTask;

23,860,269✔

3394

  ST_TASK_DLOG("vgId:%d %s start, request paras lastVer:%" PRId64, TD_VID(pVnode), __func__, req->walMetaNewReq.lastVer);

23,860,269✔

3396

  if (sStreamReaderInfo->metaBlock == NULL) {

23,860,118✔

3397

    STREAM_CHECK_RET_GOTO(createBlockForWalMetaNew((SSDataBlock**)&sStreamReaderInfo->metaBlock));

222,521✔

3398

    STREAM_CHECK_RET_GOTO(blockDataEnsureCapacity(sStreamReaderInfo->metaBlock, STREAM_RETURN_ROWS_NUM));

222,521✔

3400

  blockDataEmpty(sStreamReaderInfo->metaBlock);

23,859,444✔

3401

  resultRsp.metaBlock = sStreamReaderInfo->metaBlock;

23,858,744✔

3402

  resultRsp.ver = req->walMetaNewReq.lastVer;

23,857,094✔

3403

  STREAM_CHECK_RET_GOTO(processWalVerMetaNew(pVnode, &resultRsp, sStreamReaderInfo, req->walMetaNewReq.ctime));

23,858,739✔

3405

  ST_TASK_DLOG("vgId:%d %s get result last ver:%"PRId64" rows:%d", TD_VID(pVnode), __func__, resultRsp.ver, resultRsp.totalRows);

23,859,234✔

3406

  STREAM_CHECK_CONDITION_GOTO(resultRsp.totalRows == 0, TDB_CODE_SUCCESS);

23,858,071✔

3407

  size = tSerializeSStreamWalDataResponse(NULL, 0, &resultRsp);

415,891✔

3408

  buf = rpcMallocCont(size);

415,891✔

3409

  size = tSerializeSStreamWalDataResponse(buf, size, &resultRsp);

415,891✔

3410

  printDataBlock(sStreamReaderInfo->metaBlock, __func__, "meta", ((SStreamTask*)pTask)->streamId);

415,891✔

3411

  printDataBlock(resultRsp.deleteBlock, __func__, "delete", ((SStreamTask*)pTask)->streamId);

415,740✔

3412

  printDataBlock(resultRsp.tableBlock, __func__, "table", ((SStreamTask*)pTask)->streamId);

415,891✔

3414

end:

23,858,071✔

3415

  if (code == 0 && resultRsp.totalRows == 0) {

23,858,896✔

3416

    code = TSDB_CODE_STREAM_NO_DATA;

23,443,830✔

3417

    size = sizeof(int64_t) * 2;

23,443,830✔

3418

    buf = rpcMallocCont(size);

23,443,830✔

3419

    *(int64_t*)buf = resultRsp.ver;

23,439,490✔

3420

    *(((int64_t*)buf) + 1) = resultRsp.verTime;

23,440,315✔

3422

  SRpcMsg rsp = {

23,856,969✔

3424

  tmsgSendRsp(&rsp);

23,858,071✔

3425

  if (code == TSDB_CODE_STREAM_NO_DATA){

23,857,661✔

3426

    code = 0;

23,441,770✔

3428

  STREAM_PRINT_LOG_END_WITHID(code, lino);

23,857,661✔

3429

  blockDataDestroy(resultRsp.deleteBlock);

23,861,066✔

3430

  blockDataDestroy(resultRsp.tableBlock);

23,859,718✔

3432

  return code;

23,859,715✔

3434

static int32_t vnodeProcessStreamWalMetaDataNewReq(SVnode* pVnode, SRpcMsg* pMsg, SSTriggerPullRequestUnion* req, SStreamTriggerReaderInfo* sStreamReaderInfo) {

6,385,783✔

3435

  int32_t      code = 0;

6,385,783✔

3436

  int32_t      lino = 0;

6,385,783✔

3437

  void*        buf = NULL;

6,385,783✔

3438

  size_t       size = 0;

6,385,783✔

3439

  SSTriggerWalNewRsp resultRsp = {0};

6,385,783✔

3441

  void* pTask = sStreamReaderInfo->pTask;

6,385,238✔

3442

  ST_TASK_DLOG("vgId:%d %s start, request paras lastVer:%" PRId64, TD_VID(pVnode), __func__, req->walMetaDataNewReq.lastVer);

6,384,914✔

3444

  if (sStreamReaderInfo->metaBlock == NULL) {

6,384,914✔

3445

    STREAM_CHECK_RET_GOTO(createBlockForWalMetaNew((SSDataBlock**)&sStreamReaderInfo->metaBlock));

139,593✔

3446

    STREAM_CHECK_RET_GOTO(blockDataEnsureCapacity(sStreamReaderInfo->metaBlock, STREAM_RETURN_ROWS_NUM));

139,593✔

3449

  resultRsp.metaBlock = sStreamReaderInfo->metaBlock;

6,386,057✔

3450

  STREAM_CHECK_RET_GOTO(createOneDataBlock(sStreamReaderInfo->triggerBlock, false, (SSDataBlock**)&resultRsp.dataBlock));

6,386,602✔

3451

  resultRsp.ver = req->walMetaDataNewReq.lastVer;

6,385,067✔

3452

  resultRsp.checkAlter = true;

6,384,566✔

3453

  resultRsp.indexHash = tSimpleHashInit(8, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT));

6,384,566✔

3454

  STREAM_CHECK_NULL_GOTO(resultRsp.indexHash, terrno);

6,381,509✔

3456

  STREAM_CHECK_RET_GOTO(processWalVerMetaDataNew(pVnode, sStreamReaderInfo, &resultRsp));

6,381,509✔

3458

  STREAM_CHECK_CONDITION_GOTO(resultRsp.totalRows == 0, TDB_CODE_SUCCESS);

6,382,388✔

3459

  size = tSerializeSStreamWalDataResponse(NULL, 0, &resultRsp);

164,366✔

3460

  buf = rpcMallocCont(size);

165,884✔

3461

  size = tSerializeSStreamWalDataResponse(buf, size, &resultRsp);

165,884✔

3462

  printDataBlock(sStreamReaderInfo->metaBlock, __func__, "meta", ((SStreamTask*)pTask)->streamId);

165,610✔

3463

  printDataBlock(resultRsp.dataBlock, __func__, "data", ((SStreamTask*)pTask)->streamId);

165,610✔

3464

  printDataBlock(resultRsp.deleteBlock, __func__, "delete", ((SStreamTask*)pTask)->streamId);

165,610✔

3465

  printDataBlock(resultRsp.tableBlock, __func__, "table", ((SStreamTask*)pTask)->streamId);

165,884✔

3466

  printIndexHash(resultRsp.indexHash, pTask);

165,884✔

3468

end:

6,383,877✔

3469

  if (resultRsp.totalRows == 0) {

6,383,061✔

3470

    code = TSDB_CODE_STREAM_NO_DATA;

6,219,506✔

3471

    size = sizeof(int64_t) * 2;

6,219,506✔

3472

    buf = rpcMallocCont(size);

6,219,506✔

3473

    *(int64_t*)buf = resultRsp.ver;

6,214,287✔

3474

    *(((int64_t*)buf) + 1) = resultRsp.verTime;

6,214,016✔

3476

  SRpcMsg rsp = {

6,378,390✔

3478

  tmsgSendRsp(&rsp);

6,381,512✔

3479

  if (code == TSDB_CODE_STREAM_NO_DATA){

6,385,172✔

3480

    code = 0;

6,220,470✔

3482

  blockDataDestroy(resultRsp.dataBlock);

6,385,172✔

3483

  blockDataDestroy(resultRsp.deleteBlock);

6,380,877✔

3484

  blockDataDestroy(resultRsp.tableBlock);

6,383,122✔

3485

  tSimpleHashCleanup(resultRsp.indexHash);

6,382,851✔

3487

  STREAM_PRINT_LOG_END_WITHID(code, lino);

6,380,244✔

3489

  return code;

6,383,846✔

3492

static int32_t vnodeProcessStreamWalDataNewReq(SVnode* pVnode, SRpcMsg* pMsg, SSTriggerPullRequestUnion* req, SStreamTriggerReaderInfo* sStreamReaderInfo) {

16,463,422✔

3493

  int32_t      code = 0;

16,463,422✔

3494

  int32_t      lino = 0;

16,463,422✔

3495

  void*        buf = NULL;

16,463,422✔

3496

  size_t       size = 0;

16,463,422✔

3497

  SSTriggerWalNewRsp resultRsp = {0};

16,463,422✔

3499

  void* pTask = sStreamReaderInfo->pTask;

16,463,422✔

3500

  ST_TASK_DLOG("vgId:%d %s start, request paras size:%zu", TD_VID(pVnode), __func__, taosArrayGetSize(req->walDataNewReq.versions));

16,463,422✔

3502

  STREAM_CHECK_RET_GOTO(createOneDataBlock(sStreamReaderInfo->triggerBlock, false, (SSDataBlock**)&resultRsp.dataBlock));

16,463,422✔

3503

  resultRsp.indexHash = tSimpleHashInit(8, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT));

16,463,699✔

3504

  STREAM_CHECK_NULL_GOTO(resultRsp.indexHash, terrno);

16,462,049✔

3505

  resultRsp.uidHash = tSimpleHashInit(8, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT));

16,462,049✔

3506

  STREAM_CHECK_NULL_GOTO(resultRsp.uidHash, terrno);

16,460,947✔

3508

  STREAM_CHECK_RET_GOTO(processWalVerDataNew(pVnode, sStreamReaderInfo, req->walDataNewReq.versions, req->walDataNewReq.ranges, &resultRsp));

16,460,947✔

3509

  ST_TASK_DLOG("vgId:%d %s get result last ver:%"PRId64" rows:%d", TD_VID(pVnode), __func__, resultRsp.ver, resultRsp.totalRows);

16,463,145✔

3511

  STREAM_CHECK_CONDITION_GOTO(resultRsp.totalRows == 0, TDB_CODE_SUCCESS);

16,463,422✔

3513

  size = tSerializeSStreamWalDataResponse(NULL, 0, &resultRsp);

232,716✔

3514

  buf = rpcMallocCont(size);

232,716✔

3515

  size = tSerializeSStreamWalDataResponse(buf, size, &resultRsp);

232,716✔

3516

  printDataBlock(resultRsp.dataBlock, __func__, "data", ((SStreamTask*)pTask)->streamId);

232,716✔

3517

  printIndexHash(resultRsp.indexHash, pTask);

232,716✔

3519

end:

16,463,422✔

3520

  if (resultRsp.totalRows == 0) {

16,463,422✔

3521

    buf = rpcMallocCont(sizeof(int64_t));

16,230,983✔

3522

    *(int64_t *)buf = resultRsp.ver;

16,230,709✔

3523

    size = sizeof(int64_t);

16,230,709✔

3524

    code = TSDB_CODE_STREAM_NO_DATA;

16,230,709✔

3526

  SRpcMsg rsp = {

16,463,148✔

3528

  tmsgSendRsp(&rsp);

16,463,148✔

3529

  if (code == TSDB_CODE_STREAM_NO_DATA){

16,462,874✔

3530

    code = 0;

16,230,983✔

3533

  blockDataDestroy(resultRsp.dataBlock);

16,462,874✔

3534

  blockDataDestroy(resultRsp.deleteBlock);

16,462,597✔

3535

  blockDataDestroy(resultRsp.tableBlock);

16,460,947✔

3536

  tSimpleHashCleanup(resultRsp.indexHash);

16,462,049✔

3537

  tSimpleHashCleanup(resultRsp.uidHash);

16,462,049✔

3538

  STREAM_PRINT_LOG_END_WITHID(code, lino);

16,459,856✔

3540

  return code;

16,463,156✔

3543

static int32_t vnodeProcessStreamWalCalcDataNewReq(SVnode* pVnode, SRpcMsg* pMsg, SSTriggerPullRequestUnion* req, SStreamTriggerReaderInfo* sStreamReaderInfo) {

1,185,022✔

3544

  int32_t      code = 0;

1,185,022✔

3545

  int32_t      lino = 0;

1,185,022✔

3546

  void*        buf = NULL;

1,185,022✔

3547

  size_t       size = 0;

1,185,022✔

3548

  SSTriggerWalNewRsp resultRsp = {0};

1,185,022✔

3549

  SSDataBlock* pBlock1 = NULL;

1,185,022✔

3550

  SSDataBlock* pBlock2 = NULL;

1,185,022✔

3552

  void* pTask = sStreamReaderInfo->pTask;

1,185,022✔

3553

  ST_TASK_DLOG("vgId:%d %s start, request paras size:%zu", TD_VID(pVnode), __func__, taosArrayGetSize(req->walDataNewReq.versions));

1,185,022✔

3555

  SSDataBlock* dataBlock = sStreamReaderInfo->isVtableStream ? sStreamReaderInfo->calcBlock : sStreamReaderInfo->triggerBlock;

1,185,022✔

3556

  STREAM_CHECK_RET_GOTO(createOneDataBlock(dataBlock, false, (SSDataBlock**)&resultRsp.dataBlock));

1,185,022✔

3557

  resultRsp.isCalc = sStreamReaderInfo->isVtableStream ? true : false;

1,185,022✔

3558

  resultRsp.indexHash = tSimpleHashInit(8, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT));

1,185,022✔

3559

  STREAM_CHECK_NULL_GOTO(resultRsp.indexHash, terrno);

1,185,022✔

3560

  resultRsp.uidHash = tSimpleHashInit(8, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT));

1,185,022✔

3561

  STREAM_CHECK_NULL_GOTO(resultRsp.uidHash, terrno);

1,185,022✔

3563

  STREAM_CHECK_RET_GOTO(processWalVerDataNew(pVnode, sStreamReaderInfo, req->walDataNewReq.versions, req->walDataNewReq.ranges, &resultRsp));

1,185,022✔

3564

  STREAM_CHECK_CONDITION_GOTO(resultRsp.totalRows == 0, TDB_CODE_SUCCESS);

1,184,751✔

3566

  if (!sStreamReaderInfo->isVtableStream){

287,884✔

3567

    STREAM_CHECK_RET_GOTO(createOneDataBlock(sStreamReaderInfo->calcBlock, false, &pBlock2));

180,353✔

3569

    blockDataTransform(pBlock2, resultRsp.dataBlock);

180,624✔

3570

    blockDataDestroy(resultRsp.dataBlock);

180,353✔

3571

    resultRsp.dataBlock = pBlock2;

180,624✔

3572

    pBlock2 = NULL;

180,624✔

3575

  size = tSerializeSStreamWalDataResponse(NULL, 0, &resultRsp);

288,155✔

3576

  buf = rpcMallocCont(size);

287,884✔

3577

  size = tSerializeSStreamWalDataResponse(buf, size, &resultRsp);

287,884✔

3578

  printDataBlock(resultRsp.dataBlock, __func__, "data", ((SStreamTask*)pTask)->streamId);

287,884✔

3579

  printIndexHash(resultRsp.indexHash, pTask);

287,884✔

3581

end:

1,184,751✔

3582

  if (resultRsp.totalRows == 0) {

1,185,022✔

3583

    buf = rpcMallocCont(sizeof(int64_t));

896,867✔

3584

    *(int64_t *)buf = resultRsp.ver;

896,867✔

3585

    size = sizeof(int64_t);

896,867✔

3586

    code = TSDB_CODE_STREAM_NO_DATA;

896,867✔

3588

  SRpcMsg rsp = {

1,185,022✔

3590

  tmsgSendRsp(&rsp);

1,184,751✔

3591

  if (code == TSDB_CODE_STREAM_NO_DATA){

1,185,022✔

3592

    code = 0;

896,867✔

3595

  blockDataDestroy(pBlock1);

1,185,022✔

3596

  blockDataDestroy(pBlock2);

1,185,022✔

3597

  blockDataDestroy(resultRsp.dataBlock);

1,185,022✔

3598

  blockDataDestroy(resultRsp.deleteBlock);

1,185,022✔

3599

  blockDataDestroy(resultRsp.tableBlock);

1,185,022✔

3600

  tSimpleHashCleanup(resultRsp.indexHash);

1,185,022✔

3601

  tSimpleHashCleanup(resultRsp.uidHash);

1,185,022✔

3602

  STREAM_PRINT_LOG_END_WITHID(code, lino);

1,185,022✔

3604

  return code;

1,185,022✔

3607

static int32_t vnodeProcessStreamGroupColValueReq(SVnode* pVnode, SRpcMsg* pMsg, SSTriggerPullRequestUnion* req, SStreamTriggerReaderInfo* sStreamReaderInfo) {

359,721✔

3608

  int32_t code = 0;

359,721✔

3609

  int32_t lino = 0;

359,721✔

3610

  void*   buf = NULL;

359,721✔

3611

  size_t  size = 0;

359,721✔

3612

  SArray** gInfo = NULL;

359,721✔

3614

  void* pTask = sStreamReaderInfo->pTask;

359,721✔

3615

  ST_TASK_DLOG("vgId:%d %s start, request gid:%" PRId64, TD_VID(pVnode), __func__, req->groupColValueReq.gid);

359,721✔

3617

  gInfo = taosHashAcquire(sStreamReaderInfo->groupIdMap, &req->groupColValueReq.gid, POINTER_BYTES);

359,721✔

3618

  STREAM_CHECK_NULL_GOTO(gInfo, TSDB_CODE_STREAM_NO_CONTEXT);

359,721✔

3619

  SStreamGroupInfo pGroupInfo = {0};

359,721✔

3620

  pGroupInfo.gInfo = *gInfo;

359,721✔

3622

  size = tSerializeSStreamGroupInfo(NULL, 0, &pGroupInfo, TD_VID(pVnode));

359,721✔

3624

  buf = rpcMallocCont(size);

359,721✔

3625

  STREAM_CHECK_NULL_GOTO(buf, terrno);

359,721✔

3626

  size = tSerializeSStreamGroupInfo(buf, size, &pGroupInfo, TD_VID(pVnode));

359,721✔

3628

end:

359,721✔

3629

  taosHashRelease(sStreamReaderInfo->groupIdMap, gInfo);

359,721✔

3630

  if (code != 0) {

359,721✔

3635

  STREAM_PRINT_LOG_END_WITHID(code, lino);

359,721✔

3636

  SRpcMsg rsp = {

359,721✔

3638

  tmsgSendRsp(&rsp);

359,721✔

3640

  return code;

359,721✔

3643

static int32_t setVtableInfo(SVnode* pVnode, SArray* infos, SArray* cids, int64_t uid, uint64_t gid, int64_t ver, SMetaReader* metaReader, SStreamTriggerReaderInfo* sStreamReaderInfo) {

203,773✔

3644

  int32_t              code = 0;

203,773✔

3645

  int32_t              lino = 0;

203,773✔

3646

  void* pTask = sStreamReaderInfo->pTask;

203,773✔

3648

  VTableInfo* vTable = taosArrayReserve(infos, 1);

203,773✔

3649

  STREAM_CHECK_NULL_GOTO(vTable, terrno);

203,773✔

3650

  vTable->uid = uid;

203,773✔

3651

  vTable->gId = gid;

203,773✔

3653

  ST_TASK_DLOG("vgId:%d %s put vtable uid:%"PRId64, TD_VID(pVnode), __func__, uid);

203,773✔

3655

  code = sStreamReaderInfo->storageApi.metaReaderFn.getTableEntryByVersionUid(metaReader, ver, uid);

203,773✔

3656

  if (code != 0) {

203,773✔

3660

  if (atomic_load_8(&sStreamReaderInfo->isVtableOnlyTs) == 1) {

203,773✔

3661

    vTable->cols.nCols = metaReader->me.colRef.nCols;

11,705✔

3662

    vTable->cols.version = metaReader->me.colRef.version;

11,705✔

3663

    vTable->cols.pColRef = taosMemoryCalloc(metaReader->me.colRef.nCols, sizeof(SColRef));

11,705✔

3664

    STREAM_CHECK_NULL_GOTO(vTable->cols.pColRef, terrno);

11,705✔

3665

    for (size_t j = 0; j < metaReader->me.colRef.nCols; j++) {

70,230✔

3666

      memcpy(vTable->cols.pColRef + j, &metaReader->me.colRef.pColRef[j], sizeof(SColRef));

58,525✔

3669

    vTable->cols.nCols = taosArrayGetSize(cids);

192,068✔

3670

    vTable->cols.version = metaReader->me.colRef.version;

192,068✔

3671

    vTable->cols.pColRef = taosMemoryCalloc(taosArrayGetSize(cids), sizeof(SColRef));

192,068✔

3672

    STREAM_CHECK_NULL_GOTO(vTable->cols.pColRef, terrno);

192,068✔

3673

    for (size_t i = 0; i < taosArrayGetSize(cids); i++) {

729,155✔

3674

      for (size_t j = 0; j < metaReader->me.colRef.nCols; j++) {

2,262,449✔

3675

        if (metaReader->me.colRef.pColRef[j].hasRef &&

2,069,580✔

3676

            metaReader->me.colRef.pColRef[j].id == *(col_id_t*)taosArrayGet(cids, i)) {

1,527,440✔

3677

          memcpy(vTable->cols.pColRef + i, &metaReader->me.colRef.pColRef[j], sizeof(SColRef));

344,218✔

3678

          break;

344,218✔

3683

  tDecoderClear(&metaReader->coder);

203,773✔

3685

end:

203,773✔

3686

  return code;

203,773✔

3689

static int32_t getAllVinfo(SVnode* pVnode, SStreamMsgVTableInfo* vTableInfo, SArray* cids, int64_t ver, SMetaReader* metaReader, SStreamTriggerReaderInfo* sStreamReaderInfo){

85,834✔

3690

  int32_t              code = 0;

85,834✔

3691

  int32_t              lino = 0;

85,834✔

3692

  void* pTask = sStreamReaderInfo->pTask;

85,834✔

3693

  SArray*              pTableListArray = NULL;

85,834✔

3696

  pTableListArray = qStreamGetTableArrayList(sStreamReaderInfo);

85,834✔

3697

  STREAM_CHECK_NULL_GOTO(pTableListArray, terrno);

85,834✔

3699

  vTableInfo->infos = taosArrayInit(taosArrayGetSize(pTableListArray), sizeof(VTableInfo));

85,834✔

3700

  STREAM_CHECK_NULL_GOTO(vTableInfo->infos, terrno);

85,834✔

3702

  for (size_t i = 0; i < taosArrayGetSize(pTableListArray); i++) {

289,607✔

3703

    SStreamTableKeyInfo* pKeyInfo = taosArrayGetP(pTableListArray, i);

203,773✔

3704

    if (pKeyInfo == NULL || pKeyInfo->markedDeleted) {

203,773✔

3707

    code = setVtableInfo(pVnode, vTableInfo->infos, cids, pKeyInfo->uid, pKeyInfo->groupId, ver, metaReader, sStreamReaderInfo);

203,773✔

3708

    if (code != 0) {

203,773✔

3715

end:

85,834✔

3716

  taosArrayDestroyP(pTableListArray, taosMemFree);

85,834✔

3717

  return code;

85,834✔

3751

static int32_t vnodeProcessStreamVTableInfoReq(SVnode* pVnode, SRpcMsg* pMsg, SSTriggerPullRequestUnion* req, SStreamTriggerReaderInfo* sStreamReaderInfo) {

85,834✔

3752

  int32_t              code = 0;

85,834✔

3753

  int32_t              lino = 0;

85,834✔

3754

  void*                buf = NULL;

85,834✔

3755

  size_t               size = 0;

85,834✔

3756

  SStreamMsgVTableInfo vTableInfo = {0};

85,834✔

3757

  SMetaReader          metaReader = {0};

85,834✔

3759

  void* pTask = sStreamReaderInfo->pTask;

85,834✔

3760

  ST_TASK_DLOG("vgId:%d %s start, version:%"PRId64, TD_VID(pVnode), __func__, req->virTableInfoReq.ver);

85,834✔

3762

  SArray* cids = req->virTableInfoReq.cids;

85,834✔

3763

  STREAM_CHECK_NULL_GOTO(cids, terrno);

85,834✔

3765

  if (taosArrayGetSize(cids) == 1 && *(col_id_t*)taosArrayGet(cids, 0) == PRIMARYKEY_TIMESTAMP_COL_ID){

85,834✔

3766

    (void)atomic_val_compare_exchange_8(&sStreamReaderInfo->isVtableOnlyTs, 0, 1);

11,705✔

3768

  sStreamReaderInfo->storageApi.metaReaderFn.initReader(&metaReader, pVnode, META_READER_LOCK, &sStreamReaderInfo->storageApi.metaFn);

85,834✔

3770

  if (req->virTableInfoReq.fetchAllTable || req->virTableInfoReq.uids == NULL || taosArrayGetSize(req->virTableInfoReq.uids) == 0) {

85,834✔

3771

    STREAM_CHECK_RET_GOTO(getAllVinfo(pVnode, &vTableInfo, cids, req->virTableInfoReq.ver, &metaReader, sStreamReaderInfo));

85,834✔

3775

  ST_TASK_DLOG("vgId:%d %s end, size:%"PRIzu, TD_VID(pVnode), __func__, taosArrayGetSize(vTableInfo.infos));

85,834✔

3776

  STREAM_CHECK_RET_GOTO(buildVTableInfoRsp(&vTableInfo, &buf, &size));

85,834✔

3778

end:

85,834✔

3779

  tDestroySStreamMsgVTableInfo(&vTableInfo);

85,834✔

3780

  sStreamReaderInfo->storageApi.metaReaderFn.clearReader(&metaReader);

85,834✔

3781

  STREAM_PRINT_LOG_END_WITHID(code, lino);

85,834✔

3782

  SRpcMsg rsp = {

85,834✔

3784

  tmsgSendRsp(&rsp);

85,834✔

3785

  return code;

85,834✔

3788

static int32_t vnodeProcessStreamOTableInfoReq(SVnode* pVnode, SRpcMsg* pMsg, SSTriggerPullRequestUnion* req, SStreamTriggerReaderInfo* sStreamReaderInfo) {

146,369✔

3789

  int32_t                   code = 0;

146,369✔

3790

  int32_t                   lino = 0;

146,369✔

3791

  void*                     buf = NULL;

146,369✔

3792

  size_t                    size = 0;

146,369✔

3793

  SSTriggerOrigTableInfoRsp oTableInfo = {0};

146,369✔

3794

  SMetaReader               metaReader = {0};

146,369✔

3795

  void*                     pTask = sStreamReaderInfo->pTask;

146,369✔

3797

  ST_TASK_DLOG("vgId:%d %s start, ver:%" PRId64, TD_VID(pVnode), __func__, req->origTableInfoReq.ver);

146,369✔

3799

  SArray* cols = req->origTableInfoReq.cols;

146,369✔

3800

  STREAM_CHECK_NULL_GOTO(cols, terrno);

146,369✔

3802

  oTableInfo.cols = taosArrayInit(taosArrayGetSize(cols), sizeof(OTableInfoRsp));

146,369✔

3804

  STREAM_CHECK_NULL_GOTO(oTableInfo.cols, terrno);

146,369✔

3806

  sStreamReaderInfo->storageApi.metaReaderFn.initReader(&metaReader, pVnode, META_READER_LOCK, &sStreamReaderInfo->storageApi.metaFn);

146,369✔

3807

  for (size_t i = 0; i < taosArrayGetSize(cols); i++) {

537,407✔

3808

    OTableInfo*    oInfo = taosArrayGet(cols, i);

391,038✔

3809

    OTableInfoRsp* vTableInfo = taosArrayReserve(oTableInfo.cols, 1);

391,038✔

3810

    STREAM_CHECK_NULL_GOTO(oInfo, terrno);

391,038✔

3811

    STREAM_CHECK_NULL_GOTO(vTableInfo, terrno);

391,038✔

3812

    code = sStreamReaderInfo->storageApi.metaReaderFn.getTableEntryByVersionName(&metaReader, req->origTableInfoReq.ver, oInfo->refTableName);

391,038✔

3813

    if (code != 0) {

391,038✔

3814

      code = 0;

245✔

3815

      ST_TASK_ELOG("vgId:%d %s get table entry by name:%s failed, msg:%s", TD_VID(pVnode), __func__, oInfo->refTableName, tstrerror(code));

245✔

3816

      continue;

245✔

3818

    vTableInfo->uid = metaReader.me.uid;

390,793✔

3819

    ST_TASK_DLOG("vgId:%d %s get original uid:%"PRId64, TD_VID(pVnode), __func__, vTableInfo->uid);

390,793✔

3821

    SSchemaWrapper* sSchemaWrapper = NULL;

390,793✔

3822

    if (metaReader.me.type == TD_CHILD_TABLE) {

390,793✔

3823

      int64_t suid = metaReader.me.ctbEntry.suid;

388,646✔

3824

      vTableInfo->suid = suid;

388,646✔

3825

      tDecoderClear(&metaReader.coder);

388,646✔

3826

      STREAM_CHECK_RET_GOTO(sStreamReaderInfo->storageApi.metaReaderFn.getTableEntryByVersionUid(&metaReader, req->origTableInfoReq.ver, suid));

388,646✔

3827

      sSchemaWrapper = &metaReader.me.stbEntry.schemaRow;

388,646✔

3828

    } else if (metaReader.me.type == TD_NORMAL_TABLE) {

2,147✔

3829

      vTableInfo->suid = 0;

2,147✔

3830

      sSchemaWrapper = &metaReader.me.ntbEntry.schemaRow;

2,147✔

3835

    for (size_t j = 0; j < sSchemaWrapper->nCols; j++) {

1,343,332✔

3836

      SSchema* s = sSchemaWrapper->pSchema + j;

1,343,332✔

3837

      if (strcmp(s->name, oInfo->refColName) == 0) {

1,343,332✔

3838

        vTableInfo->cid = s->colId;

390,793✔

3839

        break;

390,793✔

3842

    if (vTableInfo->cid == 0) {

390,793✔

3846

    tDecoderClear(&metaReader.coder);

390,793✔

3849

  STREAM_CHECK_RET_GOTO(buildOTableInfoRsp(&oTableInfo, &buf, &size));

146,369✔

3851

end:

146,369✔

3852

  tDestroySTriggerOrigTableInfoRsp(&oTableInfo);

146,369✔

3853

  sStreamReaderInfo->storageApi.metaReaderFn.clearReader(&metaReader);

146,369✔

3854

  STREAM_PRINT_LOG_END_WITHID(code, lino);

146,369✔

3855

  SRpcMsg rsp = {

146,369✔

3857

  tmsgSendRsp(&rsp);

146,369✔

3858

  return code;

146,369✔

3861

static int32_t vnodeProcessStreamVTableTagInfoReq(SVnode* pVnode, SRpcMsg* pMsg, SSTriggerPullRequestUnion* req, SStreamTriggerReaderInfo* sStreamReaderInfo) {

1,173,909✔

3862

  int32_t                   code = 0;

1,173,909✔

3863

  int32_t                   lino = 0;

1,173,909✔

3864

  void*                     buf = NULL;

1,173,909✔

3865

  size_t                    size = 0;

1,173,909✔

3866

  SSDataBlock* pBlock = NULL;

1,173,909✔

3868

  SMetaReader               metaReader = {0};

1,173,909✔

3869

  SMetaReader               metaReaderStable = {0};

1,173,909✔

3870

  int64_t streamId = req->base.streamId;

1,173,909✔

3871

  stsDebug("vgId:%d %s start, ver:%"PRId64, TD_VID(pVnode), __func__, req->virTablePseudoColReq.ver);

1,173,909✔

3873

  SArray* cols = req->virTablePseudoColReq.cids;

1,173,909✔

3874

  STREAM_CHECK_NULL_GOTO(cols, terrno);

1,173,909✔

3876

  sStreamReaderInfo->storageApi.metaReaderFn.initReader(&metaReader, pVnode, META_READER_LOCK, &sStreamReaderInfo->storageApi.metaFn);

1,173,909✔

3877

  STREAM_CHECK_RET_GOTO(sStreamReaderInfo->storageApi.metaReaderFn.getTableEntryByVersionUid(&metaReader, req->virTablePseudoColReq.ver, req->virTablePseudoColReq.uid));

1,173,909✔

3879

  STREAM_CHECK_CONDITION_GOTO(metaReader.me.type != TD_VIRTUAL_CHILD_TABLE && metaReader.me.type != TD_VIRTUAL_NORMAL_TABLE, TSDB_CODE_INVALID_PARA);

1,173,909✔

3881

  STREAM_CHECK_RET_GOTO(createDataBlock(&pBlock));

1,173,909✔

3882

  if (metaReader.me.type == TD_VIRTUAL_NORMAL_TABLE) {

1,173,909✔

3883

    STREAM_CHECK_CONDITION_GOTO (taosArrayGetSize(cols) < 1 || *(col_id_t*)taosArrayGet(cols, 0) != -1, TSDB_CODE_INVALID_PARA);

2,438✔

3884

    SColumnInfoData idata = createColumnInfoData(TSDB_DATA_TYPE_BINARY, TSDB_TABLE_NAME_LEN, -1);

2,438✔

3885

    STREAM_CHECK_RET_GOTO(blockDataAppendColInfo(pBlock, &idata));

2,438✔

3886

    STREAM_CHECK_RET_GOTO(blockDataEnsureCapacity(pBlock, 1));

2,438✔

3887

    pBlock->info.rows = 1;

2,438✔

3888

    SColumnInfoData* pDst = taosArrayGet(pBlock->pDataBlock, 0);

2,438✔

3889

    STREAM_CHECK_NULL_GOTO(pDst, terrno);

2,438✔

3890

    STREAM_CHECK_RET_GOTO(varColSetVarData(pDst, 0, metaReader.me.name, strlen(metaReader.me.name), false));

2,438✔

3891

  } else if (metaReader.me.type == TD_VIRTUAL_CHILD_TABLE){

1,171,471✔

3892

    int64_t suid = metaReader.me.ctbEntry.suid;

1,171,471✔

3893

    sStreamReaderInfo->storageApi.metaReaderFn.readerReleaseLock(&metaReader);

1,171,471✔

3894

    sStreamReaderInfo->storageApi.metaReaderFn.initReader(&metaReaderStable, pVnode, META_READER_LOCK, &sStreamReaderInfo->storageApi.metaFn);

1,171,471✔

3896

    STREAM_CHECK_RET_GOTO(sStreamReaderInfo->storageApi.metaReaderFn.getTableEntryByVersionUid(&metaReaderStable, req->virTablePseudoColReq.ver, suid));

1,171,471✔

3897

    SSchemaWrapper*  sSchemaWrapper = &metaReaderStable.me.stbEntry.schemaTag;

1,170,922✔

3898

    for (size_t i = 0; i < taosArrayGetSize(cols); i++){

3,081,210✔

3899

      col_id_t* id = taosArrayGet(cols, i);

1,910,016✔

3900

      STREAM_CHECK_NULL_GOTO(id, terrno);

1,910,016✔

3901

      if (*id == -1) {

1,910,016✔

3902

        SColumnInfoData idata = createColumnInfoData(TSDB_DATA_TYPE_BINARY, TSDB_TABLE_NAME_LEN, -1);

1,164,594✔

3903

        STREAM_CHECK_RET_GOTO(blockDataAppendColInfo(pBlock, &idata));

1,164,871✔

3904

        continue;

1,164,871✔

3906

      size_t j = 0;

745,422✔

3907

      for (; j < sSchemaWrapper->nCols; j++) {

1,443,282✔

3908

        SSchema* s = sSchemaWrapper->pSchema + j;

1,443,282✔

3909

        if (s->colId == *id) {

1,443,282✔

3910

          SColumnInfoData idata = createColumnInfoData(s->type, s->bytes, s->colId);

745,694✔

3911

          STREAM_CHECK_RET_GOTO(blockDataAppendColInfo(pBlock, &idata));

745,417✔

3912

          break;

745,417✔

3915

      if (j == sSchemaWrapper->nCols) {

745,417✔

3920

    STREAM_CHECK_RET_GOTO(blockDataEnsureCapacity(pBlock, 1));

1,171,471✔

3921

    pBlock->info.rows = 1;

1,171,471✔

3923

    for (size_t i = 0; i < taosArrayGetSize(pBlock->pDataBlock); i++){

3,082,036✔

3924

      SColumnInfoData* pDst = taosArrayGet(pBlock->pDataBlock, i);

1,910,565✔

3925

      STREAM_CHECK_NULL_GOTO(pDst, terrno);

1,910,565✔

3927

      if (pDst->info.colId == -1) {

1,910,565✔

3928

        STREAM_CHECK_RET_GOTO(varColSetVarData(pDst, 0, metaReader.me.name, strlen(metaReader.me.name), false));

1,164,871✔

3929

        continue;

1,164,871✔

3931

      if (pDst->info.type == TSDB_DATA_TYPE_NULL) {

745,417✔

3936

      STagVal val = {0};

745,694✔

3937

      val.cid = pDst->info.colId;

745,694✔

3938

      const char* p = sStreamReaderInfo->storageApi.metaFn.extractTagVal(metaReader.me.ctbEntry.pTags, pDst->info.type, &val);

745,694✔

3940

      char* data = NULL;

745,417✔

3941

      if (pDst->info.type != TSDB_DATA_TYPE_JSON && p != NULL) {

745,417✔

3942

        data = tTagValToData((const STagVal*)p, false);

745,417✔

3947

      STREAM_CHECK_RET_GOTO(colDataSetVal(pDst, 0, data,

745,694✔

3950

      if ((pDst->info.type != TSDB_DATA_TYPE_JSON) && (p != NULL) && IS_VAR_DATA_TYPE(((const STagVal*)p)->type) &&

745,694✔

3952

        taosMemoryFree(data);

692,940✔

3961

  stsDebug("vgId:%d %s get result rows:%" PRId64, TD_VID(pVnode), __func__, pBlock->info.rows);

1,173,632✔

3962

  printDataBlock(pBlock, __func__, "", streamId);

1,173,632✔

3963

  STREAM_CHECK_RET_GOTO(buildRsp(pBlock, &buf, &size));

1,173,909✔

3965

end:

1,174,186✔

3966

  if(size == 0){

1,173,909✔

3969

  sStreamReaderInfo->storageApi.metaReaderFn.clearReader(&metaReaderStable);

1,173,909✔

3970

  sStreamReaderInfo->storageApi.metaReaderFn.clearReader(&metaReader);

1,173,909✔

3971

  STREAM_PRINT_LOG_END(code, lino);

1,173,632✔

3972

  SRpcMsg rsp = {

1,173,632✔

3974

  tmsgSendRsp(&rsp);

1,173,909✔

3975

  blockDataDestroy(pBlock);

1,173,909✔

3976

  return code;

1,173,909✔

3979

static int32_t vnodeProcessStreamFetchMsg(SVnode* pVnode, SRpcMsg* pMsg, SQueueInfo *pInfo) {

8,888,929✔

3980

  int32_t            code = 0;

8,888,929✔

3981

  int32_t            lino = 0;

8,888,929✔

3982

  void*              buf = NULL;

8,888,929✔

3983

  size_t             size = 0;

8,888,929✔

3984

  void*              taskAddr = NULL;

8,888,652✔

3985

  SArray*            pResList = NULL;

8,888,652✔

3986

  bool               hasNext = false;

8,888,652✔

3987

  SStreamTriggerReaderCalcInfo* sStreamReaderCalcInfo = NULL;

8,888,652✔

3989

  SResFetchReq req = {0};

8,888,652✔

3990

  STREAM_CHECK_CONDITION_GOTO(tDeserializeSResFetchReq(pMsg->pCont, pMsg->contLen, &req) < 0,

8,888,652✔

3992

  SArray* calcInfoList = (SArray*)qStreamGetReaderInfo(req.queryId, req.taskId, &taskAddr);

8,888,775✔

3993

  STREAM_CHECK_NULL_GOTO(calcInfoList, terrno);

8,888,929✔

3995

  STREAM_CHECK_CONDITION_GOTO(req.execId < 0, TSDB_CODE_INVALID_PARA);

8,888,929✔

3996

  sStreamReaderCalcInfo = taosArrayGetP(calcInfoList, req.execId);

8,888,929✔

3997

  STREAM_CHECK_NULL_GOTO(sStreamReaderCalcInfo, terrno);

8,888,929✔

3998

  sStreamReaderCalcInfo->rtInfo.execId = req.execId;

8,888,929✔

4000

  void* pTask = sStreamReaderCalcInfo->pTask;

8,888,929✔

4001

  ST_TASK_DLOG("vgId:%d %s start, execId:%d, reset:%d, pTaskInfo:%p, scan type:%d", TD_VID(pVnode), __func__, req.execId, req.reset,

8,888,929✔

4004

  if (req.reset) {

8,888,929✔

4005

    int64_t uid = 0;

5,557,608✔

4006

    if (req.dynTbname && !req.pStRtFuncInfo->isMultiGroupCalc) {

5,557,608✔

4007

      SArray* vals = req.pStRtFuncInfo->pStreamPartColVals;

187,311✔

4008

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

187,311✔

4009

        SStreamGroupValue* pValue = taosArrayGet(vals, i);

187,311✔

4010

        if (pValue != NULL && pValue->isTbname) {

187,311✔

4011

          uid = pValue->uid;

187,311✔

4012

          break;

187,311✔

4017

    SReadHandle handle = {0};

5,557,608✔

4018

    handle.vnode = pVnode;

5,557,608✔

4019

    handle.pMsgCb = &pVnode->msgCb;

5,557,608✔

4020

    handle.pWorkerCb = pInfo->workerCb;

5,557,608✔

4021

    handle.uid = uid;

5,557,608✔

4022

    handle.cacheSttStatis = true;

5,557,608✔

4024

    initStorageAPI(&handle.api);

5,557,608✔

4025

    if (QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN == nodeType(sStreamReaderCalcInfo->calcAst->pNode) ||

5,557,095✔

4026

      QUERY_NODE_PHYSICAL_PLAN_TABLE_MERGE_SCAN == nodeType(sStreamReaderCalcInfo->calcAst->pNode)){

4,410,126✔

4027

      STimeRangeNode* node = (STimeRangeNode*)((STableScanPhysiNode*)(sStreamReaderCalcInfo->calcAst->pNode))->pTimeRange;

4,064,922✔

4028

      if (node != NULL) {

4,064,630✔

4029

        STREAM_CHECK_RET_GOTO(processCalaTimeRange(sStreamReaderCalcInfo, &req, node, &handle, false));

636,647✔

4031

        ST_TASK_DLOG("vgId:%d %s no scan time range node", TD_VID(pVnode), __func__);

3,427,983✔

4034

      node = (STimeRangeNode*)((STableScanPhysiNode*)(sStreamReaderCalcInfo->calcAst->pNode))->pExtTimeRange;

4,064,851✔

4035

      if (node != NULL) {

4,064,630✔

4036

        STREAM_CHECK_RET_GOTO(processCalaTimeRange(sStreamReaderCalcInfo, &req, node, &handle, true));

85,932✔

4038

        ST_TASK_DLOG("vgId:%d %s no interp time range node", TD_VID(pVnode), __func__);

3,978,698✔

4042

    TSWAP(sStreamReaderCalcInfo->rtInfo.funcInfo, *req.pStRtFuncInfo);

5,557,537✔

4043

    sStreamReaderCalcInfo->rtInfo.funcInfo.hasPlaceHolder = sStreamReaderCalcInfo->hasPlaceHolder;

5,557,316✔

4044

    handle.streamRtInfo = &sStreamReaderCalcInfo->rtInfo;

5,557,608✔

4046

    if (sStreamReaderCalcInfo->pTaskInfo == NULL || !qNeedReset(sStreamReaderCalcInfo->pTaskInfo)) {

5,557,294✔

4047

      qDestroyTask(sStreamReaderCalcInfo->pTaskInfo);

717,048✔

4048

      STREAM_CHECK_RET_GOTO(qCreateStreamExecTaskInfo(&sStreamReaderCalcInfo->pTaskInfo,

717,340✔

4051

      STREAM_CHECK_RET_GOTO(qSetTaskId(sStreamReaderCalcInfo->pTaskInfo, req.taskId, req.queryId));

717,340✔

4053

      STREAM_CHECK_RET_GOTO(qResetTableScan(sStreamReaderCalcInfo->pTaskInfo, &handle));

4,840,268✔

4056

    STREAM_CHECK_RET_GOTO(qSetTaskId(sStreamReaderCalcInfo->pTaskInfo, req.taskId, req.queryId));

5,557,297✔

4059

  if (req.pOpParam != NULL) {

8,888,397✔

4060

    qUpdateOperatorParam(sStreamReaderCalcInfo->pTaskInfo, (void*)req.pOpParam);

451,673✔

4063

  pResList = taosArrayInit(4, POINTER_BYTES);

8,888,397✔

4064

  STREAM_CHECK_NULL_GOTO(pResList, terrno);

8,888,708✔

4065

  uint64_t ts = 0;

8,888,708✔

4066

  STREAM_CHECK_RET_GOTO(qExecTaskOpt(sStreamReaderCalcInfo->pTaskInfo, pResList, &ts, &hasNext, NULL, req.pOpParam != NULL));

8,888,708✔

4068

  for(size_t i = 0; i < taosArrayGetSize(pResList); i++){

20,918,783✔

4069

    SSDataBlock* pBlock = taosArrayGetP(pResList, i);

12,032,674✔

4070

    if (pBlock == NULL) continue;

12,032,674✔

4071

    printDataBlock(pBlock, __func__, "fetch", ((SStreamTask*)pTask)->streamId);

12,032,674✔

4080

end:

8,888,858✔

4081

  code = streamBuildFetchRsp(pResList, hasNext, &buf, &size, pVnode->config.tsdbCfg.precision);

8,888,637✔

4083

  if (sStreamReaderCalcInfo && sStreamReaderCalcInfo->rtInfo.funcInfo.isMultiGroupCalc) {

8,888,345✔

4088

  taosArrayDestroy(pResList);

8,888,345✔

4089

  streamReleaseTask(taskAddr);

8,888,929✔

4091

  if (code == TSDB_CODE_PAR_TABLE_NOT_EXIST || code == TSDB_CODE_TDB_TABLE_NOT_EXIST){

8,888,929✔

4094

  STREAM_PRINT_LOG_END(code, lino);

8,888,929✔

4095

  SRpcMsg rsp = {.msgType = TDMT_STREAM_FETCH_RSP, .info = pMsg->info, .pCont = buf, .contLen = size, .code = code};

8,888,929✔

4096

  tmsgSendRsp(&rsp);

8,888,929✔

4097

  tDestroySResFetchReq(&req);

8,888,325✔

4098

  if (TDB_CODE_SUCCESS != code) {

8,888,069✔

4102

  return code;

8,888,069✔

4105

static int32_t initTableList(SStreamTriggerReaderInfo* sStreamReaderInfo, SVnode* pVnode) {

59,604,230✔

4106

  int32_t code = 0;

59,604,230✔

4107

  if (sStreamReaderInfo->tableList.pTableList != NULL) {

59,604,230✔

4108

    return code;

59,243,530✔

4110

  taosWLockLatch(&sStreamReaderInfo->lock);

362,375✔

4111

  sStreamReaderInfo->pVnode = pVnode;

362,375✔

4112

  initStorageAPI(&sStreamReaderInfo->storageApi);

362,375✔

4113

  if (sStreamReaderInfo->tableList.pTableList == NULL) {

362,375✔

4114

    code = initStreamTableListInfo(&sStreamReaderInfo->tableList);

362,375✔

4115

    if (code == 0) {

362,375✔

4116

      code = generateTablistForStreamReader(pVnode, sStreamReaderInfo);

362,375✔

4117

      if (code != 0) {

362,375✔

4120

        sStreamReaderInfo->tableList.version = pVnode->state.applied;

362,375✔

4121

        stDebug("vgId:%d %s init table list for stream reader, table num:%zu, version:%" PRId64,

362,375✔

4126

  taosWUnLockLatch(&sStreamReaderInfo->lock);

362,375✔

4127

  return code;

362,375✔

4130

int32_t vnodeProcessStreamReaderMsg(SVnode* pVnode, SRpcMsg* pMsg, SQueueInfo *pInfo) {

68,669,342✔

4131

  int32_t                   code = 0;

68,669,342✔

4132

  int32_t                   lino = 0;

68,669,342✔

4133

  SSTriggerPullRequestUnion req = {0};

68,669,342✔

4134

  void*                     taskAddr = NULL;

68,669,342✔

4135

  bool                      sendRsp = false;

68,665,901✔

4137

  vDebug("vgId:%d, msg:%p in stream reader queue is processing", pVnode->config.vgId, pMsg);

68,665,901✔

4138

  if (!syncIsReadyForRead(pVnode->sync)) {

68,667,118✔

4139

    vnodeRedirectRpcMsg(pVnode, pMsg, terrno);

149,509✔

4140

    return 0;

149,509✔

4143

  if (pMsg->msgType == TDMT_STREAM_FETCH) {

68,517,568✔

4144

    return vnodeProcessStreamFetchMsg(pVnode, pMsg, pInfo);

8,888,929✔

4145

  } else if (pMsg->msgType == TDMT_STREAM_TRIGGER_PULL) {

59,629,021✔

4146

    void*   pReq = POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead));

59,630,356✔

4147

    int32_t len = pMsg->contLen - sizeof(SMsgHead);

59,631,452✔

4148

    STREAM_CHECK_RET_GOTO(tDeserializeSTriggerPullRequest(pReq, len, &req));

59,630,904✔

4149

    stDebug("vgId:%d %s start, type:%d, streamId:%" PRIx64 ", readerTaskId:%" PRIx64 ", sessionId:%" PRIx64 ", applied:%" PRIx64,

59,625,570✔

4151

    SStreamTriggerReaderInfo* sStreamReaderInfo = qStreamGetReaderInfo(req.base.streamId, req.base.readerTaskId, &taskAddr);

59,627,407✔

4152

    STREAM_CHECK_NULL_GOTO(sStreamReaderInfo, terrno);

59,629,569✔

4153

    STREAM_CHECK_RET_GOTO(initTableList(sStreamReaderInfo, pVnode));

59,605,084✔

4154

    sendRsp = true;

59,605,356✔

4155

    switch (req.base.type) {

59,605,356✔

4156

      case STRIGGER_PULL_SET_TABLE:

146,369✔

4157

        STREAM_CHECK_RET_GOTO(vnodeProcessStreamSetTableReq(pVnode, pMsg, &req, sStreamReaderInfo));

146,369✔

4158

        break;

146,369✔

4159

      case STRIGGER_PULL_LAST_TS:

351,844✔

4160

        STREAM_CHECK_RET_GOTO(vnodeProcessStreamLastTsReq(pVnode, pMsg, &req, sStreamReaderInfo));

351,844✔

4161

        break;

351,844✔

4162

      case STRIGGER_PULL_FIRST_TS:

333,457✔

4163

        STREAM_CHECK_RET_GOTO(vnodeProcessStreamFirstTsReq(pVnode, pMsg, &req, sStreamReaderInfo));

333,457✔

4164

        break;

333,186✔

4165

      case STRIGGER_PULL_TSDB_META:

591,841✔

4167

        STREAM_CHECK_RET_GOTO(vnodeProcessStreamTsdbMetaReq(pVnode, pMsg, &req, sStreamReaderInfo));

591,841✔

4168

        break;

591,841✔

4169

      case STRIGGER_PULL_TSDB_TS_DATA:

81,138✔

4170

        if (sStreamReaderInfo->isVtableStream) {

81,138✔

4171

          STREAM_CHECK_RET_GOTO(vnodeProcessStreamTsdbTsDataReqVTable(pVnode, pMsg, &req, sStreamReaderInfo));

3,300✔

4173

          STREAM_CHECK_RET_GOTO(vnodeProcessStreamTsdbTsDataReqNonVTable(pVnode, pMsg, &req, sStreamReaderInfo));

77,838✔

4175

        break;

81,138✔

4176

      case STRIGGER_PULL_TSDB_TRIGGER_DATA:

287,024✔

4178

        STREAM_CHECK_RET_GOTO(vnodeProcessStreamTsdbTriggerDataReq(pVnode, pMsg, &req, sStreamReaderInfo));

287,024✔

4179

        break;

159,256✔

4180

      case STRIGGER_PULL_TSDB_CALC_DATA:

7,497,027✔

4182

        STREAM_CHECK_RET_GOTO(vnodeProcessStreamTsdbCalcDataReq(pVnode, pMsg, &req, sStreamReaderInfo));

7,497,027✔

4183

        break;

7,496,292✔

4184

      case STRIGGER_PULL_TSDB_DATA:

656,020✔

4186

        STREAM_CHECK_RET_GOTO(vnodeProcessStreamTsdbVirtalDataReq(pVnode, pMsg, &req, sStreamReaderInfo));

656,020✔

4187

        break;

652,720✔

4188

      case STRIGGER_PULL_GROUP_COL_VALUE:

359,721✔

4189

        STREAM_CHECK_RET_GOTO(vnodeProcessStreamGroupColValueReq(pVnode, pMsg, &req, sStreamReaderInfo));

359,721✔

4190

        break;

359,721✔

4191

      case STRIGGER_PULL_VTABLE_INFO:

85,834✔

4192

        STREAM_CHECK_RET_GOTO(vnodeProcessStreamVTableInfoReq(pVnode, pMsg, &req, sStreamReaderInfo));

85,834✔

4193

        break;

85,834✔

4194

      case STRIGGER_PULL_VTABLE_PSEUDO_COL:

1,173,909✔

4195

        STREAM_CHECK_RET_GOTO(vnodeProcessStreamVTableTagInfoReq(pVnode, pMsg, &req, sStreamReaderInfo));

1,173,909✔

4196

        break;

1,173,909✔

4197

      case STRIGGER_PULL_OTABLE_INFO:

146,369✔

4198

        STREAM_CHECK_RET_GOTO(vnodeProcessStreamOTableInfoReq(pVnode, pMsg, &req, sStreamReaderInfo));

146,369✔

4199

        break;

146,369✔

4200

      case STRIGGER_PULL_WAL_META_NEW:

23,860,269✔

4201

        STREAM_CHECK_RET_GOTO(vnodeProcessStreamWalMetaNewReq(pVnode, pMsg, &req, sStreamReaderInfo));

23,860,269✔

4202

        break;

23,859,718✔

4203

      case STRIGGER_PULL_WAL_DATA_NEW:

16,463,422✔

4204

        STREAM_CHECK_RET_GOTO(vnodeProcessStreamWalDataNewReq(pVnode, pMsg, &req, sStreamReaderInfo));

16,463,422✔

4205

        break;

16,463,699✔

4206

      case STRIGGER_PULL_WAL_META_DATA_NEW:

6,386,605✔

4207

        STREAM_CHECK_RET_GOTO(vnodeProcessStreamWalMetaDataNewReq(pVnode, pMsg, &req, sStreamReaderInfo));

6,386,605✔

4208

        break;

6,384,451✔

4209

      case STRIGGER_PULL_WAL_CALC_DATA_NEW:

1,185,022✔

4210

        STREAM_CHECK_RET_GOTO(vnodeProcessStreamWalCalcDataNewReq(pVnode, pMsg, &req, sStreamReaderInfo));

1,185,022✔

4211

        break;

1,185,022✔

4212

      default:

3✔

4213

        vError("unknown inner msg type:%d in stream reader queue", req.base.type);

3✔

4218

    vError("unknown msg type:%d in stream reader queue", pMsg->msgType);

3✔

4221

end:

59,617,950✔

4223

  streamReleaseTask(taskAddr);

59,627,464✔

4225

  tDestroySTriggerPullRequest(&req);

59,628,561✔

4226

  STREAM_PRINT_LOG_END(code, lino);

59,624,243✔

4227

  if (!sendRsp) {

59,627,077✔

4228

    SRpcMsg rsp = {

48,970✔

4230

      .pCont = pMsg->info.rsp,

24,485✔

4231

      .contLen = pMsg->info.rspLen,

24,485✔

4234

    tmsgSendRsp(&rsp);

24,485✔

4236

  return code;

59,626,252✔

taosdata / TDengine / #5066

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

taosdata / TDengine / #5066

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