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

taosdata / TDengine / #3533

20 Nov 2024 07:11AM UTC coverage: 58.848% (-1.9%) from 60.78%
#3533

push

travis-ci

web-flow
Merge pull request #28823 from taosdata/fix/3.0/TD-32587

fix:[TD-32587]fix stmt segmentation fault

115578 of 252434 branches covered (45.79%)

Branch coverage included in aggregate %.

1 of 4 new or added lines in 1 file covered. (25.0%)

8038 existing lines in 233 files now uncovered.

194926 of 275199 relevant lines covered (70.83%)

1494459.59 hits per line

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

32.64
/source/client/src/clientStmt.c
1

2
#include "clientInt.h"
3
#include "clientLog.h"
4
#include "tdef.h"
5

6
#include "clientStmt.h"
7

8
char* gStmtStatusStr[] = {"unknown",     "init", "prepare", "settbname", "settags",
9
                          "fetchFields", "bind", "bindCol", "addBatch",  "exec"};
10

11
static FORCE_INLINE int32_t stmtAllocQNodeFromBuf(STableBufInfo* pTblBuf, void** pBuf) {
UNCOV
12
  if (pTblBuf->buffOffset < pTblBuf->buffSize) {
×
UNCOV
13
    *pBuf = (char*)pTblBuf->pCurBuff + pTblBuf->buffOffset;
×
UNCOV
14
    pTblBuf->buffOffset += pTblBuf->buffUnit;
×
15
  } else if (pTblBuf->buffIdx < taosArrayGetSize(pTblBuf->pBufList)) {
×
16
    pTblBuf->pCurBuff = taosArrayGetP(pTblBuf->pBufList, pTblBuf->buffIdx++);
×
17
    if (NULL == pTblBuf->pCurBuff) {
×
18
      return TAOS_GET_TERRNO(terrno);
×
19
    }
20
    *pBuf = pTblBuf->pCurBuff;
×
21
    pTblBuf->buffOffset = pTblBuf->buffUnit;
×
22
  } else {
23
    void* buff = taosMemoryMalloc(pTblBuf->buffSize);
×
24
    if (NULL == buff) {
×
25
      return terrno;
×
26
    }
27

28
    if (taosArrayPush(pTblBuf->pBufList, &buff) == NULL) {
×
29
      return terrno;
×
30
    }
31

32
    pTblBuf->buffIdx++;
×
33
    pTblBuf->pCurBuff = buff;
×
34
    *pBuf = buff;
×
35
    pTblBuf->buffOffset = pTblBuf->buffUnit;
×
36
  }
37

UNCOV
38
  return TSDB_CODE_SUCCESS;
×
39
}
40

UNCOV
41
bool stmtDequeue(STscStmt* pStmt, SStmtQNode** param) {
×
UNCOV
42
  while (0 == atomic_load_64(&pStmt->queue.qRemainNum)) {
×
UNCOV
43
    taosUsleep(1);
×
UNCOV
44
    return false;
×
45
  }
46

UNCOV
47
  SStmtQNode* orig = pStmt->queue.head;
×
48

UNCOV
49
  SStmtQNode* node = pStmt->queue.head->next;
×
UNCOV
50
  pStmt->queue.head = pStmt->queue.head->next;
×
51

52
  // taosMemoryFreeClear(orig);
53

UNCOV
54
  *param = node;
×
55

UNCOV
56
  (void)atomic_sub_fetch_64(&pStmt->queue.qRemainNum, 1);
×
57

UNCOV
58
  return true;
×
59
}
60

UNCOV
61
void stmtEnqueue(STscStmt* pStmt, SStmtQNode* param) {
×
UNCOV
62
  pStmt->queue.tail->next = param;
×
UNCOV
63
  pStmt->queue.tail = param;
×
64

UNCOV
65
  pStmt->stat.bindDataNum++;
×
UNCOV
66
  (void)atomic_add_fetch_64(&pStmt->queue.qRemainNum, 1);
×
UNCOV
67
}
×
68

69
static int32_t stmtCreateRequest(STscStmt* pStmt) {
144✔
70
  int32_t code = 0;
144✔
71

72
  if (pStmt->exec.pRequest == NULL) {
144✔
73
    code = buildRequest(pStmt->taos->id, pStmt->sql.sqlStr, pStmt->sql.sqlLen, NULL, false, &pStmt->exec.pRequest,
66✔
74
                        pStmt->reqid);
75
    if (pStmt->reqid != 0) {
66!
76
      pStmt->reqid++;
×
77
    }
78
    if (TSDB_CODE_SUCCESS == code) {
66!
79
      pStmt->exec.pRequest->syncQuery = true;
66✔
80
      pStmt->exec.pRequest->isStmtBind = true;
66✔
81
    }
82
  }
83

84
  return code;
144✔
85
}
86

87
int32_t stmtSwitchStatus(STscStmt* pStmt, STMT_STATUS newStatus) {
273✔
88
  int32_t code = 0;
273✔
89

90
  if (newStatus >= STMT_INIT && newStatus < STMT_MAX) {
273!
91
    STMT_LOG_SEQ(newStatus);
273✔
92
  }
93

94
  if (pStmt->errCode && newStatus != STMT_PREPARE) {
273!
95
    STMT_DLOG("stmt already failed with err: %s", tstrerror(pStmt->errCode));
×
96
    return pStmt->errCode;
×
97
  }
98

99
  switch (newStatus) {
273!
100
    case STMT_PREPARE:
43✔
101
      pStmt->errCode = 0;
43✔
102
      break;
43✔
103
    case STMT_SETTBNAME:
37✔
104
      if (STMT_STATUS_EQ(INIT)) {
37!
105
        code = TSDB_CODE_TSC_STMT_API_ERROR;
×
106
      }
107
      if (!pStmt->sql.stbInterlaceMode && (STMT_STATUS_EQ(BIND) || STMT_STATUS_EQ(BIND_COL))) {
37!
108
        code = TSDB_CODE_TSC_STMT_API_ERROR;
×
109
      }
110
      break;
37✔
111
    case STMT_SETTAGS:
5✔
112
      if (STMT_STATUS_NE(SETTBNAME) && STMT_STATUS_NE(FETCH_FIELDS)) {
5!
113
        code = TSDB_CODE_TSC_STMT_API_ERROR;
×
114
      }
115
      break;
5✔
116
    case STMT_FETCH_FIELDS:
×
117
      if (STMT_STATUS_EQ(INIT)) {
×
118
        code = TSDB_CODE_TSC_STMT_API_ERROR;
×
119
      }
120
      break;
×
121
    case STMT_BIND:
66✔
122
      if (STMT_STATUS_EQ(INIT) || STMT_STATUS_EQ(BIND_COL)) {
66!
123
        code = TSDB_CODE_TSC_STMT_API_ERROR;
×
124
      }
125
      /*
126
            if ((pStmt->sql.type == STMT_TYPE_MULTI_INSERT) && ()) {
127
              code = TSDB_CODE_TSC_STMT_API_ERROR;
128
            }
129
      */
130
      break;
66✔
131
    case STMT_BIND_COL:
×
132
      if (STMT_STATUS_EQ(INIT) || STMT_STATUS_EQ(BIND)) {
×
133
        code = TSDB_CODE_TSC_STMT_API_ERROR;
×
134
      }
135
      break;
×
136
    case STMT_ADD_BATCH:
61✔
137
      if (STMT_STATUS_NE(BIND) && STMT_STATUS_NE(BIND_COL) && STMT_STATUS_NE(FETCH_FIELDS)) {
61!
138
        code = TSDB_CODE_TSC_STMT_API_ERROR;
×
139
      }
140
      break;
61✔
141
    case STMT_EXECUTE:
61✔
142
      if (STMT_TYPE_QUERY == pStmt->sql.type) {
61✔
143
        if (STMT_STATUS_NE(ADD_BATCH) && STMT_STATUS_NE(FETCH_FIELDS) && STMT_STATUS_NE(BIND) &&
14!
144
            STMT_STATUS_NE(BIND_COL)) {
×
145
          code = TSDB_CODE_TSC_STMT_API_ERROR;
×
146
        }
147
      } else {
148
        if (STMT_STATUS_NE(ADD_BATCH) && STMT_STATUS_NE(FETCH_FIELDS)) {
47!
149
          code = TSDB_CODE_TSC_STMT_API_ERROR;
×
150
        }
151
      }
152
      break;
61✔
153
    default:
×
154
      code = TSDB_CODE_APP_ERROR;
×
155
      break;
×
156
  }
157

158
  STMT_ERR_RET(code);
273!
159

160
  pStmt->sql.status = newStatus;
273✔
161

162
  return TSDB_CODE_SUCCESS;
273✔
163
}
164

165
int32_t stmtGetTbName(TAOS_STMT* stmt, char** tbName) {
21✔
166
  STscStmt* pStmt = (STscStmt*)stmt;
21✔
167

168
  pStmt->sql.type = STMT_TYPE_MULTI_INSERT;
21✔
169

170
  if ('\0' == pStmt->bInfo.tbName[0]) {
21!
171
    tscError("no table name set");
×
172
    STMT_ERR_RET(TSDB_CODE_TSC_STMT_TBNAME_ERROR);
×
173
  }
174

175
  *tbName = pStmt->bInfo.tbName;
21✔
176

177
  return TSDB_CODE_SUCCESS;
21✔
178
}
179

180
int32_t stmtBackupQueryFields(STscStmt* pStmt) {
×
181
  SStmtQueryResInfo* pRes = &pStmt->sql.queryRes;
×
182
  pRes->numOfCols = pStmt->exec.pRequest->body.resInfo.numOfCols;
×
183
  pRes->precision = pStmt->exec.pRequest->body.resInfo.precision;
×
184

185
  int32_t size = pRes->numOfCols * sizeof(TAOS_FIELD);
×
186
  pRes->fields = taosMemoryMalloc(size);
×
187
  if (pRes->fields == NULL) {
×
188
    STMT_ERR_RET(terrno);
×
189
  }
190

191
  pRes->userFields = taosMemoryMalloc(size);
×
192
  if (pRes->userFields == NULL) {
×
193
    taosMemoryFreeClear(pRes->fields);
×
194
    STMT_ERR_RET(terrno);
×
195
  }
196

197
  (void)memcpy(pRes->fields, pStmt->exec.pRequest->body.resInfo.fields, size);
×
198
  (void)memcpy(pRes->userFields, pStmt->exec.pRequest->body.resInfo.userFields, size);
×
199

200
  return TSDB_CODE_SUCCESS;
×
201
}
202

203
int32_t stmtRestoreQueryFields(STscStmt* pStmt) {
×
204
  SStmtQueryResInfo* pRes = &pStmt->sql.queryRes;
×
205
  int32_t            size = pRes->numOfCols * sizeof(TAOS_FIELD);
×
206

207
  pStmt->exec.pRequest->body.resInfo.numOfCols = pRes->numOfCols;
×
208
  pStmt->exec.pRequest->body.resInfo.precision = pRes->precision;
×
209

210
  if (NULL == pStmt->exec.pRequest->body.resInfo.fields) {
×
211
    pStmt->exec.pRequest->body.resInfo.fields = taosMemoryMalloc(size);
×
212
    if (NULL == pStmt->exec.pRequest->body.resInfo.fields) {
×
213
      STMT_ERR_RET(terrno);
×
214
    }
215
    (void)memcpy(pStmt->exec.pRequest->body.resInfo.fields, pRes->fields, size);
×
216
  }
217

218
  if (NULL == pStmt->exec.pRequest->body.resInfo.userFields) {
×
219
    pStmt->exec.pRequest->body.resInfo.userFields = taosMemoryMalloc(size);
×
220
    if (NULL == pStmt->exec.pRequest->body.resInfo.userFields) {
×
221
      STMT_ERR_RET(terrno);
×
222
    }
223
    (void)memcpy(pStmt->exec.pRequest->body.resInfo.userFields, pRes->userFields, size);
×
224
  }
225

226
  return TSDB_CODE_SUCCESS;
×
227
}
228

229
int32_t stmtUpdateBindInfo(TAOS_STMT* stmt, STableMeta* pTableMeta, void* tags, SName* tbName, const char* sTableName,
24✔
230
                           bool autoCreateTbl) {
231
  STscStmt* pStmt = (STscStmt*)stmt;
24✔
232
  char      tbFName[TSDB_TABLE_FNAME_LEN];
233
  int32_t   code = tNameExtractFullName(tbName, tbFName);
24✔
234
  if (code != 0) {
24!
235
    return code;
×
236
  }
237

238
  (void)memcpy(&pStmt->bInfo.sname, tbName, sizeof(*tbName));
24✔
239
  tstrncpy(pStmt->bInfo.tbFName, tbFName, TSDB_TABLE_FNAME_LEN);
24✔
240
  pStmt->bInfo.tbFName[sizeof(pStmt->bInfo.tbFName) - 1] = 0;
24✔
241

242
  pStmt->bInfo.tbUid = autoCreateTbl ? 0 : pTableMeta->uid;
24✔
243
  pStmt->bInfo.tbSuid = pTableMeta->suid;
24✔
244
  pStmt->bInfo.tbVgId = pTableMeta->vgId;
24✔
245
  pStmt->bInfo.tbType = pTableMeta->tableType;
24✔
246
  pStmt->bInfo.boundTags = tags;
24✔
247
  pStmt->bInfo.tagsCached = false;
24✔
248
  tstrncpy(pStmt->bInfo.stbFName, sTableName, sizeof(pStmt->bInfo.stbFName));
24✔
249

250
  return TSDB_CODE_SUCCESS;
24✔
251
}
252

253
int32_t stmtUpdateExecInfo(TAOS_STMT* stmt, SHashObj* pVgHash, SHashObj* pBlockHash) {
24✔
254
  STscStmt* pStmt = (STscStmt*)stmt;
24✔
255

256
  pStmt->sql.pVgHash = pVgHash;
24✔
257
  pStmt->exec.pBlockHash = pBlockHash;
24✔
258

259
  return TSDB_CODE_SUCCESS;
24✔
260
}
261

262
int32_t stmtUpdateInfo(TAOS_STMT* stmt, STableMeta* pTableMeta, void* tags, SName* tbName, bool autoCreateTbl,
24✔
263
                       SHashObj* pVgHash, SHashObj* pBlockHash, const char* sTableName) {
264
  STscStmt* pStmt = (STscStmt*)stmt;
24✔
265

266
  STMT_ERR_RET(stmtUpdateBindInfo(stmt, pTableMeta, tags, tbName, sTableName, autoCreateTbl));
24!
267
  STMT_ERR_RET(stmtUpdateExecInfo(stmt, pVgHash, pBlockHash));
24!
268

269
  pStmt->sql.autoCreateTbl = autoCreateTbl;
24✔
270
  if (pStmt->sql.autoCreateTbl) {
24✔
271
    pStmt->sql.stbInterlaceMode = false;
21✔
272
  }
273

274
  return TSDB_CODE_SUCCESS;
24✔
275
}
276

277
int32_t stmtGetExecInfo(TAOS_STMT* stmt, SHashObj** pVgHash, SHashObj** pBlockHash) {
×
278
  STscStmt* pStmt = (STscStmt*)stmt;
×
279

280
  *pVgHash = pStmt->sql.pVgHash;
×
281
  pStmt->sql.pVgHash = NULL;
×
282

283
  *pBlockHash = pStmt->exec.pBlockHash;
×
284
  pStmt->exec.pBlockHash = NULL;
×
285

286
  return TSDB_CODE_SUCCESS;
×
287
}
288

289
int32_t stmtCacheBlock(STscStmt* pStmt) {
61✔
290
  if (pStmt->sql.type != STMT_TYPE_MULTI_INSERT) {
61✔
291
    return TSDB_CODE_SUCCESS;
18✔
292
  }
293

294
  uint64_t uid = pStmt->bInfo.tbUid;
43✔
295
  uint64_t cacheUid = (TSDB_CHILD_TABLE == pStmt->bInfo.tbType) ? pStmt->bInfo.tbSuid : uid;
43!
296

297
  if (taosHashGet(pStmt->sql.pTableCache, &cacheUid, sizeof(cacheUid))) {
43✔
298
    return TSDB_CODE_SUCCESS;
22✔
299
  }
300

301
  STableDataCxt** pSrc = taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
21✔
302
  if (!pSrc) {
21!
303
    return TSDB_CODE_OUT_OF_MEMORY;
×
304
  }
305
  STableDataCxt* pDst = NULL;
21✔
306

307
  STMT_ERR_RET(qCloneStmtDataBlock(&pDst, *pSrc, true));
21!
308

309
  SStmtTableCache cache = {
21✔
310
      .pDataCtx = pDst,
311
      .boundTags = pStmt->bInfo.boundTags,
21✔
312
  };
313

314
  if (taosHashPut(pStmt->sql.pTableCache, &cacheUid, sizeof(cacheUid), &cache, sizeof(cache))) {
21!
315
    return terrno;
×
316
  }
317

318
  if (pStmt->sql.autoCreateTbl) {
21!
319
    pStmt->bInfo.tagsCached = true;
21✔
320
  } else {
321
    pStmt->bInfo.boundTags = NULL;
×
322
  }
323

324
  return TSDB_CODE_SUCCESS;
21✔
325
}
326

327
int32_t stmtParseSql(STscStmt* pStmt) {
41✔
328
  pStmt->exec.pCurrBlock = NULL;
41✔
329

330
  SStmtCallback stmtCb = {
41✔
331
      .pStmt = pStmt,
332
      .getTbNameFn = stmtGetTbName,
333
      .setInfoFn = stmtUpdateInfo,
334
      .getExecInfoFn = stmtGetExecInfo,
335
  };
336

337
  STMT_ERR_RET(stmtCreateRequest(pStmt));
41!
338

339
  pStmt->stat.parseSqlNum++;
41✔
340
  STMT_ERR_RET(parseSql(pStmt->exec.pRequest, false, &pStmt->sql.pQuery, &stmtCb));
41✔
341
  pStmt->sql.siInfo.pQuery = pStmt->sql.pQuery;
38✔
342

343
  pStmt->bInfo.needParse = false;
38✔
344

345
  if (pStmt->sql.pQuery->pRoot && 0 == pStmt->sql.type) {
38✔
346
    pStmt->sql.type = STMT_TYPE_INSERT;
3✔
347
    pStmt->sql.stbInterlaceMode = false;
3✔
348
  } else if (pStmt->sql.pQuery->pPrepareRoot) {
35✔
349
    pStmt->sql.type = STMT_TYPE_QUERY;
14✔
350
    pStmt->sql.stbInterlaceMode = false;
14✔
351

352
    return TSDB_CODE_SUCCESS;
14✔
353
  }
354

355
  STableDataCxt** pSrc =
356
      (STableDataCxt**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
24✔
357
  if (NULL == pSrc || NULL == *pSrc) {
24!
358
    return TSDB_CODE_OUT_OF_MEMORY;
×
359
  }
360

361
  STableDataCxt* pTableCtx = *pSrc;
24✔
362
  if (pStmt->sql.stbInterlaceMode) {
24!
UNCOV
363
    int16_t lastIdx = -1;
×
364

UNCOV
365
    for (int32_t i = 0; i < pTableCtx->boundColsInfo.numOfBound; ++i) {
×
UNCOV
366
      if (pTableCtx->boundColsInfo.pColIndex[i] < lastIdx) {
×
367
        pStmt->sql.stbInterlaceMode = false;
×
368
        break;
×
369
      }
370

UNCOV
371
      lastIdx = pTableCtx->boundColsInfo.pColIndex[i];
×
372
    }
373
  }
374

375
  if (NULL == pStmt->sql.pBindInfo) {
24!
376
    pStmt->sql.pBindInfo = taosMemoryMalloc(pTableCtx->boundColsInfo.numOfBound * sizeof(*pStmt->sql.pBindInfo));
24✔
377
    if (NULL == pStmt->sql.pBindInfo) {
24!
378
      return terrno;
×
379
    }
380
  }
381

382
  return TSDB_CODE_SUCCESS;
24✔
383
}
384

385
int32_t stmtCleanBindInfo(STscStmt* pStmt) {
88✔
386
  pStmt->bInfo.tbUid = 0;
88✔
387
  pStmt->bInfo.tbSuid = 0;
88✔
388
  pStmt->bInfo.tbVgId = -1;
88✔
389
  pStmt->bInfo.tbType = 0;
88✔
390
  pStmt->bInfo.needParse = true;
88✔
391
  pStmt->bInfo.inExecCache = false;
88✔
392

393
  pStmt->bInfo.tbName[0] = 0;
88✔
394
  pStmt->bInfo.tbFName[0] = 0;
88✔
395
  if (!pStmt->bInfo.tagsCached) {
88✔
396
    qDestroyBoundColInfo(pStmt->bInfo.boundTags);
44✔
397
    taosMemoryFreeClear(pStmt->bInfo.boundTags);
44✔
398
  }
399
  pStmt->bInfo.stbFName[0] = 0;
88✔
400

401
  return TSDB_CODE_SUCCESS;
88✔
402
}
403

404
void stmtFreeTableBlkList(STableColsData* pTb) {
×
405
  (void)qResetStmtColumns(pTb->aCol, true);
×
406
  taosArrayDestroy(pTb->aCol);
×
407
}
×
408

UNCOV
409
void stmtResetQueueTableBuf(STableBufInfo* pTblBuf, SStmtQueue* pQueue) {
×
UNCOV
410
  pTblBuf->pCurBuff = taosArrayGetP(pTblBuf->pBufList, 0);
×
UNCOV
411
  if (NULL == pTblBuf->pCurBuff) {
×
412
    tscError("QInfo:%p, failed to get buffer from list", pTblBuf);
×
413
    return;
×
414
  }
UNCOV
415
  pTblBuf->buffIdx = 1;
×
UNCOV
416
  pTblBuf->buffOffset = sizeof(*pQueue->head);
×
417

UNCOV
418
  pQueue->head = pQueue->tail = pTblBuf->pCurBuff;
×
UNCOV
419
  pQueue->qRemainNum = 0;
×
UNCOV
420
  pQueue->head->next = NULL;
×
421
}
422

423
int32_t stmtCleanExecInfo(STscStmt* pStmt, bool keepTable, bool deepClean) {
104✔
424
  if (pStmt->sql.stbInterlaceMode) {
104!
UNCOV
425
    if (deepClean) {
×
UNCOV
426
      taosHashCleanup(pStmt->exec.pBlockHash);
×
UNCOV
427
      pStmt->exec.pBlockHash = NULL;
×
428

UNCOV
429
      if (NULL != pStmt->exec.pCurrBlock) {
×
UNCOV
430
        taosMemoryFreeClear(pStmt->exec.pCurrBlock->pData);
×
UNCOV
431
        qDestroyStmtDataBlock(pStmt->exec.pCurrBlock);
×
432
      }
433
    } else {
UNCOV
434
      pStmt->sql.siInfo.pTableColsIdx = 0;
×
UNCOV
435
      stmtResetQueueTableBuf(&pStmt->sql.siInfo.tbBuf, &pStmt->queue);
×
436
    }
437
  } else {
438
    if (STMT_TYPE_QUERY != pStmt->sql.type || deepClean) {
104✔
439
      taos_free_result(pStmt->exec.pRequest);
90✔
440
      pStmt->exec.pRequest = NULL;
90✔
441
    }
442

443
    size_t keyLen = 0;
104✔
444
    void*  pIter = taosHashIterate(pStmt->exec.pBlockHash, NULL);
104✔
445
    while (pIter) {
173✔
446
      STableDataCxt* pBlocks = *(STableDataCxt**)pIter;
69✔
447
      char*          key = taosHashGetKey(pIter, &keyLen);
69✔
448
      STableMeta*    pMeta = qGetTableMetaInDataBlock(pBlocks);
69✔
449

450
      if (keepTable && pBlocks == pStmt->exec.pCurrBlock) {
69!
451
        TSWAP(pBlocks->pData, pStmt->exec.pCurrTbData);
45✔
452
        STMT_ERR_RET(qResetStmtDataBlock(pBlocks, false));
104!
453

454
        pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
45✔
455
        continue;
45✔
456
      }
457

458
      qDestroyStmtDataBlock(pBlocks);
24✔
459
      STMT_ERR_RET(taosHashRemove(pStmt->exec.pBlockHash, key, keyLen));
24!
460

461
      pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
24✔
462
    }
463

464
    if (keepTable) {
104✔
465
      return TSDB_CODE_SUCCESS;
59✔
466
    }
467

468
    taosHashCleanup(pStmt->exec.pBlockHash);
45✔
469
    pStmt->exec.pBlockHash = NULL;
45✔
470

471
    tDestroySubmitTbData(pStmt->exec.pCurrTbData, TSDB_MSG_FLG_ENCODE);
45✔
472
    taosMemoryFreeClear(pStmt->exec.pCurrTbData);
45✔
473
  }
474

475
  STMT_ERR_RET(stmtCleanBindInfo(pStmt));
45!
476

477
  return TSDB_CODE_SUCCESS;
45✔
478
}
479

UNCOV
480
void stmtFreeTbBuf(void* buf) {
×
UNCOV
481
  void* pBuf = *(void**)buf;
×
UNCOV
482
  taosMemoryFree(pBuf);
×
UNCOV
483
}
×
484

UNCOV
485
void stmtFreeTbCols(void* buf) {
×
UNCOV
486
  SArray* pCols = *(SArray**)buf;
×
UNCOV
487
  taosArrayDestroy(pCols);
×
UNCOV
488
}
×
489

490
int32_t stmtCleanSQLInfo(STscStmt* pStmt) {
43✔
491
  STMT_DLOG_E("start to free SQL info");
43✔
492

493
  taosMemoryFree(pStmt->sql.pBindInfo);
43✔
494
  taosMemoryFree(pStmt->sql.queryRes.fields);
43✔
495
  taosMemoryFree(pStmt->sql.queryRes.userFields);
43✔
496
  taosMemoryFree(pStmt->sql.sqlStr);
43✔
497
  qDestroyQuery(pStmt->sql.pQuery);
43✔
498
  taosArrayDestroy(pStmt->sql.nodeList);
43✔
499
  taosHashCleanup(pStmt->sql.pVgHash);
43✔
500
  pStmt->sql.pVgHash = NULL;
43✔
501

502
  void* pIter = taosHashIterate(pStmt->sql.pTableCache, NULL);
43✔
503
  while (pIter) {
64✔
504
    SStmtTableCache* pCache = (SStmtTableCache*)pIter;
21✔
505

506
    qDestroyStmtDataBlock(pCache->pDataCtx);
21✔
507
    qDestroyBoundColInfo(pCache->boundTags);
21✔
508
    taosMemoryFreeClear(pCache->boundTags);
21!
509

510
    pIter = taosHashIterate(pStmt->sql.pTableCache, pIter);
21✔
511
  }
512
  taosHashCleanup(pStmt->sql.pTableCache);
43✔
513
  pStmt->sql.pTableCache = NULL;
43✔
514

515
  STMT_ERR_RET(stmtCleanExecInfo(pStmt, false, true));
43!
516
  STMT_ERR_RET(stmtCleanBindInfo(pStmt));
43!
517

518
  taos_free_result(pStmt->sql.siInfo.pRequest);
43✔
519
  taosHashCleanup(pStmt->sql.siInfo.pVgroupHash);
43✔
520
  tSimpleHashCleanup(pStmt->sql.siInfo.pTableHash);
43✔
521
  taosArrayDestroyEx(pStmt->sql.siInfo.tbBuf.pBufList, stmtFreeTbBuf);
43✔
522
  taosMemoryFree(pStmt->sql.siInfo.pTSchema);
43✔
523
  qDestroyStmtDataBlock(pStmt->sql.siInfo.pDataCtx);
43✔
524
  taosArrayDestroyEx(pStmt->sql.siInfo.pTableCols, stmtFreeTbCols);
43✔
525

526
  (void)memset(&pStmt->sql, 0, sizeof(pStmt->sql));
43✔
527
  pStmt->sql.siInfo.tableColsReady = true;
43✔
528

529
  STMT_DLOG_E("end to free SQL info");
43✔
530

531
  return TSDB_CODE_SUCCESS;
43✔
532
}
533

534
int32_t stmtTryAddTableVgroupInfo(STscStmt* pStmt, int32_t* vgId) {
×
535
  if (*vgId >= 0 && taosHashGet(pStmt->sql.pVgHash, (const char*)vgId, sizeof(*vgId))) {
×
536
    return TSDB_CODE_SUCCESS;
×
537
  }
538

539
  SVgroupInfo      vgInfo = {0};
×
540
  SRequestConnInfo conn = {.pTrans = pStmt->taos->pAppInfo->pTransporter,
×
541
                           .requestId = pStmt->exec.pRequest->requestId,
×
542
                           .requestObjRefId = pStmt->exec.pRequest->self,
×
543
                           .mgmtEps = getEpSet_s(&pStmt->taos->pAppInfo->mgmtEp)};
×
544

545
  int32_t code = catalogGetTableHashVgroup(pStmt->pCatalog, &conn, &pStmt->bInfo.sname, &vgInfo);
×
546
  if (TSDB_CODE_SUCCESS != code) {
×
547
    return code;
×
548
  }
549

550
  code =
551
      taosHashPut(pStmt->sql.pVgHash, (const char*)&vgInfo.vgId, sizeof(vgInfo.vgId), (char*)&vgInfo, sizeof(vgInfo));
×
552
  if (TSDB_CODE_SUCCESS != code) {
×
553
    return code;
×
554
  }
555

556
  *vgId = vgInfo.vgId;
×
557

558
  return TSDB_CODE_SUCCESS;
×
559
}
560

561
int32_t stmtRebuildDataBlock(STscStmt* pStmt, STableDataCxt* pDataBlock, STableDataCxt** newBlock, uint64_t uid,
×
562
                             uint64_t suid, int32_t vgId) {
563
  STMT_ERR_RET(stmtTryAddTableVgroupInfo(pStmt, &vgId));
×
564
  STMT_ERR_RET(qRebuildStmtDataBlock(newBlock, pDataBlock, uid, suid, vgId, pStmt->sql.autoCreateTbl));
×
565

566
  STMT_DLOG("tableDataCxt rebuilt, uid:%" PRId64 ", vgId:%d", uid, vgId);
×
567

568
  return TSDB_CODE_SUCCESS;
×
569
}
570

571
int32_t stmtGetFromCache(STscStmt* pStmt) {
37✔
572
  if (pStmt->sql.stbInterlaceMode && pStmt->sql.siInfo.pDataCtx) {
37!
573
    pStmt->bInfo.needParse = false;
×
574
    pStmt->bInfo.inExecCache = false;
×
575
    return TSDB_CODE_SUCCESS;
×
576
  }
577

578
  pStmt->bInfo.needParse = true;
37✔
579
  pStmt->bInfo.inExecCache = false;
37✔
580

581
  STableDataCxt** pCxtInExec = taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
37✔
582
  if (pCxtInExec) {
37✔
583
    pStmt->bInfo.needParse = false;
16✔
584
    pStmt->bInfo.inExecCache = true;
16✔
585

586
    pStmt->exec.pCurrBlock = *pCxtInExec;
16✔
587

588
    if (pStmt->sql.autoCreateTbl) {
16!
589
      tscDebug("reuse stmt block for tb %s in execBlock", pStmt->bInfo.tbFName);
16!
590
      return TSDB_CODE_SUCCESS;
16✔
591
    }
592
  }
593

594
  if (NULL == pStmt->pCatalog) {
21✔
595
    STMT_ERR_RET(catalogGetHandle(pStmt->taos->pAppInfo->clusterId, &pStmt->pCatalog));
13!
596
    pStmt->sql.siInfo.pCatalog = pStmt->pCatalog;
13✔
597
  }
598

599
  if (NULL == pStmt->sql.pTableCache || taosHashGetSize(pStmt->sql.pTableCache) <= 0) {
21!
600
    if (pStmt->bInfo.inExecCache) {
21!
601
      pStmt->bInfo.needParse = false;
×
602
      tscDebug("reuse stmt block for tb %s in execBlock", pStmt->bInfo.tbFName);
×
603
      return TSDB_CODE_SUCCESS;
×
604
    }
605

606
    tscDebug("no stmt block cache for tb %s", pStmt->bInfo.tbFName);
21!
607
    return TSDB_CODE_SUCCESS;
21✔
608
  }
609

610
  if (pStmt->sql.autoCreateTbl) {
×
611
    SStmtTableCache* pCache = taosHashGet(pStmt->sql.pTableCache, &pStmt->bInfo.tbSuid, sizeof(pStmt->bInfo.tbSuid));
×
612
    if (pCache) {
×
613
      pStmt->bInfo.needParse = false;
×
614
      pStmt->bInfo.tbUid = 0;
×
615

616
      STableDataCxt* pNewBlock = NULL;
×
617
      STMT_ERR_RET(stmtRebuildDataBlock(pStmt, pCache->pDataCtx, &pNewBlock, 0, pStmt->bInfo.tbSuid, -1));
×
618

619
      if (taosHashPut(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName), &pNewBlock,
×
620
                      POINTER_BYTES)) {
621
        STMT_ERR_RET(terrno);
×
622
      }
623

624
      pStmt->exec.pCurrBlock = pNewBlock;
×
625

626
      tscDebug("reuse stmt block for tb %s in sqlBlock, suid:0x%" PRIx64, pStmt->bInfo.tbFName, pStmt->bInfo.tbSuid);
×
627

628
      return TSDB_CODE_SUCCESS;
×
629
    }
630

631
    STMT_RET(stmtCleanBindInfo(pStmt));
×
632
  }
633

634
  uint64_t uid, suid;
635
  int32_t  vgId;
636
  int8_t   tableType;
637

638
  STableMeta*      pTableMeta = NULL;
×
639
  SRequestConnInfo conn = {.pTrans = pStmt->taos->pAppInfo->pTransporter,
×
640
                           .requestId = pStmt->exec.pRequest->requestId,
×
641
                           .requestObjRefId = pStmt->exec.pRequest->self,
×
642
                           .mgmtEps = getEpSet_s(&pStmt->taos->pAppInfo->mgmtEp)};
×
643
  int32_t          code = catalogGetTableMeta(pStmt->pCatalog, &conn, &pStmt->bInfo.sname, &pTableMeta);
×
644

645
  pStmt->stat.ctgGetTbMetaNum++;
×
646

647
  if (TSDB_CODE_PAR_TABLE_NOT_EXIST == code) {
×
648
    tscDebug("tb %s not exist", pStmt->bInfo.tbFName);
×
649
    STMT_ERR_RET(stmtCleanBindInfo(pStmt));
×
650

651
    STMT_ERR_RET(code);
×
652
  }
653

654
  STMT_ERR_RET(code);
×
655

656
  uid = pTableMeta->uid;
×
657
  suid = pTableMeta->suid;
×
658
  tableType = pTableMeta->tableType;
×
659
  pStmt->bInfo.tbVgId = pTableMeta->vgId;
×
660
  vgId = pTableMeta->vgId;
×
661

662
  taosMemoryFree(pTableMeta);
×
663

664
  uint64_t cacheUid = (TSDB_CHILD_TABLE == tableType) ? suid : uid;
×
665

666
  if (uid == pStmt->bInfo.tbUid) {
×
667
    pStmt->bInfo.needParse = false;
×
668

669
    tscDebug("tb %s is current table", pStmt->bInfo.tbFName);
×
670

671
    return TSDB_CODE_SUCCESS;
×
672
  }
673

674
  if (pStmt->bInfo.inExecCache) {
×
675
    SStmtTableCache* pCache = taosHashGet(pStmt->sql.pTableCache, &cacheUid, sizeof(cacheUid));
×
676
    if (NULL == pCache) {
×
677
      tscError("table [%s, %" PRIx64 ", %" PRIx64 "] found in exec blockHash, but not in sql blockHash",
×
678
               pStmt->bInfo.tbFName, uid, cacheUid);
679

680
      STMT_ERR_RET(TSDB_CODE_APP_ERROR);
×
681
    }
682

683
    pStmt->bInfo.needParse = false;
×
684

685
    pStmt->bInfo.tbUid = uid;
×
686
    pStmt->bInfo.tbSuid = suid;
×
687
    pStmt->bInfo.tbType = tableType;
×
688
    pStmt->bInfo.boundTags = pCache->boundTags;
×
689
    pStmt->bInfo.tagsCached = true;
×
690

691
    tscDebug("tb %s in execBlock list, set to current", pStmt->bInfo.tbFName);
×
692

693
    return TSDB_CODE_SUCCESS;
×
694
  }
695

696
  SStmtTableCache* pCache = taosHashGet(pStmt->sql.pTableCache, &cacheUid, sizeof(cacheUid));
×
697
  if (pCache) {
×
698
    pStmt->bInfo.needParse = false;
×
699

700
    pStmt->bInfo.tbUid = uid;
×
701
    pStmt->bInfo.tbSuid = suid;
×
702
    pStmt->bInfo.tbType = tableType;
×
703
    pStmt->bInfo.boundTags = pCache->boundTags;
×
704
    pStmt->bInfo.tagsCached = true;
×
705

706
    STableDataCxt* pNewBlock = NULL;
×
707
    STMT_ERR_RET(stmtRebuildDataBlock(pStmt, pCache->pDataCtx, &pNewBlock, uid, suid, vgId));
×
708

709
    if (taosHashPut(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName), &pNewBlock,
×
710
                    POINTER_BYTES)) {
711
      STMT_ERR_RET(terrno);
×
712
    }
713

714
    pStmt->exec.pCurrBlock = pNewBlock;
×
715

716
    tscDebug("tb %s in sqlBlock list, set to current", pStmt->bInfo.tbFName);
×
717

718
    return TSDB_CODE_SUCCESS;
×
719
  }
720

721
  STMT_ERR_RET(stmtCleanBindInfo(pStmt));
×
722

723
  return TSDB_CODE_SUCCESS;
×
724
}
725

726
int32_t stmtResetStmt(STscStmt* pStmt) {
8✔
727
  STMT_ERR_RET(stmtCleanSQLInfo(pStmt));
8!
728

729
  pStmt->sql.pTableCache = taosHashInit(100, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
8✔
730
  if (NULL == pStmt->sql.pTableCache) {
8!
731
    STMT_ERR_RET(terrno);
×
732
  }
733

734
  pStmt->sql.status = STMT_INIT;
8✔
735

736
  return TSDB_CODE_SUCCESS;
8✔
737
}
738

UNCOV
739
int32_t stmtAsyncOutput(STscStmt* pStmt, void* param) {
×
UNCOV
740
  SStmtQNode* pParam = (SStmtQNode*)param;
×
741

UNCOV
742
  if (pParam->restoreTbCols) {
×
UNCOV
743
    for (int32_t i = 0; i < pStmt->sql.siInfo.pTableColsIdx; ++i) {
×
UNCOV
744
      SArray** p = (SArray**)TARRAY_GET_ELEM(pStmt->sql.siInfo.pTableCols, i);
×
UNCOV
745
      *p = taosArrayInit(20, POINTER_BYTES);
×
UNCOV
746
      if (*p == NULL) {
×
747
        STMT_ERR_RET(terrno);
×
748
      }
749
    }
750

UNCOV
751
    atomic_store_8((int8_t*)&pStmt->sql.siInfo.tableColsReady, true);
×
752
  } else {
UNCOV
753
    STMT_ERR_RET(qAppendStmtTableOutput(pStmt->sql.pQuery, pStmt->sql.pVgHash, &pParam->tblData, pStmt->exec.pCurrBlock,
×
754
                                        &pStmt->sql.siInfo));
755

756
    // taosMemoryFree(pParam->pTbData);
757

UNCOV
758
    (void)atomic_sub_fetch_64(&pStmt->sql.siInfo.tbRemainNum, 1);
×
759
  }
UNCOV
760
  return TSDB_CODE_SUCCESS;
×
761
}
762

UNCOV
763
void* stmtBindThreadFunc(void* param) {
×
UNCOV
764
  setThreadName("stmtBind");
×
765

UNCOV
766
  qInfo("stmt bind thread started");
×
767

UNCOV
768
  STscStmt* pStmt = (STscStmt*)param;
×
769

UNCOV
770
  while (true) {
×
UNCOV
771
    if (atomic_load_8((int8_t*)&pStmt->queue.stopQueue)) {
×
UNCOV
772
      break;
×
773
    }
774

UNCOV
775
    SStmtQNode* asyncParam = NULL;
×
UNCOV
776
    if (!stmtDequeue(pStmt, &asyncParam)) {
×
UNCOV
777
      continue;
×
778
    }
779

UNCOV
780
    int ret = stmtAsyncOutput(pStmt, asyncParam);
×
UNCOV
781
    if (ret != 0) {
×
782
      qError("stmtAsyncOutput failed, reason:%s", tstrerror(ret));
×
783
    }
784
  }
785

UNCOV
786
  qInfo("stmt bind thread stopped");
×
787

UNCOV
788
  return NULL;
×
789
}
790

UNCOV
791
int32_t stmtStartBindThread(STscStmt* pStmt) {
×
792
  TdThreadAttr thAttr;
UNCOV
793
  if (taosThreadAttrInit(&thAttr) != 0) {
×
794
    return TSDB_CODE_TSC_INTERNAL_ERROR;
×
795
  }
UNCOV
796
  if (taosThreadAttrSetDetachState(&thAttr, PTHREAD_CREATE_JOINABLE) != 0) {
×
797
    return TSDB_CODE_TSC_INTERNAL_ERROR;
×
798
  }
799

UNCOV
800
  if (taosThreadCreate(&pStmt->bindThread, &thAttr, stmtBindThreadFunc, pStmt) != 0) {
×
801
    terrno = TAOS_SYSTEM_ERROR(errno);
×
802
    STMT_ERR_RET(terrno);
×
803
  }
804

UNCOV
805
  pStmt->bindThreadInUse = true;
×
806

UNCOV
807
  (void)taosThreadAttrDestroy(&thAttr);
×
UNCOV
808
  return TSDB_CODE_SUCCESS;
×
809
}
810

UNCOV
811
int32_t stmtInitQueue(STscStmt* pStmt) {
×
UNCOV
812
  STMT_ERR_RET(stmtAllocQNodeFromBuf(&pStmt->sql.siInfo.tbBuf, (void**)&pStmt->queue.head));
×
UNCOV
813
  pStmt->queue.tail = pStmt->queue.head;
×
814

UNCOV
815
  return TSDB_CODE_SUCCESS;
×
816
}
817

UNCOV
818
int32_t stmtInitTableBuf(STableBufInfo* pTblBuf) {
×
UNCOV
819
  pTblBuf->buffUnit = sizeof(SStmtQNode);
×
UNCOV
820
  pTblBuf->buffSize = pTblBuf->buffUnit * 1000;
×
UNCOV
821
  pTblBuf->pBufList = taosArrayInit(100, POINTER_BYTES);
×
UNCOV
822
  if (NULL == pTblBuf->pBufList) {
×
823
    return terrno;
×
824
  }
UNCOV
825
  void* buff = taosMemoryMalloc(pTblBuf->buffSize);
×
UNCOV
826
  if (NULL == buff) {
×
827
    return terrno;
×
828
  }
829

UNCOV
830
  if (taosArrayPush(pTblBuf->pBufList, &buff) == NULL) {
×
831
    return terrno;
×
832
  }
833

UNCOV
834
  pTblBuf->pCurBuff = buff;
×
UNCOV
835
  pTblBuf->buffIdx = 1;
×
UNCOV
836
  pTblBuf->buffOffset = 0;
×
837

UNCOV
838
  return TSDB_CODE_SUCCESS;
×
839
}
840

841
TAOS_STMT* stmtInit(STscObj* taos, int64_t reqid, TAOS_STMT_OPTIONS* pOptions) {
35✔
842
  STscObj*  pObj = (STscObj*)taos;
35✔
843
  STscStmt* pStmt = NULL;
35✔
844
  int32_t   code = 0;
35✔
845

846
  pStmt = taosMemoryCalloc(1, sizeof(STscStmt));
35✔
847
  if (NULL == pStmt) {
35!
848
    terrno = TSDB_CODE_OUT_OF_MEMORY;
×
849
    return NULL;
×
850
  }
851

852
  pStmt->sql.pTableCache = taosHashInit(100, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
35✔
853
  if (NULL == pStmt->sql.pTableCache) {
35!
854
    terrno = TSDB_CODE_OUT_OF_MEMORY;
×
855
    taosMemoryFree(pStmt);
×
856
    return NULL;
×
857
  }
858

859
  pStmt->taos = pObj;
35✔
860
  pStmt->bInfo.needParse = true;
35✔
861
  pStmt->sql.status = STMT_INIT;
35✔
862
  pStmt->reqid = reqid;
35✔
863
  pStmt->errCode = TSDB_CODE_SUCCESS;
35✔
864

865
  if (NULL != pOptions) {
35!
UNCOV
866
    (void)memcpy(&pStmt->options, pOptions, sizeof(pStmt->options));
×
UNCOV
867
    if (pOptions->singleStbInsert && pOptions->singleTableBindOnce) {
×
UNCOV
868
      pStmt->stbInterlaceMode = true;
×
869
    }
870
  }
871

872
  if (pStmt->stbInterlaceMode) {
35!
UNCOV
873
    pStmt->sql.siInfo.transport = taos->pAppInfo->pTransporter;
×
UNCOV
874
    pStmt->sql.siInfo.acctId = taos->acctId;
×
UNCOV
875
    pStmt->sql.siInfo.dbname = taos->db;
×
UNCOV
876
    pStmt->sql.siInfo.mgmtEpSet = getEpSet_s(&pStmt->taos->pAppInfo->mgmtEp);
×
UNCOV
877
    pStmt->sql.siInfo.pTableHash = tSimpleHashInit(100, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY));
×
UNCOV
878
    if (NULL == pStmt->sql.siInfo.pTableHash) {
×
879
      terrno = TSDB_CODE_OUT_OF_MEMORY;
×
880
      (void)stmtClose(pStmt);
×
881
      return NULL;
×
882
    }
UNCOV
883
    pStmt->sql.siInfo.pTableCols = taosArrayInit(STMT_TABLE_COLS_NUM, POINTER_BYTES);
×
UNCOV
884
    if (NULL == pStmt->sql.siInfo.pTableCols) {
×
885
      terrno = TSDB_CODE_OUT_OF_MEMORY;
×
886
      (void)stmtClose(pStmt);
×
887
      return NULL;
×
888
    }
889

UNCOV
890
    code = stmtInitTableBuf(&pStmt->sql.siInfo.tbBuf);
×
UNCOV
891
    if (TSDB_CODE_SUCCESS == code) {
×
UNCOV
892
      code = stmtInitQueue(pStmt);
×
893
    }
UNCOV
894
    if (TSDB_CODE_SUCCESS == code) {
×
UNCOV
895
      code = stmtStartBindThread(pStmt);
×
896
    }
UNCOV
897
    if (TSDB_CODE_SUCCESS != code) {
×
898
      terrno = code;
×
899
      (void)stmtClose(pStmt);
×
900
      return NULL;
×
901
    }
902
  }
903

904
  pStmt->sql.siInfo.tableColsReady = true;
35✔
905

906
  STMT_LOG_SEQ(STMT_INIT);
35✔
907

908
  tscDebug("stmt:%p initialized", pStmt);
35!
909

910
  return pStmt;
35✔
911
}
912

913
int stmtPrepare(TAOS_STMT* stmt, const char* sql, unsigned long length) {
43✔
914
  STscStmt* pStmt = (STscStmt*)stmt;
43✔
915

916
  STMT_DLOG_E("start to prepare");
43✔
917

918
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
43!
919
    return pStmt->errCode;
×
920
  }
921

922
  if (pStmt->sql.status >= STMT_PREPARE) {
43✔
923
    STMT_ERR_RET(stmtResetStmt(pStmt));
8!
924
  }
925

926
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_PREPARE));
43!
927

928
  if (length <= 0) {
43!
UNCOV
929
    length = strlen(sql);
×
930
  }
931

932
  pStmt->sql.sqlStr = taosStrndup(sql, length);
43✔
933
  if (!pStmt->sql.sqlStr) {
43!
934
    return terrno;
×
935
  }
936
  pStmt->sql.sqlLen = length;
43✔
937
  pStmt->sql.stbInterlaceMode = pStmt->stbInterlaceMode;
43✔
938

939
  char* dbName = NULL;
43✔
940
  if (qParseDbName(sql, length, &dbName)) {
43!
941
    STMT_ERR_RET(stmtSetDbName(stmt, dbName));
×
942
    taosMemoryFreeClear(dbName);
×
943
  }
944

945
  return TSDB_CODE_SUCCESS;
43✔
946
}
947

UNCOV
948
int32_t stmtInitStbInterlaceTableInfo(STscStmt* pStmt) {
×
UNCOV
949
  STableDataCxt** pSrc = taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
×
UNCOV
950
  if (!pSrc) {
×
951
    return TSDB_CODE_OUT_OF_MEMORY;
×
952
  }
UNCOV
953
  STableDataCxt* pDst = NULL;
×
954

UNCOV
955
  STMT_ERR_RET(qCloneStmtDataBlock(&pDst, *pSrc, true));
×
UNCOV
956
  pStmt->sql.siInfo.pDataCtx = pDst;
×
957

UNCOV
958
  SArray* pTblCols = NULL;
×
UNCOV
959
  for (int32_t i = 0; i < STMT_TABLE_COLS_NUM; i++) {
×
UNCOV
960
    pTblCols = taosArrayInit(20, POINTER_BYTES);
×
UNCOV
961
    if (NULL == pTblCols) {
×
962
      return terrno;
×
963
    }
964

UNCOV
965
    if (taosArrayPush(pStmt->sql.siInfo.pTableCols, &pTblCols) == NULL) {
×
966
      return terrno;
×
967
    }
968
  }
969

UNCOV
970
  pStmt->sql.siInfo.boundTags = pStmt->bInfo.boundTags;
×
971

UNCOV
972
  return TSDB_CODE_SUCCESS;
×
973
}
974

975
int stmtSetDbName(TAOS_STMT* stmt, const char* dbName) {
×
976
  STscStmt* pStmt = (STscStmt*)stmt;
×
977

978
  STMT_DLOG("start to set dbName: %s", dbName);
×
979

980
  STMT_ERR_RET(stmtCreateRequest(pStmt));
×
981

982
  // The SQL statement specifies a database name, overriding the previously specified database
983
  taosMemoryFreeClear(pStmt->exec.pRequest->pDb);
×
984
  pStmt->exec.pRequest->pDb = taosStrdup(dbName);
×
985
  if (pStmt->exec.pRequest->pDb == NULL) {
×
986
    return terrno;
×
987
  }
988
  return TSDB_CODE_SUCCESS;
×
989
}
990

991
int stmtSetTbName(TAOS_STMT* stmt, const char* tbName) {
37✔
992
  STscStmt* pStmt = (STscStmt*)stmt;
37✔
993

994
  int64_t startUs = taosGetTimestampUs();
37✔
995

996
  STMT_DLOG("start to set tbName: %s", tbName);
37✔
997

998
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
37!
999
    return pStmt->errCode;
×
1000
  }
1001

1002
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_SETTBNAME));
37!
1003

1004
  int32_t insert = 0;
37✔
1005
  STMT_ERR_RET(stmtIsInsert(stmt, &insert));
37!
1006
  if (0 == insert) {
37!
1007
    tscError("set tb name not available for none insert statement");
×
1008
    STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR);
×
1009
  }
1010

1011
  if (!pStmt->sql.stbInterlaceMode || NULL == pStmt->sql.siInfo.pDataCtx) {
37!
1012
    STMT_ERR_RET(stmtCreateRequest(pStmt));
37!
1013

1014
    STMT_ERR_RET(qCreateSName(&pStmt->bInfo.sname, tbName, pStmt->taos->acctId, pStmt->exec.pRequest->pDb,
37!
1015
                              pStmt->exec.pRequest->msgBuf, pStmt->exec.pRequest->msgBufLen));
1016
    STMT_ERR_RET(tNameExtractFullName(&pStmt->bInfo.sname, pStmt->bInfo.tbFName));
37!
1017

1018
    STMT_ERR_RET(stmtGetFromCache(pStmt));
37!
1019

1020
    if (pStmt->bInfo.needParse) {
37✔
1021
      tstrncpy(pStmt->bInfo.tbName, tbName, sizeof(pStmt->bInfo.tbName));
21✔
1022
      pStmt->bInfo.tbName[sizeof(pStmt->bInfo.tbName) - 1] = 0;
21✔
1023

1024
      STMT_ERR_RET(stmtParseSql(pStmt));
21!
1025
    }
1026
  } else {
UNCOV
1027
    tstrncpy(pStmt->bInfo.tbName, tbName, sizeof(pStmt->bInfo.tbName));
×
UNCOV
1028
    pStmt->bInfo.tbName[sizeof(pStmt->bInfo.tbName) - 1] = 0;
×
UNCOV
1029
    pStmt->exec.pRequest->requestId++;
×
UNCOV
1030
    pStmt->bInfo.needParse = false;
×
1031
  }
1032

1033
  if (pStmt->sql.stbInterlaceMode && NULL == pStmt->sql.siInfo.pDataCtx) {
37!
UNCOV
1034
    STMT_ERR_RET(stmtInitStbInterlaceTableInfo(pStmt));
×
1035
  }
1036

1037
  int64_t startUs2 = taosGetTimestampUs();
37✔
1038
  pStmt->stat.setTbNameUs += startUs2 - startUs;
37✔
1039

1040
  return TSDB_CODE_SUCCESS;
37✔
1041
}
1042

1043
int stmtSetTbTags(TAOS_STMT* stmt, TAOS_MULTI_BIND* tags) {
5✔
1044
  STscStmt* pStmt = (STscStmt*)stmt;
5✔
1045

1046
  STMT_DLOG_E("start to set tbTags");
5!
1047

1048
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
5!
1049
    return pStmt->errCode;
×
1050
  }
1051

1052
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_SETTAGS));
5!
1053

1054
  SBoundColInfo* tags_info = (SBoundColInfo*)pStmt->bInfo.boundTags;
5✔
1055
  if (tags_info->numOfBound <= 0 || tags_info->numOfCols <= 0) {
5!
1056
    tscWarn("no tags bound in sql, will not bound tags");
×
1057
    return TSDB_CODE_SUCCESS;
×
1058
  }
1059

1060
  if (pStmt->bInfo.inExecCache) {
5!
1061
    return TSDB_CODE_SUCCESS;
×
1062
  }
1063

1064
  STableDataCxt** pDataBlock =
1065
      (STableDataCxt**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
5✔
1066
  if (NULL == pDataBlock) {
5!
1067
    tscError("table %s not found in exec blockHash", pStmt->bInfo.tbFName);
×
1068
    STMT_ERR_RET(TSDB_CODE_APP_ERROR);
×
1069
  }
1070

1071
  tscDebug("start to bind stmt tag values");
5!
1072
  STMT_ERR_RET(qBindStmtTagsValue(*pDataBlock, pStmt->bInfo.boundTags, pStmt->bInfo.tbSuid, pStmt->bInfo.stbFName,
5!
1073
                                  pStmt->bInfo.sname.tname, tags, pStmt->exec.pRequest->msgBuf,
1074
                                  pStmt->exec.pRequest->msgBufLen));
1075

1076
  return TSDB_CODE_SUCCESS;
5✔
1077
}
1078

1079
int stmtFetchTagFields(STscStmt* pStmt, int32_t* fieldNum, TAOS_FIELD_E** fields) {
×
1080
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
×
1081
    return pStmt->errCode;
×
1082
  }
1083

1084
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
×
1085
    tscError("invalid operation to get query tag fileds");
×
1086
    STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR);
×
1087
  }
1088

1089
  STableDataCxt** pDataBlock =
1090
      (STableDataCxt**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
×
1091
  if (NULL == pDataBlock) {
×
1092
    tscError("table %s not found in exec blockHash", pStmt->bInfo.tbFName);
×
1093
    STMT_ERR_RET(TSDB_CODE_APP_ERROR);
×
1094
  }
1095

1096
  STMT_ERR_RET(qBuildStmtTagFields(*pDataBlock, pStmt->bInfo.boundTags, fieldNum, fields));
×
1097

1098
  return TSDB_CODE_SUCCESS;
×
1099
}
1100

1101
int stmtFetchColFields(STscStmt* pStmt, int32_t* fieldNum, TAOS_FIELD_E** fields) {
×
1102
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
×
1103
    return pStmt->errCode;
×
1104
  }
1105

1106
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
×
1107
    tscError("invalid operation to get query column fileds");
×
1108
    STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR);
×
1109
  }
1110

1111
  STableDataCxt** pDataBlock = NULL;
×
1112

1113
  if (pStmt->sql.stbInterlaceMode) {
×
1114
    pDataBlock = &pStmt->sql.siInfo.pDataCtx;
×
1115
  } else {
1116
    pDataBlock =
1117
        (STableDataCxt**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
×
1118
    if (NULL == pDataBlock) {
×
1119
      tscError("table %s not found in exec blockHash", pStmt->bInfo.tbFName);
×
1120
      STMT_ERR_RET(TSDB_CODE_APP_ERROR);
×
1121
    }
1122
  }
1123

1124
  STMT_ERR_RET(qBuildStmtColFields(*pDataBlock, fieldNum, fields));
×
1125

1126
  return TSDB_CODE_SUCCESS;
×
1127
}
1128

1129
/*
1130
SArray* stmtGetFreeCol(STscStmt* pStmt, int32_t* idx) {
1131
  while (true) {
1132
    if (pStmt->exec.smInfo.pColIdx >= STMT_COL_BUF_SIZE) {
1133
      pStmt->exec.smInfo.pColIdx = 0;
1134
    }
1135

1136
    if ((pStmt->exec.smInfo.pColIdx + 1) == atomic_load_32(&pStmt->exec.smInfo.pColFreeIdx)) {
1137
      taosUsleep(1);
1138
      continue;
1139
    }
1140

1141
    *idx = pStmt->exec.smInfo.pColIdx;
1142
    return pStmt->exec.smInfo.pCols[pStmt->exec.smInfo.pColIdx++];
1143
  }
1144
}
1145
*/
1146

UNCOV
1147
int32_t stmtAppendTablePostHandle(STscStmt* pStmt, SStmtQNode* param) {
×
UNCOV
1148
  if (NULL == pStmt->sql.siInfo.pVgroupHash) {
×
UNCOV
1149
    pStmt->sql.siInfo.pVgroupHash =
×
UNCOV
1150
        taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_ENTRY_LOCK);
×
1151
  }
UNCOV
1152
  if (NULL == pStmt->sql.siInfo.pVgroupList) {
×
UNCOV
1153
    pStmt->sql.siInfo.pVgroupList = taosArrayInit(64, POINTER_BYTES);
×
1154
  }
1155

UNCOV
1156
  if (NULL == pStmt->sql.siInfo.pRequest) {
×
UNCOV
1157
    STMT_ERR_RET(buildRequest(pStmt->taos->id, pStmt->sql.sqlStr, pStmt->sql.sqlLen, NULL, false,
×
1158
                              (SRequestObj**)&pStmt->sql.siInfo.pRequest, pStmt->reqid));
1159

UNCOV
1160
    if (pStmt->reqid != 0) {
×
1161
      pStmt->reqid++;
×
1162
    }
UNCOV
1163
    pStmt->exec.pRequest->syncQuery = true;
×
1164

UNCOV
1165
    pStmt->sql.siInfo.requestId = ((SRequestObj*)pStmt->sql.siInfo.pRequest)->requestId;
×
UNCOV
1166
    pStmt->sql.siInfo.requestSelf = ((SRequestObj*)pStmt->sql.siInfo.pRequest)->self;
×
1167
  }
1168

UNCOV
1169
  if (!pStmt->sql.siInfo.tbFromHash && pStmt->sql.siInfo.firstName[0] &&
×
UNCOV
1170
      0 == strcmp(pStmt->sql.siInfo.firstName, pStmt->bInfo.tbName)) {
×
UNCOV
1171
    pStmt->sql.siInfo.tbFromHash = true;
×
1172
  }
1173

UNCOV
1174
  if (0 == pStmt->sql.siInfo.firstName[0]) {
×
UNCOV
1175
    tstrncpy(pStmt->sql.siInfo.firstName, pStmt->bInfo.tbName, TSDB_TABLE_NAME_LEN);
×
1176
  }
1177

UNCOV
1178
  param->tblData.getFromHash = pStmt->sql.siInfo.tbFromHash;
×
UNCOV
1179
  param->next = NULL;
×
1180

UNCOV
1181
  (void)atomic_add_fetch_64(&pStmt->sql.siInfo.tbRemainNum, 1);
×
1182

UNCOV
1183
  stmtEnqueue(pStmt, param);
×
1184

UNCOV
1185
  return TSDB_CODE_SUCCESS;
×
1186
}
1187

1188
static FORCE_INLINE int32_t stmtGetTableColsFromCache(STscStmt* pStmt, SArray** pTableCols) {
1189
  while (true) {
UNCOV
1190
    if (pStmt->sql.siInfo.pTableColsIdx < taosArrayGetSize(pStmt->sql.siInfo.pTableCols)) {
×
UNCOV
1191
      *pTableCols = (SArray*)taosArrayGetP(pStmt->sql.siInfo.pTableCols, pStmt->sql.siInfo.pTableColsIdx++);
×
UNCOV
1192
      break;
×
1193
    } else {
1194
      SArray* pTblCols = NULL;
×
1195
      for (int32_t i = 0; i < 100; i++) {
×
1196
        pTblCols = taosArrayInit(20, POINTER_BYTES);
×
1197
        if (NULL == pTblCols) {
×
1198
          return terrno;
×
1199
        }
1200

1201
        if (taosArrayPush(pStmt->sql.siInfo.pTableCols, &pTblCols) == NULL) {
×
1202
          return terrno;
×
1203
        }
1204
      }
1205
    }
1206
  }
1207

UNCOV
1208
  return TSDB_CODE_SUCCESS;
×
1209
}
1210

1211
int stmtBindBatch(TAOS_STMT* stmt, TAOS_MULTI_BIND* bind, int32_t colIdx) {
66✔
1212
  STscStmt* pStmt = (STscStmt*)stmt;
66✔
1213
  int32_t   code = 0;
66✔
1214

1215
  int64_t startUs = taosGetTimestampUs();
66✔
1216

1217
  STMT_DLOG("start to bind stmt data, colIdx: %d", colIdx);
66✔
1218

1219
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
66!
1220
    return pStmt->errCode;
×
1221
  }
1222

1223
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_BIND));
66!
1224

1225
  if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 &&
66!
1226
      STMT_TYPE_MULTI_INSERT != pStmt->sql.type) {
×
1227
    pStmt->bInfo.needParse = false;
×
1228
  }
1229

1230
  if (pStmt->exec.pRequest && STMT_TYPE_QUERY == pStmt->sql.type && pStmt->sql.runTimes) {
66!
1231
    taos_free_result(pStmt->exec.pRequest);
×
1232
    pStmt->exec.pRequest = NULL;
×
1233
  }
1234

1235
  STMT_ERR_RET(stmtCreateRequest(pStmt));
66!
1236

1237
  if (pStmt->bInfo.needParse) {
66✔
1238
    STMT_ERR_RET(stmtParseSql(pStmt));
20✔
1239
  }
1240

1241
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
63✔
1242
    STMT_ERR_RET(qStmtBindParams(pStmt->sql.pQuery, bind, colIdx));
14!
1243

1244
    SParseContext ctx = {.requestId = pStmt->exec.pRequest->requestId,
14✔
1245
                         .acctId = pStmt->taos->acctId,
14✔
1246
                         .db = pStmt->exec.pRequest->pDb,
14✔
1247
                         .topicQuery = false,
1248
                         .pSql = pStmt->sql.sqlStr,
14✔
1249
                         .sqlLen = pStmt->sql.sqlLen,
14✔
1250
                         .pMsg = pStmt->exec.pRequest->msgBuf,
14✔
1251
                         .msgLen = ERROR_MSG_BUF_DEFAULT_SIZE,
1252
                         .pTransporter = pStmt->taos->pAppInfo->pTransporter,
14✔
1253
                         .pStmtCb = NULL,
1254
                         .pUser = pStmt->taos->user,
14✔
1255
                         .setQueryFp = setQueryRequest};
1256

1257
    ctx.mgmtEpSet = getEpSet_s(&pStmt->taos->pAppInfo->mgmtEp);
14✔
1258
    STMT_ERR_RET(catalogGetHandle(pStmt->taos->pAppInfo->clusterId, &ctx.pCatalog));
14!
1259

1260
    STMT_ERR_RET(qStmtParseQuerySql(&ctx, pStmt->sql.pQuery));
14!
1261

1262
    if (pStmt->sql.pQuery->haveResultSet) {
14!
1263
      STMT_ERR_RET(setResSchemaInfo(&pStmt->exec.pRequest->body.resInfo, pStmt->sql.pQuery->pResSchema,
14!
1264
                                    pStmt->sql.pQuery->numOfResCols));
1265
      taosMemoryFreeClear(pStmt->sql.pQuery->pResSchema);
14!
1266
      setResPrecision(&pStmt->exec.pRequest->body.resInfo, pStmt->sql.pQuery->precision);
14✔
1267
    }
1268

1269
    TSWAP(pStmt->exec.pRequest->dbList, pStmt->sql.pQuery->pDbList);
14✔
1270
    TSWAP(pStmt->exec.pRequest->tableList, pStmt->sql.pQuery->pTableList);
14✔
1271
    TSWAP(pStmt->exec.pRequest->targetTableList, pStmt->sql.pQuery->pTargetTableList);
14✔
1272

1273
    // if (STMT_TYPE_QUERY == pStmt->sql.queryRes) {
1274
    //   STMT_ERR_RET(stmtRestoreQueryFields(pStmt));
1275
    // }
1276

1277
    // STMT_ERR_RET(stmtBackupQueryFields(pStmt));
1278

1279
    return TSDB_CODE_SUCCESS;
14✔
1280
  }
1281

1282
  if (pStmt->sql.stbInterlaceMode && NULL == pStmt->sql.siInfo.pDataCtx) {
49!
1283
    STMT_ERR_RET(stmtInitStbInterlaceTableInfo(pStmt));
×
1284
  }
1285

1286
  STableDataCxt** pDataBlock = NULL;
49✔
1287

1288
  if (pStmt->exec.pCurrBlock) {
49✔
1289
    pDataBlock = &pStmt->exec.pCurrBlock;
25✔
1290
  } else {
1291
    pDataBlock =
1292
        (STableDataCxt**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
24✔
1293
    if (NULL == pDataBlock) {
24!
1294
      tscError("table %s not found in exec blockHash", pStmt->bInfo.tbFName);
×
1295
      STMT_ERR_RET(TSDB_CODE_TSC_STMT_CACHE_ERROR);
×
1296
    }
1297
    pStmt->exec.pCurrBlock = *pDataBlock;
24✔
1298
    if (pStmt->sql.stbInterlaceMode) {
24!
UNCOV
1299
      taosArrayDestroy(pStmt->exec.pCurrBlock->pData->aCol);
×
UNCOV
1300
      pStmt->exec.pCurrBlock->pData->aCol = NULL;
×
1301
    }
1302
  }
1303

1304
  int64_t startUs2 = taosGetTimestampUs();
49✔
1305
  pStmt->stat.bindDataUs1 += startUs2 - startUs;
49✔
1306

1307
  SStmtQNode* param = NULL;
49✔
1308
  if (pStmt->sql.stbInterlaceMode) {
49!
UNCOV
1309
    STMT_ERR_RET(stmtAllocQNodeFromBuf(&pStmt->sql.siInfo.tbBuf, (void**)&param));
×
UNCOV
1310
    STMT_ERR_RET(stmtGetTableColsFromCache(pStmt, &param->tblData.aCol));
×
UNCOV
1311
    taosArrayClear(param->tblData.aCol);
×
1312

1313
    // param->tblData.aCol = taosArrayInit(20, POINTER_BYTES);
1314

UNCOV
1315
    param->restoreTbCols = false;
×
UNCOV
1316
    tstrncpy(param->tblData.tbName, pStmt->bInfo.tbName, TSDB_TABLE_NAME_LEN);
×
1317
  }
1318

1319
  int64_t startUs3 = taosGetTimestampUs();
49✔
1320
  pStmt->stat.bindDataUs2 += startUs3 - startUs2;
49✔
1321

1322
  SArray* pCols = pStmt->sql.stbInterlaceMode ? param->tblData.aCol : (*pDataBlock)->pData->aCol;
49!
1323

1324
  if (colIdx < 0) {
49!
1325
    if (pStmt->sql.stbInterlaceMode) {
49!
UNCOV
1326
      (*pDataBlock)->pData->flags = 0;
×
UNCOV
1327
      code = qBindStmtStbColsValue(*pDataBlock, pCols, bind, pStmt->exec.pRequest->msgBuf,
×
UNCOV
1328
                                   pStmt->exec.pRequest->msgBufLen, &pStmt->sql.siInfo.pTSchema, pStmt->sql.pBindInfo);
×
1329
    } else {
1330
      code =
1331
          qBindStmtColsValue(*pDataBlock, pCols, bind, pStmt->exec.pRequest->msgBuf, pStmt->exec.pRequest->msgBufLen);
49✔
1332
    }
1333

1334
    if (code) {
49✔
1335
      tscError("qBindStmtColsValue failed, error:%s", tstrerror(code));
2!
1336
      STMT_ERR_RET(code);
2!
1337
    }
1338
  } else {
1339
    if (pStmt->sql.stbInterlaceMode) {
×
1340
      tscError("bind single column not allowed in stb insert mode");
×
1341
      STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR);
×
1342
    }
1343

1344
    if (colIdx != (pStmt->bInfo.sBindLastIdx + 1) && colIdx != 0) {
×
1345
      tscError("bind column index not in sequence");
×
1346
      STMT_ERR_RET(TSDB_CODE_APP_ERROR);
×
1347
    }
1348

1349
    pStmt->bInfo.sBindLastIdx = colIdx;
×
1350

1351
    if (0 == colIdx) {
×
1352
      pStmt->bInfo.sBindRowNum = bind->num;
×
1353
    }
1354

1355
    code = qBindStmtSingleColValue(*pDataBlock, pCols, bind, pStmt->exec.pRequest->msgBuf,
×
1356
                                   pStmt->exec.pRequest->msgBufLen, colIdx, pStmt->bInfo.sBindRowNum);
×
1357
    if (code) {
×
1358
      tscError("qBindStmtSingleColValue failed, error:%s", tstrerror(code));
×
1359
      STMT_ERR_RET(code);
×
1360
    }
1361
  }
1362

1363
  int64_t startUs4 = taosGetTimestampUs();
47✔
1364
  pStmt->stat.bindDataUs3 += startUs4 - startUs3;
47✔
1365

1366
  if (pStmt->sql.stbInterlaceMode) {
47!
UNCOV
1367
    STMT_ERR_RET(stmtAppendTablePostHandle(pStmt, param));
×
1368
  }
1369

1370
  pStmt->stat.bindDataUs4 += taosGetTimestampUs() - startUs4;
47✔
1371

1372
  return TSDB_CODE_SUCCESS;
47✔
1373
}
1374

1375
int stmtAddBatch(TAOS_STMT* stmt) {
61✔
1376
  STscStmt* pStmt = (STscStmt*)stmt;
61✔
1377

1378
  int64_t startUs = taosGetTimestampUs();
61✔
1379

1380
  STMT_DLOG_E("start to add batch");
61✔
1381

1382
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
61!
1383
    return pStmt->errCode;
×
1384
  }
1385

1386
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_ADD_BATCH));
61!
1387

1388
  if (pStmt->sql.stbInterlaceMode) {
61!
UNCOV
1389
    int64_t startUs2 = taosGetTimestampUs();
×
UNCOV
1390
    pStmt->stat.addBatchUs += startUs2 - startUs;
×
1391

UNCOV
1392
    pStmt->sql.siInfo.tableColsReady = false;
×
1393

UNCOV
1394
    SStmtQNode* param = NULL;
×
UNCOV
1395
    STMT_ERR_RET(stmtAllocQNodeFromBuf(&pStmt->sql.siInfo.tbBuf, (void**)&param));
×
UNCOV
1396
    param->restoreTbCols = true;
×
UNCOV
1397
    param->next = NULL;
×
1398

UNCOV
1399
    stmtEnqueue(pStmt, param);
×
1400

UNCOV
1401
    return TSDB_CODE_SUCCESS;
×
1402
  }
1403

1404
  STMT_ERR_RET(stmtCacheBlock(pStmt));
61!
1405

1406
  return TSDB_CODE_SUCCESS;
61✔
1407
}
1408

1409
int stmtUpdateTableUid(STscStmt* pStmt, SSubmitRsp* pRsp) {
×
1410
  tscDebug("stmt start to update tbUid, blockNum: %d", pRsp->nBlocks);
×
1411

1412
  int32_t code = 0;
×
1413
  int32_t finalCode = 0;
×
1414
  size_t  keyLen = 0;
×
1415
  void*   pIter = taosHashIterate(pStmt->exec.pBlockHash, NULL);
×
1416
  while (pIter) {
×
1417
    STableDataCxt* pBlock = *(STableDataCxt**)pIter;
×
1418
    char*          key = taosHashGetKey(pIter, &keyLen);
×
1419

1420
    STableMeta* pMeta = qGetTableMetaInDataBlock(pBlock);
×
1421
    if (pMeta->uid) {
×
1422
      pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
×
1423
      continue;
×
1424
    }
1425

1426
    SSubmitBlkRsp* blkRsp = NULL;
×
1427
    int32_t        i = 0;
×
1428
    for (; i < pRsp->nBlocks; ++i) {
×
1429
      blkRsp = pRsp->pBlocks + i;
×
1430
      if (strlen(blkRsp->tblFName) != keyLen) {
×
1431
        continue;
×
1432
      }
1433

1434
      if (strncmp(blkRsp->tblFName, key, keyLen)) {
×
1435
        continue;
×
1436
      }
1437

1438
      break;
×
1439
    }
1440

1441
    if (i < pRsp->nBlocks) {
×
1442
      tscDebug("auto created table %s uid updated from %" PRIx64 " to %" PRIx64, blkRsp->tblFName, pMeta->uid,
×
1443
               blkRsp->uid);
1444

1445
      pMeta->uid = blkRsp->uid;
×
1446
      pStmt->bInfo.tbUid = blkRsp->uid;
×
1447
    } else {
1448
      tscDebug("table %s not found in submit rsp, will update from catalog", pStmt->bInfo.tbFName);
×
1449
      if (NULL == pStmt->pCatalog) {
×
1450
        code = catalogGetHandle(pStmt->taos->pAppInfo->clusterId, &pStmt->pCatalog);
×
1451
        if (code) {
×
1452
          pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
×
1453
          finalCode = code;
×
1454
          continue;
×
1455
        }
1456
      }
1457

1458
      code = stmtCreateRequest(pStmt);
×
1459
      if (code) {
×
1460
        pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
×
1461
        finalCode = code;
×
1462
        continue;
×
1463
      }
1464

1465
      STableMeta*      pTableMeta = NULL;
×
1466
      SRequestConnInfo conn = {.pTrans = pStmt->taos->pAppInfo->pTransporter,
×
1467
                               .requestId = pStmt->exec.pRequest->requestId,
×
1468
                               .requestObjRefId = pStmt->exec.pRequest->self,
×
1469
                               .mgmtEps = getEpSet_s(&pStmt->taos->pAppInfo->mgmtEp)};
×
1470
      code = catalogGetTableMeta(pStmt->pCatalog, &conn, &pStmt->bInfo.sname, &pTableMeta);
×
1471

1472
      pStmt->stat.ctgGetTbMetaNum++;
×
1473

1474
      taos_free_result(pStmt->exec.pRequest);
×
1475
      pStmt->exec.pRequest = NULL;
×
1476

1477
      if (code || NULL == pTableMeta) {
×
1478
        pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
×
1479
        finalCode = code;
×
1480
        taosMemoryFree(pTableMeta);
×
1481
        continue;
×
1482
      }
1483

1484
      pMeta->uid = pTableMeta->uid;
×
1485
      pStmt->bInfo.tbUid = pTableMeta->uid;
×
1486
      taosMemoryFree(pTableMeta);
×
1487
    }
1488

1489
    pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
×
1490
  }
1491

1492
  return finalCode;
×
1493
}
1494

1495
/*
1496
int stmtStaticModeExec(TAOS_STMT* stmt) {
1497
  STscStmt*   pStmt = (STscStmt*)stmt;
1498
  int32_t     code = 0;
1499
  SSubmitRsp* pRsp = NULL;
1500
  if (pStmt->sql.staticMode) {
1501
    return TSDB_CODE_TSC_STMT_API_ERROR;
1502
  }
1503

1504
  STMT_DLOG_E("start to exec");
1505

1506
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_EXECUTE));
1507

1508
  STMT_ERR_RET(qBuildStmtOutputFromTbList(pStmt->sql.pQuery, pStmt->sql.pVgHash, pStmt->exec.pTbBlkList,
1509
pStmt->exec.pCurrBlock, pStmt->exec.tbBlkNum));
1510

1511
  launchQueryImpl(pStmt->exec.pRequest, pStmt->sql.pQuery, true, NULL);
1512

1513
  if (pStmt->exec.pRequest->code && NEED_CLIENT_HANDLE_ERROR(pStmt->exec.pRequest->code)) {
1514
    code = refreshMeta(pStmt->exec.pRequest->pTscObj, pStmt->exec.pRequest);
1515
    if (code) {
1516
      pStmt->exec.pRequest->code = code;
1517
    } else {
1518
      tFreeSSubmitRsp(pRsp);
1519
      STMT_ERR_RET(stmtResetStmt(pStmt));
1520
      STMT_ERR_RET(TSDB_CODE_NEED_RETRY);
1521
    }
1522
  }
1523

1524
  STMT_ERR_JRET(pStmt->exec.pRequest->code);
1525

1526
  pStmt->exec.affectedRows = taos_affected_rows(pStmt->exec.pRequest);
1527
  pStmt->affectedRows += pStmt->exec.affectedRows;
1528

1529
_return:
1530

1531
  stmtCleanExecInfo(pStmt, (code ? false : true), false);
1532

1533
  tFreeSSubmitRsp(pRsp);
1534

1535
  ++pStmt->sql.runTimes;
1536

1537
  STMT_RET(code);
1538
}
1539
*/
1540

1541
int stmtExec(TAOS_STMT* stmt) {
61✔
1542
  STscStmt*   pStmt = (STscStmt*)stmt;
61✔
1543
  int32_t     code = 0;
61✔
1544
  SSubmitRsp* pRsp = NULL;
61✔
1545

1546
  int64_t startUs = taosGetTimestampUs();
61✔
1547

1548
  STMT_DLOG_E("start to exec");
61✔
1549

1550
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
61!
1551
    return pStmt->errCode;
×
1552
  }
1553

1554
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_EXECUTE));
61!
1555

1556
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
61✔
1557
    launchQueryImpl(pStmt->exec.pRequest, pStmt->sql.pQuery, true, NULL);
14✔
1558
  } else {
1559
    if (pStmt->sql.stbInterlaceMode) {
47!
UNCOV
1560
      int64_t startTs = taosGetTimestampUs();
×
UNCOV
1561
      while (atomic_load_64(&pStmt->sql.siInfo.tbRemainNum)) {
×
UNCOV
1562
        taosUsleep(1);
×
1563
      }
UNCOV
1564
      pStmt->stat.execWaitUs += taosGetTimestampUs() - startTs;
×
1565

UNCOV
1566
      STMT_ERR_RET(qBuildStmtFinOutput(pStmt->sql.pQuery, pStmt->sql.pVgHash, pStmt->sql.siInfo.pVgroupList));
×
UNCOV
1567
      taosHashCleanup(pStmt->sql.siInfo.pVgroupHash);
×
UNCOV
1568
      pStmt->sql.siInfo.pVgroupHash = NULL;
×
UNCOV
1569
      pStmt->sql.siInfo.pVgroupList = NULL;
×
1570
    } else {
1571
      tDestroySubmitTbData(pStmt->exec.pCurrTbData, TSDB_MSG_FLG_ENCODE);
47✔
1572
      taosMemoryFreeClear(pStmt->exec.pCurrTbData);
47✔
1573

1574
      STMT_ERR_RET(qCloneCurrentTbData(pStmt->exec.pCurrBlock, &pStmt->exec.pCurrTbData));
47!
1575

1576
      STMT_ERR_RET(qBuildStmtOutput(pStmt->sql.pQuery, pStmt->sql.pVgHash, pStmt->exec.pBlockHash));
47!
1577
    }
1578

1579
    launchQueryImpl(pStmt->exec.pRequest, pStmt->sql.pQuery, true, NULL);
47✔
1580
  }
1581

1582
  if (pStmt->exec.pRequest->code && NEED_CLIENT_HANDLE_ERROR(pStmt->exec.pRequest->code)) {
61!
1583
    code = refreshMeta(pStmt->exec.pRequest->pTscObj, pStmt->exec.pRequest);
×
1584
    if (code) {
×
1585
      pStmt->exec.pRequest->code = code;
×
1586
    } else {
1587
      tFreeSSubmitRsp(pRsp);
×
1588
      STMT_ERR_RET(stmtResetStmt(pStmt));
×
1589
      STMT_ERR_RET(TSDB_CODE_NEED_RETRY);
×
1590
    }
1591
  }
1592

1593
  STMT_ERR_JRET(pStmt->exec.pRequest->code);
61✔
1594

1595
  pStmt->exec.affectedRows = taos_affected_rows(pStmt->exec.pRequest);
59✔
1596
  pStmt->affectedRows += pStmt->exec.affectedRows;
59✔
1597

1598
_return:
61✔
1599

1600
  while (0 == atomic_load_8((int8_t*)&pStmt->sql.siInfo.tableColsReady)) {
61!
1601
    taosUsleep(1);
×
1602
  }
1603

1604
  STMT_ERR_RET(stmtCleanExecInfo(pStmt, (code ? false : true), false));
61!
1605

1606
  tFreeSSubmitRsp(pRsp);
61✔
1607

1608
  ++pStmt->sql.runTimes;
61✔
1609

1610
  int64_t startUs2 = taosGetTimestampUs();
61✔
1611
  pStmt->stat.execUseUs += startUs2 - startUs;
61✔
1612

1613
  STMT_RET(code);
61✔
1614
}
1615

1616
int stmtClose(TAOS_STMT* stmt) {
35✔
1617
  STscStmt* pStmt = (STscStmt*)stmt;
35✔
1618

1619
  STMT_DLOG_E("start to free stmt");
35✔
1620

1621
  pStmt->queue.stopQueue = true;
35✔
1622

1623
  if (pStmt->bindThreadInUse) {
35!
UNCOV
1624
    (void)taosThreadJoin(pStmt->bindThread, NULL);
×
UNCOV
1625
    pStmt->bindThreadInUse = false;
×
1626
  }
1627

1628
  STMT_DLOG("stmt %p closed, stbInterlaceMode: %d, statInfo: ctgGetTbMetaNum=>%" PRId64 ", getCacheTbInfo=>%" PRId64
35✔
1629
            ", parseSqlNum=>%" PRId64 ", pStmt->stat.bindDataNum=>%" PRId64
1630
            ", settbnameAPI:%u, bindAPI:%u, addbatchAPI:%u, execAPI:%u"
1631
            ", setTbNameUs:%" PRId64 ", bindDataUs:%" PRId64 ",%" PRId64 ",%" PRId64 ",%" PRId64 " addBatchUs:%" PRId64
1632
            ", execWaitUs:%" PRId64 ", execUseUs:%" PRId64,
1633
            pStmt, pStmt->sql.stbInterlaceMode, pStmt->stat.ctgGetTbMetaNum, pStmt->stat.getCacheTbInfo,
1634
            pStmt->stat.parseSqlNum, pStmt->stat.bindDataNum, pStmt->seqIds[STMT_SETTBNAME], pStmt->seqIds[STMT_BIND],
1635
            pStmt->seqIds[STMT_ADD_BATCH], pStmt->seqIds[STMT_EXECUTE], pStmt->stat.setTbNameUs,
1636
            pStmt->stat.bindDataUs1, pStmt->stat.bindDataUs2, pStmt->stat.bindDataUs3, pStmt->stat.bindDataUs4,
1637
            pStmt->stat.addBatchUs, pStmt->stat.execWaitUs, pStmt->stat.execUseUs);
1638

1639
  STMT_ERR_RET(stmtCleanSQLInfo(pStmt));
35!
1640
  taosMemoryFree(stmt);
35✔
1641

1642
  return TSDB_CODE_SUCCESS;
35✔
1643
}
1644

1645
const char* stmtErrstr(TAOS_STMT* stmt) {
9✔
1646
  STscStmt* pStmt = (STscStmt*)stmt;
9✔
1647

1648
  if (stmt == NULL || NULL == pStmt->exec.pRequest) {
9!
1649
    return (char*)tstrerror(terrno);
2✔
1650
  }
1651

1652
  pStmt->exec.pRequest->code = terrno;
7✔
1653

1654
  return taos_errstr(pStmt->exec.pRequest);
7✔
1655
}
1656

1657
int stmtAffectedRows(TAOS_STMT* stmt) { return ((STscStmt*)stmt)->affectedRows; }
5✔
1658

1659
int stmtAffectedRowsOnce(TAOS_STMT* stmt) { return ((STscStmt*)stmt)->exec.affectedRows; }
×
1660

1661
int stmtIsInsert(TAOS_STMT* stmt, int* insert) {
77✔
1662
  STscStmt* pStmt = (STscStmt*)stmt;
77✔
1663

1664
  STMT_DLOG_E("start is insert");
77✔
1665

1666
  if (pStmt->sql.type) {
77✔
1667
    *insert = (STMT_TYPE_INSERT == pStmt->sql.type || STMT_TYPE_MULTI_INSERT == pStmt->sql.type);
54!
1668
  } else {
1669
    *insert = qIsInsertValuesSql(pStmt->sql.sqlStr, pStmt->sql.sqlLen);
23✔
1670
  }
1671

1672
  return TSDB_CODE_SUCCESS;
77✔
1673
}
1674

1675
int stmtGetTagFields(TAOS_STMT* stmt, int* nums, TAOS_FIELD_E** fields) {
×
1676
  int32_t   code = 0;
×
1677
  STscStmt* pStmt = (STscStmt*)stmt;
×
1678
  int32_t   preCode = pStmt->errCode;
×
1679

1680
  STMT_DLOG_E("start to get tag fields");
×
1681

1682
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
×
1683
    return pStmt->errCode;
×
1684
  }
1685

1686
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
×
1687
    STMT_ERRI_JRET(TSDB_CODE_TSC_STMT_API_ERROR);
×
1688
  }
1689

1690
  STMT_ERRI_JRET(stmtSwitchStatus(pStmt, STMT_FETCH_FIELDS));
×
1691

1692
  if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 &&
×
1693
      STMT_TYPE_MULTI_INSERT != pStmt->sql.type) {
×
1694
    pStmt->bInfo.needParse = false;
×
1695
  }
1696

1697
  if (pStmt->exec.pRequest && STMT_TYPE_QUERY == pStmt->sql.type && pStmt->sql.runTimes) {
×
1698
    taos_free_result(pStmt->exec.pRequest);
×
1699
    pStmt->exec.pRequest = NULL;
×
1700
  }
1701

1702
  STMT_ERRI_JRET(stmtCreateRequest(pStmt));
×
1703

1704
  if (pStmt->bInfo.needParse) {
×
1705
    STMT_ERRI_JRET(stmtParseSql(pStmt));
×
1706
  }
1707

1708
  STMT_ERRI_JRET(stmtFetchTagFields(stmt, nums, fields));
×
1709

1710
_return:
×
1711

1712
  pStmt->errCode = preCode;
×
1713

1714
  return code;
×
1715
}
1716

1717
int stmtGetColFields(TAOS_STMT* stmt, int* nums, TAOS_FIELD_E** fields) {
×
1718
  int32_t   code = 0;
×
1719
  STscStmt* pStmt = (STscStmt*)stmt;
×
1720
  int32_t   preCode = pStmt->errCode;
×
1721

1722
  STMT_DLOG_E("start to get col fields");
×
1723

1724
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
×
1725
    return pStmt->errCode;
×
1726
  }
1727

1728
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
×
1729
    STMT_ERRI_JRET(TSDB_CODE_TSC_STMT_API_ERROR);
×
1730
  }
1731

1732
  STMT_ERRI_JRET(stmtSwitchStatus(pStmt, STMT_FETCH_FIELDS));
×
1733

1734
  if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 &&
×
1735
      STMT_TYPE_MULTI_INSERT != pStmt->sql.type) {
×
1736
    pStmt->bInfo.needParse = false;
×
1737
  }
1738

1739
  if (pStmt->exec.pRequest && STMT_TYPE_QUERY == pStmt->sql.type && pStmt->sql.runTimes) {
×
1740
    taos_free_result(pStmt->exec.pRequest);
×
1741
    pStmt->exec.pRequest = NULL;
×
1742
    STMT_ERR_RET(stmtCreateRequest(pStmt));
×
1743
  }
1744

1745
  STMT_ERRI_JRET(stmtCreateRequest(pStmt));
×
1746

1747
  if (pStmt->bInfo.needParse) {
×
1748
    STMT_ERRI_JRET(stmtParseSql(pStmt));
×
1749
  }
1750

1751
  STMT_ERRI_JRET(stmtFetchColFields(stmt, nums, fields));
×
1752

1753
_return:
×
1754

1755
  pStmt->errCode = preCode;
×
1756

1757
  return code;
×
1758
}
1759

1760
int stmtGetParamNum(TAOS_STMT* stmt, int* nums) {
×
1761
  STscStmt* pStmt = (STscStmt*)stmt;
×
1762

1763
  STMT_DLOG_E("start to get param num");
×
1764

1765
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
×
1766
    return pStmt->errCode;
×
1767
  }
1768

1769
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_FETCH_FIELDS));
×
1770

1771
  if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 &&
×
1772
      STMT_TYPE_MULTI_INSERT != pStmt->sql.type) {
×
1773
    pStmt->bInfo.needParse = false;
×
1774
  }
1775

1776
  if (pStmt->exec.pRequest && STMT_TYPE_QUERY == pStmt->sql.type && pStmt->sql.runTimes) {
×
1777
    taos_free_result(pStmt->exec.pRequest);
×
1778
    pStmt->exec.pRequest = NULL;
×
1779
  }
1780

1781
  STMT_ERR_RET(stmtCreateRequest(pStmt));
×
1782

1783
  if (pStmt->bInfo.needParse) {
×
1784
    STMT_ERR_RET(stmtParseSql(pStmt));
×
1785
  }
1786

1787
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
×
1788
    *nums = taosArrayGetSize(pStmt->sql.pQuery->pPlaceholderValues);
×
1789
  } else {
1790
    STMT_ERR_RET(stmtFetchColFields(stmt, nums, NULL));
×
1791
  }
1792

1793
  return TSDB_CODE_SUCCESS;
×
1794
}
1795

1796
int stmtGetParam(TAOS_STMT* stmt, int idx, int* type, int* bytes) {
×
1797
  STscStmt* pStmt = (STscStmt*)stmt;
×
1798

1799
  STMT_DLOG_E("start to get param");
×
1800

1801
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
×
1802
    return pStmt->errCode;
×
1803
  }
1804

1805
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
×
1806
    STMT_RET(TSDB_CODE_TSC_STMT_API_ERROR);
×
1807
  }
1808

1809
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_FETCH_FIELDS));
×
1810

1811
  if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 &&
×
1812
      STMT_TYPE_MULTI_INSERT != pStmt->sql.type) {
×
1813
    pStmt->bInfo.needParse = false;
×
1814
  }
1815

1816
  if (pStmt->exec.pRequest && STMT_TYPE_QUERY == pStmt->sql.type && pStmt->sql.runTimes) {
×
1817
    taos_free_result(pStmt->exec.pRequest);
×
1818
    pStmt->exec.pRequest = NULL;
×
1819
  }
1820

1821
  STMT_ERR_RET(stmtCreateRequest(pStmt));
×
1822

1823
  if (pStmt->bInfo.needParse) {
×
1824
    STMT_ERR_RET(stmtParseSql(pStmt));
×
1825
  }
1826

1827
  int32_t       nums = 0;
×
1828
  TAOS_FIELD_E* pField = NULL;
×
1829
  STMT_ERR_RET(stmtFetchColFields(stmt, &nums, &pField));
×
1830
  if (idx >= nums) {
×
1831
    tscError("idx %d is too big", idx);
×
1832
    taosMemoryFree(pField);
×
1833
    STMT_ERR_RET(TSDB_CODE_INVALID_PARA);
×
1834
  }
1835

1836
  *type = pField[idx].type;
×
1837
  *bytes = pField[idx].bytes;
×
1838

1839
  taosMemoryFree(pField);
×
1840

1841
  return TSDB_CODE_SUCCESS;
×
1842
}
1843

1844
TAOS_RES* stmtUseResult(TAOS_STMT* stmt) {
14✔
1845
  STscStmt* pStmt = (STscStmt*)stmt;
14✔
1846

1847
  STMT_DLOG_E("start to use result");
14!
1848

1849
  if (STMT_TYPE_QUERY != pStmt->sql.type) {
14!
1850
    tscError("useResult only for query statement");
×
1851
    return NULL;
×
1852
  }
1853

1854
  return pStmt->exec.pRequest;
14✔
1855
}
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