• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

taosdata / TDengine / #4740

18 Sep 2025 04:31AM UTC coverage: 58.139% (-0.9%) from 59.028%
#4740

push

travis-ci

web-flow
fix: clear parse csv error syntax error msg (#33000)

133663 of 293099 branches covered (45.6%)

Branch coverage included in aggregate %.

0 of 2 new or added lines in 1 file covered. (0.0%)

4143 existing lines in 175 files now uncovered.

202241 of 284660 relevant lines covered (71.05%)

5584206.0 hits per line

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

71.64
/source/dnode/vnode/src/tq/tqUtil.c
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 tqSendMetaPollRsp(STqHandle* pHandle, const SRpcMsg* pMsg, const SMqPollReq* pReq,
19
                                 const SMqMetaRsp* pRsp, int32_t vgId);
20
static int32_t tqSendBatchMetaPollRsp(STqHandle* pHandle, const SRpcMsg* pMsg, const SMqPollReq* pReq,
21
                                      const SMqBatchMetaRsp* pRsp, int32_t vgId);
22

23
int32_t tqInitDataRsp(SMqDataRsp* pRsp, STqOffsetVal pOffset) {
21,682✔
24
  int32_t    code = TDB_CODE_SUCCESS;
21,682✔
25
  int32_t    lino = 0;
21,682✔
26
  tqDebug("%s called", __FUNCTION__ );
21,682!
27
  TSDB_CHECK_NULL(pRsp, code, lino, END, TSDB_CODE_INVALID_PARA);
21,682!
28

29
  pRsp->blockData = taosArrayInit(0, sizeof(void*));
21,682✔
30
  TSDB_CHECK_NULL(pRsp->blockData, code, lino, END, terrno);
21,682!
31

32
  pRsp->blockDataLen = taosArrayInit(0, sizeof(int32_t));
21,682✔
33
  TSDB_CHECK_NULL(pRsp->blockDataLen, code, lino, END, terrno);
21,682!
34

35
  tOffsetCopy(&pRsp->reqOffset, &pOffset);
21,682✔
36
  tOffsetCopy(&pRsp->rspOffset, &pOffset);
21,682✔
37
  pRsp->withTbName = 0;
21,682✔
38
  pRsp->withSchema = false;
21,682✔
39

40
END:
21,682✔
41
  if (code != 0){
21,682!
42
    tqError("%s failed at:%d, code:%s", __FUNCTION__ , lino, tstrerror(code));
×
43
  }
44
  return code;
21,682✔
45
}
46

47
static int32_t tqInitTaosxRsp(SMqDataRsp* pRsp, STqOffsetVal pOffset) {
6,842✔
48
  int32_t    code = TDB_CODE_SUCCESS;
6,842✔
49
  int32_t    lino = 0;
6,842✔
50
  tqDebug("%s called", __FUNCTION__ );
6,842!
51
  TSDB_CHECK_NULL(pRsp, code, lino, END, TSDB_CODE_INVALID_PARA);
6,844!
52
  tOffsetCopy(&pRsp->reqOffset, &pOffset);
6,844✔
53
  tOffsetCopy(&pRsp->rspOffset, &pOffset);
6,844✔
54

55
  pRsp->withTbName = 1;
6,844✔
56
  pRsp->withSchema = 1;
6,844✔
57
  pRsp->blockData = taosArrayInit(0, sizeof(void*));
6,844✔
58
  TSDB_CHECK_NULL(pRsp->blockData, code, lino, END, terrno);\
6,844!
59

60
  pRsp->blockDataLen = taosArrayInit(0, sizeof(int32_t));
6,844✔
61
  TSDB_CHECK_NULL(pRsp->blockDataLen, code, lino, END, terrno);
6,843!
62

63
  pRsp->blockTbName = taosArrayInit(0, sizeof(void*));
6,843✔
64
  TSDB_CHECK_NULL(pRsp->blockTbName, code, lino, END, terrno);
6,844!
65

66
  pRsp->blockSchema = taosArrayInit(0, sizeof(void*));
6,844✔
67
  TSDB_CHECK_NULL(pRsp->blockSchema, code, lino, END, terrno);
6,843!
68

69
END:
6,843✔
70
  if (code != 0){
6,843!
71
    tqError("%s failed at:%d, code:%s", __FUNCTION__ , lino, tstrerror(code));
×
72
    taosArrayDestroy(pRsp->blockData);
×
73
    taosArrayDestroy(pRsp->blockDataLen);
×
74
    taosArrayDestroy(pRsp->blockTbName);
×
75
    taosArrayDestroy(pRsp->blockSchema);
×
76
  }
77
  return code;
6,844✔
78
}
79

80
static int32_t extractResetOffsetVal(STqOffsetVal* pOffsetVal, STQ* pTq, STqHandle* pHandle, const SMqPollReq* pRequest,
2,289✔
81
                                     SRpcMsg* pMsg, bool* pBlockReturned) {
82
  if (pOffsetVal == NULL || pTq == NULL || pHandle == NULL || pRequest == NULL || pMsg == NULL || pBlockReturned == NULL) {
2,289!
83
    return TSDB_CODE_INVALID_PARA;
×
84
  }
85
  uint64_t   consumerId = pRequest->consumerId;
2,293✔
86
  STqOffset* pOffset = NULL;
2,293✔
87
  int32_t    code = tqMetaGetOffset(pTq, pRequest->subKey, &pOffset);
2,293✔
88
  int32_t    vgId = TD_VID(pTq->pVnode);
2,293✔
89

90
  *pBlockReturned = false;
2,293✔
91
  // In this vnode, data has been polled by consumer for this topic, so let's continue from the last offset value.
92
  if (code == 0) {
2,293✔
93
    tOffsetCopy(pOffsetVal, &pOffset->val);
110✔
94

95
    char formatBuf[TSDB_OFFSET_LEN] = {0};
110✔
96
    tFormatOffset(formatBuf, TSDB_OFFSET_LEN, pOffsetVal);
110✔
97
    tqDebug("tmq poll: consumer:0x%" PRIx64
110!
98
                ", subkey %s, vgId:%d, existed offset found, offset reset to %s and continue.QID:0x%" PRIx64,
99
            consumerId, pHandle->subKey, vgId, formatBuf, pRequest->reqId);
100
    return 0;
110✔
101
  } else {
102
    // no poll occurs in this vnode for this topic, let's seek to the right offset value.
103
    if (pRequest->reqOffset.type == TMQ_OFFSET__RESET_EARLIEST) {
2,183✔
104
      if (pRequest->useSnapshot) {
1,488✔
105
        tqDebug("tmq poll: consumer:0x%" PRIx64 ", subkey:%s, vgId:%d, (earliest) set offset to be snapshot",
832!
106
                consumerId, pHandle->subKey, vgId);
107
        if (pHandle->fetchMeta) {
832✔
108
          tqOffsetResetToMeta(pOffsetVal, 0);
109
        } else {
110
          SValue val = {0};
816!
111
          tqOffsetResetToData(pOffsetVal, 0, 0, val);
816✔
112
        }
113
      } else {
114
        walRefFirstVer(pTq->pVnode->pWal, pHandle->pRef);
656✔
115
        tqOffsetResetToLog(pOffsetVal, pHandle->pRef->refVer);
655✔
116
      }
117
    } else if (pRequest->reqOffset.type == TMQ_OFFSET__RESET_LATEST) {
695✔
118
      walRefLastVer(pTq->pVnode->pWal, pHandle->pRef);
47✔
119
      SMqDataRsp dataRsp = {0};
47✔
120
      tqOffsetResetToLog(pOffsetVal, pHandle->pRef->refVer + 1);
47✔
121

122
      code = tqInitDataRsp(&dataRsp, *pOffsetVal);
47✔
123
      if (code != 0) {
47!
124
        return code;
×
125
      }
126
      tqDebug("tmq poll: consumer:0x%" PRIx64 ", subkey %s, vgId:%d, (latest) offset reset to %" PRId64, consumerId,
47!
127
              pHandle->subKey, vgId, dataRsp.rspOffset.version);
128
      code = tqSendDataRsp(pHandle, pMsg, pRequest, &dataRsp, (pRequest->rawData == 1) ? TMQ_MSG_TYPE__POLL_RAW_DATA_RSP : TMQ_MSG_TYPE__POLL_DATA_RSP, vgId);
47!
129
      tDeleteMqDataRsp(&dataRsp);
47✔
130

131
      *pBlockReturned = true;
47✔
132
      return code;
47✔
133
    } else if (pRequest->reqOffset.type == TMQ_OFFSET__RESET_NONE) {
648!
134
      tqError("tmq poll: subkey:%s, no offset committed for consumer:0x%" PRIx64
648!
135
                  " in vg %d, subkey %s, reset none failed",
136
              pHandle->subKey, consumerId, vgId, pRequest->subKey);
137
      return TSDB_CODE_TQ_NO_COMMITTED_OFFSET;
648✔
138
    }
139
  }
140

141
  return 0;
1,487✔
142
}
143

144
static int32_t extractDataAndRspForNormalSubscribe(STQ* pTq, STqHandle* pHandle, const SMqPollReq* pRequest,
21,138✔
145
                                                   SRpcMsg* pMsg, STqOffsetVal* pOffset) {
146
  int32_t    code = TDB_CODE_SUCCESS;
21,138✔
147
  int32_t    lino = 0;
21,138✔
148
  tqDebug("%s called", __FUNCTION__ );
21,138!
149
  uint64_t consumerId = pRequest->consumerId;
21,139✔
150
  int32_t  vgId = TD_VID(pTq->pVnode);
21,139✔
151
  terrno = 0;
21,139✔
152

153
  SMqDataRsp dataRsp = {0};
21,139✔
154
  code = tqInitDataRsp(&dataRsp, *pOffset);
21,139✔
155
  TSDB_CHECK_CODE(code, lino, end);
21,139!
156

157
  code = qSetTaskId(pHandle->execHandle.task, consumerId, pRequest->reqId);
21,139✔
158
  TSDB_CHECK_CODE(code, lino, end);
21,139!
159

160
  code = tqScanData(pTq, pHandle, &dataRsp, pOffset, pRequest);
21,139✔
161
  if (code != 0 && terrno != TSDB_CODE_WAL_LOG_NOT_EXIST) {
21,139✔
162
    goto end;
676✔
163
  }
164

165
  //   till now, all data has been transferred to consumer, new data needs to push client once arrived.
166
  if (terrno == TSDB_CODE_WAL_LOG_NOT_EXIST && dataRsp.blockNum == 0) {
20,463✔
167
    // lock
168
    taosWLockLatch(&pTq->lock);
8,643✔
169
    int64_t ver = walGetCommittedVer(pTq->pVnode->pWal);
8,643✔
170
    if (dataRsp.rspOffset.version > ver) {  // check if there are data again to avoid lost data
8,643✔
171
      code = tqRegisterPushHandle(pTq, pHandle, pMsg);
8,411✔
172
      taosWUnLockLatch(&pTq->lock);
8,411✔
173
      goto end;
8,411✔
174
    }
175
    taosWUnLockLatch(&pTq->lock);
232✔
176
  }
177

178
  // reqOffset represents the current date offset, may be changed if wal not exists
179
  tOffsetCopy(&dataRsp.reqOffset, pOffset);
12,052✔
180
  code = tqSendDataRsp(pHandle, pMsg, pRequest, &dataRsp, TMQ_MSG_TYPE__POLL_DATA_RSP, vgId);
12,052✔
181

182
end:
21,136✔
183
  {
184
    char buf[TSDB_OFFSET_LEN] = {0};
21,136✔
185
    tFormatOffset(buf, TSDB_OFFSET_LEN, &dataRsp.rspOffset);
21,136✔
186
    if (code != 0){
21,139✔
187
      tqError("tmq poll: consumer:0x%" PRIx64 ", subkey %s, vgId:%d, rsp block:%d, rsp offset type:%s, QID:0x%" PRIx64 " error msg:%s, line:%d",
676!
188
              consumerId, pHandle->subKey, vgId, dataRsp.blockNum, buf, pRequest->reqId, tstrerror(code), lino);
189
    } else {
190
      tqDebug("tmq poll: consumer:0x%" PRIx64 ", subkey %s, vgId:%d, rsp block:%d, rsp offset type:%s, QID:0x%" PRIx64 " success",
20,463✔
191
              consumerId, pHandle->subKey, vgId, dataRsp.blockNum, buf, pRequest->reqId);
192
    }
193

194
    tDeleteMqDataRsp(&dataRsp);
21,142✔
195
    return code;
21,139✔
196
  }
197
}
198

199
#define PROCESS_EXCLUDED_MSG(TYPE, DECODE_FUNC, DELETE_FUNC)                                               \
200
  SDecoder decoder = {0};                                                                                  \
201
  TYPE     req = {0};                                                                                      \
202
  void*    data = POINTER_SHIFT(pHead->body, sizeof(SMsgHead));                                            \
203
  int32_t  len = pHead->bodyLen - sizeof(SMsgHead);                                                        \
204
  tDecoderInit(&decoder, data, len);                                                                       \
205
  if (DECODE_FUNC(&decoder, &req) == 0 && (req.source & TD_REQ_FROM_TAOX) != 0) {                          \
206
    tqDebug("tmq poll: consumer:0x%" PRIx64 " (epoch %d) iter log, jump meta for, vgId:%d offset %" PRId64 \
207
            " msgType %d",                                                                                 \
208
            pRequest->consumerId, pRequest->epoch, vgId, fetchVer, pHead->msgType);                        \
209
    fetchVer++;                                                                                            \
210
    DELETE_FUNC(&req);                                                                                     \
211
    tDecoderClear(&decoder);                                                                               \
212
    continue;                                                                                              \
213
  }                                                                                                        \
214
  DELETE_FUNC(&req);                                                                                       \
215
  tDecoderClear(&decoder);
216

217
static void tDeleteCommon(void* parm) {}
316✔
218

219
#define POLL_RSP_TYPE(pRequest,taosxRsp) \
220
taosxRsp.createTableNum > 0 ? TMQ_MSG_TYPE__POLL_DATA_META_RSP : \
221
(pRequest->rawData == 1 ? TMQ_MSG_TYPE__POLL_RAW_DATA_RSP : TMQ_MSG_TYPE__POLL_DATA_RSP)
222

223
static int32_t buildBatchMeta(SMqBatchMetaRsp *btMetaRsp, int16_t type, int32_t bodyLen, void* body){
3✔
224
  int32_t         code = 0;
3✔
225

226
  if (!btMetaRsp->batchMetaReq) {
3!
227
    btMetaRsp->batchMetaReq = taosArrayInit(4, POINTER_BYTES);
3✔
228
    TQ_NULL_GO_TO_END(btMetaRsp->batchMetaReq);
3!
229
    btMetaRsp->batchMetaLen = taosArrayInit(4, sizeof(int32_t));
3✔
230
    TQ_NULL_GO_TO_END(btMetaRsp->batchMetaLen);
3!
231
  }
232

233
  SMqMetaRsp tmpMetaRsp = {0};
3✔
234
  tmpMetaRsp.resMsgType = type;
3✔
235
  tmpMetaRsp.metaRspLen = bodyLen;
3✔
236
  tmpMetaRsp.metaRsp = body;
3✔
237
  uint32_t len = 0;
3✔
238
  tEncodeSize(tEncodeMqMetaRsp, &tmpMetaRsp, len, code);
3!
239
  if (TSDB_CODE_SUCCESS != code) {
3!
240
    tqError("tmq extract meta from log, tEncodeMqMetaRsp error");
×
241
    goto END;
×
242
  }
243
  int32_t tLen = sizeof(SMqRspHead) + len;
3✔
244
  void*   tBuf = taosMemoryCalloc(1, tLen);
3!
245
  TQ_NULL_GO_TO_END(tBuf);
3!
246
  void*    metaBuff = POINTER_SHIFT(tBuf, sizeof(SMqRspHead));
3✔
247
  SEncoder encoder = {0};
3✔
248
  tEncoderInit(&encoder, metaBuff, len);
3✔
249
  code = tEncodeMqMetaRsp(&encoder, &tmpMetaRsp);
3✔
250
  tEncoderClear(&encoder);
3✔
251

252
  if (code < 0) {
3!
253
    tqError("tmq extract meta from log, tEncodeMqMetaRsp error");
×
254
    goto END;
×
255
  }
256
  TQ_NULL_GO_TO_END (taosArrayPush(btMetaRsp->batchMetaReq, &tBuf));
6!
257
  TQ_NULL_GO_TO_END (taosArrayPush(btMetaRsp->batchMetaLen, &tLen));
6!
258

259
END:
3✔
260
  return code;
3✔
261
}
262

263
static int32_t buildCreateTbBatchReqBinary(SMqDataRsp *taosxRsp, void** pBuf, int32_t *len){
1✔
264
  int32_t code = 0;
1✔
265
  SVCreateTbBatchReq pReq = {0};
1✔
266
  pReq.nReqs = taosArrayGetSize(taosxRsp->createTableReq);
1✔
267
  pReq.pArray = taosArrayInit(1, sizeof(struct SVCreateTbReq));
1✔
268
  TQ_NULL_GO_TO_END(pReq.pArray);
1!
269
  for (int i = 0; i < taosArrayGetSize(taosxRsp->createTableReq); i++){
5✔
270
    void   *createTableReq = taosArrayGetP(taosxRsp->createTableReq, i);
4✔
271
    TQ_NULL_GO_TO_END(taosArrayPush(pReq.pArray, createTableReq));
8!
272
  }
273
  tEncodeSize(tEncodeSVCreateTbBatchReq, &pReq, *len, code);
1!
274
  if (code < 0) {
1!
275
    goto END;
×
276
  }
277
  *len += sizeof(SMsgHead);
1✔
278
  *pBuf = taosMemoryMalloc(*len);
1!
279
  TQ_NULL_GO_TO_END(pBuf);
1!
280
  SEncoder coder = {0};
1✔
281
  tEncoderInit(&coder, POINTER_SHIFT(*pBuf, sizeof(SMsgHead)), *len);
1✔
282
  code = tEncodeSVCreateTbBatchReq(&coder, &pReq);
1✔
283
  tEncoderClear(&coder);
1✔
284

285
END:
1✔
286
  taosArrayDestroy(pReq.pArray);
1✔
287
  return code;
1✔
288
}
289

290
#define SEND_BATCH_META_RSP \
291
tqOffsetResetToLog(&btMetaRsp.rspOffset, fetchVer);\
292
code = tqSendBatchMetaPollRsp(pHandle, pMsg, pRequest, &btMetaRsp, vgId);\
293
goto END;
294

295
#define SEND_DATA_RSP \
296
tqOffsetResetToLog(&taosxRsp.rspOffset, fetchVer);\
297
code = tqSendDataRsp(pHandle, pMsg, pRequest, &taosxRsp, POLL_RSP_TYPE(pRequest, taosxRsp), vgId);\
298
goto END;
299
static int32_t extractDataAndRspForDbStbSubscribe(STQ* pTq, STqHandle* pHandle, const SMqPollReq* pRequest,
6,843✔
300
                                                  SRpcMsg* pMsg, STqOffsetVal* offset) {
301
  int32_t         vgId = TD_VID(pTq->pVnode);
6,843✔
302
  SMqDataRsp      taosxRsp = {0};
6,843✔
303
  SMqBatchMetaRsp btMetaRsp = {0};
6,843✔
304
  int32_t         code = 0;
6,843✔
305

306
  TQ_ERR_GO_TO_END(tqInitTaosxRsp(&taosxRsp, *offset));
6,843!
307
  if (offset->type != TMQ_OFFSET__LOG) {
6,843✔
308
    TQ_ERR_GO_TO_END(tqScanTaosx(pTq, pHandle, &taosxRsp, &btMetaRsp, offset, pRequest));
151!
309

310
    if (taosArrayGetSize(btMetaRsp.batchMetaReq) > 0) {
151✔
311
      code = tqSendBatchMetaPollRsp(pHandle, pMsg, pRequest, &btMetaRsp, vgId);
14✔
312
      tqDebug("tmq poll: consumer:0x%" PRIx64 " subkey:%s vgId:%d, send meta offset type:%d,uid:%" PRId64 ",ts:%" PRId64,
14!
313
              pRequest->consumerId, pHandle->subKey, vgId, btMetaRsp.rspOffset.type, btMetaRsp.rspOffset.uid,btMetaRsp.rspOffset.ts);
314
      goto END;
14✔
315
    }
316

317
    tqDebug("taosx poll: consumer:0x%" PRIx64 " subkey:%s vgId:%d, send data blockNum:%d, offset type:%d,uid:%" PRId64",ts:%" PRId64,
137!
318
            pRequest->consumerId, pHandle->subKey, vgId, taosxRsp.blockNum, taosxRsp.rspOffset.type, taosxRsp.rspOffset.uid, taosxRsp.rspOffset.ts);
319
    if (taosxRsp.blockNum > 0) {
137✔
320
      code = tqSendDataRsp(pHandle, pMsg, pRequest, &taosxRsp, TMQ_MSG_TYPE__POLL_DATA_RSP, vgId);
133✔
321
      goto END;
133✔
322
    } else {
323
      tOffsetCopy(offset, &taosxRsp.rspOffset);
4✔
324
    }
325
  }
326

327
  if (offset->type == TMQ_OFFSET__LOG) {
6,697!
328
    walReaderVerifyOffset(pHandle->pWalReader, offset);
6,697✔
329
    int64_t fetchVer = offset->version;
6,697✔
330

331
    uint64_t st = taosGetTimestampMs();
6,696✔
332
    int      totalRows = 0;
6,696✔
333
    int32_t  totalMetaRows = 0;
6,696✔
334
    while (1) {
81,095✔
335
      int32_t savedEpoch = atomic_load_32(&pHandle->epoch);
87,791✔
336
      if (savedEpoch > pRequest->epoch) {
87,714!
337
        tqError("tmq poll: consumer:0x%" PRIx64 " (epoch %d) iter log, savedEpoch error, vgId:%d offset %" PRId64,
×
338
                pRequest->consumerId, pRequest->epoch, vgId, fetchVer);
339
        code = TSDB_CODE_TQ_INTERNAL_ERROR;
×
340
        goto END;
6,697✔
341
      }
342

343
      if (tqFetchLog(pTq, pHandle, &fetchVer, pRequest->reqId) < 0) {
87,714✔
344
        if (totalMetaRows > 0) {
6,178✔
345
          SEND_BATCH_META_RSP
2✔
346
        }
347
        SEND_DATA_RSP
6,176!
348
      }
349

350
      SWalCont* pHead = &pHandle->pWalReader->pHead->head;
81,801✔
351
      tqDebug("tmq poll: consumer:0x%" PRIx64 " (epoch %d) iter log, vgId:%d offset %" PRId64 " msgType %s",
81,801!
352
              pRequest->consumerId, pRequest->epoch, vgId, fetchVer, TMSG_INFO(pHead->msgType));
353

354
      // process meta
355
      if (pHead->msgType != TDMT_VND_SUBMIT) {
81,808✔
356
        if (totalRows > 0) {
547✔
357
          SEND_DATA_RSP
24!
358
        }
359

360
        if ((pRequest->sourceExcluded & TD_REQ_FROM_TAOX) != 0) {
523✔
361
          if (pHead->msgType == TDMT_VND_CREATE_TABLE) {
519✔
362
            PROCESS_EXCLUDED_MSG(SVCreateTbBatchReq, tDecodeSVCreateTbBatchReq, tDeleteSVCreateTbBatchReq)
165!
363
          } else if (pHead->msgType == TDMT_VND_ALTER_TABLE) {
354✔
364
            PROCESS_EXCLUDED_MSG(SVAlterTbReq, tDecodeSVAlterTbReq, tDeleteCommon)
44!
365
          } else if (pHead->msgType == TDMT_VND_CREATE_STB || pHead->msgType == TDMT_VND_ALTER_STB) {
389✔
366
            PROCESS_EXCLUDED_MSG(SVCreateStbReq, tDecodeSVCreateStbReq, tDeleteCommon)
267!
367
          } else if (pHead->msgType == TDMT_VND_DELETE) {
43✔
368
            PROCESS_EXCLUDED_MSG(SDeleteRes, tDecodeDeleteRes, tDeleteCommon)
5!
369
          }
370
        }
371

372
        tqDebug("fetch meta msg, ver:%" PRId64 ", vgId:%d, type:%s, enable batch meta:%d", pHead->version, vgId,
186!
373
                TMSG_INFO(pHead->msgType), pRequest->enableBatchMeta);
374
        if (!pRequest->enableBatchMeta && !pRequest->useSnapshot) {
186!
375
          SMqMetaRsp metaRsp = {0};
184✔
376
          tqOffsetResetToLog(&metaRsp.rspOffset, fetchVer + 1);
184✔
377
          metaRsp.resMsgType = pHead->msgType;
184✔
378
          metaRsp.metaRspLen = pHead->bodyLen;
184✔
379
          metaRsp.metaRsp = pHead->body;
184✔
380
          code = tqSendMetaPollRsp(pHandle, pMsg, pRequest, &metaRsp, vgId);
184✔
381
          goto END;
184✔
382
        }
383
        code = buildBatchMeta(&btMetaRsp, pHead->msgType, pHead->bodyLen, pHead->body);
2✔
384
        fetchVer++;
2✔
385
        if (code != 0){
2!
386
          goto END;
×
387
        }
388
        totalMetaRows++;
2✔
389
        if ((taosArrayGetSize(btMetaRsp.batchMetaReq) >= tmqRowSize) || (taosGetTimestampMs() - st > pRequest->timeout)) {
4!
390
          SEND_BATCH_META_RSP
×
391
        }
392
        continue;
2✔
393
      }
394

395
      if (totalMetaRows > 0 && pHandle->fetchMeta != ONLY_META) {
81,261!
396
        SEND_BATCH_META_RSP
×
397
      }
398

399
      // process data
400
      SPackedData submit = {
81,261✔
401
          .msgStr = POINTER_SHIFT(pHead->body, sizeof(SSubmitReq2Msg)),
81,261✔
402
          .msgLen = pHead->bodyLen - sizeof(SSubmitReq2Msg),
81,261✔
403
          .ver = pHead->version,
81,261✔
404
      };
405

406
      TQ_ERR_GO_TO_END(tqTaosxScanLog(pTq, pHandle, submit, &taosxRsp, &totalRows, pRequest));
81,261!
407

408
      if (pHandle->fetchMeta == ONLY_META && taosArrayGetSize(taosxRsp.createTableReq) > 0){
81,073✔
409
        int32_t len = 0;
1✔
410
        void *pBuf = NULL;
1✔
411
        code = buildCreateTbBatchReqBinary(&taosxRsp, &pBuf, &len);
1✔
412
        if (code == 0){
1!
413
          code = buildBatchMeta(&btMetaRsp, TDMT_VND_CREATE_TABLE, len, pBuf);
1✔
414
        }
415
        taosMemoryFree(pBuf);
1!
416
        for (int i = 0; i < taosArrayGetSize(taosxRsp.createTableReq); i++) {
5✔
417
          void* pCreateTbReq = taosArrayGetP(taosxRsp.createTableReq, i);
4✔
418
          if (pCreateTbReq != NULL) {
4!
419
            tDestroySVSubmitCreateTbReq(pCreateTbReq, TSDB_MSG_FLG_DECODE);
4✔
420
          }
421
          taosMemoryFree(pCreateTbReq);
4!
422
        }
423
        taosArrayDestroy(taosxRsp.createTableReq);
1✔
424
        taosxRsp.createTableReq = NULL;
1✔
425
        fetchVer++;
1✔
426
        if (code != 0){
1!
427
          goto END;
1✔
428
        }
429
        totalMetaRows++;
1✔
430
        if ((taosArrayGetSize(btMetaRsp.batchMetaReq) >= tmqRowSize) ||
1!
431
            (taosGetTimestampMs() - st > pRequest->timeout) ||
1!
432
            (!pRequest->enableBatchMeta && !pRequest->useSnapshot)) {
1!
433
          SEND_BATCH_META_RSP
1✔
434
        }
435
        continue;
×
436
      }
437

438
      if ((pRequest->rawData == 0 && totalRows >= pRequest->minPollRows) ||
81,072✔
439
          (taosGetTimestampMs() - st > pRequest->timeout) ||
80,830!
440
          (pRequest->rawData != 0 && (taosArrayGetSize(taosxRsp.blockData) > pRequest->minPollRows ||
80,880!
441
                                      terrno == TSDB_CODE_TMQ_RAW_DATA_SPLIT))) {
×
442
        if (terrno == TSDB_CODE_TMQ_RAW_DATA_SPLIT){
381!
UNCOV
443
          terrno = 0;
×
444
        } else{
445
          fetchVer++;
309✔
446
        }
447
        SEND_DATA_RSP
309!
448
      } else {
449
        fetchVer++;
80,756✔
450
      }
451
    }
452
  }
453

454
END:
×
455
  if (code != 0){
6,844!
456
    tqError("tmq poll: tqTaosxScanLog error. consumerId:0x%" PRIx64 ", in vgId:%d, subkey %s", pRequest->consumerId, vgId,
×
457
            pRequest->subKey);
458
  }
459
  tDeleteMqBatchMetaRsp(&btMetaRsp);
6,844✔
460
  tDeleteSTaosxRsp(&taosxRsp);
6,841✔
461
  return code;
6,843✔
462
}
463

464
int32_t tqExtractDataForMq(STQ* pTq, STqHandle* pHandle, const SMqPollReq* pRequest, SRpcMsg* pMsg) {
28,678✔
465
  if (pTq == NULL || pHandle == NULL || pRequest == NULL || pMsg == NULL) {
28,678!
466
    return TSDB_CODE_TMQ_INVALID_MSG;
×
467
  }
468
  int32_t      code = 0;
28,678✔
469
  STqOffsetVal reqOffset = {0};
28,678✔
470
  tOffsetCopy(&reqOffset, &pRequest->reqOffset);
28,678✔
471

472
  // reset the offset if needed
473
  if (IS_OFFSET_RESET_TYPE(pRequest->reqOffset.type)) {
28,678✔
474
    bool blockReturned = false;
2,295✔
475
    code = extractResetOffsetVal(&reqOffset, pTq, pHandle, pRequest, pMsg, &blockReturned);
2,295✔
476
    if (code != 0) {
2,293✔
477
      goto END;
695✔
478
    }
479

480
    // empty block returned, quit
481
    if (blockReturned) {
1,645✔
482
      goto END;
47✔
483
    }
484
  } else if (reqOffset.type == 0) {  // use the consumer specified offset
26,383!
485
    uError("req offset type is 0");
×
486
    code = TSDB_CODE_TMQ_INVALID_MSG;
×
487
    goto END;
×
488
  }
489

490
  if (pHandle->execHandle.subType == TOPIC_SUB_TYPE__COLUMN) {
27,981✔
491
    code = extractDataAndRspForNormalSubscribe(pTq, pHandle, pRequest, pMsg, &reqOffset);
21,138✔
492
  } else {
493
    code = extractDataAndRspForDbStbSubscribe(pTq, pHandle, pRequest, pMsg, &reqOffset);
6,843✔
494
  }
495

496
END:
28,668✔
497
  if (code != 0){
28,668✔
498
    uError("failed to extract data for mq, msg:%s", tstrerror(code));
1,324!
499
  }
500
  tOffsetDestroy(&reqOffset);
28,668✔
501
  return code;
28,676✔
502
}
503

504
static void initMqRspHead(SMqRspHead* pMsgHead, int32_t type, int32_t epoch, int64_t consumerId, int64_t sver,
19,438✔
505
                          int64_t ever) {
506
  if (pMsgHead == NULL) {
19,438!
507
    return;
×
508
  }
509
  pMsgHead->consumerId = consumerId;
19,438✔
510
  pMsgHead->epoch = epoch;
19,438✔
511
  pMsgHead->mqMsgType = type;
19,438✔
512
  pMsgHead->walsver = sver;
19,438✔
513
  pMsgHead->walever = ever;
19,438✔
514
}
515

516
int32_t tqSendBatchMetaPollRsp(STqHandle* pHandle, const SRpcMsg* pMsg, const SMqPollReq* pReq,
17✔
517
                               const SMqBatchMetaRsp* pRsp, int32_t vgId) {
518
  if (pHandle == NULL || pMsg == NULL || pReq == NULL || pRsp == NULL) {
17!
519
    return TSDB_CODE_TMQ_INVALID_MSG;
×
520
  }
521
  int32_t len = 0;
17✔
522
  int32_t code = 0;
17✔
523
  tEncodeSize(tEncodeMqBatchMetaRsp, pRsp, len, code);
17!
524
  if (code < 0) {
17!
525
    return TAOS_GET_TERRNO(code);
×
526
  }
527
  int32_t tlen = sizeof(SMqRspHead) + len;
17✔
528
  void*   buf = rpcMallocCont(tlen);
17✔
529
  if (buf == NULL) {
17!
530
    return TAOS_GET_TERRNO(terrno);
×
531
  }
532

533
  int64_t sver = 0, ever = 0;
17✔
534
  walReaderValidVersionRange(pHandle->execHandle.pTqReader->pWalReader, &sver, &ever);
17✔
535
  initMqRspHead(buf, TMQ_MSG_TYPE__POLL_BATCH_META_RSP, pReq->epoch, pReq->consumerId, sver, ever);
17✔
536

537
  void* abuf = POINTER_SHIFT(buf, sizeof(SMqRspHead));
17✔
538

539
  SEncoder encoder = {0};
17✔
540
  tEncoderInit(&encoder, abuf, len);
17✔
541
  code = tEncodeMqBatchMetaRsp(&encoder, pRsp);
17✔
542
  tEncoderClear(&encoder);
17✔
543
  if (code < 0) {
17!
544
    rpcFreeCont(buf);
×
545
    return TAOS_GET_TERRNO(code);
×
546
  }
547
  SRpcMsg resp = {.info = pMsg->info, .pCont = buf, .contLen = tlen, .code = 0};
17✔
548

549
  tmsgSendRsp(&resp);
17✔
550
  tqDebug("vgId:%d, from consumer:0x%" PRIx64 " (epoch %d) send rsp, res msg type: batch meta, size:%ld offset type:%d",
17!
551
          vgId, pReq->consumerId, pReq->epoch, taosArrayGetSize(pRsp->batchMetaReq), pRsp->rspOffset.type);
552

553
  return 0;
17✔
554
}
555

556
int32_t tqSendMetaPollRsp(STqHandle* pHandle, const SRpcMsg* pMsg, const SMqPollReq* pReq, const SMqMetaRsp* pRsp,
184✔
557
                          int32_t vgId) {
558
  if (pHandle == NULL || pMsg == NULL || pReq == NULL || pRsp == NULL) {
184!
559
    return TSDB_CODE_TMQ_INVALID_MSG;
×
560
  }
561
  int32_t len = 0;
184✔
562
  int32_t code = 0;
184✔
563
  tEncodeSize(tEncodeMqMetaRsp, pRsp, len, code);
184!
564
  if (code < 0) {
184!
565
    return TAOS_GET_TERRNO(code);
×
566
  }
567
  int32_t tlen = sizeof(SMqRspHead) + len;
184✔
568
  void*   buf = rpcMallocCont(tlen);
184✔
569
  if (buf == NULL) {
184!
570
    return TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);
×
571
  }
572

573
  int64_t sver = 0, ever = 0;
184✔
574
  walReaderValidVersionRange(pHandle->execHandle.pTqReader->pWalReader, &sver, &ever);
184✔
575
  initMqRspHead(buf, TMQ_MSG_TYPE__POLL_META_RSP, pReq->epoch, pReq->consumerId, sver, ever);
184✔
576

577
  void* abuf = POINTER_SHIFT(buf, sizeof(SMqRspHead));
184✔
578

579
  SEncoder encoder = {0};
184✔
580
  tEncoderInit(&encoder, abuf, len);
184✔
581
  code = tEncodeMqMetaRsp(&encoder, pRsp);
184✔
582
  tEncoderClear(&encoder);
184✔
583
  if (code < 0) {
184!
584
    rpcFreeCont(buf);
×
585
    return TAOS_GET_TERRNO(code);
×
586
  }
587

588
  SRpcMsg resp = {.info = pMsg->info, .pCont = buf, .contLen = tlen, .code = 0};
184✔
589

590
  tmsgSendRsp(&resp);
184✔
591
  tqDebug("vgId:%d, from consumer:0x%" PRIx64 " (epoch %d) send rsp, res msg type %d, offset type:%d", vgId,
184!
592
          pReq->consumerId, pReq->epoch, pRsp->resMsgType, pRsp->rspOffset.type);
593

594
  return 0;
184✔
595
}
596

597
int32_t tqDoSendDataRsp(const SRpcHandleInfo* pRpcHandleInfo, SMqDataRsp* pRsp, int32_t epoch, int64_t consumerId,
19,238✔
598
                        int32_t type, int64_t sver, int64_t ever) {
599
  if (pRpcHandleInfo == NULL || pRsp == NULL) {
19,238!
600
    return TSDB_CODE_TMQ_INVALID_MSG;
×
601
  }
602
  int32_t len = 0;
19,238✔
603
  int32_t code = 0;
19,238✔
604

605
  if (type == TMQ_MSG_TYPE__POLL_RAW_DATA_RSP){
19,238!
606
    pRsp->withSchema = 0;
×
607
  }
608
  if (type == TMQ_MSG_TYPE__POLL_DATA_RSP ||
19,238✔
609
      type == TMQ_MSG_TYPE__WALINFO_RSP ||
15!
610
      type == TMQ_MSG_TYPE__POLL_RAW_DATA_RSP) {
611
    tEncodeSize(tEncodeMqDataRsp, pRsp, len, code);
19,223!
612
  } else if (type == TMQ_MSG_TYPE__POLL_DATA_META_RSP) {
15!
613
    tEncodeSize(tEncodeSTaosxRsp, pRsp, len, code);
15!
614
  }
615

616
  if (code < 0) {
19,238!
617
    return TAOS_GET_TERRNO(code);
×
618
  }
619

620
  int32_t tlen = sizeof(SMqRspHead) + len;
19,238✔
621
  void*   buf = rpcMallocCont(tlen);
19,238✔
622
  if (buf == NULL) {
19,238!
623
    return terrno;
×
624
  }
625

626
  SMqRspHead* pHead = (SMqRspHead*)buf;
19,238✔
627
  initMqRspHead(pHead, type, epoch, consumerId, sver, ever);
19,238✔
628

629
  void* abuf = POINTER_SHIFT(buf, sizeof(SMqRspHead));
19,237✔
630

631
  SEncoder encoder = {0};
19,237✔
632
  tEncoderInit(&encoder, abuf, len);
19,237✔
633

634
  if (type == TMQ_MSG_TYPE__POLL_DATA_RSP ||
19,236✔
635
      type == TMQ_MSG_TYPE__WALINFO_RSP ||
15!
636
      type == TMQ_MSG_TYPE__POLL_RAW_DATA_RSP) {
637
    code = tEncodeMqDataRsp(&encoder, pRsp);
19,221✔
638
  } else if (type == TMQ_MSG_TYPE__POLL_DATA_META_RSP) {
15!
639
    code = tEncodeSTaosxRsp(&encoder, pRsp);
15✔
640
  }
641
  tEncoderClear(&encoder);
19,238✔
642
  if (code < 0) {
19,238!
643
    rpcFreeCont(buf);
×
644
    return TAOS_GET_TERRNO(code);
×
645
  }
646
  SRpcMsg rsp = {.info = *pRpcHandleInfo, .pCont = buf, .contLen = tlen, .code = 0};
19,238✔
647

648
  tmsgSendRsp(&rsp);
19,238✔
649
  return 0;
19,234✔
650
}
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2026 Coveralls, Inc