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

taosdata / TDengine / #3873

21 Apr 2025 07:22AM UTC coverage: 63.063% (+0.1%) from 62.968%
#3873

push

travis-ci

GitHub
docs(opc): add perssit data support (#30783)

156631 of 316378 branches covered (49.51%)

Branch coverage included in aggregate %.

242184 of 316027 relevant lines covered (76.63%)

20271838.47 hits per line

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

66.53
/source/client/src/clientStmt2.c
1
#include "clientInt.h"
2
#include "clientLog.h"
3
#include "tdef.h"
4

5
#include "clientStmt.h"
6
#include "clientStmt2.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) {
10,225✔
13
    *pBuf = (char*)pTblBuf->pCurBuff + pTblBuf->buffOffset;
10,227✔
14
    pTblBuf->buffOffset += pTblBuf->buffUnit;
10,227✔
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(TSDB_CODE_OUT_OF_MEMORY);
×
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;
10,227✔
39
}
40

41
static bool stmtDequeue(STscStmt2* pStmt, SStmtQNode** param) {
10,238✔
42
  int i = 0;
10,238✔
43
  while (0 == atomic_load_64((int64_t*)&pStmt->queue.qRemainNum)) {
56,343✔
44
    if (i < 10) {
46,118✔
45
      taosUsleep(1);
43,684✔
46
      i++;
43,671✔
47
    } else {
48
      (void)taosThreadMutexLock(&pStmt->queue.mutex);
2,434✔
49
      if (0 == atomic_load_64((int64_t*)&pStmt->queue.qRemainNum)) {
2,435✔
50
        (void)taosThreadCondWait(&pStmt->queue.waitCond, &pStmt->queue.mutex);
2,428✔
51
      }
52
      (void)taosThreadMutexUnlock(&pStmt->queue.mutex);
2,435✔
53
    }
54
  }
55
  if (pStmt->queue.stopQueue) {
10,095✔
56
    return false;
82✔
57
  }
58
  SStmtQNode* orig = pStmt->queue.head;
10,013✔
59
  SStmtQNode* node = pStmt->queue.head->next;
10,013✔
60
  pStmt->queue.head = pStmt->queue.head->next;
10,013✔
61
  *param = node;
10,013✔
62

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

65
  return true;
10,153✔
66
}
67

68
static void stmtEnqueue(STscStmt2* pStmt, SStmtQNode* param) {
10,149✔
69
  pStmt->queue.tail->next = param;
10,149✔
70
  pStmt->queue.tail = param;
10,149✔
71

72
  pStmt->stat.bindDataNum++;
10,149✔
73

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

80
static int32_t stmtCreateRequest(STscStmt2* pStmt) {
6,428✔
81
  int32_t code = 0;
6,428✔
82

83
  if (pStmt->exec.pRequest == NULL) {
6,428✔
84
    code = buildRequest(pStmt->taos->id, pStmt->sql.sqlStr, pStmt->sql.sqlLen, NULL, false, &pStmt->exec.pRequest,
179✔
85
                        pStmt->reqid);
86
    if (pStmt->reqid != 0) {
179!
87
      pStmt->reqid++;
×
88
    }
89
    pStmt->exec.pRequest->type = RES_TYPE__QUERY;
179✔
90
    if (pStmt->db != NULL) {
179✔
91
      taosMemoryFreeClear(pStmt->exec.pRequest->pDb); 
118!
92
      pStmt->exec.pRequest->pDb = taosStrdup(pStmt->db);
118!
93
    }
94
    if (TSDB_CODE_SUCCESS == code) {
179!
95
      pStmt->exec.pRequest->syncQuery = true;
179✔
96
      pStmt->exec.pRequest->isStmtBind = true;
179✔
97
    }
98
  }
99

100
  return code;
6,428✔
101
}
102

103
static int32_t stmtSwitchStatus(STscStmt2* pStmt, STMT_STATUS newStatus) {
21,010✔
104
  int32_t code = 0;
21,010✔
105

106
  if (newStatus >= STMT_INIT && newStatus < STMT_MAX) {
21,010!
107
    STMT_LOG_SEQ(newStatus);
21,018!
108
  }
109

110
  if (pStmt->errCode && newStatus != STMT_PREPARE) {
21,041!
111
    STMT_DLOG("stmt already failed with err:%s", tstrerror(pStmt->errCode));
×
112
    return pStmt->errCode;
×
113
  }
114

115
  switch (newStatus) {
21,041!
116
    case STMT_PREPARE:
153✔
117
      pStmt->errCode = 0;
153✔
118
      break;
153✔
119
    case STMT_SETTBNAME:
5,850✔
120
      if (STMT_STATUS_EQ(INIT)) {
5,850!
121
        code = TSDB_CODE_TSC_STMT_API_ERROR;
×
122
      }
123
      if (!pStmt->sql.stbInterlaceMode && (STMT_STATUS_EQ(BIND) || STMT_STATUS_EQ(BIND_COL))) {
5,850!
124
        code = TSDB_CODE_TSC_STMT_API_ERROR;
×
125
      }
126
      break;
5,850✔
127
    case STMT_SETTAGS:
132✔
128
      if (STMT_STATUS_EQ(INIT)) {
132!
129
        code = TSDB_CODE_TSC_STMT_API_ERROR;
×
130
      }
131
      break;
132✔
132
    case STMT_FETCH_FIELDS:
58✔
133
      if (STMT_STATUS_EQ(INIT)) {
58!
134
        code = TSDB_CODE_TSC_STMT_API_ERROR;
×
135
      }
136
      break;
58✔
137
    case STMT_BIND:
5,856✔
138
      if (STMT_STATUS_EQ(INIT) || STMT_STATUS_EQ(BIND_COL)) {
5,856!
139
        code = TSDB_CODE_TSC_STMT_API_ERROR;
×
140
      }
141
      /*
142
            if ((pStmt->sql.type == STMT_TYPE_MULTI_INSERT) && ()) {
143
              code = TSDB_CODE_TSC_STMT_API_ERROR;
144
            }
145
      */
146
      break;
5,856✔
147
    case STMT_BIND_COL:
×
148
      if (STMT_STATUS_EQ(INIT) || STMT_STATUS_EQ(BIND)) {
×
149
        code = TSDB_CODE_TSC_STMT_API_ERROR;
×
150
      }
151
      break;
×
152
    case STMT_ADD_BATCH:
4,526✔
153
      if (STMT_STATUS_NE(BIND) && STMT_STATUS_NE(BIND_COL) && STMT_STATUS_NE(FETCH_FIELDS)) {
4,526!
154
        code = TSDB_CODE_TSC_STMT_API_ERROR;
×
155
      }
156
      break;
4,526✔
157
    case STMT_EXECUTE:
4,466✔
158
      if (STMT_TYPE_QUERY == pStmt->sql.type) {
4,466✔
159
        if (STMT_STATUS_NE(ADD_BATCH) && STMT_STATUS_NE(FETCH_FIELDS) && STMT_STATUS_NE(BIND) &&
5!
160
            STMT_STATUS_NE(BIND_COL)) {
×
161
          code = TSDB_CODE_TSC_STMT_API_ERROR;
×
162
        }
163
      } else {
164
        if (STMT_STATUS_NE(ADD_BATCH) && STMT_STATUS_NE(FETCH_FIELDS)) {
4,461!
165
          code = TSDB_CODE_TSC_STMT_API_ERROR;
1✔
166
        }
167
      }
168
      break;
4,466✔
169
    default:
×
170
      code = TSDB_CODE_APP_ERROR;
×
171
      break;
×
172
  }
173

174
  STMT_ERR_RET(code);
21,041✔
175

176
  pStmt->sql.status = newStatus;
21,040✔
177

178
  return TSDB_CODE_SUCCESS;
21,040✔
179
}
180

181
static int32_t stmtGetTbName(TAOS_STMT2* stmt, char** tbName) {
118✔
182
  STscStmt2* pStmt = (STscStmt2*)stmt;
118✔
183

184
  pStmt->sql.type = STMT_TYPE_MULTI_INSERT;
118✔
185

186
  if ('\0' == pStmt->bInfo.tbName[0]) {
118✔
187
    tscWarn("no table name set, OK if it is a stmt get fields");
23!
188
    STMT_ERR_RET(TSDB_CODE_TSC_STMT_TBNAME_ERROR);
23!
189
  }
190

191
  *tbName = pStmt->bInfo.tbName;
95✔
192

193
  return TSDB_CODE_SUCCESS;
95✔
194
}
195

196
static int32_t stmtUpdateBindInfo(TAOS_STMT2* stmt, STableMeta* pTableMeta, void* tags, SName* tbName,
121✔
197
                                  const char* sTableName, bool autoCreateTbl, int8_t tbNameFlag) {
198
  STscStmt2* pStmt = (STscStmt2*)stmt;
121✔
199
  char       tbFName[TSDB_TABLE_FNAME_LEN];
200
  int32_t    code = tNameExtractFullName(tbName, tbFName);
121✔
201
  if (code != 0) {
122!
202
    return code;
×
203
  }
204

205
  (void)memcpy(&pStmt->bInfo.sname, tbName, sizeof(*tbName));
122✔
206
  tstrncpy(pStmt->bInfo.tbFName, tbFName, sizeof(pStmt->bInfo.tbFName));
122✔
207
  pStmt->bInfo.tbFName[sizeof(pStmt->bInfo.tbFName) - 1] = 0;
122✔
208

209
  pStmt->bInfo.tbUid = autoCreateTbl ? 0 : pTableMeta->uid;
122✔
210
  pStmt->bInfo.tbSuid = pTableMeta->suid;
122✔
211
  pStmt->bInfo.tbVgId = pTableMeta->vgId;
122✔
212
  pStmt->bInfo.tbType = pTableMeta->tableType;
122✔
213

214
  if (!pStmt->bInfo.tagsCached) {
122✔
215
    qDestroyBoundColInfo(pStmt->bInfo.boundTags);
116✔
216
    taosMemoryFreeClear(pStmt->bInfo.boundTags);
110!
217
  }
218

219
  pStmt->bInfo.boundTags = tags;
116✔
220
  pStmt->bInfo.tagsCached = false;
116✔
221
  pStmt->bInfo.tbNameFlag = tbNameFlag;
116✔
222
  tstrncpy(pStmt->bInfo.stbFName, sTableName, sizeof(pStmt->bInfo.stbFName));
116✔
223

224
  return TSDB_CODE_SUCCESS;
116✔
225
}
226

227
static int32_t stmtUpdateExecInfo(TAOS_STMT2* stmt, SHashObj* pVgHash, SHashObj* pBlockHash) {
117✔
228
  STscStmt2* pStmt = (STscStmt2*)stmt;
117✔
229

230
  pStmt->sql.pVgHash = pVgHash;
117✔
231
  pStmt->exec.pBlockHash = pBlockHash;
117✔
232

233
  return TSDB_CODE_SUCCESS;
117✔
234
}
235

236
static int32_t stmtUpdateInfo(TAOS_STMT2* stmt, STableMeta* pTableMeta, void* tags, SName* tbName, bool autoCreateTbl,
122✔
237
                              SHashObj* pVgHash, SHashObj* pBlockHash, const char* sTableName, uint8_t tbNameFlag) {
238
  STscStmt2* pStmt = (STscStmt2*)stmt;
122✔
239

240
  STMT_ERR_RET(stmtUpdateBindInfo(stmt, pTableMeta, tags, tbName, sTableName, autoCreateTbl, tbNameFlag));
122!
241
  STMT_ERR_RET(stmtUpdateExecInfo(stmt, pVgHash, pBlockHash));
117!
242

243
  pStmt->sql.autoCreateTbl = autoCreateTbl;
115✔
244

245
  return TSDB_CODE_SUCCESS;
115✔
246
}
247

248
static int32_t stmtGetExecInfo(TAOS_STMT2* stmt, SHashObj** pVgHash, SHashObj** pBlockHash) {
1✔
249
  STscStmt2* pStmt = (STscStmt2*)stmt;
1✔
250

251
  *pVgHash = pStmt->sql.pVgHash;
1✔
252
  pStmt->sql.pVgHash = NULL;
1✔
253

254
  *pBlockHash = pStmt->exec.pBlockHash;
1✔
255
  pStmt->exec.pBlockHash = NULL;
1✔
256

257
  return TSDB_CODE_SUCCESS;
1✔
258
}
259

260
static int32_t stmtParseSql(STscStmt2* pStmt) {
150✔
261
  pStmt->exec.pCurrBlock = NULL;
150✔
262

263
  SStmtCallback stmtCb = {
150✔
264
      .pStmt = pStmt,
265
      .getTbNameFn = stmtGetTbName,
266
      .setInfoFn = stmtUpdateInfo,
267
      .getExecInfoFn = stmtGetExecInfo,
268
  };
269

270
  STMT_ERR_RET(stmtCreateRequest(pStmt));
150!
271

272
  pStmt->stat.parseSqlNum++;
149✔
273
  STMT_ERR_RET(parseSql(pStmt->exec.pRequest, false, &pStmt->sql.pQuery, &stmtCb));
149✔
274
  pStmt->sql.siInfo.pQuery = pStmt->sql.pQuery;
130✔
275

276
  pStmt->bInfo.needParse = false;
130✔
277

278
  if (pStmt->sql.pQuery->pRoot && 0 == pStmt->sql.type) {
130✔
279
    pStmt->sql.type = STMT_TYPE_INSERT;
11✔
280
    pStmt->sql.stbInterlaceMode = false;
11✔
281
  } else if (pStmt->sql.pQuery->pPrepareRoot) {
119✔
282
    pStmt->sql.type = STMT_TYPE_QUERY;
6✔
283
    pStmt->sql.stbInterlaceMode = false;
6✔
284

285
    return TSDB_CODE_SUCCESS;
6✔
286
  }
287

288
  STableDataCxt** pSrc =
289
      (STableDataCxt**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
124✔
290
  if (NULL == pSrc || NULL == *pSrc) {
123!
291
    return terrno;
×
292
  }
293

294
  STableDataCxt* pTableCtx = *pSrc;
123✔
295
  // if (pStmt->sql.stbInterlaceMode) {
296
  //   int16_t lastIdx = -1;
297

298
  //   for (int32_t i = 0; i < pTableCtx->boundColsInfo.numOfBound; ++i) {
299
  //     if (pTableCtx->boundColsInfo.pColIndex[i] < lastIdx) {
300
  //       pStmt->sql.stbInterlaceMode = false;
301
  //       break;
302
  //     }
303

304
  //     lastIdx = pTableCtx->boundColsInfo.pColIndex[i];
305
  //   }
306
  // }
307

308
  if (NULL == pStmt->sql.pBindInfo) {
123✔
309
    pStmt->sql.pBindInfo = taosMemoryMalloc(pTableCtx->boundColsInfo.numOfBound * sizeof(*pStmt->sql.pBindInfo));
122!
310
    if (NULL == pStmt->sql.pBindInfo) {
123!
311
      return terrno;
×
312
    }
313
  }
314

315
  return TSDB_CODE_SUCCESS;
124✔
316
}
317

318
static int32_t stmtCleanBindInfo(STscStmt2* pStmt) {
4,718✔
319
  pStmt->bInfo.tbUid = 0;
4,718✔
320
  pStmt->bInfo.tbVgId = -1;
4,718✔
321
  pStmt->bInfo.tbType = 0;
4,718✔
322
  pStmt->bInfo.needParse = true;
4,718✔
323
  pStmt->bInfo.inExecCache = false;
4,718✔
324

325
  pStmt->bInfo.tbName[0] = 0;
4,718✔
326
  pStmt->bInfo.tbFName[0] = 0;
4,718✔
327
  if (!pStmt->bInfo.tagsCached) {
4,718✔
328
    qDestroyBoundColInfo(pStmt->bInfo.boundTags);
4,639✔
329
    taosMemoryFreeClear(pStmt->bInfo.boundTags);
4,647!
330
  }
331
  if (!pStmt->sql.autoCreateTbl) {
4,721✔
332
    pStmt->bInfo.stbFName[0] = 0;
4,646✔
333
    pStmt->bInfo.tbSuid = 0;
4,646✔
334
  }
335

336
  return TSDB_CODE_SUCCESS;
4,721✔
337
}
338

339
static void stmtFreeTableBlkList(STableColsData* pTb) {
×
340
  (void)qResetStmtColumns(pTb->aCol, true);
×
341
  taosArrayDestroy(pTb->aCol);
×
342
}
×
343

344
static void stmtResetQueueTableBuf(STableBufInfo* pTblBuf, SStmtQueue* pQueue) {
4,406✔
345
  pTblBuf->pCurBuff = taosArrayGetP(pTblBuf->pBufList, 0);
4,406✔
346
  if (NULL == pTblBuf->pCurBuff) {
4,410✔
347
    tscError("QInfo:%p, failed to get buffer from list", pTblBuf);
4!
348
    return;
×
349
  }
350
  pTblBuf->buffIdx = 1;
4,406✔
351
  pTblBuf->buffOffset = sizeof(*pQueue->head);
4,406✔
352

353
  pQueue->head = pQueue->tail = pTblBuf->pCurBuff;
4,406✔
354
  pQueue->qRemainNum = 0;
4,406✔
355
  pQueue->head->next = NULL;
4,406✔
356
}
357

358
static int32_t stmtCleanExecInfo(STscStmt2* pStmt, bool keepTable, bool deepClean) {
4,613✔
359
  if (pStmt->sql.stbInterlaceMode) {
4,613✔
360
    if (deepClean) {
4,488✔
361
      taosHashCleanup(pStmt->exec.pBlockHash);
79✔
362
      pStmt->exec.pBlockHash = NULL;
79✔
363

364
      if (NULL != pStmt->exec.pCurrBlock) {
79✔
365
        taosMemoryFreeClear(pStmt->exec.pCurrBlock->pData);
75!
366
        qDestroyStmtDataBlock(pStmt->exec.pCurrBlock);
75✔
367
        pStmt->exec.pCurrBlock = NULL;
75✔
368
      }
369
    } else {
370
      pStmt->sql.siInfo.pTableColsIdx = 0;
4,409✔
371
      stmtResetQueueTableBuf(&pStmt->sql.siInfo.tbBuf, &pStmt->queue);
4,409✔
372
    }
373
    if (NULL != pStmt->exec.pRequest) {
4,484✔
374
      pStmt->exec.pRequest->body.resInfo.numOfRows = 0;
4,482✔
375
    }
376
  } else {
377
    if (STMT_TYPE_QUERY != pStmt->sql.type || deepClean) {
127✔
378
      // if (!pStmt->options.asyncExecFn) {
379
      taos_free_result(pStmt->exec.pRequest);
122✔
380
      pStmt->exec.pRequest = NULL;
122✔
381
      //}
382
    }
383

384
    size_t keyLen = 0;
127✔
385
    void*  pIter = taosHashIterate(pStmt->exec.pBlockHash, NULL);
127✔
386
    while (pIter) {
269✔
387
      STableDataCxt* pBlocks = *(STableDataCxt**)pIter;
142✔
388
      char*          key = taosHashGetKey(pIter, &keyLen);
142✔
389
      STableMeta*    pMeta = qGetTableMetaInDataBlock(pBlocks);
142✔
390

391
      if (keepTable && pBlocks == pStmt->exec.pCurrBlock) {
142✔
392
        TSWAP(pBlocks->pData, pStmt->exec.pCurrTbData);
47✔
393
        STMT_ERR_RET(qResetStmtDataBlock(pBlocks, false));
99!
394

395
        pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
47✔
396
        continue;
47✔
397
      }
398

399
      qDestroyStmtDataBlock(pBlocks);
95✔
400
      STMT_ERR_RET(taosHashRemove(pStmt->exec.pBlockHash, key, keyLen));
95!
401

402
      pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
95✔
403
    }
404

405
    if (keepTable) {
127✔
406
      return TSDB_CODE_SUCCESS;
52✔
407
    }
408

409
    taosHashCleanup(pStmt->exec.pBlockHash);
75✔
410
    pStmt->exec.pBlockHash = NULL;
75✔
411

412
    tDestroySubmitTbData(pStmt->exec.pCurrTbData, TSDB_MSG_FLG_ENCODE);
75✔
413
    taosMemoryFreeClear(pStmt->exec.pCurrTbData);
75!
414
  }
415

416
  STMT_ERR_RET(stmtCleanBindInfo(pStmt));
4,559!
417

418
  return TSDB_CODE_SUCCESS;
4,557✔
419
}
420

421
static void stmtFreeTbBuf(void* buf) {
86✔
422
  void* pBuf = *(void**)buf;
86✔
423
  taosMemoryFree(pBuf);
86!
424
}
86✔
425

426
static void stmtFreeTbCols(void* buf) {
76,000✔
427
  SArray* pCols = *(SArray**)buf;
76,000✔
428
  taosArrayDestroy(pCols);
76,000✔
429
}
76,000✔
430

431
static int32_t stmtCleanSQLInfo(STscStmt2* pStmt) {
145✔
432
  STMT_DLOG_E("start to free SQL info");
145!
433

434
  taosMemoryFreeClear(pStmt->db);
145!
435
  taosMemoryFree(pStmt->sql.pBindInfo);
145!
436
  taosMemoryFree(pStmt->sql.queryRes.fields);
145!
437
  taosMemoryFree(pStmt->sql.queryRes.userFields);
145!
438
  taosMemoryFree(pStmt->sql.sqlStr);
145!
439
  qDestroyQuery(pStmt->sql.pQuery);
145✔
440
  taosArrayDestroy(pStmt->sql.nodeList);
145✔
441
  taosHashCleanup(pStmt->sql.pVgHash);
145✔
442
  pStmt->sql.pVgHash = NULL;
145✔
443
  if (pStmt->sql.fixValueTags) {
145✔
444
    tdDestroySVCreateTbReq(pStmt->sql.fixValueTbReq);
3!
445
  }
446

447
  void* pIter = taosHashIterate(pStmt->sql.pTableCache, NULL);
145✔
448
  while (pIter) {
160✔
449
    SStmtTableCache* pCache = (SStmtTableCache*)pIter;
15✔
450

451
    qDestroyStmtDataBlock(pCache->pDataCtx);
15✔
452
    qDestroyBoundColInfo(pCache->boundTags);
15✔
453
    taosMemoryFreeClear(pCache->boundTags);
15!
454

455
    pIter = taosHashIterate(pStmt->sql.pTableCache, pIter);
15✔
456
  }
457
  taosHashCleanup(pStmt->sql.pTableCache);
145✔
458
  pStmt->sql.pTableCache = NULL;
145✔
459

460
  STMT_ERR_RET(stmtCleanExecInfo(pStmt, false, true));
145!
461
  STMT_ERR_RET(stmtCleanBindInfo(pStmt));
145!
462

463
  taos_free_result(pStmt->sql.siInfo.pRequest);
145✔
464
  taosHashCleanup(pStmt->sql.siInfo.pVgroupHash);
145✔
465
  tSimpleHashCleanup(pStmt->sql.siInfo.pTableHash);
145✔
466
  taosArrayDestroyEx(pStmt->sql.siInfo.tbBuf.pBufList, stmtFreeTbBuf);
145✔
467
  taosMemoryFree(pStmt->sql.siInfo.pTSchema);
145!
468
  qDestroyStmtDataBlock(pStmt->sql.siInfo.pDataCtx);
145✔
469
  taosArrayDestroyEx(pStmt->sql.siInfo.pTableCols, stmtFreeTbCols);
145✔
470

471
  (void)memset(&pStmt->sql, 0, sizeof(pStmt->sql));
145✔
472
  pStmt->sql.siInfo.tableColsReady = true;
145✔
473

474
  STMT_DLOG_E("end to free SQL info");
145!
475

476
  return TSDB_CODE_SUCCESS;
145✔
477
}
478

479
static int32_t stmtTryAddTableVgroupInfo(STscStmt2* pStmt, int32_t* vgId) {
113✔
480
  if (*vgId >= 0 && taosHashGet(pStmt->sql.pVgHash, (const char*)vgId, sizeof(*vgId))) {
113✔
481
    return TSDB_CODE_SUCCESS;
15✔
482
  }
483

484
  SVgroupInfo      vgInfo = {0};
98✔
485
  SRequestConnInfo conn = {.pTrans = pStmt->taos->pAppInfo->pTransporter,
98✔
486
                           .requestId = pStmt->exec.pRequest->requestId,
98✔
487
                           .requestObjRefId = pStmt->exec.pRequest->self,
98✔
488
                           .mgmtEps = getEpSet_s(&pStmt->taos->pAppInfo->mgmtEp)};
98✔
489

490
  int32_t code = catalogGetTableHashVgroup(pStmt->pCatalog, &conn, &pStmt->bInfo.sname, &vgInfo);
98✔
491
  if (TSDB_CODE_SUCCESS != code) {
98!
492
    return code;
×
493
  }
494

495
  code =
496
      taosHashPut(pStmt->sql.pVgHash, (const char*)&vgInfo.vgId, sizeof(vgInfo.vgId), (char*)&vgInfo, sizeof(vgInfo));
98✔
497
  if (TSDB_CODE_SUCCESS != code) {
98!
498
    return code;
×
499
  }
500

501
  *vgId = vgInfo.vgId;
98✔
502

503
  return TSDB_CODE_SUCCESS;
98✔
504
}
505

506
static int32_t stmtRebuildDataBlock(STscStmt2* pStmt, STableDataCxt* pDataBlock, STableDataCxt** newBlock, uint64_t uid,
66✔
507
                                    uint64_t suid, int32_t vgId) {
508
  STMT_ERR_RET(stmtTryAddTableVgroupInfo(pStmt, &vgId));
66!
509
  STMT_ERR_RET(qRebuildStmtDataBlock(newBlock, pDataBlock, uid, suid, vgId, pStmt->sql.autoCreateTbl));
66!
510

511
  STMT_DLOG("uid:%" PRId64 ", rebuild table data context, vgId:%d", uid, vgId);
66!
512

513
  return TSDB_CODE_SUCCESS;
66✔
514
}
515

516
static int32_t stmtGetFromCache(STscStmt2* pStmt) {
185✔
517
  if (pStmt->sql.stbInterlaceMode && pStmt->sql.siInfo.pDataCtx) {
185!
518
    pStmt->bInfo.needParse = false;
×
519
    pStmt->bInfo.inExecCache = false;
×
520
    return TSDB_CODE_SUCCESS;
×
521
  }
522

523
  pStmt->bInfo.needParse = true;
185✔
524
  pStmt->bInfo.inExecCache = false;
185✔
525

526
  STableDataCxt** pCxtInExec = taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
185✔
527
  if (pCxtInExec) {
182✔
528
    pStmt->bInfo.needParse = false;
24✔
529
    pStmt->bInfo.inExecCache = true;
24✔
530

531
    pStmt->exec.pCurrBlock = *pCxtInExec;
24✔
532

533
    if (pStmt->sql.autoCreateTbl) {
24✔
534
      tscDebug("reuse stmt block for tb %s in execBlock", pStmt->bInfo.tbFName);
18!
535
      return TSDB_CODE_SUCCESS;
18✔
536
    }
537
  }
538

539
  if (NULL == pStmt->pCatalog) {
164✔
540
    STMT_ERR_RET(catalogGetHandle(pStmt->taos->pAppInfo->clusterId, &pStmt->pCatalog));
88!
541
    pStmt->sql.siInfo.pCatalog = pStmt->pCatalog;
90✔
542
  }
543

544
  if (NULL == pStmt->sql.pTableCache || taosHashGetSize(pStmt->sql.pTableCache) <= 0) {
166!
545
    if (pStmt->bInfo.inExecCache) {
94!
546
      pStmt->bInfo.needParse = false;
×
547
      tscDebug("reuse stmt block for tb %s in execBlock", pStmt->bInfo.tbFName);
×
548
      return TSDB_CODE_SUCCESS;
×
549
    }
550

551
    tscDebug("no stmt block cache for tb %s", pStmt->bInfo.tbFName);
94!
552
    return TSDB_CODE_SUCCESS;
94✔
553
  }
554

555
  if (pStmt->sql.autoCreateTbl) {
72✔
556
    SStmtTableCache* pCache = taosHashGet(pStmt->sql.pTableCache, &pStmt->bInfo.tbSuid, sizeof(pStmt->bInfo.tbSuid));
48✔
557
    if (pCache) {
48!
558
      pStmt->bInfo.needParse = false;
48✔
559
      pStmt->bInfo.tbUid = 0;
48✔
560

561
      STableDataCxt* pNewBlock = NULL;
48✔
562
      STMT_ERR_RET(stmtRebuildDataBlock(pStmt, pCache->pDataCtx, &pNewBlock, 0, pStmt->bInfo.tbSuid, -1));
48!
563

564
      if (taosHashPut(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName), &pNewBlock,
48!
565
                      POINTER_BYTES)) {
566
        STMT_ERR_RET(terrno);
×
567
      }
568

569
      pStmt->exec.pCurrBlock = pNewBlock;
48✔
570

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

573
      return TSDB_CODE_SUCCESS;
48✔
574
    }
575

576
    STMT_RET(stmtCleanBindInfo(pStmt));
×
577
  }
578

579
  uint64_t uid, suid;
580
  int32_t  vgId;
581
  int8_t   tableType;
582

583
  STableMeta*      pTableMeta = NULL;
24✔
584
  SRequestConnInfo conn = {.pTrans = pStmt->taos->pAppInfo->pTransporter,
24✔
585
                           .requestId = pStmt->exec.pRequest->requestId,
24✔
586
                           .requestObjRefId = pStmt->exec.pRequest->self,
24✔
587
                           .mgmtEps = getEpSet_s(&pStmt->taos->pAppInfo->mgmtEp)};
24✔
588
  int32_t          code = catalogGetTableMeta(pStmt->pCatalog, &conn, &pStmt->bInfo.sname, &pTableMeta);
24✔
589

590
  pStmt->stat.ctgGetTbMetaNum++;
24✔
591

592
  if (TSDB_CODE_PAR_TABLE_NOT_EXIST == code) {
24!
593
    tscDebug("tb %s not exist", pStmt->bInfo.tbFName);
×
594
    STMT_ERR_RET(stmtCleanBindInfo(pStmt));
×
595

596
    STMT_ERR_RET(code);
×
597
  }
598

599
  STMT_ERR_RET(code);
24!
600

601
  uid = pTableMeta->uid;
24✔
602
  suid = pTableMeta->suid;
24✔
603
  tableType = pTableMeta->tableType;
24✔
604
  pStmt->bInfo.tbVgId = pTableMeta->vgId;
24✔
605
  vgId = pTableMeta->vgId;
24✔
606

607
  taosMemoryFree(pTableMeta);
24!
608

609
  uint64_t cacheUid = (TSDB_CHILD_TABLE == tableType) ? suid : uid;
24!
610

611
  if (uid == pStmt->bInfo.tbUid) {
24!
612
    pStmt->bInfo.needParse = false;
×
613

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

616
    return TSDB_CODE_SUCCESS;
×
617
  }
618

619
  if (pStmt->bInfo.inExecCache) {
24✔
620
    SStmtTableCache* pCache = taosHashGet(pStmt->sql.pTableCache, &cacheUid, sizeof(cacheUid));
6✔
621
    if (NULL == pCache) {
6!
622
      tscError("table [%s, %" PRIx64 ", %" PRIx64 "] found in exec blockHash, but not in sql blockHash",
×
623
               pStmt->bInfo.tbFName, uid, cacheUid);
624

625
      STMT_ERR_RET(TSDB_CODE_APP_ERROR);
×
626
    }
627

628
    pStmt->bInfo.needParse = false;
6✔
629

630
    pStmt->bInfo.tbUid = uid;
6✔
631
    pStmt->bInfo.tbSuid = suid;
6✔
632
    pStmt->bInfo.tbType = tableType;
6✔
633
    pStmt->bInfo.boundTags = pCache->boundTags;
6✔
634
    pStmt->bInfo.tagsCached = true;
6✔
635

636
    tscDebug("tb %s in execBlock list, set to current", pStmt->bInfo.tbFName);
6!
637

638
    return TSDB_CODE_SUCCESS;
6✔
639
  }
640

641
  SStmtTableCache* pCache = taosHashGet(pStmt->sql.pTableCache, &cacheUid, sizeof(cacheUid));
18✔
642
  if (pCache) {
18!
643
    pStmt->bInfo.needParse = false;
18✔
644

645
    pStmt->bInfo.tbUid = uid;
18✔
646
    pStmt->bInfo.tbSuid = suid;
18✔
647
    pStmt->bInfo.tbType = tableType;
18✔
648
    pStmt->bInfo.boundTags = pCache->boundTags;
18✔
649
    pStmt->bInfo.tagsCached = true;
18✔
650

651
    STableDataCxt* pNewBlock = NULL;
18✔
652
    STMT_ERR_RET(stmtRebuildDataBlock(pStmt, pCache->pDataCtx, &pNewBlock, uid, suid, vgId));
18!
653

654
    if (taosHashPut(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName), &pNewBlock,
18!
655
                    POINTER_BYTES)) {
656
      STMT_ERR_RET(terrno);
×
657
    }
658

659
    pStmt->exec.pCurrBlock = pNewBlock;
18✔
660

661
    tscDebug("tb %s in sqlBlock list, set to current", pStmt->bInfo.tbFName);
18!
662

663
    return TSDB_CODE_SUCCESS;
18✔
664
  }
665

666
  STMT_ERR_RET(stmtCleanBindInfo(pStmt));
×
667

668
  return TSDB_CODE_SUCCESS;
×
669
}
670

671
static int32_t stmtResetStmt(STscStmt2* pStmt) {
×
672
  STMT_ERR_RET(stmtCleanSQLInfo(pStmt));
×
673

674
  pStmt->sql.pTableCache = taosHashInit(100, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
×
675
  if (NULL == pStmt->sql.pTableCache) {
×
676
    STMT_ERR_RET(terrno);
×
677
  }
678

679
  pStmt->sql.status = STMT_INIT;
×
680

681
  return TSDB_CODE_SUCCESS;
×
682
}
683

684
static int32_t stmtAsyncOutput(STscStmt2* pStmt, void* param) {
10,155✔
685
  SStmtQNode* pParam = (SStmtQNode*)param;
10,155✔
686

687
  if (pParam->restoreTbCols) {
10,155✔
688
    for (int32_t i = 0; i < pStmt->sql.siInfo.pTableColsIdx; ++i) {
10,148✔
689
      SArray** p = (SArray**)TARRAY_GET_ELEM(pStmt->sql.siInfo.pTableCols, i);
5,750✔
690
      *p = taosArrayInit(20, POINTER_BYTES);
5,750✔
691
      if (*p == NULL) {
5,746!
692
        STMT_ERR_RET(terrno);
×
693
      }
694
    }
695

696
    atomic_store_8((int8_t*)&pStmt->sql.siInfo.tableColsReady, true);
4,398✔
697
  } else {
698
    int code = qAppendStmtTableOutput(pStmt->sql.pQuery, pStmt->sql.pVgHash, &pParam->tblData, pStmt->exec.pCurrBlock,
5,744✔
699
                                      &pStmt->sql.siInfo, pParam->pCreateTbReq);
700
    // taosMemoryFree(pParam->pTbData);
701
    (void)atomic_sub_fetch_64(&pStmt->sql.siInfo.tbRemainNum, 1);
5,742✔
702
    STMT_ERR_RET(code);
5,749!
703
  }
704
  return TSDB_CODE_SUCCESS;
10,160✔
705
}
706

707
static void* stmtBindThreadFunc(void* param) {
86✔
708
  setThreadName("stmtBind");
86✔
709

710
  qInfo("stmt bind thread started");
86!
711

712
  STscStmt2* pStmt = (STscStmt2*)param;
86✔
713

714
  while (true) {
10,239✔
715
    if (pStmt->queue.stopQueue) {
10,325✔
716
      break;
86✔
717
    }
718

719
    SStmtQNode* asyncParam = NULL;
10,239✔
720
    if (!stmtDequeue(pStmt, &asyncParam)) {
10,239✔
721
      continue;
82✔
722
    }
723

724
    if (stmtAsyncOutput(pStmt, asyncParam) != 0) {
10,155!
725
      qError("stmt async output failed");
×
726
    }
727
  }
728

729
  qInfo("stmt bind thread stopped");
86!
730

731
  return NULL;
86✔
732
}
733

734
static int32_t stmtStartBindThread(STscStmt2* pStmt) {
86✔
735
  TdThreadAttr thAttr;
736
  if (taosThreadAttrInit(&thAttr) != 0) {
86!
737
    return TSDB_CODE_TSC_INTERNAL_ERROR;
×
738
  }
739
  if (taosThreadAttrSetDetachState(&thAttr, PTHREAD_CREATE_JOINABLE) != 0) {
86!
740
    return TSDB_CODE_TSC_INTERNAL_ERROR;
×
741
  }
742

743
  if (taosThreadCreate(&pStmt->bindThread, &thAttr, stmtBindThreadFunc, pStmt) != 0) {
86!
744
    terrno = TAOS_SYSTEM_ERROR(ERRNO);
×
745
    STMT_ERR_RET(terrno);
×
746
  }
747

748
  pStmt->bindThreadInUse = true;
86✔
749

750
  (void)taosThreadAttrDestroy(&thAttr);
86✔
751
  return TSDB_CODE_SUCCESS;
86✔
752
}
753

754
static int32_t stmtInitQueue(STscStmt2* pStmt) {
86✔
755
  (void)taosThreadCondInit(&pStmt->queue.waitCond, NULL);
86✔
756
  (void)taosThreadMutexInit(&pStmt->queue.mutex, NULL);
86✔
757
  STMT_ERR_RET(stmtAllocQNodeFromBuf(&pStmt->sql.siInfo.tbBuf, (void**)&pStmt->queue.head));
172!
758
  pStmt->queue.tail = pStmt->queue.head;
86✔
759

760
  return TSDB_CODE_SUCCESS;
86✔
761
}
762

763
static int32_t stmtIniAsyncBind(STscStmt2* pStmt) {
147✔
764
  (void)taosThreadCondInit(&pStmt->asyncBindParam.waitCond, NULL);
147✔
765
  (void)taosThreadMutexInit(&pStmt->asyncBindParam.mutex, NULL);
147✔
766
  pStmt->asyncBindParam.asyncBindNum = 0;
147✔
767

768
  return TSDB_CODE_SUCCESS;
147✔
769
}
770

771
static int32_t stmtInitTableBuf(STableBufInfo* pTblBuf) {
86✔
772
  pTblBuf->buffUnit = sizeof(SStmtQNode);
86✔
773
  pTblBuf->buffSize = pTblBuf->buffUnit * 1000;
86✔
774
  pTblBuf->pBufList = taosArrayInit(100, POINTER_BYTES);
86✔
775
  if (NULL == pTblBuf->pBufList) {
86!
776
    return terrno;
×
777
  }
778
  void* buff = taosMemoryMalloc(pTblBuf->buffSize);
86!
779
  if (NULL == buff) {
86!
780
    return terrno;
×
781
  }
782

783
  if (taosArrayPush(pTblBuf->pBufList, &buff) == NULL) {
172!
784
    return terrno;
×
785
  }
786

787
  pTblBuf->pCurBuff = buff;
86✔
788
  pTblBuf->buffIdx = 1;
86✔
789
  pTblBuf->buffOffset = 0;
86✔
790

791
  return TSDB_CODE_SUCCESS;
86✔
792
}
793

794
TAOS_STMT2* stmtInit2(STscObj* taos, TAOS_STMT2_OPTION* pOptions) {
147✔
795
  STscObj*   pObj = (STscObj*)taos;
147✔
796
  STscStmt2* pStmt = NULL;
147✔
797
  int32_t    code = 0;
147✔
798

799
  pStmt = taosMemoryCalloc(1, sizeof(STscStmt2));
147!
800
  if (NULL == pStmt) {
147!
801
    return NULL;
×
802
  }
803

804
  pStmt->sql.pTableCache = taosHashInit(100, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
147✔
805
  if (NULL == pStmt->sql.pTableCache) {
147!
806
    taosMemoryFree(pStmt);
×
807
    return NULL;
×
808
  }
809

810
  pStmt->taos = pObj;
147✔
811
  if (taos->db != NULL && taos->db[0] != '\0') {
147!
812
    pStmt->db = taosStrdup(taos->db);
51!
813
  }
814
  pStmt->bInfo.needParse = true;
147✔
815
  pStmt->sql.status = STMT_INIT;
147✔
816
  pStmt->errCode = TSDB_CODE_SUCCESS;
147✔
817

818
  if (NULL != pOptions) {
147!
819
    (void)memcpy(&pStmt->options, pOptions, sizeof(pStmt->options));
147✔
820
    if (pOptions->singleStbInsert && pOptions->singleTableBindOnce) {
147✔
821
      pStmt->stbInterlaceMode = true;
81✔
822
    }
823

824
    pStmt->reqid = pOptions->reqid;
147✔
825
  }
826

827
  if (pStmt->stbInterlaceMode) {
147✔
828
    pStmt->sql.siInfo.transport = taos->pAppInfo->pTransporter;
81✔
829
    pStmt->sql.siInfo.acctId = taos->acctId;
81✔
830
    pStmt->sql.siInfo.dbname = taos->db;
81✔
831
    pStmt->sql.siInfo.mgmtEpSet = getEpSet_s(&pStmt->taos->pAppInfo->mgmtEp);
81✔
832
    pStmt->sql.siInfo.pTableHash = tSimpleHashInit(100, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY));
81✔
833
    if (NULL == pStmt->sql.siInfo.pTableHash) {
81!
834
      (void)stmtClose2(pStmt);
×
835
      return NULL;
×
836
    }
837
    pStmt->sql.siInfo.pTableCols = taosArrayInit(STMT_TABLE_COLS_NUM, POINTER_BYTES);
81✔
838
    if (NULL == pStmt->sql.siInfo.pTableCols) {
81!
839
      terrno = terrno;
×
840
      (void)stmtClose2(pStmt);
×
841
      return NULL;
×
842
    }
843

844
    code = stmtInitTableBuf(&pStmt->sql.siInfo.tbBuf);
81✔
845
    if (TSDB_CODE_SUCCESS == code) {
81!
846
      code = stmtInitQueue(pStmt);
81✔
847
    }
848
    if (TSDB_CODE_SUCCESS == code) {
81!
849
      code = stmtStartBindThread(pStmt);
81✔
850
    }
851
    if (TSDB_CODE_SUCCESS != code) {
81!
852
      terrno = code;
×
853
      (void)stmtClose2(pStmt);
×
854
      return NULL;
×
855
    }
856
  }
857

858
  pStmt->sql.siInfo.tableColsReady = true;
147✔
859
  if (pStmt->options.asyncExecFn) {
147✔
860
    if (tsem_init(&pStmt->asyncExecSem, 0, 1) != 0) {
5!
861
      terrno = TAOS_SYSTEM_ERROR(ERRNO);
×
862
      (void)stmtClose2(pStmt);
×
863
      return NULL;
×
864
    }
865
  }
866
  code = stmtIniAsyncBind(pStmt);
147✔
867
  if (TSDB_CODE_SUCCESS != code) {
147!
868
    terrno = code;
×
869
    (void)stmtClose2(pStmt);
×
870
    return NULL;
×
871
  }
872

873
  pStmt->execSemWaited = false;
147✔
874

875
  STMT_LOG_SEQ(STMT_INIT);
147!
876

877
  tscDebug("stmt:%p initialized", pStmt);
147!
878

879
  return pStmt;
147✔
880
}
881

882
static int stmtSetDbName2(TAOS_STMT2* stmt, const char* dbName) {
54✔
883
  STscStmt2* pStmt = (STscStmt2*)stmt;
54✔
884

885
  STMT2_DLOG("dbname is specified in sql:%s", dbName);
54!
886

887
  pStmt->db = taosStrdup(dbName);
54!
888
  (void)strdequote(pStmt->db);
54✔
889
  STMT_ERR_RET(stmtCreateRequest(pStmt));
54!
890

891
  // The SQL statement specifies a database name, overriding the previously specified database
892
  taosMemoryFreeClear(pStmt->exec.pRequest->pDb);
54!
893
  pStmt->exec.pRequest->pDb = taosStrdup(pStmt->db);
54!
894
  if (pStmt->exec.pRequest->pDb == NULL) {
54!
895
    return terrno;
×
896
  }
897
  if (pStmt->sql.stbInterlaceMode) {
54✔
898
    pStmt->sql.siInfo.dbname = pStmt->db;
12✔
899
  }
900
  return TSDB_CODE_SUCCESS;
54✔
901
}
902
static int32_t stmtResetStbInterlaceCache(STscStmt2* pStmt) {
5✔
903
  int32_t code = TSDB_CODE_SUCCESS;
5✔
904

905
  if (pStmt->bindThreadInUse) {
5!
906
    pStmt->queue.stopQueue = true;
5✔
907

908
    (void)taosThreadMutexLock(&pStmt->queue.mutex);
5✔
909
    (void)atomic_add_fetch_64(&pStmt->queue.qRemainNum, 1);
5✔
910
    (void)taosThreadCondSignal(&(pStmt->queue.waitCond));
5✔
911
    (void)taosThreadMutexUnlock(&pStmt->queue.mutex);
5✔
912

913
    (void)taosThreadJoin(pStmt->bindThread, NULL);
5✔
914
    pStmt->bindThreadInUse = false;
5✔
915
    pStmt->queue.head = NULL;
5✔
916
    pStmt->queue.tail = NULL;
5✔
917
    pStmt->queue.qRemainNum = 0;
5✔
918

919
    (void)taosThreadCondDestroy(&pStmt->queue.waitCond);
5✔
920
    (void)taosThreadMutexDestroy(&pStmt->queue.mutex);
5✔
921
  }
922

923
  pStmt->sql.siInfo.pTableHash = tSimpleHashInit(100, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY));
5✔
924
  if (NULL == pStmt->sql.siInfo.pTableHash) {
5!
925
    return terrno;
×
926
  }
927

928
  pStmt->sql.siInfo.pTableCols = taosArrayInit(STMT_TABLE_COLS_NUM, POINTER_BYTES);
5✔
929
  if (NULL == pStmt->sql.siInfo.pTableCols) {
5!
930
    return terrno;
×
931
  }
932

933
  code = stmtInitTableBuf(&pStmt->sql.siInfo.tbBuf);
5✔
934

935
  if (TSDB_CODE_SUCCESS == code) {
5!
936
    code = stmtInitQueue(pStmt);
5✔
937
    pStmt->queue.stopQueue = false;
5✔
938
  }
939
  if (TSDB_CODE_SUCCESS == code) {
5!
940
    code = stmtStartBindThread(pStmt);
5✔
941
  }
942
  if (TSDB_CODE_SUCCESS != code) {
5!
943
    return code;
×
944
  }
945

946
  return TSDB_CODE_SUCCESS;
5✔
947
}
948

949
static int32_t stmtResetStmtForPrepare(STscStmt2* pStmt) {
9✔
950
  char*             db = pStmt->db;
9✔
951
  bool              stbInterlaceMode = pStmt->stbInterlaceMode;
9✔
952
  TAOS_STMT2_OPTION options = pStmt->options;
9✔
953
  uint32_t          reqid = pStmt->reqid;
9✔
954

955
  taosMemoryFree(pStmt->sql.pBindInfo);
9!
956
  pStmt->sql.pBindInfo = NULL;
9✔
957

958
  taosMemoryFree(pStmt->sql.queryRes.fields);
9!
959
  pStmt->sql.queryRes.fields = NULL;
9✔
960

961
  taosMemoryFree(pStmt->sql.queryRes.userFields);
9!
962
  pStmt->sql.queryRes.userFields = NULL;
9✔
963

964
  pStmt->sql.type = 0;
9✔
965
  pStmt->sql.runTimes = 0;
9✔
966
  taosMemoryFree(pStmt->sql.sqlStr);
9!
967
  pStmt->sql.sqlStr = NULL;
9✔
968

969
  qDestroyQuery(pStmt->sql.pQuery);
9✔
970
  pStmt->sql.pQuery = NULL;
9✔
971

972
  taosArrayDestroy(pStmt->sql.nodeList);
9✔
973
  pStmt->sql.nodeList = NULL;
9✔
974

975
  taosHashCleanup(pStmt->sql.pVgHash);
9✔
976
  pStmt->sql.pVgHash = NULL;
9✔
977

978
  if (pStmt->sql.fixValueTags) {
9!
979
    tdDestroySVCreateTbReq(pStmt->sql.fixValueTbReq);
×
980
  }
981

982
  void* pIter = taosHashIterate(pStmt->sql.pTableCache, NULL);
9✔
983
  while (pIter) {
12✔
984
    SStmtTableCache* pCache = (SStmtTableCache*)pIter;
3✔
985

986
    qDestroyStmtDataBlock(pCache->pDataCtx);
3✔
987
    qDestroyBoundColInfo(pCache->boundTags);
3✔
988
    taosMemoryFreeClear(pCache->boundTags);
3!
989

990
    pIter = taosHashIterate(pStmt->sql.pTableCache, pIter);
3✔
991
  }
992
  taosHashCleanup(pStmt->sql.pTableCache);
9✔
993
  pStmt->sql.pTableCache = taosHashInit(100, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
9✔
994
  if (NULL == pStmt->sql.pTableCache) {
9!
995
    return terrno;
×
996
  }
997

998
  STMT_ERR_RET(stmtCleanExecInfo(pStmt, false, true));
9!
999

1000
  if (pStmt->exec.pRequest) {
9✔
1001
    taos_free_result(pStmt->exec.pRequest);
3✔
1002
    pStmt->exec.pRequest = NULL;
3✔
1003
  }
1004

1005
  if (pStmt->sql.siInfo.pTableCols) {
9✔
1006
    taosArrayDestroyEx(pStmt->sql.siInfo.pTableCols, stmtFreeTbCols);
5✔
1007
    pStmt->sql.siInfo.pTableCols = NULL;
5✔
1008
  }
1009

1010
  if (pStmt->sql.siInfo.tbBuf.pBufList) {
9✔
1011
    taosArrayDestroyEx(pStmt->sql.siInfo.tbBuf.pBufList, stmtFreeTbBuf);
5✔
1012
    pStmt->sql.siInfo.tbBuf.pBufList = NULL;
5✔
1013
  }
1014

1015
  if (pStmt->sql.siInfo.pTableHash) {
9✔
1016
    tSimpleHashCleanup(pStmt->sql.siInfo.pTableHash);
5✔
1017
    pStmt->sql.siInfo.pTableHash = NULL;
5✔
1018
  }
1019

1020
  if (pStmt->sql.siInfo.pVgroupHash) {
9!
1021
    taosHashCleanup(pStmt->sql.siInfo.pVgroupHash);
×
1022
    pStmt->sql.siInfo.pVgroupHash = NULL;
×
1023
  }
1024

1025
  if (pStmt->sql.siInfo.pVgroupList) {
9!
1026
    taosArrayDestroy(pStmt->sql.siInfo.pVgroupList);
×
1027
    pStmt->sql.siInfo.pVgroupList = NULL;
×
1028
  }
1029

1030
  if (pStmt->sql.siInfo.pDataCtx) {
9✔
1031
    qDestroyStmtDataBlock(pStmt->sql.siInfo.pDataCtx);
3✔
1032
    pStmt->sql.siInfo.pDataCtx = NULL;
3✔
1033
  }
1034

1035
  if (pStmt->sql.siInfo.pTSchema) {
9✔
1036
    taosMemoryFree(pStmt->sql.siInfo.pTSchema);
3!
1037
    pStmt->sql.siInfo.pTSchema = NULL;
3✔
1038
  }
1039

1040
  if (pStmt->sql.siInfo.pRequest) {
9✔
1041
    taos_free_result(pStmt->sql.siInfo.pRequest);
3✔
1042
    pStmt->sql.siInfo.pRequest = NULL;
3✔
1043
  }
1044

1045
  if (stbInterlaceMode) {
9✔
1046
    STMT_ERR_RET(stmtResetStbInterlaceCache(pStmt));
5!
1047
  }
1048

1049
  pStmt->db = db;
9✔
1050
  pStmt->stbInterlaceMode = stbInterlaceMode;
9✔
1051
  pStmt->options = options;
9✔
1052
  pStmt->reqid = reqid;
9✔
1053

1054
  pStmt->sql.status = STMT_INIT;
9✔
1055

1056
  return TSDB_CODE_SUCCESS;
9✔
1057
}
1058

1059
int stmtPrepare2(TAOS_STMT2* stmt, const char* sql, unsigned long length) {
153✔
1060
  STscStmt2* pStmt = (STscStmt2*)stmt;
153✔
1061
  int32_t    code = 0;
153✔
1062

1063
  if (stmt == NULL || sql == NULL) {
153!
1064
    STMT2_ELOG_E("stmt or sql is NULL");
×
1065
    return TSDB_CODE_INVALID_PARA;
×
1066
  }
1067

1068
  if (pStmt->sql.status >= STMT_PREPARE) {
154✔
1069
    STMT2_DLOG("stmt status is %d, need to reset stmt2 cache before prepare", pStmt->sql.status);
9!
1070
    STMT_ERR_RET(stmtResetStmtForPrepare(pStmt));
9!
1071
  }
1072

1073
  STMT2_DLOG("start to prepare with sql:%s", sql);
154!
1074

1075
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
154✔
1076
    STMT2_ELOG("stmt errCode is not success, ErrCode: 0x%x, ErrMessage: %s\n. ", pStmt->errCode,
1!
1077
               strerror(pStmt->errCode));
1078
    return pStmt->errCode;
1✔
1079
  }
1080

1081
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_PREPARE));
153!
1082

1083
  if (length <= 0) {
153✔
1084
    length = strlen(sql);
93✔
1085
  }
1086

1087
  pStmt->sql.sqlStr = taosStrndup(sql, length);
153!
1088
  if (!pStmt->sql.sqlStr) {
153!
1089
    return terrno;
×
1090
  }
1091
  pStmt->sql.sqlLen = length;
153✔
1092
  pStmt->sql.stbInterlaceMode = pStmt->stbInterlaceMode;
153✔
1093

1094
  char* dbName = NULL;
153✔
1095
  if (qParseDbName(sql, length, &dbName)) {
153✔
1096
    STMT_ERR_RET(stmtSetDbName2(stmt, dbName));
54!
1097
    taosMemoryFreeClear(dbName);
54!
1098
  }
1099

1100
  return TSDB_CODE_SUCCESS;
151✔
1101
}
1102

1103
static int32_t stmtInitStbInterlaceTableInfo(STscStmt2* pStmt) {
75✔
1104
  STableDataCxt** pSrc = taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
75✔
1105
  if (!pSrc) {
76!
1106
    return terrno;
×
1107
  }
1108
  STableDataCxt* pDst = NULL;
76✔
1109

1110
  STMT_ERR_RET(qCloneStmtDataBlock(&pDst, *pSrc, true));
76!
1111
  pStmt->sql.siInfo.pDataCtx = pDst;
59✔
1112

1113
  SArray* pTblCols = NULL;
59✔
1114
  for (int32_t i = 0; i < STMT_TABLE_COLS_NUM; i++) {
61,388✔
1115
    pTblCols = taosArrayInit(20, POINTER_BYTES);
61,019✔
1116
    if (NULL == pTblCols) {
65,828!
1117
      return terrno;
×
1118
    }
1119

1120
    if (taosArrayPush(pStmt->sql.siInfo.pTableCols, &pTblCols) == NULL) {
127,157!
1121
      return terrno;
×
1122
    }
1123
  }
1124

1125
  pStmt->sql.siInfo.boundTags = pStmt->bInfo.boundTags;
369✔
1126

1127
  return TSDB_CODE_SUCCESS;
369✔
1128
}
1129

1130
int stmtIsInsert2(TAOS_STMT2* stmt, int* insert) {
11,748✔
1131
  STscStmt2* pStmt = (STscStmt2*)stmt;
11,748✔
1132

1133
  STMT_DLOG_E("start is insert");
11,748!
1134

1135
  if (pStmt->sql.type) {
11,751✔
1136
    *insert = (STMT_TYPE_INSERT == pStmt->sql.type || STMT_TYPE_MULTI_INSERT == pStmt->sql.type);
11,599✔
1137
  } else {
1138
    *insert = qIsInsertValuesSql(pStmt->sql.sqlStr, pStmt->sql.sqlLen);
152✔
1139
  }
1140

1141
  return TSDB_CODE_SUCCESS;
11,751✔
1142
}
1143

1144
int stmtSetTbName2(TAOS_STMT2* stmt, const char* tbName) {
5,845✔
1145
  STscStmt2* pStmt = (STscStmt2*)stmt;
5,845✔
1146

1147
  int64_t startUs = taosGetTimestampUs();
5,848✔
1148

1149
  STMT_DLOG("start to set tbName:%s", tbName);
5,848!
1150

1151
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
5,845!
1152
    return pStmt->errCode;
×
1153
  }
1154

1155
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_SETTBNAME));
5,845!
1156

1157
  int32_t insert = 0;
5,848✔
1158
  STMT_ERR_RET(stmtIsInsert2(stmt, &insert));
5,848!
1159
  if (0 == insert) {
5,845!
1160
    tscError("set tb name not available for none insert statement");
×
1161
    STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR);
×
1162
  }
1163

1164
  if (!pStmt->sql.stbInterlaceMode || NULL == pStmt->sql.siInfo.pDataCtx) {
5,845✔
1165
    STMT_ERR_RET(stmtCreateRequest(pStmt));
187!
1166

1167
    STMT_ERR_RET(qCreateSName(&pStmt->bInfo.sname, tbName, pStmt->taos->acctId, pStmt->exec.pRequest->pDb,
186!
1168
                              pStmt->exec.pRequest->msgBuf, pStmt->exec.pRequest->msgBufLen));
1169
    STMT_ERR_RET(tNameExtractFullName(&pStmt->bInfo.sname, pStmt->bInfo.tbFName));
185✔
1170

1171
    STMT_ERR_RET(stmtGetFromCache(pStmt));
185!
1172

1173
    if (pStmt->bInfo.needParse) {
184✔
1174
      tstrncpy(pStmt->bInfo.tbName, tbName, sizeof(pStmt->bInfo.tbName));
94✔
1175
      pStmt->bInfo.tbName[sizeof(pStmt->bInfo.tbName) - 1] = 0;
94✔
1176

1177
      STMT_ERR_RET(stmtParseSql(pStmt));
94!
1178
    }
1179
  } else {
1180
    tstrncpy(pStmt->bInfo.tbName, tbName, sizeof(pStmt->bInfo.tbName));
5,658✔
1181
    pStmt->bInfo.tbName[sizeof(pStmt->bInfo.tbName) - 1] = 0;
5,658✔
1182
    pStmt->exec.pRequest->requestId++;
5,658✔
1183
    pStmt->bInfo.needParse = false;
5,658✔
1184
  }
1185

1186
  if (pStmt->sql.stbInterlaceMode && NULL == pStmt->sql.siInfo.pDataCtx) {
5,841✔
1187
    STMT_ERR_RET(stmtInitStbInterlaceTableInfo(pStmt));
76!
1188
  }
1189

1190
  int64_t startUs2 = taosGetTimestampUs();
5,846✔
1191
  pStmt->stat.setTbNameUs += startUs2 - startUs;
5,846✔
1192

1193
  return TSDB_CODE_SUCCESS;
5,846✔
1194
}
1195

1196
int stmtSetTbTags2(TAOS_STMT2* stmt, TAOS_STMT2_BIND* tags, SVCreateTbReq** pCreateTbReq) {
120✔
1197
  STscStmt2* pStmt = (STscStmt2*)stmt;
120✔
1198

1199
  STMT_DLOG_E("start to set tbTags");
120!
1200

1201
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
120!
1202
    return pStmt->errCode;
×
1203
  }
1204

1205
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_SETTAGS));
120!
1206

1207
  if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 &&
120!
1208
      STMT_TYPE_MULTI_INSERT != pStmt->sql.type) {
×
1209
    pStmt->bInfo.needParse = false;
×
1210
  }
1211
  STMT_ERR_RET(stmtCreateRequest(pStmt));
120!
1212

1213
  if (pStmt->bInfo.needParse) {
120!
1214
    STMT_ERR_RET(stmtParseSql(pStmt));
×
1215
  }
1216
  if (pStmt->sql.stbInterlaceMode && NULL == pStmt->sql.siInfo.pDataCtx) {
120!
1217
    STMT_ERR_RET(stmtInitStbInterlaceTableInfo(pStmt));
×
1218
  }
1219

1220
  SBoundColInfo* tags_info = (SBoundColInfo*)pStmt->bInfo.boundTags;
120✔
1221
  // if (tags_info->numOfBound <= 0 || tags_info->numOfCols <= 0) {
1222
  //   tscWarn("no tags or cols bound in sql, will not bound tags");
1223
  //   return TSDB_CODE_SUCCESS;
1224
  // }
1225
  if (pStmt->sql.autoCreateTbl && pStmt->sql.stbInterlaceMode) {
120!
1226
    STMT_ERR_RET(qCreateSName(&pStmt->bInfo.sname, pStmt->bInfo.tbName, pStmt->taos->acctId, pStmt->exec.pRequest->pDb,
38!
1227
                              pStmt->exec.pRequest->msgBuf, pStmt->exec.pRequest->msgBufLen));
1228
    STMT_ERR_RET(tNameExtractFullName(&pStmt->bInfo.sname, pStmt->bInfo.tbFName));
38!
1229
  }
1230

1231
  STableDataCxt** pDataBlock = NULL;
120✔
1232
  if (pStmt->exec.pCurrBlock) {
120✔
1233
    pDataBlock = &pStmt->exec.pCurrBlock;
95✔
1234
  } else {
1235
    pDataBlock =
1236
        (STableDataCxt**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
25✔
1237
    if (NULL == pDataBlock) {
25!
1238
      tscError("table %s not found in exec blockHash", pStmt->bInfo.tbFName);
×
1239
      STMT_ERR_RET(TSDB_CODE_TSC_STMT_CACHE_ERROR);
×
1240
    }
1241
    // pStmt->exec.pCurrBlock = *pDataBlock;
1242
    // if (pStmt->sql.stbInterlaceMode) {
1243
    //   taosArrayDestroy(pStmt->exec.pCurrBlock->pData->aCol);
1244
    //   (*pDataBlock)->pData->aCol = NULL;
1245
    // }
1246
  }
1247
  if (pStmt->bInfo.inExecCache && !pStmt->sql.autoCreateTbl) {
120!
1248
    return TSDB_CODE_SUCCESS;
×
1249
  }
1250

1251
  tscDebug("start to bind stmt tag values");
120!
1252

1253
  void* boundTags = NULL;
120✔
1254
  if (pStmt->sql.stbInterlaceMode) {
120✔
1255
    boundTags = pStmt->sql.siInfo.boundTags;
38✔
1256
    *pCreateTbReq = taosMemoryCalloc(1, sizeof(SVCreateTbReq));
38!
1257
    if (NULL == pCreateTbReq) {
38!
1258
      return terrno;
×
1259
    }
1260
    int32_t vgId = -1;
38✔
1261
    STMT_ERR_RET(stmtTryAddTableVgroupInfo(pStmt, &vgId));
38!
1262
    (*pCreateTbReq)->uid = vgId;
38✔
1263
  } else {
1264
    boundTags = pStmt->bInfo.boundTags;
82✔
1265
  }
1266

1267
  STMT_ERR_RET(qBindStmtTagsValue2(*pDataBlock, boundTags, pStmt->bInfo.tbSuid, pStmt->bInfo.stbFName,
120✔
1268
                                   pStmt->bInfo.sname.tname, tags, pStmt->exec.pRequest->msgBuf,
1269
                                   pStmt->exec.pRequest->msgBufLen, pStmt->taos->optionInfo.charsetCxt, *pCreateTbReq));
1270

1271
  return TSDB_CODE_SUCCESS;
119✔
1272
}
1273

1274
int stmtCheckTags2(TAOS_STMT2* stmt, SVCreateTbReq** pCreateTbReq) {
12✔
1275
  STscStmt2* pStmt = (STscStmt2*)stmt;
12✔
1276

1277
  STMT_DLOG_E("start to set tbTags");
12!
1278

1279
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
12!
1280
    return pStmt->errCode;
×
1281
  }
1282

1283
  if (!pStmt->sql.stbInterlaceMode) {
12!
1284
    return TSDB_CODE_SUCCESS;
×
1285
  }
1286

1287
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_SETTAGS));
12!
1288

1289
  if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 &&
12!
1290
      STMT_TYPE_MULTI_INSERT != pStmt->sql.type) {
×
1291
    pStmt->bInfo.needParse = false;
×
1292
  }
1293
  STMT_ERR_RET(stmtCreateRequest(pStmt));
12!
1294

1295
  if (pStmt->bInfo.needParse) {
12!
1296
    STMT_ERR_RET(stmtParseSql(pStmt));
×
1297
    if (!pStmt->sql.autoCreateTbl) {
×
1298
      return TSDB_CODE_SUCCESS;
×
1299
    }
1300
  }
1301

1302
  if (pStmt->sql.stbInterlaceMode && NULL == pStmt->sql.siInfo.pDataCtx) {
12!
1303
    STMT_ERR_RET(stmtInitStbInterlaceTableInfo(pStmt));
×
1304
  }
1305

1306
  STMT_ERR_RET(qCreateSName(&pStmt->bInfo.sname, pStmt->bInfo.tbName, pStmt->taos->acctId, pStmt->exec.pRequest->pDb,
12!
1307
                            pStmt->exec.pRequest->msgBuf, pStmt->exec.pRequest->msgBufLen));
1308
  STMT_ERR_RET(tNameExtractFullName(&pStmt->bInfo.sname, pStmt->bInfo.tbFName));
12!
1309

1310
  STableDataCxt** pDataBlock = NULL;
12✔
1311
  if (pStmt->exec.pCurrBlock) {
12✔
1312
    pDataBlock = &pStmt->exec.pCurrBlock;
9✔
1313
  } else {
1314
    pDataBlock =
1315
        (STableDataCxt**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
3✔
1316
    if (NULL == pDataBlock) {
3!
1317
      tscError("table %s not found in exec blockHash", pStmt->bInfo.tbFName);
×
1318
      STMT_ERR_RET(TSDB_CODE_TSC_STMT_CACHE_ERROR);
×
1319
    }
1320
  }
1321

1322
  if (!((*pDataBlock)->pData->flags & SUBMIT_REQ_AUTO_CREATE_TABLE)) {
12!
1323
    return TSDB_CODE_SUCCESS;
×
1324
  }
1325

1326
  if (pStmt->sql.fixValueTags) {
12✔
1327
    STMT_ERR_RET(cloneSVreateTbReq(pStmt->sql.fixValueTbReq, pCreateTbReq));
9!
1328
    if ((*pCreateTbReq)->name) {
9!
1329
      taosMemoryFree((*pCreateTbReq)->name);
9!
1330
    }
1331
    (*pCreateTbReq)->name = taosStrdup(pStmt->bInfo.tbName);
9!
1332
    int32_t vgId = -1;
9✔
1333
    STMT_ERR_RET(stmtTryAddTableVgroupInfo(pStmt, &vgId));
9!
1334
    (*pCreateTbReq)->uid = vgId;
9✔
1335
    return TSDB_CODE_SUCCESS;
9✔
1336
  }
1337

1338
  if ((*pDataBlock)->pData->pCreateTbReq) {
3!
1339
    pStmt->sql.fixValueTags = true;
3✔
1340
    STMT_ERR_RET(cloneSVreateTbReq((*pDataBlock)->pData->pCreateTbReq, &pStmt->sql.fixValueTbReq));
3!
1341
    STMT_ERR_RET(cloneSVreateTbReq(pStmt->sql.fixValueTbReq, pCreateTbReq));
3!
1342
    (*pCreateTbReq)->uid = (*pDataBlock)->pMeta->vgId;
3✔
1343
  }
1344

1345
  return TSDB_CODE_SUCCESS;
3✔
1346
}
1347

1348
static int stmtFetchColFields2(STscStmt2* pStmt, int32_t* fieldNum, TAOS_FIELD_E** fields) {
×
1349
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
×
1350
    return pStmt->errCode;
×
1351
  }
1352

1353
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
×
1354
    tscError("invalid operation to get query column fileds");
×
1355
    STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR);
×
1356
  }
1357

1358
  STableDataCxt** pDataBlock = NULL;
×
1359

1360
  if (pStmt->sql.stbInterlaceMode) {
×
1361
    pDataBlock = &pStmt->sql.siInfo.pDataCtx;
×
1362
  } else {
1363
    pDataBlock =
1364
        (STableDataCxt**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
×
1365
    if (NULL == pDataBlock) {
×
1366
      tscError("table %s not found in exec blockHash", pStmt->bInfo.tbFName);
×
1367
      STMT_ERR_RET(TSDB_CODE_APP_ERROR);
×
1368
    }
1369
  }
1370

1371
  STMT_ERR_RET(qBuildStmtColFields(*pDataBlock, fieldNum, fields));
×
1372

1373
  return TSDB_CODE_SUCCESS;
×
1374
}
1375

1376
static int stmtFetchStbColFields2(STscStmt2* pStmt, int32_t* fieldNum, TAOS_FIELD_ALL** fields) {
32✔
1377
  int32_t    code = 0;
32✔
1378
  int32_t    preCode = pStmt->errCode;
32✔
1379

1380
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
32!
1381
    return pStmt->errCode;
×
1382
  }
1383

1384
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
32!
1385
    tscError("invalid operation to get query column fileds");
×
1386
    STMT_ERRI_JRET(TSDB_CODE_TSC_STMT_API_ERROR);
×
1387
  }
1388

1389
  STableDataCxt** pDataBlock = NULL;
32✔
1390
  bool            cleanStb = false;
32✔
1391

1392
  if (pStmt->sql.stbInterlaceMode && pStmt->sql.siInfo.pDataCtx != NULL) {
32✔
1393
    pDataBlock = &pStmt->sql.siInfo.pDataCtx;
7✔
1394
  } else {
1395
    cleanStb = true;
25✔
1396
    pDataBlock =
1397
        (STableDataCxt**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
25✔
1398
  }
1399

1400
  if (NULL == pDataBlock || NULL == *pDataBlock) {
32!
1401
    tscError("table %s not found in exec blockHash", pStmt->bInfo.tbFName);
×
1402
    STMT_ERRI_JRET(TSDB_CODE_APP_ERROR);
×
1403
  }
1404

1405
  STMT_ERRI_JRET(
32!
1406
      qBuildStmtStbColFields(*pDataBlock, pStmt->bInfo.boundTags, pStmt->bInfo.tbNameFlag, fieldNum, fields));
1407

1408
  if (pStmt->bInfo.tbType == TSDB_SUPER_TABLE && cleanStb) {
32!
1409
    qDestroyStmtDataBlock(*pDataBlock);
18✔
1410
    *pDataBlock = NULL;
18✔
1411
    if (taosHashRemove(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName)) != 0) {
18!
1412
      tscError("get fileds %s remove exec blockHash fail", pStmt->bInfo.tbFName);
×
1413
      STMT_ERRI_JRET(TSDB_CODE_APP_ERROR);
×
1414
    }
1415
    pStmt->sql.autoCreateTbl = false;
18✔
1416
    pStmt->bInfo.tagsCached = false;
18✔
1417
    pStmt->bInfo.sname = (SName){0};
18✔
1418
    stmtCleanBindInfo(pStmt);
18✔
1419
  }
1420

1421
_return:
14✔
1422

1423
  pStmt->errCode = preCode;
32✔
1424

1425
  return code;
32✔
1426
}
1427
/*
1428
SArray* stmtGetFreeCol(STscStmt2* pStmt, int32_t* idx) {
1429
  while (true) {
1430
    if (pStmt->exec.smInfo.pColIdx >= STMT_COL_BUF_SIZE) {
1431
      pStmt->exec.smInfo.pColIdx = 0;
1432
    }
1433

1434
    if ((pStmt->exec.smInfo.pColIdx + 1) == atomic_load_32(&pStmt->exec.smInfo.pColFreeIdx)) {
1435
      taosUsleep(1);
1436
      continue;
1437
    }
1438

1439
    *idx = pStmt->exec.smInfo.pColIdx;
1440
    return pStmt->exec.smInfo.pCols[pStmt->exec.smInfo.pColIdx++];
1441
  }
1442
}
1443
*/
1444
static int32_t stmtAppendTablePostHandle(STscStmt2* pStmt, SStmtQNode* param) {
5,736✔
1445
  if (NULL == pStmt->sql.siInfo.pVgroupHash) {
5,736✔
1446
    pStmt->sql.siInfo.pVgroupHash =
4,407✔
1447
        taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_ENTRY_LOCK);
4,403✔
1448
  }
1449
  if (NULL == pStmt->sql.siInfo.pVgroupList) {
5,740✔
1450
    pStmt->sql.siInfo.pVgroupList = taosArrayInit(64, POINTER_BYTES);
4,407✔
1451
  }
1452

1453
  if (NULL == pStmt->sql.siInfo.pRequest) {
5,735✔
1454
    STMT_ERR_RET(buildRequest(pStmt->taos->id, pStmt->sql.sqlStr, pStmt->sql.sqlLen, NULL, false,
74!
1455
                              (SRequestObj**)&pStmt->sql.siInfo.pRequest, pStmt->reqid));
1456

1457
    if (pStmt->reqid != 0) {
74!
1458
      pStmt->reqid++;
×
1459
    }
1460
    pStmt->exec.pRequest->syncQuery = true;
74✔
1461

1462
    pStmt->sql.siInfo.requestId = ((SRequestObj*)pStmt->sql.siInfo.pRequest)->requestId;
74✔
1463
    pStmt->sql.siInfo.requestSelf = ((SRequestObj*)pStmt->sql.siInfo.pRequest)->self;
74✔
1464
  }
1465

1466
  if (!pStmt->sql.siInfo.tbFromHash && pStmt->sql.siInfo.firstName[0] &&
5,735✔
1467
      0 == strcmp(pStmt->sql.siInfo.firstName, pStmt->bInfo.tbName)) {
84✔
1468
    pStmt->sql.siInfo.tbFromHash = true;
35✔
1469
  }
1470

1471
  if (0 == pStmt->sql.siInfo.firstName[0]) {
5,735✔
1472
    tstrncpy(pStmt->sql.siInfo.firstName, pStmt->bInfo.tbName, TSDB_TABLE_NAME_LEN);
72✔
1473
  }
1474

1475
  param->tblData.getFromHash = pStmt->sql.siInfo.tbFromHash;
5,735✔
1476
  param->next = NULL;
5,735✔
1477

1478
  (void)atomic_add_fetch_64(&pStmt->sql.siInfo.tbRemainNum, 1);
5,735✔
1479

1480
  stmtEnqueue(pStmt, param);
5,752✔
1481

1482
  return TSDB_CODE_SUCCESS;
5,750✔
1483
}
1484

1485
static FORCE_INLINE int32_t stmtGetTableColsFromCache(STscStmt2* pStmt, SArray** pTableCols) {
1486
  while (true) {
1487
    if (pStmt->sql.siInfo.pTableColsIdx < taosArrayGetSize(pStmt->sql.siInfo.pTableCols)) {
5,736!
1488
      *pTableCols = (SArray*)taosArrayGetP(pStmt->sql.siInfo.pTableCols, pStmt->sql.siInfo.pTableColsIdx++);
5,737✔
1489
      break;
5,737✔
1490
    } else {
1491
      SArray* pTblCols = NULL;
×
1492
      for (int32_t i = 0; i < 100; i++) {
×
1493
        pTblCols = taosArrayInit(20, POINTER_BYTES);
×
1494
        if (NULL == pTblCols) {
×
1495
          return terrno;
×
1496
        }
1497

1498
        if (taosArrayPush(pStmt->sql.siInfo.pTableCols, &pTblCols) == NULL) {
×
1499
          return terrno;
×
1500
        }
1501
      }
1502
    }
1503
  }
1504

1505
  return TSDB_CODE_SUCCESS;
5,737✔
1506
}
1507

1508
static int32_t stmtCacheBlock(STscStmt2* pStmt) {
116✔
1509
  if (pStmt->sql.type != STMT_TYPE_MULTI_INSERT) {
116✔
1510
    return TSDB_CODE_SUCCESS;
9✔
1511
  }
1512

1513
  uint64_t uid = pStmt->bInfo.tbUid;
107✔
1514
  uint64_t cacheUid = (TSDB_CHILD_TABLE == pStmt->bInfo.tbType) ? pStmt->bInfo.tbSuid : uid;
107!
1515

1516
  if (taosHashGet(pStmt->sql.pTableCache, &cacheUid, sizeof(cacheUid))) {
107✔
1517
    return TSDB_CODE_SUCCESS;
89✔
1518
  }
1519

1520
  STableDataCxt** pSrc = taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
18✔
1521
  if (!pSrc) {
18!
1522
    return terrno;
×
1523
  }
1524
  STableDataCxt* pDst = NULL;
18✔
1525

1526
  STMT_ERR_RET(qCloneStmtDataBlock(&pDst, *pSrc, true));
18!
1527

1528
  SStmtTableCache cache = {
18✔
1529
      .pDataCtx = pDst,
1530
      .boundTags = pStmt->bInfo.boundTags,
18✔
1531
  };
1532

1533
  if (taosHashPut(pStmt->sql.pTableCache, &cacheUid, sizeof(cacheUid), &cache, sizeof(cache))) {
18!
1534
    return terrno;
×
1535
  }
1536

1537
  if (pStmt->sql.autoCreateTbl) {
18✔
1538
    pStmt->bInfo.tagsCached = true;
15✔
1539
  } else {
1540
    pStmt->bInfo.boundTags = NULL;
3✔
1541
  }
1542

1543
  return TSDB_CODE_SUCCESS;
18✔
1544
}
1545

1546
static int stmtAddBatch2(TAOS_STMT2* stmt) {
4,526✔
1547
  STscStmt2* pStmt = (STscStmt2*)stmt;
4,526✔
1548

1549
  int64_t startUs = taosGetTimestampUs();
4,525✔
1550

1551
  STMT_DLOG_E("start to add batch");
4,525!
1552

1553
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
4,525!
1554
    return pStmt->errCode;
×
1555
  }
1556

1557
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_ADD_BATCH));
4,525!
1558

1559
  if (pStmt->sql.stbInterlaceMode) {
4,524✔
1560
    int64_t startUs2 = taosGetTimestampUs();
4,405✔
1561
    pStmt->stat.addBatchUs += startUs2 - startUs;
4,405✔
1562

1563
    pStmt->sql.siInfo.tableColsReady = false;
4,405✔
1564

1565
    SStmtQNode* param = NULL;
4,405✔
1566
    STMT_ERR_RET(stmtAllocQNodeFromBuf(&pStmt->sql.siInfo.tbBuf, (void**)&param));
8,810!
1567
    param->restoreTbCols = true;
4,405✔
1568
    param->next = NULL;
4,405✔
1569

1570
    if (pStmt->sql.autoCreateTbl) {
4,405✔
1571
      pStmt->bInfo.tagsCached = true;
22✔
1572
    }
1573

1574
    stmtEnqueue(pStmt, param);
4,405✔
1575

1576
    return TSDB_CODE_SUCCESS;
4,412✔
1577
  }
1578

1579
  STMT_ERR_RET(stmtCacheBlock(pStmt));
116!
1580

1581
  return TSDB_CODE_SUCCESS;
116✔
1582
}
1583
/*
1584
static int32_t stmtBackupQueryFields(STscStmt2* pStmt) {
1585
  SStmtQueryResInfo* pRes = &pStmt->sql.queryRes;
1586
  pRes->numOfCols = pStmt->exec.pRequest->body.resInfo.numOfCols;
1587
  pRes->precision = pStmt->exec.pRequest->body.resInfo.precision;
1588

1589
  int32_t size = pRes->numOfCols * sizeof(TAOS_FIELD);
1590
  pRes->fields = taosMemoryMalloc(size);
1591
  pRes->userFields = taosMemoryMalloc(size);
1592
  if (NULL == pRes->fields || NULL == pRes->userFields) {
1593
    STMT_ERR_RET(terrno);
1594
  }
1595
  (void)memcpy(pRes->fields, pStmt->exec.pRequest->body.resInfo.fields, size);
1596
  (void)memcpy(pRes->userFields, pStmt->exec.pRequest->body.resInfo.userFields, size);
1597

1598
  return TSDB_CODE_SUCCESS;
1599
}
1600

1601
static int32_t stmtRestoreQueryFields(STscStmt2* pStmt) {
1602
  SStmtQueryResInfo* pRes = &pStmt->sql.queryRes;
1603
  int32_t            size = pRes->numOfCols * sizeof(TAOS_FIELD);
1604

1605
  pStmt->exec.pRequest->body.resInfo.numOfCols = pRes->numOfCols;
1606
  pStmt->exec.pRequest->body.resInfo.precision = pRes->precision;
1607

1608
  if (NULL == pStmt->exec.pRequest->body.resInfo.fields) {
1609
    pStmt->exec.pRequest->body.resInfo.fields = taosMemoryMalloc(size);
1610
    if (NULL == pStmt->exec.pRequest->body.resInfo.fields) {
1611
      STMT_ERR_RET(terrno);
1612
    }
1613
    (void)memcpy(pStmt->exec.pRequest->body.resInfo.fields, pRes->fields, size);
1614
  }
1615

1616
  if (NULL == pStmt->exec.pRequest->body.resInfo.userFields) {
1617
    pStmt->exec.pRequest->body.resInfo.userFields = taosMemoryMalloc(size);
1618
    if (NULL == pStmt->exec.pRequest->body.resInfo.userFields) {
1619
      STMT_ERR_RET(terrno);
1620
    }
1621
    (void)memcpy(pStmt->exec.pRequest->body.resInfo.userFields, pRes->userFields, size);
1622
  }
1623

1624
  return TSDB_CODE_SUCCESS;
1625
}
1626
*/
1627
int stmtBindBatch2(TAOS_STMT2* stmt, TAOS_STMT2_BIND* bind, int32_t colIdx, SVCreateTbReq* pCreateTbReq) {
5,841✔
1628
  STscStmt2* pStmt = (STscStmt2*)stmt;
5,841✔
1629
  int32_t    code = 0;
5,841✔
1630

1631
  int64_t startUs = taosGetTimestampUs();
5,851✔
1632

1633
  STMT_DLOG("start to bind stmt data, colIdx:%d", colIdx);
5,851!
1634

1635
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
5,853!
1636
    return pStmt->errCode;
×
1637
  }
1638

1639
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_BIND));
5,853!
1640

1641
  if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 &&
5,854!
1642
      STMT_TYPE_MULTI_INSERT != pStmt->sql.type) {
×
1643
    pStmt->bInfo.needParse = false;
×
1644
  }
1645

1646
  if (pStmt->exec.pRequest && STMT_TYPE_QUERY == pStmt->sql.type && pStmt->sql.runTimes) {
5,854✔
1647
    taos_free_result(pStmt->exec.pRequest);
1✔
1648
    pStmt->exec.pRequest = NULL;
1✔
1649
  }
1650

1651
  STMT_ERR_RET(stmtCreateRequest(pStmt));
5,854!
1652
  if (pStmt->bInfo.needParse) {
5,850✔
1653
    code = stmtParseSql(pStmt);
6✔
1654
    if (code != TSDB_CODE_SUCCESS) {
6!
1655
      goto cleanup_root;
×
1656
    }
1657
  }
1658

1659
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
5,850✔
1660
    code = qStmtBindParams2(pStmt->sql.pQuery, bind, colIdx, pStmt->taos->optionInfo.charsetCxt);
5✔
1661
    if (code != TSDB_CODE_SUCCESS) {
5!
1662
      goto cleanup_root;
×
1663
    }
1664
    SParseContext ctx = {.requestId = pStmt->exec.pRequest->requestId,
5✔
1665
                         .acctId = pStmt->taos->acctId,
5✔
1666
                         .db = pStmt->exec.pRequest->pDb,
5✔
1667
                         .topicQuery = false,
1668
                         .pSql = pStmt->sql.sqlStr,
5✔
1669
                         .sqlLen = pStmt->sql.sqlLen,
5✔
1670
                         .pMsg = pStmt->exec.pRequest->msgBuf,
5✔
1671
                         .msgLen = ERROR_MSG_BUF_DEFAULT_SIZE,
1672
                         .pTransporter = pStmt->taos->pAppInfo->pTransporter,
5✔
1673
                         .pStmtCb = NULL,
1674
                         .pUser = pStmt->taos->user};
5✔
1675
    ctx.mgmtEpSet = getEpSet_s(&pStmt->taos->pAppInfo->mgmtEp);
5✔
1676
    code = catalogGetHandle(pStmt->taos->pAppInfo->clusterId, &ctx.pCatalog);
5✔
1677
    if (code != TSDB_CODE_SUCCESS) {
5!
1678
      goto cleanup_root;
×
1679
    }
1680
    code = qStmtParseQuerySql(&ctx, pStmt->sql.pQuery);
5✔
1681
    if (code != TSDB_CODE_SUCCESS) {
5!
1682
      goto cleanup_root;
×
1683
    }
1684

1685
    if (pStmt->sql.pQuery->haveResultSet) {
5!
1686
      STMT_ERR_RET(setResSchemaInfo(&pStmt->exec.pRequest->body.resInfo, pStmt->sql.pQuery->pResSchema,
10!
1687
                                    pStmt->sql.pQuery->numOfResCols, pStmt->sql.pQuery->pResExtSchema, true));
1688
      taosMemoryFreeClear(pStmt->sql.pQuery->pResSchema);
5!
1689
      taosMemoryFreeClear(pStmt->sql.pQuery->pResExtSchema);
5!
1690
      setResPrecision(&pStmt->exec.pRequest->body.resInfo, pStmt->sql.pQuery->precision);
5✔
1691
    }
1692

1693
    TSWAP(pStmt->exec.pRequest->dbList, pStmt->sql.pQuery->pDbList);
5✔
1694
    TSWAP(pStmt->exec.pRequest->tableList, pStmt->sql.pQuery->pTableList);
5✔
1695
    TSWAP(pStmt->exec.pRequest->targetTableList, pStmt->sql.pQuery->pTargetTableList);
5✔
1696

1697
    // if (STMT_TYPE_QUERY == pStmt->sql.queryRes) {
1698
    //   STMT_ERR_RET(stmtRestoreQueryFields(pStmt));
1699
    // }
1700

1701
    // STMT_ERR_RET(stmtBackupQueryFields(pStmt));
1702

1703
    return TSDB_CODE_SUCCESS;
5✔
1704

1705
  cleanup_root:
×
1706
    if (pStmt->sql.pQuery->pRoot) {
×
1707
      nodesDestroyNode(pStmt->sql.pQuery->pRoot);
×
1708
      pStmt->sql.pQuery->pRoot = NULL;
×
1709
    }
1710
    STMT_ERR_RET(code);
×
1711
  }
1712

1713
  if (pStmt->sql.stbInterlaceMode && NULL == pStmt->sql.siInfo.pDataCtx) {
5,845!
1714
    STMT_ERR_RET(stmtInitStbInterlaceTableInfo(pStmt));
×
1715
  }
1716

1717
  STableDataCxt** pDataBlock = NULL;
5,847✔
1718

1719
  if (pStmt->exec.pCurrBlock) {
5,847✔
1720
    pDataBlock = &pStmt->exec.pCurrBlock;
5,752✔
1721
  } else {
1722
    pDataBlock =
1723
        (STableDataCxt**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
95✔
1724
    if (NULL == pDataBlock) {
96!
1725
      tscError("table %s not found in exec blockHash", pStmt->bInfo.tbFName);
×
1726
      STMT_ERR_RET(TSDB_CODE_TSC_STMT_CACHE_ERROR);
×
1727
    }
1728
    pStmt->exec.pCurrBlock = *pDataBlock;
96✔
1729
    if (pStmt->sql.stbInterlaceMode) {
96✔
1730
      taosArrayDestroy(pStmt->exec.pCurrBlock->pData->aCol);
74✔
1731
      (*pDataBlock)->pData->aCol = NULL;
75✔
1732
    }
1733
    if (colIdx < -1) {
97✔
1734
      pStmt->sql.bindRowFormat = true;
1✔
1735
      taosArrayDestroy((*pDataBlock)->pData->aCol);
1✔
1736
      (*pDataBlock)->pData->aCol = taosArrayInit(20, POINTER_BYTES);
1✔
1737
    }
1738
  }
1739

1740
  int64_t startUs2 = taosGetTimestampUs();
5,847✔
1741
  pStmt->stat.bindDataUs1 += startUs2 - startUs;
5,847✔
1742

1743
  SStmtQNode* param = NULL;
5,847✔
1744
  if (pStmt->sql.stbInterlaceMode) {
5,847✔
1745
    STMT_ERR_RET(stmtAllocQNodeFromBuf(&pStmt->sql.siInfo.tbBuf, (void**)&param));
11,470!
1746
    STMT_ERR_RET(stmtGetTableColsFromCache(pStmt, &param->tblData.aCol));
11,473!
1747
    taosArrayClear(param->tblData.aCol);
5,737✔
1748

1749
    // param->tblData.aCol = taosArrayInit(20, POINTER_BYTES);
1750

1751
    param->restoreTbCols = false;
5,725✔
1752
    param->tblData.isOrdered = true;
5,725✔
1753
    param->tblData.isDuplicateTs = false;
5,725✔
1754
    tstrncpy(param->tblData.tbName, pStmt->bInfo.tbName, TSDB_TABLE_NAME_LEN);
5,725✔
1755

1756
    param->pCreateTbReq = pCreateTbReq;
5,725✔
1757
  }
1758

1759
  int64_t startUs3 = taosGetTimestampUs();
5,846✔
1760
  pStmt->stat.bindDataUs2 += startUs3 - startUs2;
5,846✔
1761

1762
  SArray* pCols = pStmt->sql.stbInterlaceMode ? param->tblData.aCol : (*pDataBlock)->pData->aCol;
5,846✔
1763

1764
  if (colIdx < 0) {
5,846✔
1765
    if (pStmt->sql.stbInterlaceMode) {
5,843✔
1766
      // (*pDataBlock)->pData->flags = 0;
1767
      (*pDataBlock)->pData->flags &= ~SUBMIT_REQ_COLUMN_DATA_FORMAT;
5,733✔
1768
      code = qBindStmtStbColsValue2(*pDataBlock, pCols, bind, pStmt->exec.pRequest->msgBuf,
5,733✔
1769
                                    pStmt->exec.pRequest->msgBufLen, &pStmt->sql.siInfo.pTSchema, pStmt->sql.pBindInfo,
5,733✔
1770
                                    pStmt->taos->optionInfo.charsetCxt);
5,733✔
1771
      param->tblData.isOrdered = (*pDataBlock)->ordered;
5,739✔
1772
      param->tblData.isDuplicateTs = (*pDataBlock)->duplicateTs;
5,739✔
1773
    } else {
1774
      if (colIdx == -1) {
111✔
1775
        if (pStmt->sql.bindRowFormat) {
109✔
1776
          tscError("can't mix bind row format and bind column format");
1!
1777
          STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR);
1!
1778
        }
1779
        code = qBindStmtColsValue2(*pDataBlock, pCols, bind, pStmt->exec.pRequest->msgBuf,
108✔
1780
                                   pStmt->exec.pRequest->msgBufLen, pStmt->taos->optionInfo.charsetCxt);
108✔
1781
      } else {
1782
        code = qBindStmt2RowValue(*pDataBlock, (*pDataBlock)->pData->aRowP, bind, pStmt->exec.pRequest->msgBuf,
2✔
1783
                                  pStmt->exec.pRequest->msgBufLen, &pStmt->sql.siInfo.pTSchema, pStmt->sql.pBindInfo,
2✔
1784
                                  pStmt->taos->optionInfo.charsetCxt);
2✔
1785
      }
1786
    }
1787

1788
    if (code) {
5,849✔
1789
      tscError("qBindStmtColsValue failed, error:%s", tstrerror(code));
1!
1790
      STMT_ERR_RET(code);
1!
1791
    }
1792
  } else {
1793
    if (pStmt->sql.stbInterlaceMode) {
6!
1794
      tscError("bind single column not allowed in stb insert mode");
×
1795
      STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR);
×
1796
    }
1797

1798
    if (pStmt->sql.bindRowFormat) {
6!
1799
      tscError("can't mix bind row format and bind column format");
×
1800
      STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR);
×
1801
    }
1802

1803
    if (colIdx != (pStmt->bInfo.sBindLastIdx + 1) && colIdx != 0) {
6!
1804
      tscError("bind column index not in sequence");
×
1805
      STMT_ERR_RET(TSDB_CODE_APP_ERROR);
×
1806
    }
1807

1808
    pStmt->bInfo.sBindLastIdx = colIdx;
6✔
1809

1810
    if (0 == colIdx) {
6✔
1811
      pStmt->bInfo.sBindRowNum = bind->num;
3✔
1812
    }
1813

1814
    code = qBindStmtSingleColValue2(*pDataBlock, pCols, bind, pStmt->exec.pRequest->msgBuf,
6✔
1815
                                    pStmt->exec.pRequest->msgBufLen, colIdx, pStmt->bInfo.sBindRowNum, pStmt->taos->optionInfo.charsetCxt);
6✔
1816
    if (code) {
6!
1817
      tscError("qBindStmtSingleColValue failed, error:%s", tstrerror(code));
×
1818
      STMT_ERR_RET(code);
×
1819
    }
1820
  }
1821

1822
  int64_t startUs4 = taosGetTimestampUs();
5,855✔
1823
  pStmt->stat.bindDataUs3 += startUs4 - startUs3;
5,855✔
1824

1825
  if (pStmt->sql.stbInterlaceMode) {
5,855✔
1826
    STMT_ERR_RET(stmtAppendTablePostHandle(pStmt, param));
5,739!
1827
  } else {
1828
    STMT_ERR_RET(stmtAddBatch2(pStmt));
116!
1829
  }
1830

1831
  pStmt->stat.bindDataUs4 += taosGetTimestampUs() - startUs4;
5,868✔
1832

1833
  return TSDB_CODE_SUCCESS;
5,868✔
1834
}
1835
/*
1836
int stmtUpdateTableUid(STscStmt2* pStmt, SSubmitRsp* pRsp) {
1837
  tscDebug("stmt start to update tbUid, blockNum:%d", pRsp->nBlocks);
1838

1839
  int32_t code = 0;
1840
  int32_t finalCode = 0;
1841
  size_t  keyLen = 0;
1842
  void*   pIter = taosHashIterate(pStmt->exec.pBlockHash, NULL);
1843
  while (pIter) {
1844
    STableDataCxt* pBlock = *(STableDataCxt**)pIter;
1845
    char*          key = taosHashGetKey(pIter, &keyLen);
1846

1847
    STableMeta* pMeta = qGetTableMetaInDataBlock(pBlock);
1848
    if (pMeta->uid) {
1849
      pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
1850
      continue;
1851
    }
1852

1853
    SSubmitBlkRsp* blkRsp = NULL;
1854
    int32_t        i = 0;
1855
    for (; i < pRsp->nBlocks; ++i) {
1856
      blkRsp = pRsp->pBlocks + i;
1857
      if (strlen(blkRsp->tblFName) != keyLen) {
1858
        continue;
1859
      }
1860

1861
      if (strncmp(blkRsp->tblFName, key, keyLen)) {
1862
        continue;
1863
      }
1864

1865
      break;
1866
    }
1867

1868
    if (i < pRsp->nBlocks) {
1869
      tscDebug("auto created table %s uid updated from %" PRIx64 " to %" PRIx64, blkRsp->tblFName, pMeta->uid,
1870
               blkRsp->uid);
1871

1872
      pMeta->uid = blkRsp->uid;
1873
      pStmt->bInfo.tbUid = blkRsp->uid;
1874
    } else {
1875
      tscDebug("table %s not found in submit rsp, will update from catalog", pStmt->bInfo.tbFName);
1876
      if (NULL == pStmt->pCatalog) {
1877
        code = catalogGetHandle(pStmt->taos->pAppInfo->clusterId, &pStmt->pCatalog);
1878
        if (code) {
1879
          pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
1880
          finalCode = code;
1881
          continue;
1882
        }
1883
      }
1884

1885
      code = stmtCreateRequest(pStmt);
1886
      if (code) {
1887
        pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
1888
        finalCode = code;
1889
        continue;
1890
      }
1891

1892
      STableMeta*      pTableMeta = NULL;
1893
      SRequestConnInfo conn = {.pTrans = pStmt->taos->pAppInfo->pTransporter,
1894
                               .requestId = pStmt->exec.pRequest->requestId,
1895
                               .requestObjRefId = pStmt->exec.pRequest->self,
1896
                               .mgmtEps = getEpSet_s(&pStmt->taos->pAppInfo->mgmtEp)};
1897
      code = catalogGetTableMeta(pStmt->pCatalog, &conn, &pStmt->bInfo.sname, &pTableMeta);
1898

1899
      pStmt->stat.ctgGetTbMetaNum++;
1900

1901
      taos_free_result(pStmt->exec.pRequest);
1902
      pStmt->exec.pRequest = NULL;
1903

1904
      if (code || NULL == pTableMeta) {
1905
        pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
1906
        finalCode = code;
1907
        taosMemoryFree(pTableMeta);
1908
        continue;
1909
      }
1910

1911
      pMeta->uid = pTableMeta->uid;
1912
      pStmt->bInfo.tbUid = pTableMeta->uid;
1913
      taosMemoryFree(pTableMeta);
1914
    }
1915

1916
    pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
1917
  }
1918

1919
  return finalCode;
1920
}
1921
*/
1922
/*
1923
int stmtStaticModeExec(TAOS_STMT* stmt) {
1924
  STscStmt2*   pStmt = (STscStmt2*)stmt;
1925
  int32_t     code = 0;
1926
  SSubmitRsp* pRsp = NULL;
1927
  if (pStmt->sql.staticMode) {
1928
    return TSDB_CODE_TSC_STMT_API_ERROR;
1929
  }
1930

1931
  STMT_DLOG_E("start to exec");
1932

1933
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_EXECUTE));
1934

1935
  STMT_ERR_RET(qBuildStmtOutputFromTbList(pStmt->sql.pQuery, pStmt->sql.pVgHash, pStmt->exec.pTbBlkList,
1936
pStmt->exec.pCurrBlock, pStmt->exec.tbBlkNum));
1937

1938
  launchQueryImpl(pStmt->exec.pRequest, pStmt->sql.pQuery, true, NULL);
1939

1940
  if (pStmt->exec.pRequest->code && NEED_CLIENT_HANDLE_ERROR(pStmt->exec.pRequest->code)) {
1941
    code = refreshMeta(pStmt->exec.pRequest->pTscObj, pStmt->exec.pRequest);
1942
    if (code) {
1943
      pStmt->exec.pRequest->code = code;
1944
    } else {
1945
      tFreeSSubmitRsp(pRsp);
1946
      STMT_ERR_RET(stmtResetStmt(pStmt));
1947
      STMT_ERR_RET(TSDB_CODE_NEED_RETRY);
1948
    }
1949
  }
1950

1951
  STMT_ERR_JRET(pStmt->exec.pRequest->code);
1952

1953
  pStmt->exec.affectedRows = taos_affected_rows(pStmt->exec.pRequest);
1954
  pStmt->affectedRows += pStmt->exec.affectedRows;
1955

1956
_return:
1957

1958
  stmtCleanExecInfo(pStmt, (code ? false : true), false);
1959

1960
  tFreeSSubmitRsp(pRsp);
1961

1962
  ++pStmt->sql.runTimes;
1963

1964
  STMT_RET(code);
1965
}
1966
*/
1967

1968
static int32_t createParseContext(const SRequestObj* pRequest, SParseContext** pCxt, SSqlCallbackWrapper* pWrapper) {
12✔
1969
  const STscObj* pTscObj = pRequest->pTscObj;
12✔
1970

1971
  *pCxt = taosMemoryCalloc(1, sizeof(SParseContext));
12!
1972
  if (*pCxt == NULL) {
12!
1973
    return terrno;
×
1974
  }
1975

1976
  **pCxt = (SParseContext){.requestId = pRequest->requestId,
12✔
1977
                           .requestRid = pRequest->self,
12✔
1978
                           .acctId = pTscObj->acctId,
12✔
1979
                           .db = pRequest->pDb,
12✔
1980
                           .topicQuery = false,
1981
                           .pSql = pRequest->sqlstr,
12✔
1982
                           .sqlLen = pRequest->sqlLen,
12✔
1983
                           .pMsg = pRequest->msgBuf,
12✔
1984
                           .msgLen = ERROR_MSG_BUF_DEFAULT_SIZE,
1985
                           .pTransporter = pTscObj->pAppInfo->pTransporter,
12✔
1986
                           .pStmtCb = NULL,
1987
                           .pUser = pTscObj->user,
12✔
1988
                           .pEffectiveUser = pRequest->effectiveUser,
12✔
1989
                           .isSuperUser = (0 == strcmp(pTscObj->user, TSDB_DEFAULT_USER)),
12✔
1990
                           .enableSysInfo = pTscObj->sysInfo,
12✔
1991
                           .async = true,
1992
                           .svrVer = pTscObj->sVer,
12✔
1993
                           .nodeOffline = (pTscObj->pAppInfo->onlineDnodes < pTscObj->pAppInfo->totalDnodes),
12✔
1994
                           .allocatorId = pRequest->allocatorRefId,
12✔
1995
                           .parseSqlFp = clientParseSql,
1996
                           .parseSqlParam = pWrapper};
1997
  int8_t biMode = atomic_load_8(&((STscObj*)pTscObj)->biMode);
12✔
1998
  (*pCxt)->biMode = biMode;
12✔
1999
  return TSDB_CODE_SUCCESS;
12✔
2000
}
2001

2002
static void asyncQueryCb(void* userdata, TAOS_RES* res, int code) {
12✔
2003
  STscStmt2*        pStmt = userdata;
12✔
2004
  __taos_async_fn_t fp = pStmt->options.asyncExecFn;
12✔
2005

2006
  pStmt->exec.affectedRows = taos_affected_rows(pStmt->exec.pRequest);
12✔
2007
  pStmt->affectedRows += pStmt->exec.affectedRows;
12✔
2008

2009
  fp(pStmt->options.userdata, res, code);
12✔
2010

2011
  while (0 == atomic_load_8((int8_t*)&pStmt->sql.siInfo.tableColsReady)) {
12!
2012
    taosUsleep(1);
×
2013
  }
2014
  (void)stmtCleanExecInfo(pStmt, (code ? false : true), false);
12✔
2015
  ++pStmt->sql.runTimes;
12✔
2016

2017
  if (tsem_post(&pStmt->asyncExecSem) != 0) {
12!
2018
    tscError("failed to post asyncExecSem");
×
2019
  }
2020
}
12✔
2021

2022
int stmtExec2(TAOS_STMT2* stmt, int* affected_rows) {
4,461✔
2023
  STscStmt2* pStmt = (STscStmt2*)stmt;
4,461✔
2024
  int32_t    code = 0;
4,461✔
2025
  int64_t    startUs = taosGetTimestampUs();
4,463✔
2026

2027
  STMT_DLOG_E("start to exec");
4,463!
2028

2029
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
4,462!
2030
    return pStmt->errCode;
×
2031
  }
2032

2033
  TSC_ERR_RET(taosThreadMutexLock(&pStmt->asyncBindParam.mutex));
4,462!
2034
  while (atomic_load_8((int8_t*)&pStmt->asyncBindParam.asyncBindNum) > 0) {
4,466!
2035
    (void)taosThreadCondWait(&pStmt->asyncBindParam.waitCond, &pStmt->asyncBindParam.mutex);
×
2036
  }
2037
  TSC_ERR_RET(taosThreadMutexUnlock(&pStmt->asyncBindParam.mutex));
4,463!
2038

2039
  if (pStmt->sql.stbInterlaceMode) {
4,465✔
2040
    STMT_ERR_RET(stmtAddBatch2(pStmt));
4,412!
2041
  }
2042

2043
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_EXECUTE));
4,466✔
2044

2045
  if (STMT_TYPE_QUERY != pStmt->sql.type) {
4,465✔
2046
    if (pStmt->sql.stbInterlaceMode) {
4,459✔
2047
      int64_t startTs = taosGetTimestampUs();
4,414✔
2048
      while (atomic_load_64(&pStmt->sql.siInfo.tbRemainNum)) {
9,020✔
2049
        taosUsleep(1);
4,620✔
2050
      }
2051
      pStmt->stat.execWaitUs += taosGetTimestampUs() - startTs;
4,402✔
2052

2053
      STMT_ERR_RET(qBuildStmtFinOutput(pStmt->sql.pQuery, pStmt->sql.pVgHash, pStmt->sql.siInfo.pVgroupList));
4,402!
2054
      taosHashCleanup(pStmt->sql.siInfo.pVgroupHash);
4,409✔
2055
      pStmt->sql.siInfo.pVgroupHash = NULL;
4,412✔
2056
      pStmt->sql.siInfo.pVgroupList = NULL;
4,412✔
2057
    } else {
2058
      tDestroySubmitTbData(pStmt->exec.pCurrTbData, TSDB_MSG_FLG_ENCODE);
47✔
2059
      taosMemoryFreeClear(pStmt->exec.pCurrTbData);
47!
2060

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

2063
      STMT_ERR_RET(qBuildStmtOutput(pStmt->sql.pQuery, pStmt->sql.pVgHash, pStmt->exec.pBlockHash));
47!
2064
    }
2065
  }
2066

2067
  SRequestObj*      pRequest = pStmt->exec.pRequest;
4,464✔
2068
  __taos_async_fn_t fp = pStmt->options.asyncExecFn;
4,464✔
2069

2070
  if (!fp) {
4,464✔
2071
    launchQueryImpl(pStmt->exec.pRequest, pStmt->sql.pQuery, true, NULL);
4,452✔
2072

2073
    if (pStmt->exec.pRequest->code && NEED_CLIENT_HANDLE_ERROR(pStmt->exec.pRequest->code)) {
4,447!
2074
      code = refreshMeta(pStmt->exec.pRequest->pTscObj, pStmt->exec.pRequest);
×
2075
      if (code) {
×
2076
        pStmt->exec.pRequest->code = code;
×
2077
      } else {
2078
        STMT_ERR_RET(stmtResetStmt(pStmt));
×
2079
        STMT_ERR_RET(TSDB_CODE_NEED_RETRY);
×
2080
      }
2081
    }
2082

2083
    STMT_ERR_JRET(pStmt->exec.pRequest->code);
4,450!
2084

2085
    pStmt->exec.affectedRows = taos_affected_rows(pStmt->exec.pRequest);
4,450✔
2086
    if (affected_rows) {
4,452✔
2087
      *affected_rows = pStmt->exec.affectedRows;
4,447✔
2088
    }
2089
    pStmt->affectedRows += pStmt->exec.affectedRows;
4,452✔
2090

2091
    while (0 == atomic_load_8((int8_t*)&pStmt->sql.siInfo.tableColsReady)) {
4,452!
2092
      taosUsleep(1);
×
2093
    }
2094

2095
    STMT_ERR_RET(stmtCleanExecInfo(pStmt, (code ? false : true), false));
4,447✔
2096

2097
    ++pStmt->sql.runTimes;
4,442✔
2098
  } else {
2099
    SSqlCallbackWrapper* pWrapper = taosMemoryCalloc(1, sizeof(SSqlCallbackWrapper));
12!
2100
    if (pWrapper == NULL) {
12!
2101
      code = terrno;
×
2102
    } else {
2103
      pWrapper->pRequest = pRequest;
12✔
2104
      pRequest->pWrapper = pWrapper;
12✔
2105
    }
2106
    if (TSDB_CODE_SUCCESS == code) {
12!
2107
      code = createParseContext(pRequest, &pWrapper->pParseCtx, pWrapper);
12✔
2108
    }
2109
    pRequest->syncQuery = false;
12✔
2110
    pRequest->body.queryFp = asyncQueryCb;
12✔
2111
    ((SSyncQueryParam*)(pRequest)->body.interParam)->userParam = pStmt;
12✔
2112

2113
    pStmt->execSemWaited = false;
12✔
2114
    launchAsyncQuery(pRequest, pStmt->sql.pQuery, NULL, pWrapper);
12✔
2115
  }
2116

2117
_return:
4,454✔
2118
  pStmt->stat.execUseUs += taosGetTimestampUs() - startUs;
4,452✔
2119

2120
  STMT_RET(code);
4,452!
2121
}
2122

2123
int stmtClose2(TAOS_STMT2* stmt) {
145✔
2124
  STscStmt2* pStmt = (STscStmt2*)stmt;
145✔
2125

2126
  STMT_DLOG_E("start to free stmt");
145!
2127

2128
  if (pStmt->bindThreadInUse) {
145✔
2129
    pStmt->queue.stopQueue = true;
81✔
2130

2131
    (void)taosThreadMutexLock(&pStmt->queue.mutex);
81✔
2132
    (void)atomic_add_fetch_64(&pStmt->queue.qRemainNum, 1);
81✔
2133
    (void)taosThreadCondSignal(&(pStmt->queue.waitCond));
81✔
2134
    (void)taosThreadMutexUnlock(&pStmt->queue.mutex);
81✔
2135

2136
    (void)taosThreadJoin(pStmt->bindThread, NULL);
81✔
2137
    pStmt->bindThreadInUse = false;
81✔
2138

2139
    (void)taosThreadCondDestroy(&pStmt->queue.waitCond);
81✔
2140
    (void)taosThreadMutexDestroy(&pStmt->queue.mutex);
81✔
2141
  }
2142

2143
  TSC_ERR_RET(taosThreadMutexLock(&pStmt->asyncBindParam.mutex));
145!
2144
  while (atomic_load_8((int8_t*)&pStmt->asyncBindParam.asyncBindNum) > 0) {
145!
2145
    (void)taosThreadCondWait(&pStmt->asyncBindParam.waitCond, &pStmt->asyncBindParam.mutex);
×
2146
  }
2147
  TSC_ERR_RET(taosThreadMutexUnlock(&pStmt->asyncBindParam.mutex));
145!
2148

2149
  (void)taosThreadCondDestroy(&pStmt->asyncBindParam.waitCond);
145✔
2150
  (void)taosThreadMutexDestroy(&pStmt->asyncBindParam.mutex);
145✔
2151

2152
  if (pStmt->options.asyncExecFn && !pStmt->execSemWaited) {
145!
2153
    if (tsem_wait(&pStmt->asyncExecSem) != 0) {
5!
2154
      tscError("failed to wait asyncExecSem");
×
2155
    }
2156
  }
2157

2158
  STMT_DLOG("stmt %p closed, stbInterlaceMode:%d, statInfo: ctgGetTbMetaNum=>%" PRId64 ", getCacheTbInfo=>%" PRId64
145!
2159
            ", parseSqlNum=>%" PRId64 ", pStmt->stat.bindDataNum=>%" PRId64
2160
            ", settbnameAPI:%u, bindAPI:%u, addbatchAPI:%u, execAPI:%u"
2161
            ", setTbNameUs:%" PRId64 ", bindDataUs:%" PRId64 ",%" PRId64 ",%" PRId64 ",%" PRId64 " addBatchUs:%" PRId64
2162
            ", execWaitUs:%" PRId64 ", execUseUs:%" PRId64,
2163
            pStmt, pStmt->sql.stbInterlaceMode, pStmt->stat.ctgGetTbMetaNum, pStmt->stat.getCacheTbInfo,
2164
            pStmt->stat.parseSqlNum, pStmt->stat.bindDataNum, pStmt->seqIds[STMT_SETTBNAME], pStmt->seqIds[STMT_BIND],
2165
            pStmt->seqIds[STMT_ADD_BATCH], pStmt->seqIds[STMT_EXECUTE], pStmt->stat.setTbNameUs,
2166
            pStmt->stat.bindDataUs1, pStmt->stat.bindDataUs2, pStmt->stat.bindDataUs3, pStmt->stat.bindDataUs4,
2167
            pStmt->stat.addBatchUs, pStmt->stat.execWaitUs, pStmt->stat.execUseUs);
2168

2169
  STMT_ERR_RET(stmtCleanSQLInfo(pStmt));
145!
2170

2171
  if (pStmt->options.asyncExecFn) {
145✔
2172
    if (tsem_destroy(&pStmt->asyncExecSem) != 0) {
5!
2173
      tscError("failed to destroy asyncExecSem");
×
2174
    }
2175
  }
2176
  taosMemoryFree(stmt);
145!
2177

2178
  return TSDB_CODE_SUCCESS;
145✔
2179
}
2180

2181
const char* stmtErrstr2(TAOS_STMT2* stmt) {
3✔
2182
  STscStmt2* pStmt = (STscStmt2*)stmt;
3✔
2183

2184
  if (stmt == NULL || NULL == pStmt->exec.pRequest) {
3!
2185
    return (char*)tstrerror(terrno);
3✔
2186
  }
2187

2188
  pStmt->exec.pRequest->code = terrno;
×
2189

2190
  return taos_errstr(pStmt->exec.pRequest);
×
2191
}
2192
/*
2193
int stmtAffectedRows(TAOS_STMT* stmt) { return ((STscStmt2*)stmt)->affectedRows; }
2194

2195
int stmtAffectedRowsOnce(TAOS_STMT* stmt) { return ((STscStmt2*)stmt)->exec.affectedRows; }
2196
*/
2197

2198
int stmtParseColFields2(TAOS_STMT2* stmt) {
47✔
2199
  int32_t    code = 0;
47✔
2200
  STscStmt2* pStmt = (STscStmt2*)stmt;
47✔
2201
  int32_t    preCode = pStmt->errCode;
47✔
2202

2203
  STMT_DLOG_E("start to get col fields");
47!
2204

2205
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
47!
2206
    return pStmt->errCode;
×
2207
  }
2208

2209
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
47!
2210
    STMT_ERRI_JRET(TSDB_CODE_TSC_STMT_API_ERROR);
×
2211
  }
2212

2213
  STMT_ERRI_JRET(stmtSwitchStatus(pStmt, STMT_FETCH_FIELDS));
47!
2214

2215
  if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 &&
47!
2216
      STMT_TYPE_MULTI_INSERT != pStmt->sql.type) {
3!
2217
    pStmt->bInfo.needParse = false;
×
2218
  }
2219
  if (pStmt->sql.stbInterlaceMode && pStmt->sql.siInfo.pDataCtx != NULL) {
47✔
2220
    pStmt->bInfo.needParse = false;
7✔
2221
  }
2222
  if (pStmt->exec.pRequest && STMT_TYPE_QUERY == pStmt->sql.type && pStmt->sql.runTimes) {
47!
2223
    taos_free_result(pStmt->exec.pRequest);
×
2224
    pStmt->exec.pRequest = NULL;
×
2225
    STMT_ERRI_JRET(stmtCreateRequest(pStmt));
×
2226
  }
2227

2228
  STMT_ERRI_JRET(stmtCreateRequest(pStmt));
47!
2229

2230
  if (pStmt->bInfo.needParse) {
47✔
2231
    STMT_ERRI_JRET(stmtParseSql(pStmt));
40✔
2232
  }
2233

2234
_return:
32✔
2235
  // compatible with previous versions
2236
  if (code == TSDB_CODE_PAR_TABLE_NOT_EXIST && (pStmt->bInfo.tbNameFlag & NO_DATA_USING_CLAUSE) == 0x0) {
47!
2237
    code = TSDB_CODE_TSC_STMT_TBNAME_ERROR;
1✔
2238
  }
2239

2240
  pStmt->errCode = preCode;
47✔
2241

2242
  return code;
47✔
2243
}
2244

2245
int stmtGetStbColFields2(TAOS_STMT2* stmt, int* nums, TAOS_FIELD_ALL** fields) {
47✔
2246
  int32_t code = stmtParseColFields2(stmt);
47✔
2247
  if (code != TSDB_CODE_SUCCESS) {
47✔
2248
    return code;
15✔
2249
  }
2250

2251
  return stmtFetchStbColFields2(stmt, nums, fields);
32✔
2252
}
2253

2254
int stmtGetParamNum2(TAOS_STMT2* stmt, int* nums) {
11✔
2255
  int32_t    code = 0;
11✔
2256
  STscStmt2* pStmt = (STscStmt2*)stmt;
11✔
2257
  int32_t    preCode = pStmt->errCode;
11✔
2258

2259
  STMT_DLOG_E("start to get param num");
11!
2260

2261
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
11!
2262
    return pStmt->errCode;
×
2263
  }
2264

2265
  STMT_ERRI_JRET(stmtSwitchStatus(pStmt, STMT_FETCH_FIELDS));
11!
2266

2267
  if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 &&
11!
2268
      STMT_TYPE_MULTI_INSERT != pStmt->sql.type) {
×
2269
    pStmt->bInfo.needParse = false;
×
2270
  }
2271

2272
  if (pStmt->exec.pRequest && STMT_TYPE_QUERY == pStmt->sql.type && pStmt->sql.runTimes) {
11!
2273
    taos_free_result(pStmt->exec.pRequest);
×
2274
    pStmt->exec.pRequest = NULL;
×
2275
  }
2276

2277
  STMT_ERRI_JRET(stmtCreateRequest(pStmt));
11!
2278

2279
  if (pStmt->bInfo.needParse) {
11!
2280
    STMT_ERRI_JRET(stmtParseSql(pStmt));
11✔
2281
  }
2282

2283
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
4!
2284
    *nums = taosArrayGetSize(pStmt->sql.pQuery->pPlaceholderValues);
4✔
2285
  } else {
2286
    STMT_ERRI_JRET(stmtFetchColFields2(stmt, nums, NULL));
×
2287
  }
2288

2289
_return:
×
2290

2291
  pStmt->errCode = preCode;
11✔
2292

2293
  return code;
11✔
2294
}
2295

2296
TAOS_RES* stmtUseResult2(TAOS_STMT2* stmt) {
5✔
2297
  STscStmt2* pStmt = (STscStmt2*)stmt;
5✔
2298

2299
  STMT_DLOG_E("start to use result");
5!
2300

2301
  if (STMT_TYPE_QUERY != pStmt->sql.type) {
5!
2302
    tscError("useResult only for query statement");
×
2303
    return NULL;
×
2304
  }
2305

2306
  return pStmt->exec.pRequest;
5✔
2307
}
2308

2309
int32_t stmtAsyncBindThreadFunc(void* args) {
×
2310
  qInfo("async stmt bind thread started");
×
2311

2312
  ThreadArgs* targs = (ThreadArgs*)args;
×
2313
  STscStmt2*  pStmt = (STscStmt2*)targs->stmt;
×
2314

2315
  int code = taos_stmt2_bind_param(targs->stmt, targs->bindv, targs->col_idx);
×
2316
  targs->fp(targs->param, NULL, code);
×
2317
  (void)taosThreadMutexLock(&(pStmt->asyncBindParam.mutex));
×
2318
  (void)atomic_sub_fetch_8(&pStmt->asyncBindParam.asyncBindNum, 1);
×
2319
  (void)taosThreadCondSignal(&(pStmt->asyncBindParam.waitCond));
×
2320
  (void)taosThreadMutexUnlock(&(pStmt->asyncBindParam.mutex));
×
2321
  taosMemoryFree(args);
×
2322

2323
  qInfo("async stmt bind thread stopped");
×
2324

2325
  return code;
×
2326
}
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