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

taosdata / TDengine / #3562

20 Dec 2024 09:57AM UTC coverage: 26.655% (-32.2%) from 58.812%
#3562

push

travis-ci

web-flow
Merge pull request #29229 from taosdata/enh/TS-5749-3.0

enh: seperate tsdb async tasks to different thread pools

21498 of 109421 branches covered (19.65%)

Branch coverage included in aggregate %.

66 of 96 new or added lines in 7 files covered. (68.75%)

39441 existing lines in 157 files now uncovered.

35007 of 102566 relevant lines covered (34.13%)

53922.97 hits per line

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

39.17
/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) {
36✔
13
    *pBuf = (char*)pTblBuf->pCurBuff + pTblBuf->buffOffset;
36✔
14
    pTblBuf->buffOffset += pTblBuf->buffUnit;
36✔
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;
36✔
39
}
40

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

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

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

52
  // taosMemoryFreeClear(orig);
53

54
  *param = node;
29✔
55

56
  (void)atomic_sub_fetch_64(&pStmt->queue.qRemainNum, 1);
29✔
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) {
102✔
70
  int32_t code = 0;
102✔
71

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

84
  return code;
102✔
85
}
86

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

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

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

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

158
  STMT_ERR_RET(code);
211!
159

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

162
  return TSDB_CODE_SUCCESS;
211✔
163
}
164

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

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

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

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

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

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

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

250
  return TSDB_CODE_SUCCESS;
20✔
251
}
252

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

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

259
  return TSDB_CODE_SUCCESS;
19✔
260
}
261

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

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

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

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

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

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

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

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

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

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

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

324
  return TSDB_CODE_SUCCESS;
16✔
325
}
326

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

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

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

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

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

345
  if (pStmt->sql.pQuery->pRoot && 0 == pStmt->sql.type) {
19!
UNCOV
346
    pStmt->sql.type = STMT_TYPE_INSERT;
×
UNCOV
347
    pStmt->sql.stbInterlaceMode = false;
×
348
  } else if (pStmt->sql.pQuery->pPrepareRoot) {
19!
349
    pStmt->sql.type = STMT_TYPE_QUERY;
×
350
    pStmt->sql.stbInterlaceMode = false;
×
351

352
    return TSDB_CODE_SUCCESS;
×
353
  }
354

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

361
  STableDataCxt* pTableCtx = *pSrc;
19✔
362
  if (pStmt->sql.stbInterlaceMode) {
19✔
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) {
19!
376
    pStmt->sql.pBindInfo = taosMemoryMalloc(pTableCtx->boundColsInfo.numOfBound * sizeof(*pStmt->sql.pBindInfo));
19!
377
    if (NULL == pStmt->sql.pBindInfo) {
20!
378
      return terrno;
×
379
    }
380
  }
381

382
  return TSDB_CODE_SUCCESS;
20✔
383
}
384

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

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

401
  return TSDB_CODE_SUCCESS;
56✔
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) {
68✔
424
  if (pStmt->sql.stbInterlaceMode) {
68✔
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) {
48!
439
      taos_free_result(pStmt->exec.pRequest);
48✔
440
      pStmt->exec.pRequest = NULL;
48✔
441
    }
442

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

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

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

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

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

464
    if (keepTable) {
48✔
465
      return TSDB_CODE_SUCCESS;
32✔
466
    }
467

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

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

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

477
  return TSDB_CODE_SUCCESS;
36✔
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) {
20✔
491
  STMT_DLOG_E("start to free SQL info");
20!
492

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

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

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

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

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

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

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

529
  STMT_DLOG_E("end to free SQL info");
20!
530

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

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

581
  STableDataCxt** pCxtInExec = taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
35✔
582
  if (pCxtInExec) {
36✔
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) {
20✔
595
    STMT_ERR_RET(catalogGetHandle(pStmt->taos->pAppInfo->clusterId, &pStmt->pCatalog));
12!
596
    pStmt->sql.siInfo.pCatalog = pStmt->pCatalog;
12✔
597
  }
598

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

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) {
32✔
744
      SArray** p = (SArray**)TARRAY_GET_ELEM(pStmt->sql.siInfo.pTableCols, i);
16✔
745
      *p = taosArrayInit(20, POINTER_BYTES);
16✔
746
      if (*p == NULL) {
16!
747
        STMT_ERR_RET(terrno);
×
748
      }
749
    }
750

751
    atomic_store_8((int8_t*)&pStmt->sql.siInfo.tableColsReady, true);
16✔
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);
16✔
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) {
842✔
771
    if (atomic_load_8((int8_t*)&pStmt->queue.stopQueue)) {
846✔
772
      break;
4✔
773
    }
774

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

780
    int ret = stmtAsyncOutput(pStmt, asyncParam);
32✔
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) {
12✔
842
  STscObj*  pObj = (STscObj*)taos;
12✔
843
  STscStmt* pStmt = NULL;
12✔
844
  int32_t   code = 0;
12✔
845

846
  pStmt = taosMemoryCalloc(1, sizeof(STscStmt));
12!
847
  if (NULL == pStmt) {
12!
848
    return NULL;
×
849
  }
850

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

857
  pStmt->taos = pObj;
12✔
858
  pStmt->bInfo.needParse = true;
12✔
859
  pStmt->sql.status = STMT_INIT;
12✔
860
  pStmt->reqid = reqid;
12✔
861
  pStmt->errCode = TSDB_CODE_SUCCESS;
12✔
862

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

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

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

900
  pStmt->sql.siInfo.tableColsReady = true;
12✔
901

902
  STMT_LOG_SEQ(STMT_INIT);
12!
903

904
  tscDebug("stmt:%p initialized", pStmt);
12!
905

906
  return pStmt;
12✔
907
}
908

909
int stmtPrepare(TAOS_STMT* stmt, const char* sql, unsigned long length) {
20✔
910
  STscStmt* pStmt = (STscStmt*)stmt;
20✔
911

912
  STMT_DLOG_E("start to prepare");
20!
913

914
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
20!
915
    return pStmt->errCode;
×
916
  }
917

918
  if (pStmt->sql.status >= STMT_PREPARE) {
20✔
919
    STMT_ERR_RET(stmtResetStmt(pStmt));
8!
920
  }
921

922
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_PREPARE));
20!
923

924
  if (length <= 0) {
20!
925
    length = strlen(sql);
×
926
  }
927

928
  pStmt->sql.sqlStr = taosStrndup(sql, length);
20!
929
  if (!pStmt->sql.sqlStr) {
20!
930
    return terrno;
×
931
  }
932
  pStmt->sql.sqlLen = length;
20✔
933
  pStmt->sql.stbInterlaceMode = pStmt->stbInterlaceMode;
20✔
934

935
  char* dbName = NULL;
20✔
936
  if (qParseDbName(sql, length, &dbName)) {
20!
937
    STMT_ERR_RET(stmtSetDbName(stmt, dbName));
×
938
    taosMemoryFreeClear(dbName);
×
939
  }
940

941
  return TSDB_CODE_SUCCESS;
20✔
942
}
943

944
int32_t stmtInitStbInterlaceTableInfo(STscStmt* pStmt) {
4✔
945
  STableDataCxt** pSrc = taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
4✔
946
  if (!pSrc) {
4!
947
    return terrno;
×
948
  }
949
  STableDataCxt* pDst = NULL;
4✔
950

951
  STMT_ERR_RET(qCloneStmtDataBlock(&pDst, *pSrc, true));
4!
952
  pStmt->sql.siInfo.pDataCtx = pDst;
4✔
953

954
  SArray* pTblCols = NULL;
4✔
955
  for (int32_t i = 0; i < STMT_TABLE_COLS_NUM; i++) {
3,307✔
956
    pTblCols = taosArrayInit(20, POINTER_BYTES);
3,296✔
957
    if (NULL == pTblCols) {
3,566!
958
      return terrno;
×
959
    }
960

961
    if (taosArrayPush(pStmt->sql.siInfo.pTableCols, &pTblCols) == NULL) {
6,869!
962
      return terrno;
×
963
    }
964
  }
965

966
  pStmt->sql.siInfo.boundTags = pStmt->bInfo.boundTags;
11✔
967

968
  return TSDB_CODE_SUCCESS;
11✔
969
}
970

971
int stmtSetDbName(TAOS_STMT* stmt, const char* dbName) {
×
972
  STscStmt* pStmt = (STscStmt*)stmt;
×
973

974
  STMT_DLOG("start to set dbName: %s", dbName);
×
975

976
  STMT_ERR_RET(stmtCreateRequest(pStmt));
×
977

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

987
int stmtSetTbName(TAOS_STMT* stmt, const char* tbName) {
48✔
988
  STscStmt* pStmt = (STscStmt*)stmt;
48✔
989

990
  int64_t startUs = taosGetTimestampUs();
48✔
991

992
  STMT_DLOG("start to set tbName: %s", tbName);
48!
993

994
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
48!
995
    return pStmt->errCode;
×
996
  }
997

998
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_SETTBNAME));
48!
999

1000
  int32_t insert = 0;
48✔
1001
  STMT_ERR_RET(stmtIsInsert(stmt, &insert));
48!
1002
  if (0 == insert) {
48!
1003
    tscError("set tb name not available for none insert statement");
×
1004
    STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR);
×
1005
  }
1006

1007
  if (!pStmt->sql.stbInterlaceMode || NULL == pStmt->sql.siInfo.pDataCtx) {
48✔
1008
    STMT_ERR_RET(stmtCreateRequest(pStmt));
36!
1009

1010
    STMT_ERR_RET(qCreateSName(&pStmt->bInfo.sname, tbName, pStmt->taos->acctId, pStmt->exec.pRequest->pDb,
36!
1011
                              pStmt->exec.pRequest->msgBuf, pStmt->exec.pRequest->msgBufLen));
1012
    STMT_ERR_RET(tNameExtractFullName(&pStmt->bInfo.sname, pStmt->bInfo.tbFName));
36!
1013

1014
    STMT_ERR_RET(stmtGetFromCache(pStmt));
36!
1015

1016
    if (pStmt->bInfo.needParse) {
36✔
1017
      tstrncpy(pStmt->bInfo.tbName, tbName, sizeof(pStmt->bInfo.tbName));
20✔
1018
      pStmt->bInfo.tbName[sizeof(pStmt->bInfo.tbName) - 1] = 0;
20✔
1019

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

1029
  if (pStmt->sql.stbInterlaceMode && NULL == pStmt->sql.siInfo.pDataCtx) {
48✔
1030
    STMT_ERR_RET(stmtInitStbInterlaceTableInfo(pStmt));
4!
1031
  }
1032

1033
  int64_t startUs2 = taosGetTimestampUs();
48✔
1034
  pStmt->stat.setTbNameUs += startUs2 - startUs;
48✔
1035

1036
  return TSDB_CODE_SUCCESS;
48✔
1037
}
1038

1039
int stmtSetTbTags(TAOS_STMT* stmt, TAOS_MULTI_BIND* tags) {
×
1040
  STscStmt* pStmt = (STscStmt*)stmt;
×
1041

1042
  STMT_DLOG_E("start to set tbTags");
×
1043

1044
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
×
1045
    return pStmt->errCode;
×
1046
  }
1047

1048
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_SETTAGS));
×
1049

1050
  SBoundColInfo* tags_info = (SBoundColInfo*)pStmt->bInfo.boundTags;
×
1051
  if (tags_info->numOfBound <= 0 || tags_info->numOfCols <= 0) {
×
1052
    tscWarn("no tags bound in sql, will not bound tags");
×
1053
    return TSDB_CODE_SUCCESS;
×
1054
  }
1055

1056
  if (pStmt->bInfo.inExecCache) {
×
1057
    return TSDB_CODE_SUCCESS;
×
1058
  }
1059

1060
  STableDataCxt** pDataBlock =
1061
      (STableDataCxt**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
×
1062
  if (NULL == pDataBlock) {
×
1063
    tscError("table %s not found in exec blockHash", pStmt->bInfo.tbFName);
×
1064
    STMT_ERR_RET(TSDB_CODE_APP_ERROR);
×
1065
  }
1066

1067
  tscDebug("start to bind stmt tag values");
×
1068
  STMT_ERR_RET(qBindStmtTagsValue(*pDataBlock, pStmt->bInfo.boundTags, pStmt->bInfo.tbSuid, pStmt->bInfo.stbFName,
×
1069
                                  pStmt->bInfo.sname.tname, tags, pStmt->exec.pRequest->msgBuf,
1070
                                  pStmt->exec.pRequest->msgBufLen, pStmt->taos->optionInfo.charsetCxt));
1071

1072
  return TSDB_CODE_SUCCESS;
×
1073
}
1074

1075
int stmtFetchTagFields(STscStmt* pStmt, int32_t* fieldNum, TAOS_FIELD_E** fields) {
×
1076
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
×
1077
    return pStmt->errCode;
×
1078
  }
1079

1080
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
×
1081
    tscError("invalid operation to get query tag fileds");
×
1082
    STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR);
×
1083
  }
1084

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

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

1094
  return TSDB_CODE_SUCCESS;
×
1095
}
1096

1097
int stmtFetchColFields(STscStmt* pStmt, int32_t* fieldNum, TAOS_FIELD_E** fields) {
×
1098
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
×
1099
    return pStmt->errCode;
×
1100
  }
1101

1102
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
×
1103
    tscError("invalid operation to get query column fileds");
×
1104
    STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR);
×
1105
  }
1106

1107
  STableDataCxt** pDataBlock = NULL;
×
1108

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

1120
  STMT_ERR_RET(qBuildStmtColFields(*pDataBlock, fieldNum, fields));
×
1121

1122
  return TSDB_CODE_SUCCESS;
×
1123
}
1124

1125
/*
1126
SArray* stmtGetFreeCol(STscStmt* pStmt, int32_t* idx) {
1127
  while (true) {
1128
    if (pStmt->exec.smInfo.pColIdx >= STMT_COL_BUF_SIZE) {
1129
      pStmt->exec.smInfo.pColIdx = 0;
1130
    }
1131

1132
    if ((pStmt->exec.smInfo.pColIdx + 1) == atomic_load_32(&pStmt->exec.smInfo.pColFreeIdx)) {
1133
      taosUsleep(1);
1134
      continue;
1135
    }
1136

1137
    *idx = pStmt->exec.smInfo.pColIdx;
1138
    return pStmt->exec.smInfo.pCols[pStmt->exec.smInfo.pColIdx++];
1139
  }
1140
}
1141
*/
1142

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

1152
  if (NULL == pStmt->sql.siInfo.pRequest) {
16✔
1153
    STMT_ERR_RET(buildRequest(pStmt->taos->id, pStmt->sql.sqlStr, pStmt->sql.sqlLen, NULL, false,
4!
1154
                              (SRequestObj**)&pStmt->sql.siInfo.pRequest, pStmt->reqid));
1155

1156
    if (pStmt->reqid != 0) {
4!
1157
      pStmt->reqid++;
×
1158
    }
1159
    pStmt->exec.pRequest->syncQuery = true;
4✔
1160

1161
    pStmt->sql.siInfo.requestId = ((SRequestObj*)pStmt->sql.siInfo.pRequest)->requestId;
4✔
1162
    pStmt->sql.siInfo.requestSelf = ((SRequestObj*)pStmt->sql.siInfo.pRequest)->self;
4✔
1163
  }
1164

1165
  if (!pStmt->sql.siInfo.tbFromHash && pStmt->sql.siInfo.firstName[0] &&
16✔
1166
      0 == strcmp(pStmt->sql.siInfo.firstName, pStmt->bInfo.tbName)) {
4!
1167
    pStmt->sql.siInfo.tbFromHash = true;
4✔
1168
  }
1169

1170
  if (0 == pStmt->sql.siInfo.firstName[0]) {
16✔
1171
    tstrncpy(pStmt->sql.siInfo.firstName, pStmt->bInfo.tbName, TSDB_TABLE_NAME_LEN);
4✔
1172
  }
1173

1174
  param->tblData.getFromHash = pStmt->sql.siInfo.tbFromHash;
16✔
1175
  param->next = NULL;
16✔
1176

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

1179
  stmtEnqueue(pStmt, param);
16✔
1180

1181
  return TSDB_CODE_SUCCESS;
16✔
1182
}
1183

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

1197
        if (taosArrayPush(pStmt->sql.siInfo.pTableCols, &pTblCols) == NULL) {
×
1198
          return terrno;
×
1199
        }
1200
      }
1201
    }
1202
  }
1203

1204
  return TSDB_CODE_SUCCESS;
16✔
1205
}
1206

1207
int stmtBindBatch(TAOS_STMT* stmt, TAOS_MULTI_BIND* bind, int32_t colIdx) {
47✔
1208
  STscStmt* pStmt = (STscStmt*)stmt;
47✔
1209
  int32_t   code = 0;
47✔
1210

1211
  int64_t startUs = taosGetTimestampUs();
48✔
1212

1213
  STMT_DLOG("start to bind stmt data, colIdx: %d", colIdx);
48!
1214

1215
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
48!
1216
    return pStmt->errCode;
×
1217
  }
1218

1219
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_BIND));
48!
1220

1221
  if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 &&
48!
1222
      STMT_TYPE_MULTI_INSERT != pStmt->sql.type) {
×
1223
    pStmt->bInfo.needParse = false;
×
1224
  }
1225

1226
  if (pStmt->exec.pRequest && STMT_TYPE_QUERY == pStmt->sql.type && pStmt->sql.runTimes) {
48!
1227
    taos_free_result(pStmt->exec.pRequest);
×
1228
    pStmt->exec.pRequest = NULL;
×
1229
  }
1230

1231
  STMT_ERR_RET(stmtCreateRequest(pStmt));
48!
1232

1233
  if (pStmt->bInfo.needParse) {
47!
UNCOV
1234
    STMT_ERR_RET(stmtParseSql(pStmt));
×
1235
  }
1236

1237
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
47!
1238
    STMT_ERR_RET(qStmtBindParams(pStmt->sql.pQuery, bind, colIdx, pStmt->taos->optionInfo.charsetCxt));
×
1239

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

1253
    ctx.mgmtEpSet = getEpSet_s(&pStmt->taos->pAppInfo->mgmtEp);
×
1254
    STMT_ERR_RET(catalogGetHandle(pStmt->taos->pAppInfo->clusterId, &ctx.pCatalog));
×
1255

1256
    STMT_ERR_RET(qStmtParseQuerySql(&ctx, pStmt->sql.pQuery));
×
1257

1258
    if (pStmt->sql.pQuery->haveResultSet) {
×
1259
      STMT_ERR_RET(setResSchemaInfo(&pStmt->exec.pRequest->body.resInfo, pStmt->sql.pQuery->pResSchema,
×
1260
                                    pStmt->sql.pQuery->numOfResCols));
1261
      taosMemoryFreeClear(pStmt->sql.pQuery->pResSchema);
×
1262
      setResPrecision(&pStmt->exec.pRequest->body.resInfo, pStmt->sql.pQuery->precision);
×
1263
    }
1264

1265
    TSWAP(pStmt->exec.pRequest->dbList, pStmt->sql.pQuery->pDbList);
×
1266
    TSWAP(pStmt->exec.pRequest->tableList, pStmt->sql.pQuery->pTableList);
×
1267
    TSWAP(pStmt->exec.pRequest->targetTableList, pStmt->sql.pQuery->pTargetTableList);
×
1268

1269
    // if (STMT_TYPE_QUERY == pStmt->sql.queryRes) {
1270
    //   STMT_ERR_RET(stmtRestoreQueryFields(pStmt));
1271
    // }
1272

1273
    // STMT_ERR_RET(stmtBackupQueryFields(pStmt));
1274

1275
    return TSDB_CODE_SUCCESS;
×
1276
  }
1277

1278
  if (pStmt->sql.stbInterlaceMode && NULL == pStmt->sql.siInfo.pDataCtx) {
47!
1279
    STMT_ERR_RET(stmtInitStbInterlaceTableInfo(pStmt));
×
1280
  }
1281

1282
  STableDataCxt** pDataBlock = NULL;
47✔
1283

1284
  if (pStmt->exec.pCurrBlock) {
47✔
1285
    pDataBlock = &pStmt->exec.pCurrBlock;
28✔
1286
  } else {
1287
    pDataBlock =
1288
        (STableDataCxt**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
19✔
1289
    if (NULL == pDataBlock) {
19!
1290
      tscError("table %s not found in exec blockHash", pStmt->bInfo.tbFName);
×
1291
      STMT_ERR_RET(TSDB_CODE_TSC_STMT_CACHE_ERROR);
×
1292
    }
1293
    pStmt->exec.pCurrBlock = *pDataBlock;
19✔
1294
    if (pStmt->sql.stbInterlaceMode) {
19✔
1295
      taosArrayDestroy(pStmt->exec.pCurrBlock->pData->aCol);
4✔
1296
      pStmt->exec.pCurrBlock->pData->aCol = NULL;
4✔
1297
    }
1298
  }
1299

1300
  int64_t startUs2 = taosGetTimestampUs();
47✔
1301
  pStmt->stat.bindDataUs1 += startUs2 - startUs;
47✔
1302

1303
  SStmtQNode* param = NULL;
47✔
1304
  if (pStmt->sql.stbInterlaceMode) {
47✔
1305
    STMT_ERR_RET(stmtAllocQNodeFromBuf(&pStmt->sql.siInfo.tbBuf, (void**)&param));
32!
1306
    STMT_ERR_RET(stmtGetTableColsFromCache(pStmt, &param->tblData.aCol));
32!
1307
    taosArrayClear(param->tblData.aCol);
16✔
1308

1309
    // param->tblData.aCol = taosArrayInit(20, POINTER_BYTES);
1310

1311
    param->restoreTbCols = false;
16✔
1312
    tstrncpy(param->tblData.tbName, pStmt->bInfo.tbName, TSDB_TABLE_NAME_LEN);
16✔
1313
  }
1314

1315
  int64_t startUs3 = taosGetTimestampUs();
47✔
1316
  pStmt->stat.bindDataUs2 += startUs3 - startUs2;
47✔
1317

1318
  SArray* pCols = pStmt->sql.stbInterlaceMode ? param->tblData.aCol : (*pDataBlock)->pData->aCol;
47✔
1319

1320
  if (colIdx < 0) {
47!
1321
    if (pStmt->sql.stbInterlaceMode) {
47✔
1322
      (*pDataBlock)->pData->flags = 0;
16✔
1323
      code = qBindStmtStbColsValue(*pDataBlock, pCols, bind, pStmt->exec.pRequest->msgBuf,
16✔
1324
                                   pStmt->exec.pRequest->msgBufLen, &pStmt->sql.siInfo.pTSchema, pStmt->sql.pBindInfo, pStmt->taos->optionInfo.charsetCxt);
16✔
1325
    } else {
1326
      code =
1327
          qBindStmtColsValue(*pDataBlock, pCols, bind, pStmt->exec.pRequest->msgBuf, pStmt->exec.pRequest->msgBufLen, pStmt->taos->optionInfo.charsetCxt);
31✔
1328
    }
1329

1330
    if (code) {
48!
1331
      tscError("qBindStmtColsValue failed, error:%s", tstrerror(code));
×
1332
      STMT_ERR_RET(code);
×
1333
    }
1334
  } else {
1335
    if (pStmt->sql.stbInterlaceMode) {
×
1336
      tscError("bind single column not allowed in stb insert mode");
×
1337
      STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR);
×
1338
    }
1339

1340
    if (colIdx != (pStmt->bInfo.sBindLastIdx + 1) && colIdx != 0) {
×
1341
      tscError("bind column index not in sequence");
×
1342
      STMT_ERR_RET(TSDB_CODE_APP_ERROR);
×
1343
    }
1344

1345
    pStmt->bInfo.sBindLastIdx = colIdx;
×
1346

1347
    if (0 == colIdx) {
×
1348
      pStmt->bInfo.sBindRowNum = bind->num;
×
1349
    }
1350

1351
    code = qBindStmtSingleColValue(*pDataBlock, pCols, bind, pStmt->exec.pRequest->msgBuf,
×
1352
                                   pStmt->exec.pRequest->msgBufLen, colIdx, pStmt->bInfo.sBindRowNum, pStmt->taos->optionInfo.charsetCxt);
×
1353
    if (code) {
×
1354
      tscError("qBindStmtSingleColValue failed, error:%s", tstrerror(code));
×
1355
      STMT_ERR_RET(code);
×
1356
    }
1357
  }
1358

1359
  int64_t startUs4 = taosGetTimestampUs();
48✔
1360
  pStmt->stat.bindDataUs3 += startUs4 - startUs3;
48✔
1361

1362
  if (pStmt->sql.stbInterlaceMode) {
48✔
1363
    STMT_ERR_RET(stmtAppendTablePostHandle(pStmt, param));
16!
1364
  }
1365

1366
  pStmt->stat.bindDataUs4 += taosGetTimestampUs() - startUs4;
48✔
1367

1368
  return TSDB_CODE_SUCCESS;
48✔
1369
}
1370

1371
int stmtAddBatch(TAOS_STMT* stmt) {
48✔
1372
  STscStmt* pStmt = (STscStmt*)stmt;
48✔
1373

1374
  int64_t startUs = taosGetTimestampUs();
48✔
1375

1376
  STMT_DLOG_E("start to add batch");
48!
1377

1378
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
48!
1379
    return pStmt->errCode;
×
1380
  }
1381

1382
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_ADD_BATCH));
48!
1383

1384
  if (pStmt->sql.stbInterlaceMode) {
48✔
1385
    int64_t startUs2 = taosGetTimestampUs();
16✔
1386
    pStmt->stat.addBatchUs += startUs2 - startUs;
16✔
1387

1388
    pStmt->sql.siInfo.tableColsReady = false;
16✔
1389

1390
    SStmtQNode* param = NULL;
16✔
1391
    STMT_ERR_RET(stmtAllocQNodeFromBuf(&pStmt->sql.siInfo.tbBuf, (void**)&param));
32!
1392
    param->restoreTbCols = true;
16✔
1393
    param->next = NULL;
16✔
1394

1395
    stmtEnqueue(pStmt, param);
16✔
1396

1397
    return TSDB_CODE_SUCCESS;
16✔
1398
  }
1399

1400
  STMT_ERR_RET(stmtCacheBlock(pStmt));
32!
1401

1402
  return TSDB_CODE_SUCCESS;
32✔
1403
}
1404

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

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

1416
    STableMeta* pMeta = qGetTableMetaInDataBlock(pBlock);
×
1417
    if (pMeta->uid) {
×
1418
      pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
×
1419
      continue;
×
1420
    }
1421

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

1430
      if (strncmp(blkRsp->tblFName, key, keyLen)) {
×
1431
        continue;
×
1432
      }
1433

1434
      break;
×
1435
    }
1436

1437
    if (i < pRsp->nBlocks) {
×
1438
      tscDebug("auto created table %s uid updated from %" PRIx64 " to %" PRIx64, blkRsp->tblFName, pMeta->uid,
×
1439
               blkRsp->uid);
1440

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

1454
      code = stmtCreateRequest(pStmt);
×
1455
      if (code) {
×
1456
        pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
×
1457
        finalCode = code;
×
1458
        continue;
×
1459
      }
1460

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

1468
      pStmt->stat.ctgGetTbMetaNum++;
×
1469

1470
      taos_free_result(pStmt->exec.pRequest);
×
1471
      pStmt->exec.pRequest = NULL;
×
1472

1473
      if (code || NULL == pTableMeta) {
×
1474
        pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
×
1475
        finalCode = code;
×
1476
        taosMemoryFree(pTableMeta);
×
1477
        continue;
×
1478
      }
1479

1480
      pMeta->uid = pTableMeta->uid;
×
1481
      pStmt->bInfo.tbUid = pTableMeta->uid;
×
1482
      taosMemoryFree(pTableMeta);
×
1483
    }
1484

1485
    pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
×
1486
  }
1487

1488
  return finalCode;
×
1489
}
1490

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

1500
  STMT_DLOG_E("start to exec");
1501

1502
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_EXECUTE));
1503

1504
  STMT_ERR_RET(qBuildStmtOutputFromTbList(pStmt->sql.pQuery, pStmt->sql.pVgHash, pStmt->exec.pTbBlkList,
1505
pStmt->exec.pCurrBlock, pStmt->exec.tbBlkNum));
1506

1507
  launchQueryImpl(pStmt->exec.pRequest, pStmt->sql.pQuery, true, NULL);
1508

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

1520
  STMT_ERR_JRET(pStmt->exec.pRequest->code);
1521

1522
  pStmt->exec.affectedRows = taos_affected_rows(pStmt->exec.pRequest);
1523
  pStmt->affectedRows += pStmt->exec.affectedRows;
1524

1525
_return:
1526

1527
  stmtCleanExecInfo(pStmt, (code ? false : true), false);
1528

1529
  tFreeSSubmitRsp(pRsp);
1530

1531
  ++pStmt->sql.runTimes;
1532

1533
  STMT_RET(code);
1534
}
1535
*/
1536

1537
int stmtExec(TAOS_STMT* stmt) {
46✔
1538
  STscStmt*   pStmt = (STscStmt*)stmt;
46✔
1539
  int32_t     code = 0;
46✔
1540
  SSubmitRsp* pRsp = NULL;
46✔
1541

1542
  int64_t startUs = taosGetTimestampUs();
48✔
1543

1544
  STMT_DLOG_E("start to exec");
48!
1545

1546
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
48!
1547
    return pStmt->errCode;
×
1548
  }
1549

1550
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_EXECUTE));
48!
1551

1552
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
47!
1553
    launchQueryImpl(pStmt->exec.pRequest, pStmt->sql.pQuery, true, NULL);
×
1554
  } else {
1555
    if (pStmt->sql.stbInterlaceMode) {
47✔
1556
      int64_t startTs = taosGetTimestampUs();
16✔
1557
      while (atomic_load_64(&pStmt->sql.siInfo.tbRemainNum)) {
43✔
1558
        taosUsleep(1);
27✔
1559
      }
1560
      pStmt->stat.execWaitUs += taosGetTimestampUs() - startTs;
16✔
1561

1562
      STMT_ERR_RET(qBuildStmtFinOutput(pStmt->sql.pQuery, pStmt->sql.pVgHash, pStmt->sql.siInfo.pVgroupList));
16!
1563
      taosHashCleanup(pStmt->sql.siInfo.pVgroupHash);
16✔
1564
      pStmt->sql.siInfo.pVgroupHash = NULL;
16✔
1565
      pStmt->sql.siInfo.pVgroupList = NULL;
16✔
1566
    } else {
1567
      tDestroySubmitTbData(pStmt->exec.pCurrTbData, TSDB_MSG_FLG_ENCODE);
31✔
1568
      taosMemoryFreeClear(pStmt->exec.pCurrTbData);
32!
1569

1570
      STMT_ERR_RET(qCloneCurrentTbData(pStmt->exec.pCurrBlock, &pStmt->exec.pCurrTbData));
32!
1571

1572
      STMT_ERR_RET(qBuildStmtOutput(pStmt->sql.pQuery, pStmt->sql.pVgHash, pStmt->exec.pBlockHash));
31!
1573
    }
1574

1575
    launchQueryImpl(pStmt->exec.pRequest, pStmt->sql.pQuery, true, NULL);
48✔
1576
  }
1577

1578
  if (pStmt->exec.pRequest->code && NEED_CLIENT_HANDLE_ERROR(pStmt->exec.pRequest->code)) {
48!
1579
    code = refreshMeta(pStmt->exec.pRequest->pTscObj, pStmt->exec.pRequest);
×
1580
    if (code) {
×
1581
      pStmt->exec.pRequest->code = code;
×
1582
    } else {
1583
      tFreeSSubmitRsp(pRsp);
×
1584
      STMT_ERR_RET(stmtResetStmt(pStmt));
×
1585
      STMT_ERR_RET(TSDB_CODE_NEED_RETRY);
×
1586
    }
1587
  }
1588

1589
  STMT_ERR_JRET(pStmt->exec.pRequest->code);
48!
1590

1591
  pStmt->exec.affectedRows = taos_affected_rows(pStmt->exec.pRequest);
48✔
1592
  pStmt->affectedRows += pStmt->exec.affectedRows;
48✔
1593

1594
_return:
48✔
1595

1596
  while (0 == atomic_load_8((int8_t*)&pStmt->sql.siInfo.tableColsReady)) {
48!
1597
    taosUsleep(1);
×
1598
  }
1599

1600
  STMT_ERR_RET(stmtCleanExecInfo(pStmt, (code ? false : true), false));
48!
1601

1602
  tFreeSSubmitRsp(pRsp);
48✔
1603

1604
  ++pStmt->sql.runTimes;
48✔
1605

1606
  int64_t startUs2 = taosGetTimestampUs();
48✔
1607
  pStmt->stat.execUseUs += startUs2 - startUs;
48✔
1608

1609
  STMT_RET(code);
48!
1610
}
1611

1612
int stmtClose(TAOS_STMT* stmt) {
12✔
1613
  STscStmt* pStmt = (STscStmt*)stmt;
12✔
1614

1615
  STMT_DLOG_E("start to free stmt");
12!
1616

1617
  pStmt->queue.stopQueue = true;
12✔
1618

1619
  if (pStmt->bindThreadInUse) {
12✔
1620
    (void)taosThreadJoin(pStmt->bindThread, NULL);
4✔
1621
    pStmt->bindThreadInUse = false;
4✔
1622
  }
1623

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

1635
  STMT_ERR_RET(stmtCleanSQLInfo(pStmt));
12!
1636
  taosMemoryFree(stmt);
12!
1637

1638
  return TSDB_CODE_SUCCESS;
12✔
1639
}
1640

UNCOV
1641
const char* stmtErrstr(TAOS_STMT* stmt) {
×
UNCOV
1642
  STscStmt* pStmt = (STscStmt*)stmt;
×
1643

UNCOV
1644
  if (stmt == NULL || NULL == pStmt->exec.pRequest) {
×
1645
    return (char*)tstrerror(terrno);
×
1646
  }
1647

UNCOV
1648
  pStmt->exec.pRequest->code = terrno;
×
1649

UNCOV
1650
  return taos_errstr(pStmt->exec.pRequest);
×
1651
}
1652

UNCOV
1653
int stmtAffectedRows(TAOS_STMT* stmt) { return ((STscStmt*)stmt)->affectedRows; }
×
1654

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

1657
int stmtIsInsert(TAOS_STMT* stmt, int* insert) {
96✔
1658
  STscStmt* pStmt = (STscStmt*)stmt;
96✔
1659

1660
  STMT_DLOG_E("start is insert");
96!
1661

1662
  if (pStmt->sql.type) {
96✔
1663
    *insert = (STMT_TYPE_INSERT == pStmt->sql.type || STMT_TYPE_MULTI_INSERT == pStmt->sql.type);
76!
1664
  } else {
1665
    *insert = qIsInsertValuesSql(pStmt->sql.sqlStr, pStmt->sql.sqlLen);
20✔
1666
  }
1667

1668
  return TSDB_CODE_SUCCESS;
96✔
1669
}
1670

1671
int stmtGetTagFields(TAOS_STMT* stmt, int* nums, TAOS_FIELD_E** fields) {
×
1672
  int32_t   code = 0;
×
1673
  STscStmt* pStmt = (STscStmt*)stmt;
×
1674
  int32_t   preCode = pStmt->errCode;
×
1675

1676
  STMT_DLOG_E("start to get tag fields");
×
1677

1678
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
×
1679
    return pStmt->errCode;
×
1680
  }
1681

1682
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
×
1683
    STMT_ERRI_JRET(TSDB_CODE_TSC_STMT_API_ERROR);
×
1684
  }
1685

1686
  STMT_ERRI_JRET(stmtSwitchStatus(pStmt, STMT_FETCH_FIELDS));
×
1687

1688
  if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 &&
×
1689
      STMT_TYPE_MULTI_INSERT != pStmt->sql.type) {
×
1690
    pStmt->bInfo.needParse = false;
×
1691
  }
1692

1693
  if (pStmt->exec.pRequest && STMT_TYPE_QUERY == pStmt->sql.type && pStmt->sql.runTimes) {
×
1694
    taos_free_result(pStmt->exec.pRequest);
×
1695
    pStmt->exec.pRequest = NULL;
×
1696
  }
1697

1698
  STMT_ERRI_JRET(stmtCreateRequest(pStmt));
×
1699

1700
  if (pStmt->bInfo.needParse) {
×
1701
    STMT_ERRI_JRET(stmtParseSql(pStmt));
×
1702
  }
1703

1704
  STMT_ERRI_JRET(stmtFetchTagFields(stmt, nums, fields));
×
1705

1706
_return:
×
1707

1708
  pStmt->errCode = preCode;
×
1709

1710
  return code;
×
1711
}
1712

1713
int stmtGetColFields(TAOS_STMT* stmt, int* nums, TAOS_FIELD_E** fields) {
×
1714
  int32_t   code = 0;
×
1715
  STscStmt* pStmt = (STscStmt*)stmt;
×
1716
  int32_t   preCode = pStmt->errCode;
×
1717

1718
  STMT_DLOG_E("start to get col fields");
×
1719

1720
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
×
1721
    return pStmt->errCode;
×
1722
  }
1723

1724
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
×
1725
    STMT_ERRI_JRET(TSDB_CODE_TSC_STMT_API_ERROR);
×
1726
  }
1727

1728
  STMT_ERRI_JRET(stmtSwitchStatus(pStmt, STMT_FETCH_FIELDS));
×
1729

1730
  if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 &&
×
1731
      STMT_TYPE_MULTI_INSERT != pStmt->sql.type) {
×
1732
    pStmt->bInfo.needParse = false;
×
1733
  }
1734

1735
  if (pStmt->exec.pRequest && STMT_TYPE_QUERY == pStmt->sql.type && pStmt->sql.runTimes) {
×
1736
    taos_free_result(pStmt->exec.pRequest);
×
1737
    pStmt->exec.pRequest = NULL;
×
1738
    STMT_ERR_RET(stmtCreateRequest(pStmt));
×
1739
  }
1740

1741
  STMT_ERRI_JRET(stmtCreateRequest(pStmt));
×
1742

1743
  if (pStmt->bInfo.needParse) {
×
1744
    STMT_ERRI_JRET(stmtParseSql(pStmt));
×
1745
  }
1746

1747
  STMT_ERRI_JRET(stmtFetchColFields(stmt, nums, fields));
×
1748

1749
_return:
×
1750

1751
  pStmt->errCode = preCode;
×
1752

1753
  return code;
×
1754
}
1755

1756
int stmtGetParamNum(TAOS_STMT* stmt, int* nums) {
×
1757
  STscStmt* pStmt = (STscStmt*)stmt;
×
1758

1759
  STMT_DLOG_E("start to get param num");
×
1760

1761
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
×
1762
    return pStmt->errCode;
×
1763
  }
1764

1765
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_FETCH_FIELDS));
×
1766

1767
  if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 &&
×
1768
      STMT_TYPE_MULTI_INSERT != pStmt->sql.type) {
×
1769
    pStmt->bInfo.needParse = false;
×
1770
  }
1771

1772
  if (pStmt->exec.pRequest && STMT_TYPE_QUERY == pStmt->sql.type && pStmt->sql.runTimes) {
×
1773
    taos_free_result(pStmt->exec.pRequest);
×
1774
    pStmt->exec.pRequest = NULL;
×
1775
  }
1776

1777
  STMT_ERR_RET(stmtCreateRequest(pStmt));
×
1778

1779
  if (pStmt->bInfo.needParse) {
×
1780
    STMT_ERR_RET(stmtParseSql(pStmt));
×
1781
  }
1782

1783
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
×
1784
    *nums = taosArrayGetSize(pStmt->sql.pQuery->pPlaceholderValues);
×
1785
  } else {
1786
    STMT_ERR_RET(stmtFetchColFields(stmt, nums, NULL));
×
1787
  }
1788

1789
  return TSDB_CODE_SUCCESS;
×
1790
}
1791

1792
int stmtGetParam(TAOS_STMT* stmt, int idx, int* type, int* bytes) {
×
1793
  STscStmt* pStmt = (STscStmt*)stmt;
×
1794

1795
  STMT_DLOG_E("start to get param");
×
1796

1797
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
×
1798
    return pStmt->errCode;
×
1799
  }
1800

1801
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
×
1802
    STMT_RET(TSDB_CODE_TSC_STMT_API_ERROR);
×
1803
  }
1804

1805
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_FETCH_FIELDS));
×
1806

1807
  if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 &&
×
1808
      STMT_TYPE_MULTI_INSERT != pStmt->sql.type) {
×
1809
    pStmt->bInfo.needParse = false;
×
1810
  }
1811

1812
  if (pStmt->exec.pRequest && STMT_TYPE_QUERY == pStmt->sql.type && pStmt->sql.runTimes) {
×
1813
    taos_free_result(pStmt->exec.pRequest);
×
1814
    pStmt->exec.pRequest = NULL;
×
1815
  }
1816

1817
  STMT_ERR_RET(stmtCreateRequest(pStmt));
×
1818

1819
  if (pStmt->bInfo.needParse) {
×
1820
    STMT_ERR_RET(stmtParseSql(pStmt));
×
1821
  }
1822

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

1832
  *type = pField[idx].type;
×
1833
  *bytes = pField[idx].bytes;
×
1834

1835
  taosMemoryFree(pField);
×
1836

1837
  return TSDB_CODE_SUCCESS;
×
1838
}
1839

1840
TAOS_RES* stmtUseResult(TAOS_STMT* stmt) {
×
1841
  STscStmt* pStmt = (STscStmt*)stmt;
×
1842

1843
  STMT_DLOG_E("start to use result");
×
1844

1845
  if (STMT_TYPE_QUERY != pStmt->sql.type) {
×
1846
    tscError("useResult only for query statement");
×
1847
    return NULL;
×
1848
  }
1849

1850
  return pStmt->exec.pRequest;
×
1851
}
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