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

taosdata / TDengine / #4876

10 Dec 2025 05:56AM UTC coverage: 64.632% (+0.2%) from 64.472%
#4876

push

travis-ci

guanshengliang
test: fix idmp case with checkDataMemLoop checked (#33862)

4 of 9 new or added lines in 3 files covered. (44.44%)

380 existing lines in 104 files now uncovered.

162866 of 251990 relevant lines covered (64.63%)

107950382.52 hits per line

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

66.4
/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* gStmt2StatusStr[] = {"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) {
23,629,942✔
13
    *pBuf = (char*)pTblBuf->pCurBuff + pTblBuf->buffOffset;
23,629,313✔
14
    pTblBuf->buffOffset += pTblBuf->buffUnit;
23,629,600✔
15
  } else if (pTblBuf->buffIdx < taosArrayGetSize(pTblBuf->pBufList)) {
611✔
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;
23,629,327✔
39
}
40

41
static bool stmtDequeue(STscStmt2* pStmt, SStmtQNode** param) {
23,632,134✔
42
  int i = 0;
23,632,134✔
43
  while (0 == atomic_load_64((int64_t*)&pStmt->queue.qRemainNum)) {
108,271,698✔
44
    if (pStmt->queue.stopQueue) {
84,660,999✔
45
      return false;
20,673✔
46
    }
47
    if (i < 10) {
84,641,129✔
48
      taosUsleep(1);
78,388,800✔
49
      i++;
78,377,948✔
50
    } else {
51
      (void)taosThreadMutexLock(&pStmt->queue.mutex);
6,252,329✔
52
      if (pStmt->queue.stopQueue) {
6,262,224✔
53
        (void)taosThreadMutexUnlock(&pStmt->queue.mutex);
×
54
        return false;
×
55
      }
56
      if (0 == atomic_load_64((int64_t*)&pStmt->queue.qRemainNum)) {
6,261,959✔
57
        (void)taosThreadCondWait(&pStmt->queue.waitCond, &pStmt->queue.mutex);
6,260,556✔
58
      }
59
      (void)taosThreadMutexUnlock(&pStmt->queue.mutex);
6,261,091✔
60
    }
61
  }
62

63
  if (pStmt->queue.stopQueue && 0 == atomic_load_64((int64_t*)&pStmt->queue.qRemainNum)) {
23,588,656✔
64
    return false;
×
65
  }
66

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

75
  SStmtQNode* node = pStmt->queue.head->next;
23,611,829✔
76
  pStmt->queue.head->next = node->next;
23,612,098✔
77
  if (pStmt->queue.tail == node) {
23,611,829✔
78
    pStmt->queue.tail = pStmt->queue.head;
14,669,195✔
79
  }
80
  node->next = NULL;
23,611,829✔
81
  *param = node;
23,611,829✔
82

83
  (void)atomic_sub_fetch_64((int64_t*)&pStmt->queue.qRemainNum, 1);
23,611,829✔
84
  (void)taosThreadMutexUnlock(&pStmt->queue.mutex);
23,613,294✔
85

86
  STMT2_TLOG("dequeue success, node:%p, remainNum:%" PRId64, node, pStmt->queue.qRemainNum);
23,612,409✔
87

88
  return true;
23,612,832✔
89
}
90

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

97
  param->next = NULL;
23,606,056✔
98

99
  (void)taosThreadMutexLock(&pStmt->queue.mutex);
23,606,325✔
100

101
  pStmt->queue.tail->next = param;
23,611,314✔
102
  pStmt->queue.tail = param;
23,611,852✔
103
  pStmt->stat.bindDataNum++;
23,610,776✔
104

105
  (void)atomic_add_fetch_64(&pStmt->queue.qRemainNum, 1);
23,611,852✔
106
  (void)taosThreadCondSignal(&(pStmt->queue.waitCond));
23,611,931✔
107

108
  (void)taosThreadMutexUnlock(&pStmt->queue.mutex);
23,610,243✔
109

110
  STMT2_TLOG("enqueue param:%p, remainNum:%" PRId64 ", restoreTbCols:%d", param, pStmt->queue.qRemainNum,
23,612,089✔
111
             param->restoreTbCols);
112
}
113

114
static int32_t stmtCreateRequest(STscStmt2* pStmt) {
28,314,591✔
115
  int32_t code = 0;
28,314,591✔
116

117
  if (pStmt->exec.pRequest == NULL) {
28,314,591✔
118
    code = buildRequest(pStmt->taos->id, pStmt->sql.sqlStr, pStmt->sql.sqlLen, NULL, false, &pStmt->exec.pRequest,
21,625✔
119
                        pStmt->reqid);
120
    if (pStmt->reqid != 0) {
21,625✔
121
      pStmt->reqid++;
241✔
122
    }
123
    pStmt->exec.pRequest->type = RES_TYPE__QUERY;
21,625✔
124
    if (pStmt->db != NULL) {
21,625✔
125
      taosMemoryFreeClear(pStmt->exec.pRequest->pDb);
21,531✔
126
      pStmt->exec.pRequest->pDb = taosStrdup(pStmt->db);
21,531✔
127
    }
128
    if (TSDB_CODE_SUCCESS == code) {
21,719✔
129
      pStmt->exec.pRequest->syncQuery = true;
21,625✔
130
      pStmt->exec.pRequest->stmtBindVersion = 2;
21,625✔
131
    }
132
    STMT2_DLOG("create request:0x%" PRIx64 ", QID:0x%" PRIx64, pStmt->exec.pRequest->self,
21,719✔
133
               pStmt->exec.pRequest->requestId);
134
  }
135

136
  return code;
28,315,198✔
137
}
138

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

142
  if (newStatus >= STMT_INIT && newStatus < STMT_MAX) {
59,266,566✔
143
    STMT2_LOG_SEQ(newStatus);
59,277,498✔
144
  }
145

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

151
  switch (newStatus) {
59,280,324✔
152
    case STMT_PREPARE:
21,531✔
153
      pStmt->errCode = 0;
21,531✔
154
      break;
14,814✔
155
    case STMT_SETTBNAME:
16,217,988✔
156
      if (STMT_STATUS_EQ(INIT)) {
16,217,988✔
157
        code = TSDB_CODE_TSC_STMT_API_ERROR;
×
158
      }
159
      if (!pStmt->sql.stbInterlaceMode && (STMT_STATUS_EQ(BIND) || STMT_STATUS_EQ(BIND_COL))) {
16,218,261✔
160
        code = TSDB_CODE_TSC_STMT_API_ERROR;
×
161
      }
162
      break;
16,218,253✔
163
    case STMT_SETTAGS:
12,039,258✔
164
      if (STMT_STATUS_EQ(INIT)) {
12,039,258✔
165
        code = TSDB_CODE_TSC_STMT_API_ERROR;
×
166
      }
167
      break;
12,039,258✔
168
    case STMT_FETCH_FIELDS:
703✔
169
      if (STMT_STATUS_EQ(INIT)) {
703✔
170
        code = TSDB_CODE_TSC_STMT_API_ERROR;
×
171
      }
172
      break;
703✔
173
    case STMT_BIND:
16,215,452✔
174
      if (STMT_STATUS_EQ(INIT) || STMT_STATUS_EQ(BIND_COL)) {
16,215,452✔
175
        code = TSDB_CODE_TSC_STMT_API_ERROR;
×
176
      }
177
      /*
178
            if ((pStmt->sql.type == STMT_TYPE_MULTI_INSERT) && ()) {
179
              code = TSDB_CODE_TSC_STMT_API_ERROR;
180
            }
181
      */
182
      break;
16,215,456✔
183
    case STMT_BIND_COL:
×
184
      if (STMT_STATUS_EQ(INIT) || STMT_STATUS_EQ(BIND)) {
×
185
        code = TSDB_CODE_TSC_STMT_API_ERROR;
×
186
      }
187
      break;
×
188
    case STMT_ADD_BATCH:
7,394,099✔
189
      if (STMT_STATUS_NE(BIND) && STMT_STATUS_NE(BIND_COL) && STMT_STATUS_NE(FETCH_FIELDS)) {
7,394,099✔
190
        code = TSDB_CODE_TSC_STMT_API_ERROR;
×
191
      }
192
      break;
7,394,099✔
193
    case STMT_EXECUTE:
7,391,293✔
194
      if (STMT_TYPE_QUERY == pStmt->sql.type) {
7,391,293✔
195
        if (STMT_STATUS_NE(ADD_BATCH) && STMT_STATUS_NE(FETCH_FIELDS) && STMT_STATUS_NE(BIND) &&
242✔
196
            STMT_STATUS_NE(BIND_COL)) {
×
197
          code = TSDB_CODE_TSC_STMT_API_ERROR;
×
198
        }
199
      } else {
200
        if (STMT_STATUS_NE(ADD_BATCH) && STMT_STATUS_NE(FETCH_FIELDS)) {
7,390,244✔
201
          code = TSDB_CODE_TSC_STMT_API_ERROR;
×
202
        }
203
      }
204
      break;
7,391,293✔
205
    default:
×
206
      code = TSDB_CODE_APP_ERROR;
×
207
      break;
×
208
  }
209

210
  STMT_ERR_RET(code);
59,273,876✔
211

212
  pStmt->sql.status = newStatus;
59,273,876✔
213

214
  return TSDB_CODE_SUCCESS;
59,280,059✔
215
}
216

217
static int32_t stmtGetTbName(TAOS_STMT2* stmt, char** tbName) {
22,086✔
218
  STscStmt2* pStmt = (STscStmt2*)stmt;
22,086✔
219

220
  pStmt->sql.type = STMT_TYPE_MULTI_INSERT;
22,086✔
221

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

227
  *tbName = pStmt->bInfo.tbName;
21,383✔
228

229
  return TSDB_CODE_SUCCESS;
21,383✔
230
}
231

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

241
  if ((tags != NULL && ((SBoundColInfo*)tags)->numOfCols == 0) || !autoCreateTbl) {
21,723✔
242
    pStmt->sql.autoCreateTbl = false;
16,901✔
243
  }
244

245
  (void)memcpy(&pStmt->bInfo.sname, tbName, sizeof(*tbName));
21,992✔
246
  tstrncpy(pStmt->bInfo.tbFName, tbFName, sizeof(pStmt->bInfo.tbFName));
21,992✔
247
  pStmt->bInfo.tbFName[sizeof(pStmt->bInfo.tbFName) - 1] = 0;
21,723✔
248

249
  pStmt->bInfo.tbUid = autoCreateTbl ? 0 : pTableMeta->uid;
21,992✔
250
  pStmt->bInfo.tbSuid = pTableMeta->suid;
21,992✔
251
  pStmt->bInfo.tbVgId = pTableMeta->vgId;
21,992✔
252
  pStmt->bInfo.tbType = pTableMeta->tableType;
21,992✔
253

254
  if (!pStmt->bInfo.tagsCached) {
21,992✔
255
    qDestroyBoundColInfo(pStmt->bInfo.boundTags);
21,534✔
256
    taosMemoryFreeClear(pStmt->bInfo.boundTags);
21,709✔
257
  }
258

259
  if (cols) {
21,709✔
260
    pStmt->bInfo.boundCols =
×
261
        tSimpleHashInit(taosArrayGetSize(cols), taosGetDefaultHashFunction(TSDB_DATA_TYPE_SMALLINT));
×
262
    if (pStmt->bInfo.boundCols) {
×
263
      for (int32_t i = 0; i < taosArrayGetSize(cols); i++) {
×
264
        SColVal* pColVal = taosArrayGet(cols, i);
×
265
        if (pColVal) {
×
266
          code = tSimpleHashPut(pStmt->bInfo.boundCols, &pColVal->cid, sizeof(int16_t), pColVal, sizeof(SColVal));
×
267
          if (code != 0) {
×
268
            return code;
×
269
          }
270
        }
271
      }
272
    }
273
  } else {
274
    pStmt->bInfo.boundCols = NULL;
21,709✔
275
  }
276
  pStmt->bInfo.boundTags = tags;
21,992✔
277
  pStmt->bInfo.tagsCached = false;
21,709✔
278
  pStmt->bInfo.tbNameFlag = tbNameFlag;
21,992✔
279
  tstrncpy(pStmt->bInfo.stbFName, sTableName, sizeof(pStmt->bInfo.stbFName));
21,440✔
280

281
  if (pTableMeta->tableType != TSDB_CHILD_TABLE && pTableMeta->tableType != TSDB_SUPER_TABLE) {
21,992✔
282
    pStmt->sql.stbInterlaceMode = false;
×
283
  }
284

285
  return TSDB_CODE_SUCCESS;
21,440✔
286
}
287

288
static int32_t stmtUpdateExecInfo(TAOS_STMT2* stmt, SHashObj* pVgHash, SHashObj* pBlockHash) {
21,709✔
289
  STscStmt2* pStmt = (STscStmt2*)stmt;
21,709✔
290

291
  pStmt->sql.pVgHash = pVgHash;
21,709✔
292
  pStmt->exec.pBlockHash = pBlockHash;
21,440✔
293

294
  return TSDB_CODE_SUCCESS;
21,723✔
295
}
296

297
static int32_t stmtUpdateInfo(TAOS_STMT2* stmt, STableMeta* pTableMeta, void* tags, SArray* cols, SName* tbName,
21,803✔
298
                              bool autoCreateTbl, SHashObj* pVgHash, SHashObj* pBlockHash, const char* sTableName,
299
                              uint8_t tbNameFlag) {
300
  STscStmt2* pStmt = (STscStmt2*)stmt;
21,803✔
301

302
  STMT_ERR_RET(stmtUpdateBindInfo(stmt, pTableMeta, tags, cols, tbName, sTableName, autoCreateTbl, tbNameFlag));
21,803✔
303
  STMT_ERR_RET(stmtUpdateExecInfo(stmt, pVgHash, pBlockHash));
21,723✔
304

305
  pStmt->sql.autoCreateTbl = autoCreateTbl;
21,723✔
306

307
  return TSDB_CODE_SUCCESS;
21,992✔
308
}
309

310
static int32_t stmtGetExecInfo(TAOS_STMT2* stmt, SHashObj** pVgHash, SHashObj** pBlockHash) {
703✔
311
  STscStmt2* pStmt = (STscStmt2*)stmt;
703✔
312

313
  *pVgHash = pStmt->sql.pVgHash;
703✔
314
  pStmt->sql.pVgHash = NULL;
703✔
315

316
  *pBlockHash = pStmt->exec.pBlockHash;
703✔
317
  pStmt->exec.pBlockHash = NULL;
703✔
318

319
  return TSDB_CODE_SUCCESS;
703✔
320
}
321

322
static int32_t stmtParseSql(STscStmt2* pStmt) {
21,676✔
323
  pStmt->exec.pCurrBlock = NULL;
21,676✔
324

325
  SStmtCallback stmtCb = {
22,234✔
326
      .pStmt = pStmt,
327
      .getTbNameFn = stmtGetTbName,
328
      .setInfoFn = stmtUpdateInfo,
329
      .getExecInfoFn = stmtGetExecInfo,
330
  };
331

332
  STMT_ERR_RET(stmtCreateRequest(pStmt));
22,234✔
333
  pStmt->exec.pRequest->stmtBindVersion = 2;
22,234✔
334

335
  pStmt->stat.parseSqlNum++;
22,234✔
336

337
  STMT2_DLOG("start to parse, QID:0x%" PRIx64, pStmt->exec.pRequest->requestId);
22,234✔
338
  STMT_ERR_RET(parseSql(pStmt->exec.pRequest, false, &pStmt->sql.pQuery, &stmtCb));
22,234✔
339

340
  pStmt->sql.siInfo.pQuery = pStmt->sql.pQuery;
22,234✔
341

342
  pStmt->bInfo.needParse = false;
22,234✔
343

344
  if (pStmt->sql.type == 0) {
22,234✔
345
    if (pStmt->sql.pQuery->pRoot && LEGAL_INSERT(nodeType(pStmt->sql.pQuery->pRoot))) {
242✔
346
      pStmt->sql.type = STMT_TYPE_INSERT;
×
347
      pStmt->sql.stbInterlaceMode = false;
×
348
    } else if (pStmt->sql.pQuery->pPrepareRoot && LEGAL_SELECT(nodeType(pStmt->sql.pQuery->pPrepareRoot))) {
242✔
349
      pStmt->sql.type = STMT_TYPE_QUERY;
242✔
350
      pStmt->sql.stbInterlaceMode = false;
242✔
351

352
      return TSDB_CODE_SUCCESS;
242✔
353
    } else {
354
      STMT2_ELOG_E("only support select or insert sql");
×
355
      if (pStmt->exec.pRequest->msgBuf) {
×
356
        tstrncpy(pStmt->exec.pRequest->msgBuf, "stmt only support select or insert", pStmt->exec.pRequest->msgBufLen);
×
357
      }
358
      return TSDB_CODE_PAR_SYNTAX_ERROR;
×
359
    }
360
  } else if (pStmt->sql.type == STMT_TYPE_QUERY) {
21,440✔
361
    pStmt->sql.stbInterlaceMode = false;
×
362
    return TSDB_CODE_SUCCESS;
×
363
  } else if (pStmt->sql.type == STMT_TYPE_INSERT) {
21,992✔
364
    pStmt->sql.stbInterlaceMode = false;
×
365
  }
366

367
  STableDataCxt** pSrc =
368
      (STableDataCxt**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
21,992✔
369
  if (NULL == pSrc || NULL == *pSrc) {
22,086✔
370
    STMT2_ELOG("fail to get exec.pBlockHash, maybe parse failed, tbFName:%s", pStmt->bInfo.tbFName);
×
371
    STMT_ERR_RET(TSDB_CODE_TSC_STMT_CACHE_ERROR);
×
372
  }
373

374
  STableDataCxt* pTableCtx = *pSrc;
22,086✔
375
  if (pStmt->sql.stbInterlaceMode && pTableCtx->pData->pCreateTbReq && (pStmt->bInfo.tbNameFlag & USING_CLAUSE) == 0) {
22,086✔
376
    STMT2_TLOG("destroy pCreateTbReq for no-using insert, tbFName:%s", pStmt->bInfo.tbFName);
×
377
    tdDestroySVCreateTbReq(pTableCtx->pData->pCreateTbReq);
×
378
    taosMemoryFreeClear(pTableCtx->pData->pCreateTbReq);
×
379
    pTableCtx->pData->pCreateTbReq = NULL;
×
380
  }
381
  // if (pStmt->sql.stbInterlaceMode) {
382
  //   int16_t lastIdx = -1;
383

384
  //   for (int32_t i = 0; i < pTableCtx->boundColsInfo.numOfBound; ++i) {
385
  //     if (pTableCtx->boundColsInfo.pColIndex[i] < lastIdx) {
386
  //       pStmt->sql.stbInterlaceMode = false;
387
  //       break;
388
  //     }
389

390
  //     lastIdx = pTableCtx->boundColsInfo.pColIndex[i];
391
  //   }
392
  // }
393

394
  if (NULL == pStmt->sql.pBindInfo) {
22,086✔
395
    pStmt->sql.pBindInfo = taosMemoryMalloc(pTableCtx->boundColsInfo.numOfBound * sizeof(*pStmt->sql.pBindInfo));
21,289✔
396
    if (NULL == pStmt->sql.pBindInfo) {
21,289✔
397
      STMT2_ELOG_E("fail to malloc pBindInfo");
×
398
      return terrno;
×
399
    }
400
  }
401

402
  return TSDB_CODE_SUCCESS;
22,086✔
403
}
404

405
static void resetRequest(STscStmt2* pStmt) {
24,351✔
406
  if (pStmt->exec.pRequest) {
24,351✔
407
    taos_free_result(pStmt->exec.pRequest);
21,625✔
408
    pStmt->exec.pRequest = NULL;
21,625✔
409
  }
410
  pStmt->asyncResultAvailable = false;
24,351✔
411
}
24,351✔
412

413
static int32_t stmtCleanBindInfo(STscStmt2* pStmt) {
7,431,517✔
414
  pStmt->bInfo.tbUid = 0;
7,431,517✔
415
  pStmt->bInfo.tbVgId = -1;
7,431,517✔
416
  pStmt->bInfo.tbType = 0;
7,431,517✔
417
  pStmt->bInfo.needParse = true;
7,431,517✔
418
  pStmt->bInfo.inExecCache = false;
7,431,517✔
419

420
  pStmt->bInfo.tbName[0] = 0;
7,431,517✔
421
  pStmt->bInfo.tbFName[0] = 0;
7,431,234✔
422
  if (!pStmt->bInfo.tagsCached) {
7,431,234✔
423
    qDestroyBoundColInfo(pStmt->bInfo.boundTags);
3,408,654✔
424
    taosMemoryFreeClear(pStmt->bInfo.boundTags);
3,408,032✔
425
  }
426

427
  if (!pStmt->bInfo.boundColsCached) {
7,430,626✔
428
    tSimpleHashCleanup(pStmt->bInfo.boundCols);
39,921✔
429
    pStmt->bInfo.boundCols = NULL;
39,921✔
430
  }
431

432
  if (!pStmt->sql.autoCreateTbl) {
7,430,074✔
433
    pStmt->bInfo.stbFName[0] = 0;
3,406,620✔
434
    pStmt->bInfo.tbSuid = 0;
3,406,351✔
435
  }
436

437
  STMT2_TLOG("finish clean bind info, tagsCached:%d, autoCreateTbl:%d", pStmt->bInfo.tagsCached,
7,430,343✔
438
             pStmt->sql.autoCreateTbl);
439

440
  return TSDB_CODE_SUCCESS;
7,430,499✔
441
}
442

443
static void stmtFreeTableBlkList(STableColsData* pTb) {
×
444
  (void)qResetStmtColumns(pTb->aCol, true);
×
445
  taosArrayDestroy(pTb->aCol);
×
446
}
×
447

448
static void stmtResetQueueTableBuf(STableBufInfo* pTblBuf, SStmtQueue* pQueue) {
7,385,803✔
449
  pTblBuf->pCurBuff = taosArrayGetP(pTblBuf->pBufList, 0);
7,385,803✔
450
  if (NULL == pTblBuf->pCurBuff) {
7,391,844✔
451
    tscError("QInfo:%p, fail to get buffer from list", pTblBuf);
2,127✔
452
    return;
×
453
  }
454
  pTblBuf->buffIdx = 1;
7,389,717✔
455
  pTblBuf->buffOffset = sizeof(*pQueue->head);
7,389,717✔
456

457
  pQueue->head = pQueue->tail = pTblBuf->pCurBuff;
7,389,448✔
458
  pQueue->qRemainNum = 0;
7,389,448✔
459
  pQueue->head->next = NULL;
7,389,165✔
460
}
461

462
static int32_t stmtCleanExecInfo(STscStmt2* pStmt, bool keepTable, bool deepClean) {
7,409,004✔
463
  if (pStmt->sql.stbInterlaceMode) {
7,409,004✔
464
    if (deepClean) {
7,408,479✔
465
      taosHashCleanup(pStmt->exec.pBlockHash);
20,673✔
466
      pStmt->exec.pBlockHash = NULL;
20,673✔
467

468
      if (NULL != pStmt->exec.pCurrBlock) {
20,673✔
469
        taosMemoryFreeClear(pStmt->exec.pCurrBlock->boundColsInfo.pColIndex);
20,673✔
470
        taosMemoryFreeClear(pStmt->exec.pCurrBlock->pData);
20,673✔
471
        qDestroyStmtDataBlock(pStmt->exec.pCurrBlock);
20,673✔
472
        pStmt->exec.pCurrBlock = NULL;
20,673✔
473
      }
474
      if (STMT_TYPE_QUERY != pStmt->sql.type) {
20,673✔
475
        resetRequest(pStmt);
20,673✔
476
      }
477
    } else {
478
      pStmt->sql.siInfo.pTableColsIdx = 0;
7,387,806✔
479
      stmtResetQueueTableBuf(&pStmt->sql.siInfo.tbBuf, &pStmt->queue);
7,387,792✔
480
      tSimpleHashClear(pStmt->sql.siInfo.pTableRowDataHash);
7,390,156✔
481
    }
482
    if (NULL != pStmt->exec.pRequest) {
7,412,630✔
483
      pStmt->exec.pRequest->body.resInfo.numOfRows = 0;
7,391,957✔
484
    }
485
  } else {
486
    if (STMT_TYPE_QUERY != pStmt->sql.type || deepClean) {
828✔
487
      resetRequest(pStmt);
72✔
488
    }
489

490
    size_t keyLen = 0;
1,904✔
491
    void*  pIter = taosHashIterate(pStmt->exec.pBlockHash, NULL);
1,904✔
492
    while (pIter) {
8,020✔
493
      STableDataCxt* pBlocks = *(STableDataCxt**)pIter;
6,116✔
494
      char*          key = taosHashGetKey(pIter, &keyLen);
6,116✔
495
      STableMeta*    pMeta = qGetTableMetaInDataBlock(pBlocks);
6,116✔
496

497
      if (keepTable && pBlocks == pStmt->exec.pCurrBlock) {
6,116✔
498
        TSWAP(pBlocks->pData, pStmt->exec.pCurrTbData);
710✔
499
        STMT_ERR_RET(qResetStmtDataBlock(pBlocks, false));
1,662✔
500

501
        pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
710✔
502
        continue;
710✔
503
      }
504

505
      qDestroyStmtDataBlock(pBlocks);
5,406✔
506
      STMT_ERR_RET(taosHashRemove(pStmt->exec.pBlockHash, key, keyLen));
5,406✔
507

508
      pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
5,406✔
509
    }
510

511
    if (keepTable) {
1,904✔
512
      STMT2_TLOG("finish clean exec info, stbInterlaceMode:%d, keepTable:%d, deepClean:%d", pStmt->sql.stbInterlaceMode,
952✔
513
                 keepTable, deepClean);
514
      return TSDB_CODE_SUCCESS;
952✔
515
    }
516

517
    taosHashCleanup(pStmt->exec.pBlockHash);
952✔
518
    pStmt->exec.pBlockHash = NULL;
952✔
519

520
    tDestroySubmitTbData(pStmt->exec.pCurrTbData, TSDB_MSG_FLG_ENCODE);
952✔
521
    taosMemoryFreeClear(pStmt->exec.pCurrTbData);
952✔
522
  }
523

524
  STMT_ERR_RET(stmtCleanBindInfo(pStmt));
7,413,582✔
525
  STMT2_TLOG("finish clean exec info, stbInterlaceMode:%d, keepTable:%d, deepClean:%d", pStmt->sql.stbInterlaceMode,
7,409,464✔
526
             keepTable, deepClean);
527

528
  return TSDB_CODE_SUCCESS;
7,409,813✔
529
}
530

531
static void stmtFreeTbBuf(void* buf) {
20,673✔
532
  void* pBuf = *(void**)buf;
20,673✔
533
  taosMemoryFree(pBuf);
20,673✔
534
}
20,673✔
535

536
static void stmtFreeTbCols(void* buf) {
20,673,000✔
537
  SArray* pCols = *(SArray**)buf;
20,673,000✔
538
  taosArrayDestroy(pCols);
20,673,000✔
539
}
20,673,000✔
540

541
static int32_t stmtCleanSQLInfo(STscStmt2* pStmt) {
19,609✔
542
  STMT2_TLOG_E("start to free SQL info");
19,609✔
543

544
  taosMemoryFree(pStmt->sql.pBindInfo);
19,609✔
545
  taosMemoryFree(pStmt->sql.queryRes.fields);
19,609✔
546
  taosMemoryFree(pStmt->sql.queryRes.userFields);
19,609✔
547
  taosMemoryFree(pStmt->sql.sqlStr);
19,609✔
548
  qDestroyQuery(pStmt->sql.pQuery);
19,609✔
549
  taosArrayDestroy(pStmt->sql.nodeList);
19,609✔
550
  taosHashCleanup(pStmt->sql.pVgHash);
19,609✔
551
  pStmt->sql.pVgHash = NULL;
19,609✔
552
  if (pStmt->sql.fixValueTags) {
19,609✔
553
    tdDestroySVCreateTbReq(pStmt->sql.fixValueTbReq);
504✔
554
  }
555

556
  void* pIter = taosHashIterate(pStmt->sql.pTableCache, NULL);
19,609✔
557
  while (pIter) {
20,319✔
558
    SStmtTableCache* pCache = (SStmtTableCache*)pIter;
710✔
559

560
    qDestroyStmtDataBlock(pCache->pDataCtx);
710✔
561
    qDestroyBoundColInfo(pCache->boundTags);
710✔
562
    taosMemoryFreeClear(pCache->boundTags);
710✔
563

564
    pIter = taosHashIterate(pStmt->sql.pTableCache, pIter);
710✔
565
  }
566
  taosHashCleanup(pStmt->sql.pTableCache);
19,609✔
567
  pStmt->sql.pTableCache = NULL;
19,609✔
568

569
  STMT_ERR_RET(stmtCleanExecInfo(pStmt, false, true));
19,609✔
570
  STMT_ERR_RET(stmtCleanBindInfo(pStmt));
19,609✔
571

572
  taos_free_result(pStmt->sql.siInfo.pRequest);
19,609✔
573
  taosHashCleanup(pStmt->sql.siInfo.pVgroupHash);
19,609✔
574
  tSimpleHashCleanup(pStmt->sql.siInfo.pTableHash);
19,609✔
575
  tSimpleHashCleanup(pStmt->sql.siInfo.pTableRowDataHash);
19,609✔
576
  taosArrayDestroyEx(pStmt->sql.siInfo.tbBuf.pBufList, stmtFreeTbBuf);
19,609✔
577
  taosMemoryFree(pStmt->sql.siInfo.pTSchema);
19,609✔
578
  qDestroyStmtDataBlock(pStmt->sql.siInfo.pDataCtx);
19,609✔
579
  taosArrayDestroyEx(pStmt->sql.siInfo.pTableCols, stmtFreeTbCols);
19,609✔
580
  pStmt->sql.siInfo.pTableCols = NULL;
19,609✔
581

582
  (void)memset(&pStmt->sql, 0, sizeof(pStmt->sql));
19,609✔
583
  pStmt->sql.siInfo.tableColsReady = true;
19,609✔
584

585
  STMT2_TLOG_E("end to free SQL info");
19,609✔
586

587
  return TSDB_CODE_SUCCESS;
19,609✔
588
}
589

590
static int32_t stmtTryAddTableVgroupInfo(STscStmt2* pStmt, int32_t* vgId) {
12,032,624✔
591
  if (*vgId >= 0 && taosHashGet(pStmt->sql.pVgHash, (const char*)vgId, sizeof(*vgId))) {
12,032,624✔
592
    return TSDB_CODE_SUCCESS;
×
593
  }
594

595
  SVgroupInfo      vgInfo = {0};
12,032,624✔
596
  SRequestConnInfo conn = {.pTrans = pStmt->taos->pAppInfo->pTransporter,
12,032,624✔
597
                           .requestId = pStmt->exec.pRequest->requestId,
12,032,624✔
598
                           .requestObjRefId = pStmt->exec.pRequest->self,
12,032,624✔
599
                           .mgmtEps = getEpSet_s(&pStmt->taos->pAppInfo->mgmtEp)};
12,032,624✔
600

601
  int32_t code = catalogGetTableHashVgroup(pStmt->pCatalog, &conn, &pStmt->bInfo.sname, &vgInfo);
12,038,018✔
602
  if (TSDB_CODE_SUCCESS != code) {
12,037,858✔
603
    STMT2_ELOG("fail to get vgroup info from catalog, code:%d", code);
×
604
    return code;
×
605
  }
606

607
  code =
608
      taosHashPut(pStmt->sql.pVgHash, (const char*)&vgInfo.vgId, sizeof(vgInfo.vgId), (char*)&vgInfo, sizeof(vgInfo));
12,037,858✔
609
  if (TSDB_CODE_SUCCESS != code) {
12,035,960✔
UNCOV
610
    STMT2_ELOG("fail to put vgroup info, code:%d", code);
×
611
    return code;
×
612
  }
613

614
  *vgId = vgInfo.vgId;
12,037,016✔
615

616
  return TSDB_CODE_SUCCESS;
12,037,016✔
617
}
618

619
int32_t stmtGetTableMetaAndValidate(STscStmt2* pStmt, uint64_t* uid, uint64_t* suid, int32_t* vgId, int8_t* tableType) {
17,170✔
620
  STableMeta*      pTableMeta = NULL;
17,170✔
621
  SRequestConnInfo conn = {.pTrans = pStmt->taos->pAppInfo->pTransporter,
17,170✔
622
                           .requestId = pStmt->exec.pRequest->requestId,
17,170✔
623
                           .requestObjRefId = pStmt->exec.pRequest->self,
17,170✔
624
                           .mgmtEps = getEpSet_s(&pStmt->taos->pAppInfo->mgmtEp)};
17,170✔
625
  int32_t          code = catalogGetTableMeta(pStmt->pCatalog, &conn, &pStmt->bInfo.sname, &pTableMeta);
17,170✔
626

627
  pStmt->stat.ctgGetTbMetaNum++;
17,170✔
628

629
  if (TSDB_CODE_PAR_TABLE_NOT_EXIST == code) {
17,170✔
630
    STMT2_ELOG("tb %s not exist", pStmt->bInfo.tbFName);
×
631
    (void)stmtCleanBindInfo(pStmt);
×
632

633
    if (!pStmt->sql.autoCreateTbl) {
×
634
      STMT2_ELOG("table %s does not exist and autoCreateTbl is disabled", pStmt->bInfo.tbFName);
×
635
      STMT_ERR_RET(TSDB_CODE_PAR_TABLE_NOT_EXIST);
×
636
    }
637

638
    STMT_ERR_RET(code);
×
639
  }
640

641
  STMT_ERR_RET(code);
16,618✔
642

643
  *uid = pTableMeta->uid;
16,618✔
644
  *suid = pTableMeta->suid;
17,170✔
645
  *tableType = pTableMeta->tableType;
17,170✔
646
  pStmt->bInfo.tbVgId = pTableMeta->vgId;
17,170✔
647
  *vgId = pTableMeta->vgId;
17,170✔
648

649
  taosMemoryFree(pTableMeta);
17,170✔
650

651
  return TSDB_CODE_SUCCESS;
17,170✔
652
}
653

654
static int32_t stmtRebuildDataBlock(STscStmt2* pStmt, STableDataCxt* pDataBlock, STableDataCxt** newBlock, uint64_t uid,
4,696✔
655
                                    uint64_t suid, int32_t vgId) {
656
  STMT_ERR_RET(stmtTryAddTableVgroupInfo(pStmt, &vgId));
4,696✔
657
  STMT_ERR_RET(qRebuildStmtDataBlock(newBlock, pDataBlock, uid, suid, vgId, pStmt->sql.autoCreateTbl));
4,696✔
658

659
  STMT2_DLOG("uid:%" PRId64 ", rebuild table data context, vgId:%d", uid, vgId);
4,696✔
660

661
  return TSDB_CODE_SUCCESS;
4,696✔
662
}
663

664
static int32_t stmtGetFromCache(STscStmt2* pStmt) {
25,985✔
665
  if (pStmt->sql.stbInterlaceMode && pStmt->sql.siInfo.pDataCtx) {
25,985✔
666
    pStmt->bInfo.needParse = false;
×
667
    pStmt->bInfo.inExecCache = false;
×
668
    return TSDB_CODE_SUCCESS;
×
669
  }
670

671
  pStmt->bInfo.needParse = true;
25,985✔
672
  pStmt->bInfo.inExecCache = false;
25,985✔
673

674
  STableDataCxt** pCxtInExec = taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
25,985✔
675
  if (pCxtInExec) {
26,079✔
676
    pStmt->bInfo.needParse = false;
×
677
    pStmt->bInfo.inExecCache = true;
×
678

679
    pStmt->exec.pCurrBlock = *pCxtInExec;
×
680

681
    if (pStmt->sql.autoCreateTbl) {
×
682
      STMT2_DLOG("reuse stmt block for tb %s in execBlock", pStmt->bInfo.tbFName);
×
683
      return TSDB_CODE_SUCCESS;
×
684
    }
685
  }
686

687
  if (NULL == pStmt->pCatalog) {
26,079✔
688
    STMT_ERR_RET(catalogGetHandle(pStmt->taos->pAppInfo->clusterId, &pStmt->pCatalog));
19,367✔
689
    pStmt->sql.siInfo.pCatalog = pStmt->pCatalog;
19,273✔
690
  }
691

692
  if (NULL == pStmt->sql.pTableCache || taosHashGetSize(pStmt->sql.pTableCache) <= 0) {
25,985✔
693
    if (pStmt->bInfo.inExecCache) {
21,383✔
694
      pStmt->bInfo.needParse = false;
×
695
      STMT2_DLOG("reuse stmt block for tb %s in execBlock", pStmt->bInfo.tbFName);
×
696
      return TSDB_CODE_SUCCESS;
×
697
    }
698

699
    STMT2_DLOG("no stmt block cache for tb %s", pStmt->bInfo.tbFName);
21,383✔
700

701
    return TSDB_CODE_SUCCESS;
21,289✔
702
  }
703

704
  if (pStmt->sql.autoCreateTbl) {
4,602✔
705
    SStmtTableCache* pCache = taosHashGet(pStmt->sql.pTableCache, &pStmt->bInfo.tbSuid, sizeof(pStmt->bInfo.tbSuid));
4,696✔
706
    if (pCache) {
4,696✔
707
      pStmt->bInfo.needParse = false;
4,696✔
708
      pStmt->bInfo.tbUid = 0;
4,696✔
709

710
      STableDataCxt* pNewBlock = NULL;
4,696✔
711
      STMT_ERR_RET(stmtRebuildDataBlock(pStmt, pCache->pDataCtx, &pNewBlock, 0, pStmt->bInfo.tbSuid, -1));
4,696✔
712

713
      if (taosHashPut(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName), &pNewBlock,
4,696✔
714
                      POINTER_BYTES)) {
715
        STMT_ERR_RET(terrno);
×
716
      }
717

718
      pStmt->exec.pCurrBlock = pNewBlock;
4,696✔
719

720
      STMT2_DLOG("reuse stmt block for tb %s in sqlBlock, suid:0x%" PRIx64, pStmt->bInfo.tbFName, pStmt->bInfo.tbSuid);
4,696✔
721

722
      return TSDB_CODE_SUCCESS;
4,696✔
723
    }
724

725
    STMT_RET(stmtCleanBindInfo(pStmt));
×
726
  }
727

728
  uint64_t uid, suid;
×
729
  int32_t  vgId;
×
730
  int8_t   tableType;
×
731

732
  STMT_ERR_RET(stmtGetTableMetaAndValidate(pStmt, &uid, &suid, &vgId, &tableType));
×
733

734
  uint64_t cacheUid = (TSDB_CHILD_TABLE == tableType) ? suid : uid;
×
735

736
  if (uid == pStmt->bInfo.tbUid) {
×
737
    pStmt->bInfo.needParse = false;
×
738

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

741
    return TSDB_CODE_SUCCESS;
×
742
  }
743

744
  if (pStmt->bInfo.inExecCache) {
×
745
    SStmtTableCache* pCache = taosHashGet(pStmt->sql.pTableCache, &cacheUid, sizeof(cacheUid));
×
746
    if (NULL == pCache) {
×
747
      STMT2_ELOG("table [%s, %" PRIx64 ", %" PRIx64 "] found in exec blockHash, but not in sql blockHash",
×
748
                 pStmt->bInfo.tbFName, uid, cacheUid);
749

750
      STMT_ERR_RET(TSDB_CODE_APP_ERROR);
×
751
    }
752

753
    pStmt->bInfo.needParse = false;
×
754

755
    pStmt->bInfo.tbUid = uid;
×
756
    pStmt->bInfo.tbSuid = suid;
×
757
    pStmt->bInfo.tbType = tableType;
×
758
    pStmt->bInfo.boundTags = pCache->boundTags;
×
759
    pStmt->bInfo.tagsCached = true;
×
760

761
    STMT2_DLOG("tb %s in execBlock list, set to current", pStmt->bInfo.tbFName);
×
762

763
    return TSDB_CODE_SUCCESS;
×
764
  }
765

766
  SStmtTableCache* pCache = taosHashGet(pStmt->sql.pTableCache, &cacheUid, sizeof(cacheUid));
×
767
  if (pCache) {
×
768
    pStmt->bInfo.needParse = false;
×
769

770
    pStmt->bInfo.tbUid = uid;
×
771
    pStmt->bInfo.tbSuid = suid;
×
772
    pStmt->bInfo.tbType = tableType;
×
773
    pStmt->bInfo.boundTags = pCache->boundTags;
×
774
    pStmt->bInfo.tagsCached = true;
×
775

776
    STableDataCxt* pNewBlock = NULL;
×
777
    STMT_ERR_RET(stmtRebuildDataBlock(pStmt, pCache->pDataCtx, &pNewBlock, uid, suid, vgId));
×
778

779
    if (taosHashPut(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName), &pNewBlock,
×
780
                    POINTER_BYTES)) {
781
      STMT_ERR_RET(terrno);
×
782
    }
783

784
    pStmt->exec.pCurrBlock = pNewBlock;
×
785

786
    tscDebug("tb %s in sqlBlock list, set to current", pStmt->bInfo.tbFName);
×
787

788
    return TSDB_CODE_SUCCESS;
×
789
  }
790

791
  STMT_ERR_RET(stmtCleanBindInfo(pStmt));
×
792

793
  return TSDB_CODE_SUCCESS;
×
794
}
795

796
static int32_t stmtResetStmt(STscStmt2* pStmt) {
×
797
  STMT_ERR_RET(stmtCleanSQLInfo(pStmt));
×
798

799
  pStmt->sql.pTableCache = taosHashInit(100, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
×
800
  if (NULL == pStmt->sql.pTableCache) {
×
801
    STMT2_ELOG("fail to allocate memory for pTableCache in stmtResetStmt:%s", tstrerror(terrno));
×
802
    STMT_ERR_RET(terrno);
×
803
  }
804

805
  pStmt->sql.status = STMT_INIT;
×
806

807
  return TSDB_CODE_SUCCESS;
×
808
}
809

810
static void stmtAsyncOutput(STscStmt2* pStmt, void* param) {
23,609,162✔
811
  SStmtQNode* pParam = (SStmtQNode*)param;
23,609,162✔
812

813
  if (pParam->restoreTbCols) {
23,609,162✔
814
    for (int32_t i = 0; i < pStmt->sql.siInfo.pTableColsIdx; ++i) {
23,609,907✔
815
      SArray** p = (SArray**)TARRAY_GET_ELEM(pStmt->sql.siInfo.pTableCols, i);
16,218,588✔
816
      *p = taosArrayInit(20, POINTER_BYTES);
16,218,588✔
817
      if (*p == NULL) {
16,218,497✔
818
        pStmt->errCode = terrno;
722✔
819
      }
820
    }
821
    atomic_store_8((int8_t*)&pStmt->sql.siInfo.tableColsReady, true);
7,391,588✔
822
    STMT2_TLOG_E("restore pTableCols finished");
7,392,183✔
823
  } else {
824
    int code = qAppendStmt2TableOutput(pStmt->sql.pQuery, pStmt->sql.pVgHash, &pParam->tblData, pStmt->exec.pCurrBlock,
16,217,299✔
825
                                       &pStmt->sql.siInfo, pParam->pCreateTbReq);
826
    // taosMemoryFree(pParam->pTbData);
827
    if (code != TSDB_CODE_SUCCESS) {
16,218,471✔
828
      STMT2_ELOG("async append stmt output failed, tbname:%s, err:%s", pParam->tblData.tbName, tstrerror(code));
×
829
      pStmt->errCode = code;
×
830
    }
831
    (void)atomic_sub_fetch_64(&pStmt->sql.siInfo.tbRemainNum, 1);
16,218,471✔
832
  }
833
}
23,613,680✔
834

835
static void* stmtBindThreadFunc(void* param) {
20,673✔
836
  setThreadName("stmt2Bind");
20,673✔
837

838
  STscStmt2* pStmt = (STscStmt2*)param;
20,673✔
839
  STMT2_ILOG_E("stmt2 bind thread started");
20,673✔
840

841
  while (true) {
23,612,808✔
842
    SStmtQNode* asyncParam = NULL;
23,633,481✔
843

844
    if (!stmtDequeue(pStmt, &asyncParam)) {
23,633,477✔
845
      if (pStmt->queue.stopQueue && 0 == atomic_load_64((int64_t*)&pStmt->queue.qRemainNum)) {
20,673✔
846
        STMT2_DLOG_E("queue is empty and stopQueue is set, thread will exit");
20,673✔
847
        break;
20,673✔
848
      }
849
      continue;
×
850
    }
851

852
    stmtAsyncOutput(pStmt, asyncParam);
23,609,654✔
853
  }
854

855
  STMT2_ILOG_E("stmt2 bind thread stopped");
20,673✔
856
  return NULL;
20,673✔
857
}
858

859
static int32_t stmtStartBindThread(STscStmt2* pStmt) {
20,673✔
860
  TdThreadAttr thAttr;
15,927✔
861
  if (taosThreadAttrInit(&thAttr) != 0) {
20,673✔
862
    return TSDB_CODE_TSC_INTERNAL_ERROR;
×
863
  }
864
  if (taosThreadAttrSetDetachState(&thAttr, PTHREAD_CREATE_JOINABLE) != 0) {
20,673✔
865
    return TSDB_CODE_TSC_INTERNAL_ERROR;
×
866
  }
867

868
  if (taosThreadCreate(&pStmt->bindThread, &thAttr, stmtBindThreadFunc, pStmt) != 0) {
20,673✔
869
    terrno = TAOS_SYSTEM_ERROR(ERRNO);
×
870
    STMT_ERR_RET(terrno);
×
871
  }
872

873
  pStmt->bindThreadInUse = true;
20,673✔
874

875
  (void)taosThreadAttrDestroy(&thAttr);
20,673✔
876
  return TSDB_CODE_SUCCESS;
20,673✔
877
}
878

879
static int32_t stmtInitQueue(STscStmt2* pStmt) {
20,673✔
880
  (void)taosThreadCondInit(&pStmt->queue.waitCond, NULL);
20,673✔
881
  (void)taosThreadMutexInit(&pStmt->queue.mutex, NULL);
20,673✔
882
  STMT_ERR_RET(stmtAllocQNodeFromBuf(&pStmt->sql.siInfo.tbBuf, (void**)&pStmt->queue.head));
41,346✔
883
  pStmt->queue.tail = pStmt->queue.head;
20,673✔
884

885
  return TSDB_CODE_SUCCESS;
20,673✔
886
}
887

888
static int32_t stmtIniAsyncBind(STscStmt2* pStmt) {
19,609✔
889
  (void)taosThreadCondInit(&pStmt->asyncBindParam.waitCond, NULL);
19,609✔
890
  (void)taosThreadMutexInit(&pStmt->asyncBindParam.mutex, NULL);
19,609✔
891
  pStmt->asyncBindParam.asyncBindNum = 0;
19,609✔
892

893
  return TSDB_CODE_SUCCESS;
19,609✔
894
}
895

896
static int32_t stmtInitTableBuf(STableBufInfo* pTblBuf) {
20,673✔
897
  pTblBuf->buffUnit = sizeof(SStmtQNode);
20,673✔
898
  pTblBuf->buffSize = pTblBuf->buffUnit * 1000;
20,673✔
899
  pTblBuf->pBufList = taosArrayInit(100, POINTER_BYTES);
20,673✔
900
  if (NULL == pTblBuf->pBufList) {
20,673✔
901
    return terrno;
×
902
  }
903
  void* buff = taosMemoryMalloc(pTblBuf->buffSize);
20,673✔
904
  if (NULL == buff) {
20,673✔
905
    return terrno;
×
906
  }
907

908
  if (taosArrayPush(pTblBuf->pBufList, &buff) == NULL) {
41,346✔
909
    return terrno;
×
910
  }
911

912
  pTblBuf->pCurBuff = buff;
20,673✔
913
  pTblBuf->buffIdx = 1;
20,673✔
914
  pTblBuf->buffOffset = 0;
20,673✔
915

916
  return TSDB_CODE_SUCCESS;
20,673✔
917
}
918

919
TAOS_STMT2* stmtInit2(STscObj* taos, TAOS_STMT2_OPTION* pOptions) {
19,609✔
920
  STscObj*   pObj = (STscObj*)taos;
19,609✔
921
  STscStmt2* pStmt = NULL;
19,609✔
922
  int32_t    code = 0;
19,609✔
923

924
  pStmt = taosMemoryCalloc(1, sizeof(STscStmt2));
19,609✔
925
  if (NULL == pStmt) {
19,609✔
926
    STMT2_ELOG_E("fail to allocate memory for pStmt");
×
927
    return NULL;
×
928
  }
929

930
  pStmt->sql.pTableCache = taosHashInit(100, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
19,609✔
931
  if (NULL == pStmt->sql.pTableCache) {
19,609✔
932
    STMT2_ELOG("fail to allocate memory for pTableCache in stmtInit2:%s", tstrerror(terrno));
×
933
    taosMemoryFree(pStmt);
×
934
    return NULL;
×
935
  }
936

937
  pStmt->taos = pObj;
19,609✔
938
  if (taos->db[0] != '\0') {
19,609✔
939
    pStmt->db = taosStrdup(taos->db);
19,515✔
940
  }
941
  pStmt->bInfo.needParse = true;
19,703✔
942
  pStmt->sql.status = STMT_INIT;
19,703✔
943
  pStmt->errCode = TSDB_CODE_SUCCESS;
19,703✔
944

945
  if (NULL != pOptions) {
19,703✔
946
    (void)memcpy(&pStmt->options, pOptions, sizeof(pStmt->options));
18,905✔
947
    if (pOptions->singleStbInsert && pOptions->singleTableBindOnce) {
18,905✔
948
      pStmt->stbInterlaceMode = true;
18,657✔
949
    }
950

951
    pStmt->reqid = pOptions->reqid;
18,905✔
952
  }
953

954
  if (pStmt->stbInterlaceMode) {
19,703✔
955
    pStmt->sql.siInfo.transport = taos->pAppInfo->pTransporter;
18,657✔
956
    pStmt->sql.siInfo.acctId = taos->acctId;
18,657✔
957
    pStmt->sql.siInfo.dbname = taos->db;
18,657✔
958
    pStmt->sql.siInfo.mgmtEpSet = getEpSet_s(&pStmt->taos->pAppInfo->mgmtEp);
18,657✔
959

960
    pStmt->sql.siInfo.pTableHash = tSimpleHashInit(100, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY));
18,657✔
961
    if (NULL == pStmt->sql.siInfo.pTableHash) {
18,657✔
962
      STMT2_ELOG("fail to allocate memory for pTableHash:%s", tstrerror(terrno));
×
963
      (void)stmtClose2(pStmt);
×
964
      return NULL;
×
965
    }
966

967
    pStmt->sql.siInfo.pTableRowDataHash = tSimpleHashInit(100, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY));
18,657✔
968
    if (NULL == pStmt->sql.siInfo.pTableRowDataHash) {
18,657✔
969
      STMT2_ELOG("fail to allocate memory for pTableRowDataHash:%s", tstrerror(terrno));
×
970
      (void)stmtClose2(pStmt);
×
971
      return NULL;
×
972
    }
973

974
    pStmt->sql.siInfo.pTableCols = taosArrayInit(STMT_TABLE_COLS_NUM, POINTER_BYTES);
18,657✔
975
    if (NULL == pStmt->sql.siInfo.pTableCols) {
18,657✔
976
      STMT2_ELOG("fail to allocate memory for pTableCols:%s", tstrerror(terrno));
×
977
      (void)stmtClose2(pStmt);
×
978
      return NULL;
×
979
    }
980

981
    code = stmtInitTableBuf(&pStmt->sql.siInfo.tbBuf);
18,657✔
982
    if (TSDB_CODE_SUCCESS == code) {
18,657✔
983
      code = stmtInitQueue(pStmt);
18,657✔
984
    }
985
    if (TSDB_CODE_SUCCESS == code) {
18,657✔
986
      code = stmtStartBindThread(pStmt);
18,657✔
987
    }
988
    if (TSDB_CODE_SUCCESS != code) {
18,657✔
989
      terrno = code;
×
990
      STMT2_ELOG("fail to init stmt2 bind thread:%s", tstrerror(code));
×
991
      (void)stmtClose2(pStmt);
×
992
      return NULL;
×
993
    }
994
  }
995

996
  pStmt->sql.siInfo.tableColsReady = true;
19,703✔
997
  if (pStmt->options.asyncExecFn) {
19,703✔
998
    if (tsem_init(&pStmt->asyncExecSem, 0, 1) != 0) {
×
999
      terrno = TAOS_SYSTEM_ERROR(ERRNO);
×
1000
      STMT2_ELOG("fail to init asyncExecSem:%s", tstrerror(terrno));
×
1001
      (void)stmtClose2(pStmt);
×
1002
      return NULL;
×
1003
    }
1004
  }
1005
  code = stmtIniAsyncBind(pStmt);
19,703✔
1006
  if (TSDB_CODE_SUCCESS != code) {
19,609✔
1007
    terrno = code;
×
1008
    STMT2_ELOG("fail to start init asyncExecSem:%s", tstrerror(code));
×
1009

1010
    (void)stmtClose2(pStmt);
×
1011
    return NULL;
×
1012
  }
1013

1014
  pStmt->execSemWaited = false;
19,609✔
1015

1016
  // STMT_LOG_SEQ(STMT_INIT);
1017

1018
  STMT2_DLOG("stmt2 initialize finished, seqId:%d, db:%s, interlaceMode:%d, asyncExec:%d", pStmt->seqId, pStmt->db,
19,609✔
1019
             pStmt->stbInterlaceMode, pStmt->options.asyncExecFn != NULL);
1020

1021
  return pStmt;
19,609✔
1022
}
1023

1024
static int stmtSetDbName2(TAOS_STMT2* stmt, const char* dbName) {
×
1025
  STscStmt2* pStmt = (STscStmt2*)stmt;
×
1026
  if (dbName == NULL || dbName[0] == '\0') {
×
1027
    STMT2_ELOG_E("dbname in sql is illegal");
×
1028
    return TSDB_CODE_TSC_SQL_SYNTAX_ERROR;
×
1029
  }
1030

1031
  STMT2_DLOG("dbname is specified in sql:%s", dbName);
×
1032
  if (pStmt->db == NULL || pStmt->db[0] == '\0') {
×
1033
    taosMemoryFreeClear(pStmt->db);
×
1034
    STMT2_DLOG("dbname:%s is by sql, not by taosconnect", dbName);
×
1035
    pStmt->db = taosStrdup(dbName);
×
1036
    (void)strdequote(pStmt->db);
×
1037
  }
1038
  STMT_ERR_RET(stmtCreateRequest(pStmt));
×
1039

1040
  // The SQL statement specifies a database name, overriding the previously specified database
1041
  taosMemoryFreeClear(pStmt->exec.pRequest->pDb);
×
1042
  pStmt->exec.pRequest->pDb = taosStrdup(dbName);
×
1043
  (void)strdequote(pStmt->exec.pRequest->pDb);
×
1044
  if (pStmt->exec.pRequest->pDb == NULL) {
×
1045
    return terrno;
×
1046
  }
1047
  if (pStmt->sql.stbInterlaceMode) {
×
1048
    pStmt->sql.siInfo.dbname = pStmt->exec.pRequest->pDb;
×
1049
  }
1050
  return TSDB_CODE_SUCCESS;
×
1051
}
1052
static int32_t stmtResetStbInterlaceCache(STscStmt2* pStmt) {
2,016✔
1053
  int32_t code = TSDB_CODE_SUCCESS;
2,016✔
1054

1055
  if (pStmt->bindThreadInUse) {
2,016✔
1056
    while (0 == atomic_load_8((int8_t*)&pStmt->sql.siInfo.tableColsReady)) {
2,016✔
1057
      taosUsleep(1);
×
1058
    }
1059
    (void)taosThreadMutexLock(&pStmt->queue.mutex);
2,016✔
1060
    pStmt->queue.stopQueue = true;
2,016✔
1061
    (void)taosThreadCondSignal(&(pStmt->queue.waitCond));
2,016✔
1062
    (void)taosThreadMutexUnlock(&pStmt->queue.mutex);
2,016✔
1063

1064
    (void)taosThreadJoin(pStmt->bindThread, NULL);
2,016✔
1065
    pStmt->bindThreadInUse = false;
2,016✔
1066
    pStmt->queue.head = NULL;
2,016✔
1067
    pStmt->queue.tail = NULL;
2,016✔
1068
    pStmt->queue.qRemainNum = 0;
2,016✔
1069

1070
    (void)taosThreadCondDestroy(&pStmt->queue.waitCond);
2,016✔
1071
    (void)taosThreadMutexDestroy(&pStmt->queue.mutex);
2,016✔
1072
  }
1073

1074
  pStmt->sql.siInfo.pTableHash = tSimpleHashInit(100, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY));
2,016✔
1075
  if (NULL == pStmt->sql.siInfo.pTableHash) {
2,016✔
1076
    return terrno;
×
1077
  }
1078

1079
  pStmt->sql.siInfo.pTableRowDataHash = tSimpleHashInit(100, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY));
2,016✔
1080
  if (NULL == pStmt->sql.siInfo.pTableRowDataHash) {
2,016✔
1081
    return terrno;
×
1082
  }
1083

1084
  pStmt->sql.siInfo.pTableCols = taosArrayInit(STMT_TABLE_COLS_NUM, POINTER_BYTES);
2,016✔
1085
  if (NULL == pStmt->sql.siInfo.pTableCols) {
2,016✔
1086
    return terrno;
×
1087
  }
1088

1089
  code = stmtInitTableBuf(&pStmt->sql.siInfo.tbBuf);
2,016✔
1090

1091
  if (TSDB_CODE_SUCCESS == code) {
2,016✔
1092
    code = stmtInitQueue(pStmt);
2,016✔
1093
    pStmt->queue.stopQueue = false;
2,016✔
1094
  }
1095
  if (TSDB_CODE_SUCCESS == code) {
2,016✔
1096
    code = stmtStartBindThread(pStmt);
2,016✔
1097
  }
1098
  if (TSDB_CODE_SUCCESS != code) {
2,016✔
1099
    return code;
×
1100
  }
1101

1102
  return TSDB_CODE_SUCCESS;
2,016✔
1103
}
1104

1105
static int32_t stmtDeepReset(STscStmt2* pStmt) {
2,016✔
1106
  char*             db = pStmt->db;
2,016✔
1107
  bool              stbInterlaceMode = pStmt->stbInterlaceMode;
2,016✔
1108
  TAOS_STMT2_OPTION options = pStmt->options;
2,016✔
1109
  uint32_t          reqid = pStmt->reqid;
2,016✔
1110

1111
  pStmt->errCode = 0;
2,016✔
1112
  if (pStmt->options.asyncExecFn && !pStmt->execSemWaited) {
2,016✔
1113
    if (tsem_wait(&pStmt->asyncExecSem) != 0) {
×
1114
      STMT2_ELOG_E("bind param wait asyncExecSem failed");
×
1115
    }
1116
    pStmt->execSemWaited = true;
×
1117
  }
1118
  pStmt->sql.autoCreateTbl = false;
2,016✔
1119
  taosMemoryFree(pStmt->sql.pBindInfo);
2,016✔
1120
  pStmt->sql.pBindInfo = NULL;
2,016✔
1121

1122
  taosMemoryFree(pStmt->sql.queryRes.fields);
2,016✔
1123
  pStmt->sql.queryRes.fields = NULL;
2,016✔
1124

1125
  taosMemoryFree(pStmt->sql.queryRes.userFields);
2,016✔
1126
  pStmt->sql.queryRes.userFields = NULL;
2,016✔
1127

1128
  pStmt->sql.type = 0;
2,016✔
1129
  pStmt->sql.runTimes = 0;
2,016✔
1130
  taosMemoryFree(pStmt->sql.sqlStr);
2,016✔
1131
  pStmt->sql.sqlStr = NULL;
2,016✔
1132

1133
  qDestroyQuery(pStmt->sql.pQuery);
2,016✔
1134
  pStmt->sql.pQuery = NULL;
2,016✔
1135

1136
  taosArrayDestroy(pStmt->sql.nodeList);
2,016✔
1137
  pStmt->sql.nodeList = NULL;
2,016✔
1138

1139
  taosHashCleanup(pStmt->sql.pVgHash);
2,016✔
1140
  pStmt->sql.pVgHash = NULL;
2,016✔
1141

1142
  if (pStmt->sql.fixValueTags) {
2,016✔
1143
    tdDestroySVCreateTbReq(pStmt->sql.fixValueTbReq);
2,016✔
1144
    pStmt->sql.fixValueTbReq = NULL;
2,016✔
1145
  }
1146
  pStmt->sql.fixValueTags = false;
2,016✔
1147

1148
  void* pIter = taosHashIterate(pStmt->sql.pTableCache, NULL);
2,016✔
1149
  while (pIter) {
2,016✔
1150
    SStmtTableCache* pCache = (SStmtTableCache*)pIter;
×
1151

1152
    qDestroyStmtDataBlock(pCache->pDataCtx);
×
1153
    qDestroyBoundColInfo(pCache->boundTags);
×
1154
    taosMemoryFreeClear(pCache->boundTags);
×
1155

1156
    pIter = taosHashIterate(pStmt->sql.pTableCache, pIter);
×
1157
  }
1158
  taosHashCleanup(pStmt->sql.pTableCache);
2,016✔
1159

1160
  if (pStmt->sql.stbInterlaceMode) {
2,016✔
1161
    pStmt->bInfo.tagsCached = false;
2,016✔
1162
  }
1163
  STMT_ERR_RET(stmtCleanExecInfo(pStmt, false, true));
2,016✔
1164

1165
  resetRequest(pStmt);
2,016✔
1166

1167
  if (pStmt->sql.siInfo.pTableCols) {
2,016✔
1168
    taosArrayDestroyEx(pStmt->sql.siInfo.pTableCols, stmtFreeTbCols);
2,016✔
1169
    pStmt->sql.siInfo.pTableCols = NULL;
2,016✔
1170
  }
1171

1172
  if (pStmt->sql.siInfo.tbBuf.pBufList) {
2,016✔
1173
    taosArrayDestroyEx(pStmt->sql.siInfo.tbBuf.pBufList, stmtFreeTbBuf);
2,016✔
1174
    pStmt->sql.siInfo.tbBuf.pBufList = NULL;
2,016✔
1175
  }
1176

1177
  if (pStmt->sql.siInfo.pTableHash) {
2,016✔
1178
    tSimpleHashCleanup(pStmt->sql.siInfo.pTableHash);
2,016✔
1179
    pStmt->sql.siInfo.pTableHash = NULL;
2,016✔
1180
  }
1181

1182
  if (pStmt->sql.siInfo.pTableRowDataHash) {
2,016✔
1183
    tSimpleHashCleanup(pStmt->sql.siInfo.pTableRowDataHash);
2,016✔
1184
    pStmt->sql.siInfo.pTableRowDataHash = NULL;
2,016✔
1185
  }
1186

1187
  if (pStmt->sql.siInfo.pVgroupHash) {
2,016✔
1188
    taosHashCleanup(pStmt->sql.siInfo.pVgroupHash);
×
1189
    pStmt->sql.siInfo.pVgroupHash = NULL;
×
1190
  }
1191

1192
  if (pStmt->sql.siInfo.pVgroupList) {
2,016✔
1193
    taosArrayDestroy(pStmt->sql.siInfo.pVgroupList);
×
1194
    pStmt->sql.siInfo.pVgroupList = NULL;
×
1195
  }
1196

1197
  if (pStmt->sql.siInfo.pDataCtx) {
2,016✔
1198
    qDestroyStmtDataBlock(pStmt->sql.siInfo.pDataCtx);
2,016✔
1199
    pStmt->sql.siInfo.pDataCtx = NULL;
2,016✔
1200
  }
1201

1202
  if (pStmt->sql.siInfo.pTSchema) {
2,016✔
1203
    taosMemoryFree(pStmt->sql.siInfo.pTSchema);
2,016✔
1204
    pStmt->sql.siInfo.pTSchema = NULL;
2,016✔
1205
  }
1206

1207
  if (pStmt->sql.siInfo.pRequest) {
2,016✔
1208
    taos_free_result(pStmt->sql.siInfo.pRequest);
2,016✔
1209
    pStmt->sql.siInfo.pRequest = NULL;
2,016✔
1210
  }
1211

1212
  if (stbInterlaceMode) {
2,016✔
1213
    STMT_ERR_RET(stmtResetStbInterlaceCache(pStmt));
2,016✔
1214
  }
1215

1216
  pStmt->db = db;
2,016✔
1217
  pStmt->stbInterlaceMode = stbInterlaceMode;
2,016✔
1218
  pStmt->options = options;
2,016✔
1219
  pStmt->reqid = reqid;
2,016✔
1220

1221
  pStmt->sql.pTableCache = taosHashInit(100, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
2,016✔
1222
  if (NULL == pStmt->sql.pTableCache) {
2,016✔
1223
    STMT2_ELOG("fail to allocate memory for pTableCache in stmtResetStmt:%s", tstrerror(terrno));
×
1224
    return terrno;
×
1225
  }
1226

1227
  pStmt->sql.status = STMT_INIT;
2,016✔
1228

1229
  return TSDB_CODE_SUCCESS;
2,016✔
1230
}
1231

1232
int stmtPrepare2(TAOS_STMT2* stmt, const char* sql, unsigned long length) {
21,531✔
1233
  STscStmt2* pStmt = (STscStmt2*)stmt;
21,531✔
1234
  int32_t    code = 0;
21,531✔
1235

1236
  STMT2_DLOG("start to prepare with sql:%s", sql);
21,531✔
1237

1238
  if (stmt == NULL || sql == NULL) {
21,531✔
1239
    STMT2_ELOG_E("stmt or sql is NULL");
×
1240
    return TSDB_CODE_INVALID_PARA;
×
1241
  }
1242

1243
  if (pStmt->sql.status >= STMT_PREPARE) {
21,625✔
1244
    STMT2_DLOG("stmt status is %d, need to reset stmt2 cache before prepare", pStmt->sql.status);
2,016✔
1245
    STMT_ERR_RET(stmtDeepReset(pStmt));
2,016✔
1246
  }
1247

1248
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
21,531✔
1249
    STMT2_ELOG("errCode is not success before, ErrCode: 0x%x, errorsyt: %s\n. ", pStmt->errCode,
×
1250
               tstrerror(pStmt->errCode));
1251
    return pStmt->errCode;
×
1252
  }
1253

1254
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_PREPARE));
21,531✔
1255

1256
  if (length <= 0) {
21,531✔
1257
    length = strlen(sql);
248✔
1258
  }
1259
  pStmt->sql.sqlStr = taosStrndup(sql, length);
21,531✔
1260
  if (!pStmt->sql.sqlStr) {
21,531✔
1261
    STMT2_ELOG("fail to allocate memory for sqlStr:%s", tstrerror(terrno));
×
1262
    STMT_ERR_RET(terrno);
×
1263
  }
1264
  pStmt->sql.sqlLen = length;
21,531✔
1265
  STMT_ERR_RET(stmtCreateRequest(pStmt));
21,531✔
1266

1267
  if (stmt2IsInsert(pStmt)) {
21,625✔
1268
    pStmt->sql.stbInterlaceMode = pStmt->stbInterlaceMode;
21,289✔
1269
    char* dbName = NULL;
21,289✔
1270
    if (qParseDbName(sql, length, &dbName)) {
21,289✔
1271
      STMT_ERR_RET(stmtSetDbName2(stmt, dbName));
×
1272
      taosMemoryFreeClear(dbName);
×
1273
    } else if (pStmt->db != NULL && pStmt->db[0] != '\0') {
21,289✔
1274
      taosMemoryFreeClear(pStmt->exec.pRequest->pDb);
21,289✔
1275
      pStmt->exec.pRequest->pDb = taosStrdup(pStmt->db);
21,289✔
1276
      if (pStmt->exec.pRequest->pDb == NULL) {
21,383✔
1277
        STMT_ERR_RET(terrno);
×
1278
      }
1279
      (void)strdequote(pStmt->exec.pRequest->pDb);
21,383✔
1280

1281
      if (pStmt->sql.stbInterlaceMode) {
21,289✔
1282
        pStmt->sql.siInfo.dbname = pStmt->exec.pRequest->pDb;
20,579✔
1283
      }
1284
    }
1285

1286
  } else if (stmt2IsSelect(pStmt)) {
336✔
1287
    pStmt->sql.stbInterlaceMode = false;
242✔
1288
    STMT_ERR_RET(stmtParseSql(pStmt));
242✔
1289
  } else {
1290
    return stmtBuildErrorMsgWithCode(pStmt, "stmt only support 'SELECT' or 'INSERT'", TSDB_CODE_PAR_SYNTAX_ERROR);
×
1291
  }
1292
  return TSDB_CODE_SUCCESS;
21,531✔
1293
}
1294

1295
static int32_t stmtInitStbInterlaceTableInfo(STscStmt2* pStmt) {
20,314✔
1296
  STableDataCxt** pSrc = taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
20,314✔
1297
  if (!pSrc) {
20,673✔
1298
    return terrno;
×
1299
  }
1300
  STableDataCxt* pDst = NULL;
20,673✔
1301

1302
  STMT_ERR_RET(qCloneStmtDataBlock(&pDst, *pSrc, true));
20,673✔
1303
  pStmt->sql.siInfo.pDataCtx = pDst;
20,673✔
1304

1305
  SArray* pTblCols = NULL;
20,673✔
1306
  for (int32_t i = 0; i < STMT_TABLE_COLS_NUM; i++) {
20,639,840✔
1307
    pTblCols = taosArrayInit(20, POINTER_BYTES);
20,619,076✔
1308
    if (NULL == pTblCols) {
20,422,345✔
1309
      return terrno;
×
1310
    }
1311

1312
    if (taosArrayPush(pStmt->sql.siInfo.pTableCols, &pTblCols) == NULL) {
41,053,015✔
1313
      return terrno;
×
1314
    }
1315
  }
1316

1317
  pStmt->sql.siInfo.boundTags = pStmt->bInfo.boundTags;
20,764✔
1318

1319
  STMT2_TLOG("init stb interlace table info, tbName:%s, pDataCtx:%p, boundTags:%p", pStmt->bInfo.tbFName,
20,764✔
1320
             pStmt->sql.siInfo.pDataCtx, pStmt->sql.siInfo.boundTags);
1321

1322
  return TSDB_CODE_SUCCESS;
20,673✔
1323
}
1324

1325
bool stmt2IsInsert(TAOS_STMT2* stmt) {
32,458,681✔
1326
  STscStmt2* pStmt = (STscStmt2*)stmt;
32,458,681✔
1327
  if (pStmt->sql.type) {
32,458,681✔
1328
    return (STMT_TYPE_INSERT == pStmt->sql.type || STMT_TYPE_MULTI_INSERT == pStmt->sql.type);
32,420,273✔
1329
  }
1330

1331
  return qIsInsertValuesSql(pStmt->sql.sqlStr, pStmt->sql.sqlLen);
39,219✔
1332
}
1333

1334
bool stmt2IsSelect(TAOS_STMT2* stmt) {
242✔
1335
  STscStmt2* pStmt = (STscStmt2*)stmt;
242✔
1336

1337
  if (pStmt->sql.type) {
242✔
1338
    return STMT_TYPE_QUERY == pStmt->sql.type;
×
1339
  }
1340
  return qIsSelectFromSql(pStmt->sql.sqlStr, pStmt->sql.sqlLen);
242✔
1341
}
1342

1343
int stmtSetTbName2(TAOS_STMT2* stmt, const char* tbName) {
16,217,533✔
1344
  STscStmt2* pStmt = (STscStmt2*)stmt;
16,217,533✔
1345

1346
  int64_t startUs = taosGetTimestampUs();
16,220,022✔
1347

1348
  STMT2_TLOG("start to set tbName:%s", tbName);
16,220,022✔
1349

1350
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
16,222,554✔
1351
    return pStmt->errCode;
×
1352
  }
1353

1354
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_SETTBNAME));
16,222,827✔
1355

1356
  int32_t insert = 0;
16,217,410✔
1357
  if (!stmt2IsInsert(stmt)) {
16,217,410✔
1358
    STMT2_ELOG_E("set tb name not available for no-insert statement");
×
1359
    STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR);
×
1360
  }
1361

1362
  if (!pStmt->sql.stbInterlaceMode || NULL == pStmt->sql.siInfo.pDataCtx) {
16,220,256✔
1363
    STMT_ERR_RET(stmtCreateRequest(pStmt));
30,067✔
1364

1365
    STMT_ERR_RET(qCreateSName2(&pStmt->bInfo.sname, tbName, pStmt->taos->acctId, pStmt->exec.pRequest->pDb,
26,079✔
1366
                               pStmt->exec.pRequest->msgBuf, pStmt->exec.pRequest->msgBufLen));
1367
    STMT_ERR_RET(tNameExtractFullName(&pStmt->bInfo.sname, pStmt->bInfo.tbFName));
26,079✔
1368

1369
    STMT_ERR_RET(stmtGetFromCache(pStmt));
25,985✔
1370

1371
    if (pStmt->bInfo.needParse) {
25,985✔
1372
      tstrncpy(pStmt->bInfo.tbName, tbName, sizeof(pStmt->bInfo.tbName));
21,289✔
1373
      pStmt->bInfo.tbName[sizeof(pStmt->bInfo.tbName) - 1] = 0;
21,289✔
1374

1375
      STMT_ERR_RET(stmtParseSql(pStmt));
21,289✔
1376
      if (!pStmt->sql.autoCreateTbl) {
21,289✔
1377
        uint64_t uid, suid;
15,927✔
1378
        int32_t  vgId;
15,644✔
1379
        int8_t   tableType;
15,644✔
1380

1381
        int32_t code = stmtGetTableMetaAndValidate(pStmt, &uid, &suid, &vgId, &tableType);
16,887✔
1382
        if (code != TSDB_CODE_SUCCESS) {
17,170✔
1383
          return code;
×
1384
        }
1385
      }
1386
    }
1387

1388
  } else {
1389
    tstrncpy(pStmt->bInfo.tbName, tbName, sizeof(pStmt->bInfo.tbName));
16,189,902✔
1390
    pStmt->bInfo.tbName[sizeof(pStmt->bInfo.tbName) - 1] = 0;
16,188,543✔
1391
    pStmt->exec.pRequest->requestId++;
16,188,543✔
1392
    pStmt->bInfo.needParse = false;
16,190,153✔
1393
  }
1394

1395
  if (pStmt->sql.stbInterlaceMode && NULL == pStmt->sql.siInfo.pDataCtx) {
16,215,618✔
1396
    STMT_ERR_RET(stmtInitStbInterlaceTableInfo(pStmt));
20,314✔
1397
  }
1398

1399
  int64_t startUs2 = taosGetTimestampUs();
16,221,152✔
1400
  pStmt->stat.setTbNameUs += startUs2 - startUs;
16,221,152✔
1401

1402
  return TSDB_CODE_SUCCESS;
16,221,959✔
1403
}
1404

1405
int stmtSetTbTags2(TAOS_STMT2* stmt, TAOS_STMT2_BIND* tags, SVCreateTbReq** pCreateTbReq) {
10,022,274✔
1406
  STscStmt2* pStmt = (STscStmt2*)stmt;
10,022,274✔
1407

1408
  STMT2_TLOG_E("start to set tbTags");
10,022,274✔
1409

1410
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
10,023,512✔
1411
    return pStmt->errCode;
×
1412
  }
1413

1414
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_SETTAGS));
10,023,512✔
1415

1416
  if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 &&
10,023,556✔
1417
      STMT_TYPE_MULTI_INSERT != pStmt->sql.type) {
×
1418
    pStmt->bInfo.needParse = false;
×
1419
  }
1420
  STMT_ERR_RET(stmtCreateRequest(pStmt));
10,023,556✔
1421

1422
  if (pStmt->bInfo.needParse) {
10,022,474✔
1423
    STMT_ERR_RET(stmtParseSql(pStmt));
×
1424
  }
1425
  if (pStmt->sql.stbInterlaceMode && NULL == pStmt->sql.siInfo.pDataCtx) {
10,022,474✔
1426
    STMT_ERR_RET(stmtInitStbInterlaceTableInfo(pStmt));
×
1427
  }
1428

1429
  SBoundColInfo* tags_info = (SBoundColInfo*)pStmt->bInfo.boundTags;
10,022,474✔
1430
  // if (tags_info->numOfBound <= 0 || tags_info->numOfCols <= 0) {
1431
  //   tscWarn("no tags or cols bound in sql, will not bound tags");
1432
  //   return TSDB_CODE_SUCCESS;
1433
  // }
1434
  if (pStmt->sql.autoCreateTbl && pStmt->sql.stbInterlaceMode) {
10,022,474✔
1435
    STMT_ERR_RET(qCreateSName2(&pStmt->bInfo.sname, pStmt->bInfo.tbName, pStmt->taos->acctId, pStmt->exec.pRequest->pDb,
10,015,466✔
1436
                               pStmt->exec.pRequest->msgBuf, pStmt->exec.pRequest->msgBufLen));
1437
    STMT_ERR_RET(tNameExtractFullName(&pStmt->bInfo.sname, pStmt->bInfo.tbFName));
10,017,044✔
1438
  }
1439

1440
  STableDataCxt** pDataBlock = NULL;
10,024,052✔
1441
  if (pStmt->exec.pCurrBlock) {
10,024,052✔
1442
    pDataBlock = &pStmt->exec.pCurrBlock;
10,022,359✔
1443
  } else {
1444
    pDataBlock =
1445
        (STableDataCxt**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
1,693✔
1446
    if (NULL == pDataBlock) {
1,693✔
1447
      STMT2_ELOG("table %s not found in exec blockHash:%p", pStmt->bInfo.tbFName, pStmt->exec.pBlockHash);
×
1448
      STMT_ERR_RET(TSDB_CODE_TSC_STMT_CACHE_ERROR);
×
1449
    }
1450
  }
1451
  if (pStmt->bInfo.inExecCache && !pStmt->sql.autoCreateTbl) {
10,024,052✔
1452
    return TSDB_CODE_SUCCESS;
×
1453
  }
1454

1455
  STMT2_TLOG_E("start to bind stmt tag values");
10,024,052✔
1456

1457
  void* boundTags = NULL;
10,020,784✔
1458
  if (pStmt->sql.stbInterlaceMode) {
10,020,784✔
1459
    boundTags = pStmt->sql.siInfo.boundTags;
10,015,378✔
1460
    *pCreateTbReq = taosMemoryCalloc(1, sizeof(SVCreateTbReq));
10,015,378✔
1461
    if (NULL == pCreateTbReq) {
10,016,622✔
1462
      return terrno;
×
1463
    }
1464
    int32_t vgId = -1;
10,016,622✔
1465
    STMT_ERR_RET(stmtTryAddTableVgroupInfo(pStmt, &vgId));
10,016,622✔
1466
    (*pCreateTbReq)->uid = vgId;
10,019,840✔
1467
  } else {
1468
    boundTags = pStmt->bInfo.boundTags;
5,406✔
1469
  }
1470

1471
  STMT_ERR_RET(qBindStmtTagsValue2(*pDataBlock, boundTags, pStmt->bInfo.tbSuid, pStmt->bInfo.stbFName,
10,025,246✔
1472
                                   pStmt->bInfo.sname.tname, tags, pStmt->exec.pRequest->msgBuf,
1473
                                   pStmt->exec.pRequest->msgBufLen, pStmt->taos->optionInfo.charsetCxt, *pCreateTbReq));
1474

1475
  return TSDB_CODE_SUCCESS;
10,025,360✔
1476
}
1477

1478
int stmtCheckTags2(TAOS_STMT2* stmt, SVCreateTbReq** pCreateTbReq) {
2,014,014✔
1479
  STscStmt2* pStmt = (STscStmt2*)stmt;
2,014,014✔
1480

1481
  STMT2_TLOG_E("start to clone createTbRequest for fixed tags");
2,014,014✔
1482

1483
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
2,014,706✔
1484
    return pStmt->errCode;
×
1485
  }
1486

1487
  if (!pStmt->sql.stbInterlaceMode) {
2,014,706✔
1488
    return TSDB_CODE_SUCCESS;
×
1489
  }
1490

1491
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_SETTAGS));
2,014,706✔
1492

1493
  if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 &&
2,015,274✔
1494
      STMT_TYPE_MULTI_INSERT != pStmt->sql.type) {
×
1495
    pStmt->bInfo.needParse = false;
×
1496
  }
1497
  STMT_ERR_RET(stmtCreateRequest(pStmt));
2,015,274✔
1498

1499
  if (pStmt->bInfo.needParse) {
2,014,014✔
1500
    STMT_ERR_RET(stmtParseSql(pStmt));
×
1501
    if (!pStmt->sql.autoCreateTbl) {
×
1502
      STMT2_WLOG_E("don't need to create table, will not check tags");
×
1503
      return TSDB_CODE_SUCCESS;
×
1504
    }
1505
  }
1506

1507
  if (pStmt->sql.stbInterlaceMode && NULL == pStmt->sql.siInfo.pDataCtx) {
2,014,014✔
1508
    STMT_ERR_RET(stmtInitStbInterlaceTableInfo(pStmt));
×
1509
  }
1510

1511
  STMT_ERR_RET(qCreateSName(&pStmt->bInfo.sname, pStmt->bInfo.tbName, pStmt->taos->acctId, pStmt->exec.pRequest->pDb,
2,014,014✔
1512
                            pStmt->exec.pRequest->msgBuf, pStmt->exec.pRequest->msgBufLen));
1513
  STMT_ERR_RET(tNameExtractFullName(&pStmt->bInfo.sname, pStmt->bInfo.tbFName));
2,014,172✔
1514

1515
  STableDataCxt** pDataBlock = NULL;
2,013,950✔
1516
  if (pStmt->exec.pCurrBlock) {
2,013,950✔
1517
    pDataBlock = &pStmt->exec.pCurrBlock;
2,011,430✔
1518
  } else {
1519
    pDataBlock =
1520
        (STableDataCxt**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
2,520✔
1521
    if (NULL == pDataBlock) {
2,520✔
1522
      STMT2_ELOG("table %s not found in exec blockHash:%p", pStmt->bInfo.tbFName, pStmt->exec.pBlockHash);
×
1523
      STMT_ERR_RET(TSDB_CODE_TSC_STMT_CACHE_ERROR);
1,324✔
1524
    }
1525
  }
1526

1527
  if (!((*pDataBlock)->pData->flags & SUBMIT_REQ_AUTO_CREATE_TABLE)) {
2,015,274✔
1528
    STMT2_DLOG_E("don't need to create, will not check tags");
×
1529
    return TSDB_CODE_SUCCESS;
×
1530
  }
1531

1532
  if (pStmt->sql.fixValueTags) {
2,015,274✔
1533
    STMT2_TLOG_E("tags are fixed, use one createTbReq");
2,013,070✔
1534
    STMT_ERR_RET(cloneSVreateTbReq(pStmt->sql.fixValueTbReq, pCreateTbReq));
2,013,070✔
1535
    if ((*pCreateTbReq)->name) {
2,012,344✔
1536
      taosMemoryFree((*pCreateTbReq)->name);
2,012,344✔
1537
    }
1538
    (*pCreateTbReq)->name = taosStrdup(pStmt->bInfo.tbName);
2,012,378✔
1539
    int32_t vgId = -1;
2,013,070✔
1540
    STMT_ERR_RET(stmtTryAddTableVgroupInfo(pStmt, &vgId));
2,013,070✔
1541
    (*pCreateTbReq)->uid = vgId;
2,013,040✔
1542
    return TSDB_CODE_SUCCESS;
2,013,040✔
1543
  }
1544

1545
  if ((*pDataBlock)->pData->pCreateTbReq) {
2,204✔
1546
    STMT2_TLOG_E("tags are fixed, set createTbReq first time");
2,520✔
1547
    pStmt->sql.fixValueTags = true;
2,520✔
1548
    STMT_ERR_RET(cloneSVreateTbReq((*pDataBlock)->pData->pCreateTbReq, &pStmt->sql.fixValueTbReq));
2,520✔
1549
    STMT_ERR_RET(cloneSVreateTbReq(pStmt->sql.fixValueTbReq, pCreateTbReq));
2,520✔
1550
    (*pCreateTbReq)->uid = (*pDataBlock)->pMeta->vgId;
2,520✔
1551
  }
1552

1553
  return TSDB_CODE_SUCCESS;
2,204✔
1554
}
1555

1556
static int stmtFetchColFields2(STscStmt2* pStmt, int32_t* fieldNum, TAOS_FIELD_E** fields) {
×
1557
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
×
1558
    return pStmt->errCode;
×
1559
  }
1560

1561
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
×
1562
    tscError("invalid operation to get query column fileds");
×
1563
    STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR);
×
1564
  }
1565

1566
  STableDataCxt** pDataBlock = NULL;
×
1567

1568
  if (pStmt->sql.stbInterlaceMode) {
×
1569
    pDataBlock = &pStmt->sql.siInfo.pDataCtx;
×
1570
  } else {
1571
    pDataBlock =
1572
        (STableDataCxt**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
×
1573
    if (NULL == pDataBlock) {
×
1574
      STMT2_ELOG("table %s not found in exec blockHash:%p", pStmt->bInfo.tbFName, pStmt->exec.pBlockHash);
×
1575
      STMT_ERR_RET(TSDB_CODE_APP_ERROR);
×
1576
    }
1577
  }
1578

1579
  STMT_ERR_RET(qBuildStmtColFields(*pDataBlock, fieldNum, fields));
×
1580

1581
  return TSDB_CODE_SUCCESS;
×
1582
}
1583

1584
static int stmtFetchStbColFields2(STscStmt2* pStmt, int32_t* fieldNum, TAOS_FIELD_ALL** fields) {
703✔
1585
  int32_t code = 0;
703✔
1586
  int32_t preCode = pStmt->errCode;
703✔
1587

1588
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
703✔
1589
    return pStmt->errCode;
×
1590
  }
1591

1592
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
703✔
1593
    STMT2_ELOG_E("stmtFetchStbColFields2 only for insert statement");
×
1594
    STMT_ERRI_JRET(TSDB_CODE_TSC_STMT_API_ERROR);
×
1595
  }
1596

1597
  STableDataCxt** pDataBlock = NULL;
703✔
1598
  bool            cleanStb = false;
703✔
1599

1600
  if (pStmt->sql.stbInterlaceMode && pStmt->sql.siInfo.pDataCtx != NULL) {
703✔
1601
    pDataBlock = &pStmt->sql.siInfo.pDataCtx;
×
1602
  } else {
1603
    cleanStb = true;
703✔
1604
    pDataBlock =
1605
        (STableDataCxt**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
703✔
1606
  }
1607

1608
  if (NULL == pDataBlock || NULL == *pDataBlock) {
703✔
1609
    STMT2_ELOG("table %s not found in exec blockHash:%p", pStmt->bInfo.tbFName, pStmt->exec.pBlockHash);
×
1610
    STMT_ERRI_JRET(TSDB_CODE_APP_ERROR);
×
1611
  }
1612

1613
  STMT_ERRI_JRET(qBuildStmtStbColFields(*pDataBlock, pStmt->bInfo.boundTags, pStmt->bInfo.boundCols,
703✔
1614
                                        pStmt->bInfo.tbNameFlag, fieldNum, fields));
1615

1616
  if (pStmt->bInfo.tbType == TSDB_SUPER_TABLE && cleanStb) {
703✔
1617
    taosMemoryFreeClear((*pDataBlock)->boundColsInfo.pColIndex);
703✔
1618
    qDestroyStmtDataBlock(*pDataBlock);
703✔
1619
    *pDataBlock = NULL;
703✔
1620
    if (taosHashRemove(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName)) != 0) {
703✔
1621
      STMT2_ELOG("fail to remove remove stb:%s exec blockHash", pStmt->bInfo.tbFName);
×
1622
      STMT_ERRI_JRET(TSDB_CODE_APP_ERROR);
×
1623
    }
1624
    pStmt->sql.autoCreateTbl = false;
703✔
1625
    pStmt->bInfo.tagsCached = false;
703✔
1626
    pStmt->bInfo.sname = (SName){0};
703✔
1627
    STMT_ERR_RET(stmtCleanBindInfo(pStmt));
703✔
1628
  }
1629

1630
_return:
×
1631

1632
  pStmt->errCode = preCode;
703✔
1633

1634
  return code;
703✔
1635
}
1636
/*
1637
SArray* stmtGetFreeCol(STscStmt2* pStmt, int32_t* idx) {
1638
  while (true) {
1639
    if (pStmt->exec.smInfo.pColIdx >= STMT_COL_BUF_SIZE) {
1640
      pStmt->exec.smInfo.pColIdx = 0;
1641
    }
1642

1643
    if ((pStmt->exec.smInfo.pColIdx + 1) == atomic_load_32(&pStmt->exec.smInfo.pColFreeIdx)) {
1644
      taosUsleep(1);
1645
      continue;
1646
    }
1647

1648
    *idx = pStmt->exec.smInfo.pColIdx;
1649
    return pStmt->exec.smInfo.pCols[pStmt->exec.smInfo.pColIdx++];
1650
  }
1651
}
1652
*/
1653
static int32_t stmtAppendTablePostHandle(STscStmt2* pStmt, SStmtQNode* param) {
16,214,747✔
1654
  if (NULL == pStmt->sql.siInfo.pVgroupHash) {
16,214,747✔
1655
    pStmt->sql.siInfo.pVgroupHash =
7,390,982✔
1656
        taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_ENTRY_LOCK);
7,389,566✔
1657
  }
1658
  if (NULL == pStmt->sql.siInfo.pVgroupList) {
16,217,508✔
1659
    pStmt->sql.siInfo.pVgroupList = taosArrayInit(64, POINTER_BYTES);
7,391,251✔
1660
  }
1661

1662
  if (NULL == pStmt->sql.siInfo.pRequest) {
16,216,589✔
1663
    STMT_ERR_RET(buildRequest(pStmt->taos->id, pStmt->sql.sqlStr, pStmt->sql.sqlLen, NULL, false,
20,673✔
1664
                              (SRequestObj**)&pStmt->sql.siInfo.pRequest, pStmt->reqid));
1665

1666
    if (pStmt->reqid != 0) {
20,579✔
1667
      pStmt->reqid++;
241✔
1668
    }
1669
    pStmt->exec.pRequest->syncQuery = true;
20,579✔
1670

1671
    pStmt->sql.siInfo.requestId = ((SRequestObj*)pStmt->sql.siInfo.pRequest)->requestId;
20,579✔
1672
    pStmt->sql.siInfo.requestSelf = ((SRequestObj*)pStmt->sql.siInfo.pRequest)->self;
20,579✔
1673
  }
1674

1675
  if (!pStmt->sql.siInfo.tbFromHash && pStmt->sql.siInfo.firstName[0] &&
16,217,033✔
1676
      0 == strcmp(pStmt->sql.siInfo.firstName, pStmt->bInfo.tbName)) {
126,691✔
1677
    pStmt->sql.siInfo.tbFromHash = true;
11,675✔
1678
  }
1679

1680
  if (0 == pStmt->sql.siInfo.firstName[0]) {
16,216,495✔
1681
    tstrncpy(pStmt->sql.siInfo.firstName, pStmt->bInfo.tbName, TSDB_TABLE_NAME_LEN);
18,563✔
1682
  }
1683

1684
  param->tblData.getFromHash = pStmt->sql.siInfo.tbFromHash;
16,216,764✔
1685
  param->next = NULL;
16,216,226✔
1686

1687
  (void)atomic_add_fetch_64(&pStmt->sql.siInfo.tbRemainNum, 1);
16,215,419✔
1688

1689
  if (pStmt->queue.stopQueue) {
16,220,189✔
1690
    STMT2_ELOG_E("bind thread already stopped, cannot enqueue");
×
1691
    return TSDB_CODE_TSC_STMT_API_ERROR;
×
1692
  }
1693
  stmtEnqueue(pStmt, param);
16,220,189✔
1694

1695
  return TSDB_CODE_SUCCESS;
16,219,838✔
1696
}
1697

1698
static FORCE_INLINE int32_t stmtGetTableColsFromCache(STscStmt2* pStmt, SArray** pTableCols) {
1699
  while (true) {
1700
    if (pStmt->sql.siInfo.pTableColsIdx < taosArrayGetSize(pStmt->sql.siInfo.pTableCols)) {
16,217,352✔
1701
      *pTableCols = (SArray*)taosArrayGetP(pStmt->sql.siInfo.pTableCols, pStmt->sql.siInfo.pTableColsIdx++);
16,217,210✔
1702
      break;
16,218,794✔
1703
    } else {
1704
      SArray* pTblCols = NULL;
×
1705
      for (int32_t i = 0; i < 100; i++) {
×
1706
        pTblCols = taosArrayInit(20, POINTER_BYTES);
×
1707
        if (NULL == pTblCols) {
×
1708
          return terrno;
×
1709
        }
1710

1711
        if (taosArrayPush(pStmt->sql.siInfo.pTableCols, &pTblCols) == NULL) {
×
1712
          return terrno;
×
1713
        }
1714
      }
1715
    }
1716
  }
1717

1718
  return TSDB_CODE_SUCCESS;
16,218,794✔
1719
}
1720

1721
static int32_t stmtCacheBlock(STscStmt2* pStmt) {
5,406✔
1722
  if (pStmt->sql.type != STMT_TYPE_MULTI_INSERT) {
5,406✔
1723
    return TSDB_CODE_SUCCESS;
×
1724
  }
1725

1726
  uint64_t uid = pStmt->bInfo.tbUid;
5,406✔
1727
  uint64_t cacheUid = (TSDB_CHILD_TABLE == pStmt->bInfo.tbType) ? pStmt->bInfo.tbSuid : uid;
5,406✔
1728

1729
  if (taosHashGet(pStmt->sql.pTableCache, &cacheUid, sizeof(cacheUid))) {
5,406✔
1730
    STMT2_TLOG("table %s already cached, no need to cache again", pStmt->bInfo.tbFName);
4,696✔
1731
    return TSDB_CODE_SUCCESS;
4,696✔
1732
  }
1733

1734
  STableDataCxt** pSrc = taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
710✔
1735
  if (!pSrc) {
710✔
1736
    STMT2_ELOG("table %s not found in exec blockHash:%p", pStmt->bInfo.tbFName, pStmt->exec.pBlockHash);
×
1737
    return terrno;
×
1738
  }
1739
  STableDataCxt* pDst = NULL;
710✔
1740

1741
  STMT_ERR_RET(qCloneStmtDataBlock(&pDst, *pSrc, true));
710✔
1742

1743
  SStmtTableCache cache = {
710✔
1744
      .pDataCtx = pDst,
1745
      .boundTags = pStmt->bInfo.boundTags,
710✔
1746
  };
1747

1748
  if (taosHashPut(pStmt->sql.pTableCache, &cacheUid, sizeof(cacheUid), &cache, sizeof(cache))) {
710✔
1749
    STMT2_ELOG("fail to put table cache:%s", tstrerror(terrno));
×
1750
    return terrno;
×
1751
  }
1752

1753
  if (pStmt->sql.autoCreateTbl) {
710✔
1754
    pStmt->bInfo.tagsCached = true;
710✔
1755
  } else {
1756
    pStmt->bInfo.boundTags = NULL;
×
1757
  }
1758

1759
  return TSDB_CODE_SUCCESS;
710✔
1760
}
1761

1762
static int stmtAddBatch2(TAOS_STMT2* stmt) {
7,396,001✔
1763
  STscStmt2* pStmt = (STscStmt2*)stmt;
7,396,001✔
1764

1765
  int64_t startUs = taosGetTimestampUs();
7,396,859✔
1766

1767
  // STMT2_TLOG_E("start to add batch");
1768

1769
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
7,396,859✔
1770
    return pStmt->errCode;
×
1771
  }
1772

1773
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_ADD_BATCH));
7,396,859✔
1774

1775
  if (pStmt->sql.stbInterlaceMode) {
7,394,804✔
1776
    int64_t startUs2 = taosGetTimestampUs();
7,391,485✔
1777
    pStmt->stat.addBatchUs += startUs2 - startUs;
7,391,485✔
1778

1779
    pStmt->sql.siInfo.tableColsReady = false;
7,391,754✔
1780

1781
    SStmtQNode* param = NULL;
7,391,485✔
1782
    STMT_ERR_RET(stmtAllocQNodeFromBuf(&pStmt->sql.siInfo.tbBuf, (void**)&param));
14,783,056✔
1783
    param->restoreTbCols = true;
7,391,571✔
1784
    param->next = NULL;
7,391,571✔
1785

1786
    if (pStmt->sql.autoCreateTbl) {
7,391,571✔
1787
      pStmt->bInfo.tagsCached = true;
4,020,414✔
1788
    }
1789
    pStmt->bInfo.boundColsCached = true;
7,391,571✔
1790

1791
    if (pStmt->queue.stopQueue) {
7,391,840✔
1792
      STMT2_ELOG_E("stmt bind thread is stopped,cannot enqueue bind request");
×
1793
      return TSDB_CODE_TSC_STMT_API_ERROR;
×
1794
    }
1795

1796
    stmtEnqueue(pStmt, param);
7,391,840✔
1797

1798
    return TSDB_CODE_SUCCESS;
7,391,909✔
1799
  }
1800

1801
  STMT_ERR_RET(stmtCacheBlock(pStmt));
5,406✔
1802

1803
  return TSDB_CODE_SUCCESS;
5,406✔
1804
}
1805
/*
1806
static int32_t stmtBackupQueryFields(STscStmt2* pStmt) {
1807
  SStmtQueryResInfo* pRes = &pStmt->sql.queryRes;
1808
  pRes->numOfCols = pStmt->exec.pRequest->body.resInfo.numOfCols;
1809
  pRes->precision = pStmt->exec.pRequest->body.resInfo.precision;
1810

1811
  int32_t size = pRes->numOfCols * sizeof(TAOS_FIELD);
1812
  pRes->fields = taosMemoryMalloc(size);
1813
  pRes->userFields = taosMemoryMalloc(size);
1814
  if (NULL == pRes->fields || NULL == pRes->userFields) {
1815
    STMT_ERR_RET(terrno);
1816
  }
1817
  (void)memcpy(pRes->fields, pStmt->exec.pRequest->body.resInfo.fields, size);
1818
  (void)memcpy(pRes->userFields, pStmt->exec.pRequest->body.resInfo.userFields, size);
1819

1820
  return TSDB_CODE_SUCCESS;
1821
}
1822

1823
static int32_t stmtRestoreQueryFields(STscStmt2* pStmt) {
1824
  SStmtQueryResInfo* pRes = &pStmt->sql.queryRes;
1825
  int32_t            size = pRes->numOfCols * sizeof(TAOS_FIELD);
1826

1827
  pStmt->exec.pRequest->body.resInfo.numOfCols = pRes->numOfCols;
1828
  pStmt->exec.pRequest->body.resInfo.precision = pRes->precision;
1829

1830
  if (NULL == pStmt->exec.pRequest->body.resInfo.fields) {
1831
    pStmt->exec.pRequest->body.resInfo.fields = taosMemoryMalloc(size);
1832
    if (NULL == pStmt->exec.pRequest->body.resInfo.fields) {
1833
      STMT_ERR_RET(terrno);
1834
    }
1835
    (void)memcpy(pStmt->exec.pRequest->body.resInfo.fields, pRes->fields, size);
1836
  }
1837

1838
  if (NULL == pStmt->exec.pRequest->body.resInfo.userFields) {
1839
    pStmt->exec.pRequest->body.resInfo.userFields = taosMemoryMalloc(size);
1840
    if (NULL == pStmt->exec.pRequest->body.resInfo.userFields) {
1841
      STMT_ERR_RET(terrno);
1842
    }
1843
    (void)memcpy(pStmt->exec.pRequest->body.resInfo.userFields, pRes->userFields, size);
1844
  }
1845

1846
  return TSDB_CODE_SUCCESS;
1847
}
1848
*/
1849

1850
int stmtBindBatch2(TAOS_STMT2* stmt, TAOS_STMT2_BIND* bind, int32_t colIdx, SVCreateTbReq* pCreateTbReq) {
16,215,911✔
1851
  STscStmt2* pStmt = (STscStmt2*)stmt;
16,215,911✔
1852
  int32_t    code = 0;
16,215,911✔
1853

1854
  int64_t startUs = taosGetTimestampUs();
16,223,759✔
1855

1856
  STMT2_TLOG("start to bind data, colIdx:%d", colIdx);
16,223,759✔
1857

1858
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
16,218,376✔
1859
    return pStmt->errCode;
×
1860
  }
1861

1862
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_BIND));
16,218,111✔
1863

1864
  if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 &&
16,216,325✔
1865
      STMT_TYPE_MULTI_INSERT != pStmt->sql.type) {
×
1866
    pStmt->bInfo.needParse = false;
×
1867
  }
1868

1869
  if (pStmt->exec.pRequest && STMT_TYPE_QUERY == pStmt->sql.type && pStmt->sql.runTimes) {
16,215,522✔
1870
    resetRequest(pStmt);
×
1871
  }
1872

1873
  STMT_ERR_RET(stmtCreateRequest(pStmt));
16,216,339✔
1874
  if (pStmt->bInfo.needParse) {
16,216,605✔
1875
    code = stmtParseSql(pStmt);
×
1876
    if (code != TSDB_CODE_SUCCESS) {
×
1877
      goto cleanup_root;
×
1878
    }
1879
  }
1880

1881
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
16,216,605✔
1882
    code = qStmtBindParams2(pStmt->sql.pQuery, bind, colIdx, pStmt->taos->optionInfo.charsetCxt);
242✔
1883
    if (code != TSDB_CODE_SUCCESS) {
242✔
1884
      goto cleanup_root;
×
1885
    }
1886
    SParseContext ctx = {.requestId = pStmt->exec.pRequest->requestId,
242✔
1887
                         .acctId = pStmt->taos->acctId,
242✔
1888
                         .db = pStmt->exec.pRequest->pDb,
242✔
1889
                         .topicQuery = false,
1890
                         .pSql = pStmt->sql.sqlStr,
242✔
1891
                         .sqlLen = pStmt->sql.sqlLen,
242✔
1892
                         .pMsg = pStmt->exec.pRequest->msgBuf,
242✔
1893
                         .msgLen = ERROR_MSG_BUF_DEFAULT_SIZE,
1894
                         .pTransporter = pStmt->taos->pAppInfo->pTransporter,
242✔
1895
                         .pStmtCb = NULL,
1896
                         .pUser = pStmt->taos->user,
242✔
1897
                         .stmtBindVersion = pStmt->exec.pRequest->stmtBindVersion};
242✔
1898
    ctx.mgmtEpSet = getEpSet_s(&pStmt->taos->pAppInfo->mgmtEp);
242✔
1899
    code = catalogGetHandle(pStmt->taos->pAppInfo->clusterId, &ctx.pCatalog);
242✔
1900
    if (code != TSDB_CODE_SUCCESS) {
242✔
1901
      goto cleanup_root;
×
1902
    }
1903
    code = qStmtParseQuerySql(&ctx, pStmt->sql.pQuery);
242✔
1904
    if (code != TSDB_CODE_SUCCESS) {
242✔
1905
      goto cleanup_root;
×
1906
    }
1907

1908
    if (pStmt->sql.pQuery->haveResultSet) {
242✔
1909
      STMT_ERR_RET(setResSchemaInfo(&pStmt->exec.pRequest->body.resInfo, pStmt->sql.pQuery->pResSchema,
484✔
1910
                                    pStmt->sql.pQuery->numOfResCols, pStmt->sql.pQuery->pResExtSchema, true));
1911
      taosMemoryFreeClear(pStmt->sql.pQuery->pResSchema);
242✔
1912
      taosMemoryFreeClear(pStmt->sql.pQuery->pResExtSchema);
242✔
1913
      setResPrecision(&pStmt->exec.pRequest->body.resInfo, pStmt->sql.pQuery->precision);
242✔
1914
    }
1915

1916
    TSWAP(pStmt->exec.pRequest->dbList, pStmt->sql.pQuery->pDbList);
242✔
1917
    TSWAP(pStmt->exec.pRequest->tableList, pStmt->sql.pQuery->pTableList);
242✔
1918
    TSWAP(pStmt->exec.pRequest->targetTableList, pStmt->sql.pQuery->pTargetTableList);
242✔
1919

1920
    // if (STMT_TYPE_QUERY == pStmt->sql.queryRes) {
1921
    //   STMT_ERR_RET(stmtRestoreQueryFields(pStmt));
1922
    // }
1923

1924
    // STMT_ERR_RET(stmtBackupQueryFields(pStmt));
1925

1926
    return TSDB_CODE_SUCCESS;
242✔
1927

1928
  cleanup_root:
×
1929
    STMT2_ELOG("parse query statment unexpected failed code:%d, need to clean node", code);
×
1930
    if (pStmt->sql.pQuery && pStmt->sql.pQuery->pRoot) {
×
1931
      nodesDestroyNode(pStmt->sql.pQuery->pRoot);
×
1932
      pStmt->sql.pQuery->pRoot = NULL;
×
1933
    }
1934
    STMT_ERR_RET(code);
×
1935
  }
1936

1937
  if (pStmt->sql.stbInterlaceMode && NULL == pStmt->sql.siInfo.pDataCtx) {
16,217,162✔
1938
    STMT_ERR_RET(stmtInitStbInterlaceTableInfo(pStmt));
×
1939
  }
1940

1941
  STableDataCxt** pDataBlock = NULL;
16,218,796✔
1942

1943
  if (pStmt->exec.pCurrBlock) {
16,218,796✔
1944
    pDataBlock = &pStmt->exec.pCurrBlock;
16,198,041✔
1945
  } else {
1946
    pDataBlock =
1947
        (STableDataCxt**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
21,289✔
1948
    if (NULL == pDataBlock) {
21,383✔
1949
      STMT2_ELOG("table %s not found in exec blockHash:%p", pStmt->bInfo.tbFName, pStmt->exec.pBlockHash);
×
1950
      STMT_ERR_RET(TSDB_CODE_TSC_STMT_CACHE_ERROR);
×
1951
    }
1952
    pStmt->exec.pCurrBlock = *pDataBlock;
21,383✔
1953
    if (pStmt->sql.stbInterlaceMode) {
21,383✔
1954
      taosArrayDestroy(pStmt->exec.pCurrBlock->pData->aCol);
20,673✔
1955
      (*pDataBlock)->pData->aCol = NULL;
20,673✔
1956
    }
1957
    if (colIdx < -1) {
21,383✔
1958
      pStmt->sql.bindRowFormat = true;
×
1959
      taosArrayDestroy((*pDataBlock)->pData->aCol);
×
1960
      (*pDataBlock)->pData->aCol = taosArrayInit(20, POINTER_BYTES);
×
1961
    }
1962
  }
1963

1964
  int64_t startUs2 = taosGetTimestampUs();
16,214,924✔
1965
  pStmt->stat.bindDataUs1 += startUs2 - startUs;
16,214,924✔
1966

1967
  SStmtQNode* param = NULL;
16,215,476✔
1968
  if (pStmt->sql.stbInterlaceMode) {
16,215,476✔
1969
    STMT_ERR_RET(stmtAllocQNodeFromBuf(&pStmt->sql.siInfo.tbBuf, (void**)&param));
32,434,867✔
1970
    STMT_ERR_RET(stmtGetTableColsFromCache(pStmt, &param->tblData.aCol));
32,435,612✔
1971
    taosArrayClear(param->tblData.aCol);
16,218,794✔
1972

1973
    // param->tblData.aCol = taosArrayInit(20, POINTER_BYTES);
1974

1975
    param->restoreTbCols = false;
16,210,053✔
1976
    param->tblData.isOrdered = true;
16,210,053✔
1977
    param->tblData.isDuplicateTs = false;
16,210,053✔
1978
    tstrncpy(param->tblData.tbName, pStmt->bInfo.tbName, TSDB_TABLE_NAME_LEN);
16,210,053✔
1979

1980
    param->pCreateTbReq = pCreateTbReq;
16,210,332✔
1981
  }
1982

1983
  int64_t startUs3 = taosGetTimestampUs();
16,221,465✔
1984
  pStmt->stat.bindDataUs2 += startUs3 - startUs2;
16,221,465✔
1985

1986
  SArray* pCols = pStmt->sql.stbInterlaceMode ? param->tblData.aCol : (*pDataBlock)->pData->aCol;
16,220,393✔
1987
  SBlobSet* pBlob = NULL;
16,221,469✔
1988
  if (colIdx < 0) {
16,220,397✔
1989
    if (pStmt->sql.stbInterlaceMode) {
16,222,619✔
1990
      (*pDataBlock)->pData->flags &= ~SUBMIT_REQ_COLUMN_DATA_FORMAT;
16,213,471✔
1991
      code = qBindStmtStbColsValue2(*pDataBlock, pCols, pStmt->bInfo.boundCols, bind, pStmt->exec.pRequest->msgBuf,
18,190,369✔
1992
                                    pStmt->exec.pRequest->msgBufLen, &pStmt->sql.siInfo.pTSchema, pStmt->sql.pBindInfo,
16,213,467✔
1993
                                    pStmt->taos->optionInfo.charsetCxt, &pBlob);
16,213,736✔
1994
      param->tblData.isOrdered = (*pDataBlock)->ordered;
16,215,813✔
1995
      param->tblData.isDuplicateTs = (*pDataBlock)->duplicateTs;
16,216,889✔
1996
    } else {
1997
      if (colIdx == -1) {
9,686✔
1998
        if (pStmt->sql.bindRowFormat) {
5,406✔
1999
          STMT2_ELOG_E("can't mix bind row format and bind column format");
×
2000
          STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR);
×
2001
        }
2002
        code = qBindStmtColsValue2(*pDataBlock, pCols, pStmt->bInfo.boundCols, bind, pStmt->exec.pRequest->msgBuf,
5,406✔
2003
                                   pStmt->exec.pRequest->msgBufLen, pStmt->taos->optionInfo.charsetCxt);
5,406✔
2004
      } else {
2005
        code =
2006
            qBindStmt2RowValue(*pDataBlock, (*pDataBlock)->pData->aRowP, pStmt->bInfo.boundCols, bind,
4,280✔
2007
                               pStmt->exec.pRequest->msgBuf, pStmt->exec.pRequest->msgBufLen,
4,280✔
2008
                               &pStmt->sql.siInfo.pTSchema, pStmt->sql.pBindInfo, pStmt->taos->optionInfo.charsetCxt);
4,280✔
2009
      }
2010
    }
2011

2012
    if (code) {
16,221,219✔
2013
      STMT2_ELOG("bind cols or rows failed, error:%s", tstrerror(code));
×
2014
      STMT_ERR_RET(code);
×
2015
    }
2016
  } else {
2017
    if (pStmt->sql.stbInterlaceMode) {
×
2018
      STMT2_ELOG_E("bind single column not allowed in stb insert mode");
×
2019
      STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR);
×
2020
    }
2021

2022
    if (pStmt->sql.bindRowFormat) {
×
2023
      STMT2_ELOG_E("can't mix bind row format and bind column format");
×
2024
      STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR);
×
2025
    }
2026

2027
    if (colIdx != (pStmt->bInfo.sBindLastIdx + 1) && colIdx != 0) {
×
2028
      STMT2_ELOG_E("bind column index not in sequence");
×
2029
      STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR);
×
2030
    }
2031

2032
    pStmt->bInfo.sBindLastIdx = colIdx;
×
2033

2034
    if (0 == colIdx) {
×
2035
      pStmt->bInfo.sBindRowNum = bind->num;
×
2036
    }
2037

2038
    code = qBindStmtSingleColValue2(*pDataBlock, pCols, bind, pStmt->exec.pRequest->msgBuf,
×
2039
                                    pStmt->exec.pRequest->msgBufLen, colIdx, pStmt->bInfo.sBindRowNum,
×
2040
                                    pStmt->taos->optionInfo.charsetCxt);
×
2041
    if (code) {
×
2042
      STMT2_ELOG("bind single col failed, error:%s", tstrerror(code));
×
2043
      STMT_ERR_RET(code);
×
2044
    }
2045
  }
2046

2047
  int64_t startUs4 = taosGetTimestampUs();
16,220,220✔
2048
  pStmt->stat.bindDataUs3 += startUs4 - startUs3;
16,220,220✔
2049

2050
  if (pStmt->stbInterlaceMode) {
16,218,875✔
2051
    if (param) param->tblData.pBlobSet = pBlob;
16,219,517✔
2052
  }
2053

2054
  if (pStmt->sql.stbInterlaceMode) {
16,220,489✔
2055
    STMT_ERR_RET(stmtAppendTablePostHandle(pStmt, param));
16,219,850✔
2056
  } else {
2057
    STMT_ERR_RET(stmtAddBatch2(pStmt));
1,259✔
2058
  }
2059

2060
  pStmt->stat.bindDataUs4 += taosGetTimestampUs() - startUs4;
16,225,583✔
2061
  return TSDB_CODE_SUCCESS;
16,225,314✔
2062
}
2063

2064
/*
2065
int stmtUpdateTableUid(STscStmt2* pStmt, SSubmitRsp* pRsp) {
2066
  tscDebug("stmt start to update tbUid, blockNum:%d", pRsp->nBlocks);
2067

2068
  int32_t code = 0;
2069
  int32_t finalCode = 0;
2070
  size_t  keyLen = 0;
2071
  void*   pIter = taosHashIterate(pStmt->exec.pBlockHash, NULL);
2072
  while (pIter) {
2073
    STableDataCxt* pBlock = *(STableDataCxt**)pIter;
2074
    char*          key = taosHashGetKey(pIter, &keyLen);
2075

2076
    STableMeta* pMeta = qGetTableMetaInDataBlock(pBlock);
2077
    if (pMeta->uid) {
2078
      pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
2079
      continue;
2080
    }
2081

2082
    SSubmitBlkRsp* blkRsp = NULL;
2083
    int32_t        i = 0;
2084
    for (; i < pRsp->nBlocks; ++i) {
2085
      blkRsp = pRsp->pBlocks + i;
2086
      if (strlen(blkRsp->tblFName) != keyLen) {
2087
        continue;
2088
      }
2089

2090
      if (strncmp(blkRsp->tblFName, key, keyLen)) {
2091
        continue;
2092
      }
2093

2094
      break;
2095
    }
2096

2097
    if (i < pRsp->nBlocks) {
2098
      tscDebug("auto created table %s uid updated from %" PRIx64 " to %" PRIx64, blkRsp->tblFName, pMeta->uid,
2099
               blkRsp->uid);
2100

2101
      pMeta->uid = blkRsp->uid;
2102
      pStmt->bInfo.tbUid = blkRsp->uid;
2103
    } else {
2104
      tscDebug("table %s not found in submit rsp, will update from catalog", pStmt->bInfo.tbFName);
2105
      if (NULL == pStmt->pCatalog) {
2106
        code = catalogGetHandle(pStmt->taos->pAppInfo->clusterId, &pStmt->pCatalog);
2107
        if (code) {
2108
          pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
2109
          finalCode = code;
2110
          continue;
2111
        }
2112
      }
2113

2114
      code = stmtCreateRequest(pStmt);
2115
      if (code) {
2116
        pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
2117
        finalCode = code;
2118
        continue;
2119
      }
2120

2121
      STableMeta*      pTableMeta = NULL;
2122
      SRequestConnInfo conn = {.pTrans = pStmt->taos->pAppInfo->pTransporter,
2123
                               .requestId = pStmt->exec.pRequest->requestId,
2124
                               .requestObjRefId = pStmt->exec.pRequest->self,
2125
                               .mgmtEps = getEpSet_s(&pStmt->taos->pAppInfo->mgmtEp)};
2126
      code = catalogGetTableMeta(pStmt->pCatalog, &conn, &pStmt->bInfo.sname, &pTableMeta);
2127

2128
      pStmt->stat.ctgGetTbMetaNum++;
2129

2130
      taos_free_result(pStmt->exec.pRequest);
2131
      pStmt->exec.pRequest = NULL;
2132

2133
      if (code || NULL == pTableMeta) {
2134
        pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
2135
        finalCode = code;
2136
        taosMemoryFree(pTableMeta);
2137
        continue;
2138
      }
2139

2140
      pMeta->uid = pTableMeta->uid;
2141
      pStmt->bInfo.tbUid = pTableMeta->uid;
2142
      taosMemoryFree(pTableMeta);
2143
    }
2144

2145
    pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
2146
  }
2147

2148
  return finalCode;
2149
}
2150
*/
2151
/*
2152
int stmtStaticModeExec(TAOS_STMT* stmt) {
2153
  STscStmt2*   pStmt = (STscStmt2*)stmt;
2154
  int32_t     code = 0;
2155
  SSubmitRsp* pRsp = NULL;
2156
  if (pStmt->sql.staticMode) {
2157
    return TSDB_CODE_TSC_STMT_API_ERROR;
2158
  }
2159

2160
  STMT_DLOG_E("start to exec");
2161

2162
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_EXECUTE));
2163

2164
  STMT_ERR_RET(qBuildStmtOutputFromTbList(pStmt->sql.pQuery, pStmt->sql.pVgHash, pStmt->exec.pTbBlkList,
2165
pStmt->exec.pCurrBlock, pStmt->exec.tbBlkNum));
2166

2167
  launchQueryImpl(pStmt->exec.pRequest, pStmt->sql.pQuery, true, NULL);
2168

2169
  if (pStmt->exec.pRequest->code && NEED_CLIENT_HANDLE_ERROR(pStmt->exec.pRequest->code)) {
2170
    code = refreshMeta(pStmt->exec.pRequest->pTscObj, pStmt->exec.pRequest);
2171
    if (code) {
2172
      pStmt->exec.pRequest->code = code;
2173
    } else {
2174
      tFreeSSubmitRsp(pRsp);
2175
      STMT_ERR_RET(stmtResetStmt(pStmt));
2176
      STMT_ERR_RET(TSDB_CODE_NEED_RETRY);
2177
    }
2178
  }
2179

2180
  STMT_ERR_JRET(pStmt->exec.pRequest->code);
2181

2182
  pStmt->exec.affectedRows = taos_affected_rows(pStmt->exec.pRequest);
2183
  pStmt->affectedRows += pStmt->exec.affectedRows;
2184

2185
_return:
2186

2187
  stmtCleanExecInfo(pStmt, (code ? false : true), false);
2188

2189
  tFreeSSubmitRsp(pRsp);
2190

2191
  ++pStmt->sql.runTimes;
2192

2193
  STMT_RET(code);
2194
}
2195
*/
2196

2197
static int32_t createParseContext(const SRequestObj* pRequest, SParseContext** pCxt, SSqlCallbackWrapper* pWrapper) {
×
2198
  const STscObj* pTscObj = pRequest->pTscObj;
×
2199

2200
  *pCxt = taosMemoryCalloc(1, sizeof(SParseContext));
×
2201
  if (*pCxt == NULL) {
×
2202
    return terrno;
×
2203
  }
2204

2205
  **pCxt = (SParseContext){.requestId = pRequest->requestId,
×
2206
                           .requestRid = pRequest->self,
×
2207
                           .acctId = pTscObj->acctId,
×
2208
                           .db = pRequest->pDb,
×
2209
                           .topicQuery = false,
2210
                           .pSql = pRequest->sqlstr,
×
2211
                           .sqlLen = pRequest->sqlLen,
×
2212
                           .pMsg = pRequest->msgBuf,
×
2213
                           .msgLen = ERROR_MSG_BUF_DEFAULT_SIZE,
2214
                           .pTransporter = pTscObj->pAppInfo->pTransporter,
×
2215
                           .pStmtCb = NULL,
2216
                           .pUser = pTscObj->user,
×
2217
                           .pEffectiveUser = pRequest->effectiveUser,
×
2218
                           .isSuperUser = (0 == strcmp(pTscObj->user, TSDB_DEFAULT_USER)),
×
2219
                           .enableSysInfo = pTscObj->sysInfo,
×
2220
                           .async = true,
2221
                           .svrVer = pTscObj->sVer,
×
2222
                           .nodeOffline = (pTscObj->pAppInfo->onlineDnodes < pTscObj->pAppInfo->totalDnodes),
×
2223
                           .allocatorId = pRequest->allocatorRefId,
×
2224
                           .parseSqlFp = clientParseSql,
2225
                           .parseSqlParam = pWrapper};
2226
  int8_t biMode = atomic_load_8(&((STscObj*)pTscObj)->biMode);
×
2227
  (*pCxt)->biMode = biMode;
×
2228
  return TSDB_CODE_SUCCESS;
×
2229
}
2230

2231
static void asyncQueryCb(void* userdata, TAOS_RES* res, int code) {
×
2232
  STscStmt2*        pStmt = userdata;
×
2233
  __taos_async_fn_t fp = pStmt->options.asyncExecFn;
×
2234
  pStmt->asyncResultAvailable = true;
×
2235
  pStmt->exec.pRequest->inCallback = true;
×
2236

2237
  if (code == TSDB_CODE_SUCCESS) {
×
2238
    pStmt->exec.affectedRows = taos_affected_rows(res);
×
2239
    pStmt->affectedRows += pStmt->exec.affectedRows;
×
2240
  }
2241

2242
  if (fp) {
×
2243
    fp(pStmt->options.userdata, res, code);
×
2244
  }
2245

2246
  while (0 == atomic_load_8((int8_t*)&pStmt->sql.siInfo.tableColsReady)) {
×
2247
    taosUsleep(1);
×
2248
  }
2249
  (void)stmtCleanExecInfo(pStmt, (code ? false : true), false);
×
2250
  ++pStmt->sql.runTimes;
×
2251
  if (pStmt->exec.pRequest != NULL) {
×
2252
    pStmt->exec.pRequest->inCallback = false;
×
2253
  }
2254

2255
  if (tsem_post(&pStmt->asyncExecSem) != 0) {
×
2256
    STMT2_ELOG_E("fail to post asyncExecSem");
×
2257
  }
2258
}
×
2259

2260
int stmtExec2(TAOS_STMT2* stmt, int* affected_rows) {
7,389,080✔
2261
  STscStmt2* pStmt = (STscStmt2*)stmt;
7,389,080✔
2262
  int32_t    code = 0;
7,389,080✔
2263
  int64_t    startUs = taosGetTimestampUs();
7,392,207✔
2264

2265
  STMT2_DLOG_E("start to exec");
7,392,207✔
2266

2267
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
7,392,886✔
2268
    return pStmt->errCode;
×
2269
  }
2270

2271
  STMT_ERR_RET(taosThreadMutexLock(&pStmt->asyncBindParam.mutex));
7,392,886✔
2272
  while (atomic_load_8((int8_t*)&pStmt->asyncBindParam.asyncBindNum) > 0) {
7,393,134✔
2273
    (void)taosThreadCondWait(&pStmt->asyncBindParam.waitCond, &pStmt->asyncBindParam.mutex);
×
2274
  }
2275
  STMT_ERR_RET(taosThreadMutexUnlock(&pStmt->asyncBindParam.mutex));
7,392,116✔
2276

2277
  if (pStmt->sql.stbInterlaceMode) {
7,392,771✔
2278
    STMT_ERR_RET(stmtAddBatch2(pStmt));
7,391,755✔
2279
  }
2280

2281
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_EXECUTE));
7,393,198✔
2282

2283
  if (STMT_TYPE_QUERY != pStmt->sql.type) {
7,391,187✔
2284
    if (pStmt->sql.stbInterlaceMode) {
7,391,192✔
2285
      int64_t startTs = taosGetTimestampUs();
7,391,957✔
2286
      // wait for stmt bind thread to finish
2287
      while (atomic_load_64(&pStmt->sql.siInfo.tbRemainNum)) {
26,533,673✔
2288
        taosUsleep(1);
19,142,298✔
2289
      }
2290

2291
      if (pStmt->errCode != TSDB_CODE_SUCCESS) {
7,390,545✔
2292
        return pStmt->errCode;
×
2293
      }
2294

2295
      pStmt->stat.execWaitUs += taosGetTimestampUs() - startTs;
7,391,772✔
2296
      STMT_ERR_RET(qBuildStmtFinOutput(pStmt->sql.pQuery, pStmt->sql.pVgHash, pStmt->sql.siInfo.pVgroupList));
7,391,772✔
2297
      taosHashCleanup(pStmt->sql.siInfo.pVgroupHash);
7,391,910✔
2298
      pStmt->sql.siInfo.pVgroupHash = NULL;
7,391,006✔
2299
      pStmt->sql.siInfo.pVgroupList = NULL;
7,391,275✔
2300
    } else {
2301
      tDestroySubmitTbData(pStmt->exec.pCurrTbData, TSDB_MSG_FLG_ENCODE);
281✔
2302
      taosMemoryFreeClear(pStmt->exec.pCurrTbData);
710✔
2303

2304
      STMT_ERR_RET(qCloneCurrentTbData(pStmt->exec.pCurrBlock, &pStmt->exec.pCurrTbData));
710✔
2305

2306
      STMT_ERR_RET(qBuildStmtOutput(pStmt->sql.pQuery, pStmt->sql.pVgHash, pStmt->exec.pBlockHash));
710✔
2307
    }
2308
  }
2309

2310
  pStmt->asyncResultAvailable = false;
7,391,711✔
2311
  SRequestObj*      pRequest = pStmt->exec.pRequest;
7,391,442✔
2312
  __taos_async_fn_t fp = pStmt->options.asyncExecFn;
7,391,442✔
2313
  STMT2_DLOG("EXEC INFO :req:0x%" PRIx64 ", QID:0x%" PRIx64 ", exec sql:%s,  conn:%" PRId64, pRequest->self,
7,391,711✔
2314
             pRequest->requestId, pStmt->sql.sqlStr, pRequest->pTscObj->id);
2315

2316
  if (!fp) {
7,391,840✔
2317
    launchQueryImpl(pStmt->exec.pRequest, pStmt->sql.pQuery, true, NULL);
7,391,840✔
2318

2319
    if (pStmt->exec.pRequest->code && NEED_CLIENT_HANDLE_ERROR(pStmt->exec.pRequest->code)) {
7,387,641✔
2320
      STMT2_ELOG_E("exec failed errorcode:NEED_CLIENT_HANDLE_ERROR, need to refresh meta and retry");
×
2321
      code = refreshMeta(pStmt->exec.pRequest->pTscObj, pStmt->exec.pRequest);
×
2322
      if (code) {
×
2323
        pStmt->exec.pRequest->code = code;
×
2324

2325
      } else {
2326
        STMT_ERR_RET(stmtResetStmt(pStmt));
×
2327
        STMT_ERR_RET(TSDB_CODE_NEED_RETRY);
×
2328
      }
2329
    }
2330

2331
    STMT_ERR_JRET(pStmt->exec.pRequest->code);
7,390,834✔
2332

2333
    pStmt->exec.affectedRows = taos_affected_rows(pStmt->exec.pRequest);
7,390,565✔
2334
    if (affected_rows) {
7,391,262✔
2335
      *affected_rows = pStmt->exec.affectedRows;
7,389,540✔
2336
    }
2337
    pStmt->affectedRows += pStmt->exec.affectedRows;
7,391,262✔
2338

2339
    // wait for stmt bind thread to finish
2340
    while (0 == atomic_load_8((int8_t*)&pStmt->sql.siInfo.tableColsReady)) {
7,405,720✔
2341
      taosUsleep(1);
15,010✔
2342
    }
2343

2344
    STMT_ERR_RET(stmtCleanExecInfo(pStmt, (code ? false : true), false));
7,390,594✔
2345

2346
    ++pStmt->sql.runTimes;
7,390,260✔
2347
  } else {
2348
    SSqlCallbackWrapper* pWrapper = taosMemoryCalloc(1, sizeof(SSqlCallbackWrapper));
×
2349
    if (pWrapper == NULL) {
×
2350
      code = terrno;
×
2351
    } else {
2352
      pWrapper->pRequest = pRequest;
×
2353
      pRequest->pWrapper = pWrapper;
×
2354
    }
2355
    if (TSDB_CODE_SUCCESS == code) {
×
2356
      code = createParseContext(pRequest, &pWrapper->pParseCtx, pWrapper);
×
2357
    }
2358
    pRequest->syncQuery = false;
×
2359
    pRequest->body.queryFp = asyncQueryCb;
×
2360
    ((SSyncQueryParam*)(pRequest)->body.interParam)->userParam = pStmt;
×
2361

2362
    pStmt->execSemWaited = false;
×
2363
    launchAsyncQuery(pRequest, pStmt->sql.pQuery, NULL, pWrapper);
×
2364
  }
2365

2366
_return:
7,389,977✔
2367
  if (code) {
7,389,977✔
2368
    STMT2_ELOG("exec failed, error:%s", tstrerror(code));
×
2369
  }
2370
  pStmt->stat.execUseUs += taosGetTimestampUs() - startUs;
7,390,739✔
2371

2372
  STMT_RET(code);
7,391,008✔
2373
}
2374

2375
int stmtClose2(TAOS_STMT2* stmt) {
19,609✔
2376
  STscStmt2* pStmt = (STscStmt2*)stmt;
19,609✔
2377

2378
  STMT2_DLOG_E("start to close stmt");
19,609✔
2379
  taosMemoryFreeClear(pStmt->db);
19,609✔
2380

2381
  if (pStmt->bindThreadInUse) {
19,609✔
2382
    // wait for stmt bind thread to finish
2383
    while (0 == atomic_load_8((int8_t*)&pStmt->sql.siInfo.tableColsReady)) {
18,657✔
2384
      taosUsleep(1);
×
2385
    }
2386

2387
    (void)taosThreadMutexLock(&pStmt->queue.mutex);
18,657✔
2388
    pStmt->queue.stopQueue = true;
18,657✔
2389
    (void)taosThreadCondSignal(&(pStmt->queue.waitCond));
18,657✔
2390
    (void)taosThreadMutexUnlock(&pStmt->queue.mutex);
18,657✔
2391

2392
    (void)taosThreadJoin(pStmt->bindThread, NULL);
18,657✔
2393
    pStmt->bindThreadInUse = false;
18,657✔
2394

2395
    (void)taosThreadCondDestroy(&pStmt->queue.waitCond);
18,657✔
2396
    (void)taosThreadMutexDestroy(&pStmt->queue.mutex);
18,657✔
2397
  }
2398

2399
  TSC_ERR_RET(taosThreadMutexLock(&pStmt->asyncBindParam.mutex));
19,609✔
2400
  while (atomic_load_8((int8_t*)&pStmt->asyncBindParam.asyncBindNum) > 0) {
19,609✔
2401
    (void)taosThreadCondWait(&pStmt->asyncBindParam.waitCond, &pStmt->asyncBindParam.mutex);
×
2402
  }
2403
  TSC_ERR_RET(taosThreadMutexUnlock(&pStmt->asyncBindParam.mutex));
19,609✔
2404

2405
  (void)taosThreadCondDestroy(&pStmt->asyncBindParam.waitCond);
19,609✔
2406
  (void)taosThreadMutexDestroy(&pStmt->asyncBindParam.mutex);
19,609✔
2407

2408
  if (pStmt->options.asyncExecFn && !pStmt->execSemWaited) {
19,609✔
2409
    if (tsem_wait(&pStmt->asyncExecSem) != 0) {
×
2410
      STMT2_ELOG_E("fail to wait asyncExecSem");
×
2411
    }
2412
  }
2413

2414
  STMT2_DLOG("stmt %p closed, stbInterlaceMode:%d, statInfo: ctgGetTbMetaNum=>%" PRId64 ", getCacheTbInfo=>%" PRId64
19,609✔
2415
             ", parseSqlNum=>%" PRId64 ", pStmt->stat.bindDataNum=>%" PRId64
2416
             ", settbnameAPI:%u, bindAPI:%u, addbatchAPI:%u, execAPI:%u"
2417
             ", setTbNameUs:%" PRId64 ", bindDataUs:%" PRId64 ",%" PRId64 ",%" PRId64 ",%" PRId64 " addBatchUs:%" PRId64
2418
             ", execWaitUs:%" PRId64 ", execUseUs:%" PRId64,
2419
             pStmt, pStmt->sql.stbInterlaceMode, pStmt->stat.ctgGetTbMetaNum, pStmt->stat.getCacheTbInfo,
2420
             pStmt->stat.parseSqlNum, pStmt->stat.bindDataNum, pStmt->seqIds[STMT_SETTBNAME], pStmt->seqIds[STMT_BIND],
2421
             pStmt->seqIds[STMT_ADD_BATCH], pStmt->seqIds[STMT_EXECUTE], pStmt->stat.setTbNameUs,
2422
             pStmt->stat.bindDataUs1, pStmt->stat.bindDataUs2, pStmt->stat.bindDataUs3, pStmt->stat.bindDataUs4,
2423
             pStmt->stat.addBatchUs, pStmt->stat.execWaitUs, pStmt->stat.execUseUs);
2424
  if (pStmt->sql.stbInterlaceMode) {
19,609✔
2425
    pStmt->bInfo.tagsCached = false;
18,657✔
2426
  }
2427
  pStmt->bInfo.boundColsCached = false;
19,609✔
2428

2429
  STMT_ERR_RET(stmtCleanSQLInfo(pStmt));
19,609✔
2430

2431
  if (pStmt->options.asyncExecFn) {
19,609✔
2432
    if (tsem_destroy(&pStmt->asyncExecSem) != 0) {
×
2433
      STMT2_ELOG_E("fail to destroy asyncExecSem");
×
2434
    }
2435
  }
2436
  taosMemoryFree(stmt);
19,609✔
2437

2438
  return TSDB_CODE_SUCCESS;
19,609✔
2439
}
2440

2441
const char* stmtErrstr2(TAOS_STMT2* stmt) {
×
2442
  STscStmt2* pStmt = (STscStmt2*)stmt;
×
2443

2444
  if (stmt == NULL || NULL == pStmt->exec.pRequest) {
×
2445
    return (char*)tstrerror(terrno);
×
2446
  }
2447

2448
  // if stmt async exec ,error code is pStmt->exec.pRequest->code
2449
  if (!(pStmt->sql.status >= STMT_EXECUTE && pStmt->options.asyncExecFn != NULL && pStmt->asyncResultAvailable)) {
×
2450
    pStmt->exec.pRequest->code = terrno;
×
2451
  }
2452

2453
  SRequestObj* pRequest = pStmt->exec.pRequest;
×
2454
  if (NULL != pRequest->msgBuf && (strlen(pRequest->msgBuf) > 0 || pRequest->code == TSDB_CODE_RPC_FQDN_ERROR)) {
×
2455
    return pRequest->msgBuf;
×
2456
  }
2457
  return (const char*)tstrerror(pRequest->code);
×
2458
}
2459
/*
2460
int stmtAffectedRows(TAOS_STMT* stmt) { return ((STscStmt2*)stmt)->affectedRows; }
2461

2462
int stmtAffectedRowsOnce(TAOS_STMT* stmt) { return ((STscStmt2*)stmt)->exec.affectedRows; }
2463
*/
2464

2465
int stmtParseColFields2(TAOS_STMT2* stmt) {
703✔
2466
  int32_t    code = 0;
703✔
2467
  STscStmt2* pStmt = (STscStmt2*)stmt;
703✔
2468
  int32_t    preCode = pStmt->errCode;
703✔
2469

2470
  STMT2_DLOG_E("start to get col fields for insert");
703✔
2471

2472
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
703✔
2473
    return pStmt->errCode;
×
2474
  }
2475

2476
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
703✔
2477
    STMT2_ELOG_E("stmtParseColFields2 only for insert");
×
2478
    STMT_ERRI_JRET(TSDB_CODE_TSC_STMT_API_ERROR);
×
2479
  }
2480

2481
  STMT_ERRI_JRET(stmtSwitchStatus(pStmt, STMT_FETCH_FIELDS));
703✔
2482

2483
  if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 &&
703✔
2484
      STMT_TYPE_MULTI_INSERT != pStmt->sql.type) {
×
2485
    pStmt->bInfo.needParse = false;
×
2486
  }
2487
  if (pStmt->sql.stbInterlaceMode && pStmt->sql.siInfo.pDataCtx != NULL) {
703✔
2488
    pStmt->bInfo.needParse = false;
×
2489
  }
2490

2491
  STMT_ERRI_JRET(stmtCreateRequest(pStmt));
703✔
2492

2493
  if (pStmt->bInfo.needParse) {
703✔
2494
    STMT_ERRI_JRET(stmtParseSql(pStmt));
703✔
2495
  }
2496

2497
_return:
703✔
2498
  // compatible with previous versions
2499
  if (code == TSDB_CODE_PAR_TABLE_NOT_EXIST && (pStmt->bInfo.tbNameFlag & NO_DATA_USING_CLAUSE) == 0x0) {
703✔
2500
    code = TSDB_CODE_TSC_STMT_TBNAME_ERROR;
×
2501
  }
2502

2503
  pStmt->errCode = preCode;
703✔
2504

2505
  return code;
703✔
2506
}
2507

2508
int stmtGetStbColFields2(TAOS_STMT2* stmt, int* nums, TAOS_FIELD_ALL** fields) {
703✔
2509
  int32_t code = stmtParseColFields2(stmt);
703✔
2510
  if (code != TSDB_CODE_SUCCESS) {
703✔
2511
    return code;
×
2512
  }
2513

2514
  return stmtFetchStbColFields2(stmt, nums, fields);
703✔
2515
}
2516

2517
int stmtGetParamNum2(TAOS_STMT2* stmt, int* nums) {
×
2518
  int32_t    code = 0;
×
2519
  STscStmt2* pStmt = (STscStmt2*)stmt;
×
2520
  int32_t    preCode = pStmt->errCode;
×
2521

2522
  STMT2_DLOG_E("start to get param num for query");
×
2523

2524
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
×
2525
    return pStmt->errCode;
×
2526
  }
2527

2528
  STMT_ERRI_JRET(stmtSwitchStatus(pStmt, STMT_FETCH_FIELDS));
×
2529

2530
  if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 &&
×
2531
      STMT_TYPE_MULTI_INSERT != pStmt->sql.type) {
×
2532
    pStmt->bInfo.needParse = false;
×
2533
  }
2534

2535
  if (pStmt->exec.pRequest && STMT_TYPE_QUERY == pStmt->sql.type && pStmt->sql.runTimes) {
×
2536
    resetRequest(pStmt);
×
2537
  }
2538

2539
  STMT_ERRI_JRET(stmtCreateRequest(pStmt));
×
2540

2541
  if (pStmt->bInfo.needParse) {
×
2542
    STMT_ERRI_JRET(stmtParseSql(pStmt));
×
2543
  }
2544

2545
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
×
2546
    *nums = taosArrayGetSize(pStmt->sql.pQuery->pPlaceholderValues);
×
2547
  } else {
2548
    STMT_ERRI_JRET(stmtFetchColFields2(stmt, nums, NULL));
×
2549
  }
2550

2551
  STMT2_TLOG("get param num success, nums:%d", *nums);
×
2552

2553
_return:
×
2554

2555
  pStmt->errCode = preCode;
×
2556

2557
  return code;
×
2558
}
2559

2560
TAOS_RES* stmtUseResult2(TAOS_STMT2* stmt) {
242✔
2561
  STscStmt2* pStmt = (STscStmt2*)stmt;
242✔
2562

2563
  STMT2_TLOG_E("start to use result");
242✔
2564

2565
  if (STMT_TYPE_QUERY != pStmt->sql.type) {
242✔
2566
    STMT2_ELOG_E("useResult only for query statement");
×
2567
    return NULL;
×
2568
  }
2569

2570
  if (pStmt->options.asyncExecFn != NULL && !pStmt->asyncResultAvailable) {
242✔
2571
    STMT2_ELOG_E("use result after callBackFn return");
×
2572
    return NULL;
×
2573
  }
2574

2575
  return pStmt->exec.pRequest;
242✔
2576
}
2577

2578
int32_t stmtAsyncBindThreadFunc(void* args) {
×
2579
  qInfo("async stmt bind thread started");
×
2580

2581
  ThreadArgs* targs = (ThreadArgs*)args;
×
2582
  STscStmt2*  pStmt = (STscStmt2*)targs->stmt;
×
2583

2584
  int code = taos_stmt2_bind_param(targs->stmt, targs->bindv, targs->col_idx);
×
2585
  targs->fp(targs->param, NULL, code);
×
2586
  (void)taosThreadMutexLock(&(pStmt->asyncBindParam.mutex));
×
2587
  (void)atomic_sub_fetch_8(&pStmt->asyncBindParam.asyncBindNum, 1);
×
2588
  (void)taosThreadCondSignal(&(pStmt->asyncBindParam.waitCond));
×
2589
  (void)taosThreadMutexUnlock(&(pStmt->asyncBindParam.mutex));
×
2590
  taosMemoryFree(args);
×
2591

2592
  qInfo("async stmt bind thread stopped");
×
2593

2594
  return code;
×
2595
}
2596

2597
void stmtBuildErrorMsg(STscStmt2* pStmt, const char* msg) {
×
2598
  if (pStmt == NULL || msg == NULL) {
×
2599
    return;
×
2600
  }
2601

2602
  if (pStmt->exec.pRequest == NULL) {
×
2603
    return;
×
2604
  }
2605

2606
  if (pStmt->exec.pRequest->msgBuf == NULL) {
×
2607
    return;
×
2608
  }
2609

2610
  size_t msgLen = strlen(msg);
×
2611
  size_t bufLen = pStmt->exec.pRequest->msgBufLen;
×
2612

2613
  if (msgLen >= bufLen) {
×
2614
    tstrncpy(pStmt->exec.pRequest->msgBuf, msg, bufLen - 1);
×
2615
    pStmt->exec.pRequest->msgBuf[bufLen - 1] = '\0';
×
2616
    pStmt->exec.pRequest->msgBufLen = bufLen - 1;
×
2617
  } else {
2618
    tstrncpy(pStmt->exec.pRequest->msgBuf, msg, bufLen);
×
2619
    pStmt->exec.pRequest->msgBufLen = msgLen;
×
2620
  }
2621

2622
  return;
×
2623
}
2624

2625
int32_t stmtBuildErrorMsgWithCode(STscStmt2* pStmt, const char* msg, int32_t errorCode) {
×
2626
  stmtBuildErrorMsg(pStmt, msg);
×
2627
  pStmt->errCode = errorCode;
×
2628

2629
  return errorCode;
×
2630
}
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