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

taosdata / TDengine / #4103

17 May 2025 02:18AM UTC coverage: 63.264% (+0.4%) from 62.905%
#4103

push

travis-ci

web-flow
Merge pull request #31110 from taosdata/3.0

merge 3.0

158149 of 318142 branches covered (49.71%)

Branch coverage included in aggregate %.

3 of 5 new or added lines in 1 file covered. (60.0%)

1725 existing lines in 138 files now uncovered.

243642 of 316962 relevant lines covered (76.87%)

16346281.8 hits per line

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

65.87
/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) {
9,575✔
13
    *pBuf = (char*)pTblBuf->pCurBuff + pTblBuf->buffOffset;
9,574✔
14
    pTblBuf->buffOffset += pTblBuf->buffUnit;
9,574✔
15
  } else if (pTblBuf->buffIdx < taosArrayGetSize(pTblBuf->pBufList)) {
1!
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;
9,574✔
39
}
40

41
static bool stmtDequeue(STscStmt2* pStmt, SStmtQNode** param) {
9,581✔
42
  int i = 0;
9,581✔
43
  while (0 == atomic_load_64((int64_t*)&pStmt->queue.qRemainNum)) {
52,966✔
44
    if (pStmt->queue.stopQueue) {
43,446✔
45
      STMT2_DLOG_E("stmt stopQueue,but remainNum is 0");
96!
46
      return false;
96✔
47
    }
48
    if (i < 10) {
43,350✔
49
      taosUsleep(1);
41,478✔
50
      i++;
41,453✔
51
    } else {
52
      (void)taosThreadMutexLock(&pStmt->queue.mutex);
1,872✔
53
      if (pStmt->queue.stopQueue) {
1,932!
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)) {
1,932✔
59
        (void)taosThreadCondWait(&pStmt->queue.waitCond, &pStmt->queue.mutex);
1,925✔
60
      }
61
      (void)taosThreadMutexUnlock(&pStmt->queue.mutex);
1,931✔
62
    }
63
  }
64

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

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

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

85
  (void)atomic_sub_fetch_64((int64_t*)&pStmt->queue.qRemainNum, 1);
9,483✔
86
  (void)taosThreadMutexUnlock(&pStmt->queue.mutex);
9,487✔
87

88
  return true;
9,488✔
89
}
90

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

96
  param->next = NULL;
9,476✔
97

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

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

104
  (void)atomic_add_fetch_64(&pStmt->queue.qRemainNum, 1);
9,481✔
105
  (void)taosThreadCondSignal(&(pStmt->queue.waitCond));
9,486✔
106

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

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

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

130
  return code;
6,019✔
131
}
132

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

136
  if (newStatus >= STMT_INIT && newStatus < STMT_MAX) {
19,698!
137
    STMT_LOG_SEQ(newStatus);
19,704!
138
  }
139

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

145
  switch (newStatus) {
19,737!
146
    case STMT_PREPARE:
163✔
147
      pStmt->errCode = 0;
163✔
148
      break;
163✔
149
    case STMT_SETTBNAME:
5,364✔
150
      if (STMT_STATUS_EQ(INIT)) {
5,364!
UNCOV
151
        code = TSDB_CODE_TSC_STMT_API_ERROR;
×
152
      }
153
      if (!pStmt->sql.stbInterlaceMode && (STMT_STATUS_EQ(BIND) || STMT_STATUS_EQ(BIND_COL))) {
5,364!
UNCOV
154
        code = TSDB_CODE_TSC_STMT_API_ERROR;
×
155
      }
156
      break;
5,364✔
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,376✔
168
      if (STMT_STATUS_EQ(INIT) || STMT_STATUS_EQ(BIND_COL)) {
5,376!
UNCOV
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,376✔
UNCOV
177
    case STMT_BIND_COL:
×
UNCOV
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,341✔
183
      if (STMT_STATUS_NE(BIND) && STMT_STATUS_NE(BIND_COL) && STMT_STATUS_NE(FETCH_FIELDS)) {
4,341!
UNCOV
184
        code = TSDB_CODE_TSC_STMT_API_ERROR;
×
185
      }
186
      break;
4,341✔
187
    case STMT_EXECUTE:
4,279✔
188
      if (STMT_TYPE_QUERY == pStmt->sql.type) {
4,279✔
189
        if (STMT_STATUS_NE(ADD_BATCH) && STMT_STATUS_NE(FETCH_FIELDS) && STMT_STATUS_NE(BIND) &&
5!
UNCOV
190
            STMT_STATUS_NE(BIND_COL)) {
×
UNCOV
191
          code = TSDB_CODE_TSC_STMT_API_ERROR;
×
192
        }
193
      } else {
194
        if (STMT_STATUS_NE(ADD_BATCH) && STMT_STATUS_NE(FETCH_FIELDS)) {
4,274!
195
          code = TSDB_CODE_TSC_STMT_API_ERROR;
1✔
196
        }
197
      }
198
      break;
4,279✔
UNCOV
199
    default:
×
UNCOV
200
      code = TSDB_CODE_APP_ERROR;
×
UNCOV
201
      break;
×
202
  }
203

204
  STMT_ERR_RET(code);
19,737✔
205

206
  pStmt->sql.status = newStatus;
19,736✔
207

208
  return TSDB_CODE_SUCCESS;
19,736✔
209
}
210

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

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

216
  if ('\0' == pStmt->bInfo.tbName[0]) {
136✔
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;
103✔
222

223
  return TSDB_CODE_SUCCESS;
103✔
224
}
225

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

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

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

244
  if (!pStmt->bInfo.tagsCached) {
143✔
245
    qDestroyBoundColInfo(pStmt->bInfo.boundTags);
140✔
246
    taosMemoryFreeClear(pStmt->bInfo.boundTags);
138!
247
  }
248

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

254
  return TSDB_CODE_SUCCESS;
139✔
255
}
256

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

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

263
  return TSDB_CODE_SUCCESS;
138✔
264
}
265

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

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

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

275
  return TSDB_CODE_SUCCESS;
137✔
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) {
171✔
291
  pStmt->exec.pCurrBlock = NULL;
171✔
292

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

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

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

306
  pStmt->bInfo.needParse = false;
150✔
307

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

315
    return TSDB_CODE_SUCCESS;
6✔
316
  }
317

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

324
  STableDataCxt* pTableCtx = *pSrc;
144✔
325
  // if (pStmt->sql.stbInterlaceMode) {
326
  //   int16_t lastIdx = -1;
327

328
  //   for (int32_t i = 0; i < pTableCtx->boundColsInfo.numOfBound; ++i) {
329
  //     if (pTableCtx->boundColsInfo.pColIndex[i] < lastIdx) {
330
  //       pStmt->sql.stbInterlaceMode = false;
331
  //       break;
332
  //     }
333

334
  //     lastIdx = pTableCtx->boundColsInfo.pColIndex[i];
335
  //   }
336
  // }
337

338
  if (NULL == pStmt->sql.pBindInfo) {
144✔
339
    pStmt->sql.pBindInfo = taosMemoryMalloc(pTableCtx->boundColsInfo.numOfBound * sizeof(*pStmt->sql.pBindInfo));
133!
340
    if (NULL == pStmt->sql.pBindInfo) {
133!
UNCOV
341
      return terrno;
×
342
    }
343
  }
344

345
  return TSDB_CODE_SUCCESS;
144✔
346
}
347

348
static int32_t stmtCleanBindInfo(STscStmt2* pStmt) {
4,551✔
349
  pStmt->bInfo.tbUid = 0;
4,551✔
350
  pStmt->bInfo.tbVgId = -1;
4,551✔
351
  pStmt->bInfo.tbType = 0;
4,551✔
352
  pStmt->bInfo.needParse = true;
4,551✔
353
  pStmt->bInfo.inExecCache = false;
4,551✔
354

355
  pStmt->bInfo.tbName[0] = 0;
4,551✔
356
  pStmt->bInfo.tbFName[0] = 0;
4,551✔
357
  if (!pStmt->bInfo.tagsCached) {
4,551✔
358
    qDestroyBoundColInfo(pStmt->bInfo.boundTags);
4,479✔
359
    taosMemoryFreeClear(pStmt->bInfo.boundTags);
4,480!
360
  }
361
  if (!pStmt->sql.autoCreateTbl) {
4,552✔
362
    pStmt->bInfo.stbFName[0] = 0;
4,446✔
363
    pStmt->bInfo.tbSuid = 0;
4,446✔
364
  }
365

366
  return TSDB_CODE_SUCCESS;
4,552✔
367
}
368

UNCOV
369
static void stmtFreeTableBlkList(STableColsData* pTb) {
×
UNCOV
370
  (void)qResetStmtColumns(pTb->aCol, true);
×
UNCOV
371
  taosArrayDestroy(pTb->aCol);
×
UNCOV
372
}
×
373

374
static void stmtResetQueueTableBuf(STableBufInfo* pTblBuf, SStmtQueue* pQueue) {
4,215✔
375
  pTblBuf->pCurBuff = taosArrayGetP(pTblBuf->pBufList, 0);
4,215✔
376
  if (NULL == pTblBuf->pCurBuff) {
4,215✔
377
    tscError("QInfo:%p, failed to get buffer from list", pTblBuf);
1!
UNCOV
378
    return;
×
379
  }
380
  pTblBuf->buffIdx = 1;
4,214✔
381
  pTblBuf->buffOffset = sizeof(*pQueue->head);
4,214✔
382

383
  pQueue->head = pQueue->tail = pTblBuf->pCurBuff;
4,214✔
384
  pQueue->qRemainNum = 0;
4,214✔
385
  pQueue->head->next = NULL;
4,214✔
386
}
387

388
static int32_t stmtCleanExecInfo(STscStmt2* pStmt, bool keepTable, bool deepClean) {
4,432✔
389
  if (pStmt->sql.stbInterlaceMode) {
4,432✔
390
    if (deepClean) {
4,306✔
391
      taosHashCleanup(pStmt->exec.pBlockHash);
89✔
392
      pStmt->exec.pBlockHash = NULL;
89✔
393

394
      if (NULL != pStmt->exec.pCurrBlock) {
89✔
395
        taosMemoryFreeClear(pStmt->exec.pCurrBlock->pData);
85!
396
        qDestroyStmtDataBlock(pStmt->exec.pCurrBlock);
85✔
397
        pStmt->exec.pCurrBlock = NULL;
85✔
398
      }
399
      if (STMT_TYPE_QUERY != pStmt->sql.type) {
89!
400
        taos_free_result(pStmt->exec.pRequest);
89✔
401
        pStmt->exec.pRequest = NULL;
89✔
402
      }
403
    } else {
404
      pStmt->sql.siInfo.pTableColsIdx = 0;
4,217✔
405
      stmtResetQueueTableBuf(&pStmt->sql.siInfo.tbBuf, &pStmt->queue);
4,217✔
406
    }
407
    if (NULL != pStmt->exec.pRequest) {
4,305✔
408
      pStmt->exec.pRequest->body.resInfo.numOfRows = 0;
4,217✔
409
    }
410
  } else {
411
    if (STMT_TYPE_QUERY != pStmt->sql.type || deepClean) {
127✔
412
      // if (!pStmt->options.asyncExecFn) {
413
      taos_free_result(pStmt->exec.pRequest);
122✔
414
      pStmt->exec.pRequest = NULL;
122✔
415
      //}
416
    }
417

418
    size_t keyLen = 0;
127✔
419
    void*  pIter = taosHashIterate(pStmt->exec.pBlockHash, NULL);
127✔
420
    while (pIter) {
269✔
421
      STableDataCxt* pBlocks = *(STableDataCxt**)pIter;
142✔
422
      char*          key = taosHashGetKey(pIter, &keyLen);
142✔
423
      STableMeta*    pMeta = qGetTableMetaInDataBlock(pBlocks);
142✔
424

425
      if (keepTable && pBlocks == pStmt->exec.pCurrBlock) {
142✔
426
        TSWAP(pBlocks->pData, pStmt->exec.pCurrTbData);
47✔
427
        STMT_ERR_RET(qResetStmtDataBlock(pBlocks, false));
99!
428

429
        pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
47✔
430
        continue;
47✔
431
      }
432

433
      qDestroyStmtDataBlock(pBlocks);
95✔
434
      STMT_ERR_RET(taosHashRemove(pStmt->exec.pBlockHash, key, keyLen));
95!
435

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

439
    if (keepTable) {
127✔
440
      return TSDB_CODE_SUCCESS;
52✔
441
    }
442

443
    taosHashCleanup(pStmt->exec.pBlockHash);
75✔
444
    pStmt->exec.pBlockHash = NULL;
75✔
445

446
    tDestroySubmitTbData(pStmt->exec.pCurrTbData, TSDB_MSG_FLG_ENCODE);
75✔
447
    taosMemoryFreeClear(pStmt->exec.pCurrTbData);
75!
448
  }
449

450
  STMT_ERR_RET(stmtCleanBindInfo(pStmt));
4,380!
451

452
  return TSDB_CODE_SUCCESS;
4,381✔
453
}
454

455
static void stmtFreeTbBuf(void* buf) {
96✔
456
  void* pBuf = *(void**)buf;
96✔
457
  taosMemoryFree(pBuf);
96!
458
}
96✔
459

460
static void stmtFreeTbCols(void* buf) {
86,000✔
461
  SArray* pCols = *(SArray**)buf;
86,000✔
462
  taosArrayDestroy(pCols);
86,000✔
463
}
86,000✔
464

465
static int32_t stmtCleanSQLInfo(STscStmt2* pStmt) {
143✔
466
  STMT2_DLOG_E("start to free SQL info");
143!
467

468
  taosMemoryFree(pStmt->sql.pBindInfo);
143!
469
  taosMemoryFree(pStmt->sql.queryRes.fields);
143!
470
  taosMemoryFree(pStmt->sql.queryRes.userFields);
143!
471
  taosMemoryFree(pStmt->sql.sqlStr);
143!
472
  qDestroyQuery(pStmt->sql.pQuery);
143✔
473
  taosArrayDestroy(pStmt->sql.nodeList);
143✔
474
  taosHashCleanup(pStmt->sql.pVgHash);
143✔
475
  pStmt->sql.pVgHash = NULL;
143✔
476
  if (pStmt->sql.fixValueTags) {
143✔
477
    tdDestroySVCreateTbReq(pStmt->sql.fixValueTbReq);
5!
478
  }
479

480
  void* pIter = taosHashIterate(pStmt->sql.pTableCache, NULL);
143✔
481
  while (pIter) {
158✔
482
    SStmtTableCache* pCache = (SStmtTableCache*)pIter;
15✔
483

484
    qDestroyStmtDataBlock(pCache->pDataCtx);
15✔
485
    qDestroyBoundColInfo(pCache->boundTags);
15✔
486
    taosMemoryFreeClear(pCache->boundTags);
15!
487

488
    pIter = taosHashIterate(pStmt->sql.pTableCache, pIter);
15✔
489
  }
490
  taosHashCleanup(pStmt->sql.pTableCache);
143✔
491
  pStmt->sql.pTableCache = NULL;
143✔
492

493
  STMT_ERR_RET(stmtCleanExecInfo(pStmt, false, true));
143!
494
  STMT_ERR_RET(stmtCleanBindInfo(pStmt));
143!
495

496
  taos_free_result(pStmt->sql.siInfo.pRequest);
143✔
497
  taosHashCleanup(pStmt->sql.siInfo.pVgroupHash);
143✔
498
  tSimpleHashCleanup(pStmt->sql.siInfo.pTableHash);
143✔
499
  taosArrayDestroyEx(pStmt->sql.siInfo.tbBuf.pBufList, stmtFreeTbBuf);
143✔
500
  taosMemoryFree(pStmt->sql.siInfo.pTSchema);
143!
501
  qDestroyStmtDataBlock(pStmt->sql.siInfo.pDataCtx);
143✔
502
  taosArrayDestroyEx(pStmt->sql.siInfo.pTableCols, stmtFreeTbCols);
143✔
503
  pStmt->sql.siInfo.pTableCols = NULL;
143✔
504

505
  (void)memset(&pStmt->sql, 0, sizeof(pStmt->sql));
143✔
506
  pStmt->sql.siInfo.tableColsReady = true;
143✔
507

508
  STMT_DLOG_E("end to free SQL info");
143!
509

510
  return TSDB_CODE_SUCCESS;
143✔
511
}
512

513
static int32_t stmtTryAddTableVgroupInfo(STscStmt2* pStmt, int32_t* vgId) {
113✔
514
  if (*vgId >= 0 && taosHashGet(pStmt->sql.pVgHash, (const char*)vgId, sizeof(*vgId))) {
113✔
515
    return TSDB_CODE_SUCCESS;
15✔
516
  }
517

518
  SVgroupInfo      vgInfo = {0};
98✔
519
  SRequestConnInfo conn = {.pTrans = pStmt->taos->pAppInfo->pTransporter,
98✔
520
                           .requestId = pStmt->exec.pRequest->requestId,
98✔
521
                           .requestObjRefId = pStmt->exec.pRequest->self,
98✔
522
                           .mgmtEps = getEpSet_s(&pStmt->taos->pAppInfo->mgmtEp)};
98✔
523

524
  int32_t code = catalogGetTableHashVgroup(pStmt->pCatalog, &conn, &pStmt->bInfo.sname, &vgInfo);
98✔
525
  if (TSDB_CODE_SUCCESS != code) {
98!
UNCOV
526
    return code;
×
527
  }
528

529
  code =
530
      taosHashPut(pStmt->sql.pVgHash, (const char*)&vgInfo.vgId, sizeof(vgInfo.vgId), (char*)&vgInfo, sizeof(vgInfo));
98✔
531
  if (TSDB_CODE_SUCCESS != code) {
98!
532
    return code;
×
533
  }
534

535
  *vgId = vgInfo.vgId;
98✔
536

537
  return TSDB_CODE_SUCCESS;
98✔
538
}
539

540
static int32_t stmtRebuildDataBlock(STscStmt2* pStmt, STableDataCxt* pDataBlock, STableDataCxt** newBlock, uint64_t uid,
66✔
541
                                    uint64_t suid, int32_t vgId) {
542
  STMT_ERR_RET(stmtTryAddTableVgroupInfo(pStmt, &vgId));
66!
543
  STMT_ERR_RET(qRebuildStmtDataBlock(newBlock, pDataBlock, uid, suid, vgId, pStmt->sql.autoCreateTbl));
66!
544

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

547
  return TSDB_CODE_SUCCESS;
66✔
548
}
549

550
static int32_t stmtGetFromCache(STscStmt2* pStmt) {
194✔
551
  if (pStmt->sql.stbInterlaceMode && pStmt->sql.siInfo.pDataCtx) {
194!
UNCOV
552
    pStmt->bInfo.needParse = false;
×
UNCOV
553
    pStmt->bInfo.inExecCache = false;
×
UNCOV
554
    return TSDB_CODE_SUCCESS;
×
555
  }
556

557
  pStmt->bInfo.needParse = true;
194✔
558
  pStmt->bInfo.inExecCache = false;
194✔
559

560
  STableDataCxt** pCxtInExec = taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
194✔
561
  if (pCxtInExec) {
193✔
562
    pStmt->bInfo.needParse = false;
24✔
563
    pStmt->bInfo.inExecCache = true;
24✔
564

565
    pStmt->exec.pCurrBlock = *pCxtInExec;
24✔
566

567
    if (pStmt->sql.autoCreateTbl) {
24✔
568
      tscDebug("reuse stmt block for tb %s in execBlock", pStmt->bInfo.tbFName);
18!
569
      return TSDB_CODE_SUCCESS;
18✔
570
    }
571
  }
572

573
  if (NULL == pStmt->pCatalog) {
175✔
574
    STMT_ERR_RET(catalogGetHandle(pStmt->taos->pAppInfo->clusterId, &pStmt->pCatalog));
87!
575
    pStmt->sql.siInfo.pCatalog = pStmt->pCatalog;
88✔
576
  }
577

578
  if (NULL == pStmt->sql.pTableCache || taosHashGetSize(pStmt->sql.pTableCache) <= 0) {
176!
579
    if (pStmt->bInfo.inExecCache) {
104!
580
      pStmt->bInfo.needParse = false;
×
UNCOV
581
      tscDebug("reuse stmt block for tb %s in execBlock", pStmt->bInfo.tbFName);
×
UNCOV
582
      return TSDB_CODE_SUCCESS;
×
583
    }
584

585
    tscDebug("no stmt block cache for tb %s", pStmt->bInfo.tbFName);
104!
586
    return TSDB_CODE_SUCCESS;
104✔
587
  }
588

589
  if (pStmt->sql.autoCreateTbl) {
72✔
590
    SStmtTableCache* pCache = taosHashGet(pStmt->sql.pTableCache, &pStmt->bInfo.tbSuid, sizeof(pStmt->bInfo.tbSuid));
48✔
591
    if (pCache) {
48!
592
      pStmt->bInfo.needParse = false;
48✔
593
      pStmt->bInfo.tbUid = 0;
48✔
594

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

598
      if (taosHashPut(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName), &pNewBlock,
48!
599
                      POINTER_BYTES)) {
UNCOV
600
        STMT_ERR_RET(terrno);
×
601
      }
602

603
      pStmt->exec.pCurrBlock = pNewBlock;
48✔
604

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

607
      return TSDB_CODE_SUCCESS;
48✔
608
    }
609

610
    STMT_RET(stmtCleanBindInfo(pStmt));
×
611
  }
612

613
  uint64_t uid, suid;
614
  int32_t  vgId;
615
  int8_t   tableType;
616

617
  STableMeta*      pTableMeta = NULL;
24✔
618
  SRequestConnInfo conn = {.pTrans = pStmt->taos->pAppInfo->pTransporter,
24✔
619
                           .requestId = pStmt->exec.pRequest->requestId,
24✔
620
                           .requestObjRefId = pStmt->exec.pRequest->self,
24✔
621
                           .mgmtEps = getEpSet_s(&pStmt->taos->pAppInfo->mgmtEp)};
24✔
622
  int32_t          code = catalogGetTableMeta(pStmt->pCatalog, &conn, &pStmt->bInfo.sname, &pTableMeta);
24✔
623

624
  pStmt->stat.ctgGetTbMetaNum++;
24✔
625

626
  if (TSDB_CODE_PAR_TABLE_NOT_EXIST == code) {
24!
UNCOV
627
    tscDebug("tb %s not exist", pStmt->bInfo.tbFName);
×
628
    STMT_ERR_RET(stmtCleanBindInfo(pStmt));
×
629

630
    STMT_ERR_RET(code);
×
631
  }
632

633
  STMT_ERR_RET(code);
24!
634

635
  uid = pTableMeta->uid;
24✔
636
  suid = pTableMeta->suid;
24✔
637
  tableType = pTableMeta->tableType;
24✔
638
  pStmt->bInfo.tbVgId = pTableMeta->vgId;
24✔
639
  vgId = pTableMeta->vgId;
24✔
640

641
  taosMemoryFree(pTableMeta);
24!
642

643
  uint64_t cacheUid = (TSDB_CHILD_TABLE == tableType) ? suid : uid;
24!
644

645
  if (uid == pStmt->bInfo.tbUid) {
24!
UNCOV
646
    pStmt->bInfo.needParse = false;
×
647

UNCOV
648
    tscDebug("tb %s is current table", pStmt->bInfo.tbFName);
×
649

UNCOV
650
    return TSDB_CODE_SUCCESS;
×
651
  }
652

653
  if (pStmt->bInfo.inExecCache) {
24✔
654
    SStmtTableCache* pCache = taosHashGet(pStmt->sql.pTableCache, &cacheUid, sizeof(cacheUid));
6✔
655
    if (NULL == pCache) {
6!
UNCOV
656
      tscError("table [%s, %" PRIx64 ", %" PRIx64 "] found in exec blockHash, but not in sql blockHash",
×
657
               pStmt->bInfo.tbFName, uid, cacheUid);
658

UNCOV
659
      STMT_ERR_RET(TSDB_CODE_APP_ERROR);
×
660
    }
661

662
    pStmt->bInfo.needParse = false;
6✔
663

664
    pStmt->bInfo.tbUid = uid;
6✔
665
    pStmt->bInfo.tbSuid = suid;
6✔
666
    pStmt->bInfo.tbType = tableType;
6✔
667
    pStmt->bInfo.boundTags = pCache->boundTags;
6✔
668
    pStmt->bInfo.tagsCached = true;
6✔
669

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

672
    return TSDB_CODE_SUCCESS;
6✔
673
  }
674

675
  SStmtTableCache* pCache = taosHashGet(pStmt->sql.pTableCache, &cacheUid, sizeof(cacheUid));
18✔
676
  if (pCache) {
18!
677
    pStmt->bInfo.needParse = false;
18✔
678

679
    pStmt->bInfo.tbUid = uid;
18✔
680
    pStmt->bInfo.tbSuid = suid;
18✔
681
    pStmt->bInfo.tbType = tableType;
18✔
682
    pStmt->bInfo.boundTags = pCache->boundTags;
18✔
683
    pStmt->bInfo.tagsCached = true;
18✔
684

685
    STableDataCxt* pNewBlock = NULL;
18✔
686
    STMT_ERR_RET(stmtRebuildDataBlock(pStmt, pCache->pDataCtx, &pNewBlock, uid, suid, vgId));
18!
687

688
    if (taosHashPut(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName), &pNewBlock,
18!
689
                    POINTER_BYTES)) {
690
      STMT_ERR_RET(terrno);
×
691
    }
692

693
    pStmt->exec.pCurrBlock = pNewBlock;
18✔
694

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

697
    return TSDB_CODE_SUCCESS;
18✔
698
  }
699

UNCOV
700
  STMT_ERR_RET(stmtCleanBindInfo(pStmt));
×
701

UNCOV
702
  return TSDB_CODE_SUCCESS;
×
703
}
704

UNCOV
705
static int32_t stmtResetStmt(STscStmt2* pStmt) {
×
706
  STMT_ERR_RET(stmtCleanSQLInfo(pStmt));
×
707

UNCOV
708
  pStmt->sql.pTableCache = taosHashInit(100, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
×
UNCOV
709
  if (NULL == pStmt->sql.pTableCache) {
×
UNCOV
710
    STMT_ERR_RET(terrno);
×
711
  }
712

UNCOV
713
  pStmt->sql.status = STMT_INIT;
×
714

UNCOV
715
  return TSDB_CODE_SUCCESS;
×
716
}
717

718
static int32_t stmtAsyncOutput(STscStmt2* pStmt, void* param) {
9,484✔
719
  SStmtQNode* pParam = (SStmtQNode*)param;
9,484✔
720

721
  if (pParam->restoreTbCols) {
9,484✔
722
    for (int32_t i = 0; i < pStmt->sql.siInfo.pTableColsIdx; ++i) {
9,485✔
723
      SArray** p = (SArray**)TARRAY_GET_ELEM(pStmt->sql.siInfo.pTableCols, i);
5,265✔
724
      *p = taosArrayInit(20, POINTER_BYTES);
5,265✔
725
      if (*p == NULL) {
5,264!
UNCOV
726
        STMT_ERR_RET(terrno);
×
727
      }
728
    }
729

730
    atomic_store_8((int8_t*)&pStmt->sql.siInfo.tableColsReady, true);
4,220✔
731
  } else {
732
    int code = qAppendStmtTableOutput(pStmt->sql.pQuery, pStmt->sql.pVgHash, &pParam->tblData, pStmt->exec.pCurrBlock,
5,258✔
733
                                      &pStmt->sql.siInfo, pParam->pCreateTbReq);
734
    // taosMemoryFree(pParam->pTbData);
735
    (void)atomic_sub_fetch_64(&pStmt->sql.siInfo.tbRemainNum, 1);
5,252✔
736
    STMT_ERR_RET(code);
5,265!
737
  }
738
  return TSDB_CODE_SUCCESS;
9,489✔
739
}
740

741
static void* stmtBindThreadFunc(void* param) {
96✔
742
  setThreadName("stmt2Bind");
96✔
743
  qInfo("stmt2 bind thread started");
96!
744

745
  STscStmt2* pStmt = (STscStmt2*)param;
96✔
746

747
  while (true) {
9,486✔
748
    SStmtQNode* asyncParam = NULL;
9,582✔
749

750
    if (!stmtDequeue(pStmt, &asyncParam)) {
9,582✔
751
      if (pStmt->queue.stopQueue && 0 == atomic_load_64((int64_t*)&pStmt->queue.qRemainNum)) {
96!
752
        STMT2_DLOG_E("queue is empty and stopQueue is set, thread will exit");
96!
753
        break;
96✔
754
      }
755
      continue;
×
756
    }
757

758
    int ret = stmtAsyncOutput(pStmt, asyncParam);
9,485✔
759
    if (ret != 0) {
9,486!
760
      STMT2_ELOG("stmtAsyncOutput failed, reason:%s", tstrerror(ret));
×
761
    }
762
  }
763

764
  qInfo("stmt2 bind thread stopped");
96!
765
  return NULL;
96✔
766
}
767

768
static int32_t stmtStartBindThread(STscStmt2* pStmt) {
96✔
769
  TdThreadAttr thAttr;
770
  if (taosThreadAttrInit(&thAttr) != 0) {
96!
UNCOV
771
    return TSDB_CODE_TSC_INTERNAL_ERROR;
×
772
  }
773
  if (taosThreadAttrSetDetachState(&thAttr, PTHREAD_CREATE_JOINABLE) != 0) {
96!
UNCOV
774
    return TSDB_CODE_TSC_INTERNAL_ERROR;
×
775
  }
776

777
  if (taosThreadCreate(&pStmt->bindThread, &thAttr, stmtBindThreadFunc, pStmt) != 0) {
95!
UNCOV
778
    terrno = TAOS_SYSTEM_ERROR(ERRNO);
×
UNCOV
779
    STMT_ERR_RET(terrno);
×
780
  }
781

782
  pStmt->bindThreadInUse = true;
96✔
783

784
  (void)taosThreadAttrDestroy(&thAttr);
96✔
785
  return TSDB_CODE_SUCCESS;
96✔
786
}
787

788
static int32_t stmtInitQueue(STscStmt2* pStmt) {
96✔
789
  (void)taosThreadCondInit(&pStmt->queue.waitCond, NULL);
96✔
790
  (void)taosThreadMutexInit(&pStmt->queue.mutex, NULL);
96✔
791
  STMT_ERR_RET(stmtAllocQNodeFromBuf(&pStmt->sql.siInfo.tbBuf, (void**)&pStmt->queue.head));
192!
792
  pStmt->queue.tail = pStmt->queue.head;
96✔
793

794
  return TSDB_CODE_SUCCESS;
96✔
795
}
796

797
static int32_t stmtIniAsyncBind(STscStmt2* pStmt) {
145✔
798
  (void)taosThreadCondInit(&pStmt->asyncBindParam.waitCond, NULL);
145✔
799
  (void)taosThreadMutexInit(&pStmt->asyncBindParam.mutex, NULL);
145✔
800
  pStmt->asyncBindParam.asyncBindNum = 0;
145✔
801

802
  return TSDB_CODE_SUCCESS;
145✔
803
}
804

805
static int32_t stmtInitTableBuf(STableBufInfo* pTblBuf) {
96✔
806
  pTblBuf->buffUnit = sizeof(SStmtQNode);
96✔
807
  pTblBuf->buffSize = pTblBuf->buffUnit * 1000;
96✔
808
  pTblBuf->pBufList = taosArrayInit(100, POINTER_BYTES);
96✔
809
  if (NULL == pTblBuf->pBufList) {
96!
UNCOV
810
    return terrno;
×
811
  }
812
  void* buff = taosMemoryMalloc(pTblBuf->buffSize);
96!
813
  if (NULL == buff) {
96!
UNCOV
814
    return terrno;
×
815
  }
816

817
  if (taosArrayPush(pTblBuf->pBufList, &buff) == NULL) {
192!
UNCOV
818
    return terrno;
×
819
  }
820

821
  pTblBuf->pCurBuff = buff;
96✔
822
  pTblBuf->buffIdx = 1;
96✔
823
  pTblBuf->buffOffset = 0;
96✔
824

825
  return TSDB_CODE_SUCCESS;
96✔
826
}
827

828
TAOS_STMT2* stmtInit2(STscObj* taos, TAOS_STMT2_OPTION* pOptions) {
144✔
829
  STscObj*   pObj = (STscObj*)taos;
144✔
830
  STscStmt2* pStmt = NULL;
144✔
831
  int32_t    code = 0;
144✔
832

833
  pStmt = taosMemoryCalloc(1, sizeof(STscStmt2));
144!
834
  if (NULL == pStmt) {
145!
UNCOV
835
    return NULL;
×
836
  }
837

838
  pStmt->sql.pTableCache = taosHashInit(100, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
145✔
839
  if (NULL == pStmt->sql.pTableCache) {
145!
UNCOV
840
    taosMemoryFree(pStmt);
×
UNCOV
841
    return NULL;
×
842
  }
843

844
  pStmt->taos = pObj;
145✔
845
  if (taos->db[0] != '\0') {
145✔
846
    pStmt->db = taosStrdup(taos->db);
51!
847
  }
848
  pStmt->bInfo.needParse = true;
145✔
849
  pStmt->sql.status = STMT_INIT;
145✔
850
  pStmt->errCode = TSDB_CODE_SUCCESS;
145✔
851

852
  if (NULL != pOptions) {
145!
853
    (void)memcpy(&pStmt->options, pOptions, sizeof(pStmt->options));
145✔
854
    if (pOptions->singleStbInsert && pOptions->singleTableBindOnce) {
145✔
855
      pStmt->stbInterlaceMode = true;
79✔
856
    }
857

858
    pStmt->reqid = pOptions->reqid;
145✔
859
  }
860

861
  if (pStmt->stbInterlaceMode) {
145✔
862
    pStmt->sql.siInfo.transport = taos->pAppInfo->pTransporter;
79✔
863
    pStmt->sql.siInfo.acctId = taos->acctId;
79✔
864
    pStmt->sql.siInfo.dbname = taos->db;
79✔
865
    pStmt->sql.siInfo.mgmtEpSet = getEpSet_s(&pStmt->taos->pAppInfo->mgmtEp);
79✔
866
    pStmt->sql.siInfo.pTableHash = tSimpleHashInit(100, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY));
79✔
867
    if (NULL == pStmt->sql.siInfo.pTableHash) {
79!
868
      (void)stmtClose2(pStmt);
×
869
      return NULL;
×
870
    }
871
    pStmt->sql.siInfo.pTableCols = taosArrayInit(STMT_TABLE_COLS_NUM, POINTER_BYTES);
79✔
872
    if (NULL == pStmt->sql.siInfo.pTableCols) {
79!
UNCOV
873
      terrno = terrno;
×
UNCOV
874
      (void)stmtClose2(pStmt);
×
UNCOV
875
      return NULL;
×
876
    }
877

878
    code = stmtInitTableBuf(&pStmt->sql.siInfo.tbBuf);
79✔
879
    if (TSDB_CODE_SUCCESS == code) {
79!
880
      code = stmtInitQueue(pStmt);
79✔
881
    }
882
    if (TSDB_CODE_SUCCESS == code) {
79!
883
      code = stmtStartBindThread(pStmt);
79✔
884
    }
885
    if (TSDB_CODE_SUCCESS != code) {
79!
UNCOV
886
      terrno = code;
×
UNCOV
887
      (void)stmtClose2(pStmt);
×
UNCOV
888
      return NULL;
×
889
    }
890
  }
891

892
  pStmt->sql.siInfo.tableColsReady = true;
145✔
893
  if (pStmt->options.asyncExecFn) {
145✔
894
    if (tsem_init(&pStmt->asyncExecSem, 0, 1) != 0) {
5!
UNCOV
895
      terrno = TAOS_SYSTEM_ERROR(ERRNO);
×
UNCOV
896
      (void)stmtClose2(pStmt);
×
UNCOV
897
      return NULL;
×
898
    }
899
  }
900
  code = stmtIniAsyncBind(pStmt);
145✔
901
  if (TSDB_CODE_SUCCESS != code) {
145!
UNCOV
902
    terrno = code;
×
UNCOV
903
    (void)stmtClose2(pStmt);
×
UNCOV
904
    return NULL;
×
905
  }
906

907
  pStmt->execSemWaited = false;
145✔
908

909
  STMT_LOG_SEQ(STMT_INIT);
145!
910

911
  tscDebug("stmt:%p initialized", pStmt);
145!
912

913
  return pStmt;
145✔
914
}
915

916
static int stmtSetDbName2(TAOS_STMT2* stmt, const char* dbName) {
64✔
917
  if (dbName == NULL || dbName[0] == '\0') {
64!
UNCOV
918
    return TSDB_CODE_TSC_SQL_SYNTAX_ERROR;
×
919
  }
920
  STscStmt2* pStmt = (STscStmt2*)stmt;
64✔
921

922
  STMT2_DLOG("dbname is specified in sql:%s", dbName);
64!
923
  if (pStmt->db == NULL || pStmt->db[0] == '\0') {
64!
924
    taosMemoryFreeClear(pStmt->db);
33!
925
    STMT2_DLOG("dbname not set by taosconnect, set by sql:%s", dbName);
33!
926
    pStmt->db = taosStrdup(dbName);
33!
927
    (void)strdequote(pStmt->db);
33✔
928
  }
929
  STMT_ERR_RET(stmtCreateRequest(pStmt));
64!
930

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

946
  if (pStmt->bindThreadInUse) {
17!
947
    while (0 == atomic_load_8((int8_t*)&pStmt->sql.siInfo.tableColsReady)) {
17!
UNCOV
948
      taosUsleep(1);
×
949
    }
950
    (void)taosThreadMutexLock(&pStmt->queue.mutex);
17✔
951
    pStmt->queue.stopQueue = true;
17✔
952
    (void)taosThreadCondSignal(&(pStmt->queue.waitCond));
17✔
953
    (void)taosThreadMutexUnlock(&pStmt->queue.mutex);
17✔
954

955
    (void)taosThreadJoin(pStmt->bindThread, NULL);
17✔
956
    pStmt->bindThreadInUse = false;
17✔
957
    pStmt->queue.head = NULL;
17✔
958
    pStmt->queue.tail = NULL;
17✔
959
    pStmt->queue.qRemainNum = 0;
17✔
960

961
    (void)taosThreadCondDestroy(&pStmt->queue.waitCond);
17✔
962
    (void)taosThreadMutexDestroy(&pStmt->queue.mutex);
17✔
963
  }
964

965
  pStmt->sql.siInfo.pTableHash = tSimpleHashInit(100, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY));
17✔
966
  if (NULL == pStmt->sql.siInfo.pTableHash) {
17!
UNCOV
967
    return terrno;
×
968
  }
969

970
  pStmt->sql.siInfo.pTableCols = taosArrayInit(STMT_TABLE_COLS_NUM, POINTER_BYTES);
17✔
971
  if (NULL == pStmt->sql.siInfo.pTableCols) {
17!
UNCOV
972
    return terrno;
×
973
  }
974

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

977
  if (TSDB_CODE_SUCCESS == code) {
17!
978
    code = stmtInitQueue(pStmt);
17✔
979
    pStmt->queue.stopQueue = false;
17✔
980
  }
981
  if (TSDB_CODE_SUCCESS == code) {
17!
982
    code = stmtStartBindThread(pStmt);
17✔
983
  }
984
  if (TSDB_CODE_SUCCESS != code) {
17!
UNCOV
985
    return code;
×
986
  }
987

988
  return TSDB_CODE_SUCCESS;
17✔
989
}
990

991
static int32_t stmtResetStmtForPrepare(STscStmt2* pStmt) {
21✔
992
  char*             db = pStmt->db;
21✔
993
  bool              stbInterlaceMode = pStmt->stbInterlaceMode;
21✔
994
  TAOS_STMT2_OPTION options = pStmt->options;
21✔
995
  uint32_t          reqid = pStmt->reqid;
21✔
996

997
  taosMemoryFree(pStmt->sql.pBindInfo);
21!
998
  pStmt->sql.pBindInfo = NULL;
21✔
999

1000
  taosMemoryFree(pStmt->sql.queryRes.fields);
21!
1001
  pStmt->sql.queryRes.fields = NULL;
21✔
1002

1003
  taosMemoryFree(pStmt->sql.queryRes.userFields);
21!
1004
  pStmt->sql.queryRes.userFields = NULL;
21✔
1005

1006
  pStmt->sql.type = 0;
21✔
1007
  pStmt->sql.runTimes = 0;
21✔
1008
  taosMemoryFree(pStmt->sql.sqlStr);
21!
1009
  pStmt->sql.sqlStr = NULL;
21✔
1010

1011
  qDestroyQuery(pStmt->sql.pQuery);
21✔
1012
  pStmt->sql.pQuery = NULL;
21✔
1013

1014
  taosArrayDestroy(pStmt->sql.nodeList);
21✔
1015
  pStmt->sql.nodeList = NULL;
21✔
1016

1017
  taosHashCleanup(pStmt->sql.pVgHash);
21✔
1018
  pStmt->sql.pVgHash = NULL;
21✔
1019

1020
  if (pStmt->sql.fixValueTags) {
21✔
1021
    tdDestroySVCreateTbReq(pStmt->sql.fixValueTbReq);
12!
1022
    pStmt->sql.fixValueTbReq = NULL;
12✔
1023
  }
1024
  pStmt->sql.fixValueTags = false;
21✔
1025

1026
  void* pIter = taosHashIterate(pStmt->sql.pTableCache, NULL);
21✔
1027
  while (pIter) {
24✔
1028
    SStmtTableCache* pCache = (SStmtTableCache*)pIter;
3✔
1029

1030
    qDestroyStmtDataBlock(pCache->pDataCtx);
3✔
1031
    qDestroyBoundColInfo(pCache->boundTags);
3✔
1032
    taosMemoryFreeClear(pCache->boundTags);
3!
1033

1034
    pIter = taosHashIterate(pStmt->sql.pTableCache, pIter);
3✔
1035
  }
1036
  taosHashCleanup(pStmt->sql.pTableCache);
21✔
1037
  pStmt->sql.pTableCache = taosHashInit(100, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
21✔
1038
  if (NULL == pStmt->sql.pTableCache) {
21!
UNCOV
1039
    return terrno;
×
1040
  }
1041

1042
  if (pStmt->sql.stbInterlaceMode) {
21✔
1043
    pStmt->bInfo.tagsCached = false;
16✔
1044
  }
1045
  STMT_ERR_RET(stmtCleanExecInfo(pStmt, false, true));
21!
1046

1047
  if (pStmt->exec.pRequest) {
21!
UNCOV
1048
    taos_free_result(pStmt->exec.pRequest);
×
UNCOV
1049
    pStmt->exec.pRequest = NULL;
×
1050
  }
1051

1052
  if (pStmt->sql.siInfo.pTableCols) {
21✔
1053
    taosArrayDestroyEx(pStmt->sql.siInfo.pTableCols, stmtFreeTbCols);
17✔
1054
    pStmt->sql.siInfo.pTableCols = NULL;
17✔
1055
  }
1056

1057
  if (pStmt->sql.siInfo.tbBuf.pBufList) {
21✔
1058
    taosArrayDestroyEx(pStmt->sql.siInfo.tbBuf.pBufList, stmtFreeTbBuf);
17✔
1059
    pStmt->sql.siInfo.tbBuf.pBufList = NULL;
17✔
1060
  }
1061

1062
  if (pStmt->sql.siInfo.pTableHash) {
21✔
1063
    tSimpleHashCleanup(pStmt->sql.siInfo.pTableHash);
17✔
1064
    pStmt->sql.siInfo.pTableHash = NULL;
17✔
1065
  }
1066

1067
  if (pStmt->sql.siInfo.pVgroupHash) {
21!
UNCOV
1068
    taosHashCleanup(pStmt->sql.siInfo.pVgroupHash);
×
UNCOV
1069
    pStmt->sql.siInfo.pVgroupHash = NULL;
×
1070
  }
1071

1072
  if (pStmt->sql.siInfo.pVgroupList) {
21!
UNCOV
1073
    taosArrayDestroy(pStmt->sql.siInfo.pVgroupList);
×
UNCOV
1074
    pStmt->sql.siInfo.pVgroupList = NULL;
×
1075
  }
1076

1077
  if (pStmt->sql.siInfo.pDataCtx) {
21✔
1078
    qDestroyStmtDataBlock(pStmt->sql.siInfo.pDataCtx);
15✔
1079
    pStmt->sql.siInfo.pDataCtx = NULL;
15✔
1080
  }
1081

1082
  if (pStmt->sql.siInfo.pTSchema) {
21✔
1083
    taosMemoryFree(pStmt->sql.siInfo.pTSchema);
15!
1084
    pStmt->sql.siInfo.pTSchema = NULL;
15✔
1085
  }
1086

1087
  if (pStmt->sql.siInfo.pRequest) {
21✔
1088
    taos_free_result(pStmt->sql.siInfo.pRequest);
15✔
1089
    pStmt->sql.siInfo.pRequest = NULL;
15✔
1090
  }
1091

1092
  if (stbInterlaceMode) {
21✔
1093
    STMT2_DLOG_E("reprepare for inter");
17!
1094
    STMT_ERR_RET(stmtResetStbInterlaceCache(pStmt));
17!
1095
  }
1096

1097
  pStmt->db = db;
21✔
1098
  pStmt->stbInterlaceMode = stbInterlaceMode;
21✔
1099
  pStmt->options = options;
21✔
1100
  pStmt->reqid = reqid;
21✔
1101

1102
  pStmt->sql.status = STMT_INIT;
21✔
1103

1104
  return TSDB_CODE_SUCCESS;
21✔
1105
}
1106

1107
int stmtPrepare2(TAOS_STMT2* stmt, const char* sql, unsigned long length) {
160✔
1108
  STscStmt2* pStmt = (STscStmt2*)stmt;
160✔
1109
  int32_t    code = 0;
160✔
1110

1111
  if (stmt == NULL || sql == NULL) {
160!
UNCOV
1112
    STMT2_ELOG_E("stmt or sql is NULL");
×
UNCOV
1113
    return TSDB_CODE_INVALID_PARA;
×
1114
  }
1115

1116
  if (pStmt->sql.status >= STMT_PREPARE) {
164✔
1117
    STMT2_DLOG("stmt status is %d, need to reset stmt2 cache before prepare", pStmt->sql.status);
21!
1118
    STMT_ERR_RET(stmtResetStmtForPrepare(pStmt));
21!
1119
  }
1120

1121
  STMT2_DLOG("start to prepare with sql:%s", sql);
164!
1122

1123
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
164✔
1124
    STMT2_ELOG("stmt errCode is not success, ErrCode: 0x%x, ErrMessage: %s\n. ", pStmt->errCode,
1!
1125
               strerror(pStmt->errCode));
1126
    return pStmt->errCode;
1✔
1127
  }
1128

1129
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_PREPARE));
163!
1130

1131
  if (length <= 0) {
163✔
1132
    length = strlen(sql);
103✔
1133
  }
1134

1135
  pStmt->sql.sqlStr = taosStrndup(sql, length);
163!
1136
  if (!pStmt->sql.sqlStr) {
163!
1137
    return terrno;
×
1138
  }
1139
  pStmt->sql.sqlLen = length;
163✔
1140
  pStmt->sql.stbInterlaceMode = pStmt->stbInterlaceMode;
163✔
1141

1142
  char* dbName = NULL;
163✔
1143
  if (qParseDbName(sql, length, &dbName)) {
163✔
1144
    STMT_ERR_RET(stmtSetDbName2(stmt, dbName));
64!
1145
    taosMemoryFreeClear(dbName);
64!
1146
  }
1147

1148
  return TSDB_CODE_SUCCESS;
162✔
1149
}
1150

1151
static int32_t stmtInitStbInterlaceTableInfo(STscStmt2* pStmt) {
86✔
1152
  STableDataCxt** pSrc = taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
86✔
1153
  if (!pSrc) {
86!
UNCOV
1154
    return terrno;
×
1155
  }
1156
  STableDataCxt* pDst = NULL;
86✔
1157

1158
  STMT_ERR_RET(qCloneStmtDataBlock(&pDst, *pSrc, true));
86!
1159
  pStmt->sql.siInfo.pDataCtx = pDst;
79✔
1160

1161
  SArray* pTblCols = NULL;
79✔
1162
  for (int32_t i = 0; i < STMT_TABLE_COLS_NUM; i++) {
74,161✔
1163
    pTblCols = taosArrayInit(20, POINTER_BYTES);
74,193✔
1164
    if (NULL == pTblCols) {
77,628!
UNCOV
1165
      return terrno;
×
1166
    }
1167

1168
    if (taosArrayPush(pStmt->sql.siInfo.pTableCols, &pTblCols) == NULL) {
151,710!
UNCOV
1169
      return terrno;
×
1170
    }
1171
  }
1172

1173
  pStmt->sql.siInfo.boundTags = pStmt->bInfo.boundTags;
26✔
1174

1175
  return TSDB_CODE_SUCCESS;
26✔
1176
}
1177

1178
int stmtIsInsert2(TAOS_STMT2* stmt, int* insert) {
10,791✔
1179
  STscStmt2* pStmt = (STscStmt2*)stmt;
10,791✔
1180

1181
  STMT_DLOG_E("start is insert");
10,791!
1182

1183
  if (pStmt->sql.type) {
10,793✔
1184
    *insert = (STMT_TYPE_INSERT == pStmt->sql.type || STMT_TYPE_MULTI_INSERT == pStmt->sql.type);
10,635✔
1185
  } else {
1186
    *insert = qIsInsertValuesSql(pStmt->sql.sqlStr, pStmt->sql.sqlLen);
158✔
1187
  }
1188

1189
  return TSDB_CODE_SUCCESS;
10,797✔
1190
}
1191

1192
int stmtSetTbName2(TAOS_STMT2* stmt, const char* tbName) {
5,362✔
1193
  STscStmt2* pStmt = (STscStmt2*)stmt;
5,362✔
1194

1195
  int64_t startUs = taosGetTimestampUs();
5,362✔
1196

1197
  STMT_DLOG("start to set tbName:%s", tbName);
5,362!
1198

1199
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
5,361!
UNCOV
1200
    return pStmt->errCode;
×
1201
  }
1202

1203
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_SETTBNAME));
5,361!
1204

1205
  int32_t insert = 0;
5,362✔
1206
  STMT_ERR_RET(stmtIsInsert2(stmt, &insert));
5,362!
1207
  if (0 == insert) {
5,365!
UNCOV
1208
    tscError("set tb name not available for none insert statement");
×
UNCOV
1209
    STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR);
×
1210
  }
1211

1212
  if (!pStmt->sql.stbInterlaceMode || NULL == pStmt->sql.siInfo.pDataCtx) {
5,365✔
1213
    STMT_ERR_RET(stmtCreateRequest(pStmt));
200!
1214

1215
    STMT_ERR_RET(qCreateSName(&pStmt->bInfo.sname, tbName, pStmt->taos->acctId, pStmt->exec.pRequest->pDb,
196!
1216
                              pStmt->exec.pRequest->msgBuf, pStmt->exec.pRequest->msgBufLen));
1217
    STMT_ERR_RET(tNameExtractFullName(&pStmt->bInfo.sname, pStmt->bInfo.tbFName));
194✔
1218

1219
    STMT_ERR_RET(stmtGetFromCache(pStmt));
194!
1220

1221
    if (pStmt->bInfo.needParse) {
194✔
1222
      tstrncpy(pStmt->bInfo.tbName, tbName, sizeof(pStmt->bInfo.tbName));
104✔
1223
      pStmt->bInfo.tbName[sizeof(pStmt->bInfo.tbName) - 1] = 0;
104✔
1224

1225
      STMT_ERR_RET(stmtParseSql(pStmt));
104!
1226
    }
1227
  } else {
1228
    tstrncpy(pStmt->bInfo.tbName, tbName, sizeof(pStmt->bInfo.tbName));
5,165✔
1229
    pStmt->bInfo.tbName[sizeof(pStmt->bInfo.tbName) - 1] = 0;
5,165✔
1230
    pStmt->exec.pRequest->requestId++;
5,165✔
1231
    pStmt->bInfo.needParse = false;
5,165✔
1232
  }
1233

1234
  if (pStmt->sql.stbInterlaceMode && NULL == pStmt->sql.siInfo.pDataCtx) {
5,360✔
1235
    STMT_ERR_RET(stmtInitStbInterlaceTableInfo(pStmt));
86!
1236
  }
1237

1238
  int64_t startUs2 = taosGetTimestampUs();
5,367✔
1239
  pStmt->stat.setTbNameUs += startUs2 - startUs;
5,367✔
1240

1241
  return TSDB_CODE_SUCCESS;
5,367✔
1242
}
1243

1244
int stmtSetTbTags2(TAOS_STMT2* stmt, TAOS_STMT2_BIND* tags, SVCreateTbReq** pCreateTbReq) {
120✔
1245
  STscStmt2* pStmt = (STscStmt2*)stmt;
120✔
1246

1247
  STMT_DLOG_E("start to set tbTags");
120!
1248

1249
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
120!
UNCOV
1250
    return pStmt->errCode;
×
1251
  }
1252

1253
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_SETTAGS));
120!
1254

1255
  if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 &&
120!
UNCOV
1256
      STMT_TYPE_MULTI_INSERT != pStmt->sql.type) {
×
UNCOV
1257
    pStmt->bInfo.needParse = false;
×
1258
  }
1259
  STMT_ERR_RET(stmtCreateRequest(pStmt));
120!
1260

1261
  if (pStmt->bInfo.needParse) {
120!
UNCOV
1262
    STMT_ERR_RET(stmtParseSql(pStmt));
×
1263
  }
1264
  if (pStmt->sql.stbInterlaceMode && NULL == pStmt->sql.siInfo.pDataCtx) {
120!
UNCOV
1265
    STMT_ERR_RET(stmtInitStbInterlaceTableInfo(pStmt));
×
1266
  }
1267

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

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

1299
  tscDebug("start to bind stmt tag values");
120!
1300

1301
  void* boundTags = NULL;
120✔
1302
  if (pStmt->sql.stbInterlaceMode) {
120✔
1303
    boundTags = pStmt->sql.siInfo.boundTags;
38✔
1304
    *pCreateTbReq = taosMemoryCalloc(1, sizeof(SVCreateTbReq));
38!
1305
    if (NULL == pCreateTbReq) {
38!
1306
      return terrno;
×
1307
    }
1308
    int32_t vgId = -1;
38✔
1309
    STMT_ERR_RET(stmtTryAddTableVgroupInfo(pStmt, &vgId));
38!
1310
    (*pCreateTbReq)->uid = vgId;
38✔
1311
  } else {
1312
    boundTags = pStmt->bInfo.boundTags;
82✔
1313
  }
1314

1315
  STMT_ERR_RET(qBindStmtTagsValue2(*pDataBlock, boundTags, pStmt->bInfo.tbSuid, pStmt->bInfo.stbFName,
120✔
1316
                                   pStmt->bInfo.sname.tname, tags, pStmt->exec.pRequest->msgBuf,
1317
                                   pStmt->exec.pRequest->msgBufLen, pStmt->taos->optionInfo.charsetCxt, *pCreateTbReq));
1318

1319
  return TSDB_CODE_SUCCESS;
119✔
1320
}
1321

1322
int stmtCheckTags2(TAOS_STMT2* stmt, SVCreateTbReq** pCreateTbReq) {
26✔
1323
  STscStmt2* pStmt = (STscStmt2*)stmt;
26✔
1324

1325
  STMT_DLOG_E("start to set tbTags");
26!
1326

1327
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
26!
UNCOV
1328
    return pStmt->errCode;
×
1329
  }
1330

1331
  if (!pStmt->sql.stbInterlaceMode) {
26!
UNCOV
1332
    return TSDB_CODE_SUCCESS;
×
1333
  }
1334

1335
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_SETTAGS));
26!
1336

1337
  if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 &&
26!
UNCOV
1338
      STMT_TYPE_MULTI_INSERT != pStmt->sql.type) {
×
1339
    pStmt->bInfo.needParse = false;
×
1340
  }
1341
  STMT_ERR_RET(stmtCreateRequest(pStmt));
26!
1342

1343
  if (pStmt->bInfo.needParse) {
26!
UNCOV
1344
    STMT_ERR_RET(stmtParseSql(pStmt));
×
UNCOV
1345
    if (!pStmt->sql.autoCreateTbl) {
×
UNCOV
1346
      return TSDB_CODE_SUCCESS;
×
1347
    }
1348
  }
1349

1350
  if (pStmt->sql.stbInterlaceMode && NULL == pStmt->sql.siInfo.pDataCtx) {
26!
UNCOV
1351
    STMT_ERR_RET(stmtInitStbInterlaceTableInfo(pStmt));
×
1352
  }
1353

1354
  STMT_ERR_RET(qCreateSName(&pStmt->bInfo.sname, pStmt->bInfo.tbName, pStmt->taos->acctId, pStmt->exec.pRequest->pDb,
26!
1355
                            pStmt->exec.pRequest->msgBuf, pStmt->exec.pRequest->msgBufLen));
1356
  STMT_ERR_RET(tNameExtractFullName(&pStmt->bInfo.sname, pStmt->bInfo.tbFName));
26!
1357

1358
  STableDataCxt** pDataBlock = NULL;
26✔
1359
  if (pStmt->exec.pCurrBlock) {
26✔
1360
    pDataBlock = &pStmt->exec.pCurrBlock;
9✔
1361
  } else {
1362
    pDataBlock =
1363
        (STableDataCxt**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
17✔
1364
    if (NULL == pDataBlock) {
17!
1365
      tscError("table %s not found in exec blockHash", pStmt->bInfo.tbFName);
×
1366
      STMT_ERR_RET(TSDB_CODE_TSC_STMT_CACHE_ERROR);
×
1367
    }
1368
  }
1369

1370
  if (!((*pDataBlock)->pData->flags & SUBMIT_REQ_AUTO_CREATE_TABLE)) {
26!
1371
    return TSDB_CODE_SUCCESS;
×
1372
  }
1373

1374
  if (pStmt->sql.fixValueTags) {
26✔
1375
    STMT_ERR_RET(cloneSVreateTbReq(pStmt->sql.fixValueTbReq, pCreateTbReq));
9!
1376
    if ((*pCreateTbReq)->name) {
9!
1377
      taosMemoryFree((*pCreateTbReq)->name);
9!
1378
    }
1379
    (*pCreateTbReq)->name = taosStrdup(pStmt->bInfo.tbName);
9!
1380
    int32_t vgId = -1;
9✔
1381
    STMT_ERR_RET(stmtTryAddTableVgroupInfo(pStmt, &vgId));
9!
1382
    (*pCreateTbReq)->uid = vgId;
9✔
1383
    return TSDB_CODE_SUCCESS;
9✔
1384
  }
1385

1386
  if ((*pDataBlock)->pData->pCreateTbReq) {
17!
1387
    pStmt->sql.fixValueTags = true;
17✔
1388
    STMT_ERR_RET(cloneSVreateTbReq((*pDataBlock)->pData->pCreateTbReq, &pStmt->sql.fixValueTbReq));
17!
1389
    STMT_ERR_RET(cloneSVreateTbReq(pStmt->sql.fixValueTbReq, pCreateTbReq));
17!
1390
    (*pCreateTbReq)->uid = (*pDataBlock)->pMeta->vgId;
17✔
1391
  }
1392

1393
  return TSDB_CODE_SUCCESS;
17✔
1394
}
1395

UNCOV
1396
static int stmtFetchColFields2(STscStmt2* pStmt, int32_t* fieldNum, TAOS_FIELD_E** fields) {
×
1397
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
×
UNCOV
1398
    return pStmt->errCode;
×
1399
  }
1400

1401
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
×
1402
    tscError("invalid operation to get query column fileds");
×
UNCOV
1403
    STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR);
×
1404
  }
1405

UNCOV
1406
  STableDataCxt** pDataBlock = NULL;
×
1407

UNCOV
1408
  if (pStmt->sql.stbInterlaceMode) {
×
UNCOV
1409
    pDataBlock = &pStmt->sql.siInfo.pDataCtx;
×
1410
  } else {
1411
    pDataBlock =
UNCOV
1412
        (STableDataCxt**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
×
UNCOV
1413
    if (NULL == pDataBlock) {
×
UNCOV
1414
      tscError("table %s not found in exec blockHash", pStmt->bInfo.tbFName);
×
UNCOV
1415
      STMT_ERR_RET(TSDB_CODE_APP_ERROR);
×
1416
    }
1417
  }
1418

UNCOV
1419
  STMT_ERR_RET(qBuildStmtColFields(*pDataBlock, fieldNum, fields));
×
1420

UNCOV
1421
  return TSDB_CODE_SUCCESS;
×
1422
}
1423

1424
static int stmtFetchStbColFields2(STscStmt2* pStmt, int32_t* fieldNum, TAOS_FIELD_ALL** fields) {
42✔
1425
  int32_t code = 0;
42✔
1426
  int32_t preCode = pStmt->errCode;
42✔
1427

1428
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
42!
1429
    return pStmt->errCode;
×
1430
  }
1431

1432
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
42!
UNCOV
1433
    tscError("invalid operation to get query column fileds");
×
UNCOV
1434
    STMT_ERRI_JRET(TSDB_CODE_TSC_STMT_API_ERROR);
×
1435
  }
1436

1437
  STableDataCxt** pDataBlock = NULL;
42✔
1438
  bool            cleanStb = false;
42✔
1439

1440
  if (pStmt->sql.stbInterlaceMode && pStmt->sql.siInfo.pDataCtx != NULL) {
42✔
1441
    pDataBlock = &pStmt->sql.siInfo.pDataCtx;
7✔
1442
  } else {
1443
    cleanStb = true;
35✔
1444
    pDataBlock =
1445
        (STableDataCxt**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
35✔
1446
  }
1447

1448
  if (NULL == pDataBlock || NULL == *pDataBlock) {
42!
UNCOV
1449
    tscError("table %s not found in exec blockHash", pStmt->bInfo.tbFName);
×
UNCOV
1450
    STMT_ERRI_JRET(TSDB_CODE_APP_ERROR);
×
1451
  }
1452

1453
  STMT_ERRI_JRET(
42!
1454
      qBuildStmtStbColFields(*pDataBlock, pStmt->bInfo.boundTags, pStmt->bInfo.tbNameFlag, fieldNum, fields));
1455

1456
  if (pStmt->bInfo.tbType == TSDB_SUPER_TABLE && cleanStb) {
42!
1457
    qDestroyStmtDataBlock(*pDataBlock);
28✔
1458
    *pDataBlock = NULL;
28✔
1459
    if (taosHashRemove(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName)) != 0) {
28!
UNCOV
1460
      tscError("get fileds %s remove exec blockHash fail", pStmt->bInfo.tbFName);
×
UNCOV
1461
      STMT_ERRI_JRET(TSDB_CODE_APP_ERROR);
×
1462
    }
1463
    pStmt->sql.autoCreateTbl = false;
28✔
1464
    pStmt->bInfo.tagsCached = false;
28✔
1465
    pStmt->bInfo.sname = (SName){0};
28✔
1466
    STMT_ERR_RET(stmtCleanBindInfo(pStmt));
28!
1467
  }
1468

1469
_return:
14✔
1470

1471
  pStmt->errCode = preCode;
42✔
1472

1473
  return code;
42✔
1474
}
1475
/*
1476
SArray* stmtGetFreeCol(STscStmt2* pStmt, int32_t* idx) {
1477
  while (true) {
1478
    if (pStmt->exec.smInfo.pColIdx >= STMT_COL_BUF_SIZE) {
1479
      pStmt->exec.smInfo.pColIdx = 0;
1480
    }
1481

1482
    if ((pStmt->exec.smInfo.pColIdx + 1) == atomic_load_32(&pStmt->exec.smInfo.pColFreeIdx)) {
1483
      taosUsleep(1);
1484
      continue;
1485
    }
1486

1487
    *idx = pStmt->exec.smInfo.pColIdx;
1488
    return pStmt->exec.smInfo.pCols[pStmt->exec.smInfo.pColIdx++];
1489
  }
1490
}
1491
*/
1492
static int32_t stmtAppendTablePostHandle(STscStmt2* pStmt, SStmtQNode* param) {
5,258✔
1493
  if (NULL == pStmt->sql.siInfo.pVgroupHash) {
5,258✔
1494
    pStmt->sql.siInfo.pVgroupHash =
4,223✔
1495
        taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_ENTRY_LOCK);
4,222✔
1496
  }
1497
  if (NULL == pStmt->sql.siInfo.pVgroupList) {
5,259✔
1498
    pStmt->sql.siInfo.pVgroupList = taosArrayInit(64, POINTER_BYTES);
4,224✔
1499
  }
1500

1501
  if (NULL == pStmt->sql.siInfo.pRequest) {
5,253✔
1502
    STMT_ERR_RET(buildRequest(pStmt->taos->id, pStmt->sql.sqlStr, pStmt->sql.sqlLen, NULL, false,
84!
1503
                              (SRequestObj**)&pStmt->sql.siInfo.pRequest, pStmt->reqid));
1504

1505
    if (pStmt->reqid != 0) {
84!
UNCOV
1506
      pStmt->reqid++;
×
1507
    }
1508
    pStmt->exec.pRequest->syncQuery = true;
84✔
1509

1510
    pStmt->sql.siInfo.requestId = ((SRequestObj*)pStmt->sql.siInfo.pRequest)->requestId;
84✔
1511
    pStmt->sql.siInfo.requestSelf = ((SRequestObj*)pStmt->sql.siInfo.pRequest)->self;
84✔
1512
  }
1513

1514
  if (!pStmt->sql.siInfo.tbFromHash && pStmt->sql.siInfo.firstName[0] &&
5,253✔
1515
      0 == strcmp(pStmt->sql.siInfo.firstName, pStmt->bInfo.tbName)) {
91✔
1516
    pStmt->sql.siInfo.tbFromHash = true;
33✔
1517
  }
1518

1519
  if (0 == pStmt->sql.siInfo.firstName[0]) {
5,253✔
1520
    tstrncpy(pStmt->sql.siInfo.firstName, pStmt->bInfo.tbName, TSDB_TABLE_NAME_LEN);
70✔
1521
  }
1522

1523
  param->tblData.getFromHash = pStmt->sql.siInfo.tbFromHash;
5,253✔
1524
  param->next = NULL;
5,253✔
1525

1526
  (void)atomic_add_fetch_64(&pStmt->sql.siInfo.tbRemainNum, 1);
5,253✔
1527

1528
  if (pStmt->queue.stopQueue) {
5,264!
UNCOV
1529
    STMT_DLOG_E("stmt bind thread is stopped,cannot enqueue bind request");
×
UNCOV
1530
    return TSDB_CODE_TSC_STMT_API_ERROR;
×
1531
  }
1532
  stmtEnqueue(pStmt, param);
5,264✔
1533

1534
  return TSDB_CODE_SUCCESS;
5,259✔
1535
}
1536

1537
static FORCE_INLINE int32_t stmtGetTableColsFromCache(STscStmt2* pStmt, SArray** pTableCols) {
1538
  while (true) {
1539
    if (pStmt->sql.siInfo.pTableColsIdx < taosArrayGetSize(pStmt->sql.siInfo.pTableCols)) {
5,255!
1540
      *pTableCols = (SArray*)taosArrayGetP(pStmt->sql.siInfo.pTableCols, pStmt->sql.siInfo.pTableColsIdx++);
5,251✔
1541
      break;
5,254✔
1542
    } else {
UNCOV
1543
      SArray* pTblCols = NULL;
×
UNCOV
1544
      for (int32_t i = 0; i < 100; i++) {
×
UNCOV
1545
        pTblCols = taosArrayInit(20, POINTER_BYTES);
×
UNCOV
1546
        if (NULL == pTblCols) {
×
UNCOV
1547
          return terrno;
×
1548
        }
1549

1550
        if (taosArrayPush(pStmt->sql.siInfo.pTableCols, &pTblCols) == NULL) {
×
UNCOV
1551
          return terrno;
×
1552
        }
1553
      }
1554
    }
1555
  }
1556

1557
  return TSDB_CODE_SUCCESS;
5,254✔
1558
}
1559

1560
static int32_t stmtCacheBlock(STscStmt2* pStmt) {
116✔
1561
  if (pStmt->sql.type != STMT_TYPE_MULTI_INSERT) {
116✔
1562
    return TSDB_CODE_SUCCESS;
9✔
1563
  }
1564

1565
  uint64_t uid = pStmt->bInfo.tbUid;
107✔
1566
  uint64_t cacheUid = (TSDB_CHILD_TABLE == pStmt->bInfo.tbType) ? pStmt->bInfo.tbSuid : uid;
107!
1567

1568
  if (taosHashGet(pStmt->sql.pTableCache, &cacheUid, sizeof(cacheUid))) {
107✔
1569
    return TSDB_CODE_SUCCESS;
89✔
1570
  }
1571

1572
  STableDataCxt** pSrc = taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
18✔
1573
  if (!pSrc) {
18!
UNCOV
1574
    return terrno;
×
1575
  }
1576
  STableDataCxt* pDst = NULL;
18✔
1577

1578
  STMT_ERR_RET(qCloneStmtDataBlock(&pDst, *pSrc, true));
18!
1579

1580
  SStmtTableCache cache = {
18✔
1581
      .pDataCtx = pDst,
1582
      .boundTags = pStmt->bInfo.boundTags,
18✔
1583
  };
1584

1585
  if (taosHashPut(pStmt->sql.pTableCache, &cacheUid, sizeof(cacheUid), &cache, sizeof(cache))) {
18!
UNCOV
1586
    return terrno;
×
1587
  }
1588

1589
  if (pStmt->sql.autoCreateTbl) {
18✔
1590
    pStmt->bInfo.tagsCached = true;
15✔
1591
  } else {
1592
    pStmt->bInfo.boundTags = NULL;
3✔
1593
  }
1594

1595
  return TSDB_CODE_SUCCESS;
18✔
1596
}
1597

1598
static int stmtAddBatch2(TAOS_STMT2* stmt) {
4,341✔
1599
  STscStmt2* pStmt = (STscStmt2*)stmt;
4,341✔
1600

1601
  int64_t startUs = taosGetTimestampUs();
4,340✔
1602

1603
  STMT_DLOG_E("start to add batch");
4,340!
1604

1605
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
4,341!
UNCOV
1606
    return pStmt->errCode;
×
1607
  }
1608

1609
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_ADD_BATCH));
4,341!
1610

1611
  if (pStmt->sql.stbInterlaceMode) {
4,340✔
1612
    int64_t startUs2 = taosGetTimestampUs();
4,224✔
1613
    pStmt->stat.addBatchUs += startUs2 - startUs;
4,224✔
1614

1615
    pStmt->sql.siInfo.tableColsReady = false;
4,224✔
1616

1617
    SStmtQNode* param = NULL;
4,224✔
1618
    STMT_ERR_RET(stmtAllocQNodeFromBuf(&pStmt->sql.siInfo.tbBuf, (void**)&param));
8,447!
1619
    param->restoreTbCols = true;
4,223✔
1620
    param->next = NULL;
4,223✔
1621

1622
    if (pStmt->sql.autoCreateTbl) {
4,223✔
1623
      pStmt->bInfo.tagsCached = true;
36✔
1624
    }
1625

1626
    if (pStmt->queue.stopQueue) {
4,223!
UNCOV
1627
      STMT_DLOG_E("stmt bind thread is stopped,cannot enqueue bind request");
×
UNCOV
1628
      return TSDB_CODE_TSC_STMT_API_ERROR;
×
1629
    }
1630

1631
    stmtEnqueue(pStmt, param);
4,223✔
1632

1633
    return TSDB_CODE_SUCCESS;
4,226✔
1634
  }
1635

1636
  STMT_ERR_RET(stmtCacheBlock(pStmt));
116!
1637

1638
  return TSDB_CODE_SUCCESS;
116✔
1639
}
1640
/*
1641
static int32_t stmtBackupQueryFields(STscStmt2* pStmt) {
1642
  SStmtQueryResInfo* pRes = &pStmt->sql.queryRes;
1643
  pRes->numOfCols = pStmt->exec.pRequest->body.resInfo.numOfCols;
1644
  pRes->precision = pStmt->exec.pRequest->body.resInfo.precision;
1645

1646
  int32_t size = pRes->numOfCols * sizeof(TAOS_FIELD);
1647
  pRes->fields = taosMemoryMalloc(size);
1648
  pRes->userFields = taosMemoryMalloc(size);
1649
  if (NULL == pRes->fields || NULL == pRes->userFields) {
1650
    STMT_ERR_RET(terrno);
1651
  }
1652
  (void)memcpy(pRes->fields, pStmt->exec.pRequest->body.resInfo.fields, size);
1653
  (void)memcpy(pRes->userFields, pStmt->exec.pRequest->body.resInfo.userFields, size);
1654

1655
  return TSDB_CODE_SUCCESS;
1656
}
1657

1658
static int32_t stmtRestoreQueryFields(STscStmt2* pStmt) {
1659
  SStmtQueryResInfo* pRes = &pStmt->sql.queryRes;
1660
  int32_t            size = pRes->numOfCols * sizeof(TAOS_FIELD);
1661

1662
  pStmt->exec.pRequest->body.resInfo.numOfCols = pRes->numOfCols;
1663
  pStmt->exec.pRequest->body.resInfo.precision = pRes->precision;
1664

1665
  if (NULL == pStmt->exec.pRequest->body.resInfo.fields) {
1666
    pStmt->exec.pRequest->body.resInfo.fields = taosMemoryMalloc(size);
1667
    if (NULL == pStmt->exec.pRequest->body.resInfo.fields) {
1668
      STMT_ERR_RET(terrno);
1669
    }
1670
    (void)memcpy(pStmt->exec.pRequest->body.resInfo.fields, pRes->fields, size);
1671
  }
1672

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

1681
  return TSDB_CODE_SUCCESS;
1682
}
1683
*/
1684
int stmtBindBatch2(TAOS_STMT2* stmt, TAOS_STMT2_BIND* bind, int32_t colIdx, SVCreateTbReq* pCreateTbReq) {
5,375✔
1685
  STscStmt2* pStmt = (STscStmt2*)stmt;
5,375✔
1686
  int32_t    code = 0;
5,375✔
1687

1688
  int64_t startUs = taosGetTimestampUs();
5,376✔
1689

1690
  STMT_DLOG("start to bind stmt data, colIdx:%d", colIdx);
5,376!
1691

1692
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
5,375!
UNCOV
1693
    return pStmt->errCode;
×
1694
  }
1695

1696
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_BIND));
5,375!
1697

1698
  if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 &&
5,377!
UNCOV
1699
      STMT_TYPE_MULTI_INSERT != pStmt->sql.type) {
×
UNCOV
1700
    pStmt->bInfo.needParse = false;
×
1701
  }
1702

1703
  if (pStmt->exec.pRequest && STMT_TYPE_QUERY == pStmt->sql.type && pStmt->sql.runTimes) {
5,377✔
1704
    taos_free_result(pStmt->exec.pRequest);
1✔
1705
    pStmt->exec.pRequest = NULL;
1✔
1706
  }
1707

1708
  STMT_ERR_RET(stmtCreateRequest(pStmt));
5,377!
1709
  if (pStmt->bInfo.needParse) {
5,375✔
1710
    code = stmtParseSql(pStmt);
6✔
1711
    if (code != TSDB_CODE_SUCCESS) {
6!
UNCOV
1712
      goto cleanup_root;
×
1713
    }
1714
  }
1715

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

1742
    if (pStmt->sql.pQuery->haveResultSet) {
5!
1743
      STMT_ERR_RET(setResSchemaInfo(&pStmt->exec.pRequest->body.resInfo, pStmt->sql.pQuery->pResSchema,
10!
1744
                                    pStmt->sql.pQuery->numOfResCols, pStmt->sql.pQuery->pResExtSchema, true));
1745
      taosMemoryFreeClear(pStmt->sql.pQuery->pResSchema);
5!
1746
      taosMemoryFreeClear(pStmt->sql.pQuery->pResExtSchema);
5!
1747
      setResPrecision(&pStmt->exec.pRequest->body.resInfo, pStmt->sql.pQuery->precision);
5✔
1748
    }
1749

1750
    TSWAP(pStmt->exec.pRequest->dbList, pStmt->sql.pQuery->pDbList);
5✔
1751
    TSWAP(pStmt->exec.pRequest->tableList, pStmt->sql.pQuery->pTableList);
5✔
1752
    TSWAP(pStmt->exec.pRequest->targetTableList, pStmt->sql.pQuery->pTargetTableList);
5✔
1753

1754
    // if (STMT_TYPE_QUERY == pStmt->sql.queryRes) {
1755
    //   STMT_ERR_RET(stmtRestoreQueryFields(pStmt));
1756
    // }
1757

1758
    // STMT_ERR_RET(stmtBackupQueryFields(pStmt));
1759

1760
    return TSDB_CODE_SUCCESS;
5✔
1761

UNCOV
1762
  cleanup_root:
×
UNCOV
1763
    if (pStmt->sql.pQuery->pRoot) {
×
UNCOV
1764
      nodesDestroyNode(pStmt->sql.pQuery->pRoot);
×
UNCOV
1765
      pStmt->sql.pQuery->pRoot = NULL;
×
1766
    }
UNCOV
1767
    STMT_ERR_RET(code);
×
1768
  }
1769

1770
  if (pStmt->sql.stbInterlaceMode && NULL == pStmt->sql.siInfo.pDataCtx) {
5,370!
UNCOV
1771
    STMT_ERR_RET(stmtInitStbInterlaceTableInfo(pStmt));
×
1772
  }
1773

1774
  STableDataCxt** pDataBlock = NULL;
5,367✔
1775

1776
  if (pStmt->exec.pCurrBlock) {
5,367✔
1777
    pDataBlock = &pStmt->exec.pCurrBlock;
5,260✔
1778
  } else {
1779
    pDataBlock =
1780
        (STableDataCxt**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
107✔
1781
    if (NULL == pDataBlock) {
107!
UNCOV
1782
      tscError("table %s not found in exec blockHash", pStmt->bInfo.tbFName);
×
UNCOV
1783
      STMT_ERR_RET(TSDB_CODE_TSC_STMT_CACHE_ERROR);
×
1784
    }
1785
    pStmt->exec.pCurrBlock = *pDataBlock;
107✔
1786
    if (pStmt->sql.stbInterlaceMode) {
107✔
1787
      taosArrayDestroy(pStmt->exec.pCurrBlock->pData->aCol);
85✔
1788
      (*pDataBlock)->pData->aCol = NULL;
85✔
1789
    }
1790
    if (colIdx < -1) {
107✔
1791
      pStmt->sql.bindRowFormat = true;
1✔
1792
      taosArrayDestroy((*pDataBlock)->pData->aCol);
1✔
1793
      (*pDataBlock)->pData->aCol = taosArrayInit(20, POINTER_BYTES);
1✔
1794
    }
1795
  }
1796

1797
  int64_t startUs2 = taosGetTimestampUs();
5,369✔
1798
  pStmt->stat.bindDataUs1 += startUs2 - startUs;
5,369✔
1799

1800
  SStmtQNode* param = NULL;
5,369✔
1801
  if (pStmt->sql.stbInterlaceMode) {
5,369✔
1802
    STMT_ERR_RET(stmtAllocQNodeFromBuf(&pStmt->sql.siInfo.tbBuf, (void**)&param));
10,510!
1803
    STMT_ERR_RET(stmtGetTableColsFromCache(pStmt, &param->tblData.aCol));
10,509!
1804
    taosArrayClear(param->tblData.aCol);
5,254✔
1805

1806
    // param->tblData.aCol = taosArrayInit(20, POINTER_BYTES);
1807

1808
    param->restoreTbCols = false;
5,248✔
1809
    param->tblData.isOrdered = true;
5,248✔
1810
    param->tblData.isDuplicateTs = false;
5,248✔
1811
    tstrncpy(param->tblData.tbName, pStmt->bInfo.tbName, TSDB_TABLE_NAME_LEN);
5,248✔
1812

1813
    param->pCreateTbReq = pCreateTbReq;
5,248✔
1814
  }
1815

1816
  int64_t startUs3 = taosGetTimestampUs();
5,369✔
1817
  pStmt->stat.bindDataUs2 += startUs3 - startUs2;
5,369✔
1818

1819
  SArray* pCols = pStmt->sql.stbInterlaceMode ? param->tblData.aCol : (*pDataBlock)->pData->aCol;
5,369✔
1820

1821
  if (colIdx < 0) {
5,369✔
1822
    if (pStmt->sql.stbInterlaceMode) {
5,366✔
1823
      // (*pDataBlock)->pData->flags = 0;
1824
      (*pDataBlock)->pData->flags &= ~SUBMIT_REQ_COLUMN_DATA_FORMAT;
5,255✔
1825
      code = qBindStmtStbColsValue2(*pDataBlock, pCols, bind, pStmt->exec.pRequest->msgBuf,
5,255✔
1826
                                    pStmt->exec.pRequest->msgBufLen, &pStmt->sql.siInfo.pTSchema, pStmt->sql.pBindInfo,
5,255✔
1827
                                    pStmt->taos->optionInfo.charsetCxt);
5,255✔
1828
      param->tblData.isOrdered = (*pDataBlock)->ordered;
5,261✔
1829
      param->tblData.isDuplicateTs = (*pDataBlock)->duplicateTs;
5,261✔
1830
    } else {
1831
      if (colIdx == -1) {
111✔
1832
        if (pStmt->sql.bindRowFormat) {
109✔
1833
          tscError("can't mix bind row format and bind column format");
1!
1834
          STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR);
1!
1835
        }
1836
        code = qBindStmtColsValue2(*pDataBlock, pCols, bind, pStmt->exec.pRequest->msgBuf,
108✔
1837
                                   pStmt->exec.pRequest->msgBufLen, pStmt->taos->optionInfo.charsetCxt);
108✔
1838
      } else {
1839
        code = qBindStmt2RowValue(*pDataBlock, (*pDataBlock)->pData->aRowP, bind, pStmt->exec.pRequest->msgBuf,
2✔
1840
                                  pStmt->exec.pRequest->msgBufLen, &pStmt->sql.siInfo.pTSchema, pStmt->sql.pBindInfo,
2✔
1841
                                  pStmt->taos->optionInfo.charsetCxt);
2✔
1842
      }
1843
    }
1844

1845
    if (code) {
5,371✔
1846
      tscError("qBindStmtColsValue failed, error:%s", tstrerror(code));
1!
1847
      STMT_ERR_RET(code);
1!
1848
    }
1849
  } else {
1850
    if (pStmt->sql.stbInterlaceMode) {
6!
UNCOV
1851
      tscError("bind single column not allowed in stb insert mode");
×
UNCOV
1852
      STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR);
×
1853
    }
1854

1855
    if (pStmt->sql.bindRowFormat) {
6!
UNCOV
1856
      tscError("can't mix bind row format and bind column format");
×
UNCOV
1857
      STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR);
×
1858
    }
1859

1860
    if (colIdx != (pStmt->bInfo.sBindLastIdx + 1) && colIdx != 0) {
6!
UNCOV
1861
      tscError("bind column index not in sequence");
×
UNCOV
1862
      STMT_ERR_RET(TSDB_CODE_APP_ERROR);
×
1863
    }
1864

1865
    pStmt->bInfo.sBindLastIdx = colIdx;
6✔
1866

1867
    if (0 == colIdx) {
6✔
1868
      pStmt->bInfo.sBindRowNum = bind->num;
3✔
1869
    }
1870

1871
    code = qBindStmtSingleColValue2(*pDataBlock, pCols, bind, pStmt->exec.pRequest->msgBuf,
6✔
1872
                                    pStmt->exec.pRequest->msgBufLen, colIdx, pStmt->bInfo.sBindRowNum,
6✔
1873
                                    pStmt->taos->optionInfo.charsetCxt);
6✔
1874
    if (code) {
6!
UNCOV
1875
      tscError("qBindStmtSingleColValue failed, error:%s", tstrerror(code));
×
UNCOV
1876
      STMT_ERR_RET(code);
×
1877
    }
1878
  }
1879

1880
  int64_t startUs4 = taosGetTimestampUs();
5,373✔
1881
  pStmt->stat.bindDataUs3 += startUs4 - startUs3;
5,373✔
1882

1883
  if (pStmt->sql.stbInterlaceMode) {
5,373✔
1884
    STMT_ERR_RET(stmtAppendTablePostHandle(pStmt, param));
5,260!
1885
  } else {
1886
    STMT_ERR_RET(stmtAddBatch2(pStmt));
116!
1887
  }
1888

1889
  pStmt->stat.bindDataUs4 += taosGetTimestampUs() - startUs4;
5,377✔
1890

1891
  return TSDB_CODE_SUCCESS;
5,377✔
1892
}
1893
/*
1894
int stmtUpdateTableUid(STscStmt2* pStmt, SSubmitRsp* pRsp) {
1895
  tscDebug("stmt start to update tbUid, blockNum:%d", pRsp->nBlocks);
1896

1897
  int32_t code = 0;
1898
  int32_t finalCode = 0;
1899
  size_t  keyLen = 0;
1900
  void*   pIter = taosHashIterate(pStmt->exec.pBlockHash, NULL);
1901
  while (pIter) {
1902
    STableDataCxt* pBlock = *(STableDataCxt**)pIter;
1903
    char*          key = taosHashGetKey(pIter, &keyLen);
1904

1905
    STableMeta* pMeta = qGetTableMetaInDataBlock(pBlock);
1906
    if (pMeta->uid) {
1907
      pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
1908
      continue;
1909
    }
1910

1911
    SSubmitBlkRsp* blkRsp = NULL;
1912
    int32_t        i = 0;
1913
    for (; i < pRsp->nBlocks; ++i) {
1914
      blkRsp = pRsp->pBlocks + i;
1915
      if (strlen(blkRsp->tblFName) != keyLen) {
1916
        continue;
1917
      }
1918

1919
      if (strncmp(blkRsp->tblFName, key, keyLen)) {
1920
        continue;
1921
      }
1922

1923
      break;
1924
    }
1925

1926
    if (i < pRsp->nBlocks) {
1927
      tscDebug("auto created table %s uid updated from %" PRIx64 " to %" PRIx64, blkRsp->tblFName, pMeta->uid,
1928
               blkRsp->uid);
1929

1930
      pMeta->uid = blkRsp->uid;
1931
      pStmt->bInfo.tbUid = blkRsp->uid;
1932
    } else {
1933
      tscDebug("table %s not found in submit rsp, will update from catalog", pStmt->bInfo.tbFName);
1934
      if (NULL == pStmt->pCatalog) {
1935
        code = catalogGetHandle(pStmt->taos->pAppInfo->clusterId, &pStmt->pCatalog);
1936
        if (code) {
1937
          pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
1938
          finalCode = code;
1939
          continue;
1940
        }
1941
      }
1942

1943
      code = stmtCreateRequest(pStmt);
1944
      if (code) {
1945
        pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
1946
        finalCode = code;
1947
        continue;
1948
      }
1949

1950
      STableMeta*      pTableMeta = NULL;
1951
      SRequestConnInfo conn = {.pTrans = pStmt->taos->pAppInfo->pTransporter,
1952
                               .requestId = pStmt->exec.pRequest->requestId,
1953
                               .requestObjRefId = pStmt->exec.pRequest->self,
1954
                               .mgmtEps = getEpSet_s(&pStmt->taos->pAppInfo->mgmtEp)};
1955
      code = catalogGetTableMeta(pStmt->pCatalog, &conn, &pStmt->bInfo.sname, &pTableMeta);
1956

1957
      pStmt->stat.ctgGetTbMetaNum++;
1958

1959
      taos_free_result(pStmt->exec.pRequest);
1960
      pStmt->exec.pRequest = NULL;
1961

1962
      if (code || NULL == pTableMeta) {
1963
        pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
1964
        finalCode = code;
1965
        taosMemoryFree(pTableMeta);
1966
        continue;
1967
      }
1968

1969
      pMeta->uid = pTableMeta->uid;
1970
      pStmt->bInfo.tbUid = pTableMeta->uid;
1971
      taosMemoryFree(pTableMeta);
1972
    }
1973

1974
    pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
1975
  }
1976

1977
  return finalCode;
1978
}
1979
*/
1980
/*
1981
int stmtStaticModeExec(TAOS_STMT* stmt) {
1982
  STscStmt2*   pStmt = (STscStmt2*)stmt;
1983
  int32_t     code = 0;
1984
  SSubmitRsp* pRsp = NULL;
1985
  if (pStmt->sql.staticMode) {
1986
    return TSDB_CODE_TSC_STMT_API_ERROR;
1987
  }
1988

1989
  STMT_DLOG_E("start to exec");
1990

1991
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_EXECUTE));
1992

1993
  STMT_ERR_RET(qBuildStmtOutputFromTbList(pStmt->sql.pQuery, pStmt->sql.pVgHash, pStmt->exec.pTbBlkList,
1994
pStmt->exec.pCurrBlock, pStmt->exec.tbBlkNum));
1995

1996
  launchQueryImpl(pStmt->exec.pRequest, pStmt->sql.pQuery, true, NULL);
1997

1998
  if (pStmt->exec.pRequest->code && NEED_CLIENT_HANDLE_ERROR(pStmt->exec.pRequest->code)) {
1999
    code = refreshMeta(pStmt->exec.pRequest->pTscObj, pStmt->exec.pRequest);
2000
    if (code) {
2001
      pStmt->exec.pRequest->code = code;
2002
    } else {
2003
      tFreeSSubmitRsp(pRsp);
2004
      STMT_ERR_RET(stmtResetStmt(pStmt));
2005
      STMT_ERR_RET(TSDB_CODE_NEED_RETRY);
2006
    }
2007
  }
2008

2009
  STMT_ERR_JRET(pStmt->exec.pRequest->code);
2010

2011
  pStmt->exec.affectedRows = taos_affected_rows(pStmt->exec.pRequest);
2012
  pStmt->affectedRows += pStmt->exec.affectedRows;
2013

2014
_return:
2015

2016
  stmtCleanExecInfo(pStmt, (code ? false : true), false);
2017

2018
  tFreeSSubmitRsp(pRsp);
2019

2020
  ++pStmt->sql.runTimes;
2021

2022
  STMT_RET(code);
2023
}
2024
*/
2025

2026
static int32_t createParseContext(const SRequestObj* pRequest, SParseContext** pCxt, SSqlCallbackWrapper* pWrapper) {
12✔
2027
  const STscObj* pTscObj = pRequest->pTscObj;
12✔
2028

2029
  *pCxt = taosMemoryCalloc(1, sizeof(SParseContext));
12!
2030
  if (*pCxt == NULL) {
12!
UNCOV
2031
    return terrno;
×
2032
  }
2033

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

2060
static void asyncQueryCb(void* userdata, TAOS_RES* res, int code) {
12✔
2061
  STscStmt2*        pStmt = userdata;
12✔
2062
  __taos_async_fn_t fp = pStmt->options.asyncExecFn;
12✔
2063

2064
  pStmt->exec.affectedRows = taos_affected_rows(pStmt->exec.pRequest);
12✔
2065
  pStmt->affectedRows += pStmt->exec.affectedRows;
12✔
2066

2067
  fp(pStmt->options.userdata, res, code);
12✔
2068

2069
  while (0 == atomic_load_8((int8_t*)&pStmt->sql.siInfo.tableColsReady)) {
12!
UNCOV
2070
    taosUsleep(1);
×
2071
  }
2072
  (void)stmtCleanExecInfo(pStmt, (code ? false : true), false);
12✔
2073
  ++pStmt->sql.runTimes;
12✔
2074

2075
  if (tsem_post(&pStmt->asyncExecSem) != 0) {
12!
UNCOV
2076
    tscError("failed to post asyncExecSem");
×
2077
  }
2078
}
12✔
2079

2080
int stmtExec2(TAOS_STMT2* stmt, int* affected_rows) {
4,277✔
2081
  STscStmt2* pStmt = (STscStmt2*)stmt;
4,277✔
2082
  int32_t    code = 0;
4,277✔
2083
  int64_t    startUs = taosGetTimestampUs();
4,274✔
2084

2085
  STMT_DLOG_E("start to exec");
4,274!
2086

2087
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
4,274!
UNCOV
2088
    return pStmt->errCode;
×
2089
  }
2090

2091
  TSC_ERR_RET(taosThreadMutexLock(&pStmt->asyncBindParam.mutex));
4,274!
2092
  while (atomic_load_8((int8_t*)&pStmt->asyncBindParam.asyncBindNum) > 0) {
4,277!
2093
    (void)taosThreadCondWait(&pStmt->asyncBindParam.waitCond, &pStmt->asyncBindParam.mutex);
×
2094
  }
2095
  TSC_ERR_RET(taosThreadMutexUnlock(&pStmt->asyncBindParam.mutex));
4,275!
2096

2097
  if (pStmt->sql.stbInterlaceMode) {
4,278✔
2098
    STMT_ERR_RET(stmtAddBatch2(pStmt));
4,225!
2099
  }
2100

2101
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_EXECUTE));
4,279✔
2102

2103
  if (STMT_TYPE_QUERY != pStmt->sql.type) {
4,278✔
2104
    if (pStmt->sql.stbInterlaceMode) {
4,273✔
2105
      int64_t startTs = taosGetTimestampUs();
4,226✔
2106
      while (atomic_load_64(&pStmt->sql.siInfo.tbRemainNum)) {
8,750✔
2107
        taosUsleep(1);
4,532✔
2108
      }
2109
      pStmt->stat.execWaitUs += taosGetTimestampUs() - startTs;
4,218✔
2110
      STMT_ERR_RET(qBuildStmtFinOutput(pStmt->sql.pQuery, pStmt->sql.pVgHash, pStmt->sql.siInfo.pVgroupList));
4,218!
2111
      taosHashCleanup(pStmt->sql.siInfo.pVgroupHash);
4,224✔
2112
      pStmt->sql.siInfo.pVgroupHash = NULL;
4,224✔
2113
      pStmt->sql.siInfo.pVgroupList = NULL;
4,224✔
2114
    } else {
2115
      tDestroySubmitTbData(pStmt->exec.pCurrTbData, TSDB_MSG_FLG_ENCODE);
47✔
2116
      taosMemoryFreeClear(pStmt->exec.pCurrTbData);
47!
2117

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

2120
      STMT_ERR_RET(qBuildStmtOutput(pStmt->sql.pQuery, pStmt->sql.pVgHash, pStmt->exec.pBlockHash));
47!
2121
    }
2122
  }
2123

2124
  SRequestObj*      pRequest = pStmt->exec.pRequest;
4,274✔
2125
  __taos_async_fn_t fp = pStmt->options.asyncExecFn;
4,274✔
2126

2127
  if (!fp) {
4,274✔
2128
    launchQueryImpl(pStmt->exec.pRequest, pStmt->sql.pQuery, true, NULL);
4,262✔
2129

2130
    if (pStmt->exec.pRequest->code && NEED_CLIENT_HANDLE_ERROR(pStmt->exec.pRequest->code)) {
4,259!
UNCOV
2131
      code = refreshMeta(pStmt->exec.pRequest->pTscObj, pStmt->exec.pRequest);
×
UNCOV
2132
      if (code) {
×
UNCOV
2133
        pStmt->exec.pRequest->code = code;
×
2134
      } else {
UNCOV
2135
        STMT_ERR_RET(stmtResetStmt(pStmt));
×
UNCOV
2136
        STMT_ERR_RET(TSDB_CODE_NEED_RETRY);
×
2137
      }
2138
    }
2139

2140
    STMT_ERR_JRET(pStmt->exec.pRequest->code);
4,260!
2141

2142
    pStmt->exec.affectedRows = taos_affected_rows(pStmt->exec.pRequest);
4,260✔
2143
    if (affected_rows) {
4,261✔
2144
      *affected_rows = pStmt->exec.affectedRows;
4,249✔
2145
    }
2146
    pStmt->affectedRows += pStmt->exec.affectedRows;
4,261✔
2147

2148
    while (0 == atomic_load_8((int8_t*)&pStmt->sql.siInfo.tableColsReady)) {
4,261!
UNCOV
2149
      taosUsleep(1);
×
2150
    }
2151

2152
    STMT_ERR_RET(stmtCleanExecInfo(pStmt, (code ? false : true), false));
4,256✔
2153

2154
    ++pStmt->sql.runTimes;
4,255✔
2155
  } else {
2156
    SSqlCallbackWrapper* pWrapper = taosMemoryCalloc(1, sizeof(SSqlCallbackWrapper));
12!
2157
    if (pWrapper == NULL) {
12!
UNCOV
2158
      code = terrno;
×
2159
    } else {
2160
      pWrapper->pRequest = pRequest;
12✔
2161
      pRequest->pWrapper = pWrapper;
12✔
2162
    }
2163
    if (TSDB_CODE_SUCCESS == code) {
12!
2164
      code = createParseContext(pRequest, &pWrapper->pParseCtx, pWrapper);
12✔
2165
    }
2166
    pRequest->syncQuery = false;
12✔
2167
    pRequest->body.queryFp = asyncQueryCb;
12✔
2168
    ((SSyncQueryParam*)(pRequest)->body.interParam)->userParam = pStmt;
12✔
2169

2170
    pStmt->execSemWaited = false;
12✔
2171
    launchAsyncQuery(pRequest, pStmt->sql.pQuery, NULL, pWrapper);
12✔
2172
  }
2173

2174
_return:
4,267✔
2175
  pStmt->stat.execUseUs += taosGetTimestampUs() - startUs;
4,264✔
2176

2177
  STMT_RET(code);
4,264!
2178
}
2179

2180
int stmtClose2(TAOS_STMT2* stmt) {
143✔
2181
  STscStmt2* pStmt = (STscStmt2*)stmt;
143✔
2182

2183
  STMT_DLOG_E("start to free stmt");
143!
2184
  taosMemoryFreeClear(pStmt->db);
143!
2185

2186
  if (pStmt->bindThreadInUse) {
143✔
2187
    while (0 == atomic_load_8((int8_t*)&pStmt->sql.siInfo.tableColsReady)) {
79!
UNCOV
2188
      taosUsleep(1);
×
2189
    }
2190
    (void)taosThreadMutexLock(&pStmt->queue.mutex);
79✔
2191
    pStmt->queue.stopQueue = true;
79✔
2192
    (void)taosThreadCondSignal(&(pStmt->queue.waitCond));
79✔
2193
    (void)taosThreadMutexUnlock(&pStmt->queue.mutex);
79✔
2194

2195
    (void)taosThreadJoin(pStmt->bindThread, NULL);
79✔
2196
    pStmt->bindThreadInUse = false;
79✔
2197

2198
    (void)taosThreadCondDestroy(&pStmt->queue.waitCond);
79✔
2199
    (void)taosThreadMutexDestroy(&pStmt->queue.mutex);
79✔
2200
  }
2201

2202
  TSC_ERR_RET(taosThreadMutexLock(&pStmt->asyncBindParam.mutex));
143!
2203
  while (atomic_load_8((int8_t*)&pStmt->asyncBindParam.asyncBindNum) > 0) {
143!
2204
    (void)taosThreadCondWait(&pStmt->asyncBindParam.waitCond, &pStmt->asyncBindParam.mutex);
×
2205
  }
2206
  TSC_ERR_RET(taosThreadMutexUnlock(&pStmt->asyncBindParam.mutex));
143!
2207

2208
  (void)taosThreadCondDestroy(&pStmt->asyncBindParam.waitCond);
143✔
2209
  (void)taosThreadMutexDestroy(&pStmt->asyncBindParam.mutex);
143✔
2210

2211
  if (pStmt->options.asyncExecFn && !pStmt->execSemWaited) {
143!
2212
    if (tsem_wait(&pStmt->asyncExecSem) != 0) {
5!
UNCOV
2213
      tscError("failed to wait asyncExecSem");
×
2214
    }
2215
  }
2216

2217
  STMT_DLOG("stmt %p closed, stbInterlaceMode:%d, statInfo: ctgGetTbMetaNum=>%" PRId64 ", getCacheTbInfo=>%" PRId64
143!
2218
            ", parseSqlNum=>%" PRId64 ", pStmt->stat.bindDataNum=>%" PRId64
2219
            ", settbnameAPI:%u, bindAPI:%u, addbatchAPI:%u, execAPI:%u"
2220
            ", setTbNameUs:%" PRId64 ", bindDataUs:%" PRId64 ",%" PRId64 ",%" PRId64 ",%" PRId64 " addBatchUs:%" PRId64
2221
            ", execWaitUs:%" PRId64 ", execUseUs:%" PRId64,
2222
            pStmt, pStmt->sql.stbInterlaceMode, pStmt->stat.ctgGetTbMetaNum, pStmt->stat.getCacheTbInfo,
2223
            pStmt->stat.parseSqlNum, pStmt->stat.bindDataNum, pStmt->seqIds[STMT_SETTBNAME], pStmt->seqIds[STMT_BIND],
2224
            pStmt->seqIds[STMT_ADD_BATCH], pStmt->seqIds[STMT_EXECUTE], pStmt->stat.setTbNameUs,
2225
            pStmt->stat.bindDataUs1, pStmt->stat.bindDataUs2, pStmt->stat.bindDataUs3, pStmt->stat.bindDataUs4,
2226
            pStmt->stat.addBatchUs, pStmt->stat.execWaitUs, pStmt->stat.execUseUs);
2227
  if (pStmt->sql.stbInterlaceMode) {
143✔
2228
    pStmt->bInfo.tagsCached = false;
73✔
2229
  }
2230

2231
  STMT_ERR_RET(stmtCleanSQLInfo(pStmt));
143!
2232

2233
  if (pStmt->options.asyncExecFn) {
143✔
2234
    if (tsem_destroy(&pStmt->asyncExecSem) != 0) {
5!
UNCOV
2235
      tscError("failed to destroy asyncExecSem");
×
2236
    }
2237
  }
2238
  taosMemoryFree(stmt);
143!
2239

2240
  return TSDB_CODE_SUCCESS;
143✔
2241
}
2242

2243
const char* stmtErrstr2(TAOS_STMT2* stmt) {
3✔
2244
  STscStmt2* pStmt = (STscStmt2*)stmt;
3✔
2245

2246
  if (stmt == NULL || NULL == pStmt->exec.pRequest) {
3!
2247
    return (char*)tstrerror(terrno);
3✔
2248
  }
2249

UNCOV
2250
  pStmt->exec.pRequest->code = terrno;
×
2251

UNCOV
2252
  SRequestObj* pRequest = pStmt->exec.pRequest;
×
UNCOV
2253
  if (NULL != pRequest->msgBuf && (strlen(pRequest->msgBuf) > 0 || pRequest->code == TSDB_CODE_RPC_FQDN_ERROR)) {
×
UNCOV
2254
    return pRequest->msgBuf;
×
2255
  }
UNCOV
2256
  return (const char*)tstrerror(pRequest->code);
×
2257
}
2258
/*
2259
int stmtAffectedRows(TAOS_STMT* stmt) { return ((STscStmt2*)stmt)->affectedRows; }
2260

2261
int stmtAffectedRowsOnce(TAOS_STMT* stmt) { return ((STscStmt2*)stmt)->exec.affectedRows; }
2262
*/
2263

2264
int stmtParseColFields2(TAOS_STMT2* stmt) {
57✔
2265
  int32_t    code = 0;
57✔
2266
  STscStmt2* pStmt = (STscStmt2*)stmt;
57✔
2267
  int32_t    preCode = pStmt->errCode;
57✔
2268

2269
  STMT_DLOG_E("start to get col fields");
57!
2270

2271
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
57!
UNCOV
2272
    return pStmt->errCode;
×
2273
  }
2274

2275
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
57!
UNCOV
2276
    STMT_ERRI_JRET(TSDB_CODE_TSC_STMT_API_ERROR);
×
2277
  }
2278

2279
  STMT_ERRI_JRET(stmtSwitchStatus(pStmt, STMT_FETCH_FIELDS));
57!
2280

2281
  if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 &&
57!
2282
      STMT_TYPE_MULTI_INSERT != pStmt->sql.type) {
3!
UNCOV
2283
    pStmt->bInfo.needParse = false;
×
2284
  }
2285
  if (pStmt->sql.stbInterlaceMode && pStmt->sql.siInfo.pDataCtx != NULL) {
57✔
2286
    pStmt->bInfo.needParse = false;
7✔
2287
  }
2288
  if (pStmt->exec.pRequest && STMT_TYPE_QUERY == pStmt->sql.type && pStmt->sql.runTimes) {
57!
UNCOV
2289
    taos_free_result(pStmt->exec.pRequest);
×
UNCOV
2290
    pStmt->exec.pRequest = NULL;
×
UNCOV
2291
    STMT_ERRI_JRET(stmtCreateRequest(pStmt));
×
2292
  }
2293

2294
  STMT_ERRI_JRET(stmtCreateRequest(pStmt));
57!
2295

2296
  if (pStmt->bInfo.needParse) {
57✔
2297
    STMT_ERRI_JRET(stmtParseSql(pStmt));
50✔
2298
  }
2299

2300
_return:
42✔
2301
  // compatible with previous versions
2302
  if (code == TSDB_CODE_PAR_TABLE_NOT_EXIST && (pStmt->bInfo.tbNameFlag & NO_DATA_USING_CLAUSE) == 0x0) {
57!
2303
    code = TSDB_CODE_TSC_STMT_TBNAME_ERROR;
1✔
2304
  }
2305

2306
  if (code != TSDB_CODE_SUCCESS) {
57✔
2307
    taos_free_result(pStmt->exec.pRequest);
15✔
2308
    pStmt->exec.pRequest = NULL;
15✔
2309
  }
2310

2311
  pStmt->errCode = preCode;
57✔
2312

2313
  return code;
57✔
2314
}
2315

2316
int stmtGetStbColFields2(TAOS_STMT2* stmt, int* nums, TAOS_FIELD_ALL** fields) {
57✔
2317
  int32_t code = stmtParseColFields2(stmt);
57✔
2318
  if (code != TSDB_CODE_SUCCESS) {
57✔
2319
    return code;
15✔
2320
  }
2321

2322
  return stmtFetchStbColFields2(stmt, nums, fields);
42✔
2323
}
2324

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

2330
  STMT_DLOG_E("start to get param num");
11!
2331

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

2336
  STMT_ERRI_JRET(stmtSwitchStatus(pStmt, STMT_FETCH_FIELDS));
11!
2337

2338
  if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 &&
11!
UNCOV
2339
      STMT_TYPE_MULTI_INSERT != pStmt->sql.type) {
×
2340
    pStmt->bInfo.needParse = false;
×
2341
  }
2342

2343
  if (pStmt->exec.pRequest && STMT_TYPE_QUERY == pStmt->sql.type && pStmt->sql.runTimes) {
11!
2344
    taos_free_result(pStmt->exec.pRequest);
×
2345
    pStmt->exec.pRequest = NULL;
×
2346
  }
2347

2348
  STMT_ERRI_JRET(stmtCreateRequest(pStmt));
11!
2349

2350
  if (pStmt->bInfo.needParse) {
11!
2351
    STMT_ERRI_JRET(stmtParseSql(pStmt));
11✔
2352
  }
2353

2354
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
4!
2355
    *nums = taosArrayGetSize(pStmt->sql.pQuery->pPlaceholderValues);
4✔
2356
  } else {
UNCOV
2357
    STMT_ERRI_JRET(stmtFetchColFields2(stmt, nums, NULL));
×
2358
  }
2359

UNCOV
2360
_return:
×
2361
  if (code != TSDB_CODE_SUCCESS) {
11✔
2362
    taos_free_result(pStmt->exec.pRequest);
7✔
2363
    pStmt->exec.pRequest = NULL;
7✔
2364
  }
2365
  pStmt->errCode = preCode;
11✔
2366

2367
  return code;
11✔
2368
}
2369

2370
TAOS_RES* stmtUseResult2(TAOS_STMT2* stmt) {
5✔
2371
  STscStmt2* pStmt = (STscStmt2*)stmt;
5✔
2372

2373
  STMT_DLOG_E("start to use result");
5!
2374

2375
  if (STMT_TYPE_QUERY != pStmt->sql.type) {
5!
UNCOV
2376
    tscError("useResult only for query statement");
×
UNCOV
2377
    return NULL;
×
2378
  }
2379

2380
  return pStmt->exec.pRequest;
5✔
2381
}
2382

UNCOV
2383
int32_t stmtAsyncBindThreadFunc(void* args) {
×
UNCOV
2384
  qInfo("async stmt bind thread started");
×
2385

UNCOV
2386
  ThreadArgs* targs = (ThreadArgs*)args;
×
UNCOV
2387
  STscStmt2*  pStmt = (STscStmt2*)targs->stmt;
×
2388

UNCOV
2389
  int code = taos_stmt2_bind_param(targs->stmt, targs->bindv, targs->col_idx);
×
UNCOV
2390
  targs->fp(targs->param, NULL, code);
×
UNCOV
2391
  (void)taosThreadMutexLock(&(pStmt->asyncBindParam.mutex));
×
UNCOV
2392
  (void)atomic_sub_fetch_8(&pStmt->asyncBindParam.asyncBindNum, 1);
×
UNCOV
2393
  (void)taosThreadCondSignal(&(pStmt->asyncBindParam.waitCond));
×
UNCOV
2394
  (void)taosThreadMutexUnlock(&(pStmt->asyncBindParam.mutex));
×
UNCOV
2395
  taosMemoryFree(args);
×
2396

UNCOV
2397
  qInfo("async stmt bind thread stopped");
×
2398

UNCOV
2399
  return code;
×
2400
}
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