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

taosdata / TDengine / #4180

27 May 2025 10:49AM UTC coverage: 63.267% (+0.4%) from 62.885%
#4180

push

travis-ci

web-flow
TD-35056 (#31227)

158426 of 318644 branches covered (49.72%)

Branch coverage included in aggregate %.

243945 of 317346 relevant lines covered (76.87%)

18276423.81 hits per line

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

62.28
/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,289✔
13
    *pBuf = (char*)pTblBuf->pCurBuff + pTblBuf->buffOffset;
10,290✔
14
    pTblBuf->buffOffset += pTblBuf->buffUnit;
10,290✔
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,290✔
39
}
40

41
static bool stmtDequeue(STscStmt2* pStmt, SStmtQNode** param) {
10,287✔
42
  int i = 0;
10,287✔
43
  while (0 == atomic_load_64((int64_t*)&pStmt->queue.qRemainNum)) {
57,985✔
44
    if (pStmt->queue.stopQueue) {
47,724✔
45
      return false;
100✔
46
    }
47
    if (i < 10) {
47,624✔
48
      taosUsleep(1);
45,028✔
49
      i++;
45,040✔
50
    } else {
51
      (void)taosThreadMutexLock(&pStmt->queue.mutex);
2,596✔
52
      if (pStmt->queue.stopQueue) {
2,657!
53
        (void)taosThreadMutexUnlock(&pStmt->queue.mutex);
×
54
        return false;
×
55
      }
56
      if (0 == atomic_load_64((int64_t*)&pStmt->queue.qRemainNum)) {
2,657✔
57
        (void)taosThreadCondWait(&pStmt->queue.waitCond, &pStmt->queue.mutex);
2,648✔
58
      }
59
      (void)taosThreadMutexUnlock(&pStmt->queue.mutex);
2,658✔
60
    }
61
  }
62

63
  if (pStmt->queue.stopQueue && 0 == atomic_load_64((int64_t*)&pStmt->queue.qRemainNum)) {
10,078!
64
    return false;
×
65
  }
66

67
  (void)taosThreadMutexLock(&pStmt->queue.mutex);
10,078✔
68
  if (pStmt->queue.head == pStmt->queue.tail) {
10,189!
69
    pStmt->queue.qRemainNum = 0;
×
70
    (void)taosThreadMutexUnlock(&pStmt->queue.mutex);
×
71
    STMT2_ELOG_E("interlace queue is empty, cannot dequeue");
×
72
    return false;
×
73
  }
74

75
  SStmtQNode* node = pStmt->queue.head->next;
10,189✔
76
  pStmt->queue.head->next = node->next;
10,189✔
77
  if (pStmt->queue.tail == node) {
10,189✔
78
    pStmt->queue.tail = pStmt->queue.head;
7,300✔
79
  }
80
  node->next = NULL;
10,189✔
81
  *param = node;
10,189✔
82

83
  (void)atomic_sub_fetch_64((int64_t*)&pStmt->queue.qRemainNum, 1);
10,189✔
84
  (void)taosThreadMutexUnlock(&pStmt->queue.mutex);
10,186✔
85

86
  STMT2_DLOG("dequeue success, node:%p, remainNum:%" PRId64, node, pStmt->queue.qRemainNum);
10,193!
87

88
  return true;
10,192✔
89
}
90

91
static void stmtEnqueue(STscStmt2* pStmt, SStmtQNode* param) {
10,179✔
92
  if (param == NULL) {
10,179!
93
    STMT2_ELOG_E("enqueue param is NULL");
×
94
    return;
×
95
  }
96

97
  param->next = NULL;
10,179✔
98

99
  (void)taosThreadMutexLock(&pStmt->queue.mutex);
10,179✔
100

101
  pStmt->queue.tail->next = param;
10,188✔
102
  pStmt->queue.tail = param;
10,188✔
103
  pStmt->stat.bindDataNum++;
10,188✔
104

105
  (void)atomic_add_fetch_64(&pStmt->queue.qRemainNum, 1);
10,188✔
106
  (void)taosThreadCondSignal(&(pStmt->queue.waitCond));
10,193✔
107

108
  (void)taosThreadMutexUnlock(&pStmt->queue.mutex);
10,193✔
109

110
  STMT2_TLOG("enqueue param:%p, remainNum:%" PRId64 ", restoreTbCols:%d", param, pStmt->queue.qRemainNum,
10,191!
111
             param->restoreTbCols);
112
}
113

114
static int32_t stmtCreateRequest(STscStmt2* pStmt) {
6,542✔
115
  int32_t code = 0;
6,542✔
116

117
  if (pStmt->exec.pRequest == NULL) {
6,542✔
118
    code = buildRequest(pStmt->taos->id, pStmt->sql.sqlStr, pStmt->sql.sqlLen, NULL, false, &pStmt->exec.pRequest,
193✔
119
                        pStmt->reqid);
120
    if (pStmt->reqid != 0) {
193!
121
      pStmt->reqid++;
×
122
    }
123
    pStmt->exec.pRequest->type = RES_TYPE__QUERY;
193✔
124
    if (pStmt->db != NULL) {
193✔
125
      taosMemoryFreeClear(pStmt->exec.pRequest->pDb);
128!
126
      pStmt->exec.pRequest->pDb = taosStrdup(pStmt->db);
128!
127
    }
128
    if (TSDB_CODE_SUCCESS == code) {
193!
129
      pStmt->exec.pRequest->syncQuery = true;
193✔
130
      pStmt->exec.pRequest->isStmtBind = true;
193✔
131
    }
132
  }
133

134
  STMT2_TLOG("stmtCreateRequest, pRequest:%p, code:%d, db:%s", pStmt->exec.pRequest, code, pStmt->db);
6,542!
135

136
  return code;
6,542✔
137
}
138

139
static int32_t stmtSwitchStatus(STscStmt2* pStmt, STMT_STATUS newStatus) {
21,101✔
140
  int32_t code = 0;
21,101✔
141

142
  if (newStatus >= STMT_INIT && newStatus < STMT_MAX) {
21,101!
143
    STMT2_ELOG("stmt already failed with err:%s", tstrerror(pStmt->errCode));
21,109!
144
    STMT_LOG_SEQ(newStatus);
21,168!
145
  }
146

147
  if (pStmt->errCode && newStatus != STMT_PREPARE) {
21,175!
148
    STMT2_ELOG("stmt already failed with err:%s, please use stmt prepare", tstrerror(pStmt->errCode));
×
149
    return pStmt->errCode;
×
150
  }
151

152
  switch (newStatus) {
21,175!
153
    case STMT_PREPARE:
167✔
154
      pStmt->errCode = 0;
167✔
155
      break;
167✔
156
    case STMT_SETTBNAME:
5,879✔
157
      if (STMT_STATUS_EQ(INIT)) {
5,879!
158
        code = TSDB_CODE_TSC_STMT_API_ERROR;
×
159
      }
160
      if (!pStmt->sql.stbInterlaceMode && (STMT_STATUS_EQ(BIND) || STMT_STATUS_EQ(BIND_COL))) {
5,879!
161
        code = TSDB_CODE_TSC_STMT_API_ERROR;
×
162
      }
163
      break;
5,879✔
164
    case STMT_SETTAGS:
146✔
165
      if (STMT_STATUS_EQ(INIT)) {
146!
166
        code = TSDB_CODE_TSC_STMT_API_ERROR;
×
167
      }
168
      break;
146✔
169
    case STMT_FETCH_FIELDS:
68✔
170
      if (STMT_STATUS_EQ(INIT)) {
68!
171
        code = TSDB_CODE_TSC_STMT_API_ERROR;
×
172
      }
173
      break;
68✔
174
    case STMT_BIND:
5,890✔
175
      if (STMT_STATUS_EQ(INIT) || STMT_STATUS_EQ(BIND_COL)) {
5,890!
176
        code = TSDB_CODE_TSC_STMT_API_ERROR;
×
177
      }
178
      /*
179
            if ((pStmt->sql.type == STMT_TYPE_MULTI_INSERT) && ()) {
180
              code = TSDB_CODE_TSC_STMT_API_ERROR;
181
            }
182
      */
183
      break;
5,890✔
184
    case STMT_BIND_COL:
×
185
      if (STMT_STATUS_EQ(INIT) || STMT_STATUS_EQ(BIND)) {
×
186
        code = TSDB_CODE_TSC_STMT_API_ERROR;
×
187
      }
188
      break;
×
189
    case STMT_ADD_BATCH:
4,544✔
190
      if (STMT_STATUS_NE(BIND) && STMT_STATUS_NE(BIND_COL) && STMT_STATUS_NE(FETCH_FIELDS)) {
4,544!
191
        code = TSDB_CODE_TSC_STMT_API_ERROR;
×
192
      }
193
      break;
4,544✔
194
    case STMT_EXECUTE:
4,481✔
195
      if (STMT_TYPE_QUERY == pStmt->sql.type) {
4,481✔
196
        if (STMT_STATUS_NE(ADD_BATCH) && STMT_STATUS_NE(FETCH_FIELDS) && STMT_STATUS_NE(BIND) &&
5!
197
            STMT_STATUS_NE(BIND_COL)) {
×
198
          code = TSDB_CODE_TSC_STMT_API_ERROR;
×
199
        }
200
      } else {
201
        if (STMT_STATUS_NE(ADD_BATCH) && STMT_STATUS_NE(FETCH_FIELDS)) {
4,476!
202
          code = TSDB_CODE_TSC_STMT_API_ERROR;
1✔
203
        }
204
      }
205
      break;
4,481✔
206
    default:
×
207
      code = TSDB_CODE_APP_ERROR;
×
208
      break;
×
209
  }
210

211
  STMT_ERR_RET(code);
21,175✔
212

213
  pStmt->sql.status = newStatus;
21,174✔
214

215
  return TSDB_CODE_SUCCESS;
21,174✔
216
}
217

218
static int32_t stmtGetTbName(TAOS_STMT2* stmt, char** tbName) {
142✔
219
  STscStmt2* pStmt = (STscStmt2*)stmt;
142✔
220

221
  pStmt->sql.type = STMT_TYPE_MULTI_INSERT;
142✔
222

223
  if ('\0' == pStmt->bInfo.tbName[0]) {
142✔
224
    tscWarn("no table name set, OK if it is a stmt get fields");
33!
225
    STMT_ERR_RET(TSDB_CODE_TSC_STMT_TBNAME_ERROR);
33!
226
  }
227

228
  *tbName = pStmt->bInfo.tbName;
109✔
229

230
  return TSDB_CODE_SUCCESS;
109✔
231
}
232

233
static int32_t stmtUpdateBindInfo(TAOS_STMT2* stmt, STableMeta* pTableMeta, void* tags, SName* tbName,
145✔
234
                                  const char* sTableName, bool autoCreateTbl, int8_t tbNameFlag) {
235
  STscStmt2* pStmt = (STscStmt2*)stmt;
145✔
236
  char       tbFName[TSDB_TABLE_FNAME_LEN];
237
  int32_t    code = tNameExtractFullName(tbName, tbFName);
145✔
238
  if (code != 0) {
147!
239
    return code;
×
240
  }
241

242
  (void)memcpy(&pStmt->bInfo.sname, tbName, sizeof(*tbName));
147✔
243
  tstrncpy(pStmt->bInfo.tbFName, tbFName, sizeof(pStmt->bInfo.tbFName));
147✔
244
  pStmt->bInfo.tbFName[sizeof(pStmt->bInfo.tbFName) - 1] = 0;
147✔
245

246
  pStmt->bInfo.tbUid = autoCreateTbl ? 0 : pTableMeta->uid;
147✔
247
  pStmt->bInfo.tbSuid = pTableMeta->suid;
147✔
248
  pStmt->bInfo.tbVgId = pTableMeta->vgId;
147✔
249
  pStmt->bInfo.tbType = pTableMeta->tableType;
147✔
250

251
  if (!pStmt->bInfo.tagsCached) {
147✔
252
    qDestroyBoundColInfo(pStmt->bInfo.boundTags);
140✔
253
    taosMemoryFreeClear(pStmt->bInfo.boundTags);
137!
254
  }
255

256
  pStmt->bInfo.boundTags = tags;
143✔
257
  pStmt->bInfo.tagsCached = false;
143✔
258
  pStmt->bInfo.tbNameFlag = tbNameFlag;
143✔
259
  tstrncpy(pStmt->bInfo.stbFName, sTableName, sizeof(pStmt->bInfo.stbFName));
143✔
260

261
  return TSDB_CODE_SUCCESS;
143✔
262
}
263

264
static int32_t stmtUpdateExecInfo(TAOS_STMT2* stmt, SHashObj* pVgHash, SHashObj* pBlockHash) {
141✔
265
  STscStmt2* pStmt = (STscStmt2*)stmt;
141✔
266

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

270
  return TSDB_CODE_SUCCESS;
141✔
271
}
272

273
static int32_t stmtUpdateInfo(TAOS_STMT2* stmt, STableMeta* pTableMeta, void* tags, SName* tbName, bool autoCreateTbl,
145✔
274
                              SHashObj* pVgHash, SHashObj* pBlockHash, const char* sTableName, uint8_t tbNameFlag) {
275
  STscStmt2* pStmt = (STscStmt2*)stmt;
145✔
276

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

280
  pStmt->sql.autoCreateTbl = autoCreateTbl;
141✔
281

282
  return TSDB_CODE_SUCCESS;
141✔
283
}
284

285
static int32_t stmtGetExecInfo(TAOS_STMT2* stmt, SHashObj** pVgHash, SHashObj** pBlockHash) {
11✔
286
  STscStmt2* pStmt = (STscStmt2*)stmt;
11✔
287

288
  *pVgHash = pStmt->sql.pVgHash;
11✔
289
  pStmt->sql.pVgHash = NULL;
11✔
290

291
  *pBlockHash = pStmt->exec.pBlockHash;
11✔
292
  pStmt->exec.pBlockHash = NULL;
11✔
293

294
  return TSDB_CODE_SUCCESS;
11✔
295
}
296

297
static int32_t stmtParseSql(STscStmt2* pStmt) {
175✔
298
  pStmt->exec.pCurrBlock = NULL;
175✔
299

300
  STMT2_DLOG_E("start to stmtParseSql");
175!
301

302
  SStmtCallback stmtCb = {
175✔
303
      .pStmt = pStmt,
304
      .getTbNameFn = stmtGetTbName,
305
      .setInfoFn = stmtUpdateInfo,
306
      .getExecInfoFn = stmtGetExecInfo,
307
  };
308

309
  STMT_ERR_RET(stmtCreateRequest(pStmt));
175!
310

311
  pStmt->stat.parseSqlNum++;
175✔
312
  STMT_ERR_RET(parseSql(pStmt->exec.pRequest, false, &pStmt->sql.pQuery, &stmtCb));
175✔
313

314
  pStmt->sql.siInfo.pQuery = pStmt->sql.pQuery;
151✔
315

316
  pStmt->bInfo.needParse = false;
151✔
317

318
  if (pStmt->sql.pQuery->pRoot && 0 == pStmt->sql.type) {
151✔
319
    pStmt->sql.type = STMT_TYPE_INSERT;
11✔
320
    pStmt->sql.stbInterlaceMode = false;
11✔
321
  } else if (pStmt->sql.pQuery->pPrepareRoot) {
140✔
322
    pStmt->sql.type = STMT_TYPE_QUERY;
6✔
323
    pStmt->sql.stbInterlaceMode = false;
6✔
324

325
    return TSDB_CODE_SUCCESS;
6✔
326
  }
327

328
  STableDataCxt** pSrc =
329
      (STableDataCxt**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
145✔
330
  if (NULL == pSrc || NULL == *pSrc) {
148!
331
    STMT2_ELOG("fail to get exec.pBlockHash, maybe parse failed, tbFName:%s", pStmt->bInfo.tbFName);
×
332
    return terrno;
×
333
  }
334

335
  STableDataCxt* pTableCtx = *pSrc;
148✔
336
  if (pStmt->sql.stbInterlaceMode && pTableCtx->pData->pCreateTbReq && (pStmt->bInfo.tbNameFlag & USING_CLAUSE) == 0) {
148✔
337
    STMT2_TLOG("destroy pCreateTbReq for no-using insert, tbFName:%s", pStmt->bInfo.tbFName);
5!
338
    tdDestroySVCreateTbReq(pTableCtx->pData->pCreateTbReq);
5!
339
    taosMemoryFreeClear(pTableCtx->pData->pCreateTbReq);
5!
340
    pTableCtx->pData->pCreateTbReq = NULL;
5✔
341
  }
342
  // if (pStmt->sql.stbInterlaceMode) {
343
  //   int16_t lastIdx = -1;
344

345
  //   for (int32_t i = 0; i < pTableCtx->boundColsInfo.numOfBound; ++i) {
346
  //     if (pTableCtx->boundColsInfo.pColIndex[i] < lastIdx) {
347
  //       pStmt->sql.stbInterlaceMode = false;
348
  //       break;
349
  //     }
350

351
  //     lastIdx = pTableCtx->boundColsInfo.pColIndex[i];
352
  //   }
353
  // }
354

355
  if (NULL == pStmt->sql.pBindInfo) {
148✔
356
    pStmt->sql.pBindInfo = taosMemoryMalloc(pTableCtx->boundColsInfo.numOfBound * sizeof(*pStmt->sql.pBindInfo));
134!
357
    if (NULL == pStmt->sql.pBindInfo) {
133!
358
      STMT2_ELOG_E("fail to malloc pBindInfo");
×
359
      return terrno;
×
360
    }
361
  }
362

363
  return TSDB_CODE_SUCCESS;
147✔
364
}
365

366
static int32_t stmtCleanBindInfo(STscStmt2* pStmt) {
4,756✔
367
  pStmt->bInfo.tbUid = 0;
4,756✔
368
  pStmt->bInfo.tbVgId = -1;
4,756✔
369
  pStmt->bInfo.tbType = 0;
4,756✔
370
  pStmt->bInfo.needParse = true;
4,756✔
371
  pStmt->bInfo.inExecCache = false;
4,756✔
372

373
  pStmt->bInfo.tbName[0] = 0;
4,756✔
374
  pStmt->bInfo.tbFName[0] = 0;
4,756✔
375
  if (!pStmt->bInfo.tagsCached) {
4,756✔
376
    qDestroyBoundColInfo(pStmt->bInfo.boundTags);
4,683✔
377
    taosMemoryFreeClear(pStmt->bInfo.boundTags);
4,689!
378
  }
379
  if (!pStmt->sql.autoCreateTbl) {
4,763✔
380
    pStmt->bInfo.stbFName[0] = 0;
4,654✔
381
    pStmt->bInfo.tbSuid = 0;
4,654✔
382
  }
383

384
  STMT2_TLOG("finish clean bind info, tagsCached:%d, autoCreateTbl:%d", pStmt->bInfo.tagsCached,
4,763!
385
             pStmt->sql.autoCreateTbl);
386

387
  return TSDB_CODE_SUCCESS;
4,761✔
388
}
389

390
static void stmtFreeTableBlkList(STableColsData* pTb) {
×
391
  (void)qResetStmtColumns(pTb->aCol, true);
×
392
  taosArrayDestroy(pTb->aCol);
×
393
}
×
394

395
static void stmtResetQueueTableBuf(STableBufInfo* pTblBuf, SStmtQueue* pQueue) {
4,425✔
396
  pTblBuf->pCurBuff = taosArrayGetP(pTblBuf->pBufList, 0);
4,425✔
397
  if (NULL == pTblBuf->pCurBuff) {
4,417✔
398
    tscError("QInfo:%p, fail to get buffer from list", pTblBuf);
5!
399
    return;
×
400
  }
401
  pTblBuf->buffIdx = 1;
4,412✔
402
  pTblBuf->buffOffset = sizeof(*pQueue->head);
4,412✔
403

404
  pQueue->head = pQueue->tail = pTblBuf->pCurBuff;
4,412✔
405
  pQueue->qRemainNum = 0;
4,412✔
406
  pQueue->head->next = NULL;
4,412✔
407
}
408

409
static int32_t stmtCleanExecInfo(STscStmt2* pStmt, bool keepTable, bool deepClean) {
4,641✔
410
  if (pStmt->sql.stbInterlaceMode) {
4,641✔
411
    if (deepClean) {
4,518✔
412
      taosHashCleanup(pStmt->exec.pBlockHash);
93✔
413
      pStmt->exec.pBlockHash = NULL;
93✔
414

415
      if (NULL != pStmt->exec.pCurrBlock) {
93✔
416
        taosMemoryFreeClear(pStmt->exec.pCurrBlock->boundColsInfo.pColIndex);
89!
417
        taosMemoryFreeClear(pStmt->exec.pCurrBlock->pData);
89!
418
        qDestroyStmtDataBlock(pStmt->exec.pCurrBlock);
89✔
419
        pStmt->exec.pCurrBlock = NULL;
89✔
420
      }
421
      if (STMT_TYPE_QUERY != pStmt->sql.type) {
93!
422
        taos_free_result(pStmt->exec.pRequest);
93✔
423
        pStmt->exec.pRequest = NULL;
93✔
424
      }
425
    } else {
426
      pStmt->sql.siInfo.pTableColsIdx = 0;
4,425✔
427
      stmtResetQueueTableBuf(&pStmt->sql.siInfo.tbBuf, &pStmt->queue);
4,425✔
428
    }
429
    if (NULL != pStmt->exec.pRequest) {
4,506✔
430
      pStmt->exec.pRequest->body.resInfo.numOfRows = 0;
4,414✔
431
    }
432
  } else {
433
    if (STMT_TYPE_QUERY != pStmt->sql.type || deepClean) {
127✔
434
      // if (!pStmt->options.asyncExecFn) {
435
      taos_free_result(pStmt->exec.pRequest);
122✔
436
      pStmt->exec.pRequest = NULL;
122✔
437
      //}
438
    }
439

440
    size_t keyLen = 0;
127✔
441
    void*  pIter = taosHashIterate(pStmt->exec.pBlockHash, NULL);
127✔
442
    while (pIter) {
269✔
443
      STableDataCxt* pBlocks = *(STableDataCxt**)pIter;
142✔
444
      char*          key = taosHashGetKey(pIter, &keyLen);
142✔
445
      STableMeta*    pMeta = qGetTableMetaInDataBlock(pBlocks);
142✔
446

447
      if (keepTable && pBlocks == pStmt->exec.pCurrBlock) {
142✔
448
        TSWAP(pBlocks->pData, pStmt->exec.pCurrTbData);
47✔
449
        STMT_ERR_RET(qResetStmtDataBlock(pBlocks, false));
99!
450

451
        pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
47✔
452
        continue;
47✔
453
      }
454

455
      qDestroyStmtDataBlock(pBlocks);
95✔
456
      STMT_ERR_RET(taosHashRemove(pStmt->exec.pBlockHash, key, keyLen));
95!
457

458
      pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
95✔
459
    }
460

461
    if (keepTable) {
127✔
462
      STMT2_TLOG("finish clean exec info, stbInterlaceMode:%d, keepTable:%d, deepClean:%d", pStmt->sql.stbInterlaceMode,
52!
463
                 keepTable, deepClean);
464
      return TSDB_CODE_SUCCESS;
52✔
465
    }
466

467
    taosHashCleanup(pStmt->exec.pBlockHash);
75✔
468
    pStmt->exec.pBlockHash = NULL;
75✔
469

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

474
  STMT_ERR_RET(stmtCleanBindInfo(pStmt));
4,581!
475
  STMT2_TLOG("finish clean exec info, stbInterlaceMode:%d, keepTable:%d, deepClean:%d", pStmt->sql.stbInterlaceMode,
4,585!
476
             keepTable, deepClean);
477

478
  return TSDB_CODE_SUCCESS;
4,586✔
479
}
480

481
static void stmtFreeTbBuf(void* buf) {
100✔
482
  void* pBuf = *(void**)buf;
100✔
483
  taosMemoryFree(pBuf);
100!
484
}
100✔
485

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

491
static int32_t stmtCleanSQLInfo(STscStmt2* pStmt) {
147✔
492
  STMT2_TLOG_E("start to free SQL info");
147!
493

494
  taosMemoryFree(pStmt->sql.pBindInfo);
147!
495
  taosMemoryFree(pStmt->sql.queryRes.fields);
147!
496
  taosMemoryFree(pStmt->sql.queryRes.userFields);
147!
497
  taosMemoryFree(pStmt->sql.sqlStr);
147!
498
  qDestroyQuery(pStmt->sql.pQuery);
147✔
499
  taosArrayDestroy(pStmt->sql.nodeList);
147✔
500
  taosHashCleanup(pStmt->sql.pVgHash);
147✔
501
  pStmt->sql.pVgHash = NULL;
147✔
502
  if (pStmt->sql.fixValueTags) {
147✔
503
    tdDestroySVCreateTbReq(pStmt->sql.fixValueTbReq);
5!
504
  }
505

506
  void* pIter = taosHashIterate(pStmt->sql.pTableCache, NULL);
147✔
507
  while (pIter) {
162✔
508
    SStmtTableCache* pCache = (SStmtTableCache*)pIter;
15✔
509

510
    qDestroyStmtDataBlock(pCache->pDataCtx);
15✔
511
    qDestroyBoundColInfo(pCache->boundTags);
15✔
512
    taosMemoryFreeClear(pCache->boundTags);
15!
513

514
    pIter = taosHashIterate(pStmt->sql.pTableCache, pIter);
15✔
515
  }
516
  taosHashCleanup(pStmt->sql.pTableCache);
147✔
517
  pStmt->sql.pTableCache = NULL;
147✔
518

519
  STMT_ERR_RET(stmtCleanExecInfo(pStmt, false, true));
147!
520
  STMT_ERR_RET(stmtCleanBindInfo(pStmt));
147!
521

522
  taos_free_result(pStmt->sql.siInfo.pRequest);
147✔
523
  taosHashCleanup(pStmt->sql.siInfo.pVgroupHash);
147✔
524
  tSimpleHashCleanup(pStmt->sql.siInfo.pTableHash);
147✔
525
  taosArrayDestroyEx(pStmt->sql.siInfo.tbBuf.pBufList, stmtFreeTbBuf);
147✔
526
  taosMemoryFree(pStmt->sql.siInfo.pTSchema);
147!
527
  qDestroyStmtDataBlock(pStmt->sql.siInfo.pDataCtx);
147✔
528
  taosArrayDestroyEx(pStmt->sql.siInfo.pTableCols, stmtFreeTbCols);
147✔
529
  pStmt->sql.siInfo.pTableCols = NULL;
147✔
530

531
  (void)memset(&pStmt->sql, 0, sizeof(pStmt->sql));
147✔
532
  pStmt->sql.siInfo.tableColsReady = true;
147✔
533

534
  STMT2_TLOG_E("end to free SQL info");
147!
535

536
  return TSDB_CODE_SUCCESS;
147✔
537
}
538

539
static int32_t stmtTryAddTableVgroupInfo(STscStmt2* pStmt, int32_t* vgId) {
113✔
540
  if (*vgId >= 0 && taosHashGet(pStmt->sql.pVgHash, (const char*)vgId, sizeof(*vgId))) {
113✔
541
    return TSDB_CODE_SUCCESS;
15✔
542
  }
543

544
  SVgroupInfo      vgInfo = {0};
98✔
545
  SRequestConnInfo conn = {.pTrans = pStmt->taos->pAppInfo->pTransporter,
98✔
546
                           .requestId = pStmt->exec.pRequest->requestId,
98✔
547
                           .requestObjRefId = pStmt->exec.pRequest->self,
98✔
548
                           .mgmtEps = getEpSet_s(&pStmt->taos->pAppInfo->mgmtEp)};
98✔
549

550
  int32_t code = catalogGetTableHashVgroup(pStmt->pCatalog, &conn, &pStmt->bInfo.sname, &vgInfo);
98✔
551
  if (TSDB_CODE_SUCCESS != code) {
98!
552
    STMT2_ELOG("fail to get vgroup info from catalog, code:%d", code);
×
553
    return code;
×
554
  }
555

556
  code =
557
      taosHashPut(pStmt->sql.pVgHash, (const char*)&vgInfo.vgId, sizeof(vgInfo.vgId), (char*)&vgInfo, sizeof(vgInfo));
98✔
558
  if (TSDB_CODE_SUCCESS != code) {
98!
559
    STMT2_ELOG("fail to put vgroup info, code:%d", code);
×
560
    return code;
×
561
  }
562

563
  *vgId = vgInfo.vgId;
98✔
564

565
  return TSDB_CODE_SUCCESS;
98✔
566
}
567

568
static int32_t stmtRebuildDataBlock(STscStmt2* pStmt, STableDataCxt* pDataBlock, STableDataCxt** newBlock, uint64_t uid,
66✔
569
                                    uint64_t suid, int32_t vgId) {
570
  STMT_ERR_RET(stmtTryAddTableVgroupInfo(pStmt, &vgId));
66!
571
  STMT_ERR_RET(qRebuildStmtDataBlock(newBlock, pDataBlock, uid, suid, vgId, pStmt->sql.autoCreateTbl));
66!
572

573
  STMT2_DLOG("uid:%" PRId64 ", rebuild table data context, vgId:%d", uid, vgId);
66!
574

575
  return TSDB_CODE_SUCCESS;
66✔
576
}
577

578
static int32_t stmtGetFromCache(STscStmt2* pStmt) {
198✔
579
  if (pStmt->sql.stbInterlaceMode && pStmt->sql.siInfo.pDataCtx) {
198!
580
    pStmt->bInfo.needParse = false;
×
581
    pStmt->bInfo.inExecCache = false;
×
582
    return TSDB_CODE_SUCCESS;
×
583
  }
584

585
  pStmt->bInfo.needParse = true;
198✔
586
  pStmt->bInfo.inExecCache = false;
198✔
587

588
  STableDataCxt** pCxtInExec = taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
198✔
589
  if (pCxtInExec) {
197✔
590
    pStmt->bInfo.needParse = false;
24✔
591
    pStmt->bInfo.inExecCache = true;
24✔
592

593
    pStmt->exec.pCurrBlock = *pCxtInExec;
24✔
594

595
    if (pStmt->sql.autoCreateTbl) {
24✔
596
      STMT2_DLOG("reuse stmt block for tb %s in execBlock", pStmt->bInfo.tbFName);
18!
597
      return TSDB_CODE_SUCCESS;
18✔
598
    }
599
  }
600

601
  if (NULL == pStmt->pCatalog) {
179✔
602
    STMT_ERR_RET(catalogGetHandle(pStmt->taos->pAppInfo->clusterId, &pStmt->pCatalog));
91!
603
    pStmt->sql.siInfo.pCatalog = pStmt->pCatalog;
93✔
604
  }
605

606
  if (NULL == pStmt->sql.pTableCache || taosHashGetSize(pStmt->sql.pTableCache) <= 0) {
181!
607
    if (pStmt->bInfo.inExecCache) {
109!
608
      pStmt->bInfo.needParse = false;
×
609
      STMT2_DLOG("reuse stmt block for tb %s in execBlock", pStmt->bInfo.tbFName);
×
610
      return TSDB_CODE_SUCCESS;
×
611
    }
612

613
    STMT2_DLOG("no stmt block cache for tb %s", pStmt->bInfo.tbFName);
109!
614
    return TSDB_CODE_SUCCESS;
109✔
615
  }
616

617
  if (pStmt->sql.autoCreateTbl) {
72✔
618
    SStmtTableCache* pCache = taosHashGet(pStmt->sql.pTableCache, &pStmt->bInfo.tbSuid, sizeof(pStmt->bInfo.tbSuid));
48✔
619
    if (pCache) {
48!
620
      pStmt->bInfo.needParse = false;
48✔
621
      pStmt->bInfo.tbUid = 0;
48✔
622

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

626
      if (taosHashPut(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName), &pNewBlock,
48!
627
                      POINTER_BYTES)) {
628
        STMT_ERR_RET(terrno);
×
629
      }
630

631
      pStmt->exec.pCurrBlock = pNewBlock;
48✔
632

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

635
      return TSDB_CODE_SUCCESS;
48✔
636
    }
637

638
    STMT_RET(stmtCleanBindInfo(pStmt));
×
639
  }
640

641
  uint64_t uid, suid;
642
  int32_t  vgId;
643
  int8_t   tableType;
644

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

652
  pStmt->stat.ctgGetTbMetaNum++;
24✔
653

654
  if (TSDB_CODE_PAR_TABLE_NOT_EXIST == code) {
24!
655
    STMT2_DLOG("tb %s not exist", pStmt->bInfo.tbFName);
×
656
    STMT_ERR_RET(stmtCleanBindInfo(pStmt));
×
657

658
    STMT_ERR_RET(code);
×
659
  }
660

661
  STMT_ERR_RET(code);
24!
662

663
  uid = pTableMeta->uid;
24✔
664
  suid = pTableMeta->suid;
24✔
665
  tableType = pTableMeta->tableType;
24✔
666
  pStmt->bInfo.tbVgId = pTableMeta->vgId;
24✔
667
  vgId = pTableMeta->vgId;
24✔
668

669
  taosMemoryFree(pTableMeta);
24!
670

671
  uint64_t cacheUid = (TSDB_CHILD_TABLE == tableType) ? suid : uid;
24!
672

673
  if (uid == pStmt->bInfo.tbUid) {
24!
674
    pStmt->bInfo.needParse = false;
×
675

676
    STMT2_DLOG("tb %s is current table", pStmt->bInfo.tbFName);
×
677

678
    return TSDB_CODE_SUCCESS;
×
679
  }
680

681
  if (pStmt->bInfo.inExecCache) {
24✔
682
    SStmtTableCache* pCache = taosHashGet(pStmt->sql.pTableCache, &cacheUid, sizeof(cacheUid));
6✔
683
    if (NULL == pCache) {
6!
684
      STMT2_ELOG("table [%s, %" PRIx64 ", %" PRIx64 "] found in exec blockHash, but not in sql blockHash",
×
685
                 pStmt->bInfo.tbFName, uid, cacheUid);
686

687
      STMT_ERR_RET(TSDB_CODE_APP_ERROR);
×
688
    }
689

690
    pStmt->bInfo.needParse = false;
6✔
691

692
    pStmt->bInfo.tbUid = uid;
6✔
693
    pStmt->bInfo.tbSuid = suid;
6✔
694
    pStmt->bInfo.tbType = tableType;
6✔
695
    pStmt->bInfo.boundTags = pCache->boundTags;
6✔
696
    pStmt->bInfo.tagsCached = true;
6✔
697

698
    STMT2_DLOG("tb %s in execBlock list, set to current", pStmt->bInfo.tbFName);
6!
699

700
    return TSDB_CODE_SUCCESS;
6✔
701
  }
702

703
  SStmtTableCache* pCache = taosHashGet(pStmt->sql.pTableCache, &cacheUid, sizeof(cacheUid));
18✔
704
  if (pCache) {
18!
705
    pStmt->bInfo.needParse = false;
18✔
706

707
    pStmt->bInfo.tbUid = uid;
18✔
708
    pStmt->bInfo.tbSuid = suid;
18✔
709
    pStmt->bInfo.tbType = tableType;
18✔
710
    pStmt->bInfo.boundTags = pCache->boundTags;
18✔
711
    pStmt->bInfo.tagsCached = true;
18✔
712

713
    STableDataCxt* pNewBlock = NULL;
18✔
714
    STMT_ERR_RET(stmtRebuildDataBlock(pStmt, pCache->pDataCtx, &pNewBlock, uid, suid, vgId));
18!
715

716
    if (taosHashPut(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName), &pNewBlock,
18!
717
                    POINTER_BYTES)) {
718
      STMT_ERR_RET(terrno);
×
719
    }
720

721
    pStmt->exec.pCurrBlock = pNewBlock;
18✔
722

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

725
    return TSDB_CODE_SUCCESS;
18✔
726
  }
727

728
  STMT_ERR_RET(stmtCleanBindInfo(pStmt));
×
729

730
  return TSDB_CODE_SUCCESS;
×
731
}
732

733
static int32_t stmtResetStmt(STscStmt2* pStmt) {
×
734
  STMT_ERR_RET(stmtCleanSQLInfo(pStmt));
×
735

736
  pStmt->sql.pTableCache = taosHashInit(100, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
×
737
  if (NULL == pStmt->sql.pTableCache) {
×
738
    STMT2_ELOG("fail to allocate memory for pTableCache in stmtResetStmt:%s", tstrerror(terrno));
×
739
    STMT_ERR_RET(terrno);
×
740
  }
741

742
  pStmt->sql.status = STMT_INIT;
×
743

744
  return TSDB_CODE_SUCCESS;
×
745
}
746

747
static int32_t stmtAsyncOutput(STscStmt2* pStmt, void* param) {
10,185✔
748
  SStmtQNode* pParam = (SStmtQNode*)param;
10,185✔
749

750
  if (pParam->restoreTbCols) {
10,185✔
751
    for (int32_t i = 0; i < pStmt->sql.siInfo.pTableColsIdx; ++i) {
10,188✔
752
      SArray** p = (SArray**)TARRAY_GET_ELEM(pStmt->sql.siInfo.pTableCols, i);
5,764✔
753
      *p = taosArrayInit(20, POINTER_BYTES);
5,764✔
754
      if (*p == NULL) {
5,763!
755
        STMT_ERR_RET(terrno);
×
756
      }
757
    }
758
    atomic_store_8((int8_t*)&pStmt->sql.siInfo.tableColsReady, true);
4,424✔
759
    STMT2_TLOG_E("restore pTableCols finished");
4,427!
760
  } else {
761
    int code = qAppendStmtTableOutput(pStmt->sql.pQuery, pStmt->sql.pVgHash, &pParam->tblData, pStmt->exec.pCurrBlock,
5,758✔
762
                                      &pStmt->sql.siInfo, pParam->pCreateTbReq);
763
    // taosMemoryFree(pParam->pTbData);
764
    (void)atomic_sub_fetch_64(&pStmt->sql.siInfo.tbRemainNum, 1);
5,757✔
765
    if (code != TSDB_CODE_SUCCESS) {
5,764!
766
      STMT2_ELOG("async append stmt output failed, tbname:%s, err:%s", pParam->tblData.tbName, tstrerror(code));
×
767
      STMT_ERR_RET(code);
×
768
    }
769
  }
770
  return TSDB_CODE_SUCCESS;
10,191✔
771
}
772

773
static void* stmtBindThreadFunc(void* param) {
100✔
774
  setThreadName("stmt2Bind");
100✔
775

776
  STscStmt2* pStmt = (STscStmt2*)param;
100✔
777
  STMT2_ILOG_E("stmt2 bind thread started");
100!
778

779
  while (true) {
10,190✔
780
    SStmtQNode* asyncParam = NULL;
10,290✔
781

782
    if (!stmtDequeue(pStmt, &asyncParam)) {
10,290✔
783
      if (pStmt->queue.stopQueue && 0 == atomic_load_64((int64_t*)&pStmt->queue.qRemainNum)) {
100!
784
        STMT2_DLOG_E("queue is empty and stopQueue is set, thread will exit");
100!
785
        break;
100✔
786
      }
787
      continue;
×
788
    }
789

790
    int ret = stmtAsyncOutput(pStmt, asyncParam);
10,189✔
791
    if (ret != 0) {
10,191!
792
      STMT2_ELOG("stmtAsyncOutput failed, reason:%s", tstrerror(ret));
×
793
    }
794
  }
795

796
  STMT2_ILOG_E("stmt2 bind thread stopped");
100!
797
  return NULL;
100✔
798
}
799

800
static int32_t stmtStartBindThread(STscStmt2* pStmt) {
100✔
801
  TdThreadAttr thAttr;
802
  if (taosThreadAttrInit(&thAttr) != 0) {
100!
803
    return TSDB_CODE_TSC_INTERNAL_ERROR;
×
804
  }
805
  if (taosThreadAttrSetDetachState(&thAttr, PTHREAD_CREATE_JOINABLE) != 0) {
100!
806
    return TSDB_CODE_TSC_INTERNAL_ERROR;
×
807
  }
808

809
  if (taosThreadCreate(&pStmt->bindThread, &thAttr, stmtBindThreadFunc, pStmt) != 0) {
100!
810
    terrno = TAOS_SYSTEM_ERROR(ERRNO);
×
811
    STMT_ERR_RET(terrno);
×
812
  }
813

814
  pStmt->bindThreadInUse = true;
100✔
815

816
  (void)taosThreadAttrDestroy(&thAttr);
100✔
817
  return TSDB_CODE_SUCCESS;
100✔
818
}
819

820
static int32_t stmtInitQueue(STscStmt2* pStmt) {
99✔
821
  (void)taosThreadCondInit(&pStmt->queue.waitCond, NULL);
99✔
822
  (void)taosThreadMutexInit(&pStmt->queue.mutex, NULL);
100✔
823
  STMT_ERR_RET(stmtAllocQNodeFromBuf(&pStmt->sql.siInfo.tbBuf, (void**)&pStmt->queue.head));
200!
824
  pStmt->queue.tail = pStmt->queue.head;
100✔
825

826
  return TSDB_CODE_SUCCESS;
100✔
827
}
828

829
static int32_t stmtIniAsyncBind(STscStmt2* pStmt) {
149✔
830
  (void)taosThreadCondInit(&pStmt->asyncBindParam.waitCond, NULL);
149✔
831
  (void)taosThreadMutexInit(&pStmt->asyncBindParam.mutex, NULL);
149✔
832
  pStmt->asyncBindParam.asyncBindNum = 0;
149✔
833

834
  return TSDB_CODE_SUCCESS;
149✔
835
}
836

837
static int32_t stmtInitTableBuf(STableBufInfo* pTblBuf) {
100✔
838
  pTblBuf->buffUnit = sizeof(SStmtQNode);
100✔
839
  pTblBuf->buffSize = pTblBuf->buffUnit * 1000;
100✔
840
  pTblBuf->pBufList = taosArrayInit(100, POINTER_BYTES);
100✔
841
  if (NULL == pTblBuf->pBufList) {
100!
842
    return terrno;
×
843
  }
844
  void* buff = taosMemoryMalloc(pTblBuf->buffSize);
100!
845
  if (NULL == buff) {
99!
846
    return terrno;
×
847
  }
848

849
  if (taosArrayPush(pTblBuf->pBufList, &buff) == NULL) {
198!
850
    return terrno;
×
851
  }
852

853
  pTblBuf->pCurBuff = buff;
99✔
854
  pTblBuf->buffIdx = 1;
99✔
855
  pTblBuf->buffOffset = 0;
99✔
856

857
  return TSDB_CODE_SUCCESS;
99✔
858
}
859

860
TAOS_STMT2* stmtInit2(STscObj* taos, TAOS_STMT2_OPTION* pOptions) {
146✔
861
  STscObj*   pObj = (STscObj*)taos;
146✔
862
  STscStmt2* pStmt = NULL;
146✔
863
  int32_t    code = 0;
146✔
864

865
  pStmt = taosMemoryCalloc(1, sizeof(STscStmt2));
146!
866
  if (NULL == pStmt) {
149!
867
    STMT2_ELOG_E("fail to allocate memory for pStmt");
×
868
    return NULL;
×
869
  }
870

871
  pStmt->sql.pTableCache = taosHashInit(100, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
149✔
872
  if (NULL == pStmt->sql.pTableCache) {
149!
873
    STMT2_ELOG("fail to allocate memory for pTableCache in stmtInit2:%s", tstrerror(terrno));
×
874
    taosMemoryFree(pStmt);
×
875
    return NULL;
×
876
  }
877

878
  pStmt->taos = pObj;
149✔
879
  if (taos->db[0] != '\0') {
149✔
880
    pStmt->db = taosStrdup(taos->db);
51!
881
  }
882
  pStmt->bInfo.needParse = true;
149✔
883
  pStmt->sql.status = STMT_INIT;
149✔
884
  pStmt->errCode = TSDB_CODE_SUCCESS;
149✔
885

886
  if (NULL != pOptions) {
149!
887
    (void)memcpy(&pStmt->options, pOptions, sizeof(pStmt->options));
149✔
888
    if (pOptions->singleStbInsert && pOptions->singleTableBindOnce) {
149✔
889
      pStmt->stbInterlaceMode = true;
83✔
890
    }
891

892
    pStmt->reqid = pOptions->reqid;
149✔
893
  }
894

895
  if (pStmt->stbInterlaceMode) {
149✔
896
    pStmt->sql.siInfo.transport = taos->pAppInfo->pTransporter;
83✔
897
    pStmt->sql.siInfo.acctId = taos->acctId;
83✔
898
    pStmt->sql.siInfo.dbname = taos->db;
83✔
899
    pStmt->sql.siInfo.mgmtEpSet = getEpSet_s(&pStmt->taos->pAppInfo->mgmtEp);
83✔
900
    pStmt->sql.siInfo.pTableHash = tSimpleHashInit(100, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY));
83✔
901
    if (NULL == pStmt->sql.siInfo.pTableHash) {
83!
902
      STMT2_ELOG("fail to allocate memory for pTableHash:%s", tstrerror(terrno));
×
903
      (void)stmtClose2(pStmt);
×
904
      return NULL;
×
905
    }
906
    pStmt->sql.siInfo.pTableCols = taosArrayInit(STMT_TABLE_COLS_NUM, POINTER_BYTES);
83✔
907
    if (NULL == pStmt->sql.siInfo.pTableCols) {
83!
908
      STMT2_ELOG("fail to allocate memory for pTableCols:%s", tstrerror(terrno));
×
909
      (void)stmtClose2(pStmt);
×
910
      return NULL;
×
911
    }
912

913
    code = stmtInitTableBuf(&pStmt->sql.siInfo.tbBuf);
83✔
914
    if (TSDB_CODE_SUCCESS == code) {
82!
915
      code = stmtInitQueue(pStmt);
82✔
916
    }
917
    if (TSDB_CODE_SUCCESS == code) {
83!
918
      code = stmtStartBindThread(pStmt);
83✔
919
    }
920
    if (TSDB_CODE_SUCCESS != code) {
83!
921
      terrno = code;
×
922
      STMT2_ELOG("fail to init stmt2 bind thread:%s", tstrerror(code));
×
923
      (void)stmtClose2(pStmt);
×
924
      return NULL;
×
925
    }
926
  }
927

928
  pStmt->sql.siInfo.tableColsReady = true;
149✔
929
  if (pStmt->options.asyncExecFn) {
149✔
930
    if (tsem_init(&pStmt->asyncExecSem, 0, 1) != 0) {
5!
931
      terrno = TAOS_SYSTEM_ERROR(ERRNO);
×
932
      STMT2_ELOG("fail to init asyncExecSem:%s", tstrerror(terrno));
×
933
      (void)stmtClose2(pStmt);
×
934
      return NULL;
×
935
    }
936
  }
937
  code = stmtIniAsyncBind(pStmt);
149✔
938
  if (TSDB_CODE_SUCCESS != code) {
149!
939
    terrno = code;
×
940
    STMT2_ELOG("fail to start init asyncExecSem:%s", tstrerror(code));
×
941

942
    (void)stmtClose2(pStmt);
×
943
    return NULL;
×
944
  }
945

946
  pStmt->execSemWaited = false;
149✔
947

948
  // STMT_LOG_SEQ(STMT_INIT);
949

950
  STMT2_DLOG("stmt2 initialize finished, seqId:%d, db:%s, interlaceMode:%d, asyncExec:%d", pStmt->seqId, pStmt->db,
149!
951
             pStmt->stbInterlaceMode, pStmt->options.asyncExecFn != NULL);
952

953
  return pStmt;
149✔
954
}
955

956
static int stmtSetDbName2(TAOS_STMT2* stmt, const char* dbName) {
64✔
957
  STscStmt2* pStmt = (STscStmt2*)stmt;
64✔
958
  if (dbName == NULL || dbName[0] == '\0') {
64!
959
    STMT2_ELOG_E("dbname in sql is illegal");
×
960
    return TSDB_CODE_TSC_SQL_SYNTAX_ERROR;
×
961
  }
962

963
  STMT2_DLOG("dbname is specified in sql:%s", dbName);
64!
964
  if (pStmt->db == NULL || pStmt->db[0] == '\0') {
64!
965
    taosMemoryFreeClear(pStmt->db);
33!
966
    STMT2_DLOG("dbname:%s is by sql, not by taosconnect", dbName);
33!
967
    pStmt->db = taosStrdup(dbName);
33!
968
    (void)strdequote(pStmt->db);
33✔
969
  }
970
  STMT_ERR_RET(stmtCreateRequest(pStmt));
64!
971

972
  // The SQL statement specifies a database name, overriding the previously specified database
973
  taosMemoryFreeClear(pStmt->exec.pRequest->pDb);
64!
974
  pStmt->exec.pRequest->pDb = taosStrdup(dbName);
64!
975
  (void)strdequote(pStmt->exec.pRequest->pDb);
64✔
976
  if (pStmt->exec.pRequest->pDb == NULL) {
64!
977
    return terrno;
×
978
  }
979
  if (pStmt->sql.stbInterlaceMode) {
64✔
980
    pStmt->sql.siInfo.dbname = pStmt->exec.pRequest->pDb;
22✔
981
  }
982
  return TSDB_CODE_SUCCESS;
64✔
983
}
984
static int32_t stmtResetStbInterlaceCache(STscStmt2* pStmt) {
17✔
985
  int32_t code = TSDB_CODE_SUCCESS;
17✔
986

987
  if (pStmt->bindThreadInUse) {
17!
988
    while (0 == atomic_load_8((int8_t*)&pStmt->sql.siInfo.tableColsReady)) {
17!
989
      taosUsleep(1);
×
990
    }
991
    (void)taosThreadMutexLock(&pStmt->queue.mutex);
17✔
992
    pStmt->queue.stopQueue = true;
17✔
993
    (void)taosThreadCondSignal(&(pStmt->queue.waitCond));
17✔
994
    (void)taosThreadMutexUnlock(&pStmt->queue.mutex);
17✔
995

996
    (void)taosThreadJoin(pStmt->bindThread, NULL);
17✔
997
    pStmt->bindThreadInUse = false;
17✔
998
    pStmt->queue.head = NULL;
17✔
999
    pStmt->queue.tail = NULL;
17✔
1000
    pStmt->queue.qRemainNum = 0;
17✔
1001

1002
    (void)taosThreadCondDestroy(&pStmt->queue.waitCond);
17✔
1003
    (void)taosThreadMutexDestroy(&pStmt->queue.mutex);
17✔
1004
  }
1005

1006
  pStmt->sql.siInfo.pTableHash = tSimpleHashInit(100, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY));
17✔
1007
  if (NULL == pStmt->sql.siInfo.pTableHash) {
17!
1008
    return terrno;
×
1009
  }
1010

1011
  pStmt->sql.siInfo.pTableCols = taosArrayInit(STMT_TABLE_COLS_NUM, POINTER_BYTES);
17✔
1012
  if (NULL == pStmt->sql.siInfo.pTableCols) {
17!
1013
    return terrno;
×
1014
  }
1015

1016
  code = stmtInitTableBuf(&pStmt->sql.siInfo.tbBuf);
17✔
1017

1018
  if (TSDB_CODE_SUCCESS == code) {
17!
1019
    code = stmtInitQueue(pStmt);
17✔
1020
    pStmt->queue.stopQueue = false;
17✔
1021
  }
1022
  if (TSDB_CODE_SUCCESS == code) {
17!
1023
    code = stmtStartBindThread(pStmt);
17✔
1024
  }
1025
  if (TSDB_CODE_SUCCESS != code) {
17!
1026
    return code;
×
1027
  }
1028

1029
  return TSDB_CODE_SUCCESS;
17✔
1030
}
1031

1032
static int32_t stmtResetStmtForPrepare(STscStmt2* pStmt) {
21✔
1033
  char*             db = pStmt->db;
21✔
1034
  bool              stbInterlaceMode = pStmt->stbInterlaceMode;
21✔
1035
  TAOS_STMT2_OPTION options = pStmt->options;
21✔
1036
  uint32_t          reqid = pStmt->reqid;
21✔
1037

1038
  taosMemoryFree(pStmt->sql.pBindInfo);
21!
1039
  pStmt->sql.pBindInfo = NULL;
21✔
1040

1041
  taosMemoryFree(pStmt->sql.queryRes.fields);
21!
1042
  pStmt->sql.queryRes.fields = NULL;
21✔
1043

1044
  taosMemoryFree(pStmt->sql.queryRes.userFields);
21!
1045
  pStmt->sql.queryRes.userFields = NULL;
21✔
1046

1047
  pStmt->sql.type = 0;
21✔
1048
  pStmt->sql.runTimes = 0;
21✔
1049
  taosMemoryFree(pStmt->sql.sqlStr);
21!
1050
  pStmt->sql.sqlStr = NULL;
21✔
1051

1052
  qDestroyQuery(pStmt->sql.pQuery);
21✔
1053
  pStmt->sql.pQuery = NULL;
21✔
1054

1055
  taosArrayDestroy(pStmt->sql.nodeList);
21✔
1056
  pStmt->sql.nodeList = NULL;
21✔
1057

1058
  taosHashCleanup(pStmt->sql.pVgHash);
21✔
1059
  pStmt->sql.pVgHash = NULL;
21✔
1060

1061
  if (pStmt->sql.fixValueTags) {
21✔
1062
    tdDestroySVCreateTbReq(pStmt->sql.fixValueTbReq);
12!
1063
    pStmt->sql.fixValueTbReq = NULL;
12✔
1064
  }
1065
  pStmt->sql.fixValueTags = false;
21✔
1066

1067
  void* pIter = taosHashIterate(pStmt->sql.pTableCache, NULL);
21✔
1068
  while (pIter) {
24✔
1069
    SStmtTableCache* pCache = (SStmtTableCache*)pIter;
3✔
1070

1071
    qDestroyStmtDataBlock(pCache->pDataCtx);
3✔
1072
    qDestroyBoundColInfo(pCache->boundTags);
3✔
1073
    taosMemoryFreeClear(pCache->boundTags);
3!
1074

1075
    pIter = taosHashIterate(pStmt->sql.pTableCache, pIter);
3✔
1076
  }
1077
  taosHashCleanup(pStmt->sql.pTableCache);
21✔
1078

1079
  if (pStmt->sql.stbInterlaceMode) {
21✔
1080
    pStmt->bInfo.tagsCached = false;
16✔
1081
  }
1082
  STMT_ERR_RET(stmtCleanExecInfo(pStmt, false, true));
21!
1083

1084
  if (pStmt->exec.pRequest) {
21!
1085
    taos_free_result(pStmt->exec.pRequest);
×
1086
    pStmt->exec.pRequest = NULL;
×
1087
  }
1088

1089
  if (pStmt->sql.siInfo.pTableCols) {
21✔
1090
    taosArrayDestroyEx(pStmt->sql.siInfo.pTableCols, stmtFreeTbCols);
17✔
1091
    pStmt->sql.siInfo.pTableCols = NULL;
17✔
1092
  }
1093

1094
  if (pStmt->sql.siInfo.tbBuf.pBufList) {
21✔
1095
    taosArrayDestroyEx(pStmt->sql.siInfo.tbBuf.pBufList, stmtFreeTbBuf);
17✔
1096
    pStmt->sql.siInfo.tbBuf.pBufList = NULL;
17✔
1097
  }
1098

1099
  if (pStmt->sql.siInfo.pTableHash) {
21✔
1100
    tSimpleHashCleanup(pStmt->sql.siInfo.pTableHash);
17✔
1101
    pStmt->sql.siInfo.pTableHash = NULL;
17✔
1102
  }
1103

1104
  if (pStmt->sql.siInfo.pVgroupHash) {
21!
1105
    taosHashCleanup(pStmt->sql.siInfo.pVgroupHash);
×
1106
    pStmt->sql.siInfo.pVgroupHash = NULL;
×
1107
  }
1108

1109
  if (pStmt->sql.siInfo.pVgroupList) {
21!
1110
    taosArrayDestroy(pStmt->sql.siInfo.pVgroupList);
×
1111
    pStmt->sql.siInfo.pVgroupList = NULL;
×
1112
  }
1113

1114
  if (pStmt->sql.siInfo.pDataCtx) {
21✔
1115
    qDestroyStmtDataBlock(pStmt->sql.siInfo.pDataCtx);
15✔
1116
    pStmt->sql.siInfo.pDataCtx = NULL;
15✔
1117
  }
1118

1119
  if (pStmt->sql.siInfo.pTSchema) {
21✔
1120
    taosMemoryFree(pStmt->sql.siInfo.pTSchema);
15!
1121
    pStmt->sql.siInfo.pTSchema = NULL;
15✔
1122
  }
1123

1124
  if (pStmt->sql.siInfo.pRequest) {
21✔
1125
    taos_free_result(pStmt->sql.siInfo.pRequest);
15✔
1126
    pStmt->sql.siInfo.pRequest = NULL;
15✔
1127
  }
1128

1129
  if (stbInterlaceMode) {
21✔
1130
    STMT_ERR_RET(stmtResetStbInterlaceCache(pStmt));
17!
1131
  }
1132

1133
  pStmt->db = db;
21✔
1134
  pStmt->stbInterlaceMode = stbInterlaceMode;
21✔
1135
  pStmt->options = options;
21✔
1136
  pStmt->reqid = reqid;
21✔
1137

1138
  pStmt->sql.pTableCache = taosHashInit(100, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
21✔
1139
  if (NULL == pStmt->sql.pTableCache) {
21!
1140
    STMT2_ELOG("fail to allocate memory for pTableCache in stmtResetStmt:%s", tstrerror(terrno));
×
1141
    return terrno;
×
1142
  }
1143

1144
  pStmt->sql.status = STMT_INIT;
21✔
1145

1146
  return TSDB_CODE_SUCCESS;
21✔
1147
}
1148

1149
int stmtPrepare2(TAOS_STMT2* stmt, const char* sql, unsigned long length) {
158✔
1150
  STscStmt2* pStmt = (STscStmt2*)stmt;
158✔
1151
  int32_t    code = 0;
158✔
1152

1153
  STMT2_DLOG("start to prepare with sql:%s", sql);
158!
1154

1155
  if (stmt == NULL || sql == NULL) {
158!
1156
    STMT2_ELOG_E("stmt or sql is NULL");
×
1157
    return TSDB_CODE_INVALID_PARA;
×
1158
  }
1159

1160
  if (pStmt->sql.status >= STMT_PREPARE) {
168✔
1161
    STMT2_DLOG("stmt status is %d, need to reset stmt2 cache before prepare", pStmt->sql.status);
21!
1162
    STMT_ERR_RET(stmtResetStmtForPrepare(pStmt));
21!
1163
  }
1164

1165
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
168✔
1166
    STMT2_ELOG("errCode is not success before, ErrCode: 0x%x, errorsyt: %s\n. ", pStmt->errCode,
1!
1167
               tstrerror(pStmt->errCode));
1168
    return pStmt->errCode;
1✔
1169
  }
1170

1171
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_PREPARE));
167!
1172

1173
  if (length <= 0) {
167✔
1174
    length = strlen(sql);
103✔
1175
  }
1176

1177
  pStmt->sql.sqlStr = taosStrndup(sql, length);
167!
1178
  if (!pStmt->sql.sqlStr) {
167!
1179
    STMT2_ELOG("fail to allocate memory for sqlStr:%s", tstrerror(terrno));
×
1180
    return terrno;
×
1181
  }
1182
  pStmt->sql.sqlLen = length;
167✔
1183
  pStmt->sql.stbInterlaceMode = pStmt->stbInterlaceMode;
167✔
1184

1185
  char* dbName = NULL;
167✔
1186
  if (qParseDbName(sql, length, &dbName)) {
167✔
1187
    STMT_ERR_RET(stmtSetDbName2(stmt, dbName));
64!
1188
    taosMemoryFreeClear(dbName);
64!
1189
  }
1190

1191
  return TSDB_CODE_SUCCESS;
163✔
1192
}
1193

1194
static int32_t stmtInitStbInterlaceTableInfo(STscStmt2* pStmt) {
89✔
1195
  STableDataCxt** pSrc = taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
89✔
1196
  if (!pSrc) {
90!
1197
    return terrno;
×
1198
  }
1199
  STableDataCxt* pDst = NULL;
90✔
1200

1201
  STMT_ERR_RET(qCloneStmtDataBlock(&pDst, *pSrc, true));
90!
1202
  pStmt->sql.siInfo.pDataCtx = pDst;
85✔
1203

1204
  SArray* pTblCols = NULL;
85✔
1205
  for (int32_t i = 0; i < STMT_TABLE_COLS_NUM; i++) {
78,292✔
1206
    pTblCols = taosArrayInit(20, POINTER_BYTES);
78,195✔
1207
    if (NULL == pTblCols) {
80,712!
1208
      return terrno;
×
1209
    }
1210

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

1216
  pStmt->sql.siInfo.boundTags = pStmt->bInfo.boundTags;
97✔
1217

1218
  STMT2_TLOG("init stb interlace table info, tbName:%s, pDataCtx:%p, boundTags:%p", pStmt->bInfo.tbFName,
97!
1219
             pStmt->sql.siInfo.pDataCtx, pStmt->sql.siInfo.boundTags);
1220

1221
  return TSDB_CODE_SUCCESS;
90✔
1222
}
1223

1224
int stmtIsInsert2(TAOS_STMT2* stmt, int* insert) {
11,813✔
1225
  STscStmt2* pStmt = (STscStmt2*)stmt;
11,813✔
1226

1227
  // STMT_DLOG_E("start is insert");
1228

1229
  if (pStmt->sql.type) {
11,813✔
1230
    *insert = (STMT_TYPE_INSERT == pStmt->sql.type || STMT_TYPE_MULTI_INSERT == pStmt->sql.type);
11,648✔
1231
  } else {
1232
    *insert = qIsInsertValuesSql(pStmt->sql.sqlStr, pStmt->sql.sqlLen);
165✔
1233
  }
1234

1235
  return TSDB_CODE_SUCCESS;
11,814✔
1236
}
1237

1238
int stmtSetTbName2(TAOS_STMT2* stmt, const char* tbName) {
5,861✔
1239
  STscStmt2* pStmt = (STscStmt2*)stmt;
5,861✔
1240

1241
  int64_t startUs = taosGetTimestampUs();
5,862✔
1242

1243
  STMT2_TLOG("start to set tbName:%s", tbName);
5,862!
1244

1245
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
5,861!
1246
    return pStmt->errCode;
×
1247
  }
1248

1249
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_SETTBNAME));
5,861!
1250

1251
  int32_t insert = 0;
5,879✔
1252
  STMT_ERR_RET(stmtIsInsert2(stmt, &insert));
5,879!
1253
  if (0 == insert) {
5,877!
1254
    STMT2_ELOG_E("set tb name not available for none insert statement");
×
1255
    STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR);
×
1256
  }
1257

1258
  if (!pStmt->sql.stbInterlaceMode || NULL == pStmt->sql.siInfo.pDataCtx) {
5,877✔
1259
    STMT_ERR_RET(stmtCreateRequest(pStmt));
200!
1260

1261
    STMT_ERR_RET(qCreateSName(&pStmt->bInfo.sname, tbName, pStmt->taos->acctId, pStmt->exec.pRequest->pDb,
200!
1262
                              pStmt->exec.pRequest->msgBuf, pStmt->exec.pRequest->msgBufLen));
1263
    STMT_ERR_RET(tNameExtractFullName(&pStmt->bInfo.sname, pStmt->bInfo.tbFName));
198✔
1264

1265
    STMT_ERR_RET(stmtGetFromCache(pStmt));
199!
1266

1267
    if (pStmt->bInfo.needParse) {
199✔
1268
      tstrncpy(pStmt->bInfo.tbName, tbName, sizeof(pStmt->bInfo.tbName));
109✔
1269
      pStmt->bInfo.tbName[sizeof(pStmt->bInfo.tbName) - 1] = 0;
109✔
1270

1271
      STMT_ERR_RET(stmtParseSql(pStmt));
109!
1272
    }
1273
  } else {
1274
    tstrncpy(pStmt->bInfo.tbName, tbName, sizeof(pStmt->bInfo.tbName));
5,677✔
1275
    pStmt->bInfo.tbName[sizeof(pStmt->bInfo.tbName) - 1] = 0;
5,677✔
1276
    pStmt->exec.pRequest->requestId++;
5,677✔
1277
    pStmt->bInfo.needParse = false;
5,677✔
1278
  }
1279

1280
  if (pStmt->sql.stbInterlaceMode && NULL == pStmt->sql.siInfo.pDataCtx) {
5,875✔
1281
    STMT_ERR_RET(stmtInitStbInterlaceTableInfo(pStmt));
90!
1282
  }
1283

1284
  int64_t startUs2 = taosGetTimestampUs();
5,873✔
1285
  pStmt->stat.setTbNameUs += startUs2 - startUs;
5,873✔
1286

1287
  return TSDB_CODE_SUCCESS;
5,873✔
1288
}
1289

1290
int stmtSetTbTags2(TAOS_STMT2* stmt, TAOS_STMT2_BIND* tags, SVCreateTbReq** pCreateTbReq) {
120✔
1291
  STscStmt2* pStmt = (STscStmt2*)stmt;
120✔
1292

1293
  STMT2_TLOG_E("start to set tbTags");
120!
1294

1295
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
120!
1296
    return pStmt->errCode;
×
1297
  }
1298

1299
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_SETTAGS));
120!
1300

1301
  if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 &&
120!
1302
      STMT_TYPE_MULTI_INSERT != pStmt->sql.type) {
×
1303
    pStmt->bInfo.needParse = false;
×
1304
  }
1305
  STMT_ERR_RET(stmtCreateRequest(pStmt));
120!
1306

1307
  if (pStmt->bInfo.needParse) {
120!
1308
    STMT_ERR_RET(stmtParseSql(pStmt));
×
1309
  }
1310
  if (pStmt->sql.stbInterlaceMode && NULL == pStmt->sql.siInfo.pDataCtx) {
120!
1311
    STMT_ERR_RET(stmtInitStbInterlaceTableInfo(pStmt));
×
1312
  }
1313

1314
  SBoundColInfo* tags_info = (SBoundColInfo*)pStmt->bInfo.boundTags;
120✔
1315
  // if (tags_info->numOfBound <= 0 || tags_info->numOfCols <= 0) {
1316
  //   tscWarn("no tags or cols bound in sql, will not bound tags");
1317
  //   return TSDB_CODE_SUCCESS;
1318
  // }
1319
  if (pStmt->sql.autoCreateTbl && pStmt->sql.stbInterlaceMode) {
120!
1320
    STMT_ERR_RET(qCreateSName(&pStmt->bInfo.sname, pStmt->bInfo.tbName, pStmt->taos->acctId, pStmt->exec.pRequest->pDb,
38!
1321
                              pStmt->exec.pRequest->msgBuf, pStmt->exec.pRequest->msgBufLen));
1322
    STMT_ERR_RET(tNameExtractFullName(&pStmt->bInfo.sname, pStmt->bInfo.tbFName));
38!
1323
  }
1324

1325
  STableDataCxt** pDataBlock = NULL;
120✔
1326
  if (pStmt->exec.pCurrBlock) {
120✔
1327
    pDataBlock = &pStmt->exec.pCurrBlock;
95✔
1328
  } else {
1329
    pDataBlock =
1330
        (STableDataCxt**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
25✔
1331
    if (NULL == pDataBlock) {
25!
1332
      STMT2_ELOG("table %s not found in exec blockHash:%p", pStmt->bInfo.tbFName, pStmt->exec.pBlockHash);
×
1333
      STMT_ERR_RET(TSDB_CODE_TSC_STMT_CACHE_ERROR);
×
1334
    }
1335
  }
1336
  if (pStmt->bInfo.inExecCache && !pStmt->sql.autoCreateTbl) {
120!
1337
    return TSDB_CODE_SUCCESS;
×
1338
  }
1339

1340
  STMT2_TLOG_E("start to bind stmt tag values");
120!
1341

1342
  void* boundTags = NULL;
120✔
1343
  if (pStmt->sql.stbInterlaceMode) {
120✔
1344
    boundTags = pStmt->sql.siInfo.boundTags;
38✔
1345
    *pCreateTbReq = taosMemoryCalloc(1, sizeof(SVCreateTbReq));
38!
1346
    if (NULL == pCreateTbReq) {
38!
1347
      return terrno;
×
1348
    }
1349
    int32_t vgId = -1;
38✔
1350
    STMT_ERR_RET(stmtTryAddTableVgroupInfo(pStmt, &vgId));
38!
1351
    (*pCreateTbReq)->uid = vgId;
38✔
1352
  } else {
1353
    boundTags = pStmt->bInfo.boundTags;
82✔
1354
  }
1355

1356
  STMT_ERR_RET(qBindStmtTagsValue2(*pDataBlock, boundTags, pStmt->bInfo.tbSuid, pStmt->bInfo.stbFName,
120✔
1357
                                   pStmt->bInfo.sname.tname, tags, pStmt->exec.pRequest->msgBuf,
1358
                                   pStmt->exec.pRequest->msgBufLen, pStmt->taos->optionInfo.charsetCxt, *pCreateTbReq));
1359

1360
  return TSDB_CODE_SUCCESS;
119✔
1361
}
1362

1363
int stmtCheckTags2(TAOS_STMT2* stmt, SVCreateTbReq** pCreateTbReq) {
26✔
1364
  STscStmt2* pStmt = (STscStmt2*)stmt;
26✔
1365

1366
  STMT2_TLOG_E("start to clone createTbRequest for fixed tags");
26!
1367

1368
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
26!
1369
    return pStmt->errCode;
×
1370
  }
1371

1372
  if (!pStmt->sql.stbInterlaceMode) {
26!
1373
    return TSDB_CODE_SUCCESS;
×
1374
  }
1375

1376
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_SETTAGS));
26!
1377

1378
  if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 &&
26!
1379
      STMT_TYPE_MULTI_INSERT != pStmt->sql.type) {
×
1380
    pStmt->bInfo.needParse = false;
×
1381
  }
1382
  STMT_ERR_RET(stmtCreateRequest(pStmt));
26!
1383

1384
  if (pStmt->bInfo.needParse) {
26!
1385
    STMT_ERR_RET(stmtParseSql(pStmt));
×
1386
    if (!pStmt->sql.autoCreateTbl) {
×
1387
      STMT2_WLOG_E("don't need to create table, will not check tags");
×
1388
      return TSDB_CODE_SUCCESS;
×
1389
    }
1390
  }
1391

1392
  if (pStmt->sql.stbInterlaceMode && NULL == pStmt->sql.siInfo.pDataCtx) {
26!
1393
    STMT_ERR_RET(stmtInitStbInterlaceTableInfo(pStmt));
×
1394
  }
1395

1396
  STMT_ERR_RET(qCreateSName(&pStmt->bInfo.sname, pStmt->bInfo.tbName, pStmt->taos->acctId, pStmt->exec.pRequest->pDb,
26!
1397
                            pStmt->exec.pRequest->msgBuf, pStmt->exec.pRequest->msgBufLen));
1398
  STMT_ERR_RET(tNameExtractFullName(&pStmt->bInfo.sname, pStmt->bInfo.tbFName));
26!
1399

1400
  STableDataCxt** pDataBlock = NULL;
26✔
1401
  if (pStmt->exec.pCurrBlock) {
26✔
1402
    pDataBlock = &pStmt->exec.pCurrBlock;
9✔
1403
  } else {
1404
    pDataBlock =
1405
        (STableDataCxt**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
17✔
1406
    if (NULL == pDataBlock) {
17!
1407
      STMT2_ELOG("table %s not found in exec blockHash:%p", pStmt->bInfo.tbFName, pStmt->exec.pBlockHash);
×
1408
      STMT_ERR_RET(TSDB_CODE_TSC_STMT_CACHE_ERROR);
×
1409
    }
1410
  }
1411

1412
  if (!((*pDataBlock)->pData->flags & SUBMIT_REQ_AUTO_CREATE_TABLE)) {
26!
1413
    STMT2_DLOG_E("don't need to create, will not check tags");
×
1414
    return TSDB_CODE_SUCCESS;
×
1415
  }
1416

1417
  if (pStmt->sql.fixValueTags) {
26✔
1418
    STMT2_TLOG_E("tags are fixed, use one createTbReq");
9!
1419
    STMT_ERR_RET(cloneSVreateTbReq(pStmt->sql.fixValueTbReq, pCreateTbReq));
9!
1420
    if ((*pCreateTbReq)->name) {
9!
1421
      taosMemoryFree((*pCreateTbReq)->name);
9!
1422
    }
1423
    (*pCreateTbReq)->name = taosStrdup(pStmt->bInfo.tbName);
9!
1424
    int32_t vgId = -1;
9✔
1425
    STMT_ERR_RET(stmtTryAddTableVgroupInfo(pStmt, &vgId));
9!
1426
    (*pCreateTbReq)->uid = vgId;
9✔
1427
    return TSDB_CODE_SUCCESS;
9✔
1428
  }
1429

1430
  if ((*pDataBlock)->pData->pCreateTbReq) {
17!
1431
    STMT2_TLOG_E("tags are fixed, set createTbReq first time");
17!
1432
    pStmt->sql.fixValueTags = true;
17✔
1433
    STMT_ERR_RET(cloneSVreateTbReq((*pDataBlock)->pData->pCreateTbReq, &pStmt->sql.fixValueTbReq));
17!
1434
    STMT_ERR_RET(cloneSVreateTbReq(pStmt->sql.fixValueTbReq, pCreateTbReq));
17!
1435
    (*pCreateTbReq)->uid = (*pDataBlock)->pMeta->vgId;
17✔
1436
  }
1437

1438
  return TSDB_CODE_SUCCESS;
17✔
1439
}
1440

1441
static int stmtFetchColFields2(STscStmt2* pStmt, int32_t* fieldNum, TAOS_FIELD_E** fields) {
×
1442
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
×
1443
    return pStmt->errCode;
×
1444
  }
1445

1446
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
×
1447
    tscError("invalid operation to get query column fileds");
×
1448
    STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR);
×
1449
  }
1450

1451
  STableDataCxt** pDataBlock = NULL;
×
1452

1453
  if (pStmt->sql.stbInterlaceMode) {
×
1454
    pDataBlock = &pStmt->sql.siInfo.pDataCtx;
×
1455
  } else {
1456
    pDataBlock =
1457
        (STableDataCxt**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
×
1458
    if (NULL == pDataBlock) {
×
1459
      STMT2_ELOG("table %s not found in exec blockHash:%p", pStmt->bInfo.tbFName, pStmt->exec.pBlockHash);
×
1460
      STMT_ERR_RET(TSDB_CODE_APP_ERROR);
×
1461
    }
1462
  }
1463

1464
  STMT_ERR_RET(qBuildStmtColFields(*pDataBlock, fieldNum, fields));
×
1465

1466
  return TSDB_CODE_SUCCESS;
×
1467
}
1468

1469
static int stmtFetchStbColFields2(STscStmt2* pStmt, int32_t* fieldNum, TAOS_FIELD_ALL** fields) {
42✔
1470
  int32_t code = 0;
42✔
1471
  int32_t preCode = pStmt->errCode;
42✔
1472

1473
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
42!
1474
    return pStmt->errCode;
×
1475
  }
1476

1477
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
42!
1478
    STMT2_ELOG_E("stmtFetchStbColFields2 only for insert statement");
×
1479
    STMT_ERRI_JRET(TSDB_CODE_TSC_STMT_API_ERROR);
×
1480
  }
1481

1482
  STableDataCxt** pDataBlock = NULL;
42✔
1483
  bool            cleanStb = false;
42✔
1484

1485
  if (pStmt->sql.stbInterlaceMode && pStmt->sql.siInfo.pDataCtx != NULL) {
42✔
1486
    pDataBlock = &pStmt->sql.siInfo.pDataCtx;
7✔
1487
  } else {
1488
    cleanStb = true;
35✔
1489
    pDataBlock =
1490
        (STableDataCxt**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
35✔
1491
  }
1492

1493
  if (NULL == pDataBlock || NULL == *pDataBlock) {
42!
1494
    STMT2_ELOG("table %s not found in exec blockHash:%p", pStmt->bInfo.tbFName, pStmt->exec.pBlockHash);
×
1495
    STMT_ERRI_JRET(TSDB_CODE_APP_ERROR);
×
1496
  }
1497

1498
  STMT_ERRI_JRET(
42!
1499
      qBuildStmtStbColFields(*pDataBlock, pStmt->bInfo.boundTags, pStmt->bInfo.tbNameFlag, fieldNum, fields));
1500

1501
  if (pStmt->bInfo.tbType == TSDB_SUPER_TABLE && cleanStb) {
42!
1502
    taosMemoryFreeClear((*pDataBlock)->boundColsInfo.pColIndex);
28!
1503
    qDestroyStmtDataBlock(*pDataBlock);
28✔
1504
    *pDataBlock = NULL;
28✔
1505
    if (taosHashRemove(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName)) != 0) {
28!
1506
      STMT2_ELOG("fail to remove remove stb:%s exec blockHash", pStmt->bInfo.tbFName);
×
1507
      STMT_ERRI_JRET(TSDB_CODE_APP_ERROR);
×
1508
    }
1509
    pStmt->sql.autoCreateTbl = false;
28✔
1510
    pStmt->bInfo.tagsCached = false;
28✔
1511
    pStmt->bInfo.sname = (SName){0};
28✔
1512
    STMT_ERR_RET(stmtCleanBindInfo(pStmt));
28!
1513
  }
1514

1515
_return:
14✔
1516

1517
  pStmt->errCode = preCode;
42✔
1518

1519
  return code;
42✔
1520
}
1521
/*
1522
SArray* stmtGetFreeCol(STscStmt2* pStmt, int32_t* idx) {
1523
  while (true) {
1524
    if (pStmt->exec.smInfo.pColIdx >= STMT_COL_BUF_SIZE) {
1525
      pStmt->exec.smInfo.pColIdx = 0;
1526
    }
1527

1528
    if ((pStmt->exec.smInfo.pColIdx + 1) == atomic_load_32(&pStmt->exec.smInfo.pColFreeIdx)) {
1529
      taosUsleep(1);
1530
      continue;
1531
    }
1532

1533
    *idx = pStmt->exec.smInfo.pColIdx;
1534
    return pStmt->exec.smInfo.pCols[pStmt->exec.smInfo.pColIdx++];
1535
  }
1536
}
1537
*/
1538
static int32_t stmtAppendTablePostHandle(STscStmt2* pStmt, SStmtQNode* param) {
5,753✔
1539
  if (NULL == pStmt->sql.siInfo.pVgroupHash) {
5,753✔
1540
    pStmt->sql.siInfo.pVgroupHash =
4,419✔
1541
        taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_ENTRY_LOCK);
4,416✔
1542
  }
1543
  if (NULL == pStmt->sql.siInfo.pVgroupList) {
5,756✔
1544
    pStmt->sql.siInfo.pVgroupList = taosArrayInit(64, POINTER_BYTES);
4,421✔
1545
  }
1546

1547
  if (NULL == pStmt->sql.siInfo.pRequest) {
5,757✔
1548
    STMT_ERR_RET(buildRequest(pStmt->taos->id, pStmt->sql.sqlStr, pStmt->sql.sqlLen, NULL, false,
88!
1549
                              (SRequestObj**)&pStmt->sql.siInfo.pRequest, pStmt->reqid));
1550

1551
    if (pStmt->reqid != 0) {
88!
1552
      pStmt->reqid++;
×
1553
    }
1554
    pStmt->exec.pRequest->syncQuery = true;
88✔
1555

1556
    pStmt->sql.siInfo.requestId = ((SRequestObj*)pStmt->sql.siInfo.pRequest)->requestId;
88✔
1557
    pStmt->sql.siInfo.requestSelf = ((SRequestObj*)pStmt->sql.siInfo.pRequest)->self;
88✔
1558
  }
1559

1560
  if (!pStmt->sql.siInfo.tbFromHash && pStmt->sql.siInfo.firstName[0] &&
5,757✔
1561
      0 == strcmp(pStmt->sql.siInfo.firstName, pStmt->bInfo.tbName)) {
95✔
1562
    pStmt->sql.siInfo.tbFromHash = true;
35✔
1563
  }
1564

1565
  if (0 == pStmt->sql.siInfo.firstName[0]) {
5,757✔
1566
    tstrncpy(pStmt->sql.siInfo.firstName, pStmt->bInfo.tbName, TSDB_TABLE_NAME_LEN);
74✔
1567
  }
1568

1569
  param->tblData.getFromHash = pStmt->sql.siInfo.tbFromHash;
5,757✔
1570
  param->next = NULL;
5,757✔
1571

1572
  (void)atomic_add_fetch_64(&pStmt->sql.siInfo.tbRemainNum, 1);
5,757✔
1573

1574
  if (pStmt->queue.stopQueue) {
5,766!
1575
    STMT2_ELOG_E("bind thread already stopped, cannot enqueue");
×
1576
    return TSDB_CODE_TSC_STMT_API_ERROR;
×
1577
  }
1578
  stmtEnqueue(pStmt, param);
5,766✔
1579

1580
  return TSDB_CODE_SUCCESS;
5,762✔
1581
}
1582

1583
static FORCE_INLINE int32_t stmtGetTableColsFromCache(STscStmt2* pStmt, SArray** pTableCols) {
1584
  while (true) {
1585
    if (pStmt->sql.siInfo.pTableColsIdx < taosArrayGetSize(pStmt->sql.siInfo.pTableCols)) {
5,764!
1586
      *pTableCols = (SArray*)taosArrayGetP(pStmt->sql.siInfo.pTableCols, pStmt->sql.siInfo.pTableColsIdx++);
5,764✔
1587
      break;
5,762✔
1588
    } else {
1589
      SArray* pTblCols = NULL;
×
1590
      for (int32_t i = 0; i < 100; i++) {
×
1591
        pTblCols = taosArrayInit(20, POINTER_BYTES);
×
1592
        if (NULL == pTblCols) {
×
1593
          return terrno;
×
1594
        }
1595

1596
        if (taosArrayPush(pStmt->sql.siInfo.pTableCols, &pTblCols) == NULL) {
×
1597
          return terrno;
×
1598
        }
1599
      }
1600
    }
1601
  }
1602

1603
  return TSDB_CODE_SUCCESS;
5,762✔
1604
}
1605

1606
static int32_t stmtCacheBlock(STscStmt2* pStmt) {
116✔
1607
  if (pStmt->sql.type != STMT_TYPE_MULTI_INSERT) {
116✔
1608
    return TSDB_CODE_SUCCESS;
9✔
1609
  }
1610

1611
  uint64_t uid = pStmt->bInfo.tbUid;
107✔
1612
  uint64_t cacheUid = (TSDB_CHILD_TABLE == pStmt->bInfo.tbType) ? pStmt->bInfo.tbSuid : uid;
107!
1613

1614
  if (taosHashGet(pStmt->sql.pTableCache, &cacheUid, sizeof(cacheUid))) {
107✔
1615
    STMT2_TLOG("table %s already cached, no need to cache again", pStmt->bInfo.tbFName);
89!
1616
    return TSDB_CODE_SUCCESS;
89✔
1617
  }
1618

1619
  STableDataCxt** pSrc = taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
18✔
1620
  if (!pSrc) {
18!
1621
    STMT2_ELOG("table %s not found in exec blockHash:%p", pStmt->bInfo.tbFName, pStmt->exec.pBlockHash);
×
1622
    return terrno;
×
1623
  }
1624
  STableDataCxt* pDst = NULL;
18✔
1625

1626
  STMT_ERR_RET(qCloneStmtDataBlock(&pDst, *pSrc, true));
18!
1627

1628
  SStmtTableCache cache = {
18✔
1629
      .pDataCtx = pDst,
1630
      .boundTags = pStmt->bInfo.boundTags,
18✔
1631
  };
1632

1633
  if (taosHashPut(pStmt->sql.pTableCache, &cacheUid, sizeof(cacheUid), &cache, sizeof(cache))) {
18!
1634
    STMT2_ELOG("fail to put table cache:%s", tstrerror(terrno));
×
1635
    return terrno;
×
1636
  }
1637

1638
  if (pStmt->sql.autoCreateTbl) {
18✔
1639
    pStmt->bInfo.tagsCached = true;
15✔
1640
  } else {
1641
    pStmt->bInfo.boundTags = NULL;
3✔
1642
  }
1643

1644
  return TSDB_CODE_SUCCESS;
18✔
1645
}
1646

1647
static int stmtAddBatch2(TAOS_STMT2* stmt) {
4,538✔
1648
  STscStmt2* pStmt = (STscStmt2*)stmt;
4,538✔
1649

1650
  int64_t startUs = taosGetTimestampUs();
4,540✔
1651

1652
  // STMT2_TLOG_E("start to add batch");
1653

1654
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
4,540!
1655
    return pStmt->errCode;
×
1656
  }
1657

1658
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_ADD_BATCH));
4,540!
1659

1660
  if (pStmt->sql.stbInterlaceMode) {
4,542✔
1661
    int64_t startUs2 = taosGetTimestampUs();
4,426✔
1662
    pStmt->stat.addBatchUs += startUs2 - startUs;
4,426✔
1663

1664
    pStmt->sql.siInfo.tableColsReady = false;
4,426✔
1665

1666
    SStmtQNode* param = NULL;
4,426✔
1667
    STMT_ERR_RET(stmtAllocQNodeFromBuf(&pStmt->sql.siInfo.tbBuf, (void**)&param));
8,852!
1668
    param->restoreTbCols = true;
4,426✔
1669
    param->next = NULL;
4,426✔
1670

1671
    if (pStmt->sql.autoCreateTbl) {
4,426✔
1672
      pStmt->bInfo.tagsCached = true;
36✔
1673
    }
1674

1675
    if (pStmt->queue.stopQueue) {
4,426!
1676
      STMT2_ELOG_E("stmt bind thread is stopped,cannot enqueue bind request");
×
1677
      return TSDB_CODE_TSC_STMT_API_ERROR;
×
1678
    }
1679

1680
    stmtEnqueue(pStmt, param);
4,426✔
1681

1682
    return TSDB_CODE_SUCCESS;
4,427✔
1683
  }
1684

1685
  STMT_ERR_RET(stmtCacheBlock(pStmt));
116!
1686

1687
  return TSDB_CODE_SUCCESS;
116✔
1688
}
1689
/*
1690
static int32_t stmtBackupQueryFields(STscStmt2* pStmt) {
1691
  SStmtQueryResInfo* pRes = &pStmt->sql.queryRes;
1692
  pRes->numOfCols = pStmt->exec.pRequest->body.resInfo.numOfCols;
1693
  pRes->precision = pStmt->exec.pRequest->body.resInfo.precision;
1694

1695
  int32_t size = pRes->numOfCols * sizeof(TAOS_FIELD);
1696
  pRes->fields = taosMemoryMalloc(size);
1697
  pRes->userFields = taosMemoryMalloc(size);
1698
  if (NULL == pRes->fields || NULL == pRes->userFields) {
1699
    STMT_ERR_RET(terrno);
1700
  }
1701
  (void)memcpy(pRes->fields, pStmt->exec.pRequest->body.resInfo.fields, size);
1702
  (void)memcpy(pRes->userFields, pStmt->exec.pRequest->body.resInfo.userFields, size);
1703

1704
  return TSDB_CODE_SUCCESS;
1705
}
1706

1707
static int32_t stmtRestoreQueryFields(STscStmt2* pStmt) {
1708
  SStmtQueryResInfo* pRes = &pStmt->sql.queryRes;
1709
  int32_t            size = pRes->numOfCols * sizeof(TAOS_FIELD);
1710

1711
  pStmt->exec.pRequest->body.resInfo.numOfCols = pRes->numOfCols;
1712
  pStmt->exec.pRequest->body.resInfo.precision = pRes->precision;
1713

1714
  if (NULL == pStmt->exec.pRequest->body.resInfo.fields) {
1715
    pStmt->exec.pRequest->body.resInfo.fields = taosMemoryMalloc(size);
1716
    if (NULL == pStmt->exec.pRequest->body.resInfo.fields) {
1717
      STMT_ERR_RET(terrno);
1718
    }
1719
    (void)memcpy(pStmt->exec.pRequest->body.resInfo.fields, pRes->fields, size);
1720
  }
1721

1722
  if (NULL == pStmt->exec.pRequest->body.resInfo.userFields) {
1723
    pStmt->exec.pRequest->body.resInfo.userFields = taosMemoryMalloc(size);
1724
    if (NULL == pStmt->exec.pRequest->body.resInfo.userFields) {
1725
      STMT_ERR_RET(terrno);
1726
    }
1727
    (void)memcpy(pStmt->exec.pRequest->body.resInfo.userFields, pRes->userFields, size);
1728
  }
1729

1730
  return TSDB_CODE_SUCCESS;
1731
}
1732
*/
1733
int stmtBindBatch2(TAOS_STMT2* stmt, TAOS_STMT2_BIND* bind, int32_t colIdx, SVCreateTbReq* pCreateTbReq) {
5,876✔
1734
  STscStmt2* pStmt = (STscStmt2*)stmt;
5,876✔
1735
  int32_t    code = 0;
5,876✔
1736

1737
  int64_t startUs = taosGetTimestampUs();
5,876✔
1738

1739
  STMT2_TLOG("start to bind data, colIdx:%d", colIdx);
5,876!
1740

1741
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
5,874!
1742
    return pStmt->errCode;
×
1743
  }
1744

1745
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_BIND));
5,874!
1746

1747
  if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 &&
5,890!
1748
      STMT_TYPE_MULTI_INSERT != pStmt->sql.type) {
×
1749
    pStmt->bInfo.needParse = false;
×
1750
  }
1751

1752
  if (pStmt->exec.pRequest && STMT_TYPE_QUERY == pStmt->sql.type && pStmt->sql.runTimes) {
5,890✔
1753
    taos_free_result(pStmt->exec.pRequest);
1✔
1754
    pStmt->exec.pRequest = NULL;
1✔
1755
  }
1756

1757
  STMT_ERR_RET(stmtCreateRequest(pStmt));
5,890!
1758
  if (pStmt->bInfo.needParse) {
5,889✔
1759
    code = stmtParseSql(pStmt);
6✔
1760
    if (code != TSDB_CODE_SUCCESS) {
6!
1761
      goto cleanup_root;
×
1762
    }
1763
  }
1764

1765
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
5,889✔
1766
    code = qStmtBindParams2(pStmt->sql.pQuery, bind, colIdx, pStmt->taos->optionInfo.charsetCxt);
5✔
1767
    if (code != TSDB_CODE_SUCCESS) {
5!
1768
      goto cleanup_root;
×
1769
    }
1770
    SParseContext ctx = {.requestId = pStmt->exec.pRequest->requestId,
5✔
1771
                         .acctId = pStmt->taos->acctId,
5✔
1772
                         .db = pStmt->exec.pRequest->pDb,
5✔
1773
                         .topicQuery = false,
1774
                         .pSql = pStmt->sql.sqlStr,
5✔
1775
                         .sqlLen = pStmt->sql.sqlLen,
5✔
1776
                         .pMsg = pStmt->exec.pRequest->msgBuf,
5✔
1777
                         .msgLen = ERROR_MSG_BUF_DEFAULT_SIZE,
1778
                         .pTransporter = pStmt->taos->pAppInfo->pTransporter,
5✔
1779
                         .pStmtCb = NULL,
1780
                         .pUser = pStmt->taos->user};
5✔
1781
    ctx.mgmtEpSet = getEpSet_s(&pStmt->taos->pAppInfo->mgmtEp);
5✔
1782
    code = catalogGetHandle(pStmt->taos->pAppInfo->clusterId, &ctx.pCatalog);
5✔
1783
    if (code != TSDB_CODE_SUCCESS) {
5!
1784
      goto cleanup_root;
×
1785
    }
1786
    code = qStmtParseQuerySql(&ctx, pStmt->sql.pQuery);
5✔
1787
    if (code != TSDB_CODE_SUCCESS) {
5!
1788
      goto cleanup_root;
×
1789
    }
1790

1791
    if (pStmt->sql.pQuery->haveResultSet) {
5!
1792
      STMT_ERR_RET(setResSchemaInfo(&pStmt->exec.pRequest->body.resInfo, pStmt->sql.pQuery->pResSchema,
10!
1793
                                    pStmt->sql.pQuery->numOfResCols, pStmt->sql.pQuery->pResExtSchema, true));
1794
      taosMemoryFreeClear(pStmt->sql.pQuery->pResSchema);
5!
1795
      taosMemoryFreeClear(pStmt->sql.pQuery->pResExtSchema);
5!
1796
      setResPrecision(&pStmt->exec.pRequest->body.resInfo, pStmt->sql.pQuery->precision);
5✔
1797
    }
1798

1799
    TSWAP(pStmt->exec.pRequest->dbList, pStmt->sql.pQuery->pDbList);
5✔
1800
    TSWAP(pStmt->exec.pRequest->tableList, pStmt->sql.pQuery->pTableList);
5✔
1801
    TSWAP(pStmt->exec.pRequest->targetTableList, pStmt->sql.pQuery->pTargetTableList);
5✔
1802

1803
    // if (STMT_TYPE_QUERY == pStmt->sql.queryRes) {
1804
    //   STMT_ERR_RET(stmtRestoreQueryFields(pStmt));
1805
    // }
1806

1807
    // STMT_ERR_RET(stmtBackupQueryFields(pStmt));
1808

1809
    return TSDB_CODE_SUCCESS;
5✔
1810

1811
  cleanup_root:
×
1812
    STMT2_ELOG("parse query statment unexpected failed code:%d, need to clean node", code);
×
1813
    if (pStmt->sql.pQuery->pRoot) {
×
1814
      nodesDestroyNode(pStmt->sql.pQuery->pRoot);
×
1815
      pStmt->sql.pQuery->pRoot = NULL;
×
1816
    }
1817
    STMT_ERR_RET(code);
×
1818
  }
1819

1820
  if (pStmt->sql.stbInterlaceMode && NULL == pStmt->sql.siInfo.pDataCtx) {
5,884!
1821
    STMT_ERR_RET(stmtInitStbInterlaceTableInfo(pStmt));
×
1822
  }
1823

1824
  STableDataCxt** pDataBlock = NULL;
5,883✔
1825

1826
  if (pStmt->exec.pCurrBlock) {
5,883✔
1827
    pDataBlock = &pStmt->exec.pCurrBlock;
5,772✔
1828
  } else {
1829
    pDataBlock =
1830
        (STableDataCxt**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
111✔
1831
    if (NULL == pDataBlock) {
111!
1832
      STMT2_ELOG("table %s not found in exec blockHash:%p", pStmt->bInfo.tbFName, pStmt->exec.pBlockHash);
×
1833
      STMT_ERR_RET(TSDB_CODE_TSC_STMT_CACHE_ERROR);
×
1834
    }
1835
    pStmt->exec.pCurrBlock = *pDataBlock;
111✔
1836
    if (pStmt->sql.stbInterlaceMode) {
111✔
1837
      taosArrayDestroy(pStmt->exec.pCurrBlock->pData->aCol);
89✔
1838
      (*pDataBlock)->pData->aCol = NULL;
88✔
1839
    }
1840
    if (colIdx < -1) {
110✔
1841
      pStmt->sql.bindRowFormat = true;
1✔
1842
      taosArrayDestroy((*pDataBlock)->pData->aCol);
1✔
1843
      (*pDataBlock)->pData->aCol = taosArrayInit(20, POINTER_BYTES);
1✔
1844
    }
1845
  }
1846

1847
  int64_t startUs2 = taosGetTimestampUs();
5,879✔
1848
  pStmt->stat.bindDataUs1 += startUs2 - startUs;
5,879✔
1849

1850
  SStmtQNode* param = NULL;
5,879✔
1851
  if (pStmt->sql.stbInterlaceMode) {
5,879✔
1852
    STMT_ERR_RET(stmtAllocQNodeFromBuf(&pStmt->sql.siInfo.tbBuf, (void**)&param));
11,527!
1853
    STMT_ERR_RET(stmtGetTableColsFromCache(pStmt, &param->tblData.aCol));
11,526!
1854
    taosArrayClear(param->tblData.aCol);
5,762✔
1855

1856
    // param->tblData.aCol = taosArrayInit(20, POINTER_BYTES);
1857

1858
    param->restoreTbCols = false;
5,761✔
1859
    param->tblData.isOrdered = true;
5,761✔
1860
    param->tblData.isDuplicateTs = false;
5,761✔
1861
    tstrncpy(param->tblData.tbName, pStmt->bInfo.tbName, TSDB_TABLE_NAME_LEN);
5,761✔
1862

1863
    param->pCreateTbReq = pCreateTbReq;
5,761✔
1864
  }
1865

1866
  int64_t startUs3 = taosGetTimestampUs();
5,880✔
1867
  pStmt->stat.bindDataUs2 += startUs3 - startUs2;
5,880✔
1868

1869
  SArray* pCols = pStmt->sql.stbInterlaceMode ? param->tblData.aCol : (*pDataBlock)->pData->aCol;
5,880✔
1870

1871
  if (colIdx < 0) {
5,880✔
1872
    if (pStmt->sql.stbInterlaceMode) {
5,875✔
1873
      // (*pDataBlock)->pData->flags = 0;
1874
      (*pDataBlock)->pData->flags &= ~SUBMIT_REQ_COLUMN_DATA_FORMAT;
5,765✔
1875
      code = qBindStmtStbColsValue2(*pDataBlock, pCols, bind, pStmt->exec.pRequest->msgBuf,
5,765✔
1876
                                    pStmt->exec.pRequest->msgBufLen, &pStmt->sql.siInfo.pTSchema, pStmt->sql.pBindInfo,
5,765✔
1877
                                    pStmt->taos->optionInfo.charsetCxt);
5,765✔
1878
      param->tblData.isOrdered = (*pDataBlock)->ordered;
5,754✔
1879
      param->tblData.isDuplicateTs = (*pDataBlock)->duplicateTs;
5,754✔
1880
    } else {
1881
      if (colIdx == -1) {
111✔
1882
        if (pStmt->sql.bindRowFormat) {
109✔
1883
          STMT2_ELOG_E("can't mix bind row format and bind column format");
1!
1884
          STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR);
1!
1885
        }
1886
        code = qBindStmtColsValue2(*pDataBlock, pCols, bind, pStmt->exec.pRequest->msgBuf,
108✔
1887
                                   pStmt->exec.pRequest->msgBufLen, pStmt->taos->optionInfo.charsetCxt);
108✔
1888
      } else {
1889
        code = qBindStmt2RowValue(*pDataBlock, (*pDataBlock)->pData->aRowP, bind, pStmt->exec.pRequest->msgBuf,
2✔
1890
                                  pStmt->exec.pRequest->msgBufLen, &pStmt->sql.siInfo.pTSchema, pStmt->sql.pBindInfo,
2✔
1891
                                  pStmt->taos->optionInfo.charsetCxt);
2✔
1892
      }
1893
    }
1894

1895
    if (code) {
5,864✔
1896
      STMT2_ELOG("bind cols or rows failed, error:%s", tstrerror(code));
1!
1897
      STMT_ERR_RET(code);
1!
1898
    }
1899
  } else {
1900
    if (pStmt->sql.stbInterlaceMode) {
6!
1901
      STMT2_ELOG_E("bind single column not allowed in stb insert mode");
×
1902
      STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR);
×
1903
    }
1904

1905
    if (pStmt->sql.bindRowFormat) {
6!
1906
      STMT2_ELOG_E("can't mix bind row format and bind column format");
×
1907
      STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR);
×
1908
    }
1909

1910
    if (colIdx != (pStmt->bInfo.sBindLastIdx + 1) && colIdx != 0) {
6!
1911
      STMT2_ELOG_E("bind column index not in sequence");
×
1912
      STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR);
×
1913
    }
1914

1915
    pStmt->bInfo.sBindLastIdx = colIdx;
6✔
1916

1917
    if (0 == colIdx) {
6✔
1918
      pStmt->bInfo.sBindRowNum = bind->num;
3✔
1919
    }
1920

1921
    code = qBindStmtSingleColValue2(*pDataBlock, pCols, bind, pStmt->exec.pRequest->msgBuf,
6✔
1922
                                    pStmt->exec.pRequest->msgBufLen, colIdx, pStmt->bInfo.sBindRowNum,
6✔
1923
                                    pStmt->taos->optionInfo.charsetCxt);
6✔
1924
    if (code) {
6!
1925
      STMT2_ELOG("bind single col failed, error:%s", tstrerror(code));
×
1926
      STMT_ERR_RET(code);
×
1927
    }
1928
  }
1929

1930
  int64_t startUs4 = taosGetTimestampUs();
5,868✔
1931
  pStmt->stat.bindDataUs3 += startUs4 - startUs3;
5,868✔
1932

1933
  if (pStmt->sql.stbInterlaceMode) {
5,868✔
1934
    STMT_ERR_RET(stmtAppendTablePostHandle(pStmt, param));
5,759!
1935
  } else {
1936
    STMT_ERR_RET(stmtAddBatch2(pStmt));
116!
1937
  }
1938

1939
  pStmt->stat.bindDataUs4 += taosGetTimestampUs() - startUs4;
5,875✔
1940

1941
  return TSDB_CODE_SUCCESS;
5,875✔
1942
}
1943
/*
1944
int stmtUpdateTableUid(STscStmt2* pStmt, SSubmitRsp* pRsp) {
1945
  tscDebug("stmt start to update tbUid, blockNum:%d", pRsp->nBlocks);
1946

1947
  int32_t code = 0;
1948
  int32_t finalCode = 0;
1949
  size_t  keyLen = 0;
1950
  void*   pIter = taosHashIterate(pStmt->exec.pBlockHash, NULL);
1951
  while (pIter) {
1952
    STableDataCxt* pBlock = *(STableDataCxt**)pIter;
1953
    char*          key = taosHashGetKey(pIter, &keyLen);
1954

1955
    STableMeta* pMeta = qGetTableMetaInDataBlock(pBlock);
1956
    if (pMeta->uid) {
1957
      pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
1958
      continue;
1959
    }
1960

1961
    SSubmitBlkRsp* blkRsp = NULL;
1962
    int32_t        i = 0;
1963
    for (; i < pRsp->nBlocks; ++i) {
1964
      blkRsp = pRsp->pBlocks + i;
1965
      if (strlen(blkRsp->tblFName) != keyLen) {
1966
        continue;
1967
      }
1968

1969
      if (strncmp(blkRsp->tblFName, key, keyLen)) {
1970
        continue;
1971
      }
1972

1973
      break;
1974
    }
1975

1976
    if (i < pRsp->nBlocks) {
1977
      tscDebug("auto created table %s uid updated from %" PRIx64 " to %" PRIx64, blkRsp->tblFName, pMeta->uid,
1978
               blkRsp->uid);
1979

1980
      pMeta->uid = blkRsp->uid;
1981
      pStmt->bInfo.tbUid = blkRsp->uid;
1982
    } else {
1983
      tscDebug("table %s not found in submit rsp, will update from catalog", pStmt->bInfo.tbFName);
1984
      if (NULL == pStmt->pCatalog) {
1985
        code = catalogGetHandle(pStmt->taos->pAppInfo->clusterId, &pStmt->pCatalog);
1986
        if (code) {
1987
          pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
1988
          finalCode = code;
1989
          continue;
1990
        }
1991
      }
1992

1993
      code = stmtCreateRequest(pStmt);
1994
      if (code) {
1995
        pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
1996
        finalCode = code;
1997
        continue;
1998
      }
1999

2000
      STableMeta*      pTableMeta = NULL;
2001
      SRequestConnInfo conn = {.pTrans = pStmt->taos->pAppInfo->pTransporter,
2002
                               .requestId = pStmt->exec.pRequest->requestId,
2003
                               .requestObjRefId = pStmt->exec.pRequest->self,
2004
                               .mgmtEps = getEpSet_s(&pStmt->taos->pAppInfo->mgmtEp)};
2005
      code = catalogGetTableMeta(pStmt->pCatalog, &conn, &pStmt->bInfo.sname, &pTableMeta);
2006

2007
      pStmt->stat.ctgGetTbMetaNum++;
2008

2009
      taos_free_result(pStmt->exec.pRequest);
2010
      pStmt->exec.pRequest = NULL;
2011

2012
      if (code || NULL == pTableMeta) {
2013
        pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
2014
        finalCode = code;
2015
        taosMemoryFree(pTableMeta);
2016
        continue;
2017
      }
2018

2019
      pMeta->uid = pTableMeta->uid;
2020
      pStmt->bInfo.tbUid = pTableMeta->uid;
2021
      taosMemoryFree(pTableMeta);
2022
    }
2023

2024
    pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
2025
  }
2026

2027
  return finalCode;
2028
}
2029
*/
2030
/*
2031
int stmtStaticModeExec(TAOS_STMT* stmt) {
2032
  STscStmt2*   pStmt = (STscStmt2*)stmt;
2033
  int32_t     code = 0;
2034
  SSubmitRsp* pRsp = NULL;
2035
  if (pStmt->sql.staticMode) {
2036
    return TSDB_CODE_TSC_STMT_API_ERROR;
2037
  }
2038

2039
  STMT_DLOG_E("start to exec");
2040

2041
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_EXECUTE));
2042

2043
  STMT_ERR_RET(qBuildStmtOutputFromTbList(pStmt->sql.pQuery, pStmt->sql.pVgHash, pStmt->exec.pTbBlkList,
2044
pStmt->exec.pCurrBlock, pStmt->exec.tbBlkNum));
2045

2046
  launchQueryImpl(pStmt->exec.pRequest, pStmt->sql.pQuery, true, NULL);
2047

2048
  if (pStmt->exec.pRequest->code && NEED_CLIENT_HANDLE_ERROR(pStmt->exec.pRequest->code)) {
2049
    code = refreshMeta(pStmt->exec.pRequest->pTscObj, pStmt->exec.pRequest);
2050
    if (code) {
2051
      pStmt->exec.pRequest->code = code;
2052
    } else {
2053
      tFreeSSubmitRsp(pRsp);
2054
      STMT_ERR_RET(stmtResetStmt(pStmt));
2055
      STMT_ERR_RET(TSDB_CODE_NEED_RETRY);
2056
    }
2057
  }
2058

2059
  STMT_ERR_JRET(pStmt->exec.pRequest->code);
2060

2061
  pStmt->exec.affectedRows = taos_affected_rows(pStmt->exec.pRequest);
2062
  pStmt->affectedRows += pStmt->exec.affectedRows;
2063

2064
_return:
2065

2066
  stmtCleanExecInfo(pStmt, (code ? false : true), false);
2067

2068
  tFreeSSubmitRsp(pRsp);
2069

2070
  ++pStmt->sql.runTimes;
2071

2072
  STMT_RET(code);
2073
}
2074
*/
2075

2076
static int32_t createParseContext(const SRequestObj* pRequest, SParseContext** pCxt, SSqlCallbackWrapper* pWrapper) {
12✔
2077
  const STscObj* pTscObj = pRequest->pTscObj;
12✔
2078

2079
  *pCxt = taosMemoryCalloc(1, sizeof(SParseContext));
12!
2080
  if (*pCxt == NULL) {
12!
2081
    return terrno;
×
2082
  }
2083

2084
  **pCxt = (SParseContext){.requestId = pRequest->requestId,
12✔
2085
                           .requestRid = pRequest->self,
12✔
2086
                           .acctId = pTscObj->acctId,
12✔
2087
                           .db = pRequest->pDb,
12✔
2088
                           .topicQuery = false,
2089
                           .pSql = pRequest->sqlstr,
12✔
2090
                           .sqlLen = pRequest->sqlLen,
12✔
2091
                           .pMsg = pRequest->msgBuf,
12✔
2092
                           .msgLen = ERROR_MSG_BUF_DEFAULT_SIZE,
2093
                           .pTransporter = pTscObj->pAppInfo->pTransporter,
12✔
2094
                           .pStmtCb = NULL,
2095
                           .pUser = pTscObj->user,
12✔
2096
                           .pEffectiveUser = pRequest->effectiveUser,
12✔
2097
                           .isSuperUser = (0 == strcmp(pTscObj->user, TSDB_DEFAULT_USER)),
12✔
2098
                           .enableSysInfo = pTscObj->sysInfo,
12✔
2099
                           .async = true,
2100
                           .svrVer = pTscObj->sVer,
12✔
2101
                           .nodeOffline = (pTscObj->pAppInfo->onlineDnodes < pTscObj->pAppInfo->totalDnodes),
12✔
2102
                           .allocatorId = pRequest->allocatorRefId,
12✔
2103
                           .parseSqlFp = clientParseSql,
2104
                           .parseSqlParam = pWrapper};
2105
  int8_t biMode = atomic_load_8(&((STscObj*)pTscObj)->biMode);
12✔
2106
  (*pCxt)->biMode = biMode;
12✔
2107
  return TSDB_CODE_SUCCESS;
12✔
2108
}
2109

2110
static void asyncQueryCb(void* userdata, TAOS_RES* res, int code) {
12✔
2111
  STscStmt2*        pStmt = userdata;
12✔
2112
  __taos_async_fn_t fp = pStmt->options.asyncExecFn;
12✔
2113

2114
  pStmt->exec.affectedRows = taos_affected_rows(pStmt->exec.pRequest);
12✔
2115
  pStmt->affectedRows += pStmt->exec.affectedRows;
12✔
2116

2117
  fp(pStmt->options.userdata, res, code);
12✔
2118

2119
  while (0 == atomic_load_8((int8_t*)&pStmt->sql.siInfo.tableColsReady)) {
12!
2120
    taosUsleep(1);
×
2121
  }
2122
  (void)stmtCleanExecInfo(pStmt, (code ? false : true), false);
12✔
2123
  ++pStmt->sql.runTimes;
12✔
2124

2125
  if (tsem_post(&pStmt->asyncExecSem) != 0) {
12!
2126
    tscError("fail to post asyncExecSem");
×
2127
  }
2128
}
12✔
2129

2130
int stmtExec2(TAOS_STMT2* stmt, int* affected_rows) {
4,476✔
2131
  STscStmt2* pStmt = (STscStmt2*)stmt;
4,476✔
2132
  int32_t    code = 0;
4,476✔
2133
  int64_t    startUs = taosGetTimestampUs();
4,475✔
2134

2135
  STMT2_DLOG_E("start to exec");
4,475!
2136

2137
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
4,475!
2138
    return pStmt->errCode;
×
2139
  }
2140

2141
  TSC_ERR_RET(taosThreadMutexLock(&pStmt->asyncBindParam.mutex));
4,475!
2142
  while (atomic_load_8((int8_t*)&pStmt->asyncBindParam.asyncBindNum) > 0) {
4,480!
2143
    (void)taosThreadCondWait(&pStmt->asyncBindParam.waitCond, &pStmt->asyncBindParam.mutex);
×
2144
  }
2145
  TSC_ERR_RET(taosThreadMutexUnlock(&pStmt->asyncBindParam.mutex));
4,475!
2146

2147
  if (pStmt->sql.stbInterlaceMode) {
4,478✔
2148
    STMT_ERR_RET(stmtAddBatch2(pStmt));
4,426!
2149
  }
2150

2151
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_EXECUTE));
4,479✔
2152

2153
  if (STMT_TYPE_QUERY != pStmt->sql.type) {
4,480✔
2154
    if (pStmt->sql.stbInterlaceMode) {
4,475✔
2155
      int64_t startTs = taosGetTimestampUs();
4,428✔
2156
      // wait for stmt bind thread to finish
2157
      while (atomic_load_64(&pStmt->sql.siInfo.tbRemainNum)) {
8,001✔
2158
        taosUsleep(1);
3,579✔
2159
      }
2160

2161
      pStmt->stat.execWaitUs += taosGetTimestampUs() - startTs;
4,418✔
2162
      STMT_ERR_RET(qBuildStmtFinOutput(pStmt->sql.pQuery, pStmt->sql.pVgHash, pStmt->sql.siInfo.pVgroupList));
4,418!
2163
      taosHashCleanup(pStmt->sql.siInfo.pVgroupHash);
4,423✔
2164
      pStmt->sql.siInfo.pVgroupHash = NULL;
4,426✔
2165
      pStmt->sql.siInfo.pVgroupList = NULL;
4,426✔
2166
    } else {
2167
      tDestroySubmitTbData(pStmt->exec.pCurrTbData, TSDB_MSG_FLG_ENCODE);
47✔
2168
      taosMemoryFreeClear(pStmt->exec.pCurrTbData);
47!
2169

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

2172
      STMT_ERR_RET(qBuildStmtOutput(pStmt->sql.pQuery, pStmt->sql.pVgHash, pStmt->exec.pBlockHash));
47!
2173
    }
2174
  }
2175

2176
  SRequestObj*      pRequest = pStmt->exec.pRequest;
4,478✔
2177
  __taos_async_fn_t fp = pStmt->options.asyncExecFn;
4,478✔
2178

2179
  if (!fp) {
4,478✔
2180
    launchQueryImpl(pStmt->exec.pRequest, pStmt->sql.pQuery, true, NULL);
4,466✔
2181

2182
    if (pStmt->exec.pRequest->code && NEED_CLIENT_HANDLE_ERROR(pStmt->exec.pRequest->code)) {
4,463!
2183
      STMT2_ELOG_E("exec failed errorcode:NEED_CLIENT_HANDLE_ERROR, need to refresh meta and retry");
×
2184
      code = refreshMeta(pStmt->exec.pRequest->pTscObj, pStmt->exec.pRequest);
×
2185
      if (code) {
×
2186
        pStmt->exec.pRequest->code = code;
×
2187

2188
      } else {
2189
        STMT_ERR_RET(stmtResetStmt(pStmt));
×
2190
        STMT_ERR_RET(TSDB_CODE_NEED_RETRY);
×
2191
      }
2192
    }
2193

2194
    STMT_ERR_JRET(pStmt->exec.pRequest->code);
4,463!
2195

2196
    pStmt->exec.affectedRows = taos_affected_rows(pStmt->exec.pRequest);
4,463✔
2197
    if (affected_rows) {
4,468✔
2198
      *affected_rows = pStmt->exec.affectedRows;
4,462✔
2199
    }
2200
    pStmt->affectedRows += pStmt->exec.affectedRows;
4,468✔
2201

2202
    // wait for stmt bind thread to finish
2203
    while (0 == atomic_load_8((int8_t*)&pStmt->sql.siInfo.tableColsReady)) {
4,468!
2204
      taosUsleep(1);
×
2205
    }
2206

2207
    STMT_ERR_RET(stmtCleanExecInfo(pStmt, (code ? false : true), false));
4,467✔
2208

2209
    ++pStmt->sql.runTimes;
4,453✔
2210
  } else {
2211
    SSqlCallbackWrapper* pWrapper = taosMemoryCalloc(1, sizeof(SSqlCallbackWrapper));
12!
2212
    if (pWrapper == NULL) {
12!
2213
      code = terrno;
×
2214
    } else {
2215
      pWrapper->pRequest = pRequest;
12✔
2216
      pRequest->pWrapper = pWrapper;
12✔
2217
    }
2218
    if (TSDB_CODE_SUCCESS == code) {
12!
2219
      code = createParseContext(pRequest, &pWrapper->pParseCtx, pWrapper);
12✔
2220
    }
2221
    pRequest->syncQuery = false;
12✔
2222
    pRequest->body.queryFp = asyncQueryCb;
12✔
2223
    ((SSyncQueryParam*)(pRequest)->body.interParam)->userParam = pStmt;
12✔
2224

2225
    pStmt->execSemWaited = false;
12✔
2226
    launchAsyncQuery(pRequest, pStmt->sql.pQuery, NULL, pWrapper);
12✔
2227
  }
2228

2229
_return:
4,465✔
2230
  if (code) {
4,465!
2231
    STMT2_ELOG("exec failed, error:%s", tstrerror(code));
×
2232
  }
2233
  pStmt->stat.execUseUs += taosGetTimestampUs() - startUs;
4,464✔
2234

2235
  STMT_RET(code);
4,464!
2236
}
2237

2238
int stmtClose2(TAOS_STMT2* stmt) {
147✔
2239
  STscStmt2* pStmt = (STscStmt2*)stmt;
147✔
2240

2241
  STMT2_DLOG_E("start to close stmt");
147!
2242
  taosMemoryFreeClear(pStmt->db);
147!
2243

2244
  if (pStmt->bindThreadInUse) {
147✔
2245
    // wait for stmt bind thread to finish
2246
    while (0 == atomic_load_8((int8_t*)&pStmt->sql.siInfo.tableColsReady)) {
83!
2247
      taosUsleep(1);
×
2248
    }
2249

2250
    (void)taosThreadMutexLock(&pStmt->queue.mutex);
83✔
2251
    pStmt->queue.stopQueue = true;
83✔
2252
    (void)taosThreadCondSignal(&(pStmt->queue.waitCond));
83✔
2253
    (void)taosThreadMutexUnlock(&pStmt->queue.mutex);
83✔
2254

2255
    (void)taosThreadJoin(pStmt->bindThread, NULL);
83✔
2256
    pStmt->bindThreadInUse = false;
83✔
2257

2258
    (void)taosThreadCondDestroy(&pStmt->queue.waitCond);
83✔
2259
    (void)taosThreadMutexDestroy(&pStmt->queue.mutex);
83✔
2260
  }
2261

2262
  TSC_ERR_RET(taosThreadMutexLock(&pStmt->asyncBindParam.mutex));
147!
2263
  while (atomic_load_8((int8_t*)&pStmt->asyncBindParam.asyncBindNum) > 0) {
147!
2264
    (void)taosThreadCondWait(&pStmt->asyncBindParam.waitCond, &pStmt->asyncBindParam.mutex);
×
2265
  }
2266
  TSC_ERR_RET(taosThreadMutexUnlock(&pStmt->asyncBindParam.mutex));
147!
2267

2268
  (void)taosThreadCondDestroy(&pStmt->asyncBindParam.waitCond);
147✔
2269
  (void)taosThreadMutexDestroy(&pStmt->asyncBindParam.mutex);
147✔
2270

2271
  if (pStmt->options.asyncExecFn && !pStmt->execSemWaited) {
147!
2272
    STMT2_TLOG_E("wait for asyncExecSem");
5!
2273
    if (tsem_wait(&pStmt->asyncExecSem) != 0) {
5!
2274
      STMT2_ELOG_E("fail to wait asyncExecSem");
×
2275
    }
2276
  }
2277

2278
  STMT2_DLOG("stmt %p closed, stbInterlaceMode:%d, statInfo: ctgGetTbMetaNum=>%" PRId64 ", getCacheTbInfo=>%" PRId64
147!
2279
             ", parseSqlNum=>%" PRId64 ", pStmt->stat.bindDataNum=>%" PRId64
2280
             ", settbnameAPI:%u, bindAPI:%u, addbatchAPI:%u, execAPI:%u"
2281
             ", setTbNameUs:%" PRId64 ", bindDataUs:%" PRId64 ",%" PRId64 ",%" PRId64 ",%" PRId64 " addBatchUs:%" PRId64
2282
             ", execWaitUs:%" PRId64 ", execUseUs:%" PRId64,
2283
             pStmt, pStmt->sql.stbInterlaceMode, pStmt->stat.ctgGetTbMetaNum, pStmt->stat.getCacheTbInfo,
2284
             pStmt->stat.parseSqlNum, pStmt->stat.bindDataNum, pStmt->seqIds[STMT_SETTBNAME], pStmt->seqIds[STMT_BIND],
2285
             pStmt->seqIds[STMT_ADD_BATCH], pStmt->seqIds[STMT_EXECUTE], pStmt->stat.setTbNameUs,
2286
             pStmt->stat.bindDataUs1, pStmt->stat.bindDataUs2, pStmt->stat.bindDataUs3, pStmt->stat.bindDataUs4,
2287
             pStmt->stat.addBatchUs, pStmt->stat.execWaitUs, pStmt->stat.execUseUs);
2288
  if (pStmt->sql.stbInterlaceMode) {
147✔
2289
    pStmt->bInfo.tagsCached = false;
77✔
2290
  }
2291

2292
  STMT_ERR_RET(stmtCleanSQLInfo(pStmt));
147!
2293

2294
  if (pStmt->options.asyncExecFn) {
147✔
2295
    if (tsem_destroy(&pStmt->asyncExecSem) != 0) {
5!
2296
      STMT2_ELOG_E("fail to destroy asyncExecSem");
×
2297
    }
2298
  }
2299
  taosMemoryFree(stmt);
147!
2300

2301
  return TSDB_CODE_SUCCESS;
147✔
2302
}
2303

2304
const char* stmtErrstr2(TAOS_STMT2* stmt) {
3✔
2305
  STscStmt2* pStmt = (STscStmt2*)stmt;
3✔
2306

2307
  if (stmt == NULL || NULL == pStmt->exec.pRequest) {
3!
2308
    return (char*)tstrerror(terrno);
3✔
2309
  }
2310

2311
  pStmt->exec.pRequest->code = terrno;
×
2312

2313
  SRequestObj* pRequest = pStmt->exec.pRequest;
×
2314
  if (NULL != pRequest->msgBuf && (strlen(pRequest->msgBuf) > 0 || pRequest->code == TSDB_CODE_RPC_FQDN_ERROR)) {
×
2315
    return pRequest->msgBuf;
×
2316
  }
2317
  return (const char*)tstrerror(pRequest->code);
×
2318
}
2319
/*
2320
int stmtAffectedRows(TAOS_STMT* stmt) { return ((STscStmt2*)stmt)->affectedRows; }
2321

2322
int stmtAffectedRowsOnce(TAOS_STMT* stmt) { return ((STscStmt2*)stmt)->exec.affectedRows; }
2323
*/
2324

2325
int stmtParseColFields2(TAOS_STMT2* stmt) {
57✔
2326
  int32_t    code = 0;
57✔
2327
  STscStmt2* pStmt = (STscStmt2*)stmt;
57✔
2328
  int32_t    preCode = pStmt->errCode;
57✔
2329

2330
  STMT2_DLOG_E("start to get col fields for insert");
57!
2331

2332
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
57!
2333
    return pStmt->errCode;
×
2334
  }
2335

2336
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
57!
2337
    STMT2_ELOG_E("stmtParseColFields2 only for insert");
×
2338
    STMT_ERRI_JRET(TSDB_CODE_TSC_STMT_API_ERROR);
×
2339
  }
2340

2341
  STMT_ERRI_JRET(stmtSwitchStatus(pStmt, STMT_FETCH_FIELDS));
57!
2342

2343
  if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 &&
57!
2344
      STMT_TYPE_MULTI_INSERT != pStmt->sql.type) {
3!
2345
    pStmt->bInfo.needParse = false;
×
2346
  }
2347
  if (pStmt->sql.stbInterlaceMode && pStmt->sql.siInfo.pDataCtx != NULL) {
57✔
2348
    pStmt->bInfo.needParse = false;
7✔
2349
  }
2350

2351
  STMT_ERRI_JRET(stmtCreateRequest(pStmt));
57!
2352

2353
  if (pStmt->bInfo.needParse) {
57✔
2354
    STMT_ERRI_JRET(stmtParseSql(pStmt));
50✔
2355
  }
2356

2357
_return:
42✔
2358
  // compatible with previous versions
2359
  if (code == TSDB_CODE_PAR_TABLE_NOT_EXIST && (pStmt->bInfo.tbNameFlag & NO_DATA_USING_CLAUSE) == 0x0) {
57!
2360
    code = TSDB_CODE_TSC_STMT_TBNAME_ERROR;
1✔
2361
  }
2362

2363
  if (code != TSDB_CODE_SUCCESS) {
57✔
2364
    STMT2_ELOG("stmt get fileds parse failed, code:%d", code);
15!
2365
    taos_free_result(pStmt->exec.pRequest);
15✔
2366
    pStmt->exec.pRequest = NULL;
15✔
2367
  }
2368

2369
  pStmt->errCode = preCode;
57✔
2370

2371
  return code;
57✔
2372
}
2373

2374
int stmtGetStbColFields2(TAOS_STMT2* stmt, int* nums, TAOS_FIELD_ALL** fields) {
57✔
2375
  int32_t code = stmtParseColFields2(stmt);
57✔
2376
  if (code != TSDB_CODE_SUCCESS) {
57✔
2377
    return code;
15✔
2378
  }
2379

2380
  return stmtFetchStbColFields2(stmt, nums, fields);
42✔
2381
}
2382

2383
int stmtGetParamNum2(TAOS_STMT2* stmt, int* nums) {
11✔
2384
  int32_t    code = 0;
11✔
2385
  STscStmt2* pStmt = (STscStmt2*)stmt;
11✔
2386
  int32_t    preCode = pStmt->errCode;
11✔
2387

2388
  STMT2_DLOG_E("start to get param num for query");
11!
2389

2390
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
11!
2391
    return pStmt->errCode;
×
2392
  }
2393

2394
  STMT_ERRI_JRET(stmtSwitchStatus(pStmt, STMT_FETCH_FIELDS));
11!
2395

2396
  if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 &&
11!
2397
      STMT_TYPE_MULTI_INSERT != pStmt->sql.type) {
×
2398
    pStmt->bInfo.needParse = false;
×
2399
  }
2400

2401
  if (pStmt->exec.pRequest && STMT_TYPE_QUERY == pStmt->sql.type && pStmt->sql.runTimes) {
11!
2402
    taos_free_result(pStmt->exec.pRequest);
×
2403
    pStmt->exec.pRequest = NULL;
×
2404
  }
2405

2406
  STMT_ERRI_JRET(stmtCreateRequest(pStmt));
11!
2407

2408
  if (pStmt->bInfo.needParse) {
11!
2409
    STMT_ERRI_JRET(stmtParseSql(pStmt));
11✔
2410
  }
2411

2412
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
4!
2413
    *nums = taosArrayGetSize(pStmt->sql.pQuery->pPlaceholderValues);
4✔
2414
  } else {
2415
    STMT_ERRI_JRET(stmtFetchColFields2(stmt, nums, NULL));
×
2416
  }
2417

2418
  STMT2_TLOG("get param num success, nums:%d", *nums);
4!
2419

2420
_return:
4✔
2421
  if (code != TSDB_CODE_SUCCESS) {
11✔
2422
    taos_free_result(pStmt->exec.pRequest);
7✔
2423
    pStmt->exec.pRequest = NULL;
7✔
2424
  }
2425
  pStmt->errCode = preCode;
11✔
2426

2427
  return code;
11✔
2428
}
2429

2430
TAOS_RES* stmtUseResult2(TAOS_STMT2* stmt) {
5✔
2431
  STscStmt2* pStmt = (STscStmt2*)stmt;
5✔
2432

2433
  STMT2_TLOG_E("start to use result");
5!
2434

2435
  if (STMT_TYPE_QUERY != pStmt->sql.type) {
5!
2436
    STMT2_ELOG_E("useResult only for query statement");
×
2437
    return NULL;
×
2438
  }
2439

2440
  return pStmt->exec.pRequest;
5✔
2441
}
2442

2443
int32_t stmtAsyncBindThreadFunc(void* args) {
×
2444
  qInfo("async stmt bind thread started");
×
2445

2446
  ThreadArgs* targs = (ThreadArgs*)args;
×
2447
  STscStmt2*  pStmt = (STscStmt2*)targs->stmt;
×
2448

2449
  int code = taos_stmt2_bind_param(targs->stmt, targs->bindv, targs->col_idx);
×
2450
  targs->fp(targs->param, NULL, code);
×
2451
  (void)taosThreadMutexLock(&(pStmt->asyncBindParam.mutex));
×
2452
  (void)atomic_sub_fetch_8(&pStmt->asyncBindParam.asyncBindNum, 1);
×
2453
  (void)taosThreadCondSignal(&(pStmt->asyncBindParam.waitCond));
×
2454
  (void)taosThreadMutexUnlock(&(pStmt->asyncBindParam.mutex));
×
2455
  taosMemoryFree(args);
×
2456

2457
  qInfo("async stmt bind thread stopped");
×
2458

2459
  return code;
×
2460
}
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