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

taosdata / TDengine / #3647

13 Mar 2025 05:26AM UTC coverage: 25.9% (-2.5%) from 28.375%
#3647

push

travis-ci

web-flow
Merge pull request #30158 from taosdata/docs/anchor-caps-30

docs: lowercase anchors for 3.0

53974 of 285572 branches covered (18.9%)

Branch coverage included in aggregate %.

92870 of 281392 relevant lines covered (33.0%)

617448.64 hits per line

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

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

41
bool stmtDequeue(STscStmt* pStmt, SStmtQNode** param) {
37✔
42
  int i = 0;
37✔
43
  while (0 == atomic_load_64((int64_t*)&pStmt->queue.qRemainNum)) {
134✔
44
    if (i < 10) {
97✔
45
      taosUsleep(1);
94✔
46
      i++;
94✔
47
    } else {
48
      (void)taosThreadMutexLock(&pStmt->queue.mutex);
3✔
49
      if (0 == atomic_load_64((int64_t*)&pStmt->queue.qRemainNum)) {
3!
50
        (void)taosThreadCondWait(&pStmt->queue.waitCond, &pStmt->queue.mutex);
3✔
51
      }
52
      (void)taosThreadMutexUnlock(&pStmt->queue.mutex);
3✔
53
    }
54
  }
55
  if (pStmt->queue.stopQueue) {
37✔
56
    return false;
1✔
57
  }
58
  SStmtQNode* orig = pStmt->queue.head;
36✔
59
  SStmtQNode* node = pStmt->queue.head->next;
36✔
60
  pStmt->queue.head = pStmt->queue.head->next;
36✔
61
  *param = node;
36✔
62

63
  (void)atomic_sub_fetch_64((int64_t*)&pStmt->queue.qRemainNum, 1);
36✔
64

65
  return true;
36✔
66
}
67

68
void stmtEnqueue(STscStmt* pStmt, SStmtQNode* param) {
36✔
69
  pStmt->queue.tail->next = param;
36✔
70
  pStmt->queue.tail = param;
36✔
71

72
  pStmt->stat.bindDataNum++;
36✔
73

74
  (void)taosThreadMutexLock(&pStmt->queue.mutex);
36✔
75
  (void)atomic_add_fetch_64(&pStmt->queue.qRemainNum, 1);
36✔
76
  (void)taosThreadCondSignal(&(pStmt->queue.waitCond));
36✔
77
  (void)taosThreadMutexUnlock(&pStmt->queue.mutex);
36✔
78
}
36✔
79

80
static int32_t stmtCreateRequest(STscStmt* pStmt) {
132✔
81
  int32_t code = 0;
132✔
82

83
  if (pStmt->exec.pRequest == NULL) {
132✔
84
    code = buildRequest(pStmt->taos->id, pStmt->sql.sqlStr, pStmt->sql.sqlLen, NULL, false, &pStmt->exec.pRequest,
27✔
85
                        pStmt->reqid);
86
    if (pStmt->reqid != 0) {
27!
87
      pStmt->reqid++;
×
88
    }
89
    if (TSDB_CODE_SUCCESS == code) {
27!
90
      pStmt->exec.pRequest->syncQuery = true;
27✔
91
      pStmt->exec.pRequest->isStmtBind = true;
27✔
92
    }
93
  }
94

95
  return code;
132✔
96
}
97

98
int32_t stmtSwitchStatus(STscStmt* pStmt, STMT_STATUS newStatus) {
209✔
99
  int32_t code = 0;
209✔
100

101
  if (newStatus >= STMT_INIT && newStatus < STMT_MAX) {
209!
102
    STMT_LOG_SEQ(newStatus);
209!
103
  }
104

105
  if (pStmt->errCode && newStatus != STMT_PREPARE) {
209!
106
    STMT_DLOG("stmt already failed with err:%s", tstrerror(pStmt->errCode));
×
107
    return pStmt->errCode;
×
108
  }
109

110
  switch (newStatus) {
209!
111
    case STMT_PREPARE:
12✔
112
      pStmt->errCode = 0;
12✔
113
      break;
12✔
114
    case STMT_SETTBNAME:
30✔
115
      if (STMT_STATUS_EQ(INIT)) {
30!
116
        code = TSDB_CODE_TSC_STMT_API_ERROR;
×
117
      }
118
      if (!pStmt->sql.stbInterlaceMode && (STMT_STATUS_EQ(BIND) || STMT_STATUS_EQ(BIND_COL))) {
30!
119
        code = TSDB_CODE_TSC_STMT_API_ERROR;
×
120
      }
121
      break;
30✔
122
    case STMT_SETTAGS:
11✔
123
      if (STMT_STATUS_NE(SETTBNAME) && STMT_STATUS_NE(FETCH_FIELDS)) {
11!
124
        code = TSDB_CODE_TSC_STMT_API_ERROR;
×
125
      }
126
      break;
11✔
127
    case STMT_FETCH_FIELDS:
11✔
128
      if (STMT_STATUS_EQ(INIT)) {
11!
129
        code = TSDB_CODE_TSC_STMT_API_ERROR;
×
130
      }
131
      break;
11✔
132
    case STMT_BIND:
85✔
133
      if (STMT_STATUS_EQ(INIT) || STMT_STATUS_EQ(BIND_COL)) {
85!
134
        code = TSDB_CODE_TSC_STMT_API_ERROR;
×
135
      }
136
      /*
137
            if ((pStmt->sql.type == STMT_TYPE_MULTI_INSERT) && ()) {
138
              code = TSDB_CODE_TSC_STMT_API_ERROR;
139
            }
140
      */
141
      break;
85✔
142
    case STMT_BIND_COL:
×
143
      if (STMT_STATUS_EQ(INIT) || STMT_STATUS_EQ(BIND)) {
×
144
        code = TSDB_CODE_TSC_STMT_API_ERROR;
×
145
      }
146
      break;
×
147
    case STMT_ADD_BATCH:
30✔
148
      if (STMT_STATUS_NE(BIND) && STMT_STATUS_NE(BIND_COL) && STMT_STATUS_NE(FETCH_FIELDS)) {
30!
149
        code = TSDB_CODE_TSC_STMT_API_ERROR;
×
150
      }
151
      break;
30✔
152
    case STMT_EXECUTE:
30✔
153
      if (STMT_TYPE_QUERY == pStmt->sql.type) {
30!
154
        if (STMT_STATUS_NE(ADD_BATCH) && STMT_STATUS_NE(FETCH_FIELDS) && STMT_STATUS_NE(BIND) &&
×
155
            STMT_STATUS_NE(BIND_COL)) {
×
156
          code = TSDB_CODE_TSC_STMT_API_ERROR;
×
157
        }
158
      } else {
159
        if (STMT_STATUS_NE(ADD_BATCH) && STMT_STATUS_NE(FETCH_FIELDS)) {
30!
160
          code = TSDB_CODE_TSC_STMT_API_ERROR;
×
161
        }
162
      }
163
      break;
30✔
164
    default:
×
165
      code = TSDB_CODE_APP_ERROR;
×
166
      break;
×
167
  }
168

169
  STMT_ERR_RET(code);
209!
170

171
  pStmt->sql.status = newStatus;
209✔
172

173
  return TSDB_CODE_SUCCESS;
209✔
174
}
175

176
int32_t stmtGetTbName(TAOS_STMT* stmt, char** tbName) {
8✔
177
  STscStmt* pStmt = (STscStmt*)stmt;
8✔
178

179
  pStmt->sql.type = STMT_TYPE_MULTI_INSERT;
8✔
180

181
  if ('\0' == pStmt->bInfo.tbName[0]) {
8✔
182
    tscError("no table name set");
2!
183
    STMT_ERR_RET(TSDB_CODE_TSC_STMT_TBNAME_ERROR);
2!
184
  }
185

186
  *tbName = pStmt->bInfo.tbName;
6✔
187

188
  return TSDB_CODE_SUCCESS;
6✔
189
}
190
/*
191
int32_t stmtBackupQueryFields(STscStmt* pStmt) {
192
  SStmtQueryResInfo* pRes = &pStmt->sql.queryRes;
193
  pRes->numOfCols = pStmt->exec.pRequest->body.resInfo.numOfCols;
194
  pRes->precision = pStmt->exec.pRequest->body.resInfo.precision;
195

196
  int32_t size = pRes->numOfCols * sizeof(TAOS_FIELD);
197
  pRes->fields = taosMemoryMalloc(size);
198
  if (pRes->fields == NULL) {
199
    STMT_ERR_RET(terrno);
200
  }
201

202
  pRes->userFields = taosMemoryMalloc(size);
203
  if (pRes->userFields == NULL) {
204
    taosMemoryFreeClear(pRes->fields);
205
    STMT_ERR_RET(terrno);
206
  }
207

208
  (void)memcpy(pRes->fields, pStmt->exec.pRequest->body.resInfo.fields, size);
209
  (void)memcpy(pRes->userFields, pStmt->exec.pRequest->body.resInfo.userFields, size);
210

211
  return TSDB_CODE_SUCCESS;
212
}
213

214
int32_t stmtRestoreQueryFields(STscStmt* pStmt) {
215
  SStmtQueryResInfo* pRes = &pStmt->sql.queryRes;
216
  int32_t            size = pRes->numOfCols * sizeof(TAOS_FIELD);
217

218
  pStmt->exec.pRequest->body.resInfo.numOfCols = pRes->numOfCols;
219
  pStmt->exec.pRequest->body.resInfo.precision = pRes->precision;
220

221
  if (NULL == pStmt->exec.pRequest->body.resInfo.fields) {
222
    pStmt->exec.pRequest->body.resInfo.fields = taosMemoryMalloc(size);
223
    if (NULL == pStmt->exec.pRequest->body.resInfo.fields) {
224
      STMT_ERR_RET(terrno);
225
    }
226
    (void)memcpy(pStmt->exec.pRequest->body.resInfo.fields, pRes->fields, size);
227
  }
228

229
  if (NULL == pStmt->exec.pRequest->body.resInfo.userFields) {
230
    pStmt->exec.pRequest->body.resInfo.userFields = taosMemoryMalloc(size);
231
    if (NULL == pStmt->exec.pRequest->body.resInfo.userFields) {
232
      STMT_ERR_RET(terrno);
233
    }
234
    (void)memcpy(pStmt->exec.pRequest->body.resInfo.userFields, pRes->userFields, size);
235
  }
236

237
  return TSDB_CODE_SUCCESS;
238
}
239
*/
240
int32_t stmtUpdateBindInfo(TAOS_STMT* stmt, STableMeta* pTableMeta, void* tags, SName* tbName, const char* sTableName,
8✔
241
                           bool autoCreateTbl) {
242
  STscStmt* pStmt = (STscStmt*)stmt;
8✔
243
  char      tbFName[TSDB_TABLE_FNAME_LEN];
244
  int32_t   code = tNameExtractFullName(tbName, tbFName);
8✔
245
  if (code != 0) {
8!
246
    return code;
×
247
  }
248

249
  (void)memcpy(&pStmt->bInfo.sname, tbName, sizeof(*tbName));
8✔
250
  tstrncpy(pStmt->bInfo.tbFName, tbFName, TSDB_TABLE_FNAME_LEN);
8✔
251
  pStmt->bInfo.tbFName[sizeof(pStmt->bInfo.tbFName) - 1] = 0;
8✔
252

253
  pStmt->bInfo.tbUid = autoCreateTbl ? 0 : pTableMeta->uid;
8✔
254
  pStmt->bInfo.tbSuid = pTableMeta->suid;
8✔
255
  pStmt->bInfo.tbVgId = pTableMeta->vgId;
8✔
256
  pStmt->bInfo.tbType = pTableMeta->tableType;
8✔
257
  pStmt->bInfo.boundTags = tags;
8✔
258
  pStmt->bInfo.tagsCached = false;
8✔
259
  tstrncpy(pStmt->bInfo.stbFName, sTableName, sizeof(pStmt->bInfo.stbFName));
8✔
260

261
  return TSDB_CODE_SUCCESS;
8✔
262
}
263

264
int32_t stmtUpdateExecInfo(TAOS_STMT* stmt, SHashObj* pVgHash, SHashObj* pBlockHash) {
8✔
265
  STscStmt* pStmt = (STscStmt*)stmt;
8✔
266

267
  pStmt->sql.pVgHash = pVgHash;
8✔
268
  pStmt->exec.pBlockHash = pBlockHash;
8✔
269

270
  return TSDB_CODE_SUCCESS;
8✔
271
}
272

273
int32_t stmtUpdateInfo(TAOS_STMT* stmt, STableMeta* pTableMeta, void* tags, SName* tbName, bool autoCreateTbl,
8✔
274
                       SHashObj* pVgHash, SHashObj* pBlockHash, const char* sTableName, bool preCtbname) {
275
  STscStmt* pStmt = (STscStmt*)stmt;
8✔
276

277
  STMT_ERR_RET(stmtUpdateBindInfo(stmt, pTableMeta, tags, tbName, sTableName, autoCreateTbl));
8!
278
  STMT_ERR_RET(stmtUpdateExecInfo(stmt, pVgHash, pBlockHash));
8!
279

280
  pStmt->sql.autoCreateTbl = autoCreateTbl;
8✔
281
  if (pStmt->sql.autoCreateTbl) {
8✔
282
    pStmt->sql.stbInterlaceMode = false;
5✔
283
  }
284

285
  return TSDB_CODE_SUCCESS;
8✔
286
}
287

288
int32_t stmtGetExecInfo(TAOS_STMT* stmt, SHashObj** pVgHash, SHashObj** pBlockHash) {
×
289
  STscStmt* pStmt = (STscStmt*)stmt;
×
290

291
  *pVgHash = pStmt->sql.pVgHash;
×
292
  pStmt->sql.pVgHash = NULL;
×
293

294
  *pBlockHash = pStmt->exec.pBlockHash;
×
295
  pStmt->exec.pBlockHash = NULL;
×
296

297
  return TSDB_CODE_SUCCESS;
×
298
}
299

300
int32_t stmtCacheBlock(STscStmt* pStmt) {
21✔
301
  if (pStmt->sql.type != STMT_TYPE_MULTI_INSERT) {
21✔
302
    return TSDB_CODE_SUCCESS;
1✔
303
  }
304

305
  uint64_t uid = pStmt->bInfo.tbUid;
20✔
306
  uint64_t cacheUid = (TSDB_CHILD_TABLE == pStmt->bInfo.tbType) ? pStmt->bInfo.tbSuid : uid;
20!
307

308
  if (taosHashGet(pStmt->sql.pTableCache, &cacheUid, sizeof(cacheUid))) {
20✔
309
    return TSDB_CODE_SUCCESS;
16✔
310
  }
311

312
  STableDataCxt** pSrc = taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
4✔
313
  if (!pSrc) {
4!
314
    return terrno;
×
315
  }
316
  STableDataCxt* pDst = NULL;
4✔
317

318
  STMT_ERR_RET(qCloneStmtDataBlock(&pDst, *pSrc, true));
4!
319

320
  SStmtTableCache cache = {
4✔
321
      .pDataCtx = pDst,
322
      .boundTags = pStmt->bInfo.boundTags,
4✔
323
  };
324

325
  if (taosHashPut(pStmt->sql.pTableCache, &cacheUid, sizeof(cacheUid), &cache, sizeof(cache))) {
4!
326
    return terrno;
×
327
  }
328

329
  if (pStmt->sql.autoCreateTbl) {
4✔
330
    pStmt->bInfo.tagsCached = true;
3✔
331
  } else {
332
    pStmt->bInfo.boundTags = NULL;
1✔
333
  }
334

335
  return TSDB_CODE_SUCCESS;
4✔
336
}
337

338
int32_t stmtParseSql(STscStmt* pStmt) {
11✔
339
  pStmt->exec.pCurrBlock = NULL;
11✔
340

341
  SStmtCallback stmtCb = {
11✔
342
      .pStmt = pStmt,
343
      .getTbNameFn = stmtGetTbName,
344
      .setInfoFn = stmtUpdateInfo,
345
      .getExecInfoFn = stmtGetExecInfo,
346
  };
347

348
  STMT_ERR_RET(stmtCreateRequest(pStmt));
11!
349

350
  pStmt->stat.parseSqlNum++;
11✔
351
  STMT_ERR_RET(parseSql(pStmt->exec.pRequest, false, &pStmt->sql.pQuery, &stmtCb));
11✔
352
  pStmt->sql.siInfo.pQuery = pStmt->sql.pQuery;
8✔
353

354
  pStmt->bInfo.needParse = false;
8✔
355

356
  if (pStmt->sql.pQuery->pRoot && 0 == pStmt->sql.type) {
8!
357
    pStmt->sql.type = STMT_TYPE_INSERT;
1✔
358
    pStmt->sql.stbInterlaceMode = false;
1✔
359
  } else if (pStmt->sql.pQuery->pPrepareRoot) {
7!
360
    pStmt->sql.type = STMT_TYPE_QUERY;
×
361
    pStmt->sql.stbInterlaceMode = false;
×
362

363
    return TSDB_CODE_SUCCESS;
×
364
  }
365

366
  STableDataCxt** pSrc =
367
      (STableDataCxt**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
8✔
368
  if (NULL == pSrc || NULL == *pSrc) {
8!
369
    return terrno;
×
370
  }
371

372
  STableDataCxt* pTableCtx = *pSrc;
8✔
373
  if (pStmt->sql.stbInterlaceMode) {
8✔
374
    int16_t lastIdx = -1;
1✔
375

376
    for (int32_t i = 0; i < pTableCtx->boundColsInfo.numOfBound; ++i) {
5✔
377
      if (pTableCtx->boundColsInfo.pColIndex[i] < lastIdx) {
4!
378
        pStmt->sql.stbInterlaceMode = false;
×
379
        break;
×
380
      }
381

382
      lastIdx = pTableCtx->boundColsInfo.pColIndex[i];
4✔
383
    }
384
  }
385

386
  if (NULL == pStmt->sql.pBindInfo) {
8!
387
    pStmt->sql.pBindInfo = taosMemoryMalloc(pTableCtx->boundColsInfo.numOfBound * sizeof(*pStmt->sql.pBindInfo));
8!
388
    if (NULL == pStmt->sql.pBindInfo) {
8!
389
      return terrno;
×
390
    }
391
  }
392

393
  return TSDB_CODE_SUCCESS;
8✔
394
}
395

396
int32_t stmtCleanBindInfo(STscStmt* pStmt) {
29✔
397
  pStmt->bInfo.tbUid = 0;
29✔
398
  pStmt->bInfo.tbSuid = 0;
29✔
399
  pStmt->bInfo.tbVgId = -1;
29✔
400
  pStmt->bInfo.tbType = 0;
29✔
401
  pStmt->bInfo.needParse = true;
29✔
402
  pStmt->bInfo.inExecCache = false;
29✔
403

404
  pStmt->bInfo.tbName[0] = 0;
29✔
405
  pStmt->bInfo.tbFName[0] = 0;
29✔
406
  if (!pStmt->bInfo.tagsCached) {
29✔
407
    qDestroyBoundColInfo(pStmt->bInfo.boundTags);
21✔
408
    taosMemoryFreeClear(pStmt->bInfo.boundTags);
21!
409
  }
410
  pStmt->bInfo.stbFName[0] = 0;
29✔
411

412
  return TSDB_CODE_SUCCESS;
29✔
413
}
414

415
void stmtFreeTableBlkList(STableColsData* pTb) {
×
416
  (void)qResetStmtColumns(pTb->aCol, true);
×
417
  taosArrayDestroy(pTb->aCol);
×
418
}
×
419

420
void stmtResetQueueTableBuf(STableBufInfo* pTblBuf, SStmtQueue* pQueue) {
9✔
421
  pTblBuf->pCurBuff = taosArrayGetP(pTblBuf->pBufList, 0);
9✔
422
  if (NULL == pTblBuf->pCurBuff) {
9!
423
    tscError("QInfo:%p, failed to get buffer from list", pTblBuf);
×
424
    return;
×
425
  }
426
  pTblBuf->buffIdx = 1;
9✔
427
  pTblBuf->buffOffset = sizeof(*pQueue->head);
9✔
428

429
  pQueue->head = pQueue->tail = pTblBuf->pCurBuff;
9✔
430
  pQueue->qRemainNum = 0;
9✔
431
  pQueue->head->next = NULL;
9✔
432
}
433

434
int32_t stmtCleanExecInfo(STscStmt* pStmt, bool keepTable, bool deepClean) {
40✔
435
  if (pStmt->sql.stbInterlaceMode) {
40✔
436
    if (deepClean) {
10✔
437
      taosHashCleanup(pStmt->exec.pBlockHash);
1✔
438
      pStmt->exec.pBlockHash = NULL;
1✔
439

440
      if (NULL != pStmt->exec.pCurrBlock) {
1!
441
        taosMemoryFreeClear(pStmt->exec.pCurrBlock->pData);
1!
442
        qDestroyStmtDataBlock(pStmt->exec.pCurrBlock);
1✔
443
      }
444
    } else {
445
      pStmt->sql.siInfo.pTableColsIdx = 0;
9✔
446
      stmtResetQueueTableBuf(&pStmt->sql.siInfo.tbBuf, &pStmt->queue);
9✔
447
    }
448
  } else {
449
    if (STMT_TYPE_QUERY != pStmt->sql.type || deepClean) {
30!
450
      taos_free_result(pStmt->exec.pRequest);
30✔
451
      pStmt->exec.pRequest = NULL;
30✔
452
    }
453

454
    size_t keyLen = 0;
30✔
455
    void*  pIter = taosHashIterate(pStmt->exec.pBlockHash, NULL);
30✔
456
    while (pIter) {
74✔
457
      STableDataCxt* pBlocks = *(STableDataCxt**)pIter;
44✔
458
      char*          key = taosHashGetKey(pIter, &keyLen);
44✔
459
      STableMeta*    pMeta = qGetTableMetaInDataBlock(pBlocks);
44✔
460

461
      if (keepTable && pBlocks == pStmt->exec.pCurrBlock) {
44✔
462
        TSWAP(pBlocks->pData, pStmt->exec.pCurrTbData);
21✔
463
        STMT_ERR_RET(qResetStmtDataBlock(pBlocks, false));
42!
464

465
        pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
21✔
466
        continue;
21✔
467
      }
468

469
      qDestroyStmtDataBlock(pBlocks);
23✔
470
      STMT_ERR_RET(taosHashRemove(pStmt->exec.pBlockHash, key, keyLen));
23!
471

472
      pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
23✔
473
    }
474

475
    if (keepTable) {
30✔
476
      return TSDB_CODE_SUCCESS;
21✔
477
    }
478

479
    taosHashCleanup(pStmt->exec.pBlockHash);
9✔
480
    pStmt->exec.pBlockHash = NULL;
9✔
481

482
    tDestroySubmitTbData(pStmt->exec.pCurrTbData, TSDB_MSG_FLG_ENCODE);
9✔
483
    taosMemoryFreeClear(pStmt->exec.pCurrTbData);
9!
484
  }
485

486
  STMT_ERR_RET(stmtCleanBindInfo(pStmt));
19!
487

488
  return TSDB_CODE_SUCCESS;
19✔
489
}
490

491
void stmtFreeTbBuf(void* buf) {
1✔
492
  void* pBuf = *(void**)buf;
1✔
493
  taosMemoryFree(pBuf);
1!
494
}
1✔
495

496
void stmtFreeTbCols(void* buf) {
1,000✔
497
  SArray* pCols = *(SArray**)buf;
1,000✔
498
  taosArrayDestroy(pCols);
1,000✔
499
}
1,000✔
500

501
int32_t stmtCleanSQLInfo(STscStmt* pStmt) {
10✔
502
  STMT_DLOG_E("start to free SQL info");
10!
503

504
  taosMemoryFree(pStmt->sql.pBindInfo);
10!
505
  taosMemoryFree(pStmt->sql.queryRes.fields);
10!
506
  taosMemoryFree(pStmt->sql.queryRes.userFields);
10!
507
  taosMemoryFree(pStmt->sql.sqlStr);
10!
508
  qDestroyQuery(pStmt->sql.pQuery);
10✔
509
  taosArrayDestroy(pStmt->sql.nodeList);
10✔
510
  taosHashCleanup(pStmt->sql.pVgHash);
10✔
511
  pStmt->sql.pVgHash = NULL;
10✔
512

513
  void* pIter = taosHashIterate(pStmt->sql.pTableCache, NULL);
10✔
514
  while (pIter) {
14✔
515
    SStmtTableCache* pCache = (SStmtTableCache*)pIter;
4✔
516

517
    qDestroyStmtDataBlock(pCache->pDataCtx);
4✔
518
    qDestroyBoundColInfo(pCache->boundTags);
4✔
519
    taosMemoryFreeClear(pCache->boundTags);
4!
520

521
    pIter = taosHashIterate(pStmt->sql.pTableCache, pIter);
4✔
522
  }
523
  taosHashCleanup(pStmt->sql.pTableCache);
10✔
524
  pStmt->sql.pTableCache = NULL;
10✔
525

526
  STMT_ERR_RET(stmtCleanExecInfo(pStmt, false, true));
10!
527
  STMT_ERR_RET(stmtCleanBindInfo(pStmt));
10!
528

529
  taos_free_result(pStmt->sql.siInfo.pRequest);
10✔
530
  taosHashCleanup(pStmt->sql.siInfo.pVgroupHash);
10✔
531
  tSimpleHashCleanup(pStmt->sql.siInfo.pTableHash);
10✔
532
  taosArrayDestroyEx(pStmt->sql.siInfo.tbBuf.pBufList, stmtFreeTbBuf);
10✔
533
  taosMemoryFree(pStmt->sql.siInfo.pTSchema);
10!
534
  qDestroyStmtDataBlock(pStmt->sql.siInfo.pDataCtx);
10✔
535
  taosArrayDestroyEx(pStmt->sql.siInfo.pTableCols, stmtFreeTbCols);
10✔
536

537
  (void)memset(&pStmt->sql, 0, sizeof(pStmt->sql));
10✔
538
  pStmt->sql.siInfo.tableColsReady = true;
10✔
539

540
  STMT_DLOG_E("end to free SQL info");
10!
541

542
  return TSDB_CODE_SUCCESS;
10✔
543
}
544

545
int32_t stmtTryAddTableVgroupInfo(STscStmt* pStmt, int32_t* vgId) {
16✔
546
  if (*vgId >= 0 && taosHashGet(pStmt->sql.pVgHash, (const char*)vgId, sizeof(*vgId))) {
16✔
547
    return TSDB_CODE_SUCCESS;
7✔
548
  }
549

550
  SVgroupInfo      vgInfo = {0};
9✔
551
  SRequestConnInfo conn = {.pTrans = pStmt->taos->pAppInfo->pTransporter,
9✔
552
                           .requestId = pStmt->exec.pRequest->requestId,
9✔
553
                           .requestObjRefId = pStmt->exec.pRequest->self,
9✔
554
                           .mgmtEps = getEpSet_s(&pStmt->taos->pAppInfo->mgmtEp)};
9✔
555

556
  int32_t code = catalogGetTableHashVgroup(pStmt->pCatalog, &conn, &pStmt->bInfo.sname, &vgInfo);
9✔
557
  if (TSDB_CODE_SUCCESS != code) {
9!
558
    return code;
×
559
  }
560

561
  code =
562
      taosHashPut(pStmt->sql.pVgHash, (const char*)&vgInfo.vgId, sizeof(vgInfo.vgId), (char*)&vgInfo, sizeof(vgInfo));
9✔
563
  if (TSDB_CODE_SUCCESS != code) {
9!
564
    return code;
×
565
  }
566

567
  *vgId = vgInfo.vgId;
9✔
568

569
  return TSDB_CODE_SUCCESS;
9✔
570
}
571

572
int32_t stmtRebuildDataBlock(STscStmt* pStmt, STableDataCxt* pDataBlock, STableDataCxt** newBlock, uint64_t uid,
16✔
573
                             uint64_t suid, int32_t vgId) {
574
  STMT_ERR_RET(stmtTryAddTableVgroupInfo(pStmt, &vgId));
16!
575
  STMT_ERR_RET(qRebuildStmtDataBlock(newBlock, pDataBlock, uid, suid, vgId, pStmt->sql.autoCreateTbl));
16!
576

577
  STMT_DLOG("uid:%" PRId64 ", rebuild table data context, vgId:%d", uid, vgId);
16!
578

579
  return TSDB_CODE_SUCCESS;
16✔
580
}
581

582
int32_t stmtGetFromCache(STscStmt* pStmt) {
22✔
583
  if (pStmt->sql.stbInterlaceMode && pStmt->sql.siInfo.pDataCtx) {
22!
584
    pStmt->bInfo.needParse = false;
×
585
    pStmt->bInfo.inExecCache = false;
×
586
    return TSDB_CODE_SUCCESS;
×
587
  }
588

589
  pStmt->bInfo.needParse = true;
22✔
590
  pStmt->bInfo.inExecCache = false;
22✔
591

592
  STableDataCxt** pCxtInExec = taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
22✔
593
  if (pCxtInExec) {
22!
594
    pStmt->bInfo.needParse = false;
×
595
    pStmt->bInfo.inExecCache = true;
×
596

597
    pStmt->exec.pCurrBlock = *pCxtInExec;
×
598

599
    if (pStmt->sql.autoCreateTbl) {
×
600
      tscDebug("reuse stmt block for tb %s in execBlock", pStmt->bInfo.tbFName);
×
601
      return TSDB_CODE_SUCCESS;
×
602
    }
603
  }
604

605
  if (NULL == pStmt->pCatalog) {
22✔
606
    STMT_ERR_RET(catalogGetHandle(pStmt->taos->pAppInfo->clusterId, &pStmt->pCatalog));
6!
607
    pStmt->sql.siInfo.pCatalog = pStmt->pCatalog;
6✔
608
  }
609

610
  if (NULL == pStmt->sql.pTableCache || taosHashGetSize(pStmt->sql.pTableCache) <= 0) {
22!
611
    if (pStmt->bInfo.inExecCache) {
6!
612
      pStmt->bInfo.needParse = false;
×
613
      tscDebug("reuse stmt block for tb %s in execBlock", pStmt->bInfo.tbFName);
×
614
      return TSDB_CODE_SUCCESS;
×
615
    }
616

617
    tscDebug("no stmt block cache for tb %s", pStmt->bInfo.tbFName);
6!
618
    return TSDB_CODE_SUCCESS;
6✔
619
  }
620

621
  if (pStmt->sql.autoCreateTbl) {
16✔
622
    SStmtTableCache* pCache = taosHashGet(pStmt->sql.pTableCache, &pStmt->bInfo.tbSuid, sizeof(pStmt->bInfo.tbSuid));
8✔
623
    if (pCache) {
8!
624
      pStmt->bInfo.needParse = false;
8✔
625
      pStmt->bInfo.tbUid = 0;
8✔
626

627
      STableDataCxt* pNewBlock = NULL;
8✔
628
      STMT_ERR_RET(stmtRebuildDataBlock(pStmt, pCache->pDataCtx, &pNewBlock, 0, pStmt->bInfo.tbSuid, -1));
8!
629

630
      if (taosHashPut(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName), &pNewBlock,
8!
631
                      POINTER_BYTES)) {
632
        STMT_ERR_RET(terrno);
×
633
      }
634

635
      pStmt->exec.pCurrBlock = pNewBlock;
8✔
636

637
      tscDebug("reuse stmt block for tb %s in sqlBlock, suid:0x%" PRIx64, pStmt->bInfo.tbFName, pStmt->bInfo.tbSuid);
8!
638

639
      return TSDB_CODE_SUCCESS;
8✔
640
    }
641

642
    STMT_RET(stmtCleanBindInfo(pStmt));
×
643
  }
644

645
  uint64_t uid, suid;
646
  int32_t  vgId;
647
  int8_t   tableType;
648

649
  STableMeta*      pTableMeta = NULL;
8✔
650
  SRequestConnInfo conn = {.pTrans = pStmt->taos->pAppInfo->pTransporter,
8✔
651
                           .requestId = pStmt->exec.pRequest->requestId,
8✔
652
                           .requestObjRefId = pStmt->exec.pRequest->self,
8✔
653
                           .mgmtEps = getEpSet_s(&pStmt->taos->pAppInfo->mgmtEp)};
8✔
654
  int32_t          code = catalogGetTableMeta(pStmt->pCatalog, &conn, &pStmt->bInfo.sname, &pTableMeta);
8✔
655

656
  pStmt->stat.ctgGetTbMetaNum++;
8✔
657

658
  if (TSDB_CODE_PAR_TABLE_NOT_EXIST == code) {
8!
659
    tscDebug("tb %s not exist", pStmt->bInfo.tbFName);
×
660
    STMT_ERR_RET(stmtCleanBindInfo(pStmt));
×
661

662
    STMT_ERR_RET(code);
×
663
  }
664

665
  STMT_ERR_RET(code);
8!
666

667
  uid = pTableMeta->uid;
8✔
668
  suid = pTableMeta->suid;
8✔
669
  tableType = pTableMeta->tableType;
8✔
670
  pStmt->bInfo.tbVgId = pTableMeta->vgId;
8✔
671
  vgId = pTableMeta->vgId;
8✔
672

673
  taosMemoryFree(pTableMeta);
8!
674

675
  uint64_t cacheUid = (TSDB_CHILD_TABLE == tableType) ? suid : uid;
8!
676

677
  if (uid == pStmt->bInfo.tbUid) {
8!
678
    pStmt->bInfo.needParse = false;
×
679

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

682
    return TSDB_CODE_SUCCESS;
×
683
  }
684

685
  if (pStmt->bInfo.inExecCache) {
8!
686
    SStmtTableCache* pCache = taosHashGet(pStmt->sql.pTableCache, &cacheUid, sizeof(cacheUid));
×
687
    if (NULL == pCache) {
×
688
      tscError("table [%s, %" PRIx64 ", %" PRIx64 "] found in exec blockHash, but not in sql blockHash",
×
689
               pStmt->bInfo.tbFName, uid, cacheUid);
690

691
      STMT_ERR_RET(TSDB_CODE_APP_ERROR);
×
692
    }
693

694
    pStmt->bInfo.needParse = false;
×
695

696
    pStmt->bInfo.tbUid = uid;
×
697
    pStmt->bInfo.tbSuid = suid;
×
698
    pStmt->bInfo.tbType = tableType;
×
699
    pStmt->bInfo.boundTags = pCache->boundTags;
×
700
    pStmt->bInfo.tagsCached = true;
×
701

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

704
    return TSDB_CODE_SUCCESS;
×
705
  }
706

707
  SStmtTableCache* pCache = taosHashGet(pStmt->sql.pTableCache, &cacheUid, sizeof(cacheUid));
8✔
708
  if (pCache) {
8!
709
    pStmt->bInfo.needParse = false;
8✔
710

711
    pStmt->bInfo.tbUid = uid;
8✔
712
    pStmt->bInfo.tbSuid = suid;
8✔
713
    pStmt->bInfo.tbType = tableType;
8✔
714
    pStmt->bInfo.boundTags = pCache->boundTags;
8✔
715
    pStmt->bInfo.tagsCached = true;
8✔
716

717
    STableDataCxt* pNewBlock = NULL;
8✔
718
    STMT_ERR_RET(stmtRebuildDataBlock(pStmt, pCache->pDataCtx, &pNewBlock, uid, suid, vgId));
8!
719

720
    if (taosHashPut(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName), &pNewBlock,
8!
721
                    POINTER_BYTES)) {
722
      STMT_ERR_RET(terrno);
×
723
    }
724

725
    pStmt->exec.pCurrBlock = pNewBlock;
8✔
726

727
    tscDebug("tb %s in sqlBlock list, set to current", pStmt->bInfo.tbFName);
8!
728

729
    return TSDB_CODE_SUCCESS;
8✔
730
  }
731

732
  STMT_ERR_RET(stmtCleanBindInfo(pStmt));
×
733

734
  return TSDB_CODE_SUCCESS;
×
735
}
736

737
int32_t stmtResetStmt(STscStmt* pStmt) {
1✔
738
  STMT_ERR_RET(stmtCleanSQLInfo(pStmt));
1!
739

740
  pStmt->sql.pTableCache = taosHashInit(100, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
1✔
741
  if (NULL == pStmt->sql.pTableCache) {
1!
742
    STMT_ERR_RET(terrno);
×
743
  }
744

745
  pStmt->sql.status = STMT_INIT;
1✔
746

747
  return TSDB_CODE_SUCCESS;
1✔
748
}
749

750
int32_t stmtAsyncOutput(STscStmt* pStmt, void* param) {
36✔
751
  SStmtQNode* pParam = (SStmtQNode*)param;
36✔
752

753
  if (pParam->restoreTbCols) {
36✔
754
    for (int32_t i = 0; i < pStmt->sql.siInfo.pTableColsIdx; ++i) {
36✔
755
      SArray** p = (SArray**)TARRAY_GET_ELEM(pStmt->sql.siInfo.pTableCols, i);
27✔
756
      *p = taosArrayInit(20, POINTER_BYTES);
27✔
757
      if (*p == NULL) {
27!
758
        STMT_ERR_RET(terrno);
×
759
      }
760
    }
761

762
    atomic_store_8((int8_t*)&pStmt->sql.siInfo.tableColsReady, true);
9✔
763
  } else {
764
    STMT_ERR_RET(qAppendStmtTableOutput(pStmt->sql.pQuery, pStmt->sql.pVgHash, &pParam->tblData, pStmt->exec.pCurrBlock,
27!
765
                                        &pStmt->sql.siInfo));
766

767
    // taosMemoryFree(pParam->pTbData);
768

769
    (void)atomic_sub_fetch_64(&pStmt->sql.siInfo.tbRemainNum, 1);
27✔
770
  }
771
  return TSDB_CODE_SUCCESS;
36✔
772
}
773

774
void* stmtBindThreadFunc(void* param) {
1✔
775
  setThreadName("stmtBind");
1✔
776

777
  qInfo("stmt bind thread started");
1!
778

779
  STscStmt* pStmt = (STscStmt*)param;
1✔
780

781
  while (true) {
37✔
782
    if (pStmt->queue.stopQueue) {
38✔
783
      break;
1✔
784
    }
785

786
    SStmtQNode* asyncParam = NULL;
37✔
787
    if (!stmtDequeue(pStmt, &asyncParam)) {
37✔
788
      continue;
1✔
789
    }
790

791
    int ret = stmtAsyncOutput(pStmt, asyncParam);
36✔
792
    if (ret != 0) {
36!
793
      qError("stmtAsyncOutput failed, reason:%s", tstrerror(ret));
×
794
    }
795
  }
796

797
  qInfo("stmt bind thread stopped");
1!
798

799
  return NULL;
1✔
800
}
801

802
int32_t stmtStartBindThread(STscStmt* pStmt) {
1✔
803
  TdThreadAttr thAttr;
804
  if (taosThreadAttrInit(&thAttr) != 0) {
1!
805
    return TSDB_CODE_TSC_INTERNAL_ERROR;
×
806
  }
807
  if (taosThreadAttrSetDetachState(&thAttr, PTHREAD_CREATE_JOINABLE) != 0) {
1!
808
    return TSDB_CODE_TSC_INTERNAL_ERROR;
×
809
  }
810

811
  if (taosThreadCreate(&pStmt->bindThread, &thAttr, stmtBindThreadFunc, pStmt) != 0) {
1!
812
    terrno = TAOS_SYSTEM_ERROR(errno);
×
813
    STMT_ERR_RET(terrno);
×
814
  }
815

816
  pStmt->bindThreadInUse = true;
1✔
817

818
  (void)taosThreadAttrDestroy(&thAttr);
1✔
819
  return TSDB_CODE_SUCCESS;
1✔
820
}
821

822
int32_t stmtInitQueue(STscStmt* pStmt) {
1✔
823
  (void)taosThreadCondInit(&pStmt->queue.waitCond, NULL);
1✔
824
  (void)taosThreadMutexInit(&pStmt->queue.mutex, NULL);
1✔
825
  STMT_ERR_RET(stmtAllocQNodeFromBuf(&pStmt->sql.siInfo.tbBuf, (void**)&pStmt->queue.head));
2!
826
  pStmt->queue.tail = pStmt->queue.head;
1✔
827

828
  return TSDB_CODE_SUCCESS;
1✔
829
}
830

831
int32_t stmtInitTableBuf(STableBufInfo* pTblBuf) {
1✔
832
  pTblBuf->buffUnit = sizeof(SStmtQNode);
1✔
833
  pTblBuf->buffSize = pTblBuf->buffUnit * 1000;
1✔
834
  pTblBuf->pBufList = taosArrayInit(100, POINTER_BYTES);
1✔
835
  if (NULL == pTblBuf->pBufList) {
1!
836
    return terrno;
×
837
  }
838
  void* buff = taosMemoryMalloc(pTblBuf->buffSize);
1!
839
  if (NULL == buff) {
1!
840
    return terrno;
×
841
  }
842

843
  if (taosArrayPush(pTblBuf->pBufList, &buff) == NULL) {
2!
844
    return terrno;
×
845
  }
846

847
  pTblBuf->pCurBuff = buff;
1✔
848
  pTblBuf->buffIdx = 1;
1✔
849
  pTblBuf->buffOffset = 0;
1✔
850

851
  return TSDB_CODE_SUCCESS;
1✔
852
}
853

854
TAOS_STMT* stmtInit(STscObj* taos, int64_t reqid, TAOS_STMT_OPTIONS* pOptions) {
10✔
855
  STscObj*  pObj = (STscObj*)taos;
10✔
856
  STscStmt* pStmt = NULL;
10✔
857
  int32_t   code = 0;
10✔
858

859
  pStmt = taosMemoryCalloc(1, sizeof(STscStmt));
10!
860
  if (NULL == pStmt) {
10!
861
    return NULL;
×
862
  }
863

864
  pStmt->sql.pTableCache = taosHashInit(100, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
10✔
865
  if (NULL == pStmt->sql.pTableCache) {
10!
866
    taosMemoryFree(pStmt);
×
867
    return NULL;
×
868
  }
869

870
  pStmt->taos = pObj;
10✔
871
  pStmt->bInfo.needParse = true;
10✔
872
  pStmt->sql.status = STMT_INIT;
10✔
873
  pStmt->reqid = reqid;
10✔
874
  pStmt->errCode = TSDB_CODE_SUCCESS;
10✔
875

876
  if (NULL != pOptions) {
10✔
877
    (void)memcpy(&pStmt->options, pOptions, sizeof(pStmt->options));
1✔
878
    if (pOptions->singleStbInsert && pOptions->singleTableBindOnce) {
1!
879
      pStmt->stbInterlaceMode = true;
1✔
880
    }
881
  }
882

883
  if (pStmt->stbInterlaceMode) {
10✔
884
    pStmt->sql.siInfo.transport = taos->pAppInfo->pTransporter;
1✔
885
    pStmt->sql.siInfo.acctId = taos->acctId;
1✔
886
    pStmt->sql.siInfo.dbname = taos->db;
1✔
887
    pStmt->sql.siInfo.mgmtEpSet = getEpSet_s(&pStmt->taos->pAppInfo->mgmtEp);
1✔
888
    pStmt->sql.siInfo.pTableHash = tSimpleHashInit(100, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY));
1✔
889
    if (NULL == pStmt->sql.siInfo.pTableHash) {
1!
890
      (void)stmtClose(pStmt);
×
891
      return NULL;
×
892
    }
893
    pStmt->sql.siInfo.pTableCols = taosArrayInit(STMT_TABLE_COLS_NUM, POINTER_BYTES);
1✔
894
    if (NULL == pStmt->sql.siInfo.pTableCols) {
1!
895
      (void)stmtClose(pStmt);
×
896
      return NULL;
×
897
    }
898

899
    code = stmtInitTableBuf(&pStmt->sql.siInfo.tbBuf);
1✔
900
    if (TSDB_CODE_SUCCESS == code) {
1!
901
      code = stmtInitQueue(pStmt);
1✔
902
    }
903
    if (TSDB_CODE_SUCCESS == code) {
1!
904
      code = stmtStartBindThread(pStmt);
1✔
905
    }
906
    if (TSDB_CODE_SUCCESS != code) {
1!
907
      terrno = code;
×
908
      (void)stmtClose(pStmt);
×
909
      return NULL;
×
910
    }
911
  }
912

913
  pStmt->sql.siInfo.tableColsReady = true;
10✔
914

915
  STMT_LOG_SEQ(STMT_INIT);
10!
916

917
  tscDebug("stmt:%p initialized", pStmt);
10!
918

919
  return pStmt;
10✔
920
}
921

922
int stmtPrepare(TAOS_STMT* stmt, const char* sql, unsigned long length) {
12✔
923
  STscStmt* pStmt = (STscStmt*)stmt;
12✔
924

925
  STMT_DLOG_E("start to prepare");
12!
926

927
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
12!
928
    return pStmt->errCode;
×
929
  }
930

931
  if (pStmt->sql.status >= STMT_PREPARE) {
12✔
932
    STMT_ERR_RET(stmtResetStmt(pStmt));
1!
933
  }
934

935
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_PREPARE));
12!
936

937
  if (length <= 0) {
12!
938
    length = strlen(sql);
12✔
939
  }
940

941
  pStmt->sql.sqlStr = taosStrndup(sql, length);
12!
942
  if (!pStmt->sql.sqlStr) {
12!
943
    return terrno;
×
944
  }
945
  pStmt->sql.sqlLen = length;
12✔
946
  pStmt->sql.stbInterlaceMode = pStmt->stbInterlaceMode;
12✔
947

948
  char* dbName = NULL;
12✔
949
  if (qParseDbName(sql, length, &dbName)) {
12✔
950
    STMT_ERR_RET(stmtSetDbName(stmt, dbName));
3!
951
    taosMemoryFreeClear(dbName);
3!
952
  }
953

954
  return TSDB_CODE_SUCCESS;
12✔
955
}
956

957
int32_t stmtInitStbInterlaceTableInfo(STscStmt* pStmt) {
1✔
958
  STableDataCxt** pSrc = taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
1✔
959
  if (!pSrc) {
1!
960
    return terrno;
×
961
  }
962
  STableDataCxt* pDst = NULL;
1✔
963

964
  STMT_ERR_RET(qCloneStmtDataBlock(&pDst, *pSrc, true));
1!
965
  pStmt->sql.siInfo.pDataCtx = pDst;
1✔
966

967
  SArray* pTblCols = NULL;
1✔
968
  for (int32_t i = 0; i < STMT_TABLE_COLS_NUM; i++) {
1,001✔
969
    pTblCols = taosArrayInit(20, POINTER_BYTES);
1,000✔
970
    if (NULL == pTblCols) {
1,000!
971
      return terrno;
×
972
    }
973

974
    if (taosArrayPush(pStmt->sql.siInfo.pTableCols, &pTblCols) == NULL) {
2,000!
975
      return terrno;
×
976
    }
977
  }
978

979
  pStmt->sql.siInfo.boundTags = pStmt->bInfo.boundTags;
1✔
980

981
  return TSDB_CODE_SUCCESS;
1✔
982
}
983

984
int stmtSetDbName(TAOS_STMT* stmt, const char* dbName) {
3✔
985
  STscStmt* pStmt = (STscStmt*)stmt;
3✔
986

987
  STMT_DLOG("start to set dbName:%s", dbName);
3!
988

989
  STMT_ERR_RET(stmtCreateRequest(pStmt));
3!
990

991
  // The SQL statement specifies a database name, overriding the previously specified database
992
  taosMemoryFreeClear(pStmt->exec.pRequest->pDb);
3!
993
  pStmt->exec.pRequest->pDb = taosStrdup(dbName);
3!
994
  if (pStmt->exec.pRequest->pDb == NULL) {
3!
995
    return terrno;
×
996
  }
997
  return TSDB_CODE_SUCCESS;
3✔
998
}
999

1000
int stmtSetTbName(TAOS_STMT* stmt, const char* tbName) {
30✔
1001
  STscStmt* pStmt = (STscStmt*)stmt;
30✔
1002

1003
  int64_t startUs = taosGetTimestampUs();
30✔
1004

1005
  STMT_DLOG("start to set tbName:%s", tbName);
30!
1006

1007
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
30!
1008
    return pStmt->errCode;
×
1009
  }
1010

1011
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_SETTBNAME));
30!
1012

1013
  int32_t insert = 0;
30✔
1014
  STMT_ERR_RET(stmtIsInsert(stmt, &insert));
30!
1015
  if (0 == insert) {
30!
1016
    tscError("set tb name not available for none insert statement");
×
1017
    STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR);
×
1018
  }
1019

1020
  if (!pStmt->sql.stbInterlaceMode || NULL == pStmt->sql.siInfo.pDataCtx) {
30✔
1021
    STMT_ERR_RET(stmtCreateRequest(pStmt));
22!
1022

1023
    STMT_ERR_RET(qCreateSName(&pStmt->bInfo.sname, tbName, pStmt->taos->acctId, pStmt->exec.pRequest->pDb,
22!
1024
                              pStmt->exec.pRequest->msgBuf, pStmt->exec.pRequest->msgBufLen));
1025
    STMT_ERR_RET(tNameExtractFullName(&pStmt->bInfo.sname, pStmt->bInfo.tbFName));
22!
1026

1027
    STMT_ERR_RET(stmtGetFromCache(pStmt));
22!
1028

1029
    if (pStmt->bInfo.needParse) {
22✔
1030
      tstrncpy(pStmt->bInfo.tbName, tbName, sizeof(pStmt->bInfo.tbName));
6✔
1031
      pStmt->bInfo.tbName[sizeof(pStmt->bInfo.tbName) - 1] = 0;
6✔
1032

1033
      STMT_ERR_RET(stmtParseSql(pStmt));
6!
1034
    }
1035
  } else {
1036
    tstrncpy(pStmt->bInfo.tbName, tbName, sizeof(pStmt->bInfo.tbName));
8✔
1037
    pStmt->bInfo.tbName[sizeof(pStmt->bInfo.tbName) - 1] = 0;
8✔
1038
    pStmt->exec.pRequest->requestId++;
8✔
1039
    pStmt->bInfo.needParse = false;
8✔
1040
  }
1041

1042
  if (pStmt->sql.stbInterlaceMode && NULL == pStmt->sql.siInfo.pDataCtx) {
30✔
1043
    STMT_ERR_RET(stmtInitStbInterlaceTableInfo(pStmt));
1!
1044
  }
1045

1046
  int64_t startUs2 = taosGetTimestampUs();
30✔
1047
  pStmt->stat.setTbNameUs += startUs2 - startUs;
30✔
1048

1049
  return TSDB_CODE_SUCCESS;
30✔
1050
}
1051

1052
int stmtSetTbTags(TAOS_STMT* stmt, TAOS_MULTI_BIND* tags) {
11✔
1053
  STscStmt* pStmt = (STscStmt*)stmt;
11✔
1054

1055
  STMT_DLOG_E("start to set tbTags");
11!
1056

1057
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
11!
1058
    return pStmt->errCode;
×
1059
  }
1060

1061
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_SETTAGS));
11!
1062

1063
  SBoundColInfo* tags_info = (SBoundColInfo*)pStmt->bInfo.boundTags;
11✔
1064
  if (tags_info->numOfBound <= 0 || tags_info->numOfCols <= 0) {
11!
1065
    tscWarn("no tags bound in sql, will not bound tags");
×
1066
    return TSDB_CODE_SUCCESS;
×
1067
  }
1068

1069
  if (pStmt->bInfo.inExecCache) {
11!
1070
    return TSDB_CODE_SUCCESS;
×
1071
  }
1072

1073
  STableDataCxt** pDataBlock =
1074
      (STableDataCxt**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
11✔
1075
  if (NULL == pDataBlock) {
11!
1076
    tscError("table %s not found in exec blockHash", pStmt->bInfo.tbFName);
×
1077
    STMT_ERR_RET(TSDB_CODE_APP_ERROR);
×
1078
  }
1079

1080
  tscDebug("start to bind stmt tag values");
11!
1081
  STMT_ERR_RET(qBindStmtTagsValue(*pDataBlock, pStmt->bInfo.boundTags, pStmt->bInfo.tbSuid, pStmt->bInfo.stbFName,
11!
1082
                                  pStmt->bInfo.sname.tname, tags, pStmt->exec.pRequest->msgBuf,
1083
                                  pStmt->exec.pRequest->msgBufLen, pStmt->taos->optionInfo.charsetCxt));
1084

1085
  return TSDB_CODE_SUCCESS;
11✔
1086
}
1087

1088
int stmtFetchTagFields(STscStmt* pStmt, int32_t* fieldNum, TAOS_FIELD_E** fields) {
2✔
1089
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
2✔
1090
    return pStmt->errCode;
1✔
1091
  }
1092

1093
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
1!
1094
    tscError("invalid operation to get query tag fileds");
×
1095
    STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR);
×
1096
  }
1097

1098
  STableDataCxt** pDataBlock =
1099
      (STableDataCxt**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
1✔
1100
  if (NULL == pDataBlock) {
1!
1101
    tscError("table %s not found in exec blockHash", pStmt->bInfo.tbFName);
×
1102
    STMT_ERR_RET(TSDB_CODE_APP_ERROR);
×
1103
  }
1104

1105
  STMT_ERR_RET(qBuildStmtTagFields(*pDataBlock, pStmt->bInfo.boundTags, fieldNum, fields));
1!
1106

1107
  return TSDB_CODE_SUCCESS;
1✔
1108
}
1109

1110
int stmtFetchColFields(STscStmt* pStmt, int32_t* fieldNum, TAOS_FIELD_E** fields) {
6✔
1111
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
6!
1112
    return pStmt->errCode;
×
1113
  }
1114

1115
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
6!
1116
    tscError("invalid operation to get query column fileds");
×
1117
    STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR);
×
1118
  }
1119

1120
  STableDataCxt** pDataBlock = NULL;
6✔
1121

1122
  if (pStmt->sql.stbInterlaceMode) {
6!
1123
    pDataBlock = &pStmt->sql.siInfo.pDataCtx;
×
1124
  } else {
1125
    pDataBlock =
1126
        (STableDataCxt**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
6✔
1127
    if (NULL == pDataBlock) {
6!
1128
      tscError("table %s not found in exec blockHash", pStmt->bInfo.tbFName);
×
1129
      STMT_ERR_RET(TSDB_CODE_APP_ERROR);
×
1130
    }
1131
  }
1132

1133
  STMT_ERR_RET(qBuildStmtColFields(*pDataBlock, fieldNum, fields));
6!
1134

1135
  return TSDB_CODE_SUCCESS;
6✔
1136
}
1137

1138
/*
1139
SArray* stmtGetFreeCol(STscStmt* pStmt, int32_t* idx) {
1140
  while (true) {
1141
    if (pStmt->exec.smInfo.pColIdx >= STMT_COL_BUF_SIZE) {
1142
      pStmt->exec.smInfo.pColIdx = 0;
1143
    }
1144

1145
    if ((pStmt->exec.smInfo.pColIdx + 1) == atomic_load_32(&pStmt->exec.smInfo.pColFreeIdx)) {
1146
      taosUsleep(1);
1147
      continue;
1148
    }
1149

1150
    *idx = pStmt->exec.smInfo.pColIdx;
1151
    return pStmt->exec.smInfo.pCols[pStmt->exec.smInfo.pColIdx++];
1152
  }
1153
}
1154
*/
1155

1156
int32_t stmtAppendTablePostHandle(STscStmt* pStmt, SStmtQNode* param) {
27✔
1157
  if (NULL == pStmt->sql.siInfo.pVgroupHash) {
27✔
1158
    pStmt->sql.siInfo.pVgroupHash =
9✔
1159
        taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_ENTRY_LOCK);
9✔
1160
  }
1161
  if (NULL == pStmt->sql.siInfo.pVgroupList) {
27✔
1162
    pStmt->sql.siInfo.pVgroupList = taosArrayInit(64, POINTER_BYTES);
9✔
1163
  }
1164

1165
  if (NULL == pStmt->sql.siInfo.pRequest) {
27✔
1166
    STMT_ERR_RET(buildRequest(pStmt->taos->id, pStmt->sql.sqlStr, pStmt->sql.sqlLen, NULL, false,
1!
1167
                              (SRequestObj**)&pStmt->sql.siInfo.pRequest, pStmt->reqid));
1168

1169
    if (pStmt->reqid != 0) {
1!
1170
      pStmt->reqid++;
×
1171
    }
1172
    pStmt->exec.pRequest->syncQuery = true;
1✔
1173

1174
    pStmt->sql.siInfo.requestId = ((SRequestObj*)pStmt->sql.siInfo.pRequest)->requestId;
1✔
1175
    pStmt->sql.siInfo.requestSelf = ((SRequestObj*)pStmt->sql.siInfo.pRequest)->self;
1✔
1176
  }
1177

1178
  if (!pStmt->sql.siInfo.tbFromHash && pStmt->sql.siInfo.firstName[0] &&
27✔
1179
      0 == strcmp(pStmt->sql.siInfo.firstName, pStmt->bInfo.tbName)) {
1!
1180
    pStmt->sql.siInfo.tbFromHash = true;
1✔
1181
  }
1182

1183
  if (0 == pStmt->sql.siInfo.firstName[0]) {
27✔
1184
    tstrncpy(pStmt->sql.siInfo.firstName, pStmt->bInfo.tbName, TSDB_TABLE_NAME_LEN);
1✔
1185
  }
1186

1187
  param->tblData.getFromHash = pStmt->sql.siInfo.tbFromHash;
27✔
1188
  param->next = NULL;
27✔
1189

1190
  (void)atomic_add_fetch_64(&pStmt->sql.siInfo.tbRemainNum, 1);
27✔
1191

1192
  stmtEnqueue(pStmt, param);
27✔
1193

1194
  return TSDB_CODE_SUCCESS;
27✔
1195
}
1196

1197
static FORCE_INLINE int32_t stmtGetTableColsFromCache(STscStmt* pStmt, SArray** pTableCols) {
1198
  while (true) {
1199
    if (pStmt->sql.siInfo.pTableColsIdx < taosArrayGetSize(pStmt->sql.siInfo.pTableCols)) {
27!
1200
      *pTableCols = (SArray*)taosArrayGetP(pStmt->sql.siInfo.pTableCols, pStmt->sql.siInfo.pTableColsIdx++);
27✔
1201
      break;
27✔
1202
    } else {
1203
      SArray* pTblCols = NULL;
×
1204
      for (int32_t i = 0; i < 100; i++) {
×
1205
        pTblCols = taosArrayInit(20, POINTER_BYTES);
×
1206
        if (NULL == pTblCols) {
×
1207
          return terrno;
×
1208
        }
1209

1210
        if (taosArrayPush(pStmt->sql.siInfo.pTableCols, &pTblCols) == NULL) {
×
1211
          return terrno;
×
1212
        }
1213
      }
1214
    }
1215
  }
1216

1217
  return TSDB_CODE_SUCCESS;
27✔
1218
}
1219

1220
int stmtBindBatch(TAOS_STMT* stmt, TAOS_MULTI_BIND* bind, int32_t colIdx) {
85✔
1221
  STscStmt* pStmt = (STscStmt*)stmt;
85✔
1222
  int32_t   code = 0;
85✔
1223

1224
  int64_t startUs = taosGetTimestampUs();
85✔
1225

1226
  STMT_DLOG("start to bind stmt data, colIdx:%d", colIdx);
85!
1227

1228
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
85!
1229
    return pStmt->errCode;
×
1230
  }
1231

1232
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_BIND));
85!
1233

1234
  if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 &&
85!
1235
      STMT_TYPE_MULTI_INSERT != pStmt->sql.type) {
×
1236
    pStmt->bInfo.needParse = false;
×
1237
  }
1238

1239
  if (pStmt->exec.pRequest && STMT_TYPE_QUERY == pStmt->sql.type && pStmt->sql.runTimes) {
85!
1240
    taos_free_result(pStmt->exec.pRequest);
×
1241
    pStmt->exec.pRequest = NULL;
×
1242
  }
1243

1244
  STMT_ERR_RET(stmtCreateRequest(pStmt));
85!
1245

1246
  if (pStmt->bInfo.needParse) {
85✔
1247
    STMT_ERR_RET(stmtParseSql(pStmt));
1!
1248
  }
1249

1250
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
85!
1251
    STMT_ERR_RET(qStmtBindParams(pStmt->sql.pQuery, bind, colIdx, pStmt->taos->optionInfo.charsetCxt));
×
1252

1253
    SParseContext ctx = {.requestId = pStmt->exec.pRequest->requestId,
×
1254
                         .acctId = pStmt->taos->acctId,
×
1255
                         .db = pStmt->exec.pRequest->pDb,
×
1256
                         .topicQuery = false,
1257
                         .pSql = pStmt->sql.sqlStr,
×
1258
                         .sqlLen = pStmt->sql.sqlLen,
×
1259
                         .pMsg = pStmt->exec.pRequest->msgBuf,
×
1260
                         .msgLen = ERROR_MSG_BUF_DEFAULT_SIZE,
1261
                         .pTransporter = pStmt->taos->pAppInfo->pTransporter,
×
1262
                         .pStmtCb = NULL,
1263
                         .pUser = pStmt->taos->user,
×
1264
                         .setQueryFp = setQueryRequest};
1265

1266
    ctx.mgmtEpSet = getEpSet_s(&pStmt->taos->pAppInfo->mgmtEp);
×
1267
    STMT_ERR_RET(catalogGetHandle(pStmt->taos->pAppInfo->clusterId, &ctx.pCatalog));
×
1268

1269
    STMT_ERR_RET(qStmtParseQuerySql(&ctx, pStmt->sql.pQuery));
×
1270

1271
    if (pStmt->sql.pQuery->haveResultSet) {
×
1272
      STMT_ERR_RET(setResSchemaInfo(&pStmt->exec.pRequest->body.resInfo, pStmt->sql.pQuery->pResSchema,
×
1273
                                    pStmt->sql.pQuery->numOfResCols));
1274
      taosMemoryFreeClear(pStmt->sql.pQuery->pResSchema);
×
1275
      setResPrecision(&pStmt->exec.pRequest->body.resInfo, pStmt->sql.pQuery->precision);
×
1276
    }
1277

1278
    TSWAP(pStmt->exec.pRequest->dbList, pStmt->sql.pQuery->pDbList);
×
1279
    TSWAP(pStmt->exec.pRequest->tableList, pStmt->sql.pQuery->pTableList);
×
1280
    TSWAP(pStmt->exec.pRequest->targetTableList, pStmt->sql.pQuery->pTargetTableList);
×
1281

1282
    // if (STMT_TYPE_QUERY == pStmt->sql.queryRes) {
1283
    //   STMT_ERR_RET(stmtRestoreQueryFields(pStmt));
1284
    // }
1285

1286
    // STMT_ERR_RET(stmtBackupQueryFields(pStmt));
1287

1288
    return TSDB_CODE_SUCCESS;
×
1289
  }
1290

1291
  if (pStmt->sql.stbInterlaceMode && NULL == pStmt->sql.siInfo.pDataCtx) {
85!
1292
    STMT_ERR_RET(stmtInitStbInterlaceTableInfo(pStmt));
×
1293
  }
1294

1295
  STableDataCxt** pDataBlock = NULL;
85✔
1296

1297
  if (pStmt->exec.pCurrBlock) {
85✔
1298
    pDataBlock = &pStmt->exec.pCurrBlock;
79✔
1299
  } else {
1300
    pDataBlock =
1301
        (STableDataCxt**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
6✔
1302
    if (NULL == pDataBlock) {
6!
1303
      tscError("table %s not found in exec blockHash", pStmt->bInfo.tbFName);
×
1304
      STMT_ERR_RET(TSDB_CODE_TSC_STMT_CACHE_ERROR);
×
1305
    }
1306
    pStmt->exec.pCurrBlock = *pDataBlock;
6✔
1307
    if (pStmt->sql.stbInterlaceMode) {
6✔
1308
      taosArrayDestroy(pStmt->exec.pCurrBlock->pData->aCol);
1✔
1309
      pStmt->exec.pCurrBlock->pData->aCol = NULL;
1✔
1310
    }
1311
  }
1312

1313
  int64_t startUs2 = taosGetTimestampUs();
85✔
1314
  pStmt->stat.bindDataUs1 += startUs2 - startUs;
85✔
1315

1316
  SStmtQNode* param = NULL;
85✔
1317
  if (pStmt->sql.stbInterlaceMode) {
85✔
1318
    STMT_ERR_RET(stmtAllocQNodeFromBuf(&pStmt->sql.siInfo.tbBuf, (void**)&param));
54!
1319
    STMT_ERR_RET(stmtGetTableColsFromCache(pStmt, &param->tblData.aCol));
54!
1320
    taosArrayClear(param->tblData.aCol);
27✔
1321

1322
    // param->tblData.aCol = taosArrayInit(20, POINTER_BYTES);
1323

1324
    param->restoreTbCols = false;
27✔
1325
    tstrncpy(param->tblData.tbName, pStmt->bInfo.tbName, TSDB_TABLE_NAME_LEN);
27✔
1326
  }
1327

1328
  int64_t startUs3 = taosGetTimestampUs();
85✔
1329
  pStmt->stat.bindDataUs2 += startUs3 - startUs2;
85✔
1330

1331
  SArray* pCols = pStmt->sql.stbInterlaceMode ? param->tblData.aCol : (*pDataBlock)->pData->aCol;
85✔
1332

1333
  if (colIdx < 0) {
85!
1334
    if (pStmt->sql.stbInterlaceMode) {
85✔
1335
      (*pDataBlock)->pData->flags = 0;
27✔
1336
      code =
1337
          qBindStmtStbColsValue(*pDataBlock, pCols, bind, pStmt->exec.pRequest->msgBuf, pStmt->exec.pRequest->msgBufLen,
27✔
1338
                                &pStmt->sql.siInfo.pTSchema, pStmt->sql.pBindInfo, pStmt->taos->optionInfo.charsetCxt);
27✔
1339
    } else {
1340
      code = qBindStmtColsValue(*pDataBlock, pCols, bind, pStmt->exec.pRequest->msgBuf, pStmt->exec.pRequest->msgBufLen,
58✔
1341
                                pStmt->taos->optionInfo.charsetCxt);
58✔
1342
    }
1343

1344
    if (code) {
85✔
1345
      tscError("qBindStmtColsValue failed, error:%s", tstrerror(code));
1!
1346
      STMT_ERR_RET(code);
1!
1347
    }
1348
  } else {
1349
    if (pStmt->sql.stbInterlaceMode) {
×
1350
      tscError("bind single column not allowed in stb insert mode");
×
1351
      STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR);
×
1352
    }
1353

1354
    if (colIdx != (pStmt->bInfo.sBindLastIdx + 1) && colIdx != 0) {
×
1355
      tscError("bind column index not in sequence");
×
1356
      STMT_ERR_RET(TSDB_CODE_APP_ERROR);
×
1357
    }
1358

1359
    pStmt->bInfo.sBindLastIdx = colIdx;
×
1360

1361
    if (0 == colIdx) {
×
1362
      pStmt->bInfo.sBindRowNum = bind->num;
×
1363
    }
1364

1365
    code =
1366
        qBindStmtSingleColValue(*pDataBlock, pCols, bind, pStmt->exec.pRequest->msgBuf, pStmt->exec.pRequest->msgBufLen,
×
1367
                                colIdx, pStmt->bInfo.sBindRowNum, pStmt->taos->optionInfo.charsetCxt);
×
1368
    if (code) {
×
1369
      tscError("qBindStmtSingleColValue failed, error:%s", tstrerror(code));
×
1370
      STMT_ERR_RET(code);
×
1371
    }
1372
  }
1373

1374
  int64_t startUs4 = taosGetTimestampUs();
84✔
1375
  pStmt->stat.bindDataUs3 += startUs4 - startUs3;
84✔
1376

1377
  if (pStmt->sql.stbInterlaceMode) {
84✔
1378
    STMT_ERR_RET(stmtAppendTablePostHandle(pStmt, param));
27!
1379
  }
1380

1381
  pStmt->stat.bindDataUs4 += taosGetTimestampUs() - startUs4;
84✔
1382

1383
  return TSDB_CODE_SUCCESS;
84✔
1384
}
1385

1386
int stmtAddBatch(TAOS_STMT* stmt) {
30✔
1387
  STscStmt* pStmt = (STscStmt*)stmt;
30✔
1388

1389
  int64_t startUs = taosGetTimestampUs();
30✔
1390

1391
  STMT_DLOG_E("start to add batch");
30!
1392

1393
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
30!
1394
    return pStmt->errCode;
×
1395
  }
1396

1397
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_ADD_BATCH));
30!
1398

1399
  if (pStmt->sql.stbInterlaceMode) {
30✔
1400
    int64_t startUs2 = taosGetTimestampUs();
9✔
1401
    pStmt->stat.addBatchUs += startUs2 - startUs;
9✔
1402

1403
    pStmt->sql.siInfo.tableColsReady = false;
9✔
1404

1405
    SStmtQNode* param = NULL;
9✔
1406
    STMT_ERR_RET(stmtAllocQNodeFromBuf(&pStmt->sql.siInfo.tbBuf, (void**)&param));
18!
1407
    param->restoreTbCols = true;
9✔
1408
    param->next = NULL;
9✔
1409

1410
    stmtEnqueue(pStmt, param);
9✔
1411

1412
    return TSDB_CODE_SUCCESS;
9✔
1413
  }
1414

1415
  STMT_ERR_RET(stmtCacheBlock(pStmt));
21!
1416

1417
  return TSDB_CODE_SUCCESS;
21✔
1418
}
1419
/*
1420
int stmtUpdateTableUid(STscStmt* pStmt, SSubmitRsp* pRsp) {
1421
  tscDebug("stmt start to update tbUid, blockNum:%d", pRsp->nBlocks);
1422

1423
  int32_t code = 0;
1424
  int32_t finalCode = 0;
1425
  size_t  keyLen = 0;
1426
  void*   pIter = taosHashIterate(pStmt->exec.pBlockHash, NULL);
1427
  while (pIter) {
1428
    STableDataCxt* pBlock = *(STableDataCxt**)pIter;
1429
    char*          key = taosHashGetKey(pIter, &keyLen);
1430

1431
    STableMeta* pMeta = qGetTableMetaInDataBlock(pBlock);
1432
    if (pMeta->uid) {
1433
      pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
1434
      continue;
1435
    }
1436

1437
    SSubmitBlkRsp* blkRsp = NULL;
1438
    int32_t        i = 0;
1439
    for (; i < pRsp->nBlocks; ++i) {
1440
      blkRsp = pRsp->pBlocks + i;
1441
      if (strlen(blkRsp->tblFName) != keyLen) {
1442
        continue;
1443
      }
1444

1445
      if (strncmp(blkRsp->tblFName, key, keyLen)) {
1446
        continue;
1447
      }
1448

1449
      break;
1450
    }
1451

1452
    if (i < pRsp->nBlocks) {
1453
      tscDebug("auto created table %s uid updated from %" PRIx64 " to %" PRIx64, blkRsp->tblFName, pMeta->uid,
1454
               blkRsp->uid);
1455

1456
      pMeta->uid = blkRsp->uid;
1457
      pStmt->bInfo.tbUid = blkRsp->uid;
1458
    } else {
1459
      tscDebug("table %s not found in submit rsp, will update from catalog", pStmt->bInfo.tbFName);
1460
      if (NULL == pStmt->pCatalog) {
1461
        code = catalogGetHandle(pStmt->taos->pAppInfo->clusterId, &pStmt->pCatalog);
1462
        if (code) {
1463
          pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
1464
          finalCode = code;
1465
          continue;
1466
        }
1467
      }
1468

1469
      code = stmtCreateRequest(pStmt);
1470
      if (code) {
1471
        pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
1472
        finalCode = code;
1473
        continue;
1474
      }
1475

1476
      STableMeta*      pTableMeta = NULL;
1477
      SRequestConnInfo conn = {.pTrans = pStmt->taos->pAppInfo->pTransporter,
1478
                               .requestId = pStmt->exec.pRequest->requestId,
1479
                               .requestObjRefId = pStmt->exec.pRequest->self,
1480
                               .mgmtEps = getEpSet_s(&pStmt->taos->pAppInfo->mgmtEp)};
1481
      code = catalogGetTableMeta(pStmt->pCatalog, &conn, &pStmt->bInfo.sname, &pTableMeta);
1482

1483
      pStmt->stat.ctgGetTbMetaNum++;
1484

1485
      taos_free_result(pStmt->exec.pRequest);
1486
      pStmt->exec.pRequest = NULL;
1487

1488
      if (code || NULL == pTableMeta) {
1489
        pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
1490
        finalCode = code;
1491
        taosMemoryFree(pTableMeta);
1492
        continue;
1493
      }
1494

1495
      pMeta->uid = pTableMeta->uid;
1496
      pStmt->bInfo.tbUid = pTableMeta->uid;
1497
      taosMemoryFree(pTableMeta);
1498
    }
1499

1500
    pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
1501
  }
1502

1503
  return finalCode;
1504
}
1505
*/
1506

1507
/*
1508
int stmtStaticModeExec(TAOS_STMT* stmt) {
1509
  STscStmt*   pStmt = (STscStmt*)stmt;
1510
  int32_t     code = 0;
1511
  SSubmitRsp* pRsp = NULL;
1512
  if (pStmt->sql.staticMode) {
1513
    return TSDB_CODE_TSC_STMT_API_ERROR;
1514
  }
1515

1516
  STMT_DLOG_E("start to exec");
1517

1518
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_EXECUTE));
1519

1520
  STMT_ERR_RET(qBuildStmtOutputFromTbList(pStmt->sql.pQuery, pStmt->sql.pVgHash, pStmt->exec.pTbBlkList,
1521
pStmt->exec.pCurrBlock, pStmt->exec.tbBlkNum));
1522

1523
  launchQueryImpl(pStmt->exec.pRequest, pStmt->sql.pQuery, true, NULL);
1524

1525
  if (pStmt->exec.pRequest->code && NEED_CLIENT_HANDLE_ERROR(pStmt->exec.pRequest->code)) {
1526
    code = refreshMeta(pStmt->exec.pRequest->pTscObj, pStmt->exec.pRequest);
1527
    if (code) {
1528
      pStmt->exec.pRequest->code = code;
1529
    } else {
1530
      tFreeSSubmitRsp(pRsp);
1531
      STMT_ERR_RET(stmtResetStmt(pStmt));
1532
      STMT_ERR_RET(TSDB_CODE_NEED_RETRY);
1533
    }
1534
  }
1535

1536
  STMT_ERR_JRET(pStmt->exec.pRequest->code);
1537

1538
  pStmt->exec.affectedRows = taos_affected_rows(pStmt->exec.pRequest);
1539
  pStmt->affectedRows += pStmt->exec.affectedRows;
1540

1541
_return:
1542

1543
  stmtCleanExecInfo(pStmt, (code ? false : true), false);
1544

1545
  tFreeSSubmitRsp(pRsp);
1546

1547
  ++pStmt->sql.runTimes;
1548

1549
  STMT_RET(code);
1550
}
1551
*/
1552

1553
int stmtExec(TAOS_STMT* stmt) {
30✔
1554
  STscStmt*   pStmt = (STscStmt*)stmt;
30✔
1555
  int32_t     code = 0;
30✔
1556
  SSubmitRsp* pRsp = NULL;
30✔
1557

1558
  int64_t startUs = taosGetTimestampUs();
30✔
1559

1560
  STMT_DLOG_E("start to exec");
30!
1561

1562
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
30!
1563
    return pStmt->errCode;
×
1564
  }
1565

1566
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_EXECUTE));
30!
1567

1568
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
30!
1569
    launchQueryImpl(pStmt->exec.pRequest, pStmt->sql.pQuery, true, NULL);
×
1570
  } else {
1571
    if (pStmt->sql.stbInterlaceMode) {
30✔
1572
      int64_t startTs = taosGetTimestampUs();
9✔
1573
      while (atomic_load_64(&pStmt->sql.siInfo.tbRemainNum)) {
19✔
1574
        taosUsleep(1);
10✔
1575
      }
1576
      pStmt->stat.execWaitUs += taosGetTimestampUs() - startTs;
9✔
1577

1578
      STMT_ERR_RET(qBuildStmtFinOutput(pStmt->sql.pQuery, pStmt->sql.pVgHash, pStmt->sql.siInfo.pVgroupList));
9!
1579
      taosHashCleanup(pStmt->sql.siInfo.pVgroupHash);
9✔
1580
      pStmt->sql.siInfo.pVgroupHash = NULL;
9✔
1581
      pStmt->sql.siInfo.pVgroupList = NULL;
9✔
1582
    } else {
1583
      tDestroySubmitTbData(pStmt->exec.pCurrTbData, TSDB_MSG_FLG_ENCODE);
21✔
1584
      taosMemoryFreeClear(pStmt->exec.pCurrTbData);
21!
1585

1586
      STMT_ERR_RET(qCloneCurrentTbData(pStmt->exec.pCurrBlock, &pStmt->exec.pCurrTbData));
21!
1587

1588
      STMT_ERR_RET(qBuildStmtOutput(pStmt->sql.pQuery, pStmt->sql.pVgHash, pStmt->exec.pBlockHash));
21!
1589
    }
1590

1591
    launchQueryImpl(pStmt->exec.pRequest, pStmt->sql.pQuery, true, NULL);
30✔
1592
  }
1593

1594
  if (pStmt->exec.pRequest->code && NEED_CLIENT_HANDLE_ERROR(pStmt->exec.pRequest->code)) {
30!
1595
    code = refreshMeta(pStmt->exec.pRequest->pTscObj, pStmt->exec.pRequest);
×
1596
    if (code) {
×
1597
      pStmt->exec.pRequest->code = code;
×
1598
    } else {
1599
      tFreeSSubmitRsp(pRsp);
×
1600
      STMT_ERR_RET(stmtResetStmt(pStmt));
×
1601
      STMT_ERR_RET(TSDB_CODE_NEED_RETRY);
×
1602
    }
1603
  }
1604

1605
  STMT_ERR_JRET(pStmt->exec.pRequest->code);
30!
1606

1607
  pStmt->exec.affectedRows = taos_affected_rows(pStmt->exec.pRequest);
30✔
1608
  pStmt->affectedRows += pStmt->exec.affectedRows;
30✔
1609

1610
_return:
30✔
1611

1612
  while (0 == atomic_load_8((int8_t*)&pStmt->sql.siInfo.tableColsReady)) {
30!
1613
    taosUsleep(1);
×
1614
  }
1615

1616
  STMT_ERR_RET(stmtCleanExecInfo(pStmt, (code ? false : true), false));
30!
1617

1618
  tFreeSSubmitRsp(pRsp);
30✔
1619

1620
  ++pStmt->sql.runTimes;
30✔
1621

1622
  int64_t startUs2 = taosGetTimestampUs();
30✔
1623
  pStmt->stat.execUseUs += startUs2 - startUs;
30✔
1624

1625
  STMT_RET(code);
30!
1626
}
1627

1628
int stmtClose(TAOS_STMT* stmt) {
9✔
1629
  STscStmt* pStmt = (STscStmt*)stmt;
9✔
1630

1631
  STMT_DLOG_E("start to free stmt");
9!
1632

1633
  pStmt->queue.stopQueue = true;
9✔
1634
  
1635
  (void)taosThreadMutexLock(&pStmt->queue.mutex);
9✔
1636
  (void)atomic_add_fetch_64(&pStmt->queue.qRemainNum, 1);
9✔
1637
  (void)taosThreadCondSignal(&(pStmt->queue.waitCond));
9✔
1638
  (void)taosThreadMutexUnlock(&pStmt->queue.mutex);
9✔
1639

1640
  if (pStmt->bindThreadInUse) {
9✔
1641
    (void)taosThreadJoin(pStmt->bindThread, NULL);
1✔
1642
    pStmt->bindThreadInUse = false;
1✔
1643
  }
1644

1645
  (void)taosThreadCondDestroy(&pStmt->queue.waitCond);
9✔
1646
  (void)taosThreadMutexDestroy(&pStmt->queue.mutex);
9✔
1647

1648
  STMT_DLOG("stmt %p closed, stbInterlaceMode:%d, statInfo: ctgGetTbMetaNum=>%" PRId64 ", getCacheTbInfo=>%" PRId64
9!
1649
            ", parseSqlNum=>%" PRId64 ", pStmt->stat.bindDataNum=>%" PRId64
1650
            ", settbnameAPI:%u, bindAPI:%u, addbatchAPI:%u, execAPI:%u"
1651
            ", setTbNameUs:%" PRId64 ", bindDataUs:%" PRId64 ",%" PRId64 ",%" PRId64 ",%" PRId64 " addBatchUs:%" PRId64
1652
            ", execWaitUs:%" PRId64 ", execUseUs:%" PRId64,
1653
            pStmt, pStmt->sql.stbInterlaceMode, pStmt->stat.ctgGetTbMetaNum, pStmt->stat.getCacheTbInfo,
1654
            pStmt->stat.parseSqlNum, pStmt->stat.bindDataNum, pStmt->seqIds[STMT_SETTBNAME], pStmt->seqIds[STMT_BIND],
1655
            pStmt->seqIds[STMT_ADD_BATCH], pStmt->seqIds[STMT_EXECUTE], pStmt->stat.setTbNameUs,
1656
            pStmt->stat.bindDataUs1, pStmt->stat.bindDataUs2, pStmt->stat.bindDataUs3, pStmt->stat.bindDataUs4,
1657
            pStmt->stat.addBatchUs, pStmt->stat.execWaitUs, pStmt->stat.execUseUs);
1658

1659
  STMT_ERR_RET(stmtCleanSQLInfo(pStmt));
9!
1660
  taosMemoryFree(stmt);
9!
1661

1662
  return TSDB_CODE_SUCCESS;
9✔
1663
}
1664

1665
const char* stmtErrstr(TAOS_STMT* stmt) {
×
1666
  STscStmt* pStmt = (STscStmt*)stmt;
×
1667

1668
  if (stmt == NULL || NULL == pStmt->exec.pRequest) {
×
1669
    return (char*)tstrerror(terrno);
×
1670
  }
1671

1672
  pStmt->exec.pRequest->code = terrno;
×
1673

1674
  return taos_errstr(pStmt->exec.pRequest);
×
1675
}
1676

1677
int stmtAffectedRows(TAOS_STMT* stmt) { return ((STscStmt*)stmt)->affectedRows; }
×
1678

1679
int stmtAffectedRowsOnce(TAOS_STMT* stmt) { return ((STscStmt*)stmt)->exec.affectedRows; }
28✔
1680

1681
int stmtIsInsert(TAOS_STMT* stmt, int* insert) {
32✔
1682
  STscStmt* pStmt = (STscStmt*)stmt;
32✔
1683

1684
  STMT_DLOG_E("start is insert");
32!
1685

1686
  if (pStmt->sql.type) {
32✔
1687
    *insert = (STMT_TYPE_INSERT == pStmt->sql.type || STMT_TYPE_MULTI_INSERT == pStmt->sql.type);
25!
1688
  } else {
1689
    *insert = qIsInsertValuesSql(pStmt->sql.sqlStr, pStmt->sql.sqlLen);
7✔
1690
  }
1691

1692
  return TSDB_CODE_SUCCESS;
32✔
1693
}
1694

1695
int stmtGetTagFields(TAOS_STMT* stmt, int* nums, TAOS_FIELD_E** fields) {
4✔
1696
  int32_t   code = 0;
4✔
1697
  STscStmt* pStmt = (STscStmt*)stmt;
4✔
1698
  int32_t   preCode = pStmt->errCode;
4✔
1699

1700
  STMT_DLOG_E("start to get tag fields");
4!
1701

1702
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
4!
1703
    return pStmt->errCode;
×
1704
  }
1705

1706
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
4!
1707
    STMT_ERRI_JRET(TSDB_CODE_TSC_STMT_API_ERROR);
×
1708
  }
1709

1710
  STMT_ERRI_JRET(stmtSwitchStatus(pStmt, STMT_FETCH_FIELDS));
4!
1711

1712
  if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 &&
4!
1713
      STMT_TYPE_MULTI_INSERT != pStmt->sql.type) {
×
1714
    pStmt->bInfo.needParse = false;
×
1715
  }
1716

1717
  if (pStmt->exec.pRequest && STMT_TYPE_QUERY == pStmt->sql.type && pStmt->sql.runTimes) {
4!
1718
    taos_free_result(pStmt->exec.pRequest);
×
1719
    pStmt->exec.pRequest = NULL;
×
1720
  }
1721

1722
  STMT_ERRI_JRET(stmtCreateRequest(pStmt));
4!
1723

1724
  if (pStmt->bInfo.needParse) {
4✔
1725
    STMT_ERRI_JRET(stmtParseSql(pStmt));
3✔
1726
  }
1727

1728
  STMT_ERRI_JRET(stmtFetchTagFields(stmt, nums, fields));
2✔
1729

1730
_return:
1✔
1731

1732
  pStmt->errCode = preCode;
4✔
1733

1734
  return code;
4✔
1735
}
1736

1737
int stmtGetColFields(TAOS_STMT* stmt, int* nums, TAOS_FIELD_E** fields) {
1✔
1738
  int32_t   code = 0;
1✔
1739
  STscStmt* pStmt = (STscStmt*)stmt;
1✔
1740
  int32_t   preCode = pStmt->errCode;
1✔
1741

1742
  STMT_DLOG_E("start to get col fields");
1!
1743

1744
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
1!
1745
    return pStmt->errCode;
×
1746
  }
1747

1748
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
1!
1749
    STMT_ERRI_JRET(TSDB_CODE_TSC_STMT_API_ERROR);
×
1750
  }
1751

1752
  STMT_ERRI_JRET(stmtSwitchStatus(pStmt, STMT_FETCH_FIELDS));
1!
1753

1754
  if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 &&
1!
1755
      STMT_TYPE_MULTI_INSERT != pStmt->sql.type) {
×
1756
    pStmt->bInfo.needParse = false;
×
1757
  }
1758

1759
  if (pStmt->exec.pRequest && STMT_TYPE_QUERY == pStmt->sql.type && pStmt->sql.runTimes) {
1!
1760
    taos_free_result(pStmt->exec.pRequest);
×
1761
    pStmt->exec.pRequest = NULL;
×
1762
    STMT_ERR_RET(stmtCreateRequest(pStmt));
×
1763
  }
1764

1765
  STMT_ERRI_JRET(stmtCreateRequest(pStmt));
1!
1766

1767
  if (pStmt->bInfo.needParse) {
1!
1768
    STMT_ERRI_JRET(stmtParseSql(pStmt));
×
1769
  }
1770

1771
  STMT_ERRI_JRET(stmtFetchColFields(stmt, nums, fields));
1!
1772

1773
_return:
1✔
1774

1775
  pStmt->errCode = preCode;
1✔
1776

1777
  return code;
1✔
1778
}
1779

1780
int stmtGetParamNum(TAOS_STMT* stmt, int* nums) {
2✔
1781
  int       code = 0;
2✔
1782
  STscStmt* pStmt = (STscStmt*)stmt;
2✔
1783
  int32_t   preCode = pStmt->errCode;
2✔
1784

1785
  STMT_DLOG_E("start to get param num");
2!
1786

1787
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
2!
1788
    return pStmt->errCode;
×
1789
  }
1790

1791
  STMT_ERRI_JRET(stmtSwitchStatus(pStmt, STMT_FETCH_FIELDS));
2!
1792

1793
  if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 &&
2!
1794
      STMT_TYPE_MULTI_INSERT != pStmt->sql.type) {
×
1795
    pStmt->bInfo.needParse = false;
×
1796
  }
1797

1798
  if (pStmt->exec.pRequest && STMT_TYPE_QUERY == pStmt->sql.type && pStmt->sql.runTimes) {
2!
1799
    taos_free_result(pStmt->exec.pRequest);
×
1800
    pStmt->exec.pRequest = NULL;
×
1801
  }
1802

1803
  STMT_ERRI_JRET(stmtCreateRequest(pStmt));
2!
1804

1805
  if (pStmt->bInfo.needParse) {
2✔
1806
    STMT_ERRI_JRET(stmtParseSql(pStmt));
1!
1807
  }
1808

1809
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
1!
1810
    *nums = taosArrayGetSize(pStmt->sql.pQuery->pPlaceholderValues);
×
1811
  } else {
1812
    STMT_ERRI_JRET(stmtFetchColFields(stmt, nums, NULL));
1!
1813
  }
1814

1815
_return:
1✔
1816

1817
  pStmt->errCode = preCode;
2✔
1818

1819
  return code;
2✔
1820
}
1821

1822
int stmtGetParam(TAOS_STMT* stmt, int idx, int* type, int* bytes) {
4✔
1823
  int       code = 0;
4✔
1824
  STscStmt* pStmt = (STscStmt*)stmt;
4✔
1825
  int32_t   preCode = pStmt->errCode;
4✔
1826

1827
  STMT_DLOG_E("start to get param");
4!
1828

1829
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
4!
1830
    return pStmt->errCode;
×
1831
  }
1832

1833
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
4!
1834
    STMT_ERRI_JRET(TSDB_CODE_TSC_STMT_API_ERROR);
×
1835
  }
1836

1837
  STMT_ERRI_JRET(stmtSwitchStatus(pStmt, STMT_FETCH_FIELDS));
4!
1838

1839
  if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 &&
4!
1840
      STMT_TYPE_MULTI_INSERT != pStmt->sql.type) {
×
1841
    pStmt->bInfo.needParse = false;
×
1842
  }
1843

1844
  if (pStmt->exec.pRequest && STMT_TYPE_QUERY == pStmt->sql.type && pStmt->sql.runTimes) {
4!
1845
    taos_free_result(pStmt->exec.pRequest);
×
1846
    pStmt->exec.pRequest = NULL;
×
1847
  }
1848

1849
  STMT_ERRI_JRET(stmtCreateRequest(pStmt));
4!
1850

1851
  if (pStmt->bInfo.needParse) {
4!
1852
    STMT_ERRI_JRET(stmtParseSql(pStmt));
×
1853
  }
1854

1855
  int32_t       nums = 0;
4✔
1856
  TAOS_FIELD_E* pField = NULL;
4✔
1857
  STMT_ERRI_JRET(stmtFetchColFields(stmt, &nums, &pField));
4!
1858
  if (idx >= nums) {
4!
1859
    tscError("idx %d is too big", idx);
×
1860
    STMT_ERRI_JRET(TSDB_CODE_INVALID_PARA);
×
1861
  }
1862

1863
  *type = pField[idx].type;
4✔
1864
  *bytes = pField[idx].bytes;
4✔
1865

1866
_return:
4✔
1867

1868
  taosMemoryFree(pField);
4!
1869
  pStmt->errCode = preCode;
4✔
1870

1871
  return code;
4✔
1872
}
1873

1874
TAOS_RES* stmtUseResult(TAOS_STMT* stmt) {
×
1875
  STscStmt* pStmt = (STscStmt*)stmt;
×
1876

1877
  STMT_DLOG_E("start to use result");
×
1878

1879
  if (STMT_TYPE_QUERY != pStmt->sql.type) {
×
1880
    tscError("useResult only for query statement");
×
1881
    return NULL;
×
1882
  }
1883

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