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

taosdata / TDengine / #3525

10 Nov 2024 03:50AM UTC coverage: 60.818% (-0.08%) from 60.898%
#3525

push

travis-ci

web-flow
Merge pull request #28709 from taosdata/main

merge: from main to 3.0 branch

118634 of 249004 branches covered (47.64%)

Branch coverage included in aggregate %.

136 of 169 new or added lines in 23 files covered. (80.47%)

542 existing lines in 129 files now uncovered.

199071 of 273386 relevant lines covered (72.82%)

15691647.46 hits per line

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

46.08
/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) {
12
  if (pTblBuf->buffOffset < pTblBuf->buffSize) {
35✔
13
    *pBuf = (char*)pTblBuf->pCurBuff + pTblBuf->buffOffset;
35✔
14
    pTblBuf->buffOffset += pTblBuf->buffUnit;
35✔
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

38
  return TSDB_CODE_SUCCESS;
35✔
39
}
40

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

47
  SStmtQNode* orig = pStmt->queue.head;
31✔
48

49
  SStmtQNode* node = pStmt->queue.head->next;
31✔
50
  pStmt->queue.head = pStmt->queue.head->next;
31✔
51

52
  // taosMemoryFreeClear(orig);
53

54
  *param = node;
31✔
55

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

58
  return true;
32✔
59
}
60

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

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

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

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

84
  return code;
121✔
85
}
86

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

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

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

99
  switch (newStatus) {
264!
100
    case STMT_PREPARE:
37✔
101
      pStmt->errCode = 0;
37✔
102
      break;
37✔
103
    case STMT_SETTBNAME:
24✔
104
      if (STMT_STATUS_EQ(INIT)) {
24!
105
        code = TSDB_CODE_TSC_STMT_API_ERROR;
×
106
      }
107
      if (!pStmt->sql.stbInterlaceMode && (STMT_STATUS_EQ(BIND) || STMT_STATUS_EQ(BIND_COL))) {
24!
108
        code = TSDB_CODE_TSC_STMT_API_ERROR;
×
109
      }
110
      break;
24✔
111
    case STMT_SETTAGS:
8✔
112
      if (STMT_STATUS_NE(SETTBNAME) && STMT_STATUS_NE(FETCH_FIELDS)) {
8!
113
        code = TSDB_CODE_TSC_STMT_API_ERROR;
×
114
      }
115
      break;
8✔
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:
75✔
122
      if (STMT_STATUS_EQ(INIT) || STMT_STATUS_EQ(BIND_COL)) {
75!
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;
75✔
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:
63✔
137
      if (STMT_STATUS_NE(BIND) && STMT_STATUS_NE(BIND_COL) && STMT_STATUS_NE(FETCH_FIELDS)) {
63!
138
        code = TSDB_CODE_TSC_STMT_API_ERROR;
×
139
      }
140
      break;
63✔
141
    case STMT_EXECUTE:
57✔
142
      if (STMT_TYPE_QUERY == pStmt->sql.type) {
57✔
143
        if (STMT_STATUS_NE(ADD_BATCH) && STMT_STATUS_NE(FETCH_FIELDS) && STMT_STATUS_NE(BIND) &&
16!
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)) {
41!
149
          code = TSDB_CODE_TSC_STMT_API_ERROR;
×
150
        }
151
      }
152
      break;
57✔
153
    default:
×
154
      code = TSDB_CODE_APP_ERROR;
×
155
      break;
×
156
  }
157

158
  STMT_ERR_RET(code);
264!
159

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

162
  return TSDB_CODE_SUCCESS;
264✔
163
}
164

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

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

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

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

177
  return TSDB_CODE_SUCCESS;
12✔
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,
16✔
230
                           bool autoCreateTbl) {
231
  STscStmt* pStmt = (STscStmt*)stmt;
16✔
232
  char      tbFName[TSDB_TABLE_FNAME_LEN];
233
  int32_t   code = tNameExtractFullName(tbName, tbFName);
16✔
234
  if (code != 0) {
16!
235
    return code;
×
236
  }
237

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

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

250
  return TSDB_CODE_SUCCESS;
16✔
251
}
252

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

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

259
  return TSDB_CODE_SUCCESS;
16✔
260
}
261

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

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

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

274
  return TSDB_CODE_SUCCESS;
16✔
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) {
48✔
290
  if (pStmt->sql.type != STMT_TYPE_MULTI_INSERT) {
48✔
291
    return TSDB_CODE_SUCCESS;
28✔
292
  }
293

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

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

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

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

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

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

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

324
  return TSDB_CODE_SUCCESS;
8✔
325
}
326

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

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

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

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

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

345
  if (pStmt->sql.pQuery->pRoot && 0 == pStmt->sql.type) {
32✔
346
    pStmt->sql.type = STMT_TYPE_INSERT;
4✔
347
    pStmt->sql.stbInterlaceMode = false;
4✔
348
  } else if (pStmt->sql.pQuery->pPrepareRoot) {
28✔
349
    pStmt->sql.type = STMT_TYPE_QUERY;
16✔
350
    pStmt->sql.stbInterlaceMode = false;
16✔
351

352
    return TSDB_CODE_SUCCESS;
16✔
353
  }
354

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

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

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

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

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

382
  return TSDB_CODE_SUCCESS;
16✔
383
}
384

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

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

401
  return TSDB_CODE_SUCCESS;
91✔
402
}
403

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

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

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

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

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

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

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

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

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

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

464
    if (keepTable) {
74✔
465
      return TSDB_CODE_SUCCESS;
40✔
466
    }
467

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

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

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

477
  return TSDB_CODE_SUCCESS;
54✔
478
}
479

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

485
void stmtFreeTbCols(void* buf) {
4,000✔
486
  SArray* pCols = *(SArray**)buf;
4,000✔
487
  taosArrayDestroy(pCols);
4,000✔
488
}
4,000✔
489

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

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

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

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

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

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

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

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

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

531
  return TSDB_CODE_SUCCESS;
37✔
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) {
11✔
572
  if (pStmt->sql.stbInterlaceMode && pStmt->sql.siInfo.pDataCtx) {
11!
573
    pStmt->bInfo.needParse = false;
×
574
    pStmt->bInfo.inExecCache = false;
×
575
    return TSDB_CODE_SUCCESS;
×
576
  }
577

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

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

UNCOV
586
    pStmt->exec.pCurrBlock = *pCxtInExec;
×
587

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

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

599
  if (NULL == pStmt->sql.pTableCache || taosHashGetSize(pStmt->sql.pTableCache) <= 0) {
12!
600
    if (pStmt->bInfo.inExecCache) {
12!
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);
12!
607
    return TSDB_CODE_SUCCESS;
12✔
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

UNCOV
726
int32_t stmtResetStmt(STscStmt* pStmt) {
×
UNCOV
727
  STMT_ERR_RET(stmtCleanSQLInfo(pStmt));
×
728

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

UNCOV
734
  pStmt->sql.status = STMT_INIT;
×
735

UNCOV
736
  return TSDB_CODE_SUCCESS;
×
737
}
738

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

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

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

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

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

763
void* stmtBindThreadFunc(void* param) {
4✔
764
  setThreadName("stmtBind");
4✔
765

766
  qInfo("stmt bind thread started");
4!
767

768
  STscStmt* pStmt = (STscStmt*)param;
4✔
769

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

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

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

786
  qInfo("stmt bind thread stopped");
4!
787

788
  return NULL;
4✔
789
}
790

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

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

805
  pStmt->bindThreadInUse = true;
4✔
806

807
  (void)taosThreadAttrDestroy(&thAttr);
4✔
808
  return TSDB_CODE_SUCCESS;
4✔
809
}
810

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

815
  return TSDB_CODE_SUCCESS;
4✔
816
}
817

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

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

834
  pTblBuf->pCurBuff = buff;
4✔
835
  pTblBuf->buffIdx = 1;
4✔
836
  pTblBuf->buffOffset = 0;
4✔
837

838
  return TSDB_CODE_SUCCESS;
4✔
839
}
840

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

846
  pStmt = taosMemoryCalloc(1, sizeof(STscStmt));
37✔
847
  if (NULL == pStmt) {
37!
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);
37✔
853
  if (NULL == pStmt->sql.pTableCache) {
37!
854
    terrno = TSDB_CODE_OUT_OF_MEMORY;
×
855
    taosMemoryFree(pStmt);
×
856
    return NULL;
×
857
  }
858

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

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

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

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

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

906
  STMT_LOG_SEQ(STMT_INIT);
37✔
907

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

910
  return pStmt;
37✔
911
}
912

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

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

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

922
  if (pStmt->sql.status >= STMT_PREPARE) {
37!
UNCOV
923
    STMT_ERR_RET(stmtResetStmt(pStmt));
×
924
  }
925

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

928
  if (length <= 0) {
37✔
929
    length = strlen(sql);
3✔
930
  }
931

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

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

945
  return TSDB_CODE_SUCCESS;
37✔
946
}
947

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

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

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

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

970
  pStmt->sql.siInfo.boundTags = pStmt->bInfo.boundTags;
11✔
971

972
  return TSDB_CODE_SUCCESS;
11✔
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) {
24✔
992
  STscStmt* pStmt = (STscStmt*)stmt;
24✔
993

994
  int64_t startUs = taosGetTimestampUs();
24✔
995

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

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

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

1004
  int32_t insert = 0;
24✔
1005
  STMT_ERR_RET(stmtIsInsert(stmt, &insert));
24!
1006
  if (0 == insert) {
24!
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) {
24✔
1012
    STMT_ERR_RET(stmtCreateRequest(pStmt));
12!
1013

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

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

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

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

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

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

1040
  return TSDB_CODE_SUCCESS;
24✔
1041
}
1042

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

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

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

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

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

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

1064
  STableDataCxt** pDataBlock =
1065
      (STableDataCxt**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
8✔
1066
  if (NULL == pDataBlock) {
8!
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");
8!
1072
  STMT_ERR_RET(qBindStmtTagsValue(*pDataBlock, pStmt->bInfo.boundTags, pStmt->bInfo.tbSuid, pStmt->bInfo.stbFName,
8!
1073
                                  pStmt->bInfo.sname.tname, tags, pStmt->exec.pRequest->msgBuf,
1074
                                  pStmt->exec.pRequest->msgBufLen));
1075

1076
  return TSDB_CODE_SUCCESS;
8✔
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

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

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

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

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

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

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

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

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

1183
  stmtEnqueue(pStmt, param);
16✔
1184

1185
  return TSDB_CODE_SUCCESS;
16✔
1186
}
1187

1188
static FORCE_INLINE int32_t stmtGetTableColsFromCache(STscStmt* pStmt, SArray** pTableCols) {
1189
  while (true) {
1190
    if (pStmt->sql.siInfo.pTableColsIdx < taosArrayGetSize(pStmt->sql.siInfo.pTableCols)) {
16!
1191
      *pTableCols = (SArray*)taosArrayGetP(pStmt->sql.siInfo.pTableCols, pStmt->sql.siInfo.pTableColsIdx++);
16✔
1192
      break;
16✔
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

1208
  return TSDB_CODE_SUCCESS;
16✔
1209
}
1210

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

1215
  int64_t startUs = taosGetTimestampUs();
75✔
1216

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

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

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

1225
  if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 &&
75!
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) {
75!
1231
    taos_free_result(pStmt->exec.pRequest);
×
1232
    pStmt->exec.pRequest = NULL;
×
1233
  }
1234

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

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

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

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

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

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

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

1269
    TSWAP(pStmt->exec.pRequest->dbList, pStmt->sql.pQuery->pDbList);
16✔
1270
    TSWAP(pStmt->exec.pRequest->tableList, pStmt->sql.pQuery->pTableList);
16✔
1271
    TSWAP(pStmt->exec.pRequest->targetTableList, pStmt->sql.pQuery->pTargetTableList);
16✔
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;
16✔
1280
  }
1281

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

1286
  STableDataCxt** pDataBlock = NULL;
56✔
1287

1288
  if (pStmt->exec.pCurrBlock) {
56✔
1289
    pDataBlock = &pStmt->exec.pCurrBlock;
40✔
1290
  } else {
1291
    pDataBlock =
1292
        (STableDataCxt**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
16✔
1293
    if (NULL == pDataBlock) {
16!
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;
16✔
1298
    if (pStmt->sql.stbInterlaceMode) {
16✔
1299
      taosArrayDestroy(pStmt->exec.pCurrBlock->pData->aCol);
4✔
1300
      pStmt->exec.pCurrBlock->pData->aCol = NULL;
4✔
1301
    }
1302
  }
1303

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

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

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

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

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

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

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

1334
    if (code) {
56✔
1335
      tscError("qBindStmtColsValue failed, error:%s", tstrerror(code));
6!
1336
      STMT_ERR_RET(code);
6!
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();
50✔
1364
  pStmt->stat.bindDataUs3 += startUs4 - startUs3;
50✔
1365

1366
  if (pStmt->sql.stbInterlaceMode) {
50✔
1367
    STMT_ERR_RET(stmtAppendTablePostHandle(pStmt, param));
16!
1368
  }
1369

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

1372
  return TSDB_CODE_SUCCESS;
50✔
1373
}
1374

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

1378
  int64_t startUs = taosGetTimestampUs();
63✔
1379

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

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

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

1388
  if (pStmt->sql.stbInterlaceMode) {
63✔
1389
    int64_t startUs2 = taosGetTimestampUs();
15✔
1390
    pStmt->stat.addBatchUs += startUs2 - startUs;
15✔
1391

1392
    pStmt->sql.siInfo.tableColsReady = false;
15✔
1393

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

1399
    stmtEnqueue(pStmt, param);
15✔
1400

1401
    return TSDB_CODE_SUCCESS;
16✔
1402
  }
1403

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

1406
  return TSDB_CODE_SUCCESS;
48✔
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) {
57✔
1542
  STscStmt*   pStmt = (STscStmt*)stmt;
57✔
1543
  int32_t     code = 0;
57✔
1544
  SSubmitRsp* pRsp = NULL;
57✔
1545

1546
  int64_t startUs = taosGetTimestampUs();
57✔
1547

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

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

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

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

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

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

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

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

1582
  if (pStmt->exec.pRequest->code && NEED_CLIENT_HANDLE_ERROR(pStmt->exec.pRequest->code)) {
57!
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);
57✔
1594

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

1598
_return:
57✔
1599

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

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

1606
  tFreeSSubmitRsp(pRsp);
57✔
1607

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

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

1613
  STMT_RET(code);
57✔
1614
}
1615

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

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

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

1623
  if (pStmt->bindThreadInUse) {
37✔
1624
    (void)taosThreadJoin(pStmt->bindThread, NULL);
4✔
1625
    pStmt->bindThreadInUse = false;
4✔
1626
  }
1627

1628
  STMT_DLOG("stmt %p closed, stbInterlaceMode: %d, statInfo: ctgGetTbMetaNum=>%" PRId64 ", getCacheTbInfo=>%" PRId64
37✔
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));
37!
1640
  taosMemoryFree(stmt);
37✔
1641

1642
  return TSDB_CODE_SUCCESS;
37✔
1643
}
1644

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

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

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

1654
  return taos_errstr(pStmt->exec.pRequest);
15✔
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) {
51✔
1662
  STscStmt* pStmt = (STscStmt*)stmt;
51✔
1663

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

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

1672
  return TSDB_CODE_SUCCESS;
51✔
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) {
16✔
1845
  STscStmt* pStmt = (STscStmt*)stmt;
16✔
1846

1847
  STMT_DLOG_E("start to use result");
16✔
1848

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

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