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

taosdata / TDengine / #4167

26 May 2025 05:50AM UTC coverage: 62.885% (-0.1%) from 63.028%
#4167

push

travis-ci

web-flow
test; basic cases (#31203)

157018 of 318118 branches covered (49.36%)

Branch coverage included in aggregate %.

242478 of 317165 relevant lines covered (76.45%)

18056696.68 hits per line

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

65.83
/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,272✔
13
    *pBuf = (char*)pTblBuf->pCurBuff + pTblBuf->buffOffset;
10,273✔
14
    pTblBuf->buffOffset += pTblBuf->buffUnit;
10,273✔
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,273✔
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,320✔
44
    if (pStmt->queue.stopQueue) {
47,042✔
45
      STMT2_DLOG_E("stmt stopQueue,but remainNum is 0");
100!
46
      return false;
100✔
47
    }
48
    if (i < 10) {
46,942✔
49
      taosUsleep(1);
44,216✔
50
      i++;
44,275✔
51
    } else {
52
      (void)taosThreadMutexLock(&pStmt->queue.mutex);
2,726✔
53
      if (pStmt->queue.stopQueue) {
2,756!
54
        (void)taosThreadMutexUnlock(&pStmt->queue.mutex);
×
55
        STMT2_DLOG_E("stmt stopQueue,but remainNum is 0");
×
56
        return false;
×
57
      }
58
      if (0 == atomic_load_64((int64_t*)&pStmt->queue.qRemainNum)) {
2,756✔
59
        (void)taosThreadCondWait(&pStmt->queue.waitCond, &pStmt->queue.mutex);
2,737✔
60
      }
61
      (void)taosThreadMutexUnlock(&pStmt->queue.mutex);
2,756✔
62
    }
63
  }
64

65
  if (pStmt->queue.stopQueue && 0 == atomic_load_64((int64_t*)&pStmt->queue.qRemainNum)) {
10,079!
66
    STMT2_DLOG_E("stmt stopQueue and queue is empty");
×
67
    return false;
×
68
  }
69

70
  (void)taosThreadMutexLock(&pStmt->queue.mutex);
10,079✔
71
  if (pStmt->queue.head == pStmt->queue.tail) {
10,188!
72
    (void)taosThreadMutexUnlock(&pStmt->queue.mutex);
×
73
    STMT2_ELOG_E("stmt bind queue is empty");
×
74
    return false;
×
75
  }
76

77
  SStmtQNode* node = pStmt->queue.head->next;
10,188✔
78
  pStmt->queue.head->next = node->next;
10,188✔
79
  if (pStmt->queue.tail == node) {
10,188✔
80
    pStmt->queue.tail = pStmt->queue.head;
5,733✔
81
  }
82
  node->next = NULL;
10,188✔
83
  *param = node;
10,188✔
84

85
  (void)atomic_sub_fetch_64((int64_t*)&pStmt->queue.qRemainNum, 1);
10,188✔
86
  (void)taosThreadMutexUnlock(&pStmt->queue.mutex);
10,193✔
87

88
  return true;
10,187✔
89
}
90

91
static void stmtEnqueue(STscStmt2* pStmt, SStmtQNode* param) {
10,180✔
92
  if (param == NULL) {
10,180!
93
    return;
×
94
  }
95

96
  param->next = NULL;
10,180✔
97

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

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

104
  (void)atomic_add_fetch_64(&pStmt->queue.qRemainNum, 1);
10,186✔
105
  (void)taosThreadCondSignal(&(pStmt->queue.waitCond));
10,191✔
106

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

110
static int32_t stmtCreateRequest(STscStmt2* pStmt) {
6,520✔
111
  int32_t code = 0;
6,520✔
112

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

130
  return code;
6,518✔
131
}
132

133
static int32_t stmtSwitchStatus(STscStmt2* pStmt, STMT_STATUS newStatus) {
21,120✔
134
  int32_t code = 0;
21,120✔
135

136
  if (newStatus >= STMT_INIT && newStatus < STMT_MAX) {
21,120!
137
    STMT_LOG_SEQ(newStatus);
21,126!
138
  }
139

140
  if (pStmt->errCode && newStatus != STMT_PREPARE) {
21,146!
141
    STMT_DLOG("stmt already failed with err:%s", tstrerror(pStmt->errCode));
×
142
    return pStmt->errCode;
×
143
  }
144

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

204
  STMT_ERR_RET(code);
21,146✔
205

206
  pStmt->sql.status = newStatus;
21,145✔
207

208
  return TSDB_CODE_SUCCESS;
21,145✔
209
}
210

211
static int32_t stmtGetTbName(TAOS_STMT2* stmt, char** tbName) {
141✔
212
  STscStmt2* pStmt = (STscStmt2*)stmt;
141✔
213

214
  pStmt->sql.type = STMT_TYPE_MULTI_INSERT;
141✔
215

216
  if ('\0' == pStmt->bInfo.tbName[0]) {
141✔
217
    tscWarn("no table name set, OK if it is a stmt get fields");
33!
218
    STMT_ERR_RET(TSDB_CODE_TSC_STMT_TBNAME_ERROR);
33!
219
  }
220

221
  *tbName = pStmt->bInfo.tbName;
108✔
222

223
  return TSDB_CODE_SUCCESS;
108✔
224
}
225

226
static int32_t stmtUpdateBindInfo(TAOS_STMT2* stmt, STableMeta* pTableMeta, void* tags, SName* tbName,
146✔
227
                                  const char* sTableName, bool autoCreateTbl, int8_t tbNameFlag) {
228
  STscStmt2* pStmt = (STscStmt2*)stmt;
146✔
229
  char       tbFName[TSDB_TABLE_FNAME_LEN];
230
  int32_t    code = tNameExtractFullName(tbName, tbFName);
146✔
231
  if (code != 0) {
147!
232
    return code;
×
233
  }
234

235
  (void)memcpy(&pStmt->bInfo.sname, tbName, sizeof(*tbName));
147✔
236
  tstrncpy(pStmt->bInfo.tbFName, tbFName, sizeof(pStmt->bInfo.tbFName));
147✔
237
  pStmt->bInfo.tbFName[sizeof(pStmt->bInfo.tbFName) - 1] = 0;
147✔
238

239
  pStmt->bInfo.tbUid = autoCreateTbl ? 0 : pTableMeta->uid;
147✔
240
  pStmt->bInfo.tbSuid = pTableMeta->suid;
147✔
241
  pStmt->bInfo.tbVgId = pTableMeta->vgId;
147✔
242
  pStmt->bInfo.tbType = pTableMeta->tableType;
147✔
243

244
  if (!pStmt->bInfo.tagsCached) {
147✔
245
    qDestroyBoundColInfo(pStmt->bInfo.boundTags);
144✔
246
    taosMemoryFreeClear(pStmt->bInfo.boundTags);
141!
247
  }
248

249
  pStmt->bInfo.boundTags = tags;
144✔
250
  pStmt->bInfo.tagsCached = false;
144✔
251
  pStmt->bInfo.tbNameFlag = tbNameFlag;
144✔
252
  tstrncpy(pStmt->bInfo.stbFName, sTableName, sizeof(pStmt->bInfo.stbFName));
144✔
253

254
  return TSDB_CODE_SUCCESS;
144✔
255
}
256

257
static int32_t stmtUpdateExecInfo(TAOS_STMT2* stmt, SHashObj* pVgHash, SHashObj* pBlockHash) {
146✔
258
  STscStmt2* pStmt = (STscStmt2*)stmt;
146✔
259

260
  pStmt->sql.pVgHash = pVgHash;
146✔
261
  pStmt->exec.pBlockHash = pBlockHash;
146✔
262

263
  return TSDB_CODE_SUCCESS;
146✔
264
}
265

266
static int32_t stmtUpdateInfo(TAOS_STMT2* stmt, STableMeta* pTableMeta, void* tags, SName* tbName, bool autoCreateTbl,
146✔
267
                              SHashObj* pVgHash, SHashObj* pBlockHash, const char* sTableName, uint8_t tbNameFlag) {
268
  STscStmt2* pStmt = (STscStmt2*)stmt;
146✔
269

270
  STMT_ERR_RET(stmtUpdateBindInfo(stmt, pTableMeta, tags, tbName, sTableName, autoCreateTbl, tbNameFlag));
146!
271
  STMT_ERR_RET(stmtUpdateExecInfo(stmt, pVgHash, pBlockHash));
145!
272

273
  pStmt->sql.autoCreateTbl = autoCreateTbl;
143✔
274

275
  return TSDB_CODE_SUCCESS;
143✔
276
}
277

278
static int32_t stmtGetExecInfo(TAOS_STMT2* stmt, SHashObj** pVgHash, SHashObj** pBlockHash) {
11✔
279
  STscStmt2* pStmt = (STscStmt2*)stmt;
11✔
280

281
  *pVgHash = pStmt->sql.pVgHash;
11✔
282
  pStmt->sql.pVgHash = NULL;
11✔
283

284
  *pBlockHash = pStmt->exec.pBlockHash;
11✔
285
  pStmt->exec.pBlockHash = NULL;
11✔
286

287
  return TSDB_CODE_SUCCESS;
11✔
288
}
289

290
static int32_t stmtParseSql(STscStmt2* pStmt) {
174✔
291
  pStmt->exec.pCurrBlock = NULL;
174✔
292

293
  SStmtCallback stmtCb = {
174✔
294
      .pStmt = pStmt,
295
      .getTbNameFn = stmtGetTbName,
296
      .setInfoFn = stmtUpdateInfo,
297
      .getExecInfoFn = stmtGetExecInfo,
298
  };
299

300
  STMT_ERR_RET(stmtCreateRequest(pStmt));
174!
301

302
  pStmt->stat.parseSqlNum++;
175✔
303
  STMT_ERR_RET(parseSql(pStmt->exec.pRequest, false, &pStmt->sql.pQuery, &stmtCb));
175✔
304

305
  pStmt->sql.siInfo.pQuery = pStmt->sql.pQuery;
153✔
306

307
  pStmt->bInfo.needParse = false;
153✔
308

309
  if (pStmt->sql.pQuery->pRoot && 0 == pStmt->sql.type) {
153✔
310
    pStmt->sql.type = STMT_TYPE_INSERT;
11✔
311
    pStmt->sql.stbInterlaceMode = false;
11✔
312
  } else if (pStmt->sql.pQuery->pPrepareRoot) {
142✔
313
    pStmt->sql.type = STMT_TYPE_QUERY;
6✔
314
    pStmt->sql.stbInterlaceMode = false;
6✔
315

316
    return TSDB_CODE_SUCCESS;
6✔
317
  }
318

319
  STableDataCxt** pSrc =
320
      (STableDataCxt**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
147✔
321
  if (NULL == pSrc || NULL == *pSrc) {
148!
322
    return terrno;
×
323
  }
324

325
  STableDataCxt* pTableCtx = *pSrc;
148✔
326
  if (pStmt->sql.stbInterlaceMode && pTableCtx->pData->pCreateTbReq && (pStmt->bInfo.tbNameFlag & USING_CLAUSE) == 0) {
148✔
327
    tdDestroySVCreateTbReq(pTableCtx->pData->pCreateTbReq);
5!
328
    taosMemoryFreeClear(pTableCtx->pData->pCreateTbReq);
5!
329
    pTableCtx->pData->pCreateTbReq = NULL;
5✔
330
  }
331
  // if (pStmt->sql.stbInterlaceMode) {
332
  //   int16_t lastIdx = -1;
333

334
  //   for (int32_t i = 0; i < pTableCtx->boundColsInfo.numOfBound; ++i) {
335
  //     if (pTableCtx->boundColsInfo.pColIndex[i] < lastIdx) {
336
  //       pStmt->sql.stbInterlaceMode = false;
337
  //       break;
338
  //     }
339

340
  //     lastIdx = pTableCtx->boundColsInfo.pColIndex[i];
341
  //   }
342
  // }
343

344
  if (NULL == pStmt->sql.pBindInfo) {
148✔
345
    pStmt->sql.pBindInfo = taosMemoryMalloc(pTableCtx->boundColsInfo.numOfBound * sizeof(*pStmt->sql.pBindInfo));
136!
346
    if (NULL == pStmt->sql.pBindInfo) {
136!
347
      return terrno;
×
348
    }
349
  }
350

351
  return TSDB_CODE_SUCCESS;
148✔
352
}
353

354
static int32_t stmtCleanBindInfo(STscStmt2* pStmt) {
4,758✔
355
  pStmt->bInfo.tbUid = 0;
4,758✔
356
  pStmt->bInfo.tbVgId = -1;
4,758✔
357
  pStmt->bInfo.tbType = 0;
4,758✔
358
  pStmt->bInfo.needParse = true;
4,758✔
359
  pStmt->bInfo.inExecCache = false;
4,758✔
360

361
  pStmt->bInfo.tbName[0] = 0;
4,758✔
362
  pStmt->bInfo.tbFName[0] = 0;
4,758✔
363
  if (!pStmt->bInfo.tagsCached) {
4,758✔
364
    qDestroyBoundColInfo(pStmt->bInfo.boundTags);
4,680✔
365
    taosMemoryFreeClear(pStmt->bInfo.boundTags);
4,685!
366
  }
367
  if (!pStmt->sql.autoCreateTbl) {
4,755✔
368
    pStmt->bInfo.stbFName[0] = 0;
4,650✔
369
    pStmt->bInfo.tbSuid = 0;
4,650✔
370
  }
371

372
  return TSDB_CODE_SUCCESS;
4,755✔
373
}
374

375
static void stmtFreeTableBlkList(STableColsData* pTb) {
×
376
  (void)qResetStmtColumns(pTb->aCol, true);
×
377
  taosArrayDestroy(pTb->aCol);
×
378
}
×
379

380
static void stmtResetQueueTableBuf(STableBufInfo* pTblBuf, SStmtQueue* pQueue) {
4,421✔
381
  pTblBuf->pCurBuff = taosArrayGetP(pTblBuf->pBufList, 0);
4,421✔
382
  if (NULL == pTblBuf->pCurBuff) {
4,422✔
383
    tscError("QInfo:%p, failed to get buffer from list", pTblBuf);
6!
384
    return;
×
385
  }
386
  pTblBuf->buffIdx = 1;
4,416✔
387
  pTblBuf->buffOffset = sizeof(*pQueue->head);
4,416✔
388

389
  pQueue->head = pQueue->tail = pTblBuf->pCurBuff;
4,416✔
390
  pQueue->qRemainNum = 0;
4,416✔
391
  pQueue->head->next = NULL;
4,416✔
392
}
393

394
static int32_t stmtCleanExecInfo(STscStmt2* pStmt, bool keepTable, bool deepClean) {
4,642✔
395
  if (pStmt->sql.stbInterlaceMode) {
4,642✔
396
    if (deepClean) {
4,517✔
397
      taosHashCleanup(pStmt->exec.pBlockHash);
93✔
398
      pStmt->exec.pBlockHash = NULL;
93✔
399

400
      if (NULL != pStmt->exec.pCurrBlock) {
93✔
401
        taosMemoryFreeClear(pStmt->exec.pCurrBlock->boundColsInfo.pColIndex);
89!
402
        taosMemoryFreeClear(pStmt->exec.pCurrBlock->pData);
89!
403
        qDestroyStmtDataBlock(pStmt->exec.pCurrBlock);
89✔
404
        pStmt->exec.pCurrBlock = NULL;
89✔
405
      }
406
      if (STMT_TYPE_QUERY != pStmt->sql.type) {
93!
407
        taos_free_result(pStmt->exec.pRequest);
93✔
408
        pStmt->exec.pRequest = NULL;
93✔
409
      }
410
    } else {
411
      pStmt->sql.siInfo.pTableColsIdx = 0;
4,424✔
412
      stmtResetQueueTableBuf(&pStmt->sql.siInfo.tbBuf, &pStmt->queue);
4,424✔
413
    }
414
    if (NULL != pStmt->exec.pRequest) {
4,510✔
415
      pStmt->exec.pRequest->body.resInfo.numOfRows = 0;
4,417✔
416
    }
417
  } else {
418
    if (STMT_TYPE_QUERY != pStmt->sql.type || deepClean) {
127✔
419
      // if (!pStmt->options.asyncExecFn) {
420
      taos_free_result(pStmt->exec.pRequest);
122✔
421
      pStmt->exec.pRequest = NULL;
122✔
422
      //}
423
    }
424

425
    size_t keyLen = 0;
127✔
426
    void*  pIter = taosHashIterate(pStmt->exec.pBlockHash, NULL);
127✔
427
    while (pIter) {
269✔
428
      STableDataCxt* pBlocks = *(STableDataCxt**)pIter;
142✔
429
      char*          key = taosHashGetKey(pIter, &keyLen);
142✔
430
      STableMeta*    pMeta = qGetTableMetaInDataBlock(pBlocks);
142✔
431

432
      if (keepTable && pBlocks == pStmt->exec.pCurrBlock) {
142✔
433
        TSWAP(pBlocks->pData, pStmt->exec.pCurrTbData);
47✔
434
        STMT_ERR_RET(qResetStmtDataBlock(pBlocks, false));
99!
435

436
        pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
47✔
437
        continue;
47✔
438
      }
439

440
      qDestroyStmtDataBlock(pBlocks);
95✔
441
      STMT_ERR_RET(taosHashRemove(pStmt->exec.pBlockHash, key, keyLen));
95!
442

443
      pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
95✔
444
    }
445

446
    if (keepTable) {
127✔
447
      return TSDB_CODE_SUCCESS;
52✔
448
    }
449

450
    taosHashCleanup(pStmt->exec.pBlockHash);
75✔
451
    pStmt->exec.pBlockHash = NULL;
75✔
452

453
    tDestroySubmitTbData(pStmt->exec.pCurrTbData, TSDB_MSG_FLG_ENCODE);
75✔
454
    taosMemoryFreeClear(pStmt->exec.pCurrTbData);
75!
455
  }
456

457
  STMT_ERR_RET(stmtCleanBindInfo(pStmt));
4,585✔
458

459
  return TSDB_CODE_SUCCESS;
4,578✔
460
}
461

462
static void stmtFreeTbBuf(void* buf) {
100✔
463
  void* pBuf = *(void**)buf;
100✔
464
  taosMemoryFree(pBuf);
100!
465
}
100✔
466

467
static void stmtFreeTbCols(void* buf) {
90,000✔
468
  SArray* pCols = *(SArray**)buf;
90,000✔
469
  taosArrayDestroy(pCols);
90,000✔
470
}
90,000✔
471

472
static int32_t stmtCleanSQLInfo(STscStmt2* pStmt) {
147✔
473
  STMT2_DLOG_E("start to free SQL info");
147!
474

475
  taosMemoryFree(pStmt->sql.pBindInfo);
147!
476
  taosMemoryFree(pStmt->sql.queryRes.fields);
147!
477
  taosMemoryFree(pStmt->sql.queryRes.userFields);
147!
478
  taosMemoryFree(pStmt->sql.sqlStr);
147!
479
  qDestroyQuery(pStmt->sql.pQuery);
147✔
480
  taosArrayDestroy(pStmt->sql.nodeList);
147✔
481
  taosHashCleanup(pStmt->sql.pVgHash);
147✔
482
  pStmt->sql.pVgHash = NULL;
147✔
483
  if (pStmt->sql.fixValueTags) {
147✔
484
    tdDestroySVCreateTbReq(pStmt->sql.fixValueTbReq);
5!
485
  }
486

487
  void* pIter = taosHashIterate(pStmt->sql.pTableCache, NULL);
147✔
488
  while (pIter) {
162✔
489
    SStmtTableCache* pCache = (SStmtTableCache*)pIter;
15✔
490

491
    qDestroyStmtDataBlock(pCache->pDataCtx);
15✔
492
    qDestroyBoundColInfo(pCache->boundTags);
15✔
493
    taosMemoryFreeClear(pCache->boundTags);
15!
494

495
    pIter = taosHashIterate(pStmt->sql.pTableCache, pIter);
15✔
496
  }
497
  taosHashCleanup(pStmt->sql.pTableCache);
147✔
498
  pStmt->sql.pTableCache = NULL;
147✔
499

500
  STMT_ERR_RET(stmtCleanExecInfo(pStmt, false, true));
147!
501
  STMT_ERR_RET(stmtCleanBindInfo(pStmt));
147!
502

503
  taos_free_result(pStmt->sql.siInfo.pRequest);
147✔
504
  taosHashCleanup(pStmt->sql.siInfo.pVgroupHash);
147✔
505
  tSimpleHashCleanup(pStmt->sql.siInfo.pTableHash);
147✔
506
  taosArrayDestroyEx(pStmt->sql.siInfo.tbBuf.pBufList, stmtFreeTbBuf);
147✔
507
  taosMemoryFree(pStmt->sql.siInfo.pTSchema);
147!
508
  qDestroyStmtDataBlock(pStmt->sql.siInfo.pDataCtx);
147✔
509
  taosArrayDestroyEx(pStmt->sql.siInfo.pTableCols, stmtFreeTbCols);
147✔
510
  pStmt->sql.siInfo.pTableCols = NULL;
147✔
511

512
  (void)memset(&pStmt->sql, 0, sizeof(pStmt->sql));
147✔
513
  pStmt->sql.siInfo.tableColsReady = true;
147✔
514

515
  STMT_DLOG_E("end to free SQL info");
147!
516

517
  return TSDB_CODE_SUCCESS;
147✔
518
}
519

520
static int32_t stmtTryAddTableVgroupInfo(STscStmt2* pStmt, int32_t* vgId) {
113✔
521
  if (*vgId >= 0 && taosHashGet(pStmt->sql.pVgHash, (const char*)vgId, sizeof(*vgId))) {
113✔
522
    return TSDB_CODE_SUCCESS;
15✔
523
  }
524

525
  SVgroupInfo      vgInfo = {0};
98✔
526
  SRequestConnInfo conn = {.pTrans = pStmt->taos->pAppInfo->pTransporter,
98✔
527
                           .requestId = pStmt->exec.pRequest->requestId,
98✔
528
                           .requestObjRefId = pStmt->exec.pRequest->self,
98✔
529
                           .mgmtEps = getEpSet_s(&pStmt->taos->pAppInfo->mgmtEp)};
98✔
530

531
  int32_t code = catalogGetTableHashVgroup(pStmt->pCatalog, &conn, &pStmt->bInfo.sname, &vgInfo);
98✔
532
  if (TSDB_CODE_SUCCESS != code) {
98!
533
    return code;
×
534
  }
535

536
  code =
537
      taosHashPut(pStmt->sql.pVgHash, (const char*)&vgInfo.vgId, sizeof(vgInfo.vgId), (char*)&vgInfo, sizeof(vgInfo));
98✔
538
  if (TSDB_CODE_SUCCESS != code) {
98!
539
    return code;
×
540
  }
541

542
  *vgId = vgInfo.vgId;
98✔
543

544
  return TSDB_CODE_SUCCESS;
98✔
545
}
546

547
static int32_t stmtRebuildDataBlock(STscStmt2* pStmt, STableDataCxt* pDataBlock, STableDataCxt** newBlock, uint64_t uid,
66✔
548
                                    uint64_t suid, int32_t vgId) {
549
  STMT_ERR_RET(stmtTryAddTableVgroupInfo(pStmt, &vgId));
66!
550
  STMT_ERR_RET(qRebuildStmtDataBlock(newBlock, pDataBlock, uid, suid, vgId, pStmt->sql.autoCreateTbl));
66!
551

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

554
  return TSDB_CODE_SUCCESS;
66✔
555
}
556

557
static int32_t stmtGetFromCache(STscStmt2* pStmt) {
195✔
558
  if (pStmt->sql.stbInterlaceMode && pStmt->sql.siInfo.pDataCtx) {
195!
559
    pStmt->bInfo.needParse = false;
×
560
    pStmt->bInfo.inExecCache = false;
×
561
    return TSDB_CODE_SUCCESS;
×
562
  }
563

564
  pStmt->bInfo.needParse = true;
195✔
565
  pStmt->bInfo.inExecCache = false;
195✔
566

567
  STableDataCxt** pCxtInExec = taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
195✔
568
  if (pCxtInExec) {
195✔
569
    pStmt->bInfo.needParse = false;
24✔
570
    pStmt->bInfo.inExecCache = true;
24✔
571

572
    pStmt->exec.pCurrBlock = *pCxtInExec;
24✔
573

574
    if (pStmt->sql.autoCreateTbl) {
24✔
575
      tscDebug("reuse stmt block for tb %s in execBlock", pStmt->bInfo.tbFName);
18!
576
      return TSDB_CODE_SUCCESS;
18✔
577
    }
578
  }
579

580
  if (NULL == pStmt->pCatalog) {
177✔
581
    STMT_ERR_RET(catalogGetHandle(pStmt->taos->pAppInfo->clusterId, &pStmt->pCatalog));
89!
582
    pStmt->sql.siInfo.pCatalog = pStmt->pCatalog;
92✔
583
  }
584

585
  if (NULL == pStmt->sql.pTableCache || taosHashGetSize(pStmt->sql.pTableCache) <= 0) {
180✔
586
    if (pStmt->bInfo.inExecCache) {
109!
587
      pStmt->bInfo.needParse = false;
×
588
      tscDebug("reuse stmt block for tb %s in execBlock", pStmt->bInfo.tbFName);
×
589
      return TSDB_CODE_SUCCESS;
×
590
    }
591

592
    tscDebug("no stmt block cache for tb %s", pStmt->bInfo.tbFName);
109!
593
    return TSDB_CODE_SUCCESS;
109✔
594
  }
595

596
  if (pStmt->sql.autoCreateTbl) {
73✔
597
    SStmtTableCache* pCache = taosHashGet(pStmt->sql.pTableCache, &pStmt->bInfo.tbSuid, sizeof(pStmt->bInfo.tbSuid));
48✔
598
    if (pCache) {
48!
599
      pStmt->bInfo.needParse = false;
48✔
600
      pStmt->bInfo.tbUid = 0;
48✔
601

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

605
      if (taosHashPut(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName), &pNewBlock,
48!
606
                      POINTER_BYTES)) {
607
        STMT_ERR_RET(terrno);
×
608
      }
609

610
      pStmt->exec.pCurrBlock = pNewBlock;
48✔
611

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

614
      return TSDB_CODE_SUCCESS;
48✔
615
    }
616

617
    STMT_RET(stmtCleanBindInfo(pStmt));
×
618
  }
619

620
  uint64_t uid, suid;
621
  int32_t  vgId;
622
  int8_t   tableType;
623

624
  STableMeta*      pTableMeta = NULL;
25✔
625
  SRequestConnInfo conn = {.pTrans = pStmt->taos->pAppInfo->pTransporter,
25✔
626
                           .requestId = pStmt->exec.pRequest->requestId,
25✔
627
                           .requestObjRefId = pStmt->exec.pRequest->self,
25✔
628
                           .mgmtEps = getEpSet_s(&pStmt->taos->pAppInfo->mgmtEp)};
25✔
629
  int32_t          code = catalogGetTableMeta(pStmt->pCatalog, &conn, &pStmt->bInfo.sname, &pTableMeta);
24✔
630

631
  pStmt->stat.ctgGetTbMetaNum++;
24✔
632

633
  if (TSDB_CODE_PAR_TABLE_NOT_EXIST == code) {
24!
634
    tscDebug("tb %s not exist", pStmt->bInfo.tbFName);
×
635
    STMT_ERR_RET(stmtCleanBindInfo(pStmt));
×
636

637
    STMT_ERR_RET(code);
×
638
  }
639

640
  STMT_ERR_RET(code);
24!
641

642
  uid = pTableMeta->uid;
24✔
643
  suid = pTableMeta->suid;
24✔
644
  tableType = pTableMeta->tableType;
24✔
645
  pStmt->bInfo.tbVgId = pTableMeta->vgId;
24✔
646
  vgId = pTableMeta->vgId;
24✔
647

648
  taosMemoryFree(pTableMeta);
24!
649

650
  uint64_t cacheUid = (TSDB_CHILD_TABLE == tableType) ? suid : uid;
24!
651

652
  if (uid == pStmt->bInfo.tbUid) {
24!
653
    pStmt->bInfo.needParse = false;
×
654

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

657
    return TSDB_CODE_SUCCESS;
×
658
  }
659

660
  if (pStmt->bInfo.inExecCache) {
24✔
661
    SStmtTableCache* pCache = taosHashGet(pStmt->sql.pTableCache, &cacheUid, sizeof(cacheUid));
6✔
662
    if (NULL == pCache) {
6!
663
      tscError("table [%s, %" PRIx64 ", %" PRIx64 "] found in exec blockHash, but not in sql blockHash",
×
664
               pStmt->bInfo.tbFName, uid, cacheUid);
665

666
      STMT_ERR_RET(TSDB_CODE_APP_ERROR);
×
667
    }
668

669
    pStmt->bInfo.needParse = false;
6✔
670

671
    pStmt->bInfo.tbUid = uid;
6✔
672
    pStmt->bInfo.tbSuid = suid;
6✔
673
    pStmt->bInfo.tbType = tableType;
6✔
674
    pStmt->bInfo.boundTags = pCache->boundTags;
6✔
675
    pStmt->bInfo.tagsCached = true;
6✔
676

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

679
    return TSDB_CODE_SUCCESS;
6✔
680
  }
681

682
  SStmtTableCache* pCache = taosHashGet(pStmt->sql.pTableCache, &cacheUid, sizeof(cacheUid));
18✔
683
  if (pCache) {
18!
684
    pStmt->bInfo.needParse = false;
18✔
685

686
    pStmt->bInfo.tbUid = uid;
18✔
687
    pStmt->bInfo.tbSuid = suid;
18✔
688
    pStmt->bInfo.tbType = tableType;
18✔
689
    pStmt->bInfo.boundTags = pCache->boundTags;
18✔
690
    pStmt->bInfo.tagsCached = true;
18✔
691

692
    STableDataCxt* pNewBlock = NULL;
18✔
693
    STMT_ERR_RET(stmtRebuildDataBlock(pStmt, pCache->pDataCtx, &pNewBlock, uid, suid, vgId));
18!
694

695
    if (taosHashPut(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName), &pNewBlock,
18!
696
                    POINTER_BYTES)) {
697
      STMT_ERR_RET(terrno);
×
698
    }
699

700
    pStmt->exec.pCurrBlock = pNewBlock;
18✔
701

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

704
    return TSDB_CODE_SUCCESS;
18✔
705
  }
706

707
  STMT_ERR_RET(stmtCleanBindInfo(pStmt));
×
708

709
  return TSDB_CODE_SUCCESS;
×
710
}
711

712
static int32_t stmtResetStmt(STscStmt2* pStmt) {
×
713
  STMT_ERR_RET(stmtCleanSQLInfo(pStmt));
×
714

715
  pStmt->sql.pTableCache = taosHashInit(100, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
×
716
  if (NULL == pStmt->sql.pTableCache) {
×
717
    STMT_ERR_RET(terrno);
×
718
  }
719

720
  pStmt->sql.status = STMT_INIT;
×
721

722
  return TSDB_CODE_SUCCESS;
×
723
}
724

725
static int32_t stmtAsyncOutput(STscStmt2* pStmt, void* param) {
10,184✔
726
  SStmtQNode* pParam = (SStmtQNode*)param;
10,184✔
727

728
  if (pParam->restoreTbCols) {
10,184✔
729
    for (int32_t i = 0; i < pStmt->sql.siInfo.pTableColsIdx; ++i) {
10,182✔
730
      SArray** p = (SArray**)TARRAY_GET_ELEM(pStmt->sql.siInfo.pTableCols, i);
5,763✔
731
      *p = taosArrayInit(20, POINTER_BYTES);
5,763✔
732
      if (*p == NULL) {
5,764!
733
        STMT_ERR_RET(terrno);
×
734
      }
735
    }
736

737
    atomic_store_8((int8_t*)&pStmt->sql.siInfo.tableColsReady, true);
4,419✔
738
  } else {
739
    int code = qAppendStmtTableOutput(pStmt->sql.pQuery, pStmt->sql.pVgHash, &pParam->tblData, pStmt->exec.pCurrBlock,
5,757✔
740
                                      &pStmt->sql.siInfo, pParam->pCreateTbReq);
741
    // taosMemoryFree(pParam->pTbData);
742
    (void)atomic_sub_fetch_64(&pStmt->sql.siInfo.tbRemainNum, 1);
5,758✔
743
    STMT_ERR_RET(code);
5,764!
744
  }
745
  return TSDB_CODE_SUCCESS;
10,190✔
746
}
747

748
static void* stmtBindThreadFunc(void* param) {
100✔
749
  setThreadName("stmt2Bind");
100✔
750
  qInfo("stmt2 bind thread started");
100!
751

752
  STscStmt2* pStmt = (STscStmt2*)param;
100✔
753

754
  while (true) {
10,189✔
755
    SStmtQNode* asyncParam = NULL;
10,289✔
756

757
    if (!stmtDequeue(pStmt, &asyncParam)) {
10,289✔
758
      if (pStmt->queue.stopQueue && 0 == atomic_load_64((int64_t*)&pStmt->queue.qRemainNum)) {
100!
759
        STMT2_DLOG_E("queue is empty and stopQueue is set, thread will exit");
100!
760
        break;
100✔
761
      }
762
      continue;
×
763
    }
764

765
    int ret = stmtAsyncOutput(pStmt, asyncParam);
10,187✔
766
    if (ret != 0) {
10,189!
767
      STMT2_ELOG("stmtAsyncOutput failed, reason:%s", tstrerror(ret));
×
768
    }
769
  }
770

771
  qInfo("stmt2 bind thread stopped");
100!
772
  return NULL;
100✔
773
}
774

775
static int32_t stmtStartBindThread(STscStmt2* pStmt) {
100✔
776
  TdThreadAttr thAttr;
777
  if (taosThreadAttrInit(&thAttr) != 0) {
100!
778
    return TSDB_CODE_TSC_INTERNAL_ERROR;
×
779
  }
780
  if (taosThreadAttrSetDetachState(&thAttr, PTHREAD_CREATE_JOINABLE) != 0) {
100!
781
    return TSDB_CODE_TSC_INTERNAL_ERROR;
×
782
  }
783

784
  if (taosThreadCreate(&pStmt->bindThread, &thAttr, stmtBindThreadFunc, pStmt) != 0) {
100!
785
    terrno = TAOS_SYSTEM_ERROR(ERRNO);
×
786
    STMT_ERR_RET(terrno);
×
787
  }
788

789
  pStmt->bindThreadInUse = true;
100✔
790

791
  (void)taosThreadAttrDestroy(&thAttr);
100✔
792
  return TSDB_CODE_SUCCESS;
100✔
793
}
794

795
static int32_t stmtInitQueue(STscStmt2* pStmt) {
100✔
796
  (void)taosThreadCondInit(&pStmt->queue.waitCond, NULL);
100✔
797
  (void)taosThreadMutexInit(&pStmt->queue.mutex, NULL);
100✔
798
  STMT_ERR_RET(stmtAllocQNodeFromBuf(&pStmt->sql.siInfo.tbBuf, (void**)&pStmt->queue.head));
200!
799
  pStmt->queue.tail = pStmt->queue.head;
100✔
800

801
  return TSDB_CODE_SUCCESS;
100✔
802
}
803

804
static int32_t stmtIniAsyncBind(STscStmt2* pStmt) {
149✔
805
  (void)taosThreadCondInit(&pStmt->asyncBindParam.waitCond, NULL);
149✔
806
  (void)taosThreadMutexInit(&pStmt->asyncBindParam.mutex, NULL);
149✔
807
  pStmt->asyncBindParam.asyncBindNum = 0;
149✔
808

809
  return TSDB_CODE_SUCCESS;
149✔
810
}
811

812
static int32_t stmtInitTableBuf(STableBufInfo* pTblBuf) {
100✔
813
  pTblBuf->buffUnit = sizeof(SStmtQNode);
100✔
814
  pTblBuf->buffSize = pTblBuf->buffUnit * 1000;
100✔
815
  pTblBuf->pBufList = taosArrayInit(100, POINTER_BYTES);
100✔
816
  if (NULL == pTblBuf->pBufList) {
100!
817
    return terrno;
×
818
  }
819
  void* buff = taosMemoryMalloc(pTblBuf->buffSize);
100!
820
  if (NULL == buff) {
100!
821
    return terrno;
×
822
  }
823

824
  if (taosArrayPush(pTblBuf->pBufList, &buff) == NULL) {
200!
825
    return terrno;
×
826
  }
827

828
  pTblBuf->pCurBuff = buff;
100✔
829
  pTblBuf->buffIdx = 1;
100✔
830
  pTblBuf->buffOffset = 0;
100✔
831

832
  return TSDB_CODE_SUCCESS;
100✔
833
}
834

835
TAOS_STMT2* stmtInit2(STscObj* taos, TAOS_STMT2_OPTION* pOptions) {
147✔
836
  STscObj*   pObj = (STscObj*)taos;
147✔
837
  STscStmt2* pStmt = NULL;
147✔
838
  int32_t    code = 0;
147✔
839

840
  pStmt = taosMemoryCalloc(1, sizeof(STscStmt2));
147!
841
  if (NULL == pStmt) {
149!
842
    return NULL;
×
843
  }
844

845
  pStmt->sql.pTableCache = taosHashInit(100, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
149✔
846
  if (NULL == pStmt->sql.pTableCache) {
149!
847
    taosMemoryFree(pStmt);
×
848
    return NULL;
×
849
  }
850

851
  pStmt->taos = pObj;
149✔
852
  if (taos->db[0] != '\0') {
149✔
853
    pStmt->db = taosStrdup(taos->db);
51!
854
  }
855
  pStmt->bInfo.needParse = true;
149✔
856
  pStmt->sql.status = STMT_INIT;
149✔
857
  pStmt->errCode = TSDB_CODE_SUCCESS;
149✔
858

859
  if (NULL != pOptions) {
149!
860
    (void)memcpy(&pStmt->options, pOptions, sizeof(pStmt->options));
149✔
861
    if (pOptions->singleStbInsert && pOptions->singleTableBindOnce) {
149✔
862
      pStmt->stbInterlaceMode = true;
83✔
863
    }
864

865
    pStmt->reqid = pOptions->reqid;
149✔
866
  }
867

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

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

899
  pStmt->sql.siInfo.tableColsReady = true;
149✔
900
  if (pStmt->options.asyncExecFn) {
149✔
901
    if (tsem_init(&pStmt->asyncExecSem, 0, 1) != 0) {
5!
902
      terrno = TAOS_SYSTEM_ERROR(ERRNO);
×
903
      (void)stmtClose2(pStmt);
×
904
      return NULL;
×
905
    }
906
  }
907
  code = stmtIniAsyncBind(pStmt);
149✔
908
  if (TSDB_CODE_SUCCESS != code) {
149!
909
    terrno = code;
×
910
    (void)stmtClose2(pStmt);
×
911
    return NULL;
×
912
  }
913

914
  pStmt->execSemWaited = false;
149✔
915

916
  STMT_LOG_SEQ(STMT_INIT);
149!
917

918
  tscDebug("stmt:%p initialized", pStmt);
149!
919

920
  return pStmt;
149✔
921
}
922

923
static int stmtSetDbName2(TAOS_STMT2* stmt, const char* dbName) {
64✔
924
  if (dbName == NULL || dbName[0] == '\0') {
64!
925
    return TSDB_CODE_TSC_SQL_SYNTAX_ERROR;
×
926
  }
927
  STscStmt2* pStmt = (STscStmt2*)stmt;
64✔
928

929
  STMT2_DLOG("dbname is specified in sql:%s", dbName);
64!
930
  if (pStmt->db == NULL || pStmt->db[0] == '\0') {
64!
931
    taosMemoryFreeClear(pStmt->db);
33!
932
    STMT2_DLOG("dbname not set by taosconnect, set by sql:%s", dbName);
33!
933
    pStmt->db = taosStrdup(dbName);
33!
934
    (void)strdequote(pStmt->db);
33✔
935
  }
936
  STMT_ERR_RET(stmtCreateRequest(pStmt));
64!
937

938
  // The SQL statement specifies a database name, overriding the previously specified database
939
  taosMemoryFreeClear(pStmt->exec.pRequest->pDb);
64!
940
  pStmt->exec.pRequest->pDb = taosStrdup(dbName);
64!
941
  (void)strdequote(pStmt->exec.pRequest->pDb);
64✔
942
  if (pStmt->exec.pRequest->pDb == NULL) {
64!
943
    return terrno;
×
944
  }
945
  if (pStmt->sql.stbInterlaceMode) {
64✔
946
    pStmt->sql.siInfo.dbname = pStmt->exec.pRequest->pDb;
22✔
947
  }
948
  return TSDB_CODE_SUCCESS;
64✔
949
}
950
static int32_t stmtResetStbInterlaceCache(STscStmt2* pStmt) {
17✔
951
  int32_t code = TSDB_CODE_SUCCESS;
17✔
952

953
  if (pStmt->bindThreadInUse) {
17!
954
    while (0 == atomic_load_8((int8_t*)&pStmt->sql.siInfo.tableColsReady)) {
17!
955
      taosUsleep(1);
×
956
    }
957
    (void)taosThreadMutexLock(&pStmt->queue.mutex);
17✔
958
    pStmt->queue.stopQueue = true;
17✔
959
    (void)taosThreadCondSignal(&(pStmt->queue.waitCond));
17✔
960
    (void)taosThreadMutexUnlock(&pStmt->queue.mutex);
17✔
961

962
    (void)taosThreadJoin(pStmt->bindThread, NULL);
17✔
963
    pStmt->bindThreadInUse = false;
17✔
964
    pStmt->queue.head = NULL;
17✔
965
    pStmt->queue.tail = NULL;
17✔
966
    pStmt->queue.qRemainNum = 0;
17✔
967

968
    (void)taosThreadCondDestroy(&pStmt->queue.waitCond);
17✔
969
    (void)taosThreadMutexDestroy(&pStmt->queue.mutex);
17✔
970
  }
971

972
  pStmt->sql.siInfo.pTableHash = tSimpleHashInit(100, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY));
17✔
973
  if (NULL == pStmt->sql.siInfo.pTableHash) {
17!
974
    return terrno;
×
975
  }
976

977
  pStmt->sql.siInfo.pTableCols = taosArrayInit(STMT_TABLE_COLS_NUM, POINTER_BYTES);
17✔
978
  if (NULL == pStmt->sql.siInfo.pTableCols) {
17!
979
    return terrno;
×
980
  }
981

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

984
  if (TSDB_CODE_SUCCESS == code) {
17!
985
    code = stmtInitQueue(pStmt);
17✔
986
    pStmt->queue.stopQueue = false;
17✔
987
  }
988
  if (TSDB_CODE_SUCCESS == code) {
17!
989
    code = stmtStartBindThread(pStmt);
17✔
990
  }
991
  if (TSDB_CODE_SUCCESS != code) {
17!
992
    return code;
×
993
  }
994

995
  return TSDB_CODE_SUCCESS;
17✔
996
}
997

998
static int32_t stmtResetStmtForPrepare(STscStmt2* pStmt) {
21✔
999
  char*             db = pStmt->db;
21✔
1000
  bool              stbInterlaceMode = pStmt->stbInterlaceMode;
21✔
1001
  TAOS_STMT2_OPTION options = pStmt->options;
21✔
1002
  uint32_t          reqid = pStmt->reqid;
21✔
1003

1004
  taosMemoryFree(pStmt->sql.pBindInfo);
21!
1005
  pStmt->sql.pBindInfo = NULL;
21✔
1006

1007
  taosMemoryFree(pStmt->sql.queryRes.fields);
21!
1008
  pStmt->sql.queryRes.fields = NULL;
21✔
1009

1010
  taosMemoryFree(pStmt->sql.queryRes.userFields);
21!
1011
  pStmt->sql.queryRes.userFields = NULL;
21✔
1012

1013
  pStmt->sql.type = 0;
21✔
1014
  pStmt->sql.runTimes = 0;
21✔
1015
  taosMemoryFree(pStmt->sql.sqlStr);
21!
1016
  pStmt->sql.sqlStr = NULL;
21✔
1017

1018
  qDestroyQuery(pStmt->sql.pQuery);
21✔
1019
  pStmt->sql.pQuery = NULL;
21✔
1020

1021
  taosArrayDestroy(pStmt->sql.nodeList);
21✔
1022
  pStmt->sql.nodeList = NULL;
21✔
1023

1024
  taosHashCleanup(pStmt->sql.pVgHash);
21✔
1025
  pStmt->sql.pVgHash = NULL;
21✔
1026

1027
  if (pStmt->sql.fixValueTags) {
21✔
1028
    tdDestroySVCreateTbReq(pStmt->sql.fixValueTbReq);
12!
1029
    pStmt->sql.fixValueTbReq = NULL;
12✔
1030
  }
1031
  pStmt->sql.fixValueTags = false;
21✔
1032

1033
  void* pIter = taosHashIterate(pStmt->sql.pTableCache, NULL);
21✔
1034
  while (pIter) {
24✔
1035
    SStmtTableCache* pCache = (SStmtTableCache*)pIter;
3✔
1036

1037
    qDestroyStmtDataBlock(pCache->pDataCtx);
3✔
1038
    qDestroyBoundColInfo(pCache->boundTags);
3✔
1039
    taosMemoryFreeClear(pCache->boundTags);
3!
1040

1041
    pIter = taosHashIterate(pStmt->sql.pTableCache, pIter);
3✔
1042
  }
1043
  taosHashCleanup(pStmt->sql.pTableCache);
21✔
1044
  pStmt->sql.pTableCache = taosHashInit(100, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
21✔
1045
  if (NULL == pStmt->sql.pTableCache) {
21!
1046
    return terrno;
×
1047
  }
1048

1049
  if (pStmt->sql.stbInterlaceMode) {
21✔
1050
    pStmt->bInfo.tagsCached = false;
16✔
1051
  }
1052
  STMT_ERR_RET(stmtCleanExecInfo(pStmt, false, true));
21!
1053

1054
  if (pStmt->exec.pRequest) {
21!
1055
    taos_free_result(pStmt->exec.pRequest);
×
1056
    pStmt->exec.pRequest = NULL;
×
1057
  }
1058

1059
  if (pStmt->sql.siInfo.pTableCols) {
21✔
1060
    taosArrayDestroyEx(pStmt->sql.siInfo.pTableCols, stmtFreeTbCols);
17✔
1061
    pStmt->sql.siInfo.pTableCols = NULL;
17✔
1062
  }
1063

1064
  if (pStmt->sql.siInfo.tbBuf.pBufList) {
21✔
1065
    taosArrayDestroyEx(pStmt->sql.siInfo.tbBuf.pBufList, stmtFreeTbBuf);
17✔
1066
    pStmt->sql.siInfo.tbBuf.pBufList = NULL;
17✔
1067
  }
1068

1069
  if (pStmt->sql.siInfo.pTableHash) {
21✔
1070
    tSimpleHashCleanup(pStmt->sql.siInfo.pTableHash);
17✔
1071
    pStmt->sql.siInfo.pTableHash = NULL;
17✔
1072
  }
1073

1074
  if (pStmt->sql.siInfo.pVgroupHash) {
21!
1075
    taosHashCleanup(pStmt->sql.siInfo.pVgroupHash);
×
1076
    pStmt->sql.siInfo.pVgroupHash = NULL;
×
1077
  }
1078

1079
  if (pStmt->sql.siInfo.pVgroupList) {
21!
1080
    taosArrayDestroy(pStmt->sql.siInfo.pVgroupList);
×
1081
    pStmt->sql.siInfo.pVgroupList = NULL;
×
1082
  }
1083

1084
  if (pStmt->sql.siInfo.pDataCtx) {
21✔
1085
    qDestroyStmtDataBlock(pStmt->sql.siInfo.pDataCtx);
15✔
1086
    pStmt->sql.siInfo.pDataCtx = NULL;
15✔
1087
  }
1088

1089
  if (pStmt->sql.siInfo.pTSchema) {
21✔
1090
    taosMemoryFree(pStmt->sql.siInfo.pTSchema);
15!
1091
    pStmt->sql.siInfo.pTSchema = NULL;
15✔
1092
  }
1093

1094
  if (pStmt->sql.siInfo.pRequest) {
21✔
1095
    taos_free_result(pStmt->sql.siInfo.pRequest);
15✔
1096
    pStmt->sql.siInfo.pRequest = NULL;
15✔
1097
  }
1098

1099
  if (stbInterlaceMode) {
21✔
1100
    STMT2_DLOG_E("reprepare for inter");
17!
1101
    STMT_ERR_RET(stmtResetStbInterlaceCache(pStmt));
17!
1102
  }
1103

1104
  pStmt->db = db;
21✔
1105
  pStmt->stbInterlaceMode = stbInterlaceMode;
21✔
1106
  pStmt->options = options;
21✔
1107
  pStmt->reqid = reqid;
21✔
1108

1109
  pStmt->sql.status = STMT_INIT;
21✔
1110

1111
  return TSDB_CODE_SUCCESS;
21✔
1112
}
1113

1114
int stmtPrepare2(TAOS_STMT2* stmt, const char* sql, unsigned long length) {
163✔
1115
  STscStmt2* pStmt = (STscStmt2*)stmt;
163✔
1116
  int32_t    code = 0;
163✔
1117

1118
  if (stmt == NULL || sql == NULL) {
163!
1119
    STMT2_ELOG_E("stmt or sql is NULL");
×
1120
    return TSDB_CODE_INVALID_PARA;
×
1121
  }
1122

1123
  if (pStmt->sql.status >= STMT_PREPARE) {
168✔
1124
    STMT2_DLOG("stmt status is %d, need to reset stmt2 cache before prepare", pStmt->sql.status);
21!
1125
    STMT_ERR_RET(stmtResetStmtForPrepare(pStmt));
21!
1126
  }
1127

1128
  STMT2_DLOG("start to prepare with sql:%s", sql);
168!
1129

1130
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
168✔
1131
    STMT2_ELOG("stmt errCode is not success, ErrCode: 0x%x, ErrMessage: %s\n. ", pStmt->errCode,
1!
1132
               strerror(pStmt->errCode));
1133
    return pStmt->errCode;
1✔
1134
  }
1135

1136
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_PREPARE));
167!
1137

1138
  if (length <= 0) {
167✔
1139
    length = strlen(sql);
103✔
1140
  }
1141

1142
  pStmt->sql.sqlStr = taosStrndup(sql, length);
167!
1143
  if (!pStmt->sql.sqlStr) {
166!
1144
    return terrno;
×
1145
  }
1146
  pStmt->sql.sqlLen = length;
166✔
1147
  pStmt->sql.stbInterlaceMode = pStmt->stbInterlaceMode;
166✔
1148

1149
  char* dbName = NULL;
166✔
1150
  if (qParseDbName(sql, length, &dbName)) {
166✔
1151
    STMT_ERR_RET(stmtSetDbName2(stmt, dbName));
64!
1152
    taosMemoryFreeClear(dbName);
64!
1153
  }
1154

1155
  return TSDB_CODE_SUCCESS;
160✔
1156
}
1157

1158
static int32_t stmtInitStbInterlaceTableInfo(STscStmt2* pStmt) {
89✔
1159
  STableDataCxt** pSrc = taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
89✔
1160
  if (!pSrc) {
90!
1161
    return terrno;
×
1162
  }
1163
  STableDataCxt* pDst = NULL;
90✔
1164

1165
  STMT_ERR_RET(qCloneStmtDataBlock(&pDst, *pSrc, true));
90!
1166
  pStmt->sql.siInfo.pDataCtx = pDst;
85✔
1167

1168
  SArray* pTblCols = NULL;
85✔
1169
  for (int32_t i = 0; i < STMT_TABLE_COLS_NUM; i++) {
76,104✔
1170
    pTblCols = taosArrayInit(20, POINTER_BYTES);
75,792✔
1171
    if (NULL == pTblCols) {
78,283!
1172
      return terrno;
×
1173
    }
1174

1175
    if (taosArrayPush(pStmt->sql.siInfo.pTableCols, &pTblCols) == NULL) {
154,302!
1176
      return terrno;
×
1177
    }
1178
  }
1179

1180
  pStmt->sql.siInfo.boundTags = pStmt->bInfo.boundTags;
312✔
1181

1182
  return TSDB_CODE_SUCCESS;
312✔
1183
}
1184

1185
int stmtIsInsert2(TAOS_STMT2* stmt, int* insert) {
11,790✔
1186
  STscStmt2* pStmt = (STscStmt2*)stmt;
11,790✔
1187

1188
  STMT_DLOG_E("start is insert");
11,790!
1189

1190
  if (pStmt->sql.type) {
11,798✔
1191
    *insert = (STMT_TYPE_INSERT == pStmt->sql.type || STMT_TYPE_MULTI_INSERT == pStmt->sql.type);
11,634✔
1192
  } else {
1193
    *insert = qIsInsertValuesSql(pStmt->sql.sqlStr, pStmt->sql.sqlLen);
164✔
1194
  }
1195

1196
  return TSDB_CODE_SUCCESS;
11,800✔
1197
}
1198

1199
int stmtSetTbName2(TAOS_STMT2* stmt, const char* tbName) {
5,864✔
1200
  STscStmt2* pStmt = (STscStmt2*)stmt;
5,864✔
1201

1202
  int64_t startUs = taosGetTimestampUs();
5,870✔
1203

1204
  STMT_DLOG("start to set tbName:%s", tbName);
5,870!
1205

1206
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
5,866!
1207
    return pStmt->errCode;
×
1208
  }
1209

1210
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_SETTBNAME));
5,866!
1211

1212
  int32_t insert = 0;
5,869✔
1213
  STMT_ERR_RET(stmtIsInsert2(stmt, &insert));
5,869!
1214
  if (0 == insert) {
5,870!
1215
    tscError("set tb name not available for none insert statement");
×
1216
    STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR);
×
1217
  }
1218

1219
  if (!pStmt->sql.stbInterlaceMode || NULL == pStmt->sql.siInfo.pDataCtx) {
5,870✔
1220
    STMT_ERR_RET(stmtCreateRequest(pStmt));
202!
1221

1222
    STMT_ERR_RET(qCreateSName(&pStmt->bInfo.sname, tbName, pStmt->taos->acctId, pStmt->exec.pRequest->pDb,
198!
1223
                              pStmt->exec.pRequest->msgBuf, pStmt->exec.pRequest->msgBufLen));
1224
    STMT_ERR_RET(tNameExtractFullName(&pStmt->bInfo.sname, pStmt->bInfo.tbFName));
198✔
1225

1226
    STMT_ERR_RET(stmtGetFromCache(pStmt));
195!
1227

1228
    if (pStmt->bInfo.needParse) {
198✔
1229
      tstrncpy(pStmt->bInfo.tbName, tbName, sizeof(pStmt->bInfo.tbName));
107✔
1230
      pStmt->bInfo.tbName[sizeof(pStmt->bInfo.tbName) - 1] = 0;
107✔
1231

1232
      STMT_ERR_RET(stmtParseSql(pStmt));
107!
1233
    }
1234
  } else {
1235
    tstrncpy(pStmt->bInfo.tbName, tbName, sizeof(pStmt->bInfo.tbName));
5,668✔
1236
    pStmt->bInfo.tbName[sizeof(pStmt->bInfo.tbName) - 1] = 0;
5,668✔
1237
    pStmt->exec.pRequest->requestId++;
5,668✔
1238
    pStmt->bInfo.needParse = false;
5,668✔
1239
  }
1240

1241
  if (pStmt->sql.stbInterlaceMode && NULL == pStmt->sql.siInfo.pDataCtx) {
5,867✔
1242
    STMT_ERR_RET(stmtInitStbInterlaceTableInfo(pStmt));
90!
1243
  }
1244

1245
  int64_t startUs2 = taosGetTimestampUs();
5,866✔
1246
  pStmt->stat.setTbNameUs += startUs2 - startUs;
5,866✔
1247

1248
  return TSDB_CODE_SUCCESS;
5,866✔
1249
}
1250

1251
int stmtSetTbTags2(TAOS_STMT2* stmt, TAOS_STMT2_BIND* tags, SVCreateTbReq** pCreateTbReq) {
120✔
1252
  STscStmt2* pStmt = (STscStmt2*)stmt;
120✔
1253

1254
  STMT_DLOG_E("start to set tbTags");
120!
1255

1256
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
120!
1257
    return pStmt->errCode;
×
1258
  }
1259

1260
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_SETTAGS));
120!
1261

1262
  if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 &&
120!
1263
      STMT_TYPE_MULTI_INSERT != pStmt->sql.type) {
×
1264
    pStmt->bInfo.needParse = false;
×
1265
  }
1266
  STMT_ERR_RET(stmtCreateRequest(pStmt));
120!
1267

1268
  if (pStmt->bInfo.needParse) {
120!
1269
    STMT_ERR_RET(stmtParseSql(pStmt));
×
1270
  }
1271
  if (pStmt->sql.stbInterlaceMode && NULL == pStmt->sql.siInfo.pDataCtx) {
120!
1272
    STMT_ERR_RET(stmtInitStbInterlaceTableInfo(pStmt));
×
1273
  }
1274

1275
  SBoundColInfo* tags_info = (SBoundColInfo*)pStmt->bInfo.boundTags;
120✔
1276
  // if (tags_info->numOfBound <= 0 || tags_info->numOfCols <= 0) {
1277
  //   tscWarn("no tags or cols bound in sql, will not bound tags");
1278
  //   return TSDB_CODE_SUCCESS;
1279
  // }
1280
  if (pStmt->sql.autoCreateTbl && pStmt->sql.stbInterlaceMode) {
120!
1281
    STMT_ERR_RET(qCreateSName(&pStmt->bInfo.sname, pStmt->bInfo.tbName, pStmt->taos->acctId, pStmt->exec.pRequest->pDb,
38!
1282
                              pStmt->exec.pRequest->msgBuf, pStmt->exec.pRequest->msgBufLen));
1283
    STMT_ERR_RET(tNameExtractFullName(&pStmt->bInfo.sname, pStmt->bInfo.tbFName));
38!
1284
  }
1285

1286
  STableDataCxt** pDataBlock = NULL;
120✔
1287
  if (pStmt->exec.pCurrBlock) {
120✔
1288
    pDataBlock = &pStmt->exec.pCurrBlock;
95✔
1289
  } else {
1290
    pDataBlock =
1291
        (STableDataCxt**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
25✔
1292
    if (NULL == pDataBlock) {
25!
1293
      tscError("table %s not found in exec blockHash", pStmt->bInfo.tbFName);
×
1294
      STMT_ERR_RET(TSDB_CODE_TSC_STMT_CACHE_ERROR);
×
1295
    }
1296
    // pStmt->exec.pCurrBlock = *pDataBlock;
1297
    // if (pStmt->sql.stbInterlaceMode) {
1298
    //   taosArrayDestroy(pStmt->exec.pCurrBlock->pData->aCol);
1299
    //   (*pDataBlock)->pData->aCol = NULL;
1300
    // }
1301
  }
1302
  if (pStmt->bInfo.inExecCache && !pStmt->sql.autoCreateTbl) {
120!
1303
    return TSDB_CODE_SUCCESS;
×
1304
  }
1305

1306
  tscDebug("start to bind stmt tag values");
120!
1307

1308
  void* boundTags = NULL;
120✔
1309
  if (pStmt->sql.stbInterlaceMode) {
120✔
1310
    boundTags = pStmt->sql.siInfo.boundTags;
38✔
1311
    *pCreateTbReq = taosMemoryCalloc(1, sizeof(SVCreateTbReq));
38!
1312
    if (NULL == pCreateTbReq) {
38!
1313
      return terrno;
×
1314
    }
1315
    int32_t vgId = -1;
38✔
1316
    STMT_ERR_RET(stmtTryAddTableVgroupInfo(pStmt, &vgId));
38!
1317
    (*pCreateTbReq)->uid = vgId;
38✔
1318
  } else {
1319
    boundTags = pStmt->bInfo.boundTags;
82✔
1320
  }
1321

1322
  STMT_ERR_RET(qBindStmtTagsValue2(*pDataBlock, boundTags, pStmt->bInfo.tbSuid, pStmt->bInfo.stbFName,
120✔
1323
                                   pStmt->bInfo.sname.tname, tags, pStmt->exec.pRequest->msgBuf,
1324
                                   pStmt->exec.pRequest->msgBufLen, pStmt->taos->optionInfo.charsetCxt, *pCreateTbReq));
1325

1326
  return TSDB_CODE_SUCCESS;
119✔
1327
}
1328

1329
int stmtCheckTags2(TAOS_STMT2* stmt, SVCreateTbReq** pCreateTbReq) {
26✔
1330
  STscStmt2* pStmt = (STscStmt2*)stmt;
26✔
1331

1332
  STMT2_TLOG_E("start to clone createTbRequest for fixed tags");
26!
1333

1334
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
26!
1335
    return pStmt->errCode;
×
1336
  }
1337

1338
  if (!pStmt->sql.stbInterlaceMode) {
26!
1339
    return TSDB_CODE_SUCCESS;
×
1340
  }
1341

1342
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_SETTAGS));
26!
1343

1344
  if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 &&
26!
1345
      STMT_TYPE_MULTI_INSERT != pStmt->sql.type) {
×
1346
    pStmt->bInfo.needParse = false;
×
1347
  }
1348
  STMT_ERR_RET(stmtCreateRequest(pStmt));
26!
1349

1350
  if (pStmt->bInfo.needParse) {
26!
1351
    STMT_ERR_RET(stmtParseSql(pStmt));
×
1352
    if (!pStmt->sql.autoCreateTbl) {
×
1353
      return TSDB_CODE_SUCCESS;
×
1354
    }
1355
  }
1356

1357
  if (pStmt->sql.stbInterlaceMode && NULL == pStmt->sql.siInfo.pDataCtx) {
26!
1358
    STMT_ERR_RET(stmtInitStbInterlaceTableInfo(pStmt));
×
1359
  }
1360

1361
  STMT_ERR_RET(qCreateSName(&pStmt->bInfo.sname, pStmt->bInfo.tbName, pStmt->taos->acctId, pStmt->exec.pRequest->pDb,
26!
1362
                            pStmt->exec.pRequest->msgBuf, pStmt->exec.pRequest->msgBufLen));
1363
  STMT_ERR_RET(tNameExtractFullName(&pStmt->bInfo.sname, pStmt->bInfo.tbFName));
26!
1364

1365
  STableDataCxt** pDataBlock = NULL;
26✔
1366
  if (pStmt->exec.pCurrBlock) {
26✔
1367
    pDataBlock = &pStmt->exec.pCurrBlock;
9✔
1368
  } else {
1369
    pDataBlock =
1370
        (STableDataCxt**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
17✔
1371
    if (NULL == pDataBlock) {
17!
1372
      tscError("table %s not found in exec blockHash", pStmt->bInfo.tbFName);
×
1373
      STMT_ERR_RET(TSDB_CODE_TSC_STMT_CACHE_ERROR);
×
1374
    }
1375
  }
1376

1377
  if (!((*pDataBlock)->pData->flags & SUBMIT_REQ_AUTO_CREATE_TABLE)) {
26!
1378
    return TSDB_CODE_SUCCESS;
×
1379
  }
1380

1381
  if (pStmt->sql.fixValueTags) {
26✔
1382
    STMT_ERR_RET(cloneSVreateTbReq(pStmt->sql.fixValueTbReq, pCreateTbReq));
9!
1383
    if ((*pCreateTbReq)->name) {
9!
1384
      taosMemoryFree((*pCreateTbReq)->name);
9!
1385
    }
1386
    (*pCreateTbReq)->name = taosStrdup(pStmt->bInfo.tbName);
9!
1387
    int32_t vgId = -1;
9✔
1388
    STMT_ERR_RET(stmtTryAddTableVgroupInfo(pStmt, &vgId));
9!
1389
    (*pCreateTbReq)->uid = vgId;
9✔
1390
    return TSDB_CODE_SUCCESS;
9✔
1391
  }
1392

1393
  if ((*pDataBlock)->pData->pCreateTbReq) {
17!
1394
    pStmt->sql.fixValueTags = true;
17✔
1395
    STMT_ERR_RET(cloneSVreateTbReq((*pDataBlock)->pData->pCreateTbReq, &pStmt->sql.fixValueTbReq));
17!
1396
    STMT_ERR_RET(cloneSVreateTbReq(pStmt->sql.fixValueTbReq, pCreateTbReq));
17!
1397
    (*pCreateTbReq)->uid = (*pDataBlock)->pMeta->vgId;
17✔
1398
  }
1399

1400
  return TSDB_CODE_SUCCESS;
17✔
1401
}
1402

1403
static int stmtFetchColFields2(STscStmt2* pStmt, int32_t* fieldNum, TAOS_FIELD_E** fields) {
×
1404
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
×
1405
    return pStmt->errCode;
×
1406
  }
1407

1408
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
×
1409
    tscError("invalid operation to get query column fileds");
×
1410
    STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR);
×
1411
  }
1412

1413
  STableDataCxt** pDataBlock = NULL;
×
1414

1415
  if (pStmt->sql.stbInterlaceMode) {
×
1416
    pDataBlock = &pStmt->sql.siInfo.pDataCtx;
×
1417
  } else {
1418
    pDataBlock =
1419
        (STableDataCxt**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
×
1420
    if (NULL == pDataBlock) {
×
1421
      tscError("table %s not found in exec blockHash", pStmt->bInfo.tbFName);
×
1422
      STMT_ERR_RET(TSDB_CODE_APP_ERROR);
×
1423
    }
1424
  }
1425

1426
  STMT_ERR_RET(qBuildStmtColFields(*pDataBlock, fieldNum, fields));
×
1427

1428
  return TSDB_CODE_SUCCESS;
×
1429
}
1430

1431
static int stmtFetchStbColFields2(STscStmt2* pStmt, int32_t* fieldNum, TAOS_FIELD_ALL** fields) {
42✔
1432
  int32_t code = 0;
42✔
1433
  int32_t preCode = pStmt->errCode;
42✔
1434

1435
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
42!
1436
    return pStmt->errCode;
×
1437
  }
1438

1439
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
42!
1440
    tscError("invalid operation to get query column fileds");
×
1441
    STMT_ERRI_JRET(TSDB_CODE_TSC_STMT_API_ERROR);
×
1442
  }
1443

1444
  STableDataCxt** pDataBlock = NULL;
42✔
1445
  bool            cleanStb = false;
42✔
1446

1447
  if (pStmt->sql.stbInterlaceMode && pStmt->sql.siInfo.pDataCtx != NULL) {
42✔
1448
    pDataBlock = &pStmt->sql.siInfo.pDataCtx;
7✔
1449
  } else {
1450
    cleanStb = true;
35✔
1451
    pDataBlock =
1452
        (STableDataCxt**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
35✔
1453
  }
1454

1455
  if (NULL == pDataBlock || NULL == *pDataBlock) {
42!
1456
    tscError("table %s not found in exec blockHash", pStmt->bInfo.tbFName);
×
1457
    STMT_ERRI_JRET(TSDB_CODE_APP_ERROR);
×
1458
  }
1459

1460
  STMT_ERRI_JRET(
42!
1461
      qBuildStmtStbColFields(*pDataBlock, pStmt->bInfo.boundTags, pStmt->bInfo.tbNameFlag, fieldNum, fields));
1462

1463
  if (pStmt->bInfo.tbType == TSDB_SUPER_TABLE && cleanStb) {
42!
1464
    taosMemoryFreeClear((*pDataBlock)->boundColsInfo.pColIndex);
28!
1465
    qDestroyStmtDataBlock(*pDataBlock);
28✔
1466
    *pDataBlock = NULL;
28✔
1467
    if (taosHashRemove(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName)) != 0) {
28!
1468
      tscError("get fileds %s remove exec blockHash fail", pStmt->bInfo.tbFName);
×
1469
      STMT_ERRI_JRET(TSDB_CODE_APP_ERROR);
×
1470
    }
1471
    pStmt->sql.autoCreateTbl = false;
28✔
1472
    pStmt->bInfo.tagsCached = false;
28✔
1473
    pStmt->bInfo.sname = (SName){0};
28✔
1474
    STMT_ERR_RET(stmtCleanBindInfo(pStmt));
28!
1475
  }
1476

1477
_return:
14✔
1478

1479
  pStmt->errCode = preCode;
42✔
1480

1481
  return code;
42✔
1482
}
1483
/*
1484
SArray* stmtGetFreeCol(STscStmt2* pStmt, int32_t* idx) {
1485
  while (true) {
1486
    if (pStmt->exec.smInfo.pColIdx >= STMT_COL_BUF_SIZE) {
1487
      pStmt->exec.smInfo.pColIdx = 0;
1488
    }
1489

1490
    if ((pStmt->exec.smInfo.pColIdx + 1) == atomic_load_32(&pStmt->exec.smInfo.pColFreeIdx)) {
1491
      taosUsleep(1);
1492
      continue;
1493
    }
1494

1495
    *idx = pStmt->exec.smInfo.pColIdx;
1496
    return pStmt->exec.smInfo.pCols[pStmt->exec.smInfo.pColIdx++];
1497
  }
1498
}
1499
*/
1500
static int32_t stmtAppendTablePostHandle(STscStmt2* pStmt, SStmtQNode* param) {
5,752✔
1501
  if (NULL == pStmt->sql.siInfo.pVgroupHash) {
5,752✔
1502
    pStmt->sql.siInfo.pVgroupHash =
4,419✔
1503
        taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_ENTRY_LOCK);
4,420✔
1504
  }
1505
  if (NULL == pStmt->sql.siInfo.pVgroupList) {
5,751✔
1506
    pStmt->sql.siInfo.pVgroupList = taosArrayInit(64, POINTER_BYTES);
4,421✔
1507
  }
1508

1509
  if (NULL == pStmt->sql.siInfo.pRequest) {
5,746✔
1510
    STMT_ERR_RET(buildRequest(pStmt->taos->id, pStmt->sql.sqlStr, pStmt->sql.sqlLen, NULL, false,
87!
1511
                              (SRequestObj**)&pStmt->sql.siInfo.pRequest, pStmt->reqid));
1512

1513
    if (pStmt->reqid != 0) {
88!
1514
      pStmt->reqid++;
×
1515
    }
1516
    pStmt->exec.pRequest->syncQuery = true;
88✔
1517

1518
    pStmt->sql.siInfo.requestId = ((SRequestObj*)pStmt->sql.siInfo.pRequest)->requestId;
88✔
1519
    pStmt->sql.siInfo.requestSelf = ((SRequestObj*)pStmt->sql.siInfo.pRequest)->self;
88✔
1520
  }
1521

1522
  if (!pStmt->sql.siInfo.tbFromHash && pStmt->sql.siInfo.firstName[0] &&
5,747✔
1523
      0 == strcmp(pStmt->sql.siInfo.firstName, pStmt->bInfo.tbName)) {
96✔
1524
    pStmt->sql.siInfo.tbFromHash = true;
35✔
1525
  }
1526

1527
  if (0 == pStmt->sql.siInfo.firstName[0]) {
5,747✔
1528
    tstrncpy(pStmt->sql.siInfo.firstName, pStmt->bInfo.tbName, TSDB_TABLE_NAME_LEN);
74✔
1529
  }
1530

1531
  param->tblData.getFromHash = pStmt->sql.siInfo.tbFromHash;
5,747✔
1532
  param->next = NULL;
5,747✔
1533

1534
  (void)atomic_add_fetch_64(&pStmt->sql.siInfo.tbRemainNum, 1);
5,747✔
1535

1536
  if (pStmt->queue.stopQueue) {
5,767!
1537
    STMT_DLOG_E("stmt bind thread is stopped,cannot enqueue bind request");
×
1538
    return TSDB_CODE_TSC_STMT_API_ERROR;
×
1539
  }
1540
  stmtEnqueue(pStmt, param);
5,767✔
1541

1542
  return TSDB_CODE_SUCCESS;
5,761✔
1543
}
1544

1545
static FORCE_INLINE int32_t stmtGetTableColsFromCache(STscStmt2* pStmt, SArray** pTableCols) {
1546
  while (true) {
1547
    if (pStmt->sql.siInfo.pTableColsIdx < taosArrayGetSize(pStmt->sql.siInfo.pTableCols)) {
5,750!
1548
      *pTableCols = (SArray*)taosArrayGetP(pStmt->sql.siInfo.pTableCols, pStmt->sql.siInfo.pTableColsIdx++);
5,752✔
1549
      break;
5,749✔
1550
    } else {
1551
      SArray* pTblCols = NULL;
×
1552
      for (int32_t i = 0; i < 100; i++) {
×
1553
        pTblCols = taosArrayInit(20, POINTER_BYTES);
×
1554
        if (NULL == pTblCols) {
×
1555
          return terrno;
×
1556
        }
1557

1558
        if (taosArrayPush(pStmt->sql.siInfo.pTableCols, &pTblCols) == NULL) {
×
1559
          return terrno;
×
1560
        }
1561
      }
1562
    }
1563
  }
1564

1565
  return TSDB_CODE_SUCCESS;
5,749✔
1566
}
1567

1568
static int32_t stmtCacheBlock(STscStmt2* pStmt) {
116✔
1569
  if (pStmt->sql.type != STMT_TYPE_MULTI_INSERT) {
116✔
1570
    return TSDB_CODE_SUCCESS;
9✔
1571
  }
1572

1573
  uint64_t uid = pStmt->bInfo.tbUid;
107✔
1574
  uint64_t cacheUid = (TSDB_CHILD_TABLE == pStmt->bInfo.tbType) ? pStmt->bInfo.tbSuid : uid;
107!
1575

1576
  if (taosHashGet(pStmt->sql.pTableCache, &cacheUid, sizeof(cacheUid))) {
107✔
1577
    return TSDB_CODE_SUCCESS;
89✔
1578
  }
1579

1580
  STableDataCxt** pSrc = taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
18✔
1581
  if (!pSrc) {
18!
1582
    return terrno;
×
1583
  }
1584
  STableDataCxt* pDst = NULL;
18✔
1585

1586
  STMT_ERR_RET(qCloneStmtDataBlock(&pDst, *pSrc, true));
18!
1587

1588
  SStmtTableCache cache = {
18✔
1589
      .pDataCtx = pDst,
1590
      .boundTags = pStmt->bInfo.boundTags,
18✔
1591
  };
1592

1593
  if (taosHashPut(pStmt->sql.pTableCache, &cacheUid, sizeof(cacheUid), &cache, sizeof(cache))) {
18!
1594
    return terrno;
×
1595
  }
1596

1597
  if (pStmt->sql.autoCreateTbl) {
18✔
1598
    pStmt->bInfo.tagsCached = true;
15✔
1599
  } else {
1600
    pStmt->bInfo.boundTags = NULL;
3✔
1601
  }
1602

1603
  return TSDB_CODE_SUCCESS;
18✔
1604
}
1605

1606
static int stmtAddBatch2(TAOS_STMT2* stmt) {
4,542✔
1607
  STscStmt2* pStmt = (STscStmt2*)stmt;
4,542✔
1608

1609
  int64_t startUs = taosGetTimestampUs();
4,543✔
1610

1611
  STMT_DLOG_E("start to add batch");
4,543!
1612

1613
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
4,543!
1614
    return pStmt->errCode;
×
1615
  }
1616

1617
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_ADD_BATCH));
4,543!
1618

1619
  if (pStmt->sql.stbInterlaceMode) {
4,541✔
1620
    int64_t startUs2 = taosGetTimestampUs();
4,422✔
1621
    pStmt->stat.addBatchUs += startUs2 - startUs;
4,422✔
1622

1623
    pStmt->sql.siInfo.tableColsReady = false;
4,422✔
1624

1625
    SStmtQNode* param = NULL;
4,422✔
1626
    STMT_ERR_RET(stmtAllocQNodeFromBuf(&pStmt->sql.siInfo.tbBuf, (void**)&param));
8,845!
1627
    param->restoreTbCols = true;
4,423✔
1628
    param->next = NULL;
4,423✔
1629

1630
    if (pStmt->sql.autoCreateTbl) {
4,423✔
1631
      pStmt->bInfo.tagsCached = true;
36✔
1632
    }
1633

1634
    if (pStmt->queue.stopQueue) {
4,423!
1635
      STMT_DLOG_E("stmt bind thread is stopped,cannot enqueue bind request");
×
1636
      return TSDB_CODE_TSC_STMT_API_ERROR;
×
1637
    }
1638

1639
    stmtEnqueue(pStmt, param);
4,423✔
1640

1641
    return TSDB_CODE_SUCCESS;
4,427✔
1642
  }
1643

1644
  STMT_ERR_RET(stmtCacheBlock(pStmt));
116!
1645

1646
  return TSDB_CODE_SUCCESS;
116✔
1647
}
1648
/*
1649
static int32_t stmtBackupQueryFields(STscStmt2* pStmt) {
1650
  SStmtQueryResInfo* pRes = &pStmt->sql.queryRes;
1651
  pRes->numOfCols = pStmt->exec.pRequest->body.resInfo.numOfCols;
1652
  pRes->precision = pStmt->exec.pRequest->body.resInfo.precision;
1653

1654
  int32_t size = pRes->numOfCols * sizeof(TAOS_FIELD);
1655
  pRes->fields = taosMemoryMalloc(size);
1656
  pRes->userFields = taosMemoryMalloc(size);
1657
  if (NULL == pRes->fields || NULL == pRes->userFields) {
1658
    STMT_ERR_RET(terrno);
1659
  }
1660
  (void)memcpy(pRes->fields, pStmt->exec.pRequest->body.resInfo.fields, size);
1661
  (void)memcpy(pRes->userFields, pStmt->exec.pRequest->body.resInfo.userFields, size);
1662

1663
  return TSDB_CODE_SUCCESS;
1664
}
1665

1666
static int32_t stmtRestoreQueryFields(STscStmt2* pStmt) {
1667
  SStmtQueryResInfo* pRes = &pStmt->sql.queryRes;
1668
  int32_t            size = pRes->numOfCols * sizeof(TAOS_FIELD);
1669

1670
  pStmt->exec.pRequest->body.resInfo.numOfCols = pRes->numOfCols;
1671
  pStmt->exec.pRequest->body.resInfo.precision = pRes->precision;
1672

1673
  if (NULL == pStmt->exec.pRequest->body.resInfo.fields) {
1674
    pStmt->exec.pRequest->body.resInfo.fields = taosMemoryMalloc(size);
1675
    if (NULL == pStmt->exec.pRequest->body.resInfo.fields) {
1676
      STMT_ERR_RET(terrno);
1677
    }
1678
    (void)memcpy(pStmt->exec.pRequest->body.resInfo.fields, pRes->fields, size);
1679
  }
1680

1681
  if (NULL == pStmt->exec.pRequest->body.resInfo.userFields) {
1682
    pStmt->exec.pRequest->body.resInfo.userFields = taosMemoryMalloc(size);
1683
    if (NULL == pStmt->exec.pRequest->body.resInfo.userFields) {
1684
      STMT_ERR_RET(terrno);
1685
    }
1686
    (void)memcpy(pStmt->exec.pRequest->body.resInfo.userFields, pRes->userFields, size);
1687
  }
1688

1689
  return TSDB_CODE_SUCCESS;
1690
}
1691
*/
1692
int stmtBindBatch2(TAOS_STMT2* stmt, TAOS_STMT2_BIND* bind, int32_t colIdx, SVCreateTbReq* pCreateTbReq) {
5,868✔
1693
  STscStmt2* pStmt = (STscStmt2*)stmt;
5,868✔
1694
  int32_t    code = 0;
5,868✔
1695

1696
  int64_t startUs = taosGetTimestampUs();
5,874✔
1697

1698
  STMT_DLOG("start to bind stmt data, colIdx:%d", colIdx);
5,874!
1699

1700
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
5,875!
1701
    return pStmt->errCode;
×
1702
  }
1703

1704
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_BIND));
5,875!
1705

1706
  if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 &&
5,874!
1707
      STMT_TYPE_MULTI_INSERT != pStmt->sql.type) {
×
1708
    pStmt->bInfo.needParse = false;
×
1709
  }
1710

1711
  if (pStmt->exec.pRequest && STMT_TYPE_QUERY == pStmt->sql.type && pStmt->sql.runTimes) {
5,874✔
1712
    taos_free_result(pStmt->exec.pRequest);
1✔
1713
    pStmt->exec.pRequest = NULL;
1✔
1714
  }
1715

1716
  STMT_ERR_RET(stmtCreateRequest(pStmt));
5,874!
1717
  if (pStmt->bInfo.needParse) {
5,869✔
1718
    code = stmtParseSql(pStmt);
6✔
1719
    if (code != TSDB_CODE_SUCCESS) {
6!
1720
      goto cleanup_root;
×
1721
    }
1722
  }
1723

1724
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
5,869✔
1725
    code = qStmtBindParams2(pStmt->sql.pQuery, bind, colIdx, pStmt->taos->optionInfo.charsetCxt);
5✔
1726
    if (code != TSDB_CODE_SUCCESS) {
5!
1727
      goto cleanup_root;
×
1728
    }
1729
    SParseContext ctx = {.requestId = pStmt->exec.pRequest->requestId,
5✔
1730
                         .acctId = pStmt->taos->acctId,
5✔
1731
                         .db = pStmt->exec.pRequest->pDb,
5✔
1732
                         .topicQuery = false,
1733
                         .pSql = pStmt->sql.sqlStr,
5✔
1734
                         .sqlLen = pStmt->sql.sqlLen,
5✔
1735
                         .pMsg = pStmt->exec.pRequest->msgBuf,
5✔
1736
                         .msgLen = ERROR_MSG_BUF_DEFAULT_SIZE,
1737
                         .pTransporter = pStmt->taos->pAppInfo->pTransporter,
5✔
1738
                         .pStmtCb = NULL,
1739
                         .pUser = pStmt->taos->user};
5✔
1740
    ctx.mgmtEpSet = getEpSet_s(&pStmt->taos->pAppInfo->mgmtEp);
5✔
1741
    code = catalogGetHandle(pStmt->taos->pAppInfo->clusterId, &ctx.pCatalog);
5✔
1742
    if (code != TSDB_CODE_SUCCESS) {
5!
1743
      goto cleanup_root;
×
1744
    }
1745
    code = qStmtParseQuerySql(&ctx, pStmt->sql.pQuery);
5✔
1746
    if (code != TSDB_CODE_SUCCESS) {
5!
1747
      goto cleanup_root;
×
1748
    }
1749

1750
    if (pStmt->sql.pQuery->haveResultSet) {
5!
1751
      STMT_ERR_RET(setResSchemaInfo(&pStmt->exec.pRequest->body.resInfo, pStmt->sql.pQuery->pResSchema,
10!
1752
                                    pStmt->sql.pQuery->numOfResCols, pStmt->sql.pQuery->pResExtSchema, true));
1753
      taosMemoryFreeClear(pStmt->sql.pQuery->pResSchema);
5!
1754
      taosMemoryFreeClear(pStmt->sql.pQuery->pResExtSchema);
5!
1755
      setResPrecision(&pStmt->exec.pRequest->body.resInfo, pStmt->sql.pQuery->precision);
5✔
1756
    }
1757

1758
    TSWAP(pStmt->exec.pRequest->dbList, pStmt->sql.pQuery->pDbList);
5✔
1759
    TSWAP(pStmt->exec.pRequest->tableList, pStmt->sql.pQuery->pTableList);
5✔
1760
    TSWAP(pStmt->exec.pRequest->targetTableList, pStmt->sql.pQuery->pTargetTableList);
5✔
1761

1762
    // if (STMT_TYPE_QUERY == pStmt->sql.queryRes) {
1763
    //   STMT_ERR_RET(stmtRestoreQueryFields(pStmt));
1764
    // }
1765

1766
    // STMT_ERR_RET(stmtBackupQueryFields(pStmt));
1767

1768
    return TSDB_CODE_SUCCESS;
5✔
1769

1770
  cleanup_root:
×
1771
    if (pStmt->sql.pQuery->pRoot) {
×
1772
      nodesDestroyNode(pStmt->sql.pQuery->pRoot);
×
1773
      pStmt->sql.pQuery->pRoot = NULL;
×
1774
    }
1775
    STMT_ERR_RET(code);
×
1776
  }
1777

1778
  if (pStmt->sql.stbInterlaceMode && NULL == pStmt->sql.siInfo.pDataCtx) {
5,864!
1779
    STMT_ERR_RET(stmtInitStbInterlaceTableInfo(pStmt));
×
1780
  }
1781

1782
  STableDataCxt** pDataBlock = NULL;
5,867✔
1783

1784
  if (pStmt->exec.pCurrBlock) {
5,867✔
1785
    pDataBlock = &pStmt->exec.pCurrBlock;
5,756✔
1786
  } else {
1787
    pDataBlock =
1788
        (STableDataCxt**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
111✔
1789
    if (NULL == pDataBlock) {
111!
1790
      tscError("table %s not found in exec blockHash", pStmt->bInfo.tbFName);
×
1791
      STMT_ERR_RET(TSDB_CODE_TSC_STMT_CACHE_ERROR);
×
1792
    }
1793
    pStmt->exec.pCurrBlock = *pDataBlock;
111✔
1794
    if (pStmt->sql.stbInterlaceMode) {
111✔
1795
      taosArrayDestroy(pStmt->exec.pCurrBlock->pData->aCol);
89✔
1796
      (*pDataBlock)->pData->aCol = NULL;
89✔
1797
    }
1798
    if (colIdx < -1) {
111✔
1799
      pStmt->sql.bindRowFormat = true;
1✔
1800
      taosArrayDestroy((*pDataBlock)->pData->aCol);
1✔
1801
      (*pDataBlock)->pData->aCol = taosArrayInit(20, POINTER_BYTES);
1✔
1802
    }
1803
  }
1804

1805
  int64_t startUs2 = taosGetTimestampUs();
5,862✔
1806
  pStmt->stat.bindDataUs1 += startUs2 - startUs;
5,862✔
1807

1808
  SStmtQNode* param = NULL;
5,862✔
1809
  if (pStmt->sql.stbInterlaceMode) {
5,862✔
1810
    STMT_ERR_RET(stmtAllocQNodeFromBuf(&pStmt->sql.siInfo.tbBuf, (void**)&param));
11,500!
1811
    STMT_ERR_RET(stmtGetTableColsFromCache(pStmt, &param->tblData.aCol));
11,499!
1812
    taosArrayClear(param->tblData.aCol);
5,749✔
1813

1814
    // param->tblData.aCol = taosArrayInit(20, POINTER_BYTES);
1815

1816
    param->restoreTbCols = false;
5,748✔
1817
    param->tblData.isOrdered = true;
5,748✔
1818
    param->tblData.isDuplicateTs = false;
5,748✔
1819
    tstrncpy(param->tblData.tbName, pStmt->bInfo.tbName, TSDB_TABLE_NAME_LEN);
5,748✔
1820

1821
    param->pCreateTbReq = pCreateTbReq;
5,748✔
1822
  }
1823

1824
  int64_t startUs3 = taosGetTimestampUs();
5,869✔
1825
  pStmt->stat.bindDataUs2 += startUs3 - startUs2;
5,869✔
1826

1827
  SArray* pCols = pStmt->sql.stbInterlaceMode ? param->tblData.aCol : (*pDataBlock)->pData->aCol;
5,869✔
1828

1829
  if (colIdx < 0) {
5,869✔
1830
    if (pStmt->sql.stbInterlaceMode) {
5,864✔
1831
      // (*pDataBlock)->pData->flags = 0;
1832
      (*pDataBlock)->pData->flags &= ~SUBMIT_REQ_COLUMN_DATA_FORMAT;
5,753✔
1833
      code = qBindStmtStbColsValue2(*pDataBlock, pCols, bind, pStmt->exec.pRequest->msgBuf,
5,753✔
1834
                                    pStmt->exec.pRequest->msgBufLen, &pStmt->sql.siInfo.pTSchema, pStmt->sql.pBindInfo,
5,753✔
1835
                                    pStmt->taos->optionInfo.charsetCxt);
5,753✔
1836
      param->tblData.isOrdered = (*pDataBlock)->ordered;
5,754✔
1837
      param->tblData.isDuplicateTs = (*pDataBlock)->duplicateTs;
5,754✔
1838
    } else {
1839
      if (colIdx == -1) {
111✔
1840
        if (pStmt->sql.bindRowFormat) {
109✔
1841
          tscError("can't mix bind row format and bind column format");
1!
1842
          STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR);
1!
1843
        }
1844
        code = qBindStmtColsValue2(*pDataBlock, pCols, bind, pStmt->exec.pRequest->msgBuf,
108✔
1845
                                   pStmt->exec.pRequest->msgBufLen, pStmt->taos->optionInfo.charsetCxt);
108✔
1846
      } else {
1847
        code = qBindStmt2RowValue(*pDataBlock, (*pDataBlock)->pData->aRowP, bind, pStmt->exec.pRequest->msgBuf,
2✔
1848
                                  pStmt->exec.pRequest->msgBufLen, &pStmt->sql.siInfo.pTSchema, pStmt->sql.pBindInfo,
2✔
1849
                                  pStmt->taos->optionInfo.charsetCxt);
2✔
1850
      }
1851
    }
1852

1853
    if (code) {
5,864✔
1854
      tscError("qBindStmtColsValue failed, error:%s", tstrerror(code));
1!
1855
      STMT_ERR_RET(code);
1!
1856
    }
1857
  } else {
1858
    if (pStmt->sql.stbInterlaceMode) {
6!
1859
      tscError("bind single column not allowed in stb insert mode");
×
1860
      STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR);
×
1861
    }
1862

1863
    if (pStmt->sql.bindRowFormat) {
6!
1864
      tscError("can't mix bind row format and bind column format");
×
1865
      STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR);
×
1866
    }
1867

1868
    if (colIdx != (pStmt->bInfo.sBindLastIdx + 1) && colIdx != 0) {
6!
1869
      tscError("bind column index not in sequence");
×
1870
      STMT_ERR_RET(TSDB_CODE_APP_ERROR);
×
1871
    }
1872

1873
    pStmt->bInfo.sBindLastIdx = colIdx;
6✔
1874

1875
    if (0 == colIdx) {
6✔
1876
      pStmt->bInfo.sBindRowNum = bind->num;
3✔
1877
    }
1878

1879
    code = qBindStmtSingleColValue2(*pDataBlock, pCols, bind, pStmt->exec.pRequest->msgBuf,
6✔
1880
                                    pStmt->exec.pRequest->msgBufLen, colIdx, pStmt->bInfo.sBindRowNum,
6✔
1881
                                    pStmt->taos->optionInfo.charsetCxt);
6✔
1882
    if (code) {
6!
1883
      tscError("qBindStmtSingleColValue failed, error:%s", tstrerror(code));
×
1884
      STMT_ERR_RET(code);
×
1885
    }
1886
  }
1887

1888
  int64_t startUs4 = taosGetTimestampUs();
5,869✔
1889
  pStmt->stat.bindDataUs3 += startUs4 - startUs3;
5,869✔
1890

1891
  if (pStmt->sql.stbInterlaceMode) {
5,869✔
1892
    STMT_ERR_RET(stmtAppendTablePostHandle(pStmt, param));
5,753!
1893
  } else {
1894
    STMT_ERR_RET(stmtAddBatch2(pStmt));
116!
1895
  }
1896

1897
  pStmt->stat.bindDataUs4 += taosGetTimestampUs() - startUs4;
5,880✔
1898

1899
  return TSDB_CODE_SUCCESS;
5,880✔
1900
}
1901
/*
1902
int stmtUpdateTableUid(STscStmt2* pStmt, SSubmitRsp* pRsp) {
1903
  tscDebug("stmt start to update tbUid, blockNum:%d", pRsp->nBlocks);
1904

1905
  int32_t code = 0;
1906
  int32_t finalCode = 0;
1907
  size_t  keyLen = 0;
1908
  void*   pIter = taosHashIterate(pStmt->exec.pBlockHash, NULL);
1909
  while (pIter) {
1910
    STableDataCxt* pBlock = *(STableDataCxt**)pIter;
1911
    char*          key = taosHashGetKey(pIter, &keyLen);
1912

1913
    STableMeta* pMeta = qGetTableMetaInDataBlock(pBlock);
1914
    if (pMeta->uid) {
1915
      pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
1916
      continue;
1917
    }
1918

1919
    SSubmitBlkRsp* blkRsp = NULL;
1920
    int32_t        i = 0;
1921
    for (; i < pRsp->nBlocks; ++i) {
1922
      blkRsp = pRsp->pBlocks + i;
1923
      if (strlen(blkRsp->tblFName) != keyLen) {
1924
        continue;
1925
      }
1926

1927
      if (strncmp(blkRsp->tblFName, key, keyLen)) {
1928
        continue;
1929
      }
1930

1931
      break;
1932
    }
1933

1934
    if (i < pRsp->nBlocks) {
1935
      tscDebug("auto created table %s uid updated from %" PRIx64 " to %" PRIx64, blkRsp->tblFName, pMeta->uid,
1936
               blkRsp->uid);
1937

1938
      pMeta->uid = blkRsp->uid;
1939
      pStmt->bInfo.tbUid = blkRsp->uid;
1940
    } else {
1941
      tscDebug("table %s not found in submit rsp, will update from catalog", pStmt->bInfo.tbFName);
1942
      if (NULL == pStmt->pCatalog) {
1943
        code = catalogGetHandle(pStmt->taos->pAppInfo->clusterId, &pStmt->pCatalog);
1944
        if (code) {
1945
          pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
1946
          finalCode = code;
1947
          continue;
1948
        }
1949
      }
1950

1951
      code = stmtCreateRequest(pStmt);
1952
      if (code) {
1953
        pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
1954
        finalCode = code;
1955
        continue;
1956
      }
1957

1958
      STableMeta*      pTableMeta = NULL;
1959
      SRequestConnInfo conn = {.pTrans = pStmt->taos->pAppInfo->pTransporter,
1960
                               .requestId = pStmt->exec.pRequest->requestId,
1961
                               .requestObjRefId = pStmt->exec.pRequest->self,
1962
                               .mgmtEps = getEpSet_s(&pStmt->taos->pAppInfo->mgmtEp)};
1963
      code = catalogGetTableMeta(pStmt->pCatalog, &conn, &pStmt->bInfo.sname, &pTableMeta);
1964

1965
      pStmt->stat.ctgGetTbMetaNum++;
1966

1967
      taos_free_result(pStmt->exec.pRequest);
1968
      pStmt->exec.pRequest = NULL;
1969

1970
      if (code || NULL == pTableMeta) {
1971
        pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
1972
        finalCode = code;
1973
        taosMemoryFree(pTableMeta);
1974
        continue;
1975
      }
1976

1977
      pMeta->uid = pTableMeta->uid;
1978
      pStmt->bInfo.tbUid = pTableMeta->uid;
1979
      taosMemoryFree(pTableMeta);
1980
    }
1981

1982
    pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
1983
  }
1984

1985
  return finalCode;
1986
}
1987
*/
1988
/*
1989
int stmtStaticModeExec(TAOS_STMT* stmt) {
1990
  STscStmt2*   pStmt = (STscStmt2*)stmt;
1991
  int32_t     code = 0;
1992
  SSubmitRsp* pRsp = NULL;
1993
  if (pStmt->sql.staticMode) {
1994
    return TSDB_CODE_TSC_STMT_API_ERROR;
1995
  }
1996

1997
  STMT_DLOG_E("start to exec");
1998

1999
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_EXECUTE));
2000

2001
  STMT_ERR_RET(qBuildStmtOutputFromTbList(pStmt->sql.pQuery, pStmt->sql.pVgHash, pStmt->exec.pTbBlkList,
2002
pStmt->exec.pCurrBlock, pStmt->exec.tbBlkNum));
2003

2004
  launchQueryImpl(pStmt->exec.pRequest, pStmt->sql.pQuery, true, NULL);
2005

2006
  if (pStmt->exec.pRequest->code && NEED_CLIENT_HANDLE_ERROR(pStmt->exec.pRequest->code)) {
2007
    code = refreshMeta(pStmt->exec.pRequest->pTscObj, pStmt->exec.pRequest);
2008
    if (code) {
2009
      pStmt->exec.pRequest->code = code;
2010
    } else {
2011
      tFreeSSubmitRsp(pRsp);
2012
      STMT_ERR_RET(stmtResetStmt(pStmt));
2013
      STMT_ERR_RET(TSDB_CODE_NEED_RETRY);
2014
    }
2015
  }
2016

2017
  STMT_ERR_JRET(pStmt->exec.pRequest->code);
2018

2019
  pStmt->exec.affectedRows = taos_affected_rows(pStmt->exec.pRequest);
2020
  pStmt->affectedRows += pStmt->exec.affectedRows;
2021

2022
_return:
2023

2024
  stmtCleanExecInfo(pStmt, (code ? false : true), false);
2025

2026
  tFreeSSubmitRsp(pRsp);
2027

2028
  ++pStmt->sql.runTimes;
2029

2030
  STMT_RET(code);
2031
}
2032
*/
2033

2034
static int32_t createParseContext(const SRequestObj* pRequest, SParseContext** pCxt, SSqlCallbackWrapper* pWrapper) {
12✔
2035
  const STscObj* pTscObj = pRequest->pTscObj;
12✔
2036

2037
  *pCxt = taosMemoryCalloc(1, sizeof(SParseContext));
12!
2038
  if (*pCxt == NULL) {
12!
2039
    return terrno;
×
2040
  }
2041

2042
  **pCxt = (SParseContext){.requestId = pRequest->requestId,
12✔
2043
                           .requestRid = pRequest->self,
12✔
2044
                           .acctId = pTscObj->acctId,
12✔
2045
                           .db = pRequest->pDb,
12✔
2046
                           .topicQuery = false,
2047
                           .pSql = pRequest->sqlstr,
12✔
2048
                           .sqlLen = pRequest->sqlLen,
12✔
2049
                           .pMsg = pRequest->msgBuf,
12✔
2050
                           .msgLen = ERROR_MSG_BUF_DEFAULT_SIZE,
2051
                           .pTransporter = pTscObj->pAppInfo->pTransporter,
12✔
2052
                           .pStmtCb = NULL,
2053
                           .pUser = pTscObj->user,
12✔
2054
                           .pEffectiveUser = pRequest->effectiveUser,
12✔
2055
                           .isSuperUser = (0 == strcmp(pTscObj->user, TSDB_DEFAULT_USER)),
12✔
2056
                           .enableSysInfo = pTscObj->sysInfo,
12✔
2057
                           .async = true,
2058
                           .svrVer = pTscObj->sVer,
12✔
2059
                           .nodeOffline = (pTscObj->pAppInfo->onlineDnodes < pTscObj->pAppInfo->totalDnodes),
12✔
2060
                           .allocatorId = pRequest->allocatorRefId,
12✔
2061
                           .parseSqlFp = clientParseSql,
2062
                           .parseSqlParam = pWrapper};
2063
  int8_t biMode = atomic_load_8(&((STscObj*)pTscObj)->biMode);
12✔
2064
  (*pCxt)->biMode = biMode;
12✔
2065
  return TSDB_CODE_SUCCESS;
12✔
2066
}
2067

2068
static void asyncQueryCb(void* userdata, TAOS_RES* res, int code) {
12✔
2069
  STscStmt2*        pStmt = userdata;
12✔
2070
  __taos_async_fn_t fp = pStmt->options.asyncExecFn;
12✔
2071

2072
  pStmt->exec.affectedRows = taos_affected_rows(pStmt->exec.pRequest);
12✔
2073
  pStmt->affectedRows += pStmt->exec.affectedRows;
12✔
2074

2075
  fp(pStmt->options.userdata, res, code);
12✔
2076

2077
  while (0 == atomic_load_8((int8_t*)&pStmt->sql.siInfo.tableColsReady)) {
12!
2078
    taosUsleep(1);
×
2079
  }
2080
  (void)stmtCleanExecInfo(pStmt, (code ? false : true), false);
12✔
2081
  ++pStmt->sql.runTimes;
12✔
2082

2083
  if (tsem_post(&pStmt->asyncExecSem) != 0) {
12!
2084
    tscError("failed to post asyncExecSem");
×
2085
  }
2086
}
12✔
2087

2088
int stmtExec2(TAOS_STMT2* stmt, int* affected_rows) {
4,477✔
2089
  STscStmt2* pStmt = (STscStmt2*)stmt;
4,477✔
2090
  int32_t    code = 0;
4,477✔
2091
  int64_t    startUs = taosGetTimestampUs();
4,479✔
2092

2093
  STMT_DLOG_E("start to exec");
4,479!
2094

2095
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
4,479!
2096
    return pStmt->errCode;
×
2097
  }
2098

2099
  TSC_ERR_RET(taosThreadMutexLock(&pStmt->asyncBindParam.mutex));
4,479!
2100
  while (atomic_load_8((int8_t*)&pStmt->asyncBindParam.asyncBindNum) > 0) {
4,481!
2101
    (void)taosThreadCondWait(&pStmt->asyncBindParam.waitCond, &pStmt->asyncBindParam.mutex);
×
2102
  }
2103
  TSC_ERR_RET(taosThreadMutexUnlock(&pStmt->asyncBindParam.mutex));
4,476!
2104

2105
  if (pStmt->sql.stbInterlaceMode) {
4,480✔
2106
    STMT_ERR_RET(stmtAddBatch2(pStmt));
4,427!
2107
  }
2108

2109
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_EXECUTE));
4,480✔
2110

2111
  if (STMT_TYPE_QUERY != pStmt->sql.type) {
4,478✔
2112
    if (pStmt->sql.stbInterlaceMode) {
4,473✔
2113
      int64_t startTs = taosGetTimestampUs();
4,427✔
2114
      while (atomic_load_64(&pStmt->sql.siInfo.tbRemainNum)) {
9,089✔
2115
        taosUsleep(1);
4,671✔
2116
      }
2117
      pStmt->stat.execWaitUs += taosGetTimestampUs() - startTs;
4,423✔
2118
      STMT_ERR_RET(qBuildStmtFinOutput(pStmt->sql.pQuery, pStmt->sql.pVgHash, pStmt->sql.siInfo.pVgroupList));
4,423!
2119
      taosHashCleanup(pStmt->sql.siInfo.pVgroupHash);
4,427✔
2120
      pStmt->sql.siInfo.pVgroupHash = NULL;
4,427✔
2121
      pStmt->sql.siInfo.pVgroupList = NULL;
4,427✔
2122
    } else {
2123
      tDestroySubmitTbData(pStmt->exec.pCurrTbData, TSDB_MSG_FLG_ENCODE);
47✔
2124
      taosMemoryFreeClear(pStmt->exec.pCurrTbData);
47!
2125

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

2128
      STMT_ERR_RET(qBuildStmtOutput(pStmt->sql.pQuery, pStmt->sql.pVgHash, pStmt->exec.pBlockHash));
47!
2129
    }
2130
  }
2131

2132
  SRequestObj*      pRequest = pStmt->exec.pRequest;
4,476✔
2133
  __taos_async_fn_t fp = pStmt->options.asyncExecFn;
4,476✔
2134

2135
  if (!fp) {
4,476✔
2136
    launchQueryImpl(pStmt->exec.pRequest, pStmt->sql.pQuery, true, NULL);
4,464✔
2137

2138
    if (pStmt->exec.pRequest->code && NEED_CLIENT_HANDLE_ERROR(pStmt->exec.pRequest->code)) {
4,464!
2139
      code = refreshMeta(pStmt->exec.pRequest->pTscObj, pStmt->exec.pRequest);
×
2140
      if (code) {
×
2141
        pStmt->exec.pRequest->code = code;
×
2142
      } else {
2143
        STMT_ERR_RET(stmtResetStmt(pStmt));
×
2144
        STMT_ERR_RET(TSDB_CODE_NEED_RETRY);
×
2145
      }
2146
    }
2147

2148
    STMT_ERR_JRET(pStmt->exec.pRequest->code);
4,466!
2149

2150
    pStmt->exec.affectedRows = taos_affected_rows(pStmt->exec.pRequest);
4,466✔
2151
    if (affected_rows) {
4,463✔
2152
      *affected_rows = pStmt->exec.affectedRows;
4,458✔
2153
    }
2154
    pStmt->affectedRows += pStmt->exec.affectedRows;
4,463✔
2155

2156
    while (0 == atomic_load_8((int8_t*)&pStmt->sql.siInfo.tableColsReady)) {
4,463!
2157
      taosUsleep(1);
×
2158
    }
2159

2160
    STMT_ERR_RET(stmtCleanExecInfo(pStmt, (code ? false : true), false));
4,464✔
2161

2162
    ++pStmt->sql.runTimes;
4,448✔
2163
  } else {
2164
    SSqlCallbackWrapper* pWrapper = taosMemoryCalloc(1, sizeof(SSqlCallbackWrapper));
12!
2165
    if (pWrapper == NULL) {
12!
2166
      code = terrno;
×
2167
    } else {
2168
      pWrapper->pRequest = pRequest;
12✔
2169
      pRequest->pWrapper = pWrapper;
12✔
2170
    }
2171
    if (TSDB_CODE_SUCCESS == code) {
12!
2172
      code = createParseContext(pRequest, &pWrapper->pParseCtx, pWrapper);
12✔
2173
    }
2174
    pRequest->syncQuery = false;
12✔
2175
    pRequest->body.queryFp = asyncQueryCb;
12✔
2176
    ((SSyncQueryParam*)(pRequest)->body.interParam)->userParam = pStmt;
12✔
2177

2178
    pStmt->execSemWaited = false;
12✔
2179
    launchAsyncQuery(pRequest, pStmt->sql.pQuery, NULL, pWrapper);
12✔
2180
  }
2181

2182
_return:
4,460✔
2183
  pStmt->stat.execUseUs += taosGetTimestampUs() - startUs;
4,463✔
2184

2185
  STMT_RET(code);
4,463!
2186
}
2187

2188
int stmtClose2(TAOS_STMT2* stmt) {
147✔
2189
  STscStmt2* pStmt = (STscStmt2*)stmt;
147✔
2190

2191
  STMT_DLOG_E("start to free stmt");
147!
2192
  taosMemoryFreeClear(pStmt->db);
147!
2193

2194
  if (pStmt->bindThreadInUse) {
147✔
2195
    while (0 == atomic_load_8((int8_t*)&pStmt->sql.siInfo.tableColsReady)) {
83!
2196
      taosUsleep(1);
×
2197
    }
2198
    (void)taosThreadMutexLock(&pStmt->queue.mutex);
83✔
2199
    pStmt->queue.stopQueue = true;
83✔
2200
    (void)taosThreadCondSignal(&(pStmt->queue.waitCond));
83✔
2201
    (void)taosThreadMutexUnlock(&pStmt->queue.mutex);
83✔
2202

2203
    (void)taosThreadJoin(pStmt->bindThread, NULL);
83✔
2204
    pStmt->bindThreadInUse = false;
83✔
2205

2206
    (void)taosThreadCondDestroy(&pStmt->queue.waitCond);
83✔
2207
    (void)taosThreadMutexDestroy(&pStmt->queue.mutex);
83✔
2208
  }
2209

2210
  TSC_ERR_RET(taosThreadMutexLock(&pStmt->asyncBindParam.mutex));
147!
2211
  while (atomic_load_8((int8_t*)&pStmt->asyncBindParam.asyncBindNum) > 0) {
147!
2212
    (void)taosThreadCondWait(&pStmt->asyncBindParam.waitCond, &pStmt->asyncBindParam.mutex);
×
2213
  }
2214
  TSC_ERR_RET(taosThreadMutexUnlock(&pStmt->asyncBindParam.mutex));
147!
2215

2216
  (void)taosThreadCondDestroy(&pStmt->asyncBindParam.waitCond);
147✔
2217
  (void)taosThreadMutexDestroy(&pStmt->asyncBindParam.mutex);
147✔
2218

2219
  if (pStmt->options.asyncExecFn && !pStmt->execSemWaited) {
147!
2220
    if (tsem_wait(&pStmt->asyncExecSem) != 0) {
5!
2221
      tscError("failed to wait asyncExecSem");
×
2222
    }
2223
  }
2224

2225
  STMT_DLOG("stmt %p closed, stbInterlaceMode:%d, statInfo: ctgGetTbMetaNum=>%" PRId64 ", getCacheTbInfo=>%" PRId64
147!
2226
            ", parseSqlNum=>%" PRId64 ", pStmt->stat.bindDataNum=>%" PRId64
2227
            ", settbnameAPI:%u, bindAPI:%u, addbatchAPI:%u, execAPI:%u"
2228
            ", setTbNameUs:%" PRId64 ", bindDataUs:%" PRId64 ",%" PRId64 ",%" PRId64 ",%" PRId64 " addBatchUs:%" PRId64
2229
            ", execWaitUs:%" PRId64 ", execUseUs:%" PRId64,
2230
            pStmt, pStmt->sql.stbInterlaceMode, pStmt->stat.ctgGetTbMetaNum, pStmt->stat.getCacheTbInfo,
2231
            pStmt->stat.parseSqlNum, pStmt->stat.bindDataNum, pStmt->seqIds[STMT_SETTBNAME], pStmt->seqIds[STMT_BIND],
2232
            pStmt->seqIds[STMT_ADD_BATCH], pStmt->seqIds[STMT_EXECUTE], pStmt->stat.setTbNameUs,
2233
            pStmt->stat.bindDataUs1, pStmt->stat.bindDataUs2, pStmt->stat.bindDataUs3, pStmt->stat.bindDataUs4,
2234
            pStmt->stat.addBatchUs, pStmt->stat.execWaitUs, pStmt->stat.execUseUs);
2235
  if (pStmt->sql.stbInterlaceMode) {
147✔
2236
    pStmt->bInfo.tagsCached = false;
77✔
2237
  }
2238

2239
  STMT_ERR_RET(stmtCleanSQLInfo(pStmt));
147!
2240

2241
  if (pStmt->options.asyncExecFn) {
147✔
2242
    if (tsem_destroy(&pStmt->asyncExecSem) != 0) {
5!
2243
      tscError("failed to destroy asyncExecSem");
×
2244
    }
2245
  }
2246
  taosMemoryFree(stmt);
147!
2247

2248
  return TSDB_CODE_SUCCESS;
147✔
2249
}
2250

2251
const char* stmtErrstr2(TAOS_STMT2* stmt) {
3✔
2252
  STscStmt2* pStmt = (STscStmt2*)stmt;
3✔
2253

2254
  if (stmt == NULL || NULL == pStmt->exec.pRequest) {
3!
2255
    return (char*)tstrerror(terrno);
3✔
2256
  }
2257

2258
  pStmt->exec.pRequest->code = terrno;
×
2259

2260
  SRequestObj* pRequest = pStmt->exec.pRequest;
×
2261
  if (NULL != pRequest->msgBuf && (strlen(pRequest->msgBuf) > 0 || pRequest->code == TSDB_CODE_RPC_FQDN_ERROR)) {
×
2262
    return pRequest->msgBuf;
×
2263
  }
2264
  return (const char*)tstrerror(pRequest->code);
×
2265
}
2266
/*
2267
int stmtAffectedRows(TAOS_STMT* stmt) { return ((STscStmt2*)stmt)->affectedRows; }
2268

2269
int stmtAffectedRowsOnce(TAOS_STMT* stmt) { return ((STscStmt2*)stmt)->exec.affectedRows; }
2270
*/
2271

2272
int stmtParseColFields2(TAOS_STMT2* stmt) {
57✔
2273
  int32_t    code = 0;
57✔
2274
  STscStmt2* pStmt = (STscStmt2*)stmt;
57✔
2275
  int32_t    preCode = pStmt->errCode;
57✔
2276

2277
  STMT_DLOG_E("start to get col fields");
57!
2278

2279
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
57!
2280
    return pStmt->errCode;
×
2281
  }
2282

2283
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
57!
2284
    STMT_ERRI_JRET(TSDB_CODE_TSC_STMT_API_ERROR);
×
2285
  }
2286

2287
  STMT_ERRI_JRET(stmtSwitchStatus(pStmt, STMT_FETCH_FIELDS));
57!
2288

2289
  if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 &&
57!
2290
      STMT_TYPE_MULTI_INSERT != pStmt->sql.type) {
3!
2291
    pStmt->bInfo.needParse = false;
×
2292
  }
2293
  if (pStmt->sql.stbInterlaceMode && pStmt->sql.siInfo.pDataCtx != NULL) {
57✔
2294
    pStmt->bInfo.needParse = false;
7✔
2295
  }
2296
  if (pStmt->exec.pRequest && STMT_TYPE_QUERY == pStmt->sql.type && pStmt->sql.runTimes) {
57!
2297
    taos_free_result(pStmt->exec.pRequest);
×
2298
    pStmt->exec.pRequest = NULL;
×
2299
    STMT_ERRI_JRET(stmtCreateRequest(pStmt));
×
2300
  }
2301

2302
  STMT_ERRI_JRET(stmtCreateRequest(pStmt));
57!
2303

2304
  if (pStmt->bInfo.needParse) {
57✔
2305
    STMT_ERRI_JRET(stmtParseSql(pStmt));
50✔
2306
  }
2307

2308
_return:
42✔
2309
  // compatible with previous versions
2310
  if (code == TSDB_CODE_PAR_TABLE_NOT_EXIST && (pStmt->bInfo.tbNameFlag & NO_DATA_USING_CLAUSE) == 0x0) {
57!
2311
    code = TSDB_CODE_TSC_STMT_TBNAME_ERROR;
1✔
2312
  }
2313

2314
  if (code != TSDB_CODE_SUCCESS) {
57✔
2315
    taos_free_result(pStmt->exec.pRequest);
15✔
2316
    pStmt->exec.pRequest = NULL;
15✔
2317
  }
2318

2319
  pStmt->errCode = preCode;
57✔
2320

2321
  return code;
57✔
2322
}
2323

2324
int stmtGetStbColFields2(TAOS_STMT2* stmt, int* nums, TAOS_FIELD_ALL** fields) {
57✔
2325
  int32_t code = stmtParseColFields2(stmt);
57✔
2326
  if (code != TSDB_CODE_SUCCESS) {
57✔
2327
    return code;
15✔
2328
  }
2329

2330
  return stmtFetchStbColFields2(stmt, nums, fields);
42✔
2331
}
2332

2333
int stmtGetParamNum2(TAOS_STMT2* stmt, int* nums) {
11✔
2334
  int32_t    code = 0;
11✔
2335
  STscStmt2* pStmt = (STscStmt2*)stmt;
11✔
2336
  int32_t    preCode = pStmt->errCode;
11✔
2337

2338
  STMT_DLOG_E("start to get param num");
11!
2339

2340
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
11!
2341
    return pStmt->errCode;
×
2342
  }
2343

2344
  STMT_ERRI_JRET(stmtSwitchStatus(pStmt, STMT_FETCH_FIELDS));
11!
2345

2346
  if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 &&
11!
2347
      STMT_TYPE_MULTI_INSERT != pStmt->sql.type) {
×
2348
    pStmt->bInfo.needParse = false;
×
2349
  }
2350

2351
  if (pStmt->exec.pRequest && STMT_TYPE_QUERY == pStmt->sql.type && pStmt->sql.runTimes) {
11!
2352
    taos_free_result(pStmt->exec.pRequest);
×
2353
    pStmt->exec.pRequest = NULL;
×
2354
  }
2355

2356
  STMT_ERRI_JRET(stmtCreateRequest(pStmt));
11!
2357

2358
  if (pStmt->bInfo.needParse) {
11!
2359
    STMT_ERRI_JRET(stmtParseSql(pStmt));
11✔
2360
  }
2361

2362
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
4!
2363
    *nums = taosArrayGetSize(pStmt->sql.pQuery->pPlaceholderValues);
4✔
2364
  } else {
2365
    STMT_ERRI_JRET(stmtFetchColFields2(stmt, nums, NULL));
×
2366
  }
2367

2368
_return:
×
2369
  if (code != TSDB_CODE_SUCCESS) {
11✔
2370
    taos_free_result(pStmt->exec.pRequest);
7✔
2371
    pStmt->exec.pRequest = NULL;
7✔
2372
  }
2373
  pStmt->errCode = preCode;
11✔
2374

2375
  return code;
11✔
2376
}
2377

2378
TAOS_RES* stmtUseResult2(TAOS_STMT2* stmt) {
5✔
2379
  STscStmt2* pStmt = (STscStmt2*)stmt;
5✔
2380

2381
  STMT_DLOG_E("start to use result");
5!
2382

2383
  if (STMT_TYPE_QUERY != pStmt->sql.type) {
5!
2384
    tscError("useResult only for query statement");
×
2385
    return NULL;
×
2386
  }
2387

2388
  return pStmt->exec.pRequest;
5✔
2389
}
2390

2391
int32_t stmtAsyncBindThreadFunc(void* args) {
×
2392
  qInfo("async stmt bind thread started");
×
2393

2394
  ThreadArgs* targs = (ThreadArgs*)args;
×
2395
  STscStmt2*  pStmt = (STscStmt2*)targs->stmt;
×
2396

2397
  int code = taos_stmt2_bind_param(targs->stmt, targs->bindv, targs->col_idx);
×
2398
  targs->fp(targs->param, NULL, code);
×
2399
  (void)taosThreadMutexLock(&(pStmt->asyncBindParam.mutex));
×
2400
  (void)atomic_sub_fetch_8(&pStmt->asyncBindParam.asyncBindNum, 1);
×
2401
  (void)taosThreadCondSignal(&(pStmt->asyncBindParam.waitCond));
×
2402
  (void)taosThreadMutexUnlock(&(pStmt->asyncBindParam.mutex));
×
2403
  taosMemoryFree(args);
×
2404

2405
  qInfo("async stmt bind thread stopped");
×
2406

2407
  return code;
×
2408
}
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