• 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/clientStmt.c
1

2
#include "clientInt.h"
3
#include "clientLog.h"
4
#include "tdef.h"
5

6
#include "clientStmt.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(terrno);
×
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
bool stmtDequeue(STscStmt* pStmt, SStmtQNode** param) {
×
42
  int i = 0;
×
43
  while (0 == atomic_load_64((int64_t*)&pStmt->queue.qRemainNum)) {
×
44
    if (i < 10) {
×
45
      taosUsleep(1);
×
46
      i++;
×
47
    } else {
48
      (void)taosThreadMutexLock(&pStmt->queue.mutex);
×
49
      if (0 == atomic_load_64((int64_t*)&pStmt->queue.qRemainNum)) {
×
50
        (void)taosThreadCondWait(&pStmt->queue.waitCond, &pStmt->queue.mutex);
×
51
      }
52
      (void)taosThreadMutexUnlock(&pStmt->queue.mutex);
×
53
    }
54
  }
55
  if (pStmt->queue.stopQueue) {
×
56
    return false;
×
57
  }
58
  SStmtQNode* orig = pStmt->queue.head;
×
59
  SStmtQNode* node = pStmt->queue.head->next;
×
60
  pStmt->queue.head = pStmt->queue.head->next;
×
61
  *param = node;
×
62

63
  (void)atomic_sub_fetch_64((int64_t*)&pStmt->queue.qRemainNum, 1);
×
64

65
  return true;
×
66
}
67

68
void stmtEnqueue(STscStmt* pStmt, SStmtQNode* param) {
×
69
  pStmt->queue.tail->next = param;
×
70
  pStmt->queue.tail = param;
×
71

72
  pStmt->stat.bindDataNum++;
×
73

74
  (void)taosThreadMutexLock(&pStmt->queue.mutex);
×
75
  (void)atomic_add_fetch_64(&pStmt->queue.qRemainNum, 1);
×
76
  (void)taosThreadCondSignal(&(pStmt->queue.waitCond));
×
77
  (void)taosThreadMutexUnlock(&pStmt->queue.mutex);
×
78
}
×
79

80
static int32_t stmtCreateRequest(STscStmt* pStmt) {
×
81
  int32_t code = 0;
×
82

83
  if (pStmt->exec.pRequest == NULL) {
×
84
    code = buildRequest(pStmt->taos->id, pStmt->sql.sqlStr, pStmt->sql.sqlLen, NULL, false, &pStmt->exec.pRequest,
×
85
                        pStmt->reqid);
86
    if (pStmt->reqid != 0) {
×
87
      pStmt->reqid++;
×
88
    }
89
    if (TSDB_CODE_SUCCESS == code) {
×
90
      pStmt->exec.pRequest->syncQuery = true;
×
91
      pStmt->exec.pRequest->isStmtBind = true;
×
92
    }
93
  }
94

95
  return code;
×
96
}
97

98
int32_t stmtSwitchStatus(STscStmt* pStmt, STMT_STATUS newStatus) {
×
99
  int32_t code = 0;
×
100

101
  if (newStatus >= STMT_INIT && newStatus < STMT_MAX) {
×
102
    STMT_LOG_SEQ(newStatus);
×
103
  }
104

105
  if (pStmt->errCode && newStatus != STMT_PREPARE) {
×
106
    STMT_DLOG("stmt already failed with err:%s", tstrerror(pStmt->errCode));
×
107
    return pStmt->errCode;
×
108
  }
109

110
  switch (newStatus) {
×
111
    case STMT_PREPARE:
×
112
      pStmt->errCode = 0;
×
113
      break;
×
114
    case STMT_SETTBNAME:
×
115
      if (STMT_STATUS_EQ(INIT)) {
×
116
        code = TSDB_CODE_TSC_STMT_API_ERROR;
×
117
      }
118
      if (!pStmt->sql.stbInterlaceMode && (STMT_STATUS_EQ(BIND) || STMT_STATUS_EQ(BIND_COL))) {
×
119
        code = TSDB_CODE_TSC_STMT_API_ERROR;
×
120
      }
121
      break;
×
122
    case STMT_SETTAGS:
×
123
      if (STMT_STATUS_NE(SETTBNAME) && STMT_STATUS_NE(FETCH_FIELDS)) {
×
124
        code = TSDB_CODE_TSC_STMT_API_ERROR;
×
125
      }
126
      break;
×
127
    case STMT_FETCH_FIELDS:
×
128
      if (STMT_STATUS_EQ(INIT)) {
×
129
        code = TSDB_CODE_TSC_STMT_API_ERROR;
×
130
      }
131
      break;
×
132
    case STMT_BIND:
×
133
      if (STMT_STATUS_EQ(INIT) || STMT_STATUS_EQ(BIND_COL)) {
×
134
        code = TSDB_CODE_TSC_STMT_API_ERROR;
×
135
      }
136
      /*
137
            if ((pStmt->sql.type == STMT_TYPE_MULTI_INSERT) && ()) {
138
              code = TSDB_CODE_TSC_STMT_API_ERROR;
139
            }
140
      */
141
      break;
×
142
    case STMT_BIND_COL:
×
143
      if (STMT_STATUS_EQ(INIT) || STMT_STATUS_EQ(BIND)) {
×
144
        code = TSDB_CODE_TSC_STMT_API_ERROR;
×
145
      }
146
      break;
×
147
    case STMT_ADD_BATCH:
×
148
      if (STMT_STATUS_NE(BIND) && STMT_STATUS_NE(BIND_COL) && STMT_STATUS_NE(FETCH_FIELDS)) {
×
149
        code = TSDB_CODE_TSC_STMT_API_ERROR;
×
150
      }
151
      break;
×
152
    case STMT_EXECUTE:
×
153
      if (STMT_TYPE_QUERY == pStmt->sql.type) {
×
154
        if (STMT_STATUS_NE(ADD_BATCH) && STMT_STATUS_NE(FETCH_FIELDS) && STMT_STATUS_NE(BIND) &&
×
155
            STMT_STATUS_NE(BIND_COL)) {
×
156
          code = TSDB_CODE_TSC_STMT_API_ERROR;
×
157
        }
158
      } else {
159
        if (STMT_STATUS_NE(ADD_BATCH) && STMT_STATUS_NE(FETCH_FIELDS)) {
×
160
          code = TSDB_CODE_TSC_STMT_API_ERROR;
×
161
        }
162
      }
163
      break;
×
164
    default:
×
165
      code = TSDB_CODE_APP_ERROR;
×
166
      break;
×
167
  }
168

169
  STMT_ERR_RET(code);
×
170

171
  pStmt->sql.status = newStatus;
×
172

173
  return TSDB_CODE_SUCCESS;
×
174
}
175

176
int32_t stmtGetTbName(TAOS_STMT* stmt, char** tbName) {
×
177
  STscStmt* pStmt = (STscStmt*)stmt;
×
178

179
  pStmt->sql.type = STMT_TYPE_MULTI_INSERT;
×
180

181
  if ('\0' == pStmt->bInfo.tbName[0]) {
×
182
    tscError("no table name set");
×
183
    STMT_ERR_RET(TSDB_CODE_TSC_STMT_TBNAME_ERROR);
×
184
  }
185

186
  *tbName = pStmt->bInfo.tbName;
×
187

188
  return TSDB_CODE_SUCCESS;
×
189
}
190
/*
191
int32_t stmtBackupQueryFields(STscStmt* pStmt) {
192
  SStmtQueryResInfo* pRes = &pStmt->sql.queryRes;
193
  pRes->numOfCols = pStmt->exec.pRequest->body.resInfo.numOfCols;
194
  pRes->precision = pStmt->exec.pRequest->body.resInfo.precision;
195

196
  int32_t size = pRes->numOfCols * sizeof(TAOS_FIELD);
197
  pRes->fields = taosMemoryMalloc(size);
198
  if (pRes->fields == NULL) {
199
    STMT_ERR_RET(terrno);
200
  }
201

202
  pRes->userFields = taosMemoryMalloc(size);
203
  if (pRes->userFields == NULL) {
204
    taosMemoryFreeClear(pRes->fields);
205
    STMT_ERR_RET(terrno);
206
  }
207

208
  (void)memcpy(pRes->fields, pStmt->exec.pRequest->body.resInfo.fields, size);
209
  (void)memcpy(pRes->userFields, pStmt->exec.pRequest->body.resInfo.userFields, size);
210

211
  return TSDB_CODE_SUCCESS;
212
}
213

214
int32_t stmtRestoreQueryFields(STscStmt* pStmt) {
215
  SStmtQueryResInfo* pRes = &pStmt->sql.queryRes;
216
  int32_t            size = pRes->numOfCols * sizeof(TAOS_FIELD_E);
217

218
  pStmt->exec.pRequest->body.resInfo.numOfCols = pRes->numOfCols;
219
  pStmt->exec.pRequest->body.resInfo.precision = pRes->precision;
220

221
  if (NULL == pStmt->exec.pRequest->body.resInfo.fields) {
222
    pStmt->exec.pRequest->body.resInfo.fields = taosMemoryMalloc(size);
223
    if (NULL == pStmt->exec.pRequest->body.resInfo.fields) {
224
      STMT_ERR_RET(terrno);
225
    }
226
    (void)memcpy(pStmt->exec.pRequest->body.resInfo.fields, pRes->fields, size);
227
  }
228

229
  if (NULL == pStmt->exec.pRequest->body.resInfo.userFields) {
230
    pStmt->exec.pRequest->body.resInfo.userFields = taosMemoryMalloc(size);
231
    if (NULL == pStmt->exec.pRequest->body.resInfo.userFields) {
232
      STMT_ERR_RET(terrno);
233
    }
234
    (void)memcpy(pStmt->exec.pRequest->body.resInfo.userFields, pRes->userFields, size);
235
  }
236

237
  return TSDB_CODE_SUCCESS;
238
}
239
*/
240
int32_t stmtUpdateBindInfo(TAOS_STMT* stmt, STableMeta* pTableMeta, void* tags, SName* tbName, const char* sTableName,
×
241
                           bool autoCreateTbl, uint8_t tbNameFlag) {
242
  STscStmt* pStmt = (STscStmt*)stmt;
×
243
  char      tbFName[TSDB_TABLE_FNAME_LEN];
244
  int32_t   code = tNameExtractFullName(tbName, tbFName);
×
245
  if (code != 0) {
×
246
    return code;
×
247
  }
248

249
  (void)memcpy(&pStmt->bInfo.sname, tbName, sizeof(*tbName));
×
250
  tstrncpy(pStmt->bInfo.tbFName, tbFName, TSDB_TABLE_FNAME_LEN);
×
251
  pStmt->bInfo.tbFName[sizeof(pStmt->bInfo.tbFName) - 1] = 0;
×
252

253
  pStmt->bInfo.tbUid = autoCreateTbl ? 0 : pTableMeta->uid;
×
254
  pStmt->bInfo.tbSuid = pTableMeta->suid;
×
255
  pStmt->bInfo.tbVgId = pTableMeta->vgId;
×
256
  pStmt->bInfo.tbType = pTableMeta->tableType;
×
257
  pStmt->bInfo.boundTags = tags;
×
258
  pStmt->bInfo.tagsCached = false;
×
259
  pStmt->bInfo.tbNameFlag = tbNameFlag;
×
260
  tstrncpy(pStmt->bInfo.stbFName, sTableName, sizeof(pStmt->bInfo.stbFName));
×
261

262
  return TSDB_CODE_SUCCESS;
×
263
}
264

265
int32_t stmtUpdateExecInfo(TAOS_STMT* stmt, SHashObj* pVgHash, SHashObj* pBlockHash) {
×
266
  STscStmt* pStmt = (STscStmt*)stmt;
×
267

268
  pStmt->sql.pVgHash = pVgHash;
×
269
  pStmt->exec.pBlockHash = pBlockHash;
×
270

271
  return TSDB_CODE_SUCCESS;
×
272
}
273

274
int32_t stmtUpdateInfo(TAOS_STMT* stmt, STableMeta* pTableMeta, void* tags, SArray* cols, SName* tbName,
×
275
                       bool autoCreateTbl, SHashObj* pVgHash, SHashObj* pBlockHash, const char* sTableName,
276
                       uint8_t tbNameFlag) {
277
  STscStmt* pStmt = (STscStmt*)stmt;
×
278

279
  STMT_ERR_RET(stmtUpdateBindInfo(stmt, pTableMeta, tags, tbName, sTableName, autoCreateTbl, tbNameFlag));
×
280
  STMT_ERR_RET(stmtUpdateExecInfo(stmt, pVgHash, pBlockHash));
×
281

282
  pStmt->sql.autoCreateTbl = autoCreateTbl;
×
283
  if (pStmt->sql.autoCreateTbl) {
×
284
    pStmt->sql.stbInterlaceMode = false;
×
285
  }
286

287
  return TSDB_CODE_SUCCESS;
×
288
}
289

290
int32_t stmtGetExecInfo(TAOS_STMT* stmt, SHashObj** pVgHash, SHashObj** pBlockHash) {
×
291
  STscStmt* pStmt = (STscStmt*)stmt;
×
292

293
  *pVgHash = pStmt->sql.pVgHash;
×
294
  pStmt->sql.pVgHash = NULL;
×
295

296
  *pBlockHash = pStmt->exec.pBlockHash;
×
297
  pStmt->exec.pBlockHash = NULL;
×
298

299
  return TSDB_CODE_SUCCESS;
×
300
}
301

302
int32_t stmtCacheBlock(STscStmt* pStmt) {
×
303
  if (pStmt->sql.type != STMT_TYPE_MULTI_INSERT) {
×
304
    return TSDB_CODE_SUCCESS;
×
305
  }
306

307
  uint64_t uid = pStmt->bInfo.tbUid;
×
308
  uint64_t cacheUid = (TSDB_CHILD_TABLE == pStmt->bInfo.tbType) ? pStmt->bInfo.tbSuid : uid;
×
309

310
  if (taosHashGet(pStmt->sql.pTableCache, &cacheUid, sizeof(cacheUid))) {
×
311
    return TSDB_CODE_SUCCESS;
×
312
  }
313

314
  STableDataCxt** pSrc = taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
×
315
  if (!pSrc) {
×
316
    return terrno;
×
317
  }
318
  STableDataCxt* pDst = NULL;
×
319

320
  STMT_ERR_RET(qCloneStmtDataBlock(&pDst, *pSrc, true));
×
321

322
  SStmtTableCache cache = {
×
323
      .pDataCtx = pDst,
324
      .boundTags = pStmt->bInfo.boundTags,
×
325
  };
326

327
  if (taosHashPut(pStmt->sql.pTableCache, &cacheUid, sizeof(cacheUid), &cache, sizeof(cache))) {
×
328
    return terrno;
×
329
  }
330

331
  if (pStmt->sql.autoCreateTbl) {
×
332
    pStmt->bInfo.tagsCached = true;
×
333
  } else {
334
    pStmt->bInfo.boundTags = NULL;
×
335
  }
336

337
  return TSDB_CODE_SUCCESS;
×
338
}
339

340
int32_t stmtParseSql(STscStmt* pStmt) {
×
341
  pStmt->exec.pCurrBlock = NULL;
×
342

343
  SStmtCallback stmtCb = {
×
344
      .pStmt = pStmt,
345
      .getTbNameFn = stmtGetTbName,
346
      .setInfoFn = stmtUpdateInfo,
347
      .getExecInfoFn = stmtGetExecInfo,
348
  };
349

350
  STMT_ERR_RET(stmtCreateRequest(pStmt));
×
351

352
  pStmt->stat.parseSqlNum++;
×
353
  STMT_ERR_RET(parseSql(pStmt->exec.pRequest, false, &pStmt->sql.pQuery, &stmtCb));
×
354
  pStmt->sql.siInfo.pQuery = pStmt->sql.pQuery;
×
355

356
  pStmt->bInfo.needParse = false;
×
357

358
  if (pStmt->sql.pQuery->pRoot && 0 == pStmt->sql.type) {
×
359
    pStmt->sql.type = STMT_TYPE_INSERT;
×
360
    pStmt->sql.stbInterlaceMode = false;
×
361
  } else if (pStmt->sql.pQuery->pPrepareRoot) {
×
362
    pStmt->sql.type = STMT_TYPE_QUERY;
×
363
    pStmt->sql.stbInterlaceMode = false;
×
364

365
    return TSDB_CODE_SUCCESS;
×
366
  }
367

368
  STableDataCxt** pSrc =
369
      (STableDataCxt**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
×
370
  if (NULL == pSrc || NULL == *pSrc) {
×
371
    return terrno;
×
372
  }
373

374
  STableDataCxt* pTableCtx = *pSrc;
×
375
  if (pStmt->sql.stbInterlaceMode) {
×
376
    int16_t lastIdx = -1;
×
377

378
    for (int32_t i = 0; i < pTableCtx->boundColsInfo.numOfBound; ++i) {
×
379
      if (pTableCtx->boundColsInfo.pColIndex[i] < lastIdx) {
×
380
        pStmt->sql.stbInterlaceMode = false;
×
381
        break;
×
382
      }
383

384
      lastIdx = pTableCtx->boundColsInfo.pColIndex[i];
×
385
    }
386
  }
387

388
  if (NULL == pStmt->sql.pBindInfo) {
×
389
    pStmt->sql.pBindInfo = taosMemoryMalloc(pTableCtx->boundColsInfo.numOfBound * sizeof(*pStmt->sql.pBindInfo));
×
390
    if (NULL == pStmt->sql.pBindInfo) {
×
391
      return terrno;
×
392
    }
393
  }
394

395
  return TSDB_CODE_SUCCESS;
×
396
}
397

398
int32_t stmtCleanBindInfo(STscStmt* pStmt) {
×
399
  pStmt->bInfo.tbUid = 0;
×
400
  pStmt->bInfo.tbSuid = 0;
×
401
  pStmt->bInfo.tbVgId = -1;
×
402
  pStmt->bInfo.tbType = 0;
×
403
  pStmt->bInfo.needParse = true;
×
404
  pStmt->bInfo.inExecCache = false;
×
405

406
  pStmt->bInfo.tbName[0] = 0;
×
407
  pStmt->bInfo.tbFName[0] = 0;
×
408
  if (!pStmt->bInfo.tagsCached) {
×
409
    qDestroyBoundColInfo(pStmt->bInfo.boundTags);
×
410
    taosMemoryFreeClear(pStmt->bInfo.boundTags);
×
411
  }
412
  pStmt->bInfo.stbFName[0] = 0;
×
413

414
  return TSDB_CODE_SUCCESS;
×
415
}
416

417
void stmtFreeTableBlkList(STableColsData* pTb) {
×
418
  (void)qResetStmtColumns(pTb->aCol, true);
×
419
  taosArrayDestroy(pTb->aCol);
×
420
}
×
421

422
void stmtResetQueueTableBuf(STableBufInfo* pTblBuf, SStmtQueue* pQueue) {
×
423
  pTblBuf->pCurBuff = taosArrayGetP(pTblBuf->pBufList, 0);
×
424
  if (NULL == pTblBuf->pCurBuff) {
×
425
    tscError("QInfo:%p, failed to get buffer from list", pTblBuf);
×
426
    return;
×
427
  }
428
  pTblBuf->buffIdx = 1;
×
429
  pTblBuf->buffOffset = sizeof(*pQueue->head);
×
430

431
  pQueue->head = pQueue->tail = pTblBuf->pCurBuff;
×
432
  pQueue->qRemainNum = 0;
×
433
  pQueue->head->next = NULL;
×
434
}
435

436
int32_t stmtCleanExecInfo(STscStmt* pStmt, bool keepTable, bool deepClean) {
×
437
  if (pStmt->sql.stbInterlaceMode) {
×
438
    if (deepClean) {
×
439
      taosHashCleanup(pStmt->exec.pBlockHash);
×
440
      pStmt->exec.pBlockHash = NULL;
×
441

442
      if (NULL != pStmt->exec.pCurrBlock) {
×
443
        taosMemoryFreeClear(pStmt->exec.pCurrBlock->pData);
×
444
        qDestroyStmtDataBlock(pStmt->exec.pCurrBlock);
×
445
      }
446
    } else {
447
      pStmt->sql.siInfo.pTableColsIdx = 0;
×
448
      stmtResetQueueTableBuf(&pStmt->sql.siInfo.tbBuf, &pStmt->queue);
×
449
      tSimpleHashClear(pStmt->sql.siInfo.pTableRowDataHash);
×
450
    }
451
  } else {
452
    if (STMT_TYPE_QUERY != pStmt->sql.type || deepClean) {
×
453
      taos_free_result(pStmt->exec.pRequest);
×
454
      pStmt->exec.pRequest = NULL;
×
455
    }
456

457
    size_t keyLen = 0;
×
458
    void*  pIter = taosHashIterate(pStmt->exec.pBlockHash, NULL);
×
459
    while (pIter) {
×
460
      STableDataCxt* pBlocks = *(STableDataCxt**)pIter;
×
461
      char*          key = taosHashGetKey(pIter, &keyLen);
×
462
      STableMeta*    pMeta = qGetTableMetaInDataBlock(pBlocks);
×
463

464
      if (keepTable && pBlocks == pStmt->exec.pCurrBlock) {
×
465
        TSWAP(pBlocks->pData, pStmt->exec.pCurrTbData);
×
466
        STMT_ERR_RET(qResetStmtDataBlock(pBlocks, false));
×
467

468
        pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
×
469
        continue;
×
470
      }
471

472
      qDestroyStmtDataBlock(pBlocks);
×
473
      STMT_ERR_RET(taosHashRemove(pStmt->exec.pBlockHash, key, keyLen));
×
474

475
      pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
×
476
    }
477

478
    if (keepTable) {
×
479
      return TSDB_CODE_SUCCESS;
×
480
    }
481

482
    taosHashCleanup(pStmt->exec.pBlockHash);
×
483
    pStmt->exec.pBlockHash = NULL;
×
484

485
    tDestroySubmitTbData(pStmt->exec.pCurrTbData, TSDB_MSG_FLG_ENCODE);
×
486
    taosMemoryFreeClear(pStmt->exec.pCurrTbData);
×
487
  }
488

489
  STMT_ERR_RET(stmtCleanBindInfo(pStmt));
×
490

491
  return TSDB_CODE_SUCCESS;
×
492
}
493

494
void stmtFreeTbBuf(void* buf) {
×
495
  void* pBuf = *(void**)buf;
×
496
  taosMemoryFree(pBuf);
×
497
}
×
498

499
void stmtFreeTbCols(void* buf) {
×
500
  SArray* pCols = *(SArray**)buf;
×
501
  taosArrayDestroy(pCols);
×
502
}
×
503

504
int32_t stmtCleanSQLInfo(STscStmt* pStmt) {
×
505
  STMT_DLOG_E("start to free SQL info");
×
506

507
  taosMemoryFree(pStmt->sql.pBindInfo);
×
508
  taosMemoryFree(pStmt->sql.queryRes.fields);
×
509
  taosMemoryFree(pStmt->sql.queryRes.userFields);
×
510
  taosMemoryFree(pStmt->sql.sqlStr);
×
511
  qDestroyQuery(pStmt->sql.pQuery);
×
512
  taosArrayDestroy(pStmt->sql.nodeList);
×
513
  taosHashCleanup(pStmt->sql.pVgHash);
×
514
  pStmt->sql.pVgHash = NULL;
×
515

516
  void* pIter = taosHashIterate(pStmt->sql.pTableCache, NULL);
×
517
  while (pIter) {
×
518
    SStmtTableCache* pCache = (SStmtTableCache*)pIter;
×
519

520
    qDestroyStmtDataBlock(pCache->pDataCtx);
×
521
    qDestroyBoundColInfo(pCache->boundTags);
×
522
    taosMemoryFreeClear(pCache->boundTags);
×
523

524
    pIter = taosHashIterate(pStmt->sql.pTableCache, pIter);
×
525
  }
526
  taosHashCleanup(pStmt->sql.pTableCache);
×
527
  pStmt->sql.pTableCache = NULL;
×
528

529
  STMT_ERR_RET(stmtCleanExecInfo(pStmt, false, true));
×
530
  STMT_ERR_RET(stmtCleanBindInfo(pStmt));
×
531

532
  taos_free_result(pStmt->sql.siInfo.pRequest);
×
533
  taosHashCleanup(pStmt->sql.siInfo.pVgroupHash);
×
534
  tSimpleHashCleanup(pStmt->sql.siInfo.pTableHash);
×
535
  tSimpleHashCleanup(pStmt->sql.siInfo.pTableRowDataHash);
×
536
  taosArrayDestroyEx(pStmt->sql.siInfo.tbBuf.pBufList, stmtFreeTbBuf);
×
537
  taosMemoryFree(pStmt->sql.siInfo.pTSchema);
×
538
  qDestroyStmtDataBlock(pStmt->sql.siInfo.pDataCtx);
×
539
  taosArrayDestroyEx(pStmt->sql.siInfo.pTableCols, stmtFreeTbCols);
×
540

541
  (void)memset(&pStmt->sql, 0, sizeof(pStmt->sql));
×
542
  pStmt->sql.siInfo.tableColsReady = true;
×
543

544
  STMT_DLOG_E("end to free SQL info");
×
545

546
  return TSDB_CODE_SUCCESS;
×
547
}
548

549
int32_t stmtTryAddTableVgroupInfo(STscStmt* pStmt, int32_t* vgId) {
×
550
  if (*vgId >= 0 && taosHashGet(pStmt->sql.pVgHash, (const char*)vgId, sizeof(*vgId))) {
×
551
    return TSDB_CODE_SUCCESS;
×
552
  }
553

554
  SVgroupInfo      vgInfo = {0};
×
555
  SRequestConnInfo conn = {.pTrans = pStmt->taos->pAppInfo->pTransporter,
×
556
                           .requestId = pStmt->exec.pRequest->requestId,
×
557
                           .requestObjRefId = pStmt->exec.pRequest->self,
×
558
                           .mgmtEps = getEpSet_s(&pStmt->taos->pAppInfo->mgmtEp)};
×
559

560
  int32_t code = catalogGetTableHashVgroup(pStmt->pCatalog, &conn, &pStmt->bInfo.sname, &vgInfo);
×
561
  if (TSDB_CODE_SUCCESS != code) {
×
562
    return code;
×
563
  }
564

565
  code =
566
      taosHashPut(pStmt->sql.pVgHash, (const char*)&vgInfo.vgId, sizeof(vgInfo.vgId), (char*)&vgInfo, sizeof(vgInfo));
×
567
  if (TSDB_CODE_SUCCESS != code) {
×
568
    return code;
×
569
  }
570

571
  *vgId = vgInfo.vgId;
×
572

573
  return TSDB_CODE_SUCCESS;
×
574
}
575

576
int32_t stmtRebuildDataBlock(STscStmt* pStmt, STableDataCxt* pDataBlock, STableDataCxt** newBlock, uint64_t uid,
×
577
                             uint64_t suid, int32_t vgId) {
578
  STMT_ERR_RET(stmtTryAddTableVgroupInfo(pStmt, &vgId));
×
579
  STMT_ERR_RET(qRebuildStmtDataBlock(newBlock, pDataBlock, uid, suid, vgId, pStmt->sql.autoCreateTbl));
×
580

581
  STMT_DLOG("uid:%" PRId64 ", rebuild table data context, vgId:%d", uid, vgId);
×
582

583
  return TSDB_CODE_SUCCESS;
×
584
}
585

586
int32_t stmtGetFromCache(STscStmt* pStmt) {
×
587
  if (pStmt->sql.stbInterlaceMode && pStmt->sql.siInfo.pDataCtx) {
×
588
    pStmt->bInfo.needParse = false;
×
589
    pStmt->bInfo.inExecCache = false;
×
590
    return TSDB_CODE_SUCCESS;
×
591
  }
592

593
  pStmt->bInfo.needParse = true;
×
594
  pStmt->bInfo.inExecCache = false;
×
595

596
  STableDataCxt** pCxtInExec = taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
×
597
  if (pCxtInExec) {
×
598
    pStmt->bInfo.needParse = false;
×
599
    pStmt->bInfo.inExecCache = true;
×
600

601
    pStmt->exec.pCurrBlock = *pCxtInExec;
×
602

603
    if (pStmt->sql.autoCreateTbl) {
×
604
      tscDebug("reuse stmt block for tb %s in execBlock", pStmt->bInfo.tbFName);
×
605
      return TSDB_CODE_SUCCESS;
×
606
    }
607
  }
608

609
  if (NULL == pStmt->pCatalog) {
×
610
    STMT_ERR_RET(catalogGetHandle(pStmt->taos->pAppInfo->clusterId, &pStmt->pCatalog));
×
611
    pStmt->sql.siInfo.pCatalog = pStmt->pCatalog;
×
612
  }
613

614
  if (NULL == pStmt->sql.pTableCache || taosHashGetSize(pStmt->sql.pTableCache) <= 0) {
×
615
    if (pStmt->bInfo.inExecCache) {
×
616
      pStmt->bInfo.needParse = false;
×
617
      tscDebug("reuse stmt block for tb %s in execBlock", pStmt->bInfo.tbFName);
×
618
      return TSDB_CODE_SUCCESS;
×
619
    }
620

621
    tscDebug("no stmt block cache for tb %s", pStmt->bInfo.tbFName);
×
622
    return TSDB_CODE_SUCCESS;
×
623
  }
624

625
  if (pStmt->sql.autoCreateTbl) {
×
626
    SStmtTableCache* pCache = taosHashGet(pStmt->sql.pTableCache, &pStmt->bInfo.tbSuid, sizeof(pStmt->bInfo.tbSuid));
×
627
    if (pCache) {
×
628
      pStmt->bInfo.needParse = false;
×
629
      pStmt->bInfo.tbUid = 0;
×
630

631
      STableDataCxt* pNewBlock = NULL;
×
632
      STMT_ERR_RET(stmtRebuildDataBlock(pStmt, pCache->pDataCtx, &pNewBlock, 0, pStmt->bInfo.tbSuid, -1));
×
633

634
      if (taosHashPut(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName), &pNewBlock,
×
635
                      POINTER_BYTES)) {
636
        STMT_ERR_RET(terrno);
×
637
      }
638

639
      pStmt->exec.pCurrBlock = pNewBlock;
×
640

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

643
      return TSDB_CODE_SUCCESS;
×
644
    }
645

646
    STMT_RET(stmtCleanBindInfo(pStmt));
×
647
  }
648

649
  uint64_t uid, suid;
650
  int32_t  vgId;
651
  int8_t   tableType;
652

653
  STableMeta*      pTableMeta = NULL;
×
654
  SRequestConnInfo conn = {.pTrans = pStmt->taos->pAppInfo->pTransporter,
×
655
                           .requestId = pStmt->exec.pRequest->requestId,
×
656
                           .requestObjRefId = pStmt->exec.pRequest->self,
×
657
                           .mgmtEps = getEpSet_s(&pStmt->taos->pAppInfo->mgmtEp)};
×
658
  int32_t          code = catalogGetTableMeta(pStmt->pCatalog, &conn, &pStmt->bInfo.sname, &pTableMeta);
×
659

660
  pStmt->stat.ctgGetTbMetaNum++;
×
661

662
  if (TSDB_CODE_PAR_TABLE_NOT_EXIST == code) {
×
663
    tscDebug("tb %s not exist", pStmt->bInfo.tbFName);
×
664
    STMT_ERR_RET(stmtCleanBindInfo(pStmt));
×
665

666
    STMT_ERR_RET(code);
×
667
  }
668

669
  STMT_ERR_RET(code);
×
670

671
  uid = pTableMeta->uid;
×
672
  suid = pTableMeta->suid;
×
673
  tableType = pTableMeta->tableType;
×
674
  pStmt->bInfo.tbVgId = pTableMeta->vgId;
×
675
  vgId = pTableMeta->vgId;
×
676

677
  taosMemoryFree(pTableMeta);
×
678

679
  uint64_t cacheUid = (TSDB_CHILD_TABLE == tableType) ? suid : uid;
×
680

681
  if (uid == pStmt->bInfo.tbUid) {
×
682
    pStmt->bInfo.needParse = false;
×
683

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

686
    return TSDB_CODE_SUCCESS;
×
687
  }
688

689
  if (pStmt->bInfo.inExecCache) {
×
690
    SStmtTableCache* pCache = taosHashGet(pStmt->sql.pTableCache, &cacheUid, sizeof(cacheUid));
×
691
    if (NULL == pCache) {
×
692
      tscError("table [%s, %" PRIx64 ", %" PRIx64 "] found in exec blockHash, but not in sql blockHash",
×
693
               pStmt->bInfo.tbFName, uid, cacheUid);
694

695
      STMT_ERR_RET(TSDB_CODE_APP_ERROR);
×
696
    }
697

698
    pStmt->bInfo.needParse = false;
×
699

700
    pStmt->bInfo.tbUid = uid;
×
701
    pStmt->bInfo.tbSuid = suid;
×
702
    pStmt->bInfo.tbType = tableType;
×
703
    pStmt->bInfo.boundTags = pCache->boundTags;
×
704
    pStmt->bInfo.tagsCached = true;
×
705

706
    tscDebug("tb %s in execBlock list, set to current", pStmt->bInfo.tbFName);
×
707

708
    return TSDB_CODE_SUCCESS;
×
709
  }
710

711
  SStmtTableCache* pCache = taosHashGet(pStmt->sql.pTableCache, &cacheUid, sizeof(cacheUid));
×
712
  if (pCache) {
×
713
    pStmt->bInfo.needParse = false;
×
714

715
    pStmt->bInfo.tbUid = uid;
×
716
    pStmt->bInfo.tbSuid = suid;
×
717
    pStmt->bInfo.tbType = tableType;
×
718
    pStmt->bInfo.boundTags = pCache->boundTags;
×
719
    pStmt->bInfo.tagsCached = true;
×
720

721
    STableDataCxt* pNewBlock = NULL;
×
722
    STMT_ERR_RET(stmtRebuildDataBlock(pStmt, pCache->pDataCtx, &pNewBlock, uid, suid, vgId));
×
723

724
    if (taosHashPut(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName), &pNewBlock,
×
725
                    POINTER_BYTES)) {
726
      STMT_ERR_RET(terrno);
×
727
    }
728

729
    pStmt->exec.pCurrBlock = pNewBlock;
×
730

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

733
    return TSDB_CODE_SUCCESS;
×
734
  }
735

736
  STMT_ERR_RET(stmtCleanBindInfo(pStmt));
×
737

738
  return TSDB_CODE_SUCCESS;
×
739
}
740

741
int32_t stmtResetStmt(STscStmt* pStmt) {
×
742
  STMT_ERR_RET(stmtCleanSQLInfo(pStmt));
×
743

744
  pStmt->sql.pTableCache = taosHashInit(100, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
×
745
  if (NULL == pStmt->sql.pTableCache) {
×
746
    STMT_ERR_RET(terrno);
×
747
  }
748

749
  if (pStmt->sql.siInfo.pTableRowDataHash) {
×
750
    tSimpleHashClear(pStmt->sql.siInfo.pTableRowDataHash);
×
751
  }
752

753
  pStmt->sql.status = STMT_INIT;
×
754

755
  return TSDB_CODE_SUCCESS;
×
756
}
757

758
int32_t stmtAsyncOutput(STscStmt* pStmt, void* param) {
×
759
  SStmtQNode* pParam = (SStmtQNode*)param;
×
760

761
  if (pParam->restoreTbCols) {
×
762
    for (int32_t i = 0; i < pStmt->sql.siInfo.pTableColsIdx; ++i) {
×
763
      SArray** p = (SArray**)TARRAY_GET_ELEM(pStmt->sql.siInfo.pTableCols, i);
×
764
      *p = taosArrayInit(20, POINTER_BYTES);
×
765
      if (*p == NULL) {
×
766
        STMT_ERR_RET(terrno);
×
767
      }
768
    }
769

770
    atomic_store_8((int8_t*)&pStmt->sql.siInfo.tableColsReady, true);
×
771
  } else {
772
    STMT_ERR_RET(qAppendStmtTableOutput(pStmt->sql.pQuery, pStmt->sql.pVgHash, &pParam->tblData, pStmt->exec.pCurrBlock,
×
773
                                        &pStmt->sql.siInfo, NULL));
774

775
    // taosMemoryFree(pParam->pTbData);
776

777
    (void)atomic_sub_fetch_64(&pStmt->sql.siInfo.tbRemainNum, 1);
×
778
  }
779
  return TSDB_CODE_SUCCESS;
×
780
}
781

782
void* stmtBindThreadFunc(void* param) {
×
783
  setThreadName("stmtBind");
×
784

785
  qInfo("stmt bind thread started");
×
786

787
  STscStmt* pStmt = (STscStmt*)param;
×
788

789
  while (true) {
×
790
    if (pStmt->queue.stopQueue) {
×
791
      break;
×
792
    }
793

794
    SStmtQNode* asyncParam = NULL;
×
795
    if (!stmtDequeue(pStmt, &asyncParam)) {
×
796
      continue;
×
797
    }
798

799
    int ret = stmtAsyncOutput(pStmt, asyncParam);
×
800
    if (ret != 0) {
×
801
      qError("stmtAsyncOutput failed, reason:%s", tstrerror(ret));
×
802
    }
803
  }
804

805
  qInfo("stmt bind thread stopped");
×
806

807
  return NULL;
×
808
}
809

810
int32_t stmtStartBindThread(STscStmt* pStmt) {
×
811
  TdThreadAttr thAttr;
812
  if (taosThreadAttrInit(&thAttr) != 0) {
×
813
    return TSDB_CODE_TSC_INTERNAL_ERROR;
×
814
  }
815
  if (taosThreadAttrSetDetachState(&thAttr, PTHREAD_CREATE_JOINABLE) != 0) {
×
816
    return TSDB_CODE_TSC_INTERNAL_ERROR;
×
817
  }
818

819
  if (taosThreadCreate(&pStmt->bindThread, &thAttr, stmtBindThreadFunc, pStmt) != 0) {
×
820
    terrno = TAOS_SYSTEM_ERROR(ERRNO);
×
821
    STMT_ERR_RET(terrno);
×
822
  }
823

824
  pStmt->bindThreadInUse = true;
×
825

826
  (void)taosThreadAttrDestroy(&thAttr);
×
827
  return TSDB_CODE_SUCCESS;
×
828
}
829

830
int32_t stmtInitQueue(STscStmt* pStmt) {
×
831
  (void)taosThreadCondInit(&pStmt->queue.waitCond, NULL);
×
832
  (void)taosThreadMutexInit(&pStmt->queue.mutex, NULL);
×
833
  STMT_ERR_RET(stmtAllocQNodeFromBuf(&pStmt->sql.siInfo.tbBuf, (void**)&pStmt->queue.head));
×
834
  pStmt->queue.tail = pStmt->queue.head;
×
835

836
  return TSDB_CODE_SUCCESS;
×
837
}
838

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

851
  if (taosArrayPush(pTblBuf->pBufList, &buff) == NULL) {
×
852
    return terrno;
×
853
  }
854

855
  pTblBuf->pCurBuff = buff;
×
856
  pTblBuf->buffIdx = 1;
×
857
  pTblBuf->buffOffset = 0;
×
858

859
  return TSDB_CODE_SUCCESS;
×
860
}
861

862
TAOS_STMT* stmtInit(STscObj* taos, int64_t reqid, TAOS_STMT_OPTIONS* pOptions) {
×
863
  STscObj*  pObj = (STscObj*)taos;
×
864
  STscStmt* pStmt = NULL;
×
865
  int32_t   code = 0;
×
866

867
  pStmt = taosMemoryCalloc(1, sizeof(STscStmt));
×
868
  if (NULL == pStmt) {
×
869
    return NULL;
×
870
  }
871

872
  pStmt->sql.pTableCache = taosHashInit(100, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
×
873
  if (NULL == pStmt->sql.pTableCache) {
×
874
    taosMemoryFree(pStmt);
×
875
    return NULL;
×
876
  }
877

878
  pStmt->taos = pObj;
×
879
  pStmt->bInfo.needParse = true;
×
880
  pStmt->sql.status = STMT_INIT;
×
881
  pStmt->reqid = reqid;
×
882
  pStmt->errCode = TSDB_CODE_SUCCESS;
×
883

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

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

913
    code = stmtInitTableBuf(&pStmt->sql.siInfo.tbBuf);
×
914
    if (TSDB_CODE_SUCCESS == code) {
×
915
      code = stmtInitQueue(pStmt);
×
916
    }
917
    if (TSDB_CODE_SUCCESS == code) {
×
918
      code = stmtStartBindThread(pStmt);
×
919
    }
920
    if (TSDB_CODE_SUCCESS != code) {
×
921
      terrno = code;
×
922
      (void)stmtClose(pStmt);
×
923
      return NULL;
×
924
    }
925
  }
926

927
  pStmt->sql.siInfo.tableColsReady = true;
×
928

929
  STMT_LOG_SEQ(STMT_INIT);
×
930

931
  tscDebug("stmt:%p initialized", pStmt);
×
932

933
  return pStmt;
×
934
}
935

936
int stmtPrepare(TAOS_STMT* stmt, const char* sql, unsigned long length) {
×
937
  STscStmt* pStmt = (STscStmt*)stmt;
×
938

939
  STMT_DLOG_E("start to prepare");
×
940

941
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
×
942
    return pStmt->errCode;
×
943
  }
944

945
  if (pStmt->sql.status >= STMT_PREPARE) {
×
946
    STMT_ERR_RET(stmtResetStmt(pStmt));
×
947
  }
948

949
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_PREPARE));
×
950

951
  if (length <= 0) {
×
952
    length = strlen(sql);
×
953
  }
954

955
  pStmt->sql.sqlStr = taosStrndup(sql, length);
×
956
  if (!pStmt->sql.sqlStr) {
×
957
    return terrno;
×
958
  }
959
  pStmt->sql.sqlLen = length;
×
960
  pStmt->sql.stbInterlaceMode = pStmt->stbInterlaceMode;
×
961

962
  char* dbName = NULL;
×
963
  if (qParseDbName(sql, length, &dbName)) {
×
964
    STMT_ERR_RET(stmtSetDbName(stmt, dbName));
×
965
    taosMemoryFreeClear(dbName);
×
966
  }
967

968
  return TSDB_CODE_SUCCESS;
×
969
}
970

971
int32_t stmtInitStbInterlaceTableInfo(STscStmt* pStmt) {
×
972
  STableDataCxt** pSrc = taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
×
973
  if (!pSrc) {
×
974
    return terrno;
×
975
  }
976
  STableDataCxt* pDst = NULL;
×
977

978
  STMT_ERR_RET(qCloneStmtDataBlock(&pDst, *pSrc, true));
×
979
  pStmt->sql.siInfo.pDataCtx = pDst;
×
980

981
  SArray* pTblCols = NULL;
×
982
  for (int32_t i = 0; i < STMT_TABLE_COLS_NUM; i++) {
×
983
    pTblCols = taosArrayInit(20, POINTER_BYTES);
×
984
    if (NULL == pTblCols) {
×
985
      return terrno;
×
986
    }
987

988
    if (taosArrayPush(pStmt->sql.siInfo.pTableCols, &pTblCols) == NULL) {
×
989
      return terrno;
×
990
    }
991
  }
992

993
  pStmt->sql.siInfo.boundTags = pStmt->bInfo.boundTags;
×
994

995
  return TSDB_CODE_SUCCESS;
×
996
}
997

998
int stmtSetDbName(TAOS_STMT* stmt, const char* dbName) {
×
999
  STscStmt* pStmt = (STscStmt*)stmt;
×
1000

1001
  STMT_DLOG("start to set dbName:%s", dbName);
×
1002

1003
  STMT_ERR_RET(stmtCreateRequest(pStmt));
×
1004

1005
  // The SQL statement specifies a database name, overriding the previously specified database
1006
  taosMemoryFreeClear(pStmt->exec.pRequest->pDb);
×
1007
  pStmt->exec.pRequest->pDb = taosStrdup(dbName);
×
1008
  if (pStmt->exec.pRequest->pDb == NULL) {
×
1009
    return terrno;
×
1010
  }
1011
  return TSDB_CODE_SUCCESS;
×
1012
}
1013

1014
int stmtSetTbName(TAOS_STMT* stmt, const char* tbName) {
×
1015
  STscStmt* pStmt = (STscStmt*)stmt;
×
1016

1017
  int64_t startUs = taosGetTimestampUs();
×
1018

1019
  STMT_DLOG("start to set tbName:%s", tbName);
×
1020

1021
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
×
1022
    return pStmt->errCode;
×
1023
  }
1024

1025
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_SETTBNAME));
×
1026

1027
  int32_t insert = 0;
×
1028
  STMT_ERR_RET(stmtIsInsert(stmt, &insert));
×
1029
  if (0 == insert) {
×
1030
    tscError("set tb name not available for none insert statement");
×
1031
    STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR);
×
1032
  }
1033

1034
  if (!pStmt->sql.stbInterlaceMode || NULL == pStmt->sql.siInfo.pDataCtx) {
×
1035
    STMT_ERR_RET(stmtCreateRequest(pStmt));
×
1036

1037
    STMT_ERR_RET(qCreateSName(&pStmt->bInfo.sname, tbName, pStmt->taos->acctId, pStmt->exec.pRequest->pDb,
×
1038
                              pStmt->exec.pRequest->msgBuf, pStmt->exec.pRequest->msgBufLen));
1039
    STMT_ERR_RET(tNameExtractFullName(&pStmt->bInfo.sname, pStmt->bInfo.tbFName));
×
1040

1041
    STMT_ERR_RET(stmtGetFromCache(pStmt));
×
1042

1043
    if (pStmt->bInfo.needParse) {
×
1044
      tstrncpy(pStmt->bInfo.tbName, tbName, sizeof(pStmt->bInfo.tbName));
×
1045
      pStmt->bInfo.tbName[sizeof(pStmt->bInfo.tbName) - 1] = 0;
×
1046

1047
      STMT_ERR_RET(stmtParseSql(pStmt));
×
1048
    }
1049
  } else {
1050
    tstrncpy(pStmt->bInfo.tbName, tbName, sizeof(pStmt->bInfo.tbName));
×
1051
    pStmt->bInfo.tbName[sizeof(pStmt->bInfo.tbName) - 1] = 0;
×
1052
    pStmt->exec.pRequest->requestId++;
×
1053
    pStmt->bInfo.needParse = false;
×
1054
  }
1055

1056
  if (pStmt->sql.stbInterlaceMode && NULL == pStmt->sql.siInfo.pDataCtx) {
×
1057
    STMT_ERR_RET(stmtInitStbInterlaceTableInfo(pStmt));
×
1058
  }
1059

1060
  int64_t startUs2 = taosGetTimestampUs();
×
1061
  pStmt->stat.setTbNameUs += startUs2 - startUs;
×
1062

1063
  return TSDB_CODE_SUCCESS;
×
1064
}
1065

1066
int stmtSetTbTags(TAOS_STMT* stmt, TAOS_MULTI_BIND* tags) {
×
1067
  STscStmt* pStmt = (STscStmt*)stmt;
×
1068

1069
  STMT_DLOG_E("start to set tbTags");
×
1070

1071
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
×
1072
    return pStmt->errCode;
×
1073
  }
1074

1075
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_SETTAGS));
×
1076

1077
  SBoundColInfo* tags_info = (SBoundColInfo*)pStmt->bInfo.boundTags;
×
1078
  if (tags_info->numOfBound <= 0 || tags_info->numOfCols <= 0) {
×
1079
    tscWarn("no tags bound in sql, will not bound tags");
×
1080
    return TSDB_CODE_SUCCESS;
×
1081
  }
1082

1083
  if (pStmt->bInfo.inExecCache) {
×
1084
    return TSDB_CODE_SUCCESS;
×
1085
  }
1086

1087
  STableDataCxt** pDataBlock =
1088
      (STableDataCxt**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
×
1089
  if (NULL == pDataBlock) {
×
1090
    tscError("table %s not found in exec blockHash", pStmt->bInfo.tbFName);
×
1091
    STMT_ERR_RET(TSDB_CODE_APP_ERROR);
×
1092
  }
1093

1094
  tscDebug("start to bind stmt tag values");
×
1095
  STMT_ERR_RET(qBindStmtTagsValue(*pDataBlock, pStmt->bInfo.boundTags, pStmt->bInfo.tbSuid, pStmt->bInfo.stbFName,
×
1096
                                  pStmt->bInfo.sname.tname, tags, pStmt->exec.pRequest->msgBuf,
1097
                                  pStmt->exec.pRequest->msgBufLen, pStmt->taos->optionInfo.charsetCxt));
1098

1099
  return TSDB_CODE_SUCCESS;
×
1100
}
1101

1102
int stmtFetchTagFields(STscStmt* pStmt, int32_t* fieldNum, TAOS_FIELD_E** fields) {
×
1103
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
×
1104
    return pStmt->errCode;
×
1105
  }
1106

1107
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
×
1108
    tscError("invalid operation to get query tag fileds");
×
1109
    STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR);
×
1110
  }
1111

1112
  STableDataCxt** pDataBlock =
1113
      (STableDataCxt**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
×
1114
  if (NULL == pDataBlock) {
×
1115
    tscError("table %s not found in exec blockHash", pStmt->bInfo.tbFName);
×
1116
    STMT_ERR_RET(TSDB_CODE_APP_ERROR);
×
1117
  }
1118

1119
  STMT_ERR_RET(qBuildStmtTagFields(*pDataBlock, pStmt->bInfo.boundTags, fieldNum, fields));
×
1120

1121
  return TSDB_CODE_SUCCESS;
×
1122
}
1123

1124
int stmtFetchColFields(STscStmt* pStmt, int32_t* fieldNum, TAOS_FIELD_E** fields) {
×
1125
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
×
1126
    return pStmt->errCode;
×
1127
  }
1128

1129
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
×
1130
    tscError("invalid operation to get query column fileds");
×
1131
    STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR);
×
1132
  }
1133

1134
  STableDataCxt** pDataBlock = NULL;
×
1135

1136
  if (pStmt->sql.stbInterlaceMode) {
×
1137
    pDataBlock = &pStmt->sql.siInfo.pDataCtx;
×
1138
  } else {
1139
    pDataBlock =
1140
        (STableDataCxt**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
×
1141
    if (NULL == pDataBlock) {
×
1142
      tscError("table %s not found in exec blockHash", pStmt->bInfo.tbFName);
×
1143
      STMT_ERR_RET(TSDB_CODE_APP_ERROR);
×
1144
    }
1145
  }
1146

1147
  STMT_ERR_RET(qBuildStmtColFields(*pDataBlock, fieldNum, fields));
×
1148

1149
  return TSDB_CODE_SUCCESS;
×
1150
}
1151

1152
/*
1153
SArray* stmtGetFreeCol(STscStmt* pStmt, int32_t* idx) {
1154
  while (true) {
1155
    if (pStmt->exec.smInfo.pColIdx >= STMT_COL_BUF_SIZE) {
1156
      pStmt->exec.smInfo.pColIdx = 0;
1157
    }
1158

1159
    if ((pStmt->exec.smInfo.pColIdx + 1) == atomic_load_32(&pStmt->exec.smInfo.pColFreeIdx)) {
1160
      taosUsleep(1);
1161
      continue;
1162
    }
1163

1164
    *idx = pStmt->exec.smInfo.pColIdx;
1165
    return pStmt->exec.smInfo.pCols[pStmt->exec.smInfo.pColIdx++];
1166
  }
1167
}
1168
*/
1169

1170
int32_t stmtAppendTablePostHandle(STscStmt* pStmt, SStmtQNode* param) {
×
1171
  if (NULL == pStmt->sql.siInfo.pVgroupHash) {
×
1172
    pStmt->sql.siInfo.pVgroupHash =
×
1173
        taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_ENTRY_LOCK);
×
1174
  }
1175
  if (NULL == pStmt->sql.siInfo.pVgroupList) {
×
1176
    pStmt->sql.siInfo.pVgroupList = taosArrayInit(64, POINTER_BYTES);
×
1177
  }
1178

1179
  if (NULL == pStmt->sql.siInfo.pRequest) {
×
1180
    STMT_ERR_RET(buildRequest(pStmt->taos->id, pStmt->sql.sqlStr, pStmt->sql.sqlLen, NULL, false,
×
1181
                              (SRequestObj**)&pStmt->sql.siInfo.pRequest, pStmt->reqid));
1182

1183
    if (pStmt->reqid != 0) {
×
1184
      pStmt->reqid++;
×
1185
    }
1186
    pStmt->exec.pRequest->syncQuery = true;
×
1187

1188
    pStmt->sql.siInfo.requestId = ((SRequestObj*)pStmt->sql.siInfo.pRequest)->requestId;
×
1189
    pStmt->sql.siInfo.requestSelf = ((SRequestObj*)pStmt->sql.siInfo.pRequest)->self;
×
1190
  }
1191

1192
  if (!pStmt->sql.siInfo.tbFromHash && pStmt->sql.siInfo.firstName[0] &&
×
1193
      0 == strcmp(pStmt->sql.siInfo.firstName, pStmt->bInfo.tbName)) {
×
1194
    pStmt->sql.siInfo.tbFromHash = true;
×
1195
  }
1196

1197
  if (0 == pStmt->sql.siInfo.firstName[0]) {
×
1198
    tstrncpy(pStmt->sql.siInfo.firstName, pStmt->bInfo.tbName, TSDB_TABLE_NAME_LEN);
×
1199
  }
1200

1201
  param->tblData.getFromHash = pStmt->sql.siInfo.tbFromHash;
×
1202
  param->next = NULL;
×
1203

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

1206
  stmtEnqueue(pStmt, param);
×
1207

1208
  return TSDB_CODE_SUCCESS;
×
1209
}
1210

1211
static FORCE_INLINE int32_t stmtGetTableColsFromCache(STscStmt* pStmt, SArray** pTableCols) {
1212
  while (true) {
1213
    if (pStmt->sql.siInfo.pTableColsIdx < taosArrayGetSize(pStmt->sql.siInfo.pTableCols)) {
×
1214
      *pTableCols = (SArray*)taosArrayGetP(pStmt->sql.siInfo.pTableCols, pStmt->sql.siInfo.pTableColsIdx++);
×
1215
      break;
×
1216
    } else {
1217
      SArray* pTblCols = NULL;
×
1218
      for (int32_t i = 0; i < 100; i++) {
×
1219
        pTblCols = taosArrayInit(20, POINTER_BYTES);
×
1220
        if (NULL == pTblCols) {
×
1221
          return terrno;
×
1222
        }
1223

1224
        if (taosArrayPush(pStmt->sql.siInfo.pTableCols, &pTblCols) == NULL) {
×
1225
          return terrno;
×
1226
        }
1227
      }
1228
    }
1229
  }
1230

1231
  return TSDB_CODE_SUCCESS;
×
1232
}
1233

1234
int stmtBindBatch(TAOS_STMT* stmt, TAOS_MULTI_BIND* bind, int32_t colIdx) {
×
1235
  STscStmt* pStmt = (STscStmt*)stmt;
×
1236
  int32_t   code = 0;
×
1237

1238
  int64_t startUs = taosGetTimestampUs();
×
1239

1240
  STMT_DLOG("start to bind stmt data, colIdx:%d", colIdx);
×
1241

1242
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
×
1243
    return pStmt->errCode;
×
1244
  }
1245

1246
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_BIND));
×
1247

1248
  if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 &&
×
1249
      STMT_TYPE_MULTI_INSERT != pStmt->sql.type) {
×
1250
    pStmt->bInfo.needParse = false;
×
1251
  }
1252

1253
  if (pStmt->exec.pRequest && STMT_TYPE_QUERY == pStmt->sql.type && pStmt->sql.runTimes) {
×
1254
    taos_free_result(pStmt->exec.pRequest);
×
1255
    pStmt->exec.pRequest = NULL;
×
1256
  }
1257

1258
  STMT_ERR_RET(stmtCreateRequest(pStmt));
×
1259

1260
  if (pStmt->bInfo.needParse) {
×
1261
    STMT_ERR_RET(stmtParseSql(pStmt));
×
1262
  }
1263

1264
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
×
1265
    STMT_ERR_RET(qStmtBindParams(pStmt->sql.pQuery, bind, colIdx, pStmt->taos->optionInfo.charsetCxt));
×
1266

1267
    SParseContext ctx = {.requestId = pStmt->exec.pRequest->requestId,
×
1268
                         .acctId = pStmt->taos->acctId,
×
1269
                         .db = pStmt->exec.pRequest->pDb,
×
1270
                         .topicQuery = false,
1271
                         .pSql = pStmt->sql.sqlStr,
×
1272
                         .sqlLen = pStmt->sql.sqlLen,
×
1273
                         .pMsg = pStmt->exec.pRequest->msgBuf,
×
1274
                         .msgLen = ERROR_MSG_BUF_DEFAULT_SIZE,
1275
                         .pTransporter = pStmt->taos->pAppInfo->pTransporter,
×
1276
                         .pStmtCb = NULL,
1277
                         .pUser = pStmt->taos->user,
×
1278
                         .setQueryFp = setQueryRequest};
1279

1280
    ctx.mgmtEpSet = getEpSet_s(&pStmt->taos->pAppInfo->mgmtEp);
×
1281
    STMT_ERR_RET(catalogGetHandle(pStmt->taos->pAppInfo->clusterId, &ctx.pCatalog));
×
1282

1283
    STMT_ERR_RET(qStmtParseQuerySql(&ctx, pStmt->sql.pQuery));
×
1284

1285
    if (pStmt->sql.pQuery->haveResultSet) {
×
1286
      STMT_ERR_RET(setResSchemaInfo(&pStmt->exec.pRequest->body.resInfo, pStmt->sql.pQuery->pResSchema,
×
1287
                                    pStmt->sql.pQuery->numOfResCols, pStmt->sql.pQuery->pResExtSchema, true));
1288
      taosMemoryFreeClear(pStmt->sql.pQuery->pResSchema);
×
1289
      taosMemoryFreeClear(pStmt->sql.pQuery->pResExtSchema);
×
1290
      setResPrecision(&pStmt->exec.pRequest->body.resInfo, pStmt->sql.pQuery->precision);
×
1291
    }
1292

1293
    TSWAP(pStmt->exec.pRequest->dbList, pStmt->sql.pQuery->pDbList);
×
1294
    TSWAP(pStmt->exec.pRequest->tableList, pStmt->sql.pQuery->pTableList);
×
1295
    TSWAP(pStmt->exec.pRequest->targetTableList, pStmt->sql.pQuery->pTargetTableList);
×
1296

1297
    // if (STMT_TYPE_QUERY == pStmt->sql.queryRes) {
1298
    //   STMT_ERR_RET(stmtRestoreQueryFields(pStmt));
1299
    // }
1300

1301
    // STMT_ERR_RET(stmtBackupQueryFields(pStmt));
1302

1303
    return TSDB_CODE_SUCCESS;
×
1304
  }
1305

1306
  if (pStmt->sql.stbInterlaceMode && NULL == pStmt->sql.siInfo.pDataCtx) {
×
1307
    STMT_ERR_RET(stmtInitStbInterlaceTableInfo(pStmt));
×
1308
  }
1309

1310
  STableDataCxt** pDataBlock = NULL;
×
1311

1312
  if (pStmt->exec.pCurrBlock) {
×
1313
    pDataBlock = &pStmt->exec.pCurrBlock;
×
1314
  } else {
1315
    pDataBlock =
1316
        (STableDataCxt**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
×
1317
    if (NULL == pDataBlock) {
×
1318
      tscError("table %s not found in exec blockHash", pStmt->bInfo.tbFName);
×
1319
      STMT_ERR_RET(TSDB_CODE_TSC_STMT_CACHE_ERROR);
×
1320
    }
1321
    pStmt->exec.pCurrBlock = *pDataBlock;
×
1322
    if (pStmt->sql.stbInterlaceMode) {
×
1323
      taosArrayDestroy(pStmt->exec.pCurrBlock->pData->aCol);
×
1324
      pStmt->exec.pCurrBlock->pData->aCol = NULL;
×
1325
    }
1326
  }
1327

1328
  int64_t startUs2 = taosGetTimestampUs();
×
1329
  pStmt->stat.bindDataUs1 += startUs2 - startUs;
×
1330

1331
  SStmtQNode* param = NULL;
×
1332
  if (pStmt->sql.stbInterlaceMode) {
×
1333
    STMT_ERR_RET(stmtAllocQNodeFromBuf(&pStmt->sql.siInfo.tbBuf, (void**)&param));
×
1334
    STMT_ERR_RET(stmtGetTableColsFromCache(pStmt, &param->tblData.aCol));
×
1335
    taosArrayClear(param->tblData.aCol);
×
1336

1337
    // param->tblData.aCol = taosArrayInit(20, POINTER_BYTES);
1338

1339
    param->restoreTbCols = false;
×
1340
    tstrncpy(param->tblData.tbName, pStmt->bInfo.tbName, TSDB_TABLE_NAME_LEN);
×
1341
  }
1342

1343
  int64_t startUs3 = taosGetTimestampUs();
×
1344
  pStmt->stat.bindDataUs2 += startUs3 - startUs2;
×
1345

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

1348
  if (colIdx < 0) {
×
1349
    if (pStmt->sql.stbInterlaceMode) {
×
1350
      (*pDataBlock)->pData->flags = 0;
×
1351
      code =
1352
          qBindStmtStbColsValue(*pDataBlock, pCols, bind, pStmt->exec.pRequest->msgBuf, pStmt->exec.pRequest->msgBufLen,
×
1353
                                &pStmt->sql.siInfo.pTSchema, pStmt->sql.pBindInfo, pStmt->taos->optionInfo.charsetCxt);
×
1354
    } else {
1355
      code = qBindStmtColsValue(*pDataBlock, pCols, bind, pStmt->exec.pRequest->msgBuf, pStmt->exec.pRequest->msgBufLen,
×
1356
                                pStmt->taos->optionInfo.charsetCxt);
×
1357
    }
1358

1359
    if (code) {
×
1360
      tscError("qBindStmtColsValue failed, error:%s", tstrerror(code));
×
1361
      STMT_ERR_RET(code);
×
1362
    }
1363
  } else {
1364
    if (pStmt->sql.stbInterlaceMode) {
×
1365
      tscError("bind single column not allowed in stb insert mode");
×
1366
      STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR);
×
1367
    }
1368

1369
    if (colIdx != (pStmt->bInfo.sBindLastIdx + 1) && colIdx != 0) {
×
1370
      tscError("bind column index not in sequence");
×
1371
      STMT_ERR_RET(TSDB_CODE_APP_ERROR);
×
1372
    }
1373

1374
    pStmt->bInfo.sBindLastIdx = colIdx;
×
1375

1376
    if (0 == colIdx) {
×
1377
      pStmt->bInfo.sBindRowNum = bind->num;
×
1378
    }
1379

1380
    code =
1381
        qBindStmtSingleColValue(*pDataBlock, pCols, bind, pStmt->exec.pRequest->msgBuf, pStmt->exec.pRequest->msgBufLen,
×
1382
                                colIdx, pStmt->bInfo.sBindRowNum, pStmt->taos->optionInfo.charsetCxt);
×
1383
    if (code) {
×
1384
      tscError("qBindStmtSingleColValue failed, error:%s", tstrerror(code));
×
1385
      STMT_ERR_RET(code);
×
1386
    }
1387
  }
1388

1389
  int64_t startUs4 = taosGetTimestampUs();
×
1390
  pStmt->stat.bindDataUs3 += startUs4 - startUs3;
×
1391

1392
  if (pStmt->sql.stbInterlaceMode) {
×
1393
    STMT_ERR_RET(stmtAppendTablePostHandle(pStmt, param));
×
1394
  }
1395

1396
  pStmt->stat.bindDataUs4 += taosGetTimestampUs() - startUs4;
×
1397

1398
  return TSDB_CODE_SUCCESS;
×
1399
}
1400

1401
int stmtAddBatch(TAOS_STMT* stmt) {
×
1402
  STscStmt* pStmt = (STscStmt*)stmt;
×
1403

1404
  int64_t startUs = taosGetTimestampUs();
×
1405

1406
  STMT_DLOG_E("start to add batch");
×
1407

1408
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
×
1409
    return pStmt->errCode;
×
1410
  }
1411

1412
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_ADD_BATCH));
×
1413

1414
  if (pStmt->sql.stbInterlaceMode) {
×
1415
    int64_t startUs2 = taosGetTimestampUs();
×
1416
    pStmt->stat.addBatchUs += startUs2 - startUs;
×
1417

1418
    pStmt->sql.siInfo.tableColsReady = false;
×
1419

1420
    SStmtQNode* param = NULL;
×
1421
    STMT_ERR_RET(stmtAllocQNodeFromBuf(&pStmt->sql.siInfo.tbBuf, (void**)&param));
×
1422
    param->restoreTbCols = true;
×
1423
    param->next = NULL;
×
1424

1425
    stmtEnqueue(pStmt, param);
×
1426

1427
    return TSDB_CODE_SUCCESS;
×
1428
  }
1429

1430
  STMT_ERR_RET(stmtCacheBlock(pStmt));
×
1431

1432
  return TSDB_CODE_SUCCESS;
×
1433
}
1434
/*
1435
int stmtUpdateTableUid(STscStmt* pStmt, SSubmitRsp* pRsp) {
1436
  tscDebug("stmt start to update tbUid, blockNum:%d", pRsp->nBlocks);
1437

1438
  int32_t code = 0;
1439
  int32_t finalCode = 0;
1440
  size_t  keyLen = 0;
1441
  void*   pIter = taosHashIterate(pStmt->exec.pBlockHash, NULL);
1442
  while (pIter) {
1443
    STableDataCxt* pBlock = *(STableDataCxt**)pIter;
1444
    char*          key = taosHashGetKey(pIter, &keyLen);
1445

1446
    STableMeta* pMeta = qGetTableMetaInDataBlock(pBlock);
1447
    if (pMeta->uid) {
1448
      pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
1449
      continue;
1450
    }
1451

1452
    SSubmitBlkRsp* blkRsp = NULL;
1453
    int32_t        i = 0;
1454
    for (; i < pRsp->nBlocks; ++i) {
1455
      blkRsp = pRsp->pBlocks + i;
1456
      if (strlen(blkRsp->tblFName) != keyLen) {
1457
        continue;
1458
      }
1459

1460
      if (strncmp(blkRsp->tblFName, key, keyLen)) {
1461
        continue;
1462
      }
1463

1464
      break;
1465
    }
1466

1467
    if (i < pRsp->nBlocks) {
1468
      tscDebug("auto created table %s uid updated from %" PRIx64 " to %" PRIx64, blkRsp->tblFName, pMeta->uid,
1469
               blkRsp->uid);
1470

1471
      pMeta->uid = blkRsp->uid;
1472
      pStmt->bInfo.tbUid = blkRsp->uid;
1473
    } else {
1474
      tscDebug("table %s not found in submit rsp, will update from catalog", pStmt->bInfo.tbFName);
1475
      if (NULL == pStmt->pCatalog) {
1476
        code = catalogGetHandle(pStmt->taos->pAppInfo->clusterId, &pStmt->pCatalog);
1477
        if (code) {
1478
          pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
1479
          finalCode = code;
1480
          continue;
1481
        }
1482
      }
1483

1484
      code = stmtCreateRequest(pStmt);
1485
      if (code) {
1486
        pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
1487
        finalCode = code;
1488
        continue;
1489
      }
1490

1491
      STableMeta*      pTableMeta = NULL;
1492
      SRequestConnInfo conn = {.pTrans = pStmt->taos->pAppInfo->pTransporter,
1493
                               .requestId = pStmt->exec.pRequest->requestId,
1494
                               .requestObjRefId = pStmt->exec.pRequest->self,
1495
                               .mgmtEps = getEpSet_s(&pStmt->taos->pAppInfo->mgmtEp)};
1496
      code = catalogGetTableMeta(pStmt->pCatalog, &conn, &pStmt->bInfo.sname, &pTableMeta);
1497

1498
      pStmt->stat.ctgGetTbMetaNum++;
1499

1500
      taos_free_result(pStmt->exec.pRequest);
1501
      pStmt->exec.pRequest = NULL;
1502

1503
      if (code || NULL == pTableMeta) {
1504
        pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
1505
        finalCode = code;
1506
        taosMemoryFree(pTableMeta);
1507
        continue;
1508
      }
1509

1510
      pMeta->uid = pTableMeta->uid;
1511
      pStmt->bInfo.tbUid = pTableMeta->uid;
1512
      taosMemoryFree(pTableMeta);
1513
    }
1514

1515
    pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
1516
  }
1517

1518
  return finalCode;
1519
}
1520
*/
1521

1522
/*
1523
int stmtStaticModeExec(TAOS_STMT* stmt) {
1524
  STscStmt*   pStmt = (STscStmt*)stmt;
1525
  int32_t     code = 0;
1526
  SSubmitRsp* pRsp = NULL;
1527
  if (pStmt->sql.staticMode) {
1528
    return TSDB_CODE_TSC_STMT_API_ERROR;
1529
  }
1530

1531
  STMT_DLOG_E("start to exec");
1532

1533
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_EXECUTE));
1534

1535
  STMT_ERR_RET(qBuildStmtOutputFromTbList(pStmt->sql.pQuery, pStmt->sql.pVgHash, pStmt->exec.pTbBlkList,
1536
pStmt->exec.pCurrBlock, pStmt->exec.tbBlkNum));
1537

1538
  launchQueryImpl(pStmt->exec.pRequest, pStmt->sql.pQuery, true, NULL);
1539

1540
  if (pStmt->exec.pRequest->code && NEED_CLIENT_HANDLE_ERROR(pStmt->exec.pRequest->code)) {
1541
    code = refreshMeta(pStmt->exec.pRequest->pTscObj, pStmt->exec.pRequest);
1542
    if (code) {
1543
      pStmt->exec.pRequest->code = code;
1544
    } else {
1545
      tFreeSSubmitRsp(pRsp);
1546
      STMT_ERR_RET(stmtResetStmt(pStmt));
1547
      STMT_ERR_RET(TSDB_CODE_NEED_RETRY);
1548
    }
1549
  }
1550

1551
  STMT_ERR_JRET(pStmt->exec.pRequest->code);
1552

1553
  pStmt->exec.affectedRows = taos_affected_rows(pStmt->exec.pRequest);
1554
  pStmt->affectedRows += pStmt->exec.affectedRows;
1555

1556
_return:
1557

1558
  stmtCleanExecInfo(pStmt, (code ? false : true), false);
1559

1560
  tFreeSSubmitRsp(pRsp);
1561

1562
  ++pStmt->sql.runTimes;
1563

1564
  STMT_RET(code);
1565
}
1566
*/
1567

1568
int stmtExec(TAOS_STMT* stmt) {
×
1569
  STscStmt*   pStmt = (STscStmt*)stmt;
×
1570
  int32_t     code = 0;
×
1571
  SSubmitRsp* pRsp = NULL;
×
1572

1573
  int64_t startUs = taosGetTimestampUs();
×
1574

1575
  STMT_DLOG_E("start to exec");
×
1576

1577
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
×
1578
    return pStmt->errCode;
×
1579
  }
1580

1581
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_EXECUTE));
×
1582

1583
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
×
1584
    launchQueryImpl(pStmt->exec.pRequest, pStmt->sql.pQuery, true, NULL);
×
1585
  } else {
1586
    if (pStmt->sql.stbInterlaceMode) {
×
1587
      int64_t startTs = taosGetTimestampUs();
×
1588
      while (atomic_load_64(&pStmt->sql.siInfo.tbRemainNum)) {
×
1589
        taosUsleep(1);
×
1590
      }
1591
      pStmt->stat.execWaitUs += taosGetTimestampUs() - startTs;
×
1592

1593
      STMT_ERR_RET(qBuildStmtFinOutput(pStmt->sql.pQuery, pStmt->sql.pVgHash, pStmt->sql.siInfo.pVgroupList));
×
1594
      taosHashCleanup(pStmt->sql.siInfo.pVgroupHash);
×
1595
      pStmt->sql.siInfo.pVgroupHash = NULL;
×
1596
      pStmt->sql.siInfo.pVgroupList = NULL;
×
1597
    } else {
1598
      tDestroySubmitTbData(pStmt->exec.pCurrTbData, TSDB_MSG_FLG_ENCODE);
×
1599
      taosMemoryFreeClear(pStmt->exec.pCurrTbData);
×
1600

1601
      STMT_ERR_RET(qCloneCurrentTbData(pStmt->exec.pCurrBlock, &pStmt->exec.pCurrTbData));
×
1602

1603
      STMT_ERR_RET(qBuildStmtOutput(pStmt->sql.pQuery, pStmt->sql.pVgHash, pStmt->exec.pBlockHash));
×
1604
    }
1605

1606
    launchQueryImpl(pStmt->exec.pRequest, pStmt->sql.pQuery, true, NULL);
×
1607
  }
1608

1609
  if (pStmt->exec.pRequest->code && NEED_CLIENT_HANDLE_ERROR(pStmt->exec.pRequest->code)) {
×
1610
    code = refreshMeta(pStmt->exec.pRequest->pTscObj, pStmt->exec.pRequest);
×
1611
    if (code) {
×
1612
      pStmt->exec.pRequest->code = code;
×
1613
    } else {
1614
      tFreeSSubmitRsp(pRsp);
×
1615
      STMT_ERR_RET(stmtResetStmt(pStmt));
×
1616
      STMT_ERR_RET(TSDB_CODE_NEED_RETRY);
×
1617
    }
1618
  }
1619

1620
  STMT_ERR_JRET(pStmt->exec.pRequest->code);
×
1621

1622
  pStmt->exec.affectedRows = taos_affected_rows(pStmt->exec.pRequest);
×
1623
  pStmt->affectedRows += pStmt->exec.affectedRows;
×
1624

1625
_return:
×
1626

1627
  while (0 == atomic_load_8((int8_t*)&pStmt->sql.siInfo.tableColsReady)) {
×
1628
    taosUsleep(1);
×
1629
  }
1630

1631
  STMT_ERR_RET(stmtCleanExecInfo(pStmt, (code ? false : true), false));
×
1632

1633
  tFreeSSubmitRsp(pRsp);
×
1634

1635
  ++pStmt->sql.runTimes;
×
1636

1637
  int64_t startUs2 = taosGetTimestampUs();
×
1638
  pStmt->stat.execUseUs += startUs2 - startUs;
×
1639

1640
  STMT_RET(code);
×
1641
}
1642

1643
int stmtClose(TAOS_STMT* stmt) {
×
1644
  STscStmt* pStmt = (STscStmt*)stmt;
×
1645

1646
  STMT_DLOG_E("start to free stmt");
×
1647

1648
  if (pStmt->bindThreadInUse) {
×
1649
    pStmt->queue.stopQueue = true;
×
1650

1651
    (void)taosThreadMutexLock(&pStmt->queue.mutex);
×
1652
    (void)atomic_add_fetch_64(&pStmt->queue.qRemainNum, 1);
×
1653
    (void)taosThreadCondSignal(&(pStmt->queue.waitCond));
×
1654
    (void)taosThreadMutexUnlock(&pStmt->queue.mutex);
×
1655

1656
    (void)taosThreadJoin(pStmt->bindThread, NULL);
×
1657
    pStmt->bindThreadInUse = false;
×
1658

1659
    (void)taosThreadCondDestroy(&pStmt->queue.waitCond);
×
1660
    (void)taosThreadMutexDestroy(&pStmt->queue.mutex);
×
1661
  }
1662

1663
  STMT_DLOG("stmt %p closed, stbInterlaceMode:%d, statInfo: ctgGetTbMetaNum=>%" PRId64 ", getCacheTbInfo=>%" PRId64
×
1664
            ", parseSqlNum=>%" PRId64 ", pStmt->stat.bindDataNum=>%" PRId64
1665
            ", settbnameAPI:%u, bindAPI:%u, addbatchAPI:%u, execAPI:%u"
1666
            ", setTbNameUs:%" PRId64 ", bindDataUs:%" PRId64 ",%" PRId64 ",%" PRId64 ",%" PRId64 " addBatchUs:%" PRId64
1667
            ", execWaitUs:%" PRId64 ", execUseUs:%" PRId64,
1668
            pStmt, pStmt->sql.stbInterlaceMode, pStmt->stat.ctgGetTbMetaNum, pStmt->stat.getCacheTbInfo,
1669
            pStmt->stat.parseSqlNum, pStmt->stat.bindDataNum, pStmt->seqIds[STMT_SETTBNAME], pStmt->seqIds[STMT_BIND],
1670
            pStmt->seqIds[STMT_ADD_BATCH], pStmt->seqIds[STMT_EXECUTE], pStmt->stat.setTbNameUs,
1671
            pStmt->stat.bindDataUs1, pStmt->stat.bindDataUs2, pStmt->stat.bindDataUs3, pStmt->stat.bindDataUs4,
1672
            pStmt->stat.addBatchUs, pStmt->stat.execWaitUs, pStmt->stat.execUseUs);
1673

1674
  STMT_ERR_RET(stmtCleanSQLInfo(pStmt));
×
1675
  taosMemoryFree(stmt);
×
1676

1677
  return TSDB_CODE_SUCCESS;
×
1678
}
1679

1680
const char* stmtErrstr(TAOS_STMT* stmt) {
×
1681
  STscStmt* pStmt = (STscStmt*)stmt;
×
1682

1683
  if (stmt == NULL || NULL == pStmt->exec.pRequest) {
×
1684
    return (char*)tstrerror(terrno);
×
1685
  }
1686

1687
  pStmt->exec.pRequest->code = terrno;
×
1688

1689
  return taos_errstr(pStmt->exec.pRequest);
×
1690
}
1691

1692
int stmtAffectedRows(TAOS_STMT* stmt) { return ((STscStmt*)stmt)->affectedRows; }
×
1693

1694
int stmtAffectedRowsOnce(TAOS_STMT* stmt) { return ((STscStmt*)stmt)->exec.affectedRows; }
×
1695

1696
int stmtIsInsert(TAOS_STMT* stmt, int* insert) {
×
1697
  STscStmt* pStmt = (STscStmt*)stmt;
×
1698

1699
  STMT_DLOG_E("start is insert");
×
1700

1701
  if (pStmt->sql.type) {
×
1702
    *insert = (STMT_TYPE_INSERT == pStmt->sql.type || STMT_TYPE_MULTI_INSERT == pStmt->sql.type);
×
1703
  } else {
1704
    *insert = qIsInsertValuesSql(pStmt->sql.sqlStr, pStmt->sql.sqlLen);
×
1705
  }
1706

1707
  return TSDB_CODE_SUCCESS;
×
1708
}
1709

1710
int stmtGetTagFields(TAOS_STMT* stmt, int* nums, TAOS_FIELD_E** fields) {
×
1711
  int32_t   code = 0;
×
1712
  STscStmt* pStmt = (STscStmt*)stmt;
×
1713
  int32_t   preCode = pStmt->errCode;
×
1714

1715
  STMT_DLOG_E("start to get tag fields");
×
1716

1717
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
×
1718
    return pStmt->errCode;
×
1719
  }
1720

1721
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
×
1722
    STMT_ERRI_JRET(TSDB_CODE_TSC_STMT_API_ERROR);
×
1723
  }
1724

1725
  STMT_ERRI_JRET(stmtSwitchStatus(pStmt, STMT_FETCH_FIELDS));
×
1726

1727
  if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 &&
×
1728
      STMT_TYPE_MULTI_INSERT != pStmt->sql.type) {
×
1729
    pStmt->bInfo.needParse = false;
×
1730
  }
1731

1732
  if (pStmt->exec.pRequest && STMT_TYPE_QUERY == pStmt->sql.type && pStmt->sql.runTimes) {
×
1733
    taos_free_result(pStmt->exec.pRequest);
×
1734
    pStmt->exec.pRequest = NULL;
×
1735
  }
1736

1737
  STMT_ERRI_JRET(stmtCreateRequest(pStmt));
×
1738

1739
  if (pStmt->bInfo.needParse) {
×
1740
    STMT_ERRI_JRET(stmtParseSql(pStmt));
×
1741
  }
1742

1743
  STMT_ERRI_JRET(stmtFetchTagFields(stmt, nums, fields));
×
1744

1745
_return:
×
1746
  // compatible with previous versions
1747
  if (code == TSDB_CODE_PAR_TABLE_NOT_EXIST && (pStmt->bInfo.tbNameFlag & NO_DATA_USING_CLAUSE) == 0x0) {
×
1748
    code = TSDB_CODE_TSC_STMT_TBNAME_ERROR;
×
1749
  }
1750

1751
  if (code != TSDB_CODE_SUCCESS) {
×
1752
    taos_free_result(pStmt->exec.pRequest);
×
1753
    pStmt->exec.pRequest = NULL;
×
1754
  }
1755
  pStmt->errCode = preCode;
×
1756

1757
  return code;
×
1758
}
1759

1760
int stmtGetColFields(TAOS_STMT* stmt, int* nums, TAOS_FIELD_E** fields) {
×
1761
  int32_t   code = 0;
×
1762
  STscStmt* pStmt = (STscStmt*)stmt;
×
1763
  int32_t   preCode = pStmt->errCode;
×
1764

1765
  STMT_DLOG_E("start to get col fields");
×
1766

1767
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
×
1768
    return pStmt->errCode;
×
1769
  }
1770

1771
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
×
1772
    STMT_ERRI_JRET(TSDB_CODE_TSC_STMT_API_ERROR);
×
1773
  }
1774

1775
  STMT_ERRI_JRET(stmtSwitchStatus(pStmt, STMT_FETCH_FIELDS));
×
1776

1777
  if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 &&
×
1778
      STMT_TYPE_MULTI_INSERT != pStmt->sql.type) {
×
1779
    pStmt->bInfo.needParse = false;
×
1780
  }
1781

1782
  if (pStmt->exec.pRequest && STMT_TYPE_QUERY == pStmt->sql.type && pStmt->sql.runTimes) {
×
1783
    taos_free_result(pStmt->exec.pRequest);
×
1784
    pStmt->exec.pRequest = NULL;
×
1785
    STMT_ERR_RET(stmtCreateRequest(pStmt));
×
1786
  }
1787

1788
  STMT_ERRI_JRET(stmtCreateRequest(pStmt));
×
1789

1790
  if (pStmt->bInfo.needParse) {
×
1791
    STMT_ERRI_JRET(stmtParseSql(pStmt));
×
1792
  }
1793

1794
  STMT_ERRI_JRET(stmtFetchColFields(stmt, nums, fields));
×
1795

1796
_return:
×
1797

1798
  if (code != TSDB_CODE_SUCCESS) {
×
1799
    taos_free_result(pStmt->exec.pRequest);
×
1800
    pStmt->exec.pRequest = NULL;
×
1801
  }
1802

1803
  pStmt->errCode = preCode;
×
1804

1805
  return code;
×
1806
}
1807

1808
int stmtGetParamNum(TAOS_STMT* stmt, int* nums) {
×
1809
  int       code = 0;
×
1810
  STscStmt* pStmt = (STscStmt*)stmt;
×
1811
  int32_t   preCode = pStmt->errCode;
×
1812

1813
  STMT_DLOG_E("start to get param num");
×
1814

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

1819
  STMT_ERRI_JRET(stmtSwitchStatus(pStmt, STMT_FETCH_FIELDS));
×
1820

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

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

1831
  STMT_ERRI_JRET(stmtCreateRequest(pStmt));
×
1832

1833
  if (pStmt->bInfo.needParse) {
×
1834
    STMT_ERRI_JRET(stmtParseSql(pStmt));
×
1835
  }
1836

1837
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
×
1838
    *nums = taosArrayGetSize(pStmt->sql.pQuery->pPlaceholderValues);
×
1839
  } else {
1840
    STMT_ERRI_JRET(stmtFetchColFields(stmt, nums, NULL));
×
1841
  }
1842

1843
_return:
×
1844

1845
  if (code != TSDB_CODE_SUCCESS) {
×
1846
    taos_free_result(pStmt->exec.pRequest);
×
1847
    pStmt->exec.pRequest = NULL;
×
1848
  }
1849

1850
  pStmt->errCode = preCode;
×
1851

1852
  return code;
×
1853
}
1854

1855
int stmtGetParam(TAOS_STMT* stmt, int idx, int* type, int* bytes) {
×
1856
  int       code = 0;
×
1857
  STscStmt* pStmt = (STscStmt*)stmt;
×
1858
  int32_t   preCode = pStmt->errCode;
×
1859

1860
  STMT_DLOG_E("start to get param");
×
1861

1862
  if (pStmt->errCode != TSDB_CODE_SUCCESS) {
×
1863
    return pStmt->errCode;
×
1864
  }
1865

1866
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
×
1867
    STMT_ERRI_JRET(TSDB_CODE_TSC_STMT_API_ERROR);
×
1868
  }
1869

1870
  STMT_ERRI_JRET(stmtSwitchStatus(pStmt, STMT_FETCH_FIELDS));
×
1871

1872
  if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 &&
×
1873
      STMT_TYPE_MULTI_INSERT != pStmt->sql.type) {
×
1874
    pStmt->bInfo.needParse = false;
×
1875
  }
1876

1877
  if (pStmt->exec.pRequest && STMT_TYPE_QUERY == pStmt->sql.type && pStmt->sql.runTimes) {
×
1878
    taos_free_result(pStmt->exec.pRequest);
×
1879
    pStmt->exec.pRequest = NULL;
×
1880
  }
1881

1882
  STMT_ERRI_JRET(stmtCreateRequest(pStmt));
×
1883

1884
  if (pStmt->bInfo.needParse) {
×
1885
    STMT_ERRI_JRET(stmtParseSql(pStmt));
×
1886
  }
1887

1888
  int32_t       nums = 0;
×
1889
  TAOS_FIELD_E* pField = NULL;
×
1890
  STMT_ERRI_JRET(stmtFetchColFields(stmt, &nums, &pField));
×
1891
  if (idx >= nums) {
×
1892
    tscError("idx %d is too big", idx);
×
1893
    STMT_ERRI_JRET(TSDB_CODE_INVALID_PARA);
×
1894
  }
1895

1896
  *type = pField[idx].type;
×
1897
  *bytes = calcSchemaBytesFromTypeBytes(pField[idx].type, pField[idx].bytes, true);
×
1898

1899
_return:
×
1900

1901
  if (code != TSDB_CODE_SUCCESS) {
×
1902
    taos_free_result(pStmt->exec.pRequest);
×
1903
    pStmt->exec.pRequest = NULL;
×
1904
  }
1905

1906
  taosMemoryFree(pField);
×
1907
  pStmt->errCode = preCode;
×
1908

1909
  return code;
×
1910
}
1911

1912
TAOS_RES* stmtUseResult(TAOS_STMT* stmt) {
×
1913
  STscStmt* pStmt = (STscStmt*)stmt;
×
1914

1915
  STMT_DLOG_E("start to use result");
×
1916

1917
  if (STMT_TYPE_QUERY != pStmt->sql.type) {
×
1918
    tscError("useResult only for query statement");
×
1919
    return NULL;
×
1920
  }
1921

1922
  return pStmt->exec.pRequest;
×
1923
}
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