taosdata/TDengine | Build 4986 | source/dnode/vnode/src/tq/tqScan.c | Coveralls

17

static int32_t tqAddRawDataToRsp(const void* rawData, SMqDataRsp* pRsp, int8_t precision) {

×

18

  int32_t    code = TDB_CODE_SUCCESS;

×

19

  int32_t    lino = 0;

×

20

  void*      buf = NULL;

×

22

  int32_t dataStrLen = sizeof(SRetrieveTableRspForTmq) + *(uint32_t *)rawData + INT_BYTES;

×

23

  buf = taosMemoryCalloc(1, dataStrLen);

×

24

  TSDB_CHECK_NULL(buf, code, lino, END, terrno);

×

26

  SRetrieveTableRspForTmq* pRetrieve = (SRetrieveTableRspForTmq*)buf;

×

27

  pRetrieve->version = RETRIEVE_TABLE_RSP_TMQ_RAW_VERSION;

×

28

  pRetrieve->precision = precision;

×

29

  pRetrieve->compressed = 0;

×

31

  memcpy(pRetrieve->data, rawData, *(uint32_t *)rawData + INT_BYTES);

×

32

  TSDB_CHECK_NULL(taosArrayPush(pRsp->blockDataLen, &dataStrLen), code, lino, END, terrno);

×

33

  TSDB_CHECK_NULL(taosArrayPush(pRsp->blockData, &buf), code, lino, END, terrno);

×

34

  pRsp->blockDataElementFree = true;

×

36

  tqTrace("tqAddRawDataToRsp add block data to block array, blockDataLen:%d, blockData:%p", dataStrLen, buf);

×

37

  END:

×

38

  if (code != TSDB_CODE_SUCCESS) {

×

39

    taosMemoryFree(buf);

×

40

    tqError("%s failed at %d, failed to add block data to response:%s", __FUNCTION__, lino, tstrerror(code));

×

42

  return code;

×

45

static int32_t tqAddBlockDataToRsp(const SSDataBlock* pBlock, SMqDataRsp* pRsp, const SSchemaWrapper* pSW, int8_t precision) {

×

46

  int32_t code = 0;

×

47

  int32_t lino = 0;

×

48

  SSchemaWrapper* pSchema = NULL;

×

50

  size_t dataEncodeBufSize = blockGetEncodeSize(pBlock);

×

51

  int32_t dataStrLen = sizeof(SRetrieveTableRspForTmq) + dataEncodeBufSize;

×

52

  void*   buf = taosMemoryCalloc(1, dataStrLen);

×

53

  TSDB_CHECK_NULL(buf, code, lino, END, terrno);

×

55

  SRetrieveTableRspForTmq* pRetrieve = (SRetrieveTableRspForTmq*)buf;

×

56

  pRetrieve->version = RETRIEVE_TABLE_RSP_TMQ_VERSION;

×

57

  pRetrieve->precision = precision;

×

58

  pRetrieve->compressed = 0;

×

59

  pRetrieve->numOfRows = htobe64((int64_t)pBlock->info.rows);

×

61

  int32_t actualLen = blockEncode(pBlock, pRetrieve->data, dataEncodeBufSize, pSW->nCols);

×

62

  TSDB_CHECK_CONDITION(actualLen >= 0, code, lino, END, terrno);

×

64

  actualLen += sizeof(SRetrieveTableRspForTmq);

×

65

  TSDB_CHECK_NULL(taosArrayPush(pRsp->blockDataLen, &actualLen), code, lino, END, terrno);

×

66

  TSDB_CHECK_NULL(taosArrayPush(pRsp->blockData, &buf), code, lino, END, terrno);

×

67

  pSchema = tCloneSSchemaWrapper(pSW);

×

68

  TSDB_CHECK_NULL(pSchema, code, lino, END, terrno);

×

69

  TSDB_CHECK_NULL(taosArrayPush(pRsp->blockSchema, &pSchema), code, lino, END, terrno);

×

70

  pSchema = NULL;

×

71

  pRsp->blockDataElementFree = true;

×

72

  tqTrace("tqAddBlockDataToRsp add block data to block array, blockDataLen:%d, blockData:%p", dataStrLen, buf);

×

74

END:

×

75

  tDeleteSchemaWrapper(pSchema);

×

76

  if (code != TSDB_CODE_SUCCESS){

×

77

    taosMemoryFree(buf);

×

78

    tqError("%s failed at line %d with msg:%s", __func__, lino, tstrerror(code));

×

80

  return code;

×

83

static int32_t tqAddTbNameToRsp(const STQ* pTq, int64_t uid, SMqDataRsp* pRsp, int32_t n) {

×

84

  int32_t    code = TDB_CODE_SUCCESS;

×

85

  int32_t    lino = 0;

×

86

  SMetaReader mr = {0};

×

88

  TSDB_CHECK_NULL(pTq, code, lino, END, TSDB_CODE_INVALID_PARA);

×

89

  TSDB_CHECK_NULL(pRsp, code, lino, END, TSDB_CODE_INVALID_PARA);

×

91

  metaReaderDoInit(&mr, pTq->pVnode->pMeta, META_READER_LOCK);

×

93

  code = metaReaderGetTableEntryByUidCache(&mr, uid);

×

94

  TSDB_CHECK_CODE(code, lino, END);

×

96

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

×

97

    char* tbName = taosStrdup(mr.me.name);

×

98

    TSDB_CHECK_NULL(tbName, code, lino, END, terrno);

×

99

    if(taosArrayPush(pRsp->blockTbName, &tbName) == NULL){

×

100

      tqError("failed to push tbName to blockTbName:%s, uid:%"PRId64, tbName, uid);

×

101

      continue;

×

103

    tqTrace("add tbName to response success tbname:%s, uid:%"PRId64, tbName, uid);

×

106

END:

×

107

  if (code != TSDB_CODE_SUCCESS) {

×

108

    tqError("%s failed at %d, failed to add tbName to response:%s, uid:%"PRId64, __FUNCTION__, lino, tstrerror(code), uid);

×

110

  metaReaderClear(&mr);

×

111

  return code;

×

114

int32_t getDataBlock(qTaskInfo_t task, const STqHandle* pHandle, int32_t vgId, SSDataBlock** res) {

×

115

  if (task == NULL || pHandle == NULL || res == NULL) {

×

116

    return TSDB_CODE_INVALID_PARA;

×

118

  uint64_t ts = 0;

×

119

  qStreamSetOpen(task);

×

121

  tqDebug("consumer:0x%" PRIx64 " vgId:%d, tmq one task start execute", pHandle->consumerId, vgId);

×

122

  int32_t code = qExecTask(task, res, &ts);

×

123

  if (code != TSDB_CODE_SUCCESS) {

×

124

    tqError("consumer:0x%" PRIx64 " vgId:%d, task exec error since %s", pHandle->consumerId, vgId, tstrerror(code));

×

127

  tqDebug("consumer:0x%" PRIx64 " vgId:%d tmq one task end executed, pDataBlock:%p", pHandle->consumerId, vgId, *res);

×

128

  return code;

×

131

static int32_t tqProcessReplayRsp(STQ* pTq, STqHandle* pHandle, SMqDataRsp* pRsp, const SMqPollReq* pRequest, SSDataBlock* pDataBlock, qTaskInfo_t task){

×

132

  int32_t code = 0;

×

133

  int32_t lino = 0;

×

135

  if (IS_OFFSET_RESET_TYPE(pRequest->reqOffset.type) && pHandle->block != NULL) {

×

136

    blockDataDestroy(pHandle->block);

×

137

    pHandle->block = NULL;

×

139

  if (pHandle->block == NULL) {

×

140

    if (pDataBlock == NULL) {

×

141

      goto END;

×

144

    STqOffsetVal offset = {0};

×

145

    code = qStreamExtractOffset(task, &offset);

×

146

    TSDB_CHECK_CODE(code, lino, END);

×

148

    pHandle->block = NULL;

×

150

    code = createOneDataBlock(pDataBlock, true, &pHandle->block);

×

151

    TSDB_CHECK_CODE(code, lino, END);

×

153

    pHandle->blockTime = offset.ts;

×

154

    tOffsetDestroy(&offset);

×

155

    int32_t vgId = TD_VID(pTq->pVnode);

×

156

    code = getDataBlock(task, pHandle, vgId, &pDataBlock);

×

157

    TSDB_CHECK_CODE(code, lino, END);

×

160

  const STqExecHandle* pExec = &pHandle->execHandle;

×

161

  code = tqAddBlockDataToRsp(pHandle->block, pRsp, &pExec->execCol.pSW, pTq->pVnode->config.tsdbCfg.precision);

×

162

  TSDB_CHECK_CODE(code, lino, END);

×

164

  pRsp->blockNum++;

×

165

  if (pDataBlock == NULL) {

×

166

    blockDataDestroy(pHandle->block);

×

167

    pHandle->block = NULL;

×

169

    code = copyDataBlock(pHandle->block, pDataBlock);

×

170

    TSDB_CHECK_CODE(code, lino, END);

×

172

    STqOffsetVal offset = {0};

×

173

    code = qStreamExtractOffset(task, &offset);

×

174

    TSDB_CHECK_CODE(code, lino, END);

×

176

    pRsp->sleepTime = offset.ts - pHandle->blockTime;

×

177

    pHandle->blockTime = offset.ts;

×

178

    tOffsetDestroy(&offset);

×

181

END:

×

182

  if (code != TSDB_CODE_SUCCESS) {

×

183

    tqError("%s failed at %d, failed to process replay response:%s", __FUNCTION__, lino, tstrerror(code));

×

185

  return code;

×

188

int32_t tqScanData(STQ* pTq, STqHandle* pHandle, SMqDataRsp* pRsp, STqOffsetVal* pOffset, const SMqPollReq* pRequest) {

×

189

  int32_t code = 0;

×

190

  int32_t lino = 0;

×

191

  TSDB_CHECK_NULL(pRsp, code, lino, END, TSDB_CODE_INVALID_PARA);

×

192

  TSDB_CHECK_NULL(pTq, code, lino, END, TSDB_CODE_INVALID_PARA);

×

193

  TSDB_CHECK_NULL(pHandle, code, lino, END, TSDB_CODE_INVALID_PARA);

×

194

  TSDB_CHECK_NULL(pOffset, code, lino, END, TSDB_CODE_INVALID_PARA);

×

195

  TSDB_CHECK_NULL(pRequest, code, lino, END, TSDB_CODE_INVALID_PARA);

×

197

  int32_t vgId = TD_VID(pTq->pVnode);

×

198

  int32_t totalRows = 0;

×

200

  const STqExecHandle* pExec = &pHandle->execHandle;

×

201

  qTaskInfo_t          task = pExec->task;

×

203

  code = qStreamPrepareScan(task, pOffset, pHandle->execHandle.subType);

×

204

  TSDB_CHECK_CODE(code, lino, END);

×

206

  qStreamSetParams(task, pRequest->sourceExcluded, pRequest->minPollRows, pRequest->timeout, pRequest->enableReplay);

×

208

    SSDataBlock* pDataBlock = NULL;

×

209

    code = getDataBlock(task, pHandle, vgId, &pDataBlock);

×

210

    TSDB_CHECK_CODE(code, lino, END);

×

212

    if (pRequest->enableReplay) {

×

213

      code = tqProcessReplayRsp(pTq, pHandle, pRsp, pRequest, pDataBlock, task);

×

214

      TSDB_CHECK_CODE(code, lino, END);

×

215

      break;

×

217

    if (pDataBlock == NULL) {

×

218

      break;

×

220

    code = tqAddBlockDataToRsp(pDataBlock, pRsp, &pExec->execCol.pSW, pTq->pVnode->config.tsdbCfg.precision);

×

221

    TSDB_CHECK_CODE(code, lino, END);

×

223

    pRsp->blockNum++;

×

224

    totalRows += pDataBlock->info.rows;

×

227

  tqDebug("consumer:0x%" PRIx64 " vgId:%d tmq task executed finished, total blocks:%d, totalRows:%d", pHandle->consumerId, vgId, pRsp->blockNum, totalRows);

×

228

  code = qStreamExtractOffset(task, &pRsp->rspOffset);

×

230

END:

×

231

  if (code != 0) {

×

232

    tqError("%s failed at %d, tmq task executed error msg:%s", __FUNCTION__, lino, tstrerror(code));

×

234

  return code;

×

237

int32_t tqScanTaosx(STQ* pTq, const STqHandle* pHandle, SMqDataRsp* pRsp, SMqBatchMetaRsp* pBatchMetaRsp, STqOffsetVal* pOffset, const SMqPollReq* pRequest) {

×

238

  int32_t code = 0;

×

239

  int32_t lino = 0;

×

240

  char* tbName = NULL;

×

241

  const STqExecHandle* pExec = &pHandle->execHandle;

×

242

  qTaskInfo_t          task = pExec->task;

×

243

  code = qStreamPrepareScan(task, pOffset, pHandle->execHandle.subType);

×

244

  TSDB_CHECK_CODE(code, lino, END);

×

246

  qStreamSetParams(task, pRequest->sourceExcluded, pRequest->minPollRows, pRequest->timeout, false);

×

248

  int32_t rowCnt = 0;

×

249

  int64_t st = taosGetTimestampMs();

×

250

  while (1) {

×

251

    SSDataBlock* pDataBlock = NULL;

×

252

    uint64_t     ts = 0;

×

253

    tqDebug("tmqsnap task start to execute");

×

254

    code = qExecTask(task, &pDataBlock, &ts);

×

255

    TSDB_CHECK_CODE(code, lino, END);

×

256

    tqDebug("tmqsnap task execute end, get %p", pDataBlock);

×

258

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

×

259

      if (pRsp->withTbName) {

×

260

        tbName = taosStrdup(qExtractTbnameFromTask(task));

×

261

        TSDB_CHECK_NULL(tbName, code, lino, END, terrno);

×

262

        TSDB_CHECK_NULL(taosArrayPush(pRsp->blockTbName, &tbName), code, lino, END, terrno);

×

263

        tqDebug("vgId:%d, add tbname:%s to rsp msg", pTq->pVnode->config.vgId, tbName);

×

264

        tbName = NULL;

×

267

      code = tqAddBlockDataToRsp(pDataBlock, pRsp, qExtractSchemaFromTask(task), pTq->pVnode->config.tsdbCfg.precision);

×

268

      TSDB_CHECK_CODE(code, lino, END);

×

270

      pRsp->blockNum++;

×

271

      rowCnt += pDataBlock->info.rows;

×

272

      if (rowCnt <= pRequest->minPollRows && (taosGetTimestampMs() - st <= pRequest->timeout)) {

×

273

        continue;

×

278

    SMqBatchMetaRsp* tmp = qStreamExtractMetaMsg(task);

×

279

    if (taosArrayGetSize(tmp->batchMetaReq) > 0) {

×

280

      code = qStreamExtractOffset(task, &tmp->rspOffset);

×

281

      TSDB_CHECK_CODE(code, lino, END);

×

282

      *pBatchMetaRsp = *tmp;

×

283

      tqDebug("tmqsnap task get meta");

×

284

      break;

×

287

    if (pDataBlock == NULL) {

×

288

      code = qStreamExtractOffset(task, pOffset);

×

289

      TSDB_CHECK_CODE(code, lino, END);

×

291

      if (pOffset->type == TMQ_OFFSET__SNAPSHOT_DATA) {

×

292

        continue;

×

295

      tqDebug("tmqsnap vgId: %d, tsdb consume over, switch to wal, ver %" PRId64, TD_VID(pTq->pVnode), pHandle->snapshotVer + 1);

×

296

      code = qStreamExtractOffset(task, &pRsp->rspOffset);

×

297

      break;

×

300

    if (pRsp->blockNum > 0) {

×

301

      tqDebug("tmqsnap task exec exited, get data");

×

302

      code = qStreamExtractOffset(task, &pRsp->rspOffset);

×

303

      break;

×

306

  tqDebug("%s:%d success", __FUNCTION__, lino);

×

307

END:

×

308

  if (code != 0){

×

309

    tqError("%s failed at %d, vgId:%d, task exec error since %s", __FUNCTION__ , lino, pTq->pVnode->config.vgId, tstrerror(code));

×

311

  taosMemoryFree(tbName);

×

312

  return code;

×

315

static void tqProcessSubData(STQ* pTq, STqHandle* pHandle, SMqDataRsp* pRsp, int32_t* totalRows,

×

317

  int32_t code = 0;

×

318

  int32_t lino = 0;

×

319

  SArray* pBlocks = NULL;

×

320

  SArray* pSchemas = NULL;

×

322

  STqExecHandle* pExec = &pHandle->execHandle;

×

323

  STqReader* pReader = pExec->pTqReader;

×

325

  pBlocks = taosArrayInit(0, sizeof(SSDataBlock));

×

326

  TSDB_CHECK_NULL(pBlocks, code, lino, END, terrno);

×

327

  pSchemas = taosArrayInit(0, sizeof(void*));

×

328

  TSDB_CHECK_NULL(pSchemas, code, lino, END, terrno);

×

330

  SSubmitTbData* pSubmitTbData = NULL;

×

331

  code = tqRetrieveTaosxBlock(pReader, pRsp, pBlocks, pSchemas, &pSubmitTbData, rawList, pHandle->fetchMeta);

×

332

  TSDB_CHECK_CODE(code, lino, END);

×

333

  bool tmp = (pSubmitTbData->flags & pRequest->sourceExcluded) != 0;

×

334

  TSDB_CHECK_CONDITION(!tmp, code, lino, END, TSDB_CODE_SUCCESS);

×

336

  if (pHandle->fetchMeta == ONLY_META){

×

337

    goto END;

×

340

  int32_t blockNum = taosArrayGetSize(pBlocks) == 0 ? 1 : taosArrayGetSize(pBlocks);

×

341

  if (pRsp->withTbName) {

×

342

    int64_t uid = pExec->pTqReader->lastBlkUid;

×

343

    code = tqAddTbNameToRsp(pTq, uid, pRsp, blockNum);

×

344

    TSDB_CHECK_CODE(code, lino, END);

×

347

  TSDB_CHECK_CONDITION(!tmp, code, lino, END, TSDB_CODE_SUCCESS);

×

348

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

×

349

    if (taosArrayGetSize(pBlocks) == 0){

×

350

      void* rawData = taosArrayGetP(rawList, pReader->nextBlk - 1);

×

351

      if (rawData == NULL) {

×

352

        continue;

×

354

      if (tqAddRawDataToRsp(rawData, pRsp, pTq->pVnode->config.tsdbCfg.precision) != 0){

×

355

        tqError("vgId:%d, failed to add block to rsp msg", pTq->pVnode->config.vgId);

×

356

        continue;

×

358

      *totalRows += *(uint32_t *)rawData + INT_BYTES; // bytes actually

×

360

      SSDataBlock* pBlock = taosArrayGet(pBlocks, i);

×

361

      if (pBlock == NULL) {

×

362

        continue;

×

365

      SSchemaWrapper* pSW = (SSchemaWrapper*)taosArrayGetP(pSchemas, i);

×

366

      if (tqAddBlockDataToRsp(pBlock, pRsp, pSW, pTq->pVnode->config.tsdbCfg.precision) != 0){

×

367

        tqError("vgId:%d, failed to add block to rsp msg", pTq->pVnode->config.vgId);

×

368

        continue;

×

370

      *totalRows += pBlock->info.rows;

×

373

    pRsp->blockNum++;

×

375

  tqTrace("vgId:%d, process sub data success, response blocknum:%d, rows:%d", pTq->pVnode->config.vgId, pRsp->blockNum, *totalRows);

×

376

END:

×

377

  if (code != 0) {

×

378

    tqError("%s failed at %d, failed to process sub data:%s", __FUNCTION__, lino, tstrerror(code));

×

380

  taosArrayDestroyEx(pBlocks, (FDelete)blockDataFreeRes);

×

381

  taosArrayDestroyP(pSchemas, (FDelete)tDeleteSchemaWrapper);

×

382

×

384

static void preProcessSubmitMsg(STqHandle* pHandle, const SMqPollReq* pRequest, SArray** rawList){

×

385

  STqExecHandle* pExec = &pHandle->execHandle;

×

386

  STqReader* pReader = pExec->pTqReader;

×

387

  int32_t blockSz = taosArrayGetSize(pReader->submit.aSubmitTbData);

×

388

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

×

389

    SSubmitTbData* pSubmitTbData = taosArrayGet(pReader->submit.aSubmitTbData, i);

×

390

    if (pSubmitTbData== NULL){

×

391

      taosArrayDestroy(*rawList);

×

392

      *rawList = NULL;

×

393

      return;

×

396

    int64_t uid = pSubmitTbData->uid;

×

397

    if (pRequest->rawData) {

×

398

      if (taosHashGet(pRequest->uidHash, &uid, LONG_BYTES) != NULL) {

×

399

        tqDebug("poll rawdata split,uid:%" PRId64 " is already exists", uid);

×

400

        terrno = TSDB_CODE_TMQ_RAW_DATA_SPLIT;

×

401

        return;

×

403

        int32_t code = taosHashPut(pRequest->uidHash, &uid, LONG_BYTES, &uid, LONG_BYTES);

×

404

        if (code != 0) {

×

405

          tqError("failed to add table uid to hash, code:%d, uid:%" PRId64, code, uid);

×

410

    if (pSubmitTbData->pCreateTbReq == NULL){

×

411

      continue;

×

414

    int64_t createTime = INT64_MAX;

×

415

    int64_t *cTime = (int64_t*)taosHashGet(pHandle->tableCreateTimeHash, &uid, LONG_BYTES);

×

416

    if (cTime != NULL){

×

417

      createTime = *cTime;

×

419

      createTime = metaGetTableCreateTime(pReader->pVnode->pMeta, uid, 1);

×

420

      if (createTime != INT64_MAX){

×

421

        int32_t code = taosHashPut(pHandle->tableCreateTimeHash, &uid, LONG_BYTES, &createTime, LONG_BYTES);

×

422

        if (code != 0){

×

423

          tqError("failed to add table create time to hash,code:%d, uid:%"PRId64, code, uid);

×

427

    if (pHandle->fetchMeta == WITH_DATA || pSubmitTbData->ctimeMs > createTime){

×

428

      tDestroySVSubmitCreateTbReq(pSubmitTbData->pCreateTbReq, TSDB_MSG_FLG_DECODE);

×

429

      taosMemoryFreeClear(pSubmitTbData->pCreateTbReq);

×

431

      taosArrayDestroy(*rawList);

×

432

      *rawList = NULL;

×

437

int32_t tqTaosxScanLog(STQ* pTq, STqHandle* pHandle, SPackedData submit, SMqDataRsp* pRsp, int32_t* totalRows, const SMqPollReq* pRequest) {

×

438

  int32_t code = 0;

×

439

  int32_t lino = 0;

×

440

  SDecoder decoder = {0};

×

441

  STqExecHandle* pExec = &pHandle->execHandle;

×

442

  STqReader* pReader = pExec->pTqReader;

×

443

  SArray *rawList = NULL;

×

444

  if (pRequest->rawData){

×

445

    rawList = taosArrayInit(0, POINTER_BYTES);

×

446

    TSDB_CHECK_NULL(rawList, code, lino, END, terrno);

×

448

  code = tqReaderSetSubmitMsg(pReader, submit.msgStr, submit.msgLen, submit.ver, rawList, &decoder);

×

449

  TSDB_CHECK_CODE(code, lino, END);

×

450

  preProcessSubmitMsg(pHandle, pRequest, &rawList);

×

452

  if (pRequest->rawData != 0 && terrno == TSDB_CODE_TMQ_RAW_DATA_SPLIT){

×

453

    goto END;

×

457

  if (pRequest->rawData != 0 && *totalRows > 0 && pRsp->createTableNum == 0 && rawList == NULL){

×

458

    tqDebug("poll rawdata split,vgId:%d, this wal submit data contains metadata and previous data is data", pTq->pVnode->config.vgId);

×

459

    terrno = TSDB_CODE_TMQ_RAW_DATA_SPLIT;

×

460

    goto END;

×

464

  if (pRequest->rawData != 0 && pRsp->createTableNum > 0 && rawList != NULL){

×

465

    tqDebug("poll rawdata split,vgId:%d, this wal submit data is data and previous data is metadata", pTq->pVnode->config.vgId);

×

466

    terrno = TSDB_CODE_TMQ_RAW_DATA_SPLIT;

×

467

    goto END;

×

470

  if (pExec->subType == TOPIC_SUB_TYPE__TABLE) {

×

471

    while (tqNextBlockImpl(pReader, NULL)) {

×

472

      tqProcessSubData(pTq, pHandle, pRsp, totalRows, pRequest, rawList);

×

474

  } else if (pExec->subType == TOPIC_SUB_TYPE__DB) {

×

475

    while (tqNextDataBlockFilterOut(pReader, pExec->execDb.pFilterOutTbUid)) {

×

476

      tqProcessSubData(pTq, pHandle, pRsp, totalRows, pRequest, rawList);

×

480

END:

×

481

  tDecoderClear(&decoder);

×

482

  tqReaderClearSubmitMsg(pReader);

×

483

  taosArrayDestroy(rawList);

×

484

  if (code != 0){

×

485

    tqError("%s failed at %d, failed to scan log:%s", __FUNCTION__, lino, tstrerror(code));

×

487

  return code;

×

taosdata / TDengine / #4986

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

1	/*
2	* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
3	*
4	* This program is free software: you can use, redistribute, and/or modify
5	* it under the terms of the GNU Affero General Public License, version 3
6	* or later ("AGPL"), as published by the Free Software Foundation.
7	*
8	* This program is distributed in the hope that it will be useful, but WITHOUT
9	* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
10	* FITNESS FOR A PARTICULAR PURPOSE.
11	*
12	* You should have received a copy of the GNU Affero General Public License
13	* along with this program. If not, see <http://www.gnu.org/licenses/>.
14	*/
15
16	#include "tq.h"
17	static int32_t tqAddRawDataToRsp(const void* rawData, SMqDataRsp* pRsp, int8_t precision) {	×
18	int32_t code = TDB_CODE_SUCCESS;	×
19	int32_t lino = 0;	×
20	void* buf = NULL;	×
21
22	int32_t dataStrLen = sizeof(SRetrieveTableRspForTmq) + (uint32_t )rawData + INT_BYTES;	×
23	buf = taosMemoryCalloc(1, dataStrLen);	×
24	TSDB_CHECK_NULL(buf, code, lino, END, terrno);	×
25
26	SRetrieveTableRspForTmq* pRetrieve = (SRetrieveTableRspForTmq*)buf;	×
27	pRetrieve->version = RETRIEVE_TABLE_RSP_TMQ_RAW_VERSION;	×
28	pRetrieve->precision = precision;	×
29	pRetrieve->compressed = 0;	×
30
31	memcpy(pRetrieve->data, rawData, (uint32_t )rawData + INT_BYTES);	×
32	TSDB_CHECK_NULL(taosArrayPush(pRsp->blockDataLen, &dataStrLen), code, lino, END, terrno);	×
33	TSDB_CHECK_NULL(taosArrayPush(pRsp->blockData, &buf), code, lino, END, terrno);	×
34	pRsp->blockDataElementFree = true;	×
35
36	tqTrace("tqAddRawDataToRsp add block data to block array, blockDataLen:%d, blockData:%p", dataStrLen, buf);	×
37	END:	×
38	if (code != TSDB_CODE_SUCCESS) {	×
39	taosMemoryFree(buf);	×
40	tqError("%s failed at %d, failed to add block data to response:%s", __FUNCTION__, lino, tstrerror(code));	×
41	}
42	return code;	×
43	}
44
45	static int32_t tqAddBlockDataToRsp(const SSDataBlock* pBlock, SMqDataRsp* pRsp, const SSchemaWrapper* pSW, int8_t precision) {	×
46	int32_t code = 0;	×
47	int32_t lino = 0;	×
48	SSchemaWrapper* pSchema = NULL;	×
49
50	size_t dataEncodeBufSize = blockGetEncodeSize(pBlock);	×
51	int32_t dataStrLen = sizeof(SRetrieveTableRspForTmq) + dataEncodeBufSize;	×
52	void* buf = taosMemoryCalloc(1, dataStrLen);	×
53	TSDB_CHECK_NULL(buf, code, lino, END, terrno);	×
54
55	SRetrieveTableRspForTmq* pRetrieve = (SRetrieveTableRspForTmq*)buf;	×
56	pRetrieve->version = RETRIEVE_TABLE_RSP_TMQ_VERSION;	×
57	pRetrieve->precision = precision;	×
58	pRetrieve->compressed = 0;	×
59	pRetrieve->numOfRows = htobe64((int64_t)pBlock->info.rows);	×
60
61	int32_t actualLen = blockEncode(pBlock, pRetrieve->data, dataEncodeBufSize, pSW->nCols);	×
62	TSDB_CHECK_CONDITION(actualLen >= 0, code, lino, END, terrno);	×
63
64	actualLen += sizeof(SRetrieveTableRspForTmq);	×
65	TSDB_CHECK_NULL(taosArrayPush(pRsp->blockDataLen, &actualLen), code, lino, END, terrno);	×
66	TSDB_CHECK_NULL(taosArrayPush(pRsp->blockData, &buf), code, lino, END, terrno);	×
67	pSchema = tCloneSSchemaWrapper(pSW);	×
68	TSDB_CHECK_NULL(pSchema, code, lino, END, terrno);	×
69	TSDB_CHECK_NULL(taosArrayPush(pRsp->blockSchema, &pSchema), code, lino, END, terrno);	×
70	pSchema = NULL;	×
71	pRsp->blockDataElementFree = true;	×
72	tqTrace("tqAddBlockDataToRsp add block data to block array, blockDataLen:%d, blockData:%p", dataStrLen, buf);	×
73
74	END:	×
75	tDeleteSchemaWrapper(pSchema);	×
76	if (code != TSDB_CODE_SUCCESS){	×
77	taosMemoryFree(buf);	×
78	tqError("%s failed at line %d with msg:%s", __func__, lino, tstrerror(code));	×
79	}
80	return code;	×
81	}
82
83	static int32_t tqAddTbNameToRsp(const STQ* pTq, int64_t uid, SMqDataRsp* pRsp, int32_t n) {	×
84	int32_t code = TDB_CODE_SUCCESS;	×
85	int32_t lino = 0;	×
86	SMetaReader mr = {0};	×
87
88	TSDB_CHECK_NULL(pTq, code, lino, END, TSDB_CODE_INVALID_PARA);	×
89	TSDB_CHECK_NULL(pRsp, code, lino, END, TSDB_CODE_INVALID_PARA);	×
90
91	metaReaderDoInit(&mr, pTq->pVnode->pMeta, META_READER_LOCK);	×
92
93	code = metaReaderGetTableEntryByUidCache(&mr, uid);	×
94	TSDB_CHECK_CODE(code, lino, END);	×
95
96	for (int32_t i = 0; i < n; i++) {	×
97	char* tbName = taosStrdup(mr.me.name);	×
98	TSDB_CHECK_NULL(tbName, code, lino, END, terrno);	×
99	if(taosArrayPush(pRsp->blockTbName, &tbName) == NULL){	×
100	tqError("failed to push tbName to blockTbName:%s, uid:%"PRId64, tbName, uid);	×
101	continue;	×
102	}
103	tqTrace("add tbName to response success tbname:%s, uid:%"PRId64, tbName, uid);	×
104	}
105
106	END:	×
107	if (code != TSDB_CODE_SUCCESS) {	×
108	tqError("%s failed at %d, failed to add tbName to response:%s, uid:%"PRId64, __FUNCTION__, lino, tstrerror(code), uid);	×
109	}
110	metaReaderClear(&mr);	×
111	return code;	×
112	}
113
114	int32_t getDataBlock(qTaskInfo_t task, const STqHandle* pHandle, int32_t vgId, SSDataBlock** res) {	×
115	if (task == NULL \|\| pHandle == NULL \|\| res == NULL) {	×
116	return TSDB_CODE_INVALID_PARA;	×
117	}
118	uint64_t ts = 0;	×
119	qStreamSetOpen(task);	×
120
121	tqDebug("consumer:0x%" PRIx64 " vgId:%d, tmq one task start execute", pHandle->consumerId, vgId);	×
122	int32_t code = qExecTask(task, res, &ts);	×
123	if (code != TSDB_CODE_SUCCESS) {	×
124	tqError("consumer:0x%" PRIx64 " vgId:%d, task exec error since %s", pHandle->consumerId, vgId, tstrerror(code));	×
125	}
126
127	tqDebug("consumer:0x%" PRIx64 " vgId:%d tmq one task end executed, pDataBlock:%p", pHandle->consumerId, vgId, *res);	×
128	return code;	×
129	}
130
131	static int32_t tqProcessReplayRsp(STQ* pTq, STqHandle* pHandle, SMqDataRsp* pRsp, const SMqPollReq* pRequest, SSDataBlock* pDataBlock, qTaskInfo_t task){	×
132	int32_t code = 0;	×
133	int32_t lino = 0;	×
134
135	if (IS_OFFSET_RESET_TYPE(pRequest->reqOffset.type) && pHandle->block != NULL) {	×
136	blockDataDestroy(pHandle->block);	×
137	pHandle->block = NULL;	×
138	}
139	if (pHandle->block == NULL) {	×
140	if (pDataBlock == NULL) {	×
141	goto END;	×
142	}
143
144	STqOffsetVal offset = {0};	×
145	code = qStreamExtractOffset(task, &offset);	×
146	TSDB_CHECK_CODE(code, lino, END);	×
147
148	pHandle->block = NULL;	×
149
150	code = createOneDataBlock(pDataBlock, true, &pHandle->block);	×
151	TSDB_CHECK_CODE(code, lino, END);	×
152
153	pHandle->blockTime = offset.ts;	×
154	tOffsetDestroy(&offset);	×
155	int32_t vgId = TD_VID(pTq->pVnode);	×
156	code = getDataBlock(task, pHandle, vgId, &pDataBlock);	×
157	TSDB_CHECK_CODE(code, lino, END);	×
158	}
159
160	const STqExecHandle* pExec = &pHandle->execHandle;	×
161	code = tqAddBlockDataToRsp(pHandle->block, pRsp, &pExec->execCol.pSW, pTq->pVnode->config.tsdbCfg.precision);	×
162	TSDB_CHECK_CODE(code, lino, END);	×
163
164	pRsp->blockNum++;	×
165	if (pDataBlock == NULL) {	×
166	blockDataDestroy(pHandle->block);	×
167	pHandle->block = NULL;	×
168	} else {
169	code = copyDataBlock(pHandle->block, pDataBlock);	×
170	TSDB_CHECK_CODE(code, lino, END);	×
171
172	STqOffsetVal offset = {0};	×
173	code = qStreamExtractOffset(task, &offset);	×
174	TSDB_CHECK_CODE(code, lino, END);	×
175
176	pRsp->sleepTime = offset.ts - pHandle->blockTime;	×
177	pHandle->blockTime = offset.ts;	×
178	tOffsetDestroy(&offset);	×
179	}
180
181	END:	×
182	if (code != TSDB_CODE_SUCCESS) {	×
183	tqError("%s failed at %d, failed to process replay response:%s", __FUNCTION__, lino, tstrerror(code));	×
184	}
185	return code;	×
186	}
187
188	int32_t tqScanData(STQ* pTq, STqHandle* pHandle, SMqDataRsp* pRsp, STqOffsetVal* pOffset, const SMqPollReq* pRequest) {	×
189	int32_t code = 0;	×
190	int32_t lino = 0;	×
191	TSDB_CHECK_NULL(pRsp, code, lino, END, TSDB_CODE_INVALID_PARA);	×
192	TSDB_CHECK_NULL(pTq, code, lino, END, TSDB_CODE_INVALID_PARA);	×
193	TSDB_CHECK_NULL(pHandle, code, lino, END, TSDB_CODE_INVALID_PARA);	×
194	TSDB_CHECK_NULL(pOffset, code, lino, END, TSDB_CODE_INVALID_PARA);	×
195	TSDB_CHECK_NULL(pRequest, code, lino, END, TSDB_CODE_INVALID_PARA);	×
196
197	int32_t vgId = TD_VID(pTq->pVnode);	×
198	int32_t totalRows = 0;	×
199
200	const STqExecHandle* pExec = &pHandle->execHandle;	×
201	qTaskInfo_t task = pExec->task;	×
202
203	code = qStreamPrepareScan(task, pOffset, pHandle->execHandle.subType);	×
204	TSDB_CHECK_CODE(code, lino, END);	×
205
206	qStreamSetParams(task, pRequest->sourceExcluded, pRequest->minPollRows, pRequest->timeout, pRequest->enableReplay);	×
207	do {
208	SSDataBlock* pDataBlock = NULL;	×
209	code = getDataBlock(task, pHandle, vgId, &pDataBlock);	×
210	TSDB_CHECK_CODE(code, lino, END);	×
211
212	if (pRequest->enableReplay) {	×
213	code = tqProcessReplayRsp(pTq, pHandle, pRsp, pRequest, pDataBlock, task);	×
214	TSDB_CHECK_CODE(code, lino, END);	×
215	break;	×
216	}
217	if (pDataBlock == NULL) {	×
218	break;	×
219	}
220	code = tqAddBlockDataToRsp(pDataBlock, pRsp, &pExec->execCol.pSW, pTq->pVnode->config.tsdbCfg.precision);	×
221	TSDB_CHECK_CODE(code, lino, END);	×
222
223	pRsp->blockNum++;	×
224	totalRows += pDataBlock->info.rows;	×
225	} while(0);
226
227	tqDebug("consumer:0x%" PRIx64 " vgId:%d tmq task executed finished, total blocks:%d, totalRows:%d", pHandle->consumerId, vgId, pRsp->blockNum, totalRows);	×
228	code = qStreamExtractOffset(task, &pRsp->rspOffset);	×
229
230	END:	×
231	if (code != 0) {	×
232	tqError("%s failed at %d, tmq task executed error msg:%s", __FUNCTION__, lino, tstrerror(code));	×
233	}
234	return code;	×
235	}
236
237	int32_t tqScanTaosx(STQ* pTq, const STqHandle* pHandle, SMqDataRsp* pRsp, SMqBatchMetaRsp* pBatchMetaRsp, STqOffsetVal* pOffset, const SMqPollReq* pRequest) {	×
238	int32_t code = 0;	×
239	int32_t lino = 0;	×
240	char* tbName = NULL;	×
241	const STqExecHandle* pExec = &pHandle->execHandle;	×
242	qTaskInfo_t task = pExec->task;	×
243	code = qStreamPrepareScan(task, pOffset, pHandle->execHandle.subType);	×
244	TSDB_CHECK_CODE(code, lino, END);	×
245
246	qStreamSetParams(task, pRequest->sourceExcluded, pRequest->minPollRows, pRequest->timeout, false);	×
247
248	int32_t rowCnt = 0;	×
249	int64_t st = taosGetTimestampMs();	×
250	while (1) {	×
251	SSDataBlock* pDataBlock = NULL;	×
252	uint64_t ts = 0;	×
253	tqDebug("tmqsnap task start to execute");	×
254	code = qExecTask(task, &pDataBlock, &ts);	×
255	TSDB_CHECK_CODE(code, lino, END);	×
256	tqDebug("tmqsnap task execute end, get %p", pDataBlock);	×
257
258	if (pDataBlock != NULL && pDataBlock->info.rows > 0) {	×
259	if (pRsp->withTbName) {	×
260	tbName = taosStrdup(qExtractTbnameFromTask(task));	×
261	TSDB_CHECK_NULL(tbName, code, lino, END, terrno);	×
262	TSDB_CHECK_NULL(taosArrayPush(pRsp->blockTbName, &tbName), code, lino, END, terrno);	×
263	tqDebug("vgId:%d, add tbname:%s to rsp msg", pTq->pVnode->config.vgId, tbName);	×
264	tbName = NULL;	×
265	}
266
267	code = tqAddBlockDataToRsp(pDataBlock, pRsp, qExtractSchemaFromTask(task), pTq->pVnode->config.tsdbCfg.precision);	×
268	TSDB_CHECK_CODE(code, lino, END);	×
269
270	pRsp->blockNum++;	×
271	rowCnt += pDataBlock->info.rows;	×
272	if (rowCnt <= pRequest->minPollRows && (taosGetTimestampMs() - st <= pRequest->timeout)) {	×
273	continue;	×
274	}
275	}
276
277	// get meta
278	SMqBatchMetaRsp* tmp = qStreamExtractMetaMsg(task);	×
279	if (taosArrayGetSize(tmp->batchMetaReq) > 0) {	×
280	code = qStreamExtractOffset(task, &tmp->rspOffset);	×
281	TSDB_CHECK_CODE(code, lino, END);	×
282	pBatchMetaRsp = tmp;	×
283	tqDebug("tmqsnap task get meta");	×
284	break;	×
285	}
286
287	if (pDataBlock == NULL) {	×
288	code = qStreamExtractOffset(task, pOffset);	×
289	TSDB_CHECK_CODE(code, lino, END);	×
290
291	if (pOffset->type == TMQ_OFFSET__SNAPSHOT_DATA) {	×
292	continue;	×
293	}
294
295	tqDebug("tmqsnap vgId: %d, tsdb consume over, switch to wal, ver %" PRId64, TD_VID(pTq->pVnode), pHandle->snapshotVer + 1);	×
296	code = qStreamExtractOffset(task, &pRsp->rspOffset);	×
297	break;	×
298	}
299
300	if (pRsp->blockNum > 0) {	×
301	tqDebug("tmqsnap task exec exited, get data");	×
302	code = qStreamExtractOffset(task, &pRsp->rspOffset);	×
303	break;	×
304	}
305	}
306	tqDebug("%s:%d success", __FUNCTION__, lino);	×
307	END:	×
308	if (code != 0){	×
309	tqError("%s failed at %d, vgId:%d, task exec error since %s", __FUNCTION__ , lino, pTq->pVnode->config.vgId, tstrerror(code));	×
310	}
311	taosMemoryFree(tbName);	×
312	return code;	×
313	}
314
315	static void tqProcessSubData(STQ* pTq, STqHandle* pHandle, SMqDataRsp* pRsp, int32_t* totalRows,	×
316	const SMqPollReq* pRequest, SArray* rawList){
317	int32_t code = 0;	×
318	int32_t lino = 0;	×
319	SArray* pBlocks = NULL;	×
320	SArray* pSchemas = NULL;	×
321
322	STqExecHandle* pExec = &pHandle->execHandle;	×
323	STqReader* pReader = pExec->pTqReader;	×
324
325	pBlocks = taosArrayInit(0, sizeof(SSDataBlock));	×
326	TSDB_CHECK_NULL(pBlocks, code, lino, END, terrno);	×
327	pSchemas = taosArrayInit(0, sizeof(void*));	×
328	TSDB_CHECK_NULL(pSchemas, code, lino, END, terrno);	×
329
330	SSubmitTbData* pSubmitTbData = NULL;	×
331	code = tqRetrieveTaosxBlock(pReader, pRsp, pBlocks, pSchemas, &pSubmitTbData, rawList, pHandle->fetchMeta);	×
332	TSDB_CHECK_CODE(code, lino, END);	×
333	bool tmp = (pSubmitTbData->flags & pRequest->sourceExcluded) != 0;	×
334	TSDB_CHECK_CONDITION(!tmp, code, lino, END, TSDB_CODE_SUCCESS);	×
335
336	if (pHandle->fetchMeta == ONLY_META){	×
337	goto END;	×
338	}
339
340	int32_t blockNum = taosArrayGetSize(pBlocks) == 0 ? 1 : taosArrayGetSize(pBlocks);	×
341	if (pRsp->withTbName) {	×
342	int64_t uid = pExec->pTqReader->lastBlkUid;	×
343	code = tqAddTbNameToRsp(pTq, uid, pRsp, blockNum);	×
344	TSDB_CHECK_CODE(code, lino, END);	×
345	}
346
347	TSDB_CHECK_CONDITION(!tmp, code, lino, END, TSDB_CODE_SUCCESS);	×
348	for (int32_t i = 0; i < blockNum; i++) {	×
349	if (taosArrayGetSize(pBlocks) == 0){	×
350	void* rawData = taosArrayGetP(rawList, pReader->nextBlk - 1);	×
351	if (rawData == NULL) {	×
352	continue;	×
353	}
354	if (tqAddRawDataToRsp(rawData, pRsp, pTq->pVnode->config.tsdbCfg.precision) != 0){	×
355	tqError("vgId:%d, failed to add block to rsp msg", pTq->pVnode->config.vgId);	×
356	continue;	×
357	}
358	totalRows += (uint32_t *)rawData + INT_BYTES; // bytes actually	×
359	} else {
360	SSDataBlock* pBlock = taosArrayGet(pBlocks, i);	×
361	if (pBlock == NULL) {	×
362	continue;	×
363	}
364
365	SSchemaWrapper* pSW = (SSchemaWrapper*)taosArrayGetP(pSchemas, i);	×
366	if (tqAddBlockDataToRsp(pBlock, pRsp, pSW, pTq->pVnode->config.tsdbCfg.precision) != 0){	×
367	tqError("vgId:%d, failed to add block to rsp msg", pTq->pVnode->config.vgId);	×
368	continue;	×
369	}
370	*totalRows += pBlock->info.rows;	×
371	}
372
373	pRsp->blockNum++;	×
374	}
375	tqTrace("vgId:%d, process sub data success, response blocknum:%d, rows:%d", pTq->pVnode->config.vgId, pRsp->blockNum, *totalRows);	×
376	END:	×
377	if (code != 0) {	×
378	tqError("%s failed at %d, failed to process sub data:%s", __FUNCTION__, lino, tstrerror(code));	×
379	}
380	taosArrayDestroyEx(pBlocks, (FDelete)blockDataFreeRes);	×
381	taosArrayDestroyP(pSchemas, (FDelete)tDeleteSchemaWrapper);	×
382	}	×
383
384	static void preProcessSubmitMsg(STqHandle* pHandle, const SMqPollReq* pRequest, SArray** rawList){	×
385	STqExecHandle* pExec = &pHandle->execHandle;	×
386	STqReader* pReader = pExec->pTqReader;	×
387	int32_t blockSz = taosArrayGetSize(pReader->submit.aSubmitTbData);	×
388	for (int32_t i = 0; i < blockSz; i++){	×
389	SSubmitTbData* pSubmitTbData = taosArrayGet(pReader->submit.aSubmitTbData, i);	×
390	if (pSubmitTbData== NULL){	×
391	taosArrayDestroy(*rawList);	×
392	*rawList = NULL;	×
393	return;	×
394	}
395
396	int64_t uid = pSubmitTbData->uid;	×
397	if (pRequest->rawData) {	×
398	if (taosHashGet(pRequest->uidHash, &uid, LONG_BYTES) != NULL) {	×
399	tqDebug("poll rawdata split,uid:%" PRId64 " is already exists", uid);	×
400	terrno = TSDB_CODE_TMQ_RAW_DATA_SPLIT;	×
401	return;	×
402	} else {
403	int32_t code = taosHashPut(pRequest->uidHash, &uid, LONG_BYTES, &uid, LONG_BYTES);	×
404	if (code != 0) {	×
405	tqError("failed to add table uid to hash, code:%d, uid:%" PRId64, code, uid);	×
406	}
407	}
408	}
409
410	if (pSubmitTbData->pCreateTbReq == NULL){	×
411	continue;	×
412	}
413
414	int64_t createTime = INT64_MAX;	×
415	int64_t cTime = (int64_t)taosHashGet(pHandle->tableCreateTimeHash, &uid, LONG_BYTES);	×
416	if (cTime != NULL){	×
417	createTime = *cTime;	×
418	} else{
419	createTime = metaGetTableCreateTime(pReader->pVnode->pMeta, uid, 1);	×
420	if (createTime != INT64_MAX){	×
421	int32_t code = taosHashPut(pHandle->tableCreateTimeHash, &uid, LONG_BYTES, &createTime, LONG_BYTES);	×
422	if (code != 0){	×
423	tqError("failed to add table create time to hash,code:%d, uid:%"PRId64, code, uid);	×
424	}
425	}
426	}
427	if (pHandle->fetchMeta == WITH_DATA \|\| pSubmitTbData->ctimeMs > createTime){	×
428	tDestroySVSubmitCreateTbReq(pSubmitTbData->pCreateTbReq, TSDB_MSG_FLG_DECODE);	×
429	taosMemoryFreeClear(pSubmitTbData->pCreateTbReq);	×
430	} else{
431	taosArrayDestroy(*rawList);	×
432	*rawList = NULL;	×
433	}
434	}
435	}
436
437	int32_t tqTaosxScanLog(STQ* pTq, STqHandle* pHandle, SPackedData submit, SMqDataRsp* pRsp, int32_t* totalRows, const SMqPollReq* pRequest) {	×
438	int32_t code = 0;	×
439	int32_t lino = 0;	×
440	SDecoder decoder = {0};	×
441	STqExecHandle* pExec = &pHandle->execHandle;	×
442	STqReader* pReader = pExec->pTqReader;	×
443	SArray *rawList = NULL;	×
444	if (pRequest->rawData){	×
445	rawList = taosArrayInit(0, POINTER_BYTES);	×
446	TSDB_CHECK_NULL(rawList, code, lino, END, terrno);	×
447	}
448	code = tqReaderSetSubmitMsg(pReader, submit.msgStr, submit.msgLen, submit.ver, rawList, &decoder);	×
449	TSDB_CHECK_CODE(code, lino, END);	×
450	preProcessSubmitMsg(pHandle, pRequest, &rawList);	×
451	// data could not contains same uid data in rawdata mode
452	if (pRequest->rawData != 0 && terrno == TSDB_CODE_TMQ_RAW_DATA_SPLIT){	×
453	goto END;	×
454	}
455
456	// this submit data is metadata and previous data is rawdata
457	if (pRequest->rawData != 0 && *totalRows > 0 && pRsp->createTableNum == 0 && rawList == NULL){	×
458	tqDebug("poll rawdata split,vgId:%d, this wal submit data contains metadata and previous data is data", pTq->pVnode->config.vgId);	×
459	terrno = TSDB_CODE_TMQ_RAW_DATA_SPLIT;	×
460	goto END;	×
461	}
462
463	// this submit data is rawdata and previous data is metadata
464	if (pRequest->rawData != 0 && pRsp->createTableNum > 0 && rawList != NULL){	×
465	tqDebug("poll rawdata split,vgId:%d, this wal submit data is data and previous data is metadata", pTq->pVnode->config.vgId);	×
466	terrno = TSDB_CODE_TMQ_RAW_DATA_SPLIT;	×
467	goto END;	×
468	}
469
470	if (pExec->subType == TOPIC_SUB_TYPE__TABLE) {	×
471	while (tqNextBlockImpl(pReader, NULL)) {	×
472	tqProcessSubData(pTq, pHandle, pRsp, totalRows, pRequest, rawList);	×
473	}
474	} else if (pExec->subType == TOPIC_SUB_TYPE__DB) {	×
475	while (tqNextDataBlockFilterOut(pReader, pExec->execDb.pFilterOutTbUid)) {	×
476	tqProcessSubData(pTq, pHandle, pRsp, totalRows, pRequest, rawList);	×
477	}
478	}
479
480	END:	×
481	tDecoderClear(&decoder);	×
482	tqReaderClearSubmitMsg(pReader);	×
483	taosArrayDestroy(rawList);	×
484	if (code != 0){	×
485	tqError("%s failed at %d, failed to scan log:%s", __FUNCTION__, lino, tstrerror(code));	×
486	}
487	return code;	×
488	}

taosdata / TDengine / #4986

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