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

taosdata / TDengine / #4513

17 Jul 2025 02:02AM UTC coverage: 31.359% (-31.1%) from 62.446%
#4513

push

travis-ci

web-flow
Merge pull request #31914 from taosdata/fix/3.0/compare-ans-failed

fix:Convert line endings from LF to CRLF for ans file

68541 of 301034 branches covered (22.77%)

Branch coverage included in aggregate %.

117356 of 291771 relevant lines covered (40.22%)

602262.98 hits per line

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

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

5
#include "clientStmt.h"
6
#include "clientStmt2.h"
7
/*
8
char* gStmtStatusStr[] = {"unknown",     "init", "prepare", "settbname", "settags",
9
                          "fetchFields", "bind", "bindCol", "addBatch",  "exec"};
10
*/
11
static FORCE_INLINE int32_t stmtAllocQNodeFromBuf(STableBufInfo* pTblBuf, void** pBuf) {
12
  if (pTblBuf->buffOffset < pTblBuf->buffSize) {
×
13
    *pBuf = (char*)pTblBuf->pCurBuff + pTblBuf->buffOffset;
×
14
    pTblBuf->buffOffset += pTblBuf->buffUnit;
×
15
  } else if (pTblBuf->buffIdx < taosArrayGetSize(pTblBuf->pBufList)) {
×
16
    pTblBuf->pCurBuff = taosArrayGetP(pTblBuf->pBufList, pTblBuf->buffIdx++);
×
17
    if (NULL == pTblBuf->pCurBuff) {
×
18
      return TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);
×
19
    }
20
    *pBuf = pTblBuf->pCurBuff;
×
21
    pTblBuf->buffOffset = pTblBuf->buffUnit;
×
22
  } else {
23
    void* buff = taosMemoryMalloc(pTblBuf->buffSize);
×
24
    if (NULL == buff) {
×
25
      return terrno;
×
26
    }
27

28
    if (taosArrayPush(pTblBuf->pBufList, &buff) == NULL) {
×
29
      return terrno;
×
30
    }
31

32
    pTblBuf->buffIdx++;
×
33
    pTblBuf->pCurBuff = buff;
×
34
    *pBuf = buff;
×
35
    pTblBuf->buffOffset = pTblBuf->buffUnit;
×
36
  }
37

38
  return TSDB_CODE_SUCCESS;
×
39
}
40

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

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

67
  (void)taosThreadMutexLock(&pStmt->queue.mutex);
×
68
  if (pStmt->queue.head == pStmt->queue.tail) {
×
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;
×
76
  pStmt->queue.head->next = node->next;
×
77
  if (pStmt->queue.tail == node) {
×
78
    pStmt->queue.tail = pStmt->queue.head;
×
79
  }
80
  node->next = NULL;
×
81
  *param = node;
×
82

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

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

88
  return true;
×
89
}
90

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

97
  param->next = NULL;
×
98

99
  (void)taosThreadMutexLock(&pStmt->queue.mutex);
×
100

101
  pStmt->queue.tail->next = param;
×
102
  pStmt->queue.tail = param;
×
103
  pStmt->stat.bindDataNum++;
×
104

105
  (void)atomic_add_fetch_64(&pStmt->queue.qRemainNum, 1);
×
106
  (void)taosThreadCondSignal(&(pStmt->queue.waitCond));
×
107

108
  (void)taosThreadMutexUnlock(&pStmt->queue.mutex);
×
109

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

114
static int32_t stmtCreateRequest(STscStmt2* pStmt) {
×
115
  int32_t code = 0;
×
116

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

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

136
  return code;
×
137
}
138

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

142
  if (newStatus >= STMT_INIT && newStatus < STMT_MAX) {
×
143
    STMT_LOG_SEQ(newStatus);
×
144
  }
145

146
  if (pStmt->errCode && newStatus != STMT_PREPARE) {
×
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) {
×
152
    case STMT_PREPARE:
×
153
      pStmt->errCode = 0;
×
154
      break;
×
155
    case STMT_SETTBNAME:
×
156
      if (STMT_STATUS_EQ(INIT)) {
×
157
        code = TSDB_CODE_TSC_STMT_API_ERROR;
×
158
      }
159
      if (!pStmt->sql.stbInterlaceMode && (STMT_STATUS_EQ(BIND) || STMT_STATUS_EQ(BIND_COL))) {
×
160
        code = TSDB_CODE_TSC_STMT_API_ERROR;
×
161
      }
162
      break;
×
163
    case STMT_SETTAGS:
×
164
      if (STMT_STATUS_EQ(INIT)) {
×
165
        code = TSDB_CODE_TSC_STMT_API_ERROR;
×
166
      }
167
      break;
×
168
    case STMT_FETCH_FIELDS:
×
169
      if (STMT_STATUS_EQ(INIT)) {
×
170
        code = TSDB_CODE_TSC_STMT_API_ERROR;
×
171
      }
172
      break;
×
173
    case STMT_BIND:
×
174
      if (STMT_STATUS_EQ(INIT) || STMT_STATUS_EQ(BIND_COL)) {
×
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;
×
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:
×
189
      if (STMT_STATUS_NE(BIND) && STMT_STATUS_NE(BIND_COL) && STMT_STATUS_NE(FETCH_FIELDS)) {
×
190
        code = TSDB_CODE_TSC_STMT_API_ERROR;
×
191
      }
192
      break;
×
193
    case STMT_EXECUTE:
×
194
      if (STMT_TYPE_QUERY == pStmt->sql.type) {
×
195
        if (STMT_STATUS_NE(ADD_BATCH) && STMT_STATUS_NE(FETCH_FIELDS) && STMT_STATUS_NE(BIND) &&
×
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)) {
×
201
          code = TSDB_CODE_TSC_STMT_API_ERROR;
×
202
        }
203
      }
204
      break;
×
205
    default:
×
206
      code = TSDB_CODE_APP_ERROR;
×
207
      break;
×
208
  }
209

210
  STMT_ERR_RET(code);
×
211

212
  pStmt->sql.status = newStatus;
×
213

214
  return TSDB_CODE_SUCCESS;
×
215
}
216

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

220
  pStmt->sql.type = STMT_TYPE_MULTI_INSERT;
×
221

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

227
  *tbName = pStmt->bInfo.tbName;
×
228

229
  return TSDB_CODE_SUCCESS;
×
230
}
231

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

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

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

250
  if (!pStmt->bInfo.tagsCached) {
×
251
    qDestroyBoundColInfo(pStmt->bInfo.boundTags);
×
252
    taosMemoryFreeClear(pStmt->bInfo.boundTags);
×
253
  }
254

255
  if (cols) {
×
256
    pStmt->bInfo.boundCols =
×
257
        tSimpleHashInit(taosArrayGetSize(cols), taosGetDefaultHashFunction(TSDB_DATA_TYPE_SMALLINT));
×
258
    if (pStmt->bInfo.boundCols) {
×
259
      for (int32_t i = 0; i < taosArrayGetSize(cols); i++) {
×
260
        SColVal* pColVal = taosArrayGet(cols, i);
×
261
        if (pColVal) {
×
262
          tSimpleHashPut(pStmt->bInfo.boundCols, &pColVal->cid, sizeof(int16_t), pColVal, sizeof(SColVal));
×
263
        }
264
      }
265
    }
266
  } else {
267
    pStmt->bInfo.boundCols = NULL;
×
268
  }
269
  pStmt->bInfo.boundTags = tags;
×
270
  pStmt->bInfo.tagsCached = false;
×
271
  pStmt->bInfo.tbNameFlag = tbNameFlag;
×
272
  tstrncpy(pStmt->bInfo.stbFName, sTableName, sizeof(pStmt->bInfo.stbFName));
×
273

274
  return TSDB_CODE_SUCCESS;
×
275
}
276

277
static int32_t stmtUpdateExecInfo(TAOS_STMT2* stmt, SHashObj* pVgHash, SHashObj* pBlockHash) {
×
278
  STscStmt2* pStmt = (STscStmt2*)stmt;
×
279

280
  pStmt->sql.pVgHash = pVgHash;
×
281
  pStmt->exec.pBlockHash = pBlockHash;
×
282

283
  return TSDB_CODE_SUCCESS;
×
284
}
285

286
static int32_t stmtUpdateInfo(TAOS_STMT2* stmt, STableMeta* pTableMeta, void* tags, SArray* cols, SName* tbName,
×
287
                              bool autoCreateTbl, SHashObj* pVgHash, SHashObj* pBlockHash, const char* sTableName,
288
                              uint8_t tbNameFlag) {
289
  STscStmt2* pStmt = (STscStmt2*)stmt;
×
290

291
  STMT_ERR_RET(stmtUpdateBindInfo(stmt, pTableMeta, tags, cols, tbName, sTableName, autoCreateTbl, tbNameFlag));
×
292
  STMT_ERR_RET(stmtUpdateExecInfo(stmt, pVgHash, pBlockHash));
×
293

294
  pStmt->sql.autoCreateTbl = autoCreateTbl;
×
295

296
  return TSDB_CODE_SUCCESS;
×
297
}
298

299
static int32_t stmtGetExecInfo(TAOS_STMT2* stmt, SHashObj** pVgHash, SHashObj** pBlockHash) {
×
300
  STscStmt2* pStmt = (STscStmt2*)stmt;
×
301

302
  *pVgHash = pStmt->sql.pVgHash;
×
303
  pStmt->sql.pVgHash = NULL;
×
304

305
  *pBlockHash = pStmt->exec.pBlockHash;
×
306
  pStmt->exec.pBlockHash = NULL;
×
307

308
  return TSDB_CODE_SUCCESS;
×
309
}
310

311
static int32_t stmtParseSql(STscStmt2* pStmt) {
×
312
  pStmt->exec.pCurrBlock = NULL;
×
313

314
  STMT2_DLOG_E("start to stmtParseSql");
×
315

316
  SStmtCallback stmtCb = {
×
317
      .pStmt = pStmt,
318
      .getTbNameFn = stmtGetTbName,
319
      .setInfoFn = stmtUpdateInfo,
320
      .getExecInfoFn = stmtGetExecInfo,
321
  };
322

323
  STMT_ERR_RET(stmtCreateRequest(pStmt));
×
324
  pStmt->exec.pRequest->isStmtBind = true;
×
325

326
  pStmt->stat.parseSqlNum++;
×
327
  STMT_ERR_RET(parseSql(pStmt->exec.pRequest, false, &pStmt->sql.pQuery, &stmtCb));
×
328

329
  pStmt->sql.siInfo.pQuery = pStmt->sql.pQuery;
×
330

331
  pStmt->bInfo.needParse = false;
×
332

333
  if (pStmt->sql.type == 0) {
×
334
    if (pStmt->sql.pQuery->pRoot && LEGAL_INSERT(nodeType(pStmt->sql.pQuery->pRoot))) {
×
335
      pStmt->sql.type = STMT_TYPE_INSERT;
×
336
      pStmt->sql.stbInterlaceMode = false;
×
337
    } else if (pStmt->sql.pQuery->pPrepareRoot && LEGAL_SELECT(nodeType(pStmt->sql.pQuery->pPrepareRoot))) {
×
338
      pStmt->sql.type = STMT_TYPE_QUERY;
×
339
      pStmt->sql.stbInterlaceMode = false;
×
340

341
      return TSDB_CODE_SUCCESS;
×
342
    } else {
343
      pStmt->bInfo.needParse = true;
×
344
      STMT2_ELOG_E("only support select or insert sql");
×
345
      if (pStmt->exec.pRequest->msgBuf) {
×
346
        tstrncpy(pStmt->exec.pRequest->msgBuf, "stmt only support select or insert", pStmt->exec.pRequest->msgBufLen);
×
347
      }
348
      return TSDB_CODE_TSC_STMT_API_ERROR;
×
349
    }
350
  } else if (pStmt->sql.type == STMT_TYPE_QUERY) {
×
351
    pStmt->sql.stbInterlaceMode = false;
×
352
    return TSDB_CODE_SUCCESS;
×
353
  } else if (pStmt->sql.type == STMT_TYPE_INSERT) {
×
354
    pStmt->sql.stbInterlaceMode = false;
×
355
  }
356

357
  STableDataCxt** pSrc =
358
      (STableDataCxt**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
×
359
  if (NULL == pSrc || NULL == *pSrc) {
×
360
    STMT2_ELOG("fail to get exec.pBlockHash, maybe parse failed, tbFName:%s", pStmt->bInfo.tbFName);
×
361
    STMT_ERR_RET(TSDB_CODE_TSC_STMT_CACHE_ERROR);
×
362
  }
363

364
  STableDataCxt* pTableCtx = *pSrc;
×
365
  if (pStmt->sql.stbInterlaceMode && pTableCtx->pData->pCreateTbReq && (pStmt->bInfo.tbNameFlag & USING_CLAUSE) == 0) {
×
366
    STMT2_TLOG("destroy pCreateTbReq for no-using insert, tbFName:%s", pStmt->bInfo.tbFName);
×
367
    tdDestroySVCreateTbReq(pTableCtx->pData->pCreateTbReq);
×
368
    taosMemoryFreeClear(pTableCtx->pData->pCreateTbReq);
×
369
    pTableCtx->pData->pCreateTbReq = NULL;
×
370
  }
371
  // if (pStmt->sql.stbInterlaceMode) {
372
  //   int16_t lastIdx = -1;
373

374
  //   for (int32_t i = 0; i < pTableCtx->boundColsInfo.numOfBound; ++i) {
375
  //     if (pTableCtx->boundColsInfo.pColIndex[i] < lastIdx) {
376
  //       pStmt->sql.stbInterlaceMode = false;
377
  //       break;
378
  //     }
379

380
  //     lastIdx = pTableCtx->boundColsInfo.pColIndex[i];
381
  //   }
382
  // }
383

384
  if (NULL == pStmt->sql.pBindInfo) {
×
385
    pStmt->sql.pBindInfo = taosMemoryMalloc(pTableCtx->boundColsInfo.numOfBound * sizeof(*pStmt->sql.pBindInfo));
×
386
    if (NULL == pStmt->sql.pBindInfo) {
×
387
      STMT2_ELOG_E("fail to malloc pBindInfo");
×
388
      return terrno;
×
389
    }
390
  }
391

392
  return TSDB_CODE_SUCCESS;
×
393
}
394

395
static int32_t stmtCleanBindInfo(STscStmt2* pStmt) {
×
396
  pStmt->bInfo.tbUid = 0;
×
397
  pStmt->bInfo.tbVgId = -1;
×
398
  pStmt->bInfo.tbType = 0;
×
399
  pStmt->bInfo.needParse = true;
×
400
  pStmt->bInfo.inExecCache = false;
×
401

402
  pStmt->bInfo.tbName[0] = 0;
×
403
  pStmt->bInfo.tbFName[0] = 0;
×
404
  if (!pStmt->bInfo.tagsCached) {
×
405
    qDestroyBoundColInfo(pStmt->bInfo.boundTags);
×
406
    taosMemoryFreeClear(pStmt->bInfo.boundTags);
×
407
  }
408

409
  if (pStmt->bInfo.boundCols) {
×
410
    tSimpleHashCleanup(pStmt->bInfo.boundCols);
×
411
    pStmt->bInfo.boundCols = NULL;
×
412
  }
413

414
  if (!pStmt->sql.autoCreateTbl) {
×
415
    pStmt->bInfo.stbFName[0] = 0;
×
416
    pStmt->bInfo.tbSuid = 0;
×
417
  }
418

419
  STMT2_TLOG("finish clean bind info, tagsCached:%d, autoCreateTbl:%d", pStmt->bInfo.tagsCached,
×
420
             pStmt->sql.autoCreateTbl);
421

422
  return TSDB_CODE_SUCCESS;
×
423
}
424

425
static void stmtFreeTableBlkList(STableColsData* pTb) {
×
426
  (void)qResetStmtColumns(pTb->aCol, true);
×
427
  taosArrayDestroy(pTb->aCol);
×
428
}
×
429

430
static void stmtResetQueueTableBuf(STableBufInfo* pTblBuf, SStmtQueue* pQueue) {
×
431
  pTblBuf->pCurBuff = taosArrayGetP(pTblBuf->pBufList, 0);
×
432
  if (NULL == pTblBuf->pCurBuff) {
×
433
    tscError("QInfo:%p, fail to get buffer from list", pTblBuf);
×
434
    return;
×
435
  }
436
  pTblBuf->buffIdx = 1;
×
437
  pTblBuf->buffOffset = sizeof(*pQueue->head);
×
438

439
  pQueue->head = pQueue->tail = pTblBuf->pCurBuff;
×
440
  pQueue->qRemainNum = 0;
×
441
  pQueue->head->next = NULL;
×
442
}
443

444
static int32_t stmtCleanExecInfo(STscStmt2* pStmt, bool keepTable, bool deepClean) {
×
445
  if (pStmt->sql.stbInterlaceMode) {
×
446
    if (deepClean) {
×
447
      taosHashCleanup(pStmt->exec.pBlockHash);
×
448
      pStmt->exec.pBlockHash = NULL;
×
449

450
      if (NULL != pStmt->exec.pCurrBlock) {
×
451
        taosMemoryFreeClear(pStmt->exec.pCurrBlock->boundColsInfo.pColIndex);
×
452
        taosMemoryFreeClear(pStmt->exec.pCurrBlock->pData);
×
453
        qDestroyStmtDataBlock(pStmt->exec.pCurrBlock);
×
454
        pStmt->exec.pCurrBlock = NULL;
×
455
      }
456
      if (STMT_TYPE_QUERY != pStmt->sql.type) {
×
457
        taos_free_result(pStmt->exec.pRequest);
×
458
        pStmt->exec.pRequest = NULL;
×
459
      }
460
    } else {
461
      pStmt->sql.siInfo.pTableColsIdx = 0;
×
462
      stmtResetQueueTableBuf(&pStmt->sql.siInfo.tbBuf, &pStmt->queue);
×
463
      tSimpleHashClear(pStmt->sql.siInfo.pTableRowDataHash);
×
464
    }
465
    if (NULL != pStmt->exec.pRequest) {
×
466
      pStmt->exec.pRequest->body.resInfo.numOfRows = 0;
×
467
    }
468
  } else {
469
    if (STMT_TYPE_QUERY != pStmt->sql.type || deepClean) {
×
470
      // if (!pStmt->options.asyncExecFn) {
471
      taos_free_result(pStmt->exec.pRequest);
×
472
      pStmt->exec.pRequest = NULL;
×
473
      //}
474
    }
475

476
    size_t keyLen = 0;
×
477
    void*  pIter = taosHashIterate(pStmt->exec.pBlockHash, NULL);
×
478
    while (pIter) {
×
479
      STableDataCxt* pBlocks = *(STableDataCxt**)pIter;
×
480
      char*          key = taosHashGetKey(pIter, &keyLen);
×
481
      STableMeta*    pMeta = qGetTableMetaInDataBlock(pBlocks);
×
482

483
      if (keepTable && pBlocks == pStmt->exec.pCurrBlock) {
×
484
        TSWAP(pBlocks->pData, pStmt->exec.pCurrTbData);
×
485
        STMT_ERR_RET(qResetStmtDataBlock(pBlocks, false));
×
486

487
        pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
×
488
        continue;
×
489
      }
490

491
      qDestroyStmtDataBlock(pBlocks);
×
492
      STMT_ERR_RET(taosHashRemove(pStmt->exec.pBlockHash, key, keyLen));
×
493

494
      pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
×
495
    }
496

497
    if (keepTable) {
×
498
      STMT2_TLOG("finish clean exec info, stbInterlaceMode:%d, keepTable:%d, deepClean:%d", pStmt->sql.stbInterlaceMode,
×
499
                 keepTable, deepClean);
500
      return TSDB_CODE_SUCCESS;
×
501
    }
502

503
    taosHashCleanup(pStmt->exec.pBlockHash);
×
504
    pStmt->exec.pBlockHash = NULL;
×
505

506
    tDestroySubmitTbData(pStmt->exec.pCurrTbData, TSDB_MSG_FLG_ENCODE);
×
507
    taosMemoryFreeClear(pStmt->exec.pCurrTbData);
×
508
  }
509

510
  STMT_ERR_RET(stmtCleanBindInfo(pStmt));
×
511
  STMT2_TLOG("finish clean exec info, stbInterlaceMode:%d, keepTable:%d, deepClean:%d", pStmt->sql.stbInterlaceMode,
×
512
             keepTable, deepClean);
513

514
  return TSDB_CODE_SUCCESS;
×
515
}
516

517
static void stmtFreeTbBuf(void* buf) {
×
518
  void* pBuf = *(void**)buf;
×
519
  taosMemoryFree(pBuf);
×
520
}
×
521

522
static void stmtFreeTbCols(void* buf) {
×
523
  SArray* pCols = *(SArray**)buf;
×
524
  taosArrayDestroy(pCols);
×
525
}
×
526

527
static int32_t stmtCleanSQLInfo(STscStmt2* pStmt) {
×
528
  STMT2_TLOG_E("start to free SQL info");
×
529

530
  taosMemoryFree(pStmt->sql.pBindInfo);
×
531
  taosMemoryFree(pStmt->sql.queryRes.fields);
×
532
  taosMemoryFree(pStmt->sql.queryRes.userFields);
×
533
  taosMemoryFree(pStmt->sql.sqlStr);
×
534
  qDestroyQuery(pStmt->sql.pQuery);
×
535
  taosArrayDestroy(pStmt->sql.nodeList);
×
536
  taosHashCleanup(pStmt->sql.pVgHash);
×
537
  pStmt->sql.pVgHash = NULL;
×
538
  if (pStmt->sql.fixValueTags) {
×
539
    tdDestroySVCreateTbReq(pStmt->sql.fixValueTbReq);
×
540
  }
541

542
  void* pIter = taosHashIterate(pStmt->sql.pTableCache, NULL);
×
543
  while (pIter) {
×
544
    SStmtTableCache* pCache = (SStmtTableCache*)pIter;
×
545

546
    qDestroyStmtDataBlock(pCache->pDataCtx);
×
547
    qDestroyBoundColInfo(pCache->boundTags);
×
548
    taosMemoryFreeClear(pCache->boundTags);
×
549

550
    pIter = taosHashIterate(pStmt->sql.pTableCache, pIter);
×
551
  }
552
  taosHashCleanup(pStmt->sql.pTableCache);
×
553
  pStmt->sql.pTableCache = NULL;
×
554

555
  STMT_ERR_RET(stmtCleanExecInfo(pStmt, false, true));
×
556
  STMT_ERR_RET(stmtCleanBindInfo(pStmt));
×
557

558
  taos_free_result(pStmt->sql.siInfo.pRequest);
×
559
  taosHashCleanup(pStmt->sql.siInfo.pVgroupHash);
×
560
  tSimpleHashCleanup(pStmt->sql.siInfo.pTableHash);
×
561
  tSimpleHashCleanup(pStmt->sql.siInfo.pTableRowDataHash);
×
562
  taosArrayDestroyEx(pStmt->sql.siInfo.tbBuf.pBufList, stmtFreeTbBuf);
×
563
  taosMemoryFree(pStmt->sql.siInfo.pTSchema);
×
564
  qDestroyStmtDataBlock(pStmt->sql.siInfo.pDataCtx);
×
565
  taosArrayDestroyEx(pStmt->sql.siInfo.pTableCols, stmtFreeTbCols);
×
566
  pStmt->sql.siInfo.pTableCols = NULL;
×
567

568
  (void)memset(&pStmt->sql, 0, sizeof(pStmt->sql));
×
569
  pStmt->sql.siInfo.tableColsReady = true;
×
570

571
  STMT2_TLOG_E("end to free SQL info");
×
572

573
  return TSDB_CODE_SUCCESS;
×
574
}
575

576
static int32_t stmtTryAddTableVgroupInfo(STscStmt2* pStmt, int32_t* vgId) {
×
577
  if (*vgId >= 0 && taosHashGet(pStmt->sql.pVgHash, (const char*)vgId, sizeof(*vgId))) {
×
578
    return TSDB_CODE_SUCCESS;
×
579
  }
580

581
  SVgroupInfo      vgInfo = {0};
×
582
  SRequestConnInfo conn = {.pTrans = pStmt->taos->pAppInfo->pTransporter,
×
583
                           .requestId = pStmt->exec.pRequest->requestId,
×
584
                           .requestObjRefId = pStmt->exec.pRequest->self,
×
585
                           .mgmtEps = getEpSet_s(&pStmt->taos->pAppInfo->mgmtEp)};
×
586

587
  int32_t code = catalogGetTableHashVgroup(pStmt->pCatalog, &conn, &pStmt->bInfo.sname, &vgInfo);
×
588
  if (TSDB_CODE_SUCCESS != code) {
×
589
    STMT2_ELOG("fail to get vgroup info from catalog, code:%d", code);
×
590
    return code;
×
591
  }
592

593
  code =
594
      taosHashPut(pStmt->sql.pVgHash, (const char*)&vgInfo.vgId, sizeof(vgInfo.vgId), (char*)&vgInfo, sizeof(vgInfo));
×
595
  if (TSDB_CODE_SUCCESS != code) {
×
596
    STMT2_ELOG("fail to put vgroup info, code:%d", code);
×
597
    return code;
×
598
  }
599

600
  *vgId = vgInfo.vgId;
×
601

602
  return TSDB_CODE_SUCCESS;
×
603
}
604

605
static int32_t stmtRebuildDataBlock(STscStmt2* pStmt, STableDataCxt* pDataBlock, STableDataCxt** newBlock, uint64_t uid,
×
606
                                    uint64_t suid, int32_t vgId) {
607
  STMT_ERR_RET(stmtTryAddTableVgroupInfo(pStmt, &vgId));
×
608
  STMT_ERR_RET(qRebuildStmtDataBlock(newBlock, pDataBlock, uid, suid, vgId, pStmt->sql.autoCreateTbl));
×
609

610
  STMT2_DLOG("uid:%" PRId64 ", rebuild table data context, vgId:%d", uid, vgId);
×
611

612
  return TSDB_CODE_SUCCESS;
×
613
}
614

615
static int32_t stmtGetFromCache(STscStmt2* pStmt) {
×
616
  if (pStmt->sql.stbInterlaceMode && pStmt->sql.siInfo.pDataCtx) {
×
617
    pStmt->bInfo.needParse = false;
×
618
    pStmt->bInfo.inExecCache = false;
×
619
    return TSDB_CODE_SUCCESS;
×
620
  }
621

622
  pStmt->bInfo.needParse = true;
×
623
  pStmt->bInfo.inExecCache = false;
×
624

625
  STableDataCxt** pCxtInExec = taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
×
626
  if (pCxtInExec) {
×
627
    pStmt->bInfo.needParse = false;
×
628
    pStmt->bInfo.inExecCache = true;
×
629

630
    pStmt->exec.pCurrBlock = *pCxtInExec;
×
631

632
    if (pStmt->sql.autoCreateTbl) {
×
633
      STMT2_DLOG("reuse stmt block for tb %s in execBlock", pStmt->bInfo.tbFName);
×
634
      return TSDB_CODE_SUCCESS;
×
635
    }
636
  }
637

638
  if (NULL == pStmt->pCatalog) {
×
639
    STMT_ERR_RET(catalogGetHandle(pStmt->taos->pAppInfo->clusterId, &pStmt->pCatalog));
×
640
    pStmt->sql.siInfo.pCatalog = pStmt->pCatalog;
×
641
  }
642

643
  if (NULL == pStmt->sql.pTableCache || taosHashGetSize(pStmt->sql.pTableCache) <= 0) {
×
644
    if (pStmt->bInfo.inExecCache) {
×
645
      pStmt->bInfo.needParse = false;
×
646
      STMT2_DLOG("reuse stmt block for tb %s in execBlock", pStmt->bInfo.tbFName);
×
647
      return TSDB_CODE_SUCCESS;
×
648
    }
649

650
    STMT2_DLOG("no stmt block cache for tb %s", pStmt->bInfo.tbFName);
×
651
    return TSDB_CODE_SUCCESS;
×
652
  }
653

654
  if (pStmt->sql.autoCreateTbl) {
×
655
    SStmtTableCache* pCache = taosHashGet(pStmt->sql.pTableCache, &pStmt->bInfo.tbSuid, sizeof(pStmt->bInfo.tbSuid));
×
656
    if (pCache) {
×
657
      pStmt->bInfo.needParse = false;
×
658
      pStmt->bInfo.tbUid = 0;
×
659

660
      STableDataCxt* pNewBlock = NULL;
×
661
      STMT_ERR_RET(stmtRebuildDataBlock(pStmt, pCache->pDataCtx, &pNewBlock, 0, pStmt->bInfo.tbSuid, -1));
×
662

663
      if (taosHashPut(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName), &pNewBlock,
×
664
                      POINTER_BYTES)) {
665
        STMT_ERR_RET(terrno);
×
666
      }
667

668
      pStmt->exec.pCurrBlock = pNewBlock;
×
669

670
      STMT2_DLOG("reuse stmt block for tb %s in sqlBlock, suid:0x%" PRIx64, pStmt->bInfo.tbFName, pStmt->bInfo.tbSuid);
×
671

672
      return TSDB_CODE_SUCCESS;
×
673
    }
674

675
    STMT_RET(stmtCleanBindInfo(pStmt));
×
676
  }
677

678
  uint64_t uid, suid;
679
  int32_t  vgId;
680
  int8_t   tableType;
681

682
  STableMeta*      pTableMeta = NULL;
×
683
  SRequestConnInfo conn = {.pTrans = pStmt->taos->pAppInfo->pTransporter,
×
684
                           .requestId = pStmt->exec.pRequest->requestId,
×
685
                           .requestObjRefId = pStmt->exec.pRequest->self,
×
686
                           .mgmtEps = getEpSet_s(&pStmt->taos->pAppInfo->mgmtEp)};
×
687
  int32_t          code = catalogGetTableMeta(pStmt->pCatalog, &conn, &pStmt->bInfo.sname, &pTableMeta);
×
688

689
  pStmt->stat.ctgGetTbMetaNum++;
×
690

691
  if (TSDB_CODE_PAR_TABLE_NOT_EXIST == code) {
×
692
    STMT2_DLOG("tb %s not exist", pStmt->bInfo.tbFName);
×
693
    STMT_ERR_RET(stmtCleanBindInfo(pStmt));
×
694

695
    STMT_ERR_RET(code);
×
696
  }
697

698
  STMT_ERR_RET(code);
×
699

700
  uid = pTableMeta->uid;
×
701
  suid = pTableMeta->suid;
×
702
  tableType = pTableMeta->tableType;
×
703
  pStmt->bInfo.tbVgId = pTableMeta->vgId;
×
704
  vgId = pTableMeta->vgId;
×
705

706
  taosMemoryFree(pTableMeta);
×
707

708
  uint64_t cacheUid = (TSDB_CHILD_TABLE == tableType) ? suid : uid;
×
709

710
  if (uid == pStmt->bInfo.tbUid) {
×
711
    pStmt->bInfo.needParse = false;
×
712

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

715
    return TSDB_CODE_SUCCESS;
×
716
  }
717

718
  if (pStmt->bInfo.inExecCache) {
×
719
    SStmtTableCache* pCache = taosHashGet(pStmt->sql.pTableCache, &cacheUid, sizeof(cacheUid));
×
720
    if (NULL == pCache) {
×
721
      STMT2_ELOG("table [%s, %" PRIx64 ", %" PRIx64 "] found in exec blockHash, but not in sql blockHash",
×
722
                 pStmt->bInfo.tbFName, uid, cacheUid);
723

724
      STMT_ERR_RET(TSDB_CODE_APP_ERROR);
×
725
    }
726

727
    pStmt->bInfo.needParse = false;
×
728

729
    pStmt->bInfo.tbUid = uid;
×
730
    pStmt->bInfo.tbSuid = suid;
×
731
    pStmt->bInfo.tbType = tableType;
×
732
    pStmt->bInfo.boundTags = pCache->boundTags;
×
733
    pStmt->bInfo.tagsCached = true;
×
734

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

737
    return TSDB_CODE_SUCCESS;
×
738
  }
739

740
  SStmtTableCache* pCache = taosHashGet(pStmt->sql.pTableCache, &cacheUid, sizeof(cacheUid));
×
741
  if (pCache) {
×
742
    pStmt->bInfo.needParse = false;
×
743

744
    pStmt->bInfo.tbUid = uid;
×
745
    pStmt->bInfo.tbSuid = suid;
×
746
    pStmt->bInfo.tbType = tableType;
×
747
    pStmt->bInfo.boundTags = pCache->boundTags;
×
748
    pStmt->bInfo.tagsCached = true;
×
749

750
    STableDataCxt* pNewBlock = NULL;
×
751
    STMT_ERR_RET(stmtRebuildDataBlock(pStmt, pCache->pDataCtx, &pNewBlock, uid, suid, vgId));
×
752

753
    if (taosHashPut(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName), &pNewBlock,
×
754
                    POINTER_BYTES)) {
755
      STMT_ERR_RET(terrno);
×
756
    }
757

758
    pStmt->exec.pCurrBlock = pNewBlock;
×
759

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

762
    return TSDB_CODE_SUCCESS;
×
763
  }
764

765
  STMT_ERR_RET(stmtCleanBindInfo(pStmt));
×
766

767
  return TSDB_CODE_SUCCESS;
×
768
}
769

770
static int32_t stmtResetStmt(STscStmt2* pStmt) {
×
771
  STMT_ERR_RET(stmtCleanSQLInfo(pStmt));
×
772

773
  pStmt->sql.pTableCache = taosHashInit(100, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
×
774
  if (NULL == pStmt->sql.pTableCache) {
×
775
    STMT2_ELOG("fail to allocate memory for pTableCache in stmtResetStmt:%s", tstrerror(terrno));
×
776
    STMT_ERR_RET(terrno);
×
777
  }
778

779
  pStmt->sql.status = STMT_INIT;
×
780

781
  return TSDB_CODE_SUCCESS;
×
782
}
783

784
static int32_t stmtAsyncOutput(STscStmt2* pStmt, void* param) {
×
785
  SStmtQNode* pParam = (SStmtQNode*)param;
×
786

787
  if (pParam->restoreTbCols) {
×
788
    for (int32_t i = 0; i < pStmt->sql.siInfo.pTableColsIdx; ++i) {
×
789
      SArray** p = (SArray**)TARRAY_GET_ELEM(pStmt->sql.siInfo.pTableCols, i);
×
790
      *p = taosArrayInit(20, POINTER_BYTES);
×
791
      if (*p == NULL) {
×
792
        STMT_ERR_RET(terrno);
×
793
      }
794
    }
795
    atomic_store_8((int8_t*)&pStmt->sql.siInfo.tableColsReady, true);
×
796
    STMT2_TLOG_E("restore pTableCols finished");
×
797
  } else {
798
    int code = qAppendStmtTableOutput(pStmt->sql.pQuery, pStmt->sql.pVgHash, &pParam->tblData, pStmt->exec.pCurrBlock,
×
799
                                      &pStmt->sql.siInfo, pParam->pCreateTbReq);
800
    // taosMemoryFree(pParam->pTbData);
801
    (void)atomic_sub_fetch_64(&pStmt->sql.siInfo.tbRemainNum, 1);
×
802
    if (code != TSDB_CODE_SUCCESS) {
×
803
      STMT2_ELOG("async append stmt output failed, tbname:%s, err:%s", pParam->tblData.tbName, tstrerror(code));
×
804
      STMT_ERR_RET(code);
×
805
    }
806
  }
807
  return TSDB_CODE_SUCCESS;
×
808
}
809

810
static void* stmtBindThreadFunc(void* param) {
×
811
  setThreadName("stmt2Bind");
×
812

813
  STscStmt2* pStmt = (STscStmt2*)param;
×
814
  STMT2_ILOG_E("stmt2 bind thread started");
×
815

816
  while (true) {
×
817
    SStmtQNode* asyncParam = NULL;
×
818

819
    if (!stmtDequeue(pStmt, &asyncParam)) {
×
820
      if (pStmt->queue.stopQueue && 0 == atomic_load_64((int64_t*)&pStmt->queue.qRemainNum)) {
×
821
        STMT2_DLOG_E("queue is empty and stopQueue is set, thread will exit");
×
822
        break;
×
823
      }
824
      continue;
×
825
    }
826

827
    int ret = stmtAsyncOutput(pStmt, asyncParam);
×
828
    if (ret != 0) {
×
829
      STMT2_ELOG("stmtAsyncOutput failed, reason:%s", tstrerror(ret));
×
830
    }
831
  }
832

833
  STMT2_ILOG_E("stmt2 bind thread stopped");
×
834
  return NULL;
×
835
}
836

837
static int32_t stmtStartBindThread(STscStmt2* pStmt) {
×
838
  TdThreadAttr thAttr;
839
  if (taosThreadAttrInit(&thAttr) != 0) {
×
840
    return TSDB_CODE_TSC_INTERNAL_ERROR;
×
841
  }
842
  if (taosThreadAttrSetDetachState(&thAttr, PTHREAD_CREATE_JOINABLE) != 0) {
×
843
    return TSDB_CODE_TSC_INTERNAL_ERROR;
×
844
  }
845

846
  if (taosThreadCreate(&pStmt->bindThread, &thAttr, stmtBindThreadFunc, pStmt) != 0) {
×
847
    terrno = TAOS_SYSTEM_ERROR(ERRNO);
×
848
    STMT_ERR_RET(terrno);
×
849
  }
850

851
  pStmt->bindThreadInUse = true;
×
852

853
  (void)taosThreadAttrDestroy(&thAttr);
×
854
  return TSDB_CODE_SUCCESS;
×
855
}
856

857
static int32_t stmtInitQueue(STscStmt2* pStmt) {
×
858
  (void)taosThreadCondInit(&pStmt->queue.waitCond, NULL);
×
859
  (void)taosThreadMutexInit(&pStmt->queue.mutex, NULL);
×
860
  STMT_ERR_RET(stmtAllocQNodeFromBuf(&pStmt->sql.siInfo.tbBuf, (void**)&pStmt->queue.head));
×
861
  pStmt->queue.tail = pStmt->queue.head;
×
862

863
  return TSDB_CODE_SUCCESS;
×
864
}
865

866
static int32_t stmtIniAsyncBind(STscStmt2* pStmt) {
×
867
  (void)taosThreadCondInit(&pStmt->asyncBindParam.waitCond, NULL);
×
868
  (void)taosThreadMutexInit(&pStmt->asyncBindParam.mutex, NULL);
×
869
  pStmt->asyncBindParam.asyncBindNum = 0;
×
870

871
  return TSDB_CODE_SUCCESS;
×
872
}
873

874
static int32_t stmtInitTableBuf(STableBufInfo* pTblBuf) {
×
875
  pTblBuf->buffUnit = sizeof(SStmtQNode);
×
876
  pTblBuf->buffSize = pTblBuf->buffUnit * 1000;
×
877
  pTblBuf->pBufList = taosArrayInit(100, POINTER_BYTES);
×
878
  if (NULL == pTblBuf->pBufList) {
×
879
    return terrno;
×
880
  }
881
  void* buff = taosMemoryMalloc(pTblBuf->buffSize);
×
882
  if (NULL == buff) {
×
883
    return terrno;
×
884
  }
885

886
  if (taosArrayPush(pTblBuf->pBufList, &buff) == NULL) {
×
887
    return terrno;
×
888
  }
889

890
  pTblBuf->pCurBuff = buff;
×
891
  pTblBuf->buffIdx = 1;
×
892
  pTblBuf->buffOffset = 0;
×
893

894
  return TSDB_CODE_SUCCESS;
×
895
}
896

897
TAOS_STMT2* stmtInit2(STscObj* taos, TAOS_STMT2_OPTION* pOptions) {
×
898
  STscObj*   pObj = (STscObj*)taos;
×
899
  STscStmt2* pStmt = NULL;
×
900
  int32_t    code = 0;
×
901

902
  pStmt = taosMemoryCalloc(1, sizeof(STscStmt2));
×
903
  if (NULL == pStmt) {
×
904
    STMT2_ELOG_E("fail to allocate memory for pStmt");
×
905
    return NULL;
×
906
  }
907

908
  pStmt->sql.pTableCache = taosHashInit(100, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
×
909
  if (NULL == pStmt->sql.pTableCache) {
×
910
    STMT2_ELOG("fail to allocate memory for pTableCache in stmtInit2:%s", tstrerror(terrno));
×
911
    taosMemoryFree(pStmt);
×
912
    return NULL;
×
913
  }
914

915
  pStmt->taos = pObj;
×
916
  if (taos->db[0] != '\0') {
×
917
    pStmt->db = taosStrdup(taos->db);
×
918
  }
919
  pStmt->bInfo.needParse = true;
×
920
  pStmt->sql.status = STMT_INIT;
×
921
  pStmt->errCode = TSDB_CODE_SUCCESS;
×
922

923
  if (NULL != pOptions) {
×
924
    (void)memcpy(&pStmt->options, pOptions, sizeof(pStmt->options));
×
925
    if (pOptions->singleStbInsert && pOptions->singleTableBindOnce) {
×
926
      pStmt->stbInterlaceMode = true;
×
927
    }
928

929
    pStmt->reqid = pOptions->reqid;
×
930
  }
931

932
  if (pStmt->stbInterlaceMode) {
×
933
    pStmt->sql.siInfo.transport = taos->pAppInfo->pTransporter;
×
934
    pStmt->sql.siInfo.acctId = taos->acctId;
×
935
    pStmt->sql.siInfo.dbname = taos->db;
×
936
    pStmt->sql.siInfo.mgmtEpSet = getEpSet_s(&pStmt->taos->pAppInfo->mgmtEp);
×
937

938
    pStmt->sql.siInfo.pTableHash = tSimpleHashInit(100, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY));
×
939
    if (NULL == pStmt->sql.siInfo.pTableHash) {
×
940
      STMT2_ELOG("fail to allocate memory for pTableHash:%s", tstrerror(terrno));
×
941
      (void)stmtClose2(pStmt);
×
942
      return NULL;
×
943
    }
944

945
    pStmt->sql.siInfo.pTableRowDataHash = tSimpleHashInit(100, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY));
×
946
    if (NULL == pStmt->sql.siInfo.pTableRowDataHash) {
×
947
      STMT2_ELOG("fail to allocate memory for pTableRowDataHash:%s", tstrerror(terrno));
×
948
      (void)stmtClose2(pStmt);
×
949
      return NULL;
×
950
    }
951

952
    pStmt->sql.siInfo.pTableCols = taosArrayInit(STMT_TABLE_COLS_NUM, POINTER_BYTES);
×
953
    if (NULL == pStmt->sql.siInfo.pTableCols) {
×
954
      STMT2_ELOG("fail to allocate memory for pTableCols:%s", tstrerror(terrno));
×
955
      (void)stmtClose2(pStmt);
×
956
      return NULL;
×
957
    }
958

959
    code = stmtInitTableBuf(&pStmt->sql.siInfo.tbBuf);
×
960
    if (TSDB_CODE_SUCCESS == code) {
×
961
      code = stmtInitQueue(pStmt);
×
962
    }
963
    if (TSDB_CODE_SUCCESS == code) {
×
964
      code = stmtStartBindThread(pStmt);
×
965
    }
966
    if (TSDB_CODE_SUCCESS != code) {
×
967
      terrno = code;
×
968
      STMT2_ELOG("fail to init stmt2 bind thread:%s", tstrerror(code));
×
969
      (void)stmtClose2(pStmt);
×
970
      return NULL;
×
971
    }
972
  }
973

974
  pStmt->sql.siInfo.tableColsReady = true;
×
975
  if (pStmt->options.asyncExecFn) {
×
976
    if (tsem_init(&pStmt->asyncExecSem, 0, 1) != 0) {
×
977
      terrno = TAOS_SYSTEM_ERROR(ERRNO);
×
978
      STMT2_ELOG("fail to init asyncExecSem:%s", tstrerror(terrno));
×
979
      (void)stmtClose2(pStmt);
×
980
      return NULL;
×
981
    }
982
  }
983
  code = stmtIniAsyncBind(pStmt);
×
984
  if (TSDB_CODE_SUCCESS != code) {
×
985
    terrno = code;
×
986
    STMT2_ELOG("fail to start init asyncExecSem:%s", tstrerror(code));
×
987

988
    (void)stmtClose2(pStmt);
×
989
    return NULL;
×
990
  }
991

992
  pStmt->execSemWaited = false;
×
993

994
  // STMT_LOG_SEQ(STMT_INIT);
995

996
  STMT2_DLOG("stmt2 initialize finished, seqId:%d, db:%s, interlaceMode:%d, asyncExec:%d", pStmt->seqId, pStmt->db,
×
997
             pStmt->stbInterlaceMode, pStmt->options.asyncExecFn != NULL);
998

999
  return pStmt;
×
1000
}
1001

1002
static int stmtSetDbName2(TAOS_STMT2* stmt, const char* dbName) {
×
1003
  STscStmt2* pStmt = (STscStmt2*)stmt;
×
1004
  if (dbName == NULL || dbName[0] == '\0') {
×
1005
    STMT2_ELOG_E("dbname in sql is illegal");
×
1006
    return TSDB_CODE_TSC_SQL_SYNTAX_ERROR;
×
1007
  }
1008

1009
  STMT2_DLOG("dbname is specified in sql:%s", dbName);
×
1010
  if (pStmt->db == NULL || pStmt->db[0] == '\0') {
×
1011
    taosMemoryFreeClear(pStmt->db);
×
1012
    STMT2_DLOG("dbname:%s is by sql, not by taosconnect", dbName);
×
1013
    pStmt->db = taosStrdup(dbName);
×
1014
    (void)strdequote(pStmt->db);
×
1015
  }
1016
  STMT_ERR_RET(stmtCreateRequest(pStmt));
×
1017

1018
  // The SQL statement specifies a database name, overriding the previously specified database
1019
  taosMemoryFreeClear(pStmt->exec.pRequest->pDb);
×
1020
  pStmt->exec.pRequest->pDb = taosStrdup(dbName);
×
1021
  (void)strdequote(pStmt->exec.pRequest->pDb);
×
1022
  if (pStmt->exec.pRequest->pDb == NULL) {
×
1023
    return terrno;
×
1024
  }
1025
  if (pStmt->sql.stbInterlaceMode) {
×
1026
    pStmt->sql.siInfo.dbname = pStmt->exec.pRequest->pDb;
×
1027
  }
1028
  return TSDB_CODE_SUCCESS;
×
1029
}
1030
static int32_t stmtResetStbInterlaceCache(STscStmt2* pStmt) {
×
1031
  int32_t code = TSDB_CODE_SUCCESS;
×
1032

1033
  if (pStmt->bindThreadInUse) {
×
1034
    while (0 == atomic_load_8((int8_t*)&pStmt->sql.siInfo.tableColsReady)) {
×
1035
      taosUsleep(1);
×
1036
    }
1037
    (void)taosThreadMutexLock(&pStmt->queue.mutex);
×
1038
    pStmt->queue.stopQueue = true;
×
1039
    (void)taosThreadCondSignal(&(pStmt->queue.waitCond));
×
1040
    (void)taosThreadMutexUnlock(&pStmt->queue.mutex);
×
1041

1042
    (void)taosThreadJoin(pStmt->bindThread, NULL);
×
1043
    pStmt->bindThreadInUse = false;
×
1044
    pStmt->queue.head = NULL;
×
1045
    pStmt->queue.tail = NULL;
×
1046
    pStmt->queue.qRemainNum = 0;
×
1047

1048
    (void)taosThreadCondDestroy(&pStmt->queue.waitCond);
×
1049
    (void)taosThreadMutexDestroy(&pStmt->queue.mutex);
×
1050
  }
1051

1052
  pStmt->sql.siInfo.pTableHash = tSimpleHashInit(100, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY));
×
1053
  if (NULL == pStmt->sql.siInfo.pTableHash) {
×
1054
    return terrno;
×
1055
  }
1056

1057
  pStmt->sql.siInfo.pTableRowDataHash = tSimpleHashInit(100, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY));
×
1058
  if (NULL == pStmt->sql.siInfo.pTableRowDataHash) {
×
1059
    return terrno;
×
1060
  }
1061

1062
  pStmt->sql.siInfo.pTableCols = taosArrayInit(STMT_TABLE_COLS_NUM, POINTER_BYTES);
×
1063
  if (NULL == pStmt->sql.siInfo.pTableCols) {
×
1064
    return terrno;
×
1065
  }
1066

1067
  code = stmtInitTableBuf(&pStmt->sql.siInfo.tbBuf);
×
1068

1069
  if (TSDB_CODE_SUCCESS == code) {
×
1070
    code = stmtInitQueue(pStmt);
×
1071
    pStmt->queue.stopQueue = false;
×
1072
  }
1073
  if (TSDB_CODE_SUCCESS == code) {
×
1074
    code = stmtStartBindThread(pStmt);
×
1075
  }
1076
  if (TSDB_CODE_SUCCESS != code) {
×
1077
    return code;
×
1078
  }
1079

1080
  return TSDB_CODE_SUCCESS;
×
1081
}
1082

1083
static int32_t stmtDeepReset(STscStmt2* pStmt) {
×
1084
  char*             db = pStmt->db;
×
1085
  bool              stbInterlaceMode = pStmt->stbInterlaceMode;
×
1086
  TAOS_STMT2_OPTION options = pStmt->options;
×
1087
  uint32_t          reqid = pStmt->reqid;
×
1088

1089
  if (pStmt->options.asyncExecFn && !pStmt->execSemWaited) {
×
1090
    if (tsem_wait(&pStmt->asyncExecSem) != 0) {
×
1091
      STMT2_ELOG_E("bind param wait asyncExecSem failed");
×
1092
    }
1093
    pStmt->execSemWaited = true;
×
1094
  }
1095
  pStmt->sql.autoCreateTbl = false;
×
1096
  taosMemoryFree(pStmt->sql.pBindInfo);
×
1097
  pStmt->sql.pBindInfo = NULL;
×
1098

1099
  taosMemoryFree(pStmt->sql.queryRes.fields);
×
1100
  pStmt->sql.queryRes.fields = NULL;
×
1101

1102
  taosMemoryFree(pStmt->sql.queryRes.userFields);
×
1103
  pStmt->sql.queryRes.userFields = NULL;
×
1104

1105
  pStmt->sql.type = 0;
×
1106
  pStmt->sql.runTimes = 0;
×
1107
  taosMemoryFree(pStmt->sql.sqlStr);
×
1108
  pStmt->sql.sqlStr = NULL;
×
1109

1110
  qDestroyQuery(pStmt->sql.pQuery);
×
1111
  pStmt->sql.pQuery = NULL;
×
1112

1113
  taosArrayDestroy(pStmt->sql.nodeList);
×
1114
  pStmt->sql.nodeList = NULL;
×
1115

1116
  taosHashCleanup(pStmt->sql.pVgHash);
×
1117
  pStmt->sql.pVgHash = NULL;
×
1118

1119
  if (pStmt->sql.fixValueTags) {
×
1120
    tdDestroySVCreateTbReq(pStmt->sql.fixValueTbReq);
×
1121
    pStmt->sql.fixValueTbReq = NULL;
×
1122
  }
1123
  pStmt->sql.fixValueTags = false;
×
1124

1125
  void* pIter = taosHashIterate(pStmt->sql.pTableCache, NULL);
×
1126
  while (pIter) {
×
1127
    SStmtTableCache* pCache = (SStmtTableCache*)pIter;
×
1128

1129
    qDestroyStmtDataBlock(pCache->pDataCtx);
×
1130
    qDestroyBoundColInfo(pCache->boundTags);
×
1131
    taosMemoryFreeClear(pCache->boundTags);
×
1132

1133
    pIter = taosHashIterate(pStmt->sql.pTableCache, pIter);
×
1134
  }
1135
  taosHashCleanup(pStmt->sql.pTableCache);
×
1136

1137
  if (pStmt->sql.stbInterlaceMode) {
×
1138
    pStmt->bInfo.tagsCached = false;
×
1139
  }
1140
  STMT_ERR_RET(stmtCleanExecInfo(pStmt, false, true));
×
1141

1142
  if (pStmt->exec.pRequest) {
×
1143
    taos_free_result(pStmt->exec.pRequest);
×
1144
    pStmt->exec.pRequest = NULL;
×
1145
  }
1146

1147
  if (pStmt->sql.siInfo.pTableCols) {
×
1148
    taosArrayDestroyEx(pStmt->sql.siInfo.pTableCols, stmtFreeTbCols);
×
1149
    pStmt->sql.siInfo.pTableCols = NULL;
×
1150
  }
1151

1152
  if (pStmt->sql.siInfo.tbBuf.pBufList) {
×
1153
    taosArrayDestroyEx(pStmt->sql.siInfo.tbBuf.pBufList, stmtFreeTbBuf);
×
1154
    pStmt->sql.siInfo.tbBuf.pBufList = NULL;
×
1155
  }
1156

1157
  if (pStmt->sql.siInfo.pTableHash) {
×
1158
    tSimpleHashCleanup(pStmt->sql.siInfo.pTableHash);
×
1159
    pStmt->sql.siInfo.pTableHash = NULL;
×
1160
  }
1161

1162
  if (pStmt->sql.siInfo.pTableRowDataHash) {
×
1163
    tSimpleHashCleanup(pStmt->sql.siInfo.pTableRowDataHash);
×
1164
    pStmt->sql.siInfo.pTableRowDataHash = NULL;
×
1165
  }
1166

1167
  if (pStmt->sql.siInfo.pVgroupHash) {
×
1168
    taosHashCleanup(pStmt->sql.siInfo.pVgroupHash);
×
1169
    pStmt->sql.siInfo.pVgroupHash = NULL;
×
1170
  }
1171

1172
  if (pStmt->sql.siInfo.pVgroupList) {
×
1173
    taosArrayDestroy(pStmt->sql.siInfo.pVgroupList);
×
1174
    pStmt->sql.siInfo.pVgroupList = NULL;
×
1175
  }
1176

1177
  if (pStmt->sql.siInfo.pDataCtx) {
×
1178
    qDestroyStmtDataBlock(pStmt->sql.siInfo.pDataCtx);
×
1179
    pStmt->sql.siInfo.pDataCtx = NULL;
×
1180
  }
1181

1182
  if (pStmt->sql.siInfo.pTSchema) {
×
1183
    taosMemoryFree(pStmt->sql.siInfo.pTSchema);
×
1184
    pStmt->sql.siInfo.pTSchema = NULL;
×
1185
  }
1186

1187
  if (pStmt->sql.siInfo.pRequest) {
×
1188
    taos_free_result(pStmt->sql.siInfo.pRequest);
×
1189
    pStmt->sql.siInfo.pRequest = NULL;
×
1190
  }
1191

1192
  if (stbInterlaceMode) {
×
1193
    STMT_ERR_RET(stmtResetStbInterlaceCache(pStmt));
×
1194
  }
1195

1196
  pStmt->db = db;
×
1197
  pStmt->stbInterlaceMode = stbInterlaceMode;
×
1198
  pStmt->options = options;
×
1199
  pStmt->reqid = reqid;
×
1200

1201
  pStmt->sql.pTableCache = taosHashInit(100, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
×
1202
  if (NULL == pStmt->sql.pTableCache) {
×
1203
    STMT2_ELOG("fail to allocate memory for pTableCache in stmtResetStmt:%s", tstrerror(terrno));
×
1204
    return terrno;
×
1205
  }
1206

1207
  pStmt->sql.status = STMT_INIT;
×
1208

1209
  return TSDB_CODE_SUCCESS;
×
1210
}
1211

1212
int stmtPrepare2(TAOS_STMT2* stmt, const char* sql, unsigned long length) {
×
1213
  STscStmt2* pStmt = (STscStmt2*)stmt;
×
1214
  int32_t    code = 0;
×
1215

1216
  STMT2_DLOG("start to prepare with sql:%s", sql);
×
1217

1218
  if (stmt == NULL || sql == NULL) {
×
1219
    STMT2_ELOG_E("stmt or sql is NULL");
×
1220
    return TSDB_CODE_INVALID_PARA;
×
1221
  }
1222

1223
  if (pStmt->sql.status >= STMT_PREPARE) {
×
1224
    STMT2_DLOG("stmt status is %d, need to reset stmt2 cache before prepare", pStmt->sql.status);
×
1225
    STMT_ERR_RET(stmtDeepReset(pStmt));
×
1226
  }
1227

1228
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
×
1229
    STMT2_ELOG("errCode is not success before, ErrCode: 0x%x, errorsyt: %s\n. ", pStmt->errCode,
×
1230
               tstrerror(pStmt->errCode));
1231
    return pStmt->errCode;
×
1232
  }
1233

1234
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_PREPARE));
×
1235

1236
  if (length <= 0) {
×
1237
    length = strlen(sql);
×
1238
  }
1239

1240
  pStmt->sql.sqlStr = taosStrndup(sql, length);
×
1241
  if (!pStmt->sql.sqlStr) {
×
1242
    STMT2_ELOG("fail to allocate memory for sqlStr:%s", tstrerror(terrno));
×
1243
    return terrno;
×
1244
  }
1245
  pStmt->sql.sqlLen = length;
×
1246
  pStmt->sql.stbInterlaceMode = pStmt->stbInterlaceMode;
×
1247

1248
  char* dbName = NULL;
×
1249
  if (qParseDbName(sql, length, &dbName)) {
×
1250
    STMT_ERR_RET(stmtSetDbName2(stmt, dbName));
×
1251
    taosMemoryFreeClear(dbName);
×
1252
  } else if (pStmt->db != NULL && pStmt->db[0] != '\0') {
×
1253
    STMT_ERR_RET(stmtCreateRequest(pStmt));
×
1254

1255
    taosMemoryFreeClear(pStmt->exec.pRequest->pDb);
×
1256
    pStmt->exec.pRequest->pDb = taosStrdup(pStmt->db);
×
1257
    if (pStmt->exec.pRequest->pDb == NULL) {
×
1258
      return terrno;
×
1259
    }
1260
    (void)strdequote(pStmt->exec.pRequest->pDb);
×
1261

1262
    if (pStmt->sql.stbInterlaceMode) {
×
1263
      pStmt->sql.siInfo.dbname = pStmt->exec.pRequest->pDb;
×
1264
    }
1265
  }
1266
  return TSDB_CODE_SUCCESS;
×
1267
}
1268

1269
static int32_t stmtInitStbInterlaceTableInfo(STscStmt2* pStmt) {
×
1270
  STableDataCxt** pSrc = taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
×
1271
  if (!pSrc) {
×
1272
    return terrno;
×
1273
  }
1274
  STableDataCxt* pDst = NULL;
×
1275

1276
  STMT_ERR_RET(qCloneStmtDataBlock(&pDst, *pSrc, true));
×
1277
  pStmt->sql.siInfo.pDataCtx = pDst;
×
1278

1279
  SArray* pTblCols = NULL;
×
1280
  for (int32_t i = 0; i < STMT_TABLE_COLS_NUM; i++) {
×
1281
    pTblCols = taosArrayInit(20, POINTER_BYTES);
×
1282
    if (NULL == pTblCols) {
×
1283
      return terrno;
×
1284
    }
1285

1286
    if (taosArrayPush(pStmt->sql.siInfo.pTableCols, &pTblCols) == NULL) {
×
1287
      return terrno;
×
1288
    }
1289
  }
1290

1291
  pStmt->sql.siInfo.boundTags = pStmt->bInfo.boundTags;
×
1292

1293
  STMT2_TLOG("init stb interlace table info, tbName:%s, pDataCtx:%p, boundTags:%p", pStmt->bInfo.tbFName,
×
1294
             pStmt->sql.siInfo.pDataCtx, pStmt->sql.siInfo.boundTags);
1295

1296
  return TSDB_CODE_SUCCESS;
×
1297
}
1298

1299
int stmtIsInsert2(TAOS_STMT2* stmt, int* insert) {
×
1300
  STscStmt2* pStmt = (STscStmt2*)stmt;
×
1301

1302
  // STMT_DLOG_E("start is insert");
1303

1304
  if (pStmt->sql.type) {
×
1305
    *insert = (STMT_TYPE_INSERT == pStmt->sql.type || STMT_TYPE_MULTI_INSERT == pStmt->sql.type);
×
1306
  } else {
1307
    *insert = qIsInsertValuesSql(pStmt->sql.sqlStr, pStmt->sql.sqlLen);
×
1308
  }
1309

1310
  return TSDB_CODE_SUCCESS;
×
1311
}
1312

1313
int stmtSetTbName2(TAOS_STMT2* stmt, const char* tbName) {
×
1314
  STscStmt2* pStmt = (STscStmt2*)stmt;
×
1315

1316
  int64_t startUs = taosGetTimestampUs();
×
1317

1318
  STMT2_TLOG("start to set tbName:%s", tbName);
×
1319

1320
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
×
1321
    return pStmt->errCode;
×
1322
  }
1323

1324
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_SETTBNAME));
×
1325

1326
  int32_t insert = 0;
×
1327
  STMT_ERR_RET(stmtIsInsert2(stmt, &insert));
×
1328
  if (0 == insert) {
×
1329
    STMT2_ELOG_E("set tb name not available for none insert statement");
×
1330
    STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR);
×
1331
  }
1332

1333
  if (!pStmt->sql.stbInterlaceMode || NULL == pStmt->sql.siInfo.pDataCtx) {
×
1334
    STMT_ERR_RET(stmtCreateRequest(pStmt));
×
1335

1336
    STMT_ERR_RET(qCreateSName(&pStmt->bInfo.sname, tbName, pStmt->taos->acctId, pStmt->exec.pRequest->pDb,
×
1337
                              pStmt->exec.pRequest->msgBuf, pStmt->exec.pRequest->msgBufLen));
1338
    STMT_ERR_RET(tNameExtractFullName(&pStmt->bInfo.sname, pStmt->bInfo.tbFName));
×
1339

1340
    STMT_ERR_RET(stmtGetFromCache(pStmt));
×
1341

1342
    if (pStmt->bInfo.needParse) {
×
1343
      tstrncpy(pStmt->bInfo.tbName, tbName, sizeof(pStmt->bInfo.tbName));
×
1344
      pStmt->bInfo.tbName[sizeof(pStmt->bInfo.tbName) - 1] = 0;
×
1345

1346
      STMT_ERR_RET(stmtParseSql(pStmt));
×
1347
    }
1348
  } else {
1349
    tstrncpy(pStmt->bInfo.tbName, tbName, sizeof(pStmt->bInfo.tbName));
×
1350
    pStmt->bInfo.tbName[sizeof(pStmt->bInfo.tbName) - 1] = 0;
×
1351
    pStmt->exec.pRequest->requestId++;
×
1352
    pStmt->bInfo.needParse = false;
×
1353
  }
1354

1355
  if (pStmt->sql.stbInterlaceMode && NULL == pStmt->sql.siInfo.pDataCtx) {
×
1356
    STMT_ERR_RET(stmtInitStbInterlaceTableInfo(pStmt));
×
1357
  }
1358

1359
  int64_t startUs2 = taosGetTimestampUs();
×
1360
  pStmt->stat.setTbNameUs += startUs2 - startUs;
×
1361

1362
  return TSDB_CODE_SUCCESS;
×
1363
}
1364

1365
int stmtSetTbTags2(TAOS_STMT2* stmt, TAOS_STMT2_BIND* tags, SVCreateTbReq** pCreateTbReq) {
×
1366
  STscStmt2* pStmt = (STscStmt2*)stmt;
×
1367

1368
  STMT2_TLOG_E("start to set tbTags");
×
1369

1370
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
×
1371
    return pStmt->errCode;
×
1372
  }
1373

1374
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_SETTAGS));
×
1375

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

1382
  if (pStmt->bInfo.needParse) {
×
1383
    STMT_ERR_RET(stmtParseSql(pStmt));
×
1384
  }
1385
  if (pStmt->sql.stbInterlaceMode && NULL == pStmt->sql.siInfo.pDataCtx) {
×
1386
    STMT_ERR_RET(stmtInitStbInterlaceTableInfo(pStmt));
×
1387
  }
1388

1389
  SBoundColInfo* tags_info = (SBoundColInfo*)pStmt->bInfo.boundTags;
×
1390
  // if (tags_info->numOfBound <= 0 || tags_info->numOfCols <= 0) {
1391
  //   tscWarn("no tags or cols bound in sql, will not bound tags");
1392
  //   return TSDB_CODE_SUCCESS;
1393
  // }
1394
  if (pStmt->sql.autoCreateTbl && pStmt->sql.stbInterlaceMode) {
×
1395
    STMT_ERR_RET(qCreateSName(&pStmt->bInfo.sname, pStmt->bInfo.tbName, pStmt->taos->acctId, pStmt->exec.pRequest->pDb,
×
1396
                              pStmt->exec.pRequest->msgBuf, pStmt->exec.pRequest->msgBufLen));
1397
    STMT_ERR_RET(tNameExtractFullName(&pStmt->bInfo.sname, pStmt->bInfo.tbFName));
×
1398
  }
1399

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

1415
  STMT2_TLOG_E("start to bind stmt tag values");
×
1416

1417
  void* boundTags = NULL;
×
1418
  if (pStmt->sql.stbInterlaceMode) {
×
1419
    boundTags = pStmt->sql.siInfo.boundTags;
×
1420
    *pCreateTbReq = taosMemoryCalloc(1, sizeof(SVCreateTbReq));
×
1421
    if (NULL == pCreateTbReq) {
×
1422
      return terrno;
×
1423
    }
1424
    int32_t vgId = -1;
×
1425
    STMT_ERR_RET(stmtTryAddTableVgroupInfo(pStmt, &vgId));
×
1426
    (*pCreateTbReq)->uid = vgId;
×
1427
  } else {
1428
    boundTags = pStmt->bInfo.boundTags;
×
1429
  }
1430

1431
  STMT_ERR_RET(qBindStmtTagsValue2(*pDataBlock, boundTags, pStmt->bInfo.tbSuid, pStmt->bInfo.stbFName,
×
1432
                                   pStmt->bInfo.sname.tname, tags, pStmt->exec.pRequest->msgBuf,
1433
                                   pStmt->exec.pRequest->msgBufLen, pStmt->taos->optionInfo.charsetCxt, *pCreateTbReq));
1434

1435
  return TSDB_CODE_SUCCESS;
×
1436
}
1437

1438
int stmtCheckTags2(TAOS_STMT2* stmt, SVCreateTbReq** pCreateTbReq) {
×
1439
  STscStmt2* pStmt = (STscStmt2*)stmt;
×
1440

1441
  STMT2_TLOG_E("start to clone createTbRequest for fixed tags");
×
1442

1443
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
×
1444
    return pStmt->errCode;
×
1445
  }
1446

1447
  if (!pStmt->sql.stbInterlaceMode) {
×
1448
    return TSDB_CODE_SUCCESS;
×
1449
  }
1450

1451
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_SETTAGS));
×
1452

1453
  if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 &&
×
1454
      STMT_TYPE_MULTI_INSERT != pStmt->sql.type) {
×
1455
    pStmt->bInfo.needParse = false;
×
1456
  }
1457
  STMT_ERR_RET(stmtCreateRequest(pStmt));
×
1458

1459
  if (pStmt->bInfo.needParse) {
×
1460
    STMT_ERR_RET(stmtParseSql(pStmt));
×
1461
    if (!pStmt->sql.autoCreateTbl) {
×
1462
      STMT2_WLOG_E("don't need to create table, will not check tags");
×
1463
      return TSDB_CODE_SUCCESS;
×
1464
    }
1465
  }
1466

1467
  if (pStmt->sql.stbInterlaceMode && NULL == pStmt->sql.siInfo.pDataCtx) {
×
1468
    STMT_ERR_RET(stmtInitStbInterlaceTableInfo(pStmt));
×
1469
  }
1470

1471
  STMT_ERR_RET(qCreateSName(&pStmt->bInfo.sname, pStmt->bInfo.tbName, pStmt->taos->acctId, pStmt->exec.pRequest->pDb,
×
1472
                            pStmt->exec.pRequest->msgBuf, pStmt->exec.pRequest->msgBufLen));
1473
  STMT_ERR_RET(tNameExtractFullName(&pStmt->bInfo.sname, pStmt->bInfo.tbFName));
×
1474

1475
  STableDataCxt** pDataBlock = NULL;
×
1476
  if (pStmt->exec.pCurrBlock) {
×
1477
    pDataBlock = &pStmt->exec.pCurrBlock;
×
1478
  } else {
1479
    pDataBlock =
1480
        (STableDataCxt**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
×
1481
    if (NULL == pDataBlock) {
×
1482
      STMT2_ELOG("table %s not found in exec blockHash:%p", pStmt->bInfo.tbFName, pStmt->exec.pBlockHash);
×
1483
      STMT_ERR_RET(TSDB_CODE_TSC_STMT_CACHE_ERROR);
×
1484
    }
1485
  }
1486

1487
  if (!((*pDataBlock)->pData->flags & SUBMIT_REQ_AUTO_CREATE_TABLE)) {
×
1488
    STMT2_DLOG_E("don't need to create, will not check tags");
×
1489
    return TSDB_CODE_SUCCESS;
×
1490
  }
1491

1492
  if (pStmt->sql.fixValueTags) {
×
1493
    STMT2_TLOG_E("tags are fixed, use one createTbReq");
×
1494
    STMT_ERR_RET(cloneSVreateTbReq(pStmt->sql.fixValueTbReq, pCreateTbReq));
×
1495
    if ((*pCreateTbReq)->name) {
×
1496
      taosMemoryFree((*pCreateTbReq)->name);
×
1497
    }
1498
    (*pCreateTbReq)->name = taosStrdup(pStmt->bInfo.tbName);
×
1499
    int32_t vgId = -1;
×
1500
    STMT_ERR_RET(stmtTryAddTableVgroupInfo(pStmt, &vgId));
×
1501
    (*pCreateTbReq)->uid = vgId;
×
1502
    return TSDB_CODE_SUCCESS;
×
1503
  }
1504

1505
  if ((*pDataBlock)->pData->pCreateTbReq) {
×
1506
    STMT2_TLOG_E("tags are fixed, set createTbReq first time");
×
1507
    pStmt->sql.fixValueTags = true;
×
1508
    STMT_ERR_RET(cloneSVreateTbReq((*pDataBlock)->pData->pCreateTbReq, &pStmt->sql.fixValueTbReq));
×
1509
    STMT_ERR_RET(cloneSVreateTbReq(pStmt->sql.fixValueTbReq, pCreateTbReq));
×
1510
    (*pCreateTbReq)->uid = (*pDataBlock)->pMeta->vgId;
×
1511
  }
1512

1513
  return TSDB_CODE_SUCCESS;
×
1514
}
1515

1516
static int stmtFetchColFields2(STscStmt2* pStmt, int32_t* fieldNum, TAOS_FIELD_E** fields) {
×
1517
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
×
1518
    return pStmt->errCode;
×
1519
  }
1520

1521
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
×
1522
    tscError("invalid operation to get query column fileds");
×
1523
    STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR);
×
1524
  }
1525

1526
  STableDataCxt** pDataBlock = NULL;
×
1527

1528
  if (pStmt->sql.stbInterlaceMode) {
×
1529
    pDataBlock = &pStmt->sql.siInfo.pDataCtx;
×
1530
  } else {
1531
    pDataBlock =
1532
        (STableDataCxt**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
×
1533
    if (NULL == pDataBlock) {
×
1534
      STMT2_ELOG("table %s not found in exec blockHash:%p", pStmt->bInfo.tbFName, pStmt->exec.pBlockHash);
×
1535
      STMT_ERR_RET(TSDB_CODE_APP_ERROR);
×
1536
    }
1537
  }
1538

1539
  STMT_ERR_RET(qBuildStmtColFields(*pDataBlock, fieldNum, fields));
×
1540

1541
  return TSDB_CODE_SUCCESS;
×
1542
}
1543

1544
static int stmtFetchStbColFields2(STscStmt2* pStmt, int32_t* fieldNum, TAOS_FIELD_ALL** fields) {
×
1545
  int32_t code = 0;
×
1546
  int32_t preCode = pStmt->errCode;
×
1547

1548
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
×
1549
    return pStmt->errCode;
×
1550
  }
1551

1552
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
×
1553
    STMT2_ELOG_E("stmtFetchStbColFields2 only for insert statement");
×
1554
    STMT_ERRI_JRET(TSDB_CODE_TSC_STMT_API_ERROR);
×
1555
  }
1556

1557
  STableDataCxt** pDataBlock = NULL;
×
1558
  bool            cleanStb = false;
×
1559

1560
  if (pStmt->sql.stbInterlaceMode && pStmt->sql.siInfo.pDataCtx != NULL) {
×
1561
    pDataBlock = &pStmt->sql.siInfo.pDataCtx;
×
1562
  } else {
1563
    cleanStb = true;
×
1564
    pDataBlock =
1565
        (STableDataCxt**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
×
1566
  }
1567

1568
  if (NULL == pDataBlock || NULL == *pDataBlock) {
×
1569
    STMT2_ELOG("table %s not found in exec blockHash:%p", pStmt->bInfo.tbFName, pStmt->exec.pBlockHash);
×
1570
    STMT_ERRI_JRET(TSDB_CODE_APP_ERROR);
×
1571
  }
1572

1573
  STMT_ERRI_JRET(qBuildStmtStbColFields(*pDataBlock, pStmt->bInfo.boundTags, pStmt->bInfo.boundCols,
×
1574
                                        pStmt->bInfo.tbNameFlag, fieldNum, fields));
1575

1576
  if (pStmt->bInfo.tbType == TSDB_SUPER_TABLE && cleanStb) {
×
1577
    taosMemoryFreeClear((*pDataBlock)->boundColsInfo.pColIndex);
×
1578
    qDestroyStmtDataBlock(*pDataBlock);
×
1579
    *pDataBlock = NULL;
×
1580
    if (taosHashRemove(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName)) != 0) {
×
1581
      STMT2_ELOG("fail to remove remove stb:%s exec blockHash", pStmt->bInfo.tbFName);
×
1582
      STMT_ERRI_JRET(TSDB_CODE_APP_ERROR);
×
1583
    }
1584
    pStmt->sql.autoCreateTbl = false;
×
1585
    pStmt->bInfo.tagsCached = false;
×
1586
    pStmt->bInfo.sname = (SName){0};
×
1587
    STMT_ERR_RET(stmtCleanBindInfo(pStmt));
×
1588
  }
1589

1590
_return:
×
1591

1592
  pStmt->errCode = preCode;
×
1593

1594
  return code;
×
1595
}
1596
/*
1597
SArray* stmtGetFreeCol(STscStmt2* pStmt, int32_t* idx) {
1598
  while (true) {
1599
    if (pStmt->exec.smInfo.pColIdx >= STMT_COL_BUF_SIZE) {
1600
      pStmt->exec.smInfo.pColIdx = 0;
1601
    }
1602

1603
    if ((pStmt->exec.smInfo.pColIdx + 1) == atomic_load_32(&pStmt->exec.smInfo.pColFreeIdx)) {
1604
      taosUsleep(1);
1605
      continue;
1606
    }
1607

1608
    *idx = pStmt->exec.smInfo.pColIdx;
1609
    return pStmt->exec.smInfo.pCols[pStmt->exec.smInfo.pColIdx++];
1610
  }
1611
}
1612
*/
1613
static int32_t stmtAppendTablePostHandle(STscStmt2* pStmt, SStmtQNode* param) {
×
1614
  if (NULL == pStmt->sql.siInfo.pVgroupHash) {
×
1615
    pStmt->sql.siInfo.pVgroupHash =
×
1616
        taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_ENTRY_LOCK);
×
1617
  }
1618
  if (NULL == pStmt->sql.siInfo.pVgroupList) {
×
1619
    pStmt->sql.siInfo.pVgroupList = taosArrayInit(64, POINTER_BYTES);
×
1620
  }
1621

1622
  if (NULL == pStmt->sql.siInfo.pRequest) {
×
1623
    STMT_ERR_RET(buildRequest(pStmt->taos->id, pStmt->sql.sqlStr, pStmt->sql.sqlLen, NULL, false,
×
1624
                              (SRequestObj**)&pStmt->sql.siInfo.pRequest, pStmt->reqid));
1625

1626
    if (pStmt->reqid != 0) {
×
1627
      pStmt->reqid++;
×
1628
    }
1629
    pStmt->exec.pRequest->syncQuery = true;
×
1630

1631
    pStmt->sql.siInfo.requestId = ((SRequestObj*)pStmt->sql.siInfo.pRequest)->requestId;
×
1632
    pStmt->sql.siInfo.requestSelf = ((SRequestObj*)pStmt->sql.siInfo.pRequest)->self;
×
1633
  }
1634

1635
  if (!pStmt->sql.siInfo.tbFromHash && pStmt->sql.siInfo.firstName[0] &&
×
1636
      0 == strcmp(pStmt->sql.siInfo.firstName, pStmt->bInfo.tbName)) {
×
1637
    pStmt->sql.siInfo.tbFromHash = true;
×
1638
  }
1639

1640
  if (0 == pStmt->sql.siInfo.firstName[0]) {
×
1641
    tstrncpy(pStmt->sql.siInfo.firstName, pStmt->bInfo.tbName, TSDB_TABLE_NAME_LEN);
×
1642
  }
1643

1644
  param->tblData.getFromHash = pStmt->sql.siInfo.tbFromHash;
×
1645
  param->next = NULL;
×
1646

1647
  (void)atomic_add_fetch_64(&pStmt->sql.siInfo.tbRemainNum, 1);
×
1648

1649
  if (pStmt->queue.stopQueue) {
×
1650
    STMT2_ELOG_E("bind thread already stopped, cannot enqueue");
×
1651
    return TSDB_CODE_TSC_STMT_API_ERROR;
×
1652
  }
1653
  stmtEnqueue(pStmt, param);
×
1654

1655
  return TSDB_CODE_SUCCESS;
×
1656
}
1657

1658
static FORCE_INLINE int32_t stmtGetTableColsFromCache(STscStmt2* pStmt, SArray** pTableCols) {
1659
  while (true) {
1660
    if (pStmt->sql.siInfo.pTableColsIdx < taosArrayGetSize(pStmt->sql.siInfo.pTableCols)) {
×
1661
      *pTableCols = (SArray*)taosArrayGetP(pStmt->sql.siInfo.pTableCols, pStmt->sql.siInfo.pTableColsIdx++);
×
1662
      break;
×
1663
    } else {
1664
      SArray* pTblCols = NULL;
×
1665
      for (int32_t i = 0; i < 100; i++) {
×
1666
        pTblCols = taosArrayInit(20, POINTER_BYTES);
×
1667
        if (NULL == pTblCols) {
×
1668
          return terrno;
×
1669
        }
1670

1671
        if (taosArrayPush(pStmt->sql.siInfo.pTableCols, &pTblCols) == NULL) {
×
1672
          return terrno;
×
1673
        }
1674
      }
1675
    }
1676
  }
1677

1678
  return TSDB_CODE_SUCCESS;
×
1679
}
1680

1681
static int32_t stmtCacheBlock(STscStmt2* pStmt) {
×
1682
  if (pStmt->sql.type != STMT_TYPE_MULTI_INSERT) {
×
1683
    return TSDB_CODE_SUCCESS;
×
1684
  }
1685

1686
  uint64_t uid = pStmt->bInfo.tbUid;
×
1687
  uint64_t cacheUid = (TSDB_CHILD_TABLE == pStmt->bInfo.tbType) ? pStmt->bInfo.tbSuid : uid;
×
1688

1689
  if (taosHashGet(pStmt->sql.pTableCache, &cacheUid, sizeof(cacheUid))) {
×
1690
    STMT2_TLOG("table %s already cached, no need to cache again", pStmt->bInfo.tbFName);
×
1691
    return TSDB_CODE_SUCCESS;
×
1692
  }
1693

1694
  STableDataCxt** pSrc = taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
×
1695
  if (!pSrc) {
×
1696
    STMT2_ELOG("table %s not found in exec blockHash:%p", pStmt->bInfo.tbFName, pStmt->exec.pBlockHash);
×
1697
    return terrno;
×
1698
  }
1699
  STableDataCxt* pDst = NULL;
×
1700

1701
  STMT_ERR_RET(qCloneStmtDataBlock(&pDst, *pSrc, true));
×
1702

1703
  SStmtTableCache cache = {
×
1704
      .pDataCtx = pDst,
1705
      .boundTags = pStmt->bInfo.boundTags,
×
1706
  };
1707

1708
  if (taosHashPut(pStmt->sql.pTableCache, &cacheUid, sizeof(cacheUid), &cache, sizeof(cache))) {
×
1709
    STMT2_ELOG("fail to put table cache:%s", tstrerror(terrno));
×
1710
    return terrno;
×
1711
  }
1712

1713
  if (pStmt->sql.autoCreateTbl) {
×
1714
    pStmt->bInfo.tagsCached = true;
×
1715
  } else {
1716
    pStmt->bInfo.boundTags = NULL;
×
1717
  }
1718

1719
  return TSDB_CODE_SUCCESS;
×
1720
}
1721

1722
static int stmtAddBatch2(TAOS_STMT2* stmt) {
×
1723
  STscStmt2* pStmt = (STscStmt2*)stmt;
×
1724

1725
  int64_t startUs = taosGetTimestampUs();
×
1726

1727
  // STMT2_TLOG_E("start to add batch");
1728

1729
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
×
1730
    return pStmt->errCode;
×
1731
  }
1732

1733
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_ADD_BATCH));
×
1734

1735
  if (pStmt->sql.stbInterlaceMode) {
×
1736
    int64_t startUs2 = taosGetTimestampUs();
×
1737
    pStmt->stat.addBatchUs += startUs2 - startUs;
×
1738

1739
    pStmt->sql.siInfo.tableColsReady = false;
×
1740

1741
    SStmtQNode* param = NULL;
×
1742
    STMT_ERR_RET(stmtAllocQNodeFromBuf(&pStmt->sql.siInfo.tbBuf, (void**)&param));
×
1743
    param->restoreTbCols = true;
×
1744
    param->next = NULL;
×
1745

1746
    if (pStmt->sql.autoCreateTbl) {
×
1747
      pStmt->bInfo.tagsCached = true;
×
1748
    }
1749

1750
    if (pStmt->queue.stopQueue) {
×
1751
      STMT2_ELOG_E("stmt bind thread is stopped,cannot enqueue bind request");
×
1752
      return TSDB_CODE_TSC_STMT_API_ERROR;
×
1753
    }
1754

1755
    stmtEnqueue(pStmt, param);
×
1756

1757
    return TSDB_CODE_SUCCESS;
×
1758
  }
1759

1760
  STMT_ERR_RET(stmtCacheBlock(pStmt));
×
1761

1762
  return TSDB_CODE_SUCCESS;
×
1763
}
1764
/*
1765
static int32_t stmtBackupQueryFields(STscStmt2* pStmt) {
1766
  SStmtQueryResInfo* pRes = &pStmt->sql.queryRes;
1767
  pRes->numOfCols = pStmt->exec.pRequest->body.resInfo.numOfCols;
1768
  pRes->precision = pStmt->exec.pRequest->body.resInfo.precision;
1769

1770
  int32_t size = pRes->numOfCols * sizeof(TAOS_FIELD);
1771
  pRes->fields = taosMemoryMalloc(size);
1772
  pRes->userFields = taosMemoryMalloc(size);
1773
  if (NULL == pRes->fields || NULL == pRes->userFields) {
1774
    STMT_ERR_RET(terrno);
1775
  }
1776
  (void)memcpy(pRes->fields, pStmt->exec.pRequest->body.resInfo.fields, size);
1777
  (void)memcpy(pRes->userFields, pStmt->exec.pRequest->body.resInfo.userFields, size);
1778

1779
  return TSDB_CODE_SUCCESS;
1780
}
1781

1782
static int32_t stmtRestoreQueryFields(STscStmt2* pStmt) {
1783
  SStmtQueryResInfo* pRes = &pStmt->sql.queryRes;
1784
  int32_t            size = pRes->numOfCols * sizeof(TAOS_FIELD);
1785

1786
  pStmt->exec.pRequest->body.resInfo.numOfCols = pRes->numOfCols;
1787
  pStmt->exec.pRequest->body.resInfo.precision = pRes->precision;
1788

1789
  if (NULL == pStmt->exec.pRequest->body.resInfo.fields) {
1790
    pStmt->exec.pRequest->body.resInfo.fields = taosMemoryMalloc(size);
1791
    if (NULL == pStmt->exec.pRequest->body.resInfo.fields) {
1792
      STMT_ERR_RET(terrno);
1793
    }
1794
    (void)memcpy(pStmt->exec.pRequest->body.resInfo.fields, pRes->fields, size);
1795
  }
1796

1797
  if (NULL == pStmt->exec.pRequest->body.resInfo.userFields) {
1798
    pStmt->exec.pRequest->body.resInfo.userFields = taosMemoryMalloc(size);
1799
    if (NULL == pStmt->exec.pRequest->body.resInfo.userFields) {
1800
      STMT_ERR_RET(terrno);
1801
    }
1802
    (void)memcpy(pStmt->exec.pRequest->body.resInfo.userFields, pRes->userFields, size);
1803
  }
1804

1805
  return TSDB_CODE_SUCCESS;
1806
}
1807
*/
1808
int stmtBindBatch2(TAOS_STMT2* stmt, TAOS_STMT2_BIND* bind, int32_t colIdx, SVCreateTbReq* pCreateTbReq) {
×
1809
  STscStmt2* pStmt = (STscStmt2*)stmt;
×
1810
  int32_t    code = 0;
×
1811

1812
  int64_t startUs = taosGetTimestampUs();
×
1813

1814
  STMT2_TLOG("start to bind data, colIdx:%d", colIdx);
×
1815

1816
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
×
1817
    return pStmt->errCode;
×
1818
  }
1819

1820
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_BIND));
×
1821

1822
  if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 &&
×
1823
      STMT_TYPE_MULTI_INSERT != pStmt->sql.type) {
×
1824
    pStmt->bInfo.needParse = false;
×
1825
  }
1826

1827
  if (pStmt->exec.pRequest && STMT_TYPE_QUERY == pStmt->sql.type && pStmt->sql.runTimes) {
×
1828
    taos_free_result(pStmt->exec.pRequest);
×
1829
    pStmt->exec.pRequest = NULL;
×
1830
  }
1831

1832
  STMT_ERR_RET(stmtCreateRequest(pStmt));
×
1833
  if (pStmt->bInfo.needParse) {
×
1834
    code = stmtParseSql(pStmt);
×
1835
    if (code != TSDB_CODE_SUCCESS) {
×
1836
      goto cleanup_root;
×
1837
    }
1838
  }
1839

1840
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
×
1841
    code = qStmtBindParams2(pStmt->sql.pQuery, bind, colIdx, pStmt->taos->optionInfo.charsetCxt);
×
1842
    if (code != TSDB_CODE_SUCCESS) {
×
1843
      goto cleanup_root;
×
1844
    }
1845
    SParseContext ctx = {.requestId = pStmt->exec.pRequest->requestId,
×
1846
                         .acctId = pStmt->taos->acctId,
×
1847
                         .db = pStmt->exec.pRequest->pDb,
×
1848
                         .topicQuery = false,
1849
                         .pSql = pStmt->sql.sqlStr,
×
1850
                         .sqlLen = pStmt->sql.sqlLen,
×
1851
                         .pMsg = pStmt->exec.pRequest->msgBuf,
×
1852
                         .msgLen = ERROR_MSG_BUF_DEFAULT_SIZE,
1853
                         .pTransporter = pStmt->taos->pAppInfo->pTransporter,
×
1854
                         .pStmtCb = NULL,
1855
                         .pUser = pStmt->taos->user};
×
1856
    ctx.mgmtEpSet = getEpSet_s(&pStmt->taos->pAppInfo->mgmtEp);
×
1857
    code = catalogGetHandle(pStmt->taos->pAppInfo->clusterId, &ctx.pCatalog);
×
1858
    if (code != TSDB_CODE_SUCCESS) {
×
1859
      goto cleanup_root;
×
1860
    }
1861
    code = qStmtParseQuerySql(&ctx, pStmt->sql.pQuery);
×
1862
    if (code != TSDB_CODE_SUCCESS) {
×
1863
      goto cleanup_root;
×
1864
    }
1865

1866
    if (pStmt->sql.pQuery->haveResultSet) {
×
1867
      STMT_ERR_RET(setResSchemaInfo(&pStmt->exec.pRequest->body.resInfo, pStmt->sql.pQuery->pResSchema,
×
1868
                                    pStmt->sql.pQuery->numOfResCols, pStmt->sql.pQuery->pResExtSchema, true));
1869
      taosMemoryFreeClear(pStmt->sql.pQuery->pResSchema);
×
1870
      taosMemoryFreeClear(pStmt->sql.pQuery->pResExtSchema);
×
1871
      setResPrecision(&pStmt->exec.pRequest->body.resInfo, pStmt->sql.pQuery->precision);
×
1872
    }
1873

1874
    TSWAP(pStmt->exec.pRequest->dbList, pStmt->sql.pQuery->pDbList);
×
1875
    TSWAP(pStmt->exec.pRequest->tableList, pStmt->sql.pQuery->pTableList);
×
1876
    TSWAP(pStmt->exec.pRequest->targetTableList, pStmt->sql.pQuery->pTargetTableList);
×
1877

1878
    // if (STMT_TYPE_QUERY == pStmt->sql.queryRes) {
1879
    //   STMT_ERR_RET(stmtRestoreQueryFields(pStmt));
1880
    // }
1881

1882
    // STMT_ERR_RET(stmtBackupQueryFields(pStmt));
1883

1884
    return TSDB_CODE_SUCCESS;
×
1885

1886
  cleanup_root:
×
1887
    STMT2_ELOG("parse query statment unexpected failed code:%d, need to clean node", code);
×
1888
    if (pStmt->sql.pQuery && pStmt->sql.pQuery->pRoot) {
×
1889
      nodesDestroyNode(pStmt->sql.pQuery->pRoot);
×
1890
      pStmt->sql.pQuery->pRoot = NULL;
×
1891
    }
1892
    STMT_ERR_RET(code);
×
1893
  }
1894

1895
  if (pStmt->sql.stbInterlaceMode && NULL == pStmt->sql.siInfo.pDataCtx) {
×
1896
    STMT_ERR_RET(stmtInitStbInterlaceTableInfo(pStmt));
×
1897
  }
1898

1899
  STableDataCxt** pDataBlock = NULL;
×
1900

1901
  if (pStmt->exec.pCurrBlock) {
×
1902
    pDataBlock = &pStmt->exec.pCurrBlock;
×
1903
  } else {
1904
    pDataBlock =
1905
        (STableDataCxt**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
×
1906
    if (NULL == pDataBlock) {
×
1907
      STMT2_ELOG("table %s not found in exec blockHash:%p", pStmt->bInfo.tbFName, pStmt->exec.pBlockHash);
×
1908
      STMT_ERR_RET(TSDB_CODE_TSC_STMT_CACHE_ERROR);
×
1909
    }
1910
    pStmt->exec.pCurrBlock = *pDataBlock;
×
1911
    if (pStmt->sql.stbInterlaceMode) {
×
1912
      taosArrayDestroy(pStmt->exec.pCurrBlock->pData->aCol);
×
1913
      (*pDataBlock)->pData->aCol = NULL;
×
1914
    }
1915
    if (colIdx < -1) {
×
1916
      pStmt->sql.bindRowFormat = true;
×
1917
      taosArrayDestroy((*pDataBlock)->pData->aCol);
×
1918
      (*pDataBlock)->pData->aCol = taosArrayInit(20, POINTER_BYTES);
×
1919
    }
1920
  }
1921

1922
  int64_t startUs2 = taosGetTimestampUs();
×
1923
  pStmt->stat.bindDataUs1 += startUs2 - startUs;
×
1924

1925
  SStmtQNode* param = NULL;
×
1926
  if (pStmt->sql.stbInterlaceMode) {
×
1927
    STMT_ERR_RET(stmtAllocQNodeFromBuf(&pStmt->sql.siInfo.tbBuf, (void**)&param));
×
1928
    STMT_ERR_RET(stmtGetTableColsFromCache(pStmt, &param->tblData.aCol));
×
1929
    taosArrayClear(param->tblData.aCol);
×
1930

1931
    // param->tblData.aCol = taosArrayInit(20, POINTER_BYTES);
1932

1933
    param->restoreTbCols = false;
×
1934
    param->tblData.isOrdered = true;
×
1935
    param->tblData.isDuplicateTs = false;
×
1936
    tstrncpy(param->tblData.tbName, pStmt->bInfo.tbName, TSDB_TABLE_NAME_LEN);
×
1937

1938
    param->pCreateTbReq = pCreateTbReq;
×
1939
  }
1940

1941
  int64_t startUs3 = taosGetTimestampUs();
×
1942
  pStmt->stat.bindDataUs2 += startUs3 - startUs2;
×
1943

1944
  SArray* pCols = pStmt->sql.stbInterlaceMode ? param->tblData.aCol : (*pDataBlock)->pData->aCol;
×
1945

1946
  if (colIdx < 0) {
×
1947
    if (pStmt->sql.stbInterlaceMode) {
×
1948
      // (*pDataBlock)->pData->flags = 0;
1949
      (*pDataBlock)->pData->flags &= ~SUBMIT_REQ_COLUMN_DATA_FORMAT;
×
1950
      code = qBindStmtStbColsValue2(*pDataBlock, pCols, pStmt->bInfo.boundCols, bind, pStmt->exec.pRequest->msgBuf,
×
1951
                                    pStmt->exec.pRequest->msgBufLen, &pStmt->sql.siInfo.pTSchema, pStmt->sql.pBindInfo,
×
1952
                                    pStmt->taos->optionInfo.charsetCxt);
×
1953
      param->tblData.isOrdered = (*pDataBlock)->ordered;
×
1954
      param->tblData.isDuplicateTs = (*pDataBlock)->duplicateTs;
×
1955
    } else {
1956
      if (colIdx == -1) {
×
1957
        if (pStmt->sql.bindRowFormat) {
×
1958
          STMT2_ELOG_E("can't mix bind row format and bind column format");
×
1959
          STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR);
×
1960
        }
1961
        code = qBindStmtColsValue2(*pDataBlock, pCols, pStmt->bInfo.boundCols, bind, pStmt->exec.pRequest->msgBuf,
×
1962
                                   pStmt->exec.pRequest->msgBufLen, pStmt->taos->optionInfo.charsetCxt);
×
1963
      } else {
1964
        code =
1965
            qBindStmt2RowValue(*pDataBlock, (*pDataBlock)->pData->aRowP, pStmt->bInfo.boundCols, bind,
×
1966
                               pStmt->exec.pRequest->msgBuf, pStmt->exec.pRequest->msgBufLen,
×
1967
                               &pStmt->sql.siInfo.pTSchema, pStmt->sql.pBindInfo, pStmt->taos->optionInfo.charsetCxt);
×
1968
      }
1969
    }
1970

1971
    if (code) {
×
1972
      STMT2_ELOG("bind cols or rows failed, error:%s", tstrerror(code));
×
1973
      STMT_ERR_RET(code);
×
1974
    }
1975
  } else {
1976
    if (pStmt->sql.stbInterlaceMode) {
×
1977
      STMT2_ELOG_E("bind single column not allowed in stb insert mode");
×
1978
      STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR);
×
1979
    }
1980

1981
    if (pStmt->sql.bindRowFormat) {
×
1982
      STMT2_ELOG_E("can't mix bind row format and bind column format");
×
1983
      STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR);
×
1984
    }
1985

1986
    if (colIdx != (pStmt->bInfo.sBindLastIdx + 1) && colIdx != 0) {
×
1987
      STMT2_ELOG_E("bind column index not in sequence");
×
1988
      STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR);
×
1989
    }
1990

1991
    pStmt->bInfo.sBindLastIdx = colIdx;
×
1992

1993
    if (0 == colIdx) {
×
1994
      pStmt->bInfo.sBindRowNum = bind->num;
×
1995
    }
1996

1997
    code = qBindStmtSingleColValue2(*pDataBlock, pCols, bind, pStmt->exec.pRequest->msgBuf,
×
1998
                                    pStmt->exec.pRequest->msgBufLen, colIdx, pStmt->bInfo.sBindRowNum,
×
1999
                                    pStmt->taos->optionInfo.charsetCxt);
×
2000
    if (code) {
×
2001
      STMT2_ELOG("bind single col failed, error:%s", tstrerror(code));
×
2002
      STMT_ERR_RET(code);
×
2003
    }
2004
  }
2005

2006
  int64_t startUs4 = taosGetTimestampUs();
×
2007
  pStmt->stat.bindDataUs3 += startUs4 - startUs3;
×
2008

2009
  if (pStmt->sql.stbInterlaceMode) {
×
2010
    STMT_ERR_RET(stmtAppendTablePostHandle(pStmt, param));
×
2011
  } else {
2012
    STMT_ERR_RET(stmtAddBatch2(pStmt));
×
2013
  }
2014

2015
  pStmt->stat.bindDataUs4 += taosGetTimestampUs() - startUs4;
×
2016

2017
  return TSDB_CODE_SUCCESS;
×
2018
}
2019
/*
2020
int stmtUpdateTableUid(STscStmt2* pStmt, SSubmitRsp* pRsp) {
2021
  tscDebug("stmt start to update tbUid, blockNum:%d", pRsp->nBlocks);
2022

2023
  int32_t code = 0;
2024
  int32_t finalCode = 0;
2025
  size_t  keyLen = 0;
2026
  void*   pIter = taosHashIterate(pStmt->exec.pBlockHash, NULL);
2027
  while (pIter) {
2028
    STableDataCxt* pBlock = *(STableDataCxt**)pIter;
2029
    char*          key = taosHashGetKey(pIter, &keyLen);
2030

2031
    STableMeta* pMeta = qGetTableMetaInDataBlock(pBlock);
2032
    if (pMeta->uid) {
2033
      pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
2034
      continue;
2035
    }
2036

2037
    SSubmitBlkRsp* blkRsp = NULL;
2038
    int32_t        i = 0;
2039
    for (; i < pRsp->nBlocks; ++i) {
2040
      blkRsp = pRsp->pBlocks + i;
2041
      if (strlen(blkRsp->tblFName) != keyLen) {
2042
        continue;
2043
      }
2044

2045
      if (strncmp(blkRsp->tblFName, key, keyLen)) {
2046
        continue;
2047
      }
2048

2049
      break;
2050
    }
2051

2052
    if (i < pRsp->nBlocks) {
2053
      tscDebug("auto created table %s uid updated from %" PRIx64 " to %" PRIx64, blkRsp->tblFName, pMeta->uid,
2054
               blkRsp->uid);
2055

2056
      pMeta->uid = blkRsp->uid;
2057
      pStmt->bInfo.tbUid = blkRsp->uid;
2058
    } else {
2059
      tscDebug("table %s not found in submit rsp, will update from catalog", pStmt->bInfo.tbFName);
2060
      if (NULL == pStmt->pCatalog) {
2061
        code = catalogGetHandle(pStmt->taos->pAppInfo->clusterId, &pStmt->pCatalog);
2062
        if (code) {
2063
          pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
2064
          finalCode = code;
2065
          continue;
2066
        }
2067
      }
2068

2069
      code = stmtCreateRequest(pStmt);
2070
      if (code) {
2071
        pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
2072
        finalCode = code;
2073
        continue;
2074
      }
2075

2076
      STableMeta*      pTableMeta = NULL;
2077
      SRequestConnInfo conn = {.pTrans = pStmt->taos->pAppInfo->pTransporter,
2078
                               .requestId = pStmt->exec.pRequest->requestId,
2079
                               .requestObjRefId = pStmt->exec.pRequest->self,
2080
                               .mgmtEps = getEpSet_s(&pStmt->taos->pAppInfo->mgmtEp)};
2081
      code = catalogGetTableMeta(pStmt->pCatalog, &conn, &pStmt->bInfo.sname, &pTableMeta);
2082

2083
      pStmt->stat.ctgGetTbMetaNum++;
2084

2085
      taos_free_result(pStmt->exec.pRequest);
2086
      pStmt->exec.pRequest = NULL;
2087

2088
      if (code || NULL == pTableMeta) {
2089
        pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
2090
        finalCode = code;
2091
        taosMemoryFree(pTableMeta);
2092
        continue;
2093
      }
2094

2095
      pMeta->uid = pTableMeta->uid;
2096
      pStmt->bInfo.tbUid = pTableMeta->uid;
2097
      taosMemoryFree(pTableMeta);
2098
    }
2099

2100
    pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
2101
  }
2102

2103
  return finalCode;
2104
}
2105
*/
2106
/*
2107
int stmtStaticModeExec(TAOS_STMT* stmt) {
2108
  STscStmt2*   pStmt = (STscStmt2*)stmt;
2109
  int32_t     code = 0;
2110
  SSubmitRsp* pRsp = NULL;
2111
  if (pStmt->sql.staticMode) {
2112
    return TSDB_CODE_TSC_STMT_API_ERROR;
2113
  }
2114

2115
  STMT_DLOG_E("start to exec");
2116

2117
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_EXECUTE));
2118

2119
  STMT_ERR_RET(qBuildStmtOutputFromTbList(pStmt->sql.pQuery, pStmt->sql.pVgHash, pStmt->exec.pTbBlkList,
2120
pStmt->exec.pCurrBlock, pStmt->exec.tbBlkNum));
2121

2122
  launchQueryImpl(pStmt->exec.pRequest, pStmt->sql.pQuery, true, NULL);
2123

2124
  if (pStmt->exec.pRequest->code && NEED_CLIENT_HANDLE_ERROR(pStmt->exec.pRequest->code)) {
2125
    code = refreshMeta(pStmt->exec.pRequest->pTscObj, pStmt->exec.pRequest);
2126
    if (code) {
2127
      pStmt->exec.pRequest->code = code;
2128
    } else {
2129
      tFreeSSubmitRsp(pRsp);
2130
      STMT_ERR_RET(stmtResetStmt(pStmt));
2131
      STMT_ERR_RET(TSDB_CODE_NEED_RETRY);
2132
    }
2133
  }
2134

2135
  STMT_ERR_JRET(pStmt->exec.pRequest->code);
2136

2137
  pStmt->exec.affectedRows = taos_affected_rows(pStmt->exec.pRequest);
2138
  pStmt->affectedRows += pStmt->exec.affectedRows;
2139

2140
_return:
2141

2142
  stmtCleanExecInfo(pStmt, (code ? false : true), false);
2143

2144
  tFreeSSubmitRsp(pRsp);
2145

2146
  ++pStmt->sql.runTimes;
2147

2148
  STMT_RET(code);
2149
}
2150
*/
2151

2152
static int32_t createParseContext(const SRequestObj* pRequest, SParseContext** pCxt, SSqlCallbackWrapper* pWrapper) {
×
2153
  const STscObj* pTscObj = pRequest->pTscObj;
×
2154

2155
  *pCxt = taosMemoryCalloc(1, sizeof(SParseContext));
×
2156
  if (*pCxt == NULL) {
×
2157
    return terrno;
×
2158
  }
2159

2160
  **pCxt = (SParseContext){.requestId = pRequest->requestId,
×
2161
                           .requestRid = pRequest->self,
×
2162
                           .acctId = pTscObj->acctId,
×
2163
                           .db = pRequest->pDb,
×
2164
                           .topicQuery = false,
2165
                           .pSql = pRequest->sqlstr,
×
2166
                           .sqlLen = pRequest->sqlLen,
×
2167
                           .pMsg = pRequest->msgBuf,
×
2168
                           .msgLen = ERROR_MSG_BUF_DEFAULT_SIZE,
2169
                           .pTransporter = pTscObj->pAppInfo->pTransporter,
×
2170
                           .pStmtCb = NULL,
2171
                           .pUser = pTscObj->user,
×
2172
                           .pEffectiveUser = pRequest->effectiveUser,
×
2173
                           .isSuperUser = (0 == strcmp(pTscObj->user, TSDB_DEFAULT_USER)),
×
2174
                           .enableSysInfo = pTscObj->sysInfo,
×
2175
                           .async = true,
2176
                           .svrVer = pTscObj->sVer,
×
2177
                           .nodeOffline = (pTscObj->pAppInfo->onlineDnodes < pTscObj->pAppInfo->totalDnodes),
×
2178
                           .allocatorId = pRequest->allocatorRefId,
×
2179
                           .parseSqlFp = clientParseSql,
2180
                           .parseSqlParam = pWrapper};
2181
  int8_t biMode = atomic_load_8(&((STscObj*)pTscObj)->biMode);
×
2182
  (*pCxt)->biMode = biMode;
×
2183
  return TSDB_CODE_SUCCESS;
×
2184
}
2185

2186
static void asyncQueryCb(void* userdata, TAOS_RES* res, int code) {
×
2187
  STscStmt2*        pStmt = userdata;
×
2188
  __taos_async_fn_t fp = pStmt->options.asyncExecFn;
×
2189
  pStmt->asyncExecCb = true;
×
2190

2191
  pStmt->exec.affectedRows = taos_affected_rows(pStmt->exec.pRequest);
×
2192
  pStmt->affectedRows += pStmt->exec.affectedRows;
×
2193

2194
  fp(pStmt->options.userdata, res, code);
×
2195

2196
  while (0 == atomic_load_8((int8_t*)&pStmt->sql.siInfo.tableColsReady)) {
×
2197
    taosUsleep(1);
×
2198
  }
2199
  (void)stmtCleanExecInfo(pStmt, (code ? false : true), false);
×
2200
  ++pStmt->sql.runTimes;
×
2201

2202
  if (tsem_post(&pStmt->asyncExecSem) != 0) {
×
2203
    STMT2_ELOG_E("fail to post asyncExecSem");
×
2204
  }
2205
}
×
2206

2207
int stmtExec2(TAOS_STMT2* stmt, int* affected_rows) {
×
2208
  STscStmt2* pStmt = (STscStmt2*)stmt;
×
2209
  int32_t    code = 0;
×
2210
  int64_t    startUs = taosGetTimestampUs();
×
2211

2212
  STMT2_DLOG_E("start to exec");
×
2213

2214
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
×
2215
    return pStmt->errCode;
×
2216
  }
2217

2218
  STMT_ERR_RET(taosThreadMutexLock(&pStmt->asyncBindParam.mutex));
×
2219
  while (atomic_load_8((int8_t*)&pStmt->asyncBindParam.asyncBindNum) > 0) {
×
2220
    (void)taosThreadCondWait(&pStmt->asyncBindParam.waitCond, &pStmt->asyncBindParam.mutex);
×
2221
  }
2222
  STMT_ERR_RET(taosThreadMutexUnlock(&pStmt->asyncBindParam.mutex));
×
2223

2224
  if (pStmt->sql.stbInterlaceMode) {
×
2225
    STMT_ERR_RET(stmtAddBatch2(pStmt));
×
2226
  }
2227

2228
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_EXECUTE));
×
2229
  pStmt->asyncExecCb = false;
×
2230

2231
  if (STMT_TYPE_QUERY != pStmt->sql.type) {
×
2232
    if (pStmt->sql.stbInterlaceMode) {
×
2233
      int64_t startTs = taosGetTimestampUs();
×
2234
      // wait for stmt bind thread to finish
2235
      while (atomic_load_64(&pStmt->sql.siInfo.tbRemainNum)) {
×
2236
        taosUsleep(1);
×
2237
      }
2238

2239
      pStmt->stat.execWaitUs += taosGetTimestampUs() - startTs;
×
2240
      STMT_ERR_RET(qBuildStmtFinOutput(pStmt->sql.pQuery, pStmt->sql.pVgHash, pStmt->sql.siInfo.pVgroupList));
×
2241
      taosHashCleanup(pStmt->sql.siInfo.pVgroupHash);
×
2242
      pStmt->sql.siInfo.pVgroupHash = NULL;
×
2243
      pStmt->sql.siInfo.pVgroupList = NULL;
×
2244
    } else {
2245
      tDestroySubmitTbData(pStmt->exec.pCurrTbData, TSDB_MSG_FLG_ENCODE);
×
2246
      taosMemoryFreeClear(pStmt->exec.pCurrTbData);
×
2247

2248
      STMT_ERR_RET(qCloneCurrentTbData(pStmt->exec.pCurrBlock, &pStmt->exec.pCurrTbData));
×
2249

2250
      STMT_ERR_RET(qBuildStmtOutput(pStmt->sql.pQuery, pStmt->sql.pVgHash, pStmt->exec.pBlockHash));
×
2251
    }
2252
  }
2253

2254
  SRequestObj*      pRequest = pStmt->exec.pRequest;
×
2255
  __taos_async_fn_t fp = pStmt->options.asyncExecFn;
×
2256

2257
  if (!fp) {
×
2258
    launchQueryImpl(pStmt->exec.pRequest, pStmt->sql.pQuery, true, NULL);
×
2259

2260
    if (pStmt->exec.pRequest->code && NEED_CLIENT_HANDLE_ERROR(pStmt->exec.pRequest->code)) {
×
2261
      STMT2_ELOG_E("exec failed errorcode:NEED_CLIENT_HANDLE_ERROR, need to refresh meta and retry");
×
2262
      code = refreshMeta(pStmt->exec.pRequest->pTscObj, pStmt->exec.pRequest);
×
2263
      if (code) {
×
2264
        pStmt->exec.pRequest->code = code;
×
2265

2266
      } else {
2267
        STMT_ERR_RET(stmtResetStmt(pStmt));
×
2268
        STMT_ERR_RET(TSDB_CODE_NEED_RETRY);
×
2269
      }
2270
    }
2271

2272
    STMT_ERR_JRET(pStmt->exec.pRequest->code);
×
2273

2274
    pStmt->exec.affectedRows = taos_affected_rows(pStmt->exec.pRequest);
×
2275
    if (affected_rows) {
×
2276
      *affected_rows = pStmt->exec.affectedRows;
×
2277
    }
2278
    pStmt->affectedRows += pStmt->exec.affectedRows;
×
2279

2280
    // wait for stmt bind thread to finish
2281
    while (0 == atomic_load_8((int8_t*)&pStmt->sql.siInfo.tableColsReady)) {
×
2282
      taosUsleep(1);
×
2283
    }
2284

2285
    STMT_ERR_RET(stmtCleanExecInfo(pStmt, (code ? false : true), false));
×
2286

2287
    ++pStmt->sql.runTimes;
×
2288
  } else {
2289
    SSqlCallbackWrapper* pWrapper = taosMemoryCalloc(1, sizeof(SSqlCallbackWrapper));
×
2290
    if (pWrapper == NULL) {
×
2291
      code = terrno;
×
2292
    } else {
2293
      pWrapper->pRequest = pRequest;
×
2294
      pRequest->pWrapper = pWrapper;
×
2295
    }
2296
    if (TSDB_CODE_SUCCESS == code) {
×
2297
      code = createParseContext(pRequest, &pWrapper->pParseCtx, pWrapper);
×
2298
    }
2299
    pRequest->syncQuery = false;
×
2300
    pRequest->body.queryFp = asyncQueryCb;
×
2301
    ((SSyncQueryParam*)(pRequest)->body.interParam)->userParam = pStmt;
×
2302

2303
    pStmt->execSemWaited = false;
×
2304
    launchAsyncQuery(pRequest, pStmt->sql.pQuery, NULL, pWrapper);
×
2305
  }
2306

2307
_return:
×
2308
  if (code) {
×
2309
    STMT2_ELOG("exec failed, error:%s", tstrerror(code));
×
2310
  }
2311
  pStmt->stat.execUseUs += taosGetTimestampUs() - startUs;
×
2312

2313
  STMT_RET(code);
×
2314
}
2315

2316
int stmtClose2(TAOS_STMT2* stmt) {
×
2317
  STscStmt2* pStmt = (STscStmt2*)stmt;
×
2318

2319
  STMT2_DLOG_E("start to close stmt");
×
2320
  taosMemoryFreeClear(pStmt->db);
×
2321

2322
  if (pStmt->bindThreadInUse) {
×
2323
    // wait for stmt bind thread to finish
2324
    while (0 == atomic_load_8((int8_t*)&pStmt->sql.siInfo.tableColsReady)) {
×
2325
      taosUsleep(1);
×
2326
    }
2327

2328
    (void)taosThreadMutexLock(&pStmt->queue.mutex);
×
2329
    pStmt->queue.stopQueue = true;
×
2330
    (void)taosThreadCondSignal(&(pStmt->queue.waitCond));
×
2331
    (void)taosThreadMutexUnlock(&pStmt->queue.mutex);
×
2332

2333
    (void)taosThreadJoin(pStmt->bindThread, NULL);
×
2334
    pStmt->bindThreadInUse = false;
×
2335

2336
    (void)taosThreadCondDestroy(&pStmt->queue.waitCond);
×
2337
    (void)taosThreadMutexDestroy(&pStmt->queue.mutex);
×
2338
  }
2339

2340
  TSC_ERR_RET(taosThreadMutexLock(&pStmt->asyncBindParam.mutex));
×
2341
  while (atomic_load_8((int8_t*)&pStmt->asyncBindParam.asyncBindNum) > 0) {
×
2342
    (void)taosThreadCondWait(&pStmt->asyncBindParam.waitCond, &pStmt->asyncBindParam.mutex);
×
2343
  }
2344
  TSC_ERR_RET(taosThreadMutexUnlock(&pStmt->asyncBindParam.mutex));
×
2345

2346
  (void)taosThreadCondDestroy(&pStmt->asyncBindParam.waitCond);
×
2347
  (void)taosThreadMutexDestroy(&pStmt->asyncBindParam.mutex);
×
2348

2349
  if (pStmt->options.asyncExecFn && !pStmt->execSemWaited) {
×
2350
    if (tsem_wait(&pStmt->asyncExecSem) != 0) {
×
2351
      STMT2_ELOG_E("fail to wait asyncExecSem");
×
2352
    }
2353
  }
2354

2355
  STMT2_DLOG("stmt %p closed, stbInterlaceMode:%d, statInfo: ctgGetTbMetaNum=>%" PRId64 ", getCacheTbInfo=>%" PRId64
×
2356
             ", parseSqlNum=>%" PRId64 ", pStmt->stat.bindDataNum=>%" PRId64
2357
             ", settbnameAPI:%u, bindAPI:%u, addbatchAPI:%u, execAPI:%u"
2358
             ", setTbNameUs:%" PRId64 ", bindDataUs:%" PRId64 ",%" PRId64 ",%" PRId64 ",%" PRId64 " addBatchUs:%" PRId64
2359
             ", execWaitUs:%" PRId64 ", execUseUs:%" PRId64,
2360
             pStmt, pStmt->sql.stbInterlaceMode, pStmt->stat.ctgGetTbMetaNum, pStmt->stat.getCacheTbInfo,
2361
             pStmt->stat.parseSqlNum, pStmt->stat.bindDataNum, pStmt->seqIds[STMT_SETTBNAME], pStmt->seqIds[STMT_BIND],
2362
             pStmt->seqIds[STMT_ADD_BATCH], pStmt->seqIds[STMT_EXECUTE], pStmt->stat.setTbNameUs,
2363
             pStmt->stat.bindDataUs1, pStmt->stat.bindDataUs2, pStmt->stat.bindDataUs3, pStmt->stat.bindDataUs4,
2364
             pStmt->stat.addBatchUs, pStmt->stat.execWaitUs, pStmt->stat.execUseUs);
2365
  if (pStmt->sql.stbInterlaceMode) {
×
2366
    pStmt->bInfo.tagsCached = false;
×
2367
  }
2368

2369
  STMT_ERR_RET(stmtCleanSQLInfo(pStmt));
×
2370

2371
  if (pStmt->options.asyncExecFn) {
×
2372
    if (tsem_destroy(&pStmt->asyncExecSem) != 0) {
×
2373
      STMT2_ELOG_E("fail to destroy asyncExecSem");
×
2374
    }
2375
  }
2376
  taosMemoryFree(stmt);
×
2377

2378
  return TSDB_CODE_SUCCESS;
×
2379
}
2380

2381
const char* stmtErrstr2(TAOS_STMT2* stmt) {
×
2382
  STscStmt2* pStmt = (STscStmt2*)stmt;
×
2383

2384
  if (stmt == NULL || NULL == pStmt->exec.pRequest) {
×
2385
    return (char*)tstrerror(terrno);
×
2386
  }
2387

2388
  // if stmt async exec ,error code is pStmt->exec.pRequest->code
2389
  if (!(pStmt->sql.status >= STMT_EXECUTE && pStmt->options.asyncExecFn != NULL && pStmt->asyncExecCb)) {
×
2390
    pStmt->exec.pRequest->code = terrno;
×
2391
  }
2392

2393
  SRequestObj* pRequest = pStmt->exec.pRequest;
×
2394
  if (NULL != pRequest->msgBuf && (strlen(pRequest->msgBuf) > 0 || pRequest->code == TSDB_CODE_RPC_FQDN_ERROR)) {
×
2395
    return pRequest->msgBuf;
×
2396
  }
2397
  return (const char*)tstrerror(pRequest->code);
×
2398
}
2399
/*
2400
int stmtAffectedRows(TAOS_STMT* stmt) { return ((STscStmt2*)stmt)->affectedRows; }
2401

2402
int stmtAffectedRowsOnce(TAOS_STMT* stmt) { return ((STscStmt2*)stmt)->exec.affectedRows; }
2403
*/
2404

2405
int stmtParseColFields2(TAOS_STMT2* stmt) {
×
2406
  int32_t    code = 0;
×
2407
  STscStmt2* pStmt = (STscStmt2*)stmt;
×
2408
  int32_t    preCode = pStmt->errCode;
×
2409

2410
  STMT2_DLOG_E("start to get col fields for insert");
×
2411

2412
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
×
2413
    return pStmt->errCode;
×
2414
  }
2415

2416
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
×
2417
    STMT2_ELOG_E("stmtParseColFields2 only for insert");
×
2418
    STMT_ERRI_JRET(TSDB_CODE_TSC_STMT_API_ERROR);
×
2419
  }
2420

2421
  STMT_ERRI_JRET(stmtSwitchStatus(pStmt, STMT_FETCH_FIELDS));
×
2422

2423
  if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 &&
×
2424
      STMT_TYPE_MULTI_INSERT != pStmt->sql.type) {
×
2425
    pStmt->bInfo.needParse = false;
×
2426
  }
2427
  if (pStmt->sql.stbInterlaceMode && pStmt->sql.siInfo.pDataCtx != NULL) {
×
2428
    pStmt->bInfo.needParse = false;
×
2429
  }
2430

2431
  STMT_ERRI_JRET(stmtCreateRequest(pStmt));
×
2432

2433
  if (pStmt->bInfo.needParse) {
×
2434
    STMT_ERRI_JRET(stmtParseSql(pStmt));
×
2435
  }
2436

2437
_return:
×
2438
  // compatible with previous versions
2439
  if (code == TSDB_CODE_PAR_TABLE_NOT_EXIST && (pStmt->bInfo.tbNameFlag & NO_DATA_USING_CLAUSE) == 0x0) {
×
2440
    code = TSDB_CODE_TSC_STMT_TBNAME_ERROR;
×
2441
  }
2442

2443
  if (code != TSDB_CODE_SUCCESS) {
×
2444
    STMT2_ELOG("stmt get fileds parse failed, code:%d", code);
×
2445
    taos_free_result(pStmt->exec.pRequest);
×
2446
    pStmt->exec.pRequest = NULL;
×
2447
  }
2448

2449
  pStmt->errCode = preCode;
×
2450

2451
  return code;
×
2452
}
2453

2454
int stmtGetStbColFields2(TAOS_STMT2* stmt, int* nums, TAOS_FIELD_ALL** fields) {
×
2455
  int32_t code = stmtParseColFields2(stmt);
×
2456
  if (code != TSDB_CODE_SUCCESS) {
×
2457
    return code;
×
2458
  }
2459

2460
  return stmtFetchStbColFields2(stmt, nums, fields);
×
2461
}
2462

2463
int stmtGetParamNum2(TAOS_STMT2* stmt, int* nums) {
×
2464
  int32_t    code = 0;
×
2465
  STscStmt2* pStmt = (STscStmt2*)stmt;
×
2466
  int32_t    preCode = pStmt->errCode;
×
2467

2468
  STMT2_DLOG_E("start to get param num for query");
×
2469

2470
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
×
2471
    return pStmt->errCode;
×
2472
  }
2473

2474
  STMT_ERRI_JRET(stmtSwitchStatus(pStmt, STMT_FETCH_FIELDS));
×
2475

2476
  if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 &&
×
2477
      STMT_TYPE_MULTI_INSERT != pStmt->sql.type) {
×
2478
    pStmt->bInfo.needParse = false;
×
2479
  }
2480

2481
  if (pStmt->exec.pRequest && STMT_TYPE_QUERY == pStmt->sql.type && pStmt->sql.runTimes) {
×
2482
    taos_free_result(pStmt->exec.pRequest);
×
2483
    pStmt->exec.pRequest = NULL;
×
2484
  }
2485

2486
  STMT_ERRI_JRET(stmtCreateRequest(pStmt));
×
2487

2488
  if (pStmt->bInfo.needParse) {
×
2489
    STMT_ERRI_JRET(stmtParseSql(pStmt));
×
2490
  }
2491

2492
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
×
2493
    *nums = taosArrayGetSize(pStmt->sql.pQuery->pPlaceholderValues);
×
2494
  } else {
2495
    STMT_ERRI_JRET(stmtFetchColFields2(stmt, nums, NULL));
×
2496
  }
2497

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

2500
_return:
×
2501
  if (code != TSDB_CODE_SUCCESS) {
×
2502
    taos_free_result(pStmt->exec.pRequest);
×
2503
    pStmt->exec.pRequest = NULL;
×
2504
  }
2505
  pStmt->errCode = preCode;
×
2506

2507
  return code;
×
2508
}
2509

2510
TAOS_RES* stmtUseResult2(TAOS_STMT2* stmt) {
×
2511
  STscStmt2* pStmt = (STscStmt2*)stmt;
×
2512

2513
  STMT2_TLOG_E("start to use result");
×
2514

2515
  if (STMT_TYPE_QUERY != pStmt->sql.type) {
×
2516
    STMT2_ELOG_E("useResult only for query statement");
×
2517
    return NULL;
×
2518
  }
2519

2520
  return pStmt->exec.pRequest;
×
2521
}
2522

2523
int32_t stmtAsyncBindThreadFunc(void* args) {
×
2524
  qInfo("async stmt bind thread started");
×
2525

2526
  ThreadArgs* targs = (ThreadArgs*)args;
×
2527
  STscStmt2*  pStmt = (STscStmt2*)targs->stmt;
×
2528

2529
  int code = taos_stmt2_bind_param(targs->stmt, targs->bindv, targs->col_idx);
×
2530
  targs->fp(targs->param, NULL, code);
×
2531
  (void)taosThreadMutexLock(&(pStmt->asyncBindParam.mutex));
×
2532
  (void)atomic_sub_fetch_8(&pStmt->asyncBindParam.asyncBindNum, 1);
×
2533
  (void)taosThreadCondSignal(&(pStmt->asyncBindParam.waitCond));
×
2534
  (void)taosThreadMutexUnlock(&(pStmt->asyncBindParam.mutex));
×
2535
  taosMemoryFree(args);
×
2536

2537
  qInfo("async stmt bind thread stopped");
×
2538

2539
  return code;
×
2540
}
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