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

18

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

×

19

  int32_t code = 0;

×

20

  int32_t lino = 0;

×

22

  size_t dataEncodeBufSize = blockGetEncodeSize(pBlock);

×

23

  int32_t dataStrLen = sizeof(SRetrieveTableRspForTmq) + dataEncodeBufSize;

×

24

  void*   buf = taosMemoryCalloc(1, dataStrLen);

×

25

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

×

27

  SRetrieveTableRspForTmq* pRetrieve = (SRetrieveTableRspForTmq*)buf;

×

28

  pRetrieve->version = 1;

×

29

  pRetrieve->precision = precision;

×

30

  pRetrieve->compressed = 0;

×

31

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

×

33

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

×

34

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

×

36

  actualLen += sizeof(SRetrieveTableRspForTmq);

×

37

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

×

38

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

×

40

  buf = NULL;

×

41

END:

×

42

  if (code != 0){

×

43

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

×

45

  taosMemoryFree(buf);

×

46

  return code;

×

49

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

×

50

  int32_t    code = TDB_CODE_SUCCESS;

×

51

  int32_t    lino = 0;

×

52

  SMetaReader mr = {0};

×

54

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

×

55

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

×

57

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

×

59

  code = metaReaderGetTableEntryByUidCache(&mr, uid);

×

60

  TSDB_CHECK_CODE(code, lino, END);

×

62

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

×

63

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

×

64

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

×

65

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

×

66

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

×

67

      continue;

×

69

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

×

72

END:

×

73

  if (code != TSDB_CODE_SUCCESS) {

×

74

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

×

76

  metaReaderClear(&mr);

×

77

  return code;

×

80

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

×

81

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

×

82

    return TSDB_CODE_INVALID_PARA;

×

84

  uint64_t ts = 0;

×

85

  qStreamSetOpen(task);

×

87

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

×

88

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

×

89

  if (code != TSDB_CODE_SUCCESS) {

×

90

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

×

91

    return code;

×

94

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

×

95

  return 0;

×

98

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

×

99

  int32_t code = 0;

×

100

  int32_t lino = 0;

×

102

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

×

103

    blockDataDestroy(pHandle->block);

×

104

    pHandle->block = NULL;

×

106

  if (pHandle->block == NULL) {

×

107

    if (pDataBlock == NULL) {

×

108

      goto END;

×

111

    STqOffsetVal offset = {0};

×

112

    code = qStreamExtractOffset(task, &offset);

×

113

    TSDB_CHECK_CODE(code, lino, END);

×

115

    pHandle->block = NULL;

×

117

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

×

118

    TSDB_CHECK_CODE(code, lino, END);

×

120

    pHandle->blockTime = offset.ts;

×

121

    tOffsetDestroy(&offset);

×

122

    int32_t vgId = TD_VID(pTq->pVnode);

×

123

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

×

124

    TSDB_CHECK_CODE(code, lino, END);

×

127

  const STqExecHandle* pExec = &pHandle->execHandle;

×

128

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

×

129

  TSDB_CHECK_CODE(code, lino, END);

×

131

  pRsp->blockNum++;

×

132

  if (pDataBlock == NULL) {

×

133

    blockDataDestroy(pHandle->block);

×

134

    pHandle->block = NULL;

×

136

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

×

137

    TSDB_CHECK_CODE(code, lino, END);

×

139

    STqOffsetVal offset = {0};

×

140

    code = qStreamExtractOffset(task, &offset);

×

141

    TSDB_CHECK_CODE(code, lino, END);

×

143

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

×

144

    pHandle->blockTime = offset.ts;

×

145

    tOffsetDestroy(&offset);

×

148

END:

×

149

  if (code != TSDB_CODE_SUCCESS) {

×

150

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

×

152

  return code;

×

155

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

×

156

  int32_t code = 0;

×

157

  int32_t lino = 0;

×

158

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

×

159

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

×

160

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

×

161

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

×

162

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

×

164

  int32_t vgId = TD_VID(pTq->pVnode);

×

165

  int32_t totalRows = 0;

×

167

  const STqExecHandle* pExec = &pHandle->execHandle;

×

168

  qTaskInfo_t          task = pExec->task;

×

170

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

×

171

  TSDB_CHECK_CODE(code, lino, END);

×

173

  qStreamSetSourceExcluded(task, pRequest->sourceExcluded);

×

174

  int64_t st = taosGetTimestampMs();

×

175

  while (1) {

×

176

    SSDataBlock* pDataBlock = NULL;

×

177

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

×

178

    TSDB_CHECK_CODE(code, lino, END);

×

180

    if (pRequest->enableReplay) {

×

181

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

×

182

      TSDB_CHECK_CODE(code, lino, END);

×

183

      break;

×

185

    if (pDataBlock == NULL) {

×

186

      break;

×

188

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

×

189

    TSDB_CHECK_CODE(code, lino, END);

×

191

    pRsp->blockNum++;

×

192

    totalRows += pDataBlock->info.rows;

×

193

    if (totalRows >= tmqRowSize || (taosGetTimestampMs() - st > TMIN(TQ_POLL_MAX_TIME, pRequest->timeout))) {

×

198

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

×

199

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

×

201

END:

×

202

  if (code != 0) {

×

203

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

×

205

  return code;

×

208

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

×

209

  int32_t code = 0;

×

210

  int32_t lino = 0;

×

211

  char* tbName = NULL;

×

212

  SSchemaWrapper* pSW = NULL;

×

213

  const STqExecHandle* pExec = &pHandle->execHandle;

×

214

  qTaskInfo_t          task = pExec->task;

×

215

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

×

216

  TSDB_CHECK_CODE(code, lino, END);

×

218

  int32_t rowCnt = 0;

×

219

  int64_t st = taosGetTimestampMs();

×

220

  while (1) {

×

221

    SSDataBlock* pDataBlock = NULL;

×

222

    uint64_t     ts = 0;

×

223

    tqDebug("tmqsnap task start to execute");

×

224

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

×

225

    TSDB_CHECK_CODE(code, lino, END);

×

226

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

×

228

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

×

229

      if (pRsp->withTbName) {

×

230

        tbName = taosStrdup(qExtractTbnameFromTask(task));

×

231

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

×

232

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

×

233

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

×

234

        tbName = NULL;

×

236

      if (pRsp->withSchema) {

×

237

        SSchemaWrapper* pSW = tCloneSSchemaWrapper(qExtractSchemaFromTask(task));

×

238

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

×

239

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

×

240

        pSW = NULL;

×

243

      code = tqAddBlockDataToRsp(pDataBlock, pRsp, taosArrayGetSize(pDataBlock->pDataBlock),

×

244

                                 pTq->pVnode->config.tsdbCfg.precision);

×

245

      TSDB_CHECK_CODE(code, lino, END);

×

247

      pRsp->blockNum++;

×

248

      rowCnt += pDataBlock->info.rows;

×

249

      if (rowCnt <= tmqRowSize && (taosGetTimestampMs() - st <= TMIN(TQ_POLL_MAX_TIME, timeout))) {

×

250

        continue;

×

255

    SMqBatchMetaRsp* tmp = qStreamExtractMetaMsg(task);

×

256

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

×

257

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

×

258

      TSDB_CHECK_CODE(code, lino, END);

×

259

      *pBatchMetaRsp = *tmp;

×

260

      tqDebug("tmqsnap task get meta");

×

261

      break;

×

264

    if (pDataBlock == NULL) {

×

265

      code = qStreamExtractOffset(task, pOffset);

×

266

      TSDB_CHECK_CODE(code, lino, END);

×

268

      if (pOffset->type == TMQ_OFFSET__SNAPSHOT_DATA) {

×

269

        continue;

×

272

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

×

273

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

×

274

      break;

×

277

    if (pRsp->blockNum > 0) {

×

278

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

×

279

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

×

280

      break;

×

283

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

×

284

END:

×

285

  if (code != 0){

×

286

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

×

288

  taosMemoryFree(pSW);

×

289

  taosMemoryFree(tbName);

×

290

  return code;

×

293

static int32_t buildCreateTbInfo(SMqDataRsp* pRsp, SVCreateTbReq* pCreateTbReq){

×

294

  int32_t code = 0;

×

295

  int32_t lino = 0;

×

296

  void*   createReq = NULL;

×

297

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

×

298

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

×

300

  if (pRsp->createTableNum == 0) {

×

301

    pRsp->createTableLen = taosArrayInit(0, sizeof(int32_t));

×

302

    TSDB_CHECK_NULL(pRsp->createTableLen, code, lino, END, terrno);

×

303

    pRsp->createTableReq = taosArrayInit(0, sizeof(void*));

×

304

    TSDB_CHECK_NULL(pRsp->createTableReq, code, lino, END, terrno);

×

307

  uint32_t len = 0;

×

308

  tEncodeSize(tEncodeSVCreateTbReq, pCreateTbReq, len, code);

×

309

  TSDB_CHECK_CODE(code, lino, END);

×

310

  createReq = taosMemoryCalloc(1, len);

×

311

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

×

313

  SEncoder encoder = {0};

×

314

  tEncoderInit(&encoder, createReq, len);

×

315

  code = tEncodeSVCreateTbReq(&encoder, pCreateTbReq);

×

316

  tEncoderClear(&encoder);

×

317

  TSDB_CHECK_CODE(code, lino, END);

×

318

  TSDB_CHECK_NULL(taosArrayPush(pRsp->createTableLen, &len), code, lino, END, terrno);

×

319

  TSDB_CHECK_NULL(taosArrayPush(pRsp->createTableReq, &createReq), code, lino, END, terrno);

×

320

  pRsp->createTableNum++;

×

321

  tqDebug("build create table info msg success");

×

323

END:

×

324

  if (code != 0){

×

325

    tqError("%s failed at %d, failed to build create table info msg:%s", __FUNCTION__, lino, tstrerror(code));

×

326

    taosMemoryFree(createReq);

×

328

  return code;

×

331

static void tqProcessSubData(STQ* pTq, STqHandle* pHandle, SMqDataRsp* pRsp, int32_t* totalRows, int8_t sourceExcluded){

×

332

  int32_t code = 0;

×

333

  int32_t lino = 0;

×

334

  SArray* pBlocks = NULL;

×

335

  SArray* pSchemas = NULL;

×

337

  STqExecHandle* pExec = &pHandle->execHandle;

×

338

  STqReader* pReader = pExec->pTqReader;

×

340

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

×

341

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

×

342

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

×

343

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

×

345

  SSubmitTbData* pSubmitTbDataRet = NULL;

×

346

  int64_t createTime = INT64_MAX;

×

347

  code = tqRetrieveTaosxBlock(pReader, pBlocks, pSchemas, &pSubmitTbDataRet, &createTime);

×

348

  TSDB_CHECK_CODE(code, lino, END);

×

349

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

×

350

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

×

351

  if (pRsp->withTbName) {

×

352

    int64_t uid = pExec->pTqReader->lastBlkUid;

×

353

    code = tqAddTbNameToRsp(pTq, uid, pRsp, taosArrayGetSize(pBlocks));

×

354

    TSDB_CHECK_CODE(code, lino, END);

×

356

  if (pHandle->fetchMeta != WITH_DATA && pSubmitTbDataRet->pCreateTbReq != NULL) {

×

357

    if (pSubmitTbDataRet->ctimeMs - createTime <= 1000) {  // judge if table is already created to avoid sending crateTbReq

×

358

      code = buildCreateTbInfo(pRsp, pSubmitTbDataRet->pCreateTbReq);

×

359

      TSDB_CHECK_CODE(code, lino, END);

×

362

  tmp = (pHandle->fetchMeta == ONLY_META && pSubmitTbDataRet->pCreateTbReq == NULL);

×

363

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

×

364

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

×

365

    SSDataBlock* pBlock = taosArrayGet(pBlocks, i);

×

366

    if (pBlock == NULL) {

×

367

      continue;

×

369

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

×

370

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

×

371

      continue;

×

373

    *totalRows += pBlock->info.rows;

×

374

    blockDataFreeRes(pBlock);

×

375

    SSchemaWrapper* pSW = taosArrayGetP(pSchemas, i);

×

376

    if (taosArrayPush(pRsp->blockSchema, &pSW) == NULL){

×

377

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

×

378

      continue;

×

380

    pRsp->blockNum++;

×

382

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

×

383

END:

×

384

  if (code != 0){

×

385

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

×

386

    taosArrayDestroyEx(pBlocks, (FDelete)blockDataFreeRes);

×

387

    taosArrayDestroyP(pSchemas, (FDelete)tDeleteSchemaWrapper);

×

389

    taosArrayDestroy(pBlocks);

×

390

    taosArrayDestroy(pSchemas);

×

392

×

394

int32_t tqTaosxScanLog(STQ* pTq, STqHandle* pHandle, SPackedData submit, SMqDataRsp* pRsp, int32_t* totalRows, int8_t sourceExcluded) {

×

395

  int32_t code = 0;

×

396

  int32_t lino = 0;

×

397

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

×

398

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

×

399

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

×

400

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

×

401

  STqExecHandle* pExec = &pHandle->execHandle;

×

402

  STqReader* pReader = pExec->pTqReader;

×

403

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

×

404

  TSDB_CHECK_CODE(code, lino, END);

×

406

  if (pExec->subType == TOPIC_SUB_TYPE__TABLE) {

×

407

    while (tqNextBlockImpl(pReader, NULL)) {

×

408

      tqProcessSubData(pTq, pHandle, pRsp, totalRows, sourceExcluded);

×

410

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

×

411

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

×

412

      tqProcessSubData(pTq, pHandle, pRsp, totalRows, sourceExcluded);

×

416

END:

×

417

  if (code != 0){

×

418

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

×

420

  return code;

×

taosdata / TDengine / #3599

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
18	static int32_t tqAddBlockDataToRsp(const SSDataBlock* pBlock, SMqDataRsp* pRsp, int32_t numOfCols, int8_t precision) {	×
19	int32_t code = 0;	×
20	int32_t lino = 0;	×
21
22	size_t dataEncodeBufSize = blockGetEncodeSize(pBlock);	×
23	int32_t dataStrLen = sizeof(SRetrieveTableRspForTmq) + dataEncodeBufSize;	×
24	void* buf = taosMemoryCalloc(1, dataStrLen);	×
25	TSDB_CHECK_NULL(buf, code, lino, END, terrno);	×
26
27	SRetrieveTableRspForTmq* pRetrieve = (SRetrieveTableRspForTmq*)buf;	×
28	pRetrieve->version = 1;	×
29	pRetrieve->precision = precision;	×
30	pRetrieve->compressed = 0;	×
31	pRetrieve->numOfRows = htobe64((int64_t)pBlock->info.rows);	×
32
33	int32_t actualLen = blockEncode(pBlock, pRetrieve->data, dataEncodeBufSize, numOfCols);	×
34	TSDB_CHECK_CONDITION(actualLen >= 0, code, lino, END, terrno);	×
35
36	actualLen += sizeof(SRetrieveTableRspForTmq);	×
37	TSDB_CHECK_NULL(taosArrayPush(pRsp->blockDataLen, &actualLen), code, lino, END, terrno);	×
38	TSDB_CHECK_NULL(taosArrayPush(pRsp->blockData, &buf), code, lino, END, terrno);	×
39
40	buf = NULL;	×
41	END:	×
42	if (code != 0){	×
43	tqError("%s failed at line %d with msg:%s", __func__, lino, tstrerror(code));	×
44	}
45	taosMemoryFree(buf);	×
46	return code;	×
47	}
48
49	static int32_t tqAddTbNameToRsp(const STQ* pTq, int64_t uid, SMqDataRsp* pRsp, int32_t n) {	×
50	int32_t code = TDB_CODE_SUCCESS;	×
51	int32_t lino = 0;	×
52	SMetaReader mr = {0};	×
53
54	TSDB_CHECK_NULL(pTq, code, lino, END, TSDB_CODE_INVALID_PARA);	×
55	TSDB_CHECK_NULL(pRsp, code, lino, END, TSDB_CODE_INVALID_PARA);	×
56
57	metaReaderDoInit(&mr, pTq->pVnode->pMeta, META_READER_LOCK);	×
58
59	code = metaReaderGetTableEntryByUidCache(&mr, uid);	×
60	TSDB_CHECK_CODE(code, lino, END);	×
61
62	for (int32_t i = 0; i < n; i++) {	×
63	char* tbName = taosStrdup(mr.me.name);	×
64	TSDB_CHECK_NULL(tbName, code, lino, END, terrno);	×
65	if(taosArrayPush(pRsp->blockTbName, &tbName) == NULL){	×
66	tqError("failed to push tbName to blockTbName:%s, uid:%"PRId64, tbName, uid);	×
67	continue;	×
68	}
69	tqDebug("add tbName to response success tbname:%s, uid:%"PRId64, tbName, uid);	×
70	}
71
72	END:	×
73	if (code != TSDB_CODE_SUCCESS) {	×
74	tqError("%s failed at %d, failed to add tbName to response:%s, uid:%"PRId64, __FUNCTION__, lino, tstrerror(code), uid);	×
75	}
76	metaReaderClear(&mr);	×
77	return code;	×
78	}
79
80	int32_t getDataBlock(qTaskInfo_t task, const STqHandle* pHandle, int32_t vgId, SSDataBlock** res) {	×
81	if (task == NULL \|\| pHandle == NULL \|\| res == NULL) {	×
82	return TSDB_CODE_INVALID_PARA;	×
83	}
84	uint64_t ts = 0;	×
85	qStreamSetOpen(task);	×
86
87	tqDebug("consumer:0x%" PRIx64 " vgId:%d, tmq one task start execute", pHandle->consumerId, vgId);	×
88	int32_t code = qExecTask(task, res, &ts);	×
89	if (code != TSDB_CODE_SUCCESS) {	×
90	tqError("consumer:0x%" PRIx64 " vgId:%d, task exec error since %s", pHandle->consumerId, vgId, tstrerror(code));	×
91	return code;	×
92	}
93
94	tqDebug("consumer:0x%" PRIx64 " vgId:%d tmq one task end executed, pDataBlock:%p", pHandle->consumerId, vgId, *res);	×
95	return 0;	×
96	}
97
98	static int32_t tqProcessReplayRsp(STQ* pTq, STqHandle* pHandle, SMqDataRsp* pRsp, const SMqPollReq* pRequest, SSDataBlock* pDataBlock, qTaskInfo_t task){	×
99	int32_t code = 0;	×
100	int32_t lino = 0;	×
101
102	if (IS_OFFSET_RESET_TYPE(pRequest->reqOffset.type) && pHandle->block != NULL) {	×
103	blockDataDestroy(pHandle->block);	×
104	pHandle->block = NULL;	×
105	}
106	if (pHandle->block == NULL) {	×
107	if (pDataBlock == NULL) {	×
108	goto END;	×
109	}
110
111	STqOffsetVal offset = {0};	×
112	code = qStreamExtractOffset(task, &offset);	×
113	TSDB_CHECK_CODE(code, lino, END);	×
114
115	pHandle->block = NULL;	×
116
117	code = createOneDataBlock(pDataBlock, true, &pHandle->block);	×
118	TSDB_CHECK_CODE(code, lino, END);	×
119
120	pHandle->blockTime = offset.ts;	×
121	tOffsetDestroy(&offset);	×
122	int32_t vgId = TD_VID(pTq->pVnode);	×
123	code = getDataBlock(task, pHandle, vgId, &pDataBlock);	×
124	TSDB_CHECK_CODE(code, lino, END);	×
125	}
126
127	const STqExecHandle* pExec = &pHandle->execHandle;	×
128	code = tqAddBlockDataToRsp(pHandle->block, pRsp, pExec->numOfCols, pTq->pVnode->config.tsdbCfg.precision);	×
129	TSDB_CHECK_CODE(code, lino, END);	×
130
131	pRsp->blockNum++;	×
132	if (pDataBlock == NULL) {	×
133	blockDataDestroy(pHandle->block);	×
134	pHandle->block = NULL;	×
135	} else {
136	code = copyDataBlock(pHandle->block, pDataBlock);	×
137	TSDB_CHECK_CODE(code, lino, END);	×
138
139	STqOffsetVal offset = {0};	×
140	code = qStreamExtractOffset(task, &offset);	×
141	TSDB_CHECK_CODE(code, lino, END);	×
142
143	pRsp->sleepTime = offset.ts - pHandle->blockTime;	×
144	pHandle->blockTime = offset.ts;	×
145	tOffsetDestroy(&offset);	×
146	}
147
148	END:	×
149	if (code != TSDB_CODE_SUCCESS) {	×
150	tqError("%s failed at %d, failed to process replay response:%s", __FUNCTION__, lino, tstrerror(code));	×
151	}
152	return code;	×
153	}
154
155	int32_t tqScanData(STQ* pTq, STqHandle* pHandle, SMqDataRsp* pRsp, STqOffsetVal* pOffset, const SMqPollReq* pRequest) {	×
156	int32_t code = 0;	×
157	int32_t lino = 0;	×
158	TSDB_CHECK_NULL(pRsp, code, lino, END, TSDB_CODE_INVALID_PARA);	×
159	TSDB_CHECK_NULL(pTq, code, lino, END, TSDB_CODE_INVALID_PARA);	×
160	TSDB_CHECK_NULL(pHandle, code, lino, END, TSDB_CODE_INVALID_PARA);	×
161	TSDB_CHECK_NULL(pOffset, code, lino, END, TSDB_CODE_INVALID_PARA);	×
162	TSDB_CHECK_NULL(pRequest, code, lino, END, TSDB_CODE_INVALID_PARA);	×
163
164	int32_t vgId = TD_VID(pTq->pVnode);	×
165	int32_t totalRows = 0;	×
166
167	const STqExecHandle* pExec = &pHandle->execHandle;	×
168	qTaskInfo_t task = pExec->task;	×
169
170	code = qStreamPrepareScan(task, pOffset, pHandle->execHandle.subType);	×
171	TSDB_CHECK_CODE(code, lino, END);	×
172
173	qStreamSetSourceExcluded(task, pRequest->sourceExcluded);	×
174	int64_t st = taosGetTimestampMs();	×
175	while (1) {	×
176	SSDataBlock* pDataBlock = NULL;	×
177	code = getDataBlock(task, pHandle, vgId, &pDataBlock);	×
178	TSDB_CHECK_CODE(code, lino, END);	×
179
180	if (pRequest->enableReplay) {	×
181	code = tqProcessReplayRsp(pTq, pHandle, pRsp, pRequest, pDataBlock, task);	×
182	TSDB_CHECK_CODE(code, lino, END);	×
183	break;	×
184	}
185	if (pDataBlock == NULL) {	×
186	break;	×
187	}
188	code = tqAddBlockDataToRsp(pDataBlock, pRsp, pExec->numOfCols, pTq->pVnode->config.tsdbCfg.precision);	×
189	TSDB_CHECK_CODE(code, lino, END);	×
190
191	pRsp->blockNum++;	×
192	totalRows += pDataBlock->info.rows;	×
193	if (totalRows >= tmqRowSize \|\| (taosGetTimestampMs() - st > TMIN(TQ_POLL_MAX_TIME, pRequest->timeout))) {	×
194	break;
195	}
196	}
197
198	tqDebug("consumer:0x%" PRIx64 " vgId:%d tmq task executed finished, total blocks:%d, totalRows:%d", pHandle->consumerId, vgId, pRsp->blockNum, totalRows);	×
199	code = qStreamExtractOffset(task, &pRsp->rspOffset);	×
200
201	END:	×
202	if (code != 0) {	×
203	tqError("%s failed at %d, tmq task executed error msg:%s", __FUNCTION__, lino, tstrerror(code));	×
204	}
205	return code;	×
206	}
207
208	int32_t tqScanTaosx(STQ* pTq, const STqHandle* pHandle, SMqDataRsp* pRsp, SMqBatchMetaRsp* pBatchMetaRsp, STqOffsetVal* pOffset, int64_t timeout) {	×
209	int32_t code = 0;	×
210	int32_t lino = 0;	×
211	char* tbName = NULL;	×
212	SSchemaWrapper* pSW = NULL;	×
213	const STqExecHandle* pExec = &pHandle->execHandle;	×
214	qTaskInfo_t task = pExec->task;	×
215	code = qStreamPrepareScan(task, pOffset, pHandle->execHandle.subType);	×
216	TSDB_CHECK_CODE(code, lino, END);	×
217
218	int32_t rowCnt = 0;	×
219	int64_t st = taosGetTimestampMs();	×
220	while (1) {	×
221	SSDataBlock* pDataBlock = NULL;	×
222	uint64_t ts = 0;	×
223	tqDebug("tmqsnap task start to execute");	×
224	code = qExecTask(task, &pDataBlock, &ts);	×
225	TSDB_CHECK_CODE(code, lino, END);	×
226	tqDebug("tmqsnap task execute end, get %p", pDataBlock);	×
227
228	if (pDataBlock != NULL && pDataBlock->info.rows > 0) {	×
229	if (pRsp->withTbName) {	×
230	tbName = taosStrdup(qExtractTbnameFromTask(task));	×
231	TSDB_CHECK_NULL(tbName, code, lino, END, terrno);	×
232	TSDB_CHECK_NULL(taosArrayPush(pRsp->blockTbName, &tbName), code, lino, END, terrno);	×
233	tqDebug("vgId:%d, add tbname:%s to rsp msg", pTq->pVnode->config.vgId, tbName);	×
234	tbName = NULL;	×
235	}
236	if (pRsp->withSchema) {	×
237	SSchemaWrapper* pSW = tCloneSSchemaWrapper(qExtractSchemaFromTask(task));	×
238	TSDB_CHECK_NULL(pSW, code, lino, END, terrno);	×
239	TSDB_CHECK_NULL(taosArrayPush(pRsp->blockSchema, &pSW), code, lino, END, terrno);	×
240	pSW = NULL;	×
241	}
242
243	code = tqAddBlockDataToRsp(pDataBlock, pRsp, taosArrayGetSize(pDataBlock->pDataBlock),	×
244	pTq->pVnode->config.tsdbCfg.precision);	×
245	TSDB_CHECK_CODE(code, lino, END);	×
246
247	pRsp->blockNum++;	×
248	rowCnt += pDataBlock->info.rows;	×
249	if (rowCnt <= tmqRowSize && (taosGetTimestampMs() - st <= TMIN(TQ_POLL_MAX_TIME, timeout))) {	×
250	continue;	×
251	}
252	}
253
254	// get meta
255	SMqBatchMetaRsp* tmp = qStreamExtractMetaMsg(task);	×
256	if (taosArrayGetSize(tmp->batchMetaReq) > 0) {	×
257	code = qStreamExtractOffset(task, &tmp->rspOffset);	×
258	TSDB_CHECK_CODE(code, lino, END);	×
259	pBatchMetaRsp = tmp;	×
260	tqDebug("tmqsnap task get meta");	×
261	break;	×
262	}
263
264	if (pDataBlock == NULL) {	×
265	code = qStreamExtractOffset(task, pOffset);	×
266	TSDB_CHECK_CODE(code, lino, END);	×
267
268	if (pOffset->type == TMQ_OFFSET__SNAPSHOT_DATA) {	×
269	continue;	×
270	}
271
272	tqDebug("tmqsnap vgId: %d, tsdb consume over, switch to wal, ver %" PRId64, TD_VID(pTq->pVnode), pHandle->snapshotVer + 1);	×
273	code = qStreamExtractOffset(task, &pRsp->rspOffset);	×
274	break;	×
275	}
276
277	if (pRsp->blockNum > 0) {	×
278	tqDebug("tmqsnap task exec exited, get data");	×
279	code = qStreamExtractOffset(task, &pRsp->rspOffset);	×
280	break;	×
281	}
282	}
283	tqDebug("%s:%d success", __FUNCTION__, lino);	×
284	END:	×
285	if (code != 0){	×
286	tqError("%s failed at %d, vgId:%d, task exec error since %s", __FUNCTION__ , lino, pTq->pVnode->config.vgId, tstrerror(code));	×
287	}
288	taosMemoryFree(pSW);	×
289	taosMemoryFree(tbName);	×
290	return code;	×
291	}
292
293	static int32_t buildCreateTbInfo(SMqDataRsp* pRsp, SVCreateTbReq* pCreateTbReq){	×
294	int32_t code = 0;	×
295	int32_t lino = 0;	×
296	void* createReq = NULL;	×
297	TSDB_CHECK_NULL(pRsp, code, lino, END, TSDB_CODE_INVALID_PARA);	×
298	TSDB_CHECK_NULL(pCreateTbReq, code, lino, END, TSDB_CODE_INVALID_PARA);	×
299
300	if (pRsp->createTableNum == 0) {	×
301	pRsp->createTableLen = taosArrayInit(0, sizeof(int32_t));	×
302	TSDB_CHECK_NULL(pRsp->createTableLen, code, lino, END, terrno);	×
303	pRsp->createTableReq = taosArrayInit(0, sizeof(void*));	×
304	TSDB_CHECK_NULL(pRsp->createTableReq, code, lino, END, terrno);	×
305	}
306
307	uint32_t len = 0;	×
308	tEncodeSize(tEncodeSVCreateTbReq, pCreateTbReq, len, code);	×
309	TSDB_CHECK_CODE(code, lino, END);	×
310	createReq = taosMemoryCalloc(1, len);	×
311	TSDB_CHECK_NULL(createReq, code, lino, END, terrno);	×
312
313	SEncoder encoder = {0};	×
314	tEncoderInit(&encoder, createReq, len);	×
315	code = tEncodeSVCreateTbReq(&encoder, pCreateTbReq);	×
316	tEncoderClear(&encoder);	×
317	TSDB_CHECK_CODE(code, lino, END);	×
318	TSDB_CHECK_NULL(taosArrayPush(pRsp->createTableLen, &len), code, lino, END, terrno);	×
319	TSDB_CHECK_NULL(taosArrayPush(pRsp->createTableReq, &createReq), code, lino, END, terrno);	×
320	pRsp->createTableNum++;	×
321	tqDebug("build create table info msg success");	×
322
323	END:	×
324	if (code != 0){	×
325	tqError("%s failed at %d, failed to build create table info msg:%s", __FUNCTION__, lino, tstrerror(code));	×
326	taosMemoryFree(createReq);	×
327	}
328	return code;	×
329	}
330
331	static void tqProcessSubData(STQ* pTq, STqHandle* pHandle, SMqDataRsp* pRsp, int32_t* totalRows, int8_t sourceExcluded){	×
332	int32_t code = 0;	×
333	int32_t lino = 0;	×
334	SArray* pBlocks = NULL;	×
335	SArray* pSchemas = NULL;	×
336
337	STqExecHandle* pExec = &pHandle->execHandle;	×
338	STqReader* pReader = pExec->pTqReader;	×
339
340	pBlocks = taosArrayInit(0, sizeof(SSDataBlock));	×
341	TSDB_CHECK_NULL(pBlocks, code, lino, END, terrno);	×
342	pSchemas = taosArrayInit(0, sizeof(void*));	×
343	TSDB_CHECK_NULL(pSchemas, code, lino, END, terrno);	×
344
345	SSubmitTbData* pSubmitTbDataRet = NULL;	×
346	int64_t createTime = INT64_MAX;	×
347	code = tqRetrieveTaosxBlock(pReader, pBlocks, pSchemas, &pSubmitTbDataRet, &createTime);	×
348	TSDB_CHECK_CODE(code, lino, END);	×
349	bool tmp = (pSubmitTbDataRet->flags & sourceExcluded) != 0;	×
350	TSDB_CHECK_CONDITION(!tmp, code, lino, END, TSDB_CODE_SUCCESS);	×
351	if (pRsp->withTbName) {	×
352	int64_t uid = pExec->pTqReader->lastBlkUid;	×
353	code = tqAddTbNameToRsp(pTq, uid, pRsp, taosArrayGetSize(pBlocks));	×
354	TSDB_CHECK_CODE(code, lino, END);	×
355	}
356	if (pHandle->fetchMeta != WITH_DATA && pSubmitTbDataRet->pCreateTbReq != NULL) {	×
357	if (pSubmitTbDataRet->ctimeMs - createTime <= 1000) { // judge if table is already created to avoid sending crateTbReq	×
358	code = buildCreateTbInfo(pRsp, pSubmitTbDataRet->pCreateTbReq);	×
359	TSDB_CHECK_CODE(code, lino, END);	×
360	}
361	}
362	tmp = (pHandle->fetchMeta == ONLY_META && pSubmitTbDataRet->pCreateTbReq == NULL);	×
363	TSDB_CHECK_CONDITION(!tmp, code, lino, END, TSDB_CODE_SUCCESS);	×
364	for (int32_t i = 0; i < taosArrayGetSize(pBlocks); i++) {	×
365	SSDataBlock* pBlock = taosArrayGet(pBlocks, i);	×
366	if (pBlock == NULL) {	×
367	continue;	×
368	}
369	if (tqAddBlockDataToRsp(pBlock, pRsp, taosArrayGetSize(pBlock->pDataBlock), pTq->pVnode->config.tsdbCfg.precision) != 0){	×
370	tqError("vgId:%d, failed to add block to rsp msg", pTq->pVnode->config.vgId);	×
371	continue;	×
372	}
373	*totalRows += pBlock->info.rows;	×
374	blockDataFreeRes(pBlock);	×
375	SSchemaWrapper* pSW = taosArrayGetP(pSchemas, i);	×
376	if (taosArrayPush(pRsp->blockSchema, &pSW) == NULL){	×
377	tqError("vgId:%d, failed to add schema to rsp msg", pTq->pVnode->config.vgId);	×
378	continue;	×
379	}
380	pRsp->blockNum++;	×
381	}
382	tqDebug("vgId:%d, process sub data success, response blocknum:%d, rows:%d", pTq->pVnode->config.vgId, pRsp->blockNum, *totalRows);	×
383	END:	×
384	if (code != 0){	×
385	tqError("%s failed at %d, failed to process sub data:%s", __FUNCTION__, lino, tstrerror(code));	×
386	taosArrayDestroyEx(pBlocks, (FDelete)blockDataFreeRes);	×
387	taosArrayDestroyP(pSchemas, (FDelete)tDeleteSchemaWrapper);	×
388	} else {
389	taosArrayDestroy(pBlocks);	×
390	taosArrayDestroy(pSchemas);	×
391	}
392	}	×
393
394	int32_t tqTaosxScanLog(STQ* pTq, STqHandle* pHandle, SPackedData submit, SMqDataRsp* pRsp, int32_t* totalRows, int8_t sourceExcluded) {	×
395	int32_t code = 0;	×
396	int32_t lino = 0;	×
397	TSDB_CHECK_NULL(pRsp, code, lino, END, TSDB_CODE_INVALID_PARA);	×
398	TSDB_CHECK_NULL(pTq, code, lino, END, TSDB_CODE_INVALID_PARA);	×
399	TSDB_CHECK_NULL(pHandle, code, lino, END, TSDB_CODE_INVALID_PARA);	×
400	TSDB_CHECK_NULL(totalRows, code, lino, END, TSDB_CODE_INVALID_PARA);	×
401	STqExecHandle* pExec = &pHandle->execHandle;	×
402	STqReader* pReader = pExec->pTqReader;	×
403	code = tqReaderSetSubmitMsg(pReader, submit.msgStr, submit.msgLen, submit.ver);	×
404	TSDB_CHECK_CODE(code, lino, END);	×
405
406	if (pExec->subType == TOPIC_SUB_TYPE__TABLE) {	×
407	while (tqNextBlockImpl(pReader, NULL)) {	×
408	tqProcessSubData(pTq, pHandle, pRsp, totalRows, sourceExcluded);	×
409	}
410	} else if (pExec->subType == TOPIC_SUB_TYPE__DB) {	×
411	while (tqNextDataBlockFilterOut(pReader, pExec->execDb.pFilterOutTbUid)) {	×
412	tqProcessSubData(pTq, pHandle, pRsp, totalRows, sourceExcluded);	×
413	}
414	}
415
416	END:	×
417	if (code != 0){	×
418	tqError("%s failed at %d, failed to scan log:%s", __FUNCTION__, lino, tstrerror(code));	×
419	}
420	return code;	×
421	}

taosdata / TDengine / #3599

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